Add zero-hop impulse advert. Closes #83.

app/AGENTS.md

@@ -168,7 +168,7 @@ app/
 - `GET /radio/config` — includes `path_hash_mode`, `path_hash_mode_supported`, and advert-location on/off
 - `PATCH /radio/config` — may update `path_hash_mode` (`0..2`) when firmware supports it
 - `PUT /radio/private-key`
-- `POST /radio/advertise`
+- `POST /radio/advertise` — manual advert send; request body may set `mode` to `flood` or `zero_hop` (defaults to `flood`)
 - `POST /radio/discover` — short mesh discovery sweep for nearby repeaters/sensors
 - `POST /radio/disconnect`
 - `POST /radio/reboot`

app/radio_sync.py

@@ -14,6 +14,7 @@ import logging
 import math
 import time
 from contextlib import asynccontextmanager
+from typing import Literal
 
 from meshcore import EventType, MeshCore
 
@@ -37,6 +38,8 @@ logger = logging.getLogger(__name__)
 
 DEFAULT_MAX_CHANNELS = 40
 
+AdvertMode = Literal["flood", "zero_hop"]
+
 
 def _contact_sync_debug_fields(contact: Contact) -> dict[str, object]:
     """Return key contact fields for sync failure diagnostics."""
@@ -686,7 +689,12 @@ async def stop_message_polling():
         logger.info("Stopped periodic message polling")
 
 
-async def send_advertisement(mc: MeshCore, *, force: bool = False) -> bool:
+async def send_advertisement(
+    mc: MeshCore,
+    *,
+    force: bool = False,
+    mode: AdvertMode = "flood",
+) -> bool:
     """Send an advertisement to announce presence on the mesh.
 
     Respects the configured advert_interval - won't send if not enough time
@@ -695,11 +703,15 @@ async def send_advertisement(mc: MeshCore, *, force: bool = False) -> bool:
     Args:
         mc: The MeshCore instance to use for the advertisement.
         force: If True, send immediately regardless of interval.
+        mode: Advertisement mode. Flood adverts use the persisted flood-advert
+            throttle state; zero-hop adverts currently send immediately.
 
     Returns True if successful, False otherwise (including if throttled).
     """
-    # Check if enough time has elapsed (unless forced)
-    if not force:
+    use_flood = mode == "flood"
+
+    # Only flood adverts currently participate in persisted throttle state.
+    if use_flood and not force:
         settings = await AppSettingsRepository.get()
         interval = settings.advert_interval
         last_time = settings.last_advert_time
@@ -726,18 +738,19 @@ async def send_advertisement(mc: MeshCore, *, force: bool = False) -> bool:
             return False
 
     try:
-        result = await mc.commands.send_advert(flood=True)
+        result = await mc.commands.send_advert(flood=use_flood)
         if result.type == EventType.OK:
-            # Update last_advert_time in database
-            now = int(time.time())
-            await AppSettingsRepository.update(last_advert_time=now)
-            logger.info("Advertisement sent successfully")
+            if use_flood:
+                # Track flood advert timing for periodic/startup throttling.
+                now = int(time.time())
+                await AppSettingsRepository.update(last_advert_time=now)
+            logger.info("%s advertisement sent successfully", mode.replace("_", "-"))
             return True
         else:
-            logger.warning("Failed to send advertisement: %s", result.payload)
+            logger.warning("Failed to send %s advertisement: %s", mode, result.payload)
             return False
     except Exception as e:
-        logger.warning("Error sending advertisement: %s", e, exc_info=True)
+        logger.warning("Error sending %s advertisement: %s", mode, e, exc_info=True)
         return False
 
 

app/routers/radio.py

@@ -32,6 +32,7 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/radio", tags=["radio"])
 
 AdvertLocationSource = Literal["off", "current"]
+RadioAdvertMode = Literal["flood", "zero_hop"]
 DiscoveryNodeType: TypeAlias = Literal["repeater", "sensor"]
 DISCOVERY_WINDOW_SECONDS = 8.0
 _DISCOVERY_TARGET_BITS = {
@@ -104,6 +105,13 @@ class PrivateKeyUpdate(BaseModel):
     private_key: str = Field(description="Private key as hex string")
 
 
+class RadioAdvertiseRequest(BaseModel):
+    mode: RadioAdvertMode = Field(
+        default="flood",
+        description="Advertisement mode: flood through repeaters or zero-hop local only",
+    )
+
+
 def _monotonic() -> float:
     return time.monotonic()
 
@@ -266,24 +274,25 @@ async def set_private_key(update: PrivateKeyUpdate) -> dict:
 
 
 @router.post("/advertise")
-async def send_advertisement() -> dict:
-    """Send a flood advertisement to announce presence on the mesh.
+async def send_advertisement(request: RadioAdvertiseRequest | None = None) -> dict:
+    """Send an advertisement to announce presence on the mesh.
 
-    Manual advertisement requests always send immediately, updating the
-    last_advert_time which affects when the next periodic/startup advert
-    can occur.
+    Manual advertisement requests always send immediately. Flood adverts update
+    the shared flood-advert timing state used by periodic/startup advertising;
+    zero-hop adverts currently do not.
 
     Returns:
         status: "ok" if sent successfully
     """
     require_connected()
+    mode: RadioAdvertMode = request.mode if request is not None else "flood"
 
-    logger.info("Sending flood advertisement")
+    logger.info("Sending %s advertisement", mode.replace("_", "-"))
     async with radio_manager.radio_operation("manual_advertisement") as mc:
-        success = await do_send_advertisement(mc, force=True)
+        success = await do_send_advertisement(mc, force=True, mode=mode)
 
     if not success:
-        raise HTTPException(status_code=500, detail="Failed to send advertisement")
+        raise HTTPException(status_code=500, detail=f"Failed to send {mode} advertisement")
 
     return {"status": "ok"}