polyc-query 2026.9.0

use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;

use arrow::array::{Array, Int64Array, StringArray};
use datafusion::catalog::TableProvider;
use datafusion::datasource::empty::EmptyTable;
use datafusion::error::DataFusionError;
use datafusion::execution::context::SessionContext;
use futures::StreamExt;
use polyc_state::query_audit::{ErrorClass, QueryOutcome, Truncation};

use super::admission::CoreExecutionAdmissionInput;
use super::{CoreExecutionAdmission, CoreExecutionError, request_context};
use crate::core_resolution::{CoreParameter, CorePlanOutcome, CoreTable, arrow_schema};
use crate::session::QueryScope;

mod support;

use support::Harness;

const DEFAULT_ARTIFACT_RANGE_BYTES: u64 = 4 * 1024 * 1024;

#[test]
fn aggregate_execution_admission_refuses_zero() {
    let error = CoreExecutionAdmission::try_from(CoreExecutionAdmissionInput {
        max_concurrent_executions: 0,
    })
    .unwrap_err();
    assert!(matches!(error, CoreExecutionError::InvalidComposition(_)));
}

/// A limit above the permit ceiling is refused, not asserted on.
///
/// The value comes from a deployment variable with no upper bound of its own,
/// and the semaphore asserts its ceiling in its constructor. Without this the
/// refusal would be a panic inside a dependency, in a module that promises no
/// deployment value panics.
#[test]
fn aggregate_execution_admission_refuses_more_than_the_permit_ceiling() {
    let error = CoreExecutionAdmission::try_from(CoreExecutionAdmissionInput {
        max_concurrent_executions: usize::MAX,
    })
    .unwrap_err();
    assert!(matches!(error, CoreExecutionError::InvalidComposition(_)));
}

#[tokio::test]
#[allow(
    clippy::significant_drop_tightening,
    reason = "the bound query must remain alive to prove bind-time I/O separately from execution I/O"
)]
async fn signed_exact_parquet_streams_only_visible_rows() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "visible-exact",
            "SELECT text AS first, role, text AS duplicate FROM messages ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let authority = harness.visible_authority();

    let bound = authority.bind(prepared).await.expect("exact files bind");
    let reads_after_binding = harness.store.range_calls();
    assert!(
        reads_after_binding > 0,
        "binding proves complete files and footers"
    );

    let mut stream = bound.execute();
    let first = stream
        .next()
        .await
        .expect("one batch")
        .expect("verified batch");
    assert_eq!(
        harness.store.range_calls(),
        reads_after_binding,
        "DataFusion scans zero-copy slices of the retained verified image",
    );
    assert_eq!(first.schema().field(0).name(), "first");
    assert_eq!(first.schema().field(1).name(), "role");
    assert_eq!(first.schema().field(2).name(), "duplicate");
    let left = first
        .column(0)
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap();
    let duplicate = first
        .column(2)
        .as_any()
        .downcast_ref::<StringArray>()
        .unwrap();
    let values = (0..left.len())
        .map(|index| left.value(index).to_owned())
        .collect::<Vec<_>>();
    assert_eq!(values, ["visible-one", "visible-two", "visible-three"]);
    assert_eq!(left, duplicate);
    assert!(stream.next().await.is_none());
}

#[tokio::test]
async fn successful_eof_waits_for_durable_completion() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "terminal-success",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let mut stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();

    let mut rows = 0;
    while rows < 3 {
        rows += stream.next().await.unwrap().unwrap().num_rows();
    }
    assert!(
        harness.state.completion_for("terminal-success").is_none(),
        "a released batch is not a successful terminal"
    );
    assert!(stream.next().await.is_none());
    let completion = harness
        .state
        .completion_for("terminal-success")
        .expect("EOF follows a durable completion");
    assert_eq!(completion.outcome(), QueryOutcome::Succeeded);
    assert_eq!(harness.state.completion_attempts(), 1);
}

#[tokio::test]
async fn ambiguous_completion_settles_the_exact_receipt() {
    let harness = Harness::new();
    harness.state.arm_completion_response_loss();
    let prepared = harness
        .prepare(
            "terminal-ambiguous",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let rows = harness
        .collect_text(harness.visible_authority(), prepared)
        .await;

    assert_eq!(rows.len(), 3);
    assert_eq!(
        harness.state.completion_attempts(),
        1,
        "receipt settlement must not construct or send a second command"
    );
    assert_eq!(harness.state.completion_receipt_reads(), 1);
    assert_eq!(
        harness
            .state
            .completion_for("terminal-ambiguous")
            .unwrap()
            .outcome(),
        QueryOutcome::Succeeded
    );
}

#[tokio::test]
async fn dropped_stream_records_cancellation_and_never_success() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "terminal-drop",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let mut stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();
    let delivered = stream.next().await.unwrap().unwrap().num_rows();
    assert!(delivered > 0, "the consumer received rows before dropping");
    drop(stream);

    let completion = harness.await_completion("terminal-drop").await;
    assert_eq!(
        completion.outcome(),
        QueryOutcome::Failed(ErrorClass::Cancelled)
    );
    assert_eq!(
        completion.rows(),
        polyc_state::query_audit::RowCount::new(u64::try_from(delivered).unwrap()),
        "a cancelled terminal reports what the consumer received, never zero"
    );
}

#[tokio::test]
async fn drop_while_eof_is_pending_selects_cancellation_before_dispatch() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "terminal-pending-drop",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let dispatch = prepared.pause_completion_dispatch();
    let mut stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();
    let mut rows = 0;
    while rows < 3 {
        rows += stream.next().await.unwrap().unwrap().num_rows();
    }

    let mut eof = Box::pin(stream.next());
    tokio::select! {
        () = dispatch.wait() => {}
        result = &mut eof => panic!("EOF escaped before terminal dispatch: {result:?}"),
    }
    drop(eof);
    drop(stream);
    dispatch.resume();

    let completion = harness.await_completion("terminal-pending-drop").await;
    assert_eq!(
        completion.outcome(),
        QueryOutcome::Failed(ErrorClass::Cancelled)
    );
}

#[tokio::test]
async fn completion_outage_exposes_no_successful_eof() {
    let harness = Harness::new();
    harness.state.set_completion_outage();
    let prepared = harness
        .prepare(
            "terminal-outage",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let mut stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();
    let mut rows = 0;
    while rows < 3 {
        rows += stream.next().await.unwrap().unwrap().num_rows();
    }

    let error = stream.next().await.unwrap().unwrap_err();
    assert!(matches!(
        error,
        CoreExecutionError::AuditCompletionUnavailable
    ));
    assert!(harness.state.completion_for("terminal-outage").is_none());
    assert_eq!(harness.state.completion_attempts(), 3);
}

#[tokio::test]
async fn legacy_dependency_is_refused_before_artifact_reads() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "legacy-before-artifact",
            "SELECT messages.text, tool_calls.name \
             FROM messages JOIN tool_calls USING (partition, turn_id)",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;

    let Err(error) = harness.visible_authority().bind(prepared).await else {
        panic!("a legacy dependency has no bounded execution provider");
    };

    assert!(matches!(
        error,
        CoreExecutionError::LegacyProviderUnavailable
    ));
    assert_eq!(
        harness.store.range_calls(),
        0,
        "the refusal must precede every manifest or segment read",
    );
}

