nxd_core/adapters/

plugin_supervisor.rs

use crate::plugin_protocol::{
	MAX_EVENTS, MAX_MESSAGE_BYTES, PROTOCOL_VERSION, v1, validate_canonical_json, validate_metadata,
};
use crate::ports::provider::Provider;
use crate::ports::provider::{ProviderErrorCategory, ProviderFailure};
use futures_util::{StreamExt, stream::BoxStream};
use hyper_util::rt::TokioIo;
use std::collections::BTreeSet;
use std::collections::HashMap;
use std::ffi::OsString;
use std::fs;
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::io::AsyncReadExt;
use tokio::net::UnixStream;
use tokio::process::{Child, Command};
use tonic::transport::{Channel, Endpoint};
use tower::service_fn;
use v1::provider_service_client::ProviderServiceClient;

static SESSION_SEQUENCE: AtomicU64 = AtomicU64::new(0);

#[derive(Clone, Debug)]
pub struct PluginCommand {
	pub program: PathBuf,
	pub arguments: Vec<OsString>,
	pub mode: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum PluginError {
	RunDirectory,
	InvalidCommand,
	Spawn,
	StartupTimeout,
	ProcessExited,
	Incompatible,
	Protocol,
	ProtocolViolation(&'static str),
	RpcTimeout,
	EventFlood,
	VerificationFailed,
	ProviderReported(ProviderFailure),
	Cancelled,
}

impl std::fmt::Display for PluginError {
	fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		write!(formatter, "provider plugin error: {self:?}")
	}
}

impl std::error::Error for PluginError {}

pub struct PluginSession {
	child: Option<Child>,
	client: ProviderClient,
	run_dir: Option<PrivateRunDirectory>,
	stderr_task: Option<tokio::task::JoinHandle<()>>,
	request_timeout: Duration,
	provider_instance: String,
	provider_version: String,
	capabilities: BTreeSet<String>,
	/// Provider-owned resource kinds and schemas from handshake.
	resource_kinds: Vec<v1::ResourceKindDescriptor>,
}

#[derive(Clone)]
enum ProviderClient {
	External(ProviderServiceClient<Channel>),
	Linked(Arc<dyn Provider>),
}

impl ProviderClient {
	async fn get_metadata(
		&mut self,
		request: v1::GetMetadataRequest,
	) -> Result<v1::GetMetadataResponse, tonic::Status> {
		match self {
			Self::External(client) => client.get_metadata(request).await.map(tonic::Response::into_inner),
			Self::Linked(provider) => provider.get_metadata(request).await,
		}
	}

	async fn observe(
		&mut self,
		request: v1::ObserveRequest,
	) -> Result<v1::ObserveResponse, tonic::Status> {
		match self {
			Self::External(client) => client.observe(request).await.map(tonic::Response::into_inner),
			Self::Linked(provider) => provider.observe(request).await,
		}
	}

	async fn capture(
		&mut self,
		request: v1::CaptureRequest,
	) -> Result<v1::CaptureResponse, tonic::Status> {
		match self {
			Self::External(client) => client.capture(request).await.map(tonic::Response::into_inner),
			Self::Linked(provider) => provider.capture(request).await,
		}
	}

	async fn plan(&mut self, request: v1::PlanRequest) -> Result<v1::PlanResponse, tonic::Status> {
		match self {
			Self::External(client) => client.plan(request).await.map(tonic::Response::into_inner),
			Self::Linked(provider) => provider.plan(request).await,
		}
	}

	async fn apply(
		&mut self,
		request: v1::ApplyRequest,
	) -> Result<BoxStream<'static, Result<v1::ApplyResponse, tonic::Status>>, tonic::Status> {
		match self {
			Self::External(client) => {
				client.apply(request).await.map(|response| response.into_inner().boxed())
			}
			Self::Linked(provider) => provider.apply(request).await,
		}
	}

