Add aircraft fusion engine + fused live state (M3.1)

[kninetimmy]

What & why

First slice of M3 (Network fusion) — the fusion engine, the keystone the rest of M3 (network ADS-B adapter, APRS-IS/APRS fusion, AIS, filters, TOI) builds on. Until now LiveState keyed tracks purely by id, and adapters set id == correlation_key, so a network observation of the same airframe would overwrite the local one. This makes same-identity local-RF + network observations fuse into one track with full provenance (FUSION-FR-001…007, PRD §11.4/§15).

Scope is deliberately fusion-only: the real network ADS-B adapter is the next slice (M3.2). The demo source simulates the network contributor so the whole path is exercised no-hardware.

How it works

• src/aether/fusion/ (new), pure & deterministic — now is always injected (FUSION-FR-007), no hidden clock, no per-source if/elif:
  • freshness.py — per-source windows (PRD §15.4: local 5/30/60s, network 15/60/120s) measured on observed_at (§8.4); future-skew cap so a far-future timestamp can't pin a group in memory.
  • precedence.py — PRD §15.2 ladder (fresh local > fresh network > stale local > stale network > expired) with deterministic tie-breaks; network fills fields local lacks (FR-003); geometry chosen whole, never averaged (§15.5).
  • engine.py — FusionEngine/FusionGroup: one group per correlation_key, merged provenance, fusion metadata in attributes["fusion"] (active source, per-field source, last-local-RF, contributors, conflicts).
• state/live.py — correlation-keyed tracks fuse (fused id == correlation_key); None-key tracks stay keyed by id (FR-006, no proximity merge). expire() removes a fused track only when all contributors expire and re-upserts on a LOCAL→NET handoff (FR-004). One source record → one StateChange (websocket contract preserved).
• hub.py / main.py — clock read once at the edge; a 1s background expiry sweep, exception-isolated.
• demo_publisher.py — adds a demo-net leg + 4 scenario aircraft (fused / handoff / local-only / network-only) so the no-hardware demo shows fusion.
• Frontend — fused provenance surfaced in TrackList; display-only All / Local-RF / Network filter (store + pure visibleTracks selector) that never affects ingestion.
• No schema_version bump — all fusion metadata lives in attributes["fusion"].

Failure isolation

• A fused-record validation error degrades to the raw record rekeyed to its correlation_key (no stranded client ghost track).
• A poison fusion group is dropped rather than aborting the whole expiry sweep (so None-key tracks/features still age out).

These two + a memory-pin via future timestamps were found by the review pass and fixed before this PR.

Tests / verification

• scripts/check.sh green: ruff + ruff-format + mypy strict + 160 pytest + receive-only Dire Wolf tripwire.
• Frontend green: eslint + tsc -b + 31 vitest + vite build.
• New: fusion freshness/precedence/engine units, live-state fusion + expiry, demo-ws integration, frontend selectors + TrackList.

Reviewer notes / known limitations (carried as follow-ups, not blockers)

• Demo LOCAL→NET handoff can't be observed over the 0.05s-tick ws integration test (real 60s window); the strict flip is asserted at the engine-unit level instead.
• The conflicts diagnostic uses a squared-degree geometry epsilon, not great-circle distance (diagnostic-only field).
• The Local-RF filter counts a stale-but-not-expired local contributor as local (intentional; may revisit for a stricter live-local mode in M3.4).

How this was built

Authored via an ultracode workflow: 3-lens design panel → synthesized spec → implementation → 5-dimension adversarial review → fix pass. Both gates re-run from ground truth before this PR.

🤖 Generated with Claude Code

[gemini-code-assist]

Code Review

