polyc-query 2026.9.0

use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Mutex, PoisonError};
use std::time::Duration;

use super::*;
use polyc_state::cancel::CancellationToken;
use polyc_state::context::CallContext;
use polyc_state::deadline::{Deadline, MonotonicInstant};
use polyc_state::feed::SourceCheckpoint;
use polyc_state::id::OperationFamily;
use polyc_state::immutable::{ContentReference, Generation, ObjectDescriptor, Retention};
use polyc_state::journal::{JournalAttestation, JournalDirectorySnapshotId};
use polyc_state::page::PageCompleteness;
use polyc_state::projection::artifact::{ExactObjectRef, ObjectNamespace};
use polyc_state::projection::{ProjectionGeneration, ProjectionHead, PublisherFence, PublisherId};
use polyc_state::query_audit::{QueryAuditWrite, memory::MemoryQueryAudit};
use polyc_state::revision::{CommitRoot, JournalPosition, JournalSource, PartitionIncarnation};

#[derive(Debug, Default)]
struct Counts {
    snapshots: usize,
    pages: usize,
    releases: usize,
    source_reads: usize,
    resolutions: usize,
    audit_writes: usize,
    lifecycle: Vec<&'static str>,
    operation_contexts: BTreeSet<usize>,
    subcall_budgets: Vec<Duration>,
    subcall_audiences: BTreeSet<String>,
}

#[derive(Debug)]
struct FakeState {
    sources: Mutex<BTreeMap<PartitionId, JournalSourceHead>>,
    resolutions: Mutex<BTreeMap<PartitionId, ProjectionResolution>>,
    snapshot: Mutex<Vec<PartitionId>>,
    audit: MemoryQueryAudit,
    intents: Mutex<Vec<BeginQueryAudit>>,
    counts: Mutex<Counts>,
    outage: Mutex<bool>,
    refuse_audit: Mutex<bool>,
    cross_permit: Mutex<bool>,
    settlements: Mutex<usize>,
}

impl FakeState {
    fn new(partitions: &[&str]) -> Self {
        let owner = OwnerId::new("projector");
        let mut sources = BTreeMap::new();
        let mut resolutions = BTreeMap::new();
        for (ordinal, partition) in partitions.iter().enumerate() {
            let partition = PartitionId::new(*partition);
            let source = source(partition.clone(), u8::try_from(ordinal + 1).unwrap());
            sources.insert(
                partition.clone(),
                JournalSourceHead::new(source.clone(), JournalPosition::new(20)),
            );
            resolutions.insert(
                partition.clone(),
                ProjectionResolution::new(
                    ProjectionHead::Current(Box::new(manifest(
                        partition,
                        &source,
                        owner.clone(),
                        Classification::Confidential,
                        1,
                        1,
                    ))),
                    None,
                    ProjectionGeneration::new(1),
                ),
            );
        }
        Self {
            sources: Mutex::new(sources),
            resolutions: Mutex::new(resolutions),
            snapshot: Mutex::new(
                partitions
                    .iter()
                    .map(|partition| PartitionId::new(*partition))
                    .collect(),
            ),
            audit: MemoryQueryAudit::new(),
            intents: Mutex::new(Vec::new()),
            counts: Mutex::new(Counts::default()),
            outage: Mutex::new(false),
            refuse_audit: Mutex::new(false),
            cross_permit: Mutex::new(false),
            settlements: Mutex::new(0),
        }
    }

    fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
        mutex.lock().unwrap_or_else(PoisonError::into_inner)
    }

    fn set_resolution(&self, partition: &str, resolution: ProjectionResolution) {
        Self::lock(&self.resolutions).insert(PartitionId::new(partition), resolution);
    }

    fn set_outage(&self) {
        *Self::lock(&self.outage) = true;
    }

    fn cross_permit(&self) {
        *Self::lock(&self.cross_permit) = true;
    }

    fn settlements(&self) -> usize {
        *Self::lock(&self.settlements)
    }

    fn refuse_audit(&self) {
        *Self::lock(&self.refuse_audit) = true;
    }

    fn source_read_count(&self) -> usize {
        Self::lock(&self.counts).source_reads
    }

    fn resolved_partitions(&self) -> Vec<PartitionId> {
        let intents = Self::lock(&self.intents);
        intents
            .last()
            .map(|intent| {
                intent
                    .source()
                    .pins()
                    .iter()
                    .filter_map(|pin| match pin {
                        SourcePin::Projected(pin) => Some(pin.manifest().key().source().clone()),
                        SourcePin::Journal(_) | SourcePin::Authoritative(_) => None,
                    })
                    .collect()
            })
            .unwrap_or_default()
    }

    fn last_source_pins(&self) -> Vec<SourcePin> {
        Self::lock(&self.intents)
            .last()
            .map(|intent| intent.source().pins().to_vec())
            .unwrap_or_default()
    }

    fn source_lifecycle(&self) -> Vec<&'static str> {
        Self::lock(&self.counts).lifecycle.clone()
    }

    fn observe_operation(&self, operation: &CoreOperationContext) {
        let address = std::ptr::from_ref(operation).addr();
        let declared = operation
            .declared()
            .expect("the planning seam preflights each metadata call");
        let mut counts = Self::lock(&self.counts);
        counts.operation_contexts.insert(address);
        counts.subcall_budgets.push(declared.budget);
        counts
            .subcall_audiences
            .insert(declared.audience.as_str().to_owned());
    }
}

