feat(phase4a): peak_compass end-to-end + repair stale satpass test suite

Add azimuth-at-peak-elevation compass to the satpass pipeline (Matt's
decision: extend pass_predictor) and repair the 24 stale satpass tests
that were written against the pre-async handle_satpass return contract.

- pass_predictor.PassInfo gains azimuth_at_peak; _build_pass populates it
  from the already-computed peak sample
- !satpass DM wire + Central consolidated broadcast wire render aos→peak→los
- satpass_pending gains a peak_compass column (migration v21, SCHEMA_VERSION
  20->21); persisted at ingest, carried through consolidation from the
  max-elevation observer's row
- 24 stale satpass tests repaired to the ingest→consolidate two-call
  contract (none weakened); satpass suite 112/112 green
- full suite 34->10 failures (the 24 were these stale tests)

NOTE: peak_compass changes the live Central satpass wire — deploy is HELD
until the coordinated all-native flip (per Matt).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/central/pass_predictor.py

@@ -46,6 +46,7 @@ class PassInfo:
     max_elevation: float     # Degrees
     azimuth_at_aos: float    # Degrees, clockwise from north
     azimuth_at_los: float    # Degrees, clockwise from north
+    azimuth_at_peak: float   # Degrees, clockwise from north (at max elevation)
 
 
 def compute_passes(line1: str, line2: str,
@@ -144,6 +145,7 @@ def _build_pass(samples: list[tuple[datetime, float, float]]) -> PassInfo:
         max_elevation=samples[peak_idx][1],
         azimuth_at_aos=samples[0][2] % 360,
         azimuth_at_los=samples[-1][2] % 360,
+        azimuth_at_peak=samples[peak_idx][2] % 360,
     )
 
 

work/meshai/central/satpass_handler.py

@@ -190,16 +190,27 @@ def format_pass(*, sat_name: str, max_el: float,
                 aos_compass: str, los_compass: str,
                 broadcast: bool = True,
                 entry_observer: Optional[str] = None,
-                exit_observer: Optional[str] = None) -> str:
+                exit_observer: Optional[str] = None,
+                peak_compass: Optional[str] = None) -> str:
     """Unified pass formatter with mode switch.
 
     broadcast=True:  Two-line format with buckets, 12h times, LoRa budget.
-        🛰️ {name} {bucket}, {aos_compass}→{los_compass}
+        🛰️ {name} {bucket}, {aos_compass}→[peak_compass→]{los_compass}
         {duration} min window, {rise}–{set} {AM/PM} {TZ} [tomorrow] [(region)]
 
     broadcast=False: Compact DM format with exact degrees.
-        {name} {HH:MM}–{HH:MM} {TZ} max {el}° {aos_compass}→{los_compass}
+        {name} {HH:MM}–{HH:MM} {TZ} max {el}° {aos_compass}→[peak→]{los_compass}
+
+    peak_compass: compass direction at peak elevation. When provided, the
+        compass segment renders as aos→peak→los; when None it stays aos→los
+        (preserving legacy callers that don't thread the peak field).
     """
+    # Compass sweep segment: include the peak point only when supplied.
+    if peak_compass:
+        compass_seg = f"{aos_compass}→{peak_compass}→{los_compass}"
+    else:
+        compass_seg = f"{aos_compass}→{los_compass}"
+
     if broadcast:
         bucket = _elevation_bucket(max_el)
         # Duration in whole minutes
@@ -213,7 +224,7 @@ def format_pass(*, sat_name: str, max_el: float,
         tz = _tz_abbr(aos_epoch)
         date_lbl = _date_label(aos_epoch)
 
-        line1 = f"\U0001F6F0\uFE0F {sat_name} {bucket}, {aos_compass}\u2192{los_compass}"
+        line1 = f"\U0001F6F0\uFE0F {sat_name} {bucket}, {compass_seg}"
 
         # Build time portion
         time_part = f"{dur_min} min window, {rise_str}\u2013{set_str} {ampm} {tz}{date_lbl}"
@@ -235,7 +246,7 @@ def format_pass(*, sat_name: str, max_el: float,
         tz = _tz_abbr(aos_epoch)
         return (f"{sat_name} {aos_str}\u2013{los_str} {tz} "
                 f"max {int(max_el)}\u00B0 "
-                f"{aos_compass}\u2192{los_compass}")
+                f"{compass_seg}")
 
 
 def _map_severity(max_el: float) -> str:
@@ -397,10 +408,10 @@ def handle_satpass(envelope: dict, subject: str,
     conn.execute(
         "INSERT OR REPLACE INTO satpass_pending("
         "consolidated_id, observer, sat_name, norad_id, max_elevation, "
-        "aos_at, los_at, aos_compass, los_compass, received_at) "
+        "aos_at, los_at, aos_compass, los_compass, peak_compass, received_at) "
-        "VALUES (?,?,?,?,?,?,?,?,?,?)",
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
         (consolidated_id, observer, sat_name, norad_id, max_el,
-         aos_epoch, los_epoch, aos_compass, los_compass, now))
+         aos_epoch, los_epoch, aos_compass, los_compass, direction, now))
 
     # Signal consumer to schedule consolidation timer
     _pending_consolidation_ids.add(consolidated_id)