	async fn verify(
		&mut self,
		request: v1::VerifyRequest,
	) -> Result<v1::VerifyResponse, tonic::Status> {
		match self {
			Self::External(client) => client.verify(request).await.map(tonic::Response::into_inner),
			Self::Linked(provider) => provider.verify(request).await,
		}
	}

	async fn cancel(
		&mut self,
		request: v1::CancelRequest,
	) -> Result<v1::CancelResponse, tonic::Status> {
		match self {
			Self::External(client) => client.cancel(request).await.map(tonic::Response::into_inner),
			Self::Linked(provider) => provider.cancel(request).await,
		}
	}
}

impl PluginSession {
	pub async fn start(
		command: PluginCommand,
		expected_provider_kind: &str,
		provider_instance: &str,
		startup_timeout: Duration,
		request_timeout: Duration,
	) -> Result<Self, PluginError> {
		if !command.program.is_absolute()
			|| !command.program.is_file()
			|| expected_provider_kind.is_empty()
			|| provider_instance.is_empty()
		{
			return Err(PluginError::InvalidCommand);
		}
		let run_dir = create_run_dir()?;
		let socket = run_dir.0.join("provider.sock");
		let mut child_command = Command::new(command.program);
		child_command
			.env_clear()
			.args(command.arguments)
			.arg("--socket")
			.arg(&socket)
			.arg("--mode")
			.arg(command.mode)
			.stdin(Stdio::null())
			.stdout(Stdio::null())
			.stderr(Stdio::piped())
			.kill_on_drop(true);
		let mut child = child_command.spawn().map_err(|_| PluginError::Spawn)?;
		let mut stderr = child.stderr.take().ok_or(PluginError::Spawn)?;
		let stderr_task = tokio::spawn(async move {
			let mut output = Vec::new();
			let _ = (&mut stderr).take(8192).read_to_end(&mut output).await;
			output.fill(0);
		});
		let channel = connect(&socket, &mut child, startup_timeout).await?;
		let mut session = Self {
			child: Some(child),
			client: ProviderClient::External(
				ProviderServiceClient::new(channel)
					.max_decoding_message_size(MAX_MESSAGE_BYTES)
					.max_encoding_message_size(MAX_MESSAGE_BYTES),
			),
			run_dir: Some(run_dir),
			stderr_task: Some(stderr_task),
			request_timeout,
			provider_instance: provider_instance.to_string(),
			provider_version: String::new(),
			capabilities: BTreeSet::new(),
			resource_kinds: Vec::new(),
		};
		session.handshake(expected_provider_kind).await?;
		Ok(session)
	}

	pub async fn linked(
		provider: Arc<dyn Provider>,
		expected_provider_kind: &str,
		provider_instance: &str,
		request_timeout: Duration,
	) -> Result<Self, PluginError> {
		if expected_provider_kind.is_empty() || provider_instance.is_empty() {
			return Err(PluginError::InvalidCommand);
		}
		let mut session = Self {
			child: None,
			client: ProviderClient::Linked(provider),
			run_dir: None,
			stderr_task: None,
			request_timeout,
			provider_instance: provider_instance.to_string(),
			provider_version: String::new(),
			capabilities: BTreeSet::new(),
			resource_kinds: Vec::new(),
		};
		session.handshake(expected_provider_kind).await?;
		Ok(session)
	}

	async fn handshake(&mut self, expected_provider_kind: &str) -> Result<(), PluginError> {
		let metadata = tokio::time::timeout(
			self.request_timeout,
			self.client.get_metadata(v1::GetMetadataRequest {
				context: Some(request_context(&self.provider_instance, "metadata", self.request_timeout)),
			}),
		)
		.await
		.map_err(|_| PluginError::RpcTimeout)?
		.map_err(plugin_status_error)?;
		validate_metadata(&metadata).map_err(|_| PluginError::Incompatible)?;
		let capabilities = metadata.capabilities.iter().map(String::as_str).collect::<BTreeSet<_>>();
		if !["observe", "plan", "apply", "verify"]
			.into_iter()
			.all(|required| capabilities.contains(required))
		{
			return Err(PluginError::Incompatible);
		}
		if metadata.provider_kind != expected_provider_kind {
			return Err(PluginError::Incompatible);
		}
		self.provider_version = metadata.provider_version;
		self.capabilities = capabilities.into_iter().map(str::to_string).collect();
		self.resource_kinds = metadata.resource_kinds;
		Ok(())
	}

