graphrefly-operators 0.0.7

//! Integration tests for `flow` operators (Slice C-3, D024).
//!
//! Each test references the canonical-spec rule it covers in its name
//! or comments. Test naming convention: `<operator>_<scenario>`.

mod common;

use graphrefly_core::Core;
use graphrefly_operators::flow::{
    element_at, find, first, last, last_with_default, skip, take, take_while,
};

use common::{OpRuntime, RecordedEvent, TestValue};

// ---------------------------------------------------------------------
// take(count) — emits first `count` DATA then self-completes
// ---------------------------------------------------------------------

#[test]
fn take_emits_first_n_then_self_completes() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let taken = take(rt.core(), source, 2).into_node();

    let rec = rt.subscribe_recorder(taken);
    rt.emit_int(source, 10);
    rt.emit_int(source, 20);
    rt.emit_int(source, 30); // post-complete: must not surface

    assert_eq!(
        rec.data_values(),
        vec![TestValue::Int(10), TestValue::Int(20)]
    );
    assert!(
        rec.events().contains(&RecordedEvent::Complete),
        "expected Complete in {:?}",
        rec.events()
    );
}

#[test]
fn take_zero_self_completes_on_first_fire_with_no_data() {
    // D027: take(0) is allowed; first fire emits zero items then
    // immediately self-completes. Subscribers see [Start, Complete]
    // (no Data, no Dirty — R1.3.1.a one-DIRTY-per-wave applies to DATA
    // waves; pure-terminal waves don't need a preceding DIRTY).
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let taken = take(rt.core(), source, 0).into_node();

    let rec = rt.subscribe_recorder(taken);
    rt.emit_int(source, 99);

    assert!(rec.data_values().is_empty(), "events={:?}", rec.events());
    assert!(
        rec.events().contains(&RecordedEvent::Complete),
        "expected Complete in {:?}",
        rec.events()
    );
}

#[test]
fn take_propagates_upstream_complete_when_count_not_reached() {
    // Standard auto-cascade: take(5) sees only 2 DATAs then upstream
    // COMPLETE → propagates COMPLETE without a self-trigger. The
    // downstream subscriber should see [Data(1), Data(2), Complete].
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let taken = take(rt.core(), source, 5).into_node();

    let rec = rt.subscribe_recorder(taken);
    rt.emit_int(source, 1);
    rt.emit_int(source, 2);
    rt.core().complete(source);

    assert_eq!(
        rec.data_values(),
        vec![TestValue::Int(1), TestValue::Int(2)]
    );
    assert!(rec.events().contains(&RecordedEvent::Complete));
}

#[test]
fn take_resubscribable_resets_counter_on_lifecycle_reset() {
    // After self-completion, marking a node resubscribable and
    // re-subscribing resets `count_emitted` via
    // `OperatorScratch::release_handles` + fresh scratch install in
    // `reset_for_fresh_lifecycle`. The cached value from the prior
    // cycle replays once via the per-tier handshake (documented
    // Subscribe behavior, independent of operator-state reset); fresh
    // emits then proceed through a clean take(2) cycle.
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let taken = take(rt.core(), source, 2).into_node();
    rt.core().set_resubscribable(taken, true);

    // Cycle 1: emit 2 values, take completes.
    let rec1 = rt.subscribe_recorder(taken);
    rt.emit_int(source, 1);
    rt.emit_int(source, 2);
    let cycle1_data: Vec<_> = rec1
        .data_values()
        .into_iter()
        .filter(|v| matches!(v, TestValue::Int(n) if *n < 100))
        .collect();
    assert_eq!(cycle1_data, vec![TestValue::Int(1), TestValue::Int(2)]);
    assert!(rec1.events().contains(&RecordedEvent::Complete));
    drop(rec1);

    // Cycle 2: invalidate source first so the resubscribe's activation
    // re-walk doesn't consume one of the take(2) quota slots from the
    // cached value (without invalidate, source.cache=2 would re-deliver
    // through fire_op_take and count_emitted would reach 1 before any
    // fresh emits — the count would still be reset by
    // reset_for_fresh_lifecycle, but immediately re-incremented by
    // activation).
    rt.core().invalidate(source);

    let rec2 = rt.subscribe_recorder(taken);
    rt.emit_int(source, 100);
    rt.emit_int(source, 200);
    rt.emit_int(source, 300); // ignored — quota hit at 100/200
    let cycle2_fresh: Vec<_> = rec2
        .data_values()
        .into_iter()
        .filter(|v| matches!(v, TestValue::Int(n) if *n >= 100))
        .collect();
    assert_eq!(
        cycle2_fresh,
        vec![TestValue::Int(100), TestValue::Int(200)],
        "fresh cycle data wrong; events={:?}",
        rec2.events()
    );
    let complete_count = rec2
        .events()
        .iter()
        .filter(|e| matches!(e, RecordedEvent::Complete))
        .count();
    assert_eq!(
        complete_count, 1,
        "cycle 2 must self-complete after 2 fresh emits"
    );
}