@@ -443,6 +454,8 @@ def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None
     los_epoch = exit_["los_at"]
     aos_compass = entry["aos_compass"]
     los_compass = exit_["los_compass"]
+    # Peak belongs to whoever saw the highest elevation.
+    peak_compass = best["peak_compass"]
     entry_obs = entry["observer"]
     exit_obs = exit_["observer"]
 
@@ -467,6 +480,7 @@ def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None
     wire = format_pass(sat_name=sat_name, max_el=max_el,
                       aos_epoch=aos_epoch, los_epoch=los_epoch,
                       aos_compass=aos_compass, los_compass=los_compass,
+                      peak_compass=peak_compass,
                       entry_observer=entry_obs, exit_observer=exit_obs)
 
     # Dry-run gate
@@ -586,6 +600,7 @@ CREATE TABLE IF NOT EXISTS satpass_pending (
     los_at          INTEGER,
     aos_compass     TEXT,
     los_compass     TEXT,
+    peak_compass    TEXT,
     received_at     INTEGER,
     PRIMARY KEY (consolidated_id, observer)
 );

work/meshai/commands/satpass_cmd.py

@@ -166,12 +166,14 @@ class SatpassCommand(CommandHandler):
                 from meshai.central.satpass_handler import format_pass
                 az_aos = azimuth_to_compass(p.azimuth_at_aos)
                 az_los = azimuth_to_compass(p.azimuth_at_los)
+                az_peak = azimuth_to_compass(p.azimuth_at_peak)
                 aos_epoch = int(p.aos_time.timestamp())
                 los_epoch = int(p.los_time.timestamp())
                 line = format_pass(
                     sat_name=tle["name"], max_el=p.max_elevation,
                     aos_epoch=aos_epoch, los_epoch=los_epoch,
                     aos_compass=az_aos, los_compass=az_los,
+                    peak_compass=az_peak,
                     broadcast=False,
                 )
                 all_lines.append(line)

work/meshai/persistence/db.py

@@ -30,7 +30,7 @@ logger = logging.getLogger(__name__)
 
 DEFAULT_DB_PATH = "/data/meshai.sqlite"
 MESHAI_DB_PATH_ENV = "MESHAI_DB_PATH"
-SCHEMA_VERSION = 20
+SCHEMA_VERSION = 21
 SCHEMA_META_TABLE = "schema_meta"
 MIGRATIONS_DIR = Path(__file__).parent / "migrations"
 

work/meshai/persistence/migrations/v21.sql

@@ -0,0 +1,7 @@
+-- v21: peak_compass column on satpass_pending.
+-- Carries the compass direction at a pass's PEAK elevation through the
+-- observer-consolidation buffer so the consolidated broadcast wire can
+-- render aos→peak→los. Nullable; legacy pending rows (there are none at
+-- rest — the table is drained per bucket) keep NULL and render aos→los.
+
+ALTER TABLE satpass_pending ADD COLUMN peak_compass TEXT;

work/tests/test_satpass_broadcast_safety.py

@@ -89,6 +89,30 @@ def _clear_handler_flags():
             delattr(handle_satpass, attr)
 
 