#[tokio::test]
async fn exact_provider_executes_duplicate_and_zero_column_projections() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "direct-provider-projection",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let bound = harness.visible_authority().bind(prepared).await.unwrap();
    let provider = bound.provider(CoreTable::Messages);
    let context = SessionContext::new();
    let reads_after_binding = harness.store.range_calls();
    let projection = vec![6, 4, 6];
    let plan = provider
        .scan(&context.state(), Some(&projection), &[], None)
        .await
        .unwrap();
    assert_eq!(harness.store.range_calls(), reads_after_binding);
    let batches = datafusion::physical_plan::collect(plan, context.task_ctx())
        .await
        .unwrap();
    assert_eq!(
        batches
            .iter()
            .map(arrow::record_batch::RecordBatch::num_rows)
            .sum::<usize>(),
        3
    );
    assert!(batches.iter().all(|batch| batch.num_columns() == 3));
    for batch in &batches {
        assert_eq!(batch.column(0), batch.column(2));
    }

    let empty = Vec::new();
    let plan = provider
        .scan(&context.state(), Some(&empty), &[], None)
        .await
        .unwrap();
    let batches = datafusion::physical_plan::collect(plan, context.task_ctx())
        .await
        .unwrap();
    assert_eq!(
        batches
            .iter()
            .map(arrow::record_batch::RecordBatch::num_rows)
            .sum::<usize>(),
        3
    );
    assert!(batches.iter().all(|batch| batch.num_columns() == 0));
    drop(bound);
}

#[tokio::test]
async fn zero_column_scan_preserves_row_count() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "zero-column",
            "SELECT COUNT(*) AS held FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let mut stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();

    let batch = stream.next().await.unwrap().unwrap();
    let count = batch
        .column(0)
        .as_any()
        .downcast_ref::<Int64Array>()
        .unwrap();
    assert_eq!(count.value(0), 3);
}

#[tokio::test]
async fn unsupported_filter_is_evaluated_above_the_exact_scan() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "filter-above-scan",
            "SELECT text FROM messages WHERE position > 2 ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;

    let rows = harness
        .collect_text(harness.visible_authority(), prepared)
        .await;
    assert_eq!(rows, ["visible-two", "visible-three"]);
}

#[tokio::test]
async fn fleet_combines_disjoint_realms_while_visible_cannot() {
    let harness = Harness::new();
    let visible = harness
        .prepare(
            "realm-visible",
            "SELECT text FROM messages ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let fleet = harness
        .prepare(
            "realm-fleet",
            "SELECT text FROM messages ORDER BY position",
            QueryScope::Fleet,
            10,
        )
        .await;

    let visible_rows = harness
        .collect_text(harness.visible_authority(), visible)
        .await;
    let fleet_rows = harness.collect_text(harness.fleet_authority(), fleet).await;
    assert_eq!(
        visible_rows,
        ["visible-one", "visible-two", "visible-three"]
    );
    assert_eq!(
        fleet_rows,
        [
            "visible-one",
            "visible-two",
            "visible-three",
            "fleet-secret"
        ]
    );
}

#[tokio::test]
async fn unreferenced_objects_and_scope_widening_add_no_files() {
    let harness = Harness::new();
    harness.store.add_unreferenced_object();
    let prepared = harness
        .prepare(
            "unreferenced",
            "SELECT text FROM messages ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    // The current revalidation scope is Fleet, a widening from the audited
    // conversation. Exact tokens and the fresh catalog remain fixed.
    let rows = harness
        .collect_text(harness.visible_authority(), prepared)
        .await;

    assert_eq!(rows, ["visible-one", "visible-two", "visible-three"]);
}

#[tokio::test]
async fn permit_refusal_performs_no_artifact_io() {
    let harness = Harness::new();
    harness.state.refuse_audit();
    let outcome = harness
        .try_prepare(
            "refused",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;

    assert!(outcome.is_err());
    assert_eq!(harness.store.total_calls(), 0);
}

#[tokio::test]
async fn exact_digest_failure_refuses_before_a_row() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "corrupt",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    harness.store.corrupt_visible_messages();

    let error = harness
        .visible_authority()
        .bind(prepared)
        .await
        .err()
        .expect("complete-file proof catches corruption");
    assert!(matches!(error, CoreExecutionError::Artifact(_)));
    assert_eq!(
        harness.state.completion_for("corrupt").unwrap().outcome(),
        QueryOutcome::Failed(ErrorClass::Internal),
        "stored bytes that disagree with their signed digest are corruption, not caller syntax"
    );
    assert_eq!(harness.reserved_memory(), 0);
}

#[tokio::test]
async fn dropping_a_bound_query_before_execute_records_cancellation() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "bound-drop",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let bound = harness.visible_authority().bind(prepared).await.unwrap();
    drop(bound);

    let completion = harness.await_completion("bound-drop").await;
    assert_eq!(
        completion.outcome(),
        QueryOutcome::Failed(ErrorClass::Cancelled)
    );
}

#[tokio::test]
async fn exact_generation_and_length_disagreement_refuse_before_a_row() {
    for (query, defect) in [
        (
            "wrong-generation",
            Harness::wrong_generation as fn(&Harness),
        ),
        ("wrong-length", Harness::wrong_length as fn(&Harness)),
    ] {
        let harness = Harness::new();
        let prepared = harness
            .prepare(
                query,
                "SELECT text FROM messages",
                QueryScope::Conversations(vec!["a".to_owned()]),
                10,
            )
            .await;
        defect(&harness);

        let error = harness
            .visible_authority()
            .bind(prepared)
            .await
            .err()
            .expect("exact metadata disagreement is closed");
        assert!(matches!(error, CoreExecutionError::Artifact(_)));
    }
}

#[tokio::test]
async fn first_release_revalidation_fails_closed_on_scope_narrowing() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "narrowed",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let mut stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();
    harness
        .revalidator
        .set(QueryScope::Conversations(Vec::new()));

    let error = stream.next().await.unwrap().unwrap_err();
    assert!(matches!(error, CoreExecutionError::AuthorityNarrowed));
    assert_eq!(
        harness.state.completion_for("narrowed").unwrap().outcome(),
        QueryOutcome::Failed(ErrorClass::Denied)
    );
    assert!(stream.next().await.is_none());
}

