fix(engine): stop search deliveries double-collecting ETB/landfall triggers

crates/engine/src/game/engine_priority.rs

@@ -100,17 +100,13 @@ pub(crate) fn run_post_action_pipeline_from(
             }
         }
         // A completed logical owner has already collected its segment and
-        // settlement contexts into the existing deferred queue. The owner is
-        // intentionally gone before the trailing completion event, so use those
-        // exact queued occurrences to keep the generic scan from rediscovering
-        // them while still allowing every unrelated event through.
-        let deferred_logical_zone_events: Vec<_> = state
-            .deferred_triggers
-            .iter()
-            .flat_map(|context| context.trigger_events.iter())
-            .filter(|event| matches!(event, GameEvent::ZoneChanged { .. }))
-            .collect();
-        let unconsumed_events = triggers::filter_consumed_trigger_events_from(
+        // settlement contexts into the deferred queue, and a paused owner that
+        // drained may instead have claimed them in the consumed ledger.
+        // `filter_already_collected_trigger_events_from` is the single authority
+        // for both (CR 603.2c), shared with the search-delivery park family so
+        // the two collectors cannot drift.
+        let unconsumed_events = triggers::filter_already_collected_trigger_events_from(
+            state,
             events,
             event_start,
             &consumed_trigger_events,
@@ -121,7 +117,6 @@ pub(crate) fn run_post_action_pipeline_from(
                 !matches!(event, GameEvent::PhaseChanged { .. })
                     && !state.deferred_entry_events.contains(event)
                     && !retained_logical_zone_events.contains(event)
-                    && !deferred_logical_zone_events.contains(event)
             })
             .cloned()
             .collect();

crates/engine/src/game/engine_resolution_choices.rs

@@ -567,6 +567,13 @@ fn batch_or_drain_observer_triggers(
     events: &mut Vec<GameEvent>,
     event_slice_start: usize,
     event_slice_end: usize,
+    // CR 603.2c: `true` declares that `events[event_slice_start..event_slice_end]`
+    // is exactly one completed logical zone-change owner's completion slice.
+    // `LogicalZoneChangeGroup::append_delivery_events` retains EVERY `ZoneChanged`
+    // in the slice it is handed, so within such a slice a blanket drop is
+    // equivalent to per-occurrence suppression. A collector whose slice is not
+    // owner-bounded must use
+    // `triggers::filter_already_collected_trigger_events_from` instead.
     zone_changes_are_logically_owned: bool,
 ) -> Option<ResolutionChoiceOutcome> {
     if matches!(state.waiting_for, WaitingFor::Priority { .. }) {
@@ -611,15 +618,30 @@ fn batch_or_drain_observer_triggers(
 /// continuation drains, park ETB/dies/discards observers for the next priority
 /// checkpoint instead of dispatching them while the test harness (or UI) may
 /// still be inside the same `SelectCards` action (issue #5336).
+///
+/// CR 603.2c: this slice spans the whole continuation drain, so it holds both
+/// the delivery's logical zone-change owner's occurrences (already collected by
+/// `change_zone::resolve` / `zone_pipeline::move_objects_simultaneously_then`)
+/// AND zone changes no owner allocated a group for. It is therefore NOT
+/// owner-bounded and cannot blanket-drop `ZoneChanged` the way
+/// `batch_or_drain_observer_triggers` does; it consults the shared ownership
+/// authority instead. That authority's ledger half applies to every event kind,
+/// matching the generic priority scan. Without it a fetched land's landfall/ETB
+/// observers fire twice.
 fn park_search_observer_triggers(
     state: &mut GameState,
     events: &[GameEvent],
     events_before_drain: usize,
 ) -> ResolutionChoiceOutcome {
-    let trigger_events: Vec<GameEvent> = events[events_before_drain..]
-        .iter()
+    let uncollected_events = super::triggers::filter_already_collected_trigger_events_from(
+        state,
+        events,
+        events_before_drain,
+        &state.consumed_before_priority_trigger_events,
+    );
+    let trigger_events: Vec<GameEvent> = uncollected_events
+        .into_iter()
         .filter(|ev| !matches!(ev, GameEvent::PhaseChanged { .. }))
-        .cloned()
         .collect();
     if !trigger_events.is_empty() {
         super::triggers::collect_triggers_into_deferred(state, &trigger_events);

crates/engine/src/game/triggers.rs

@@ -8276,6 +8276,107 @@ pub(crate) fn filter_consumed_trigger_events(
     filter_consumed_trigger_events_from(events, 0, consumed)
 }
 
+/// CR 603.2c: Remove from `events[event_start..]` the occurrences a trigger
+/// collector has already taken, so a second collector over the same raw slice
+/// cannot fire the same observers twice.
+///
+/// Two witnesses answer "already collected", and neither is sufficient alone:
+///
+/// 1. `consumed` — occurrences explicitly claimed by
+///    [`mark_logical_zone_events_consumed_before_priority`]. Required wherever an
+///    intervening `drain_deferred_trigger_queue` has already emptied
+///    `deferred_triggers`. Only three owners mark, and their ordinals are NOT
+///    uniformly exact: `effects/mod.rs` passes the whole action buffer (exact);
+///    `zone_pipeline.rs`'s synchronous-completion site passes a sub-slice, so its
+///    ordinals are rebased; and `zone_pipeline.rs`'s batch-drain site passes the
+///    whole buffer only after `drain_pending_batch_deliveries` has moved it out
+///    and re-assembled it, so its ordinals are computed against a re-ordered
+///    buffer. See the warning on [`filter_consumed_trigger_events_from`].
+/// 2. `state.deferred_triggers` — the `ZoneChanged` values carried by contexts
+///    that [`complete_logical_zone_trigger_collection`] and
+///    [`append_and_collect_logical_zone_trigger_segment`] already queued. This is
+///    the only witness for the four owners that deliberately do NOT mark
+///    (`effects/change_zone.rs` x2, `engine_resolution_choices.rs` x2). Do NOT
+///    "fix" that asymmetry by adding `mark_`: claiming an occurrence also hides
+///    it from `check_delayed_triggers` (`engine_priority.rs`), which would
+///    silently kill the CR 603.7b leaves-the-battlefield delayed family (an
+///    ability triggers only the next time its trigger event occurs; hide the
+///    event and it never triggers).
+///
+/// WITNESS 2 IS A BOUND, NOT AN OCCURRENCE COUNT. `deferred_triggers` holds one
+/// context per matching observer, not one entry per occurrence (every zone-change
+/// collection site pushes one `PendingTriggerContext::batched` per matched
+/// `(object_id, trig_idx)`, and for a non-batched trigger `matched.trigger_events`
+/// is the singleton `vec![event.clone()]`, so one context is one witness copy; a
+/// batched trigger carries its whole matched batch, which yields more witnesses,
+/// never fewer), so N observers of ONE occurrence contribute N copies of that
+/// value. Consuming
+/// witnesses one-for-one therefore removes at most `min(queued_copies,
+/// slice_copies)` — never more than the set-membership filter this replaces at
+/// the priority scan, which removed every copy. It is NOT occurrence-exact and
+/// does NOT by itself discharge CR 603.2c's second sentence ("it can trigger
+/// repeatedly if one event contains multiple occurrences"): if a slice holds a
+/// byte-identical `ZoneChanged` that no owner collected alongside one that two
+/// observers saw, both are dropped. At the priority scan that residual is no
+/// larger than the filter this replaces; at the search-delivery park there is no
+/// prior `ZoneChanged` filter at all, so the residual is new there and is bounded
+/// by byte-identical `ZoneChanged` duplicates being unreachable inside one
+/// collector slice. An occurrence-exact witness is NOT available here: the only
+/// exact record is `LogicalZoneChangeGroup::all_origin_occurrences`, and a
+/// completed owner's group is a caller-owned local that is gone before this runs
+/// (`GameState` holds a group only inside the two *paused* frames,
+/// `PendingChangeZoneIteration` and `PendingBatchDeliveries`).
+///
+/// A collector whose slice is provably exactly one owner's completion slice does
+/// NOT need this — a blanket `ZoneChanged` drop is equivalent there, and that is
+/// what `engine_resolution_choices::batch_or_drain_observer_triggers`
+/// (owner-bounded slice + `zone_changes_are_logically_owned`) and the resumed
+/// `ChangeZone` drain in `effects/mod.rs` do. `park_search_observer_triggers`'
+/// slice spans a whole continuation drain and can hold zone changes no owner
+/// allocated a group for, so it must consult this instead.
+///
+/// Three further raw-slice collectors exist. [`park_observer_triggers_if_paused`]
+/// and [`collect_and_drain_observer_triggers_if_settled`] are not on any path that
+/// follows a logical zone-change owner today. The third — `engine_priority`'s
+/// exile-return pass — ALREADY follows one (`check_exile_returns` delivers through
+/// `zone_pipeline::move_objects_simultaneously_then`, which completes and marks),
+/// and it applies the ledger half ONLY, not the queued-context witness. It is safe
+/// today solely because that owner marks; if `zone_pipeline` ever stops marking —
+/// which is the right call for four of the seven owners, per witness 2 above — that
+/// collector is exposed. A future caller that puts any of the three after a
+/// `complete_logical_zone_trigger_collection` must route it through here.
+pub(crate) fn filter_already_collected_trigger_events_from(
+    state: &GameState,
+    events: &[GameEvent],
+    event_start: usize,
+    consumed: &[ConsumedTriggerEventOccurrence],
+) -> Vec<GameEvent> {
+    let mut queued_zone_change_witnesses: Vec<&GameEvent> = state
+        .deferred_triggers
+        .iter()
+        .flat_map(|context| context.trigger_events.iter())
+        .filter(|event| matches!(event, GameEvent::ZoneChanged { .. }))
+        .collect();
+    filter_consumed_trigger_events_from(events, event_start, consumed)
+        .into_iter()
+        .filter(|event| {
+            if !matches!(event, GameEvent::ZoneChanged { .. }) {
+                return true;
+            }
+            match queued_zone_change_witnesses
+                .iter()
+                .position(|queued| *queued == event)
+            {
+                Some(index) => {
+                    queued_zone_change_witnesses.remove(index);
+                    false
+                }
+                None => true,
+            }
+        })
+        .collect()
+}
+
 /// CR 603.2c + CR 510.2: Expand a multi-fire `WheneverEvent` `DamageDone`
 /// trigger's aggregate `CombatDamageDealtToPlayer` matches into one synthetic
 /// per-source `DamageDealt` event per matching (source, defending player)

crates/engine/src/game/triggers_dedup_regression_tests.rs

@@ -3646,3 +3646,194 @@ fn order_triggers_apnap_three_players() {
         );
     }
 }
+
+// ---------------------------------------------------------------------------
+// CR 603.2c: the shared "already collected" authority
+// (`filter_already_collected_trigger_events_from`).
+//
+// These pin the exact semantics of the queued-context witness, which is a BOUND
+// and not an occurrence count: `deferred_triggers` holds one context per matching
+// observer, so N observers of ONE occurrence contribute N copies of that value.
+// ---------------------------------------------------------------------------
+
+/// A byte-identical `ZoneChanged` builder — `ZoneChangeRecord::test_minimal` is
+/// fully deterministic, so two calls with the same arguments compare equal.
+fn zone_change_event(object_id: ObjectId) -> GameEvent {
+    GameEvent::ZoneChanged {
+        object_id,
+        from: Some(Zone::Library),
+        to: Zone::Battlefield,
+        record: Box::new(ZoneChangeRecord::test_minimal(
+            object_id,
+            Some(Zone::Library),
+            Zone::Battlefield,
+        )),
+    }
+}
+
+/// One queued context carrying exactly one copy of `event`, matching the
+/// one-witness-copy-per-matched-observer shape
+/// `collect_pending_triggers_with_collection` produces. That function builds
+/// `PendingTriggerContext::batched(matched.pending, matched.trigger_events)`, but
+/// for a non-batched trigger `matched.trigger_events` is the singleton
+/// `vec![event.clone()]` — so a `::single` context is the same one-copy shape and
+/// is used here because `::batched` is private to `triggers`.
+fn queued_context_for(event: GameEvent) -> PendingTriggerContext {
+    PendingTriggerContext::single(PendingTrigger {
+        source_id: ObjectId(99),
+        controller: PlayerId(0),
+        condition: None,
+        ability: Box::new(ResolvedAbility::new(
+            Effect::Draw {
+                count: QuantityExpr::Fixed { value: 1 },
+                target: TargetFilter::Controller,
+            },
+            Vec::new(),
+            ObjectId(99),
+            PlayerId(0),
+        )),
+        timestamp: 0,
+        target_constraints: Vec::new(),
+        distribute: None,
+        trigger_event: Some(event),
+        modal: None,
+        mode_abilities: Vec::new(),
+        description: None,
+        may_trigger_origin: None,
+        subject_match_count: None,
+        die_result: None,
+        provenance: None,
+    })
+}
+
+fn zone_change_count(events: &[GameEvent]) -> usize {
+    events
+        .iter()
+        .filter(|event| matches!(event, GameEvent::ZoneChanged { .. }))
+        .count()
+}
+
+/// U1 — the queued witness is COUNT-LIMITED, not set membership.
+///
+/// Two byte-identical `ZoneChanged` in the slice against ONE queued context
+/// carrying that value must leave exactly one survivor. Set membership would
+/// return 0, and so would a blanket `ZoneChanged` drop; both are wrong, because
+/// the second occurrence belongs to no owner.
+#[test]
+fn owner_collected_filter_consumes_one_witness_per_queued_context() {
+    let mut state = setup();
+    let event = zone_change_event(ObjectId(7));
+    let events = vec![event.clone(), event.clone()];
+    state.deferred_triggers.push(queued_context_for(event));
+
+    assert_eq!(
+        zone_change_count(&events),
+        2,
+        "the slice must really hold two byte-identical ZoneChanged"
+    );
+    assert_eq!(
+        state.deferred_triggers.len(),
+        1,
+        "exactly one context must be queued"
+    );
+
+    let survivors = filter_already_collected_trigger_events_from(&state, &events, 0, &[]);
+    assert_eq!(
+        zone_change_count(&survivors),
+        1,
+        "CR 603.2c: one queued witness consumes one copy, not every copy"
+    );
+}
+
+/// U2 — the consumed-occurrence ledger alone suppresses, with an empty queue.
+///
+/// This is the witness that survives an intervening `drain_deferred_trigger_queue`.
+/// Production isolation of this case at the search-delivery park is an open gap;
+/// it is evidenced here at the authority layer.
+#[test]
+fn owner_collected_filter_honors_consumed_ledger_with_empty_queue() {
+    let state = setup();
+    let claimed = zone_change_event(ObjectId(7));
+    let other = zone_change_event(ObjectId(8));
+    let events = vec![claimed.clone(), other.clone()];
+
+    assert!(
+        state.deferred_triggers.is_empty(),
+        "the queued-context witness must be absent so the ledger is isolated"
+    );
+
+    let consumed = vec![ConsumedTriggerEventOccurrence {
+        event: claimed.clone(),
+        occurrence: 0,
+    }];
+    let survivors = filter_already_collected_trigger_events_from(&state, &events, 0, &consumed);
+    assert_eq!(
+        survivors,
+        vec![other],
+        "the ledger-claimed occurrence is removed and the unrelated one survives"
+    );
+}
+
+/// U3 — the queued witness never touches a non-`ZoneChanged` event.
+#[test]
+fn owner_collected_filter_never_drops_non_zone_change_events() {
+    let mut state = setup();
+    let zone_change = zone_change_event(ObjectId(7));
+    let life = GameEvent::LifeChanged {
+        player_id: PlayerId(0),
+        amount: -1,
+    };
+    let events = vec![zone_change.clone(), life.clone()];
+    state
+        .deferred_triggers
+        .push(queued_context_for(zone_change));
+
+    let survivors = filter_already_collected_trigger_events_from(&state, &events, 0, &[]);
+    assert!(
+        !survivors.is_empty(),
+        "the non-zone event must not be swept away with the zone change"
+    );
+    assert_eq!(
+        survivors,
+        vec![life],
+        "only the owner-collected ZoneChanged is removed"
+    );
+}
+
+/// U4 — the witness counts CONTEXT COPIES, not occurrences.
+///
+/// Two observers of ONE occurrence queue two contexts, each carrying that same
+/// single value. A slice holding two byte-identical copies therefore loses BOTH.
+/// This is the documented bound in
+/// `filter_already_collected_trigger_events_from`'s contract, made executable so
+/// no future author can re-assert occurrence-exactness without deliberately
+/// updating this row.
+#[test]
+fn owner_collected_filter_counts_contexts_not_occurrences() {
+    let mut state = setup();
+    let event = zone_change_event(ObjectId(7));
+    let events = vec![event.clone(), event.clone()];
+    state
+        .deferred_triggers
+        .push(queued_context_for(event.clone()));
+    state.deferred_triggers.push(queued_context_for(event));
+
+    assert_eq!(
+        zone_change_count(&events),
+        2,
+        "the slice must really hold two byte-identical ZoneChanged"
+    );
+    assert_eq!(
+        state.deferred_triggers.len(),
+        2,
+        "two observers of one occurrence queue two contexts"
+    );
+
+    let survivors = filter_already_collected_trigger_events_from(&state, &events, 0, &[]);
+    assert_eq!(
+        zone_change_count(&survivors),
+        0,
+        "the queued witness is a min(queued_copies, slice_copies) BOUND, and is \
+         NOT occurrence-exact"
+    );
+}