#[async_trait]
impl CoreMetadataAuthority for FakeState {
    async fn create_directory_snapshot(
        &self,
        operation: &CoreOperationContext,
    ) -> Result<JournalDirectorySnapshot, CoreResolutionError> {
        self.observe_operation(operation);
        let sequence = {
            let mut counts = Self::lock(&self.counts);
            counts.snapshots += 1;
            counts.snapshots
        };
        let count = Self::lock(&self.snapshot).len();
        Ok(JournalDirectorySnapshot::new(
            JournalDirectorySnapshotId::new(format!("snapshot-{sequence}")),
            u64::try_from(count).unwrap(),
        ))
    }

    async fn directory_page(
        &self,
        operation: &CoreOperationContext,
        request: ListJournalDirectorySnapshot,
    ) -> Result<JournalDirectoryPage, CoreResolutionError> {
        self.observe_operation(operation);
        Self::lock(&self.counts).pages += 1;
        let all = Self::lock(&self.snapshot);
        let partitions = all
            .iter()
            .filter(|partition| request.start_after().is_none_or(|after| *partition > after))
            .take(request.limit() as usize)
            .cloned()
            .collect::<Vec<_>>();
        let consumed = request.start_after().map_or(0, |after| {
            all.iter().filter(|partition| *partition <= after).count()
        }) + partitions.len();
        let truncated = consumed < all.len();
        Ok(JournalDirectoryPage::new(
            request.snapshot().clone(),
            partitions.clone(),
            truncated.then(|| partitions.last().unwrap().clone()),
            if truncated {
                PageCompleteness::Truncated
            } else {
                PageCompleteness::Complete
            },
        ))
    }

    async fn release_directory_snapshot(
        &self,
        operation: &CoreOperationContext,
        _request: ReleaseJournalDirectorySnapshot,
    ) -> Result<(), CoreResolutionError> {
        self.observe_operation(operation);
        Self::lock(&self.counts).releases += 1;
        Ok(())
    }

    async fn source_head(
        &self,
        operation: &CoreOperationContext,
        request: GetJournalSource,
    ) -> Result<Option<JournalSourceHead>, CoreResolutionError> {
        self.observe_operation(operation);
        Self::lock(&self.counts).source_reads += 1;
        Self::lock(&self.counts).lifecycle.push("source");
        if *Self::lock(&self.outage) {
            return Err(StateError::Unavailable {
                family: OperationFamily::new("journal"),
                reach: polyc_state::error::OutageReach::NoDurableEffect,
            }
            .into());
        }
        Ok(Self::lock(&self.sources).get(request.partition()).cloned())
    }

    async fn resolve_manifest(
        &self,
        operation: &CoreOperationContext,
        request: ResolveManifest,
    ) -> Result<ProjectionResolution, CoreResolutionError> {
        self.observe_operation(operation);
        Self::lock(&self.counts).resolutions += 1;
        Self::lock(&self.counts).lifecycle.push("projection");
        Ok(Self::lock(&self.resolutions)
            .get(request.key().source())
            .cloned()
            .unwrap_or_else(|| {
                ProjectionResolution::new(
                    ProjectionHead::Absent,
                    None,
                    ProjectionGeneration::ORIGIN,
                )
            }))
    }

    async fn begin_audit(
        &self,
        operation: &CoreOperationContext,
        command: BeginQueryAudit,
    ) -> Result<BeginOutcome<CoreExecutionPermit>, CoreResolutionError> {
        self.observe_operation(operation);
        Self::lock(&self.counts).audit_writes += 1;
        Self::lock(&self.counts).lifecycle.push("audit");
        Self::lock(&self.intents).push(command.clone());
        if *Self::lock(&self.refuse_audit) {
            return Err(StateError::Unavailable {
                family: polyc_state::query_audit::family(),
                reach: polyc_state::error::OutageReach::PossiblyApplied,
            }
            .into());
        }
        let command = if *Self::lock(&self.cross_permit) {
            BeginQueryAudit::new(
                QueryId::new("another-query"),
                command.namespace().clone(),
                command.requester().clone(),
                command.shape(),
                command.source().clone(),
                command.metadata().digest(),
                command.metadata().envelope().clone(),
            )
        } else {
            command
        };
        match self.audit.begin(command, &live_context())? {
            BeginOutcome::Granted(permit) => {
                Ok(BeginOutcome::Granted(CoreExecutionPermit::from(permit)))
            }
            BeginOutcome::AlreadyRecorded(receipt) => Ok(BeginOutcome::AlreadyRecorded(receipt)),
        }
    }

    async fn complete_audit(
        &self,
        _operation: &CoreCompletionContext,
        _command: &CoreCompletionCommand,
    ) -> Result<Receipt, CoreResolutionError> {
        *Self::lock(&self.settlements) += 1;
        Err(CoreResolutionError::InvalidComposition)
    }

    async fn completion_receipt(
        &self,
        _operation: &CoreCompletionContext,
        _command: &CoreCompletionCommand,
    ) -> Result<Option<Receipt>, CoreResolutionError> {
        *Self::lock(&self.settlements) += 1;
        Err(CoreResolutionError::InvalidComposition)
    }
}

fn live_context() -> CallContext {
    CallContext::new(
        Deadline::at(MonotonicInstant::from_nanos(u64::MAX)),
        CancellationToken::new(),
    )
}

fn source(partition: PartitionId, byte: u8) -> JournalSource {
    JournalSource::new(
        partition,
        PartitionIncarnation::from_bytes([byte; PartitionIncarnation::LEN]),
    )
}

fn manifest(
    partition: PartitionId,
    source: &JournalSource,
    owner: OwnerId,
    classification: Classification,
    schema_version: u32,
    fact_version: u32,
) -> ProjectionManifest {
    manifest_for_family(
        "conversation-core/v1",
        partition,
        source,
        owner,
        classification,
        schema_version,
        fact_version,
    )
}