#[tokio::test]
async fn first_release_revalidation_fails_closed_on_source_recreation() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "recreated",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let mut stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();
    harness.state.recreate_source();

    let error = stream.next().await.unwrap().unwrap_err();
    assert!(matches!(error, CoreExecutionError::SourceChanged(_)));
    assert_eq!(
        harness.state.completion_for("recreated").unwrap().outcome(),
        QueryOutcome::Failed(ErrorClass::Unavailable)
    );
}

#[tokio::test]
async fn first_release_revalidation_fails_closed_on_state_outage() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "revalidation-outage",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let mut stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();
    harness.state.set_outage();

    let error = stream.next().await.unwrap().unwrap_err();
    assert!(matches!(error, CoreExecutionError::Resolution(_)));
    assert_eq!(
        harness
            .state
            .completion_for("revalidation-outage")
            .unwrap()
            .outcome(),
        QueryOutcome::Failed(ErrorClass::Unavailable)
    );
}

#[tokio::test]
async fn periodic_revalidation_stops_subsequent_batches() {
    for defect in ["narrow", "recreate", "outage"] {
        let harness = Harness::new();
        let prepared = harness
            .prepare(
                &format!("periodic-{defect}"),
                "SELECT text FROM messages",
                QueryScope::Fleet,
                10,
            )
            .await;
        let mut stream = harness
            .fleet_authority()
            .bind(prepared)
            .await
            .unwrap()
            .execute();
        let first = stream.next().await.unwrap().unwrap();
        assert!(first.num_rows() > 0);
        tokio::time::sleep(Duration::from_millis(2)).await;
        match defect {
            "narrow" => harness
                .revalidator
                .set(QueryScope::Conversations(vec!["a".to_owned()])),
            "recreate" => harness.state.recreate_source(),
            "outage" => harness.state.set_outage(),
            _ => unreachable!(),
        }

        let error = stream.next().await.unwrap().unwrap_err();
        assert!(matches!(
            (defect, error),
            ("narrow", CoreExecutionError::AuthorityNarrowed)
                | ("recreate", CoreExecutionError::SourceChanged(_))
                | ("outage", CoreExecutionError::Resolution(_))
        ));
    }
}

#[tokio::test]
async fn row_cap_releases_exactly_the_cap() {
    let harness = Harness::new();
    for (query, cap, expected) in [
        ("cap-minus-one", 2, 2),
        ("cap", 3, 3),
        ("cap-plus-one", 4, 3),
    ] {
        let prepared = harness
            .prepare(
                query,
                "SELECT text FROM messages ORDER BY position",
                QueryScope::Conversations(vec!["a".to_owned()]),
                cap,
            )
            .await;
        let rows = harness
            .collect_text(harness.visible_authority(), prepared)
            .await;
        assert_eq!(rows.len(), expected);
        let completion = harness.state.completion_for(query).unwrap();
        assert_eq!(
            completion.truncation(),
            if cap < 3 {
                Truncation::TruncatedAt(u64::try_from(expected).unwrap())
            } else {
                Truncation::Complete
            }
        );
    }
}