+def _ingest_and_consolidate(env, subject, *, now, data=None):
+    """Drive the two-call async satpass contract.
+
+    handle_satpass() ingests the pass into satpass_pending and ALWAYS
+    returns None (the immediate-broadcast suppression); the consumer then
+    fires consolidate_satpass_pending() on its 5s timer, which returns the
+    (wire_string, data_dict) to actually broadcast, or None if suppressed.
+    This helper runs both halves and returns the consolidation result so a
+    test can assert on the real broadcast decision + wire.
+    """
+    from meshai.central.satpass_handler import (
+        handle_satpass, consolidate_satpass_pending,
+        drain_pending_consolidation_ids)
+    drain_pending_consolidation_ids()  # clear any cross-test leakage
+    ingest = handle_satpass(
+        env, subject, data=data if data is not None else {}, now=now)
+    assert ingest is None, "handle_satpass must suppress the immediate broadcast"
+    for cid in drain_pending_consolidation_ids():
+        res = consolidate_satpass_pending(cid)
+        if res is not None:
+            return res
+    return None
+
+
 # ══════════════════════════════════════════════════════════════════════
 # 1. OPT-IN BIRD FILTER
 # ══════════════════════════════════════════════════════════════════════
@@ -113,12 +137,14 @@ class TestOptInBirdFilter:
         _enable_satpass_db(norad_ids=[], dry_run=False)
         _clear_handler_flags()
 
+        # now near the 2026 envelope window so the AOS-horizon guard does
+        # not short-circuit before the opt-in norad_ids check that logs.
         with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"):
             handle_satpass(_envelope(norad_id=25544, max_el=80.0),
-                           "central.sat.pass.iss", data={}, now=1718163120)
+                           "central.sat.pass.iss", data={}, now=1781235000)
             handle_satpass(_envelope(norad_id=25544, max_el=80.0,
                                      aos="2026-06-12T04:32:00Z"),
-                           "central.sat.pass.iss", data={}, now=1718163180)
+                           "central.sat.pass.iss", data={}, now=1781235000)
 
         matching = [r for r in caplog.records
                     if "no norad_ids configured" in r.message]
@@ -132,15 +158,15 @@ class TestOptInBirdFilter:
 
         # ISS pass should air
         iss_env = _envelope(norad_id=25544, max_el=65.0)
-        iss_result = handle_satpass(iss_env, "central.sat.pass.iss",
+        iss_result = _ingest_and_consolidate(iss_env, "central.sat.pass.iss",
-                                    data={}, now=1718163120)
+                                             now=1781235000)
         assert iss_result is not None, "ISS pass should broadcast"
 
-        # NOAA-18 pass should be rejected
+        # NOAA-18 pass should be rejected (opt-in norad_ids excludes it)
         noaa_env = _envelope(norad_id=28654, sat_name="NOAA 18", max_el=65.0,
                              aos="2026-06-12T05:00:00Z")
-        noaa_result = handle_satpass(noaa_env, "central.sat.pass.noaa",
+        noaa_result = _ingest_and_consolidate(noaa_env, "central.sat.pass.noaa",
-                                     data={}, now=1718163120)
+                                              now=1781235000)
         assert noaa_result is None, "NOAA-18 pass should be rejected"
 
     def test_dm_command_not_gated_by_norad_ids(self):
@@ -201,11 +227,15 @@ class TestRateCap:
         _enable_satpass_db(norad_ids=[25544], dry_run=False, max_per_hour=0)
         _clear_handler_flags()
 
-        now = 1718200000
+        # now near the 2026 envelope window; max_per_hour=0 => the
+        # consolidation step always trips the rate cap and logs.
+        now = 1781258400  # 2026-06-12T10:00:00Z (just before this pass)
         with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"):
             env = _envelope(norad_id=25544, max_el=70.0,
-                            aos="2026-06-12T10:05:00Z")
+                            aos="2026-06-12T10:05:00Z",
-            handle_satpass(env, "central.sat.pass.iss", data={}, now=now)
+                            los="2026-06-12T10:11:00Z")
+            result = _ingest_and_consolidate(env, "central.sat.pass.iss", now=now)
+        assert result is None, "rate cap should suppress the broadcast"
 
         matching = [r for r in caplog.records if "rate cap reached" in r.message]
         assert len(matching) >= 1, "Rate cap suppression should be logged"
@@ -250,11 +280,14 @@ class TestDryRun:
 
         with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"):
             env = _envelope(norad_id=25544, sat_name="ISS", max_el=70.0)
-            handle_satpass(env, "central.sat.pass.iss", data={},
+            result = _ingest_and_consolidate(env, "central.sat.pass.iss",
-                           now=1718163120)
+                                             now=1781235000)
+        assert result is None, "dry_run should suppress dispatch"
 