#[allow(clippy::too_many_arguments)]
fn manifest_for_family(
    family: &str,
    partition: PartitionId,
    source: &JournalSource,
    owner: OwnerId,
    classification: Classification,
    schema_version: u32,
    fact_version: u32,
) -> ProjectionManifest {
    let key = ProjectionKey::new(FamilyId::new(family), partition);
    let generation = ProjectionGeneration::new(1);
    let reference =
        ContentReference::try_new(format!("projection/{}/manifest", key.source().as_str()))
            .unwrap();
    ProjectionManifest::new(
        key.clone(),
        generation,
        SourceCheckpoint::try_new(
            source.clone(),
            JournalPosition::new(10),
            JournalPosition::new(20),
            20,
            JournalAttestation::new(
                CommitRoot::from_bytes([7; CommitRoot::LEN]),
                22,
                vec![8; 64],
                vec![9; 32],
            ),
        )
        .unwrap(),
        schema_version,
        fact_version,
        ObjectDescriptor::new(
            key.object(),
            Generation::new(1),
            ContentDigest::from_bytes([5; ContentDigest::LEN]),
            owner,
            classification,
            Retention::For(Duration::from_mins(1)),
            128,
            reference.clone(),
        ),
        ExactObjectRef::try_new(
            ObjectNamespace::try_new("conversation-visible").unwrap(),
            reference,
            11,
        )
        .unwrap(),
        PublisherId::new("projector-a"),
        PublisherFence::new(key, source.incarnation(), 1),
    )
}

fn request(sql: &str) -> CoreQueryRequest {
    CoreQueryRequest {
        sql: sql.to_owned(),
        parameters: Vec::new(),
        consistency: CoreConsistency::Projected,
        requested_bounds: CoreRequestedBounds::unbounded(),
    }
}

fn audit(query: &str) -> CoreAuditContext {
    audit_with_declared(
        query,
        &DeclaredCall::live(Audience::new("state"), Duration::MAX),
    )
}

fn audit_with_declared(query: &str, declared: &DeclaredCall) -> CoreAuditContext {
    audit_for_bounds(query, declared, CoreRequestedBounds::unbounded())
}

fn audit_for_bounds(
    query: &str,
    declared: &DeclaredCall,
    requested: CoreRequestedBounds,
) -> CoreAuditContext {
    audit_for_policy(
        query,
        declared,
        &QueryLimits::default(),
        ProjectedCorePolicy::default(),
        requested,
    )
}

fn audit_for_policy(
    query: &str,
    declared: &DeclaredCall,
    limits: &QueryLimits,
    policy: ProjectedCorePolicy,
    requested: CoreRequestedBounds,
) -> CoreAuditContext {
    CoreAuditContext::from_scoped(
        QueryId::new(query),
        RequesterId::new("persona:verified-a"),
        declared,
        EffectiveCoreBounds::mint(limits, policy, requested).unwrap(),
    )
}

fn authority(state: Arc<FakeState>) -> CorePlanningAuthority {
    authority_with_policy(state, ProjectedCorePolicy::default())
}

fn authority_with_policy(
    state: Arc<FakeState>,
    policy: ProjectedCorePolicy,
) -> CorePlanningAuthority {
    CorePlanningAuthority::new(
        NamespaceId::new("tenant-a"),
        OwnerId::new("projector"),
        policy,
        state,
    )
    .unwrap()
}

#[tokio::test]
async fn schema_only_planning_derives_dependencies_without_a_physical_scan() {
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let logical = compiler
        .compile(
            "SELECT turns.turn_id, messages.text FROM turns JOIN messages USING (turn_id)",
            &[],
            false,
        )
        .await
        .expect("schema-only plan");

    assert_eq!(
        logical.dependencies(),
        &[CoreTable::Turns, CoreTable::Messages]
    );
    assert_eq!(compiler.physical_scan_count(), 0);
}

#[tokio::test]
async fn schema_only_planning_closes_mixed_dependencies_without_a_physical_scan() {
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let logical = compiler
        .compile(
            "SELECT messages.text, tool_calls.name \
             FROM messages JOIN tool_calls USING (partition, turn_id)",
            &[],
            false,
        )
        .await
        .expect("mixed schema-only plan");

    assert_eq!(logical.dependencies(), &[CoreTable::Messages]);
    assert_eq!(logical.legacy_dependencies(), &[LegacyTable::ToolCalls]);
    assert_eq!(compiler.physical_scan_count(), 0);
}

#[tokio::test]
async fn mixed_sources_are_complete_before_the_durable_intent() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let outcome = authority(Arc::clone(&state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit("q-mixed"),
            request(
                "SELECT messages.text, tool_calls.name \
                 FROM messages JOIN tool_calls USING (partition, turn_id)",
            ),
        )
        .await
        .expect("mixed plan");

    assert!(matches!(outcome, CorePlanOutcome::Granted(_)));
    let pins = state.last_source_pins();
    assert_eq!(pins.len(), 2);
    assert!(matches!(pins[0], SourcePin::Projected(_)));
    assert!(matches!(pins[1], SourcePin::Journal(_)));
    assert_eq!(state.source_lifecycle(), ["source", "projection", "audit"]);
    assert_eq!(compiler.physical_scan_count(), 0);
}

#[tokio::test]
async fn legacy_only_planning_does_not_resolve_a_core_manifest() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let outcome = authority(Arc::clone(&state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit("q-legacy"),
            request("SELECT name FROM tool_calls"),
        )
        .await
        .expect("legacy-only plan");

    assert!(matches!(outcome, CorePlanOutcome::Granted(_)));
    let pins = state.last_source_pins();
    assert_eq!(pins.len(), 1);
    assert!(matches!(pins[0], SourcePin::Journal(_)));
    assert_eq!(state.source_lifecycle(), ["source", "audit"]);
    assert_eq!(compiler.physical_scan_count(), 0);
}