#[tokio::test]
async fn release_byte_bound_refuses_before_release() {
    let harness = Harness::with_release_bounds(1, 1);
    let prepared = harness
        .prepare(
            "release-byte-bound",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let mut stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();

    let error = stream.next().await.unwrap().unwrap_err();
    assert!(matches!(error, CoreExecutionError::ReleaseBound { .. }));
}

#[tokio::test]
#[allow(
    clippy::significant_drop_tightening,
    reason = "the test observes the bound query's live pool reservation before consuming it"
)]
async fn retained_scan_performs_no_backend_reads_and_drop_releases_memory() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "drop-cancels",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let bound = harness.visible_authority().bind(prepared).await.unwrap();
    let calls = harness.store.range_calls();
    assert!(bound.source_decode_reservation.size() > 0);
    assert!(harness.reserved_memory() > 0);
    let mut stream = bound.execute();
    assert!(stream.next().await.unwrap().is_ok());
    assert_eq!(harness.store.range_calls(), calls);
    drop(stream);
    tokio::time::timeout(Duration::from_secs(1), async {
        while harness.reserved_memory() != 0 {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("the cancelled producer releases its source reservation");
}

#[tokio::test]
async fn source_decode_limit_refuses_before_segment_body_reads() {
    let harness = Harness::with_source_decode_bound(1);
    let prepared = harness
        .prepare(
            "decode-limit",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;

    let error = harness
        .visible_authority()
        .bind(prepared)
        .await
        .err()
        .expect("the authenticated envelope exceeds the deployment limit");

    assert!(matches!(
        error,
        CoreExecutionError::SourceDecodeBound { .. }
    ));
    assert_eq!(harness.store.segment_range_calls(), 0);
    assert_eq!(harness.reserved_memory(), 0);
}

#[tokio::test]
async fn aggregate_source_decode_reservation_uses_the_shared_runtime_pool() {
    let reservations = Harness::visible_message_decode_reservations();
    assert!(reservations.len() >= 2);
    let aggregate = reservations.iter().copied().sum::<u64>() + DEFAULT_ARTIFACT_RANGE_BYTES + 1;
    let pool_limit = usize::try_from(aggregate - 1).unwrap();
    assert!(reservations.iter().all(|value| *value < pool_limit as u64));
    let harness = Harness::with_runtime_memory(pool_limit);
    let prepared = harness
        .prepare(
            "aggregate-decode-pool",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;

    let error = harness
        .visible_authority()
        .bind(prepared)
        .await
        .err()
        .expect("the summed reservation exceeds the shared pool");

    assert!(matches!(
        error,
        CoreExecutionError::DataFusion(DataFusionError::ResourcesExhausted(_))
    ));
    assert_eq!(harness.store.segment_range_calls(), 0);
    assert_eq!(harness.reserved_memory(), 0);
}

#[tokio::test]
async fn original_deadline_stops_retained_decode_before_release() {
    let harness = Harness::new();
    let prepared = harness
        .prepare_with_timeout(
            "range-deadline",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            Duration::from_millis(100),
        )
        .await;
    let mut stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();
    tokio::time::sleep(Duration::from_millis(110)).await;

    let error = stream.next().await.unwrap().unwrap_err();
    assert!(matches!(error, CoreExecutionError::Deadline));
    assert_eq!(
        harness
            .state
            .completion_for("range-deadline")
            .unwrap()
            .outcome(),
        QueryOutcome::Failed(ErrorClass::Deadline),
        "terminal settlement has a fresh server-owned budget"
    );
}

#[tokio::test]
async fn expired_deadline_refuses_before_artifact_io() {
    let harness = Harness::new();
    let prepared = harness
        .prepare_with_timeout(
            "bind-deadline",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            Duration::from_millis(20),
        )
        .await;
    tokio::time::sleep(Duration::from_millis(25)).await;

    let error = harness
        .visible_authority()
        .bind(prepared)
        .await
        .err()
        .expect("expired operation cannot enter artifacts");
    assert!(matches!(error, CoreExecutionError::Deadline));
    assert_eq!(harness.store.total_calls(), 0);
}

#[tokio::test]
async fn original_deadline_stops_a_pending_bind_time_artifact_read() {
    let harness = Harness::new();
    let prepared = harness
        .prepare_with_timeout(
            "bind-range-deadline",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            Duration::from_millis(100),
        )
        .await;
    harness.store.pause_ranges();

    let error = harness
        .visible_authority()
        .bind(prepared)
        .await
        .err()
        .expect("the original deadline stops stalled artifact admission");

    assert!(matches!(error, CoreExecutionError::Deadline));
    assert!(harness.store.range_calls() > 0);
    assert_eq!(harness.reserved_memory(), 0);
}

#[tokio::test]
async fn aggregate_execution_admission_queues_before_artifact_reads_and_releases_on_drop() {
    let harness = Harness::with_concurrency(1);
    let first = harness
        .prepare(
            "admission-first",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let first = harness.visible_authority().bind(first).await.unwrap();
    let calls_while_held = harness.store.total_calls();
    let second = harness
        .prepare_with_timeout(
            "admission-second",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            Duration::from_millis(100),
        )
        .await;

    let error = harness
        .visible_authority()
        .bind(second)
        .await
        .err()
        .expect("the sole aggregate slot remains owned");
    assert!(matches!(error, CoreExecutionError::Deadline));
    assert_eq!(harness.store.total_calls(), calls_while_held);

    drop(first);
    let third = harness
        .prepare(
            "admission-third",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let rebound = harness.visible_authority().bind(third).await.unwrap();
    drop(rebound);
    assert!(harness.store.total_calls() > calls_while_held);
}

#[tokio::test]
async fn typed_parameter_plan_executes_without_replacement_sql() {
    let harness = Harness::new();
    let prepared = harness
        .prepare_with_parameters(
            "typed-real-plan",
            "SELECT text FROM messages WHERE position > $1 ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            vec![CoreParameter::UInt64(2)],
        )
        .await;
    let rows = harness
        .collect_text(harness.visible_authority(), prepared)
        .await;

    assert_eq!(rows, ["visible-two", "visible-three"]);
}

#[test]
fn fresh_request_catalogs_cannot_observe_each_other() {
    let family = polyc_projection::family::conversation_core();
    let turns: Arc<dyn TableProvider> = Arc::new(EmptyTable::new(arrow_schema(
        family
            .table(polyc_projection::family::CONVERSATION_TURNS)
            .unwrap(),
    )));
    let messages: Arc<dyn TableProvider> = Arc::new(EmptyTable::new(arrow_schema(
        family
            .table(polyc_projection::family::CONVERSATION_MESSAGES)
            .unwrap(),
    )));
    let base = SessionContext::new().state();
    let first = request_context(&base, &BTreeMap::from([(CoreTable::Turns, turns)])).unwrap();
    let second =
        request_context(&base, &BTreeMap::from([(CoreTable::Messages, messages)])).unwrap();

    assert!(first.table_exist("turns").unwrap());
    assert!(!first.table_exist("messages").unwrap());
    assert!(second.table_exist("messages").unwrap());
    assert!(!second.table_exist("turns").unwrap());
}

#[tokio::test]
async fn prepared_realm_cannot_be_replaced_after_audit() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "crossed-realm",
            "SELECT text FROM messages",
            QueryScope::Fleet,
            10,
        )
        .await;
    let calls = harness.store.total_calls();

    let error = harness
        .visible_authority()
        .bind(prepared)
        .await
        .err()
        .expect("authority realm is retained in prepared state");
    assert!(matches!(error, CoreExecutionError::RealmMismatch));
    assert_eq!(harness.store.total_calls(), calls);
}

#[tokio::test]
async fn exact_audit_replay_never_yields_a_second_execution_capability() {
    let harness = Harness::new();
    let scope = QueryScope::Conversations(vec!["a".to_owned()]);
    let first = harness
        .try_prepare("replay", "SELECT text FROM messages", scope.clone(), 10)
        .await
        .unwrap();
    let second = harness
        .try_prepare("replay", "SELECT text FROM messages", scope, 10)
        .await
        .unwrap();

    assert!(matches!(first, CorePlanOutcome::Granted(_)));
    assert!(matches!(second, CorePlanOutcome::AlreadyRecorded(_)));
    assert_eq!(harness.store.total_calls(), 0);
}

/// A consumer that stops polling must not pin the permit past its deadline.
///
/// The producer parks waiting for a readiness token. Without a deadline arm it
/// would hold the execution admission slot, the memory reservation, and the
/// durable intent forever. The original deadline must end it.
#[tokio::test]
async fn an_idle_consumer_settles_the_deadline_and_releases_its_slot() {
    let harness = Harness::new();
    let prepared = harness
        .prepare_with_timeout(
            "idle-consumer",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            Duration::from_millis(60),
        )
        .await;
    let stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();

    // Hold the stream without ever polling it.
    let completion = harness.await_completion("idle-consumer").await;
    assert_eq!(
        completion.outcome(),
        QueryOutcome::Failed(ErrorClass::Deadline)
    );
    assert_eq!(
        completion.rows(),
        polyc_state::query_audit::RowCount::new(0)
    );
    drop(stream);
    assert_eq!(
        harness.reserved_memory(),
        0,
        "the expired producer releases its source reservation"
    );
}

/// A descheduled producer must not cost the audit its exact row count.
///
/// The consumer receives rows and then withdraws while the producer is held
/// away from the guardian for far longer than the scheduling grace this
/// lifecycle used to depend on. The delivered counts live in the latch, and
/// the guardian waits for the measured terminal inside its own settlement
/// budget, so the durable completion still reports what the consumer received.
#[tokio::test]
async fn a_descheduled_producer_still_records_the_exact_delivered_rows() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "descheduled-report",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let mut stream = {
        let mut bound = harness.visible_authority().bind(prepared).await.unwrap();
        // Four times the 50ms grace the previous lifecycle allowed.
        bound.delay_report_for_test(Duration::from_millis(200));
        bound.execute()
    };

    let delivered = stream.next().await.unwrap().unwrap().num_rows();
    assert!(
        delivered > 0,
        "the consumer received rows before withdrawing"
    );
    drop(stream);

    let completion = harness.await_completion("descheduled-report").await;
    assert_eq!(
        completion.outcome(),
        QueryOutcome::Failed(ErrorClass::Cancelled)
    );
    assert_eq!(
        completion.rows(),
        polyc_state::query_audit::RowCount::new(u64::try_from(delivered).unwrap()),
        "a withdrawn query reports what the consumer received, whatever the scheduler did"
    );
}

/// A measured failure survives a withdrawal that races the dispatch boundary.
///
/// The producer measures a deadline. The consumer then withdraws while the
/// guardian is held inside the pre-dispatch window. Exactly one transition
/// wins, and a measured deadline is never relabelled as a cancellation.
#[tokio::test]
async fn a_measured_deadline_survives_a_withdrawal_at_the_dispatch_boundary() {
    let harness = Harness::new();
    let prepared = harness
        .prepare_with_timeout(
            "deadline-vs-withdrawal",
            "SELECT text FROM messages",
            QueryScope::Conversations(vec!["a".to_owned()]),
            Duration::from_millis(60),
        )
        .await;
    let dispatch = prepared.pause_completion_dispatch();
    let stream = harness
        .visible_authority()
        .bind(prepared)
        .await
        .unwrap()
        .execute();

    // The producer's readiness wait expires, so it measures a deadline.
    dispatch.wait().await;
    drop(stream);
    dispatch.resume();

    let completion = harness.await_completion("deadline-vs-withdrawal").await;
    assert_eq!(
        completion.outcome(),
        QueryOutcome::Failed(ErrorClass::Deadline),
        "a withdrawal after a measured deadline never rewrites its class"
    );
}

/// A successful stream releases exactly one terminal frame and then ends.
///
/// The success path used to return its terminal without closing the stream, so
/// every later poll found the row stream exhausted and built the same terminal
/// again. A caller that drained the stream received an endless run of
/// terminals; over the wire it broke the transport, because nothing may follow
/// a terminal frame.
///
/// The count is the assertion. Removing the close in `release_terminal` makes
/// this fail on the second frame rather than at some later bound.
#[tokio::test]
#[allow(
    clippy::significant_drop_tightening,
    reason = "the bound query outlives the stream; dropping it early records cancellation, not the success this case is about"
)]
async fn a_successful_stream_releases_one_terminal_and_then_ends() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "one-terminal",
            "SELECT text FROM messages ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let bound = harness
        .visible_authority()
        .bind(prepared)
        .await
        .expect("exact files bind");

    let mut stream =
        crate::core_service::ProjectedResultStream::start(bound.execute(), 64 * 1024, u64::MAX)
            .expect("a representable frame ceiling");

    let mut kinds = Vec::new();
    let mut released_rows = 0_u64;
    let mut released_bytes = 0_u64;
    let mut terminal = None;
    // Bounded so a stream that never ends fails as a wrong count rather than
    // hanging the suite.
    for _ in 0..32 {
        match stream.next_frame().await {
            Some(frame) => kinds.push(match frame.expect("a produced frame") {
                polyc_query_model::ResultFrame::Schema(_) => "schema",
                polyc_query_model::ResultFrame::Data(data) => {
                    released_rows += data.rows();
                    released_bytes += data.arrow_ipc().len() as u64;
                    "data"
                }
                polyc_query_model::ResultFrame::Terminal(frame) => {
                    terminal = Some(frame);
                    "terminal"
                }
            }),
            None => break,
        }
    }

    assert_eq!(
        kinds.iter().filter(|kind| **kind == "terminal").count(),
        1,
        "exactly one terminal: {kinds:?}"
    );
    assert_eq!(
        kinds.last(),
        Some(&"terminal"),
        "the terminal is last: {kinds:?}"
    );
    assert!(
        stream.next_frame().await.is_none(),
        "the stream stays ended after its terminal"
    );

    // The wire contract, which both peers enforce: a terminal's totals are the
    // opaque bytes and rows released across the data frames. Reporting
    // execution's release budget instead broke every stream carrying a row,
    // while an empty result still passed because both numbers were zero.
    let terminal = terminal.expect("the stream released a terminal");
    assert!(released_rows > 0, "this case releases rows to count");
    assert_eq!(
        terminal.rows(),
        released_rows,
        "terminal rows are released rows"
    );
    assert_eq!(
        terminal.result_bytes(),
        released_bytes,
        "terminal bytes are the bytes released across data frames"
    );
}