	pub fn run_dir(&self) -> &Path {
		self.run_dir.as_ref().map(|directory| directory.0.as_path()).unwrap_or(Path::new(""))
	}

	pub fn provider_version(&self) -> &str {
		&self.provider_version
	}

	pub fn capabilities(&self) -> &BTreeSet<String> {
		&self.capabilities
	}

	/// Advertised provider-owned resource kind schemas from handshake.
	pub fn resource_kinds(&self) -> &[v1::ResourceKindDescriptor] {
		&self.resource_kinds
	}

	pub async fn observe(
		&mut self,
		provider_config_json: Vec<u8>,
		selection_json: Vec<u8>,
		secrets: HashMap<String, Vec<u8>>,
	) -> Result<Vec<u8>, PluginError> {
		validate_canonical_json(&provider_config_json).map_err(|_| PluginError::Protocol)?;
		validate_canonical_json(&selection_json).map_err(|_| PluginError::Protocol)?;
		let response = tokio::time::timeout(
			self.request_timeout,
			self.client.observe(v1::ObserveRequest {
				context: Some(request_context(&self.provider_instance, "observe", self.request_timeout)),
				provider_config_json,
				secrets,
				selection_json,
			}),
		)
		.await
		.map_err(|_| PluginError::RpcTimeout)?
		.map_err(plugin_status_error)?
		.observations_json;
		validate_canonical_json(&response).map_err(|_| PluginError::Protocol)?;
		Ok(response)
	}

	pub async fn capture(
		&mut self,
		provider_config_json: Vec<u8>,
		resource_ids: Vec<String>,
		secrets: HashMap<String, Vec<u8>>,
	) -> Result<Vec<v1::CaptureArtifact>, PluginError> {
		if !self.capabilities.contains("capture") {
			return Err(PluginError::Incompatible);
		}
		validate_canonical_json(&provider_config_json).map_err(|_| PluginError::Protocol)?;
		if resource_ids.is_empty()
			|| resource_ids.len() > MAX_EVENTS
			|| resource_ids.iter().any(|resource_id| resource_id.is_empty())
		{
			return Err(PluginError::Protocol);
		}
		let artifacts = tokio::time::timeout(
			self.request_timeout,
			self.client.capture(v1::CaptureRequest {
				context: Some(request_context(&self.provider_instance, "capture", self.request_timeout)),
				provider_config_json,
				resource_ids,
				secrets,
			}),
		)
		.await
		.map_err(|_| PluginError::RpcTimeout)?
		.map_err(plugin_status_error)?
		.artifacts;
		if artifacts.len() > MAX_EVENTS {
			return Err(PluginError::Protocol);
		}
		Ok(artifacts)
	}

	pub async fn plan(
		&mut self,
		desired_json: Vec<u8>,
		observations_json: Vec<u8>,
		secrets: HashMap<String, Vec<u8>>,
	) -> Result<Vec<v1::PlannedAction>, PluginError> {
		self
			.plan_with_context(
				desired_json,
				observations_json,
				secrets,
				HashMap::new(),
				v1::EnrollmentStrategy::Unspecified,
				v1::PlanningMode::Unspecified,
			)
			.await
	}

