Merge remote-tracking branch 'origin/develop' into work/T-652-external-workdir-interactive-rw

crates/protocol/Cargo.toml

@@ -7,7 +7,7 @@ license.workspace = true
 [features]
 default = ["stream"]
 stream = ["dep:tokio"]
-typescript = ["dep:ts-rs"]
+typescript = ["dep:ts-rs", "json-schema"]
 json-schema = ["dep:schemars"]
 
 [dependencies]

crates/protocol/examples/generate_typescript.rs

@@ -1,11 +1,38 @@
 #[cfg(feature = "typescript")]
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let path = protocol::typescript::generated_typescript_path();
-    if let Some(parent) = path.parent() {
-        std::fs::create_dir_all(parent)?;
+    let check = std::env::args()
+        .skip(1)
+        .any(|argument| argument == "--check");
+    let outputs = [
+        (
+            protocol::typescript::generated_typescript_path(),
+            protocol::typescript::generated_protocol_types(),
+        ),
+        (
+            protocol::typescript::generated_runtime_validator_path(),
+            protocol::typescript::generated_protocol_validator(),
+        ),
+    ];
+
+    for (path, expected) in outputs {
+        if check {
+            let actual = std::fs::read_to_string(&path)?;
+            if actual != expected {
+                return Err(format!(
+                    "generated protocol artifact is stale: {}; run without --check",
+                    path.display()
+                )
+                .into());
+            }
+            println!("current {}", path.display());
+        } else {
+            if let Some(parent) = path.parent() {
+                std::fs::create_dir_all(parent)?;
+            }
+            std::fs::write(&path, expected)?;
+            println!("wrote {}", path.display());
+        }
     }
-    std::fs::write(&path, protocol::typescript::generated_protocol_types())?;
-    println!("wrote {}", path.display());
     Ok(())
 }
 

crates/protocol/src/lib.rs

@@ -83,6 +83,7 @@ impl AuthenticatedInputSource {
 /// of an accepted id must retain the same command kind.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct WorkerCommandEnvelope {
     pub command_id: u64,
 }
