DM Execution — End-to-End Flow

DM Execution — End-to-End Flow

How a Direct Message (DM) goes from a registered type to a delivered push, across the three specs that own the DM stack (PD-3 registry, PD-1 executor, PD-2 delivery) and the code that implements them in this repo.

Status (2026-06-17). PD-3 (registry) is merged and live. PD-1 (executor) is feature-complete across a review-ready PR stack: the executor core + audit adapter + CCS enricher are merged to main, and the remaining adapters — runner concretes (PR-G #228), flag checker (PR-H #229), Eppo resolver (PR-I #230), cron concretes (PR-J #231), dispatch send_time + cap + content (PR-L1 #232 / PR-L2 #233 / PR-K #235), futurefire queue (PR-M #234), and the unified-worker composition root that ties them together (PR-F #237) — are all open, green, and mergeable, pending review. Once the stack merges, the cron path runs end-to-end behind the DMExecutorEnabled dark-launch gate. Architecture note: ADR-0001 removed the PD-1→PD-2 SQS handoff — PD-1 now posts the notification gateway directly, and PD-2’s logic runs inline as the dispatch step (§5). This doc reflects the post-ADR-0001 flow and marks each piece [live] (on main), [review-ready] (in the open PR stack), or [planned] (not yet built).

Source specs: specs/PD-1.md (scheduled execution), specs/PD-2.md (delivery contract), specs/PD-3.md (type registry).

---

1. The 10,000-foot view

                         ┌──────────────────────────────────────────────┐
   PD-3  (registry)      │  dm/types/*.yaml  →  Go-canonical registration │
   what a DM type IS     │  loaded once at process startup                │
                         └───────────────────────┬──────────────────────┘
                                                  │ registry.Registry (in-process)
                                                  ▼
   PD-1  (executor)      ┌──────────────────────────────────────────────────┐
   when + how it FIRES   │  firing modes ── cron tick / Kafka msg / one-off   │
                         │        │                                            │
                         │        ▼  one FireEvent per (user, instant)         │
                         │  7-stage pipeline                                  │
                         │   stages 1-6 ──► stage 7 = DISPATCH (inline, PD-2) │
                         │        │                  │                         │
                         │        │                  ▼ POST /v1/gateway/notify │
                         │        ▼ (always) one ScheduleEvent audit row       │
                         └────────┬───────────────────┬───────────────────────┘
                                  │ DynamoDB (sync)    │ HTTPS (direct, ADR-0001)
                                  ▼                    ▼
              {env}-dm-schedule-events        notification-service (Growth-owned)
                       │ DDB Streams              gateway: dedup / quota /
                       ▼                          quiet-hours / scheduling
              Snowflake (PLT-629 funnel)             │
                                                     ▼  APNs / FCM → push / in-app

Three load-bearing ideas tie it together:

1. One registration, read in-process. PD-3 YAML compiles to a Go struct; the executor and the dispatch step consume the same struct in the same process. A new DM type ships by adding one YAML file — no executor code change.
2. Two firing modes, one pipeline. A cron tick and a Kafka message both produce the same internal FireEvent; everything after that is shared.
3. Every fire is auditable. Each fire that starts the pipeline writes exactly one ScheduleEvent row, synchronously, regardless of outcome — the audit write is the durability guarantee.
4. Delivery is the gateway’s job, not ours (ADR-0001). Stage 7 posts notification-service directly; the gateway owns dedup, quota, quiet-hours, and scheduling. No queue, no separate consumer.

All of this — the executor, the inline dispatch step, the registry, the Valkey cap — runs in the single cmd/unified-worker Go binary under one LifecycleManager. The only new external surfaces are the DynamoDB audit/scheduling tables (no SQS queue, per ADR-0001).

---

2. PD-3 — the type registry (upstream input) [live]

A DM type is a static YAML record under dm/types/. At process startup the loader (internal/dm/registry/loader.go) parses each file into a canonical Go registry.DMTypeRegistration and indexes them by type_id into a registry.Registry. The executor reads this map; it never parses YAML itself.

What a registration carries (PD-3 §4), and where each field is consumed downstream:

Field	Owner stage	Used by
trigger (cron | kafka)	firing mode	PD-1 cron loop / Kafka handler
candidate_query (cypher | callable | event_payload)	stage 1	PD-1 CandidateRunner
eligibility (callable ref + params)	stage 3	PD-1 and PD-2 (re-run at dispatch)
ranking (optional callable)	stage 4	PD-1 RankingRunner
min_items_gate (optional int)	stage 3	PD-1 pipeline
copy (static Jinja | llm_prompt)	stage 5	PD-1 CopyRenderer
send_time (immediate | scheduled | window)	delivery	PD-2 scheduled_time translation
priority, frequency_cap	stage 7 dispatch	gateway tie-break + per-type cap
idempotency_key_template	stage 7 dispatch	rendered → request_id; the gateway owns the 24h dedupe
experiment, flag	stages 6–7	arm resolution + flag check, both at fire time (ADR-0001: no dispatch re-check)
analytics_labels, reporting_metrics	stage 7 dispatch	carried into the producer="pd2" audit → PS-5 → PLT-629

The registration is the single source of truth for the rollout triple (flag, experiment, analytics). PD-1 reads it once in-process; the dispatch step consumes the same struct, so nothing re-reads the registry off the hot path.

---

3. PD-1 — firing a DM [core live; full adapter set review-ready]

3.1 The FireEvent — the unit of work

Both firing modes converge on one internal type (internal/dm/executor/types.go):

type FireEvent struct {
    DMRunID      string       // uuid4 — groups all fires from one tick / message
    SessionID    string       // uuid4 — fresh per (user, fire)
    UserID       string       // from candidate row OR trigger payload — NEVER from HTTP input
    TypeID       string       // PD-3 type_id
    TriggerKind  TriggerKind  // cron | kafka | one_off
    FireTime     time.Time
    EventPayload []byte       // raw Kafka bytes (nil for cron)
    OneOffFireID *string      // set for future-fire-queue fires
}

Identity invariant (PD-1 constraint): UserID is always sourced from the candidate-query row (cron) or the trigger event payload (Kafka) — never from inbound HTTP headers or query params on a production fire path.

3.2 Firing modes

Cron mode [live] — internal/dm/executor/cron.go. A 1-minute ticker walks every trigger.kind: cron registration:

tick(now)
  for each cron type:
    due?  (DueChecker evaluates the cron expr in the type's tz)   ── no ─► skip
    mint dm_run_id = uuid4()
    claim (type_id, fire_minute_bucket) via conditional write     ── lost ─► skip   (multi-replica safety)
    run candidate_query ONCE → N candidate rows
    for each row: mint session_id, build FireEvent, enqueue to worker pool

The RunClaimer (conditional DDB write keyed on (type_id, fire_minute_bucket)) guarantees exactly one replica fires a given (type, minute) — CGW runs multi-replica on Fargate, so without it every replica would fire every type. The claimer + due-checker are interfaces in cron.go; their concrete DynamoDB/cron-parser implementations (internal/dm/cronsched/) are [review-ready] (PR-J #231), wired into the executor by the composition root (PR-F #237).

Candidate-query binding (cron). Weekly Forecast’s candidate_query is kind: callable (weeklyforecast.queries:strict_with_cold_start_fallback). It is declared in the YAML but registered into the symbol table at worker startup (PR-F #237) — not at package init — because it needs the live Neo4j client and the experiment-membership store. The bound callable reproduces the live handler’s audience exactly (experiment audience → strict shelf query → cold-start fallback → stitch). With the ref bound first, the registry loads with WithCallableResolution() so any unbound ref fails fast at startup rather than at fire time.

Kafka mode [planned] — for trigger.kind: kafka types, PD-1 registers a handler on CGW’s internal/kafka/ framework. One message → one dm_run_id, fanning out 0..N candidates via the type’s extractor (e.g. Earnings Coach: one receipt → one candidate per item). Backpressure is a bounded intake buffer that must not block the Kafka commit loop (PD-1 §4.3).

One-off mode [queue review-ready; route planned] — POST /scheduler/one-off-fire \{user_id, type_id, fire_at\} (admin-scoped, kill-switch-gated) writes a row to {env}-dm-future-fires; a 30s scanner claims due rows and runs them through the same pipeline. Types whose candidate_query.kind == event_payload are rejected (422 type_not_one_off_fireable). The store/scanner/reaper + the one-off-fireable validation land in PR-M #234; the HTTP route and the scanner’s wiring into the unified-worker composition root are a follow-up (intentionally not in PR-F #237).

3.3 The 7-stage pipeline

internal/dm/executor/pipeline.go — Pipeline.Run walks one FireEvent through seven stages. Each stage either STOPs with a terminal outcome or passes downstream. The stage dependencies are interfaces (deps.go) so the orchestrator is testable with fakes and the real adapters (CCS, DDB, gateway-dispatch, Eppo) land independently.

FireEvent
  │
  1. candidate_query   ── empty ───────────────► skipped_no_candidates
  │  (CandidateRunner)    error ───────────────► failed (reason="candidate_query")
  │
  2. CCS enrich        ── batch error ─────────► failed (reason="ccs")
  │  (Enricher)            single-cand error ──► drop that candidate, continue
  │
  3. eligibility       ── all dropped ─────────► skipped_eligibility
  │  (EligibilityRunner)  kept < min_items_gate ► skipped_min_items_gate
  │
  4. ranking (opt)     ── empty after rank ────► skipped_no_candidates
  │  (RankingRunner)      [skipped if type declares no ranking]
  │
  5. copy render       ── llm_prompt (v1) ─────► skipped_unsupported_copy
  │  (CopyRenderer)       render error ────────► failed (reason="copy")
  │
  6. cohort flag       ── off ─────────────────► skipped_flag_off
  │  re-check             error ───────────────► failed (reason="flag")
  │  (FlagChecker)        [catches cohort changes since the candidate query]
  │
  7. dispatch          ── gateway error (after  ► failed (reason="dispatch")
     (Dispatcher)         inline retry) ──────┘
                       └─ gateway 202 ──────────► dispatched (notification_id + scheduled_at set)
                       └─ gateway 429 / dropped ► skipped_quota / dropped_no_devices (PD-2 §4.6)

ADR-0001: stage 7 was an SQS handoff to PD-2. It is now a direct dispatch — the stage maps the assembled fields to a notification-service NotifyRequest, translates send_time → scheduled_time, enforces the per-type cap, and POSTs /v1/gateway/notify via the notification-service/pkg/client SDK, with inline retry on transient 5xx. The interface is executor.Dispatcher (Dispatch(ctx, HandoffMessage) → DispatchResult), implemented by internal/dm/dispatch/. There is no SQS queue and no separate consumer process — PD-2’s logic runs inline here. The pipeline records the dispatcher’s returned outcome, not a hardcoded dispatched. See docs/decisions/0001-drop-pd1-sqs-handoff.md.

Two details worth knowing:

• Experiment arm injection (stage 7). The registration’s analytics_labels.experiment_arm is the template "{experiment_arm}", not a value to ship. resolveArm asks the ExperimentResolver (internal/dm/experiment/, [review-ready] PR-I #230, backed by the membership store) for the per-user arm; if no resolver is wired it falls back to the experiment’s declared control arm. The literal placeholder is never shipped downstream.
• Idempotency key (stage 7). renderIdempotencyKey substitutes the PD-3 §4.8 placeholder vocabulary ({user_id}, {type_id}, {week_start_iso}, {day_iso}, {fire_time_iso}) into the type’s template. {receipt_id} is Kafka-only; a cron type using it is a registration bug, surfaced as an error.

3.4 The assembled dispatch payload (executor.HandoffMessage)

buildHandoff assembles the payload, copying the rollout triple from the registration verbatim. (ADR-0001) This was the PD-1 → PD-2 SQS message; it is now the internal carrier the dispatcher maps to a notification-service NotifyRequest (no longer a wire format). Shape:

{
  "dm_run_id", "session_id", "type_id", "user_id", "user_timezone", "fire_time",
  "idempotency_key", "priority", "frequency_cap",
  "experiment": { "name", "arm" },          // arm = resolved per-user arm
  "flag":       { "name", "kill_switch" },
  "analytics_labels": { "type", "vertical", "cohort", "experiment_arm", "extra" },
  "reporting_metrics": [ ... ],
  "eligibility_ref", "eligibility_params",
  "bff_payload": { /* PS-6 BFF object */ }
}