	pub async fn plan_with_context(
		&mut self,
		desired_json: Vec<u8>,
		observations_json: Vec<u8>,
		secrets: HashMap<String, Vec<u8>>,
		resource_contexts: HashMap<String, v1::ResourcePlanContext>,
		enrollment_strategy: v1::EnrollmentStrategy,
		planning_mode: v1::PlanningMode,
	) -> Result<Vec<v1::PlannedAction>, PluginError> {
		validate_canonical_json(&desired_json).map_err(|_| PluginError::Protocol)?;
		validate_canonical_json(&observations_json).map_err(|_| PluginError::Protocol)?;
		let actions = tokio::time::timeout(
			self.request_timeout,
			self.client.plan(v1::PlanRequest {
				context: Some(request_context(&self.provider_instance, "plan", self.request_timeout)),
				desired_json,
				observations_json,
				secrets,
				resource_contexts,
				enrollment_strategy: enrollment_strategy.into(),
				planning_mode: planning_mode.into(),
			}),
		)
		.await
		.map_err(|_| PluginError::RpcTimeout)?
		.map_err(plugin_status_error)?
		.actions;
		if actions.len() > MAX_EVENTS {
			return Err(PluginError::Protocol);
		}
		for action in &actions {
			if action.action_id.is_empty()
				|| action.resource_id.is_empty()
				|| action.secret_references.len() > MAX_EVENTS
			{
				return Err(PluginError::Protocol);
			}
			validate_canonical_json(&action.details_json).map_err(|_| PluginError::Protocol)?;
		}
		Ok(actions)
	}

	pub async fn apply(
		&mut self,
		action: v1::PlannedAction,
		secrets: HashMap<String, Vec<u8>>,
		run_id: &str,
	) -> Result<Vec<v1::ApplyResponse>, PluginError> {
		self.apply_with_progress(action, secrets, run_id, crate::Logger::silent()).await
	}

