Levandor

Cohort First-Mover → NATS Migration (cohort.signals ingestion lane)

11 min read

The cohort first-mover ENTRY signal no longer runs in-process inside the watcher. It now publishes to NATS subject cohort.signals and is consumed by the same long-running Pmv2Consumer that already handles @weather — one unified ingestion lane routed by the envelope source field. The in-process runner::run_entry was deleted. Built + reviewed + gate-green on autotrade-foundation; NOT merged, NOT deployed.

For Agents — state at a glance

Branch autotrade-foundation (off main). 4 commits in order: 1166f51 (T1 wire parse → Context, fail-closed) → 3c589c1 (T2 serialize_cohort + roundtrip parity) → 9505840 (T3 publish_cohort, Nats-Msg-Id dedup) → bdafd7f (T4 cutover + delete run_entry). Gate green as reported by the orchestrator: 428 polymarket-fetch + 89 polymarket-data. Each task spec- and code-quality-reviewed; T4 also trade-safety reviewed = SAFE. Spec: docs/superpowers/specs/2026-06-17-cohort-firstmover-nats-migration-design.md. Plan: docs/superpowers/plans/2026-06-17-cohort-firstmover-nats-migration.md. This sits directly on top of pmv2-position-manager-nats-consumer (which built the consumer + weather lane). That note’s wire::parse_signal, engine Context arm, and the single submit_live site are untouched here — only the cohort transport changed from in-process call to NATS publish.

What changed (before → after)

Before — the cohort entry signal was ingested in-process by the watcher, synchronously, every scheduled run:

run_pmv2_positions → run_autotrade_entry → runner::run_entry → engine::ingest_entry

After — the watcher publishes the entry signal to NATS; a separate long-running process consumes it:

run_pmv2_positions → run_autotrade_entry → nats::publish_cohort ──▶ cohort.signals
                                                                        │
   Pmv2Consumer (one durable, both subjects) ◀── weather.signals ──────┘
        └─ wire::parse_signal (routes on `source`) → engine::ingest_entry → submit_live

runner::run_entry is gone. The cohort and @weather now share one ingestion lane; wire::parse_signal dispatches on the envelope source field (first_mover_core ⇒ cohort Context, weather ⇒ Directive).

T1 — wire parse routes first_mover_core → Context, fail-closed (1166f51)

crates/polymarket-fetch/src/pmv2/autotrade/wire.rs. parse_signal now recognizes source == "first_mover_core" and builds an Execution::Context TradeSignal. The fail policy is fail-CLOSED (Reject) on any capital-critical missing/empty field — because a malformed cohort entry that slips through can trap capital:

FieldIf missing/emptyWhy it’s capital-critical
origin_refRejectEmpty ⇒ cohort exits never fire → position can never be auto-closed = capital trap
outcome_labelRejectEmpty ⇒ token_for fails ⇒ wrong/absent token
reference_priceRejectDrives the chase gate; absent ⇒ unsafe sizing/entry
convictionRejectSizing input
outcome_indexRejectSelects the outcome

max_price and valid_until_ts became Option on the parse path because the cohort omits them:

valid_until_ts = None is exact parity, not a degradation

The cohort never set an expiry. Treating valid_until_ts = None (no staleness deadline) is exact behavioral parity with the old in-process path AND avoids clock-skew wrongful drops — a tight TTL on a clock-skewed consumer could Stale-drop a perfectly valid entry. max_price is likewise Option (cohort derives its limit from gates, not a producer cap).

T2 — serialize_cohort + the headline cohort_roundtrip_parity test (3c589c1)

crates/polymarket-fetch/src/pmv2/autotrade/wire.rs (serialize_cohort at wire.rs:158, test at wire.rs:463). The migration’s central safety proof:

The parity invariant

serialize_cohort ∘ parse_signal is byte-equal to sources::enter_signal over the full persisted + exit-read field set — i.e. the new wire round-trip reconstructs the exact TradeSignal the old in-process emitter produced. The fixtures use distinct event_slug ≠ slug values so a field-collapse bug (one overwriting the other) would be caught, not masked.

The reference point is autotrade-sources-signal-emitter’s enter_signal — the original first-mover emitter. As long as the round-trip equals it, the cutover cannot change what the engine sees.

Subtlety: tick/min/neg_risk None→0.0→Some(0.0) is immaterial

At detection time the watcher has no CLOB data, so enter_signal leaves tick_size/min_size/neg_risk = None. serialize_cohort emits 0.0 / 0.0 / false on the wire; parse_signal rebuilds Some(0.0) / Some(0.0) / Some(false). This is immaterial because the engine Context arm re-resolves tick/min/neg_risk from the live CLOB during plan_context and never reads the signal’s values for them. The round-trip is not bit-identical on these three fields, but the divergence is provably unused.

T3 — nats::publish_cohort, JetStream dedup (9505840)