#[tokio::test]
async fn mixed_pin_bound_refuses_before_source_resolution() {
    let conversations = (0..17)
        .map(|index| format!("c-{index}"))
        .collect::<Vec<_>>();
    let partitions = conversations
        .iter()
        .map(|conversation| format!("conv-{conversation}"))
        .collect::<Vec<_>>();
    let partition_refs = partitions.iter().map(String::as_str).collect::<Vec<_>>();
    let state = Arc::new(FakeState::new(&partition_refs));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let error = authority(Arc::clone(&state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(conversations),
            false,
            audit("q-mixed-bound"),
            request(
                "SELECT messages.text, tool_calls.name \
                 FROM messages JOIN tool_calls USING (partition, turn_id)",
            ),
        )
        .await
        .expect_err("two pins per partition exceed the complete source bound");

    assert!(matches!(
        error,
        CoreResolutionError::State(StateError::BoundsExceeded { requested: 34, .. })
    ));
    assert!(state.source_lifecycle().is_empty());
}

#[tokio::test]
async fn a_missing_core_projection_never_falls_back_to_the_journal_pin() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    state.set_resolution(
        "conv-a",
        ProjectionResolution::new(ProjectionHead::Absent, None, ProjectionGeneration::ORIGIN),
    );
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let error = authority(Arc::clone(&state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit("q-no-core-fallback"),
            request(
                "SELECT messages.text, tool_calls.name \
                 FROM messages JOIN tool_calls USING (partition, turn_id)",
            ),
        )
        .await
        .expect_err("the projected dependency stays mandatory");

    assert!(matches!(error, CoreResolutionError::MissingProjection(_)));
    assert_eq!(state.source_lifecycle(), ["source", "projection"]);
    assert!(state.last_source_pins().is_empty());
}

#[tokio::test]
async fn schema_only_planning_cannot_see_an_ambient_catalog_table() {
    let context = SessionContext::new();
    let ambient_scans = Arc::new(AtomicUsize::new(0));
    context
        .register_table(
            "ambient_secret",
            Arc::new(SchemaOnlyTable {
                schema: Arc::new(Schema::new(vec![Field::new(
                    "secret",
                    DataType::Utf8,
                    false,
                )])),
                scans: ambient_scans,
            }),
        )
        .unwrap();
    let compiler = CatalogCompiler::new(context.state());
    let error = compiler
        .compile("SELECT secret FROM ambient_secret", &[], false)
        .await
        .expect_err("the compiler installs a fresh closed catalog");

    assert!(matches!(error, CoreResolutionError::DataFusion(_)));
    assert_eq!(compiler.physical_scan_count(), 0);
}

#[tokio::test]
async fn sql_predicates_cannot_widen_the_authority_scope() {
    let state = Arc::new(FakeState::new(&["conv-a", "conv-b"]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let outcome = authority(Arc::clone(&state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit("q-scope"),
            request("SELECT turn_id FROM turns WHERE partition = 'conv-b'"),
        )
        .await
        .expect("authority scope plans");

    assert!(matches!(outcome, CorePlanOutcome::Granted(_)));
    assert_eq!(
        state.resolved_partitions(),
        vec![PartitionId::new("conv-a")]
    );
    assert_eq!(compiler.physical_scan_count(), 0);
}

#[tokio::test]
async fn exact_intent_replay_returns_evidence_without_a_second_permit() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let planning = authority(Arc::clone(&state));
    let scope = QueryScope::Conversations(vec!["a".to_owned()]);
    let first = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &scope,
            false,
            audit("q-replay"),
            request("SELECT * FROM turns"),
        )
        .await
        .expect("new intent");
    let replay = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &scope,
            false,
            audit("q-replay"),
            request("SELECT * FROM turns"),
        )
        .await
        .expect("exact replay");

    assert!(matches!(first, CorePlanOutcome::Granted(_)));
    assert!(matches!(replay, CorePlanOutcome::AlreadyRecorded(_)));
    assert_eq!(compiler.physical_scan_count(), 0);
}

#[tokio::test]
async fn bounds_and_freshness_refuse_before_source_reads_or_scans() {
    let state = Arc::new(FakeState::new(&[]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let planning = authority(Arc::clone(&state));
    let too_many = (0..=MAX_SOURCE_PINS)
        .map(|index| format!("c-{index}"))
        .collect();
    let bound = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(too_many),
            false,
            audit("q-bound"),
            request("SELECT * FROM messages"),
        )
        .await
        .expect_err("the source vector is never truncated");
    assert!(matches!(
        bound,
        CoreResolutionError::State(StateError::BoundsExceeded { .. })
    ));

    let mut fresh = request("SELECT * FROM turns");
    fresh.consistency = CoreConsistency::RequireProjectedThrough(JournalPosition::new(41));
    let freshness = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit("q-fresh"),
            fresh,
        )
        .await
        .expect_err("freshness is recognized before State reads");
    assert!(matches!(
        freshness,
        CoreResolutionError::FreshnessUnsupported { position }
            if position == JournalPosition::new(41)
    ));
    assert_eq!(state.source_read_count(), 0);
    assert_eq!(compiler.physical_scan_count(), 0);

    let duplicate = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned(), "a".to_owned()]),
            false,
            audit("q-duplicate"),
            request("SELECT * FROM turns"),
        )
        .await
        .expect_err("one source cannot appear twice");
    assert!(matches!(duplicate, CoreResolutionError::DuplicatePartition));

    let empty_conversation = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec![String::new()]),
            false,
            audit("q-empty-conversation"),
            request("SELECT * FROM turns"),
        )
        .await
        .expect_err("an empty conversation cannot become conv-");
    assert!(matches!(
        empty_conversation,
        CoreResolutionError::EmptyConversationIdentity
    ));
    assert_eq!(state.source_read_count(), 0);
}