	pub async fn apply_with_progress(
		&mut self,
		action: v1::PlannedAction,
		secrets: HashMap<String, Vec<u8>>,
		run_id: &str,
		logger: crate::Logger,
	) -> Result<Vec<v1::ApplyResponse>, PluginError> {
		let mut declared = action.secret_references.iter().map(String::as_str).collect::<BTreeSet<_>>();
		if let Some(contract) = action.contract.as_ref() {
			for input in &contract.inputs {
				if input.input_name.is_empty() || !declared.insert(&input.input_name) {
					return Err(PluginError::ProtocolViolation(
						"action input declarations are empty or overlap",
					));
				}
			}
		}
		let supplied = secrets.keys().map(String::as_str).collect::<BTreeSet<_>>();
		if declared != supplied {
			return Err(PluginError::ProtocolViolation(
				"action secrets do not match the reviewed declarations",
			));
		}
		let action_id = action.action_id.clone();
		let action_timeout = action
			.contract
			.as_ref()
			.map(|c| Duration::from_secs(c.timeout_seconds as u64))
			.filter(|d| !d.is_zero())
			.unwrap_or(self.request_timeout)
			.max(self.request_timeout);
		let mut apply_client = self.client.clone();
		let apply_request = v1::ApplyRequest {
			context: Some(request_context(&self.provider_instance, &action_id, action_timeout)),
			action: Some(action),
			secrets,
		};
		let apply_future = tokio::time::timeout(action_timeout, apply_client.apply(apply_request));
		tokio::pin!(apply_future);
		let response = loop {
			tokio::select! {
				result = &mut apply_future => {
					break result
						.map_err(|_| PluginError::RpcTimeout)?
						.map_err(plugin_status_error)?;
				}
				_ = tokio::time::sleep(Duration::from_millis(100)) => {
					if crate::execution::journal::cancellation_requested(run_id) {
						let mut cancel_client = self.client.clone();
						let _ = cancel_client.cancel(v1::CancelRequest {
							context: Some(request_context(&self.provider_instance, "cancel", action_timeout)),
							run_id: format!("operation/{action_id}"),
						}).await;
						// Give the provider a bounded cleanup window before Drop
						// terminates the supervised process.
						let _ = tokio::time::timeout(Duration::from_secs(5), &mut apply_future).await;
						return Err(PluginError::Cancelled);
					}
				}
			}
		};
		let mut stream = response;
		// Stream/log progress without retaining every event (long Nix builds can
		// emit thousands of lines). Only the terminal Completed event is kept for
		// confidential_outputs handoff.
		let mut completed: Option<v1::ApplyResponse> = None;
		let mut expected_sequence = 1_u32;
		let mut progress_count = 0_usize;
		loop {
			let event = tokio::select! {
				result = tokio::time::timeout(action_timeout, stream.next()) => {
					result
						.map_err(|_| PluginError::RpcTimeout)?
						.transpose()
						.map_err(plugin_status_error)?
				}
				_ = tokio::time::sleep(Duration::from_millis(100)) => {
					if crate::execution::journal::cancellation_requested(run_id) {
						let mut cancel_client = self.client.clone();
						let _ = cancel_client.cancel(v1::CancelRequest {
							context: Some(request_context(&self.provider_instance, "cancel", action_timeout)),
							run_id: format!("operation/{action_id}"),
						}).await;
						return Err(PluginError::Cancelled);
					}
					continue;
				}
			};
			let Some(mut event) = event else { break };
			if event.sequence != expected_sequence {
				return Err(PluginError::ProtocolViolation("apply event sequence is not contiguous"));
			}
			if event.action_id != action_id {
				return Err(PluginError::ProtocolViolation(
					"apply event action identity does not match the reviewed action",
				));
			}
			if event.phase == v1::EventPhase::Unspecified as i32 {
				return Err(PluginError::ProtocolViolation("apply event phase is unspecified"));
			}
			if event.safe_message.len() > MAX_MESSAGE_BYTES {
				return Err(PluginError::ProtocolViolation("apply event message exceeds its bound"));
			}
			event.safe_message = logger.sanitize(&event.safe_message);
			let failure_phase = matches!(
				v1::EventPhase::try_from(event.phase),
				Ok(v1::EventPhase::Failed | v1::EventPhase::Cancelled)
			);
			let provider_failure = match (failure_phase, event.error.take()) {
				(true, Some(error)) => Some(provider_failure_from_wire(error, &logger)?),
				(true, None) | (false, Some(_)) => {
					return Err(PluginError::ProtocolViolation(
						"apply event error payload does not match its phase",
					));
				}
				(false, None) => None,
			};
			if !event.safe_message.trim().is_empty() {
				if event.phase == v1::EventPhase::Completed as i32 {
					// Provider completion is action-local. Only the application
					// scheduler may report host/operation success after the entire
					// reviewed dependency graph has completed.
					logger.info(&event.safe_message);
				} else if event.phase == v1::EventPhase::Failed as i32 {
					logger.failure(&event.safe_message);
				} else if event.detail {
					logger.detail(&event.safe_message);
				} else {
					logger.info(&event.safe_message);
				}
			}
			if let Some(failure) = provider_failure {
				return Err(PluginError::ProviderReported(failure));
			}
			if ((!event.confidential_outputs.is_empty() || !event.public_outputs.is_empty())
				&& event.phase != v1::EventPhase::Completed as i32)
				|| event.confidential_outputs.len() > MAX_EVENTS
				|| event.public_outputs.len() > MAX_EVENTS
				|| event.confidential_outputs.iter().any(|(name, value)| {
					name.is_empty()
						|| name.chars().any(char::is_control)
						|| value.is_empty()
						|| value.len() > MAX_MESSAGE_BYTES
				}) {
				return Err(PluginError::ProtocolViolation(
					"apply event confidential outputs violate their contract",
				));
			}
			if event.public_outputs.iter().any(|(name, value)| {
				name.is_empty()
					|| name.chars().any(char::is_control)
					|| value.is_empty()
					|| value.len() > MAX_MESSAGE_BYTES
					|| event.confidential_outputs.contains_key(name)
			}) {
				return Err(PluginError::ProtocolViolation(
					"apply event public outputs violate their contract",
				));
			}
			expected_sequence = expected_sequence.saturating_add(1);
			if event.phase == v1::EventPhase::Completed as i32 {
				completed = Some(event);
			} else {
				progress_count = progress_count.saturating_add(1);
				// Flood guard on streamed progress only (not retained).
				if progress_count > MAX_EVENTS.saturating_mul(64) {
					return Err(PluginError::EventFlood);
				}
			}
		}
		let Some(completed) = completed else {
			return Err(PluginError::ProtocolViolation("apply stream ended without a completed event"));
		};
		Ok(vec![completed])
	}