crates/polymarket-fetch/src/pmv2/autotrade/nats.rs (publish_cohort at nats.rs:53). Serialize via serialize_cohort, then publish_with_headers with Nats-Msg-Id = signal_id (nats.rs:50), then flush. Reuses the existing nats::connect from the consumer adapter.

The Nats-Msg-Id = signal_id header is the first of two dedup layers: it engages JetStream server-side message dedup so a retried publish of the same signal_id is collapsed by the broker before it ever reaches the consumer.

T4 — Cutover: publish + delete run_entry (bdafd7f, trade-safety reviewed)

crates/polymarket-fetch/src/pmv2/positions.rs (run_autotrade_entry at positions.rs:608). The watcher now publishes instead of ingesting in-process.

Publish gating — run_autotrade_entry publishes only when active AND Core tier AND !breaker:

The old !ready preflight clause was intentionally dropped

The previous in-process gate included !ready (an entry-preflight check). It was removed from the publisher because preflight is now a consume-time concern — the consumer (pmv2-position-manager-nats-consumer) is where reserve/dedup/mode/submit live. The publisher’s job is only “should this entry be offered to the lane at all” (active + Core + breaker-clear). See the load-bearing gotcha below for the consequence.

Bounded retry — PUBLISH_RETRIES = 3, 200ms backoff. A dropped publish after retries is a logged MISSED entry = an under-trade, never a double. No Telegram alert, no breaker bump — a missed first-mover entry is a benign no-fill, not a failure.

Publisher lifecycle — built once per run (positions.rs:484), gated on the cohort lane being active, from env NATS_URL + NATS_COHORT_NKEY_SEED. Absent env ⇒ dormant ⇒ nothing published (logs cohort publisher dormant (NATS_URL / NATS_COHORT_NKEY_SEED unset)).

Deletions in T4:

  • runner::run_entry — the in-process entry path (now superseded by publish → consume).
  • safety::ready — the now-dead in-process entry-preflight fn.
  • AutotradeDeps.mode — an unused field.

For Agents — verified deletions

grep "fn run_entry" and grep "fn ready" safety.rs both return nothing on autotrade-foundation. publish_cohort / serialize_cohort / the cohort_roundtrip_parity test all present. run_autotrade_entry reads NATS_COHORT_NKEY_SEED (positions.rs:484).

Why it’s safe (trade-safety verdict: SAFE)

No double-LIVE order — two independent dedup layers on signal_id

  1. NATS header Nats-Msg-Id = signal_id → JetStream server-side dedup (T3).
  2. DB backstop ON CONFLICT(dedup_key) DO NOTHING on signal_id → ≤1 order per signal_id, ever.

The money-path is untouched: the engine Context decision/submit arm, wire::parse_signal’s validation, and the single submit_live site are all unchanged — only transport changed (in-process call → NATS publish). A dormant cohort (global mode off, or env unset) publishes nothing.

This dedup continuity inherits directly from pmv2-position-manager-nats-consumer’s T1: signal_id == legacy orders::dedup_key(.., "enter", ..), the exact 6-discriminator string the old path produced — so even running old + new side-by-side could not double-order.

⚠️ The load-bearing gotcha — entry-reconcile + preflight must be restored before arming ANY lane

Deleting autotrade_ready() removed BOTH in-process entry-preflight AND per-tick reconcile_entries

The watcher-side autotrade_ready() did two jobs: entry-preflight AND the per-tick reconcile_entries. Deleting it (T4) removed both from the watcher loop.

Safe TODAY because: the cohort is dormant / global mode off, no live orders exist, and /autotrade reconcile still works manually.

BUT the Pre-arm consumer-preflight gate — the already-documented blocker shared with @weather — must now ALSO restore entry-reconcile + preflight before arming ANY lane live. Otherwise the first live lane (cohort or weather) will have stuck pending/submitted orders that nothing auto-reconciles. This is no longer weather-only scope; the cohort migration folded entry-reconcile into the same pre-arm gate. Treat “restore per-tick entry-reconcile + preflight in the consumer” as a hard prerequisite to the first live arm of either lane.

Deploy / coordination (NOT done — this is T5, remaining)

Stream: ADD a subject, do NOT rename

The JetStream stream stays WEATHER_SIGNALS (its name is unchanged — renaming would orphan the durable). ADD cohort.signals as a second subject on that stream. The pmv2 durable must consume BOTH weather.signals AND cohort.signals.