#[tokio::test]
async fn parameter_mismatch_refuses_before_state_metadata() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let planning = authority(Arc::clone(&state));
    for (index, request) in [
        request("SELECT * FROM turns WHERE turn_id = $1"),
        request("SELECT * FROM turns"),
    ]
    .into_iter()
    .enumerate()
    {
        let mut request = request;
        if index == 1 {
            request.parameters.push(CoreParameter::UInt64(1));
        }
        let error = planning
            .plan(
                &compiler,
                &QueryLimits::default(),
                &QueryScope::Conversations(vec!["a".to_owned()]),
                false,
                audit(&format!("q-param-mismatch-{index}")),
                request,
            )
            .await
            .expect_err("missing, surplus, and invalid typed parameters fail in planning");
        assert!(matches!(error, CoreResolutionError::ParameterMismatch));
    }

    let mut wrong_type = request("SELECT * FROM turns WHERE source_position = $1");
    wrong_type
        .parameters
        .push(CoreParameter::Utf8("not-a-number".to_owned()));
    let error = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit("q-param-type"),
            wrong_type,
        )
        .await
        .expect_err("a typed value must satisfy the planned column type");
    assert!(matches!(error, CoreResolutionError::DataFusion(_)));
    assert_eq!(state.source_read_count(), 0);
    assert_eq!(compiler.physical_scan_count(), 0);
}

#[tokio::test]
async fn typed_parameter_values_and_variants_bind_the_audit_shape() {
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let cases = [
        CoreParameter::Utf8("a".to_owned()),
        CoreParameter::Utf8("b".to_owned()),
        CoreParameter::UInt64(1),
        CoreParameter::Boolean(true),
        CoreParameter::Null,
    ];
    let mut shapes = BTreeSet::new();
    let mut encodings = BTreeSet::new();
    for parameter in cases {
        let state = Arc::new(FakeState::new(&["conv-a"]));
        let mut parameterized = request("SELECT * FROM turns WHERE $1 = $1");
        parameterized.parameters.push(parameter.clone());
        let outcome = authority(Arc::clone(&state))
            .plan(
                &compiler,
                &QueryLimits::default(),
                &QueryScope::Conversations(vec!["a".to_owned()]),
                false,
                audit("q-typed"),
                parameterized,
            )
            .await
            .expect("closed typed parameter plans");
        let CorePlanOutcome::Granted(prepared) = outcome else {
            panic!("a distinct query state records a new intent");
        };
        assert!(
            prepared
                .compiled
                .plan
                .get_parameter_names()
                .unwrap()
                .is_empty(),
            "the retained plan owns bound literals, not placeholders or raw parameters"
        );
        let mut canonical = Vec::new();
        push_parameters(&mut canonical, &[parameter]);
        encodings.insert(canonical);
        shapes.insert(FakeState::lock(&state.intents)[0].shape());
    }
    assert_eq!(shapes.len(), 5);
    assert_eq!(encodings.len(), 5);
    assert_eq!(compiler.physical_scan_count(), 0);
}

#[tokio::test]
async fn expired_or_cancelled_operation_stops_before_metadata_reads() {
    let declarations = [
        DeclaredCall::bounded(Audience::new("state"), Duration::ZERO),
        DeclaredCall::live(Audience::new("state"), Duration::MAX).withdrawn(),
    ];
    for (index, declared) in declarations.iter().enumerate() {
        let state = Arc::new(FakeState::new(&["conv-a"]));
        let compiler = CatalogCompiler::new(SessionContext::new().state());
        let error = authority(Arc::clone(&state))
            .plan(
                &compiler,
                &QueryLimits::default(),
                &QueryScope::Conversations(vec!["a".to_owned()]),
                false,
                audit_with_declared(&format!("q-context-{index}"), declared),
                request("SELECT * FROM turns"),
            )
            .await
            .expect_err("spent operation context cannot reach State metadata");

        assert!(matches!(
            error,
            CoreResolutionError::State(
                StateError::DeadlineExpired { .. } | StateError::Cancelled { .. }
            )
        ));
        let counts = {
            let counts = FakeState::lock(&state.counts);
            (
                counts.snapshots,
                counts.pages,
                counts.releases,
                counts.source_reads,
                counts.resolutions,
                counts.audit_writes,
            )
        };
        assert_eq!(counts, (0, 0, 0, 0, 0, 0));
        assert_eq!(compiler.physical_scan_count(), 0);
    }
}