The dispatcher reads user_timezone + the type’s send_time to compute scheduled_time, maps idempotency_key → request_id, type_id → notif_type, bff_payload → content, and keeps the rollout-triple metadata for the producer="pd2" audit (not on the wire — funnel joins on dm_run_id). See §5.

---

4. The audit trail — every fire, recorded [live]

Pipeline.Run is wrapped so that after the stage walk it always calls the Auditor exactly once, on every path including failures. The audit write is on the critical path: if it fails, Run returns the error and the cron loop surfaces it via Observer.AuditWriteFailed (→ dm_schedule_event_put_errors_total, a paging alert).

func (p *Pipeline) Run(ctx, fe) (FireResult, error) {
    res := p.run(ctx, fe)                       // the 7-stage walk above
    if err := p.Auditor.Record(ctx, fe, res); err != nil {
        return res, fmt.Errorf("audit write failed ...: %w", err)   // durability violated
    }
    return res, nil
}

4.1 The adapter

internal/dm/audit/ is PD-1’s concrete Auditor: a synchronous blocking PutItem to {env}-dm-schedule-events, one row per fire per user. This is deliberately not the async, fail-open posture of the older internal/dmaudit/ (buffered queue + worker pool + drop-on-overflow). That posture is right for the high-throughput notification send loop; it’s wrong here, where the row is the audit guarantee and PD-1 fires are low-frequency (one cron tick fans out a bounded audience).

