Notification and push fan-out

The shared delivery mechanism that everything else on this site ultimately calls into: one hub-side event becomes zero or more delivered notifications across push (APNs), Telegram, and the app's poll surface. This page is the generic fan-out; the native alert pipeline is a specific, higher-level policy layer (severity, quiet hours, dedup) that sits in front of the same fanOut call.

flowchart TD
    EV["Event origination:\ntask/chat/alert/approval"] --> WRITE["durable fact written first\n(chat_turns / task_evidence /\nalerts row)"]

    WRITE --> SEAM{"which store-owned\npush seam?"}
    SEAM -->|alert| ALERTSTORE["AlertStore.Ingest:\npost-commit routePush"]
    SEAM -->|approval| APPROVALSTORE["Items.routeApprovalPush"]
    SEAM -->|"chat turn (app transport)"| TICK["PushNotifier.Tick\n(3s poll, not event-driven)"]
    SEAM -->|"chat turn (non-app transport)"| TELEGRAM["ProcessTurn: send()\nsynchronous Telegram delivery"]

    ALERTSTORE --> ROUTE["RouteAlert: severity routing\n(see native alert pipeline)"]
    APPROVALSTORE --> FANOUT
    ROUTE --> FANOUT

    TICK --> QUERY["query chat_turns JOIN\ntelegram_channels WHERE\ntransport=app AND status IN\n(done,error) AND notified_at IS NULL"]
    QUERY --> FANOUT
    QUERY --> STAMP["notified_at = now()\n(stamped after attempt,\nsuccess or not)"]

    TELEGRAM -->|"send fails"| SORRY["best-effort 'sorry' message,\nturn marked failed"]

    FANOUT["fanOut: list all\nregistered devices"] --> PUSH["Pusher.Push per device\n(APNSClient in prod)"]
    PUSH -->|success| NEXT["next device"]
    PUSH -->|"ErrDeviceGone (410)"| DELGONE["delete device row,\nno retry, no ledger entry"]
    PUSH -->|"other error"| LOGFAIL["log + fail row\nin runs ledger,\nno automatic retry"]

    FANOUT --> TRUNC["truncatePush:\nrune-safe cut at 180,\nbreak on last space"]

    APPROVALSTORE --> CALLBACK["Telegram tap or app confirm:\nCallbackRouter.HandleCallback"]
    CALLBACK --> CBTYPE{"cb.Data prefix"}
    CBTYPE -->|"audit:"| AUDIT["queueAudit:\nenqueue agent run,\nedit message"]
    CBTYPE -->|"act:cancel:"| CANCEL["ChatActions.Cancel"]
    CBTYPE -->|"act:confirm:"| CLAIMCB["Claim: atomic\nexactly-once re-auth"]
    CLAIMCB --> EXEC["executeChatAction:\nshared executor for both\nTelegram and app-transport confirm"]

    ROUTE -.->|"quiet-held warnings\n+ pending approvals"| SWEEPQ["SweepQuietFlush:\none combined push\nwhen quiet window ends"]
    ROUTE -.->|"unacked criticals"| SWEEPC["SweepCriticalRenotify:\n+15m/+2h, cap 3"]
    APPROVALSTORE -.->|"still-pending approval"| SWEEPA["SweepApprovalReminders:\none +4h reminder"]
    SWEEPQ --> FANOUT
    SWEEPC --> FANOUT
    SWEEPA --> FANOUT

Node annotations

Event origination

Distinct producers write a durable fact first rather than pushing inline: internal/hub/agent_runs_api.go (L529-541), internal/hub/dispatch.go (L778), internal/hub/items.go (L150, an approval-eligible task_evidence row), internal/hub/alert_webhook.go (L150), and internal/hub/chat_worker.go:ProcessTurn (L84, a chat_turns row) all follow this pattern -- the notification is derived from state, not fired as a side effect of a request handler.

Store-owned push seam

