From 778f97f750327589dbc4dd7237429f45bf74ba3e Mon Sep 17 00:00:00 2001 From: Wolfie Date: Sun, 30 Aug 2026 18:32:15 -0700 Subject: [PATCH 1/2] Generate protocol catalogs and negotiate compatibility --- .github/workflows/ci.yml | 1 + .github/workflows/release.yml | 1 + docs/adr/0001-agenttab-runtime.md | 5 + docs/protocol-development.md | 48 +++ docs/verification.md | 1 + host-rs/crates/agenttab-host/src/native.rs | 88 ++++- host-rs/crates/agenttab-host/src/runtime.rs | 10 + host-rs/crates/agenttab-host/src/server.rs | 62 ++- host-rs/crates/agenttab-host/src/task.rs | 4 + .../crates/agenttab-protocol/src/generated.rs | 128 ++++++ host-rs/crates/agenttab-protocol/src/lib.rs | 374 ++++++++++++++---- package.json | 2 + packages/extension/src/generated/protocol.ts | 49 +++ packages/extension/src/native.ts | 72 +++- packages/extension/src/protocol.ts | 163 +++++--- packages/extension/test/extension.test.ts | 178 +++++++++ packages/mcp/src/server.ts | 26 +- packages/omp/src/index.ts | 26 +- packages/sdk-python/agenttab/__init__.py | 6 + .../agenttab/_generated_protocol.py | 40 ++ packages/sdk-python/agenttab/client.py | 135 +++++-- packages/sdk-python/tests/test_client.py | 170 +++++++- .../sdk-typescript/src/generated/protocol.ts | 101 +++++ packages/sdk-typescript/src/index.ts | 200 ++++++++-- packages/sdk-typescript/test/client.test.ts | 126 ++++++ protocol/agenttab-v1.json | 136 +++++++ schemas/native/v1/message.schema.json | 69 +++- schemas/rpc/v1/connection.schema.json | 54 ++- scripts/generate_protocol.py | 332 ++++++++++++++++ tests/architecture/verify_protocol_schemas.py | 80 +++- 30 files changed, 2397 insertions(+), 290 deletions(-) create mode 100644 docs/protocol-development.md create mode 100644 host-rs/crates/agenttab-protocol/src/generated.rs create mode 100644 packages/extension/src/generated/protocol.ts create mode 100644 packages/sdk-python/agenttab/_generated_protocol.py create mode 100644 packages/sdk-typescript/src/generated/protocol.ts create mode 100644 protocol/agenttab-v1.json create mode 100644 scripts/generate_protocol.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1f3a9a5..4ef0051 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -75,6 +75,7 @@ jobs: run: | PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_permissions.py cargo test --locked --manifest-path tests/architecture/ipc-probe/Cargo.toml + PYTHONDONTWRITEBYTECODE=1 python3 scripts/generate_protocol.py --check PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_protocol_schemas.py PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_identity.py PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_forbidden_surface.py diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index f5da33d..18c420e 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -114,6 +114,7 @@ jobs: bun run workspace:test bun run workspace:build PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=packages/sdk-python python3 -m unittest discover -s packages/sdk-python/tests -v + PYTHONDONTWRITEBYTECODE=1 python3 scripts/generate_protocol.py --check PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_protocol_schemas.py PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_identity.py PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_forbidden_surface.py diff --git a/docs/adr/0001-agenttab-runtime.md b/docs/adr/0001-agenttab-runtime.md index 904bb03..121f9c1 100644 --- a/docs/adr/0001-agenttab-runtime.md +++ b/docs/adr/0001-agenttab-runtime.md @@ -80,6 +80,11 @@ TCP and bearer-token access are not Standard transport. They exist only behind t AgentTab Core RPC and the host-to-extension native protocol are separately versioned. They MUST NOT silently downgrade across an incompatible version. +The canonical protocol catalog, generated artifact workflow, and additive capability negotiation are +documented in [`docs/protocol-development.md`](../protocol-development.md). An incompatible hello or +connection receives an explicit compatibility frame before disconnect; v1 peers that omit capability +fields retain the legacy handshake shape. + MCP, OMP, CLI, TypeScript, and Python are adapters over Core RPC. They are not alternate hosts. The public Standard surface has exactly seven tools: 1. `browser_open` diff --git a/docs/protocol-development.md b/docs/protocol-development.md new file mode 100644 index 0000000..8fafab8 --- /dev/null +++ b/docs/protocol-development.md @@ -0,0 +1,48 @@ +# Protocol development + +AgentTab keeps wire behavior reviewable without asking contributors to synchronize lists by hand. +The canonical catalog is [`protocol/agenttab-v1.json`](../protocol/agenttab-v1.json). It owns protocol +names and versions, feature names, frame limits, outcomes, RPC method metadata, native methods and +events, and the mapping from each RPC method to its JSON Schema input. + +The JSON Schemas under [`schemas/`](../schemas/) remain the canonical structural definitions for +requests, responses, connection negotiation, native messages, and tool inputs. The generator checks +that their method branches, mutation requirements, outcomes, schema IDs, and registered files agree +with the catalog. MCP consumes the input schemas directly. Rust, TypeScript, Python, and the extension +consume small committed generated catalogs so release artifacts do not need Python at runtime. + +## Changing the protocol + +1. Edit the catalog and the affected JSON Schemas together. +2. Run `python3 scripts/generate_protocol.py` (or `bun run protocol:generate`). +3. Review the generated diff; generated files are committed. +4. Run `python3 scripts/generate_protocol.py --check` and the normal workspace tests. + +CI and release jobs run the check offline and fail when generated files drift. Do not edit files marked +`@generated` directly. + +## Compatibility contract + +Core `connect` and native `hello` messages may include `supported_versions` and +`supported_features`. Version 1 messages that omit both fields retain their original behavior and +receive the original acknowledgement shape. When a peer advertises capabilities, the host selects an +overlapping version and returns only the feature intersection. Feature names are additive; code must +not infer support for an unadvertised feature. + +An incompatible Core connection receives a bounded `kind: "incompatible"` frame before the host +closes the stream. An incompatible native hello receives the corresponding native frame. Both include +the requested protocol/version, host-supported versions, and a recovery instruction. This replaces an +ambiguous EOF while still refusing silent major-version downgrade. + +Capability-advertising clients first send the enriched v1 handshake. If a peer closes or rejects that +transport before sending any acknowledgement, the TypeScript and Python SDKs retry once on a new +connection with the exact legacy-v1 handshake. The extension does the same and keeps using legacy v1 +for that service-worker lifetime. A timeout, any Core response bytes, any received native message, or +an explicit `incompatible` frame never triggers the fallback. This bounded asymmetric-upgrade path lets +an auto-updated client reach an older strict-v1 host without turning real incompatibility into a retry +loop. + +Version and feature fields are optional specifically so the current v1 clients and extension continue +to connect to a newer host. A future major may advertise several supported versions, but it must use +the version selected by the host for every subsequent frame. Removing or changing an existing feature, +method, outcome, or schema constraint requires a new protocol major and a migration note. diff --git a/docs/verification.md b/docs/verification.md index 325afbc..eb76cb4 100644 --- a/docs/verification.md +++ b/docs/verification.md @@ -15,6 +15,7 @@ PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=packages/sdk-python python3 -m unittest dis PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_permissions.py cargo test --locked --manifest-path tests/architecture/ipc-probe/Cargo.toml PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_protocol_schemas.py +PYTHONDONTWRITEBYTECODE=1 python3 scripts/generate_protocol.py --check PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_identity.py PYTHONDONTWRITEBYTECODE=1 python3 tests/architecture/verify_forbidden_surface.py cargo fmt --all --manifest-path host-rs/Cargo.toml -- --check diff --git a/host-rs/crates/agenttab-host/src/native.rs b/host-rs/crates/agenttab-host/src/native.rs index 878f519..005df6c 100644 --- a/host-rs/crates/agenttab-host/src/native.rs +++ b/host-rs/crates/agenttab-host/src/native.rs @@ -1,11 +1,12 @@ use crate::handoff::HandoffState; use crate::lifecycle::Lifecycle; use agenttab_protocol::{ - native_close_task, native_command, native_event_ack, native_event_ack_result, native_ready, - read_frame, write_frame, NativeDisconnectEvent, NativeDisconnectRecovery, NativeEvent, - NativeEventName, NativeEventPayload, NativeHandoff, NativeHello, NativeOriginPolicy, - NativeResponse, NativeStagedCommit, NativeTab, Outcome, ProtocolError, RpcError, RuntimeState, - EXTENSION_TO_HOST_MAX_BYTES, HOST_TO_EXTENSION_MAX_BYTES, NATIVE_PROTOCOL, PROTOCOL_VERSION, + native_close_task, native_command, native_event_ack, native_event_ack_result, + native_incompatible, native_ready, read_frame, write_frame, NativeDisconnectEvent, + NativeDisconnectRecovery, NativeEvent, NativeEventName, NativeEventPayload, NativeHandoff, + NativeHello, NativeOriginPolicy, NativeResponse, NativeStagedCommit, NativeTab, Outcome, + ProtocolError, RpcError, RuntimeState, EXTENSION_TO_HOST_MAX_BYTES, + HOST_TO_EXTENSION_MAX_BYTES, NATIVE_PROTOCOL, NATIVE_VERSION, }; use parking_lot::{Mutex, RwLock}; use serde_json::Value; @@ -166,19 +167,38 @@ impl StdioNative { .get("protocol") .and_then(Value::as_str) .unwrap_or_default(); - let version = value + let requested_version = value .get("version") .and_then(Value::as_u64) - .unwrap_or_default() as u16; - if protocol != NATIVE_PROTOCOL || version != PROTOCOL_VERSION { + .unwrap_or_default(); + let kind = value + .get("kind") + .and_then(Value::as_str) + .map(str::to_owned); + let supports_host_version = value + .get("supported_versions") + .and_then(Value::as_array) + .is_some_and(|versions| { + versions + .iter() + .any(|version| version.as_u64() == Some(u64::from(NATIVE_VERSION))) + }); + if protocol != NATIVE_PROTOCOL + || (requested_version != u64::from(NATIVE_VERSION) + && !(kind.as_deref() == Some("hello") && supports_host_version)) + { + if kind.as_deref() == Some("hello") { + self.write_value(&native_incompatible(protocol, requested_version))?; + } return Err(ProtocolError::UnsupportedProtocol { protocol: protocol.into(), - version, + version: u16::try_from(requested_version).unwrap_or_default(), }); } - match value.get("kind").and_then(Value::as_str) { + match kind.as_deref() { Some("hello") => { let hello = NativeHello::parse(value)?; + let features = hello.negotiated_features(); self.lifecycle.begin_reconciliation(); if let Some(sink) = self.event_sink.read().clone() { sink.reconcile(&hello.inventory, &hello.staged_commits, &hello.handoff) @@ -192,7 +212,7 @@ impl StdioNative { } else { RuntimeState::Ready }; - self.write_value(&native_ready(state))?; + self.write_value(&native_ready(state, features))?; } Some("response") => { let response = NativeResponse::parse(value)?; @@ -543,13 +563,15 @@ mod tests { let native = StdioNative::new(output.clone(), lifecycle.clone(), handoff.clone()); let hello = json!({ "protocol": NATIVE_PROTOCOL, - "version": PROTOCOL_VERSION, + "version": NATIVE_VERSION, "kind": "hello", "extension_version": "0.2.0", "inventory": [], "paused": false, "handoff": {"active": false}, - "staged_commits": [] + "staged_commits": [], + "supported_versions": [1], + "supported_features": ["event_ack_v1", "future_feature"] }); let mut input = Vec::new(); write_frame(&mut input, &hello, EXTENSION_TO_HOST_MAX_BYTES).unwrap(); @@ -560,6 +582,38 @@ mod tests { .unwrap() .unwrap(); assert_eq!(ready["kind"], "ready"); + assert_eq!(ready["features"], json!(["event_ack_v1"])); + } + + #[test] + fn incompatible_native_hello_receives_recovery_before_failure() { + let lifecycle = Arc::new(Lifecycle::default()); + let handoff = Arc::new(HandoffState::default()); + let output = SharedWriter::default(); + let native = StdioNative::new(output.clone(), lifecycle, handoff); + let result = native.handle_inbound(json!({ + "protocol": NATIVE_PROTOCOL, + "version": 2, + "kind": "hello", + "extension_version": "3.0.0", + "inventory": [], + "paused": false, + "handoff": {"active": false}, + "staged_commits": [], + "supported_versions": [2], + "supported_features": [] + })); + assert!(matches!( + result, + Err(ProtocolError::UnsupportedProtocol { .. }) + )); + let bytes = output.bytes.lock().clone(); + let incompatible = read_frame(&mut bytes.as_slice(), HOST_TO_EXTENSION_MAX_BYTES) + .unwrap() + .unwrap(); + assert_eq!(incompatible["kind"], "incompatible"); + assert_eq!(incompatible["requested_version"], 2); + assert_eq!(incompatible["supported_versions"], json!([1])); } #[test] fn handoff_clear_is_acknowledged_only_after_sink_applies_it() { @@ -576,7 +630,7 @@ mod tests { native .handle_inbound(json!({ "protocol": NATIVE_PROTOCOL, - "version": PROTOCOL_VERSION, + "version": NATIVE_VERSION, "kind": "event", "event": "handoff_changed", "event_id": "handoff-clear-0001", @@ -596,7 +650,7 @@ mod tests { .unwrap(), json!({ "protocol": NATIVE_PROTOCOL, - "version": PROTOCOL_VERSION, + "version": NATIVE_VERSION, "kind": "event_ack", "event": "handoff_changed", "event_id": "handoff-clear-0001", @@ -618,7 +672,7 @@ mod tests { native .handle_inbound(json!({ "protocol": NATIVE_PROTOCOL, - "version": PROTOCOL_VERSION, + "version": NATIVE_VERSION, "kind": "event", "event": "popup_commit_approved", "event_id": event_id, @@ -650,7 +704,7 @@ mod tests { native .handle_inbound(json!({ "protocol": NATIVE_PROTOCOL, - "version": PROTOCOL_VERSION, + "version": NATIVE_VERSION, "kind": "response", "request_id": request_id, "outcome": "completed", diff --git a/host-rs/crates/agenttab-host/src/runtime.rs b/host-rs/crates/agenttab-host/src/runtime.rs index 3bf2017..43cdc8e 100644 --- a/host-rs/crates/agenttab-host/src/runtime.rs +++ b/host-rs/crates/agenttab-host/src/runtime.rs @@ -1884,6 +1884,8 @@ mod tests { kind: ConnectKind::Connect, conversation_id: None, resume_capability: None, + supported_versions: None, + supported_features: None, }) .unwrap(); (temp, runtime, connection) @@ -1909,6 +1911,8 @@ mod tests { kind: ConnectKind::Connect, conversation_id: None, resume_capability: None, + supported_versions: None, + supported_features: None, }) .unwrap(); (temp, runtime, connection) @@ -1947,6 +1951,8 @@ mod tests { kind: ConnectKind::Connect, conversation_id: None, resume_capability: None, + supported_versions: None, + supported_features: None, }) .unwrap(); (temp, runtime, connection, upload_root) @@ -2238,6 +2244,8 @@ mod tests { kind: ConnectKind::Connect, conversation_id: None, resume_capability: None, + supported_versions: None, + supported_features: None, }) .unwrap(); let response = runtime.handle( @@ -2296,6 +2304,8 @@ mod tests { kind: ConnectKind::Connect, conversation_id: None, resume_capability: None, + supported_versions: None, + supported_features: None, }) .unwrap(); own_tab(&runtime, &connection, 7); diff --git a/host-rs/crates/agenttab-host/src/server.rs b/host-rs/crates/agenttab-host/src/server.rs index 31292b1..df25852 100644 --- a/host-rs/crates/agenttab-host/src/server.rs +++ b/host-rs/crates/agenttab-host/src/server.rs @@ -1,7 +1,7 @@ use crate::runtime::{request_lock_scope, RequestLockScope, Runtime}; use agenttab_protocol::{ - ConnectionInit, Outcome, ResumeCapabilityConfirm, RpcError, RpcMethod, RpcResponse, - CLIENT_TO_HOST_MAX_BYTES, HOST_TO_CLIENT_MAX_BYTES, + connection_incompatible, ConnectionInit, Outcome, ProtocolError, ResumeCapabilityConfirm, + RpcError, RpcMethod, RpcResponse, CLIENT_TO_HOST_MAX_BYTES, HOST_TO_CLIENT_MAX_BYTES, }; #[cfg(all(test, unix))] use agenttab_protocol::{PROTOCOL_VERSION, RPC_PROTOCOL}; @@ -540,8 +540,27 @@ where let Some(init_value) = read_frame_async(&mut stream, CLIENT_TO_HOST_MAX_BYTES).await? else { return Ok(()); }; - let init = ConnectionInit::parse(init_value) - .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + let init = match ConnectionInit::parse(init_value.clone()) { + Ok(init) => init, + Err(ProtocolError::UnsupportedProtocol { .. }) => { + let requested_protocol = init_value + .get("protocol") + .and_then(Value::as_str) + .unwrap_or(""); + let requested_version = init_value + .get("version") + .and_then(Value::as_u64) + .unwrap_or_default(); + write_frame_async( + &mut stream, + &connection_incompatible(requested_protocol, requested_version), + HOST_TO_CLIENT_MAX_BYTES, + ) + .await?; + return Ok(()); + } + Err(error) => return Err(io::Error::new(io::ErrorKind::InvalidData, error)), + }; let resume_capability_supplied = init.resume_capability.is_some(); let (connection, ack) = runtime .connect(init) @@ -1002,6 +1021,8 @@ mod tests { kind: agenttab_protocol::ConnectKind::Connect, conversation_id: None, resume_capability, + supported_versions: None, + supported_features: None, } } @@ -1040,6 +1061,37 @@ mod tests { .unwrap() } + #[tokio::test] + async fn incompatible_connection_receives_a_recovery_frame_before_eof() { + let temp = tempfile::tempdir().unwrap(); + let (runtime, paths) = test_runtime(&temp); + let (mut client, server) = start_test_connection(runtime).await; + write_frame_async( + &mut client, + &serde_json::json!({ + "protocol": RPC_PROTOCOL, + "version": 2, + "kind": "connect", + "supported_versions": [2], + "supported_features": [] + }), + CLIENT_TO_HOST_MAX_BYTES, + ) + .await + .unwrap(); + + let response = read_frame_async(&mut client, HOST_TO_CLIENT_MAX_BYTES) + .await + .unwrap() + .unwrap(); + assert_eq!(response["kind"], "incompatible"); + assert_eq!(response["requested_version"], 2); + assert_eq!(response["supported_versions"], serde_json::json!([1])); + assert!(response["recovery"].as_str().unwrap().contains("Update")); + assert_eq!(task_count(&paths), 0); + assert!(server.await.unwrap().is_ok()); + } + #[tokio::test] async fn explicit_invalid_resume_capability_closes_before_a_pipelined_rpc_can_create_a_task() { let temp = tempfile::tempdir().unwrap(); @@ -1376,6 +1428,8 @@ mod tests { kind: agenttab_protocol::ConnectKind::Connect, conversation_id: None, resume_capability: None, + supported_versions: None, + supported_features: None, }) .unwrap(), CLIENT_TO_HOST_MAX_BYTES, diff --git a/host-rs/crates/agenttab-host/src/task.rs b/host-rs/crates/agenttab-host/src/task.rs index 2b4ff2a..4d43acf 100644 --- a/host-rs/crates/agenttab-host/src/task.rs +++ b/host-rs/crates/agenttab-host/src/task.rs @@ -32,6 +32,7 @@ impl ConnectionContext { journal: &Arc, runtime_state: RuntimeState, ) -> Result<(Arc, ConnectionAck), JournalError> { + let features = init.negotiated_features(); let resumed_lease = match init.resume_capability.as_deref() { Some(capability) => journal.resume_task(capability)?, None => None, @@ -48,6 +49,7 @@ impl ConnectionContext { .as_ref() .map(|lease| lease.resume_capability.clone()), state: runtime_state, + features, }; let context = Arc::new(Self { connection_id: ack.connection_id, @@ -184,6 +186,8 @@ mod tests { kind: ConnectKind::Connect, conversation_id: Some("conversation".into()), resume_capability: capability, + supported_versions: None, + supported_features: None, } } diff --git a/host-rs/crates/agenttab-protocol/src/generated.rs b/host-rs/crates/agenttab-protocol/src/generated.rs new file mode 100644 index 0000000..c415a65 --- /dev/null +++ b/host-rs/crates/agenttab-protocol/src/generated.rs @@ -0,0 +1,128 @@ +// @generated by scripts/generate_protocol.py; do not edit. + +pub const RPC_PROTOCOL: &str = "agenttab.rpc"; +pub const NATIVE_PROTOCOL: &str = "agenttab.native"; +pub const RPC_VERSION: u16 = 1; +pub const NATIVE_VERSION: u16 = 1; +pub const PROTOCOL_VERSION: u16 = RPC_VERSION; +pub const RPC_SUPPORTED_VERSIONS: &[u16] = &[1]; +pub const NATIVE_SUPPORTED_VERSIONS: &[u16] = &[1]; +pub const RPC_FEATURES: &[&str] = &[ + "commit_staging_v1", + "idempotency_v1", + "operation_outcomes_v1", + "task_resume_v1", +]; +pub const NATIVE_FEATURES: &[&str] = &[ + "commit_staging_v1", + "disconnect_recovery_v1", + "event_ack_v1", + "inventory_reconciliation_v1", +]; +pub const RPC_METHOD_NAMES: &[&str] = &[ + "browser_open", + "browser_snapshot", + "browser_act", + "browser_wait", + "browser_tabs", + "browser_handoff", + "browser_commit", + "browser_developer", + "agenttab.status", +]; +pub const MUTATING_RPC_METHOD_NAMES: &[&str] = &[ + "browser_open", + "browser_act", + "browser_handoff", + "browser_commit", + "browser_developer", +]; +pub const NATIVE_METHOD_NAMES: &[&str] = &[ + "browser_open", + "browser_snapshot", + "browser_act", + "browser_wait", + "browser_tabs", + "browser_handoff", + "browser_commit", + "browser_developer", + "commit_review_bind", + "commit_review_abandon", +]; +pub const NATIVE_EVENT_NAMES: &[&str] = &[ + "inventory", + "ownership_revoked", + "tab_removed", + "group_membership_changed", + "pause_changed", + "handoff_changed", + "commit_expired", + "commit_abandoned", + "popup_commit_approved", + "popup_commit_abandoned", + "extension_disconnected", +]; +pub const OUTCOME_NAMES: &[&str] = &[ + "completed", + "not_started", + "unknown", + "needs_user", + "commit_required", +]; +pub const CLIENT_TO_HOST_MAX_BYTES: usize = 1048576; +pub const HOST_TO_CLIENT_MAX_BYTES: usize = 1048576; +pub const HOST_TO_EXTENSION_MAX_BYTES: usize = 1048576; +pub const EXTENSION_TO_HOST_MAX_BYTES: usize = 67108864; + +pub const RPC_SCHEMA_ASSETS: &[(&str, &str)] = &[ + ( + "request", + include_str!("../../../../schemas/rpc/v1/request.schema.json"), + ), + ( + "response", + include_str!("../../../../schemas/rpc/v1/response.schema.json"), + ), + ( + "connection", + include_str!("../../../../schemas/rpc/v1/connection.schema.json"), + ), + ( + "browser_open", + include_str!("../../../../schemas/rpc/v1/browser-open.schema.json"), + ), + ( + "browser_snapshot", + include_str!("../../../../schemas/rpc/v1/browser-snapshot.schema.json"), + ), + ( + "browser_act", + include_str!("../../../../schemas/rpc/v1/browser-act.schema.json"), + ), + ( + "browser_wait", + include_str!("../../../../schemas/rpc/v1/browser-wait.schema.json"), + ), + ( + "browser_tabs", + include_str!("../../../../schemas/rpc/v1/browser-tabs.schema.json"), + ), + ( + "browser_handoff", + include_str!("../../../../schemas/rpc/v1/browser-handoff.schema.json"), + ), + ( + "browser_commit", + include_str!("../../../../schemas/rpc/v1/browser-commit.schema.json"), + ), + ( + "browser_developer", + include_str!("../../../../schemas/rpc/v1/browser-developer.schema.json"), + ), + ( + "status", + include_str!("../../../../schemas/rpc/v1/status.schema.json"), + ), +]; + +pub const NATIVE_SCHEMA: &str = include_str!("../../../../schemas/native/v1/message.schema.json"); diff --git a/host-rs/crates/agenttab-protocol/src/lib.rs b/host-rs/crates/agenttab-protocol/src/lib.rs index 3ab8e17..1564abb 100644 --- a/host-rs/crates/agenttab-protocol/src/lib.rs +++ b/host-rs/crates/agenttab-protocol/src/lib.rs @@ -4,13 +4,8 @@ use std::io::{self, Read, Write}; use thiserror::Error; use uuid::Uuid; -pub const RPC_PROTOCOL: &str = "agenttab.rpc"; -pub const NATIVE_PROTOCOL: &str = "agenttab.native"; -pub const PROTOCOL_VERSION: u16 = 1; -pub const CLIENT_TO_HOST_MAX_BYTES: usize = 1024 * 1024; -pub const HOST_TO_CLIENT_MAX_BYTES: usize = 1024 * 1024; -pub const HOST_TO_EXTENSION_MAX_BYTES: usize = 1024 * 1024; -pub const EXTENSION_TO_HOST_MAX_BYTES: usize = 64 * 1024 * 1024; +mod generated; +pub use generated::*; const MAX_URL_CHARS: usize = 2_048; const MAX_SELECTOR_CHARS: usize = 2_048; @@ -30,59 +25,6 @@ const MAX_DEVELOPER_PARAM_KEY_CHARS: usize = 64; const MAX_DEVELOPER_VALUE_CHARS: usize = 512; const MAX_DEVELOPER_ARRAY_ITEMS: usize = 16; -pub const RPC_SCHEMA_ASSETS: &[(&str, &str)] = &[ - ( - "request", - include_str!("../../../../schemas/rpc/v1/request.schema.json"), - ), - ( - "response", - include_str!("../../../../schemas/rpc/v1/response.schema.json"), - ), - ( - "connection", - include_str!("../../../../schemas/rpc/v1/connection.schema.json"), - ), - ( - "status", - include_str!("../../../../schemas/rpc/v1/status.schema.json"), - ), - ( - "browser_open", - include_str!("../../../../schemas/rpc/v1/browser-open.schema.json"), - ), - ( - "browser_snapshot", - include_str!("../../../../schemas/rpc/v1/browser-snapshot.schema.json"), - ), - ( - "browser_act", - include_str!("../../../../schemas/rpc/v1/browser-act.schema.json"), - ), - ( - "browser_wait", - include_str!("../../../../schemas/rpc/v1/browser-wait.schema.json"), - ), - ( - "browser_tabs", - include_str!("../../../../schemas/rpc/v1/browser-tabs.schema.json"), - ), - ( - "browser_handoff", - include_str!("../../../../schemas/rpc/v1/browser-handoff.schema.json"), - ), - ( - "browser_commit", - include_str!("../../../../schemas/rpc/v1/browser-commit.schema.json"), - ), - ( - "browser_developer", - include_str!("../../../../schemas/rpc/v1/browser-developer.schema.json"), - ), -]; - -pub const NATIVE_SCHEMA: &str = include_str!("../../../../schemas/native/v1/message.schema.json"); - #[derive(Debug, Error)] pub enum ProtocolError { #[error("I/O error: {0}")] @@ -217,7 +159,7 @@ impl RpcRequest { validate_serialized_request_limit(&value)?; let explicit_null_idempotency = value.get("idempotency_key").is_some_and(Value::is_null); let request: Self = serde_json::from_value(value)?; - if request.protocol != RPC_PROTOCOL || request.version != PROTOCOL_VERSION { + if request.protocol != RPC_PROTOCOL || request.version != RPC_VERSION { return Err(ProtocolError::UnsupportedProtocol { protocol: request.protocol.clone(), version: request.version, @@ -941,7 +883,7 @@ impl RpcResponse { pub fn success(request_id: impl Into, outcome: Outcome, result: Value) -> Self { Self { protocol: RPC_PROTOCOL.into(), - version: PROTOCOL_VERSION, + version: RPC_VERSION, request_id: request_id.into(), ok: true, outcome, @@ -954,7 +896,7 @@ impl RpcResponse { pub fn failure(request_id: impl Into, outcome: Outcome, error: RpcError) -> Self { Self { protocol: RPC_PROTOCOL.into(), - version: PROTOCOL_VERSION, + version: RPC_VERSION, request_id: request_id.into(), outcome, ok: false, @@ -984,18 +926,55 @@ pub struct ConnectionInit { pub conversation_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub resume_capability: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub supported_versions: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub supported_features: Option>, } impl ConnectionInit { pub fn parse(value: Value) -> Result { validate_serialized_request_limit(&value)?; let message: Self = serde_json::from_value(value)?; - if message.protocol != RPC_PROTOCOL || message.version != PROTOCOL_VERSION { + if message.protocol != RPC_PROTOCOL { return Err(ProtocolError::UnsupportedProtocol { protocol: message.protocol.clone(), version: message.version, }); } + let supported_versions = message + .supported_versions + .as_deref() + .unwrap_or(std::slice::from_ref(&message.version)); + if supported_versions.is_empty() + || supported_versions.len() > 16 + || supported_versions.contains(&0) + || !supported_versions.contains(&message.version) + || has_duplicates(supported_versions) + { + return Err(ProtocolError::InvalidConnection( + "supported_versions must contain 1 to 16 unique positive versions including version" + .into(), + )); + } + if !supported_versions.contains(&RPC_VERSION) { + return Err(ProtocolError::UnsupportedProtocol { + protocol: message.protocol.clone(), + version: message.version, + }); + } + if message.supported_features.as_ref().is_some_and(|features| { + features.len() > 64 + || features + .iter() + .any(|feature| feature.is_empty() || feature.chars().count() > 128) + || has_duplicates(features) + }) { + return Err(ProtocolError::InvalidConnection( + "supported_features must contain at most 64 unique names of 1 to 128 characters" + .into(), + )); + } if message .conversation_id .as_deref() @@ -1014,6 +993,25 @@ impl ConnectionInit { } Ok(message) } + + pub fn negotiated_features(&self) -> Option> { + self.supported_features.as_ref().map(|supported| { + RPC_FEATURES + .iter() + .filter(|feature| { + supported + .iter() + .any(|candidate| candidate.as_str() == **feature) + }) + .map(|feature| (*feature).into()) + .collect() + }) + } +} + +fn has_duplicates(values: &[T]) -> bool { + let mut seen = std::collections::HashSet::with_capacity(values.len()); + values.iter().any(|value| !seen.insert(value)) } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -1030,7 +1028,7 @@ impl ResumeCapabilityConfirm { pub fn parse(value: Value) -> Result { validate_serialized_request_limit(&value)?; let message: Self = serde_json::from_value(value)?; - if message.protocol != RPC_PROTOCOL || message.version != PROTOCOL_VERSION { + if message.protocol != RPC_PROTOCOL || message.version != RPC_VERSION { return Err(ProtocolError::UnsupportedProtocol { protocol: message.protocol.clone(), version: message.version, @@ -1081,6 +1079,8 @@ pub struct ConnectionAck { #[serde(default, skip_serializing_if = "Option::is_none")] pub resume_capability: Option, pub state: RuntimeState, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub features: Option>, } impl ConnectionAck { @@ -1089,6 +1089,23 @@ impl ConnectionAck { } } +pub fn connection_incompatible(requested_protocol: &str, requested_version: u64) -> Value { + let requested_protocol: String = requested_protocol.chars().take(128).collect(); + let message = format!( + "AgentTab Core does not support {requested_protocol:?} protocol version {requested_version}" + ); + serde_json::json!({ + "protocol": RPC_PROTOCOL, + "version": RPC_VERSION, + "kind": "incompatible", + "requested_protocol": requested_protocol, + "requested_version": requested_version, + "supported_versions": RPC_SUPPORTED_VERSIONS, + "message": message, + "recovery": "Update the AgentTab client and host to releases with an overlapping protocol major.", + }) +} + #[derive(Debug, Clone, Copy, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum ConnectedKind { @@ -1179,16 +1196,53 @@ pub struct NativeHello { pub paused: bool, pub handoff: NativeHandoff, pub staged_commits: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub supported_versions: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub supported_features: Option>, } impl NativeHello { pub fn parse(value: Value) -> Result { let hello: Self = serde_json::from_value(value)?; - if hello.protocol != NATIVE_PROTOCOL || hello.version != PROTOCOL_VERSION { + if hello.protocol != NATIVE_PROTOCOL { return Err(ProtocolError::UnsupportedProtocol { protocol: hello.protocol.clone(), version: hello.version, }); } + let supported_versions = hello + .supported_versions + .as_deref() + .unwrap_or(std::slice::from_ref(&hello.version)); + if supported_versions.is_empty() + || supported_versions.len() > 16 + || supported_versions.contains(&0) + || !supported_versions.contains(&hello.version) + || has_duplicates(supported_versions) + { + return Err(ProtocolError::InvalidNativeMessage( + "supported_versions must contain 1 to 16 unique positive versions including version" + .into(), + )); + } + if !supported_versions.contains(&NATIVE_VERSION) { + return Err(ProtocolError::UnsupportedProtocol { + protocol: hello.protocol.clone(), + version: hello.version, + }); + } + if hello.supported_features.as_ref().is_some_and(|features| { + features.len() > 64 + || features + .iter() + .any(|feature| feature.is_empty() || feature.chars().count() > 128) + || has_duplicates(features) + }) { + return Err(ProtocolError::InvalidNativeMessage( + "supported_features must contain at most 64 unique names of 1 to 128 characters" + .into(), + )); + } if hello.extension_version.is_empty() { return Err(ProtocolError::InvalidNativeMessage( "extension_version must not be empty".into(), @@ -1201,6 +1255,20 @@ impl NativeHello { } Ok(hello) } + + pub fn negotiated_features(&self) -> Option> { + self.supported_features.as_ref().map(|supported| { + NATIVE_FEATURES + .iter() + .filter(|feature| { + supported + .iter() + .any(|candidate| candidate.as_str() == **feature) + }) + .map(|feature| (*feature).into()) + .collect() + }) + } } #[derive(Debug, Clone, Copy, Serialize, Deserialize)] @@ -1222,7 +1290,7 @@ pub struct NativeDisconnectRecovery { impl NativeDisconnectRecovery { pub fn parse(value: Value) -> Result { let recovery: Self = serde_json::from_value(value)?; - if recovery.protocol != NATIVE_PROTOCOL || recovery.version != PROTOCOL_VERSION { + if recovery.protocol != NATIVE_PROTOCOL || recovery.version != NATIVE_VERSION { return Err(ProtocolError::UnsupportedProtocol { protocol: recovery.protocol.clone(), version: recovery.version, @@ -1261,7 +1329,7 @@ pub struct NativeCloseTask { impl NativeCloseTask { pub fn parse(value: Value) -> Result { let command: Self = serde_json::from_value(value)?; - if command.protocol != NATIVE_PROTOCOL || command.version != PROTOCOL_VERSION { + if command.protocol != NATIVE_PROTOCOL || command.version != NATIVE_VERSION { return Err(ProtocolError::UnsupportedProtocol { protocol: command.protocol.clone(), version: command.version, @@ -1303,7 +1371,7 @@ impl NativeResponse { )); } let response: Self = serde_json::from_value(value)?; - if response.protocol != NATIVE_PROTOCOL || response.version != PROTOCOL_VERSION { + if response.protocol != NATIVE_PROTOCOL || response.version != NATIVE_VERSION { return Err(ProtocolError::UnsupportedProtocol { protocol: response.protocol.clone(), version: response.version, @@ -1456,7 +1524,7 @@ pub struct NativeDisconnectEvent { impl NativeEvent { pub fn parse(value: Value) -> Result<(Self, NativeEventPayload), ProtocolError> { let event: Self = serde_json::from_value(value)?; - if event.protocol != NATIVE_PROTOCOL || event.version != PROTOCOL_VERSION { + if event.protocol != NATIVE_PROTOCOL || event.version != NATIVE_VERSION { return Err(ProtocolError::UnsupportedProtocol { protocol: event.protocol.clone(), version: event.version, @@ -1605,7 +1673,7 @@ pub fn native_command( ) -> Value { let mut command = serde_json::json!({ "protocol": NATIVE_PROTOCOL, - "version": PROTOCOL_VERSION, + "version": NATIVE_VERSION, "kind": "command", "request_id": request_id, "connection_id": connection_id, @@ -1621,7 +1689,7 @@ pub fn native_command( pub fn native_close_task(request_id: Uuid, task_id: Uuid) -> Value { serde_json::json!({ "protocol": NATIVE_PROTOCOL, - "version": PROTOCOL_VERSION, + "version": NATIVE_VERSION, "kind": "close_task", "request_id": request_id, "task_id": task_id, @@ -1631,7 +1699,7 @@ pub fn native_close_task(request_id: Uuid, task_id: Uuid) -> Value { pub fn native_event_ack(event: NativeEventName, event_id: &str) -> Value { serde_json::json!({ "protocol": NATIVE_PROTOCOL, - "version": PROTOCOL_VERSION, + "version": NATIVE_VERSION, "kind": "event_ack", "event": event, "event_id": event_id, @@ -1665,13 +1733,34 @@ pub fn native_event_ack_result( value } -pub fn native_ready(state: RuntimeState) -> Value { - serde_json::json!({ +pub fn native_ready(state: RuntimeState, features: Option>) -> Value { + let mut message = serde_json::json!({ "protocol": NATIVE_PROTOCOL, - "version": PROTOCOL_VERSION, + "version": NATIVE_VERSION, "kind": "ready", "host_version": env!("CARGO_PKG_VERSION"), "state": state, + }); + if let Some(features) = features { + message["features"] = serde_json::json!(features); + } + message +} + +pub fn native_incompatible(requested_protocol: &str, requested_version: u64) -> Value { + let requested_protocol: String = requested_protocol.chars().take(128).collect(); + let message = format!( + "AgentTab host does not support {requested_protocol:?} native protocol version {requested_version}" + ); + serde_json::json!({ + "protocol": NATIVE_PROTOCOL, + "version": NATIVE_VERSION, + "kind": "incompatible", + "requested_protocol": requested_protocol, + "requested_version": requested_version, + "supported_versions": NATIVE_SUPPORTED_VERSIONS, + "message": message, + "recovery": "Update the AgentTab extension and host to releases with an overlapping native protocol major.", }) } @@ -1718,6 +1807,75 @@ mod tests { } } + #[test] + fn generated_catalog_matches_runtime_enums() { + let methods = [ + RpcMethod::BrowserOpen, + RpcMethod::BrowserSnapshot, + RpcMethod::BrowserAct, + RpcMethod::BrowserWait, + RpcMethod::BrowserTabs, + RpcMethod::BrowserHandoff, + RpcMethod::BrowserCommit, + RpcMethod::BrowserDeveloper, + RpcMethod::AgenttabStatus, + ]; + assert_eq!( + methods.map(|method| method.to_string()).to_vec(), + RPC_METHOD_NAMES + .iter() + .map(|method| (*method).to_owned()) + .collect::>() + ); + assert_eq!( + methods + .into_iter() + .filter(|method| method.is_mutation()) + .map(|method| method.to_string()) + .collect::>(), + MUTATING_RPC_METHOD_NAMES + .iter() + .map(|method| (*method).to_owned()) + .collect::>() + ); + let outcomes = [ + Outcome::Completed, + Outcome::NotStarted, + Outcome::Unknown, + Outcome::NeedsUser, + Outcome::CommitRequired, + ] + .map(|outcome| serde_json::to_value(outcome).unwrap()); + assert_eq!( + outcomes + .iter() + .map(|outcome| outcome.as_str().unwrap()) + .collect::>(), + OUTCOME_NAMES.to_vec() + ); + let native_events = [ + NativeEventName::Inventory, + NativeEventName::OwnershipRevoked, + NativeEventName::TabRemoved, + NativeEventName::GroupMembershipChanged, + NativeEventName::PauseChanged, + NativeEventName::HandoffChanged, + NativeEventName::CommitExpired, + NativeEventName::CommitAbandoned, + NativeEventName::PopupCommitApproved, + NativeEventName::PopupCommitAbandoned, + NativeEventName::ExtensionDisconnected, + ] + .map(|event| serde_json::to_value(event).unwrap()); + assert_eq!( + native_events + .iter() + .map(|event| event.as_str().unwrap()) + .collect::>(), + NATIVE_EVENT_NAMES.to_vec() + ); + } + #[test] fn request_envelope_rejects_unknown_fields_and_non_v7_mutation_keys() { let unknown = json!({ @@ -1971,6 +2129,31 @@ mod tests { #[test] fn connection_runtime_constraints_match_schema() { + let negotiated = ConnectionInit::parse(json!({ + "protocol": RPC_PROTOCOL, + "version": 2, + "kind": "connect", + "supported_versions": [1, 2], + "supported_features": ["task_resume_v1", "future_feature"] + })) + .unwrap(); + assert_eq!( + negotiated.negotiated_features().unwrap(), + vec!["task_resume_v1"] + ); + assert!(matches!( + ConnectionInit::parse(json!({ + "protocol": RPC_PROTOCOL, + "version": 2, + "kind": "connect", + "supported_versions": [2] + })), + Err(ProtocolError::UnsupportedProtocol { .. }) + )); + let incompatible = connection_incompatible(RPC_PROTOCOL, 2); + assert_eq!(incompatible["kind"], "incompatible"); + assert_eq!(incompatible["supported_versions"], json!([1])); + assert!(matches!( ConnectionInit::parse(json!({ "protocol": RPC_PROTOCOL, @@ -2029,6 +2212,45 @@ mod tests { assert_eq!(confirmed.value()["kind"], "resume_confirmed"); } + #[test] + fn native_hello_negotiates_features_and_versions() { + let hello = NativeHello::parse(json!({ + "protocol": NATIVE_PROTOCOL, + "version": 2, + "kind": "hello", + "extension_version": "2.0.0", + "inventory": [], + "paused": false, + "handoff": {"active": false}, + "staged_commits": [], + "supported_versions": [1, 2], + "supported_features": ["event_ack_v1", "future_feature"] + })) + .unwrap(); + assert_eq!( + hello.negotiated_features().unwrap(), + vec!["event_ack_v1"] + ); + assert!(matches!( + NativeHello::parse(json!({ + "protocol": NATIVE_PROTOCOL, + "version": 2, + "kind": "hello", + "extension_version": "2.0.0", + "inventory": [], + "paused": false, + "handoff": {"active": false}, + "staged_commits": [], + "supported_versions": [2] + })), + Err(ProtocolError::UnsupportedProtocol { .. }) + )); + assert_eq!( + native_ready(RuntimeState::Ready, Some(vec!["event_ack_v1".into()]))["features"], + json!(["event_ack_v1"]) + ); + } + #[test] fn maximum_schema_bound_requests_fit_the_one_megabyte_client_frame() { let escaped = |count| "\0".repeat(count); diff --git a/package.json b/package.json index 4211598..12608d4 100644 --- a/package.json +++ b/package.json @@ -6,6 +6,8 @@ "packages/*" ], "scripts": { + "protocol:generate": "python3 scripts/generate_protocol.py", + "protocol:check": "python3 scripts/generate_protocol.py --check", "extension:build": "bun run --cwd packages/extension build", "extension:check": "bun run --cwd packages/extension typecheck && bun run --cwd packages/extension test && bun run --cwd packages/extension build", "workspace:typecheck": "bun run --cwd packages/extension typecheck && bun run --cwd packages/sdk-typescript typecheck && bun run --cwd packages/mcp typecheck && bun run --cwd packages/omp typecheck && bun run --cwd packages/installer typecheck && bun run --cwd packages/site typecheck", diff --git a/packages/extension/src/generated/protocol.ts b/packages/extension/src/generated/protocol.ts new file mode 100644 index 0000000..bc0fcac --- /dev/null +++ b/packages/extension/src/generated/protocol.ts @@ -0,0 +1,49 @@ +// @generated by scripts/generate_protocol.py; do not edit. + +export const NATIVE_PROTOCOL = "agenttab.native" as const; +export const PROTOCOL_VERSION = 1 as const; +export const NATIVE_SUPPORTED_VERSIONS = [ + 1, +] as const; +export const NATIVE_FEATURES = [ + "commit_staging_v1", + "disconnect_recovery_v1", + "event_ack_v1", + "inventory_reconciliation_v1", +] as const; +export const OUTCOMES = [ + "completed", + "not_started", + "unknown", + "needs_user", + "commit_required", +] as const; +export const NATIVE_METHODS = { + "browser_open": true, + "browser_snapshot": true, + "browser_act": true, + "browser_wait": true, + "browser_tabs": true, + "browser_handoff": true, + "browser_commit": true, + "browser_developer": true, + "commit_review_bind": true, + "commit_review_abandon": true +} as const; +export const NATIVE_EVENTS = { + "inventory": true, + "ownership_revoked": true, + "tab_removed": true, + "group_membership_changed": true, + "pause_changed": true, + "handoff_changed": true, + "commit_expired": true, + "commit_abandoned": true, + "popup_commit_approved": true, + "popup_commit_abandoned": true, + "extension_disconnected": true +} as const; + +export type NativeMethod = keyof typeof NATIVE_METHODS; +export type NativeEventName = keyof typeof NATIVE_EVENTS; +export type GeneratedOutcome = typeof OUTCOMES[number]; diff --git a/packages/extension/src/native.ts b/packages/extension/src/native.ts index b948f05..d57b5b6 100644 --- a/packages/extension/src/native.ts +++ b/packages/extension/src/native.ts @@ -33,6 +33,14 @@ interface ReadyReconciliation { promise: Promise; } +interface NativeHandshakeAttempt { + port: NativePort; + advertisedCapabilities: boolean; + helloSent: boolean; + receivedMessage: boolean; + locallyClosed: boolean; +} + interface NativePort { postMessage(message: unknown): void; disconnect(): void; @@ -75,6 +83,10 @@ export class NativeBridge { private reconnectFallbackAtMs: number | null = null; private readonly pendingEventAcks = new Map(); private readyReconciliation: ReadyReconciliation | null = null; + private protocolIncompatible = false; + // Do not alternate handshake shapes inside the existing reconnect loop. + private useLegacyV1Handshake = false; + private handshakeAttempt: NativeHandshakeAttempt | null = null; constructor( private readonly scheduler: MutationScheduler, private readonly ownership: OwnershipLedger, @@ -86,7 +98,7 @@ export class NativeBridge { ) { } async connect(): Promise { - if (this.port) return; + if (this.protocolIncompatible || this.port) return; this.clearReconnectTimer(); let port: NativePort; try { @@ -98,16 +110,24 @@ export class NativeBridge { this.port = port; this.ready = false; this.readyReconciliation = null; + const handshakeAttempt: NativeHandshakeAttempt = { + port, + advertisedCapabilities: !this.useLegacyV1Handshake, + helloSent: false, + receivedMessage: false, + locallyClosed: false, + }; + this.handshakeAttempt = handshakeAttempt; this.ensureReconnectAlarm(RECONNECT_ALARM_FLOOR_MS); port.onMessage.addListener((message: unknown) => { void this.onMessage(port, message).catch(() => { - if (this.port === port) port.disconnect(); + if (this.port === port) this.disconnectLocally(port); }); }); port.onDisconnect.addListener(() => void this.onDisconnect(port)); try { const state = await readState(); - port.postMessage(nativeHello( + const hello = nativeHello( chrome.runtime.getManifest().version, nativeInventory(await this.ownership.inventory()), state.paused, @@ -127,7 +147,10 @@ export class NativeBridge { approved: _approved, ...staged }) => staged), - )); + handshakeAttempt.advertisedCapabilities, + ); + handshakeAttempt.helloSent = true; + port.postMessage(hello); } catch { this.onDisconnect(port); } @@ -222,13 +245,14 @@ export class NativeBridge { this.clearReconnectTimer(); if (this.port) { if (this.ready) return; - this.port.disconnect(); + this.disconnectLocally(this.port); return; } await this.connect(); } private async onMessage(port: NativePort, message: unknown): Promise { if (this.port !== port) return; + if (this.handshakeAttempt?.port === port) this.handshakeAttempt.receivedMessage = true; let parsed: NativeInboundMessage; try { parsed = parseInboundNativeMessage(message); @@ -243,7 +267,7 @@ export class NativeBridge { port.postMessage(failed(message.request_id, "invalid_request", error instanceof Error ? error.message : String(error))); return; } - if (this.port === port) port.disconnect(); + if (this.port === port) this.disconnectLocally(port); return; } if (parsed.kind === "event_ack") { @@ -257,9 +281,18 @@ export class NativeBridge { } return; } + if (parsed.kind === "incompatible") { + this.protocolIncompatible = true; + this.clearReconnectTimer(); + this.reconnectFallbackAtMs = null; + await chrome.alarms.clear(RECONNECT_ALARM); + console.error(`${parsed.message} ${parsed.recovery}`); + this.disconnectLocally(port); + return; + } if (parsed.kind === "ready") { if (this.readyReconciliation?.port === port) { - port.disconnect(); + this.disconnectLocally(port); return; } const promise = this.reconcileReady(port, parsed); @@ -278,7 +311,7 @@ export class NativeBridge { if (this.port !== port) return; } if (!this.ready) { - if (this.port === port) port.disconnect(); + if (this.port === port) this.disconnectLocally(port); return; } let response: NativeResponse; @@ -308,10 +341,11 @@ export class NativeBridge { await this.discardStages(parsed.discard_staged_tokens ?? []); if (this.port !== port) return; } catch { - if (this.port === port) port.disconnect(); + if (this.port === port) this.disconnectLocally(port); return; } this.ready = true; + this.handshakeAttempt = null; this.reconnectAttempt = 0; this.clearReconnectTimer(); this.reconnectFallbackAtMs = null; @@ -335,9 +369,17 @@ export class NativeBridge { private onDisconnect(port: NativePort): void { if (this.port !== port) return; + const attempt = this.handshakeAttempt?.port === port ? this.handshakeAttempt : null; + const retryLegacyHandshake = + !this.protocolIncompatible && + attempt?.advertisedCapabilities === true && + attempt.helloSent && + !attempt.receivedMessage && + !attempt.locallyClosed; this.port = null; this.ready = false; this.readyReconciliation = null; + this.handshakeAttempt = null; for (const [eventId, pending] of this.pendingEventAcks) { clearTimeout(pending.timeout); pending.reject(Object.assign(new Error("AgentTab host disconnected before acknowledging the staged action"), { @@ -346,12 +388,22 @@ export class NativeBridge { this.pendingEventAcks.delete(eventId); } this.scheduler.disconnect(); + if (retryLegacyHandshake) { + this.useLegacyV1Handshake = true; + void this.connect(); + return; + } void this.discardStages([]); this.scheduleReconnect(); } + private disconnectLocally(port: NativePort): void { + if (this.handshakeAttempt?.port === port) this.handshakeAttempt.locallyClosed = true; + port.disconnect(); + } + private scheduleReconnect(): void { - if (this.port || this.cancelReconnectTimer) return; + if (this.protocolIncompatible || this.port || this.cancelReconnectTimer) return; const delay = Math.min(1000 * 2 ** this.reconnectAttempt, RECONNECT_MAX_MS); this.reconnectAttempt += 1; this.ensureReconnectAlarm(delay); diff --git a/packages/extension/src/protocol.ts b/packages/extension/src/protocol.ts index 1dc7f22..f428874 100644 --- a/packages/extension/src/protocol.ts +++ b/packages/extension/src/protocol.ts @@ -1,69 +1,32 @@ import { hasOnlyKeys, isBoundedString, isIntegerInRange, isRecord } from "./type-guards"; +import { + NATIVE_EVENTS, + NATIVE_FEATURES, + NATIVE_METHODS, + NATIVE_PROTOCOL, + NATIVE_SUPPORTED_VERSIONS, + PROTOCOL_VERSION, + type GeneratedOutcome, + type NativeEventName, + type NativeMethod, +} from "./generated/protocol"; + +export { + NATIVE_EVENTS, + NATIVE_FEATURES, + NATIVE_METHODS, + NATIVE_PROTOCOL, + NATIVE_SUPPORTED_VERSIONS, + PROTOCOL_VERSION, + type NativeEventName, + type NativeMethod, +} from "./generated/protocol"; -export const NATIVE_PROTOCOL = "agenttab.native"; -export const PROTOCOL_VERSION = 1; export const SNAPSHOT_TEXT_MAX_BYTES = 1_000_000; export const SCREENSHOT_MAX_BYTES = 750_000; export const SCREENSHOT_MAX_DIMENSION = 16_384; -export type NativeMethod = - | "browser_open" - | "browser_snapshot" - | "browser_act" - | "browser_wait" - | "browser_tabs" - | "browser_handoff" - | "browser_commit" - | "browser_developer" - | "commit_review_bind" - | "commit_review_abandon"; - -const CORE_METHODS: Record = { - browser_open: true, - browser_snapshot: true, - browser_act: true, - browser_wait: true, - browser_tabs: true, - browser_handoff: true, - browser_commit: true, - browser_developer: true, - commit_review_bind: true, - commit_review_abandon: true, -}; - -const NATIVE_EVENTS: Record = { - inventory: true, - ownership_revoked: true, - tab_removed: true, - group_membership_changed: true, - pause_changed: true, - handoff_changed: true, - commit_expired: true, - commit_abandoned: true, - popup_commit_approved: true, - popup_commit_abandoned: true, - extension_disconnected: true, -}; - -export type NativeEventName = - | "inventory" - | "ownership_revoked" - | "tab_removed" - | "group_membership_changed" - | "pause_changed" - | "handoff_changed" - | "commit_expired" - | "commit_abandoned" - | "popup_commit_approved" - | "popup_commit_abandoned" - | "extension_disconnected"; - -export type Outcome = - | "completed" - | "not_started" - | "unknown" - | "needs_user" - | "commit_required"; +export type Outcome = GeneratedOutcome; export interface NativeOriginPolicy { tab_id: number; @@ -111,9 +74,25 @@ export interface NativeReady { host_version: string; state: "ready" | "paused"; discard_staged_tokens?: string[]; + features?: string[]; } -export type NativeInboundMessage = NativeDispatchCommand | NativeReady | NativeEventAck; +export interface NativeIncompatible { + protocol: typeof NATIVE_PROTOCOL; + version: number; + kind: "incompatible"; + requested_protocol: string; + requested_version: number; + supported_versions: number[]; + message: string; + recovery: string; +} + +export type NativeInboundMessage = + | NativeDispatchCommand + | NativeReady + | NativeEventAck + | NativeIncompatible; export interface RpcError { code: string; message: string; @@ -517,7 +496,7 @@ export function parseCommand(value: unknown): NativeCommand { !UUID_PATTERN.test(value.connection_id) || !UUID_PATTERN.test(value.task_id)) { throw new Error("native command IDs must be UUIDs"); } - if (typeof value.method !== "string" || !Object.hasOwn(CORE_METHODS, value.method)) { + if (typeof value.method !== "string" || !Object.hasOwn(NATIVE_METHODS, value.method)) { throw new Error("native command method is unsupported"); } const originPolicy = value.origin_policy === undefined @@ -553,11 +532,48 @@ function parseCloseTask(value: unknown): NativeCloseTask { } export function parseInboundNativeMessage(value: unknown): NativeInboundMessage { - if (!isRecord(value) || value.protocol !== NATIVE_PROTOCOL || value.version !== PROTOCOL_VERSION || typeof value.kind !== "string") { + if (!isRecord(value) || value.protocol !== NATIVE_PROTOCOL || typeof value.kind !== "string") { + throw new Error("native message protocol or version mismatch"); + } + if (value.kind !== "incompatible" && value.version !== PROTOCOL_VERSION) { throw new Error("native message protocol or version mismatch"); } if (value.kind === "command") return parseCommand(value); if (value.kind === "close_task") return parseCloseTask(value); + if (value.kind === "incompatible") { + if ( + !hasOnlyKeys(value, [ + "protocol", + "version", + "kind", + "requested_protocol", + "requested_version", + "supported_versions", + "message", + "recovery", + ]) || + !isIntegerInRange(value.version, 1, 65_535) || + !isBoundedString(value.requested_protocol, 0, 128) || + !isIntegerInRange(value.requested_version, 0, 65_535) || + !Array.isArray(value.supported_versions) || + value.supported_versions.length === 0 || + !value.supported_versions.every((version) => isIntegerInRange(version, 1, 65_535)) || + !isBoundedString(value.message, 1, 2_000) || + !isBoundedString(value.recovery, 1, 2_000) + ) { + throw new Error("native incompatibility message is invalid"); + } + return { + protocol: NATIVE_PROTOCOL, + version: value.version, + kind: "incompatible", + requested_protocol: value.requested_protocol, + requested_version: value.requested_version, + supported_versions: [...value.supported_versions] as number[], + message: value.message, + recovery: value.recovery, + }; + } if (value.kind === "event_ack") { if ( !hasOnlyKeys( @@ -616,12 +632,19 @@ export function parseInboundNativeMessage(value: unknown): NativeInboundMessage } if ( value.kind !== "ready" || - !hasOnlyKeys(value, ["protocol", "version", "kind", "host_version", "state"], ["discard_staged_tokens"]) || + !hasOnlyKeys( + value, + ["protocol", "version", "kind", "host_version", "state"], + ["discard_staged_tokens", "features"], + ) || !isBoundedString(value.host_version, 1, 128) || (value.state !== "ready" && value.state !== "paused") || (value.discard_staged_tokens !== undefined && (!Array.isArray(value.discard_staged_tokens) || - !value.discard_staged_tokens.every((token) => isBoundedString(token, 16, 256)))) + !value.discard_staged_tokens.every((token) => isBoundedString(token, 16, 256)))) || + (value.features !== undefined && + (!Array.isArray(value.features) || + !value.features.every((feature) => isBoundedString(feature, 1, 128)))) ) { throw new Error("native ready message is invalid"); } @@ -634,6 +657,9 @@ export function parseInboundNativeMessage(value: unknown): NativeInboundMessage ...(value.discard_staged_tokens === undefined ? {} : { discard_staged_tokens: [...value.discard_staged_tokens] as string[] }), + ...(value.features === undefined + ? {} + : { features: [...value.features] as string[] }), }; } export function completed(requestId: string, result: unknown): NativeResponse { @@ -699,6 +725,7 @@ export function nativeHello( paused: boolean, handoff: NativeHandoff, stagedCommits: PublicStagedCommit[], + advertiseCapabilities = true, ) { return { protocol: NATIVE_PROTOCOL, @@ -709,6 +736,12 @@ export function nativeHello( paused, handoff, staged_commits: stagedCommits, + ...(advertiseCapabilities + ? { + supported_versions: NATIVE_SUPPORTED_VERSIONS, + supported_features: NATIVE_FEATURES, + } + : {}), }; } diff --git a/packages/extension/test/extension.test.ts b/packages/extension/test/extension.test.ts index da6a015..9ab8458 100644 --- a/packages/extension/test/extension.test.ts +++ b/packages/extension/test/extension.test.ts @@ -964,6 +964,31 @@ describe("native protocol", () => { connection_id: NATIVE_CONNECTION_ID, })).toThrow("unknown fields"); }); + + test("parses explicit native compatibility and negotiated features", () => { + expect(parseInboundNativeMessage({ + protocol: "agenttab.native", + version: 1, + kind: "incompatible", + requested_protocol: "agenttab.native", + requested_version: 2, + supported_versions: [1], + message: "Native protocol version 2 is unsupported", + recovery: "Update AgentTab.", + })).toMatchObject({ + kind: "incompatible", + requested_version: 2, + supported_versions: [1], + }); + expect(parseInboundNativeMessage({ + protocol: "agenttab.native", + version: 1, + kind: "ready", + host_version: "2.0.0", + state: "ready", + features: ["event_ack_v1"], + })).toMatchObject({ kind: "ready", features: ["event_ack_v1"] }); + }); }); describe("automation route classification", () => { @@ -3099,6 +3124,152 @@ describe("handoff and pause barriers", () => { }); describe("native bridge transport", () => { + test("falls back once to legacy v1 after a pre-ready close and keeps recovery bounded", async () => { + const scheduler = new MutationScheduler(); + await seedTask(TASK_A, [44]); + const stagedToken = "native-token-survives-v1-fallback"; + await mutateState((state) => { + state.stagedCommits[stagedToken] = { + native_token: stagedToken, + task_id: TASK_A, + tab_id: 44, + page_revision: 1, + effect: "Submit the staged form", + fingerprint: "f".repeat(64), + expires_at_ms: Date.now() + 60_000, + action: { kind: "click", ref: "r1-1" }, + preview: { kind: "click" }, + }; + }); + const ownership = new OwnershipLedger(scheduler, new RevisionTracker(), () => undefined); + const capabilityPort = new MockNativePort(); + const legacyPort = new MockNativePort(); + const recoveredPort = new MockNativePort(); + const scheduled: Array<{ callback: () => void; delayMs: number }> = []; + const discarded: Array = []; + nativePort = capabilityPort; + const bridge = new NativeBridge( + scheduler, + ownership, + async () => { + throw new Error("command handler must not run"); + }, + undefined, + undefined, + async (tokens) => { + discarded.push([...tokens]); + if (tokens.length === 0) { + await mutateState((state) => { + state.stagedCommits = {}; + }); + } + }, + { + now: () => 123_000, + schedule: (callback, delayMs) => { + scheduled.push({ callback, delayMs }); + return () => undefined; + }, + }, + ); + + await bridge.connect(); + expect(capabilityPort.posted[0]).toMatchObject({ + kind: "hello", + supported_versions: [1], + }); + + nativePort = legacyPort; + capabilityPort.disconnect(); + await waitForCondition(() => legacyPort.posted.length === 1); + const legacyHello = legacyPort.posted[0] as Record; + expect(legacyHello).toMatchObject({ + protocol: "agenttab.native", + version: 1, + kind: "hello", + extension_version: "2.0.0", + }); + expect(legacyHello).not.toHaveProperty("supported_versions"); + expect(legacyHello).not.toHaveProperty("supported_features"); + expect(legacyHello.staged_commits).toEqual([{ + native_token: stagedToken, + task_id: TASK_A, + tab_id: 44, + page_revision: 1, + effect: "Submit the staged form", + fingerprint: "f".repeat(64), + expires_at_ms: expect.any(Number), + }]); + expect((await readState()).stagedCommits[stagedToken]).toBeDefined(); + expect(discarded).toHaveLength(0); + expect(scheduled).toHaveLength(0); + + nativePort = recoveredPort; + legacyPort.disconnect(); + await waitForCondition(() => scheduled.length === 1); + await waitForCondition(() => discarded.length === 1); + expect(discarded).toEqual([[]]); + expect((await readState()).stagedCommits[stagedToken]).toBeUndefined(); + expect(scheduled[0]?.delayMs).toBe(1_000); + expect(recoveredPort.posted).toHaveLength(0); + + scheduled[0]?.callback(); + await waitForCondition(() => recoveredPort.posted.length === 1); + expect(recoveredPort.posted[0]).not.toHaveProperty("supported_versions"); + }); + + test("treats an explicit native incompatibility as terminal without legacy fallback", async () => { + const scheduler = new MutationScheduler(); + const ownership = new OwnershipLedger(scheduler, new RevisionTracker(), () => undefined); + const capabilityPort = new MockNativePort(); + const unusedLegacyPort = new MockNativePort(); + const scheduled: Array<() => void> = []; + nativePort = capabilityPort; + const bridge = new NativeBridge( + scheduler, + ownership, + async () => { + throw new Error("command handler must not run"); + }, + undefined, + undefined, + undefined, + { + now: () => 123_000, + schedule: (callback) => { + scheduled.push(callback); + return () => undefined; + }, + }, + ); + const loggedErrors: unknown[][] = []; + const originalConsoleError = console.error; + console.error = (...values: unknown[]) => { + loggedErrors.push(values); + }; + try { + await bridge.connect(); + nativePort = unusedLegacyPort; + capabilityPort.receive({ + protocol: "agenttab.native", + version: 1, + kind: "incompatible", + requested_protocol: "agenttab.native", + requested_version: 2, + supported_versions: [1], + message: "Native protocol version 2 is unsupported", + recovery: "Update AgentTab.", + }); + await waitForCondition(() => capabilityPort.disconnectCount === 1); + + expect(unusedLegacyPort.posted).toHaveLength(0); + expect(scheduled).toHaveLength(0); + expect(loggedErrors).toHaveLength(1); + } finally { + console.error = originalConsoleError; + } + }); + test("reconciles hello, resets backoff only after ready, and pauses on disconnect", async () => { const scheduler = new MutationScheduler(); const revisions = new RevisionTracker(); @@ -3133,6 +3304,13 @@ describe("native bridge transport", () => { paused: false, handoff: { active: false }, staged_commits: [], + supported_versions: [1], + supported_features: [ + "commit_staging_v1", + "disconnect_recovery_v1", + "event_ack_v1", + "inventory_reconciliation_v1", + ], }); expect(scheduler.isAccepting()).toBe(false); diff --git a/packages/mcp/src/server.ts b/packages/mcp/src/server.ts index 6905d94..3f4c8f4 100644 --- a/packages/mcp/src/server.ts +++ b/packages/mcp/src/server.ts @@ -4,6 +4,8 @@ import { AgentTabClient, AgentTabError, AgentTabTransportError, + MUTATING_RPC_METHODS, + RPC_TOOL_METADATA, createUuidV7, createResumeCapabilityStore, type MethodParams, @@ -27,13 +29,7 @@ export const MCP_MAX_LINE_BYTES = 1024 * 1024 + 64 * 1024; export const MCP_INLINE_RESULT_MAX_BYTES = 8 * 1024; const IDEMPOTENCY_KEY_CACHE_MAX_ENTRIES = 4_096; -const MUTATIONS = new Set([ - "browser_open", - "browser_act", - "browser_handoff", - "browser_commit", - "browser_developer", -]); +const MUTATIONS = new Set(MUTATING_RPC_METHODS); const UUID_V7 = /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; class InvocationIdempotencyKeys { @@ -151,44 +147,44 @@ const schema = (value: Record): Record => { export const STANDARD_TOOLS: readonly Tool[] = [ { name: "browser_open", - description: "Create a task tab, create an unfocused task-owned window, or explicitly adopt the active tab.", + description: RPC_TOOL_METADATA.browser_open.description, inputSchema: schema(openSchema), }, { name: "browser_snapshot", - description: "Read an accessibility snapshot, bounded text or HTML, or a screenshot from a task-owned tab.", + description: RPC_TOOL_METADATA.browser_snapshot.description, inputSchema: schema(snapshotSchema), }, { name: "browser_act", - description: "Run an ordered batch of typed actions against one task-owned tab and page revision.", + description: RPC_TOOL_METADATA.browser_act.description, inputSchema: schema(actSchema), }, { name: "browser_wait", - description: "Wait for one schema-defined load, URL, text, selector, network-idle, or download condition.", + description: RPC_TOOL_METADATA.browser_wait.description, inputSchema: schema(waitSchema), }, { name: "browser_tabs", - description: "List only tabs owned by this task connection.", + description: RPC_TOOL_METADATA.browser_tabs.description, inputSchema: schema(tabsSchema), }, { name: "browser_handoff", - description: "Pause all agent actions and give the user control for credentials, MFA, CAPTCHA, or other human-only input.", + description: RPC_TOOL_METADATA.browser_handoff.description, inputSchema: schema(handoffSchema), }, { name: "browser_commit", - description: "Execute one previously staged consequential action after semantic review.", + description: RPC_TOOL_METADATA.browser_commit.description, inputSchema: schema(commitSchema), }, ] as const; export const DEVELOPER_TOOL: Tool = { name: "browser_developer", - description: "Run an explicitly enabled developer-mode action outside the Standard tool surface.", + description: RPC_TOOL_METADATA.browser_developer.description, inputSchema: schema(developerSchema), }; diff --git a/packages/omp/src/index.ts b/packages/omp/src/index.ts index bbcb4a2..6db7fe0 100644 --- a/packages/omp/src/index.ts +++ b/packages/omp/src/index.ts @@ -3,6 +3,8 @@ import { AgentTabClient, AgentTabError, AgentTabTransportError, + MUTATING_RPC_METHODS, + RPC_TOOL_METADATA, SCREENSHOT_MAX_BYTES, SCREENSHOT_MAX_DIMENSION, SNAPSHOT_TEXT_MAX_BYTES, @@ -59,13 +61,7 @@ interface ToolExecutionContext { type ClientFactory = (context?: ToolExecutionContext) => Promise; -const MUTATIONS = new Set([ - "browser_open", - "browser_act", - "browser_handoff", - "browser_commit", - "browser_developer", -]); +const MUTATIONS = new Set(MUTATING_RPC_METHODS); const INLINE_RESULT_MAX_BYTES = 8 * 1024; const IDEMPOTENCY_KEY_CACHE_MAX_ENTRIES = 4_096; const UUID_V7 = /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; @@ -137,7 +133,7 @@ const DEFINITIONS: ReadonlyArray<{ { name: "browser_open", label: "Browser Open", - description: "Create a task tab, create an unfocused task-owned window, or explicitly adopt the active tab.", + description: RPC_TOOL_METADATA.browser_open.description, approval: "write", schema: (z) => z.union([ z.object({ @@ -158,7 +154,7 @@ const DEFINITIONS: ReadonlyArray<{ { name: "browser_snapshot", label: "Browser Snapshot", - description: "Read an accessibility snapshot, bounded text or HTML, or a screenshot from a task-owned tab.", + description: RPC_TOOL_METADATA.browser_snapshot.description, approval: "read", schema: (z) => z.union([ z.object({ @@ -190,7 +186,7 @@ const DEFINITIONS: ReadonlyArray<{ { name: "browser_act", label: "Browser Act", - description: "Run an ordered batch of typed actions against one task-owned tab and page revision.", + description: RPC_TOOL_METADATA.browser_act.description, approval: "write", schema: (z) => { const ref = z.string().min(1).max(256); @@ -231,7 +227,7 @@ const DEFINITIONS: ReadonlyArray<{ { name: "browser_wait", label: "Browser Wait", - description: "Wait for one schema-defined load, URL, text, selector, network-idle, or download condition.", + description: RPC_TOOL_METADATA.browser_wait.description, approval: "read", schema: (z) => z.object({ tab_id: z.number().int().min(0), @@ -245,14 +241,14 @@ const DEFINITIONS: ReadonlyArray<{ { name: "browser_tabs", label: "Browser Tabs", - description: "List only tabs owned by this task connection.", + description: RPC_TOOL_METADATA.browser_tabs.description, approval: "read", schema: (z) => z.object({}).strict(), }, { name: "browser_handoff", label: "Browser Handoff", - description: "Give the user control for credentials, MFA, CAPTCHA, or other human-only input.", + description: RPC_TOOL_METADATA.browser_handoff.description, approval: "write", schema: (z) => z.object({ tab_id: z.number().int().min(0), @@ -268,7 +264,7 @@ const DEFINITIONS: ReadonlyArray<{ { name: "browser_commit", label: "Browser Commit", - description: "Execute one previously staged consequential action after semantic review.", + description: RPC_TOOL_METADATA.browser_commit.description, approval: "write", schema: (z) => z.object({ staged_token: z.string().min(32).max(256) }).strict(), }, @@ -277,7 +273,7 @@ const DEFINITIONS: ReadonlyArray<{ const DEVELOPER = { name: "browser_developer" as const, label: "Browser Developer", - description: "Run an explicitly enabled developer-mode action outside the Standard tool surface.", + description: RPC_TOOL_METADATA.browser_developer.description, approval: "write" as const, schema: (z: ZodApi) => z.object({ action: z.string().min(1).max(128), diff --git a/packages/sdk-python/agenttab/__init__.py b/packages/sdk-python/agenttab/__init__.py index 3a40673..cbfec2c 100644 --- a/packages/sdk-python/agenttab/__init__.py +++ b/packages/sdk-python/agenttab/__init__.py @@ -1,5 +1,6 @@ from .client import ( AgentTabClient, + AgentTabCompatibilityError, AgentTabError, AgentTabTransportError, CLIENT_TO_HOST_MAX_BYTES, @@ -8,7 +9,9 @@ DEFAULT_REQUEST_TIMEOUT, HOST_TO_CLIENT_MAX_BYTES, LONG_OPERATION_TRANSPORT_GRACE, + RPC_FEATURES, RPC_PROTOCOL, + RPC_SUPPORTED_VERSIONS, RPC_VERSION, encode_frame, read_frame, @@ -19,6 +22,7 @@ __all__ = [ "AgentTabClient", + "AgentTabCompatibilityError", "AgentTabError", "AgentTabTransportError", "CLIENT_TO_HOST_MAX_BYTES", @@ -27,7 +31,9 @@ "DEFAULT_REQUEST_TIMEOUT", "HOST_TO_CLIENT_MAX_BYTES", "LONG_OPERATION_TRANSPORT_GRACE", + "RPC_FEATURES", "RPC_PROTOCOL", + "RPC_SUPPORTED_VERSIONS", "RPC_VERSION", "encode_frame", "read_frame", diff --git a/packages/sdk-python/agenttab/_generated_protocol.py b/packages/sdk-python/agenttab/_generated_protocol.py new file mode 100644 index 0000000..41358c4 --- /dev/null +++ b/packages/sdk-python/agenttab/_generated_protocol.py @@ -0,0 +1,40 @@ +# @generated by scripts/generate_protocol.py; do not edit. + +RPC_PROTOCOL = "agenttab.rpc" +RPC_VERSION = 1 +RPC_SUPPORTED_VERSIONS = ( + 1, +) +RPC_FEATURES = ( + "commit_staging_v1", + "idempotency_v1", + "operation_outcomes_v1", + "task_resume_v1", +) +RPC_METHODS = ( + "browser_open", + "browser_snapshot", + "browser_act", + "browser_wait", + "browser_tabs", + "browser_handoff", + "browser_commit", + "browser_developer", + "agenttab.status", +) +MUTATIONS = frozenset(( + "browser_open", + "browser_act", + "browser_handoff", + "browser_commit", + "browser_developer", +)) +OUTCOMES = ( + "completed", + "not_started", + "unknown", + "needs_user", + "commit_required", +) +CLIENT_TO_HOST_MAX_BYTES = 1048576 +HOST_TO_CLIENT_MAX_BYTES = 1048576 diff --git a/packages/sdk-python/agenttab/client.py b/packages/sdk-python/agenttab/client.py index cf7efc0..4848a91 100644 --- a/packages/sdk-python/agenttab/client.py +++ b/packages/sdk-python/agenttab/client.py @@ -13,26 +13,24 @@ import uuid from hashlib import sha256 from pathlib import Path -from typing import Any, BinaryIO, Literal, Mapping, Protocol +from typing import Any, BinaryIO, Callable, Literal, Mapping, Protocol + +from ._generated_protocol import ( + CLIENT_TO_HOST_MAX_BYTES, + HOST_TO_CLIENT_MAX_BYTES, + MUTATIONS, + RPC_FEATURES, + RPC_PROTOCOL, + RPC_SUPPORTED_VERSIONS, + RPC_VERSION, +) -RPC_PROTOCOL = "agenttab.rpc" -RPC_VERSION = 1 -CLIENT_TO_HOST_MAX_BYTES = 1024 * 1024 -HOST_TO_CLIENT_MAX_BYTES = 1024 * 1024 DEFAULT_REQUEST_TIMEOUT = 30.0 DEFAULT_BROWSER_WAIT_TIMEOUT = 30.0 DEFAULT_BROWSER_HANDOFF_TIMEOUT = 300.0 # Core reserves five seconds after a long operation's declared timeout. Keep a # second, bounded five-second margin for its response to cross the transports. LONG_OPERATION_TRANSPORT_GRACE = 10.0 -MUTATIONS = { - "browser_open", - "browser_act", - "browser_handoff", - "browser_commit", - "browser_developer", -} - JsonObject = dict[str, Any] @@ -70,6 +68,20 @@ def __init__(self, response: Mapping[str, Any]) -> None: self.details = details.get("details") +class AgentTabCompatibilityError(RuntimeError): + def __init__(self, response: Mapping[str, Any]) -> None: + super().__init__(str(response.get("message", "AgentTab protocol is incompatible"))) + self.requested_protocol = str(response.get("requested_protocol", "")) + self.requested_version = int(response.get("requested_version", 0)) + supported = response.get("supported_versions") + self.supported_versions = tuple(supported) if isinstance(supported, list) else () + self.recovery = str(response.get("recovery", "")) + + +class _RetryLegacyV1Handshake(RuntimeError): + pass + + TransportErrorCode = Literal[ "request_timeout", "connection_closed", @@ -277,6 +289,7 @@ def _read_exact( stream: BinaryIO | socket.socket, size: int, deadline: float | None = None, + on_data: Callable[[], None] | None = None, ) -> bytes: chunks: list[bytes] = [] remaining = size @@ -300,6 +313,8 @@ def _read_exact( chunk = stream.read(remaining) if not chunk: raise EOFError("AgentTab connection closed during a frame") + if on_data is not None: + on_data() chunks.append(chunk) remaining -= len(chunk) return b"".join(chunks) @@ -310,12 +325,15 @@ def read_frame( limit: int = HOST_TO_CLIENT_MAX_BYTES, *, timeout: float | None = None, + on_data: Callable[[], None] | None = None, ) -> JsonObject: deadline = None if timeout is None else time.monotonic() + timeout - declared = struct.unpack(" limit: raise ValueError(f"AgentTab frame declares {declared} bytes; limit is {limit}") - value = json.loads(_read_exact(stream, declared, deadline).decode("utf-8")) + value = json.loads( + _read_exact(stream, declared, deadline, on_data).decode("utf-8") + ) if not isinstance(value, dict): raise ValueError("AgentTab frame must contain a JSON object") return value @@ -611,14 +629,27 @@ def connect( address = endpoint or resolve_endpoint() negotiated: tuple[BinaryIO | socket.socket, JsonObject] | None = None attempted_capability: str | None = None + advertise_capabilities = True for capability in candidates: - stream, connection = cls._negotiate_connection( - address, - conversation_id=conversation_id, - resume_capability=capability, - connect_timeout=connect_timeout, - request_timeout=request_timeout, - ) + try: + stream, connection = cls._negotiate_connection( + address, + conversation_id=conversation_id, + resume_capability=capability, + connect_timeout=connect_timeout, + request_timeout=request_timeout, + advertise_capabilities=advertise_capabilities, + ) + except _RetryLegacyV1Handshake: + advertise_capabilities = False + stream, connection = cls._negotiate_connection( + address, + conversation_id=conversation_id, + resume_capability=capability, + connect_timeout=connect_timeout, + request_timeout=request_timeout, + advertise_capabilities=False, + ) if capability is not None and connection.get("resumed") is not True: stream.close() if capability_store is not None and capability != active_capability: @@ -675,6 +706,7 @@ def _negotiate_connection( resume_capability: str | None, connect_timeout: float, request_timeout: float, + advertise_capabilities: bool = True, ) -> tuple[BinaryIO | socket.socket, JsonObject]: if os.name == "nt": stream: BinaryIO | socket.socket = _open_windows_named_pipe(address) @@ -688,16 +720,27 @@ def _negotiate_connection( "version": RPC_VERSION, "kind": "connect", } + if advertise_capabilities: + request["supported_versions"] = list(RPC_SUPPORTED_VERSIONS) + request["supported_features"] = list(RPC_FEATURES) if conversation_id: request["conversation_id"] = conversation_id if resume_capability: request["resume_capability"] = resume_capability + request_started = False + received_response_data = False + + def mark_response_data() -> None: + nonlocal received_response_data + received_response_data = True + try: negotiation_deadline = ( time.monotonic() + connect_timeout if _uses_windows_named_pipe(stream) else None ) + request_started = True cls._write( stream, encode_frame(request), @@ -710,11 +753,45 @@ def _negotiate_connection( ) if remaining is not None and remaining <= 0: raise TimeoutError("AgentTab named-pipe connection negotiation timed out") - connection = read_frame(stream, timeout=remaining) + connection = read_frame( + stream, + timeout=remaining, + on_data=mark_response_data, + ) + if ( + connection.get("protocol") == RPC_PROTOCOL + and isinstance(connection.get("version"), int) + and not isinstance(connection.get("version"), bool) + and connection.get("version", 0) > 0 + and connection.get("kind") == "incompatible" + and isinstance(connection.get("requested_protocol"), str) + and isinstance(connection.get("requested_version"), int) + and not isinstance(connection.get("requested_version"), bool) + and isinstance(connection.get("supported_versions"), list) + and all( + isinstance(version, int) + and not isinstance(version, bool) + and version > 0 + for version in connection["supported_versions"] + ) + and isinstance(connection.get("message"), str) + and isinstance(connection.get("recovery"), str) + ): + raise AgentTabCompatibilityError(connection) if ( connection.get("protocol") != RPC_PROTOCOL or connection.get("version") != RPC_VERSION or connection.get("kind") != "connected" + or ( + connection.get("features") is not None + and ( + not isinstance(connection.get("features"), list) + or not all( + isinstance(feature, str) + for feature in connection["features"] + ) + ) + ) ): raise RuntimeError("AgentTab returned an invalid connection acknowledgement") if connection.get("resumed") and ( @@ -725,8 +802,18 @@ def _negotiate_connection( raise RuntimeError( "AgentTab returned an invalid resumed connection acknowledgement" ) - except Exception: + except Exception as error: stream.close() + if ( + advertise_capabilities + and request_started + and not received_response_data + and isinstance( + error, + (EOFError, BrokenPipeError, ConnectionResetError, ConnectionAbortedError), + ) + ): + raise _RetryLegacyV1Handshake() from error raise if isinstance(stream, socket.socket): stream.settimeout(request_timeout) diff --git a/packages/sdk-python/tests/test_client.py b/packages/sdk-python/tests/test_client.py index c25c870..c73d662 100644 --- a/packages/sdk-python/tests/test_client.py +++ b/packages/sdk-python/tests/test_client.py @@ -15,11 +15,13 @@ from agenttab import ( AgentTabClient, + AgentTabCompatibilityError, AgentTabError, AgentTabTransportError, DEFAULT_BROWSER_HANDOFF_TIMEOUT, DEFAULT_BROWSER_WAIT_TIMEOUT, LONG_OPERATION_TRANSPORT_GRACE, + RPC_FEATURES, encode_frame, read_frame, resolve_transport_timeout, @@ -175,6 +177,8 @@ def serve() -> None: try: hello = read_frame(connection) self.assertEqual(hello["kind"], "connect") + self.assertEqual(hello["supported_versions"], [1]) + self.assertEqual(hello["supported_features"], list(RPC_FEATURES)) connection.sendall(encode_frame({ "protocol": "agenttab.rpc", "version": 1, @@ -207,6 +211,130 @@ def serve() -> None: server.close() self.assertRegex(str(captured[0]["idempotency_key"]), r"-7[0-9a-f]{3}-") + def test_compatibility_error_preserves_recovery_details(self) -> None: + error = AgentTabCompatibilityError({ + "message": "AgentTab Core does not support protocol version 2", + "requested_protocol": "agenttab.rpc", + "requested_version": 2, + "supported_versions": [1], + "recovery": "Update AgentTab.", + }) + self.assertEqual(error.requested_protocol, "agenttab.rpc") + self.assertEqual(error.requested_version, 2) + self.assertEqual(error.supported_versions, (1,)) + self.assertEqual(error.recovery, "Update AgentTab.") + + def test_pre_ack_rejection_retries_with_exact_legacy_v1_connect(self) -> None: + with tempfile.TemporaryDirectory() as root: + endpoint = str(Path(root) / "agenttab.sock") + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(endpoint) + server.listen(2) + captured: list[dict[str, object]] = [] + + def serve() -> None: + for attempt in range(2): + connection, _ = server.accept() + try: + captured.append(read_frame(connection)) + if attempt == 0: + continue + connection.sendall(encode_frame({ + "protocol": "agenttab.rpc", + "version": 1, + "kind": "connected", + "connection_id": "018f22b2-4126-7c1a-8c31-3f45a783da42", + "resumed": False, + "state": "ready", + })) + finally: + connection.close() + + worker = threading.Thread(target=serve) + worker.start() + client = AgentTabClient.connect( + endpoint=endpoint, + conversation_id="conversation-for-both-attempts", + ) + client.close() + worker.join(timeout=2) + server.close() + + self.assertEqual(len(captured), 2) + self.assertEqual(captured[0]["supported_versions"], [1]) + self.assertEqual(captured[0]["supported_features"], list(RPC_FEATURES)) + self.assertEqual(captured[1], { + "protocol": "agenttab.rpc", + "version": 1, + "kind": "connect", + "conversation_id": "conversation-for-both-attempts", + }) + + def test_legacy_v1_connect_rejection_is_not_retried_again(self) -> None: + with tempfile.TemporaryDirectory() as root: + endpoint = str(Path(root) / "agenttab.sock") + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(endpoint) + server.listen(2) + captured: list[dict[str, object]] = [] + + def serve() -> None: + for _attempt in range(2): + connection, _ = server.accept() + try: + captured.append(read_frame(connection)) + finally: + connection.close() + + worker = threading.Thread(target=serve) + worker.start() + with self.assertRaises((EOFError, ConnectionResetError)): + AgentTabClient.connect(endpoint=endpoint) + worker.join(timeout=2) + server.close() + + self.assertEqual(len(captured), 2) + self.assertIn("supported_versions", captured[0]) + self.assertNotIn("supported_versions", captured[1]) + self.assertNotIn("supported_features", captured[1]) + + def test_explicit_incompatibility_never_falls_back_to_legacy_v1(self) -> None: + with tempfile.TemporaryDirectory() as root: + endpoint = str(Path(root) / "agenttab.sock") + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(endpoint) + server.listen(2) + captured: list[dict[str, object]] = [] + + def serve() -> None: + connection, _ = server.accept() + try: + captured.append(read_frame(connection)) + connection.sendall(encode_frame({ + "protocol": "agenttab.rpc", + "version": 1, + "kind": "incompatible", + "requested_protocol": "agenttab.rpc", + "requested_version": 2, + "supported_versions": [1], + "message": "AgentTab Core does not support protocol version 2", + "recovery": "Update AgentTab.", + })) + finally: + connection.close() + + worker = threading.Thread(target=serve) + worker.start() + with self.assertRaises(AgentTabCompatibilityError): + AgentTabClient.connect(endpoint=endpoint) + worker.join(timeout=2) + server.settimeout(0.05) + with self.assertRaises(socket.timeout): + server.accept() + server.close() + + self.assertEqual(len(captured), 1) + def test_file_transport_enforces_request_deadline(self) -> None: class BlockingStream(io.RawIOBase): def __init__(self) -> None: @@ -327,19 +455,51 @@ def close(self) -> None: with patch("agenttab.client.os.name", "nt"), patch( "agenttab.client._open_windows_named_pipe", return_value=stream, - ): + ) as open_pipe: with self.assertRaises(TimeoutError): - AgentTabClient._negotiate_connection( - r"\\.\pipe\agenttab-test", - conversation_id=None, - resume_capability=None, + AgentTabClient.connect( + endpoint=r"\\.\pipe\agenttab-test", connect_timeout=0.05, request_timeout=1.0, ) + self.assertEqual(open_pipe.call_count, 1) self.assertTrue(stream.closed) self.assertEqual(len(stream.timeouts), 1) self.assertLessEqual(stream.timeouts[0] or 0, 0.05) + def test_partial_connection_response_never_triggers_legacy_fallback(self) -> None: + class TruncatedWindowsPipe: + _agenttab_windows_pipe = True + + def __init__(self) -> None: + self.response = io.BytesIO(b"\x01") + self.closed = False + + def read(self, size: int, _timeout: float | None = None) -> bytes: + return self.response.read(size) + + def write(self, payload: bytes, _timeout: float | None = None) -> int: + return len(payload) + + def flush(self) -> None: + return None + + def close(self) -> None: + self.closed = True + + stream = TruncatedWindowsPipe() + with patch("agenttab.client.os.name", "nt"), patch( + "agenttab.client._open_windows_named_pipe", + return_value=stream, + ) as open_pipe: + with self.assertRaises(EOFError): + AgentTabClient.connect( + endpoint=r"\\.\pipe\agenttab-test", + connect_timeout=0.05, + ) + self.assertEqual(open_pipe.call_count, 1) + self.assertTrue(stream.closed) + def test_generated_mutation_key_survives_timeout_and_reuses(self) -> None: with tempfile.TemporaryDirectory() as root: endpoint = str(Path(root) / "agenttab.sock") diff --git a/packages/sdk-typescript/src/generated/protocol.ts b/packages/sdk-typescript/src/generated/protocol.ts new file mode 100644 index 0000000..37c7aa7 --- /dev/null +++ b/packages/sdk-typescript/src/generated/protocol.ts @@ -0,0 +1,101 @@ +// @generated by scripts/generate_protocol.py; do not edit. + +export const RPC_PROTOCOL = "agenttab.rpc" as const; +export const RPC_VERSION = 1 as const; +export const RPC_SUPPORTED_VERSIONS = [ + 1, +] as const; +export const RPC_FEATURES = [ + "commit_staging_v1", + "idempotency_v1", + "operation_outcomes_v1", + "task_resume_v1", +] as const; +export const RPC_METHODS = [ + "browser_open", + "browser_snapshot", + "browser_act", + "browser_wait", + "browser_tabs", + "browser_handoff", + "browser_commit", + "browser_developer", + "agenttab.status", +] as const; +export const MUTATING_RPC_METHODS = [ + "browser_open", + "browser_act", + "browser_handoff", + "browser_commit", + "browser_developer", +] as const; +export const OUTCOMES = [ + "completed", + "not_started", + "unknown", + "needs_user", + "commit_required", +] as const; +export const CLIENT_TO_HOST_MAX_BYTES = 1048576; +export const HOST_TO_CLIENT_MAX_BYTES = 1048576; + +export const RPC_TOOL_METADATA = { + "browser_open": { + "mutation": true, + "schema": "browser-open.schema.json", + "exposure": "standard", + "description": "Create a task tab, create an unfocused task-owned window, or explicitly adopt the active tab." + }, + "browser_snapshot": { + "mutation": false, + "schema": "browser-snapshot.schema.json", + "exposure": "standard", + "description": "Read an accessibility snapshot, bounded text or HTML, or a screenshot from a task-owned tab." + }, + "browser_act": { + "mutation": true, + "schema": "browser-act.schema.json", + "exposure": "standard", + "description": "Run an ordered batch of typed actions against one task-owned tab and page revision." + }, + "browser_wait": { + "mutation": false, + "schema": "browser-wait.schema.json", + "exposure": "standard", + "description": "Wait for one schema-defined load, URL, text, selector, network-idle, or download condition." + }, + "browser_tabs": { + "mutation": false, + "schema": "browser-tabs.schema.json", + "exposure": "standard", + "description": "List only tabs owned by this task connection." + }, + "browser_handoff": { + "mutation": true, + "schema": "browser-handoff.schema.json", + "exposure": "standard", + "description": "Pause agent actions and give the user control for credentials, MFA, CAPTCHA, or other human-only input." + }, + "browser_commit": { + "mutation": true, + "schema": "browser-commit.schema.json", + "exposure": "standard", + "description": "Execute one previously staged consequential action after semantic review." + }, + "browser_developer": { + "mutation": true, + "schema": "browser-developer.schema.json", + "exposure": "developer", + "description": "Run an explicitly enabled developer-mode action outside the Standard tool surface." + }, + "agenttab.status": { + "mutation": false, + "schema": "status.schema.json", + "exposure": "internal", + "description": "Read Core readiness without creating or resuming a browser task." + } +} as const; + +export type GeneratedRpcMethod = typeof RPC_METHODS[number]; +export type GeneratedMutationMethod = typeof MUTATING_RPC_METHODS[number]; +export type GeneratedOutcome = typeof OUTCOMES[number]; diff --git a/packages/sdk-typescript/src/index.ts b/packages/sdk-typescript/src/index.ts index b0cb289..a58839f 100644 --- a/packages/sdk-typescript/src/index.ts +++ b/packages/sdk-typescript/src/index.ts @@ -5,11 +5,31 @@ import { chmod, lstat, mkdir, open, readFile, rename, rm } from "node:fs/promise import { homedir, platform } from "node:os"; import { createConnection, type Socket } from "node:net"; import { join } from "node:path"; - -export const RPC_PROTOCOL = "agenttab.rpc" as const; -export const RPC_VERSION = 1 as const; -export const CLIENT_TO_HOST_MAX_BYTES = 1024 * 1024; -export const HOST_TO_CLIENT_MAX_BYTES = 1024 * 1024; +import { + CLIENT_TO_HOST_MAX_BYTES, + HOST_TO_CLIENT_MAX_BYTES, + MUTATING_RPC_METHODS, + RPC_FEATURES, + RPC_PROTOCOL, + RPC_SUPPORTED_VERSIONS, + RPC_VERSION, + type GeneratedMutationMethod, + type GeneratedOutcome, + type GeneratedRpcMethod, +} from "./generated/protocol"; + +export { + CLIENT_TO_HOST_MAX_BYTES, + HOST_TO_CLIENT_MAX_BYTES, + MUTATING_RPC_METHODS, + OUTCOMES, + RPC_FEATURES, + RPC_METHODS, + RPC_PROTOCOL, + RPC_SUPPORTED_VERSIONS, + RPC_TOOL_METADATA, + RPC_VERSION, +} from "./generated/protocol"; export const STANDARD_ACTION_VALUE_MAX_CHARS = 2048; export const DEFAULT_REQUEST_TIMEOUT_MS = 30_000; @@ -102,13 +122,8 @@ export interface MethodParams { "agenttab.status": Record; } -export type RpcMethod = keyof MethodParams; -export type MutationMethod = - | "browser_open" - | "browser_act" - | "browser_handoff" - | "browser_commit" - | "browser_developer"; +export type RpcMethod = GeneratedRpcMethod; +export type MutationMethod = GeneratedMutationMethod; export interface ConnectionAck { protocol: typeof RPC_PROTOCOL; @@ -119,9 +134,37 @@ export interface ConnectionAck { task_id?: string; resume_capability?: string; state?: "starting" | "reconciling" | "ready" | "paused" | "terminal"; + features?: string[]; +} + +export interface ProtocolIncompatible { + protocol: typeof RPC_PROTOCOL; + version: number; + kind: "incompatible"; + requested_protocol: string; + requested_version: number; + supported_versions: number[]; + message: string; + recovery: string; +} + +export class AgentTabCompatibilityError extends Error { + readonly requestedProtocol: string; + readonly requestedVersion: number; + readonly supportedVersions: readonly number[]; + readonly recovery: string; + + constructor(message: ProtocolIncompatible) { + super(message.message); + this.name = "AgentTabCompatibilityError"; + this.requestedProtocol = message.requested_protocol; + this.requestedVersion = message.requested_version; + this.supportedVersions = message.supported_versions; + this.recovery = message.recovery; + } } -export type Outcome = "completed" | "not_started" | "unknown" | "needs_user" | "commit_required"; +export type Outcome = GeneratedOutcome; export interface RpcResponse { protocol: typeof RPC_PROTOCOL; @@ -381,13 +424,7 @@ export interface ClientOptions { capabilityStore?: ResumeCapabilityStore; } -const MUTATIONS = new Set([ - "browser_open", - "browser_act", - "browser_handoff", - "browser_commit", - "browser_developer", -]); +const MUTATIONS = new Set(MUTATING_RPC_METHODS); function longOperationTimeoutMs( method: RpcMethod, @@ -517,38 +554,64 @@ interface NegotiatedConnection { connected: ConnectionAck; } +class RetryLegacyV1Handshake extends Error { + constructor() { + super("AgentTab rejected capability negotiation before acknowledging the connection"); + this.name = "RetryLegacyV1Handshake"; + } +} + async function negotiateConnection( endpoint: string, timeoutMs: number, conversationId: string | undefined, resumeCapability: string | undefined, + advertiseCapabilities: boolean, ): Promise { const socket = createConnection(endpoint); const decoder = new FrameDecoder(); const connected = await new Promise((resolve, reject) => { - const timer = setTimeout(() => { - socket.destroy(); - reject(new Error(`Timed out after ${timeoutMs} ms connecting to AgentTab at ${endpoint}`)); - }, timeoutMs); - const closed = () => fail(new Error("AgentTab closed during connection negotiation")); - const fail = (error: Error) => { + let handshakeSent = false; + let receivedData = false; + let settled = false; + const cleanup = () => { clearTimeout(timer); - socket.off("error", fail); - socket.off("close", closed); + socket.off("error", rejectedBeforeAck); + socket.off("close", closedBeforeAck); + socket.removeAllListeners("data"); + }; + const fail = (error: Error) => { + if (settled) return; + settled = true; + cleanup(); + socket.destroy(); reject(error); }; + const retryOrFail = (error: Error) => { + fail( + advertiseCapabilities && handshakeSent && !receivedData + ? new RetryLegacyV1Handshake() + : error, + ); + }; + const rejectedBeforeAck = (error: Error) => retryOrFail(error); + const closedBeforeAck = () => retryOrFail(new Error("AgentTab closed during connection negotiation")); + const timer = setTimeout(() => { + fail(new Error(`Timed out after ${timeoutMs} ms connecting to AgentTab at ${endpoint}`)); + }, timeoutMs); const accept = (value: ConnectionAck) => { - clearTimeout(timer); - socket.off("error", fail); - socket.off("close", closed); - socket.removeAllListeners("data"); + if (settled) return; + settled = true; + cleanup(); resolve(value); }; - socket.once("error", fail); - socket.once("close", closed); + socket.once("error", rejectedBeforeAck); + socket.once("close", closedBeforeAck); socket.on("data", (chunk) => { + receivedData = true; try { for (const value of decoder.push(chunk)) { + if (isProtocolIncompatible(value)) throw new AgentTabCompatibilityError(value); if (!isConnectionAck(value)) throw new Error("AgentTab sent a response before connection negotiation completed"); accept(value); return; @@ -559,13 +622,25 @@ async function negotiateConnection( } }); socket.once("connect", () => { - socket.write(encodeFrame({ - protocol: RPC_PROTOCOL, - version: RPC_VERSION, - kind: "connect", - ...(conversationId ? { conversation_id: conversationId } : {}), - ...(resumeCapability ? { resume_capability: resumeCapability } : {}), - })); + try { + const frame = encodeFrame({ + protocol: RPC_PROTOCOL, + version: RPC_VERSION, + kind: "connect", + ...(advertiseCapabilities + ? { + supported_versions: RPC_SUPPORTED_VERSIONS, + supported_features: RPC_FEATURES, + } + : {}), + ...(conversationId ? { conversation_id: conversationId } : {}), + ...(resumeCapability ? { resume_capability: resumeCapability } : {}), + }); + handshakeSent = true; + socket.write(frame); + } catch (error) { + retryOrFail(error instanceof Error ? error : new Error(String(error))); + } }); }); return { socket, decoder, connected }; @@ -658,8 +733,28 @@ export class AgentTabClient { .filter((value, index, values) => values.indexOf(value) === index); let attemptedCapability: string | undefined; let negotiated: NegotiatedConnection | undefined; + let advertiseCapabilities = true; for (const capability of candidates.length === 0 ? [undefined] : candidates) { - const attempt = await negotiateConnection(endpoint, timeoutMs, options.conversationId, capability); + let attempt: NegotiatedConnection; + try { + attempt = await negotiateConnection( + endpoint, + timeoutMs, + options.conversationId, + capability, + advertiseCapabilities, + ); + } catch (error) { + if (!(error instanceof RetryLegacyV1Handshake)) throw error; + advertiseCapabilities = false; + attempt = await negotiateConnection( + endpoint, + timeoutMs, + options.conversationId, + capability, + false, + ); + } if (capability && !attempt.connected.resumed) { attempt.socket.destroy(); if (capability !== activeCapability) continue; @@ -940,15 +1035,34 @@ function isConnectionAck(value: unknown): value is ConnectionAck { return ( isRecord(value) && value.protocol === RPC_PROTOCOL && - value.version === RPC_VERSION && + typeof value.version === "number" && + Number.isInteger(value.version) && + value.version > 0 && value.kind === "connected" && typeof value.connection_id === "string" && typeof value.resumed === "boolean" && + (value.features === undefined || + (Array.isArray(value.features) && value.features.every((feature) => typeof feature === "string"))) && (!value.resumed || (typeof value.task_id === "string" && typeof value.resume_capability === "string")) ); } +function isProtocolIncompatible(value: unknown): value is ProtocolIncompatible { + return ( + isRecord(value) && + value.protocol === RPC_PROTOCOL && + value.version === RPC_VERSION && + value.kind === "incompatible" && + typeof value.requested_protocol === "string" && + typeof value.requested_version === "number" && + Array.isArray(value.supported_versions) && + value.supported_versions.every((version) => Number.isInteger(version) && version > 0) && + typeof value.message === "string" && + typeof value.recovery === "string" + ); +} + function isResumeCapabilityConfirmed(value: unknown, connectionId: string): boolean { return ( isRecord(value) && diff --git a/packages/sdk-typescript/test/client.test.ts b/packages/sdk-typescript/test/client.test.ts index e8eae49..d824513 100644 --- a/packages/sdk-typescript/test/client.test.ts +++ b/packages/sdk-typescript/test/client.test.ts @@ -5,12 +5,14 @@ import { join } from "node:path"; import { createServer, type Server, type Socket } from "node:net"; import { AgentTabClient, + AgentTabCompatibilityError, AgentTabError, AgentTabTransportError, DEFAULT_BROWSER_HANDOFF_TIMEOUT_MS, DEFAULT_BROWSER_WAIT_TIMEOUT_MS, FrameDecoder, LONG_OPERATION_TRANSPORT_GRACE_MS, + RPC_FEATURES, createUuidV7, createResumeCapabilityStore, encodeFrame, @@ -157,6 +159,130 @@ describe("Core RPC transport deadlines", () => { }); }); +test("advertises generated capabilities and surfaces incompatible majors", async () => { + let connect: Record | undefined; + let connectionCount = 0; + const endpoint = await listen((socket) => { + connectionCount += 1; + const decoder = new FrameDecoder(); + socket.on("data", (chunk) => { + for (const value of decoder.push(chunk) as Array>) { + connect = value; + socket.write(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + kind: "incompatible", + requested_protocol: "agenttab.rpc", + requested_version: 2, + supported_versions: [1], + message: "AgentTab Core does not support protocol version 2", + recovery: "Update the AgentTab client and host.", + })); + } + }); + }); + + let error: unknown; + try { + await AgentTabClient.connect({ endpoint }); + } catch (cause) { + error = cause; + } + expect(error).toBeInstanceOf(AgentTabCompatibilityError); + expect(error).toMatchObject({ + requestedProtocol: "agenttab.rpc", + requestedVersion: 2, + supportedVersions: [1], + }); + expect(connect).toMatchObject({ + kind: "connect", + supported_versions: [1], + supported_features: [...RPC_FEATURES], + }); + expect(connectionCount).toBe(1); +}); + +test("retries once with the exact legacy-v1 handshake after a pre-ack rejection", async () => { + const connects: Array> = []; + let connectionCount = 0; + const endpoint = await listen((socket) => { + connectionCount += 1; + const attempt = connectionCount; + const decoder = new FrameDecoder(); + socket.on("data", (chunk) => { + for (const value of decoder.push(chunk) as Array>) { + connects.push(value); + if (attempt === 1) { + socket.end(); + return; + } + socket.write(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + kind: "connected", + connection_id: "018f22b2-4126-7c1a-8c31-3f45a783da42", + resumed: false, + state: "ready", + })); + } + }); + }); + + const client = await AgentTabClient.connect({ + endpoint, + conversationId: "conversation-for-both-attempts", + }); + client.close(); + + expect(connectionCount).toBe(2); + expect(connects[0]).toMatchObject({ + kind: "connect", + conversation_id: "conversation-for-both-attempts", + supported_versions: [1], + supported_features: [...RPC_FEATURES], + }); + expect(connects[1]).toEqual({ + protocol: "agenttab.rpc", + version: 1, + kind: "connect", + conversation_id: "conversation-for-both-attempts", + }); +}); + +test("does not retry again when the exact legacy-v1 handshake is also rejected", async () => { + const connects: Array> = []; + const endpoint = await listen((socket) => { + const decoder = new FrameDecoder(); + socket.on("data", (chunk) => { + for (const value of decoder.push(chunk) as Array>) { + connects.push(value); + socket.end(); + } + }); + }); + + await expect(AgentTabClient.connect({ endpoint })).rejects.toThrow( + "AgentTab closed during connection negotiation", + ); + expect(connects).toHaveLength(2); + expect(connects[0]).toHaveProperty("supported_versions", [1]); + expect(connects[1]).not.toHaveProperty("supported_versions"); + expect(connects[1]).not.toHaveProperty("supported_features"); +}); + +test("does not reinterpret a negotiation timeout as a legacy-v1 rejection", async () => { + let connectionCount = 0; + const endpoint = await listen((socket) => { + connectionCount += 1; + socket.on("data", () => undefined); + }); + + await expect(AgentTabClient.connect({ endpoint, connectTimeoutMs: 100 })).rejects.toThrow( + "Timed out after 100 ms", + ); + expect(connectionCount).toBe(1); +}); + if (false) { // @ts-expect-error Standard browser actions never expose an agent-controlled focus transition. const forbiddenFocusAction: BrowserAction = { kind: "focus" }; diff --git a/protocol/agenttab-v1.json b/protocol/agenttab-v1.json new file mode 100644 index 0000000..3dc8ed2 --- /dev/null +++ b/protocol/agenttab-v1.json @@ -0,0 +1,136 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "schema_version": 1, + "rpc": { + "name": "agenttab.rpc", + "version": 1, + "supported_versions": [1], + "features": [ + "commit_staging_v1", + "idempotency_v1", + "operation_outcomes_v1", + "task_resume_v1" + ], + "frame_limits": { + "client_to_host": 1048576, + "host_to_client": 1048576 + } + }, + "native": { + "name": "agenttab.native", + "version": 1, + "supported_versions": [1], + "features": [ + "commit_staging_v1", + "disconnect_recovery_v1", + "event_ack_v1", + "inventory_reconciliation_v1" + ], + "frame_limits": { + "host_to_extension": 1048576, + "extension_to_host": 67108864 + }, + "methods": [ + "browser_open", + "browser_snapshot", + "browser_act", + "browser_wait", + "browser_tabs", + "browser_handoff", + "browser_commit", + "browser_developer", + "commit_review_bind", + "commit_review_abandon" + ], + "events": [ + "inventory", + "ownership_revoked", + "tab_removed", + "group_membership_changed", + "pause_changed", + "handoff_changed", + "commit_expired", + "commit_abandoned", + "popup_commit_approved", + "popup_commit_abandoned", + "extension_disconnected" + ] + }, + "outcomes": [ + "completed", + "not_started", + "unknown", + "needs_user", + "commit_required" + ], + "rpc_methods": [ + { + "name": "browser_open", + "mutation": true, + "schema": "browser-open.schema.json", + "exposure": "standard", + "description": "Create a task tab, create an unfocused task-owned window, or explicitly adopt the active tab." + }, + { + "name": "browser_snapshot", + "mutation": false, + "schema": "browser-snapshot.schema.json", + "exposure": "standard", + "description": "Read an accessibility snapshot, bounded text or HTML, or a screenshot from a task-owned tab." + }, + { + "name": "browser_act", + "mutation": true, + "schema": "browser-act.schema.json", + "exposure": "standard", + "description": "Run an ordered batch of typed actions against one task-owned tab and page revision." + }, + { + "name": "browser_wait", + "mutation": false, + "schema": "browser-wait.schema.json", + "exposure": "standard", + "description": "Wait for one schema-defined load, URL, text, selector, network-idle, or download condition." + }, + { + "name": "browser_tabs", + "mutation": false, + "schema": "browser-tabs.schema.json", + "exposure": "standard", + "description": "List only tabs owned by this task connection." + }, + { + "name": "browser_handoff", + "mutation": true, + "schema": "browser-handoff.schema.json", + "exposure": "standard", + "description": "Pause agent actions and give the user control for credentials, MFA, CAPTCHA, or other human-only input." + }, + { + "name": "browser_commit", + "mutation": true, + "schema": "browser-commit.schema.json", + "exposure": "standard", + "description": "Execute one previously staged consequential action after semantic review." + }, + { + "name": "browser_developer", + "mutation": true, + "schema": "browser-developer.schema.json", + "exposure": "developer", + "description": "Run an explicitly enabled developer-mode action outside the Standard tool surface." + }, + { + "name": "agenttab.status", + "mutation": false, + "schema": "status.schema.json", + "exposure": "internal", + "description": "Read Core readiness without creating or resuming a browser task." + } + ], + "schema_assets": [ + "request.schema.json", + "response.schema.json", + "connection.schema.json" + ] +} diff --git a/schemas/native/v1/message.schema.json b/schemas/native/v1/message.schema.json index 59b12a6..e37b425 100644 --- a/schemas/native/v1/message.schema.json +++ b/schemas/native/v1/message.schema.json @@ -8,7 +8,7 @@ "required": ["protocol", "version", "kind"], "properties": { "protocol": { "const": "agenttab.native" }, - "version": { "const": 1 }, + "version": { "type": "integer", "minimum": 1, "maximum": 65535 }, "kind": { "type": "string" } } }, @@ -136,7 +136,20 @@ "inventory": { "type": "array", "items": { "$ref": "#/$defs/tab" } }, "paused": { "type": "boolean" }, "handoff": { "$ref": "#/$defs/handoff" }, - "staged_commits": { "type": "array", "items": { "$ref": "#/$defs/staged" } } + "staged_commits": { "type": "array", "items": { "$ref": "#/$defs/staged" } }, + "supported_versions": { + "type": "array", + "items": { "type": "integer", "minimum": 1, "maximum": 65535 }, + "minItems": 1, + "maxItems": 16, + "uniqueItems": true + }, + "supported_features": { + "type": "array", + "items": { "type": "string", "minLength": 1, "maxLength": 128 }, + "maxItems": 64, + "uniqueItems": true + } }, "additionalProperties": false } @@ -149,9 +162,15 @@ "type": "object", "required": ["kind", "host_version", "state"], "properties": { - "protocol": {}, "version": {}, "kind": { "const": "ready" }, + "protocol": {}, "version": { "const": 1 }, "kind": { "const": "ready" }, "host_version": { "type": "string", "minLength": 1 }, - "state": { "enum": ["ready", "paused"] } + "state": { "enum": ["ready", "paused"] }, + "features": { + "type": "array", + "items": { "type": "string", "minLength": 1, "maxLength": 128 }, + "maxItems": 64, + "uniqueItems": true + } }, "additionalProperties": false } @@ -164,7 +183,7 @@ "type": "object", "required": ["kind", "request_id", "connection_id", "task_id", "method", "params"], "properties": { - "protocol": {}, "version": {}, "kind": { "const": "command" }, + "protocol": {}, "version": { "const": 1 }, "kind": { "const": "command" }, "request_id": { "type": "string", "format": "uuid" }, "connection_id": { "type": "string", "format": "uuid" }, "task_id": { "type": "string", "format": "uuid" }, @@ -183,7 +202,7 @@ "type": "object", "required": ["kind", "request_id", "task_id"], "properties": { - "protocol": {}, "version": {}, "kind": { "const": "close_task" }, + "protocol": {}, "version": { "const": 1 }, "kind": { "const": "close_task" }, "request_id": { "type": "string", "format": "uuid" }, "task_id": { "type": "string", "format": "uuid" } }, @@ -198,7 +217,7 @@ "type": "object", "required": ["kind", "request_id", "outcome"], "properties": { - "protocol": {}, "version": {}, "kind": { "const": "response" }, + "protocol": {}, "version": { "const": 1 }, "kind": { "const": "response" }, "request_id": { "type": "string", "format": "uuid" }, "outcome": { "enum": ["completed", "not_started", "unknown", "needs_user", "commit_required"] }, "result": {}, @@ -240,7 +259,7 @@ { "required": ["kind", "event", "event_id"], "properties": { - "protocol": {}, "version": {}, "kind": { "const": "event_ack" }, + "protocol": {}, "version": { "const": 1 }, "kind": { "const": "event_ack" }, "event": { "enum": ["handoff_changed", "popup_commit_approved", "popup_commit_abandoned"] }, "event_id": { "type": "string", "minLength": 1, "maxLength": 256 }, "outcome": { "enum": ["completed", "not_started", "unknown"] }, @@ -286,7 +305,7 @@ "type": "object", "required": ["kind", "event", "payload"], "properties": { - "protocol": {}, "version": {}, "kind": { "const": "event" }, + "protocol": {}, "version": { "const": 1 }, "kind": { "const": "event" }, "event": { "enum": [ "inventory", @@ -376,13 +395,43 @@ "type": "object", "required": ["kind", "reason", "pending_outcome"], "properties": { - "protocol": {}, "version": {}, "kind": { "const": "disconnect_recovery" }, + "protocol": {}, "version": { "const": 1 }, "kind": { "const": "disconnect_recovery" }, "reason": { "type": "string", "minLength": 1 }, "pending_outcome": { "const": "unknown" } }, "additionalProperties": false } ] + }, + { + "allOf": [ + { "$ref": "#/$defs/base" }, + { + "type": "object", + "required": [ + "kind", + "requested_protocol", + "requested_version", + "supported_versions", + "message", + "recovery" + ], + "properties": { + "protocol": {}, "version": { "const": 1 }, "kind": { "const": "incompatible" }, + "requested_protocol": { "type": "string", "maxLength": 128 }, + "requested_version": { "type": "integer", "minimum": 0 }, + "supported_versions": { + "type": "array", + "items": { "type": "integer", "minimum": 1, "maximum": 65535 }, + "minItems": 1, + "uniqueItems": true + }, + "message": { "type": "string", "minLength": 1 }, + "recovery": { "type": "string", "minLength": 1 } + }, + "additionalProperties": false + } + ] } ] } diff --git a/schemas/rpc/v1/connection.schema.json b/schemas/rpc/v1/connection.schema.json index 553ac5d..ed307e1 100644 --- a/schemas/rpc/v1/connection.schema.json +++ b/schemas/rpc/v1/connection.schema.json @@ -8,10 +8,23 @@ "required": ["protocol", "version", "kind"], "properties": { "protocol": { "const": "agenttab.rpc" }, - "version": { "const": 1 }, + "version": { "type": "integer", "minimum": 1, "maximum": 65535 }, "kind": { "const": "connect" }, "conversation_id": { "type": "string", "minLength": 1, "maxLength": 256 }, - "resume_capability": { "type": "string", "minLength": 32, "maxLength": 64 } + "resume_capability": { "type": "string", "minLength": 32, "maxLength": 64 }, + "supported_versions": { + "type": "array", + "items": { "type": "integer", "minimum": 1, "maximum": 65535 }, + "minItems": 1, + "maxItems": 16, + "uniqueItems": true + }, + "supported_features": { + "type": "array", + "items": { "type": "string", "minLength": 1, "maxLength": 128 }, + "maxItems": 64, + "uniqueItems": true + } }, "additionalProperties": false }, @@ -26,7 +39,13 @@ "resumed": { "type": "boolean" }, "task_id": { "type": "string", "format": "uuid" }, "resume_capability": { "type": "string", "minLength": 32, "maxLength": 64 }, - "state": { "enum": ["starting", "reconciling", "ready", "paused", "terminal"] } + "state": { "enum": ["starting", "reconciling", "ready", "paused", "terminal"] }, + "features": { + "type": "array", + "items": { "type": "string", "minLength": 1, "maxLength": 128 }, + "maxItems": 64, + "uniqueItems": true + } }, "allOf": [ { @@ -63,6 +82,35 @@ "connection_id": { "type": "string", "format": "uuid" } }, "additionalProperties": false + }, + { + "type": "object", + "required": [ + "protocol", + "version", + "kind", + "requested_protocol", + "requested_version", + "supported_versions", + "message", + "recovery" + ], + "properties": { + "protocol": { "const": "agenttab.rpc" }, + "version": { "const": 1 }, + "kind": { "const": "incompatible" }, + "requested_protocol": { "type": "string", "maxLength": 128 }, + "requested_version": { "type": "integer", "minimum": 0 }, + "supported_versions": { + "type": "array", + "items": { "type": "integer", "minimum": 1, "maximum": 65535 }, + "minItems": 1, + "uniqueItems": true + }, + "message": { "type": "string", "minLength": 1 }, + "recovery": { "type": "string", "minLength": 1 } + }, + "additionalProperties": false } ] } diff --git a/scripts/generate_protocol.py b/scripts/generate_protocol.py new file mode 100644 index 0000000..9ca0b5e --- /dev/null +++ b/scripts/generate_protocol.py @@ -0,0 +1,332 @@ +#!/usr/bin/env python3 +"""Generate small cross-language protocol catalogs from protocol/agenttab-v1.json.""" + +from __future__ import annotations + +import argparse +import difflib +import json +import sys +from pathlib import Path +from typing import Any + + +ROOT = Path(__file__).resolve().parents[1] +MANIFEST_PATH = ROOT / "protocol" / "agenttab-v1.json" +RPC_SCHEMA_ROOT = ROOT / "schemas" / "rpc" / "v1" + + +def load_json(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise ValueError(f"{path.relative_to(ROOT)} must contain a JSON object") + return value + + +def quoted(value: str) -> str: + return json.dumps(value, ensure_ascii=False) + + +def rust_slice(values: list[str]) -> str: + return "&[\n" + "".join(f" {quoted(value)},\n" for value in values) + "]" + + +def ts_array(values: list[str]) -> str: + return "[\n" + "".join(f" {quoted(value)},\n" for value in values) + "] as const" + + +def py_tuple(values: list[str]) -> str: + if not values: + return "()" + return "(\n" + "".join(f" {quoted(value)},\n" for value in values) + ")" + + +def validate_manifest(manifest: dict[str, Any]) -> None: + expected_top = { + "$schema", + "schema_version", + "rpc", + "native", + "outcomes", + "rpc_methods", + "schema_assets", + } + if set(manifest) != expected_top: + raise ValueError("protocol manifest has missing or unknown top-level fields") + if manifest["schema_version"] != 1: + raise ValueError("unsupported protocol manifest schema_version") + if manifest["outcomes"] != list(dict.fromkeys(manifest["outcomes"])): + raise ValueError("outcomes must be unique") + + methods = manifest["rpc_methods"] + if not isinstance(methods, list) or not methods: + raise ValueError("rpc_methods must be a non-empty array") + names: list[str] = [] + schemas: list[str] = [] + for method in methods: + if not isinstance(method, dict) or set(method) != { + "name", + "mutation", + "schema", + "exposure", + "description", + }: + raise ValueError("each rpc method must contain the canonical five fields") + if method["exposure"] not in {"standard", "developer", "internal"}: + raise ValueError(f"invalid exposure for {method['name']}") + names.append(method["name"]) + schemas.append(method["schema"]) + if len(names) != len(set(names)): + raise ValueError("rpc method names must be unique") + if len(schemas) != len(set(schemas)): + raise ValueError("rpc method schemas must be unique") + + for section_name in ("rpc", "native"): + section = manifest[section_name] + versions = section["supported_versions"] + if ( + not isinstance(versions, list) + or not versions + or versions != sorted(set(versions)) + or section["version"] not in versions + ): + raise ValueError(f"{section_name}.supported_versions must be sorted, unique, and include version") + features = section["features"] + if features != sorted(set(features)): + raise ValueError(f"{section_name}.features must be sorted and unique") + if not section["name"] or not isinstance(section["name"], str): + raise ValueError(f"{section_name}.name must be a non-empty string") + if any( + not isinstance(limit, int) or limit < 1 + for limit in section["frame_limits"].values() + ): + raise ValueError(f"{section_name}.frame_limits must be positive integers") + native = manifest["native"] + for field in ("methods", "events"): + values = native[field] + if not values or values != list(dict.fromkeys(values)): + raise ValueError(f"native.{field} must be non-empty and unique") + + request = load_json(RPC_SCHEMA_ROOT / "request.schema.json") + rpc = manifest["rpc"] + for envelope_name in ("request", "response"): + envelope = load_json(RPC_SCHEMA_ROOT / f"{envelope_name}.schema.json") + if ( + envelope["properties"]["protocol"].get("const") != rpc["name"] + or envelope["properties"]["version"].get("const") != rpc["version"] + ): + raise ValueError(f"{envelope_name}.schema.json protocol/version must match the manifest") + request_methods = request["properties"]["method"]["enum"] + if request_methods != names: + raise ValueError("request.schema.json method order must match protocol manifest") + mutation_branch = request["allOf"][0]["if"]["properties"]["method"]["enum"] + mutations = [method["name"] for method in methods if method["mutation"]] + if mutation_branch != mutations: + raise ValueError("request.schema.json mutation methods must match protocol manifest") + branch_refs = { + branch["if"]["properties"]["method"]["const"]: branch["then"]["properties"]["params"]["$ref"] + for branch in request["allOf"][1:] + } + if branch_refs != {method["name"]: method["schema"] for method in methods}: + raise ValueError("request.schema.json parameter refs must match protocol manifest") + + response = load_json(RPC_SCHEMA_ROOT / "response.schema.json") + if response["properties"]["outcome"]["enum"] != manifest["outcomes"]: + raise ValueError("response.schema.json outcomes must match protocol manifest") + for filename in [*manifest["schema_assets"], *schemas]: + path = RPC_SCHEMA_ROOT / filename + if not path.is_file(): + raise ValueError(f"missing registered RPC schema: {filename}") + schema = load_json(path) + expected_id = f"https://agenttab.dev/schemas/rpc/v1/{filename}" + if schema.get("$id") != expected_id: + raise ValueError(f"{filename} has a non-canonical $id") + native_schema = load_json(ROOT / "schemas" / "native" / "v1" / "message.schema.json") + native_base = native_schema["$defs"]["base"]["properties"] + if native_base["protocol"].get("const") != native["name"]: + raise ValueError("native message schema protocol must match the manifest") + for branch in native_schema["oneOf"]: + properties = branch["allOf"][-1]["properties"] + kind = properties["kind"]["const"] + if kind != "hello" and properties["version"].get("const") != native["version"]: + raise ValueError(f"native {kind} schema version must match the manifest") + + +def render_rust(manifest: dict[str, Any]) -> str: + rpc = manifest["rpc"] + native = manifest["native"] + methods = manifest["rpc_methods"] + mutations = [method["name"] for method in methods if method["mutation"]] + schema_names = [*manifest["schema_assets"], *[method["schema"] for method in methods]] + assets = [] + for filename in schema_names: + logical_name = filename.removesuffix(".schema.json").replace("-", "_") + assets.append( + " (\n" + f" {quoted(logical_name)},\n" + f' include_str!("../../../../schemas/rpc/v1/{filename}"),\n' + " ),\n" + ) + return "".join( + [ + "// @generated by scripts/generate_protocol.py; do not edit.\n\n", + f"pub const RPC_PROTOCOL: &str = {quoted(rpc['name'])};\n", + f"pub const NATIVE_PROTOCOL: &str = {quoted(native['name'])};\n", + f"pub const RPC_VERSION: u16 = {rpc['version']};\n", + f"pub const NATIVE_VERSION: u16 = {native['version']};\n", + "pub const PROTOCOL_VERSION: u16 = RPC_VERSION;\n", + f"pub const RPC_SUPPORTED_VERSIONS: &[u16] = &{rpc['supported_versions']};\n", + f"pub const NATIVE_SUPPORTED_VERSIONS: &[u16] = &{native['supported_versions']};\n", + f"pub const RPC_FEATURES: &[&str] = {rust_slice(rpc['features'])};\n", + f"pub const NATIVE_FEATURES: &[&str] = {rust_slice(native['features'])};\n", + f"pub const RPC_METHOD_NAMES: &[&str] = {rust_slice([method['name'] for method in methods])};\n", + f"pub const MUTATING_RPC_METHOD_NAMES: &[&str] = {rust_slice(mutations)};\n", + f"pub const NATIVE_METHOD_NAMES: &[&str] = {rust_slice(native['methods'])};\n", + f"pub const NATIVE_EVENT_NAMES: &[&str] = {rust_slice(native['events'])};\n", + f"pub const OUTCOME_NAMES: &[&str] = {rust_slice(manifest['outcomes'])};\n", + f"pub const CLIENT_TO_HOST_MAX_BYTES: usize = {rpc['frame_limits']['client_to_host']};\n", + f"pub const HOST_TO_CLIENT_MAX_BYTES: usize = {rpc['frame_limits']['host_to_client']};\n", + f"pub const HOST_TO_EXTENSION_MAX_BYTES: usize = {native['frame_limits']['host_to_extension']};\n", + f"pub const EXTENSION_TO_HOST_MAX_BYTES: usize = {native['frame_limits']['extension_to_host']};\n\n", + "pub const RPC_SCHEMA_ASSETS: &[(&str, &str)] = &[\n", + *assets, + "];\n\n", + 'pub const NATIVE_SCHEMA: &str = include_str!("../../../../schemas/native/v1/message.schema.json");\n', + ] + ) + + +def render_sdk_ts(manifest: dict[str, Any]) -> str: + rpc = manifest["rpc"] + methods = manifest["rpc_methods"] + mutations = [method["name"] for method in methods if method["mutation"]] + metadata = { + method["name"]: { + "mutation": method["mutation"], + "schema": method["schema"], + "exposure": method["exposure"], + "description": method["description"], + } + for method in methods + } + return "".join( + [ + "// @generated by scripts/generate_protocol.py; do not edit.\n\n", + f"export const RPC_PROTOCOL = {quoted(rpc['name'])} as const;\n", + f"export const RPC_VERSION = {rpc['version']} as const;\n", + f"export const RPC_SUPPORTED_VERSIONS = {ts_array(rpc['supported_versions'])};\n", + f"export const RPC_FEATURES = {ts_array(rpc['features'])};\n", + f"export const RPC_METHODS = {ts_array([method['name'] for method in methods])};\n", + f"export const MUTATING_RPC_METHODS = {ts_array(mutations)};\n", + f"export const OUTCOMES = {ts_array(manifest['outcomes'])};\n", + f"export const CLIENT_TO_HOST_MAX_BYTES = {rpc['frame_limits']['client_to_host']};\n", + f"export const HOST_TO_CLIENT_MAX_BYTES = {rpc['frame_limits']['host_to_client']};\n\n", + "export const RPC_TOOL_METADATA = ", + json.dumps(metadata, indent=2, ensure_ascii=False), + " as const;\n\n", + "export type GeneratedRpcMethod = typeof RPC_METHODS[number];\n", + "export type GeneratedMutationMethod = typeof MUTATING_RPC_METHODS[number];\n", + "export type GeneratedOutcome = typeof OUTCOMES[number];\n", + ] + ) + + +def render_extension_ts(manifest: dict[str, Any]) -> str: + native = manifest["native"] + method_record = {value: True for value in native["methods"]} + event_record = {value: True for value in native["events"]} + return "".join( + [ + "// @generated by scripts/generate_protocol.py; do not edit.\n\n", + f"export const NATIVE_PROTOCOL = {quoted(native['name'])} as const;\n", + f"export const PROTOCOL_VERSION = {native['version']} as const;\n", + f"export const NATIVE_SUPPORTED_VERSIONS = {ts_array(native['supported_versions'])};\n", + f"export const NATIVE_FEATURES = {ts_array(native['features'])};\n", + f"export const OUTCOMES = {ts_array(manifest['outcomes'])};\n", + "export const NATIVE_METHODS = ", + json.dumps(method_record, indent=2), + " as const;\n", + "export const NATIVE_EVENTS = ", + json.dumps(event_record, indent=2), + " as const;\n\n", + "export type NativeMethod = keyof typeof NATIVE_METHODS;\n", + "export type NativeEventName = keyof typeof NATIVE_EVENTS;\n", + "export type GeneratedOutcome = typeof OUTCOMES[number];\n", + ] + ) + + +def render_python(manifest: dict[str, Any]) -> str: + rpc = manifest["rpc"] + methods = manifest["rpc_methods"] + mutations = [method["name"] for method in methods if method["mutation"]] + return "".join( + [ + "# @generated by scripts/generate_protocol.py; do not edit.\n\n", + f"RPC_PROTOCOL = {quoted(rpc['name'])}\n", + f"RPC_VERSION = {rpc['version']}\n", + f"RPC_SUPPORTED_VERSIONS = {py_tuple(rpc['supported_versions'])}\n", + f"RPC_FEATURES = {py_tuple(rpc['features'])}\n", + f"RPC_METHODS = {py_tuple([method['name'] for method in methods])}\n", + f"MUTATIONS = frozenset({py_tuple(mutations)})\n", + f"OUTCOMES = {py_tuple(manifest['outcomes'])}\n", + f"CLIENT_TO_HOST_MAX_BYTES = {rpc['frame_limits']['client_to_host']}\n", + f"HOST_TO_CLIENT_MAX_BYTES = {rpc['frame_limits']['host_to_client']}\n", + ] + ) + + +def outputs(manifest: dict[str, Any]) -> dict[Path, str]: + return { + ROOT / "host-rs" / "crates" / "agenttab-protocol" / "src" / "generated.rs": render_rust(manifest), + ROOT / "packages" / "sdk-typescript" / "src" / "generated" / "protocol.ts": render_sdk_ts(manifest), + ROOT / "packages" / "extension" / "src" / "generated" / "protocol.ts": render_extension_ts(manifest), + ROOT / "packages" / "sdk-python" / "agenttab" / "_generated_protocol.py": render_python(manifest), + } + + +def check(rendered: dict[Path, str]) -> int: + changed = False + for path, expected in rendered.items(): + actual = path.read_text(encoding="utf-8") if path.exists() else "" + if actual == expected: + continue + changed = True + print(f"generated protocol artifact is stale: {path.relative_to(ROOT)}", file=sys.stderr) + diff = difflib.unified_diff( + actual.splitlines(), + expected.splitlines(), + fromfile=str(path.relative_to(ROOT)), + tofile=f"generated:{path.relative_to(ROOT)}", + lineterm="", + ) + for line in list(diff)[:80]: + print(line, file=sys.stderr) + if changed: + print("run: python3 scripts/generate_protocol.py", file=sys.stderr) + return 1 + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--check", action="store_true", help="fail instead of updating stale generated files") + args = parser.parse_args() + try: + manifest = load_json(MANIFEST_PATH) + validate_manifest(manifest) + rendered = outputs(manifest) + if args.check: + return check(rendered) + for path, content in rendered.items(): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + print(path.relative_to(ROOT)) + return 0 + except (KeyError, OSError, TypeError, ValueError, json.JSONDecodeError) as error: + print(f"protocol generation failed: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/architecture/verify_protocol_schemas.py b/tests/architecture/verify_protocol_schemas.py index 4a19292..260f8e8 100755 --- a/tests/architecture/verify_protocol_schemas.py +++ b/tests/architecture/verify_protocol_schemas.py @@ -208,7 +208,22 @@ def verify_core_messages(schemas: dict[Path, dict], registry: Registry) -> int: ) connection_validator.validate( - {"protocol": "agenttab.rpc", "version": 1, "kind": "connect"} + { + "protocol": "agenttab.rpc", + "version": 2, + "kind": "connect", + "supported_versions": [1, 2], + "supported_features": ["task_resume_v1"], + } + ) + connection_validator.validate( + { + "protocol": "agenttab.rpc", + "version": 1, + "kind": "connect", + "supported_versions": [1], + "supported_features": ["task_resume_v1"], + } ) connected = { "protocol": "agenttab.rpc", @@ -217,6 +232,7 @@ def verify_core_messages(schemas: dict[Path, dict], registry: Registry) -> int: "connection_id": "018f47a0-7b10-7abc-8def-0123456789ab", "resumed": False, "state": "ready", + "features": ["task_resume_v1"], } connection_validator.validate(connected) missing_state = dict(connected) @@ -235,7 +251,29 @@ def verify_core_messages(schemas: dict[Path, dict], registry: Registry) -> int: resume_capability="r" * 32, ) ) - return len(requests) + 10 + connection_validator.validate( + { + "protocol": "agenttab.rpc", + "version": 1, + "kind": "incompatible", + "requested_protocol": "agenttab.rpc", + "requested_version": 2, + "supported_versions": [1], + "message": "AgentTab Core does not support protocol version 2", + "recovery": "Update AgentTab.", + } + ) + expect_invalid( + connection_validator, + { + "protocol": "agenttab.rpc", + "version": 1, + "kind": "connect", + "supported_versions": [1, 1], + }, + "duplicate supported versions", + ) + return len(requests) + 16 def verify_native_messages(schemas: dict[Path, dict], registry: Registry) -> int: @@ -256,7 +294,43 @@ def verify_native_messages(schemas: dict[Path, dict], registry: Registry) -> int missing_task = dict(close_task) missing_task.pop("task_id") expect_invalid(native_validator, missing_task, "close_task task binding") - return 3 + native_validator.validate( + { + "protocol": "agenttab.native", + "version": 2, + "kind": "hello", + "extension_version": "2.0.0", + "inventory": [], + "paused": False, + "handoff": {"active": False}, + "staged_commits": [], + "supported_versions": [1, 2], + "supported_features": ["event_ack_v1"], + } + ) + native_validator.validate( + { + "protocol": "agenttab.native", + "version": 1, + "kind": "ready", + "host_version": "2.0.0", + "state": "ready", + "features": ["event_ack_v1"], + } + ) + native_validator.validate( + { + "protocol": "agenttab.native", + "version": 1, + "kind": "incompatible", + "requested_protocol": "agenttab.native", + "requested_version": 2, + "supported_versions": [1], + "message": "Native protocol version 2 is unsupported", + "recovery": "Update AgentTab.", + } + ) + return 6 def main() -> int: From 946f7d261a1120273a00172331b2b57bb42afe18 Mon Sep 17 00:00:00 2001 From: Wolfie Date: Sun, 30 Aug 2026 18:38:52 -0700 Subject: [PATCH 2/2] Satisfy the Rust formatting gate --- host-rs/crates/agenttab-host/src/native.rs | 5 +---- host-rs/crates/agenttab-protocol/src/lib.rs | 5 +---- 2 files changed, 2 insertions(+), 8 deletions(-) diff --git a/host-rs/crates/agenttab-host/src/native.rs b/host-rs/crates/agenttab-host/src/native.rs index 005df6c..6ed5a6a 100644 --- a/host-rs/crates/agenttab-host/src/native.rs +++ b/host-rs/crates/agenttab-host/src/native.rs @@ -171,10 +171,7 @@ impl StdioNative { .get("version") .and_then(Value::as_u64) .unwrap_or_default(); - let kind = value - .get("kind") - .and_then(Value::as_str) - .map(str::to_owned); + let kind = value.get("kind").and_then(Value::as_str).map(str::to_owned); let supports_host_version = value .get("supported_versions") .and_then(Value::as_array) diff --git a/host-rs/crates/agenttab-protocol/src/lib.rs b/host-rs/crates/agenttab-protocol/src/lib.rs index 1564abb..7b0381b 100644 --- a/host-rs/crates/agenttab-protocol/src/lib.rs +++ b/host-rs/crates/agenttab-protocol/src/lib.rs @@ -2227,10 +2227,7 @@ mod tests { "supported_features": ["event_ack_v1", "future_feature"] })) .unwrap(); - assert_eq!( - hello.negotiated_features().unwrap(), - vec!["event_ack_v1"] - ); + assert_eq!(hello.negotiated_features().unwrap(), vec!["event_ack_v1"]); assert!(matches!( NativeHello::parse(json!({ "protocol": NATIVE_PROTOCOL,