// ---------------------------------------------------------------------
// skip(count) — drops first `count` DATA, emits the rest
// ---------------------------------------------------------------------

#[test]
fn skip_drops_first_n_then_emits_remaining() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let skipped = skip(rt.core(), source, 2).into_node();

    let rec = rt.subscribe_recorder(skipped);
    rt.emit_int(source, 1);
    rt.emit_int(source, 2);
    rt.emit_int(source, 3);
    rt.emit_int(source, 4);

    assert_eq!(
        rec.data_values(),
        vec![TestValue::Int(3), TestValue::Int(4)]
    );
}

#[test]
fn skip_full_window_settles_dirty_resolved_per_d018() {
    // While the skip window swallows every input of a wave, the wave
    // still produced DIRTY upstream → operator queues RESOLVED to
    // settle (D018 pattern).
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let skipped = skip(rt.core(), source, 3).into_node();

    let rec = rt.subscribe_recorder(skipped);
    rt.emit_int(source, 1);

    let events = rec.events();
    assert!(
        rec.data_values().is_empty(),
        "skip window swallowed; no Data expected. events={events:?}"
    );
    assert!(
        events.contains(&RecordedEvent::Resolved),
        "expected Resolved settle for full-skip wave: {events:?}"
    );
}

// ---------------------------------------------------------------------
// take_while(predicate) — predicate-gated truncation
// ---------------------------------------------------------------------

#[test]
fn take_while_emits_until_first_false_then_self_completes() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let binding = rt.binding.clone();
    let taken = take_while(rt.core(), &rt.op_binding, source, move |h| {
        binding.deref(h).int() < 10
    })
    .into_node();

    let rec = rt.subscribe_recorder(taken);
    rt.emit_int(source, 3);
    rt.emit_int(source, 7);
    rt.emit_int(source, 12); // first false → complete here, no Data
    rt.emit_int(source, 1); // post-complete: ignored

    assert_eq!(
        rec.data_values(),
        vec![TestValue::Int(3), TestValue::Int(7)],
        "events={:?}",
        rec.events()
    );
    assert!(
        rec.events().contains(&RecordedEvent::Complete),
        "expected Complete, events={:?}",
        rec.events()
    );
}

// ---------------------------------------------------------------------
// last(source) / last_with_default(source, default)
// ---------------------------------------------------------------------

#[test]
fn last_emits_buffered_latest_on_upstream_complete() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let n = last(rt.core(), source).into_node();

    let rec = rt.subscribe_recorder(n);
    rt.emit_int(source, 1);
    rt.emit_int(source, 2);
    rt.emit_int(source, 3);

    // No emit yet — buffering silently.
    assert!(
        rec.data_values().is_empty(),
        "last should not emit before upstream Complete: {:?}",
        rec.events()
    );

    rt.core().complete(source);

    assert_eq!(rec.data_values(), vec![TestValue::Int(3)]);
    assert!(rec.events().contains(&RecordedEvent::Complete));
}

#[test]
fn last_no_default_on_empty_stream_emits_only_complete() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let n = last(rt.core(), source).into_node();

    let rec = rt.subscribe_recorder(n);
    rt.core().complete(source);

    assert!(
        rec.data_values().is_empty(),
        "no default + no DATA → only Complete; events={:?}",
        rec.events()
    );
    assert!(rec.events().contains(&RecordedEvent::Complete));
}

#[test]
fn last_with_default_on_empty_stream_emits_default() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let default = rt.intern_int(42);
    let n = last_with_default(rt.core(), source, default)
        .unwrap()
        .into_node();

    let rec = rt.subscribe_recorder(n);
    rt.core().complete(source);

    assert_eq!(rec.data_values(), vec![TestValue::Int(42)]);
    assert!(rec.events().contains(&RecordedEvent::Complete));
}

#[test]
fn last_with_default_prefers_latest_over_default() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let default = rt.intern_int(42);
    let n = last_with_default(rt.core(), source, default)
        .unwrap()
        .into_node();

    let rec = rt.subscribe_recorder(n);
    rt.emit_int(source, 7);
    rt.core().complete(source);

    assert_eq!(rec.data_values(), vec![TestValue::Int(7)]);
}

#[test]
fn last_propagates_upstream_error() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let n = last(rt.core(), source).into_node();

    let rec = rt.subscribe_recorder(n);
    rt.emit_int(source, 1);
    let err_h = rt.binding.intern(TestValue::Str("boom".into()));
    rt.core().error(source, err_h);

    let saw_error = rec
        .events()
        .iter()
        .any(|e| matches!(e, RecordedEvent::Error(TestValue::Str(s)) if s == "boom"));
    assert!(
        saw_error,
        "expected Error('boom'); events={:?}",
        rec.events()
    );
    // No Data — upstream Error short-circuits the buffered emit.
    assert!(
        rec.data_values().is_empty(),
        "last should not emit Data on Error: {:?}",
        rec.events()
    );
}

