Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

feat: integrate Worker state authority

Browse Source
This commit is contained in:
Hare
2026-09-06 11:00:00 +09:00
parent 101a0acb6b 88683a8d8f
commit ba9c885f52
43 changed files with 3127 additions and 993 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/hosts.rs
+14 -9
View File
@@ -36,8 +36,6 @@ use worker_runtime::config_bundle::{
ConfigBundleMetadata, ConfigBundleProvenance, ConfigProfileDescriptor,
};
use worker_runtime::error::RuntimeError as EmbeddedRuntimeError;
#[cfg(test)]
use worker_runtime::execution::WorkerExecutionRunState;
use worker_runtime::fs_store::FsRuntimeStoreOptions;
use worker_runtime::http_server::{
RUNTIME_PING_PERMISSION, RUNTIME_WORKSPACE_SCOPE_HEADER,
@@ -245,7 +243,10 @@ pub struct WorkerSummary {
#[serde(default)]
pub tags: Vec<String>,
pub workspace: WorkerWorkspaceSummary,
/// Runtime catalog lifecycle compatibility state.
pub state: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub worker_state: Option<protocol::WorkerStateSnapshot>,
pub last_seen_at: Option<String>,
#[serde(default)]
pub pinned: bool,
@@ -337,6 +338,7 @@ pub(crate) fn workspace_worker_summary(
workspace_id: summary.workspace.workspace_id,
},
state: summary.state,
worker_state: summary.worker_state,
last_seen_at: summary.last_seen_at,
pinned: summary.pinned,
retention_state: summary.retention_state,
@@ -2000,6 +2002,7 @@ impl EmbeddedWorkerRuntime {
workspace_id: summary.workspace_id.clone(),
},
state: embedded_worker_status_label(summary.status).to_string(),
worker_state: summary.worker_state.clone(),
last_seen_at: None,
pinned: false,
retention_state: "transient".to_string(),
@@ -2039,6 +2042,7 @@ impl EmbeddedWorkerRuntime {
workspace_id: detail.workspace_id.clone(),
},
state: embedded_worker_status_label(detail.status).to_string(),
worker_state: detail.worker_state.clone(),
last_seen_at: None,
pinned: false,
retention_state: "transient".to_string(),
@@ -3342,6 +3346,7 @@ impl RemoteWorkerRuntime {
workspace_id: summary.workspace_id.clone(),
},
state: embedded_worker_status_label(summary.status).to_string(),
worker_state: summary.worker_state.clone(),
last_seen_at: None,
pinned: false,
retention_state: "transient".to_string(),
@@ -3385,6 +3390,7 @@ impl RemoteWorkerRuntime {
workspace_id: detail.workspace_id.clone(),
},
state: embedded_worker_status_label(detail.status).to_string(),
worker_state: detail.worker_state.clone(),
last_seen_at: None,
pinned: false,
retention_state: "transient".to_string(),
@@ -4732,6 +4738,7 @@ pub fn placeholder_worker(host_id: impl Into<String>) -> WorkerSummary {
workspace_id: None,
},
state: "unsupported".to_string(),
worker_state: None,
last_seen_at: None,
pinned: false,
retention_state: "transient".to_string(),
@@ -5170,7 +5177,6 @@ mod tests {
request.worker_ref,
self.backend_id(),
),
run_state: WorkerExecutionRunState::Idle,
working_directory: request
.working_directory
.as_ref()
@@ -5199,8 +5205,8 @@ mod tests {
let content = input.content;
std::thread::spawn(move || {
std::thread::sleep(std::time::Duration::from_millis(10));
let _ = context.publish_protocol_event(protocol::Event::Status {
status: protocol::WorkerStatus::Running,
let _ = context.publish_protocol_event(protocol::Event::WorkerState {
snapshot: protocol::WorkerStatus::Running.into(),
});
let _ = context.publish_protocol_event(protocol::Event::TextDone {
text: format!("echo: {content}"),
@@ -5208,14 +5214,13 @@ mod tests {
let _ = context.publish_protocol_event(protocol::Event::RunEnd {
result: protocol::RunResult::Finished,
});
let _ = context.publish_protocol_event(protocol::Event::Status {
status: protocol::WorkerStatus::Idle,
let _ = context.publish_protocol_event(protocol::Event::WorkerState {
snapshot: protocol::WorkerStatus::Idle.into(),
});
});
if let Some(submission_request_id) = submission_request_id {
worker_runtime::execution::WorkerExecutionResult::accepted_submission(
worker_runtime::execution::WorkerExecutionOperation::Input,
WorkerExecutionRunState::Busy,
submission_request_id,
uuid::Uuid::now_v7().to_string(),
protocol::SubmissionDisposition::Started,
@@ -5223,7 +5228,6 @@ mod tests {
} else {
worker_runtime::execution::WorkerExecutionResult::accepted(
worker_runtime::execution::WorkerExecutionOperation::Input,
WorkerExecutionRunState::Busy,
)
}
}
@@ -5256,6 +5260,7 @@ mod tests {
workspace_id: None,
},
state: "available".to_string(),
worker_state: None,
last_seen_at: None,
pinned: false,
retention_state: "transient".to_string(),
crates/workspace-server/src/runtime_subscription_tests.rs
+53 -20
View File
@@ -6,7 +6,7 @@ use worker_runtime::catalog::{
};
use worker_runtime::execution::{
WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult,
WorkerExecutionRunState, WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult,
WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult,
};
use worker_runtime::identity::WorkerId;
use worker_runtime::profile_archive::{ProfileSourceArchiveRef, ProfileSourceGraphSummary};
@@ -22,7 +22,6 @@ impl WorkerExecutionBackend for TestExecutionBackend {
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
WorkerExecutionSpawnResult::connected(
WorkerExecutionHandle::new(request.worker_ref, self.backend_id()),
WorkerExecutionRunState::Idle,
None,
)
}
@@ -35,24 +34,17 @@ impl WorkerExecutionBackend for TestExecutionBackend {
if let Some(submission_request_id) = input.submission_request_id {
WorkerExecutionResult::accepted_submission(
WorkerExecutionOperation::Input,
WorkerExecutionRunState::Busy,
submission_request_id,
uuid::Uuid::now_v7().to_string(),
protocol::SubmissionDisposition::Started,
)
} else {
WorkerExecutionResult::accepted(
WorkerExecutionOperation::Input,
WorkerExecutionRunState::Busy,
)
WorkerExecutionResult::accepted(WorkerExecutionOperation::Input)
}
}
fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult {
WorkerExecutionResult::accepted(
WorkerExecutionOperation::Stop,
WorkerExecutionRunState::Stopped,
)
WorkerExecutionResult::accepted(WorkerExecutionOperation::Stop)
}
}
@@ -199,8 +191,8 @@ async fn equal_downstream_selectors_share_one_upstream_subscription() {
runtime
.observe_worker_event(
&worker.worker_ref,
protocol::Event::Status {
status: protocol::WorkerStatus::Running,
protocol::Event::WorkerState {
snapshot: protocol::WorkerStatus::Running.into(),
},
)
.unwrap();
@@ -210,7 +202,16 @@ async fn equal_downstream_selectors_share_one_upstream_subscription() {
BrokerSubscriptionEvent::Event {
payload: SubscriptionEventPayload::WorkerUpserted { ref worker },
..
} if worker.state == SubscriptionWorkerState::Running
} if worker.state == SubscriptionWorkerState::Idle
&& matches!(
worker.worker_state,
Some(protocol::WorkerStateSnapshot {
state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run(
protocol::WorkerRunState::Running
)),
..
})
)
));
}
let mut late = broker.subscribe("runtime-test", selector.clone()).unwrap();
@@ -220,7 +221,18 @@ async fn equal_downstream_selectors_share_one_upstream_subscription() {
assert!(matches!(
snapshot,
SubscriptionSnapshot::Workers { workers }
if workers.iter().any(|worker| worker.state == SubscriptionWorkerState::Running)
if workers.iter().any(|worker| {
worker.state == SubscriptionWorkerState::Idle
&& matches!(
worker.worker_state,
Some(protocol::WorkerStateSnapshot {
state: protocol::WorkerState::Busy(
protocol::WorkerBusyState::Run(protocol::WorkerRunState::Running)
),
..
})
)
})
));
drop(late);
@@ -339,14 +351,24 @@ async fn embedded_runtime_uses_in_process_subscription_source() {
runtime
.observe_worker_event(
&worker.worker_ref,
protocol::Event::Status {
status: protocol::WorkerStatus::Running,
protocol::Event::WorkerState {
snapshot: protocol::WorkerStatus::Running.into(),
},
)
.unwrap();
assert!(matches!(next_event(&mut subscription).await,
BrokerSubscriptionEvent::Event { payload: SubscriptionEventPayload::WorkerUpserted { worker }, .. }
if worker.runtime_id.as_deref() == Some("embedded-worker-runtime") && worker.state == SubscriptionWorkerState::Running));
BrokerSubscriptionEvent::Event { payload: SubscriptionEventPayload::WorkerUpserted { worker }, .. }
if worker.runtime_id.as_deref() == Some("embedded-worker-runtime")
&& worker.state == SubscriptionWorkerState::Idle
&& matches!(
worker.worker_state,
Some(protocol::WorkerStateSnapshot {
state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run(
protocol::WorkerRunState::Running
)),
..
})
)));
let mut late = broker
.subscribe(
"embedded-worker-runtime",
@@ -359,7 +381,18 @@ async fn embedded_runtime_uses_in_process_subscription_source() {
assert!(matches!(
snapshot,
SubscriptionSnapshot::Workers { workers }
if workers.iter().any(|worker| worker.state == SubscriptionWorkerState::Running)
if workers.iter().any(|worker| {
worker.state == SubscriptionWorkerState::Idle
&& matches!(
worker.worker_state,
Some(protocol::WorkerStateSnapshot {
state: protocol::WorkerState::Busy(
protocol::WorkerBusyState::Run(protocol::WorkerRunState::Running)
),
..
})
)
})
));
runtime
crates/workspace-server/src/server.rs
+42 -9
View File
@@ -15766,6 +15766,7 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> WorkerSummary
singleton_key: None,
tags: Vec::new(),
state: "missing".to_string(),
worker_state: None,
last_seen_at: Some(record.updated_at.clone()),
pinned: record.retention_state == "pinned",
retention_state: record.retention_state.clone(),
@@ -19422,7 +19423,6 @@ mod tests {
request.worker_ref,
self.backend_id(),
),
run_state: worker_runtime::execution::WorkerExecutionRunState::Idle,
working_directory,
}
}
@@ -19438,17 +19438,43 @@ mod tests {
.push((handle.worker_ref().clone(), method));
worker_runtime::execution::WorkerExecutionResult::accepted(
worker_runtime::execution::WorkerExecutionOperation::ProtocolMethod,
worker_runtime::execution::WorkerExecutionRunState::Idle,
)
}
fn worker_snapshot(
&self,
handle: &worker_runtime::execution::WorkerExecutionHandle,
) -> Option<protocol::Event> {
Some(protocol::Event::Snapshot {
session: protocol::SessionSnapshot {
pending_submissions: protocol::PendingSubmissionsSnapshot::default(),
entries: Vec::new(),
},
greeting: protocol::Greeting {
worker_name: handle.worker_ref().worker_id.to_string(),
cwd: String::new(),
provider: "deterministic-workspace-server-test".to_string(),
model: "deterministic-workspace-server-test".to_string(),
scope_summary: "test execution snapshot".to_string(),
tools: Vec::new(),
context_window: 0,
context_tokens: 0,
},
state: protocol::WorkerStateSnapshot::initial(1),
in_flight: protocol::InFlightSnapshot {
blocks: Vec::new(),
commands: Vec::new(),
},
internal_workers: Vec::new(),
})
}
fn stop_worker(
&self,
_handle: &worker_runtime::execution::WorkerExecutionHandle,
) -> worker_runtime::execution::WorkerExecutionResult {
worker_runtime::execution::WorkerExecutionResult::accepted(
worker_runtime::execution::WorkerExecutionOperation::Stop,
worker_runtime::execution::WorkerExecutionRunState::Stopped,
)
}
@@ -19458,7 +19484,6 @@ mod tests {
) -> worker_runtime::execution::WorkerExecutionResult {
worker_runtime::execution::WorkerExecutionResult::accepted(
worker_runtime::execution::WorkerExecutionOperation::Cancel,
worker_runtime::execution::WorkerExecutionRunState::Stopped,
)
}
@@ -19495,16 +19520,16 @@ mod tests {
if let Some(submission_request_id) = submission_request_id {
worker_runtime::execution::WorkerExecutionResult::accepted_submission(
worker_runtime::execution::WorkerExecutionOperation::Input,
worker_runtime::execution::WorkerExecutionRunState::Idle,
submission_request_id,
uuid::Uuid::now_v7().to_string(),
protocol::SubmissionDisposition::Started,
)
.with_worker_state(protocol::WorkerStateSnapshot::initial(1))
} else {
worker_runtime::execution::WorkerExecutionResult::accepted(
worker_runtime::execution::WorkerExecutionOperation::Input,
worker_runtime::execution::WorkerExecutionRunState::Idle,
)
.with_worker_state(protocol::WorkerStateSnapshot::initial(1))
}
}
}
@@ -25570,6 +25595,7 @@ mod tests {
workspace_id: Some(TEST_WORKSPACE_ID.to_string()),
},
state: "idle".to_string(),
worker_state: None,
last_seen_at: None,
pinned: false,
retention_state: "normal".to_string(),
@@ -25666,6 +25692,7 @@ mod tests {
workspace_id: Some(TEST_WORKSPACE_ID.to_string()),
},
state: "idle".to_string(),
worker_state: None,
last_seen_at: None,
pinned: false,
retention_state: "normal".to_string(),
@@ -28556,7 +28583,13 @@ mod tests {
protocol::subscription::SubscriptionFramePayload::WorkerProtocol(
protocol::subscription::SubscriptionWorkerProtocolMethod {
subscription_id: second_protocol_subscription_id,
method: protocol::Method::Resume,
method: protocol::Method::Resume {
command: protocol::WorkerCommandEnvelope {
command_id: 1,
expected_execution_generation: 1,
expected_worker_state_revision: 0,
},
},
},
),
);
@@ -28573,7 +28606,7 @@ mod tests {
.iter()
.any(|(worker_ref, method)| {
worker_ref.worker_id.to_string() == worker_id
&& matches!(method, protocol::Method::Resume)
&& matches!(method, protocol::Method::Resume { .. })
})
{
break;
@@ -28586,7 +28619,7 @@ mod tests {
let protocol_methods = execution_backend.protocol_methods();
assert!(protocol_methods.iter().any(|(worker_ref, method)| {
worker_ref.worker_id.to_string() == worker_id
&& matches!(method, protocol::Method::Resume)
&& matches!(method, protocol::Method::Resume { .. })
}));
server.abort();
}