diff --git a/apps/oppa/src-tauri/src/connection.rs b/apps/oppa/src-tauri/src/connection.rs index 3cf092e..9917d72 100644 --- a/apps/oppa/src-tauri/src/connection.rs +++ b/apps/oppa/src-tauri/src/connection.rs @@ -1,7 +1,7 @@ use std::{ collections::VecDeque, sync::{Arc, atomic::Ordering}, - time::Duration, + time::{Duration, SystemTime, UNIX_EPOCH}, }; use async_trait::async_trait; @@ -11,14 +11,16 @@ use oppa_agent::{ }; use oppa_core::Timestamp; use oppa_protocol::{ - AgentHeartbeat, AgentHello, AgentMessage, AgentMessageKind, GatewayAuthenticationResponse, - GatewayAuthenticationServerMessage, PrinterInventory, ProtocolVersion, ServerMessage, - ServerMessageKind, + AgentHeartbeat, AgentHello, AgentMessage, AgentMessageKind, CacheUpdate, + GatewayAuthenticationResponse, GatewayAuthenticationServerMessage, HandshakePlanner, + LegacyServerRecord, PrinterInventory, ProtocolVersion, ServerMessage, ServerMessageKind, + project_inventory, }; use oppa_transport::{ AgentTransport, BackoffPolicy, ReconnectBackoff, TransportConfig, TransportError, WebSocketTransport, }; +use serde::{Deserialize, Serialize}; use tokio::sync::{mpsc, oneshot}; use uuid::Uuid; @@ -32,6 +34,63 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(12); const SERVER_HELLO_TIMEOUT: Duration = Duration::from_secs(20); const IDLE_TIMEOUT: Duration = Duration::from_secs(90); const OUTBOUND_BUFFER_CAPACITY: usize = 256; +/// Setting that remembers which paired server accepted only the base protocol. +const PROTOCOL_NEGOTIATION_SETTING: &str = "openprinter-protocol-negotiation-v1"; + +/// Persisted form of [`LegacyServerRecord`], keyed by the paired server identity. +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct StoredNegotiation { + server_id: String, + legacy_until_ms: u64, +} + +/// Negotiation state for one connection: what the hello offered and what the server selected. +struct ConnectionNegotiation { + offered: Vec, + selected: Option, +} + +fn now_millis() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |duration| { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) + }) +} + +async fn load_legacy_record( + service: &DesktopService, + server_id: &str, +) -> Option { + let value = service + .storage + .setting(PROTOCOL_NEGOTIATION_SETTING) + .await + .ok()??; + let stored: StoredNegotiation = serde_json::from_value(value).ok()?; + (stored.server_id == server_id).then_some(LegacyServerRecord { + legacy_until_ms: stored.legacy_until_ms, + }) +} + +async fn store_cache_update(service: &DesktopService, server_id: &str, update: CacheUpdate) { + let value = match update { + CacheUpdate::Unchanged => return, + CacheUpdate::Clear => serde_json::Value::Null, + CacheUpdate::Store(record) => serde_json::json!(StoredNegotiation { + server_id: server_id.to_owned(), + legacy_until_ms: record.legacy_until_ms, + }), + }; + if let Err(error) = service + .storage + .set_setting(PROTOCOL_NEGOTIATION_SETTING, &value) + .await + { + service.log.warn("transport", error.to_string()); + } +} /// Host request that needs confirmation from the connection actor. pub struct OutboundRequest { @@ -93,6 +152,8 @@ pub async fn run_connection_supervisor( }; let mut automatic_reconnect = true; let mut retry_after = None; + // Planned across attempts of one reconnect cycle, so a rejected extended offer can fall back at once. + let mut handshake: Option = None; loop { if service.shutdown.is_cancelled() { @@ -322,7 +383,15 @@ pub async fn run_connection_supervisor( } } - let hello = hello_message(&service, &connection.agent_id); + let planner = if let Some(planner) = handshake.take() { + planner + } else { + let record = load_legacy_record(&service, &connection.server_id).await; + HandshakePlanner::new(record, now_millis()) + }; + let offered = planner.offered().to_vec(); + handshake = Some(planner); + let hello = hello_message(&service, &connection.agent_id, &offered); let hello_message_id = hello.message_id.clone(); if let Err(error) = transport.send(hello).await { let _provider_operation = service.provider_operation.lock().await; @@ -344,6 +413,10 @@ pub async fn run_connection_supervisor( ); let mut handshaken = false; + let mut negotiation = ConnectionNegotiation { + offered, + selected: None, + }; let mut awaiting_handshake = VecDeque::new(); let mut connection_ended = false; let hello_timeout = tokio::time::sleep(SERVER_HELLO_TIMEOUT); @@ -392,7 +465,14 @@ pub async fn run_connection_supervisor( if service.provider_generation.load(Ordering::Acquire) != provider_generation { automatic_reconnect = false; connection_ended = true; - } else if let Err(error) = send_inventory(&service, &mut transport, None).await { + } else if let Err(error) = send_inventory( + &service, + &mut transport, + None, + negotiation.selected.unwrap_or(ProtocolVersion::BASE), + ) + .await + { service.set_connection_error(error.to_string()).await; retry_after = next_retry_delay(&mut backoff); if retry_after.is_none() { @@ -457,6 +537,7 @@ pub async fn run_connection_supervisor( &mut awaiting_handshake, gateway_endpoint.as_str(), provider_generation, + &mut negotiation, ) .await { @@ -476,6 +557,13 @@ pub async fn run_connection_supervisor( } if !was_handshaken && handshaken { backoff.reset(); + // The handshake completed, so the next reconnect cycle plans from a fresh state. + if let (Some(planner), Some(selected)) = + (handshake.take(), negotiation.selected) + { + let update = planner.on_accepted(selected, now_millis()); + store_cache_update(&service, &connection.server_id, update).await; + } } } Err(error) => { @@ -487,9 +575,18 @@ pub async fn run_connection_supervisor( continue; } let retryable = error.is_retryable(); + // A connection that ends after an extended hello but before server.hello is how a + // base-only server refuses that hello. Retry at once with the base offer. A timeout + // never reaches this branch, so it cannot trigger a fallback. + let fell_back = !handshaken + && handshake + .as_mut() + .is_some_and(HandshakePlanner::on_closed_before_hello); service.set_connection_error(error.to_string()).await; reject_pending(&mut awaiting_handshake, &error.to_string()); - if retryable { + if fell_back { + retry_after = Some(Duration::ZERO); + } else if retryable { retry_after = backoff .next_delay() .ok() @@ -540,6 +637,7 @@ async fn handle_server_message( awaiting_handshake: &mut VecDeque, gateway_url: &str, provider_generation: u64, + negotiation: &mut ConnectionNegotiation, ) -> MessageDisposition { let _provider_operation = service.provider_operation.lock().await; if service.provider_generation.load(Ordering::Acquire) != provider_generation { @@ -554,6 +652,17 @@ async fn handle_server_message( .await; return MessageDisposition::Reconnect { delay: None }; } + if *handshaken + && message.protocol_version == ProtocolVersion::EXTENDED + && negotiation.selected != Some(ProtocolVersion::EXTENDED) + { + service + .set_connection_error( + "Gateway used an extended protocol version that was not negotiated.", + ) + .await; + return MessageDisposition::Reconnect { delay: None }; + } match &message.kind { ServerMessageKind::Hello(hello) => { if *handshaken || message.correlation_id.as_deref() != Some(hello_message_id) { @@ -562,6 +671,19 @@ async fn handle_server_message( .await; return MessageDisposition::Reconnect { delay: None }; } + // The server must select a version the agent offered in this attempt. + if !negotiation + .offered + .contains(&hello.selected_protocol_version) + { + service + .set_connection_error( + "Gateway selected a protocol version the agent did not offer.", + ) + .await; + return MessageDisposition::Reconnect { delay: None }; + } + negotiation.selected = Some(hello.selected_protocol_version); *handshaken = true; *service.last_connection_at.write().await = Some(Timestamp::now().to_string()); *service.connected_service.write().await = Some(ConnectedServiceSummary { @@ -583,7 +705,9 @@ async fn handle_server_message( ), ); - if let Err(error) = send_inventory(service, transport, None).await { + if let Err(error) = + send_inventory(service, transport, None, hello.selected_protocol_version).await + { service.set_connection_error(error.to_string()).await; return MessageDisposition::Reconnect { delay: None }; } @@ -656,8 +780,13 @@ async fn handle_server_message( } } ServerMessageKind::RequestPrinterInventory(_) => { - if let Err(error) = - send_inventory(service, transport, Some(message.message_id.clone())).await + if let Err(error) = send_inventory( + service, + transport, + Some(message.message_id.clone()), + negotiation.selected.unwrap_or(ProtocolVersion::BASE), + ) + .await { service.set_connection_error(error.to_string()).await; MessageDisposition::Reconnect { delay: None } @@ -669,7 +798,14 @@ async fn handle_server_message( if let Err(error) = service.refresh_printers().await { service.log.warn("discovery", error.to_string()); } - if let Err(error) = send_inventory(service, transport, None).await { + if let Err(error) = send_inventory( + service, + transport, + None, + negotiation.selected.unwrap_or(ProtocolVersion::BASE), + ) + .await + { service.set_connection_error(error.to_string()).await; MessageDisposition::Reconnect { delay: None } } else { @@ -821,23 +957,29 @@ fn desktop_job_state(state: oppa_core::JobState) -> DesktopJobState { } } -fn hello_message(service: &DesktopService, agent_id: &str) -> AgentMessage { +fn hello_message( + service: &DesktopService, + agent_id: &str, + offered: &[ProtocolVersion], +) -> AgentMessage { envelope( AgentMessageKind::Hello(AgentHello { agent_id: agent_id.to_owned(), agent_version: env!("CARGO_PKG_VERSION").to_owned(), product_id: service.product.product_id.to_string(), product_version: env!("CARGO_PKG_VERSION").to_owned(), - supported_protocol_versions: vec![ProtocolVersion::CURRENT], + supported_protocol_versions: offered.to_vec(), }), None, ) } +/// Sends the printer inventory projected for the negotiated protocol version. async fn send_inventory( service: &DesktopService, transport: &mut WebSocketTransport, correlation_id: Option, + version: ProtocolVersion, ) -> Result<(), TransportError> { let printers = service .catalog @@ -848,16 +990,17 @@ async fn send_inventory( .send(envelope( AgentMessageKind::PrinterInventory(PrinterInventory { revision: service.catalog.revision(), - printers, + printers: project_inventory(&printers, version), }), correlation_id, )) .await } +/// Labels each envelope with the lowest protocol version its content needs. fn envelope(kind: AgentMessageKind, correlation_id: Option) -> AgentMessage { AgentMessage { - protocol_version: ProtocolVersion::CURRENT, + protocol_version: kind.required_protocol_version(), message_id: format!("message_{}", Uuid::new_v4()), sent_at: Timestamp::now().to_string(), correlation_id, diff --git a/apps/www/content/docs/print-documents.mdx b/apps/www/content/docs/print-documents.mdx index d58cf89..eb6dad9 100644 --- a/apps/www/content/docs/print-documents.mdx +++ b/apps/www/content/docs/print-documents.mdx @@ -56,6 +56,10 @@ them, and the spooler chooses the submission route. ## PDF documents +PDF jobs require an agent that negotiated protocol version 2. Agents +that only speak the base protocol still receive receipts. See +[Protocol](/docs/protocol). + A job can reference a PDF instead of a receipt. The job's `document` member is a PDF when it has a `format` value of `"pdf"`; receipts keep their original shape. diff --git a/apps/www/content/docs/protocol.mdx b/apps/www/content/docs/protocol.mdx index 02f2055..39f9936 100644 --- a/apps/www/content/docs/protocol.mdx +++ b/apps/www/content/docs/protocol.mdx @@ -135,7 +135,7 @@ differs from the job, and stops at the declared byte length. ## Capability advertisement -Printer descriptors advertise what the printer can accept in the +Printer descriptors may advertise what the printer can accept in the optional `capabilities` object: - `documentFormats`: `receipt` and/or `pdf` @@ -145,19 +145,60 @@ optional `capabilities` object: - `printOptions`: explicit booleans for `copies`, `duplex`, `orientation`, `scaling`, and `pageRanges` -A server should send a PDF job only to a printer whose -`documentFormats` include `pdf`. An agent that receives a PDF it -cannot route fails the job with a non-retryable code instead of -guessing. - -### Compatibility - -Document capabilities are additive optional fields. Earlier agents -never send them, and absent fields mean receipt-only behavior. Servers -built from earlier schemas reject unknown fields, so they reject an -inventory that includes `documentFormats` or the other new capability -fields. Upgrade `@openprinter/protocol` and the server SDK before -connecting agents from this release to such a server. +These fields are part of protocol version 2. An agent sends them only +in a session that negotiated version 2. A server should send a PDF job +only to a printer whose `documentFormats` include `pdf`. An agent that +receives a PDF it cannot route fails the job with a non-retryable code +instead of guessing. + +## Protocol versions and negotiation + +Protocol version `"1"` is the base protocol, and it is unchanged from +earlier releases. Protocol version `"2"` adds PDF print jobs and +printer document capabilities. Peers negotiate the version during the +hello exchange, and independent upgrades are supported in both +directions: + +| Agent | Server | Result | +| --------------- | --------------- | ----------------------------------------------------------------------------------------------------- | +| Earlier release | Any | Base protocol, identical to earlier releases | +| Current release | Earlier release | Base protocol. The agent's extended hello is refused, and the agent retries once with the base offer. | +| Current release | Current release | Version 2 negotiated. Document capabilities and PDF jobs are exchanged. | + +The rules that make this safe: + +- **Handshake frames use envelope version `"1"`.** `agent.hello` + offers `supportedProtocolVersions`. Current agents offer + `["2", "1"]`, and earlier agents offer `["1"]`. `server.hello` lists + only the selected version in `supportedProtocolVersions` and names + it in `selectedProtocolVersion`. An earlier agent therefore receives + exactly the reply it expects. +- **Every other message uses the lowest version its content needs.** + Receipts, reports, heartbeats, and base-protocol inventory use + `"1"`. A PDF job or a printer with document capabilities uses `"2"`. +- **Receivers enforce both rules.** A `"1"` envelope that carries + version-2 content is rejected. A `"2"` envelope is rejected unless + version 2 was negotiated in that session. A server never sends + version-2 content to a session that negotiated version 1; `sendJob` + returns `protocol-version-unsupported` instead. +- **Earlier servers close a rejected extended hello without + replying.** The agent treats a connection that closes after its + hello and before `server.hello` as a refusal. It retries once with + the base offer, and after a successful base handshake it remembers + the server for 24 hours. During that time it does not offer the + extended protocol again, so a server upgrade is picked up within a + day without re-pairing. A timeout waiting for `server.hello` is not + treated as a refusal. +- **Base-only sessions never carry extended fields.** The agent + removes document capabilities from its inventory before sending it, + so an earlier server receives the same inventory it always did. +- **Agents never infer support from version strings.** The only + evidence is the explicit `selectedProtocolVersion` in + `server.hello`. + +Servers can restrict the versions they accept with the +`protocolVersions` option. Use `["1"]` to serve only base-protocol +agents; `"1"` is always required. ## Delivery states diff --git a/crates/oppa-agent/README.md b/crates/oppa-agent/README.md index 2075633..962eecd 100644 --- a/crates/oppa-agent/README.md +++ b/crates/oppa-agent/README.md @@ -88,6 +88,13 @@ accepts a protocol `PrintJob` so it can reuse the same validation, printer resol spooler boundaries, but it intentionally creates no server-job row, outbox row, job event, or outbound protocol report. A host may keep separate desktop-only test history. +## Protocol negotiation + +The agent's connection layer, not this crate, owns the hello exchange. It offers +`[V2, V1]` and falls back once to `[V1]` when a server closes an extended hello +without replying. See `oppa_protocol::HandshakePlanner`. `Agent` reports use the +base envelope, and PDF jobs are received only in a negotiated version-2 session. + ## PDF jobs A job whose `document` is a PDF follows the same receive, durable-storage, diff --git a/crates/oppa-agent/src/processor.rs b/crates/oppa-agent/src/processor.rs index fba1880..f5e7497 100644 --- a/crates/oppa-agent/src/processor.rs +++ b/crates/oppa-agent/src/processor.rs @@ -9,8 +9,8 @@ use oppa_printer::{ }; use oppa_protocol::{ AgentMessage, AgentMessageKind, FailureDetail, JobDocument, JobFailed, JobReceived, JobStatus, - JobSubmitted, PdfDocument, PrintDocument, PrintJob, PrintSettings, ProtocolVersion, - ServerMessage, ServerMessageKind, Validate, ValidationError, + JobSubmitted, PdfDocument, PrintDocument, PrintJob, PrintSettings, ServerMessage, + ServerMessageKind, Validate, ValidationError, }; use oppa_renderer::{ DocumentRenderer, PageRenderOptions, RenderTarget, RenderedDocument, RendererError, @@ -1064,7 +1064,7 @@ fn agent_message( kind: AgentMessageKind, ) -> Result { let message = AgentMessage { - protocol_version: ProtocolVersion::CURRENT, + protocol_version: kind.required_protocol_version(), message_id: format!("msg_{}", Uuid::new_v4().simple()), sent_at: Timestamp::now().to_string(), correlation_id: Some(correlation_id.to_owned()), diff --git a/crates/oppa-agent/tests/job_lifecycle.rs b/crates/oppa-agent/tests/job_lifecycle.rs index bde7f32..bdbc267 100644 --- a/crates/oppa-agent/tests/job_lifecycle.rs +++ b/crates/oppa-agent/tests/job_lifecycle.rs @@ -215,7 +215,7 @@ fn virtual_printer() -> PrinterRef { fn print_job_message(message_id: &str, job_id: &str, idempotency_key: &str) -> ServerMessage { ServerMessage { - protocol_version: ProtocolVersion::CURRENT, + protocol_version: ProtocolVersion::V1, message_id: message_id.to_owned(), sent_at: "2026-07-28T10:00:00Z".to_owned(), correlation_id: None, diff --git a/crates/oppa-agent/tests/pdf_jobs.rs b/crates/oppa-agent/tests/pdf_jobs.rs index 58b7cdb..4990fa4 100644 --- a/crates/oppa-agent/tests/pdf_jobs.rs +++ b/crates/oppa-agent/tests/pdf_jobs.rs @@ -139,7 +139,7 @@ fn pdf_job_message( settings: Option, ) -> ServerMessage { ServerMessage { - protocol_version: oppa_protocol::ProtocolVersion::CURRENT, + protocol_version: oppa_protocol::ProtocolVersion::V2, message_id: format!("msg_{job_id}"), sent_at: "2026-07-28T10:00:00Z".to_owned(), correlation_id: None, diff --git a/crates/oppa-protocol/README.md b/crates/oppa-protocol/README.md index 78dfa90..c266a17 100644 --- a/crates/oppa-protocol/README.md +++ b/crates/oppa-protocol/README.md @@ -40,6 +40,26 @@ optional `PrintSettings`. Bytes never travel in a message. `custom_paper_size`, and `print_options` fields. Absent fields mean receipt-only behavior and are never inferred. +## Version negotiation + +`ProtocolVersion` is `V1` (the base protocol) or `V2` (PDF jobs and document +capabilities). Handshake frames always use `V1`. Other envelopes use the lowest +version their content needs, through `AgentMessageKind::required_protocol_version` +and `ServerMessageKind::required_protocol_version`. Decoders reject a `V1` +envelope that carries version-2 content. + +`decode_agent_message_for` and `decode_server_message_for` decode as a peer that +understands only the supplied versions. A base-only peer uses +`&[ProtocolVersion::BASE]`, which rejects an extended hello exactly as the +original decoder did. + +`negotiation` contains the agent's `HandshakePlanner`, which decides each hello +offer, falls back once to the base offer after a refused extended hello, and +returns what to persist. It also contains `select_protocol_version` and +`project_inventory`, which strips version-2 fields for base-only sessions. +Tests in `tests/negotiation.rs` cover each combination of agent and server +generation. + ## Source of truth and development The TypeBox schemas in `../../packages/protocol/src/schemas` are canonical. Their deterministic JSON diff --git a/crates/oppa-protocol/src/codec.rs b/crates/oppa-protocol/src/codec.rs index fc1fb03..2d067d1 100644 --- a/crates/oppa-protocol/src/codec.rs +++ b/crates/oppa-protocol/src/codec.rs @@ -2,11 +2,10 @@ use serde::Serialize; use serde_json::{Map, Value}; use crate::{ - AgentMessage, MAX_WIRE_MESSAGE_BYTES, PROTOCOL_VERSION, ProtocolError, ProtocolMessage, - ServerMessage, Validate, ValidationError, + AgentMessage, AgentMessageKind, MAX_WIRE_MESSAGE_BYTES, ProtocolError, ProtocolMessage, + ProtocolVersion, ServerMessage, ServerMessageKind, Validate, ValidationError, }; -const SUPPORTED_VERSIONS: &[&str] = &[PROTOCOL_VERSION]; const JSON_SAFE_INTEGER: f64 = 9_007_199_254_740_991.0; const ENVELOPE_FIELDS: &[&str] = &[ "protocolVersion", @@ -19,25 +18,50 @@ const ENVELOPE_FIELDS: &[&str] = &[ /// Decodes UTF-8 JSON and validates an agent-to-server message. pub fn decode_agent_message(input: &[u8]) -> Result { - let value = decode_value(input)?; + decode_agent_message_for(input, ProtocolVersion::SUPPORTED) +} + +/// Decodes an agent message as a peer that understands only `supported` versions would. +/// +/// A base-only peer passes `&[ProtocolVersion::BASE]`. It then rejects a hello that offers the +/// extended version and an envelope labelled with it, as the original v1 decoder did. +pub fn decode_agent_message_for( + input: &[u8], + supported: &[ProtocolVersion], +) -> Result { + let value = decode_value(input, supported)?; validate_envelope_shape(&value)?; let message: AgentMessage = deserialize_message(value)?; message.validate()?; + if let AgentMessageKind::Hello(hello) = &message.kind { + ensure_offers_supported(&hello.supported_protocol_versions, supported)?; + } Ok(message) } /// Decodes UTF-8 JSON and validates a server-to-agent message. pub fn decode_server_message(input: &[u8]) -> Result { - let value = decode_value(input)?; + decode_server_message_for(input, ProtocolVersion::SUPPORTED) +} + +/// Decodes a server message as a peer that understands only `supported` versions would. +pub fn decode_server_message_for( + input: &[u8], + supported: &[ProtocolVersion], +) -> Result { + let value = decode_value(input, supported)?; validate_envelope_shape(&value)?; let message: ServerMessage = deserialize_message(value)?; message.validate()?; + if let ServerMessageKind::Hello(hello) = &message.kind { + ensure_offers_supported(&hello.supported_protocol_versions, supported)?; + } Ok(message) } /// Decodes UTF-8 JSON and routes a validated message by its discriminator. pub fn decode_protocol_message(input: &[u8]) -> Result { - let value = decode_value(input)?; + let value = decode_value(input, ProtocolVersion::SUPPORTED)?; validate_envelope_shape(&value)?; let message_type = value .get("type") @@ -75,14 +99,29 @@ pub fn encode_protocol_message(message: &ProtocolMessage) -> Result, Pro } } -fn decode_value(input: &[u8]) -> Result { +fn decode_value(input: &[u8], supported: &[ProtocolVersion]) -> Result { enforce_size(input.len())?; let mut value: Value = serde_json::from_slice(input).map_err(|_| ProtocolError::InvalidJson)?; normalize_integral_numbers(&mut value); - assert_supported_version(&value)?; + assert_supported_version(&value, supported)?; Ok(value) } +/// Rejects a handshake that lists a version the decoding peer does not support. +fn ensure_offers_supported( + listed: &[ProtocolVersion], + supported: &[ProtocolVersion], +) -> Result<(), ProtocolError> { + if listed.iter().all(|version| supported.contains(version)) { + return Ok(()); + } + Err(ValidationError::new( + "supportedProtocolVersions", + "lists a protocol version this peer does not support", + ) + .into()) +} + fn normalize_integral_numbers(value: &mut Value) { match value { Value::Number(number) if number.as_i64().is_none() && number.as_u64().is_none() => { @@ -112,11 +151,17 @@ fn normalize_integral_numbers(value: &mut Value) { } } -fn assert_supported_version(value: &Value) -> Result<(), ProtocolError> { +fn assert_supported_version( + value: &Value, + supported: &[ProtocolVersion], +) -> Result<(), ProtocolError> { let Some(version) = value.get("protocolVersion") else { return Ok(()); }; - if version == &Value::from(PROTOCOL_VERSION) { + if supported + .iter() + .any(|candidate| version == &Value::from(candidate.as_str())) + { return Ok(()); } @@ -130,7 +175,7 @@ fn assert_supported_version(value: &Value) -> Result<(), ProtocolError> { if let Some(received) = received { return Err(ProtocolError::UnsupportedProtocolVersion { received, - supported: SUPPORTED_VERSIONS, + supported: supported.iter().map(|version| version.as_str()).collect(), }); } Ok(()) diff --git a/crates/oppa-protocol/src/document.rs b/crates/oppa-protocol/src/document.rs index 7309445..58bbb0b 100644 --- a/crates/oppa-protocol/src/document.rs +++ b/crates/oppa-protocol/src/document.rs @@ -236,6 +236,12 @@ impl JobDocument { Self::Pdf(_) => "pdf", } } + + /// Returns whether this document needs the extended protocol. + #[must_use] + pub const fn is_pdf(&self) -> bool { + matches!(self, Self::Pdf(_)) + } } impl Validate for JobDocument { diff --git a/crates/oppa-protocol/src/error.rs b/crates/oppa-protocol/src/error.rs index e8815aa..c944326 100644 --- a/crates/oppa-protocol/src/error.rs +++ b/crates/oppa-protocol/src/error.rs @@ -56,8 +56,8 @@ pub enum ProtocolError { UnsupportedProtocolVersion { /// Primitive version value rendered without the rest of the payload. received: String, - /// Versions understood by this crate. - supported: &'static [&'static str], + /// Versions understood by the decoder that rejected the envelope. + supported: Vec<&'static str>, }, /// Serialization failed after semantic validation. Encoding(String), diff --git a/crates/oppa-protocol/src/lib.rs b/crates/oppa-protocol/src/lib.rs index 1d45483..327b01f 100644 --- a/crates/oppa-protocol/src/lib.rs +++ b/crates/oppa-protocol/src/lib.rs @@ -13,6 +13,7 @@ mod document; mod error; mod job; mod message; +mod negotiation; mod pdf; mod printer; mod validation; @@ -25,8 +26,9 @@ pub use auth::{ PairingResponse, PublicEd25519Jwk, SIGNATURE_ALGORITHM, }; pub use codec::{ - decode_agent_message, decode_protocol_message, decode_server_message, encode_agent_message, - encode_protocol_message, encode_server_message, + decode_agent_message, decode_agent_message_for, decode_protocol_message, decode_server_message, + decode_server_message_for, encode_agent_message, encode_protocol_message, + encode_server_message, }; pub use document::{ BarcodeFormat, ImageMediaType, JobDocument, PrintDocument, PrintSection, ReceiptWidth, @@ -41,6 +43,10 @@ pub use message::{ OpenPrinterBrandMetadata, PrinterInventory, PrinterInventoryChanged, ProtocolMessage, ProtocolVersion, RequestPrinterInventory, ServerHello, ServerMessage, ServerMessageKind, }; +pub use negotiation::{ + CacheUpdate, HandshakePlanner, LEGACY_RETRY_INTERVAL_MS, LegacyServerRecord, project_inventory, + select_protocol_version, +}; pub use pdf::{ CustomPaperSize, Duplex, MAX_COPIES, MAX_CUSTOM_PAPER_MM, MAX_PAGE_RANGE_BYTES, MAX_SCALE_PERCENT, MIN_COPIES, MIN_CUSTOM_PAPER_MM, MIN_SCALE_PERCENT, Orientation, PaperSize, @@ -53,7 +59,8 @@ pub use printer::{ }; pub use validation::{Metadata, Validate}; -/// The only wire protocol version understood by this release. +/// The base wire protocol version. Every peer understands it, and it is the only version used by +/// authentication frames and discovery documents. pub const PROTOCOL_VERSION: &str = "1"; /// Maximum UTF-8 size accepted by any protocol decoder. diff --git a/crates/oppa-protocol/src/message.rs b/crates/oppa-protocol/src/message.rs index 2fbe5f2..26a7efe 100644 --- a/crates/oppa-protocol/src/message.rs +++ b/crates/oppa-protocol/src/message.rs @@ -3,30 +3,52 @@ use std::fmt; use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; use crate::{ - MAX_PRINTERS_PER_INVENTORY, MAX_WIRE_MESSAGE_BYTES, PROTOCOL_VERSION, PrintJob, + MAX_PRINTERS_PER_INVENTORY, MAX_WIRE_MESSAGE_BYTES, PrintJob, PrinterCapabilities, PrinterDescriptor, Validate, ValidationError, validation::{validate_brand_name, validate_identifier, validate_string, validate_timestamp}, }; const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991; -/// Validated OpenPrinter protocol version. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct ProtocolVersion; +/// OpenPrinter protocol version named in an envelope or a hello negotiation list. +/// +/// `V1` is the base protocol that every peer speaks. `V2` adds PDF print jobs and +/// printer document capabilities. Handshake frames always use `V1`, and every other +/// message uses the lowest version whose contract covers its content. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub enum ProtocolVersion { + /// The base protocol, identical to the original v1 wire contract. + V1, + /// The extended protocol: PDF print jobs and printer document capabilities. + V2, +} impl ProtocolVersion { - /// Current protocol version used by new messages. - pub const CURRENT: Self = Self; + /// Version used by handshake frames and by messages without extended content. + pub const BASE: Self = Self::V1; + /// Version required for extended content. Peers negotiate it during the hello exchange. + pub const EXTENDED: Self = Self::V2; + /// Every version this crate can decode. + pub const SUPPORTED: &'static [Self] = &[Self::V1, Self::V2]; /// Returns the stable string wire representation. - pub const fn get(self) -> &'static str { - PROTOCOL_VERSION + pub const fn as_str(self) -> &'static str { + match self { + Self::V1 => "1", + Self::V2 => "2", + } } } impl Default for ProtocolVersion { fn default() -> Self { - Self::CURRENT + Self::BASE + } +} + +impl fmt::Display for ProtocolVersion { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.as_str()) } } @@ -35,7 +57,7 @@ impl Serialize for ProtocolVersion { where S: Serializer, { - serializer.serialize_str(PROTOCOL_VERSION) + serializer.serialize_str(self.as_str()) } } @@ -50,17 +72,17 @@ impl<'de> Deserialize<'de> for ProtocolVersion { type Value = ProtocolVersion; fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(formatter, "protocol version {PROTOCOL_VERSION}") + formatter.write_str("protocol version 1 or 2") } fn visit_str(self, value: &str) -> Result where E: de::Error, { - if value == PROTOCOL_VERSION { - Ok(ProtocolVersion) - } else { - Err(E::custom(format!("unsupported protocol version {value}"))) + match value { + "1" => Ok(ProtocolVersion::V1), + "2" => Ok(ProtocolVersion::V2), + _ => Err(E::custom(format!("unsupported protocol version {value}"))), } } } @@ -69,6 +91,65 @@ impl<'de> Deserialize<'de> for ProtocolVersion { } } +/// Returns whether a printer descriptor carries any capability introduced by the extended protocol. +fn descriptor_uses_extended_protocol(printer: &PrinterDescriptor) -> bool { + printer + .capabilities + .as_ref() + .is_some_and(PrinterCapabilities::uses_extended_protocol) +} + +impl AgentMessageKind { + /// Returns the lowest protocol version whose contract covers this message. + /// + /// Envelopes should be labelled with this value, so receipts and reports stay on the base protocol. + #[must_use] + pub fn required_protocol_version(&self) -> ProtocolVersion { + required_version(self) + } +} + +impl ServerMessageKind { + /// Returns the lowest protocol version whose contract covers this message. + #[must_use] + pub fn required_protocol_version(&self) -> ProtocolVersion { + required_server_version(self) + } +} + +/// Lowest protocol version whose contract covers a message. Handshake frames, receipts, +/// reports, and base-protocol inventory need only `V1`. +pub(crate) fn required_version(kind: &AgentMessageKind) -> ProtocolVersion { + match kind { + AgentMessageKind::PrinterInventory(inventory) + if inventory + .printers + .iter() + .any(descriptor_uses_extended_protocol) => + { + ProtocolVersion::EXTENDED + } + AgentMessageKind::PrinterInventoryChanged(changed) + if changed + .added + .iter() + .chain(&changed.updated) + .any(descriptor_uses_extended_protocol) => + { + ProtocolVersion::EXTENDED + } + _ => ProtocolVersion::BASE, + } +} + +/// Lowest protocol version whose contract covers a server message. +pub(crate) fn required_server_version(kind: &ServerMessageKind) -> ProtocolVersion { + match kind { + ServerMessageKind::PrintJob(job) if job.document.is_pdf() => ProtocolVersion::EXTENDED, + _ => ProtocolVersion::BASE, + } +} + /// All lifecycle terms reserved by OpenPrinter v1. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] @@ -486,6 +567,11 @@ impl Validate for AgentMessage { } _ => {} } + validate_envelope_version( + self.protocol_version, + matches!(self.kind, AgentMessageKind::Hello(_)), + required_version(&self.kind), + )?; self.kind.validate() } } @@ -536,6 +622,15 @@ impl Validate for ServerHello { &self.supported_protocol_versions, "supportedProtocolVersions", )?; + if !self + .supported_protocol_versions + .contains(&self.selected_protocol_version) + { + return Err(ValidationError::new( + "selectedProtocolVersion", + "must be one of the listed supported versions", + )); + } if !(5_000..=300_000).contains(&self.heartbeat_interval_ms) { return Err(ValidationError::new( "heartbeatIntervalMs", @@ -770,6 +865,11 @@ impl Validate for ServerMessage { } (_, false) => {} } + validate_envelope_version( + self.protocol_version, + matches!(self.kind, ServerMessageKind::Hello(_)), + required_server_version(&self.kind), + )?; self.kind.validate() } } @@ -816,16 +916,43 @@ fn validate_envelope( } fn validate_versions(versions: &[ProtocolVersion], path: &str) -> Result<(), ValidationError> { - if versions.is_empty() || versions.len() > 8 { + if versions.is_empty() || versions.len() > ProtocolVersion::SUPPORTED.len() { return Err(ValidationError::new( path, - "must contain between 1 and 8 versions", + format!( + "must contain between 1 and {} versions", + ProtocolVersion::SUPPORTED.len() + ), + )); + } + for (index, version) in versions.iter().enumerate() { + if versions[..index].contains(version) { + return Err(ValidationError::new( + path, + "must not contain duplicate versions", + )); + } + } + Ok(()) +} + +/// Enforces the envelope-version rules: handshake frames use the base version, and content that +/// needs the extended protocol must not be labelled with the base version. +fn validate_envelope_version( + version: ProtocolVersion, + handshake: bool, + required: ProtocolVersion, +) -> Result<(), ValidationError> { + if handshake && version != ProtocolVersion::BASE { + return Err(ValidationError::new( + "protocolVersion", + "handshake frames must use protocol version 1", )); } - if versions.len() > 1 { + if version == ProtocolVersion::BASE && required == ProtocolVersion::EXTENDED { return Err(ValidationError::new( - path, - "must not contain duplicate versions", + "protocolVersion", + "this content requires protocol version 2", )); } Ok(()) diff --git a/crates/oppa-protocol/src/negotiation.rs b/crates/oppa-protocol/src/negotiation.rs new file mode 100644 index 0000000..fb8bf60 --- /dev/null +++ b/crates/oppa-protocol/src/negotiation.rs @@ -0,0 +1,276 @@ +//! Protocol version negotiation rules shared by agents, servers, and conformance tests. +//! +//! An extended agent offers `[V2, V1]`. A base-only server cannot read that hello, so it closes the +//! connection without replying. The agent then retries once with `[V1]` on a fresh connection. When +//! that retry succeeds, the agent remembers that the server is base-only for +//! [`LEGACY_RETRY_INTERVAL_MS`], so later connections skip the failed attempt. After the interval +//! the agent offers the extended protocol again, so a server upgrade is picked up without re-pairing. +//! +//! The planner never infers support from application or server version strings. It uses only the +//! explicit outcome of each hello exchange. + +use crate::{PrinterDescriptor, ProtocolVersion}; + +/// How long an agent keeps offering only the base protocol to a server that rejected the extended offer. +pub const LEGACY_RETRY_INTERVAL_MS: u64 = 24 * 60 * 60 * 1_000; + +/// Chooses the highest version both peers support. +/// +/// The choice is made from the server's own list, so a peer cannot select a version the server did +/// not enable. Returns `None` when the lists share no version. +#[must_use] +pub fn select_protocol_version( + server_supported: &[ProtocolVersion], + offered: &[ProtocolVersion], +) -> Option { + [ProtocolVersion::EXTENDED, ProtocolVersion::BASE] + .into_iter() + .find(|version| server_supported.contains(version) && offered.contains(version)) +} + +/// Returns the inventory a session that negotiated `version` may carry. +/// +/// Base-only sessions receive descriptors without extended capabilities, so the peer never sees +/// fields it cannot decode. +#[must_use] +pub fn project_inventory( + printers: &[PrinterDescriptor], + version: ProtocolVersion, +) -> Vec { + printers + .iter() + .map(|printer| match version { + ProtocolVersion::V1 => printer.without_extended_capabilities(), + ProtocolVersion::V2 => printer.clone(), + }) + .collect() +} + +/// The agent's stored knowledge that one server accepts only the base protocol. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct LegacyServerRecord { + /// Unix time in milliseconds after which the extended offer is tried again. + pub legacy_until_ms: u64, +} + +/// What the host should persist after a completed hello exchange. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CacheUpdate { + /// Keep the stored state as it is. + Unchanged, + /// Remember that this server accepted only the base protocol. + Store(LegacyServerRecord), + /// Forget any stored base-only record because the server accepted the extended protocol. + Clear, +} + +/// Plans the hello offers for one agent across connection attempts. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HandshakePlanner { + offered: Vec, + had_record: bool, + may_fall_back: bool, + fell_back: bool, +} + +impl HandshakePlanner { + /// Creates the plan for the first connection attempt. + /// + /// `record` is the stored base-only record for this server, if any. While it is current, the + /// agent offers only the base protocol and does not attempt the extended protocol. + #[must_use] + pub fn new(record: Option, now_ms: u64) -> Self { + let legacy_is_current = record.is_some_and(|record| record.legacy_until_ms > now_ms); + if legacy_is_current { + Self { + offered: vec![ProtocolVersion::BASE], + had_record: true, + may_fall_back: false, + fell_back: false, + } + } else { + Self { + offered: vec![ProtocolVersion::EXTENDED, ProtocolVersion::BASE], + had_record: record.is_some(), + may_fall_back: true, + fell_back: false, + } + } + } + + /// Versions the next hello should offer. + #[must_use] + pub fn offered(&self) -> &[ProtocolVersion] { + &self.offered + } + + /// Records that a connection ended after the hello was sent and before `server.hello`. + /// + /// Returns `true` when the caller should retry immediately with [`Self::offered`], which is the + /// base-only offer. Returns `false` when no fallback applies, and normal reconnect backoff should run. + /// A timeout waiting for `server.hello` is not a rejection and must not call this method. + pub fn on_closed_before_hello(&mut self) -> bool { + if !self.may_fall_back || self.fell_back { + return false; + } + self.offered = vec![ProtocolVersion::BASE]; + self.fell_back = true; + true + } + + /// Records the version selected by `server.hello` and returns what to persist. + #[must_use] + pub fn on_accepted(&self, selected: ProtocolVersion, now_ms: u64) -> CacheUpdate { + match selected { + ProtocolVersion::BASE if self.fell_back => CacheUpdate::Store(LegacyServerRecord { + legacy_until_ms: now_ms.saturating_add(LEGACY_RETRY_INTERVAL_MS), + }), + ProtocolVersion::EXTENDED if self.had_record => CacheUpdate::Clear, + _ => CacheUpdate::Unchanged, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{PrinterAvailability, PrinterCapabilities, PrinterConnection, PrinterKind}; + + const NOW: u64 = 1_800_000_000_000; + + fn printer(extended: bool) -> PrinterDescriptor { + PrinterDescriptor { + id: "printer_office_001".to_owned(), + fingerprint: "system:CUPS:office".to_owned(), + name: "Office".to_owned(), + kind: PrinterKind::Local, + connection: PrinterConnection::System { + system_name: "office".to_owned(), + }, + capabilities: Some(PrinterCapabilities { + media_widths: vec![crate::ReceiptWidth::Mm80], + system_driver: Some(true), + esc_pos: None, + raster: true, + cut: false, + qr: true, + barcode: true, + document_formats: extended.then(|| vec![crate::DocumentFormat::Receipt]), + paper_sizes: None, + custom_paper_size: None, + print_options: None, + }), + enabled: true, + availability: PrinterAvailability::Online, + } + } + + #[test] + fn selection_prefers_the_extended_version_only_when_both_peers_have_it() { + use ProtocolVersion::{V1 as BASE, V2 as EXTENDED}; + assert_eq!( + select_protocol_version(&[BASE, EXTENDED], &[EXTENDED, BASE]), + Some(EXTENDED) + ); + assert_eq!( + select_protocol_version(&[BASE], &[EXTENDED, BASE]), + Some(BASE) + ); + assert_eq!( + select_protocol_version(&[BASE, EXTENDED], &[BASE]), + Some(BASE) + ); + assert_eq!(select_protocol_version(&[EXTENDED], &[BASE]), None); + } + + #[test] + fn projection_removes_extended_fields_only_for_base_sessions() { + let printers = vec![printer(true)]; + let base = project_inventory(&printers, ProtocolVersion::BASE); + assert!( + base[0] + .capabilities + .as_ref() + .unwrap() + .document_formats + .is_none() + ); + let extended = project_inventory(&printers, ProtocolVersion::EXTENDED); + assert!( + extended[0] + .capabilities + .as_ref() + .unwrap() + .document_formats + .is_some() + ); + } + + #[test] + fn a_fresh_agent_offers_the_extended_protocol_first() { + let planner = HandshakePlanner::new(None, NOW); + assert_eq!( + planner.offered(), + &[ProtocolVersion::EXTENDED, ProtocolVersion::BASE] + ); + } + + #[test] + fn a_rejected_extended_offer_falls_back_once_and_is_remembered() { + let mut planner = HandshakePlanner::new(None, NOW); + assert!( + planner.on_closed_before_hello(), + "the first rejection retries at once" + ); + assert_eq!(planner.offered(), &[ProtocolVersion::BASE]); + assert!( + !planner.on_closed_before_hello(), + "a second failure uses normal backoff" + ); + + assert_eq!( + planner.on_accepted(ProtocolVersion::BASE, NOW), + CacheUpdate::Store(LegacyServerRecord { + legacy_until_ms: NOW + LEGACY_RETRY_INTERVAL_MS, + }) + ); + } + + #[test] + fn a_base_handshake_without_a_fallback_stores_nothing() { + let planner = HandshakePlanner::new(None, NOW); + assert_eq!( + planner.on_accepted(ProtocolVersion::BASE, NOW), + CacheUpdate::Unchanged + ); + } + + #[test] + fn a_current_legacy_record_skips_the_extended_offer() { + let record = LegacyServerRecord { + legacy_until_ms: NOW + 1_000, + }; + let mut planner = HandshakePlanner::new(Some(record), NOW); + assert_eq!(planner.offered(), &[ProtocolVersion::BASE]); + assert!( + !planner.on_closed_before_hello(), + "no extended attempt means no fallback" + ); + } + + #[test] + fn an_expired_legacy_record_retries_the_extended_protocol() { + let record = LegacyServerRecord { + legacy_until_ms: NOW - 1, + }; + let planner = HandshakePlanner::new(Some(record), NOW); + assert_eq!( + planner.offered(), + &[ProtocolVersion::EXTENDED, ProtocolVersion::BASE] + ); + assert_eq!( + planner.on_accepted(ProtocolVersion::EXTENDED, NOW), + CacheUpdate::Clear + ); + } +} diff --git a/crates/oppa-protocol/src/printer.rs b/crates/oppa-protocol/src/printer.rs index 934bfff..8b593a9 100644 --- a/crates/oppa-protocol/src/printer.rs +++ b/crates/oppa-protocol/src/printer.rs @@ -127,6 +127,35 @@ pub struct PrintOptionsCapability { pub page_ranges: bool, } +impl PrinterCapabilities { + /// Returns whether any capability introduced by the extended protocol is present. + #[must_use] + pub const fn uses_extended_protocol(&self) -> bool { + self.document_formats.is_some() + || self.paper_sizes.is_some() + || self.custom_paper_size.is_some() + || self.print_options.is_some() + } +} + +impl PrinterDescriptor { + /// Returns a copy without extended-protocol capabilities. + /// + /// A session that negotiated only the base protocol receives inventory projected this way, so + /// the peer never sees fields it cannot decode. + #[must_use] + pub fn without_extended_capabilities(&self) -> Self { + let mut projected = self.clone(); + if let Some(capabilities) = projected.capabilities.as_mut() { + capabilities.document_formats = None; + capabilities.paper_sizes = None; + capabilities.custom_paper_size = None; + capabilities.print_options = None; + } + projected + } +} + impl Validate for PrinterCapabilities { fn validate(&self) -> Result<(), ValidationError> { if self.media_widths.is_empty() || self.media_widths.len() > 2 { diff --git a/crates/oppa-protocol/tests/documents.rs b/crates/oppa-protocol/tests/documents.rs index a0975ef..8072ad6 100644 --- a/crates/oppa-protocol/tests/documents.rs +++ b/crates/oppa-protocol/tests/documents.rs @@ -9,9 +9,15 @@ use serde_json::{Value, json}; const SHA256: &str = "9f2c7e1a4b8d03e5f6a7b8c9d0e1f2a34b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e"; +/// Wraps a document in a print job, labelled with the lowest protocol version that covers it. fn job_with_document(document: Value) -> Value { + let version = if document.get("format").is_some() { + "2" + } else { + "1" + }; json!({ - "protocolVersion": "1", + "protocolVersion": version, "messageId": "msg_doc_001", "sentAt": "2026-07-28T10:02:00Z", "type": "server.print_job", diff --git a/crates/oppa-protocol/tests/fixtures.rs b/crates/oppa-protocol/tests/fixtures.rs index 3044693..423e7c4 100644 --- a/crates/oppa-protocol/tests/fixtures.rs +++ b/crates/oppa-protocol/tests/fixtures.rs @@ -146,8 +146,8 @@ fn every_invalid_fixture_is_rejected() { result, Err(ProtocolError::UnsupportedProtocolVersion { received, - supported: &["1"], - }) if received == "99" + supported, + }) if received == "99" && supported == ["1", "2"] ), "{} must produce a version error", path.display(), diff --git a/crates/oppa-protocol/tests/negotiation.rs b/crates/oppa-protocol/tests/negotiation.rs new file mode 100644 index 0000000..f738fac --- /dev/null +++ b/crates/oppa-protocol/tests/negotiation.rs @@ -0,0 +1,312 @@ +//! Compatibility tests for protocol negotiation between an agent and each server generation. +//! +//! Peers are modelled with the same codecs they use in production. A base-only server decodes with +//! `decode_*_for(&[ProtocolVersion::BASE])`, which matches the original v1 decoder. A new server +//! decodes with the default decoder, which accepts every version. + +use std::fs; + +use oppa_protocol::{ + AgentHello, AgentMessage, AgentMessageKind, CacheUpdate, HandshakePlanner, LegacyServerRecord, + OpenPrinterBrandMetadata, PrinterInventory, ProtocolError, ProtocolVersion, ServerHello, + ServerMessage, ServerMessageKind, decode_agent_message, decode_agent_message_for, + decode_server_message_for, encode_agent_message, encode_server_message, project_inventory, + select_protocol_version, +}; + +const NOW_MS: u64 = 1_800_000_000_000; +const BASE: ProtocolVersion = ProtocolVersion::V1; +const EXTENDED: ProtocolVersion = ProtocolVersion::V2; + +fn fixture(path: &str) -> Vec { + let full = format!( + "{}/../../protocol/fixtures/{path}", + env!("CARGO_MANIFEST_DIR") + ); + fs::read(&full).unwrap_or_else(|error| panic!("{full} must be readable: {error}")) +} + +fn agent_hello(offered: Vec) -> AgentMessage { + AgentMessage { + protocol_version: BASE, + message_id: "msg_agent_hello_scenario".to_owned(), + sent_at: "2026-08-01T09:00:00Z".to_owned(), + correlation_id: None, + kind: AgentMessageKind::Hello(AgentHello { + agent_id: "agent_demo_001".to_owned(), + agent_version: "1.0.0".to_owned(), + product_id: "oppa.default".to_owned(), + product_version: "1.0.0".to_owned(), + supported_protocol_versions: offered, + }), + } +} + +/// What a server replies to a hello once it has selected a version. +fn server_hello(selected: ProtocolVersion) -> ServerMessage { + ServerMessage { + protocol_version: BASE, + message_id: "msg_server_hello_scenario".to_owned(), + sent_at: "2026-08-01T09:00:01Z".to_owned(), + correlation_id: Some("msg_agent_hello_scenario".to_owned()), + kind: ServerMessageKind::Hello(ServerHello { + server_id: "server_example_001".to_owned(), + server_version: "1.0.0".to_owned(), + brand: OpenPrinterBrandMetadata { + name: "Scenario service".to_owned(), + }, + session_id: "session_scenario".to_owned(), + supported_protocol_versions: vec![selected], + selected_protocol_version: selected, + heartbeat_interval_ms: 30_000, + max_message_bytes: 2_097_152, + }), + } +} + +/// The original, base-protocol inventory fixture: a receipt printer without extended fields. +fn base_inventory() -> AgentMessage { + decode_agent_message(&fixture("agent/printer-inventory.json")).expect("base inventory fixture") +} + +/// A printer with extended document capabilities, carried in an extended-version inventory. +fn extended_inventory() -> AgentMessage { + decode_agent_message(&fixture("agent/printer-inventory-documents.json")) + .expect("extended inventory fixture") +} + +/// Outcome of one connection attempt from the agent's perspective. +enum Attempt { + /// The server replied with `server.hello` carrying the selected version. + Accepted(ProtocolVersion), + /// The server closed the connection before replying. + ClosedBeforeHello, +} + +/// A base-only server: decodes with the original v1 rules and closes on any rejected hello. +fn base_only_server(hello_bytes: &[u8]) -> Attempt { + match decode_agent_message_for(hello_bytes, &[BASE]) { + Ok(hello) => { + let AgentMessageKind::Hello(hello) = hello.kind else { + panic!("the hello fixture must be a hello"); + }; + let selected = select_protocol_version(&[BASE], &hello.supported_protocol_versions) + .expect("base-only server accepts a base offer"); + Attempt::Accepted(selected) + } + Err(_) => Attempt::ClosedBeforeHello, + } +} + +/// A new server: accepts every version this crate implements. +fn new_server(hello_bytes: &[u8]) -> Attempt { + let hello = decode_agent_message(hello_bytes).expect("new server decodes every valid hello"); + let AgentMessageKind::Hello(hello) = hello.kind else { + panic!("the hello must be a hello"); + }; + let selected = select_protocol_version( + ProtocolVersion::SUPPORTED, + &hello.supported_protocol_versions, + ) + .expect("new server shares a version with every agent"); + Attempt::Accepted(selected) +} + +fn encode_hello(message: &AgentMessage) -> Vec { + encode_agent_message(message).expect("agent hello encodes") +} + +/// Runs the agent's connection loop against a server, following the planner's fallback rule. +fn connect( + planner: &mut HandshakePlanner, + server: impl Fn(&[u8]) -> Attempt, +) -> (ProtocolVersion, Vec) { + let mut attempts = Vec::new(); + loop { + let hello = encode_hello(&agent_hello(planner.offered().to_vec())); + let outcome = server(&hello); + let closed = matches!(outcome, Attempt::ClosedBeforeHello); + attempts.push(outcome); + if closed && planner.on_closed_before_hello() { + continue; + } + let Some(Attempt::Accepted(selected)) = attempts.last() else { + panic!("the connection ended and no fallback applies"); + }; + return (*selected, attempts); + } +} + +#[test] +fn old_agent_and_new_server_negotiate_the_base_protocol_with_the_original_handshake() { + // An original agent offers only the base protocol, exactly as its fixture does. + let hello = fixture("agent/agent-hello.json"); + let Attempt::Accepted(selected) = new_server(&hello) else { + panic!("a new server accepts an original agent"); + }; + assert_eq!(selected, BASE); + + // The reply lists only the selected version, which base-only agents can decode. + let reply = encode_server_message(&server_hello(selected)).expect("server hello encodes"); + let decoded = decode_server_message_for(&reply, &[BASE]).expect("base agent decodes the reply"); + let ServerMessageKind::Hello(decoded) = decoded.kind else { + panic!("reply must be a hello"); + }; + assert_eq!(decoded.supported_protocol_versions, vec![BASE]); +} + +#[test] +fn old_agent_inventory_stays_byte_compatible_with_the_base_protocol() { + let inventory = base_inventory(); + assert_eq!(inventory.protocol_version, BASE); + let encoded = encode_agent_message(&inventory).expect("base inventory encodes"); + let decoded_by_base_server = decode_agent_message_for(&encoded, &[BASE]) + .expect("a base-only server accepts the original inventory"); + assert_eq!(decoded_by_base_server, inventory); +} + +#[test] +fn new_agent_and_new_server_negotiate_the_extended_protocol_and_exchange_document_capabilities() { + let mut planner = HandshakePlanner::new(None, NOW_MS); + let (selected, attempts) = connect(&mut planner, new_server); + assert_eq!(selected, EXTENDED); + assert_eq!(attempts.len(), 1, "no fallback is needed"); + + // Extended inventory is labelled with the extended version and accepted by the new server. + let inventory = extended_inventory(); + assert_eq!(inventory.protocol_version, EXTENDED); + let encoded = encode_agent_message(&inventory).expect("extended inventory encodes"); + let decoded = decode_agent_message(&encoded).expect("new server accepts extended inventory"); + assert_eq!(decoded, inventory); + + // The same inventory cannot be labelled with the base version, because it needs the extension. + let mut mislabelled = inventory.clone(); + mislabelled.protocol_version = BASE; + assert!(matches!( + encode_agent_message(&mislabelled), + Err(ProtocolError::InvalidMessage(_)) + )); +} + +#[test] +fn new_agent_with_a_base_only_server_falls_back_once_and_then_completes_the_base_handshake() { + let mut planner = HandshakePlanner::new(None, NOW_MS); + let (selected, attempts) = connect(&mut planner, base_only_server); + + assert_eq!(selected, BASE); + assert!( + matches!(attempts[0], Attempt::ClosedBeforeHello), + "the extended offer is refused" + ); + assert!( + matches!(attempts[1], Attempt::Accepted(BASE)), + "the base retry completes" + ); + + // The agent remembers the server for the next connections. + assert_eq!( + planner.on_accepted(selected, NOW_MS), + CacheUpdate::Store(LegacyServerRecord { + legacy_until_ms: NOW_MS + oppa_protocol::LEGACY_RETRY_INTERVAL_MS, + }) + ); +} + +#[test] +fn a_remembered_base_only_server_is_not_probed_again_until_the_interval_expires() { + let record = LegacyServerRecord { + legacy_until_ms: NOW_MS + 60_000, + }; + let mut planner = HandshakePlanner::new(Some(record), NOW_MS); + let (selected, attempts) = connect(&mut planner, base_only_server); + assert_eq!(selected, BASE); + assert_eq!( + attempts.len(), + 1, + "no extended attempt is sent to a remembered server" + ); + + let mut after_interval = HandshakePlanner::new(Some(record), NOW_MS + 61_000); + let (_, retried) = connect(&mut after_interval, base_only_server); + assert!( + matches!(retried[0], Attempt::ClosedBeforeHello), + "the extended offer is tried again" + ); +} + +#[test] +fn a_base_only_session_never_carries_extended_fields() { + let printers = match extended_inventory().kind { + AgentMessageKind::PrinterInventory(PrinterInventory { printers, .. }) => printers, + other => panic!("unexpected inventory kind {other:?}"), + }; + let projected = project_inventory(&printers, BASE); + let inventory = AgentMessage { + protocol_version: BASE, + message_id: "msg_projected_inventory".to_owned(), + sent_at: "2026-08-01T09:02:00Z".to_owned(), + correlation_id: None, + kind: AgentMessageKind::PrinterInventory(PrinterInventory { + revision: 8, + printers: projected, + }), + }; + + let encoded = encode_agent_message(&inventory).expect("projected inventory encodes"); + let text = String::from_utf8(encoded.clone()).expect("UTF-8"); + for field in [ + "documentFormats", + "paperSizes", + "customPaperSize", + "printOptions", + ] { + assert!( + !text.contains(field), + "projected inventory must not contain {field}" + ); + } + decode_agent_message_for(&encoded, &[BASE]).expect("base-only server accepts the projection"); +} + +#[test] +fn a_base_only_server_rejects_extended_content_even_under_an_extended_envelope() { + let encoded = encode_agent_message(&extended_inventory()).expect("extended inventory encodes"); + assert!(matches!( + decode_agent_message_for(&encoded, &[BASE]), + Err(ProtocolError::UnsupportedProtocolVersion { .. }) + )); +} + +#[test] +fn servers_cannot_negotiate_a_version_the_agent_does_not_offer() { + let selected = select_protocol_version(ProtocolVersion::SUPPORTED, &[BASE]); + assert_eq!(selected, Some(BASE)); + assert_eq!(select_protocol_version(&[EXTENDED], &[BASE]), None); +} + +#[test] +fn new_server_replies_are_rejected_by_a_base_agent_when_they_select_the_extended_version() { + // Only a version the agent offered may be selected. A base-only agent never offers the extended version. + let reply = encode_server_message(&server_hello(EXTENDED)).expect("extended reply encodes"); + assert!(decode_server_message_for(&reply, &[BASE]).is_err()); +} + +#[test] +fn the_planner_and_cache_update_agree_with_the_selection_rule() { + let mut planner = HandshakePlanner::new(None, NOW_MS); + let (selected, _) = connect(&mut planner, new_server); + assert_eq!( + planner.on_accepted(selected, NOW_MS), + CacheUpdate::Unchanged + ); +} + +#[test] +fn an_extended_hello_round_trips_through_the_default_decoder() { + let hello = agent_hello(vec![EXTENDED, BASE]); + let encoded = encode_hello(&hello); + assert_eq!( + decode_agent_message(&encoded).expect("hello decodes"), + hello + ); +} diff --git a/packages/protocol/AGENTS.md b/packages/protocol/AGENTS.md index 9bb0d25..dc7b9d8 100644 --- a/packages/protocol/AGENTS.md +++ b/packages/protocol/AGENTS.md @@ -12,7 +12,13 @@ must remain bounded, and codecs must reject unsupported versions before routing. Never add secrets, local filesystem paths, commands, scripts, or generic proxy requests to the protocol. -Protocol version `"1"` is a string. Discovery, pairing, and gateway +Protocol version `"1"` is the base protocol. Version `"2"` adds PDF +jobs and printer document capabilities, and peers negotiate it during +the hello exchange. Handshake frames always use envelope version +`"1"`. Other envelopes use the lowest version their content needs, so +receipts stay byte-compatible with base-protocol peers. Keep +`requiredProtocolVersion` in `src/versioning.ts` in sync with the Rust +`required_protocol_version` methods. Discovery, pairing, and gateway authentication schemas are part of the canonical contract. The protocol may model a public Ed25519 JWK and opaque challenge frames, but private-key generation, storage, and signing belong outside this diff --git a/packages/protocol/src/codec.ts b/packages/protocol/src/codec.ts index 6e88342..8431459 100644 --- a/packages/protocol/src/codec.ts +++ b/packages/protocol/src/codec.ts @@ -1,7 +1,7 @@ import type { TSchema } from '@sinclair/typebox'; import { Value } from '@sinclair/typebox/value'; -import { MAX_AUTH_MESSAGE_BYTES, MAX_WIRE_MESSAGE_BYTES, PROTOCOL_VERSION } from './constants.js'; +import { MAX_AUTH_MESSAGE_BYTES, MAX_WIRE_MESSAGE_BYTES, PROTOCOL_VERSION, PROTOCOL_VERSIONS } from './constants.js'; import { ProtocolError, ProtocolValidationError, @@ -35,6 +35,7 @@ import { type ServerMessage, } from './schemas/messages.js'; import { PrinterDescriptorSchema, type PrinterDescriptor } from './schemas/printer.js'; +import { requiredProtocolVersion, type ProtocolVersionName } from './versioning.js'; const encoder = new TextEncoder(); const decoder = new TextDecoder('utf-8', { fatal: true }); @@ -53,23 +54,74 @@ function normalizedIssues(schema: TSchema, value: unknown): ProtocolIssue[] { return issues; } -function assertSupportedVersion(value: unknown): void { +/** Options that restrict decoding to the protocol versions a peer understands. */ +export interface ProtocolDecodeOptions { + /** + * Versions this peer can read. Defaults to every version this package supports. + * + * A peer restricted to `["1"]` rejects any hello offering `"2"`, which is how base-protocol + * servers behave. Negotiating clients use this to test that legacy peers reject offers they cannot read. + */ + readonly supportedVersions?: readonly ProtocolVersionName[]; +} + +function supportedVersionsOf(options: ProtocolDecodeOptions | undefined): readonly ProtocolVersionName[] { + return options?.supportedVersions ?? PROTOCOL_VERSIONS; +} + +function assertKnownVersion(value: unknown, supported: readonly ProtocolVersionName[]): void { if (typeof value !== 'object' || value === null || !Object.prototype.hasOwnProperty.call(value, 'protocolVersion')) { return; } - const version = (value as { protocolVersion?: unknown }).protocolVersion; if ( - version !== PROTOCOL_VERSION && + !supported.includes(version as ProtocolVersionName) && (version === null || typeof version === 'string' || typeof version === 'number' || typeof version === 'boolean') ) { - throw new UnsupportedProtocolVersionError(version); + throw new UnsupportedProtocolVersionError(version, supported); + } +} + +/** + * Enforces the envelope-version rules after schema validation. + * + * Handshake frames must use the base version and may list only versions the peer supports. Other + * messages must use a version the peer supports, and a message that needs the extended protocol + * must not be labelled with the base version. + */ +function assertVersionRules( + message: { type: string; protocolVersion: string; payload: unknown }, + supported: readonly ProtocolVersionName[], +): void { + if (!supported.includes(message.protocolVersion as ProtocolVersionName)) { + throw new UnsupportedProtocolVersionError(message.protocolVersion, supported); + } + if (message.type === 'agent.hello' || message.type === 'server.hello') { + if (message.protocolVersion !== PROTOCOL_VERSION) { + throw new ProtocolValidationError('Invalid handshake', [ + { path: '/protocolVersion', message: 'handshake frames must use protocol version 1' }, + ]); + } + const payload = message.payload as { supportedProtocolVersions: string[]; selectedProtocolVersion?: string }; + const listed = [...payload.supportedProtocolVersions, payload.selectedProtocolVersion ?? PROTOCOL_VERSION]; + if (listed.some((version) => !supported.includes(version as ProtocolVersionName))) { + throw new ProtocolValidationError('Invalid handshake', [ + { path: '/payload/supportedProtocolVersions', message: 'lists a protocol version this peer does not support' }, + ]); + } + return; + } + const required = requiredProtocolVersion(message); + if (required !== message.protocolVersion && message.protocolVersion === PROTOCOL_VERSION) { + throw new ProtocolValidationError('Invalid message', [ + { path: '/protocolVersion', message: `this content requires protocol version ${required}` }, + ]); } } function parseWithSchema(schema: TSchema, value: unknown, description: string, checkVersion = false): T { if (checkVersion) { - assertSupportedVersion(value); + assertKnownVersion(value, PROTOCOL_VERSIONS); } if (!Value.Check(schema, value)) { throw new ProtocolValidationError(`Invalid ${description}`, normalizedIssues(schema, value)); @@ -77,6 +129,20 @@ function parseWithSchema(schema: TSchema, value: unknown, description: string return value as T; } +/** Validates a message envelope against its schema and the version rules for the peer's supported versions. */ +function parseMessage( + schema: TSchema, + value: unknown, + description: string, + options: ProtocolDecodeOptions | undefined, +): T { + const supported = supportedVersionsOf(options); + assertKnownVersion(value, supported); + const message = parseWithSchema(schema, value, description); + assertVersionRules(message, supported); + return message; +} + function byteLength(value: string): number { return encoder.encode(value).byteLength; } @@ -136,22 +202,22 @@ function encodeJson(message: T, validate: (value: unknown) => T): string { } /** Validates an unknown value as an agent-to-server message. */ -export function parseAgentMessage(value: unknown): AgentMessage { - return parseWithSchema(AgentMessageSchema, value, 'agent message', true); +export function parseAgentMessage(value: unknown, options?: ProtocolDecodeOptions): AgentMessage { + return parseMessage(AgentMessageSchema, value, 'agent message', options); } /** Validates an unknown value as a server-to-agent message. */ -export function parseServerMessage(value: unknown): ServerMessage { - return parseWithSchema(ServerMessageSchema, value, 'server message', true); +export function parseServerMessage(value: unknown, options?: ProtocolDecodeOptions): ServerMessage { + return parseMessage(ServerMessageSchema, value, 'server message', options); } /** Validates an unknown value as either OpenPrinter message direction. */ -export function parseProtocolMessage(value: unknown): OpenPrinterProtocolMessage { - return parseWithSchema( +export function parseProtocolMessage(value: unknown, options?: ProtocolDecodeOptions): OpenPrinterProtocolMessage { + return parseMessage( OpenPrinterProtocolSchema, value, 'OpenPrinter protocol message', - true, + options, ); } @@ -236,18 +302,21 @@ export function parseGatewayAuthenticationServerMessage(value: unknown): Gateway } /** Decodes UTF-8 JSON and validates an agent-to-server message. */ -export function decodeAgentMessage(input: string | Uint8Array): AgentMessage { - return parseAgentMessage(decodeJson(input)); +export function decodeAgentMessage(input: string | Uint8Array, options?: ProtocolDecodeOptions): AgentMessage { + return parseAgentMessage(decodeJson(input), options); } /** Decodes UTF-8 JSON and validates a server-to-agent message. */ -export function decodeServerMessage(input: string | Uint8Array): ServerMessage { - return parseServerMessage(decodeJson(input)); +export function decodeServerMessage(input: string | Uint8Array, options?: ProtocolDecodeOptions): ServerMessage { + return parseServerMessage(decodeJson(input), options); } /** Decodes UTF-8 JSON and validates either protocol message direction. */ -export function decodeProtocolMessage(input: string | Uint8Array): OpenPrinterProtocolMessage { - return parseProtocolMessage(decodeJson(input)); +export function decodeProtocolMessage( + input: string | Uint8Array, + options?: ProtocolDecodeOptions, +): OpenPrinterProtocolMessage { + return parseProtocolMessage(decodeJson(input), options); } /** Decodes and validates one bounded client authentication frame. */ diff --git a/packages/protocol/src/constants.ts b/packages/protocol/src/constants.ts index e4d0b2e..7a7ac81 100644 --- a/packages/protocol/src/constants.ts +++ b/packages/protocol/src/constants.ts @@ -6,6 +6,17 @@ */ export const PROTOCOL_VERSION = '1' as const; +/** + * Extended protocol version. It adds PDF print jobs and printer document capabilities. + * + * Peers negotiate it during the hello exchange. A `"2"` envelope is valid only in a session that + * negotiated it, and handshake frames always use `"1"` so every peer can read them. + */ +export const EXTENDED_PROTOCOL_VERSION = '2' as const; + +/** Every protocol version this package can decode, in ascending order. */ +export const PROTOCOL_VERSIONS = [PROTOCOL_VERSION, EXTENDED_PROTOCOL_VERSION] as const; + /** Default discovery endpoint relative to an OpenPrinter server base URL. */ export const OPENPRINTER_DISCOVERY_PATH = '/.well-known/openprinter' as const; diff --git a/packages/protocol/src/errors.ts b/packages/protocol/src/errors.ts index 6bcd0bc..f671b2a 100644 --- a/packages/protocol/src/errors.ts +++ b/packages/protocol/src/errors.ts @@ -52,14 +52,17 @@ export class UnsupportedProtocolVersionError extends ProtocolError { readonly supportedVersions: readonly string[]; /** Creates an explicit version-negotiation failure. */ - constructor(receivedVersion: string | number | boolean | null) { + constructor( + receivedVersion: string | number | boolean | null, + supportedVersions: readonly string[] = PROTOCOL_VERSIONS, + ) { super( 'unsupported_protocol_version', - `Unsupported protocol version ${String(receivedVersion)}; supported versions: ${PROTOCOL_VERSION}`, + `Unsupported protocol version ${String(receivedVersion)}; supported versions: ${supportedVersions.join(', ')}`, ); this.name = 'UnsupportedProtocolVersionError'; this.receivedVersion = receivedVersion; - this.supportedVersions = [PROTOCOL_VERSION]; + this.supportedVersions = [...supportedVersions]; } } -import { PROTOCOL_VERSION } from './constants.js'; +import { PROTOCOL_VERSIONS } from './constants.js'; diff --git a/packages/protocol/src/index.ts b/packages/protocol/src/index.ts index 9f30937..de35f6d 100644 --- a/packages/protocol/src/index.ts +++ b/packages/protocol/src/index.ts @@ -18,3 +18,5 @@ export * from './schemas/job.js'; export * from './schemas/messages.js'; /** Printer connection, capability, and descriptor schemas and types. */ export * from './schemas/printer.js'; +/** Protocol version negotiation rules and projections shared by both peers. */ +export * from './versioning.js'; diff --git a/packages/protocol/src/schemas/common.ts b/packages/protocol/src/schemas/common.ts index fdabbd2..88da8bd 100644 --- a/packages/protocol/src/schemas/common.ts +++ b/packages/protocol/src/schemas/common.ts @@ -1,6 +1,6 @@ import { Type, type Static } from '@sinclair/typebox'; -import { MAX_METADATA_ENTRIES, PROTOCOL_VERSION } from '../constants.js'; +import { MAX_METADATA_ENTRIES, PROTOCOL_VERSIONS } from '../constants.js'; /** A stable opaque identifier used on the wire. */ export const IdentifierSchema = Type.String({ @@ -62,8 +62,8 @@ export const JobStatusSchema = Type.Union([ /** A lifecycle status that never implies verified physical output. */ export type JobStatus = Static; -/** The singleton protocol version schema used by every v1 envelope. */ -export const ProtocolVersionSchema = Type.Literal(PROTOCOL_VERSION); +/** A protocol version named in an envelope or a hello negotiation list. */ +export const ProtocolVersionSchema = Type.Union(PROTOCOL_VERSIONS.map((version) => Type.Literal(version))); /** TypeScript representation of the current protocol version. */ export type ProtocolVersion = Static; diff --git a/packages/protocol/src/versioning.ts b/packages/protocol/src/versioning.ts new file mode 100644 index 0000000..8480cb4 --- /dev/null +++ b/packages/protocol/src/versioning.ts @@ -0,0 +1,86 @@ +import { EXTENDED_PROTOCOL_VERSION, PROTOCOL_VERSION } from './constants.js'; +import type { PROTOCOL_VERSIONS } from './constants.js'; +import type { PrinterCapabilities, PrinterDescriptor } from './schemas/printer.js'; + +/** A protocol version string understood by this package. */ +export type ProtocolVersionName = (typeof PROTOCOL_VERSIONS)[number]; + +/** Message types exchanged before a protocol version is selected. They always use the base version. */ +const HANDSHAKE_TYPES: ReadonlySet = new Set(['agent.hello', 'server.hello']); + +/** Printer capability keys introduced by the extended protocol. */ +const EXTENDED_CAPABILITY_KEYS = ['documentFormats', 'paperSizes', 'customPaperSize', 'printOptions'] as const; + +/** Minimal shape shared by every envelope, used to decide which version a message needs. */ +interface VersionedEnvelope { + readonly type: string; + readonly payload: unknown; +} + +/** Returns whether a printer descriptor advertises any capability introduced by the extended protocol. */ +export function hasExtendedCapabilities(printer: PrinterDescriptor): boolean { + const capabilities = printer.capabilities as Partial | undefined; + if (capabilities === undefined) { + return false; + } + return EXTENDED_CAPABILITY_KEYS.some((key) => capabilities[key] !== undefined); +} + +/** + * Returns the lowest protocol version whose contract covers a message. + * + * Handshake frames and every message that uses only base-protocol content return `"1"`. A PDF print + * job or a printer descriptor with extended capabilities requires `"2"`. Peers use this rule to + * choose envelope versions, so receipts and reports remain byte-identical to the base protocol. + */ +export function requiredProtocolVersion(message: VersionedEnvelope): ProtocolVersionName { + if (HANDSHAKE_TYPES.has(message.type)) { + return PROTOCOL_VERSION; + } + if (message.type === 'server.print_job') { + const document = (message.payload as { document?: { format?: string } }).document; + return document?.format === 'pdf' ? EXTENDED_PROTOCOL_VERSION : PROTOCOL_VERSION; + } + if (message.type === 'agent.printer_inventory') { + const printers = (message.payload as { printers: PrinterDescriptor[] }).printers; + return printers.some(hasExtendedCapabilities) ? EXTENDED_PROTOCOL_VERSION : PROTOCOL_VERSION; + } + if (message.type === 'agent.printer_inventory_changed') { + const { added, updated } = message.payload as { added: PrinterDescriptor[]; updated: PrinterDescriptor[] }; + return [...added, ...updated].some(hasExtendedCapabilities) ? EXTENDED_PROTOCOL_VERSION : PROTOCOL_VERSION; + } + return PROTOCOL_VERSION; +} + +/** + * Selects the highest protocol version both peers support. + * + * Returns `null` when the lists share no version. The selection is always taken from the server's + * own list, so a peer cannot negotiate a version the server did not enable. + */ +export function selectProtocolVersion( + serverSupported: readonly ProtocolVersionName[], + offered: readonly ProtocolVersionName[], +): ProtocolVersionName | null { + if (serverSupported.includes(EXTENDED_PROTOCOL_VERSION) && offered.includes(EXTENDED_PROTOCOL_VERSION)) { + return EXTENDED_PROTOCOL_VERSION; + } + return serverSupported.includes(PROTOCOL_VERSION) && offered.includes(PROTOCOL_VERSION) ? PROTOCOL_VERSION : null; +} + +/** + * Removes capabilities introduced by the extended protocol. + * + * A session that negotiated only the base protocol receives printer descriptors projected this way, + * so it never sees fields its peer cannot decode. + */ +export function withoutExtendedCapabilities(printer: PrinterDescriptor): PrinterDescriptor { + if (!hasExtendedCapabilities(printer) || printer.capabilities === undefined) { + return printer; + } + const capabilities: Record = { ...printer.capabilities }; + for (const key of EXTENDED_CAPABILITY_KEYS) { + delete capabilities[key]; + } + return { ...printer, capabilities: capabilities as PrinterCapabilities }; +} diff --git a/packages/protocol/test/protocol.test.ts b/packages/protocol/test/protocol.test.ts index 41672c4..b6dd3bf 100644 --- a/packages/protocol/test/protocol.test.ts +++ b/packages/protocol/test/protocol.test.ts @@ -328,7 +328,7 @@ describe('runtime validation', () => { expect(error).toMatchObject({ code: 'unsupported_protocol_version', receivedVersion: '99', - supportedVersions: ['1'], + supportedVersions: ['1', '2'], }); expect(JSON.stringify(error)).not.toContain('server.heartbeat'); } diff --git a/packages/protocol/test/versioning.test.ts b/packages/protocol/test/versioning.test.ts new file mode 100644 index 0000000..9052a56 --- /dev/null +++ b/packages/protocol/test/versioning.test.ts @@ -0,0 +1,155 @@ +import { readFileSync } from 'node:fs'; + +import { describe, expect, it } from 'vitest'; + +import { + EXTENDED_PROTOCOL_VERSION, + PROTOCOL_VERSION, + ProtocolValidationError, + UnsupportedProtocolVersionError, + decodeAgentMessage, + decodeServerMessage, + encodeAgentMessage, + requiredProtocolVersion, + selectProtocolVersion, + withoutExtendedCapabilities, + type PrinterDescriptor, +} from '../src/index.js'; + +const fixtureRoot = new URL('../../../protocol/fixtures/', import.meta.url); + +function fixture(path: string): Record { + return JSON.parse(readFileSync(new URL(path, fixtureRoot), 'utf8')) as Record; +} + +function withEnvelope(message: Record, protocolVersion: string): string { + return JSON.stringify({ ...message, protocolVersion }); +} + +const receiptPrinter: PrinterDescriptor = { + id: 'printer_receipt_001', + fingerprint: 'system:CUPS:receipt-001', + name: 'Front desk receipt', + kind: 'local', + connection: { type: 'system', systemName: 'receipt-001' }, + capabilities: { mediaWidths: [80], raster: false, cut: true, qr: true, barcode: true }, + enabled: true, + availability: 'online', +}; + +const officePrinter: PrinterDescriptor = { + ...receiptPrinter, + id: 'printer_office_001', + capabilities: { + mediaWidths: [80], + raster: true, + cut: false, + qr: true, + barcode: true, + documentFormats: ['receipt', 'pdf'], + }, +}; + +describe('required protocol version', () => { + it('uses the base version for handshake frames, receipts, and base inventory', () => { + expect(requiredProtocolVersion(fixture('agent/agent-hello-extended-offer.json') as never)).toBe(PROTOCOL_VERSION); + expect(requiredProtocolVersion(fixture('server/server-hello-extended.json') as never)).toBe(PROTOCOL_VERSION); + expect(requiredProtocolVersion(fixture('server/print-job.json') as never)).toBe(PROTOCOL_VERSION); + expect( + requiredProtocolVersion({ + type: 'agent.printer_inventory', + payload: { revision: 1, printers: [receiptPrinter] }, + }), + ).toBe(PROTOCOL_VERSION); + }); + + it('requires the extended version for PDF jobs and document capabilities', () => { + expect(requiredProtocolVersion(fixture('server/print-job-pdf.json') as never)).toBe(EXTENDED_PROTOCOL_VERSION); + expect( + requiredProtocolVersion({ type: 'agent.printer_inventory', payload: { revision: 1, printers: [officePrinter] } }), + ).toBe(EXTENDED_PROTOCOL_VERSION); + expect( + requiredProtocolVersion({ + type: 'agent.printer_inventory_changed', + payload: { revision: 2, added: [], updated: [officePrinter], removedPrinterIds: [] }, + }), + ).toBe(EXTENDED_PROTOCOL_VERSION); + }); +}); + +describe('version selection', () => { + it('selects the extended version only when both peers support it', () => { + expect(selectProtocolVersion(['1', '2'], ['2', '1'])).toBe('2'); + expect(selectProtocolVersion(['1'], ['2', '1'])).toBe('1'); + expect(selectProtocolVersion(['1', '2'], ['1'])).toBe('1'); + expect(selectProtocolVersion(['2'], ['1'])).toBeNull(); + }); +}); + +describe('envelope rules', () => { + it('rejects extended content labelled with the base version', () => { + const job = withEnvelope(fixture('server/print-job-pdf.json'), PROTOCOL_VERSION); + expect(() => decodeServerMessage(job)).toThrow(ProtocolValidationError); + }); + + it('accepts extended content labelled with the extended version', () => { + const job = withEnvelope(fixture('server/print-job-pdf.json'), EXTENDED_PROTOCOL_VERSION); + expect(decodeServerMessage(job)).toMatchObject({ protocolVersion: '2', type: 'server.print_job' }); + }); + + it('refuses to encode an envelope that does not match its content', () => { + const message = decodeServerMessage(withEnvelope(fixture('server/print-job-pdf.json'), EXTENDED_PROTOCOL_VERSION)); + expect(() => encodeAgentMessage({ ...message, protocolVersion: PROTOCOL_VERSION } as never)).toThrow(); + }); + + it('requires handshake frames to use the base version', () => { + const hello = withEnvelope(fixture('agent/agent-hello-extended-offer.json'), EXTENDED_PROTOCOL_VERSION); + expect(() => decodeAgentMessage(hello)).toThrow(ProtocolValidationError); + }); +}); + +describe('legacy peers', () => { + const legacy = { supportedVersions: [PROTOCOL_VERSION] as const }; + + it('rejects an extended hello offer, as a base-only peer must', () => { + const offer = JSON.stringify(fixture('agent/agent-hello-extended-offer.json')); + expect(() => decodeAgentMessage(offer, legacy)).toThrow(ProtocolValidationError); + }); + + it('accepts the original hello, which every peer understands', () => { + const original = JSON.stringify(fixture('agent/agent-hello.json')); + expect(decodeAgentMessage(original, legacy)).toMatchObject({ type: 'agent.hello' }); + }); + + it('rejects an extended envelope with the version error a base-only peer reports', () => { + const job = withEnvelope(fixture('server/print-job.json'), EXTENDED_PROTOCOL_VERSION); + expect(() => decodeServerMessage(job, legacy)).toThrow(UnsupportedProtocolVersionError); + }); + + it('accepts a base-only session agent message unchanged', () => { + const original = JSON.stringify(fixture('agent/printer-inventory.json')); + expect(decodeAgentMessage(original, legacy)).toMatchObject({ type: 'agent.printer_inventory' }); + }); +}); + +describe('projection for base-only sessions', () => { + it('removes every extended capability and leaves the rest of the descriptor intact', () => { + const projected = withoutExtendedCapabilities(officePrinter); + expect(projected.capabilities).toEqual({ mediaWidths: [80], raster: true, cut: false, qr: true, barcode: true }); + expect(projected.id).toBe(officePrinter.id); + expect(withoutExtendedCapabilities(receiptPrinter)).toBe(receiptPrinter); + }); + + it('produces an inventory that a base-only peer accepts', () => { + const inventory = { + protocolVersion: PROTOCOL_VERSION, + messageId: 'msg_projected_inventory', + sentAt: '2026-08-01T09:00:00.000Z', + type: 'agent.printer_inventory', + payload: { revision: 3, printers: [withoutExtendedCapabilities(officePrinter)] }, + }; + expect(decodeAgentMessage(JSON.stringify(inventory), { supportedVersions: [PROTOCOL_VERSION] })).toMatchObject({ + type: 'agent.printer_inventory', + }); + }); +}); diff --git a/packages/server/README.md b/packages/server/README.md index 917195a..75c7cfe 100644 --- a/packages/server/README.md +++ b/packages/server/README.md @@ -108,6 +108,14 @@ const result = await session.sendJob({ `sendJob` handing a frame to a live transport is not durable server storage. Keep the job in application-owned storage until the agent acknowledgement and apply application retry policy outside this SDK. +## Protocol versions + +`createOpenPrinterServer` accepts `protocolVersions`. The default is every +version this SDK implements, `["2", "1"]`. The server selects the highest version +the agent offers. Use `["1"]` to serve only base-protocol agents. Agents on the +base protocol then negotiate `"1"` automatically, and PDF jobs are refused with +`protocol-version-unsupported` for sessions without version 2. + ## Job documents A PDF job is sent with `document: { format: 'pdf', source, settings }`. The diff --git a/packages/server/src/internal.ts b/packages/server/src/internal.ts index dd708f1..7a3cb75 100644 --- a/packages/server/src/internal.ts +++ b/packages/server/src/internal.ts @@ -3,7 +3,10 @@ import { OPENPRINTER_DISCOVERY_PATH, OPENPRINTER_GATEWAY_PATH, OPENPRINTER_PAIRING_PATH, + PROTOCOL_VERSION, + PROTOCOL_VERSIONS, type OpenPrinterBrandMetadata, + type ProtocolVersionName, } from '@openprinter/protocol'; import { OpenPrinterServerConfigurationError } from './errors.js'; @@ -20,6 +23,7 @@ const encoder = new TextEncoder(); /** Validated server configuration shared by all accepted sessions. */ export interface ResolvedOpenPrinterServerOptions { readonly authenticationTimeoutMs: number; + readonly protocolVersions: readonly ProtocolVersionName[]; readonly brand: OpenPrinterBrandMetadata; readonly callbackTimeoutMs: number; readonly handshakeTimeoutMs: number; @@ -65,6 +69,15 @@ export function validateOptions( const heartbeatIntervalMs = options.heartbeatIntervalMs ?? 15_000; const heartbeatTimeoutMs = options.heartbeatTimeoutMs ?? 45_000; const maxMessageBytes = options.maxMessageBytes ?? MAX_WIRE_MESSAGE_BYTES; + const protocolVersions = options.protocolVersions ?? PROTOCOL_VERSIONS; + if (!protocolVersions.includes(PROTOCOL_VERSION) || new Set(protocolVersions).size !== protocolVersions.length) { + throw new OpenPrinterServerConfigurationError( + 'protocolVersions must list each version once and include "1", which every agent can use.', + ); + } + if (protocolVersions.some((version) => !PROTOCOL_VERSIONS.includes(version))) { + throw new OpenPrinterServerConfigurationError('protocolVersions contains an unknown protocol version.'); + } const serverId = options.serverId ?? 'openprinter-server'; const serverVersion = options.serverVersion ?? '0.1.0'; const transportTimeoutMs = options.transportTimeoutMs ?? 10_000; @@ -100,6 +113,7 @@ export function validateOptions( return { brand, authenticationTimeoutMs, + protocolVersions, callbackTimeoutMs, challengeTtlMs, handshakeTimeoutMs, diff --git a/packages/server/src/session.ts b/packages/server/src/session.ts index d6260d9..9ff96d8 100644 --- a/packages/server/src/session.ts +++ b/packages/server/src/session.ts @@ -5,17 +5,21 @@ import { decodeAgentMessage, decodeBase64Url, decodeGatewayAuthenticationResponse, + EXTENDED_PROTOCOL_VERSION, encodeBase64Url, encodeGatewayAuthenticationServerMessage, encodeServerMessage, OPENPRINTER_SIGNATURE_ALGORITHM, PROTOCOL_VERSION, ProtocolError, + requiredProtocolVersion, + selectProtocolVersion, type AgentMessage, type GatewayAuthenticationChallenge, type GatewayAuthenticationFailureCode, type PrintJob, type PrinterDescriptor, + type ProtocolVersionName, } from '@openprinter/protocol'; import { @@ -91,6 +95,7 @@ export class OpenPrinterSessionImplementation implements OpenPrinterSe #challengeConsumed = false; #authenticationTimer: ReturnType | null = null; #hello: AgentMessageOf<'agent.hello'>['payload'] | null = null; + #protocolVersion: ProtocolVersionName = PROTOCOL_VERSION; #connectedAtMs: number | null = null; #lastSeenAtMs = Date.now(); #lastHeartbeatAtMs = this.#lastSeenAtMs; @@ -143,7 +148,17 @@ export class OpenPrinterSessionImplementation implements OpenPrinterSe } public async send(message: OpenPrinterApplicationMessage): Promise { - const encoded = encodeServerMessage(message); + // Each envelope carries the lowest version its content needs. Extended content is refused + // in a session that did not negotiate the extended protocol, so the peer never receives it. + const protocolVersion = requiredProtocolVersion(message); + if ( + this.#state === 'connected' && + protocolVersion === EXTENDED_PROTOCOL_VERSION && + this.#protocolVersion !== EXTENDED_PROTOCOL_VERSION + ) { + return this.#deliveryFailure('protocol-version-unsupported', false); + } + const encoded = encodeServerMessage({ ...message, protocolVersion }); const encodedBytes = utf8ByteLength(encoded); if (encodedBytes > this.#resolved.maxMessageBytes) { throw new ProtocolError( @@ -414,13 +429,25 @@ export class OpenPrinterSessionImplementation implements OpenPrinterSe let message: AgentMessage; try { - message = decodeAgentMessage(input); + message = decodeAgentMessage(input, { supportedVersions: this.#resolved.protocolVersions }); } catch (error) { const protocolError = error instanceof Error ? error : new Error('The agent message was invalid.'); await this.#protocolViolation(protocolErrorCode(error), protocolError); return; } + if ( + message.protocolVersion === EXTENDED_PROTOCOL_VERSION && + this.#state === 'connected' && + this.#protocolVersion !== EXTENDED_PROTOCOL_VERSION + ) { + await this.#protocolViolation( + 'unsupported-protocol-version', + new Error('The agent used an extended protocol version that was not negotiated.'), + ); + return; + } + this.#lastSeenAtMs = Date.now(); if (this.#state === 'handshaking') { @@ -450,7 +477,8 @@ export class OpenPrinterSessionImplementation implements OpenPrinterSe return; } - if (!hello.payload.supportedProtocolVersions.includes(PROTOCOL_VERSION)) { + const selected = selectProtocolVersion(this.#resolved.protocolVersions, hello.payload.supportedProtocolVersions); + if (selected === null) { await this.#protocolViolation( 'unsupported-protocol-version', new Error('The peers do not share a supported protocol version.'), @@ -467,8 +495,9 @@ export class OpenPrinterSessionImplementation implements OpenPrinterSe serverVersion: this.#resolved.serverVersion, brand: this.#resolved.brand, sessionId: this.sessionId, - supportedProtocolVersions: [PROTOCOL_VERSION], - selectedProtocolVersion: PROTOCOL_VERSION, + // Lists only the selected version, which keeps the handshake readable by base-protocol agents. + supportedProtocolVersions: [selected], + selectedProtocolVersion: selected, heartbeatIntervalMs: this.#resolved.heartbeatIntervalMs, maxMessageBytes: this.#resolved.maxMessageBytes, }, @@ -482,6 +511,7 @@ export class OpenPrinterSessionImplementation implements OpenPrinterSe this.#clearHandshakeTimer(); const now = Date.now(); + this.#protocolVersion = selected; this.#hello = hello.payload; this.#connectedAtMs = now; this.#lastHeartbeatAtMs = now; @@ -890,18 +920,19 @@ export class OpenPrinterSessionImplementation implements OpenPrinterSe sessionId: this.sessionId, ...(identity.metadata === undefined ? {} : { metadata: identity.metadata }), hello: structuredClone(this.#hello), + protocolVersion: this.#protocolVersion, connectedAt: new Date(this.#connectedAtMs).toISOString(), lastSeenAt: new Date(this.#lastSeenAtMs).toISOString(), printerRevision: this.#printerRevision, }; } - #deliveryFailure(reason: DeliveryFailure['reason']): DeliveryFailure { + #deliveryFailure(reason: DeliveryFailure['reason'], retryable = true): DeliveryFailure { return { ok: false, agentId: this.#identity?.agentId ?? 'unauthenticated', reason, - retryable: true, + retryable, }; } diff --git a/packages/server/src/types.ts b/packages/server/src/types.ts index 5b45c7b..ded3c5f 100644 --- a/packages/server/src/types.ts +++ b/packages/server/src/types.ts @@ -8,6 +8,7 @@ import type { PublicEd25519Jwk, PrintJob, PrinterDescriptor, + ProtocolVersionName, ServerMessage, } from '@openprinter/protocol'; @@ -47,6 +48,8 @@ export interface ConnectedAgent { readonly metadata?: Readonly; /** Validated agent hello payload. */ readonly hello: Readonly['payload']>; + /** Protocol version selected for this session during the hello exchange. */ + readonly protocolVersion: ProtocolVersionName; /** UTC time at which the protocol handshake completed. */ readonly connectedAt: string; /** UTC time of the most recently validated agent message. */ @@ -294,6 +297,13 @@ export interface OpenPrinterServerOptions { readonly serverVersion?: string; /** Framework-independent endpoint path configuration. */ readonly paths?: Partial; + /** + * Protocol versions this server accepts. Defaults to every version this SDK implements. + * + * Must include `"1"`. Use `["1"]` to serve only base-protocol agents; agents that offer the extended + * protocol then negotiate the base version instead. + */ + readonly protocolVersions?: readonly ProtocolVersionName[]; /** Storage for temporary pairing grants; defaults to an in-memory implementation. */ readonly pairingCodeStore?: PairingCodeStore; /** Storage for agent public keys; defaults to an in-memory implementation. */ @@ -357,7 +367,12 @@ export interface DeliverySuccess { } /** Reasons a message could not be handed to an agent transport. */ -export type DeliveryFailureReason = 'agent-offline' | 'session-not-ready' | 'connection-closed' | 'transport-error'; +export type DeliveryFailureReason = + | 'agent-offline' + | 'session-not-ready' + | 'connection-closed' + | 'transport-error' + | 'protocol-version-unsupported'; /** Structured, non-queued delivery failure owned by the host application. */ export interface DeliveryFailure { @@ -367,8 +382,8 @@ export interface DeliveryFailure { readonly agentId: string; /** Stable reason suitable for application queue policy. */ readonly reason: DeliveryFailureReason; - /** Signals that a later application-owned retry may succeed. */ - readonly retryable: true; + /** Whether a later application-owned retry may succeed; false when the session cannot carry the message. */ + readonly retryable: boolean; } /** Immediate transport handoff result; it is not a persistence acknowledgement. */ diff --git a/packages/server/test/negotiation.test.ts b/packages/server/test/negotiation.test.ts new file mode 100644 index 0000000..7bfb078 --- /dev/null +++ b/packages/server/test/negotiation.test.ts @@ -0,0 +1,361 @@ +import { generateKeyPairSync, sign } from 'node:crypto'; +import { readFileSync } from 'node:fs'; + +import { + EXTENDED_PROTOCOL_VERSION, + PROTOCOL_VERSION, + decodeBase64Url, + decodeGatewayAuthenticationServerMessage, + decodeServerMessage, + encodeGatewayAuthenticationResponse, + type GatewayAuthenticationChallenge, + type OpenPrinterPairingRequest, + type PrintJob, + type ProtocolVersionName, + type ServerMessage, +} from '@openprinter/protocol'; +import { describe, expect, it } from 'vitest'; + +import { + createOpenPrinterServer, + type OpenPrinterServer, + type OpenPrinterTransportCloseRequest, +} from '../src/index.js'; + +/* + * Compatibility matrix for protocol negotiation: + * + * old agent (offers ["1"]) + new server → base protocol, unchanged wire + * new agent (offers ["2", "1"]) + new server → extended protocol + * new agent (offers ["2", "1"]) + base-only server (["1"]) → closed before server.hello; + * the agent retries with ["1"] + */ + +const fixtureRoot = new URL('../../../protocol/fixtures/', import.meta.url); + +function fixture(path: string): T { + return JSON.parse(readFileSync(new URL(path, fixtureRoot), 'utf8')) as T; +} + +interface Credential { + readonly privateKey: ReturnType['privateKey']; + readonly publicKey: OpenPrinterPairingRequest['credential']['publicKey']; +} + +function credential(): Credential { + const pair = generateKeyPairSync('ed25519'); + const jwk = pair.publicKey.export({ format: 'jwk' }); + return { privateKey: pair.privateKey, publicKey: { kty: 'OKP', crv: 'Ed25519', x: jwk.x! } }; +} + +function wireRecorder() { + const sent: string[] = []; + const closes: OpenPrinterTransportCloseRequest[] = []; + return { + sent, + closes, + transport: { + send(message: string) { + sent.push(message); + }, + close(request: OpenPrinterTransportCloseRequest) { + closes.push(request); + }, + }, + }; +} + +async function waitFor(predicate: () => boolean): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 1)); + } + throw new Error('Timed out waiting for test state.'); +} + +function serverWith(protocolVersions?: readonly ProtocolVersionName[]) { + return createOpenPrinterServer>({ + brand: { name: 'Negotiation test service' }, + serverId: 'negotiation-service', + serverVersion: '1.0.0', + heartbeatIntervalMs: 60_000, + heartbeatTimeoutMs: 120_000, + ...(protocolVersions === undefined ? {} : { protocolVersions }), + }); +} + +async function pair(openprinter: OpenPrinterServer>, key: Credential) { + const pairing = await openprinter.createPairingCode(); + const result = await openprinter.pair({ + protocolVersion: PROTOCOL_VERSION, + code: pairing.code, + agent: { name: 'Front Counter', version: '1.0.0', platform: 'test', installationId: 'installation-01' }, + credential: { algorithm: 'Ed25519', publicKey: key.publicKey }, + }); + if ('error' in result) throw new Error(result.error.code); + return result; +} + +/** Runs gateway authentication on a fresh socket and returns the session plus its wire recorder. */ +async function authenticatedSession( + openprinter: OpenPrinterServer>, + key: Credential, + agent: Awaited>, +) { + const wire = wireRecorder(); + const session = openprinter.accept({ transport: wire.transport }); + await waitFor(() => wire.sent.length === 1); + const challenge = decodeGatewayAuthenticationServerMessage(wire.sent[0]!) as GatewayAuthenticationChallenge; + const signature = sign(null, decodeBase64Url(challenge.payload), key.privateKey).toString('base64url'); + await session.receive( + encodeGatewayAuthenticationResponse({ + type: 'auth.response', + challengeId: challenge.challengeId, + agentId: agent.agentId, + keyId: agent.keyId, + algorithm: 'Ed25519', + signature, + }), + ); + await waitFor(() => wire.sent.length === 2); + return { session, wire }; +} + +/** Sends a hello and waits for the server's reply or close. */ +async function hello( + session: { receive(message: string): Promise }, + wire: ReturnType, + agentId: string, + offered: readonly ProtocolVersionName[], +): Promise { + const sentBefore = wire.sent.length; + const message = fixture>('agent/agent-hello.json'); + await session.receive( + JSON.stringify({ + ...message, + messageId: `hello-${offered.join('-')}`, + payload: { ...(message.payload as object), agentId, supportedProtocolVersions: offered }, + }), + ); + await waitFor(() => wire.sent.length > sentBefore || wire.closes.length > 0); + if (wire.sent.length === sentBefore) return null; + return decodeServerMessage(wire.sent[sentBefore]!); +} + +function receiptPrinter(id: string) { + return { + id, + fingerprint: `system:CUPS:${id}`, + name: 'Front desk receipt', + kind: 'local', + connection: { type: 'system', systemName: id }, + capabilities: { + mediaWidths: [80], + systemDriver: false, + escPos: true, + raster: false, + cut: true, + qr: true, + barcode: true, + }, + enabled: true, + availability: 'online', + }; +} + +function officePrinter(id: string) { + return { + ...receiptPrinter(id), + name: 'Office laser', + capabilities: { + mediaWidths: [80], + systemDriver: true, + escPos: false, + raster: true, + cut: false, + qr: true, + barcode: true, + documentFormats: ['receipt', 'pdf'], + paperSizes: ['a4', 'letter'], + customPaperSize: true, + printOptions: { copies: true, duplex: false, orientation: true, scaling: true, pageRanges: true }, + }, + }; +} + +function inventory(version: ProtocolVersionName, printers: unknown[]): string { + return JSON.stringify({ + protocolVersion: version, + messageId: `inventory-${version}-${printers.length}`, + sentAt: '2026-08-01T09:01:00.000Z', + type: 'agent.printer_inventory', + payload: { revision: 1, printers }, + }); +} + +function receiptJob(): PrintJob { + return fixture<{ payload: PrintJob }>('server/print-job.json').payload; +} + +function pdfJob(): PrintJob { + return fixture<{ payload: PrintJob }>('server/print-job-pdf.json').payload; +} + +describe('old agent with a new server', () => { + it('negotiates the base protocol with a handshake identical to the original wire shape', async () => { + const openprinter = serverWith(); + const key = credential(); + const agent = await pair(openprinter, key); + const { session, wire } = await authenticatedSession(openprinter, key, agent); + + const reply = await hello(session, wire, agent.agentId, [PROTOCOL_VERSION]); + + expect(reply).toMatchObject({ + protocolVersion: PROTOCOL_VERSION, + type: 'server.hello', + payload: { supportedProtocolVersions: [PROTOCOL_VERSION], selectedProtocolVersion: PROTOCOL_VERSION }, + }); + expect(session.getAgent()?.protocolVersion).toBe(PROTOCOL_VERSION); + }); + + it('accepts base-protocol inventory and delivers receipts with base envelopes', async () => { + const openprinter = serverWith(); + const key = credential(); + const agent = await pair(openprinter, key); + const { session, wire } = await authenticatedSession(openprinter, key, agent); + await hello(session, wire, agent.agentId, [PROTOCOL_VERSION]); + + await session.receive(inventory(PROTOCOL_VERSION, [receiptPrinter('printer_receipt_001')])); + expect(session.getPrinters()).toHaveLength(1); + + const delivered = await session.sendJob(receiptJob()); + expect(delivered).toMatchObject({ ok: true }); + const outbound = JSON.parse(wire.sent[wire.sent.length - 1]!) as { payload: { document: object } }; + expect(outbound).toMatchObject({ type: 'server.print_job', protocolVersion: PROTOCOL_VERSION }); + expect(outbound.payload.document).not.toHaveProperty('format'); + }); + + it('refuses PDF jobs without sending them, because the session cannot carry them', async () => { + const openprinter = serverWith(); + const key = credential(); + const agent = await pair(openprinter, key); + const { session, wire } = await authenticatedSession(openprinter, key, agent); + await hello(session, wire, agent.agentId, [PROTOCOL_VERSION]); + const sentBefore = wire.sent.length; + + const result = await session.sendJob(pdfJob()); + + expect(result).toEqual({ + ok: false, + agentId: agent.agentId, + reason: 'protocol-version-unsupported', + retryable: false, + }); + expect(wire.sent).toHaveLength(sentBefore); + }); + + it('closes the session when it sends extended content, even under a labelled extended envelope', async () => { + const openprinter = serverWith(); + const key = credential(); + const agent = await pair(openprinter, key); + const { session, wire } = await authenticatedSession(openprinter, key, agent); + await hello(session, wire, agent.agentId, [PROTOCOL_VERSION]); + + await session.receive(inventory(EXTENDED_PROTOCOL_VERSION, [officePrinter('printer_office_001')])); + + expect(session.state).toBe('closed'); + expect(session.getPrinters()).toHaveLength(0); + }); +}); + +describe('new agent with a new server', () => { + it('negotiates the extended protocol and exchanges document capabilities', async () => { + const openprinter = serverWith(); + const key = credential(); + const agent = await pair(openprinter, key); + const { session, wire } = await authenticatedSession(openprinter, key, agent); + + const reply = await hello(session, wire, agent.agentId, [EXTENDED_PROTOCOL_VERSION, PROTOCOL_VERSION]); + + expect(reply).toMatchObject({ + protocolVersion: PROTOCOL_VERSION, + type: 'server.hello', + payload: { + supportedProtocolVersions: [EXTENDED_PROTOCOL_VERSION], + selectedProtocolVersion: EXTENDED_PROTOCOL_VERSION, + }, + }); + expect(session.getAgent()?.protocolVersion).toBe(EXTENDED_PROTOCOL_VERSION); + + await session.receive(inventory(EXTENDED_PROTOCOL_VERSION, [officePrinter('printer_office_001')])); + expect(session.getPrinters()[0]?.capabilities).toMatchObject({ documentFormats: ['receipt', 'pdf'] }); + }); + + it('rejects document capabilities labelled with the base envelope', async () => { + const openprinter = serverWith(); + const key = credential(); + const agent = await pair(openprinter, key); + const { session, wire } = await authenticatedSession(openprinter, key, agent); + await hello(session, wire, agent.agentId, [EXTENDED_PROTOCOL_VERSION, PROTOCOL_VERSION]); + + await session.receive(inventory(PROTOCOL_VERSION, [officePrinter('printer_office_001')])); + + expect(session.state).toBe('closed'); + }); + + it('delivers PDF jobs with the extended envelope and receipts with the base envelope', async () => { + const openprinter = serverWith(); + const key = credential(); + const agent = await pair(openprinter, key); + const { session, wire } = await authenticatedSession(openprinter, key, agent); + await hello(session, wire, agent.agentId, [EXTENDED_PROTOCOL_VERSION, PROTOCOL_VERSION]); + + expect(await session.sendJob(pdfJob())).toMatchObject({ ok: true }); + expect(JSON.parse(wire.sent[wire.sent.length - 1]!)).toMatchObject({ protocolVersion: EXTENDED_PROTOCOL_VERSION }); + + expect(await session.sendJob(receiptJob())).toMatchObject({ ok: true }); + expect(JSON.parse(wire.sent[wire.sent.length - 1]!)).toMatchObject({ protocolVersion: PROTOCOL_VERSION }); + }); +}); + +describe('new agent with a base-only server', () => { + it('closes an extended hello without replying, so the agent can fall back to the base protocol', async () => { + const openprinter = serverWith([PROTOCOL_VERSION]); + const key = credential(); + const agent = await pair(openprinter, key); + const { session, wire } = await authenticatedSession(openprinter, key, agent); + + const reply = await hello(session, wire, agent.agentId, [EXTENDED_PROTOCOL_VERSION, PROTOCOL_VERSION]); + + expect(reply).toBeNull(); + expect(wire.closes).toHaveLength(1); + expect(session.state).toBe('closed'); + }); + + it('completes the base handshake on a fresh connection after the fallback', async () => { + const openprinter = serverWith([PROTOCOL_VERSION]); + const key = credential(); + const agent = await pair(openprinter, key); + + const rejected = await authenticatedSession(openprinter, key, agent); + expect( + await hello(rejected.session, rejected.wire, agent.agentId, [EXTENDED_PROTOCOL_VERSION, PROTOCOL_VERSION]), + ).toBeNull(); + + const retry = await authenticatedSession(openprinter, key, agent); + const reply = await hello(retry.session, retry.wire, agent.agentId, [PROTOCOL_VERSION]); + + expect(reply).toMatchObject({ + type: 'server.hello', + payload: { supportedProtocolVersions: [PROTOCOL_VERSION], selectedProtocolVersion: PROTOCOL_VERSION }, + }); + expect(retry.session.getAgent()?.protocolVersion).toBe(PROTOCOL_VERSION); + }); +}); + +describe('server configuration', () => { + it('requires the base protocol so every agent can still connect', () => { + expect(() => serverWith([EXTENDED_PROTOCOL_VERSION])).toThrow(/include "1"/); + expect(() => serverWith(['1', '1'])).toThrow(/once/); + }); +}); diff --git a/protocol/README.md b/protocol/README.md index 11d3e99..dc89865 100644 --- a/protocol/README.md +++ b/protocol/README.md @@ -8,5 +8,11 @@ This directory contains language-neutral artifacts shared by the canonical TypeB - `fixtures/server` contains valid server-to-agent messages. - `fixtures/invalid` contains payloads every implementation must reject. +Protocol version `"1"` is the base protocol. Version `"2"` adds PDF print jobs and printer +document capabilities. Handshake frames always use envelope version `"1"`. Other messages use the +lowest version their content needs, and receivers reject extended content under a `"1"` envelope. +The `agent-hello-extended-offer.json` and `server-hello-extended.json` fixtures show a negotiated +extended handshake. Earlier peers reject the extended offer, which is covered in the protocol tests. + Do not hand-edit the generated schema. Run `pnpm --filter @openprinter/protocol schema:generate`, then run both protocol test suites whenever the wire contract changes. diff --git a/protocol/fixtures/agent/agent-hello-extended-offer.json b/protocol/fixtures/agent/agent-hello-extended-offer.json new file mode 100644 index 0000000..ccf1990 --- /dev/null +++ b/protocol/fixtures/agent/agent-hello-extended-offer.json @@ -0,0 +1,13 @@ +{ + "protocolVersion": "1", + "messageId": "msg_agent_hello_extended_001", + "sentAt": "2026-07-28T10:00:00Z", + "type": "agent.hello", + "payload": { + "agentId": "agent_demo_001", + "agentVersion": "0.1.0", + "productId": "oppa.default", + "productVersion": "1.0.0", + "supportedProtocolVersions": ["2", "1"] + } +} diff --git a/protocol/fixtures/agent/printer-inventory-documents.json b/protocol/fixtures/agent/printer-inventory-documents.json index f9688ec..558625b 100644 --- a/protocol/fixtures/agent/printer-inventory-documents.json +++ b/protocol/fixtures/agent/printer-inventory-documents.json @@ -1,5 +1,5 @@ { - "protocolVersion": "1", + "protocolVersion": "2", "messageId": "msg_inventory_documents_001", "correlationId": "msg_inventory_request_003", "sentAt": "2026-07-28T10:00:06Z", diff --git a/protocol/fixtures/invalid/pdf-job-declared-base-version.json b/protocol/fixtures/invalid/pdf-job-declared-base-version.json new file mode 100644 index 0000000..18b46d4 --- /dev/null +++ b/protocol/fixtures/invalid/pdf-job-declared-base-version.json @@ -0,0 +1,35 @@ +{ + "protocolVersion": "1", + "messageId": "msg_pdf_declared_base", + "sentAt": "2026-07-28T10:03:00Z", + "type": "server.print_job", + "payload": { + "jobId": "job_pdf_label_001", + "idempotencyKey": "label_001_v1", + "printerId": "printer_office_001", + "createdAt": "2026-07-28T10:02:58Z", + "document": { + "format": "pdf", + "source": { + "type": "url", + "url": "https://documents.example.com/v1/documents/doc_label_001", + "sha256": "9f2c7e1a4b8d03e5f6a7b8c9d0e1f2a34b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e", + "byteLength": 48213 + }, + "settings": { + "copies": 2, + "paper": { + "widthMm": 50, + "heightMm": 30 + }, + "orientation": "landscape", + "scaling": "fit", + "pageRanges": "1-2,4", + "duplex": "simplex" + } + }, + "metadata": { + "route": "shipping" + } + } +} diff --git a/protocol/fixtures/invalid/printer-inventory-extended-declared-base-version.json b/protocol/fixtures/invalid/printer-inventory-extended-declared-base-version.json new file mode 100644 index 0000000..b6bbc73 --- /dev/null +++ b/protocol/fixtures/invalid/printer-inventory-extended-declared-base-version.json @@ -0,0 +1,43 @@ +{ + "protocolVersion": "1", + "messageId": "msg_inventory_declared_base", + "correlationId": "msg_inventory_request_003", + "sentAt": "2026-07-28T10:00:06Z", + "type": "agent.printer_inventory", + "payload": { + "revision": 8, + "printers": [ + { + "id": "printer_office_001", + "fingerprint": "system:CUPS:office-001", + "name": "Office laser", + "kind": "local", + "connection": { + "type": "system", + "systemName": "office-001" + }, + "capabilities": { + "mediaWidths": [80], + "systemDriver": true, + "escPos": false, + "raster": true, + "cut": false, + "qr": true, + "barcode": true, + "documentFormats": ["receipt", "pdf"], + "paperSizes": ["a4", "a5", "letter", "legal"], + "customPaperSize": true, + "printOptions": { + "copies": true, + "duplex": true, + "orientation": true, + "scaling": true, + "pageRanges": true + } + }, + "enabled": true, + "availability": "online" + } + ] + } +} diff --git a/protocol/fixtures/server/print-job-pdf-loopback.json b/protocol/fixtures/server/print-job-pdf-loopback.json index c5f764b..39896bd 100644 --- a/protocol/fixtures/server/print-job-pdf-loopback.json +++ b/protocol/fixtures/server/print-job-pdf-loopback.json @@ -1,5 +1,5 @@ { - "protocolVersion": "1", + "protocolVersion": "2", "messageId": "msg_print_job_pdf_002", "sentAt": "2026-07-28T10:03:00Z", "type": "server.print_job", diff --git a/protocol/fixtures/server/print-job-pdf.json b/protocol/fixtures/server/print-job-pdf.json index c386e8f..5240179 100644 --- a/protocol/fixtures/server/print-job-pdf.json +++ b/protocol/fixtures/server/print-job-pdf.json @@ -1,5 +1,5 @@ { - "protocolVersion": "1", + "protocolVersion": "2", "messageId": "msg_print_job_pdf_001", "sentAt": "2026-07-28T10:03:00Z", "type": "server.print_job", diff --git a/protocol/fixtures/server/server-hello-extended.json b/protocol/fixtures/server/server-hello-extended.json new file mode 100644 index 0000000..61808a2 --- /dev/null +++ b/protocol/fixtures/server/server-hello-extended.json @@ -0,0 +1,19 @@ +{ + "protocolVersion": "1", + "messageId": "msg_server_hello_extended_001", + "correlationId": "msg_agent_hello_extended_001", + "sentAt": "2026-07-28T10:00:00Z", + "type": "server.hello", + "payload": { + "serverId": "server_example_001", + "serverVersion": "0.1.0", + "brand": { + "name": "OpenPrinter Node.js example" + }, + "sessionId": "session_001", + "supportedProtocolVersions": ["2"], + "selectedProtocolVersion": "2", + "heartbeatIntervalMs": 30000, + "maxMessageBytes": 2097152 + } +} diff --git a/protocol/schema/openprinter.schema.json b/protocol/schema/openprinter.schema.json index 58eed2f..4223153 100644 --- a/protocol/schema/openprinter.schema.json +++ b/protocol/schema/openprinter.schema.json @@ -43,8 +43,16 @@ }, "supportedProtocolVersions": { "items": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "maxItems": 8, "minItems": 1, @@ -62,8 +70,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -115,8 +131,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -767,8 +791,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -2022,8 +2054,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -2094,8 +2134,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -2174,8 +2222,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -2273,8 +2329,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -2407,8 +2471,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -2480,8 +2552,16 @@ "type": "integer" }, "selectedProtocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "serverId": { "maxLength": 128, @@ -2502,8 +2582,16 @@ }, "supportedProtocolVersions": { "items": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "maxItems": 8, "minItems": 1, @@ -2524,8 +2612,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -2572,8 +2668,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -3048,8 +3152,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -3101,8 +3213,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -3139,8 +3259,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -3202,8 +3330,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30, @@ -3265,8 +3401,16 @@ "type": "object" }, "protocolVersion": { - "const": "1", - "type": "string" + "anyOf": [ + { + "const": "1", + "type": "string" + }, + { + "const": "2", + "type": "string" + } + ] }, "sentAt": { "maxLength": 30,