This pull request implements in-process aircraft fusion (M3.1), which merges same-identity local-RF and network observations into a single track based on a shared correlation key. It introduces a stateful backend fusion engine, integrates it into the live state and periodic expiry tasks, and updates the frontend to support provenance filtering and display fusion metadata badges. The review feedback highlights several opportunities to improve robustness, specifically recommending timezone normalization of incoming timestamps to prevent timezone-naive/aware comparison crashes, and suggesting defensive checks for potentially missing provenance and contributor arrays to avoid runtime TypeErrors.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Normalize observed_at and received_at to UTC if they are timezone-naive. This prevents potential TypeError crashes when comparing naive and aware datetimes (e.g., if existing.observed_at is aware but record.observed_at is naive), avoids timezone-shifted comparisons in _sort_key using .timestamp(), and ensures consistent sorting in _merge_attributes.

    def update(self, record: TrackRecord) -> bool:
        """Add/replace this source's contribution. Returns False if discarded (out-of-order).

        An older observation from a source we already have a newer one for is
        dropped for fusion (the track must not jump backwards); an equal or newer
        ``observed_at`` replaces it. Duplicates are therefore idempotent.
        """
        local_rf = _is_local_rf(record)
        observed_at = record.observed_at
        if observed_at.tzinfo is None:
            observed_at = observed_at.replace(tzinfo=UTC)
        received_at = record.received_at
        if received_at.tzinfo is None:
            received_at = received_at.replace(tzinfo=UTC)

        existing = self.contributors.get(record.source)
        if existing is not None and observed_at < existing.observed_at:
            return False
        self.contributors[record.source] = Contributor(
            source=record.source,
            record=record,
            local_rf=local_rf,
            observed_at=observed_at,
            received_at=received_at,
        )
        if local_rf and (
            self.last_local_rf_at is None or observed_at > self.last_local_rf_at
        ):
            self.last_local_rf_at = observed_at
        self._dirty = True
        return True

[gemini-code-assist]

Import UTC from datetime to allow timezone normalization of incoming track record timestamps.

Suggested change

	import logging
	from dataclasses import dataclass, field
	from datetime import datetime, timedelta
	from typing import Any
	import logging
	from dataclasses import dataclass, field
	from datetime import UTC, datetime, timedelta
	from typing import Any

[gemini-code-assist]

Add a defensive check for contrib.record.provenance being None or empty. If provenance is missing or set to None on the incoming record, iterating over it directly will raise a TypeError and crash the fusion process.

Suggested change

	entries: list[Provenance] = []
	for source in sorted(self.contributors):
	contrib = self.contributors[source]
	own = next((p for p in contrib.record.provenance if p.source == source), None)
	entries: list[Provenance] = []
	for source in sorted(self.contributors):
	contrib = self.contributors[source]
	own = next((p for p in (contrib.record.provenance or []) if p.source == source), None)

[gemini-code-assist]

Add a defensive check for meta.contributors being null or undefined. If a malformed track record is received where contributors is missing, calling .map() directly will throw a TypeError and crash the React render tree.

Suggested change

	function fusionTooltip(track: TrackRecord): string | undefined {
	const meta = fusionMeta(track);
	if (!meta) return undefined;
	const sources = meta.contributors.map((c) => c.source).join(", ");
	const lastLocal = meta.last_local_rf_at ?? "never";
	return `Sources: ${sources}\nActive: ${meta.active_source}\nLast local RF: ${lastLocal}`;
	}
	function fusionTooltip(track: TrackRecord): string | undefined {
	const meta = fusionMeta(track);
	if (!meta) return undefined;
	const sources = (meta.contributors ?? []).map((c) => c.source).join(", ");
	const lastLocal = meta.last_local_rf_at ?? "never";
	return `Sources: ${sources}\nActive: ${meta.active_source}\nLast local RF: ${lastLocal}`;
	}

[kninetimmy]

Thanks for the review. Declining all four — each is already defended at the schema boundary:

• Normalize observed_at/received_at to UTC + import UTC (engine.py): every UtcDatetime field runs _ensure_utc (src/aether/schema/common.py), which raises on naive input and astimezone(UTC)-normalizes aware input. The engine can only ever see aware-UTC datetimes, so naive/aware comparison crashes are impossible. Adding tz handling in the engine would be dead code that masks a schema-contract violation the PRD wants to fail loudly (§14.1). (Note: there's no _sort_key/.timestamp() in engine.py.)
• contrib.record.provenance or [] (engine.py:253): provenance: list[Provenance] = Field(default_factory=list) with extra="forbid" — never None on a validated TrackRecord.
• meta.contributors ?? [] (TrackList.tsx): _fusion_block always emits contributors as a list; the guard defends against our own backend. Harmless but unnecessary; leaving it out.

CI green, no functional changes needed.