	pub async fn verify(
		&mut self,
		action: v1::PlannedAction,
		secrets: HashMap<String, Vec<u8>>,
	) -> Result<(), PluginError> {
		self.verify_with_outputs(action, secrets, HashMap::new(), HashMap::new()).await
	}

	pub async fn verify_with_outputs(
		&mut self,
		action: v1::PlannedAction,
		secrets: HashMap<String, Vec<u8>>,
		confidential_outputs: HashMap<String, Vec<u8>>,
		public_outputs: HashMap<String, Vec<u8>>,
	) -> Result<(), PluginError> {
		let action_timeout = action
			.contract
			.as_ref()
			.map(|c| Duration::from_secs(c.timeout_seconds as u64))
			.filter(|d| !d.is_zero())
			.unwrap_or(self.request_timeout)
			.max(self.request_timeout);
		let result = tokio::time::timeout(
			action_timeout,
			self.client.verify(v1::VerifyRequest {
				context: Some(request_context(&self.provider_instance, "verify", action_timeout)),
				action: Some(action),
				secrets,
				confidential_outputs,
				public_outputs,
			}),
		)
		.await
		.map_err(|_| PluginError::RpcTimeout)?
		.map_err(plugin_status_error)?;
		if let Some(error) = result.error {
			let logger = crate::process::Logger::silent();
			return Err(PluginError::ProviderReported(provider_failure_from_wire(error, &logger)?));
		}
		if !result.satisfied {
			return Err(PluginError::VerificationFailed);
		}
		validate_canonical_json(&result.observation_json).map_err(|_| PluginError::Protocol)?;
		Ok(())
	}

	pub async fn cancel(&mut self, run_id: &str) -> Result<(), PluginError> {
		tokio::time::timeout(
			self.request_timeout,
			self.client.cancel(v1::CancelRequest {
				context: Some(request_context(&self.provider_instance, "cancel", self.request_timeout)),
				run_id: run_id.to_string(),
			}),
		)
		.await
		.map_err(|_| PluginError::RpcTimeout)?
		.map_err(plugin_status_error)?;
		if let Some(child) = self.child.as_mut() {
			child.start_kill().map_err(|_| PluginError::ProcessExited)?;
			tokio::time::timeout(self.request_timeout, child.wait())
				.await
				.map_err(|_| PluginError::RpcTimeout)?
				.map_err(|_| PluginError::ProcessExited)?;
		}
		Ok(())
	}
}

impl Drop for PluginSession {
	fn drop(&mut self) {
		if let Some(child) = self.child.as_mut() {
			let _ = child.start_kill();
		}
		if let Some(stderr_task) = self.stderr_task.as_ref() {
			stderr_task.abort();
		}
	}
}

struct PrivateRunDirectory(PathBuf);

impl Drop for PrivateRunDirectory {
	fn drop(&mut self) {
		let _ = fs::remove_dir_all(&self.0);
	}
}

fn create_run_dir() -> Result<PrivateRunDirectory, PluginError> {
	let sequence = SESSION_SEQUENCE.fetch_add(1, Ordering::Relaxed);
	let run_dir = PathBuf::from("/tmp").join(format!("nxd-plugin-{}-{sequence}", std::process::id()));
	fs::create_dir(&run_dir).map_err(|_| PluginError::RunDirectory)?;
	fs::set_permissions(&run_dir, fs::Permissions::from_mode(0o700))
		.map_err(|_| PluginError::RunDirectory)?;
	Ok(PrivateRunDirectory(run_dir))
}

