Record exact PostgreSQL command authority proof

crates/decodex-postgres/src/error.rs

@@ -122,7 +122,9 @@ impl From<refinery::Error> for StoreError {
 
 impl From<tokio_postgres::Error> for StoreError {
 	fn from(error: tokio_postgres::Error) -> Self {
-		if error.as_db_error().is_some_and(|database| {
+		if error.code().is_some_and(|code| code.code() == "DX001") {
+			Self::IdempotencyConflict
+		} else if error.as_db_error().is_some_and(|database| {
 			database.code() == &tokio_postgres::error::SqlState::CHECK_VIOLATION
 				&& database.constraint().is_some_and(|name| name.contains("no_credentials"))
 		}) {

crates/decodex-postgres/src/exact_commands.rs

@@ -0,0 +1,81 @@
+use tokio_postgres::{Row, error::SqlState, types::ToSql};
+
+use crate::{PostgresStore, StoreError};
+
+pub(crate) const EXACT_COMMAND_PROTOCOL: &str = "decodex/exact-command/1";
+const MAX_EXACT_ATTEMPTS: usize = 4;
+
+impl PostgresStore {
+	pub(crate) async fn execute_exact_with_retry(
+		&self,
+		statement: &str,
+		parameters: &[&(dyn ToSql + Sync)],
+	) -> Result<Vec<u8>, StoreError> {
+		let mut last_retryable = None;
+
+		for _ in 0..MAX_EXACT_ATTEMPTS {
+			let mut client = match self.pool().get().await {
+				Ok(client) => client,
+				Err(error) => return Err(StoreError::Pool(error)),
+			};
+			let transaction = match client.transaction().await {
+				Ok(transaction) => transaction,
+				Err(error) if is_retryable_exact_database_error(&error) => {
+					last_retryable = Some(error);
+					continue;
+				},
+				Err(error) => return Err(StoreError::from(error)),
+			};
+			let row = match transaction.query_one(statement, parameters).await {
+				Ok(row) => row,
+				Err(error) if is_retryable_exact_database_error(&error) => {
+					last_retryable = Some(error);
+					continue;
+				},
+				Err(error) => return Err(StoreError::from(error)),
+			};
+			let response = response_bytes(row)?;
+
+			match transaction.commit().await {
+				Ok(()) => return Ok(response),
+				Err(error) if is_retryable_exact_database_error(&error) => {
+					last_retryable = Some(error);
+				},
+				Err(error) => return Err(StoreError::from(error)),
+			}
+		}
+
+		Err(StoreError::Database(
+			last_retryable.expect(
+				"an exhausted exact retry loop retains its classified infrastructure failure",
+			),
+		))
+	}
+}
+
+fn response_bytes(row: Row) -> Result<Vec<u8>, StoreError> {
+	let response: Option<Vec<u8>> = row.get(0);
+
+	response.ok_or_else(|| StoreError::Incompatible("exact command response is null".into()))
+}
+
+pub(crate) fn validate_exact_key(key: &str) -> Result<(), StoreError> {
+	if key.is_empty() || key.len() > 256 {
+		return Err(StoreError::InvalidInput("idempotency key must contain 1..=256 bytes"));
+	}
+
+	crate::ensure_credential_negative_text(key)
+}
+
+fn is_retryable_exact_database_error(error: &tokio_postgres::Error) -> bool {
+	let Some(code) = error.code() else {
+		return false;
+	};
+
+	code == &SqlState::T_R_SERIALIZATION_FAILURE
+		|| code == &SqlState::T_R_DEADLOCK_DETECTED
+		|| code == &SqlState::ADMIN_SHUTDOWN
+		|| code == &SqlState::CRASH_SHUTDOWN
+		|| code == &SqlState::CANNOT_CONNECT_NOW
+		|| code.code().starts_with("08")
+}

crates/decodex-postgres/src/lib.rs

@@ -3,32 +3,44 @@
 //! This crate owns the XY-1267 persistence foundation and XY-1271 Conversation history:
 //! immutable migrations, idempotent optimistic transactions, expiring leases, transactional
 //! activity/outbox evidence, inert account/quota-window metadata, normalized history, blob
-//! references, Context Packs, and inert transition proposals. It does not select accounts,
-//! route work, store credentials, dispatch transitions, or expose protocol/client behavior.
+//! references, Context Packs, inert transition proposals, exact in-transaction receipts, and
+//! immutable global RoleProfiles. It does not select accounts, route work, store credentials,
+//! dispatch transitions, or expose protocol/client behavior.
 
 mod accounts;
 mod authority;
 mod conversations;
 mod error;
+mod exact_commands;
 mod leases;
 mod migrations;
 mod outbox;
 mod policies;
 mod programs;
 mod project_agents;
 mod quota;
+mod role_profiles;
+mod runtime_sessions;
 #[cfg(unix)] mod socket;
 mod types;
 
 pub use self::{
 	conversations::{
-		BlobReclaimPage, ContextPackRecord, CreateArtifact, CreateConversation,
-		CreateRuntimeSession, HistoryCursor, HistoryEntry, HistoryPage, PersistContextPack,
-		ProposeTransition, RecordHistoryItem, StoredArtifact, StoredConversation,
-		StoredRuntimeSession,
+		BlobReclaimPage, ContextPackRecord, CreateArtifact, CreateConversation, HistoryCursor,
+		HistoryEntry, HistoryPage, PersistContextPack, ProposeTransition, RecordHistoryItem,
+		StoredArtifact, StoredConversation,
 	},
 	error::{BootstrapFailure, StoreError},
 	programs::{ObjectiveRecord, ProgramRecord, UpdateProgramContext},
+	role_profiles::{
+		BootstrapRoleProfiles, RoleProfileCommandOutcome, RoleProfileConfiguration,
+		RoleProfileRejection, RoleProfileRevision, RoleProfileRole,
+	},
+	runtime_sessions::{
+		CreateRuntimeSession, CreateRuntimeSessionAccountSnapshot, RuntimeSessionAccountSnapshot,
+		RuntimeSessionCommandEffect, RuntimeSessionCommandOutcome, RuntimeSessionProfileSnapshot,
+		RuntimeSessionRejection, StoredRuntimeSession,
+	},
 	types::{
 		AccountMetadata, AccountMutation, ActivityRecord, CommandIdentity, CreateProject,
 		HypotheticalFallbackFact, LeaseClaim, OutboxClaim, OutboxReconciliation, OutboxState,
@@ -107,6 +119,13 @@ impl PostgresStore {
 		authority::configured_authority_sql_fixture()
 	}
 
+	/// Return the closed execution-path query and allowed function identities for fixtures.
+	#[cfg(feature = "test-support")]
+	#[doc(hidden)]
+	pub fn execution_path_contract_fixture() -> (&'static str, Vec<&'static str>) {
+		authority::execution_path_contract_fixture()
+	}
+
 	/// Apply the production connection-startup invariant to an isolated raw fixture.
 	#[cfg(feature = "test-support")]
 	#[doc(hidden)]
@@ -250,6 +269,64 @@ impl PostgresStore {
 		result
 	}
 
+	/// Apply the immutable migration ledger only through V8 for the V9 upgrade proof.
+	#[cfg(all(unix, feature = "test-support"))]
+	#[doc(hidden)]
+	pub async fn migrate_fixture_through_v8(
+		mut config: Config,
+		expected_peer_uid: u32,
+	) -> Result<(), StoreError> {
+		validate_connection(&config)?;
+
+		let connector = verified_socket_connect(&config, expected_peer_uid)?;
+
+		pin_session_search_path(&mut config);
+
+		let manager = Manager::from_connect(
+			config,
+			connector.clone(),
+			ManagerConfig { recycling_method: RecyclingMethod::Fast },
+		);
+		let pool = Pool::builder(manager).max_size(1).build()?;
+		let mut client = checkout(&pool, &connector).await?;
+		let result = migrations::run_through_v8(&mut client).await;
+
+		drop(client);
+
+		pool.close();
+
+		result
+	}
+
+	/// Apply the immutable migration ledger only through V9 for the V10 upgrade proof.
+	#[cfg(all(unix, feature = "test-support"))]
+	#[doc(hidden)]
+	pub async fn migrate_fixture_through_v9(
+		mut config: Config,
+		expected_peer_uid: u32,
+	) -> Result<(), StoreError> {
+		validate_connection(&config)?;
+
+		let connector = verified_socket_connect(&config, expected_peer_uid)?;
+
+		pin_session_search_path(&mut config);
+
+		let manager = Manager::from_connect(
+			config,
+			connector.clone(),
+			ManagerConfig { recycling_method: RecyclingMethod::Fast },
+		);
+		let pool = Pool::builder(manager).max_size(1).build()?;
+		let mut client = checkout(&pool, &connector).await?;
+		let result = migrations::run_through_v9(&mut client).await;
+
+		drop(client);
+
+		pool.close();
+
+		result
+	}
+
 	#[cfg(not(unix))]
 	pub async fn migrate(_config: Config, _expected_peer_uid: u32) -> Result<(), StoreError> {
 		Err(StoreError::Incompatible("PostgreSQL Unix sockets require a Unix host".into()))

crates/decodex-postgres/src/migrations.rs

@@ -8,7 +8,7 @@ use deadpool_postgres::Client;
 use crate::{REQUIRED_POSTGRES_MAJOR, StoreError};
 use embedded::migrations;
 
-const EXPECTED_LATEST_MIGRATION_VERSION: i32 = 8;
+const EXPECTED_LATEST_MIGRATION_VERSION: i32 = 10;
 
 pub(crate) async fn run(client: &mut Client) -> Result<(), StoreError> {
 	migrations::runner().run_async(&mut ***client).await?;
@@ -23,6 +23,20 @@ pub(crate) async fn run_through_v7(client: &mut Client) -> Result<(), StoreError
 	Ok(())
 }
 
+#[cfg(feature = "test-support")]
+pub(crate) async fn run_through_v8(client: &mut Client) -> Result<(), StoreError> {
+	migrations::runner().set_target(Target::Version(8)).run_async(&mut ***client).await?;
+
+	Ok(())
+}
+
+#[cfg(feature = "test-support")]
+pub(crate) async fn run_through_v9(client: &mut Client) -> Result<(), StoreError> {
+	migrations::runner().set_target(Target::Version(9)).run_async(&mut ***client).await?;
+
+	Ok(())
+}
+
 pub(crate) async fn verify(client: &Client) -> Result<(), StoreError> {
 	let row = client
 		.query_one(
@@ -66,7 +80,7 @@ pub(crate) async fn verify(client: &Client) -> Result<(), StoreError> {
 		!= Some(EXPECTED_LATEST_MIGRATION_VERSION)
 	{
 		return Err(StoreError::Incompatible(
-			"embedded migration inventory does not end at the canonical V8 ledger".into(),
+			"embedded migration inventory does not end at the canonical V10 ledger".into(),
 		));
 	}
 	if actual.len() != expected.len() {