+        # dry-run logging now happens in the consolidation step; the message
+        # is "DRY-RUN would air (consolidated, N observers): <wire>".
         matching = [r for r in caplog.records
-                    if r.message.startswith("DRY-RUN would air:")]
+                    if r.message.startswith("DRY-RUN would air")]
         assert len(matching) == 1, "Should log DRY-RUN wire text once"
         assert "ISS" in matching[0].message
 
@@ -264,11 +297,12 @@ class TestDryRun:
         _enable_satpass_db(norad_ids=[25544], dry_run=False)
         _clear_handler_flags()
 
-        data = {}
         env = _envelope(norad_id=25544, max_el=70.0)
-        result = handle_satpass(env, "central.sat.pass.iss", data=data,
+        result = _ingest_and_consolidate(env, "central.sat.pass.iss",
-                                now=1718163120)
+                                         now=1781235000)
         assert result is not None, "dry_run=False should dispatch"
+        # commit callback now rides on the consolidation result's data dict.
+        wire, data = result
         assert "_on_broadcast_committed" in data
 
     def test_dry_run_default_is_true(self):
@@ -322,8 +356,13 @@ class TestBroadcastWireFormat:
         lines = wire.split("\n")
         assert len(lines) == 2
 
+        # No peak_compass passed here \u2192 the sweep stays aos\u2192los (SW\u2192NE).
         assert lines[0] == "\U0001F6F0\uFE0F ISS high pass, SW\u2192NE"
-        assert lines[1] == "6 minute window, 8:38\u20138:44 PM MDT"
+        # line2 renders "min window" + a date qualifier for a pass not
+        # occurring today (the fixed 2026-06-12 date is always in the past
+        # relative to run time).
+        assert lines[1].startswith("6 min window, 8:38\u20138:44 PM MDT")
+        assert lines[1] == "6 min window, 8:38\u20138:44 PM MDT Fri Jun 12"
 
     def test_bucket_overhead_at_60(self):
         """max_el=60 should be 'overhead'."""
@@ -464,7 +503,7 @@ class TestNoradIdTypeCoercion:
         _enable_satpass_db(norad_ids=["25544"], dry_run=False)
 
         env = _envelope(norad_id=25544, max_el=80.0)
-        result = handle_satpass(env, "test.subject", data={}, now=1781235000)
+        result = _ingest_and_consolidate(env, "test.subject", now=1781235000)
         assert result is not None, "string norad_id should match int wire"
 
     def test_mixed_int_and_string_norad_ids(self):
@@ -475,13 +514,13 @@ class TestNoradIdTypeCoercion:
 
         # int in list, int on wire
         env_iss = _envelope(norad_id=25544, max_el=65.0)
-        result_iss = handle_satpass(env_iss, "test.subject", data={}, now=1781235000)
+        result_iss = _ingest_and_consolidate(env_iss, "test.subject", now=1781235000)
         assert result_iss is not None, "int norad_id in mixed list should match"
 
         # string in list, int on wire
         env_noaa = _envelope(norad_id=22825, sat_name="NOAA 15", max_el=65.0,
                              aos="2026-06-12T05:32:00Z", los="2026-06-12T05:38:00Z")
-        result_noaa = handle_satpass(env_noaa, "test.subject", data={}, now=1781235000)
+        result_noaa = _ingest_and_consolidate(env_noaa, "test.subject", now=1781235000)
         assert result_noaa is not None, "string norad_id in mixed list should match int wire"
 
     def test_garbage_entries_skipped_without_crash(self):
@@ -493,7 +532,7 @@ class TestNoradIdTypeCoercion:
 
         env = _envelope(norad_id=25544, max_el=80.0)
         # Must not raise, and the valid entry should still match
-        result = handle_satpass(env, "test.subject", data={}, now=1781235000)
+        result = _ingest_and_consolidate(env, "test.subject", now=1781235000)
         assert result is not None, "valid entry should match despite garbage siblings"
 
     def test_all_garbage_norad_ids_matches_nothing(self):
@@ -513,7 +552,7 @@ class TestNoradIdTypeCoercion:
         _enable_satpass_db(norad_ids=[25544], dry_run=False)
 
         env = _envelope(norad_id=25544, max_el=80.0)
