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

runtime: route notifications through worker inbox

Browse Source
This commit is contained in:
Hare
2026-08-05 01:44:57 +09:00
parent dd2ca54874
commit f98a123e40
15 changed files with 614 additions and 811 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/interaction.rs
+23 -3
View File
@@ -8,7 +8,7 @@ use serde::{Deserialize, Serialize};
#[serde(rename_all = "snake_case")]
pub enum WorkerInputKind {
User,
System,
Notify,
Compact,
ListRewindTargets,
RegisterPeer,
@@ -38,15 +38,35 @@ impl WorkerInput {
}
}
pub fn system(content: impl Into<String>) -> Self {
pub fn notify(content: impl Into<String>) -> Self {
Self {
kind: WorkerInputKind::System,
kind: WorkerInputKind::Notify,
content: content.into(),
segments: None,
}
}
}
#[cfg(test)]
mod tests {
use super::WorkerInput;
#[test]
fn notify_is_an_operation_and_legacy_system_kind_is_rejected() {
assert_eq!(
serde_json::to_value(WorkerInput::notify("message")).unwrap(),
serde_json::json!({ "kind": "notify", "content": "message" })
);
assert!(
serde_json::from_value::<WorkerInput>(serde_json::json!({
"kind": "system",
"content": "message"
}))
.is_err()
);
}
}
/// Acknowledgement returned after input is accepted into the Worker.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerInteractionAck {
crates/worker-runtime/src/runtime.rs
+5 -5
View File
@@ -2392,9 +2392,9 @@ fn input_protocol_event(input: &WorkerInput) -> protocol::Event {
}]
}),
},
WorkerInputKind::System => protocol::Event::SystemItem {
WorkerInputKind::Notify => protocol::Event::SystemItem {
item: serde_json::json!({
"kind": "embedded_worker_system_input",
"kind": "embedded_worker_notification",
"content": input.content.clone(),
}),
},
@@ -3148,7 +3148,7 @@ mod tests {
fn create_worker_rejects_system_initial_input_without_persisting_worker() {
let runtime = runtime_with_backend();
let mut request = task_request("system initial input");
request.initial_input = Some(WorkerInput::system("role/system belongs in config bundle"));
request.initial_input = Some(WorkerInput::notify("role/system belongs in config bundle"));
let error = runtime.create_worker(request).unwrap_err();
assert!(matches!(
@@ -3390,7 +3390,7 @@ mod tests {
.send_input(&detail.worker_ref, WorkerInput::user("hello"))
.unwrap();
runtime
.send_input(&detail.worker_ref, WorkerInput::system("note"))
.send_input(&detail.worker_ref, WorkerInput::notify("note"))
.unwrap();
let observations = runtime
@@ -3550,7 +3550,7 @@ mod tests {
.send_input(&worker.worker_ref, WorkerInput::user("first"))
.unwrap();
runtime
.send_input(&worker.worker_ref, WorkerInput::system("second"))
.send_input(&worker.worker_ref, WorkerInput::notify("second"))
.unwrap();
runtime
.stop_worker(&worker.worker_ref, Some("finished".to_string()))
crates/worker-runtime/src/worker_backend.rs
+76 -4
View File
@@ -793,6 +793,14 @@ fn method_starts_turn(method: &Method) -> bool {
)
}
fn accepted_notify_run_state(status: WorkerStatus, auto_run: bool) -> WorkerExecutionRunState {
match status {
WorkerStatus::Running => WorkerExecutionRunState::Busy,
WorkerStatus::Idle if auto_run => WorkerExecutionRunState::Busy,
WorkerStatus::Idle | WorkerStatus::Paused => WorkerExecutionRunState::Idle,
}
}
fn accepted_run_state_for_method(method: &Method) -> WorkerExecutionRunState {
match method {
Method::Run { .. }
@@ -1098,6 +1106,29 @@ where
}
};
if input.kind == WorkerInputKind::Notify {
let status = worker.shared_state.get_status();
let accepted_run_state = accepted_notify_run_state(status, true);
let claimed_here = status == WorkerStatus::Idle
&& busy
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
.is_ok();
let result = self.send_method(
WorkerExecutionOperation::Input,
worker,
Method::Notify {
message: input.content,
auto_run: true,
},
accepted_run_state,
);
if claimed_here && result.outcome != crate::execution::WorkerExecutionOutcome::Accepted
{
busy.store(false, Ordering::SeqCst);
}
return result;
}
if worker.shared_state.get_status() != WorkerStatus::Idle
|| busy
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
@@ -1115,10 +1146,9 @@ where
.segments
.unwrap_or_else(|| vec![Segment::text(input.content.trim().to_string())]),
},
WorkerInputKind::System => Method::Notify {
message: input.content,
auto_run: true,
},
WorkerInputKind::Notify => {
unreachable!("Notify input is dispatched before the turn-start busy guard")
}
WorkerInputKind::Compact => Method::Compact,
WorkerInputKind::ListRewindTargets => Method::ListRewindTargets,
WorkerInputKind::RegisterPeer => Method::RegisterPeer {
@@ -1159,6 +1189,28 @@ where
}
};
if let Method::Notify { auto_run, .. } = &method {
let auto_run = *auto_run;
let status = worker.shared_state.get_status();
let accepted_run_state = accepted_notify_run_state(status, auto_run);
let claimed_here = status == WorkerStatus::Idle
&& auto_run
&& busy
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
.is_ok();
let result = self.send_method(
WorkerExecutionOperation::ProtocolMethod,
worker,
method,
accepted_run_state,
);
if claimed_here && result.outcome != crate::execution::WorkerExecutionOutcome::Accepted
{
busy.store(false, Ordering::SeqCst);
}
return result;
}
let starts_turn = method_starts_turn(&method);
if starts_turn
&& (worker.shared_state.get_status() != WorkerStatus::Idle
@@ -1298,6 +1350,26 @@ mod tests {
use manifest::{Scope, WorkerManifest};
use session_store::WorkerMetadataStore;
#[test]
fn notify_run_state_allows_running_worker_inbox_delivery() {
assert_eq!(
accepted_notify_run_state(WorkerStatus::Running, true),
WorkerExecutionRunState::Busy
);
assert_eq!(
accepted_notify_run_state(WorkerStatus::Idle, true),
WorkerExecutionRunState::Busy
);
assert_eq!(
accepted_notify_run_state(WorkerStatus::Idle, false),
WorkerExecutionRunState::Idle
);
assert_eq!(
accepted_notify_run_state(WorkerStatus::Paused, true),
WorkerExecutionRunState::Idle
);
}
#[derive(Clone)]
struct MockClient {
responses: Arc<Vec<Vec<LlmEvent>>>,
crates/worker/src/controller.rs
+51 -10
View File
@@ -285,6 +285,7 @@ impl WorkerController {
worker.push_notify(
"Restored Worker state contained unreachable delegated child Workers; their delegated write scopes were reclaimed before resume."
.to_string(),
false,
);
}
@@ -883,7 +884,7 @@ async fn controller_loop<C, St>(
)
.await;
let parent_originated = run.is_parent_originated();
let (new_status, shutdown) = match run {
let (mut new_status, shutdown) = match run {
PendingRun::Run(input) => {
drive_turn(
worker.run(input),
@@ -930,6 +931,11 @@ async fn controller_loop<C, St>(
.await
}
};
if !shutdown && new_status == WorkerStatus::Idle && notify_buffer.has_auto_run_pending()
{
pending = Some(PendingRun::RunForNotification(protocol::InvokeKind::Notify));
new_status = WorkerStatus::Running;
}
finish_controller_run(
&mut worker,
&shared_state,
@@ -985,13 +991,13 @@ async fn controller_loop<C, St>(
// `LogEntry::SystemItem` entry — drained out of the
// notify buffer + broadcast through the sink. No
// separate echo here.
worker.push_notify(message);
// RUNNING / Paused: the buffer push is the entire
// operation; an in-flight turn (or the next
// Resume/Run) will drain it at its next
// pending_history_appends. IDLE: only `auto_run`
// notifications stage RunForNotification; weak progress
// notices stay queued until an explicit run/resume.
worker.push_notify(message, auto_run);
// RUNNING: the in-flight turn drains the buffer at its next
// pending_history_appends; if an auto-run notification remains
// at turn end, the Controller stages a follow-up notification
// turn. Paused notifications remain queued until Resume/Run.
// IDLE: `auto_run` notifications stage RunForNotification;
// weak progress notices stay queued until an explicit run.
if should_auto_run_notification(shared_state.get_status(), auto_run) {
pending = Some(PendingRun::RunForNotification(protocol::InvokeKind::Notify));
}
@@ -1385,11 +1391,11 @@ where
.into(),
});
}
Some(Method::Notify { message, .. }) => {
Some(Method::Notify { message, auto_run }) => {
// Live echo arrives via `Event::SystemItem` once
// the in-flight turn's next `pending_history_appends`
// drains this entry through the interceptor.
notify_buffer.push_notify(message);
notify_buffer.push_notify(message, auto_run);
}
Some(Method::ListCompletions { .. }) => {}
Some(Method::ListWorkers | Method::RestoreWorker { .. } | Method::RegisterPeer { .. }) => {
@@ -1904,6 +1910,41 @@ mod tests {
assert_eq!(env.notify_buffer.len(), 1);
}
#[tokio::test]
async fn running_auto_run_notify_remains_staged_for_followup_turn() {
let mut env = make_env().await;
env._method_tx
.send(Method::Notify {
message: "continue".into(),
auto_run: true,
})
.await
.expect("send notify");
let worker_future = async {
tokio::time::sleep(Duration::from_millis(50)).await;
Ok::<_, WorkerError>(WorkerRunResult::Finished)
};
let (status, shutdown) = drive_turn(
worker_future,
&mut env.method_rx,
&env.event_tx,
&env.cancel_tx,
&env.shared_state,
&env.notify_buffer,
Some(&env.parent_socket_path),
"parent",
&env.spawned_registry,
false,
)
.await;
assert_eq!(status, WorkerStatus::Idle);
assert!(!shutdown);
assert_eq!(env.notify_buffer.len(), 1);
assert!(env.notify_buffer.has_auto_run_pending());
}
#[tokio::test]
async fn compact_method_is_rejected_while_running() {
let mut env = make_env().await;
crates/worker/src/ipc/interceptor.rs
+6 -4
View File
@@ -220,7 +220,9 @@ impl Interceptor for WorkerInterceptor {
// simply be skipped from the SystemItem batch.
warn!(error = %e, "failed to render notify_wrapper; using raw message");
let fallback = match &entry {
super::notify_buffer::PendingNotify::Notify { message } => message.clone(),
super::notify_buffer::PendingNotify::Notify { message, .. } => {
message.clone()
}
super::notify_buffer::PendingNotify::WorkerEvent { event } => {
session_store::render_worker_event(event)
}
@@ -1019,8 +1021,8 @@ mod tests {
async fn pending_history_appends_drains_buffer_into_items() {
let registry = Arc::new(HookRegistryBuilder::new().build());
let buffer = NotifyBuffer::new();
buffer.push_notify("first".into());
buffer.push_notify("second".into());
buffer.push_notify("first".into(), false);
buffer.push_notify("second".into(), false);
let interceptor = WorkerInterceptor::new(
registry,
@@ -1057,7 +1059,7 @@ mod tests {
// anything itself.
let registry = Arc::new(HookRegistryBuilder::new().build());
let buffer = NotifyBuffer::new();
buffer.push_notify("msg".into());
buffer.push_notify("msg".into(), false);
let interceptor = WorkerInterceptor::new(
registry,
crates/worker/src/ipc/notify_buffer.rs
+22 -9
View File
@@ -41,7 +41,7 @@ const CAPACITY: usize = 128;
/// is available.
#[derive(Debug, Clone)]
pub enum PendingNotify {
Notify { message: String },
Notify { message: String, auto_run: bool },
WorkerEvent { event: WorkerEvent },
}
@@ -61,8 +61,8 @@ impl NotifyBuffer {
/// Push a notify entry onto the queue. If the queue is full, the
/// oldest entry is dropped and a `tracing::warn` is emitted — the
/// caller should never hit this in normal operation.
pub fn push_notify(&self, message: String) {
self.push_entry(PendingNotify::Notify { message });
pub fn push_notify(&self, message: String, auto_run: bool) {
self.push_entry(PendingNotify::Notify { message, auto_run });
}
/// Push a typed worker-event entry onto the queue.
@@ -89,6 +89,15 @@ impl NotifyBuffer {
q.drain(..).collect()
}
/// Whether an undrained `Method::Notify { auto_run: true }` remains.
pub fn has_auto_run_pending(&self) -> bool {
self.inner
.lock()
.expect("notify buffer poisoned")
.iter()
.any(|entry| matches!(entry, PendingNotify::Notify { auto_run: true, .. }))
}
/// Number of pending entries. Primarily for tests.
pub fn len(&self) -> usize {
self.inner.lock().expect("notify buffer poisoned").len()
@@ -107,7 +116,7 @@ pub(crate) fn build_system_item(
prompts: &PromptCatalog,
) -> Result<SystemItem, CatalogError> {
match entry {
PendingNotify::Notify { message } => {
PendingNotify::Notify { message, .. } => {
let body = prompts.notify_wrapper(message)?;
Ok(SystemItem::Notification {
message: message.clone(),
@@ -132,12 +141,15 @@ mod tests {
#[test]
fn push_then_drain_preserves_order() {
let buf = NotifyBuffer::new();
buf.push_notify("one".into());
buf.push_notify("two".into());
buf.push_notify("one".into(), false);
assert!(!buf.has_auto_run_pending());
buf.push_notify("two".into(), true);
assert!(buf.has_auto_run_pending());
let drained = buf.drain();
assert!(!buf.has_auto_run_pending());
assert_eq!(drained.len(), 2);
match &drained[0] {
PendingNotify::Notify { message } => assert_eq!(message, "one"),
PendingNotify::Notify { message, .. } => assert_eq!(message, "one"),
other => panic!("unexpected: {other:?}"),
}
assert!(buf.is_empty());
@@ -147,12 +159,12 @@ mod tests {
fn capacity_drops_oldest() {
let buf = NotifyBuffer::new();
for i in 0..(CAPACITY + 5) {
buf.push_notify(format!("msg{i}"));
buf.push_notify(format!("msg{i}"), false);
}
let drained = buf.drain();
assert_eq!(drained.len(), CAPACITY);
match &drained[0] {
PendingNotify::Notify { message } => assert_eq!(message, "msg5"),
PendingNotify::Notify { message, .. } => assert_eq!(message, "msg5"),
other => panic!("unexpected: {other:?}"),
}
}
@@ -161,6 +173,7 @@ mod tests {
fn build_system_item_for_notify_carries_wrapper_body() {
let entry = PendingNotify::Notify {
message: "hello".into(),
auto_run: false,
};
let catalog = PromptCatalog::builtins_only().unwrap();
let item = build_system_item(&entry, &catalog).unwrap();
crates/worker/src/worker.rs
+3 -2
View File
@@ -1597,8 +1597,8 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
/// `Item::system_message` just before the next LLM request, via
/// `WorkerInterceptor::pending_history_appends`. See [`NotifyBuffer`]
/// for overflow behaviour and the lane-of-record rationale.
pub fn push_notify(&self, message: String) {
self.pending_notifies.push_notify(message);
pub fn push_notify(&self, message: String, auto_run: bool) {
self.pending_notifies.push_notify(message, auto_run);
}
/// Push an agent-visible typed `WorkerEvent` entry onto the pending buffer.
@@ -4364,6 +4364,7 @@ where
self.push_notify(
"Restored Worker state contained missing or unreachable delegated child Workers; their delegated write scopes were reclaimed before resume."
.to_string(),
false,
);
Ok(())
}
crates/workspace-server/src/hosts.rs
+4 -4
View File
@@ -496,7 +496,7 @@ pub struct WorkerLifecycleResult {
#[serde(rename_all = "snake_case")]
pub enum WorkerInputKind {
User,
System,
Notify,
Compact,
ListRewindTargets,
RegisterPeer,
@@ -2115,7 +2115,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
let input = EmbeddedWorkerInput {
kind: match request.kind {
WorkerInputKind::User => EmbeddedWorkerInputKind::User,
WorkerInputKind::System => EmbeddedWorkerInputKind::System,
WorkerInputKind::Notify => EmbeddedWorkerInputKind::Notify,
WorkerInputKind::Compact => EmbeddedWorkerInputKind::Compact,
WorkerInputKind::ListRewindTargets => EmbeddedWorkerInputKind::ListRewindTargets,
WorkerInputKind::RegisterPeer => EmbeddedWorkerInputKind::RegisterPeer,
@@ -3086,7 +3086,7 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
let input = EmbeddedWorkerInput {
kind: match request.kind {
WorkerInputKind::User => EmbeddedWorkerInputKind::User,
WorkerInputKind::System => EmbeddedWorkerInputKind::System,
WorkerInputKind::Notify => EmbeddedWorkerInputKind::Notify,
WorkerInputKind::Compact => EmbeddedWorkerInputKind::Compact,
WorkerInputKind::ListRewindTargets => EmbeddedWorkerInputKind::ListRewindTargets,
WorkerInputKind::RegisterPeer => EmbeddedWorkerInputKind::RegisterPeer,
@@ -4538,7 +4538,7 @@ mod tests {
.expect("test backend should connect");
let mut request = embedded_spawn_request();
request.initial_input = Some(EmbeddedWorkerInput {
kind: EmbeddedWorkerInputKind::System,
kind: EmbeddedWorkerInputKind::Notify,
content: "system/role instruction belongs in profile".to_string(),
segments: None,
});
crates/workspace-server/src/server.rs
+361 -342
View File
@@ -1,7 +1,7 @@
use std::collections::{BTreeMap, HashMap, HashSet};
use std::path::{Component, Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade};
use axum::extract::{Path as AxumPath, Query, State};
@@ -17,7 +17,6 @@ use memory::backend::{
MemoryConsolidationOutput,
};
use protocol::stream::{decode_method, encode_event};
use rusqlite::OptionalExtension;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use ticket::{
@@ -240,6 +239,12 @@ impl ServerConfig {
}
}
const ORCHESTRATOR_ATTENTION_TICKET_LIMIT: usize = 20;
const ORCHESTRATOR_ATTENTION_PROMPT: &str = include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../resources/prompts/internal/workspace_orchestrator_queue_attention.md"
));
#[derive(Clone)]
pub struct WorkspaceApi {
config: ServerConfig,
@@ -248,6 +253,7 @@ pub struct WorkspaceApi {
runtime: Arc<RuntimeRegistry>,
companion: Arc<CompanionConsole>,
orchestrator_spawn_lock: Arc<std::sync::Mutex<()>>,
orchestrator_attention_fingerprint: Arc<Mutex<Option<String>>>,
observation_proxy: BackendObservationProxy,
runtime_subscription_broker: RuntimeSubscriptionBroker,
resource_broker: BackendResourceBroker,
@@ -348,6 +354,7 @@ impl WorkspaceApi {
runtime,
companion,
orchestrator_spawn_lock: Arc::new(std::sync::Mutex::new(())),
orchestrator_attention_fingerprint: Arc::new(Mutex::new(None)),
observation_proxy,
runtime_subscription_broker,
resource_broker,
@@ -963,21 +970,9 @@ pub async fn serve(
listener: TcpListener,
) -> Result<()> {
let api = WorkspaceApi::new(config, store).await?;
let dispatcher_api = api.clone();
let dispatcher_workspace_id = dispatcher_api.config.workspace_id.clone();
let dispatcher = tokio::spawn(async move {
loop {
let api = dispatcher_api.clone();
let workspace_id = dispatcher_workspace_id.clone();
let _ = tokio::task::spawn_blocking(move || {
dispatch_pending_ticket_notifications(&api, &workspace_id)
})
.await;
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
}
});
let orchestrator_hook = tokio::spawn(run_orchestrator_turn_end_hook(api.clone()));
let result = axum::serve(listener, build_router(api)).await;
dispatcher.abort();
orchestrator_hook.abort();
result?;
Ok(())
}
@@ -2169,7 +2164,27 @@ async fn scoped_queue_ticket(
browser_ticket_backend(&api)?
.queue_ready(TicketIdOrSlug::Id(path.id.clone()), queued_by)
.map_err(Error::from)?;
browser_ticket_detail(&api, &path.id)
let Json(ticket) = browser_ticket_detail(&api, &path.id)?;
notify_ticket_recipients(
&api,
&path.workspace_id,
&path.id,
ticket
.events
.last()
.map(|event| event.sequence as i64)
.unwrap_or_default(),
ticket
.events
.last()
.map(|event| event.kind.as_str())
.unwrap_or("state_changed"),
"queue_ready",
TicketWorkflowState::Ready.as_str(),
ticket.state.as_str(),
None,
);
Ok(Json(ticket))
}
async fn scoped_close_ticket(
@@ -2204,44 +2219,36 @@ async fn execute_worker_ticket_rest_operation(
let operation_kind = ticket_mutation_operation_kind(&operation);
let is_mutation = operation_kind != "read";
let target = ticket_mutation_target(&operation).cloned();
let read_target = ticket_read_target(&operation).cloned();
let source = authenticate_worker_mutation_source(api, workspace_id, &headers)?;
let before = target.as_ref().and_then(|id| backend.show(id.clone()).ok());
let previous_state = before
.as_ref()
.map(|ticket| ticket.meta.workflow_state.as_str().to_string())
.unwrap_or_else(|| ticket_operation_initial_state(&operation));
bind_worker_ticket_operation_source(&source, &mut operation);
let source_context = worker_ticket_source_context(api, workspace_id, &source, before.as_ref());
backend = backend
.with_event_attributes(source_context.attributes(operation_kind))
.with_mutation_hook(build_ticket_notification_hook(
api,
source_context,
operation_kind,
before
.as_ref()
.map(|ticket| ticket.meta.workflow_state.as_str().to_string())
.unwrap_or_else(|| ticket_operation_initial_state(&operation)),
));
backend = backend.with_event_attributes(source_context.attributes(operation_kind));
let result = execute_ticket_backend_operation(&backend, operation).map_err(Error::from)?;
if let Some(read_target) = read_target.as_ref()
&& let Ok(ticket) = backend.show(read_target.clone())
&& let Some(event_index) = ticket.events.last().and_then(|event| {
event
.attributes
.get("event_sequence")
.and_then(|value| value.parse::<i64>().ok())
})
if is_mutation
&& let Some(target) = target
&& let Ok(ticket) = backend.show(target)
{
api.store.upsert_ticket_notification_cursor(
let event = ticket.events.last();
notify_ticket_recipients(
api,
workspace_id,
&ticket.meta.id,
&source.runtime_id,
&source.worker_id,
event_index,
&Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
)?;
}
if is_mutation {
dispatch_pending_ticket_notifications(api, workspace_id);
event
.and_then(|event| event.attributes.get("event_sequence"))
.and_then(|value| value.parse::<i64>().ok())
.unwrap_or(ticket.events.len() as i64),
event.map(|event| event.kind.as_str()).unwrap_or("mutation"),
operation_kind,
&previous_state,
ticket.meta.workflow_state.as_str(),
Some((source.runtime_id, source.worker_id)),
);
}
Ok(result)
}
@@ -2791,13 +2798,6 @@ fn bind_worker_ticket_operation_source(
}
}
fn ticket_read_target(operation: &TicketBackendOperation) -> Option<&TicketIdOrSlug> {
match operation {
TicketBackendOperation::Show { id } => Some(id),
_ => None,
}
}
fn ticket_mutation_operation_kind(operation: &TicketBackendOperation) -> &'static str {
match operation {
TicketBackendOperation::Create { .. } => "create",
@@ -2830,12 +2830,10 @@ fn ticket_operation_initial_state(operation: &TicketBackendOperation) -> String
#[derive(Debug, Clone)]
struct WorkerTicketSourceContext {
workspace_id: String,
runtime_id: String,
worker_id: String,
actor_role: String,
assignment_id: Option<String>,
orchestrator: Option<(String, String)>,
}
impl WorkerTicketSourceContext {
@@ -2887,7 +2885,6 @@ fn worker_ticket_source_context(
});
let actor_role = worker_source_actor_role(is_current_assignment, is_orchestrator);
WorkerTicketSourceContext {
workspace_id: workspace_id.to_string(),
runtime_id: source.runtime_id.clone(),
worker_id: source.worker_id.clone(),
actor_role: actor_role.to_string(),
@@ -2895,94 +2892,63 @@ fn worker_ticket_source_context(
(assignment.runtime_id == source.runtime_id && assignment.worker_id == source.worker_id)
.then_some(assignment.assignment_id)
}),
orchestrator: orchestrator.map(|worker| (worker.runtime_id, worker.worker_id)),
}
}
fn build_ticket_notification_hook(
_api: &WorkspaceApi,
source: WorkerTicketSourceContext,
operation_kind: &'static str,
previous_state: String,
) -> Arc<ticket::SqliteTicketMutationHook> {
let invoked = AtomicBool::new(false);
let notification_id = new_id("tnfy");
Arc::new(move |conn, event| {
if invoked.swap(true, Ordering::SeqCst) {
return Ok(());
}
let current_state: String = conn
.query_row(
"SELECT workflow_state FROM typed_tickets WHERE workspace_id = ?1 AND ticket_id = ?2",
rusqlite::params![source.workspace_id, event.ticket_id],
|row| row.get(0),
)
.map_err(|error| ticket::TicketError::Conflict(format!("read committed Ticket state for outbox: {error}")))?;
conn.execute(
r#"INSERT INTO ticket_notification_outbox (
notification_id, workspace_id, ticket_id, event_sequence,
source_runtime_id, source_worker_id, previous_state, current_state, created_at,
event_kind, source_operation_kind, source_actor_role, source_assignment_id
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)"#,
rusqlite::params![
notification_id,
source.workspace_id,
event.ticket_id,
event.event_index,
source.runtime_id,
source.worker_id,
previous_state,
current_state,
Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
event.event_kind.as_str(),
operation_kind,
source.actor_role,
source.assignment_id,
],
)
.map_err(|error| {
ticket::TicketError::Conflict(format!("insert Ticket notification outbox: {error}"))
})?;
fn notify_ticket_recipients(
api: &WorkspaceApi,
workspace_id: &str,
ticket_id: &str,
event_sequence: i64,
event_kind: &str,
source_operation_kind: &str,
previous_state: &str,
current_state: &str,
source: Option<(String, String)>,
) {
let mut recipients = Vec::new();
if let Some(assignment) = api
.store
.get_current_ticket_worker_assignment(workspace_id, ticket_id)
.ok()
.flatten()
{
recipients.push((assignment.runtime_id, assignment.worker_id));
}
if (matches!(previous_state, "queued" | "inprogress")
|| matches!(current_state, "queued" | "inprogress"))
&& let Some(orchestrator) = find_workspace_orchestrator(api)
{
recipients.push((orchestrator.runtime_id, orchestrator.worker_id));
}
recipients.sort();
recipients.dedup();
let assigned: Option<(String, String)> = conn
.query_row(
r#"SELECT runtime_id, worker_id FROM ticket_current_worker_assignments
WHERE workspace_id = ?1 AND ticket_id = ?2"#,
rusqlite::params![source.workspace_id, event.ticket_id],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()
.map_err(|error| {
ticket::TicketError::Conflict(format!(
"resolve assigned notification recipient: {error}"
))
})?;
let mut recipients = Vec::new();
if let Some((runtime_id, worker_id)) = assigned {
if runtime_id != source.runtime_id || worker_id != source.worker_id {
recipients.push((runtime_id, worker_id, "assigned"));
}
}
if (matches!(previous_state.as_str(), "queued" | "inprogress")
|| matches!(current_state.as_str(), "queued" | "inprogress"))
&& let Some((runtime_id, worker_id)) = &source.orchestrator
&& (*runtime_id != source.runtime_id || *worker_id != source.worker_id)
let source_fields = source
.as_ref()
.map(|(runtime_id, worker_id)| {
format!(" source_runtime_id={runtime_id} source_worker_id={worker_id}")
})
.unwrap_or_default();
for (runtime_id, worker_id) in recipients {
if source
.as_ref()
.is_some_and(|source| source.0 == runtime_id && source.1 == worker_id)
{
recipients.push((runtime_id.clone(), worker_id.clone(), "orchestrator"));
continue;
}
recipients.sort();
recipients.dedup_by(|left, right| left.0 == right.0 && left.1 == right.1);
for (runtime_id, worker_id, recipient_kind) in recipients {
conn.execute(
r#"INSERT OR IGNORE INTO ticket_notification_deliveries (
notification_id, recipient_runtime_id, recipient_worker_id, recipient_kind, attempts
) VALUES (?1, ?2, ?3, ?4, 0)"#,
rusqlite::params![notification_id, runtime_id, worker_id, recipient_kind],
)
.map_err(|error| ticket::TicketError::Conflict(format!("insert Ticket notification delivery: {error}")))?;
}
Ok(())
})
let _ = api.runtime.send_input(
&runtime_id,
&worker_id,
WorkerInputRequest {
kind: WorkerInputKind::Notify,
content: format!(
"Ticket notification: workspace_id={workspace_id} ticket_id={ticket_id} event_sequence={event_sequence} event_kind={event_kind} source_operation_kind={source_operation_kind}.{source_fields} Reread the Ticket before acting.",
),
segments: None,
},
);
}
}
fn authenticate_worker_mutation_source(
@@ -3011,109 +2977,186 @@ fn authenticate_worker_mutation_source(
})
}
fn dispatch_pending_ticket_notifications(api: &WorkspaceApi, workspace_id: &str) {
let Ok(deliveries) = api
.store
.list_pending_ticket_notification_deliveries(workspace_id, 100)
else {
async fn run_orchestrator_turn_end_hook(api: WorkspaceApi) {
let Ok(mut subscription) = api.runtime_subscription_broker.subscribe(
EMBEDDED_WORKER_RUNTIME_ID,
protocol::subscription::EventSubscriptionSelector::RuntimeWorkers,
) else {
return;
};
for delivery in deliveries {
let Some((current_runtime_id, current_worker_id)) =
current_ticket_notification_recipient(api, &delivery)
else {
continue;
};
if current_runtime_id == delivery.source_runtime_id
&& current_worker_id == delivery.source_worker_id
{
let _ = api.store.mark_ticket_notification_delivered(
&delivery.notification_id,
&delivery.recipient_runtime_id,
&delivery.recipient_worker_id,
&Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
);
continue;
}
if current_runtime_id != delivery.recipient_runtime_id
|| current_worker_id != delivery.recipient_worker_id
{
let _ = api.store.reroute_ticket_notification_delivery(
&delivery.notification_id,
&delivery.recipient_runtime_id,
&delivery.recipient_worker_id,
&current_runtime_id,
&current_worker_id,
);
continue;
}
if api
.store
.get_ticket_notification_cursor(
&delivery.workspace_id,
&delivery.ticket_id,
&current_runtime_id,
&current_worker_id,
)
.ok()
.flatten()
.is_some_and(|cursor| cursor >= delivery.event_sequence)
{
let _ = api.store.mark_ticket_notification_delivered(
&delivery.notification_id,
&delivery.recipient_runtime_id,
&delivery.recipient_worker_id,
&Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
);
continue;
}
let result = api.runtime.send_input(
&delivery.recipient_runtime_id,
&delivery.recipient_worker_id,
WorkerInputRequest {
kind: WorkerInputKind::System,
content: format!(
"Ticket notification: workspace_id={} ticket_id={} event_sequence={} event_kind={} source_operation_kind={} source_runtime_id={} source_worker_id={}. Reread the Ticket before acting.",
delivery.workspace_id,
delivery.ticket_id,
delivery.event_sequence,
delivery.event_kind,
delivery.source_operation_kind,
delivery.source_runtime_id,
delivery.source_worker_id
),
segments: None,
},
);
match result {
Ok(result) if result.state == WorkerOperationState::Accepted => {
let _ = api.store.mark_ticket_notification_delivered(
&delivery.notification_id,
&delivery.recipient_runtime_id,
&delivery.recipient_worker_id,
&Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
);
let mut worker_states = HashMap::new();
while let Some(update) = subscription.recv().await {
match update {
crate::runtime_subscription::BrokerSubscriptionEvent::Snapshot { snapshot, .. } => {
if let protocol::subscription::SubscriptionSnapshot::Workers { workers } = snapshot
{
worker_states.clear();
for worker in workers {
let worker_id = worker.worker_id.to_string();
maybe_dispatch_orchestrator_turn_end(&api, &worker_id, None, worker.state);
worker_states.insert(worker_id, worker.state);
}
}
}
Ok(result) => {
let _ = api.store.mark_ticket_notification_failed(
&delivery.notification_id,
&delivery.recipient_runtime_id,
&delivery.recipient_worker_id,
&format!("Runtime rejected notification: {:?}", result.diagnostics),
);
crate::runtime_subscription::BrokerSubscriptionEvent::Event { payload, .. } => {
match payload {
protocol::subscription::SubscriptionEventPayload::WorkerUpserted { worker } => {
let worker_id = worker.worker_id.to_string();
let previous = worker_states.insert(worker_id.clone(), worker.state);
maybe_dispatch_orchestrator_turn_end(
&api,
&worker_id,
previous,
worker.state,
);
}
protocol::subscription::SubscriptionEventPayload::WorkerRemoved {
worker_id,
..
} => {
worker_states.remove(worker_id.as_str());
}
_ => {}
}
}
Err(error) => {
let _ = api.store.mark_ticket_notification_failed(
&delivery.notification_id,
&delivery.recipient_runtime_id,
&delivery.recipient_worker_id,
&error.into_error().to_string(),
);
crate::runtime_subscription::BrokerSubscriptionEvent::Disconnected { .. } => {
worker_states.clear();
}
crate::runtime_subscription::BrokerSubscriptionEvent::Rejected { .. }
| crate::runtime_subscription::BrokerSubscriptionEvent::Closed { .. } => return,
}
}
}
fn maybe_dispatch_orchestrator_turn_end(
api: &WorkspaceApi,
worker_id: &str,
previous: Option<protocol::subscription::SubscriptionWorkerState>,
current: protocol::subscription::SubscriptionWorkerState,
) {
use protocol::subscription::SubscriptionWorkerState;
if current != SubscriptionWorkerState::Idle
|| !matches!(
previous,
None | Some(SubscriptionWorkerState::Running)
| Some(SubscriptionWorkerState::Stopped)
| Some(SubscriptionWorkerState::Paused)
)
{
return;
}
let Some(orchestrator) = find_workspace_orchestrator(api) else {
return;
};
if orchestrator.runtime_id == EMBEDDED_WORKER_RUNTIME_ID && orchestrator.worker_id == worker_id
{
dispatch_orchestrator_queue_attention(api);
}
}
fn dispatch_orchestrator_queue_attention(api: &WorkspaceApi) {
let Some(orchestrator) = find_workspace_orchestrator(api) else {
return;
};
let Ok(backend) = browser_ticket_backend(api) else {
return;
};
let Ok(mut queued) = backend.list(ticket::TicketListQuery::states([
ticket::TicketListState::Queued,
])) else {
return;
};
queued.sort_by(|left, right| left.id.cmp(&right.id));
if queued.is_empty() {
*api.orchestrator_attention_fingerprint
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
return;
}
let Ok(inprogress) = backend.list(ticket::TicketListQuery::states([
ticket::TicketListState::InProgress,
])) else {
return;
};
if !inprogress.is_empty() {
return;
}
let fingerprint = queued
.iter()
.map(|ticket| ticket.id.as_str())
.collect::<Vec<_>>()
.join("|");
if api
.orchestrator_attention_fingerprint
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.as_deref()
== Some(fingerprint.as_str())
{
return;
}
let shown = queued
.iter()
.take(ORCHESTRATOR_ATTENTION_TICKET_LIMIT)
.map(|ticket| {
format!(
"- {} — {}",
bounded_orchestrator_attention_text(&ticket.id, 80),
bounded_orchestrator_attention_text(&ticket.title, 240)
)
})
.collect::<Vec<_>>()
.join("\n");
let omitted = queued
.len()
.saturating_sub(ORCHESTRATOR_ATTENTION_TICKET_LIMIT);
let omitted_line = if omitted == 0 {
String::new()
} else {
format!("Additional queued Tickets omitted from this notice: {omitted}\n")
};
let content = ORCHESTRATOR_ATTENTION_PROMPT
.replace("{{omitted_line}}", &omitted_line)
.replace("{{workspace_id}}", &api.config.workspace_id)
.replace("{{ticket_lines}}", &shown);
let accepted = api
.runtime
.send_input(
&orchestrator.runtime_id,
&orchestrator.worker_id,
WorkerInputRequest {
kind: WorkerInputKind::Notify,
content,
segments: None,
},
)
.is_ok_and(|result| result.state == WorkerOperationState::Accepted);
if accepted {
*api.orchestrator_attention_fingerprint
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(fingerprint);
}
}
fn bounded_orchestrator_attention_text(input: &str, max_chars: usize) -> String {
let mut output = String::new();
for (index, character) in input.chars().enumerate() {
if index == max_chars {
output.push('…');
break;
}
output.push(if character.is_control() {
' '
} else {
character
});
}
output
}
fn find_workspace_orchestrator(api: &WorkspaceApi) -> Option<WorkerSummary> {
let is_orchestrator = |worker: &WorkerSummary| {
worker.singleton_key.as_deref() == Some(crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY)
@@ -3140,24 +3183,6 @@ fn find_workspace_orchestrator(api: &WorkspaceApi) -> Option<WorkerSummary> {
None
}
fn current_ticket_notification_recipient(
api: &WorkspaceApi,
delivery: &crate::store::TicketNotificationDeliveryRecord,
) -> Option<(String, String)> {
match delivery.recipient_kind.as_str() {
"assigned" => api
.store
.get_current_ticket_worker_assignment(&delivery.workspace_id, &delivery.ticket_id)
.ok()
.flatten()
.map(|assignment| (assignment.runtime_id, assignment.worker_id)),
"orchestrator" => {
find_workspace_orchestrator(api).map(|worker| (worker.runtime_id, worker.worker_id))
}
_ => None,
}
}
#[derive(Debug, Clone, Serialize)]
struct MemoryDocumentResponse {
body_md: String,
@@ -3701,6 +3726,10 @@ async fn scoped_start_workspace_orchestrator(
restored.diagnostics,
));
}
*api.orchestrator_attention_fingerprint
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
dispatch_orchestrator_queue_attention(&api);
return Ok(Json(workspace_orchestrator_response(&api, "restored")));
}
@@ -3734,6 +3763,10 @@ async fn scoped_start_workspace_orchestrator(
result.diagnostics,
));
}
*api.orchestrator_attention_fingerprint
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
dispatch_orchestrator_queue_attention(&api);
Ok(Json(workspace_orchestrator_response(&api, "created")))
}
@@ -4726,7 +4759,6 @@ async fn scoped_restore_runtime_worker(
.into());
}
assign_ticket_worker_from_lifecycle(&api, assignment, &runtime_id, &worker_id)?;
dispatch_pending_ticket_notifications(&api, &workspace_id);
return Ok(Json(WorkerRestoreResponse {
workspace_id,
runtime_id,
@@ -4747,7 +4779,6 @@ async fn scoped_restore_runtime_worker(
if let Some(assignment) = assignment_request.as_ref() {
assign_ticket_worker_from_lifecycle(&api, assignment, &runtime_id, &worker_id)?;
}
dispatch_pending_ticket_notifications(&api, &workspace_id);
Ok(response)
}
@@ -6420,7 +6451,6 @@ async fn create_runtime_worker(
if let Some(assignment) = request.ticket_assignment.as_ref() {
if let Some(worker) = existing_lifecycle_assignment_worker(&api, assignment, &runtime_id)? {
assign_ticket_worker_from_lifecycle(&api, assignment, &runtime_id, &worker.worker_id)?;
dispatch_pending_ticket_notifications(&api, api.workspace_id());
return Ok(Json(WorkerSpawnResult {
state: WorkerOperationState::Accepted,
worker: Some(worker),
@@ -9885,7 +9915,7 @@ mod tests {
}
#[tokio::test]
async fn authenticated_worker_ticket_mutation_routes_durable_assignment_notification() {
async fn authenticated_worker_ticket_mutation_notifies_current_assignment() {
let dir = tempfile::tempdir().unwrap();
let api = test_api(dir.path()).await;
let spawn = |name: &str| WorkerSpawnRequest {
@@ -9996,47 +10026,6 @@ mod tests {
);
assert!(committed_event.attributes.contains_key("event_id"));
assert!(committed_event.attributes.contains_key("event_sequence"));
let event_sequence = committed_event
.attributes
.get("event_sequence")
.unwrap()
.parse::<i64>()
.unwrap();
let response = build_router(api.clone())
.oneshot(
Request::builder()
.method("GET")
.uri(format!(
"/api/w/{TEST_WORKSPACE_ID}/tickets/{}/record",
ticket_ref.id
))
.header("x-yoi-runtime-id", EMBEDDED_WORKER_RUNTIME_ID)
.header("x-yoi-worker-id", &source_worker.worker_id)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
api.store
.get_ticket_notification_cursor(
TEST_WORKSPACE_ID,
&ticket_ref.id,
EMBEDDED_WORKER_RUNTIME_ID,
&source_worker.worker_id,
)
.unwrap(),
Some(event_sequence)
);
assert!(
api.store
.list_pending_ticket_notification_deliveries(TEST_WORKSPACE_ID, 10)
.unwrap()
.is_empty(),
"accepted Runtime system input must complete the outbox delivery"
);
let Json(stale_report) = execute_worker_ticket_test_operation(
State(api.clone()),
AxumPath(ScopedWorkspacePath {
@@ -10132,7 +10121,7 @@ mod tests {
}
#[tokio::test]
async fn queued_ticket_mutation_targets_current_orchestrator() {
async fn queued_ticket_mutation_succeeds_without_orchestrator() {
let dir = tempfile::tempdir().unwrap();
let api = test_api(dir.path()).await;
let source = api
@@ -10163,41 +10152,6 @@ mod tests {
.unwrap()
.worker
.unwrap();
let orchestrator = api
.runtime
.spawn_worker(
EMBEDDED_WORKER_RUNTIME_ID,
WorkerSpawnRequest {
requested_worker_name: Some("workspace-orchestrator".to_string()),
intent: WorkerSpawnIntent::WorkspaceOrchestrator,
acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
expected_segments: 0,
},
profile: ProfileSelector::Builtin("builtin:orchestrator".to_string()),
ticket_assignment: None,
initial_input: None,
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
resolved_config_bundle: None,
resolved_workspace_api: Some(test_worker_workspace_api(
EMBEDDED_WORKER_RUNTIME_ID,
)),
},
)
.unwrap()
.worker
.unwrap();
api.runtime
.stop_worker(
EMBEDDED_WORKER_RUNTIME_ID,
&orchestrator.worker_id,
WorkerLifecycleRequest {
reason: Some("test pending delivery".to_string()),
ticket_assignment: None,
},
)
.unwrap();
let backend = browser_ticket_backend(&api).unwrap();
let mut input = ticket::NewTicket::new("Queued notification");
input.workflow_state = Some(TicketWorkflowState::Queued);
@@ -10224,16 +10178,82 @@ mod tests {
)
.await
.unwrap();
}
#[tokio::test]
async fn orchestrator_running_to_idle_recovers_queued_ticket_without_notification_memory() {
let dir = tempfile::tempdir().unwrap();
let api = test_api(dir.path()).await;
let backend = browser_ticket_backend(&api).unwrap();
let ticket_ref = backend
.create(ticket::NewTicket::new("Recover queued work"))
.unwrap();
backend
.mark_intake_ready(
TicketIdOrSlug::Id(ticket_ref.id.clone()),
ticket::TicketIntakeSummary {
author: Some("intake".to_string()),
body: MarkdownText::new("Ready"),
references: Vec::new(),
},
ticket::TicketStateChange {
from: "planning".to_string(),
to: "ready".to_string(),
reason: "ready".to_string(),
author: Some("intake".to_string()),
body: MarkdownText::new("Ready"),
references: Vec::new(),
},
)
.unwrap();
backend
.queue_ready(TicketIdOrSlug::Id(ticket_ref.id.clone()), "browser-user")
.unwrap();
*api.orchestrator_attention_fingerprint.lock().unwrap() = Some(ticket_ref.id.clone());
let Json(started) = scoped_start_workspace_orchestrator(
State(api.clone()),
AxumPath(ScopedWorkspacePath {
workspace_id: TEST_WORKSPACE_ID.to_string(),
}),
)
.await
.unwrap();
assert!(started.online);
assert_eq!(
api.store
.count_ticket_notification_deliveries_for_recipient(
TEST_WORKSPACE_ID,
&ticket_ref.id,
EMBEDDED_WORKER_RUNTIME_ID,
&orchestrator.worker_id,
)
.unwrap(),
1
api.orchestrator_attention_fingerprint
.lock()
.unwrap()
.as_deref(),
Some(ticket_ref.id.as_str())
);
*api.orchestrator_attention_fingerprint.lock().unwrap() = None;
let worker_id = started.worker.as_ref().unwrap().worker_id.clone();
maybe_dispatch_orchestrator_turn_end(
&api,
&worker_id,
Some(protocol::subscription::SubscriptionWorkerState::Idle),
protocol::subscription::SubscriptionWorkerState::Idle,
);
assert!(
api.orchestrator_attention_fingerprint
.lock()
.unwrap()
.is_none()
);
maybe_dispatch_orchestrator_turn_end(
&api,
&worker_id,
Some(protocol::subscription::SubscriptionWorkerState::Running),
protocol::subscription::SubscriptionWorkerState::Idle,
);
assert_eq!(
api.orchestrator_attention_fingerprint
.lock()
.unwrap()
.as_deref(),
Some(ticket_ref.id.as_str())
);
}
@@ -10551,7 +10571,6 @@ mod tests {
.unwrap();
assert_eq!(queued.state, "queued");
assert_eq!(queued.queued_by.as_deref(), Some("browser-user"));
let Json(reviewed) = scoped_review_ticket(
State(api.clone()),
AxumPath(path()),
crates/workspace-server/src/store.rs
+31 -417
View File
@@ -122,6 +122,11 @@ const MIGRATIONS: &[Migration] = &[
name: "remove per-Worker Workspace credentials",
apply: remove_worker_workspace_credentials,
},
Migration {
version: 22,
name: "drop Ticket notification outbox",
apply: drop_ticket_notification_tables,
},
];
struct Migration {
@@ -296,31 +301,6 @@ pub struct TicketWorkerAssignmentUpdate {
pub previous: Option<TicketWorkerAssignmentRecord>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TicketNotificationRecipient {
pub runtime_id: String,
pub worker_id: String,
pub recipient_kind: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TicketNotificationDeliveryRecord {
pub notification_id: String,
pub workspace_id: String,
pub ticket_id: String,
pub event_sequence: i64,
pub event_kind: String,
pub source_operation_kind: String,
pub source_actor_role: String,
pub source_assignment_id: Option<String>,
pub source_runtime_id: String,
pub source_worker_id: String,
pub recipient_runtime_id: String,
pub recipient_worker_id: String,
pub recipient_kind: String,
pub attempts: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkdirRegistryRecord {
pub workspace_id: String,
@@ -625,70 +605,6 @@ pub trait ControlPlaneStore: Send + Sync {
limit: usize,
) -> Result<Vec<TicketWorkerAssignmentEventRecord>>;
fn enqueue_ticket_notification(
&self,
notification_id: &str,
workspace_id: &str,
ticket_id: &str,
event_sequence: i64,
source_runtime_id: &str,
source_worker_id: &str,
previous_state: &str,
current_state: &str,
created_at: &str,
recipients: &[TicketNotificationRecipient],
) -> Result<()>;
fn list_pending_ticket_notification_deliveries(
&self,
workspace_id: &str,
limit: usize,
) -> Result<Vec<TicketNotificationDeliveryRecord>>;
fn count_ticket_notification_deliveries_for_recipient(
&self,
workspace_id: &str,
ticket_id: &str,
runtime_id: &str,
worker_id: &str,
) -> Result<usize>;
fn mark_ticket_notification_delivered(
&self,
notification_id: &str,
recipient_runtime_id: &str,
recipient_worker_id: &str,
delivered_at: &str,
) -> Result<()>;
fn mark_ticket_notification_failed(
&self,
notification_id: &str,
recipient_runtime_id: &str,
recipient_worker_id: &str,
error: &str,
) -> Result<()>;
fn reroute_ticket_notification_delivery(
&self,
notification_id: &str,
old_runtime_id: &str,
old_worker_id: &str,
new_runtime_id: &str,
new_worker_id: &str,
) -> Result<()>;
fn upsert_ticket_notification_cursor(
&self,
workspace_id: &str,
ticket_id: &str,
runtime_id: &str,
worker_id: &str,
event_index: i64,
updated_at: &str,
) -> Result<()>;
fn get_ticket_notification_cursor(
&self,
workspace_id: &str,
ticket_id: &str,
runtime_id: &str,
worker_id: &str,
) -> Result<Option<i64>>;
fn upsert_workdir_registry(&self, record: &WorkdirRegistryRecord) -> Result<()>;
fn get_workdir_registry(
&self,
@@ -2183,239 +2099,6 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
})
}
fn enqueue_ticket_notification(
&self,
notification_id: &str,
workspace_id: &str,
ticket_id: &str,
event_sequence: i64,
source_runtime_id: &str,
source_worker_id: &str,
previous_state: &str,
current_state: &str,
created_at: &str,
recipients: &[TicketNotificationRecipient],
) -> Result<()> {
self.with_conn(|conn| {
let tx = conn.unchecked_transaction()?;
tx.execute(
r#"INSERT INTO ticket_notification_outbox (
notification_id, workspace_id, ticket_id, event_sequence,
source_runtime_id, source_worker_id, previous_state, current_state, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)"#,
params![
notification_id,
workspace_id,
ticket_id,
event_sequence,
source_runtime_id,
source_worker_id,
previous_state,
current_state,
created_at,
],
)?;
for recipient in recipients {
tx.execute(
r#"INSERT OR IGNORE INTO ticket_notification_deliveries (
notification_id, recipient_runtime_id, recipient_worker_id,
recipient_kind, attempts
) VALUES (?1, ?2, ?3, ?4, 0)"#,
params![
notification_id,
recipient.runtime_id,
recipient.worker_id,
recipient.recipient_kind,
],
)?;
}
tx.commit()?;
Ok(())
})
}
fn list_pending_ticket_notification_deliveries(
&self,
workspace_id: &str,
limit: usize,
) -> Result<Vec<TicketNotificationDeliveryRecord>> {
self.with_conn(|conn| {
let mut stmt = conn.prepare(
r#"SELECT o.notification_id, o.workspace_id, o.ticket_id, o.event_sequence,
o.event_kind, o.source_operation_kind, o.source_actor_role,
o.source_assignment_id, o.source_runtime_id, o.source_worker_id,
d.recipient_runtime_id, d.recipient_worker_id, d.recipient_kind, d.attempts
FROM ticket_notification_deliveries AS d
JOIN ticket_notification_outbox AS o ON o.notification_id = d.notification_id
WHERE o.workspace_id = ?1 AND d.delivered_at IS NULL
ORDER BY o.created_at ASC, o.notification_id ASC
LIMIT ?2"#,
)?;
let rows = stmt.query_map(params![workspace_id, limit as i64], |row| {
Ok(TicketNotificationDeliveryRecord {
notification_id: row.get(0)?,
workspace_id: row.get(1)?,
ticket_id: row.get(2)?,
event_sequence: row.get(3)?,
event_kind: row.get(4)?,
source_operation_kind: row.get(5)?,
source_actor_role: row.get(6)?,
source_assignment_id: row.get(7)?,
source_runtime_id: row.get(8)?,
source_worker_id: row.get(9)?,
recipient_runtime_id: row.get(10)?,
recipient_worker_id: row.get(11)?,
recipient_kind: row.get(12)?,
attempts: row.get(13)?,
})
})?;
rows.collect::<std::result::Result<Vec<_>, _>>()
.map_err(Error::from)
})
}
fn count_ticket_notification_deliveries_for_recipient(
&self,
workspace_id: &str,
ticket_id: &str,
runtime_id: &str,
worker_id: &str,
) -> Result<usize> {
self.with_conn(|conn| {
let count = conn.query_row(
r#"SELECT COUNT(*)
FROM ticket_notification_deliveries AS d
JOIN ticket_notification_outbox AS o ON o.notification_id = d.notification_id
WHERE o.workspace_id = ?1 AND o.ticket_id = ?2
AND d.recipient_runtime_id = ?3 AND d.recipient_worker_id = ?4"#,
params![workspace_id, ticket_id, runtime_id, worker_id],
|row| row.get::<_, i64>(0),
)?;
Ok(count as usize)
})
}
fn mark_ticket_notification_delivered(
&self,
notification_id: &str,
recipient_runtime_id: &str,
recipient_worker_id: &str,
delivered_at: &str,
) -> Result<()> {
self.with_conn(|conn| {
conn.execute(
r#"UPDATE ticket_notification_deliveries
SET delivered_at = ?4, last_error = NULL, attempts = attempts + 1
WHERE notification_id = ?1 AND recipient_runtime_id = ?2 AND recipient_worker_id = ?3"#,
params![notification_id, recipient_runtime_id, recipient_worker_id, delivered_at],
)?;
Ok(())
})
}
fn mark_ticket_notification_failed(
&self,
notification_id: &str,
recipient_runtime_id: &str,
recipient_worker_id: &str,
error: &str,
) -> Result<()> {
self.with_conn(|conn| {
conn.execute(
r#"UPDATE ticket_notification_deliveries
SET last_error = ?4, attempts = attempts + 1
WHERE notification_id = ?1 AND recipient_runtime_id = ?2 AND recipient_worker_id = ?3"#,
params![notification_id, recipient_runtime_id, recipient_worker_id, error],
)?;
Ok(())
})
}
fn reroute_ticket_notification_delivery(
&self,
notification_id: &str,
old_runtime_id: &str,
old_worker_id: &str,
new_runtime_id: &str,
new_worker_id: &str,
) -> Result<()> {
self.with_conn(|conn| {
let tx = conn.unchecked_transaction()?;
let recipient_kind: Option<String> = tx
.query_row(
r#"SELECT recipient_kind FROM ticket_notification_deliveries
WHERE notification_id = ?1 AND recipient_runtime_id = ?2 AND recipient_worker_id = ?3"#,
params![notification_id, old_runtime_id, old_worker_id],
|row| row.get(0),
)
.optional()?;
if let Some(recipient_kind) = recipient_kind {
tx.execute(
r#"INSERT OR IGNORE INTO ticket_notification_deliveries (
notification_id, recipient_runtime_id, recipient_worker_id, recipient_kind, attempts
) VALUES (?1, ?2, ?3, ?4, 0)"#,
params![notification_id, new_runtime_id, new_worker_id, recipient_kind],
)?;
tx.execute(
r#"DELETE FROM ticket_notification_deliveries
WHERE notification_id = ?1 AND recipient_runtime_id = ?2 AND recipient_worker_id = ?3"#,
params![notification_id, old_runtime_id, old_worker_id],
)?;
}
tx.commit()?;
Ok(())
})
}
fn upsert_ticket_notification_cursor(
&self,
workspace_id: &str,
ticket_id: &str,
runtime_id: &str,
worker_id: &str,
event_index: i64,
updated_at: &str,
) -> Result<()> {
self.with_conn(|conn| {
conn.execute(
r#"INSERT INTO ticket_notification_cursors (
workspace_id, ticket_id, runtime_id, worker_id, last_event_index, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6)
ON CONFLICT(workspace_id, ticket_id, runtime_id, worker_id) DO UPDATE SET
last_event_index = MAX(last_event_index, excluded.last_event_index),
updated_at = excluded.updated_at"#,
params![
workspace_id,
ticket_id,
runtime_id,
worker_id,
event_index,
updated_at
],
)?;
Ok(())
})
}
fn get_ticket_notification_cursor(
&self,
workspace_id: &str,
ticket_id: &str,
runtime_id: &str,
worker_id: &str,
) -> Result<Option<i64>> {
self.with_conn(|conn| {
conn.query_row(
r#"SELECT last_event_index FROM ticket_notification_cursors
WHERE workspace_id = ?1 AND ticket_id = ?2 AND runtime_id = ?3 AND worker_id = ?4"#,
params![workspace_id, ticket_id, runtime_id, worker_id],
|row| row.get(0),
)
.optional()
.map_err(Error::from)
})
}
fn upsert_workdir_registry(&self, record: &WorkdirRegistryRecord) -> Result<()> {
self.with_conn(|conn| {
conn.execute(
@@ -3343,6 +3026,17 @@ fn remove_worker_workspace_credentials(conn: &Connection) -> Result<()> {
Ok(())
}
fn drop_ticket_notification_tables(conn: &Connection) -> Result<()> {
conn.execute_batch(
r#"
DROP TABLE IF EXISTS ticket_notification_cursors;
DROP TABLE IF EXISTS ticket_notification_deliveries;
DROP TABLE IF EXISTS ticket_notification_outbox;
"#,
)?;
Ok(())
}
fn create_objective_event_tables(conn: &Connection) -> Result<()> {
conn.execute_batch(
r#"
@@ -4014,6 +3708,9 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
"ticket_targets",
"ticket_target_paths",
"ticket_worker_links",
"ticket_notification_outbox",
"ticket_notification_deliveries",
"ticket_notification_cursors",
] {
assert!(!table_exists(&conn, table).unwrap(), "{table} still exists");
}
@@ -4025,7 +3722,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
let db = dir.path().join("control-plane.sqlite");
let store = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 21);
assert_eq!(store.schema_version().await.unwrap(), 22);
assert!(
!store
.with_conn(|conn| table_exists(conn, "worker_workspace_credentials"))
@@ -4042,7 +3739,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
store.upsert_workspace(&record).await.unwrap();
let reopened = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(reopened.schema_version().await.unwrap(), 21);
assert_eq!(reopened.schema_version().await.unwrap(), 22);
assert_eq!(
reopened.get_workspace("local-dev").await.unwrap(),
Some(record)
@@ -4288,95 +3985,6 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
);
}
#[tokio::test]
async fn notification_outbox_is_durable() {
let dir = tempfile::tempdir().unwrap();
let db = dir.path().join("server.db");
let store = SqliteWorkspaceStore::open(&db).unwrap();
store
.upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-a".to_string(),
owner_account_id: None,
display_name: "Workspace A".to_string(),
state: "active".to_string(),
created_at: "2026-07-31T00:00:00Z".to_string(),
updated_at: "2026-07-31T00:00:00Z".to_string(),
})
.await
.unwrap();
store
.enqueue_ticket_notification(
"notification-1",
"workspace-a",
"ticket-1",
4,
"runtime-1",
"worker-1",
"queued",
"inprogress",
"2026-07-31T00:00:02Z",
&[TicketNotificationRecipient {
runtime_id: "runtime-1".to_string(),
worker_id: "worker-2".to_string(),
recipient_kind: "assigned".to_string(),
}],
)
.unwrap();
drop(store);
let store = SqliteWorkspaceStore::open(&db).unwrap();
let pending = store
.list_pending_ticket_notification_deliveries("workspace-a", 10)
.unwrap();
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].event_sequence, 4);
store
.reroute_ticket_notification_delivery(
"notification-1",
"runtime-1",
"worker-2",
"runtime-2",
"worker-3",
)
.unwrap();
assert_eq!(
store
.count_ticket_notification_deliveries_for_recipient(
"workspace-a",
"ticket-1",
"runtime-1",
"worker-2",
)
.unwrap(),
0
);
assert_eq!(
store
.count_ticket_notification_deliveries_for_recipient(
"workspace-a",
"ticket-1",
"runtime-2",
"worker-3",
)
.unwrap(),
1
);
store
.mark_ticket_notification_delivered(
"notification-1",
"runtime-2",
"worker-3",
"2026-07-31T00:00:03Z",
)
.unwrap();
assert!(
store
.list_pending_ticket_notification_deliveries("workspace-a", 10)
.unwrap()
.is_empty()
);
}
#[test]
fn fresh_schema_matches_workspace_db_v0_boundaries() {
let conn = Connection::open_in_memory().unwrap();
@@ -4427,6 +4035,9 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
"ticket_targets",
"ticket_target_paths",
"ticket_worker_links",
"ticket_notification_outbox",
"ticket_notification_deliveries",
"ticket_notification_cursors",
] {
assert!(
!tables.contains(forbidden),
@@ -4579,7 +4190,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
.unwrap();
let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 21);
assert_eq!(store.schema_version().await.unwrap(), 22);
store
.with_conn(|conn| {
@@ -4616,6 +4227,9 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
"ticket_targets",
"ticket_target_paths",
"ticket_worker_links",
"ticket_notification_outbox",
"ticket_notification_deliveries",
"ticket_notification_cursors",
] {
assert!(
!tables.contains(forbidden),
@@ -4765,7 +4379,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn repository_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 21);
assert_eq!(store.schema_version().await.unwrap(), 22);
let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(),
owner_account_id: None,
@@ -4803,7 +4417,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn memory_authority_records_round_trip_and_close_staging() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 21);
assert_eq!(store.schema_version().await.unwrap(), 22);
let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(),
owner_account_id: None,
@@ -4981,7 +4595,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn account_and_login_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 21);
assert_eq!(store.schema_version().await.unwrap(), 22);
let now = "2026-07-22T00:00:00Z".to_string();
let account = AccountRecord {
account_id: "acct-user-alice".to_string(),