use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
use std::time::Duration;
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use async_trait::async_trait;
use datafusion::catalog::{
CatalogProvider, CatalogProviderList, MemoryCatalogProvider, MemoryCatalogProviderList,
MemorySchemaProvider, Session, TableProvider,
};
use datafusion::common::Column;
use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion};
use datafusion::datasource::TableType;
use datafusion::error::DataFusionError;
use datafusion::execution::context::{SessionContext, SessionState};
use datafusion::execution::session_state::SessionStateBuilder;
use datafusion::logical_expr::expr::ScalarFunction;
use datafusion::logical_expr::{Expr, LogicalPlan, LogicalPlanBuilder};
use datafusion::physical_plan::ExecutionPlan;
use datafusion::scalar::ScalarValue;
use polyc_projection::family::{
ADMIN_MODEL_CHANGES, CREDENTIAL_LIFECYCLE_CREDENTIALS, CREDENTIAL_LIFECYCLE_KEYS,
CREDENTIAL_LIFECYCLE_REFUSALS, DELEGATION_HANDOFF_SIGNERS, DELEGATION_HANDOFFS,
EXECUTION_MODEL_CALL, EXECUTION_SUMMARY, EXECUTION_TOOL_CALLS, EXECUTION_TURN_DISPATCH,
EXECUTION_TURN_FAILED, EXECUTION_USAGE, FINANCIAL_OUTBOUND_PAYMENTS, FINANCIAL_PAYMENTS,
FINANCIAL_REFUSALS, FINANCIAL_SETTLEMENTS, FINANCIAL_WALLET_LINK_LIFECYCLE, FamilyEntry,
LogicalField, LogicalType, MEMORY_CORROBORATIONS, MEMORY_EXTRACTIONS, MEMORY_FACTS,
MEMORY_FENCES, MEMORY_INVALIDATIONS, MEMORY_LISTS, MEMORY_PORTABLE_FACTS,
MEMORY_PORTABLE_INVALIDATIONS, MEMORY_PORTABLE_LISTS, MEMORY_PROFILE_REWRITES,
MEMORY_PROVENANCE_IDENTITY, MEMORY_SUMMARIES, MEMORY_UNKNOWN, PERSONA_DIRECTORY_IDENTITIES,
PERSONA_DIRECTORY_IDENTITIES_CURRENT, PERSONA_DIRECTORY_PARTICIPATIONS,
PERSONA_DIRECTORY_PARTICIPATIONS_CURRENT, PERSONA_DIRECTORY_PROFILES,
PERSONA_DIRECTORY_PROFILES_CURRENT, PERSONA_DIRECTORY_REFUSALS, PERSONA_DIRECTORY_VISIBILITY,
PERSONA_MEMORY, QUERY_AUDIT_COMPLETIONS, QUERY_AUDIT_INTENTS, QUERY_AUDIT_SOURCE_PINS,
ROUTINE_FIRES, ROUTINE_LIFECYCLE_EVENTS, ROUTINE_SETUP_COMPLETIONS, SECURITY_APPROVAL_DETAILS,
SECURITY_APPROVALS, SECURITY_ATTRIBUTION, SECURITY_ATTRIBUTION_PROVENANCE,
SECURITY_GRANT_REPLAY_SIGNERS, SECURITY_GRANT_REPLAYS, SECURITY_ROUTINE_GRANTS,
TRACE_APPROVALS, TRACE_BOUNDARIES, TRACE_FAILURES, TRACE_HANDOFFS, TRACE_LISTS, TRACE_MESSAGES,
TRACE_PAYMENT_ATTEMPTS, TRACE_PAYMENT_RECEIPTS, TRACE_PAYMENT_REFUSALS, TRACE_QUESTIONS,
TRACE_RECORDS, TRACE_SIGNERS, TRACE_STEPS, TRACE_SUBAGENTS, TRACE_TOOL_CALLS,
TRACE_TOOL_RESULTS, TRACE_TURNS, TRACE_WALLET_LINKS, TRACE_WARNINGS, TableId, TableSchema,
VisibleAudience, administrator_audit, conversation_core, conversation_delegation,
conversation_execution, conversation_financial, conversation_security, conversation_trace,
credential_lifecycle, observed_routines, persona_directory, persona_memory, query_audit,
routine_lifecycle,
};
use polyc_state::command::CommandEnvelope;
use polyc_state::context::CallContext;
use polyc_state::deadline::{Clock, ProductionClock};
use polyc_state::digest::ContentDigest;
use polyc_state::error::{BoundKind, StateError};
use polyc_state::id::{Audience, NamespaceId, OperationFamily, OwnerId, PartitionId, Purpose};
use polyc_state::immutable::Classification;
use polyc_state::journal::{
GetJournalSource, JournalAnchor, JournalDirectoryPage, JournalDirectorySnapshot,
JournalSourceHead, ListJournalDirectorySnapshot, MAX_DIRECTORY_PAGE_PARTITIONS,
ReleaseJournalDirectorySnapshot,
};
use polyc_state::projection::{
FamilyId, ProjectionCatalogError, ProjectionHead, ProjectionKey, ProjectionManifest,
ProjectionResolution, ResolveManifest,
};
use polyc_state::query_audit::{
BeginOutcome, BeginQueryAudit, CompleteQueryAudit, ExecutionPermit, MAX_SOURCE_PINS,
ProjectionPin, QueryAuditError, QueryCompletion, QueryId, RequesterId, SourcePin,
SourceSnapshot,
};
use polyc_state::receipt::Receipt;
use polyc_state::revision::JournalPosition;
use polyc_state_connect::query_audit::{RemoteCompleteQueryAudit, RemoteExecutionPermit};
use polyc_state_connect::wire::DeclaredCall;
use crate::core_execution::PermitGuardian;
use crate::limits::QueryLimits;
use crate::statement_gate::{AllowedStatement, StatementRejected, check_statement_allowed};
use polyc_query_credential::session::QueryScope;
const SHAPE_DOMAIN: &[u8] = b"polychrome.query.conversation-core-shape.v1\0";
const BOUNDS_DOMAIN: &[u8] = b"polychrome.query.conversation-core-bounds.v1\0";
const CORE_PARTITION_PREFIX: &str = polyc_projection::family::CONVERSATION_PARTITION_PREFIX;
const DEFAULT_CORE_RESULT_RELEASE_BYTES: u64 = 32 * 1024 * 1024;
const DEFAULT_CORE_RESPONSE_FRAME_BYTES: u64 = 256 * 1024;
const DEFAULT_CORE_ARTIFACT_FILE_BYTES: u64 = 256 * 1024 * 1024;
const DEFAULT_CORE_ARTIFACT_RANGE_BYTES: u64 = 4 * 1024 * 1024;
const DEFAULT_CORE_SOURCE_DECODE_BYTES: u64 = 512 * 1024 * 1024;
pub(crate) const PROJECTED_FAMILIES: &[FamilyEntry] = polyc_projection::family::ALL_FAMILIES;
fn sql_servable_families() -> impl Iterator<Item = FamilyEntry> {
PROJECTED_FAMILIES
.iter()
.copied()
.filter(|family| family.family_str() != polyc_projection::family::SEARCH_INDEX)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) enum CoreTable {
Turns,
Messages,
Usage,
ModelCall,
ToolCalls,
TurnFailed,
Summary,
Handoffs,
HandoffSigners,
Approvals,
ApprovalDetails,
Attribution,
AttributionProvenance,
TraceTurns,
TraceSteps,
TraceBoundaries,
TraceMessages,
TraceToolCalls,
TraceToolResults,
TraceApprovals,
TraceQuestions,
TraceSubagents,
TraceHandoffs,
TracePaymentAttempts,
TracePaymentReceipts,
TracePaymentRefusals,
TraceWalletLinks,
TraceRecords,
TraceFailures,
TraceWarnings,
TraceLists,
TraceSigners,
CredentialLifecycle,
CredentialKeyLifecycle,
CredentialLifecycleRefusals,
AdminModelChanges,
QueryAuditIntents,
QueryAuditCompletions,
QueryAuditSourcePins,
MemoryFacts,
MemoryPortableFacts,
MemoryInvalidations,
MemoryPortableInvalidations,
MemoryLists,
MemoryPortableLists,
MemoryCorroborations,
MemoryExtractions,
MemoryProfileRewrites,
MemorySummaries,
MemoryProvenanceIdentity,
MemoryFences,
MemoryUnknown,
FinancialPayments,
FinancialOutboundPayments,
FinancialRefusals,
FinancialWalletLinkLifecycle,
FinancialSettlements,
RoutineLifecycle,
RoutineSetup,
RoutineFires,
PersonaProfiles,
PersonaIdentities,
PersonaParticipations,
PersonaVisibility,
PersonaRefusals,
PersonaProfilesCurrent,
PersonaIdentitiesCurrent,
PersonaParticipationsCurrent,
ObservedRoutines,
SecurityGrantReplays,
SecurityGrantReplaySigners,
SecurityRoutineGrants,
TurnDispatch,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) enum CoreRealm {
Visible,
Fleet,
}
impl CoreRealm {
pub(crate) const fn from_scope(scope: &QueryScope) -> Self {
match scope {
QueryScope::Fleet => Self::Fleet,
QueryScope::Conversations { .. } => Self::Visible,
}
}
}
impl CoreTable {
pub(crate) const ALL: [Self; 73] = [
Self::Turns,
Self::Messages,
Self::Usage,
Self::ModelCall,
Self::ToolCalls,
Self::TurnFailed,
Self::Summary,
Self::Handoffs,
Self::HandoffSigners,
Self::Approvals,
Self::ApprovalDetails,
Self::Attribution,
Self::AttributionProvenance,
Self::TraceTurns,
Self::TraceSteps,
Self::TraceBoundaries,
Self::TraceMessages,
Self::TraceToolCalls,
Self::TraceToolResults,
Self::TraceApprovals,
Self::TraceQuestions,
Self::TraceSubagents,
Self::TraceHandoffs,
Self::TracePaymentAttempts,
Self::TracePaymentReceipts,
Self::TracePaymentRefusals,
Self::TraceWalletLinks,
Self::TraceRecords,
Self::TraceFailures,
Self::TraceWarnings,
Self::TraceLists,
Self::TraceSigners,
Self::CredentialLifecycle,
Self::CredentialKeyLifecycle,
Self::CredentialLifecycleRefusals,
Self::AdminModelChanges,
Self::QueryAuditIntents,
Self::QueryAuditCompletions,
Self::QueryAuditSourcePins,
Self::MemoryFacts,
Self::MemoryPortableFacts,
Self::MemoryInvalidations,
Self::MemoryPortableInvalidations,
Self::MemoryLists,
Self::MemoryPortableLists,
Self::MemoryCorroborations,
Self::MemoryExtractions,
Self::MemoryProfileRewrites,
Self::MemorySummaries,
Self::MemoryProvenanceIdentity,
Self::MemoryFences,
Self::MemoryUnknown,
Self::FinancialPayments,
Self::FinancialOutboundPayments,
Self::FinancialRefusals,
Self::FinancialWalletLinkLifecycle,
Self::FinancialSettlements,
Self::RoutineLifecycle,
Self::RoutineSetup,
Self::RoutineFires,
Self::PersonaProfiles,
Self::PersonaIdentities,
Self::PersonaParticipations,
Self::PersonaVisibility,
Self::PersonaRefusals,
Self::PersonaProfilesCurrent,
Self::PersonaIdentitiesCurrent,
Self::PersonaParticipationsCurrent,
Self::ObservedRoutines,
Self::SecurityGrantReplays,
Self::SecurityGrantReplaySigners,
Self::SecurityRoutineGrants,
Self::TurnDispatch,
];
pub(crate) fn from_name(name: &str) -> Result<Self, CoreResolutionError> {
match name {
"turns" => Ok(Self::Turns),
"messages" => Ok(Self::Messages),
"usage" => Ok(Self::Usage),
"model_call" => Ok(Self::ModelCall),
"tool_calls" => Ok(Self::ToolCalls),
"turn_failed" => Ok(Self::TurnFailed),
"summary" => Ok(Self::Summary),
"handoffs" => Ok(Self::Handoffs),
"handoff_signers" => Ok(Self::HandoffSigners),
"approvals" => Ok(Self::Approvals),
"approval_details" => Ok(Self::ApprovalDetails),
"attribution" => Ok(Self::Attribution),
"attribution_provenance" => Ok(Self::AttributionProvenance),
"trace_turns" => Ok(Self::TraceTurns),
"trace_steps" => Ok(Self::TraceSteps),
"trace_boundaries" => Ok(Self::TraceBoundaries),
"trace_messages" => Ok(Self::TraceMessages),
"trace_tool_calls" => Ok(Self::TraceToolCalls),
"trace_tool_results" => Ok(Self::TraceToolResults),
"trace_approvals" => Ok(Self::TraceApprovals),
"trace_questions" => Ok(Self::TraceQuestions),
"trace_subagents" => Ok(Self::TraceSubagents),
"trace_handoffs" => Ok(Self::TraceHandoffs),
"trace_payment_attempts" => Ok(Self::TracePaymentAttempts),
"trace_payment_receipts" => Ok(Self::TracePaymentReceipts),
"trace_payment_refusals" => Ok(Self::TracePaymentRefusals),
"trace_wallet_links" => Ok(Self::TraceWalletLinks),
"trace_records" => Ok(Self::TraceRecords),
"trace_failures" => Ok(Self::TraceFailures),
"trace_warnings" => Ok(Self::TraceWarnings),
"trace_lists" => Ok(Self::TraceLists),
"trace_signers" => Ok(Self::TraceSigners),
"credential_lifecycle" => Ok(Self::CredentialLifecycle),
"credential_key_lifecycle" => Ok(Self::CredentialKeyLifecycle),
"credential_lifecycle_refusals" => Ok(Self::CredentialLifecycleRefusals),
"admin_model_changes" => Ok(Self::AdminModelChanges),
"query_audit_intents" => Ok(Self::QueryAuditIntents),
"query_audit_completions" => Ok(Self::QueryAuditCompletions),
"query_audit_source_pins" => Ok(Self::QueryAuditSourcePins),
"memory_facts" => Ok(Self::MemoryFacts),
"memory_portable_facts" => Ok(Self::MemoryPortableFacts),
"memory_invalidations" => Ok(Self::MemoryInvalidations),
"memory_portable_invalidations" => Ok(Self::MemoryPortableInvalidations),
"memory_lists" => Ok(Self::MemoryLists),
"memory_portable_lists" => Ok(Self::MemoryPortableLists),
"memory_corroborations" => Ok(Self::MemoryCorroborations),
"memory_extractions" => Ok(Self::MemoryExtractions),
"memory_profile_rewrites" => Ok(Self::MemoryProfileRewrites),
"memory_summaries" => Ok(Self::MemorySummaries),
"memory_provenance_identity" => Ok(Self::MemoryProvenanceIdentity),
"memory_fences" => Ok(Self::MemoryFences),
"memory_unknown" => Ok(Self::MemoryUnknown),
"payments" => Ok(Self::FinancialPayments),
"outbound_payments" => Ok(Self::FinancialOutboundPayments),
"refusals" => Ok(Self::FinancialRefusals),
"wallet_link_lifecycle" => Ok(Self::FinancialWalletLinkLifecycle),
"settlements" => Ok(Self::FinancialSettlements),
"routine_lifecycle" => Ok(Self::RoutineLifecycle),
"routine_setup" => Ok(Self::RoutineSetup),
"fires" => Ok(Self::RoutineFires),
"persona_profiles" => Ok(Self::PersonaProfiles),
"persona_identities" => Ok(Self::PersonaIdentities),
"persona_participations" => Ok(Self::PersonaParticipations),
"persona_visibility" => Ok(Self::PersonaVisibility),
"persona_refusals" => Ok(Self::PersonaRefusals),
"persona_profiles_current" => Ok(Self::PersonaProfilesCurrent),
"persona_identities_current" => Ok(Self::PersonaIdentitiesCurrent),
"persona_participations_current" => Ok(Self::PersonaParticipationsCurrent),
"observed_routines" => Ok(Self::ObservedRoutines),
"grant_replays" => Ok(Self::SecurityGrantReplays),
"grant_replay_signers" => Ok(Self::SecurityGrantReplaySigners),
"routine_grant_mutations" => Ok(Self::SecurityRoutineGrants),
"turn_dispatch" => Ok(Self::TurnDispatch),
other => Err(CoreResolutionError::UnknownDependency(other.to_owned())),
}
}
pub(crate) const fn name(self) -> &'static str {
match self {
Self::Turns => "turns",
Self::Messages => "messages",
Self::Usage => "usage",
Self::ModelCall => "model_call",
Self::ToolCalls => "tool_calls",
Self::TurnFailed => "turn_failed",
Self::Summary => "summary",
Self::Handoffs => "handoffs",
Self::HandoffSigners => "handoff_signers",
Self::Approvals => "approvals",
Self::ApprovalDetails => "approval_details",
Self::Attribution => "attribution",
Self::AttributionProvenance => "attribution_provenance",
Self::TraceTurns => "trace_turns",
Self::TraceSteps => "trace_steps",
Self::TraceBoundaries => "trace_boundaries",
Self::TraceMessages => "trace_messages",
Self::TraceToolCalls => "trace_tool_calls",
Self::TraceToolResults => "trace_tool_results",
Self::TraceApprovals => "trace_approvals",
Self::TraceQuestions => "trace_questions",
Self::TraceSubagents => "trace_subagents",
Self::TraceHandoffs => "trace_handoffs",
Self::TracePaymentAttempts => "trace_payment_attempts",
Self::TracePaymentReceipts => "trace_payment_receipts",
Self::TracePaymentRefusals => "trace_payment_refusals",
Self::TraceWalletLinks => "trace_wallet_links",
Self::TraceRecords => "trace_records",
Self::TraceFailures => "trace_failures",
Self::TraceWarnings => "trace_warnings",
Self::TraceLists => "trace_lists",
Self::TraceSigners => "trace_signers",
Self::CredentialLifecycle => "credential_lifecycle",
Self::CredentialKeyLifecycle => "credential_key_lifecycle",
Self::CredentialLifecycleRefusals => "credential_lifecycle_refusals",
Self::AdminModelChanges => "admin_model_changes",
Self::QueryAuditIntents => "query_audit_intents",
Self::QueryAuditCompletions => "query_audit_completions",
Self::QueryAuditSourcePins => "query_audit_source_pins",
Self::MemoryFacts => "memory_facts",
Self::MemoryPortableFacts => "memory_portable_facts",
Self::MemoryInvalidations => "memory_invalidations",
Self::MemoryPortableInvalidations => "memory_portable_invalidations",
Self::MemoryLists => "memory_lists",
Self::MemoryPortableLists => "memory_portable_lists",
Self::MemoryCorroborations => "memory_corroborations",
Self::MemoryExtractions => "memory_extractions",
Self::MemoryProfileRewrites => "memory_profile_rewrites",
Self::MemorySummaries => "memory_summaries",
Self::MemoryProvenanceIdentity => "memory_provenance_identity",
Self::MemoryFences => "memory_fences",
Self::MemoryUnknown => "memory_unknown",
Self::FinancialPayments => "payments",
Self::FinancialOutboundPayments => "outbound_payments",
Self::FinancialRefusals => "refusals",
Self::FinancialWalletLinkLifecycle => "wallet_link_lifecycle",
Self::FinancialSettlements => "settlements",
Self::RoutineLifecycle => "routine_lifecycle",
Self::RoutineSetup => "routine_setup",
Self::RoutineFires => "fires",
Self::PersonaProfiles => "persona_profiles",
Self::PersonaIdentities => "persona_identities",
Self::PersonaParticipations => "persona_participations",
Self::PersonaVisibility => "persona_visibility",
Self::PersonaRefusals => "persona_refusals",
Self::PersonaProfilesCurrent => "persona_profiles_current",
Self::PersonaIdentitiesCurrent => "persona_identities_current",
Self::PersonaParticipationsCurrent => "persona_participations_current",
Self::ObservedRoutines => "observed_routines",
Self::SecurityGrantReplays => "grant_replays",
Self::SecurityGrantReplaySigners => "grant_replay_signers",
Self::SecurityRoutineGrants => "routine_grant_mutations",
Self::TurnDispatch => "turn_dispatch",
}
}
pub(crate) const fn family(self) -> FamilyEntry {
match self {
Self::Turns | Self::Messages => conversation_core(),
Self::Usage
| Self::ModelCall
| Self::ToolCalls
| Self::TurnFailed
| Self::Summary
| Self::TurnDispatch => conversation_execution(),
Self::Handoffs | Self::HandoffSigners => conversation_delegation(),
Self::Approvals
| Self::ApprovalDetails
| Self::Attribution
| Self::AttributionProvenance
| Self::SecurityGrantReplays
| Self::SecurityGrantReplaySigners
| Self::SecurityRoutineGrants => conversation_security(),
Self::TraceTurns
| Self::TraceSteps
| Self::TraceBoundaries
| Self::TraceMessages
| Self::TraceToolCalls
| Self::TraceToolResults
| Self::TraceApprovals
| Self::TraceQuestions
| Self::TraceSubagents
| Self::TraceHandoffs
| Self::TracePaymentAttempts
| Self::TracePaymentReceipts
| Self::TracePaymentRefusals
| Self::TraceWalletLinks
| Self::TraceRecords
| Self::TraceFailures
| Self::TraceWarnings
| Self::TraceLists
| Self::TraceSigners => conversation_trace(),
Self::CredentialLifecycle
| Self::CredentialKeyLifecycle
| Self::CredentialLifecycleRefusals => credential_lifecycle(),
Self::AdminModelChanges => administrator_audit(),
Self::QueryAuditIntents | Self::QueryAuditCompletions | Self::QueryAuditSourcePins => {
query_audit()
}
Self::MemoryFacts
| Self::MemoryPortableFacts
| Self::MemoryInvalidations
| Self::MemoryPortableInvalidations
| Self::MemoryLists
| Self::MemoryPortableLists
| Self::MemoryCorroborations
| Self::MemoryExtractions
| Self::MemoryProfileRewrites
| Self::MemorySummaries
| Self::MemoryProvenanceIdentity
| Self::MemoryFences
| Self::MemoryUnknown => persona_memory(),
Self::FinancialPayments
| Self::FinancialOutboundPayments
| Self::FinancialRefusals
| Self::FinancialWalletLinkLifecycle
| Self::FinancialSettlements => conversation_financial(),
Self::RoutineLifecycle | Self::RoutineSetup | Self::RoutineFires => routine_lifecycle(),
Self::PersonaProfiles
| Self::PersonaIdentities
| Self::PersonaParticipations
| Self::PersonaVisibility
| Self::PersonaRefusals
| Self::PersonaProfilesCurrent
| Self::PersonaIdentitiesCurrent
| Self::PersonaParticipationsCurrent => persona_directory(),
Self::ObservedRoutines => observed_routines(),
}
}
pub(crate) const fn table(self) -> TableId {
match self {
Self::Turns => polyc_projection::family::CONVERSATION_TURNS,
Self::Messages => polyc_projection::family::CONVERSATION_MESSAGES,
Self::Usage => EXECUTION_USAGE,
Self::ModelCall => EXECUTION_MODEL_CALL,
Self::ToolCalls => EXECUTION_TOOL_CALLS,
Self::TurnFailed => EXECUTION_TURN_FAILED,
Self::Summary => EXECUTION_SUMMARY,
Self::Handoffs => DELEGATION_HANDOFFS,
Self::HandoffSigners => DELEGATION_HANDOFF_SIGNERS,
Self::Approvals => SECURITY_APPROVALS,
Self::ApprovalDetails => SECURITY_APPROVAL_DETAILS,
Self::Attribution => SECURITY_ATTRIBUTION,
Self::AttributionProvenance => SECURITY_ATTRIBUTION_PROVENANCE,
Self::TraceTurns => TRACE_TURNS,
Self::TraceSteps => TRACE_STEPS,
Self::TraceBoundaries => TRACE_BOUNDARIES,
Self::TraceMessages => TRACE_MESSAGES,
Self::TraceToolCalls => TRACE_TOOL_CALLS,
Self::TraceToolResults => TRACE_TOOL_RESULTS,
Self::TraceApprovals => TRACE_APPROVALS,
Self::TraceQuestions => TRACE_QUESTIONS,
Self::TraceSubagents => TRACE_SUBAGENTS,
Self::TraceHandoffs => TRACE_HANDOFFS,
Self::TracePaymentAttempts => TRACE_PAYMENT_ATTEMPTS,
Self::TracePaymentReceipts => TRACE_PAYMENT_RECEIPTS,
Self::TracePaymentRefusals => TRACE_PAYMENT_REFUSALS,
Self::TraceWalletLinks => TRACE_WALLET_LINKS,
Self::TraceRecords => TRACE_RECORDS,
Self::TraceFailures => TRACE_FAILURES,
Self::TraceWarnings => TRACE_WARNINGS,
Self::TraceLists => TRACE_LISTS,
Self::TraceSigners => TRACE_SIGNERS,
Self::CredentialLifecycle => CREDENTIAL_LIFECYCLE_CREDENTIALS,
Self::CredentialKeyLifecycle => CREDENTIAL_LIFECYCLE_KEYS,
Self::CredentialLifecycleRefusals => CREDENTIAL_LIFECYCLE_REFUSALS,
Self::AdminModelChanges => ADMIN_MODEL_CHANGES,
Self::QueryAuditIntents => QUERY_AUDIT_INTENTS,
Self::QueryAuditCompletions => QUERY_AUDIT_COMPLETIONS,
Self::QueryAuditSourcePins => QUERY_AUDIT_SOURCE_PINS,
Self::MemoryFacts => MEMORY_FACTS,
Self::MemoryPortableFacts => MEMORY_PORTABLE_FACTS,
Self::MemoryInvalidations => MEMORY_INVALIDATIONS,
Self::MemoryPortableInvalidations => MEMORY_PORTABLE_INVALIDATIONS,
Self::MemoryLists => MEMORY_LISTS,
Self::MemoryPortableLists => MEMORY_PORTABLE_LISTS,
Self::MemoryCorroborations => MEMORY_CORROBORATIONS,
Self::MemoryExtractions => MEMORY_EXTRACTIONS,
Self::MemoryProfileRewrites => MEMORY_PROFILE_REWRITES,
Self::MemorySummaries => MEMORY_SUMMARIES,
Self::MemoryProvenanceIdentity => MEMORY_PROVENANCE_IDENTITY,
Self::MemoryFences => MEMORY_FENCES,
Self::MemoryUnknown => MEMORY_UNKNOWN,
Self::FinancialPayments => FINANCIAL_PAYMENTS,
Self::FinancialOutboundPayments => FINANCIAL_OUTBOUND_PAYMENTS,
Self::FinancialRefusals => FINANCIAL_REFUSALS,
Self::FinancialWalletLinkLifecycle => FINANCIAL_WALLET_LINK_LIFECYCLE,
Self::FinancialSettlements => FINANCIAL_SETTLEMENTS,
Self::RoutineLifecycle => ROUTINE_LIFECYCLE_EVENTS,
Self::RoutineSetup => ROUTINE_SETUP_COMPLETIONS,
Self::RoutineFires => ROUTINE_FIRES,
Self::PersonaProfiles => PERSONA_DIRECTORY_PROFILES,
Self::PersonaIdentities => PERSONA_DIRECTORY_IDENTITIES,
Self::PersonaParticipations => PERSONA_DIRECTORY_PARTICIPATIONS,
Self::PersonaVisibility => PERSONA_DIRECTORY_VISIBILITY,
Self::PersonaRefusals => PERSONA_DIRECTORY_REFUSALS,
Self::PersonaProfilesCurrent => PERSONA_DIRECTORY_PROFILES_CURRENT,
Self::PersonaIdentitiesCurrent => PERSONA_DIRECTORY_IDENTITIES_CURRENT,
Self::PersonaParticipationsCurrent => PERSONA_DIRECTORY_PARTICIPATIONS_CURRENT,
Self::ObservedRoutines => polyc_projection::family::OBSERVED_ROUTINES_TABLE,
Self::SecurityGrantReplays => SECURITY_GRANT_REPLAYS,
Self::SecurityGrantReplaySigners => SECURITY_GRANT_REPLAY_SIGNERS,
Self::SecurityRoutineGrants => SECURITY_ROUTINE_GRANTS,
Self::TurnDispatch => EXECUTION_TURN_DISPATCH,
}
}
pub(crate) fn physical_schema(self) -> &'static TableSchema {
self.family()
.table(self.table())
.expect("each closed table handle belongs to its family")
}
pub(crate) const fn is_trace(self) -> bool {
matches!(
self,
Self::TraceTurns
| Self::TraceSteps
| Self::TraceBoundaries
| Self::TraceMessages
| Self::TraceToolCalls
| Self::TraceToolResults
| Self::TraceApprovals
| Self::TraceQuestions
| Self::TraceSubagents
| Self::TraceHandoffs
| Self::TracePaymentAttempts
| Self::TracePaymentReceipts
| Self::TracePaymentRefusals
| Self::TraceWalletLinks
| Self::TraceRecords
| Self::TraceFailures
| Self::TraceWarnings
| Self::TraceLists
| Self::TraceSigners
)
}
pub(crate) const fn is_credential_lifecycle(self) -> bool {
matches!(
self,
Self::CredentialLifecycle
| Self::CredentialKeyLifecycle
| Self::CredentialLifecycleRefusals
)
}
pub(crate) const fn is_financial(self) -> bool {
matches!(
self,
Self::FinancialPayments
| Self::FinancialOutboundPayments
| Self::FinancialRefusals
| Self::FinancialWalletLinkLifecycle
| Self::FinancialSettlements
)
}
pub(crate) const fn is_query_audit(self) -> bool {
matches!(
self,
Self::QueryAuditIntents | Self::QueryAuditCompletions | Self::QueryAuditSourcePins
)
}
pub(crate) const fn is_persona_memory(self) -> bool {
matches!(
self,
Self::MemoryFacts
| Self::MemoryPortableFacts
| Self::MemoryInvalidations
| Self::MemoryPortableInvalidations
| Self::MemoryLists
| Self::MemoryPortableLists
| Self::MemoryCorroborations
| Self::MemoryExtractions
| Self::MemoryProfileRewrites
| Self::MemorySummaries
| Self::MemoryProvenanceIdentity
| Self::MemoryFences
| Self::MemoryUnknown
)
}
pub(crate) const fn is_routine_lifecycle(self) -> bool {
matches!(
self,
Self::RoutineLifecycle | Self::RoutineSetup | Self::RoutineFires
)
}
pub(crate) const fn is_persona_directory(self) -> bool {
matches!(
self,
Self::PersonaProfiles
| Self::PersonaIdentities
| Self::PersonaParticipations
| Self::PersonaVisibility
| Self::PersonaRefusals
| Self::PersonaProfilesCurrent
| Self::PersonaIdentitiesCurrent
| Self::PersonaParticipationsCurrent
)
}
pub(crate) const fn is_security_grant_replay(self) -> bool {
matches!(
self,
Self::SecurityGrantReplays | Self::SecurityGrantReplaySigners
)
}
pub(crate) const fn is_security_routine_grants(self) -> bool {
matches!(self, Self::SecurityRoutineGrants)
}
pub(crate) const fn is_observed_routines(self) -> bool {
matches!(self, Self::ObservedRoutines)
}
pub(crate) const fn publishes_physical_columns(self) -> bool {
matches!(self, Self::Turns | Self::Messages)
|| self.is_trace()
|| self.is_credential_lifecycle()
|| self.is_query_audit()
|| self.is_financial()
|| self.is_persona_memory()
|| self.is_routine_lifecycle()
|| self.is_persona_directory()
|| self.is_observed_routines()
|| self.is_security_grant_replay()
|| self.is_security_routine_grants()
}
#[allow(
clippy::too_many_lines,
reason = "one arm per table; the schema is the contract, and the exhaustive \
unreachable-arm enumeration (one name per closed-column family) is what \
pushed this over the line budget, not added logic — splitting it would only \
move the same list and hide which tables restore a null"
)]
pub(crate) fn public_schema(self) -> SchemaRef {
if self.publishes_physical_columns() {
return arrow_schema(self.physical_schema());
}
let fields = match self {
Self::Turns
| Self::Messages
| Self::TraceTurns
| Self::TraceSteps
| Self::TraceBoundaries
| Self::TraceMessages
| Self::TraceToolCalls
| Self::TraceToolResults
| Self::TraceApprovals
| Self::TraceQuestions
| Self::TraceSubagents
| Self::TraceHandoffs
| Self::TracePaymentAttempts
| Self::TracePaymentReceipts
| Self::TracePaymentRefusals
| Self::TraceWalletLinks
| Self::TraceRecords
| Self::TraceFailures
| Self::TraceWarnings
| Self::TraceLists
| Self::TraceSigners
| Self::CredentialLifecycle
| Self::CredentialKeyLifecycle
| Self::CredentialLifecycleRefusals
| Self::QueryAuditIntents
| Self::QueryAuditCompletions
| Self::QueryAuditSourcePins
| Self::MemoryFacts
| Self::MemoryPortableFacts
| Self::MemoryInvalidations
| Self::MemoryPortableInvalidations
| Self::MemoryLists
| Self::MemoryPortableLists
| Self::MemoryCorroborations
| Self::MemoryExtractions
| Self::MemoryProfileRewrites
| Self::MemorySummaries
| Self::MemoryProvenanceIdentity
| Self::MemoryFences
| Self::MemoryUnknown
| Self::FinancialPayments
| Self::FinancialOutboundPayments
| Self::FinancialRefusals
| Self::FinancialWalletLinkLifecycle
| Self::FinancialSettlements
| Self::RoutineLifecycle
| Self::RoutineSetup
| Self::RoutineFires
| Self::PersonaProfiles
| Self::PersonaIdentities
| Self::PersonaParticipations
| Self::PersonaVisibility
| Self::PersonaRefusals
| Self::PersonaProfilesCurrent
| Self::PersonaIdentitiesCurrent
| Self::PersonaParticipationsCurrent
| Self::ObservedRoutines
| Self::SecurityGrantReplays
| Self::SecurityGrantReplaySigners
| Self::SecurityRoutineGrants => unreachable!("returned above"),
Self::AdminModelChanges => return administrator_audit_public_schema(),
Self::Usage => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("turn_id", DataType::Utf8, true),
Field::new("input_tokens", DataType::UInt64, false),
Field::new("output_tokens", DataType::UInt64, false),
],
Self::ModelCall => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("turn_id", DataType::Utf8, true),
Field::new("provider", DataType::Utf8, false),
Field::new("model", DataType::Utf8, false),
Field::new("captured_clock_unix_ms", DataType::UInt64, false),
],
Self::ToolCalls => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("turn_id", DataType::Utf8, true),
Field::new("tool_call_id", DataType::Utf8, false),
Field::new("block_type", DataType::Utf8, false),
Field::new("name", DataType::Utf8, false),
Field::new("arguments", DataType::Utf8, true),
Field::new("result", DataType::Utf8, true),
Field::new("first_party", DataType::Boolean, true),
Field::new("internal_only", DataType::Boolean, false),
Field::new("trust", DataType::Utf8, false),
],
Self::TurnFailed => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("turn_id", DataType::Utf8, true),
Field::new("failure_kind", DataType::Utf8, false),
Field::new("message", DataType::Utf8, false),
],
Self::TurnDispatch => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("turn_id", DataType::Utf8, true),
Field::new("occurrence", DataType::Utf8, false),
Field::new("visibility", DataType::Utf8, false),
Field::new("visibility_source", DataType::Utf8, false),
Field::new("source_turn_id", DataType::Utf8, false),
Field::new("edge_asserted_visibility", DataType::Utf8, false),
],
Self::Approvals
| Self::ApprovalDetails
| Self::Attribution
| Self::AttributionProvenance => return security_public_schema(self),
Self::Summary => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("turn_id", DataType::Utf8, true),
Field::new("text", DataType::Utf8, false),
Field::new("covers_through_position", DataType::UInt64, false),
],
Self::Handoffs => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("turn_id", DataType::Utf8, true),
Field::new("phase", DataType::Utf8, false),
Field::new("child_conversation_id", DataType::Utf8, true),
Field::new("child_agent_id", DataType::Utf8, true),
Field::new("carried_count", DataType::UInt64, true),
Field::new("reason", DataType::Utf8, true),
Field::new("parent_agent_id", DataType::Utf8, true),
Field::new("denial_reason", DataType::Utf8, true),
Field::new("allowed", DataType::Utf8, true),
Field::new("signature_status", DataType::Utf8, false),
],
Self::HandoffSigners => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("signed_by", DataType::FixedSizeBinary(32), false),
Field::new("signer_key_id", DataType::Utf8, false),
],
};
Arc::new(Schema::new(fields))
}
#[allow(
clippy::too_many_lines,
reason = "one arm per table; the SQL is the contract and splitting it hides which tables have a view"
)]
pub(crate) const fn public_view_sql(self) -> Option<&'static str> {
match self {
Self::Turns
| Self::Messages
| Self::TraceTurns
| Self::TraceSteps
| Self::TraceBoundaries
| Self::TraceMessages
| Self::TraceToolCalls
| Self::TraceToolResults
| Self::TraceApprovals
| Self::TraceQuestions
| Self::TraceSubagents
| Self::TraceHandoffs
| Self::TracePaymentAttempts
| Self::TracePaymentReceipts
| Self::TracePaymentRefusals
| Self::TraceWalletLinks
| Self::TraceRecords
| Self::TraceFailures
| Self::TraceWarnings
| Self::TraceLists
| Self::TraceSigners
| Self::CredentialLifecycle
| Self::CredentialKeyLifecycle
| Self::CredentialLifecycleRefusals
| Self::QueryAuditIntents
| Self::QueryAuditCompletions
| Self::QueryAuditSourcePins
| Self::MemoryFacts
| Self::MemoryPortableFacts
| Self::MemoryInvalidations
| Self::MemoryPortableInvalidations
| Self::MemoryLists
| Self::MemoryPortableLists
| Self::MemoryCorroborations
| Self::MemoryExtractions
| Self::MemoryProfileRewrites
| Self::MemorySummaries
| Self::MemoryProvenanceIdentity
| Self::MemoryFences
| Self::MemoryUnknown
| Self::FinancialPayments
| Self::FinancialOutboundPayments
| Self::FinancialRefusals
| Self::FinancialWalletLinkLifecycle
| Self::FinancialSettlements
| Self::RoutineLifecycle
| Self::RoutineSetup
| Self::RoutineFires
| Self::PersonaProfiles
| Self::PersonaIdentities
| Self::PersonaParticipations
| Self::PersonaVisibility
| Self::PersonaRefusals
| Self::PersonaProfilesCurrent
| Self::PersonaIdentitiesCurrent
| Self::PersonaParticipationsCurrent
| Self::ObservedRoutines
| Self::SecurityGrantReplays
| Self::SecurityGrantReplaySigners
| Self::SecurityRoutineGrants => None,
Self::AdminModelChanges => Some(
"SELECT partition, position, signature_status, \
CASE WHEN signature_status = 'malformed' THEN NULL \
ELSE principal END AS principal, \
CASE WHEN signature_status = 'malformed' THEN NULL \
ELSE previous_provider END AS previous_provider, \
CASE WHEN signature_status = 'malformed' THEN NULL \
ELSE previous_model END AS previous_model, \
CASE WHEN signature_status = 'malformed' THEN NULL \
ELSE new_provider END AS new_provider, \
CASE WHEN signature_status = 'malformed' THEN NULL \
ELSE new_model END AS new_model, \
CASE WHEN signature_status = 'malformed' THEN NULL \
ELSE changed_at_ms END AS changed_at_ms \
FROM __projection_admin_model_changes",
),
Self::Approvals => Some(
"SELECT partition, position, turn_id, request_id, tool_name, args_json, \
outcome, \
CASE WHEN outcome = 'unanswered' THEN CAST(NULL AS VARCHAR) \
ELSE NULLIF(response_reason, '') END AS response_reason, \
CASE WHEN outcome = 'unanswered' THEN CAST(NULL AS VARCHAR) \
ELSE NULLIF(signature_status, '') END AS signature_status, \
routine_grant, \
NULLIF(tool_descriptor_hash, '') AS tool_descriptor_hash, \
NULLIF(grant_scope, '') AS grant_scope \
FROM __projection_approvals",
),
Self::ApprovalDetails => Some(
"SELECT partition, position, \
NULLIF(request_reason, '') AS request_reason, \
NULLIF(request_sandbox_mode, '') AS request_sandbox_mode, \
signer_public_key, \
NULLIF(modified_args_json, '') AS modified_args_json, \
approved_for_session, \
NULLIF(caller, '') AS caller, \
NULLIF(approver, '') AS approver, \
NULLIF(response_sandbox_mode, '') AS response_sandbox_mode, \
NULLIF(injected_context, '') AS injected_context \
FROM __projection_approval_details",
),
Self::Attribution => Some(
"SELECT partition, position, NULLIF(turn_id, '') AS turn_id, \
persona_id, role FROM __projection_attribution",
),
Self::AttributionProvenance => Some(
"SELECT partition, position, \
NULLIF(identity_provider, '') AS identity_provider, \
NULLIF(identity_scope, '') AS identity_scope, \
NULLIF(identity_external_id, '') AS identity_external_id, \
NULLIF(identity_display_name, '') AS identity_display_name, \
NULLIF(asserting_edge_id, '') AS asserting_edge_id, \
NULLIF(signer_pk_hex, '') AS signer_pk_hex, \
NULLIF(signature_hex, '') AS signature_hex \
FROM __projection_attribution_provenance",
),
Self::Handoffs => Some(
"SELECT partition, position, NULLIF(turn_id, '') AS turn_id, phase, \
CASE WHEN phase = 'handoff' THEN child_conversation_id \
ELSE CAST(NULL AS VARCHAR) END AS child_conversation_id, \
NULLIF(child_agent_id, '') AS child_agent_id, \
CASE WHEN phase = 'handoff' THEN carried_count \
ELSE CAST(NULL AS BIGINT UNSIGNED) END AS carried_count, \
NULLIF(reason, '') AS reason, \
CASE WHEN phase = 'handoff_denied' THEN parent_agent_id \
ELSE CAST(NULL AS VARCHAR) END AS parent_agent_id, \
CASE WHEN phase = 'handoff_denied' THEN denial_reason \
ELSE CAST(NULL AS VARCHAR) END AS denial_reason, \
CASE WHEN phase = 'handoff_denied' THEN allowed \
ELSE CAST(NULL AS VARCHAR) END AS allowed, \
signature_status FROM __projection_handoffs",
),
Self::HandoffSigners => Some(
"SELECT partition, position, signed_by, signer_key_id \
FROM __projection_handoff_signers",
),
Self::Usage => Some(
"SELECT partition, position, NULLIF(turn_id, '') AS turn_id, \
input_tokens, output_tokens FROM __projection_usage",
),
Self::ModelCall => Some(
"SELECT partition, position, NULLIF(turn_id, '') AS turn_id, provider, model, \
captured_clock_unix_ms FROM __projection_model_call",
),
Self::ToolCalls => Some(
"SELECT partition, position, NULLIF(turn_id, '') AS turn_id, tool_call_id, \
block_type, name, \
CASE WHEN block_type = 'call' THEN arguments ELSE CAST(NULL AS VARCHAR) END \
AS arguments, \
CASE WHEN block_type = 'result' THEN result ELSE CAST(NULL AS VARCHAR) END \
AS result, \
CASE WHEN block_type = 'result' THEN first_party ELSE CAST(NULL AS BOOLEAN) END \
AS first_party, \
internal_only, trust FROM __projection_tool_calls",
),
Self::TurnFailed => Some(
"SELECT partition, position, NULLIF(turn_id, '') AS turn_id, failure_kind, \
message FROM __projection_turn_failed",
),
Self::TurnDispatch => Some(
"SELECT partition, position, NULLIF(turn_id, '') AS turn_id, occurrence, \
visibility, visibility_source, source_turn_id, edge_asserted_visibility \
FROM __projection_turn_dispatch",
),
Self::Summary => Some(
"SELECT partition, position, NULLIF(summary_id, '') AS turn_id, text, \
covers_through_position FROM __projection_summary",
),
}
}
pub(crate) fn physical_name(self) -> String {
format!("__projection_{}", self.name())
}
pub(crate) const fn visible_in(self, realm: CoreRealm) -> bool {
!matches!(
(self, realm),
(
Self::Summary
| Self::HandoffSigners
| Self::ApprovalDetails
| Self::AttributionProvenance
| Self::TraceSigners
| Self::CredentialLifecycle
| Self::CredentialKeyLifecycle
| Self::CredentialLifecycleRefusals
| Self::AdminModelChanges
| Self::QueryAuditIntents
| Self::QueryAuditCompletions
| Self::QueryAuditSourcePins
| Self::MemoryProvenanceIdentity
| Self::MemoryFences
| Self::MemoryUnknown
| Self::FinancialSettlements
| Self::RoutineLifecycle
| Self::RoutineSetup
| Self::RoutineFires
| Self::PersonaProfiles
| Self::PersonaIdentities
| Self::PersonaParticipations
| Self::PersonaVisibility
| Self::PersonaRefusals
| Self::PersonaProfilesCurrent
| Self::PersonaIdentitiesCurrent
| Self::PersonaParticipationsCurrent
| Self::ObservedRoutines
| Self::SecurityGrantReplaySigners,
CoreRealm::Visible
)
)
}
}
fn administrator_audit_public_schema() -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("signature_status", DataType::Utf8, false),
Field::new("principal", DataType::Utf8, true),
Field::new("previous_provider", DataType::Utf8, true),
Field::new("previous_model", DataType::Utf8, true),
Field::new("new_provider", DataType::Utf8, true),
Field::new("new_model", DataType::Utf8, true),
Field::new("changed_at_ms", DataType::UInt64, true),
]))
}
#[allow(
clippy::too_many_lines,
reason = "the exhaustive unreachable-arm enumeration (one name per non-security table) is \
what pushed this over the line budget, not added logic"
)]
fn security_public_schema(table: CoreTable) -> SchemaRef {
let fields = match table {
CoreTable::Approvals => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("turn_id", DataType::Utf8, false),
Field::new("request_id", DataType::Utf8, false),
Field::new("tool_name", DataType::Utf8, false),
Field::new("args_json", DataType::Utf8, false),
Field::new("outcome", DataType::Utf8, false),
Field::new("response_reason", DataType::Utf8, true),
Field::new("signature_status", DataType::Utf8, true),
Field::new("routine_grant", DataType::Boolean, false),
Field::new("tool_descriptor_hash", DataType::Utf8, true),
Field::new("grant_scope", DataType::Utf8, true),
],
CoreTable::ApprovalDetails => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("request_reason", DataType::Utf8, true),
Field::new("request_sandbox_mode", DataType::Utf8, true),
Field::new("signer_public_key", DataType::FixedSizeBinary(32), false),
Field::new("modified_args_json", DataType::Utf8, true),
Field::new("approved_for_session", DataType::Boolean, false),
Field::new("caller", DataType::Utf8, true),
Field::new("approver", DataType::Utf8, true),
Field::new("response_sandbox_mode", DataType::Utf8, true),
Field::new("injected_context", DataType::Utf8, true),
],
CoreTable::Attribution => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("turn_id", DataType::Utf8, true),
Field::new("persona_id", DataType::Utf8, false),
Field::new("role", DataType::Utf8, false),
],
CoreTable::AttributionProvenance => vec![
Field::new("partition", DataType::Utf8, false),
Field::new("position", DataType::UInt64, false),
Field::new("identity_provider", DataType::Utf8, true),
Field::new("identity_scope", DataType::Utf8, true),
Field::new("identity_external_id", DataType::Utf8, true),
Field::new("identity_display_name", DataType::Utf8, true),
Field::new("asserting_edge_id", DataType::Utf8, true),
Field::new("signer_pk_hex", DataType::Utf8, true),
Field::new("signature_hex", DataType::Utf8, true),
],
CoreTable::Turns
| CoreTable::Messages
| CoreTable::Usage
| CoreTable::ModelCall
| CoreTable::ToolCalls
| CoreTable::TurnFailed
| CoreTable::Summary
| CoreTable::Handoffs
| CoreTable::HandoffSigners
| CoreTable::TraceTurns
| CoreTable::TraceSteps
| CoreTable::TraceBoundaries
| CoreTable::TraceMessages
| CoreTable::TraceToolCalls
| CoreTable::TraceToolResults
| CoreTable::TraceApprovals
| CoreTable::TraceQuestions
| CoreTable::TraceSubagents
| CoreTable::TraceHandoffs
| CoreTable::TracePaymentAttempts
| CoreTable::TracePaymentReceipts
| CoreTable::TracePaymentRefusals
| CoreTable::TraceWalletLinks
| CoreTable::TraceRecords
| CoreTable::TraceFailures
| CoreTable::TraceWarnings
| CoreTable::TraceLists
| CoreTable::TraceSigners
| CoreTable::CredentialLifecycle
| CoreTable::CredentialKeyLifecycle
| CoreTable::CredentialLifecycleRefusals
| CoreTable::AdminModelChanges
| CoreTable::QueryAuditIntents
| CoreTable::QueryAuditCompletions
| CoreTable::QueryAuditSourcePins
| CoreTable::MemoryFacts
| CoreTable::MemoryPortableFacts
| CoreTable::MemoryInvalidations
| CoreTable::MemoryPortableInvalidations
| CoreTable::MemoryLists
| CoreTable::MemoryPortableLists
| CoreTable::MemoryCorroborations
| CoreTable::MemoryExtractions
| CoreTable::MemoryProfileRewrites
| CoreTable::TurnDispatch
| CoreTable::MemorySummaries
| CoreTable::MemoryProvenanceIdentity
| CoreTable::MemoryFences
| CoreTable::MemoryUnknown
| CoreTable::FinancialPayments
| CoreTable::FinancialOutboundPayments
| CoreTable::FinancialRefusals
| CoreTable::FinancialWalletLinkLifecycle
| CoreTable::FinancialSettlements
| CoreTable::RoutineLifecycle
| CoreTable::RoutineSetup
| CoreTable::RoutineFires
| CoreTable::PersonaProfiles
| CoreTable::PersonaIdentities
| CoreTable::PersonaParticipations
| CoreTable::PersonaVisibility
| CoreTable::PersonaRefusals
| CoreTable::PersonaProfilesCurrent
| CoreTable::PersonaIdentitiesCurrent
| CoreTable::PersonaParticipationsCurrent
| CoreTable::ObservedRoutines
| CoreTable::SecurityGrantReplays
| CoreTable::SecurityGrantReplaySigners
| CoreTable::SecurityRoutineGrants => {
unreachable!("only the security tables reach this helper")
}
};
Arc::new(Schema::new(fields))
}
pub(crate) struct CompiledCoreQuery {
normalized_plan: String,
dependencies: Vec<CoreTable>,
statement: AllowedStatement,
explain_enabled: bool,
plan: LogicalPlan,
base_state: SessionState,
}
impl fmt::Debug for CompiledCoreQuery {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("CompiledCoreQuery")
.field("dependencies", &self.dependencies)
.field("statement", &self.statement)
.field("explain_enabled", &self.explain_enabled)
.finish_non_exhaustive()
}
}
impl CompiledCoreQuery {
#[cfg(test)]
pub(crate) fn dependencies(&self) -> &[CoreTable] {
&self.dependencies
}
pub(crate) fn into_parts(self) -> CompiledCoreParts {
let Self {
normalized_plan,
dependencies,
statement,
explain_enabled,
plan,
base_state,
} = self;
CompiledCoreParts {
normalized_plan,
dependencies,
statement,
explain_enabled,
plan,
base_state,
}
}
}
pub(crate) struct CompiledCoreParts {
pub(crate) normalized_plan: String,
pub(crate) dependencies: Vec<CoreTable>,
pub(crate) statement: AllowedStatement,
pub(crate) explain_enabled: bool,
pub(crate) plan: LogicalPlan,
pub(crate) base_state: SessionState,
}
#[derive(Debug)]
struct SchemaOnlyTable {
schema: SchemaRef,
scans: Arc<AtomicUsize>,
}
#[async_trait]
impl TableProvider for SchemaOnlyTable {
fn schema(&self) -> SchemaRef {
Arc::clone(&self.schema)
}
fn table_type(&self) -> TableType {
TableType::Base
}
async fn scan(
&self,
_state: &dyn Session,
_projection: Option<&Vec<usize>>,
_filters: &[Expr],
_limit: Option<usize>,
) -> Result<Arc<dyn ExecutionPlan>, DataFusionError> {
self.scans.fetch_add(1, Ordering::SeqCst);
Err(DataFusionError::Plan(
"schema-only tables cannot create a physical scan".to_owned(),
))
}
}
struct ArrowLogicalType(LogicalType);
impl From<ArrowLogicalType> for DataType {
fn from(value: ArrowLogicalType) -> Self {
let ArrowLogicalType(logical) = value;
match logical {
LogicalType::Utf8 => Self::Utf8,
LogicalType::FixedBytes { len } => {
Self::FixedSizeBinary(i32::try_from(len).unwrap_or(i32::MAX))
}
LogicalType::UInt64 => Self::UInt64,
LogicalType::Boolean => Self::Boolean,
}
}
}
pub(crate) fn arrow_schema(table: &TableSchema) -> SchemaRef {
let fields = table
.fields()
.iter()
.map(|field: &LogicalField| {
Field::new(
field.name(),
DataType::from(ArrowLogicalType(field.logical_type())),
field.nullable(),
)
})
.collect::<Vec<_>>();
Arc::new(Schema::new(fields))
}
#[derive(Debug)]
pub(crate) struct CatalogCompiler {
state: SessionState,
scans: Arc<AtomicUsize>,
}
impl CatalogCompiler {
pub(crate) fn new(state: SessionState) -> Self {
Self {
state,
scans: Arc::new(AtomicUsize::new(0)),
}
}
pub(crate) async fn compile(
&self,
sql: &str,
parameters: &[CoreParameter],
allow_explain: bool,
) -> Result<CompiledCoreQuery, CoreResolutionError> {
let statement =
check_statement_allowed(sql, allow_explain).map_err(CoreResolutionError::Statement)?;
let catalog_name = self.state.config_options().catalog.default_catalog.clone();
let schema_name = self.state.config_options().catalog.default_schema.clone();
let catalog_list = Arc::new(MemoryCatalogProviderList::new());
let catalog = Arc::new(MemoryCatalogProvider::new());
catalog.register_schema(&schema_name, Arc::new(MemorySchemaProvider::new()))?;
catalog_list.register_catalog(catalog_name, catalog);
let state = SessionStateBuilder::new_from_existing(self.state.clone())
.with_catalog_list(catalog_list)
.build();
let context = SessionContext::new_with_state(state);
for family in sql_servable_families() {
for table in family.tables() {
context.register_table(
table.table().as_str(),
Arc::new(SchemaOnlyTable {
schema: CoreTable::from_name(table.table().as_str())?.public_schema(),
scans: Arc::clone(&self.scans),
}),
)?;
}
}
let dataframe = context.sql(sql).await?;
let actual = dataframe
.logical_plan()
.get_parameter_names()?
.into_iter()
.collect::<BTreeSet<_>>();
let expected = (1..=parameters.len())
.map(|index| format!("${index}"))
.collect::<BTreeSet<_>>();
if actual != expected {
return Err(CoreResolutionError::ParameterMismatch);
}
let values = parameters
.iter()
.map(BoundCoreParameter)
.map(ScalarValue::from)
.collect::<Vec<_>>();
let dataframe = dataframe.with_param_values(values)?;
let plan = coerce_json_union_outputs(dataframe.logical_plan().clone())?;
let mut dependencies = BTreeSet::new();
plan.apply(|node| {
if let LogicalPlan::TableScan(scan) = node {
dependencies.insert(
CoreTable::from_name(scan.table_name.table())
.map_err(|error| DataFusionError::Plan(error.to_string()))?,
);
}
Ok(TreeNodeRecursion::Continue)
})?;
if dependencies.is_empty() {
return Err(CoreResolutionError::NoSourceDependency);
}
let normalized_plan = plan.display_indent().to_string();
Ok(CompiledCoreQuery {
normalized_plan,
dependencies: dependencies.into_iter().collect(),
statement,
explain_enabled: allow_explain,
plan,
base_state: self.state.clone(),
})
}
#[cfg(test)]
fn physical_scan_count(&self) -> usize {
self.scans.load(Ordering::SeqCst)
}
}
fn coerce_json_union_outputs(plan: LogicalPlan) -> Result<LogicalPlan, DataFusionError> {
let schema = plan.schema().clone();
let mut wraps = Vec::with_capacity(schema.fields().len());
let mut needs_wrap = false;
for index in 0..schema.fields().len() {
let (qualifier, field) = schema.qualified_field(index);
let reference = Expr::Column(Column::new(qualifier.cloned(), field.name()));
if field.data_type() == &*polyc_query_json::JSON_UNION_DATA_TYPE {
needs_wrap = true;
wraps.push(
Expr::ScalarFunction(ScalarFunction::new_udf(
polyc_query_json::udfs::json_union_to_text_udf(),
vec![reference],
))
.alias(field.name().clone()),
);
} else {
wraps.push(reference);
}
}
if !needs_wrap {
return Ok(plan);
}
LogicalPlanBuilder::from(plan).project(wraps)?.build()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CoreConsistency {
Projected,
RequireProjectedThrough(JournalPosition),
}
#[derive(Clone, PartialEq, Eq)]
pub(crate) enum CoreParameter {
Utf8(String),
UInt64(u64),
Boolean(bool),
Null,
}
impl fmt::Debug for CoreParameter {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(match self {
Self::Utf8(_) => "utf8",
Self::UInt64(_) => "uint64",
Self::Boolean(_) => "boolean",
Self::Null => "null",
})
}
}
struct BoundCoreParameter<'a>(&'a CoreParameter);
impl From<BoundCoreParameter<'_>> for ScalarValue {
fn from(value: BoundCoreParameter<'_>) -> Self {
let BoundCoreParameter(parameter) = value;
match parameter {
CoreParameter::Utf8(value) => Self::Utf8(Some(value.clone())),
CoreParameter::UInt64(value) => Self::UInt64(Some(*value)),
CoreParameter::Boolean(value) => Self::Boolean(Some(*value)),
CoreParameter::Null => Self::Null,
}
}
}
pub(crate) struct CoreQueryRequest {
pub(crate) sql: String,
pub(crate) parameters: Vec<CoreParameter>,
pub(crate) consistency: CoreConsistency,
requested_bounds: CoreRequestedBounds,
}
impl fmt::Debug for CoreQueryRequest {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("CoreQueryRequest")
.field("sql_bytes", &self.sql.len())
.field("parameters", &self.parameters.len())
.field("consistency", &self.consistency)
.field("requested_bounds", &self.requested_bounds)
.finish()
}
}
impl CoreQueryRequest {
pub(crate) const fn new(
sql: String,
parameters: Vec<CoreParameter>,
consistency: CoreConsistency,
requested_bounds: CoreRequestedBounds,
) -> Self {
Self {
sql,
parameters,
consistency,
requested_bounds,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct CoreRequestedBounds {
timeout: Duration,
rows: u64,
result_release_bytes: u64,
response_frame_bytes: u64,
}
impl CoreRequestedBounds {
pub(crate) const fn from_requested(
timeout: Duration,
rows: u64,
result_release_bytes: u64,
response_frame_bytes: u64,
) -> Self {
Self {
timeout,
rows,
result_release_bytes,
response_frame_bytes,
}
}
#[cfg(test)]
pub(crate) const fn unbounded() -> Self {
Self {
timeout: Duration::MAX,
rows: u64::MAX,
result_release_bytes: u64::MAX,
response_frame_bytes: u64::MAX,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[allow(
clippy::struct_field_names,
reason = "every bound names its byte unit at the deployment boundary"
)]
pub(crate) struct ProjectedCorePolicy {
result_release_bytes: u64,
response_frame_bytes: u64,
manifest_bytes: u64,
artifact_file_bytes: u64,
artifact_range_bytes: u64,
source_decode_bytes: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[allow(
clippy::struct_field_names,
reason = "every bound names its byte unit at the deployment boundary"
)]
pub(crate) struct ProjectedCorePolicyInput {
pub(crate) result_release_bytes: u64,
pub(crate) response_frame_bytes: u64,
pub(crate) manifest_bytes: u64,
pub(crate) artifact_file_bytes: u64,
pub(crate) artifact_range_bytes: u64,
pub(crate) source_decode_bytes: u64,
}
impl TryFrom<ProjectedCorePolicyInput> for ProjectedCorePolicy {
type Error = CoreResolutionError;
fn try_from(value: ProjectedCorePolicyInput) -> Result<Self, Self::Error> {
let ProjectedCorePolicyInput {
result_release_bytes,
response_frame_bytes,
manifest_bytes,
artifact_file_bytes,
artifact_range_bytes,
source_decode_bytes,
} = value;
let policy = Self {
result_release_bytes,
response_frame_bytes,
manifest_bytes,
artifact_file_bytes,
artifact_range_bytes,
source_decode_bytes,
};
policy.validate()?;
Ok(policy)
}
}
impl ProjectedCorePolicy {
const fn validate(self) -> Result<(), CoreResolutionError> {
if self.result_release_bytes == 0
|| self.response_frame_bytes == 0
|| self.manifest_bytes == 0
|| self.artifact_file_bytes == 0
|| self.artifact_range_bytes == 0
|| self.source_decode_bytes == 0
|| self.response_frame_bytes > self.result_release_bytes
|| self.artifact_range_bytes > self.artifact_file_bytes
{
return Err(CoreResolutionError::InvalidBounds);
}
Ok(())
}
}
impl Default for ProjectedCorePolicy {
fn default() -> Self {
Self {
result_release_bytes: DEFAULT_CORE_RESULT_RELEASE_BYTES,
response_frame_bytes: DEFAULT_CORE_RESPONSE_FRAME_BYTES,
manifest_bytes: polyc_state::projection::artifact::MAX_MANIFEST_BYTES,
artifact_file_bytes: DEFAULT_CORE_ARTIFACT_FILE_BYTES,
artifact_range_bytes: DEFAULT_CORE_ARTIFACT_RANGE_BYTES,
source_decode_bytes: DEFAULT_CORE_SOURCE_DECODE_BYTES,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct EffectiveCoreBounds {
timeout: Duration,
rows: u64,
result_release_bytes: u64,
response_frame_bytes: u64,
manifest_bytes: u64,
artifact_file_bytes: u64,
artifact_range_bytes: u64,
source_decode_bytes: u64,
}
impl EffectiveCoreBounds {
fn mint(
limits: &QueryLimits,
policy: ProjectedCorePolicy,
requested: CoreRequestedBounds,
) -> Result<Self, CoreResolutionError> {
let query_row_ceiling = u64::try_from(limits.row_cap).unwrap_or(u64::MAX);
let effective = Self {
timeout: limits.timeout.min(requested.timeout),
rows: query_row_ceiling.min(requested.rows),
result_release_bytes: policy
.result_release_bytes
.min(requested.result_release_bytes),
response_frame_bytes: policy
.response_frame_bytes
.min(requested.response_frame_bytes),
manifest_bytes: policy.manifest_bytes,
artifact_file_bytes: policy.artifact_file_bytes,
artifact_range_bytes: policy.artifact_range_bytes,
source_decode_bytes: policy.source_decode_bytes,
};
if effective.timeout.is_zero()
|| effective.rows == 0
|| effective.result_release_bytes == 0
|| effective.response_frame_bytes == 0
|| effective.manifest_bytes == 0
|| effective.artifact_file_bytes == 0
|| effective.artifact_range_bytes == 0
|| effective.source_decode_bytes == 0
|| effective.artifact_range_bytes > effective.artifact_file_bytes
|| effective.response_frame_bytes > effective.result_release_bytes
{
return Err(CoreResolutionError::InvalidBounds);
}
Ok(effective)
}
fn canonical_bytes(self) -> Vec<u8> {
let mut bytes = BOUNDS_DOMAIN.to_vec();
bytes.extend_from_slice(
&u64::try_from(self.timeout.as_nanos())
.unwrap_or(u64::MAX)
.to_be_bytes(),
);
for value in [
self.rows,
self.result_release_bytes,
self.response_frame_bytes,
self.manifest_bytes,
self.artifact_file_bytes,
self.artifact_range_bytes,
self.source_decode_bytes,
] {
bytes.extend_from_slice(&value.to_be_bytes());
}
bytes
}
pub(crate) const fn timeout(self) -> Duration {
self.timeout
}
pub(crate) const fn rows(self) -> u64 {
self.rows
}
pub(crate) const fn response_frame_bytes(self) -> u64 {
self.response_frame_bytes
}
pub(crate) const fn result_release_bytes(self) -> u64 {
self.result_release_bytes
}
pub(crate) const fn manifest_bytes(self) -> u64 {
self.manifest_bytes
}
pub(crate) const fn artifact_file_bytes(self) -> u64 {
self.artifact_file_bytes
}
pub(crate) const fn artifact_range_bytes(self) -> u64 {
self.artifact_range_bytes
}
pub(crate) const fn source_decode_bytes(self) -> u64 {
self.source_decode_bytes
}
}
pub(crate) struct CoreAuditContext {
query: QueryId,
requester: RequesterId,
bounds: EffectiveCoreBounds,
operation: CoreOperationContext,
}
impl fmt::Debug for CoreAuditContext {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("CoreAuditContext")
.field("bounds", &self.bounds)
.finish_non_exhaustive()
}
}
impl CoreAuditContext {
pub(crate) fn from_scoped(
query: QueryId,
requester: RequesterId,
declared: &DeclaredCall,
bounds: EffectiveCoreBounds,
) -> Self {
Self {
query,
requester,
bounds,
operation: CoreOperationContext::from_declared(declared, bounds.timeout),
}
}
}
#[derive(Debug)]
pub(crate) struct CoreOperationContext {
context: CallContext,
clock: ProductionClock,
audience: Audience,
}
impl CoreOperationContext {
pub(crate) fn from_declared(declared: &DeclaredCall, timeout_ceiling: Duration) -> Self {
let clock = ProductionClock::new();
let mut clamped = declared.clone();
clamped.budget = clamped.budget.min(timeout_ceiling);
let context = clamped.origin_relative_context().in_frame(clock.now());
Self {
context,
clock,
audience: polyc_state_connect::state_audience(),
}
}
#[cfg(test)]
#[cfg(test)]
pub(crate) fn for_test(timeout: Duration) -> Self {
Self::from_declared(
&DeclaredCall::live(polyc_state_connect::state_audience(), timeout),
timeout,
)
}
pub(crate) fn check(&self) -> Result<(), CoreResolutionError> {
self.context
.check(self.clock.now(), &core_operation_family())
.map_err(CoreResolutionError::from)
}
pub(crate) fn remaining(&self) -> Result<Duration, CoreResolutionError> {
self.check()?;
Ok(self.context.remaining(self.clock.now()))
}
pub(crate) fn declared(&self) -> Result<DeclaredCall, CoreResolutionError> {
self.check()?;
Ok(DeclaredCall::bounded(
self.audience.clone(),
self.context.remaining(self.clock.now()),
))
}
#[cfg(test)]
pub(crate) fn local_context(&self) -> Result<&CallContext, CoreResolutionError> {
self.check()?;
Ok(&self.context)
}
}
#[derive(Debug)]
pub(crate) struct CoreCompletionContext(CoreOperationContext);
impl CoreCompletionContext {
pub(crate) fn server_owned(timeout: Duration) -> Self {
let declared = DeclaredCall::live(polyc_state_connect::state_audience(), timeout);
Self(CoreOperationContext::from_declared(&declared, timeout))
}
pub(crate) fn check(&self) -> Result<(), CoreResolutionError> {
self.0.check()
}
pub(crate) fn remaining(&self) -> Result<Duration, CoreResolutionError> {
self.0.remaining()
}
pub(crate) fn declared(&self) -> Result<DeclaredCall, CoreResolutionError> {
self.0.declared()
}
#[cfg(test)]
pub(crate) fn local_context(&self) -> Result<&CallContext, CoreResolutionError> {
self.0.local_context()
}
}
fn core_operation_family() -> OperationFamily {
OperationFamily::new("query.conversation-core.resolve")
}
#[async_trait]
pub(crate) trait CoreMetadataAuthority: Send + Sync {
async fn create_directory_snapshot(
&self,
operation: &CoreOperationContext,
) -> Result<JournalDirectorySnapshot, CoreResolutionError>;
async fn directory_page(
&self,
operation: &CoreOperationContext,
request: ListJournalDirectorySnapshot,
) -> Result<JournalDirectoryPage, CoreResolutionError>;
async fn release_directory_snapshot(
&self,
operation: &CoreOperationContext,
request: ReleaseJournalDirectorySnapshot,
) -> Result<(), CoreResolutionError>;
async fn source_head(
&self,
operation: &CoreOperationContext,
request: GetJournalSource,
) -> Result<Option<JournalSourceHead>, CoreResolutionError>;
async fn versioned_source_head(
&self,
operation: &CoreOperationContext,
scope: &polyc_state::command::CommandScope,
) -> Result<polyc_state::versioned::VersionedSourceHead, CoreResolutionError>;
async fn persona_memory_source_head(
&self,
operation: &CoreOperationContext,
partition: &polyc_state::persona_memory::journal::MemoryJournalPartition,
) -> Result<polyc_state::persona_memory::journal::PersonaMemorySourceHead, CoreResolutionError>;
async fn persona_memory_directory_page(
&self,
operation: &CoreOperationContext,
after: Option<&str>,
limit: u32,
) -> Result<polyc_state::persona_memory::journal::MemoryLineagePage, CoreResolutionError>;
async fn create_versioned_directory_snapshot(
&self,
operation: &CoreOperationContext,
family: &str,
) -> Result<polyc_state::versioned::VersionedDirectorySnapshot, CoreResolutionError>;
async fn versioned_directory_page(
&self,
operation: &CoreOperationContext,
request: &polyc_state::versioned::ListVersionedDirectorySnapshot,
) -> Result<polyc_state::versioned::VersionedDirectoryPage, CoreResolutionError>;
async fn release_versioned_directory_snapshot(
&self,
operation: &CoreOperationContext,
snapshot: &polyc_state::versioned::VersionedDirectorySnapshotId,
) -> Result<(), CoreResolutionError>;
async fn observed_head(
&self,
operation: &CoreOperationContext,
collection: &polyc_state::observation::CollectionId,
) -> Result<Option<polyc_state::observation::ObservationHead>, CoreResolutionError>;
async fn observed_collections(
&self,
operation: &CoreOperationContext,
kind: polyc_state::observation::CollectionKind,
) -> Result<polyc_state::observation::ObservedCollectionListing, CoreResolutionError>;
async fn resolve_manifest(
&self,
operation: &CoreOperationContext,
request: ResolveManifest,
) -> Result<ProjectionResolution, CoreResolutionError>;
async fn begin_audit(
&self,
operation: &CoreOperationContext,
command: BeginQueryAudit,
) -> Result<BeginOutcome<CoreExecutionPermit>, CoreResolutionError>;
async fn complete_audit(
&self,
operation: &CoreCompletionContext,
command: &CoreCompletionCommand,
) -> Result<Receipt, CoreResolutionError>;
async fn completion_receipt(
&self,
operation: &CoreCompletionContext,
command: &CoreCompletionCommand,
) -> Result<Option<Receipt>, CoreResolutionError>;
}
#[derive(Clone, PartialEq, Eq)]
pub(crate) enum CoreCompletionCommand {
Local(CompleteQueryAudit),
Remote(RemoteCompleteQueryAudit),
}
impl fmt::Debug for CoreCompletionCommand {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(match self {
Self::Local(_) => "local",
Self::Remote(_) => "remote",
})
}
}
impl CoreCompletionCommand {
pub(crate) const fn metadata(&self) -> &polyc_state::command::CommandMetadata {
match self {
Self::Local(command) => command.metadata(),
Self::Remote(command) => command.metadata(),
}
}
}
#[derive(PartialEq, Eq)]
pub(crate) enum CoreExecutionPermit {
Local(ExecutionPermit),
Remote(RemoteExecutionPermit),
}
impl fmt::Debug for CoreExecutionPermit {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(match self {
Self::Local(_) => "local",
Self::Remote(_) => "remote",
})
}
}
impl From<ExecutionPermit> for CoreExecutionPermit {
fn from(value: ExecutionPermit) -> Self {
Self::Local(value)
}
}
impl From<RemoteExecutionPermit> for CoreExecutionPermit {
fn from(value: RemoteExecutionPermit) -> Self {
Self::Remote(value)
}
}
impl CoreExecutionPermit {
const fn query(&self) -> &QueryId {
match self {
Self::Local(permit) => permit.query(),
Self::Remote(permit) => permit.query(),
}
}
const fn namespace(&self) -> &NamespaceId {
match self {
Self::Local(permit) => permit.namespace(),
Self::Remote(permit) => permit.namespace(),
}
}
pub(crate) const fn source(&self) -> &SourceSnapshot {
match self {
Self::Local(permit) => permit.source(),
Self::Remote(permit) => permit.source(),
}
}
pub(crate) fn into_completion(
self,
completion: QueryCompletion,
) -> Result<CoreCompletionCommand, CoreResolutionError> {
match self {
Self::Local(permit) => {
let digest = digest(&permit.completion_canonical_bytes(&completion));
Ok(CoreCompletionCommand::Local(CompleteQueryAudit::new(
permit,
completion,
digest,
audit_envelope(),
)))
}
Self::Remote(permit) => {
let digest = digest(&permit.completion_canonical_bytes(&completion));
Ok(CoreCompletionCommand::Remote(permit.into_completion(
completion,
digest,
audit_envelope(),
)?))
}
}
}
}
pub(crate) struct CorePlanningAuthority {
namespace: NamespaceId,
projection_owner: OwnerId,
policy: ProjectedCorePolicy,
metadata: Arc<dyn CoreMetadataAuthority>,
query_audit_lineage: std::sync::Mutex<Option<polyc_state::revision::PartitionIncarnation>>,
}
impl fmt::Debug for CorePlanningAuthority {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("CorePlanningAuthority")
.field("namespace", &self.namespace)
.field("projection_owner", &self.projection_owner)
.field("policy", &self.policy)
.finish_non_exhaustive()
}
}
impl CorePlanningAuthority {
pub(crate) fn new(
namespace: NamespaceId,
projection_owner: OwnerId,
policy: ProjectedCorePolicy,
metadata: Arc<dyn CoreMetadataAuthority>,
) -> Result<Self, CoreResolutionError> {
if namespace.is_empty() || projection_owner.is_empty() {
return Err(CoreResolutionError::InvalidComposition);
}
policy.validate()?;
Ok(Self {
namespace,
projection_owner,
policy,
metadata,
query_audit_lineage: std::sync::Mutex::new(None),
})
}
pub(crate) fn effective_bounds(
&self,
limits: &QueryLimits,
requested: CoreRequestedBounds,
) -> Result<EffectiveCoreBounds, CoreResolutionError> {
EffectiveCoreBounds::mint(limits, self.policy, requested)
}
pub(crate) async fn plan(
&self,
compiler: &CatalogCompiler,
limits: &QueryLimits,
scope: &QueryScope,
allow_explain: bool,
audit: CoreAuditContext,
request: CoreQueryRequest,
) -> Result<CorePlanOutcome, CoreResolutionError> {
self.plan_inner(
compiler,
limits,
scope,
allow_explain,
audit,
request,
false,
)
.await
}
pub(crate) async fn plan_composite_trace_memory(
&self,
compiler: &CatalogCompiler,
limits: &QueryLimits,
scope: &QueryScope,
allow_explain: bool,
audit: CoreAuditContext,
request: CoreQueryRequest,
) -> Result<CorePlanOutcome, CoreResolutionError> {
self.plan_inner(compiler, limits, scope, allow_explain, audit, request, true)
.await
}
#[allow(
clippy::too_many_arguments,
clippy::too_many_lines,
reason = "the private planner names every authority-bearing input and keeps the complete admission-to-audit transaction in one review boundary"
)]
async fn plan_inner(
&self,
compiler: &CatalogCompiler,
limits: &QueryLimits,
scope: &QueryScope,
allow_explain: bool,
audit: CoreAuditContext,
request: CoreQueryRequest,
allow_composite_trace_memory: bool,
) -> Result<CorePlanOutcome, CoreResolutionError> {
if let CoreConsistency::RequireProjectedThrough(position) = request.consistency {
return Err(CoreResolutionError::FreshnessUnsupported { position });
}
if audit.bounds != self.effective_bounds(limits, request.requested_bounds)? {
return Err(CoreResolutionError::InvalidBounds);
}
audit.operation.check()?;
let logical = compiler
.compile(&request.sql, &request.parameters, allow_explain)
.await?;
let realm = CoreRealm::from_scope(scope);
if logical
.dependencies
.iter()
.any(|dependency| !dependency.visible_in(realm))
{
return Err(CoreResolutionError::TableOutsideRealm);
}
if logical
.dependencies
.iter()
.any(|dependency| !family_readable_in(realm, dependency.family()))
{
return Err(CoreResolutionError::FamilyOutsideRealm);
}
check_persona_memory_admission(
&logical.dependencies,
scope,
realm,
allow_composite_trace_memory,
)?;
let partitions = self.authorized_partitions(scope, &audit.operation).await?;
let family_count = projected_families(&logical.dependencies).len();
validate_source_pin_count(partitions.len(), family_count, false)?;
validate_memory_source_pin_count(scope)?;
let sources = self.resolve_sources(&partitions, &audit.operation).await?;
let manifests = self
.resolve_all(
&logical.dependencies,
&partitions,
&sources,
scope,
&audit.operation,
)
.await?;
let source = hybrid_source_snapshot(&manifests, &sources, false)?;
let shape = shape_digest(
&audit,
&request,
&logical,
&partitions,
realm,
&self.namespace,
&self.projection_owner,
&source,
audit.bounds,
);
let placeholder = ContentDigest::from_bytes([0; ContentDigest::LEN]);
let envelope = audit_envelope();
let draft = BeginQueryAudit::new(
audit.query.clone(),
self.namespace.clone(),
audit.requester.clone(),
shape,
source.clone(),
placeholder,
envelope.clone(),
);
let expected_query = audit.query.clone();
let command = BeginQueryAudit::new(
audit.query,
self.namespace.clone(),
audit.requester,
shape,
source.clone(),
digest(&draft.canonical_bytes()),
envelope,
);
audit.operation.check()?;
match self.metadata.begin_audit(&audit.operation, command).await? {
BeginOutcome::Granted(permit) => {
let crossed = permit.query() != &expected_query
|| permit.namespace() != &self.namespace
|| permit.source() != &source;
let guardian = PermitGuardian::new(permit, Arc::clone(&self.metadata));
if crossed {
guardian.abandon();
return Err(CoreResolutionError::CrossedPermit);
}
Ok(CorePlanOutcome::Granted(Box::new(PreparedCoreQuery {
guardian,
compiled: logical,
manifests,
partitions,
scope: scope.clone(),
realm,
metadata: Arc::clone(&self.metadata),
operation: audit.operation,
bounds: audit.bounds,
})))
}
BeginOutcome::AlreadyRecorded(receipt) => Ok(CorePlanOutcome::AlreadyRecorded(receipt)),
}
}
async fn authorized_partitions(
&self,
scope: &QueryScope,
operation: &CoreOperationContext,
) -> Result<Vec<PartitionId>, CoreResolutionError> {
match scope {
QueryScope::Conversations { conversations, .. } => {
if conversations.iter().any(String::is_empty) {
return Err(CoreResolutionError::EmptyConversationIdentity);
}
let mut partitions = conversations
.iter()
.map(|conversation| {
PartitionId::new(format!("{CORE_PARTITION_PREFIX}{conversation}"))
})
.collect::<Vec<_>>();
canonical_partitions(&mut partitions)?;
Ok(partitions)
}
QueryScope::Fleet => self.fleet_partitions(operation).await,
}
}
async fn fleet_partitions(
&self,
operation: &CoreOperationContext,
) -> Result<Vec<PartitionId>, CoreResolutionError> {
operation.check()?;
let snapshot = self.metadata.create_directory_snapshot(operation).await?;
let id = snapshot.id().clone();
let result = self.read_snapshot(&snapshot, operation).await;
let release = self
.metadata
.release_directory_snapshot(operation, ReleaseJournalDirectorySnapshot::new(id.clone()))
.await;
match (result, release) {
(Ok(partitions), Ok(())) => Ok(partitions),
(Err(error), _) | (Ok(_), Err(error)) => Err(error),
}
}
async fn read_snapshot(
&self,
snapshot: &JournalDirectorySnapshot,
operation: &CoreOperationContext,
) -> Result<Vec<PartitionId>, CoreResolutionError> {
let mut partitions = Vec::new();
let mut observed = 0_u64;
let mut after = None;
loop {
let mut request = ListJournalDirectorySnapshot::new(
snapshot.id().clone(),
MAX_DIRECTORY_PAGE_PARTITIONS,
);
if let Some(cursor) = after.take() {
request = request.after(cursor);
}
operation.check()?;
let page = self
.metadata
.directory_page(operation, request.clone())
.await?;
observed = snapshot.validate_page(&request, &page, observed)?;
partitions.extend(
page.partitions()
.iter()
.filter(|partition| is_conversation_partition(partition))
.cloned(),
);
if partitions.len() > MAX_SOURCE_PINS as usize {
return Err(source_bound(partitions.len()));
}
if page.is_truncated() {
after = page.next_after().cloned();
} else {
break;
}
}
canonical_partitions(&mut partitions)?;
Ok(partitions)
}
async fn versioned_namespaces(
&self,
authority: polyc_projection::family::AuthorityFamily,
operation: &CoreOperationContext,
) -> Result<(Vec<String>, polyc_state::revision::PartitionIncarnation), CoreResolutionError>
{
let aggregate = state_authority(authority).aggregate();
let snapshot = self
.metadata
.create_versioned_directory_snapshot(operation, aggregate)
.await?;
let mut found: Vec<String> = Vec::new();
let mut after: Option<String> = None;
let outcome = loop {
if let Err(error) = operation.check() {
break Err(error);
}
let mut request = polyc_state::versioned::ListVersionedDirectorySnapshot::new(
snapshot.id().clone(),
VERSIONED_DIRECTORY_PAGE,
);
if let Some(cursor) = after.clone() {
request = request.after(cursor);
}
let page = match self
.metadata
.versioned_directory_page(operation, &request)
.await
{
Ok(page) => page,
Err(error) => break Err(error),
};
if page.lineage() != snapshot.lineage() {
break Err(CoreResolutionError::Superseded(PartitionId::new(aggregate)));
}
found.extend(page.namespaces().iter().cloned());
if found.len() > MAX_SOURCE_PINS as usize {
break Err(source_bound(found.len()));
}
match page.next_after() {
Some(next) => after = Some(next.to_owned()),
None => break Ok(()),
}
};
let released = self
.metadata
.release_versioned_directory_snapshot(operation, snapshot.id())
.await;
if let Err(release_error) = released {
if let Err(traversal_error) = outcome {
return Err(traversal_error);
}
return Err(release_error);
}
outcome?;
let expected = usize::try_from(snapshot.namespace_count()).unwrap_or(usize::MAX);
if found.len() != expected {
return Err(CoreResolutionError::IncompatibleDescriptor(
PartitionId::new(aggregate),
));
}
Ok((found, snapshot.lineage()))
}
async fn resolve_versioned_family(
&self,
family: FamilyEntry,
operation: &CoreOperationContext,
) -> Result<Vec<ProjectionManifest>, CoreResolutionError> {
let polyc_projection::family::SourceKind::Versioned { family: authority } = family.source()
else {
return Err(CoreResolutionError::IncompatibleDescriptor(
PartitionId::new(family.family_str()),
));
};
let (namespaces, lineage) = self.versioned_namespaces(authority, operation).await?;
let mut manifests = Vec::with_capacity(namespaces.len());
for namespace in &namespaces {
operation.check()?;
let (key, source) = versioned_pin(family.family_str(), authority, namespace, lineage);
let resolution = self
.metadata
.resolve_manifest(
operation,
ResolveManifest::new(
key.clone(),
polyc_state::feed::ProjectionSource::Versioned(source.clone()),
self.projection_owner.clone(),
),
)
.await?;
let Some(manifest) = resolution.current() else {
if resolution.is_superseded() {
return Err(CoreResolutionError::Superseded(key.source().clone()));
}
continue;
};
let versions = family.versions();
if manifest.key() != &key
|| manifest.evidence().source()
!= polyc_state::feed::ProjectionSource::Versioned(source)
|| manifest.schema_version() != versions.schema().get()
|| manifest.fact_version() != versions.fact_model().get()
|| manifest.object_descriptor().owner() != &self.projection_owner
|| manifest.object_descriptor().classification() != Classification::Confidential
{
return Err(CoreResolutionError::IncompatibleDescriptor(
key.source().clone(),
));
}
manifest.validate_structure()?;
manifests.push(manifest.clone());
}
Ok(manifests)
}
async fn resolve_persona_memory_family(
&self,
family: FamilyEntry,
scope: &QueryScope,
operation: &CoreOperationContext,
) -> Result<Vec<ProjectionManifest>, CoreResolutionError> {
let partitions = self
.persona_memory_scope_partitions(scope, operation)
.await?;
let mut manifests = Vec::with_capacity(partitions.len());
let versions = family.versions();
for partition in partitions {
operation.check()?;
let head = self
.metadata
.persona_memory_source_head(operation, &partition)
.await?;
let source = polyc_state::persona_memory::journal::PersonaMemorySource::new(
partition.clone(),
head.incarnation(),
);
let key = ProjectionKey::new(
FamilyId::new(family.family_str()),
source.projection_partition().clone(),
);
let resolution = self
.metadata
.resolve_manifest(
operation,
ResolveManifest::new(
key.clone(),
polyc_state::feed::ProjectionSource::PersonaMemory(source.clone()),
self.projection_owner.clone(),
),
)
.await?;
let Some(manifest) = resolution.current() else {
if resolution.is_superseded() {
return Err(CoreResolutionError::Superseded(key.source().clone()));
}
return Err(CoreResolutionError::MissingProjection(key.source().clone()));
};
if manifest.key() != &key
|| manifest.evidence().source()
!= polyc_state::feed::ProjectionSource::PersonaMemory(source)
|| manifest.schema_version() != versions.schema().get()
|| manifest.fact_version() != versions.fact_model().get()
|| manifest.object_descriptor().owner() != &self.projection_owner
|| manifest.object_descriptor().classification() != Classification::Confidential
{
return Err(CoreResolutionError::IncompatibleDescriptor(
key.source().clone(),
));
}
manifest.validate_structure()?;
manifests.push(manifest.clone());
}
Ok(manifests)
}
async fn persona_memory_scope_partitions(
&self,
scope: &QueryScope,
operation: &CoreOperationContext,
) -> Result<
Vec<polyc_state::persona_memory::journal::MemoryJournalPartition>,
CoreResolutionError,
> {
match scope {
QueryScope::Conversations { memory, .. } => memory
.partitions()
.into_iter()
.map(|persona_id| {
polyc_state::persona_memory::journal::MemoryJournalPartition::parse(format!(
"persona-{persona_id}-mem"
))
.map_err(CoreResolutionError::from)
})
.collect(),
QueryScope::Fleet => {
let mut found = Vec::new();
let mut after: Option<String> = None;
loop {
operation.check()?;
let page = self
.metadata
.persona_memory_directory_page(operation, after.as_deref(), MAX_SOURCE_PINS)
.await?;
found.extend(page.entries().iter().map(|entry| entry.partition().clone()));
if found.len() > MAX_SOURCE_PINS as usize {
return Err(source_bound(found.len()));
}
match page.next() {
Some(next) => after = Some(next.to_owned()),
None => return Ok(found),
}
}
}
}
}
async fn resolve_observed_family(
&self,
family: FamilyEntry,
operation: &CoreOperationContext,
) -> Result<Vec<ProjectionManifest>, CoreResolutionError> {
let polyc_projection::family::SourceKind::Observed {
collection: collection_kind,
} = family.source()
else {
return Err(CoreResolutionError::SourceKindUnsupported);
};
let state_kind = observed_state_collection_kind(collection_kind);
let listing = self
.metadata
.observed_collections(operation, state_kind)
.await?;
if !listing.complete() {
return Err(source_bound(listing.collections().len()));
}
let mut manifests = Vec::with_capacity(listing.collections().len());
let versions = family.versions();
for collection in listing.collections() {
operation.check()?;
let Some(head) = self.metadata.observed_head(operation, collection).await? else {
continue;
};
let key = ProjectionKey::new(
FamilyId::new(family.family_str()),
head.source().projection_partition().clone(),
);
let source = polyc_state::feed::ProjectionSource::Observed(head.source().clone());
let resolution = self
.metadata
.resolve_manifest(
operation,
ResolveManifest::new(
key.clone(),
source.clone(),
self.projection_owner.clone(),
),
)
.await?;
let Some(manifest) = resolution.current() else {
if resolution.is_superseded() {
return Err(CoreResolutionError::Superseded(key.source().clone()));
}
return Err(CoreResolutionError::MissingProjection(key.source().clone()));
};
if manifest.key() != &key
|| manifest.evidence().source() != source
|| manifest.schema_version() != versions.schema().get()
|| manifest.fact_version() != versions.fact_model().get()
|| manifest.object_descriptor().owner() != &self.projection_owner
|| manifest.object_descriptor().classification() != Classification::Confidential
{
return Err(CoreResolutionError::IncompatibleDescriptor(
key.source().clone(),
));
}
manifest.validate_structure()?;
manifests.push(manifest.clone());
}
Ok(manifests)
}
async fn resolve_all(
&self,
dependencies: &[CoreTable],
partitions: &[PartitionId],
sources: &[JournalSourceHead],
scope: &QueryScope,
operation: &CoreOperationContext,
) -> Result<Vec<ProjectionManifest>, CoreResolutionError> {
if partitions.len() != sources.len() {
return Err(CoreResolutionError::SourceVectorMismatch);
}
let families = projected_families(dependencies);
let mut manifests = Vec::with_capacity(partitions.len().saturating_mul(families.len()));
for family in families.values() {
match family.source() {
polyc_projection::family::SourceKind::Versioned { .. } => {
manifests.extend(self.resolve_versioned_family(*family, operation).await?);
continue;
}
polyc_projection::family::SourceKind::JournalFixed { partition } => {
manifests.extend(
self.resolve_fixed_family(*family, partition, operation)
.await?,
);
continue;
}
polyc_projection::family::SourceKind::QueryAudit => {
manifests.extend(self.resolve_query_audit_family(*family, operation).await?);
continue;
}
polyc_projection::family::SourceKind::PersonaMemory => {
manifests.extend(
self.resolve_persona_memory_family(*family, scope, operation)
.await?,
);
continue;
}
polyc_projection::family::SourceKind::JournalDirectory { .. } => {}
polyc_projection::family::SourceKind::Observed { .. } => {
manifests.extend(self.resolve_observed_family(*family, operation).await?);
continue;
}
}
let versions = family.versions();
for (partition, source) in partitions.iter().zip(sources) {
if source.source().partition() != partition {
return Err(CoreResolutionError::SourceMismatch(partition.clone()));
}
let key = ProjectionKey::new(FamilyId::new(family.family_str()), partition.clone());
operation.check()?;
let resolution = self
.metadata
.resolve_manifest(
operation,
ResolveManifest::new(
key.clone(),
polyc_state::feed::ProjectionSource::Journal(source.source().clone()),
self.projection_owner.clone(),
),
)
.await?;
let manifest = resolution.current().ok_or_else(|| {
if resolution.is_superseded() {
CoreResolutionError::Superseded(partition.clone())
} else {
CoreResolutionError::MissingProjection(partition.clone())
}
})?;
if !family.source().is_journal_directory()
|| manifest.key() != &key
|| manifest
.evidence()
.as_journal()
.map(polyc_state::feed::SourceCheckpoint::source)
!= Some(source.source())
|| manifest.schema_version() != versions.schema().get()
|| manifest.fact_version() != versions.fact_model().get()
|| manifest.object_descriptor().owner() != &self.projection_owner
|| manifest.object_descriptor().classification() != Classification::Confidential
{
return Err(CoreResolutionError::IncompatibleDescriptor(
partition.clone(),
));
}
manifest.validate_structure()?;
manifests.push(manifest.clone());
}
}
manifests.sort_by(|left, right| left.key().cmp(right.key()));
if manifests
.windows(2)
.any(|pair| pair[0].key() >= pair[1].key())
{
return Err(CoreResolutionError::DuplicateDescriptor);
}
Ok(manifests)
}
async fn resolve_query_audit_family(
&self,
family: FamilyEntry,
operation: &CoreOperationContext,
) -> Result<Vec<ProjectionManifest>, CoreResolutionError> {
let partition = PartitionId::new(polyc_projection::family::QUERY_AUDIT_SOURCE);
operation.check()?;
let key = ProjectionKey::new(FamilyId::new(family.family_str()), partition.clone());
let mut asked = self.query_audit_lineage_guess();
let resolution = self
.resolve_query_audit_manifest(operation, &key, asked)
.await?;
let resolution = match resolution.head() {
ProjectionHead::Superseded { source, .. } => {
let polyc_state::feed::ProjectionSource::QueryAudit(revealed) = source.as_ref()
else {
return Err(CoreResolutionError::IncompatibleDescriptor(partition));
};
asked = revealed.incarnation();
self.remember_query_audit_lineage(asked);
self.resolve_query_audit_manifest(operation, &key, asked)
.await?
}
_ => resolution,
};
let Some(manifest) = resolution.current() else {
if resolution.is_superseded() {
return Err(CoreResolutionError::Superseded(key.source().clone()));
}
return Err(CoreResolutionError::MissingProjection(partition));
};
let source = manifest.evidence().source();
let versions = family.versions();
let bound_to_asked = matches!(
&source,
polyc_state::feed::ProjectionSource::QueryAudit(observed)
if observed.incarnation() == asked
);
if !bound_to_asked
|| manifest.key() != &key
|| manifest.schema_version() != versions.schema().get()
|| manifest.fact_version() != versions.fact_model().get()
|| manifest.object_descriptor().owner() != &self.projection_owner
|| manifest.object_descriptor().classification() != Classification::Confidential
{
return Err(CoreResolutionError::IncompatibleDescriptor(
key.source().clone(),
));
}
self.remember_query_audit_lineage(asked);
manifest.validate_structure()?;
Ok(vec![manifest.clone()])
}
fn query_audit_lineage_guess(&self) -> polyc_state::revision::PartitionIncarnation {
self.query_audit_lineage
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.unwrap_or(polyc_state::revision::PartitionIncarnation::from_bytes(
[0; polyc_state::revision::PartitionIncarnation::LEN],
))
}
fn remember_query_audit_lineage(&self, lineage: polyc_state::revision::PartitionIncarnation) {
*self
.query_audit_lineage
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(lineage);
}
async fn resolve_query_audit_manifest(
&self,
operation: &CoreOperationContext,
key: &ProjectionKey,
lineage: polyc_state::revision::PartitionIncarnation,
) -> Result<ProjectionResolution, CoreResolutionError> {
let source = polyc_state::feed::ProjectionSource::QueryAudit(
polyc_state::query_audit::AuditSource::new(lineage),
);
self.metadata
.resolve_manifest(
operation,
ResolveManifest::new(key.clone(), source, self.projection_owner.clone()),
)
.await
}
async fn resolve_fixed_family(
&self,
family: FamilyEntry,
partition: &str,
operation: &CoreOperationContext,
) -> Result<Vec<ProjectionManifest>, CoreResolutionError> {
let partition = PartitionId::new(partition.to_owned());
operation.check()?;
let head = self
.metadata
.source_head(operation, GetJournalSource::new(partition.clone()))
.await?
.ok_or_else(|| CoreResolutionError::MissingSource(partition.clone()))?;
if head.source().partition() != &partition {
return Err(CoreResolutionError::SourceMismatch(partition));
}
let key = ProjectionKey::new(FamilyId::new(family.family_str()), partition.clone());
let source = polyc_state::feed::ProjectionSource::Journal(head.source().clone());
let resolution = self
.metadata
.resolve_manifest(
operation,
ResolveManifest::new(key.clone(), source.clone(), self.projection_owner.clone()),
)
.await?;
let Some(manifest) = resolution.current() else {
if resolution.is_superseded() {
return Err(CoreResolutionError::Superseded(key.source().clone()));
}
return Err(CoreResolutionError::MissingProjection(partition));
};
let versions = family.versions();
if manifest.key() != &key
|| manifest.evidence().source() != source
|| manifest.schema_version() != versions.schema().get()
|| manifest.fact_version() != versions.fact_model().get()
|| manifest.object_descriptor().owner() != &self.projection_owner
|| manifest.object_descriptor().classification() != Classification::Confidential
{
return Err(CoreResolutionError::IncompatibleDescriptor(
key.source().clone(),
));
}
manifest.validate_structure()?;
Ok(vec![manifest.clone()])
}
async fn resolve_sources(
&self,
partitions: &[PartitionId],
operation: &CoreOperationContext,
) -> Result<Vec<JournalSourceHead>, CoreResolutionError> {
let mut sources = Vec::with_capacity(partitions.len());
for partition in partitions {
operation.check()?;
let source = self
.metadata
.source_head(operation, GetJournalSource::new(partition.clone()))
.await?
.ok_or_else(|| CoreResolutionError::MissingSource(partition.clone()))?;
if source.source().partition() != partition {
return Err(CoreResolutionError::SourceMismatch(partition.clone()));
}
sources.push(source);
}
Ok(sources)
}
}
fn check_persona_memory_admission(
dependencies: &[CoreTable],
scope: &QueryScope,
realm: CoreRealm,
allow_composite_trace_memory: bool,
) -> Result<(), CoreResolutionError> {
if realm == CoreRealm::Fleet {
return Ok(());
}
let has_memory_table = dependencies
.iter()
.any(|dependency| dependency.family().family_str() == PERSONA_MEMORY);
let has_other_table = dependencies
.iter()
.any(|dependency| dependency.family().family_str() != PERSONA_MEMORY);
if has_memory_table && has_other_table && !allow_composite_trace_memory {
return Err(CoreResolutionError::FamilyOutsideRealm);
}
if !has_memory_table {
return Ok(());
}
let has_owner_table = dependencies.iter().any(|dependency| {
dependency.physical_schema().audience() == VisibleAudience::PartitionOwner
});
if !has_owner_table {
return Ok(());
}
let QueryScope::Conversations { memory, .. } = scope else {
return Err(CoreResolutionError::TableOutsideAudience);
};
if memory.owner.is_none() {
return Err(CoreResolutionError::TableOutsideAudience);
}
Ok(())
}
fn validate_memory_source_pin_count(scope: &QueryScope) -> Result<(), CoreResolutionError> {
let QueryScope::Conversations { memory, .. } = scope else {
return Ok(());
};
let memory_partitions = memory.partitions().len();
if memory_partitions == 0 {
return Ok(());
}
validate_source_pin_count(memory_partitions, 1, false)
}
fn canonical_partitions(partitions: &mut [PartitionId]) -> Result<(), CoreResolutionError> {
if partitions.len() > MAX_SOURCE_PINS as usize {
return Err(source_bound(partitions.len()));
}
partitions.sort();
if partitions.iter().any(PartitionId::is_empty)
|| partitions.windows(2).any(|pair| pair[0] >= pair[1])
{
return Err(CoreResolutionError::DuplicatePartition);
}
Ok(())
}
fn is_conversation_partition(partition: &PartitionId) -> bool {
partition
.as_str()
.strip_prefix(CORE_PARTITION_PREFIX)
.is_some_and(|suffix| !suffix.is_empty())
}
fn source_bound(requested: usize) -> CoreResolutionError {
StateError::BoundsExceeded {
bound: BoundKind::CommandRecords,
limit: u64::from(MAX_SOURCE_PINS),
requested: u64::try_from(requested).unwrap_or(u64::MAX),
}
.into()
}
const fn family_readable_in(realm: CoreRealm, family: FamilyEntry) -> bool {
match realm {
CoreRealm::Fleet => true,
CoreRealm::Visible => {
family.source().is_journal_directory()
|| matches!(
family.source(),
polyc_projection::family::SourceKind::PersonaMemory
)
}
}
}
pub(crate) fn catalog_tables(realm: CoreRealm, scope: &QueryScope) -> Vec<CoreTable> {
all_tables()
.filter(|table| {
sql_servable_families().any(|family| family == table.family())
&& table.visible_in(realm)
&& family_readable_in(realm, table.family())
&& check_persona_memory_admission(&[*table], scope, realm, false).is_ok()
})
.collect()
}
pub(crate) fn all_tables() -> impl Iterator<Item = CoreTable> + 'static {
CoreTable::ALL.into_iter()
}
fn versioned_pin(
projected: &str,
authority: polyc_projection::family::AuthorityFamily,
namespace: &str,
lineage: polyc_state::revision::PartitionIncarnation,
) -> (ProjectionKey, polyc_state::feed::VersionedSource) {
let scope = polyc_state::versioned::authority::scope(
&NamespaceId::new(namespace),
state_authority(authority),
);
let source = polyc_state::feed::VersionedSource::new(scope, lineage);
let key = ProjectionKey::new(
FamilyId::new(projected),
source.projection_partition().clone(),
);
(key, source)
}
const VERSIONED_DIRECTORY_PAGE: u32 = 64;
pub(crate) const fn state_authority(
family: polyc_projection::family::AuthorityFamily,
) -> polyc_state::versioned::authority::AuthorityFamily {
match family {
polyc_projection::family::AuthorityFamily::Credentials => {
polyc_state::versioned::authority::AuthorityFamily::Credentials
}
polyc_projection::family::AuthorityFamily::Persona => {
polyc_state::versioned::authority::AuthorityFamily::Persona
}
}
}
const fn observed_state_collection_kind(
kind: polyc_projection::family::ObservedCollectionKind,
) -> polyc_state::observation::CollectionKind {
match kind {
polyc_projection::family::ObservedCollectionKind::Routines => {
polyc_state::observation::CollectionKind::Routines
}
}
}
fn projected_families(dependencies: &[CoreTable]) -> BTreeMap<&'static str, FamilyEntry> {
dependencies
.iter()
.map(|dependency| dependency.family())
.map(|family| (family.family_str(), family))
.collect()
}
fn hybrid_source_snapshot(
manifests: &[ProjectionManifest],
sources: &[JournalSourceHead],
journal: bool,
) -> Result<SourceSnapshot, CoreResolutionError> {
let mut pins = manifests
.iter()
.cloned()
.map(ProjectionPin::new)
.map(SourcePin::Projected)
.collect::<Vec<_>>();
if journal {
pins.extend(sources.iter().map(|source| {
SourcePin::Journal(JournalAnchor::new(
source.source().clone(),
source.position(),
))
}));
}
SourceSnapshot::try_new(pins).map_err(CoreResolutionError::from)
}
fn validate_source_pin_count(
partitions: usize,
projected_families: usize,
journal: bool,
) -> Result<(), CoreResolutionError> {
let kinds = projected_families.saturating_add(usize::from(journal));
let requested = partitions.saturating_mul(kinds);
if requested > MAX_SOURCE_PINS as usize {
return Err(source_bound(requested));
}
Ok(())
}
fn digest(bytes: &[u8]) -> ContentDigest {
ContentDigest::from_bytes(*blake3::hash(bytes).as_bytes())
}
fn audit_envelope() -> CommandEnvelope {
CommandEnvelope::new(
Purpose::new("conversation-core-query"),
Audience::new("state"),
polyc_state::query_audit::command_bounds(),
)
}
fn push(bytes: &mut Vec<u8>, value: &[u8]) {
bytes.extend_from_slice(&u64::try_from(value.len()).unwrap_or(u64::MAX).to_be_bytes());
bytes.extend_from_slice(value);
}
fn push_parameters(bytes: &mut Vec<u8>, parameters: &[CoreParameter]) {
bytes.extend_from_slice(
&u64::try_from(parameters.len())
.unwrap_or(u64::MAX)
.to_be_bytes(),
);
for parameter in parameters {
match parameter {
CoreParameter::Utf8(value) => {
bytes.push(0);
push(bytes, value.as_bytes());
}
CoreParameter::UInt64(value) => {
bytes.push(1);
bytes.extend_from_slice(&value.to_be_bytes());
}
CoreParameter::Boolean(value) => {
bytes.push(2);
bytes.push(u8::from(*value));
}
CoreParameter::Null => bytes.push(3),
}
}
}
#[allow(clippy::too_many_arguments)]
fn shape_digest(
audit: &CoreAuditContext,
request: &CoreQueryRequest,
compiled: &CompiledCoreQuery,
partitions: &[PartitionId],
realm: CoreRealm,
namespace: &NamespaceId,
owner: &OwnerId,
source: &SourceSnapshot,
bounds: EffectiveCoreBounds,
) -> ContentDigest {
let mut bytes = SHAPE_DOMAIN.to_vec();
push(&mut bytes, compiled.normalized_plan.as_bytes());
bytes.push(match compiled.statement {
AllowedStatement::Query => 0,
AllowedStatement::Explain => 1,
});
bytes.push(u8::from(compiled.explain_enabled));
bytes.extend_from_slice(
&u64::try_from(compiled.dependencies.len())
.unwrap_or(u64::MAX)
.to_be_bytes(),
);
for dependency in &compiled.dependencies {
push(&mut bytes, dependency.name().as_bytes());
}
let families = projected_families(&compiled.dependencies);
bytes.extend_from_slice(
&u64::try_from(families.len())
.unwrap_or(u64::MAX)
.to_be_bytes(),
);
for family in families.values() {
push(&mut bytes, family.family_str().as_bytes());
push(&mut bytes, &family.fingerprint());
}
push(&mut bytes, namespace.as_str().as_bytes());
push(&mut bytes, owner.as_str().as_bytes());
push(&mut bytes, audit.query.as_str().as_bytes());
push(&mut bytes, audit.requester.as_str().as_bytes());
push_parameters(&mut bytes, &request.parameters);
bytes.push(match realm {
CoreRealm::Visible => 0,
CoreRealm::Fleet => 1,
});
bytes.extend_from_slice(
&u64::try_from(partitions.len())
.unwrap_or(u64::MAX)
.to_be_bytes(),
);
for partition in partitions {
push(&mut bytes, partition.as_str().as_bytes());
}
match request.consistency {
CoreConsistency::Projected => bytes.push(0),
CoreConsistency::RequireProjectedThrough(position) => {
bytes.push(1);
bytes.extend_from_slice(&position.get().to_be_bytes());
}
}
push(&mut bytes, &source.canonical_bytes());
push(&mut bytes, &bounds.canonical_bytes());
digest(&bytes)
}
#[derive(Debug)]
pub(crate) enum CorePlanOutcome {
Granted(Box<PreparedCoreQuery>),
#[allow(
dead_code,
reason = "State's deduplicated receipt stays attached to the closed planning outcome even though the transport currently reports only the already-recorded shape"
)]
AlreadyRecorded(Box<Receipt>),
}
pub(crate) struct PreparedCoreQuery {
guardian: PermitGuardian,
compiled: CompiledCoreQuery,
manifests: Vec<ProjectionManifest>,
partitions: Vec<PartitionId>,
scope: QueryScope,
realm: CoreRealm,
metadata: Arc<dyn CoreMetadataAuthority>,
operation: CoreOperationContext,
bounds: EffectiveCoreBounds,
}
impl fmt::Debug for PreparedCoreQuery {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("PreparedCoreQuery")
.field("guardian", &self.guardian)
.field("compiled", &self.compiled)
.field("manifests", &self.manifests.len())
.field("partitions", &self.partitions.len())
.field("realm", &self.realm)
.field("bounds", &self.bounds)
.finish_non_exhaustive()
}
}
impl PreparedCoreQuery {
#[cfg(test)]
pub(crate) fn pause_completion_dispatch(&self) -> crate::core_execution::GuardianDispatchPause {
self.guardian.pause_dispatch()
}
#[cfg(test)]
pub(crate) fn with_expired_operation_for_test(mut self) -> Self {
self.operation = CoreOperationContext::for_test(Duration::ZERO);
self
}
#[cfg(test)]
pub(crate) fn with_fresh_operation_for_test(mut self, budget: Duration) -> Self {
self.operation = CoreOperationContext::for_test(budget);
self
}
pub(crate) fn into_parts(self) -> PreparedCoreParts {
let Self {
guardian,
compiled,
manifests,
partitions,
scope,
realm,
metadata,
operation,
bounds,
} = self;
PreparedCoreParts {
guardian,
compiled,
manifests,
partitions,
scope,
realm,
metadata,
operation,
bounds,
}
}
}
pub(crate) struct PreparedCoreParts {
pub(crate) guardian: PermitGuardian,
pub(crate) compiled: CompiledCoreQuery,
pub(crate) manifests: Vec<ProjectionManifest>,
pub(crate) partitions: Vec<PartitionId>,
pub(crate) scope: QueryScope,
pub(crate) realm: CoreRealm,
pub(crate) metadata: Arc<dyn CoreMetadataAuthority>,
pub(crate) operation: CoreOperationContext,
pub(crate) bounds: EffectiveCoreBounds,
}
#[derive(thiserror::Error)]
pub(crate) enum CoreResolutionError {
#[error("the query statement was refused")]
Statement(StatementRejected),
#[error("schema-only planning failed")]
DataFusion(#[from] DataFusionError),
#[error("query has no declared projected or journal dependency")]
NoSourceDependency,
#[error("the query names a table outside the conversation-core catalog")]
UnknownDependency(String),
#[error("the query names a table outside the authorized physical realm")]
TableOutsideRealm,
#[error("a visible session named a family that only the Fleet realm reads")]
FamilyOutsideRealm,
#[error("a persona-memory owner-audience table was addressed outside its owner's partition")]
TableOutsideAudience,
#[error("the typed parameter vector does not match the SQL placeholders")]
ParameterMismatch,
#[error("core planning composition has an empty namespace or owner")]
InvalidComposition,
#[error("the verified query scope has no durable audit attribution")]
InvalidAttribution,
#[error("the verified conversation identity is empty")]
EmptyConversationIdentity,
#[error("the requested projection freshness is not implemented")]
FreshnessUnsupported { position: JournalPosition },
#[error("projected query bounds are empty, crossed, or exceed their parent posture")]
InvalidBounds,
#[error("audit authority returned a permit for another query, tenant, or source")]
CrossedPermit,
#[error("the durable query completion disagrees with the presented command")]
CompletionReceiptMismatch,
#[error("the current source is absent for an authorized partition")]
MissingSource(PartitionId),
#[error("this build does not resolve a family of this source kind")]
SourceKindUnsupported,
#[error("the current source vector does not match the authorized partitions")]
SourceVectorMismatch,
#[error("a current source response named another partition")]
SourceMismatch(PartitionId),
#[error("the current projection is absent for an authorized partition")]
MissingProjection(PartitionId),
#[error("the current projection belongs to a recreated source")]
Superseded(PartitionId),
#[error("the projection descriptor is incompatible with this build")]
IncompatibleDescriptor(PartitionId),
#[error("authority scope contains a duplicate or empty partition")]
DuplicatePartition,
#[error("resolved descriptors contain a duplicate key")]
DuplicateDescriptor,
#[error("State metadata refused descriptor planning")]
State(#[from] StateError),
#[error("State projection catalog refused descriptor planning")]
Projection(#[from] ProjectionCatalogError),
#[error("State query audit refused descriptor planning")]
Audit(#[from] QueryAuditError),
}
impl fmt::Debug for CoreResolutionError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(match self {
Self::Statement(_) => "Statement",
Self::DataFusion(_) => "DataFusion",
Self::NoSourceDependency => "NoSourceDependency",
Self::UnknownDependency(_) => "UnknownDependency",
Self::TableOutsideRealm => "TableOutsideRealm",
Self::FamilyOutsideRealm => "FamilyOutsideRealm",
Self::TableOutsideAudience => "TableOutsideAudience",
Self::ParameterMismatch => "ParameterMismatch",
Self::InvalidComposition => "InvalidComposition",
Self::InvalidAttribution => "InvalidAttribution",
Self::EmptyConversationIdentity => "EmptyConversationIdentity",
Self::FreshnessUnsupported { .. } => "FreshnessUnsupported",
Self::InvalidBounds => "InvalidBounds",
Self::CrossedPermit => "CrossedPermit",
Self::CompletionReceiptMismatch => "CompletionReceiptMismatch",
Self::MissingSource(_) => "MissingSource",
Self::SourceKindUnsupported => "SourceKindUnsupported",
Self::SourceVectorMismatch => "SourceVectorMismatch",
Self::SourceMismatch(_) => "SourceMismatch",
Self::MissingProjection(_) => "MissingProjection",
Self::Superseded(_) => "Superseded",
Self::IncompatibleDescriptor(_) => "IncompatibleDescriptor",
Self::DuplicatePartition => "DuplicatePartition",
Self::DuplicateDescriptor => "DuplicateDescriptor",
Self::State(_) => "State",
Self::Projection(_) => "Projection",
Self::Audit(_) => "Audit",
})
}
}
#[cfg(test)]
mod tests;