-        result = handle_satpass(env, "test.subject", data={}, now=1781235000)
+        result = _ingest_and_consolidate(env, "test.subject", now=1781235000)
         assert result is not None, "pure int norad_id should still match"
 
     def test_string_norad_id_rejects_non_matching(self):
@@ -567,7 +606,7 @@ class TestStalenessGuard:
         los = datetime.fromtimestamp(los_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 
         env = _envelope(norad_id=25544, max_el=70.0, aos=aos, los=los)
-        result = handle_satpass(env, "test.subject", data={}, now=now)
+        result = _ingest_and_consolidate(env, "test.subject", now=now)
         assert result is not None, "ongoing pass (los in future) should broadcast"
 
     def test_future_pass_broadcasts(self):
@@ -583,7 +622,7 @@ class TestStalenessGuard:
         los = datetime.fromtimestamp(los_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 
         env = _envelope(norad_id=25544, max_el=55.0, aos=aos, los=los)
-        result = handle_satpass(env, "test.subject", data={}, now=now)
+        result = _ingest_and_consolidate(env, "test.subject", now=now)
         assert result is not None, "future pass should broadcast"
 
     def test_none_los_falls_through(self):
@@ -601,7 +640,7 @@ class TestStalenessGuard:
         # Patch los to None by removing los_time from inner data
         env["data"]["data"]["los_time"] = None
 
-        result = handle_satpass(env, "test.subject", data={}, now=now)
+        result = _ingest_and_consolidate(env, "test.subject", now=now)
         # Should not be rejected by staleness guard — falls through to
         # normal handling (wire produced or other filter applies)
         # We just verify it does NOT crash and is not rejected as stale

work/tests/test_satpass_compass_fallback.py

@@ -145,6 +145,27 @@ def _clear_handler_flags():
             delattr(handle_satpass, attr)
 
 
+def _ingest_and_consolidate(env, subject, *, now, data=None):
+    """Drive the two-call async satpass contract (ingest → consolidate).
+
+    handle_satpass() ingests the pass and returns None; the consumer then
+    runs consolidate_satpass_pending(), which yields the (wire, data) to
+    broadcast (or None). Returns the consolidation result so a test can
+    assert on the real broadcast wire.
+    """
+    from meshai.central.satpass_handler import (
+        handle_satpass, consolidate_satpass_pending,
+        drain_pending_consolidation_ids)
+    drain_pending_consolidation_ids()
+    assert handle_satpass(
+        env, subject, data=data if data is not None else {}, now=now) is None
+    for cid in drain_pending_consolidation_ids():
+        res = consolidate_satpass_pending(cid)
+        if res is not None:
+            return res
+    return None
+
+
 # ── (a) satpass_predict envelope: raw azimuths → non-empty compass ───
 
 def test_satpass_predict_compass_from_raw_azimuths():
@@ -156,29 +177,26 @@ def test_satpass_predict_compass_from_raw_azimuths():
     _clear_handler_flags()
 
     now = 1781330520  # well before the AO-27 pass window
-    wire = handle_satpass(
+    result = _ingest_and_consolidate(
         AO27_PREDICT_ENVELOPE,
         "central.sat.pass.us.id.filer",
-        data={},
         now=now,
     )
-    assert wire is not None, "handler returned None for satpass_predict envelope"
+    assert result is not None, "handler returned None for satpass_predict envelope"
+    wire, _ = result
 
     lines = wire.split("\n")
     assert len(lines) == 2, f"Expected 2 lines, got {len(lines)}: {wire!r}"
 
-    # Line 1 must contain non-empty compass directions
+    # Line 1 format: name bucket, aos→peak→los. Raw azimuths convert
-    # 163.2° → S (8-point compass), 348.7° → N
+    # to a non-empty 3-point compass sweep.
-    assert "S" in lines[0].split(",")[-1], f"Expected S (from 163.2°) in compass portion: {lines[0]!r}"
+    compass = lines[0].split(", ")[-1]
-    assert "\u2192" in lines[0], f"Expected → arrow in line 1: {lines[0]!r}"
+    parts = compass.split("\u2192")
-
+    assert len(parts) == 3, f"Expected aos->peak->los sweep, got {parts!r}"
-    # Extract the compass portion: after bucket comma, before newline
+    # 163.2 -> S, 245.0 -> SW (peak), 348.7 -> N  (8-point compass)
-    # Format: 🛰️ {name} {bucket}, {aos_compass}→{los_compass}
+    assert parts[0] == "S", f"Expected S (from 163.2): {parts!r}"
-    arrow_idx = lines[0].index("\u2192")
+    assert parts[1] == "SW", f"Expected SW (from peak 245.0): {parts!r}"
-    los_part = lines[0][arrow_idx + 1:]
+    assert parts[2] == "N", f"Expected N (from 348.7): {parts!r}"
-    assert los_part != "", f"los_compass is empty in line 1: {lines[0]!r}"
-    # 348.7° → N in 8-point compass
-    assert los_part == "N", f"Expected N (from 348.7°) after arrow: {los_part!r}"
 
 
 # ── (b) n2yo envelope: precomputed _compass strings used as-is ───────
@@ -192,21 +210,22 @@ def test_n2yo_precomputed_compass_unchanged():
     _clear_handler_flags()
 
     now = 1781065800  # before NOAA-18 pass window
-    wire = handle_satpass(
+    result = _ingest_and_consolidate(
         N2YO_ENVELOPE,
         "central.sat.pass.us.id.filer",
-        data={},
         now=now,
     )
-    assert wire is not None, "handler returned None for n2yo envelope"
+    assert result is not None, "handler returned None for n2yo envelope"
+    wire, _ = result
 
     lines = wire.split("\n")
-    # Must use the precomputed strings: SE→N
+    # Must use the precomputed strings verbatim: SE→ENE→N
-    assert "SE" in lines[0], f"Expected SE from precomputed _compass: {lines[0]!r}"
+    compass = lines[0].split(", ")[-1]
-    # The los_compass from precomputed is "N"
+    parts = compass.split("\u2192")
-    arrow_idx = lines[0].index("\u2192")
+    assert len(parts) == 3, f"Expected aos->peak->los sweep, got {parts!r}"
-    los_part = lines[0][arrow_idx + 1:]
+    assert parts[0] == "SE", f"Expected precomputed aos SE: {parts!r}"
-    assert los_part == "N", f"Expected precomputed los_compass N, got {los_part!r}"
+    assert parts[1] == "ENE", f"Expected precomputed peak ENE: {parts!r}"
+    assert parts[2] == "N", f"Expected precomputed los N: {parts!r}"
 
 
 # ── (c) envelope with neither → empty compass, no crash ──────────────
@@ -228,16 +247,16 @@ def test_no_compass_no_azimuth_no_crash():
         d.pop(key, None)
 
     now = 1781330520
-    wire = handle_satpass(
+    result = _ingest_and_consolidate(
         env,
         "central.sat.pass.us.id.filer",
-        data={},
         now=now,
     )
-    assert wire is not None, "handler crashed or returned None — should produce wire with empty compass"
+    assert result is not None, "handler crashed or returned None — should produce wire with empty compass"
+    wire, _ = result
 
     lines = wire.split("\n")
     assert len(lines) == 2
-    # Arrow should still be present with empty directions: "→" or similar
+    # Arrow still present even with empty (peak empty -> no peak segment):
-    assert "\u2192" in lines[0], f"Expected → in line 1 even with empty compass: {lines[0]!r}"
+    assert "\u2192" in lines[0], f"Expected arrow in line 1 even with empty compass: {lines[0]!r}"
     # No crash = test passes

work/tests/test_satpass_handler.py

@@ -53,6 +53,7 @@ def mock_adapter_config():
     cfg.norad_ids = [25544]  # must be non-empty for opt-in
     cfg.dry_run = False
     cfg.max_broadcasts_per_hour = 4
+    cfg.max_aos_horizon_hours = 24
     with patch("meshai.central.satpass_handler.adapter_config") as mock:
         mock.satpass = cfg
         from meshai.central.satpass_handler import handle_satpass
@@ -63,18 +64,38 @@ def mock_adapter_config():
         yield cfg
 
 
+def _ingest_and_consolidate(env, subject, *, now, data=None):
+    """Drive the two-call async satpass contract (ingest -> consolidate).
+
+    handle_satpass() ingests the pass into satpass_pending and ALWAYS
+    returns None; the consumer then runs consolidate_satpass_pending(),
+    which returns the (wire, data) to broadcast (or None if suppressed).
+    Returns the consolidation result so a test can assert on the real wire.
+    """
+    from meshai.central.satpass_handler import (
+        handle_satpass, consolidate_satpass_pending,
+        drain_pending_consolidation_ids)
+    drain_pending_consolidation_ids()
+    assert handle_satpass(
+        env, subject, data=data if data is not None else {}, now=now) is None
+    for cid in drain_pending_consolidation_ids():
+        res = consolidate_satpass_pending(cid)
+        if res is not None:
+            return res
+    return None
+
+
 class TestSatpassHandler:
     """Tests for handle_satpass function."""
 
-    def test_high_elevation_pass_broadcasts(self, mock_db, mock_adapter_config):
+    def test_high_elevation_pass_broadcasts(self, mock_adapter_config):
-        """A pass with high elevation should broadcast."""
+        """A pass with high elevation should broadcast (via consolidation)."""
-        from meshai.central.satpass_handler import handle_satpass
-
         env = _envelope(max_el=75.0)
-        result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)
+        result = _ingest_and_consolidate(env, "central.sat.pass.iss",
-
+                                         now=1781235000)
         assert result is not None
-        assert "ISS" in result
+        wire, _ = result
+        assert "ISS" in wire
 
     def test_low_elevation_pass_filtered(self, mock_db, mock_adapter_config):
         """A pass below min_elevation should be filtered."""
@@ -96,14 +117,12 @@ class TestSatpassHandler:
 
         assert result is None
 
-    def test_observer_filter_allows_match(self, mock_db, mock_adapter_config):
+    def test_observer_filter_allows_match(self, mock_adapter_config):
         """A pass for configured observer should broadcast."""
-        from meshai.central.satpass_handler import handle_satpass
-
         mock_adapter_config.observers = ["Boise", "Magic Valley"]
         env = _envelope(observer="Boise", max_el=45.0)
-        result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)
+        result = _ingest_and_consolidate(env, "central.sat.pass.iss",
-
+                                         now=1781235000)
         assert result is not None
 
     def test_norad_id_filter(self, mock_db, mock_adapter_config):
@@ -116,48 +135,49 @@ class TestSatpassHandler:
 
         assert result is None
 
-    def test_dedup_blocks_second_broadcast(self, mock_db, mock_adapter_config):
+    def test_dedup_blocks_second_broadcast(self, mock_adapter_config):
         """Second pass in same hour bucket should be deduplicated."""
-        from meshai.central.satpass_handler import handle_satpass
-
-        # First call returns no existing broadcast
-        mock_db.execute.return_value.fetchone.return_value = None
-
         env = _envelope(max_el=60.0)
-        result1 = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)
+
+        # First pass consolidates into a real broadcast.
+        result1 = _ingest_and_consolidate(env, "central.sat.pass.iss",
+                                          now=1781235000)
         assert result1 is not None
 
-        # Second call simulates existing broadcast
+        # Simulate the broadcast committing (marks satpass_events broadcast).
-        mock_db.execute.return_value.fetchone.return_value = {"last_broadcast_at": 1718163120}
+        _wire, data1 = result1
+        data1["_on_broadcast_committed"](1781235010)
 
-        result2 = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163180)
+        # Second pass in the same (norad, aos-hour) bucket must be deduped.
+        result2 = _ingest_and_consolidate(env, "central.sat.pass.iss",
+                                          now=1781235060)
         assert result2 is None
 
-    def test_wire_format(self, mock_db, mock_adapter_config):
+    def test_wire_format(self, mock_adapter_config):
         """Wire format should have 2 lines with correct info."""
-        from meshai.central.satpass_handler import handle_satpass
-
         env = _envelope(sat_name="ISS", max_el=75, observer="Boise",
-                        direction="NW-SE", aos_compass="SW", los_compass="NE")
+                        direction="S", aos_compass="SW", los_compass="NE")
-        result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)
+        result = _ingest_and_consolidate(env, "central.sat.pass.iss",
+                                         now=1781235000)
+        assert result is not None
+        wire, _ = result
 