Remaining T5 deploy/coordination tasks (owned by @deploy / @weather):

  1. Stream: add subject cohort.signals to stream WEATHER_SIGNALS (do not rename). Bind the pmv2 durable to both subjects.
  2. Provision a cohort-publisher NKey (distinct publisher identity with pub perms on cohort.signals).
  3. Watcher env: put NATS_URL + NATS_COHORT_NKEY_SEED in the watcher process env (the publisher reads these; absent ⇒ dormant).
  4. @weather: cohort is a 2nd source on the same lane, no schema change — the envelope fields the cohort omits (max_price, valid_until_ts, etc.) are Option / default at schema v1, so the existing v1 envelope absorbs the cohort with no version bump.
  5. Deploy the branch (still off main) and restore the pre-arm entry-reconcile + preflight gate (see the load-bearing gotcha) before any live arm.

Review hardening (post-cutover 3-lens review → 39c01dd)

After the T4 cutover (bdafd7f) was committed, a 3-lens adversarial review (bugs / gaps / DRY) was run at the operator’s request. It found one real silent-failure landmine plus operator-visibility gaps, all fixed in 39c01dd (gate green: 429 polymarket-fetch + 89 polymarket-data; trade-safety reviewed = SAFE). Single-order safety is unchanged.

The landmine — core-NATS publish silently dropped every cohort entry into nothing

The T3 nats::publish_cohort originally published through the CORE async-nats client (client.publish_with_headers(...) + client.flush()). Core NATS publish is fire-and-forget: it returns NO JetStream PubAck and does NOT error when no stream’s subject filter captures the subject.

Consequence: if @deploy did not bind cohort.signals into the WEATHER_SIGNALS stream (or set a filter_subject that excludes it), every cohort entry would publish “successfully” into nothing — the consumer never sees it, zero error is logged anywhere, and the operator believes the lane is live. This is exactly the T5 deploy footgun (the “ADD a subject, do not rename” coordination step), and the old code could not detect it.

The fix (39c01dd) — publish through JetStream + await the ack. publish_cohort now builds a JetStream context (async_nats::jetstream::new(client.clone())), calls publish_with_headers(...).await to get a PublishAckFuture, then .awaits that future for the PubAck, the whole thing wrapped in a 5s tokio::time::timeout. Compiler-pinned against the vendored async-nats 0.49.1: a missing/misfiltered subject yields NO_RESPONDERS → StreamNotFound, i.e. a loud Err that the bounded retry sees and then surfaces. Persistence is now confirmed (PubAck) rather than assumed. Single-order safety is unchanged — Nats-Msg-Id = signal_id stream-dedup + the authoritative DB ON CONFLICT(dedup_key) DO NOTHING backstop both still apply.

Operator-visibility fixes — a dropped first-mover entry is a PERMANENT miss

The existing telegram notify closure was threaded into the entry path so the two failure modes are no longer silent:

  1. “Armed but unconfigured” lane — cohort active but the watcher’s NATS_URL / NATS_COHORT_NKEY_SEED are unset — now warns + telegrams instead of dropping entries silently at debug (which was previously indistinguishable from intentional dormancy).
  2. Retry-exhaustion now telegrams a “first-mover entry LOST” alarm.

Why it matters: re-detection on the next tick is impossible — the end-of-run position snapshot upserts unconditionally, so the entry that triggered the publish is already recorded as a held position and will not re-fire. A dropped publish is therefore a PERMANENT miss, not a transient one. (Contrast T4’s framing of a dropped publish as a benign under-trade: it is benign for capital safety — never a double — but it is not self-healing, so it must be visible.)

A small pure classifier publisher_intent(at_active, has_url, has_seed) -> {Dormant, Unconfigured, Ready} (unit-tested) drives the Dormant-vs-Unconfigured distinction. Also named two literals: COHORT_SUBJECT, and wire::SCHEMA_VERSION_MAX (replacing a hardcoded 1, so the producer tracks the parser’s max).

T5 coordination addendum (supersedes the bare "provision a NKey" step)

The cohort-publisher NKey must permit JetStream publish to cohort.signals — i.e. publish + receive the PubAck on its inbox — not merely core publish. A core-publish-only permission would now surface as a publish failure (loud) rather than a silent drop, but the correct grant is JetStream-publish.

What the review verified CLEAN (do not re-litigate)

  • End-to-end field fidelity — the wire round-trip carries every persisted/exit-read field (the cohort_roundtrip_parity invariant from T2 holds end-to-end).
  • Dedup identity — Nats-Msg-Id == DB dedup_key == the legacy orders::dedup_key(.., "enter", ..) key; one identity across all three layers.
  • No double-order — the two independent dedup layers on signal_id are intact.
  • No worthwhile duplication (DRY lens) and zero comments / zero dead-code.

Reusable lesson — async-nats core publish is fire-and-forget

async_nats::Client::publish* (core) gives you no ack and no error on a missing subject — Msg-Id stream-dedup still applies, but delivery is unconfirmed. Whenever you need delivery confirmation OR loud-on-missing-subject behavior, publish via async_nats::jetstream::Context::publish* and .await the PublishAckFuture (ideally under a timeout). See the cross-ref under TOPICS Gotchas / Bugs Fixed.

Graph View

Backlinks