fix(prism-emit): carry positive scores across epochs

crates/prism-challenge/src/orchestrator.rs

@@ -7,7 +7,8 @@
 //! [`Orchestrator::run_emitter`]) assigns every newly-finalized row to the
 //! next chain-epoch boundary's D24 set via the emission outbox
 //! (`emitted_epoch` watermark + emit cursor), so independent same-epoch
-//! scorers all land and cross-epoch evals score exactly once.
+//! scorers all land and each scoring run is assigned exactly once. Positive
+//! scores then carry into later epochs' competition sets until superseded.
 //! All state lives in the store, so the API is a pure projection and restarts
 //! sweep orphans.
 

crates/prism-emit/src/lib.rs

@@ -3,25 +3,33 @@
 //! Semantics (normative: `docs/PRISM.md` § Leaf emission):
 //!
 //! - **One D24-complete leaf set per chain epoch**, emitted by the first
-//! emitter tick that observes the epoch. The set carries every submission
-//! finalized since the previously emitted epoch — the outbox batch
-//! (`kind IS NOT NULL AND emitted_epoch IS NULL`) — competition-aggregated
+//! emitter tick that observes the epoch. The competition input is the
+//! union of (a) every submission finalized since the previously emitted
+//! epoch — the outbox batch (`kind IS NOT NULL AND emitted_epoch IS NULL`)
+//! — and (b) every still-active positive lattice score
+//! (`kind = 'score' AND score > 0`), competition-aggregated
 //! (`prism-registry`), plus `NoScore(NotAttempted)` for every other
 //! expected participant so the epoch is sealable on its own.
 //! - A row's acceptance epoch (`prism_submission.epoch`) is **intake
 //! metadata only**: a submission that finalizes after an epoch boundary
-//! (prod trains up to 6h ≫ 72-min epochs) lands in the set emitted at the
+//! (prod trains up to 6h ≫ 72-min epochs) first enters the outbox at the
 //! next boundary, never in its acceptance epoch.
-//! - **Exactly-once per scoring run**: the batch is assigned
-//! (`emitted_epoch = E`, sticky) before the submit, and the per-netuid
-//! emit cursor advances only after the full set landed. A crash between
-//! submit and cursor advance replays the sticky assigned set on the next
-//! tick; gateway leaves are first-write-wins with identical values under
-//! replay, so the retry converges. Retried/re-scored rows re-enter the
-//! outbox (`reset_for_retry` clears the watermark).
-//! - Epochs during a master outage carry no set; the first epoch after
-//! recovery includes the whole backlog (the seal always pins fresh
-//! epochs — stale ones can never Match on-chain anyway).
+//! - **Exactly-once outbox assignment per scoring run**: the fresh batch is
+//! assigned (`emitted_epoch = E`, sticky) before the submit, and the
+//! per-netuid emit cursor advances only after the full set landed. A crash
+//! between submit and cursor advance replays the sticky assigned set on
+//! the next tick; gateway leaves are first-write-wins with identical
+//! values under replay, so the retry converges. Retried/re-scored rows
+//! re-enter the outbox (`reset_for_retry` clears the watermark).
+//! - **Positive scores carry forward**: after outbox assignment, a
+//! `Score(v>0)` row keeps participating in every later epoch's
+//! competition set until a better/valid score supersedes it via `max`
+//! (lattice-proportional — not WTA). Empty or reject-only fresh batches
+//! therefore do not burn the prism share.
+//! - Epochs during a master outage carry no *new* outbox rows; the first
+//! epoch after recovery still includes active positive scores plus any
+//! backlog (the seal always pins fresh epochs — stale ones can never
+//! Match on-chain anyway).
 
 #![forbid(unsafe_code)]
 #![allow(clippy::missing_errors_doc)]