-        lines = result.split("\n")
+        lines = wire.split("\n")
         assert len(lines) == 2
         assert "ISS" in lines[0]
         assert "overhead" in lines[0]
-        assert "SW" in lines[0]
+        assert "SW" in lines[0]        # aos_compass
-        assert "NE" in lines[0]
+        assert "S" in lines[0]         # peak_compass (new)
-        assert "minute window" in lines[1]
+        assert "NE" in lines[0]        # los_compass
+        assert "min window" in lines[1]
 
-    def test_commit_callback_attached(self, mock_db, mock_adapter_config):
+    def test_commit_callback_attached(self, mock_adapter_config):
-        """Broadcast should attach commit callback."""
+        """Broadcast should attach commit callback (on the consolidation data)."""
-        from meshai.central.satpass_handler import handle_satpass
-
-        data = {}
         env = _envelope(max_el=60.0)
-        result = handle_satpass(env, "central.sat.pass.iss", data=data, now=1718163120)
+        result = _ingest_and_consolidate(env, "central.sat.pass.iss",
-
+                                         now=1781235000)
         assert result is not None
+        _wire, data = result
         assert "_on_broadcast_committed" in data
         assert "_broadcast_audit" in data
         assert data["_broadcast_audit"]["table"] == "satpass_events"