internal/hub/alert_store.go:AlertStore.Ingest (L63-102) commits the alert transition in a transaction, then, only after commit, calls routePush (L108) -> AlertPusher.RouteAlert. internal/hub/items.go: Items.routeApprovalPush (L146) does the equivalent for a newly-created approval. Both are nil-safe: if no pusher was ever wired (e.g. in tests), delivery is silently skipped rather than erroring the write.

PushNotifier.Tick (chat-turn watermark loop)

internal/hub/push_notifier.go:Tick (L96), run on a ticker (Run, L79, default 3 seconds) -- a polling trigger path, not event-driven, specifically for transport='app' turns. The query selects chat_turns joined to telegram_channels where transport='app', status IN ('done','error'), and notified_at IS NULL (L97-102): a durable, restart-safe "unnotified" cursor. notified_at is stamped after the fan-out attempt regardless of success (L135), so a poisoned payload can never wedge the queue on retry.

Channel selection (decided at the source, not in fan-out)

internal/hub/chat_worker.go:ProcessTurn (L36) reads transport via ChannelSession (L37): transport=="app" (L46) skips Telegram send/typing entirely -- the reply is only persisted for the app's own poll plus the Tick push above. Any other transport delivers synchronously over Telegram (cmd/chat-worker/main.go:send, ~L114); a send error there marks the turn failed (chat_worker.go:79, L92-98) with a best-effort "sorry" reply, itself only sent for non-app transports.

Confirm/cancel chat actions

internal/hub/chat_callbacks.go:CallbackRouter.HandleCallback (L51). Branches on the cb.Data prefix: "audit:" -> queueAudit (L124, enqueues an agent run and edits the originating message), "act:cancel:" -> ChatActions.Cancel (L78), "act:confirm:" -> Claim (L99, an atomic exactly-once re-auth gate) then executeChatAction (L150/158) -- the same deterministic executor used by both a Telegram button tap and the app's own confirm endpoint (chat_actions_api.go), so a result can never drift between the two channels.

fanOut / Pusher.Push

internal/hub/push_notifier.go:fanOut (L284). Lists every registered device (n.devices.List), then calls n.pusher.Push per device (Pusher interface, *APNSClient in production). errors.Is(err, ErrDeviceGone) (L295, a 410 from Apple) deletes the device row outright -- no retry, no ledger entry, since the device is gone for good. Any other error (L302) is logged and recorded as a fail row in the runs ledger (n.runs.Record, L304-309); there is no automatic retry -- the next real-world trigger of the same event class, or one of the scheduled sweeps below, is the only re-delivery path.

truncatePush

internal/hub/push_notifier.go:truncatePush (L324), applied by every fanOut caller: a rune-safe cut at pushBodyMax=180, breaking on the last space within truncateLookback=40 runes so a push never lands mid-word.

Coalescing / renotify sweeps

SweepQuietFlush (notification_router.go, L239) folds every held warning alert and held approval into a single combined push once the quiet window ends -- never one push per item. SweepCriticalRenotify (L195) re-pushes unacked criticals at +15m and +2h, then caps at criticalRenotifyCap=3 (L43), after which it becomes badge-only. SweepApprovalReminders (L299) sends exactly one +4h reminder per still-pending approval, silently consuming the slot if already resolved by the time the sweep runs (L332-337). These are the retry/renotify layer; they are distinct from the per-attempt device retry (or lack of it) inside fanOut itself.

What is explicitly not part of this fan-out

internal/hub/inbox_api.go, inbox_actions_api.go, and more_api.go contain no notify/fanout/subscribe/channel logic (verified by grep -- zero matches). The inbox is a pull/poll surface the app reads (often reached via a deep link carried in a push), not a delivery channel that anything fans out into.

Related pages

• Native alert pipeline -- the severity-aware policy layer that decides whether/when an alert reaches this fan-out at all.
• Landing queue -- one real producer of the events that flow through here (CI-red comments, post-merge-break pages).
• App deploy / OTA path -- another producer (deploy failure pages, OTA "update available" pushes).