@@ -58,7 +66,7 @@ pub struct EmitSummary {
  pub epoch: u64,
  /// Full D24 set size (== `|expected|`).
  pub leaves: usize,
- /// Scored rows the set was computed from (0 = pure `NoScore` fill).
+ /// Fresh outbox rows assigned to this epoch (0 = carry-only or empty).
  pub batch: usize,
  /// The submitted signed leaves (test introspection).
  pub signed: BTreeMap<Hotkey, LeafV1>,
@@ -156,9 +164,11 @@ impl EpochEmitter {
  expected: &ExpectedSet,
  ) -> Result<EmitSummary, EmitError> {
  let n = batch.len();
+ let active = self.store.active_score_rows(self.netuid).await?;
+ let competition = merge_competition_rows(&batch, &active);
  let owners: BTreeMap<String, String> =
  self.store.arch_owners().await?.into_iter().collect();
- let signed = build_epoch_leaves(&self.sk, epoch, expected, &batch, &owners)?;
+ let signed = build_epoch_leaves(&self.sk, epoch, expected, &competition, &owners)?;
  challenge_common::submit_signed_leaf_set(&self.gateway, &signed)
  .await
  .map_err(|e| EmitError::Submit(e.to_string()))?;
@@ -173,6 +183,18 @@ impl EpochEmitter {
  }
 }
 
+/// Union of the fresh outbox batch and active positive scores for competition.
+///
+/// Duplicates (a just-assigned `Score(v>0)` row also appears in `active`) are
+/// harmless: [`prism_registry::competition_scores`] takes per-hotkey/`arch`
+/// maxima.
+fn merge_competition_rows(batch: &[EpochScoreRow], active: &[EpochScoreRow]) -> Vec<EpochScoreRow> {
+ let mut out = Vec::with_capacity(batch.len().saturating_add(active.len()));
+ out.extend_from_slice(batch);
+ out.extend_from_slice(active);
+ out
+}
+
 /// Build the signed D24 set for `epoch`: the batch competition-aggregated
 /// (owner + challenger credits, max lattice) plus `NoScore(NotAttempted)`
 /// for every expected participant without a score this epoch. Batch rows

crates/prism-emit/tests/epoch_semantics.rs

@@ -5,9 +5,10 @@
 //! (append-only first-write-wins leaves + acceptance-epoch strict match);
 //! (b) a submission accepted in epoch X but finalized in X+k never scored.
 //!
-//! Post-fix semantics: one D24-complete set per chain epoch, batch = every
-//! row finalized since the previous emitted epoch, exactly-once via the
-//! `emitted_epoch` watermark + emit cursor.
+//! Post-fix semantics: one D24-complete set per chain epoch, fresh batch =
+//! every row finalized since the previous emitted epoch (exactly-once via
+//! the `emitted_epoch` watermark + emit cursor), unioned with active
+//! positive scores so winners keep weight until a better score supersedes.
 
 #![forbid(unsafe_code)]
 #![allow(clippy::expect_used, clippy::unwrap_used)]
@@ -152,7 +153,8 @@ async fn independent_same_epoch_scorers_both_land() {
 }
 
 /// (b) A submission accepted in epoch X but finalized in X+3 lands in the
-/// next boundary's set (leaf epoch ≠ acceptance epoch), exactly once.
+/// next boundary's set (leaf epoch ≠ acceptance epoch). Outbox assignment
+/// is exactly-once; the positive score then carries into later epochs.
 #[tokio::test]
 async fn late_finalize_scores_exactly_once() {
  let store = Arc::new(MemoryPrismStore::new());
@@ -178,17 +180,17 @@ async fn late_finalize_scores_exactly_once() {
  "stamped with the emit epoch"
  );
 
- // Next epoch: the row is already emitted — it must NOT score again.
+ // Next epoch: outbox is empty (already assigned), but the positive score carries.
  let s8 = em.tick(8, &exp).await.unwrap().expect("epoch 8 emits");
  assert_eq!(s8.batch, 0, "no new finals since epoch 7");
-assert_not_attempted(&leaf_soa(&s8, 0xAA));
+assert_score(&leaf_soa(&s8, 0xAA), 500_000);
  assert_eq!(store.emit_batch(541, 7).await.unwrap().len(), 1);
  assert!(store.emit_batch(541, 8).await.unwrap().is_empty());
  assert!(store.pending_emit_epochs(541).await.unwrap().is_empty());
 }
 
-/// No double-emission: a same-epoch tick is a no-op, and a later epoch's
-/// batch contains only rows finalized since (never a replay of old scores).
+/// Outbox assignment is exactly-once per finalize; positive scores still
+/// carry into later epochs alongside any new finals (lattice max).
 #[tokio::test]
 async fn no_double_emission_across_epochs() {
  let store = Arc::new(MemoryPrismStore::new());
@@ -212,7 +214,7 @@ async fn no_double_emission_across_epochs() {
  "same epoch is a no-op"
  );
 
- // A second scorer finalizes during epoch 7; only that row is in the next batch.
+ // A second scorer finalizes during epoch 7; only that row is freshly assigned.
  store
  .insert_queued(&scored_row(
  "sub-b",
@@ -223,13 +225,52 @@ async fn no_double_emission_across_epochs() {
  .await
  .unwrap();
  let s8 = em.tick(8, &exp).await.unwrap().expect("epoch 8 emits");
- assert_eq!(s8.batch, 1, "only the new row");
-assert_not_attempted(&leaf_soa(&s8, 0xAA));
+ assert_eq!(s8.batch, 1, "only the new row is freshly assigned");
+assert_score(&leaf_soa(&s8, 0xAA), 100_000);
  assert_score(&leaf_soa(&s8, 0xBB), 900_000);
  assert_eq!(store.emit_batch(541, 7).await.unwrap().len(), 1);
  assert_eq!(store.emit_batch(541, 8).await.unwrap().len(), 1);
 }
 
+/// Prod incident regression: winner emitted at epoch N, epoch N+1 seals with
+/// only a Score(0) reject — winner weights must still appear (no burn).
+#[tokio::test]
+async fn positive_score_carries_when_next_epoch_is_reject_only() {
+ let store = Arc::new(MemoryPrismStore::new());
+ store
+ .insert_queued(&scored_row(
+ "winner",
+ &hk(0xAA),
+ 24353,
+ FinalScore::Score(800_000),
+ ))
+ .await
+ .unwrap();
+
+ let em = dry_emitter(&store);
+ let exp = expected(&[0xAA, 0xBB]);
+ let s = em.tick(24353, &exp).await.unwrap().expect("epoch 24353");
+ assert_eq!(s.batch, 1);
+ assert_score(&leaf_soa(&s, 0xAA), 800_000);
+
+ store
+ .insert_queued(&scored_row(
+ "reject",
+ &hk(0xBB),
+ 24354,
+ FinalScore::Score(0),
+ ))
+ .await
+ .unwrap();
+ let s2 = em.tick(24354, &exp).await.unwrap().expect("epoch 24354");
+ assert_eq!(s2.batch, 1, "reject is freshly assigned once");
+ assert_score(&leaf_soa(&s2, 0xAA), 800_000);
+ assert_score(&leaf_soa(&s2, 0xBB), 0);
+ assert_eq!(store.emit_batch(541, 24353).await.unwrap().len(), 1);
+ assert_eq!(store.emit_batch(541, 24354).await.unwrap().len(), 1);
+ assert_eq!(store.active_score_rows(541).await.unwrap().len(), 1);
+}
+
 /// Competition credit semantics (prism-registry) are preserved through the
 /// outbox — both when owner + challenger finalize in the same epoch and when
 /// the challenger lands in a later epoch (the run6 scenario, minus the
@@ -328,7 +369,8 @@ async fn crash_replays_sticky_assignment() {
  assert!(store.pending_emit_epochs(541).await.unwrap().is_empty());
  assert_eq!(s.epoch, 11);
  assert_eq!(s.batch, 0, "the backlog was already assigned to epoch 9");
- assert_not_attempted(&leaf_soa(&s, 0xAA));
+ // Active positive score still carries into the live epoch after recovery.
+ assert_score(&leaf_soa(&s, 0xAA), 700_000);
  // The epoch-9 replay kept the score (sticky batch content).
  assert_eq!(store.emit_batch(541, 9).await.unwrap().len(), 1);
 }

crates/prism-store/src/dbprism.rs

@@ -356,6 +356,10 @@ impl PrismStore for DbPrismStore {
  .await
  }
 
+ async fn active_score_rows(&self, netuid: u16) -> Result<Vec<EpochScoreRow>, StoreError> {
+ crate::emit::active_score_rows(&self.pool, i32::from(netuid)).await
+ }
+
  async fn emit_cursor(&self, netuid: u16) -> Result<Option<u64>, StoreError> {
  crate::emit::emit_cursor(&self.pool, i32::from(netuid)).await
  }

crates/prism-store/src/emit.rs

@@ -7,6 +7,8 @@
 //! Outbox contract (see `docs/PRISM.md` § Leaf emission):
 //! - [`assign_emit_batch`] stamps every scored-not-yet-emitted row with the
 //! target leaf epoch and returns the batch (sticky assignment);
+//! - [`active_score_rows`] returns every positive lattice score for the netuid
+//! so the emitter can re-include the live champion set each epoch (carry);
 //! - the emitter submits one D24-complete set for that epoch, then advances
 //! [`set_emit_cursor`];
 //! - [`pending_emit_epochs`] finds assigned-but-incomplete epochs so a crash
@@ -80,6 +82,28 @@ pub(crate) async fn emit_batch(
  Ok(rows_to_epoch(rows))
 }
 
+/// Positive lattice scores still eligible for epoch-close competition carry.
+///
+/// `Score(0)` rejects and `NoScore` absences are excluded — they must not
+/// displace a prior valid winner when an epoch's fresh outbox is empty or
+/// burn-only. Competition aggregation takes `max` over the union of the
+/// fresh batch and this set, so a better later score supersedes naturally.
+pub(crate) async fn active_score_rows(
+ pool: &PgPool,
+ netuid: i32,
+) -> Result<Vec<EpochScoreRow>, StoreError> {
+ let rows: Vec<EmitSqlRow> = sqlx::query_as(
+ "SELECT miner_hotkey, arch_id, kind, score, absence_reason \
+ FROM prism_submission \
+ WHERE netuid = $1 AND kind = 'score' AND score > 0",
+ )
+ .bind(netuid)
+ .fetch_all(pool)
+ .await
+ .map_err(backend)?;
+ Ok(rows_to_epoch(rows))
+}
+
 /// Highest leaf epoch whose set fully landed for `netuid` (`None` = never).
 pub(crate) async fn emit_cursor(pool: &PgPool, netuid: i32) -> Result<Option<u64>, StoreError> {
  let row: Option<(i64,)> =

crates/prism-store/src/store.rs

@@ -311,6 +311,12 @@ pub trait PrismStore: Send + Sync + std::fmt::Debug {
  /// after a crash between submit and cursor advance).
  async fn emit_batch(&self, netuid: u16, epoch: u64) -> Result<Vec<EpochScoreRow>, StoreError>;
 
+ /// Positive lattice scores (`Score(v)` with `v > 0`) for competition
+ /// carry-forward across epochs. Outbox assignment stays exactly-once;
+ /// these rows are re-read every tick so a prior winner is not burned
+ /// when a later epoch's fresh batch is empty or reject-only.
+ async fn active_score_rows(&self, netuid: u16) -> Result<Vec<EpochScoreRow>, StoreError>;
+
  /// Highest leaf epoch whose set fully landed (`None` = never emitted).
  async fn emit_cursor(&self, netuid: u16) -> Result<Option<u64>, StoreError>;
 
@@ -633,6 +639,25 @@ impl PrismStore for MemoryPrismStore {
  .collect())
  }
 
+ async fn active_score_rows(&self, netuid: u16) -> Result<Vec<EpochScoreRow>, StoreError> {
+ let rows = self
+ .rows
+ .lock()
+ .map_err(|_| StoreError::Backend("poison".into()))?;
+ Ok(rows
+ .iter()
+ .filter(|r| r.netuid == netuid)
+ .filter_map(|r| match &r.final_score {
+ Some(FinalScore::Score(v)) if *v > 0 => Some(EpochScoreRow {
+ miner_hotkey: r.miner_hotkey.clone(),
+ arch_id: r.arch_id.clone(),
+ final_score: FinalScore::Score(*v),
+ }),
+ _ => None,
+ })
+ .collect())
+ }
+
  async fn emit_cursor(&self, netuid: u16) -> Result<Option<u64>, StoreError> {
  Ok(self
  .cursors
@@ -936,13 +961,20 @@ mod tests {
  assert_eq!(s.pending_emit_epochs(541).await.unwrap(), vec![9]);
  s.set_emit_cursor(541, 9).await.unwrap();
  assert_eq!(s.emit_cursor(541).await.unwrap(), Some(9));
+ // Positive scores remain active for carry after outbox assignment.
+ assert_eq!(s.active_score_rows(541).await.unwrap().len(), 1);
  // Monotonic: a stale cursor write never regresses.
  s.set_emit_cursor(541, 4).await.unwrap();
  assert_eq!(s.emit_cursor(541).await.unwrap(), Some(9));
  assert!(s.pending_emit_epochs(541).await.unwrap().is_empty());
  // Unscored rows never enter a batch.
  s.insert_queued(&row("b", "22")).await.unwrap();
  assert!(s.assign_emit_batch(541, 10).await.unwrap().is_empty());
+ // Score(0) rejects are not active carry rows.
+ let mut zero = row("c", "33");
+ zero.final_score = Some(FinalScore::Score(0));
+ s.insert_queued(&zero).await.unwrap();
+ assert_eq!(s.active_score_rows(541).await.unwrap().len(), 1);
  }
 
  #[tokio::test]

docs/COMPLETENESS.md

@@ -104,7 +104,7 @@ Agent/operator contracts: root [`AGENTS.md`](../AGENTS.md), [`deploy/AGENTS.md`]
 | design rating / elimination | done | Integer Elo (K=32), bottom 20% / 4-round cooldown, exact-E leaves. |
 | design API | done | Harness/quota/runs/viewer/annotate/ops on `:8093`. |
 | prism Lium backend | done | `PRISM_FORCE_SIM=false` in staging; the binary logs `eval_backend=lium`. API key is mounted from a file so it never appears in `docker inspect`. |
-| prism orchestration | done | DB-backed claim/execute/review/similarity/score state machine (`prism_submission` + append-only `prism_stage_event`), sweeper (7h grace), boot recovery, epoch-close batched D24 leaf emission (`prism-emit` outbox: `emitted_epoch` watermark + `prism_emit_cursor`, migration 0012). |
+| prism orchestration | done | DB-backed claim/execute/review/similarity/score state machine (`prism_submission` + append-only `prism_stage_event`), sweeper (7h grace), boot recovery, epoch-close batched D24 leaf emission (`prism-emit` outbox: `emitted_epoch` watermark + `prism_emit_cursor` + positive-score carry each epoch, migration 0012). |
 | prism recipe v1 | done | `prism-recipe` contract, fineweb-edu pinned shard (URL + SHA-256, harness re-verifies), 6h train / 7h pod caps, baseline sources, recipe pin hex on the API. |
 | prism LLM review | done | `prism-review` quality + similarity prompts (versioned), OpenRouter client (key file only, never env), deterministic sim fallback; anti-copy forces `Copied`/`Suspicious` → Score 0. |
 | prism API | done | Full status surface: submissions list/detail/events/status/jobs/recipe/baseline, idempotent accept. |