async fn connect(
	socket: &Path,
	child: &mut Child,
	timeout: Duration,
) -> Result<Channel, PluginError> {
	let socket = socket.to_path_buf();
	tokio::time::timeout(timeout, async {
		loop {
			if child.try_wait().map_err(|_| PluginError::ProcessExited)?.is_some() {
				return Err(PluginError::ProcessExited);
			}
			let path = socket.clone();
			// Accept modest trailer sizes so a misbehaving provider status does not
			// immediately GOAWAY; providers still truncate Status messages.
			let endpoint = Endpoint::from_static("http://[::]:50051")
				.connect_timeout(Duration::from_millis(100))
				.http2_max_header_list_size(64 * 1024);
			if let Ok(channel) = endpoint
				.connect_with_connector(service_fn(move |_| {
					let path = path.clone();
					async move { UnixStream::connect(path).await.map(TokioIo::new) }
				}))
				.await
			{
				return Ok(channel);
			}
			tokio::time::sleep(Duration::from_millis(20)).await;
		}
	})
	.await
	.map_err(|_| PluginError::StartupTimeout)?
}

fn provider_failure_from_wire(
	error: v1::ProviderError,
	logger: &crate::process::Logger,
) -> Result<ProviderFailure, PluginError> {
	let category = match v1::ErrorCategory::try_from(error.category) {
		Ok(v1::ErrorCategory::InvalidRequest) => ProviderErrorCategory::InvalidRequest,
		Ok(v1::ErrorCategory::Incompatible) => ProviderErrorCategory::Incompatible,
		Ok(v1::ErrorCategory::Timeout) => ProviderErrorCategory::Timeout,
		Ok(v1::ErrorCategory::Cancelled) => ProviderErrorCategory::Cancelled,
		Ok(v1::ErrorCategory::Provider) => ProviderErrorCategory::Provider,
		Ok(v1::ErrorCategory::Transport) => ProviderErrorCategory::Transport,
		Ok(v1::ErrorCategory::Internal) => ProviderErrorCategory::Internal,
		Ok(v1::ErrorCategory::Unspecified) | Err(_) => {
			return Err(PluginError::ProtocolViolation(
				"provider failure has an unspecified error category",
			));
		}
	};
	let safe_message = provider_safe_message(logger, &error.safe_message);
	if safe_message.is_empty() || safe_message.len() > MAX_MESSAGE_BYTES {
		return Err(PluginError::ProtocolViolation(
			"provider failure message is empty or exceeds its bound",
		));
	}
	Ok(ProviderFailure {
		category,
		retryable: error.retryable,
		ambiguous: error.ambiguous,
		safe_message,
	})
}

fn plugin_status_error(error: tonic::Status) -> PluginError {
	let (category, retryable) = match error.code() {
		tonic::Code::InvalidArgument
		| tonic::Code::OutOfRange
		| tonic::Code::AlreadyExists
		| tonic::Code::NotFound => (ProviderErrorCategory::InvalidRequest, false),
		tonic::Code::Unimplemented => (ProviderErrorCategory::Incompatible, false),
		tonic::Code::DeadlineExceeded => (ProviderErrorCategory::Timeout, true),
		tonic::Code::Cancelled => (ProviderErrorCategory::Cancelled, false),
		tonic::Code::Unavailable => (ProviderErrorCategory::Transport, true),
		tonic::Code::FailedPrecondition
		| tonic::Code::Aborted
		| tonic::Code::PermissionDenied
		| tonic::Code::Unauthenticated => (ProviderErrorCategory::Provider, false),
		tonic::Code::Unknown
		| tonic::Code::Internal
		| tonic::Code::DataLoss
		| tonic::Code::ResourceExhausted
		| tonic::Code::Ok => (ProviderErrorCategory::Internal, false),
	};
	let logger = crate::process::Logger::silent();
	let safe_message = provider_safe_message(&logger, error.message());
	if safe_message.is_empty() {
		return PluginError::ProtocolViolation("provider returned an empty gRPC status message");
	}
	PluginError::ProviderReported(ProviderFailure {
		category,
		retryable,
		ambiguous: false,
		safe_message,
	})
}