work/tests/test_satpass_wire_fields.py

@@ -96,6 +96,26 @@ def _enable_satpass(norad_ids=None):
     invalidate_cache()
 
 
+def _ingest_and_consolidate(env, subject, *, now, data=None):
+    """Drive the two-call async satpass contract (ingest -> consolidate).
+
+    handle_satpass() ingests the pass and returns None; the consumer then
+    runs consolidate_satpass_pending(), which returns the (wire, data) to
+    broadcast (or None). Returns the consolidation result.
+    """
+    from meshai.central.satpass_handler import (
+        handle_satpass, consolidate_satpass_pending,
+        drain_pending_consolidation_ids)
+    drain_pending_consolidation_ids()
+    assert handle_satpass(
+        env, subject, data=data if data is not None else {}, now=now) is None
+    for cid in drain_pending_consolidation_ids():
+        res = consolidate_satpass_pending(cid)
+        if res is not None:
+            return res
+    return None
+
+
 # ── Handler produces correct satpass_events row ─────────────────────
 
 def test_noaa18_envelope_produces_satpass_event():
@@ -110,13 +130,13 @@ def test_noaa18_envelope_produces_satpass_event():
         del handle_satpass._no_norad_ids_logged
 
     now = 1781065800  # before NOAA18 envelope los_time