// ---------------------------------------------------------------------
// Sugar: first / find / element_at
// ---------------------------------------------------------------------

#[test]
fn first_alias_for_take_one() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let n = first(rt.core(), source).into_node();

    let rec = rt.subscribe_recorder(n);
    rt.emit_int(source, 7);
    rt.emit_int(source, 8); // ignored — already complete

    assert_eq!(rec.data_values(), vec![TestValue::Int(7)]);
    assert!(rec.events().contains(&RecordedEvent::Complete));
}

#[test]
fn find_emits_first_matching_then_completes() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let binding = rt.binding.clone();
    let n = find(rt.core(), &rt.op_binding, source, move |h| {
        binding.deref(h).int() > 5
    })
    .into_node();

    let rec = rt.subscribe_recorder(n);
    rt.emit_int(source, 1);
    rt.emit_int(source, 3);
    rt.emit_int(source, 8); // first > 5 → emit + complete
    rt.emit_int(source, 9); // ignored

    assert_eq!(rec.data_values(), vec![TestValue::Int(8)]);
    assert!(rec.events().contains(&RecordedEvent::Complete));
}

#[test]
fn element_at_emits_indexed_value_then_completes() {
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let n = element_at(rt.core(), source, 2).into_node();

    let rec = rt.subscribe_recorder(n);
    rt.emit_int(source, 10);
    rt.emit_int(source, 20);
    rt.emit_int(source, 30); // index 2 — emit + complete
    rt.emit_int(source, 40); // ignored

    assert_eq!(rec.data_values(), vec![TestValue::Int(30)]);
    assert!(rec.events().contains(&RecordedEvent::Complete));
}

// ---------------------------------------------------------------------
// Refcount / lifecycle discipline
// ---------------------------------------------------------------------

#[test]
fn last_with_default_releases_default_on_core_drop() {
    // The default handle is retained for LastState's lifetime; on
    // Core drop, OperatorScratch::release_handles releases it.
    let binding;
    let default_handle;
    {
        let rt = OpRuntime::new();
        binding = rt.binding.clone();
        let source = rt.state_int(None);
        default_handle = rt.intern_int(99);
        // Caller's intern share: 1.
        assert_eq!(binding.refcount_of(default_handle), 1);
        let _n = last_with_default(rt.core(), source, default_handle)
            .unwrap()
            .into_node();
        // Core retained inside register_operator: now 2.
        assert_eq!(
            binding.refcount_of(default_handle),
            2,
            "register_operator must retain the default handle"
        );
    }
    // Core dropped → release_handles called → refcount drops to 1
    // (caller's original intern share remains).
    assert_eq!(
        binding.refcount_of(default_handle),
        1,
        "Core drop must release the default handle's share"
    );
}