fn provider_safe_message(logger: &crate::process::Logger, message: &str) -> String {
	logger.sanitize(message).split_whitespace().collect::<Vec<_>>().join(" ")
}

fn request_context(
	provider_instance: &str,
	operation: &str,
	deadline_from_now: Duration,
) -> v1::RequestContext {
	let deadline_unix_ms = SystemTime::now()
		.duration_since(UNIX_EPOCH)
		.unwrap_or_default()
		.saturating_add(deadline_from_now)
		.as_millis() as u64;
	v1::RequestContext {
		protocol_version: PROTOCOL_VERSION.to_string(),
		provider_instance: provider_instance.to_string(),
		operation_id: format!("operation/{operation}"),
		deadline_unix_ms,
	}
}

#[cfg(test)]
mod tests {
	use super::*;

	#[test]
	fn provider_failure_preserves_typed_retry_and_ambiguity() {
		let logger = crate::process::Logger::silent();
		let failure = provider_failure_from_wire(
			v1::ProviderError {
				category: v1::ErrorCategory::Transport as i32,
				safe_message: "endpoint not ready".into(),
				retryable: true,
				ambiguous: true,
			},
			&logger,
		)
		.expect("typed provider failure");
		assert_eq!(failure.category, ProviderErrorCategory::Transport);
		assert!(failure.retryable);
		assert!(failure.ambiguous);
		assert!(!failure.may_retry(), "ambiguous failures must never retry");
	}

	#[test]
	fn provider_failure_sanitizes_multiline_detail_without_losing_its_type() {
		let logger = crate::process::Logger::silent();
		let failure = provider_failure_from_wire(
			v1::ProviderError {
				category: v1::ErrorCategory::Provider as i32,
				safe_message: "QGA file read failed:\npermission denied".into(),
				retryable: false,
				ambiguous: false,
			},
			&logger,
		)
		.expect("sanitized typed provider failure");
		assert_eq!(failure.category, ProviderErrorCategory::Provider);
		assert_eq!(failure.safe_message, "QGA file read failed: permission denied");
	}

	#[test]
	fn malformed_provider_failure_is_protocol_error() {
		let logger = crate::process::Logger::silent();
		for error in [
			v1::ProviderError {
				category: v1::ErrorCategory::Unspecified as i32,
				safe_message: "missing category".into(),
				retryable: false,
				ambiguous: false,
			},
			v1::ProviderError {
				category: 999,
				safe_message: "unknown category".into(),
				retryable: false,
				ambiguous: false,
			},
			v1::ProviderError {
				category: v1::ErrorCategory::Provider as i32,
				safe_message: String::new(),
				retryable: false,
				ambiguous: false,
			},
		] {
			assert!(matches!(
				provider_failure_from_wire(error, &logger),
				Err(PluginError::ProtocolViolation(_))
			));
		}
	}

	#[test]
	fn grpc_status_retry_policy_is_code_based() {
		let PluginError::ProviderReported(unavailable) =
			plugin_status_error(tonic::Status::unavailable("temporary"))
		else {
			panic!("unavailable maps to a typed provider failure");
		};
		assert!(unavailable.may_retry());

		let PluginError::ProviderReported(auth) =
			plugin_status_error(tonic::Status::permission_denied("denied"))
		else {
			panic!("permission denied maps to a typed provider failure");
		};
		assert!(!auth.may_retry());
		assert!(matches!(
			plugin_status_error(tonic::Status::unknown("")),
			PluginError::ProtocolViolation("provider returned an empty gRPC status message")
		));
	}
}