/// A frame ceiling too small for the schema is the caller's bound, not a
/// server fault.
///
/// `frame_bytes` is caller-chosen and the protocol accepts any value from one
/// byte up. The encoder used to measure the schema against the protocol's
/// 4 MiB maximum alone, so a one-byte ceiling produced a schema frame the
/// stream's own pump then refused: the server logged a broken-contract error
/// and answered `internal` for a bound the caller picked.
#[tokio::test]
#[allow(
    clippy::significant_drop_tightening,
    reason = "the bound query outlives the refused stream; dropping it early records a withdrawal this case says nothing about"
)]
async fn a_frame_ceiling_below_the_schema_is_the_callers_bound() {
    let harness = Harness::new();
    let prepared = harness
        .prepare(
            "ceiling-below-schema",
            "SELECT text FROM messages ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let bound = harness
        .visible_authority()
        .bind(prepared)
        .await
        .expect("exact files bind");

    let Err(refusal) =
        crate::core_service::ProjectedResultStream::start(bound.execute(), 1, u64::MAX)
    else {
        panic!("a one-byte ceiling cannot carry a schema");
    };
    let crate::core_service::QueryServiceError::Encode(encode) = refusal else {
        panic!("a ceiling refusal is an encode refusal: {refusal:?}");
    };
    assert!(
        matches!(
            encode,
            crate::core_service::FrameEncodeError::SchemaTooLarge { .. }
        ),
        "the schema is measured against the caller's ceiling: {encode:?}"
    );
    assert_eq!(
        encode.class(),
        polyc_query_model::ErrorClass::Bounds,
        "a ceiling the caller chose is the caller's bound"
    );

    // The refusal drops the row stream, which withdraws through the latch. The
    // durable record must still name the bound: the caller withdrew nothing,
    // and a query recorded as cancelled is unreconcilable against one that was
    // refused.
    let completion = harness.await_completion("ceiling-below-schema").await;
    assert_eq!(
        completion.outcome(),
        polyc_state::query_audit::QueryOutcome::Failed(
            polyc_state::query_audit::ErrorClass::Bounds
        ),
        "the durable record says the bound refused it, not that the caller withdrew"
    );
}

/// A row over the caller's ceiling ends the stream with a bounds terminal, and
/// the durable record says so.
///
/// The encode fault used to return `Err` from `next_frame`, which ended the
/// stream with no terminal and left the latch untouched. Dropping the stream
/// then recorded a withdrawal, so the durable audit said the caller cancelled
/// a query the caller had not cancelled.
///
/// The ceiling here is exactly the encoded schema's own size, so the schema
/// fits and no data frame can: a data frame carries the schema message plus a
/// record batch. That holds whatever the fixture rows contain.
#[tokio::test]
#[allow(
    clippy::significant_drop_tightening,
    reason = "each bound query must outlive the stream it produced; dropping one early records a withdrawal these cases exist to rule out"
)]
async fn a_row_over_the_ceiling_ends_with_a_bounds_terminal() {
    let harness = Harness::new();
    let measured = harness
        .prepare(
            "ceiling-measure",
            "SELECT text FROM messages ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let measured = harness
        .visible_authority()
        .bind(measured)
        .await
        .expect("exact files bind");
    let mut measuring =
        crate::core_service::ProjectedResultStream::start(measured.execute(), 64 * 1024, u64::MAX)
            .expect("a representable frame ceiling");
    let polyc_query_model::ResultFrame::Schema(schema) = measuring
        .next_frame()
        .await
        .expect("a first frame")
        .expect("a produced frame")
    else {
        panic!("the first frame is the schema");
    };
    let schema_bytes = schema.arrow_ipc().len() as u64;
    drop(measuring);

    let prepared = harness
        .prepare(
            "row-over-ceiling",
            "SELECT text FROM messages ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let bound = harness
        .visible_authority()
        .bind(prepared)
        .await
        .expect("exact files bind");
    let mut stream =
        crate::core_service::ProjectedResultStream::start(bound.execute(), schema_bytes, u64::MAX)
            .expect("the schema fits its own size");

    let mut kinds = Vec::new();
    let mut terminal = None;
    for _ in 0..32 {
        match stream.next_frame().await {
            Some(frame) => kinds.push(match frame.expect("no frame is an error") {
                polyc_query_model::ResultFrame::Schema(_) => "schema",
                polyc_query_model::ResultFrame::Data(_) => "data",
                polyc_query_model::ResultFrame::Terminal(frame) => {
                    terminal = Some(frame);
                    "terminal"
                }
            }),
            None => break,
        }
    }

    assert_eq!(
        kinds,
        vec!["schema", "terminal"],
        "no data frame fits, and the stream still ends with one terminal"
    );
    let terminal = terminal.expect("the stream released a terminal");
    assert_eq!(
        terminal.outcome(),
        polyc_query_model::QueryOutcome::Failed(polyc_query_model::ErrorClass::Bounds),
        "the caller's ceiling is the caller's bound"
    );
    assert_eq!(terminal.rows(), 0, "no row was released");

    // Dropping the stream withdraws through the latch. That withdrawal must
    // not overwrite the failure already measured: the caller withdrew nothing,
    // and a durable record saying otherwise is the defect this case pins.
    drop(stream);
    let completion = harness.await_completion("row-over-ceiling").await;
    assert_eq!(
        completion.outcome(),
        polyc_state::query_audit::QueryOutcome::Failed(
            polyc_state::query_audit::ErrorClass::Bounds
        ),
        "the durable record says the bound refused it, not that the caller withdrew"
    );
}

/// A framing failure cannot replace the producer terminal that already became
/// the durable record.
///
/// The producer places one batch in front of the consumer, then reaches its
/// deadline while it waits for another readiness token. The consumer reads
/// that buffered batch only afterwards. Its frame ceiling makes the first row
/// unrepresentable, so the consumer measures `Bounds` after the producer has
/// already selected `Deadline`. The caller must receive the selected deadline,
/// not a locally built bounds terminal that the audit rejected.
#[tokio::test]
async fn a_late_framing_failure_takes_the_producers_selected_terminal() {
    let harness = Harness::new();

    let measured = harness
        .prepare(
            "late-frame-measure",
            "SELECT text FROM messages ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            10,
        )
        .await;
    let mut measuring = crate::core_service::ProjectedResultStream::start(
        harness
            .visible_authority()
            .bind(measured)
            .await
            .expect("exact files bind")
            .execute(),
        64 * 1024,
        u64::MAX,
    )
    .expect("a representable frame ceiling");
    let polyc_query_model::ResultFrame::Schema(schema) = measuring
        .next_frame()
        .await
        .expect("a first frame")
        .expect("a produced frame")
    else {
        panic!("the first frame is the schema");
    };
    let schema_bytes = schema.arrow_ipc().len() as u64;
    drop(measuring);

    let prepared = harness
        .prepare_with_timeout(
            "late-framing-failure",
            "SELECT text FROM messages ORDER BY position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            Duration::from_millis(200),
        )
        .await;
    let mut stream = crate::core_service::ProjectedResultStream::start(
        harness
            .visible_authority()
            .bind(prepared)
            .await
            .expect("exact files bind")
            .execute(),
        schema_bytes,
        u64::MAX,
    )
    .expect("the schema fits its own size");
    assert!(matches!(
        stream.next_frame().await,
        Some(Ok(polyc_query_model::ResultFrame::Schema(_)))
    ));

    // Request the batch without dequeuing it. Once it is buffered, the
    // producer waits for another readiness token and its own deadline wins.
    stream.request_buffered_batch();
    tokio::time::timeout(Duration::from_secs(1), async {
        while stream.buffered_frames() == 0 {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("the producer buffered one batch");
    tokio::time::timeout(Duration::from_secs(1), async {
        while !stream.terminal_selected() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("the producer selected its deadline");

    let terminal = match stream.next_frame().await.expect("a terminal follows") {
        Ok(polyc_query_model::ResultFrame::Terminal(frame)) => frame,
        Ok(polyc_query_model::ResultFrame::Data(data)) => {
            panic!("a terminal query released {} rows", data.rows())
        }
        Ok(polyc_query_model::ResultFrame::Schema(_)) => {
            panic!("the schema is released exactly once")
        }
        Err(error) => panic!("no frame is an error: {error}"),
    };
    assert_eq!(
        terminal.outcome(),
        polyc_query_model::QueryOutcome::Failed(polyc_query_model::ErrorClass::Deadline),
        "the caller takes the terminal that already became the record"
    );
    assert_eq!(terminal.rows(), 0, "the buffered batch reached no caller");

    drop(stream);
    let completion = harness.await_completion("late-framing-failure").await;
    assert_one_result(&terminal, &completion);
}

/// Asserts the protocol terminal and the durable completion describe one
/// result: same outcome, same rows, same truncation, same premises.
///
/// This is the invariant the release boundary exists for. The two carry
/// different vocabularies, so the comparison is spelled out rather than
/// derived from one side.
fn assert_one_result(
    terminal: &polyc_query_model::TerminalFrame,
    completion: &polyc_state::query_audit::QueryCompletion,
) {
    let durable_outcome = match terminal.outcome() {
        polyc_query_model::QueryOutcome::Succeeded => {
            polyc_state::query_audit::QueryOutcome::Succeeded
        }
        polyc_query_model::QueryOutcome::Failed(class) => {
            polyc_state::query_audit::QueryOutcome::Failed(crate::core_evidence::durable_class_of(
                class,
            ))
        }
    };
    assert_eq!(completion.outcome(), durable_outcome, "same outcome");
    assert_eq!(
        completion.rows(),
        polyc_state::query_audit::RowCount::new(terminal.rows()),
        "same rows"
    );
    let durable_truncation = match terminal.truncation() {
        polyc_query_model::Truncation::Complete => polyc_state::query_audit::Truncation::Complete,
        polyc_query_model::Truncation::TruncatedAt(rows) => {
            polyc_state::query_audit::Truncation::TruncatedAt(rows)
        }
    };
    assert_eq!(
        completion.truncation(),
        durable_truncation,
        "same truncation"
    );
    assert_eq!(
        terminal.source(),
        &crate::core_evidence::evidence_of(completion.source())
            .expect("the source vector is representable"),
        "same premises"
    );
}

/// Binds one query and frames it under the two given ceilings.
async fn byte_ceiling_stream(
    harness: &Harness,
    query: &str,
    frame_ceiling: u64,
    release_ceiling: u64,
) -> crate::core_service::ProjectedResultStream {
    let prepared = harness
        .prepare(
            query,
            // Wide enough that one row's frame is well under the whole
            // result's. The three-row fixture alone cannot be split:
            // Arrow's fixed overhead dominates it, so every prefix encodes
            // to the same size and no ceiling below the total admits
            // anything.
            "SELECT m.text FROM messages m, messages n ORDER BY m.position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            64,
        )
        .await;
    let bound = harness
        .visible_authority()
        .bind(prepared)
        .await
        .expect("exact files bind");
    crate::core_service::ProjectedResultStream::start(
        bound.execute(),
        frame_ceiling,
        release_ceiling,
    )
    .expect("a schema under both ceilings")
}

/// Drains a stream, returning each data frame's rows and encoded bytes,
/// the terminal, and the frames the encoder built to produce them.
async fn drain_frames(
    stream: &mut crate::core_service::ProjectedResultStream,
) -> (Vec<(u64, u64)>, polyc_query_model::TerminalFrame, u64) {
    let mut frames = Vec::new();
    let mut terminal = None;
    for _ in 0..64 {
        match stream.next_frame().await {
            Some(frame) => match frame.expect("no frame is an error") {
                polyc_query_model::ResultFrame::Schema(_) => {}
                polyc_query_model::ResultFrame::Data(data) => {
                    frames.push((data.rows(), data.arrow_ipc().len() as u64));
                }
                polyc_query_model::ResultFrame::Terminal(frame) => terminal = Some(frame),
            },
            None => break,
        }
        // Framing never runs ahead of release. A framer that encoded the
        // whole batch up front satisfies this only once the caller has drained
        // it, so the check is made after every frame rather than at the end.
        assert!(
            stream.frames_built() <= u64::try_from(frames.len()).unwrap() + 1,
            "the encoder builds at most one frame beyond the released ones"
        );
    }
    let built = stream.frames_built();
    (
        frames,
        terminal.expect("the stream released a terminal"),
        built,
    )
}

/// The caller's release ceiling is measured in the unit the protocol defines,
/// and one accounting decides both the terminal and the durable record.
///
/// The engine bounds release by a batch's decoded size in memory; the wire
/// contract bounds the same caller number by the encoded bytes summed across
/// data frames. Every data frame repeats the schema message, so the encoded
/// total grows with the frame count — and the caller chooses that count
/// through `frame_bytes`. A small enough `frame_bytes` therefore made this
/// server produce a stream its own contract refuses.
///
/// Stopping short is only half of it. The rows of a whole batch were counted
/// as delivered the moment the batch left the result stream, before its frames
/// were built, and stopping mid-batch then dropped the stream — which
/// withdraws. So the caller could read a truncated success while the durable
/// audit recorded `Failed(Cancelled)` over every row of the last batch,
/// including the ones no caller ever saw.
///
/// Both ceilings come from the stream itself. One frame carrying every row is
/// measured first; a ceiling one byte under that forces a split, and the same
/// number as the release ceiling then admits the first frame and no other,
/// because two frames carry the schema message twice.
#[tokio::test]
#[allow(
    clippy::significant_drop_tightening,
    reason = "each bound query outlives the stream it produced; dropping one early records a withdrawal this case exists to rule out"
)]
async fn release_stops_at_the_callers_encoded_byte_ceiling() {
    let harness = Harness::new();

    // One frame carrying the whole result, under ceilings that cannot bind.
    let mut whole =
        byte_ceiling_stream(&harness, "byte-ceiling-measure", 64 * 1024, u64::MAX).await;
    let (frames, terminal, _) = drain_frames(&mut whole).await;
    assert_eq!(
        frames.len(),
        1,
        "the whole result fits one frame: {frames:?}"
    );
    let (all_rows, all_bytes) = frames[0];
    assert!(all_rows > 1, "this case needs a result that can be split");
    assert_eq!(
        terminal.truncation(),
        polyc_query_model::Truncation::Complete,
        "the unbounded run is not truncated"
    );
    drop(whole);

    // One byte under that splits the result, and the same number as the
    // release ceiling admits the first frame alone.
    let ceiling = all_bytes - 1;
    let mut bounded = byte_ceiling_stream(&harness, "encoded-byte-ceiling", ceiling, ceiling).await;
    let (frames, terminal, built) = drain_frames(&mut bounded).await;

    let released_rows: u64 = frames.iter().map(|(rows, _)| rows).sum();
    let released_bytes: u64 = frames.iter().map(|(_, bytes)| bytes).sum();
    assert!(!frames.is_empty(), "the caller receives what fits");
    assert!(
        released_bytes <= ceiling,
        "released {released_bytes} bytes, over the caller's {ceiling} byte ceiling"
    );
    assert!(
        released_rows < all_rows,
        "the ceiling actually bound: {released_rows} of {all_rows} rows"
    );
    assert_eq!(
        terminal.result_bytes(),
        released_bytes,
        "the terminal reports the bytes this stream released"
    );
    assert_eq!(terminal.rows(), released_rows, "and the rows");
    assert_eq!(
        terminal.outcome(),
        polyc_query_model::QueryOutcome::Succeeded,
        "a caller's own ceiling is not a failure"
    );
    assert_eq!(
        terminal.truncation(),
        polyc_query_model::Truncation::TruncatedAt(released_rows),
        "stopping at the caller's byte ceiling is a truncation"
    );

    // Framing stops where release stops. One frame beyond the released prefix
    // is built, and that one is what discovers the ceiling; the rest of the
    // batch is never encoded.
    assert_eq!(
        built,
        u64::try_from(frames.len()).unwrap() + 1,
        "the encoder builds the released frames and the one that did not fit"
    );

    // Dropping the stream stops the producer. It must not relabel a bound the
    // consumer settled itself.
    drop(bounded);
    let completion = harness.await_completion("encoded-byte-ceiling").await;
    assert_ne!(
        completion.outcome(),
        polyc_state::query_audit::QueryOutcome::Failed(
            polyc_state::query_audit::ErrorClass::Cancelled
        ),
        "the caller withdrew nothing; it read exactly the result it asked for"
    );
    assert_eq!(
        completion.outcome(),
        polyc_state::query_audit::QueryOutcome::Succeeded,
        "the durable outcome is the terminal's outcome"
    );
    assert_eq!(
        completion.rows(),
        polyc_state::query_audit::RowCount::new(released_rows),
        "the durable row count is the rows the caller received"
    );
    assert_one_result(&terminal, &completion);
}

/// A caller that abandons the stream mid-batch is recorded as withdrawing,
/// and its row count is what the caller actually received.
///
/// One `DataFusion` batch becomes several protocol frames. The rows of the whole
/// batch were counted as delivered when the batch left the result stream —
/// before any frame was built — so a caller that took one frame and dropped
/// the stream had every row of that batch recorded against it, including the
/// ones it never saw. The count belongs to whoever knows what was released.
///
/// This case also holds the line the bounded-success path must not cross:
/// abandonment is still a withdrawal, and still reports `Cancelled`.
#[tokio::test]
async fn an_abandoned_stream_records_only_the_rows_the_caller_received() {
    let harness = Harness::new();

    // A frame ceiling that splits the result, and a release ceiling that does
    // not bind: the stop here is the caller's, not a bound's.
    let mut whole = byte_ceiling_stream(&harness, "abandon-measure", 64 * 1024, u64::MAX).await;
    let (frames, _terminal, _built) = drain_frames(&mut whole).await;
    let (all_rows, all_bytes) = frames[0];
    assert!(all_rows > 1, "this case needs a result that can be split");
    drop(whole);

    let mut abandoned =
        byte_ceiling_stream(&harness, "abandon-midway", all_bytes - 1, u64::MAX).await;
    let mut released_rows = 0;
    for _ in 0..2 {
        match abandoned.next_frame().await.expect("a frame") {
            Ok(polyc_query_model::ResultFrame::Data(data)) => released_rows += data.rows(),
            Ok(_schema) => {}
            Err(error) => panic!("no frame is an error: {error}"),
        }
    }
    assert!(released_rows > 0, "the caller received a frame");
    assert!(
        released_rows < all_rows,
        "and stopped with rows still to come: {released_rows} of {all_rows}"
    );
    drop(abandoned);

    let completion = harness.await_completion("abandon-midway").await;
    assert_eq!(
        completion.outcome(),
        polyc_state::query_audit::QueryOutcome::Failed(
            polyc_state::query_audit::ErrorClass::Cancelled
        ),
        "abandoning a stream is still a withdrawal"
    );
    assert_eq!(
        completion.rows(),
        polyc_state::query_audit::RowCount::new(released_rows),
        "a withdrawal reports the rows the caller received, not the rows the \
         batch carried"
    );
    assert_eq!(
        completion.truncation(),
        polyc_state::query_audit::Truncation::Complete,
        "abandoning is not stopping at a ceiling; the outcome carries that, \
         not the truncation"
    );
}

/// Framing stops where the durable record stops.
///
/// A deadline can settle a query while this stream still holds a batch it has
/// only partly framed. The record is fixed at that moment, so every frame
/// released after it carries rows the record does not count — and the caller
/// would read a terminal whose row count the audit contradicts. The consumer
/// therefore discards what it is holding once the query has settled.
///
/// The frame ceiling is measured from the result itself, so the batch really
/// does split; the deadline then fires while the producer waits for the
/// consumer's next readiness token, which it never sends.
#[tokio::test]
#[allow(
    clippy::significant_drop_tightening,
    reason = "the bound query outlives the stream; the case turns on what the producer does while the consumer still holds a batch"
)]
async fn framing_stops_when_the_query_settles_under_the_caller() {
    let harness = Harness::new();

    // One frame carrying the whole result, to derive a ceiling that splits it.
    let mut whole = byte_ceiling_stream(&harness, "settle-measure", 64 * 1024, u64::MAX).await;
    let (frames, _terminal, _built) = drain_frames(&mut whole).await;
    let (all_rows, all_bytes) = frames[0];
    assert!(all_rows > 1, "this case needs a result that can be split");
    drop(whole);

    let prepared = harness
        .prepare_with_timeout(
            "settle-under-caller",
            "SELECT m.text FROM messages m, messages n ORDER BY m.position",
            QueryScope::Conversations(vec!["a".to_owned()]),
            Duration::from_millis(200),
        )
        .await;
    let bound = harness
        .visible_authority()
        .bind(prepared)
        .await
        .expect("exact files bind");
    let mut stream =
        crate::core_service::ProjectedResultStream::start(bound.execute(), all_bytes - 1, u64::MAX)
            .expect("a schema under both ceilings");

    // Schema, then one data frame. The rest of the batch stays unframed.
    let mut released_rows = 0;
    for _ in 0..2 {
        match stream.next_frame().await.expect("a frame") {
            Ok(polyc_query_model::ResultFrame::Data(data)) => released_rows += data.rows(),
            Ok(_schema) => {}
            Err(error) => panic!("no frame is an error: {error}"),
        }
    }
    assert!(released_rows > 0, "the caller received a frame");
    assert!(released_rows < all_rows, "with rows still unframed");
    let built_before = stream.frames_built();

    // The producer is waiting for a readiness token this consumer will not
    // send, so its own deadline is what ends the query.
    tokio::time::sleep(Duration::from_millis(260)).await;

    let terminal = loop {
        match stream.next_frame().await.expect("a terminal follows") {
            Ok(polyc_query_model::ResultFrame::Terminal(frame)) => break frame,
            Ok(polyc_query_model::ResultFrame::Data(data)) => {
                panic!("a settled query released {} more rows", data.rows())
            }
            Ok(polyc_query_model::ResultFrame::Schema(_)) => {}
            Err(error) => panic!("no frame is an error: {error}"),
        }
    };
    assert_eq!(
        stream.frames_built(),
        built_before + 1,
        "one frame is built and refused admission; the rest of the batch is not framed"
    );
    assert_eq!(
        terminal.outcome(),
        polyc_query_model::QueryOutcome::Failed(polyc_query_model::ErrorClass::Deadline),
        "the producer's measured deadline is the terminal"
    );
    assert_eq!(
        terminal.rows(),
        released_rows,
        "and reports what was released"
    );

    drop(stream);
    let completion = harness.await_completion("settle-under-caller").await;
    assert_eq!(
        completion.outcome(),
        polyc_state::query_audit::QueryOutcome::Failed(
            polyc_state::query_audit::ErrorClass::Deadline
        ),
        "a measured deadline outranks the withdrawal that follows it"
    );
    assert_eq!(
        completion.rows(),
        polyc_state::query_audit::RowCount::new(released_rows),
        "and the record counts exactly the rows the caller received"
    );
    assert_one_result(&terminal, &completion);
}