The row shape lives in a shared codec, internal/dm/scheduleevent/, because the dispatch step writes the same table (producer: "pd2") and the two writers must not drift into divergent schemas. The codec owns:

• the Row struct (DynamoDB dynamodbav tags matching the FSD-declared keys),
• the deterministic Snowflake dedupe key event_id = dm_run_id:user_id — deterministic so a retried write dedupes in Snowflake rather than double-counting the fire,
• fire_time normalized to RFC3339 UTC (the Snowflake create timestamp),
• the 90-day TTL stamp.

(ADR-0001) The dispatch row carries notification_id (gateway-assigned, was sqs_message_id) and scheduled_at — the wall-clock instant the gateway will deliver, returned in the gateway’s 202. Persisting scheduled_at lets the funnel answer “fired Monday 00:00 UTC → scheduled for Monday 08:00 user-local,” which the old fire_time-only row could not.

4.2 Table + outcomes

{env}-dm-schedule-events — PK dm_run_id, SK user_id. DDB Streams → MSK Connect → S3 → Snowflake (DM_SCHEDULE schema, dedupe on event_id). PLT-629’s funnel queries Snowflake, filtering by producer.

PD-1’s outcome enum (internal/dm/executor/types.go): dispatched, skipped_no_candidates, skipped_eligibility, skipped_min_items_gate, skipped_flag_off, skipped_kill_switch, skipped_unsupported_copy, failed. The enum is producer-extensible — PD-2 writes its own outcomes (skipped_weekly_cap, …) to the same table.

