fix: align browser validation with Rust wire shape
This commit is contained in:
@@ -757,7 +757,6 @@ pub struct InternalWorkerSnapshot {
|
||||
pub revision: u64,
|
||||
pub session: SessionSnapshot,
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub status: WorkerStatus,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub error: Option<String>,
|
||||
@@ -811,13 +810,10 @@ pub struct PendingSubmissionSummary {
|
||||
pub struct PendingSubmissionsSnapshot {
|
||||
pub revision: u64,
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub notification_count: u32,
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub head_id: Option<String>,
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub submissions: Vec<PendingSubmissionSummary>,
|
||||
}
|
||||
|
||||
@@ -831,7 +827,6 @@ pub struct PendingSubmissionsSnapshot {
|
||||
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
|
||||
pub struct SessionSnapshot {
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub pending_submissions: PendingSubmissionsSnapshot,
|
||||
pub entries: Vec<SessionSnapshotEntry>,
|
||||
}
|
||||
@@ -1099,7 +1094,6 @@ pub enum Event {
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
disposition: Option<ToolResultDisposition>,
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
is_error: bool,
|
||||
},
|
||||
/// Token accounting for one LLM request.
|
||||
@@ -1328,7 +1322,6 @@ pub enum CompletionKind {
|
||||
pub struct CompletionEntry {
|
||||
pub value: String,
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub is_dir: bool,
|
||||
}
|
||||
|
||||
@@ -1560,11 +1553,9 @@ pub struct Greeting {
|
||||
pub tools: Vec<String>,
|
||||
/// Model context window in tokens. Always filled by the Worker greeting.
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub context_window: u64,
|
||||
/// Estimated current session context tokens at connect time.
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub context_tokens: u64,
|
||||
}
|
||||
|
||||
@@ -1677,7 +1668,6 @@ pub struct ScopeRule {
|
||||
/// Which path identity an allow rule uses when symbolic links are
|
||||
/// encountered. Deny rules always inspect both identities.
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub symlink_policy: SymlinkPolicy,
|
||||
}
|
||||
|
||||
|
||||
@@ -645,7 +645,6 @@ pub struct SubscriptionWorker {
|
||||
/// Freshness of the Runtime-backed observation carried by this projection.
|
||||
/// `Unavailable` preserves catalog membership without claiming that execution stopped.
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub availability: SubscriptionWorkerAvailability,
|
||||
/// Producer-owned monotonic revision for this Worker subject.
|
||||
pub subject_revision: u64,
|
||||
@@ -656,7 +655,6 @@ pub struct SubscriptionWorker {
|
||||
/// Runtime catalog lifecycle compatibility projection; not foreground-state authority.
|
||||
pub state: SubscriptionWorkerState,
|
||||
#[serde(default)]
|
||||
#[cfg_attr(feature = "json-schema", schemars(required))]
|
||||
pub has_running_internal_workers: bool,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub workspace_id: Option<String>,
|
||||
|
||||
@@ -156,6 +156,8 @@ pub fn generated_protocol_validator() -> String {
|
||||
.expect("SubscriptionFrame schema must serialize");
|
||||
close_object_shapes(&mut schema);
|
||||
let schema = serde_json::to_string(&schema).expect("SubscriptionFrame schema must serialize");
|
||||
let fixtures = serde_json::to_string(&rust_serialized_subscription_frame_fixtures())
|
||||
.expect("SubscriptionFrame fixtures must serialize");
|
||||
include_str!("typescript_validator_template.ts")
|
||||
.replace(
|
||||
"__GENERATED_PREAMBLE__",
|
||||
@@ -202,6 +204,58 @@ pub fn generated_protocol_validator() -> String {
|
||||
&MAX_REJECTION_MESSAGE_BYTES.to_string(),
|
||||
)
|
||||
.replace("__SUBSCRIPTION_FRAME_SCHEMA__", &schema)
|
||||
.replace("__RUST_SERIALIZED_FIXTURES__", &fixtures)
|
||||
}
|
||||
|
||||
fn rust_serialized_subscription_frame_fixtures() -> Vec<SubscriptionFrame> {
|
||||
let request_id =
|
||||
SubscriptionRequestId::new("fixture-request").expect("fixture request id must be valid");
|
||||
let subscription_id =
|
||||
SubscriptionId::new("fixture-subscription").expect("fixture subscription id must be valid");
|
||||
let worker_id =
|
||||
SubscriptionWorkerId::new("fixture-worker").expect("fixture worker id must be valid");
|
||||
let worker = SubscriptionWorker {
|
||||
worker_id: worker_id.clone(),
|
||||
runtime_id: Some("fixture-runtime".to_string()),
|
||||
resource_key: Some("fixture-worker-resource".to_string()),
|
||||
availability: SubscriptionWorkerAvailability::default(),
|
||||
subject_revision: 1,
|
||||
worker_state: None,
|
||||
state: SubscriptionWorkerState::Idle,
|
||||
has_running_internal_workers: false,
|
||||
workspace_id: Some("fixture-workspace".to_string()),
|
||||
display_name: None,
|
||||
profile: None,
|
||||
workdir_attachments: Vec::new(),
|
||||
};
|
||||
let subscribed = SubscriptionFrame::new(SubscriptionFramePayload::Response(
|
||||
SubscriptionResponse::Subscribed {
|
||||
request_id,
|
||||
subscription_id: subscription_id.clone(),
|
||||
selector: EventSubscriptionSelector::WorkspaceWorkers,
|
||||
snapshot_revision: 1,
|
||||
snapshot: SubscriptionSnapshot::Workers {
|
||||
workers: vec![worker],
|
||||
},
|
||||
},
|
||||
));
|
||||
let pending =
|
||||
SubscriptionFrame::new(SubscriptionFramePayload::Event(SubscriptionEvent::Event {
|
||||
subscription_id,
|
||||
subject_revision: 2,
|
||||
payload: SubscriptionEventPayload::WorkerProtocol {
|
||||
worker_id,
|
||||
event: Event::PendingSubmissionsChanged {
|
||||
pending: PendingSubmissionsSnapshot::default(),
|
||||
},
|
||||
},
|
||||
}));
|
||||
for fixture in [&subscribed, &pending] {
|
||||
fixture
|
||||
.validate()
|
||||
.expect("Rust-serialized Browser compatibility fixture must validate");
|
||||
}
|
||||
vec![subscribed, pending]
|
||||
}
|
||||
|
||||
fn close_object_shapes(schema: &mut Value) {
|
||||
@@ -218,6 +272,9 @@ fn close_object_shapes(schema: &mut Value) {
|
||||
for value in object.values_mut() {
|
||||
close_object_shapes(value);
|
||||
}
|
||||
normalize_null_default(object);
|
||||
normalize_integer_format(object);
|
||||
require_serialized_defaults(object);
|
||||
if object.get("type").and_then(Value::as_str) == Some("object")
|
||||
&& object.contains_key("properties")
|
||||
{
|
||||
@@ -225,6 +282,98 @@ fn close_object_shapes(schema: &mut Value) {
|
||||
}
|
||||
}
|
||||
|
||||
fn normalize_null_default(schema: &mut Map<String, Value>) {
|
||||
if schema.get("default") != Some(&Value::Null) || schema.contains_key("anyOf") {
|
||||
return;
|
||||
}
|
||||
let Some(schema_type) = schema.get_mut("type") else {
|
||||
return;
|
||||
};
|
||||
match schema_type {
|
||||
Value::String(value) if value != "null" => {
|
||||
*schema_type = Value::Array(vec![
|
||||
Value::String(value.clone()),
|
||||
Value::String("null".into()),
|
||||
]);
|
||||
}
|
||||
Value::Array(values) if !values.iter().any(|value| value.as_str() == Some("null")) => {
|
||||
values.push(Value::String("null".into()));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
fn normalize_integer_format(schema: &mut Map<String, Value>) {
|
||||
let is_integer = match schema.get("type") {
|
||||
Some(Value::String(value)) => value == "integer",
|
||||
Some(Value::Array(values)) => values.iter().any(|value| value.as_str() == Some("integer")),
|
||||
_ => false,
|
||||
};
|
||||
if !is_integer {
|
||||
return;
|
||||
}
|
||||
let Some(format) = schema.get("format").and_then(Value::as_str) else {
|
||||
return;
|
||||
};
|
||||
let (minimum, maximum) = match format {
|
||||
"uint16" => (0.0, u16::MAX as f64),
|
||||
"uint32" => (0.0, u32::MAX as f64),
|
||||
"uint" | "uint64" => (0.0, 9_007_199_254_740_991.0),
|
||||
"int32" => (i32::MIN as f64, i32::MAX as f64),
|
||||
"int64" => (-9_007_199_254_740_991.0, 9_007_199_254_740_991.0),
|
||||
other => panic!("unsupported integer format in SubscriptionFrame schema: {other}"),
|
||||
};
|
||||
tighten_numeric_bound(schema, "minimum", minimum, f64::max);
|
||||
tighten_numeric_bound(schema, "maximum", maximum, f64::min);
|
||||
}
|
||||
|
||||
fn tighten_numeric_bound(
|
||||
schema: &mut Map<String, Value>,
|
||||
keyword: &str,
|
||||
contract_bound: f64,
|
||||
tighten: fn(f64, f64) -> f64,
|
||||
) {
|
||||
let bound = schema
|
||||
.get(keyword)
|
||||
.and_then(Value::as_f64)
|
||||
.map_or(contract_bound, |existing| tighten(existing, contract_bound));
|
||||
schema.insert(
|
||||
keyword.to_string(),
|
||||
Value::Number(serde_json::Number::from_f64(bound).expect("finite JSON Schema bound")),
|
||||
);
|
||||
}
|
||||
|
||||
fn require_serialized_defaults(schema: &mut Map<String, Value>) {
|
||||
if schema.get("type").and_then(Value::as_str) != Some("object") {
|
||||
return;
|
||||
}
|
||||
let Some(properties) = schema.get("properties").and_then(Value::as_object) else {
|
||||
return;
|
||||
};
|
||||
let defaulted = properties
|
||||
.iter()
|
||||
.filter(|(_, property)| {
|
||||
property
|
||||
.as_object()
|
||||
.is_some_and(|value| value.contains_key("default"))
|
||||
})
|
||||
.map(|(key, _)| Value::String(key.clone()))
|
||||
.collect::<Vec<_>>();
|
||||
if defaulted.is_empty() {
|
||||
return;
|
||||
}
|
||||
let required = schema
|
||||
.entry("required")
|
||||
.or_insert_with(|| Value::Array(Vec::new()))
|
||||
.as_array_mut()
|
||||
.expect("object required must be an array");
|
||||
for key in defaulted {
|
||||
if !required.contains(&key) {
|
||||
required.push(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Serde flatten is represented by schemars as object properties beside `oneOf`.
|
||||
/// Move those common properties into each object alternative so standard JSON
|
||||
/// Schema `additionalProperties: false` has the intended exact-shape semantics.
|
||||
@@ -319,6 +468,41 @@ mod tests {
|
||||
assert!(generated.contains("repository_key"), "{generated}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn generated_schema_matches_serialized_defaults_and_rust_integer_widths() {
|
||||
let mut schema = serde_json::to_value(schemars::schema_for!(SubscriptionFrame)).unwrap();
|
||||
close_object_shapes(&mut schema);
|
||||
|
||||
let worker_required = schema
|
||||
.pointer("/$defs/SubscriptionWorker/required")
|
||||
.and_then(Value::as_array)
|
||||
.unwrap();
|
||||
assert!(worker_required.contains(&Value::String("availability".into())));
|
||||
assert!(worker_required.contains(&Value::String("has_running_internal_workers".into())));
|
||||
|
||||
let pending_required = schema
|
||||
.pointer("/$defs/PendingSubmissionsSnapshot/required")
|
||||
.and_then(Value::as_array)
|
||||
.unwrap();
|
||||
for field in ["notification_count", "head_id", "submissions"] {
|
||||
assert!(pending_required.contains(&Value::String(field.into())));
|
||||
}
|
||||
assert_eq!(
|
||||
schema.pointer("/$defs/PendingSubmissionsSnapshot/properties/head_id/type"),
|
||||
Some(&serde_json::json!(["string", "null"]))
|
||||
);
|
||||
assert!(
|
||||
generated_protocol_validator().contains("\"format\":\"uint32\",\"maximum\":4294967295")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[should_panic(expected = "unsupported integer format")]
|
||||
fn generated_schema_rejects_unknown_integer_formats() {
|
||||
let mut unsupported = serde_json::json!({"type": "integer", "format": "uint128"});
|
||||
close_object_shapes(&mut unsupported);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn generated_protocol_types_are_current() {
|
||||
let expected = generated_protocol_types();
|
||||
|
||||
@@ -21,6 +21,8 @@ const MAX_RESOURCE_ID_BYTES = __MAX_RESOURCE_ID_BYTES__;
|
||||
const MAX_WORKER_IDS_PER_SELECTOR = __MAX_WORKER_IDS_PER_SELECTOR__;
|
||||
const MAX_REJECTION_MESSAGE_BYTES = __MAX_REJECTION_MESSAGE_BYTES__;
|
||||
const subscriptionFrameSchema: unknown = __SUBSCRIPTION_FRAME_SCHEMA__;
|
||||
export const RUST_SERIALIZED_SUBSCRIPTION_FRAME_FIXTURES: readonly unknown[] =
|
||||
__RUST_SERIALIZED_FIXTURES__;
|
||||
const utf8 = new TextEncoder();
|
||||
|
||||
type JsonObject = Record<string, unknown>;
|
||||
@@ -99,10 +101,12 @@ function withinAggregateLimits(root: unknown): boolean {
|
||||
);
|
||||
}
|
||||
if (!isObject(value)) return false;
|
||||
const entries = Object.values(value);
|
||||
const entries = Object.entries(value);
|
||||
return (
|
||||
entries.length <= MAX_SUBSCRIPTION_COLLECTION_ITEMS &&
|
||||
entries.every((entry) => visit(entry, depth + 1))
|
||||
entries.every(
|
||||
([key, entry]) => visit(key, depth + 1) && visit(entry, depth + 1),
|
||||
)
|
||||
);
|
||||
};
|
||||
return visit(root, 0);
|
||||
|
||||
@@ -17,7 +17,7 @@ The generator fails closed on unsupported or lossy OpenAPI constructs, including
|
||||
|
||||
## Workspace subscription / Worker protocol
|
||||
|
||||
`protocol.ts` and `protocol-validator.ts` are generated together from the Serde/`schemars` closure of `protocol::subscription::SubscriptionFrame`. Rust DTOs and validation constants in `crates/protocol` remain the sole wire authority. The validator closes every generated object shape, rejects unknown tagged variants and unsafe integers, applies Rust-owned aggregate bounds, and mirrors the Browser-relevant `SubscriptionFrame::validate()` selector/payload checks. The Workspace multiplexer additionally checks live event payloads against its locally retained selector.
|
||||
`protocol.ts` and `protocol-validator.ts` are generated together from the Serde/`schemars` closure of `protocol::subscription::SubscriptionFrame`. Rust DTOs and validation constants in `crates/protocol` remain the sole wire authority. The generator normalizes Serde defaults to the actual serialized required/nullable shape, tightens every supported Rust integer format to its JavaScript-safe wire range, and fails on an unknown integer format. The validator closes every generated object shape, rejects unknown tagged variants and unsafe integers, applies Rust-owned aggregate bounds to object keys as well as values, and mirrors the Browser-relevant `SubscriptionFrame::validate()` selector/payload checks. Rust-serialized compatibility fixtures are embedded for Browser regression tests. The Workspace multiplexer additionally checks live event payloads against its locally retained selector.
|
||||
|
||||
Inbound Browser frames only are validated: Browser-to-Server methods remain compile-time typed and are decoded plus semantically validated by the Rust server. Invalid inbound frames close the shared socket with a fixed, non-reflective diagnostic; the existing reconnect path clears routing state and requests fresh snapshots. A valid `subscription_closed` frame continues to isolate and resubscribe only that subscription.
|
||||
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -1,6 +1,8 @@
|
||||
import type { EventSubscriptionSelector, SubscriptionFrame } from '$lib/generated/protocol';
|
||||
import {
|
||||
decodeSubscriptionFrame,
|
||||
MAX_SUBSCRIPTION_STRING_BYTES,
|
||||
RUST_SERIALIZED_SUBSCRIPTION_FRAME_FIXTURES,
|
||||
subscriptionFrameMatchesSelector,
|
||||
} from '$lib/generated/protocol-validator';
|
||||
|
||||
@@ -40,6 +42,13 @@ Deno.test('generated subscription validator accepts a current exact frame', () =
|
||||
assert(result.value.frame === 'response', 'expected response frame');
|
||||
});
|
||||
|
||||
Deno.test('Rust-serialized compatibility fixtures satisfy the generated Browser contract', () => {
|
||||
for (const fixture of RUST_SERIALIZED_SUBSCRIPTION_FRAME_FIXTURES) {
|
||||
const result = decode(fixture);
|
||||
assert(result.ok, `Rust fixture was rejected: ${JSON.stringify(result)}`);
|
||||
}
|
||||
});
|
||||
|
||||
Deno.test('generated subscription validator rejects malformed and newer shapes', () => {
|
||||
const cases: Array<[string, unknown]> = [
|
||||
['malformed json', '{'],
|
||||
@@ -52,6 +61,19 @@ Deno.test('generated subscription validator rejects malformed and newer shapes',
|
||||
],
|
||||
['wrong field type', subscribedFrame({ snapshot_revision: '1' })],
|
||||
['unsafe integer', subscribedFrame({ snapshot_revision: Number.MAX_SAFE_INTEGER + 1 })],
|
||||
[
|
||||
'missing serialized Worker defaults',
|
||||
subscribedFrame({
|
||||
snapshot: {
|
||||
topic: 'workers',
|
||||
data: {
|
||||
workers: [
|
||||
{ worker_id: 'worker-1', subject_revision: 1, state: 'idle' },
|
||||
],
|
||||
},
|
||||
},
|
||||
}),
|
||||
],
|
||||
['unsupported version', { ...subscribedFrame(), protocol_version: 2 }],
|
||||
[
|
||||
'newer Worker event',
|
||||
@@ -71,6 +93,37 @@ Deno.test('generated subscription validator rejects malformed and newer shapes',
|
||||
},
|
||||
},
|
||||
],
|
||||
[
|
||||
'out-of-range Rust u32',
|
||||
{
|
||||
protocol_version: 1,
|
||||
frame: 'event',
|
||||
message: {
|
||||
event: 'event',
|
||||
data: {
|
||||
subscription_id: 'subscription-1',
|
||||
subject_revision: 2,
|
||||
payload: {
|
||||
event: 'worker_protocol',
|
||||
data: {
|
||||
worker_id: 'worker-1',
|
||||
event: {
|
||||
event: 'llm_retry',
|
||||
data: {
|
||||
llm_call: 1,
|
||||
failed_attempt: 4_294_967_296,
|
||||
max_attempts: 2,
|
||||
wait_ms: 1,
|
||||
elapsed_ms: 1,
|
||||
error: 'retry',
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
],
|
||||
];
|
||||
|
||||
for (const [name, value] of cases) {
|
||||
@@ -161,6 +214,32 @@ Deno.test('generated subscription validator enforces Rust semantic bounds and se
|
||||
}
|
||||
});
|
||||
|
||||
Deno.test('aggregate limits include keys in unconstrained Worker JSON values', () => {
|
||||
const oversizedKey = 'k'.repeat(MAX_SUBSCRIPTION_STRING_BYTES + 1);
|
||||
const result = decode({
|
||||
protocol_version: 1,
|
||||
frame: 'event',
|
||||
message: {
|
||||
event: 'event',
|
||||
data: {
|
||||
subscription_id: 'subscription-1',
|
||||
subject_revision: 2,
|
||||
payload: {
|
||||
event: 'worker_protocol',
|
||||
data: {
|
||||
worker_id: 'worker-1',
|
||||
event: {
|
||||
event: 'system_item',
|
||||
data: { item: { [oversizedKey]: true } },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
assert(!result.ok && result.reason === 'aggregate_limit', 'oversized object key passed');
|
||||
});
|
||||
|
||||
Deno.test('stateful selector validation fences a routed event before projection', () => {
|
||||
const frame: SubscriptionFrame = {
|
||||
protocol_version: 1,
|
||||
|
||||
Reference in New Issue
Block a user