#[test]
fn last_releases_buffered_latest_on_lifecycle_reset() {
    // Buffered `latest` is retained for LastState's lifetime; on
    // resubscribable terminal cycle reset, release_handles releases it
    // and a fresh LastState (latest=NO_HANDLE) is installed.
    //
    // Slice C-3 /qa T1 — verifies the refcount path explicitly via
    // `refcount_of`, not just "no panic on re-subscribe". The
    // diagnostic intern in cycle 1 holds an extra share of the
    // buffered handle; we observe Core's slot share via the refcount
    // delta between cycle 1 (buffered) and post-reset (released).
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let n = last(rt.core(), source).into_node();
    rt.core().set_resubscribable(n, true);

    // Diagnostic intern: hold an extra share of `5` so its refcount
    // doesn't drop to zero when LastState releases its share. This
    // lets `refcount_of` survive across the lifecycle reset for
    // observation.
    let observed_handle = rt.intern_int(5);
    // After intern_int: refcount = 1 (our diagnostic share).
    assert_eq!(rt.binding.refcount_of(observed_handle), 1);

    let rec1 = rt.subscribe_recorder(n);
    rt.emit_int(source, 5);
    rt.settle(); // D246: pump deferred ops so refcount accounting settles.
                 // After emit + wave drain: source's cache holds one share, Last's
                 // dep_records[0].prev_data holds one (transferred from data_batch
                 // by clear_wave_state's wave-end rotation), LastState.latest
                 // holds one (from fire_op_last's retain), and our diagnostic
                 // share remains. Total: 4.
    assert_eq!(
        rt.binding.refcount_of(observed_handle),
        4,
        "after emit + wave drain: 1 (diag) + 1 (source cache) + 1 (prev_data) + 1 (LastState.latest)"
    );

    rt.core().complete(source);
    rt.settle(); // D246: pump deferred ops so refcount accounting settles.
                 // Terminal-aware fire_op_last emits Data(5) + Complete: it
                 // retains the buffered handle once more, commit_emission_verbatim
                 // consumes that retain into Last.cache. Refcount becomes 5.
    assert_eq!(rec1.data_values(), vec![TestValue::Int(5)]);
    drop(rec1);
    // D246: `drop(rec1)` posts a deferred unsubscribe (binding-layer
    // SubGuard); pump it owner-side so the last-sub-leaves Phase G
    // cache-clear fires BEFORE the rec2 re-subscribe below.
    rt.settle();
    // After drop(rec1): rec1's Subscription drops. Phase G (D119,
    // 2026-05-10) fires Core cache-clear on the last-sub-leaves
    // transition. For compute nodes (R2.2.7 / R2.2.8 ROM rule, D119),
    // `cache` is released and per-dep `prev_data` is released:
    //   - Last.cache released: 5 → 4
    //   - prev_data released:  4 → 3
    //   - LastState.latest stays in op_scratch (Phase G touches
    //     NodeRecord state only, not operator scratch)
    // Refcount after drop(rec1): 3 (diag + source.cache + LastState.latest).

    // Re-subscribing triggers reset_for_fresh_lifecycle (resubscribable
    // + terminal — note D118 R2.2.7.a now allows reset regardless of
    // TEARDOWN state) AND re-activation (subscribers count went 1→0→1,
    // _rec2 is a "first subscriber" again):
    //   - Phase 2 makes a fresh LastState (latest=NO_HANDLE,
    //     default=NO_HANDLE) — no retain on `5`.
    //   - Phase 3 releases old LastState.latest's share (3 → 2).
    //   - Phase 5 prev_data release: skipped (Phase G already cleared).
    //   - Then activation re-walks source dep, deliver_data_to_consumer
    //     retains source.cache for the new dep_records[0].data_batch
    //     (2 → 3). Wave-end rotates data_batch into prev_data (no net
    //     retain change). fire_op_last buffers source.cache into the
    //     new LastState.latest (+1 retain: 3 → 4). `Last.cache` stays
    //     at NO_HANDLE because the source is already terminal — no new
    //     Data is emitted to fill it.
    // Final: 1 (diag) + 1 (source.cache) + 1 (re-activated prev_data) +
    //        1 (re-activated LastState.latest) = 4.
    let _rec2 = rt.subscribe_recorder(n);
    rt.settle(); // D246: pump deferred ops so refcount accounting settles.
    assert_eq!(
        rt.binding.refcount_of(observed_handle),
        4,
        "after Phase G cache-clear + reset + re-activation: Last.cache \
         released by Phase G; prev_data + LastState.latest replaced by \
         re-activation's fresh retains. (Pre-D119 expected 5 because \
         Last.cache survived; D119 clears compute cache on deactivation \
         per R2.2.7 / R2.2.8 ROM rule.)"
    );
}

// ---------------------------------------------------------------------
// Integration: pipeline composition
// ---------------------------------------------------------------------

#[test]
fn take_after_skip_produces_window() {
    // skip(2) → take(3) → emits inputs 3..=5
    let rt = OpRuntime::new();
    let source = rt.state_int(None);
    let skipped = skip(rt.core(), source, 2).into_node();
    let windowed = take(rt.core(), skipped, 3).into_node();

    let rec = rt.subscribe_recorder(windowed);
    for v in 1..=10 {
        rt.emit_int(source, v);
    }

    assert_eq!(
        rec.data_values(),
        vec![TestValue::Int(3), TestValue::Int(4), TestValue::Int(5)]
    );
    assert!(rec.events().contains(&RecordedEvent::Complete));
}

// ---------------------------------------------------------------------
// Send + Sync / OperatorScratch trait-object behavior
// ---------------------------------------------------------------------

// D248/D249/S2c: `FlowRegistration` + `Core` are intentionally
// `!Send + !Sync` under full single-owner (the substrate `Sink`
// dropped `Send + Sync`; `Core` holds the subscriber map + the
// owner-only `Rc<DeferQueue>`). The prior
// `flow_registration_is_send_and_sync` assertion was shared-Core-era
// legacy and is deleted — the single-owner contract ("owned & driven
// on one thread") used to be asserted by the deleted `scheduling_groups.rs`
// test file (D253 (S5) deleted the `SchedulingGroupId` surface entirely;
// the single-owner contract is now structural via `Core: !Send + !Sync`).
#[test]
fn flow_registration_type_touch() {
    let _ = core::marker::PhantomData::<graphrefly_operators::flow::FlowRegistration>;
    let _ = core::marker::PhantomData::<Core>;
}