Audit boundary nuance. “Every fire produces a ScheduleEvent” applies only to fires whose pipeline started. A kill-switch that’s off at process start or at the cron-tick gate means no candidate query runs and no row is written — there’s nothing to audit. skipped_kill_switch exists only for the rare mid-pipeline flip. Also: in cron mode, a run-level candidate-query failure (e.g. Neo4j down at 08:00) happens before any FireEvent exists, so it’s reported via Observer.CandidateQueryFailed, not as an audit row — the per-fire pipeline never started.

4.3 Other PD-1 tables

• {env}-dm-schedule-runs [review-ready] — the cron claim lease (PK type_id, SK fire_minute_bucket, 24h TTL). Operational state, no Snowflake projection. Written by the RunClaimer concrete (PR-J #231), wired by PR-F #237.
• {env}-dm-future-fires [review-ready; worker wiring deferred] — the one-off queue (PK fire_at_bucket, SK future_fire_id, status-index GSI). The store + 30s scanner + 5-min reaper land in PR-M #234; the worker route + POST /scheduler/one-off-fire endpoint that drive them are a follow-up (not in the PR-F composition root).

---

5. PD-2 — dispatch, inline [review-ready]

ADR-0001 reshaped this layer. PD-2 was an SQS consumer reading {env}-dm-delivery-queue in a separate worker pool, re-validating eligibility/flags at dispatch. That is removed. There is no queue, no consumer, and no dispatch-time re-validation. PD-2’s logic now runs inline as stage 7 (internal/dm/dispatch/, executor.Dispatcher). Why: the gateway already owns intake/dedup/quota/quiet-hours/ scheduling, and the kill-switch + cohort flag are evaluated at fire time (stage 6) — so nothing needs to re-run after a dwell. Full rationale + the retry-durability tradeoff: docs/decisions/0001-drop-pd1-sqs-handoff.md.

Implementation status (review-ready stack). The dispatch adapter (internal/dm/dispatch/) implements the gateway-client mechanics, send_time → scheduled_time for all three kinds (immediate / scheduled / window, PR-L1 #232), the per-type weekly cap via the Valkey Capper (PR-L2 #233), and the bff_payload → NotificationContent extraction (PS-6 schema, PR-K #235). One piece remains a follow-up: the producer="pd2" audit UpdateItem (needs an UpdateItem method on the Recorder). Until it lands, PD-1’s producer="pd1" fire-time row is the system of record per fire, and the dispatch outcome it records is accurate.

The dispatch step, per fire:

HandoffMessage (from stage 6)
  → per-type weekly cap check     (Valkey Lua atomic reserve: dm_weekly:{user}:{type})
        └─ over cap ─────────────► skipped_weekly_cap; audit; STOP
  → map → NotifyRequest           (PD-2 §4.1 field mapping)
  → send_time → scheduled_time    (PD-3 immediate/scheduled/window → instant in user tz)
  → POST /v1/gateway/notify       ──►  notification-service (12 guardrails, Growth-owned)
        │  (inline bounded retry on 5xx / 408 / network)
        ├─ 202 ──────────────────► dispatched; persist notification_id + scheduled_at;
        │                          keep the reserved cap slot
        ├─ 429 quota ────────────► skipped_quota; release cap slot
        ├─ dropped: no devices ──► dropped_no_devices; release cap slot
        ├─ 400 ──────────────────► failed_schema
        └─ 5xx after max retries ► failed (reason="dispatch"); release cap slot
  → audit UpdateItem (producer="pd2", same (dm_run_id,user_id) row PD-1 wrote)

What stays the same as the old PD-2 design (only the host changed, queue → inline):

• Field mapping (PD-2 §4.1): idempotency_key → request_id (the gateway owns the 24h dedupe window — no PD-2-side idempotency store), type_id → notif_type, dm_run_id → analytics_label, bff_payload → content. The rollout-triple metadata stays in the audit, not on the wire — funnel joins on dm_run_id.
• send_time translation (PD-2 §4.3): immediate → no scheduled_time (gateway sends now); scheduled "MON 08:00" tz:user → the wall-clock instant in the user’s timezone; window → a deterministic instant in the allowed range. The gateway holds the deferred send in its Valkey scheduler — PD-1 does not buffer it.
• Per-type weekly cap (PD-2 §4.5): still enforced via the Valkey Lua reserve-then- release script; the daily-100 cap remains the gateway’s.

What’s gone: the SQS queue, the DLQ, the separate consumer pool, and dispatch-time re-validation. Retry is inline (bounded backoff), not SQS visibility-timeout redelivery — see the tradeoff note in the ADR.

---

6. Past notification-service — gateway → APNs/FCM → device, and how a DM presents [Growth-owned]

Everything past POST /v1/gateway/notify is owned by Growth, not by the DM stack. PD-2 §6 describes it only normatively — as platform-facing behavior a DM producer can depend on — and explicitly puts the internals out of scope (the 6-service notification stack, its DynamoDB/Valkey schemas, and APNs/FCM dispatch). This section documents the contract surface, not Growth’s implementation. The whole gateway path already runs in production today (repurchase nudges + weekly forecast ship through it daily); the DM stack is new, the delivery rail is not.

6.1 The gateway pipeline (12 guardrail steps)

A single POST /v1/gateway/notify runs the notification gateway’s 12-step guardrail pipeline before anything reaches a device (PD-2 §6):

idempotency (request_id, 24h) → expiration → device presence → opt-out → quiet hours
→ consolidation → quota (daily-100) → collision resolution (by score)
→ scheduling → quota increment → consolidation group → processed-mark

The response PD-2 acts on:

Gateway response	Means	PD-2
202 Accepted	queued for dispatch (fresh or idempotent cache-hit — currently indistinguishable; §8)	dispatched, INCR cap
429	gateway daily quota hit	skipped_quota, ack (no retry)
dropped: no devices	user has no registered device token	dropped_no_devices, ack
400	bad request (terminal)	failed, ack
503 / 502 / 408 / network	transient	return to queue → retry → DLQ after max-receive

6.2 Gateway → APNs/FCM → device

Past the gateway, dispatch is Growth-internal and out of scope for the DM specs (PD-2 §2.2, §3.1). At the contract level, the path is:

gateway (202, queued)
   │  scheduled_time honored; quiet-hours-adjusted
   ▼
dispatcher → sender ──► APNs   (iOS)   ──► Apple push servers ──► device, app in background/closed
                   └──► FCM    (Android)──► Google push servers ──► device
   │  per-platform payload built here (aps{alert,sound,badge,...} for APNs)
   ▼
device receives push  ──tap──►  deep-link into the Fetch app  ──►  the DM episode

Two things the DM stack relies on but does not build:

• Token → device. Apple/Google identify the device by a push token the app registered with device-registry. The gateway’s “device presence” guardrail is what turns “no token” into dropped: no devices before a wire call to Apple. APNs delivery itself (Apple’s store-and-forward, the feedback/unregistered-token channel) is Apple’s, surfaced back to Growth’s sender, never to PD-1/PD-2.
• The push is a pointer, not the content. The durable artifact is the episode (a 90-day in-app object, live since PLT-372), not the APNs payload. The push is a best-effort nudge to open the app; the episode is what renders. A DM with dropped: no devices still persists its episode — the user sees it on next app open. (See §6.4.)

6.3 notif_type and presentation — what the field does not do

A natural reading of the wire is “notif_type tells the phone how to present the DM.” That is not what notif_type is, and it’s worth stating plainly because the diagram shorthand invites the confusion:

• On the wire, notif_type is just type_id copied verbatim (PD-2 §4.1) — it names the DM type (weekly-forecast), the semantic identity, not a UI surface or a popup style.
• The specs define no mechanism by which a single wire field selects an iOS presentation mode (toast vs. modal vs. FST vs. silent). PD-2 §6 lists “per-platform presentation modes” as explicitly out of scope / Growth-owned.

So how would a DM present differently (small toast vs. large toast vs. non-dismissible modal vs. push-only)? That’s an open, in-design question across the stack (CGW PR #212), not a shipped contract. The current design direction:

• Introduce a distinct presentation.surface field (PD-3-declared) rather than overloading notif_type — because notif_type is already taken by type_id and reusing it would collide.
• presentation.surface is static per DM type (≈1:1 tier→surface), with one dynamic hook — presentation.push.kind: always | none | selective — where selective is a deterministic, side-effect-free callable (re-derivable by PD-2 at dispatch, same §4.3 rule as eligibility).
• The surface must be persisted onto the episode/BFF object (PS-6 / PLT-681), not just the transient NotifyRequest — otherwise app-open re-rendering of an unread DM has no surface to read. The episode object has no presentation field today; that’s the gap.

Status: presentation is proposed, not built. Three follow-ups gate it: (a) PS-6 adds a presentation block to the BFF/episode object + POST /episodes; (b) PD-3 declares the presentation.surface enum + push.kind discriminator (additive); (c) Growth confirms the notif_type-vs-surface naming and the unread-episode-list contract. Until those land, every DM presents through the default push + in-app episode surface — there is no per-type presentation switching in v1.

6.4 Read/unread lives in the episode layer, not the DM stack

PD-1/PD-2/PD-3 do not track read/unread/ignore — and shouldn’t. That state lives in the episode layer, already in production (PLT-372): GET /api/v1/episodes/bff/unread, /unread/count, POST /episodes/ignore/{id}. Because every DM persists as a 90-day episode regardless of push success, “show me unread DMs on app open” works today — the only gap is the missing presentation field above (§6.3), which is what would let app-open render a tier-specific surface rather than the default list item.

---

7. Worked example — Weekly Forecast, Monday 08:00

1. PD-3: dm/types/weekly-forecast.yaml is loaded at startup → registry.Types["weekly-forecast"].
2. PD-1 cron tick (08:00 local, evaluated per-tz): DueChecker.Due → true. A dm_run_id is minted; one replica wins the (weekly-forecast, 2026-…T08:00) claim.
3. Candidate query runs once → N users with a repurchase-due shelf, one candidate row each.
4. Fan-out: N FireEvents (fresh session_id each) into the 32-worker pool.
5. Per fire, stages 1–6: CCS enrich → eligibility (min_shelf_items: 3 gate) → (no ranking) → Jinja copy → cohort-flag re-check → resolve Eppo arm.
6. Stage 7 — dispatch (inline, ADR-0001): render idempotency_key = abc123:weekly:2026-06-15 → per-type cap reserve → translate MON 08:00 tz:user to the user’s wall-clock instant → POST /v1/gateway/notify with scheduled_time set. The gateway accepts (202), holds the send in its Valkey scheduler until 08:00 local, runs its 12 guardrails at release (§6.1), and dispatches via APNs/FCM (§6.2) → push; the durable episode renders in-app on open (§6.4).
7. Audit: one ScheduleEvent row per user, written synchronously before Run returns — dispatched (with notification_id + scheduled_at) or a skipped_* / failed outcome.
8. Analytics: DDB Streams ship every row to Snowflake; PLT-629’s funnel reads the merged (dm_run_id, user_id) row (PD-1 fire fields + the dispatch outcome).

---

8. Where the code lives

Concern	Package	Status
Registry (load PD-3 YAML → Go)	internal/dm/registry/	live
FireEvent, outcomes, pipeline	internal/dm/executor/	live
Stage dependency interfaces	internal/dm/executor/deps.go	live
Audit adapter (sync PutItem)	internal/dm/audit/	live
Shared row codec (dispatch writes it too)	internal/dm/scheduleevent/	live
CCS enricher (stage 2)	internal/dm/ccs/	live (#222)
Runner adapters (candidate / eligibility / ranking)	internal/dm/runner/	review-ready (PR-G #228)
Flag checker (stage 6)	internal/dm/flagcheck/	review-ready (PR-H #229)
Eppo experiment resolver (stage 7)	internal/dm/experiment/	review-ready (PR-I #230)
Cron concretes (due-check + claimer)	internal/dm/cronsched/	review-ready (PR-J #231)
Dispatch adapter (stage 7: gateway client + send_time + cap + content)	internal/dm/dispatch/, internal/dm/cap/	review-ready (PR-L1 #232 / PR-L2 #233 / PR-K #235; ADR-0001 — was SQS handoff)
Weekly Forecast type migration (eligibility, copy, candidate bind)	internal/dm/types/weeklyforecast/	review-ready (PR-K #235; candidate bind in PR-F #237)
Future-fire queue (store + scanner + reaper)	internal/dm/futurefire/	review-ready (PR-M #234); worker route + one-off endpoint deferred
Executor Observer (logs + §7.5 OTel counters)	cmd/unified-worker/dm_observer.go	review-ready (PR-F #237)
Worker registration / composition root	cmd/unified-worker/main.go, dm_executor_component.go	review-ready (PR-F #237)
PD-2 SQS consumer	internal/dm/delivery/	removed (ADR-0001 — folded into dispatch)

---

9. References

• specs/PD-1.md — Scheduled DM Execution (PLT-685)
• specs/PD-2.md — DM Delivery & notification-service Contract (PLT-686)
• specs/PD-3.md — DM Type Registry & Rollout (PLT-751)
• docs/decisions/0001-drop-pd1-sqs-handoff.md — ADR-0001: drop the SQS handoff, dispatch the gateway directly (the architecture change this doc reflects)
• docs/repurchase-nudge-dm.md — the pre-PD-stack hand-rolled handler this stack generalizes
• Unified-worker architecture: .claude/CLAUDE.md
• CGW PR #212 — open design: DM presentation/surface + App Open re-surfacing (the §6.3 follow-ups)
• notification-gateway behavior: PD-2 §6 (normative, Growth-owned) — the 12-step guardrail pipeline and gateway response contract
• Episodes (read/unread, BFF object): PLT-372 (live) — GET /api/v1/episodes/bff/unread, POST /episodes/ignore/{id}