-    wire = handle_satpass(
+    result = _ingest_and_consolidate(
         NOAA18_ENVELOPE,
         "central.sat.pass.us.id.filer",
-        data={},
         now=now,
     )
-    assert wire is not None, "handler returned None -- field extraction failed"
+    assert result is not None, "handler returned None -- field extraction failed"
+    wire, _ = result
 
     conn = get_db()
     rows = conn.execute(
@@ -147,25 +167,26 @@ def test_noaa18_wire_message_format():
     if hasattr(handle_satpass, "_no_norad_ids_logged"):
         del handle_satpass._no_norad_ids_logged
 
-    wire = handle_satpass(
+    result = _ingest_and_consolidate(
         NOAA18_ENVELOPE,
         "central.sat.pass.us.id.filer",
-        data={},
         now=1781065800,  # before NOAA18 envelope los_time
     )
-    assert wire is not None
+    assert result is not None
+    wire, _ = result
 
     lines = wire.split("\n")
     assert len(lines) == 2, f"Expected 2 lines, got {len(lines)}: {wire!r}"
 
-    # Line 1: satellite name + bucket + compass directions
+    # Line 1: satellite name + bucket + compass sweep (aos->peak->los)
     assert "NOAA 18" in lines[0]
     assert "low pass" in lines[0]  # 22.69 < 30 = low pass
-    assert "SE" in lines[0]  # aos_compass
+    assert "SE" in lines[0]   # aos_compass
-    assert "N" in lines[0]   # los_compass
+    assert "ENE" in lines[0]  # peak_compass (new)
+    assert "N" in lines[0]    # los_compass
 
-    # Line 2: duration + time window
+    # Line 2: duration + time window ("min window")
-    assert "minute window" in lines[1]
+    assert "min window" in lines[1]
 
 
 def test_missing_norad_id_rejected():
@@ -230,13 +251,12 @@ def test_observer_fallback_to_slug():
     del env["data"]["data"]["observer_name"]
     # observer_slug = "filer" still present
 
-    wire = handle_satpass(
+    result = _ingest_and_consolidate(
         env,
         "central.sat.pass.us.id.filer",
-        data={},
         now=1781065800,  # before NOAA18 envelope los_time
     )
-    assert wire is not None
+    assert result is not None
     # Observer name stored in DB, not in broadcast wire format
     from meshai.persistence import get_db
     conn = get_db()