#[tokio::test]
async fn effective_bounds_are_stable_clamped_and_permit_owned() {
    let default_state = Arc::new(FakeState::new(&["conv-a"]));
    let oversized_state = Arc::new(FakeState::new(&["conv-a"]));
    let lower_state = Arc::new(FakeState::new(&["conv-a"]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let declared = DeclaredCall::live(Audience::new("query"), Duration::from_mins(2));
    let default_limits = QueryLimits::default();
    let mut oversized = CoreRequestedBounds::unbounded();
    oversized.timeout = Duration::from_mins(1);
    oversized.rows = u64::try_from(default_limits.row_cap).unwrap() * 2;
    oversized.result_release_bytes = DEFAULT_CORE_RESULT_RELEASE_BYTES * 2;
    oversized.response_frame_bytes = DEFAULT_CORE_RESPONSE_FRAME_BYTES * 2;
    let mut lower = CoreRequestedBounds::unbounded();
    lower.rows = 100;

    let default = authority(Arc::clone(&default_state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit_for_bounds("q-bounds", &declared, CoreRequestedBounds::unbounded()),
            request("SELECT * FROM turns"),
        )
        .await
        .unwrap();
    let mut oversized_request = request("SELECT * FROM turns");
    oversized_request.requested_bounds = oversized;
    authority(Arc::clone(&oversized_state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit_for_bounds("q-bounds", &declared, oversized),
            oversized_request,
        )
        .await
        .unwrap();
    let mut lower_request = request("SELECT * FROM turns");
    lower_request.requested_bounds = lower;
    authority(Arc::clone(&lower_state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit_for_bounds("q-bounds", &declared, lower),
            lower_request,
        )
        .await
        .unwrap();

    let default_intent = FakeState::lock(&default_state.intents)[0].clone();
    let oversized_intent = FakeState::lock(&oversized_state.intents)[0].clone();
    let lower_intent = FakeState::lock(&lower_state.intents)[0].clone();
    assert_eq!(default_intent.shape(), oversized_intent.shape());
    assert_ne!(default_intent.shape(), lower_intent.shape());
    let CorePlanOutcome::Granted(prepared) = default else {
        panic!("a new bounded intent grants one permit");
    };
    assert_eq!(
        prepared.bounds.rows,
        u64::try_from(default_limits.row_cap).unwrap()
    );
    assert_eq!(prepared.bounds.timeout, default_limits.timeout);
    assert_eq!(
        prepared.bounds.result_release_bytes,
        DEFAULT_CORE_RESULT_RELEASE_BYTES
    );
    assert_eq!(
        prepared.bounds.artifact_file_bytes,
        DEFAULT_CORE_ARTIFACT_FILE_BYTES
    );
}

#[tokio::test]
async fn scoped_limits_and_deployment_policy_cannot_be_widened_or_crossed() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let limits = QueryLimits {
        timeout: Duration::from_secs(7),
        row_cap: 23,
        ..QueryLimits::default()
    };
    let policy = ProjectedCorePolicy::try_from(ProjectedCorePolicyInput {
        result_release_bytes: 8 * 1024,
        response_frame_bytes: 1024,
        manifest_bytes: 4096,
        artifact_file_bytes: 8192,
        artifact_range_bytes: 2048,
        source_decode_bytes: 16 * 1024,
    })
    .unwrap();
    let planning = authority_with_policy(Arc::clone(&state), policy);
    let effective = planning
        .effective_bounds(&limits, CoreRequestedBounds::unbounded())
        .unwrap();
    assert_eq!(effective.timeout, limits.timeout);
    assert_eq!(effective.rows, 23);
    assert_eq!(effective.result_release_bytes, 8 * 1024);
    assert_eq!(effective.response_frame_bytes, 1024);
    assert_eq!(effective.manifest_bytes, 4096);
    assert_eq!(effective.artifact_file_bytes, 8192);
    assert_eq!(effective.artifact_range_bytes, 2048);

    let crossed = planning
        .plan(
            &compiler,
            &limits,
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit_for_policy(
                "q-crossed-policy",
                &DeclaredCall::live(Audience::new("query"), Duration::MAX),
                &limits,
                ProjectedCorePolicy::default(),
                CoreRequestedBounds::unbounded(),
            ),
            request("SELECT * FROM turns"),
        )
        .await
        .expect_err("bounds minted under another deployment policy fail closed");
    assert!(matches!(crossed, CoreResolutionError::InvalidBounds));

    let wider_limits = QueryLimits {
        timeout: Duration::from_mins(1),
        row_cap: 1000,
        ..QueryLimits::default()
    };
    let widened = planning
        .plan(
            &compiler,
            &limits,
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit_for_policy(
                "q-crossed-limits",
                &DeclaredCall::live(Audience::new("query"), Duration::MAX),
                &wider_limits,
                policy,
                CoreRequestedBounds::unbounded(),
            ),
            request("SELECT * FROM turns"),
        )
        .await
        .expect_err("bounds minted above the scoped query limits fail closed");
    assert!(matches!(widened, CoreResolutionError::InvalidBounds));
    assert_eq!(state.source_read_count(), 0);
    assert_eq!(compiler.physical_scan_count(), 0);
}

#[tokio::test]
async fn audit_retry_digest_does_not_depend_on_remaining_time() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let planning = authority(Arc::clone(&state));
    let scope = QueryScope::Conversations(vec!["a".to_owned()]);
    let first_call = DeclaredCall::live(Audience::new("query"), Duration::from_mins(2));
    let retry_call = DeclaredCall::live(Audience::new("query"), Duration::from_secs(20));
    let first = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &scope,
            false,
            audit_with_declared("q-time-retry", &first_call),
            request("SELECT * FROM messages"),
        )
        .await
        .unwrap();
    let retry = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &scope,
            false,
            audit_with_declared("q-time-retry", &retry_call),
            request("SELECT * FROM messages"),
        )
        .await
        .unwrap();

    assert!(matches!(first, CorePlanOutcome::Granted(_)));
    assert!(matches!(retry, CorePlanOutcome::AlreadyRecorded(_)));
    let evidence = {
        let intents = FakeState::lock(&state.intents);
        (
            intents[0].shape(),
            intents[1].shape(),
            intents[0].metadata().digest(),
            intents[1].metadata().digest(),
        )
    };
    assert_eq!(evidence.0, evidence.1);
    assert_eq!(evidence.2, evidence.3);
}

#[tokio::test]
async fn unsorted_authority_scope_produces_the_same_source_and_shape() {
    let first_state = Arc::new(FakeState::new(&["conv-a", "conv-b"]));
    let second_state = Arc::new(FakeState::new(&["conv-a", "conv-b"]));
    let explain_enabled_state = Arc::new(FakeState::new(&["conv-a", "conv-b"]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    authority(Arc::clone(&first_state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["b".to_owned(), "a".to_owned()]),
            false,
            audit("q-deterministic"),
            request("SELECT * FROM messages"),
        )
        .await
        .unwrap();
    authority(Arc::clone(&second_state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned(), "b".to_owned()]),
            false,
            audit("q-deterministic"),
            request("SELECT * FROM messages"),
        )
        .await
        .unwrap();
    authority(Arc::clone(&explain_enabled_state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned(), "b".to_owned()]),
            true,
            audit("q-deterministic"),
            request("SELECT * FROM messages"),
        )
        .await
        .unwrap();

    let first = FakeState::lock(&first_state.intents)[0].clone();
    let second = FakeState::lock(&second_state.intents)[0].clone();
    let explain_enabled = FakeState::lock(&explain_enabled_state.intents)[0].clone();
    assert_eq!(first.shape(), second.shape());
    assert_eq!(first.source(), second.source());
    assert_eq!(first.canonical_bytes(), second.canonical_bytes());
    assert_ne!(first.shape(), explain_enabled.shape());
}

#[tokio::test]
async fn fleet_uses_and_releases_one_anchored_directory_snapshot() {
    let state = Arc::new(FakeState::new(&[
        "conv-b",
        "conv-",
        "conv-a",
        "routine-scheduler",
    ]));
    sort_snapshot(&state);
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let outcome = authority(Arc::clone(&state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Fleet,
            true,
            audit("q-fleet"),
            request("SELECT * FROM messages"),
        )
        .await
        .expect("anchored Fleet plan");

    let CorePlanOutcome::Granted(prepared) = outcome else {
        panic!("new fleet intent grants one permit");
    };
    assert_eq!(
        prepared.partitions,
        vec![PartitionId::new("conv-a"), PartitionId::new("conv-b")]
    );
    let counts = {
        let counts = FakeState::lock(&state.counts);
        (
            counts.snapshots,
            counts.pages,
            counts.releases,
            counts.operation_contexts.len(),
            counts.subcall_budgets.clone(),
            counts.subcall_audiences.clone(),
        )
    };
    assert_eq!((counts.0, counts.1, counts.2), (1, 1, 1));
    assert_eq!(counts.3, 1, "cleanup reuses the anchored operation context");
    assert!(counts.4.windows(2).all(|pair| pair[1] <= pair[0]));
    assert_eq!(
        counts.5,
        BTreeSet::from([polyc_state_connect::STATE_AUDIENCE.to_owned()])
    );
}

#[tokio::test]
async fn fleet_retry_ignores_ephemeral_directory_snapshot_identity() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let planning = authority(Arc::clone(&state));
    let first = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Fleet,
            false,
            audit("q-fleet-retry"),
            request("SELECT * FROM turns"),
        )
        .await
        .expect("first snapshot records the intent");
    let retry = planning
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Fleet,
            false,
            audit("q-fleet-retry"),
            request("SELECT * FROM turns"),
        )
        .await
        .expect("a later snapshot id does not change the request shape");

    assert!(matches!(first, CorePlanOutcome::Granted(_)));
    assert!(matches!(retry, CorePlanOutcome::AlreadyRecorded(_)));
    let counts = {
        let counts = FakeState::lock(&state.counts);
        (counts.snapshots, counts.pages, counts.releases)
    };
    assert_eq!(counts, (2, 2, 2));
    assert_eq!(compiler.physical_scan_count(), 0);
}