@@ -95,6 +96,7 @@ impl WorkerCommandEnvelope {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum WorkerCommandKind {
     Resume,
@@ -106,6 +108,7 @@ pub enum WorkerCommandKind {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum WorkerCommandDisposition {
     Accepted,
@@ -116,6 +119,7 @@ pub enum WorkerCommandDisposition {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct WorkerCommandAcknowledgement {
     pub command_id: u64,
     pub command: WorkerCommandKind,
@@ -212,6 +216,7 @@ impl From<WorkerStatus> for WorkerStateSnapshot {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "method", content = "params", rename_all = "snake_case")]
 pub enum Method {
     /// Durably accept typed input for immediate activation while the Worker is Idle.
@@ -228,6 +233,7 @@ pub enum Method {
     /// Authenticated transport form of Submit. Trusted adapters replace
     /// public Submit before forwarding it to the Worker.
     #[cfg_attr(feature = "typescript", ts(skip))]
+    #[cfg_attr(feature = "json-schema", schemars(skip))]
     SubmitTracked {
         submission_request_id: String,
         input: Vec<Segment>,
@@ -245,6 +251,7 @@ pub enum Method {
     },
     /// Authenticated transport form of Notify.
     #[cfg_attr(feature = "typescript", ts(skip))]
+    #[cfg_attr(feature = "json-schema", schemars(skip))]
     NotifyTracked {
         notification_request_id: String,
         message: String,
@@ -344,6 +351,7 @@ pub enum Method {
 /// child Worker).
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum WorkerEvent {
     /// Child finished one turn and is back to IDLE.
@@ -685,6 +693,7 @@ impl Method {
 /// protocol subjects.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum InternalWorkerKind {
     SubWorker,
@@ -694,6 +703,7 @@ pub enum InternalWorkerKind {
 /// Stable parent-owned lifecycle for one compaction run.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct CompactionLifecycle {
     pub schema_version: u32,
     pub compaction_id: String,
@@ -715,6 +725,7 @@ pub struct CompactionLifecycle {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum CompactionLifecycleState {
     Running,
@@ -727,6 +738,7 @@ pub enum CompactionLifecycleState {
 /// `name` is display-only; `session_id` is the identity used by clients.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct InternalWorkerRef {
     pub session_id: String,
     pub name: String,
@@ -739,6 +751,7 @@ pub struct InternalWorkerRef {
 /// `Event::InternalWorker` updates that raced with parent snapshot assembly.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct InternalWorkerSnapshot {
     pub worker: InternalWorkerRef,
     pub revision: u64,
@@ -755,6 +768,7 @@ pub struct InternalWorkerSnapshot {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum ToolResultDisposition {
     #[default]
@@ -768,6 +782,7 @@ pub enum ToolResultDisposition {
 /// Durable acceptance result for one idempotent Submit request.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum SubmissionDisposition {
     Started,
@@ -780,6 +795,7 @@ pub enum SubmissionDisposition {
 /// provenance remain in the session log and are intentionally not exposed.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct PendingSubmissionSummary {
     pub submission_id: String,
     pub accepted_at_ms: u64,
@@ -790,6 +806,7 @@ pub struct PendingSubmissionSummary {
 /// Revisioned session-owned FIFO projection used by snapshots and live events.
 #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct PendingSubmissionsSnapshot {
     pub revision: u64,
     #[serde(default)]
@@ -807,6 +824,7 @@ pub struct PendingSubmissionsSnapshot {
 /// to replay the durable log format.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct SessionSnapshot {
     #[serde(default)]
     pub pending_submissions: PendingSubmissionsSnapshot,
@@ -815,6 +833,7 @@ pub struct SessionSnapshot {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum SessionEntryProvenance {
     HumanInput,
@@ -829,6 +848,7 @@ pub enum SessionEntryProvenance {
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct SessionSnapshotEntry {
     /// Stable identity from durable history metadata, or a deterministic
     /// identity derived from the legacy segment and log position.
@@ -844,6 +864,7 @@ pub struct SessionSnapshotEntry {
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum SessionSnapshotEntryData {
     UserInput {
@@ -881,6 +902,7 @@ pub enum SessionSnapshotEntryData {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum SessionMessageRole {
     User,
@@ -889,6 +911,7 @@ pub enum SessionMessageRole {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum SessionContentPart {
     Text { text: String },
@@ -897,6 +920,7 @@ pub enum SessionContentPart {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct SessionToolAttachment {
     pub media_type: String,
     /// Base64-encoded durable attachment body. Public snapshots preserve the
@@ -906,6 +930,7 @@ pub struct SessionToolAttachment {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "event", content = "data", rename_all = "snake_case")]
 pub enum Event {
     /// Durable Submit acceptance. New submissions are accepted only while Idle,
@@ -1233,6 +1258,7 @@ pub enum Event {
 /// human-readable.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct Alert {
     pub level: AlertLevel,
     pub source: AlertSource,
@@ -1243,6 +1269,7 @@ pub struct Alert {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct MemoryWorkerEvent {
     pub worker: String,
     pub status: String,
@@ -1257,6 +1284,7 @@ pub struct MemoryWorkerEvent {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum AlertLevel {
     Warn,
@@ -1265,6 +1293,7 @@ pub enum AlertLevel {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum AlertSource {
     Worker,
@@ -1278,6 +1307,7 @@ pub enum AlertSource {
 /// Mirrors the completion prefix sigil: `@` → `File`.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum CompletionKind {
     File,
@@ -1288,6 +1318,7 @@ pub enum CompletionKind {
 /// `value` is a path (file kind).
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct CompletionEntry {
     pub value: String,
     #[serde(default)]
@@ -1296,14 +1327,17 @@ pub struct CompletionEntry {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct RewindTargetId {
     #[cfg_attr(feature = "typescript", ts(type = "string"))]
+    #[cfg_attr(feature = "json-schema", schemars(with = "String"))]
     pub segment_id: uuid::Uuid,
     pub user_input_entry_index: usize,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct RewindTarget {
     pub id: RewindTargetId,
     pub expected_head_entries: usize,
@@ -1320,6 +1354,7 @@ pub struct RewindTarget {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct RewindSummary {
     pub truncated_to_entries: usize,
     pub discarded_entries: usize,
@@ -1330,6 +1365,7 @@ pub struct RewindSummary {
 /// state only and are never appended to Worker history.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum CommandStatus {
     Running,
@@ -1341,6 +1377,7 @@ pub enum CommandStatus {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum CommandStream {
     Stdout,
@@ -1349,6 +1386,7 @@ pub enum CommandStream {
 
 #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct CommandStreamSlice {
     pub start_offset: u64,
     pub end_offset: u64,
@@ -1358,6 +1396,7 @@ pub struct CommandStreamSlice {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct CommandSnapshot {
     pub command_id: String,
     pub tool_call_id: Option<String>,
@@ -1372,6 +1411,7 @@ pub struct CommandSnapshot {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum CommandEvent {
     Started {
@@ -1399,6 +1439,7 @@ pub enum CommandEvent {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum CompactionPhase {
     Preparing,
@@ -1408,6 +1449,7 @@ pub enum CompactionPhase {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum CompactionTrigger {
     Manual,
@@ -1417,6 +1459,7 @@ pub enum CompactionTrigger {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct InFlightCompaction {
     pub phase: CompactionPhase,
     pub started_at_ms: u64,
@@ -1432,6 +1475,7 @@ pub struct InFlightCompaction {
 /// entries.
 #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct InFlightSnapshot {
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub blocks: Vec<InFlightBlock>,
@@ -1453,6 +1497,7 @@ impl InFlightSnapshot {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum InFlightBlock {
     Text {
@@ -1476,6 +1521,7 @@ pub enum InFlightBlock {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum InFlightToolCallState {
     #[default]
@@ -1497,6 +1543,7 @@ impl InFlightToolCallState {
 /// their own view of the manifest.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct Greeting {
     pub worker_name: String,
     pub cwd: String,
@@ -1518,6 +1565,7 @@ pub struct Greeting {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum WorkerStatus {
     #[default]
@@ -1529,6 +1577,7 @@ pub enum WorkerStatus {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum TurnResult {
     Finished,
@@ -1543,6 +1592,7 @@ pub enum TurnResult {
 /// following Turn entry, not by the marker itself.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum InvokeKind {
     /// `Method::Submit` — a user submission.
@@ -1563,6 +1613,7 @@ pub enum InvokeKind {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum RunResult {
     Finished,
@@ -1577,6 +1628,7 @@ pub enum RunResult {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum ErrorCode {
     AlreadyRunning,
@@ -1599,6 +1651,7 @@ pub enum ErrorCode {
 /// A single allow or deny rule inside a scope configuration.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct ScopeRule {
     /// Target path. Must be absolute by the time a `Scope` is built from
     /// this rule — relative paths are resolved per-layer against the
@@ -1629,6 +1682,7 @@ fn default_recursive() -> bool {
 /// path as presented through the Workdir, even when it aliases another target.
 #[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "lowercase")]
 pub enum SymlinkPolicy {
     #[default]
@@ -1644,6 +1698,7 @@ pub enum SymlinkPolicy {
 /// below** the stated level.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "lowercase")]
 pub enum Permission {
     Read,

crates/protocol/src/subscription.rs

@@ -13,6 +13,13 @@ use serde::{Deserialize, Deserializer, Serialize};
 use crate::Event as WorkerProtocolEvent;
 
 pub const SUBSCRIPTION_PROTOCOL_VERSION: u16 = 1;
+/// Browser/runtime wire guardrails. These bounds are generated into the Workspace
+/// Browser validator so aggregate limits remain part of the Rust-owned contract.
+pub const MAX_SUBSCRIPTION_FRAME_JSON_BYTES: usize = 16 * 1024 * 1024;
+pub const MAX_SUBSCRIPTION_STRING_BYTES: usize = 4 * 1024 * 1024;
+pub const MAX_SUBSCRIPTION_COLLECTION_ITEMS: usize = 100_000;
+pub const MAX_SUBSCRIPTION_VALUE_DEPTH: usize = 64;
+pub const MAX_SUBSCRIPTION_VALUE_NODES: usize = 250_000;
 pub const MAX_CORRELATION_ID_BYTES: usize = 128;
 pub const MAX_RESOURCE_ID_BYTES: usize = 256;
 pub const MAX_WORKER_IDS_PER_SELECTOR: usize = 256;
@@ -104,6 +111,7 @@ macro_rules! bounded_identifier {
     ($name:ident, $field:literal, $max:expr) => {
         #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
         #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+        #[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
         pub struct $name(String);
 
         impl $name {
@@ -203,6 +211,7 @@ fn validate_rejection_message(message: &str) -> Result<(), SubscriptionValidatio
 /// semantics is lexical, so equivalent selectors aggregate to one upstream key.
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct SubscriptionWorkerIds(Vec<SubscriptionWorkerId>);
 
 impl SubscriptionWorkerIds {
@@ -257,6 +266,7 @@ impl<'de> Deserialize<'de> for SubscriptionWorkerIds {
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "topic", rename_all = "snake_case")]
 pub enum EventSubscriptionSelector {
     RuntimeWorkers,
@@ -308,6 +318,7 @@ impl EventSubscriptionSelector {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct SubscriptionFrame {
     pub protocol_version: u16,
     #[serde(flatten)]
@@ -336,6 +347,7 @@ impl SubscriptionFrame {
 /// remain typed within each lane.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "frame", content = "message", rename_all = "snake_case")]
 pub enum SubscriptionFramePayload {
     Request(SubscriptionRequest),
@@ -357,6 +369,7 @@ impl SubscriptionFramePayload {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct SubscriptionWorkerProtocolMethod {
     pub subscription_id: SubscriptionId,
     pub method: crate::Method,
@@ -370,6 +383,7 @@ impl SubscriptionWorkerProtocolMethod {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "method", content = "params", rename_all = "snake_case")]
 pub enum SubscriptionRequest {
     SubscribeEvents {
@@ -405,6 +419,7 @@ impl SubscriptionRequest {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "result", content = "payload", rename_all = "snake_case")]
 pub enum SubscriptionResponse {
     Subscribed {
@@ -467,6 +482,7 @@ impl SubscriptionResponse {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum SubscriptionRejectionCode {
     InvalidRequest,
@@ -480,6 +496,7 @@ pub enum SubscriptionRejectionCode {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum SubscriptionTerminationCode {
     Lagged,
@@ -490,6 +507,7 @@ pub enum SubscriptionTerminationCode {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "event", content = "data", rename_all = "snake_case")]
 pub enum SubscriptionEvent {
     Event {
@@ -551,6 +569,7 @@ impl SubscriptionEvent {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum SubscriptionWorkerState {
     Idle,
@@ -564,8 +583,8 @@ pub enum SubscriptionWorkerState {
 /// `Unavailable` deliberately does not imply `Stopped`: it means the last observed
 /// lifecycle state is retained while the Runtime observation source is unavailable.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
-#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(rename_all = "snake_case")]
 pub enum SubscriptionWorkerAvailability {
     #[default]
@@ -575,6 +594,7 @@ pub enum SubscriptionWorkerAvailability {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct SubscriptionWorkerWorkdirAttachment {
     /// Stable Worker-local routing alias.
     pub alias: String,
@@ -582,6 +602,7 @@ pub struct SubscriptionWorkerWorkdirAttachment {
     /// omit this field and require `repository_key` from the Server projection.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     #[cfg_attr(feature = "typescript", ts(skip))]
+    #[cfg_attr(feature = "json-schema", schemars(skip))]
     pub repository_id: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub repository_key: Option<String>,
@@ -610,6 +631,7 @@ impl SubscriptionWorkerWorkdirAttachment {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct SubscriptionWorker {
     pub worker_id: SubscriptionWorkerId,
     /// Set by the Workspace Server when projecting a Runtime-owned Worker to clients.
@@ -674,12 +696,14 @@ impl SubscriptionWorker {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct SubscriptionWorkdir {
     pub working_directory_id: SubscriptionWorkdirId,
     /// Runtime-internal Repository id. Workspace-facing TypeScript contracts
     /// omit this field and require `repository_key` from the Server projection.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     #[cfg_attr(feature = "typescript", ts(skip))]
+    #[cfg_attr(feature = "json-schema", schemars(skip))]
     pub repository_id: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub repository_key: Option<String>,
@@ -709,6 +733,7 @@ impl SubscriptionWorkdir {
 /// enter this DTO; Workspace Server must resolve the required Repository key.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 pub struct WorkspaceSubscriptionWorkdir {
     pub working_directory_id: SubscriptionWorkdirId,
     pub repository_key: String,
@@ -726,6 +751,7 @@ impl WorkspaceSubscriptionWorkdir {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "topic", content = "data", rename_all = "snake_case")]
 pub enum SubscriptionSnapshot {
     Workers {
@@ -789,6 +815,7 @@ impl SubscriptionSnapshot {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
 #[serde(tag = "event", content = "data", rename_all = "snake_case")]
 pub enum SubscriptionEventPayload {
     WorkerUpserted {

crates/protocol/src/typescript.rs

@@ -1,5 +1,6 @@
 use std::path::PathBuf;
 
+use serde_json::{Map, Value};
 use ts_rs::{Config, TS};
 
 use crate::{
@@ -17,9 +18,13 @@ use crate::{
     WorkerCommandDisposition, WorkerCommandEnvelope, WorkerCommandKind, WorkerEvent,
     WorkerMaintenanceState, WorkerRunState, WorkerState, WorkerStateSnapshot, WorkerStatus,
     subscription::{
-        EventSubscriptionSelector, SubscriptionEvent, SubscriptionEventPayload, SubscriptionFrame,
-        SubscriptionFramePayload, SubscriptionId, SubscriptionRejectionCode, SubscriptionRequest,
-        SubscriptionRequestId, SubscriptionResponse, SubscriptionSnapshot,
+        EventSubscriptionSelector, MAX_CORRELATION_ID_BYTES, MAX_REJECTION_MESSAGE_BYTES,
+        MAX_RESOURCE_ID_BYTES, MAX_SUBSCRIPTION_COLLECTION_ITEMS,
+        MAX_SUBSCRIPTION_FRAME_JSON_BYTES, MAX_SUBSCRIPTION_STRING_BYTES,
+        MAX_SUBSCRIPTION_VALUE_DEPTH, MAX_SUBSCRIPTION_VALUE_NODES, MAX_WORKER_IDS_PER_SELECTOR,
+        SUBSCRIPTION_PROTOCOL_VERSION, SubscriptionEvent, SubscriptionEventPayload,
+        SubscriptionFrame, SubscriptionFramePayload, SubscriptionId, SubscriptionRejectionCode,
+        SubscriptionRequest, SubscriptionRequestId, SubscriptionResponse, SubscriptionSnapshot,
         SubscriptionTerminationCode, SubscriptionWorkdirId, SubscriptionWorker,
         SubscriptionWorkerAvailability, SubscriptionWorkerId, SubscriptionWorkerIds,
         SubscriptionWorkerProtocolMethod, SubscriptionWorkerState,
@@ -28,12 +33,19 @@ use crate::{
 };
 
 const GENERATED_RELATIVE_PATH: &str = "../../web/workspace/src/lib/generated/protocol.ts";
+const GENERATED_VALIDATOR_RELATIVE_PATH: &str =
+    "../../web/workspace/src/lib/generated/protocol-validator.ts";
 
 /// Repository-relative destination for the Workspace web protocol bindings.
 pub fn generated_typescript_path() -> PathBuf {
     PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(GENERATED_RELATIVE_PATH)
 }
 
+/// Repository-relative destination for the generated Workspace Browser validator.
+pub fn generated_runtime_validator_path() -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(GENERATED_VALIDATOR_RELATIVE_PATH)
+}
+
 /// Render Workspace web TypeScript bindings for the Worker wire protocol DTOs.
 ///
 /// Rust DTOs in this crate remain the source of truth; this function is used by
@@ -134,6 +146,293 @@ pub fn generated_protocol_types() -> String {
     normalize_typescript(output)
 }
 
+/// Render the strict runtime validator paired with the generated declarations.
+///
+/// `schemars` projects the reachable Serde shape from [`SubscriptionFrame`]. The
+/// normalization closes every object and expands flattened enum payloads before
+/// embedding that schema into the dependency-free Browser validator template.
+pub fn generated_protocol_validator() -> String {
+    let mut schema = serde_json::to_value(schemars::schema_for!(SubscriptionFrame))
+        .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__",
+            "// @generated by `cargo run -p protocol --example generate_typescript --features typescript`\n// Do not edit by hand. Rust DTO and validation authority lives in `crates/protocol`.",
+        )
+        .replace(
+            "__PROTOCOL_VERSION__",
+            &SUBSCRIPTION_PROTOCOL_VERSION.to_string(),
+        )
+        .replace(
+            "__MAX_FRAME_BYTES__",
+            &MAX_SUBSCRIPTION_FRAME_JSON_BYTES.to_string(),
+        )
+        .replace(
+            "__MAX_STRING_BYTES__",
+            &MAX_SUBSCRIPTION_STRING_BYTES.to_string(),
+        )
+        .replace(
+            "__MAX_COLLECTION_ITEMS__",
+            &MAX_SUBSCRIPTION_COLLECTION_ITEMS.to_string(),
+        )
+        .replace(
+            "__MAX_VALUE_DEPTH__",
+            &MAX_SUBSCRIPTION_VALUE_DEPTH.to_string(),
+        )
+        .replace(
+            "__MAX_VALUE_NODES__",
+            &MAX_SUBSCRIPTION_VALUE_NODES.to_string(),
+        )
+        .replace(
+            "__MAX_CORRELATION_ID_BYTES__",
+            &MAX_CORRELATION_ID_BYTES.to_string(),
+        )
+        .replace(
+            "__MAX_RESOURCE_ID_BYTES__",
+            &MAX_RESOURCE_ID_BYTES.to_string(),
+        )
+        .replace(
+            "__MAX_WORKER_IDS_PER_SELECTOR__",
+            &MAX_WORKER_IDS_PER_SELECTOR.to_string(),
+        )
+        .replace(
+            "__MAX_REJECTION_MESSAGE_BYTES__",
+            &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) {
+    let Value::Object(object) = schema else {
+        if let Value::Array(values) = schema {
+            for value in values {
+                close_object_shapes(value);
+            }
+        }
+        return;
+    };
+
+    expand_flattened_object(object);
+    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")
+    {
+        object.insert("additionalProperties".to_string(), Value::Bool(false));
+    }
+}
+
+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.
+fn expand_flattened_object(object: &mut Map<String, Value>) {
+    if object.get("type").and_then(Value::as_str) != Some("object")
+        || !object.contains_key("properties")
+    {
+        return;
+    }
+    let Some(alternatives) = object.get("oneOf").and_then(Value::as_array) else {
+        return;
+    };
+    if !alternatives.iter().all(|alternative| {
+        alternative
+            .as_object()
+            .and_then(|value| value.get("type"))
+            .and_then(Value::as_str)
+            == Some("object")
+    }) {
+        return;
+    }
+
+    let base_properties = object
+        .remove("properties")
+        .and_then(|value| value.as_object().cloned())
+        .unwrap_or_default();
+    let base_required = object
+        .remove("required")
+        .and_then(|value| value.as_array().cloned())
+        .unwrap_or_default();
+    object.remove("type");
+    let alternatives = object
+        .get_mut("oneOf")
+        .and_then(Value::as_array_mut)
+        .expect("oneOf was checked above");
+    for alternative in alternatives {
+        let alternative = alternative
+            .as_object_mut()
+            .expect("oneOf object alternative was checked above");
+        let properties = alternative
+            .entry("properties")
+            .or_insert_with(|| Value::Object(Map::new()))
+            .as_object_mut()
+            .expect("object alternative properties must be an object");
+        for (key, value) in &base_properties {
+            assert!(
+                properties.insert(key.clone(), value.clone()).is_none(),
+                "flattened schema property {key:?} collides with an enum payload"
+            );
+        }
+        let required = alternative
+            .entry("required")
+            .or_insert_with(|| Value::Array(Vec::new()))
+            .as_array_mut()
+            .expect("object alternative required must be an array");
+        required.extend(base_required.iter().cloned());
+    }
+}
+
 fn normalize_typescript(output: String) -> String {
     let mut lines = output.lines().map(str::trim_end).collect::<Vec<_>>();
     while lines.last() == Some(&"") {
@@ -169,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();
@@ -180,4 +514,18 @@ mod tests {
             "generated TypeScript protocol bindings are stale; run `cargo run -p protocol --example generate_typescript --features typescript`"
         );
     }
+
+    #[test]
+    fn generated_protocol_validator_is_current_and_strict() {
+        let expected = generated_protocol_validator();
+        assert!(expected.contains("\"additionalProperties\":false"));
+        assert!(expected.contains("MAX_SUBSCRIPTION_FRAME_JSON_BYTES = 16777216"));
+        let path = generated_runtime_validator_path();
+        let actual = std::fs::read_to_string(&path)
+            .unwrap_or_else(|err| panic!("failed to read {}: {err}", path.display()));
+        assert_eq!(
+            actual, expected,
+            "generated TypeScript protocol validator is stale; run `cargo run -p protocol --example generate_typescript --features typescript`"
+        );
+    }
 }

crates/protocol/src/typescript_validator_template.ts

@@ -0,0 +1,434 @@
+__GENERATED_PREAMBLE__
+
+import type {
+  EventSubscriptionSelector,
+  SubscriptionEventPayload,
+  SubscriptionFrame,
+  SubscriptionSnapshot,
+  SubscriptionWorker,
+  WorkspaceSubscriptionWorkdir,
+} from './protocol';
+
+export const SUBSCRIPTION_PROTOCOL_VERSION = __PROTOCOL_VERSION__;
+export const MAX_SUBSCRIPTION_FRAME_JSON_BYTES = __MAX_FRAME_BYTES__;
+export const MAX_SUBSCRIPTION_STRING_BYTES = __MAX_STRING_BYTES__;
+export const MAX_SUBSCRIPTION_COLLECTION_ITEMS = __MAX_COLLECTION_ITEMS__;
+export const MAX_SUBSCRIPTION_VALUE_DEPTH = __MAX_VALUE_DEPTH__;
+export const MAX_SUBSCRIPTION_VALUE_NODES = __MAX_VALUE_NODES__;
+
+const MAX_CORRELATION_ID_BYTES = __MAX_CORRELATION_ID_BYTES__;
+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>;
+type RejectionReason =
+  | 'invalid_transport'
+  | 'frame_too_large'
+  | 'invalid_json'
+  | 'aggregate_limit'
+  | 'invalid_structure'
+  | 'invalid_semantics';
+
+export type SubscriptionFrameDecodeResult =
+  | { ok: true; value: SubscriptionFrame }
+  | { ok: false; reason: RejectionReason };
+
+/**
+ * Decode an untrusted Workspace WebSocket message without reflecting its content.
+ * A successful result has passed aggregate limits, the exact generated JSON schema,
+ * and the Rust protocol's Browser-relevant semantic checks.
+ */
+export function decodeSubscriptionFrame(input: unknown): SubscriptionFrameDecodeResult {
+  if (typeof input !== 'string') return { ok: false, reason: 'invalid_transport' };
+  if (utf8.encode(input).byteLength > MAX_SUBSCRIPTION_FRAME_JSON_BYTES) {
+    return { ok: false, reason: 'frame_too_large' };
+  }
+
+  let candidate: unknown;
+  try {
+    candidate = JSON.parse(input) as unknown;
+  } catch {
+    return { ok: false, reason: 'invalid_json' };
+  }
+  if (!withinAggregateLimits(candidate)) return { ok: false, reason: 'aggregate_limit' };
+  if (!validateSchema(candidate, subscriptionFrameSchema)) {
+    return { ok: false, reason: 'invalid_structure' };
+  }
+
+  // The assertion follows generated structural validation; it is not a parse-time cast.
+  const frame = candidate as SubscriptionFrame;
+  if (!validateInboundFrameSemantics(frame)) {
+    return { ok: false, reason: 'invalid_semantics' };
+  }
+  return { ok: true, value: frame };
+}
+
+/** Validate stateful selector/payload consistency after the multiplexer resolves routing. */
+export function subscriptionFrameMatchesSelector(
+  frame: SubscriptionFrame,
+  selector: EventSubscriptionSelector,
+): boolean {
+  if (frame.frame === 'response' && frame.message.result === 'subscribed') {
+    return selectorsEqual(frame.message.payload.selector, selector);
+  }
+  if (frame.frame === 'event' && frame.message.event === 'event') {
+    return eventPayloadMatchesSelector(frame.message.data.payload, selector);
+  }
+  return true;
+}
+
+function withinAggregateLimits(root: unknown): boolean {
+  let nodes = 0;
+  const visit = (value: unknown, depth: number): boolean => {
+    nodes += 1;
+    if (nodes > MAX_SUBSCRIPTION_VALUE_NODES || depth > MAX_SUBSCRIPTION_VALUE_DEPTH) return false;
+    if (typeof value === 'string') {
+      return utf8.encode(value).byteLength <= MAX_SUBSCRIPTION_STRING_BYTES;
+    }
+    if (typeof value === 'number') {
+      return Number.isFinite(value) && (!Number.isInteger(value) || Number.isSafeInteger(value));
+    }
+    if (value === null || typeof value === 'boolean') return true;
+    if (Array.isArray(value)) {
+      return (
+        value.length <= MAX_SUBSCRIPTION_COLLECTION_ITEMS &&
+        value.every((entry) => visit(entry, depth + 1))
+      );
+    }
+    if (!isObject(value)) return false;
+    const entries = Object.entries(value);
+    return (
+      entries.length <= MAX_SUBSCRIPTION_COLLECTION_ITEMS &&
+      entries.every(
+        ([key, entry]) => visit(key, depth + 1) && visit(entry, depth + 1),
+      )
+    );
+  };
+  return visit(root, 0);
+}
+
+function validateSchema(value: unknown, rawSchema: unknown): boolean {
+  if (rawSchema === true) return true;
+  if (rawSchema === false || !isObject(rawSchema)) return false;
+  const schema = rawSchema;
+
+  if (typeof schema.$ref === 'string') {
+    const target = resolveLocalReference(schema.$ref);
+    if (target === undefined || !validateSchema(value, target)) return false;
+  }
+  if ('const' in schema && !jsonEqual(value, schema.const)) return false;
+  if (Array.isArray(schema.enum) && !schema.enum.some((entry) => jsonEqual(value, entry))) {
+    return false;
+  }
+  if (Array.isArray(schema.oneOf)) {
+    let matches = 0;
+    for (const alternative of schema.oneOf) {
+      if (validateSchema(value, alternative)) matches += 1;
+    }
+    if (matches !== 1) return false;
+  }
+  if (Array.isArray(schema.anyOf) && !schema.anyOf.some((entry) => validateSchema(value, entry))) {
+    return false;
+  }
+
+  if (schema.type !== undefined && !matchesType(value, schema.type)) return false;
+  if (typeof value === 'number') {
+    if (schema.type === 'integer' && !Number.isSafeInteger(value)) return false;
+    if (typeof schema.minimum === 'number' && value < schema.minimum) return false;
+    if (typeof schema.maximum === 'number' && value > schema.maximum) return false;
+  }
+  if (typeof value === 'string') {
+    const length = [...value].length;
+    if (typeof schema.minLength === 'number' && length < schema.minLength) return false;
+    if (typeof schema.maxLength === 'number' && length > schema.maxLength) return false;
+  }
+  if (Array.isArray(value)) {
+    if (typeof schema.minItems === 'number' && value.length < schema.minItems) return false;
+    if (typeof schema.maxItems === 'number' && value.length > schema.maxItems) return false;
+    if (schema.items !== undefined && !value.every((entry) => validateSchema(entry, schema.items))) {
+      return false;
+    }
+  }
+  if (isObject(value) && schema.properties !== undefined) {
+    if (!isObject(schema.properties)) return false;
+    const properties = schema.properties;
+    if (Array.isArray(schema.required)) {
+      for (const key of schema.required) {
+        if (typeof key !== 'string' || !Object.prototype.hasOwnProperty.call(value, key)) return false;
+      }
+    }
+    for (const [key, entry] of Object.entries(value)) {
+      if (Object.prototype.hasOwnProperty.call(properties, key)) {
+        if (!validateSchema(entry, properties[key])) return false;
+      } else if (schema.additionalProperties === false) {
+        return false;
+      } else if (isObject(schema.additionalProperties) || typeof schema.additionalProperties === 'boolean') {
+        if (!validateSchema(entry, schema.additionalProperties)) return false;
+      }
+    }
+  }
+  return true;
+}
+
+function resolveLocalReference(reference: string): unknown {
+  if (!reference.startsWith('#/')) return undefined;
+  let current: unknown = subscriptionFrameSchema;
+  for (const encoded of reference.slice(2).split('/')) {
+    if (!isObject(current)) return undefined;
+    const key = encoded.replaceAll('~1', '/').replaceAll('~0', '~');
+    current = current[key];
+  }
+  return current;
+}
+
+function matchesType(value: unknown, rawType: unknown): boolean {
+  if (Array.isArray(rawType)) return rawType.some((entry) => matchesType(value, entry));
+  switch (rawType) {
+    case 'null':
+      return value === null;
+    case 'boolean':
+      return typeof value === 'boolean';
+    case 'string':
+      return typeof value === 'string';
+    case 'number':
+      return typeof value === 'number' && Number.isFinite(value);
+    case 'integer':
+      return typeof value === 'number' && Number.isSafeInteger(value);
+    case 'array':
+      return Array.isArray(value);
+    case 'object':
+      return isObject(value);
+    default:
+      return false;
+  }
+}
+
+function isObject(value: unknown): value is JsonObject {
+  return value !== null && typeof value === 'object' && !Array.isArray(value);
+}
+
+function jsonEqual(left: unknown, right: unknown): boolean {
+  if (left === right) return true;
+  if (Array.isArray(left) && Array.isArray(right)) {
+    return left.length === right.length && left.every((entry, index) => jsonEqual(entry, right[index]));
+  }
+  if (isObject(left) && isObject(right)) {
+    const leftKeys = Object.keys(left);
+    const rightKeys = Object.keys(right);
+    return (
+      leftKeys.length === rightKeys.length &&
+      leftKeys.every(
+        (key) => Object.prototype.hasOwnProperty.call(right, key) && jsonEqual(left[key], right[key]),
+      )
+    );
+  }
+  return false;
+}
+
+function validateInboundFrameSemantics(frame: SubscriptionFrame): boolean {
+  if (frame.protocol_version !== SUBSCRIPTION_PROTOCOL_VERSION) return false;
+  if (frame.frame === 'request' || frame.frame === 'worker_protocol') return false;
+  if (frame.frame === 'response') {
+    const message = frame.message;
+    if (!validateIdentifier(message.payload.request_id, MAX_CORRELATION_ID_BYTES)) return false;
+    if (message.result === 'subscribed') {
+      return (
+        validateIdentifier(message.payload.subscription_id, MAX_CORRELATION_ID_BYTES) &&
+        validateSelector(message.payload.selector) &&
+        snapshotMatchesSelector(message.payload.snapshot, message.payload.selector)
+      );
+    }
+    if (message.result === 'unsubscribed') {
+      return validateIdentifier(message.payload.subscription_id, MAX_CORRELATION_ID_BYTES);
+    }
+    return (
+      (message.payload.subscription_id == null ||
+        validateIdentifier(message.payload.subscription_id, MAX_CORRELATION_ID_BYTES)) &&
+      validateBoundedMessage(message.payload.message)
+    );
+  }
+
+  const message = frame.message;
+  if (!validateIdentifier(message.data.subscription_id, MAX_CORRELATION_ID_BYTES)) return false;
+  if (message.event === 'subscription_closed') {
+    return validateBoundedMessage(message.data.message);
+  }
+  return validateEventPayload(message.data.payload);
+}
+
+function validateSelector(selector: EventSubscriptionSelector): boolean {
+  if (selector.topic === 'worker_lifecycle') {
+    if (
+      selector.worker_ids.length === 0 ||
+      selector.worker_ids.length > MAX_WORKER_IDS_PER_SELECTOR
+    ) {
+      return false;
+    }
+    const ids = new Set<string>();
+    for (const workerId of selector.worker_ids) {
+      if (!validateIdentifier(workerId, MAX_RESOURCE_ID_BYTES) || ids.has(workerId)) return false;
+      ids.add(workerId);
+    }
+    return true;
+  }
+  if (selector.topic === 'worker_protocol') {
+    return (
+      validateIdentifier(selector.worker_id, MAX_RESOURCE_ID_BYTES) &&
+      (selector.runtime_id == null ||
+        validateIdentifier(selector.runtime_id, MAX_RESOURCE_ID_BYTES))
+    );
+  }
+  return true;
+}
+
+function snapshotMatchesSelector(
+  snapshot: SubscriptionSnapshot,
+  selector: EventSubscriptionSelector,
+): boolean {
+  if (!validateSelector(selector)) return false;
+  if (selector.topic === 'runtime_workers' || selector.topic === 'workspace_workers') {
+    return snapshot.topic === 'workers' && validateWorkers(snapshot.data.workers);
+  }
+  if (selector.topic === 'worker_lifecycle') {
+    const selected = new Set(selector.worker_ids);
+    return (
+      snapshot.topic === 'workers' &&
+      validateWorkers(snapshot.data.workers) &&
+      snapshot.data.workers.every((worker) => selected.has(worker.worker_id))
+    );
+  }
+  if (selector.topic === 'worker_protocol') {
+    return snapshot.topic === 'worker_protocol' && snapshot.data.worker_id === selector.worker_id;
+  }
+  return (
+    snapshot.topic === 'workspace_workdirs' &&
+    snapshot.data.workdirs.every(validateWorkspaceWorkdir)
+  );
+}
+
+function validateEventPayload(payload: SubscriptionEventPayload): boolean {
+  switch (payload.event) {
+    case 'worker_upserted':
+      return validateWorker(payload.data.worker);
+    case 'worker_removed':
+      return (
+        validateIdentifier(payload.data.worker_id, MAX_RESOURCE_ID_BYTES) &&
+        (payload.data.runtime_id == null ||
+          validateIdentifier(payload.data.runtime_id, MAX_RESOURCE_ID_BYTES))
+      );
+    case 'worker_protocol':
+      return validateIdentifier(payload.data.worker_id, MAX_RESOURCE_ID_BYTES);
+    case 'workdir_upserted':
+      return validateWorkspaceWorkdir(payload.data.workdir);
+    case 'workdir_removed':
+      return validateIdentifier(payload.data.working_directory_id, MAX_RESOURCE_ID_BYTES);
+  }
+}
+
+function eventPayloadMatchesSelector(
+  payload: SubscriptionEventPayload,
+  selector: EventSubscriptionSelector,
+): boolean {
+  if (selector.topic === 'runtime_workers' || selector.topic === 'workspace_workers') {
+    return payload.event === 'worker_upserted' || payload.event === 'worker_removed';
+  }
+  if (selector.topic === 'worker_lifecycle') {
+    return (
+      (payload.event === 'worker_upserted' && selector.worker_ids.includes(payload.data.worker.worker_id)) ||
+      (payload.event === 'worker_removed' && selector.worker_ids.includes(payload.data.worker_id))
+    );
+  }
+  if (selector.topic === 'worker_protocol') {
+    return payload.event === 'worker_protocol' && payload.data.worker_id === selector.worker_id;
+  }
+  return payload.event === 'workdir_upserted' || payload.event === 'workdir_removed';
+}
+
+function selectorsEqual(
+  left: EventSubscriptionSelector,
+  right: EventSubscriptionSelector,
+): boolean {
+  if (left.topic !== right.topic) return false;
+  if (left.topic === 'worker_lifecycle' && right.topic === 'worker_lifecycle') {
+    const rightIds = new Set(right.worker_ids);
+    return left.worker_ids.length === rightIds.size && left.worker_ids.every((workerId) => rightIds.has(workerId));
+  }
+  if (left.topic === 'worker_protocol' && right.topic === 'worker_protocol') {
+    return (
+      left.worker_id === right.worker_id &&
+      (left.runtime_id ?? null) === (right.runtime_id ?? null)
+    );
+  }
+  return true;
+}
+
+function validateWorkers(workers: SubscriptionWorker[]): boolean {
+  const identities = new Set<string>();
+  for (const worker of workers) {
+    const identity = `${worker.runtime_id ?? ''}\u0000${worker.worker_id}`;
+    if (!validateWorker(worker) || identities.has(identity)) return false;
+    identities.add(identity);
+  }
+  return true;
+}
+
+function validateWorker(worker: SubscriptionWorker): boolean {
+  if (!validateIdentifier(worker.worker_id, MAX_RESOURCE_ID_BYTES)) return false;
+  if (worker.runtime_id != null && !validateIdentifier(worker.runtime_id, MAX_RESOURCE_ID_BYTES)) {
+    return false;
+  }
+  if (worker.resource_key != null && !validateIdentifier(worker.resource_key, MAX_RESOURCE_ID_BYTES)) {
+    return false;
+  }
+  const aliases = new Set<string>();
+  const workdirIds = new Set<string>();
+  for (const attachment of worker.workdir_attachments ?? []) {
+    if (
+      !validateIdentifier(attachment.alias, MAX_RESOURCE_ID_BYTES) ||
+      !validateIdentifier(attachment.working_directory_id, MAX_RESOURCE_ID_BYTES) ||
+      attachment.repository_key == null ||
+      !validateRepositoryKey(attachment.repository_key) ||
+      aliases.has(attachment.alias) ||
+      workdirIds.has(attachment.working_directory_id)
+    ) {
+      return false;
+    }
+    aliases.add(attachment.alias);
+    workdirIds.add(attachment.working_directory_id);
+  }
+  return true;
+}
+
+function validateWorkspaceWorkdir(workdir: WorkspaceSubscriptionWorkdir): boolean {
+  return (
+    validateIdentifier(workdir.working_directory_id, MAX_RESOURCE_ID_BYTES) &&
+    validateRepositoryKey(workdir.repository_key) &&
+    validateIdentifier(workdir.state, MAX_RESOURCE_ID_BYTES)
+  );
+}
+
+function validateIdentifier(value: string, maxBytes: number): boolean {
+  return (
+    value.length > 0 &&
+    utf8.encode(value).byteLength <= maxBytes &&
+    value.trim() === value &&
+    !/\p{Cc}/u.test(value)
+  );
+}
+
+function validateRepositoryKey(value: string): boolean {
+  return (
+    utf8.encode(value).byteLength <= 64 &&
+    /^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$/.test(value)
+  );
+}
+
+function validateBoundedMessage(value: string): boolean {
+  const bytes = utf8.encode(value).byteLength;
+  return bytes > 0 && bytes <= MAX_REJECTION_MESSAGE_BYTES;
+}

crates/worker-runtime/src/worker_backend.rs

@@ -1016,12 +1016,12 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
         worker_ref: &WorkerRef,
     ) -> Result<session_store::RetainedSessionSnapshot, session_store::RetainedSnapshotReadError>
     {
-        let Some(root) = self.worker_aggregate_root.as_ref() else {
-            return Err(session_store::RetainedSnapshotReadError::RetentionMissing);
-        };
+        let aggregate_dir = self
+            .worker_aggregate_dir(worker_ref)
+            .map_err(|_| session_store::RetainedSnapshotReadError::RetentionMissing)?;
         let worker_name = Self::runtime_worker_name_for_ref(worker_ref);
         session_store::read_retained_session_snapshot(
-            &root.join(&worker_name),
+            &aggregate_dir,
             &worker_name,
             session_store::DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES,
         )
@@ -3135,11 +3135,15 @@ mod tests {
     use crate::working_directory::RuntimeGitMaterializer;
     use agen::Engine;
     use agen::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
-    use agen::llm_client::{ClientError, LlmClient, Request};
+    use agen::llm_client::{ClientError, LlmClient, Request, RequestConfig};
     use async_trait::async_trait;
     use futures::{Stream, StreamExt};
     use manifest::{Scope, WorkerManifest};
-    use session_store::{LogEntry, WorkerMetadataStore};
+    use session_store::{
+        LogEntry, LoggedContentPart, LoggedHistoryEntry, LoggedItem, LoggedRole,
+        LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin,
+        Store, WorkerActiveSegmentRef, WorkerMetadata, WorkerMetadataStore,
+    };
 
     #[test]
     fn production_source_has_no_repository_derived_runtime_store() {
@@ -3155,6 +3159,217 @@ mod tests {
         }
     }
 
+    fn persisted_files(root: &Path) -> BTreeMap<PathBuf, Vec<u8>> {
+        fn collect(root: &Path, current: &Path, files: &mut BTreeMap<PathBuf, Vec<u8>>) {
+            if !current.exists() {
+                return;
+            }
+            for entry in fs::read_dir(current).unwrap() {
+                let path = entry.unwrap().path();
+                if path.is_dir() {
+                    collect(root, &path, files);
+                } else {
+                    files.insert(
+                        path.strip_prefix(root).unwrap().to_path_buf(),
+                        fs::read(path).unwrap(),
+                    );
+                }
+            }
+        }
+
+        let mut files = BTreeMap::new();
+        collect(root, root, &mut files);
+        files
+    }
+
+    fn write_retained_profile_worker(
+        runtime_store_dir: &Path,
+        worker_ref: &WorkerRef,
+    ) -> (session_store::SessionId, session_store::SegmentId, String) {
+        let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(worker_ref);
+        let aggregate_dir = runtime_store_dir
+            .join("workers")
+            .join(worker_ref.worker_id.to_string());
+        let aggregate = WorkerAggregateStore::new(&aggregate_dir, &worker_name).unwrap();
+        let session_id = session_store::new_session_id();
+        let segment_id = session_store::new_segment_id();
+        aggregate
+            .write(&WorkerMetadata::new(
+                &worker_name,
+                Some(WorkerActiveSegmentRef::active_segment(
+                    session_id, segment_id,
+                )),
+            ))
+            .unwrap();
+        let entry_id = LoggedSessionHistoryEntryId::new();
+        let expected_entry_id = entry_id.0.clone();
+        let log = [LogEntry::AnnotatedSegmentStart {
+            ts: 17,
+            session_id,
+            system_prompt: None,
+            config: RequestConfig::default(),
+            history: vec![LoggedHistoryEntry {
+                item: LoggedItem::Message {
+                    role: LoggedRole::Assistant,
+                    content: vec![LoggedContentPart::Text {
+                        text: "retained response".to_string(),
+                    }],
+                },
+                metadata: LoggedSessionHistoryMetadata {
+                    entry_id,
+                    origin: LoggedSessionHistoryOrigin::LegacyUnknown,
+                    derivation: None,
+                },
+            }],
+            forked_from: None,
+            compacted_from: None,
+        }];
+        WorkerSessionStore::new(aggregate_dir.join("session"))
+            .unwrap()
+            .create_segment(session_id, segment_id, &log)
+            .unwrap();
+        (session_id, segment_id, expected_entry_id)
+    }
+
+    #[test]
+    fn profile_factory_reads_retained_session_from_canonical_worker_aggregate() {
+        let root = tempfile::tempdir().unwrap();
+        let runtime_store_dir = root.path().join("runtime");
+        let worker_ref = WorkerRef::new(WorkerId::now_v7());
+        let (session_id, segment_id, entry_id) =
+            write_retained_profile_worker(&runtime_store_dir, &worker_ref);
+        let canonical_dir = runtime_store_dir
+            .join("workers")
+            .join(worker_ref.worker_id.to_string());
+        let legacy_prefixed_dir = runtime_store_dir.join("workers").join(
+            ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref),
+        );
+        assert!(!legacy_prefixed_dir.exists());
+        let files_before = persisted_files(&runtime_store_dir);
+        let backend = WorkerRuntimeExecutionBackend::new(
+            ProfileRuntimeWorkerFactory::new(root.path().join("profiles"))
+                .with_runtime_store_dir(&runtime_store_dir),
+        )
+        .unwrap();
+
+        let retained = backend.worker_session(WorkerSessionObservationRequest {
+            worker_ref: worker_ref.clone(),
+        });
+
+        let runtime_api::WorkerSessionAvailability::RetainedSnapshot { identity, snapshot } =
+            retained
+        else {
+            panic!("canonical aggregate must produce a retained snapshot");
+        };
+        assert_eq!(identity.session_id, session_id.to_string());
+        assert_eq!(identity.segment_id, segment_id.to_string());
+        assert_eq!(identity.entry_count, 1);
+        assert_eq!(snapshot.entries.len(), 1);
+        assert_eq!(snapshot.entries[0].entry_id, entry_id);
+        assert_eq!(
+            snapshot.entries[0].data,
+            protocol::SessionSnapshotEntryData::Message {
+                role: protocol::SessionMessageRole::Assistant,
+                content: vec![protocol::SessionContentPart::Text {
+                    text: "retained response".to_string(),
+                }],
+            }
+        );
+        assert!(canonical_dir.is_dir());
+        assert!(!legacy_prefixed_dir.exists());
+        assert_eq!(persisted_files(&runtime_store_dir), files_before);
+        assert!(backend.workers.lock().unwrap().is_empty());
+        assert!(
+            backend
+                .factory
+                .observation_hub
+                .workers
+                .lock()
+                .unwrap()
+                .is_empty()
+        );
+    }
+
+    #[test]
+    fn profile_factory_retained_read_keeps_missing_aggregate_typed_and_absent() {
+        let root = tempfile::tempdir().unwrap();
+        let runtime_store_dir = root.path().join("runtime");
+        let worker_ref = WorkerRef::new(WorkerId::now_v7());
+        let unconfigured_backend = WorkerRuntimeExecutionBackend::new(
+            ProfileRuntimeWorkerFactory::new(root.path().join("unconfigured-profiles")),
+        )
+        .unwrap();
+        assert!(matches!(
+            unconfigured_backend.worker_session(WorkerSessionObservationRequest {
+                worker_ref: worker_ref.clone(),
+            }),
+            runtime_api::WorkerSessionAvailability::Unavailable {
+                reason: runtime_api::WorkerSessionUnavailableReason::RetentionMissing,
+                ..
+            }
+        ));
+        assert!(!root.path().join("unconfigured-profiles").exists());
+        assert!(unconfigured_backend.workers.lock().unwrap().is_empty());
+
+        let canonical_dir = runtime_store_dir
+            .join("workers")
+            .join(worker_ref.worker_id.to_string());
+        let backend = WorkerRuntimeExecutionBackend::new(
+            ProfileRuntimeWorkerFactory::new(root.path().join("profiles"))
+                .with_runtime_store_dir(&runtime_store_dir),
+        )
+        .unwrap();
+
+        let unavailable = backend.worker_session(WorkerSessionObservationRequest {
+            worker_ref: worker_ref.clone(),
+        });
+
+        assert!(matches!(
+            unavailable,
+            runtime_api::WorkerSessionAvailability::Unavailable {
+                reason: runtime_api::WorkerSessionUnavailableReason::RetentionMissing,
+                ..
+            }
+        ));
+        assert!(!canonical_dir.exists());
+        assert!(!runtime_store_dir.exists());
+        assert!(backend.workers.lock().unwrap().is_empty());
+    }
+
+    #[test]
+    fn profile_factory_retained_read_validates_metadata_worker_identity() {
+        let root = tempfile::tempdir().unwrap();
+        let runtime_store_dir = root.path().join("runtime");
+        let worker_ref = WorkerRef::new(WorkerId::now_v7());
+        let aggregate_dir = runtime_store_dir
+            .join("workers")
+            .join(worker_ref.worker_id.to_string());
+        fs::create_dir_all(&aggregate_dir).unwrap();
+        fs::write(
+            aggregate_dir.join("metadata.json"),
+            serde_json::to_vec(&WorkerMetadata::new("different-worker", None)).unwrap(),
+        )
+        .unwrap();
+        let files_before = persisted_files(&runtime_store_dir);
+        let backend = WorkerRuntimeExecutionBackend::new(
+            ProfileRuntimeWorkerFactory::new(root.path().join("profiles"))
+                .with_runtime_store_dir(&runtime_store_dir),
+        )
+        .unwrap();
+
+        let unavailable = backend.worker_session(WorkerSessionObservationRequest { worker_ref });
+
+        assert!(matches!(
+            unavailable,
+            runtime_api::WorkerSessionAvailability::Unavailable {
+                reason: runtime_api::WorkerSessionUnavailableReason::CorruptLog,
+                ref message,
+            } if message == "retained session log is corrupt"
+        ));
+        assert_eq!(persisted_files(&runtime_store_dir), files_before);
+        assert!(backend.workers.lock().unwrap().is_empty());
+    }
+
     #[test]
     fn restored_router_uses_workspace_proxy_for_logical_only_attachment() {
         let root = tempfile::tempdir().unwrap();