fn sort_snapshot(state: &FakeState) {
    FakeState::lock(&state.snapshot).sort();
}

#[tokio::test]
async fn descriptor_and_state_failures_stop_before_audit_or_physical_scan() {
    let cases = [
        (
            OwnerId::new("crossed-owner"),
            Classification::Confidential,
            1,
            1,
        ),
        (OwnerId::new("projector"), Classification::Internal, 1, 1),
        (
            OwnerId::new("projector"),
            Classification::Confidential,
            2,
            1,
        ),
        (
            OwnerId::new("projector"),
            Classification::Confidential,
            1,
            2,
        ),
    ];
    for (index, (owner, classification, schema, fact)) in cases.into_iter().enumerate() {
        let state = Arc::new(FakeState::new(&["conv-a"]));
        let current_source = FakeState::lock(&state.sources)
            .get(&PartitionId::new("conv-a"))
            .unwrap()
            .source()
            .clone();
        let broken = manifest(
            PartitionId::new("conv-a"),
            &current_source,
            owner,
            classification,
            schema,
            fact,
        );
        state.set_resolution(
            "conv-a",
            ProjectionResolution::new(
                ProjectionHead::Current(Box::new(broken)),
                None,
                ProjectionGeneration::new(1),
            ),
        );
        let compiler = CatalogCompiler::new(SessionContext::new().state());
        let error = authority(Arc::clone(&state))
            .plan(
                &compiler,
                &QueryLimits::default(),
                &QueryScope::Conversations(vec!["a".to_owned()]),
                false,
                audit(&format!("q-broken-{index}")),
                request("SELECT * FROM turns"),
            )
            .await
            .expect_err("incompatible descriptor fails closed");
        assert!(matches!(
            error,
            CoreResolutionError::IncompatibleDescriptor(_)
        ));
        assert_eq!(FakeState::lock(&state.counts).audit_writes, 0);
        assert_eq!(compiler.physical_scan_count(), 0);
    }
}

#[tokio::test]
async fn another_projection_family_cannot_satisfy_conversation_core() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    let current_source = FakeState::lock(&state.sources)
        .get(&PartitionId::new("conv-a"))
        .unwrap()
        .source()
        .clone();
    let wrong_family = manifest_for_family(
        "conversation-search/v1",
        PartitionId::new("conv-a"),
        &current_source,
        OwnerId::new("projector"),
        Classification::Confidential,
        1,
        1,
    );
    state.set_resolution(
        "conv-a",
        ProjectionResolution::new(
            ProjectionHead::Current(Box::new(wrong_family)),
            None,
            ProjectionGeneration::new(1),
        ),
    );
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let error = authority(Arc::clone(&state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit("q-wrong-family"),
            request("SELECT * FROM turns"),
        )
        .await
        .expect_err("another projection family cannot satisfy core");
    assert!(matches!(
        error,
        CoreResolutionError::IncompatibleDescriptor(_)
    ));
    assert_eq!(FakeState::lock(&state.counts).audit_writes, 0);
}

#[tokio::test]
async fn source_recreation_and_crossed_source_responses_fail_closed() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    let recreated = source(PartitionId::new("conv-a"), 9);
    let broken = manifest(
        PartitionId::new("conv-a"),
        &recreated,
        OwnerId::new("projector"),
        Classification::Confidential,
        1,
        1,
    );
    state.set_resolution(
        "conv-a",
        ProjectionResolution::new(
            ProjectionHead::Current(Box::new(broken)),
            None,
            ProjectionGeneration::new(1),
        ),
    );
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    let recreated_error = authority(Arc::clone(&state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit("q-recreated"),
            request("SELECT * FROM turns"),
        )
        .await
        .expect_err("a descriptor for another incarnation is not current");
    assert!(matches!(
        recreated_error,
        CoreResolutionError::IncompatibleDescriptor(_)
    ));
    assert_eq!(FakeState::lock(&state.counts).audit_writes, 0);

    let crossed = Arc::new(FakeState::new(&["conv-a"]));
    FakeState::lock(&crossed.sources).insert(
        PartitionId::new("conv-a"),
        JournalSourceHead::new(
            source(PartitionId::new("conv-b"), 1),
            JournalPosition::new(20),
        ),
    );
    let crossed_error = authority(Arc::clone(&crossed))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit("q-crossed-source"),
            request("SELECT * FROM turns"),
        )
        .await
        .expect_err("source responses bind the requested partition");
    assert!(matches!(
        crossed_error,
        CoreResolutionError::SourceMismatch(_)
    ));
    assert_eq!(FakeState::lock(&crossed.counts).resolutions, 0);
}

#[tokio::test]
async fn missing_superseded_outage_and_audit_refusal_fail_closed() {
    for head in [
        ProjectionHead::Absent,
        ProjectionHead::Superseded {
            generation: ProjectionGeneration::new(1),
            source: Box::new(source(PartitionId::new("conv-a"), 9)),
        },
    ] {
        let state = Arc::new(FakeState::new(&["conv-a"]));
        state.set_resolution(
            "conv-a",
            ProjectionResolution::new(head, None, ProjectionGeneration::new(1)),
        );
        let compiler = CatalogCompiler::new(SessionContext::new().state());
        let error = authority(Arc::clone(&state))
            .plan(
                &compiler,
                &QueryLimits::default(),
                &QueryScope::Conversations(vec!["a".to_owned()]),
                false,
                audit("q-head"),
                request("SELECT * FROM turns"),
            )
            .await
            .expect_err("missing current projection fails closed");
        assert!(matches!(
            error,
            CoreResolutionError::MissingProjection(_) | CoreResolutionError::Superseded(_)
        ));
    }

    let outage = Arc::new(FakeState::new(&["conv-a"]));
    outage.set_outage();
    let compiler = CatalogCompiler::new(SessionContext::new().state());
    assert!(
        authority(Arc::clone(&outage))
            .plan(
                &compiler,
                &QueryLimits::default(),
                &QueryScope::Conversations(vec!["a".to_owned()]),
                false,
                audit("q-outage"),
                request("SELECT * FROM turns"),
            )
            .await
            .is_err()
    );

    let refused = Arc::new(FakeState::new(&["conv-a"]));
    refused.refuse_audit();
    assert!(
        authority(Arc::clone(&refused))
            .plan(
                &compiler,
                &QueryLimits::default(),
                &QueryScope::Conversations(vec!["a".to_owned()]),
                false,
                audit("q-refused"),
                request("SELECT * FROM turns"),
            )
            .await
            .is_err()
    );
    assert_eq!(compiler.physical_scan_count(), 0);
}

/// A crossed permit addresses another query's audit trail.
///
/// State granted a permit whose query identity is not the one planning asked
/// for. Completing it would write this query's terminal onto that other trail.
/// Planning must refuse and settle nothing, leaving both intents for State's
/// reconciler.
#[tokio::test]
async fn a_crossed_permit_settles_nothing_and_refuses() {
    let state = Arc::new(FakeState::new(&["conv-a"]));
    state.cross_permit();
    let compiler = CatalogCompiler::new(SessionContext::new().state());

    let error = authority(Arc::clone(&state))
        .plan(
            &compiler,
            &QueryLimits::default(),
            &QueryScope::Conversations(vec!["a".to_owned()]),
            false,
            audit("q-crossed-permit"),
            request("SELECT text FROM messages"),
        )
        .await
        .expect_err("a crossed permit is refused");

    assert!(matches!(error, CoreResolutionError::CrossedPermit));
    // The guardian actor is spawned before the crossing is known, so give it
    // the chance it would need to settle. It must still settle nothing.
    tokio::time::sleep(Duration::from_millis(50)).await;
    assert_eq!(
        state.settlements(),
        0,
        "an abandoned guardian never presents a completion for another trail"
    );
}