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

feat: add Subject async chat with explicit durable publication

Browse Source
This commit is contained in:
Hare
2026-10-29 22:27:23 +09:00
parent 7ad6030498
commit 882316fb3b
54 changed files with 6187 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/interaction.rs
+5
View File
@@ -26,6 +26,9 @@ impl WorkerInputKind {
pub struct WorkerInput {
pub kind: WorkerInputKind,
pub content: String,
/// Provenance stamped by the authenticated Backend, never by public/model input.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub source: Option<protocol::AuthenticatedInputSource>,
/// Authenticated client-generated idempotency key. Runtime generates one
/// only for trusted internal callers that omit it.
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -39,6 +42,7 @@ impl WorkerInput {
Self {
kind: WorkerInputKind::User,
content: content.into(),
source: None,
submission_request_id: None,
segments: None,
}
@@ -48,6 +52,7 @@ impl WorkerInput {
Self {
kind: WorkerInputKind::Notify,
content: content.into(),
source: None,
submission_request_id: None,
segments: None,
}
crates/worker-runtime/src/runtime.rs
+5
View File
@@ -8184,6 +8184,7 @@ mod tests {
let input = WorkerInput {
kind: WorkerInputKind::User,
content: String::new(),
source: None,
submission_request_id: None,
segments: Some(vec![protocol::Segment::Flow {
selector: "builtin:coder-review".to_string(),
@@ -8197,6 +8198,7 @@ mod tests {
let input = WorkerInput {
kind: WorkerInputKind::User,
content: String::new(),
source: None,
submission_request_id: None,
segments: Some(Vec::new()),
};
@@ -8212,6 +8214,7 @@ mod tests {
request.initial_input = Some(WorkerInput {
kind: WorkerInputKind::User,
content: String::new(),
source: None,
submission_request_id: None,
segments: Some(vec![protocol::Segment::Flow {
selector: "builtin:coder-review".to_string(),
@@ -13678,6 +13681,7 @@ mod tests {
request.initial_input = Some(WorkerInput {
kind: WorkerInputKind::User,
content: String::new(),
source: None,
submission_request_id: None,
segments: Some(vec![
protocol::Segment::Flow {
@@ -13721,6 +13725,7 @@ mod tests {
let input = WorkerInput {
kind: WorkerInputKind::User,
content: String::new(),
source: None,
submission_request_id: None,
segments: Some(vec![protocol::Segment::Flow {
selector: "builtin:coder-review".to_string(),
crates/worker-runtime/src/worker_backend.rs
+25 -3
View File
@@ -6542,9 +6542,11 @@ where
Method::NotifyTracked {
notification_request_id: notification_request_id.clone(),
message: input.content,
source: protocol::AuthenticatedInputSource::Backend {
operation_id: notification_request_id.clone(),
},
source: input.source.unwrap_or_else(|| {
protocol::AuthenticatedInputSource::Backend {
operation_id: notification_request_id.clone(),
}
}),
},
notification_request_id,
);
@@ -9872,6 +9874,11 @@ mod tests {
enum MockResponse {
Complete(Vec<LlmEvent>),
Hang(Vec<LlmEvent>),
Gated {
started: Arc<tokio::sync::Notify>,
release: Arc<tokio::sync::Notify>,
events: Vec<LlmEvent>,
},
}
#[derive(Clone)]
@@ -9917,6 +9924,20 @@ mod tests {
MockResponse::Complete(events) => {
Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok))))
}
MockResponse::Gated {
started,
release,
events,
} => {
started.notify_one();
Ok(Box::pin(
futures::stream::once(async move {
release.notified().await;
futures::stream::iter(events.into_iter().map(Ok))
})
.flatten(),
))
}
MockResponse::Hang(events) => Ok(Box::pin(
futures::stream::iter(events.into_iter().map(Ok))
.chain(futures::stream::pending()),
@@ -10940,6 +10961,7 @@ mod tests {
#[cfg(feature = "fs-store")]
include!("worker_backend_restore_failure_tests.rs");
include!("worker_backend_restore_candidate_tests.rs");
include!("worker_backend_notify_transport_tests.rs");
struct PendingFailureFactory {
restore_calls: Arc<AtomicUsize>,
crates/worker-runtime/src/worker_backend_notify_transport_tests.rs
+250
View File
@@ -0,0 +1,250 @@
// Concrete Runtime -> Controller -> filesystem receipt/history tests. Only model
// I/O is gated; the Runtime adapter and notification acceptance owner are real.
mod notify_transport {
use super::*;
const WAIT: Duration = Duration::from_secs(5);
async fn accepted(events: &mut tokio::sync::broadcast::Receiver<Event>, id: &str) {
tokio::time::timeout(WAIT, async {
loop {
match events.recv().await.unwrap() {
Event::NotificationAccepted {
notification_request_id,
} if notification_request_id == id => return,
Event::NotificationRejected { message, .. } | Event::Error { message, .. } => {
panic!("notification rejected: {message}")
}
_ => {}
}
}
})
.await
.expect("durable notification acceptance");
}
#[test]
fn busy_notify_retains_source_scoped_receipts_and_commits_provenance_before_next_model_input() {
let started = Arc::new(tokio::sync::Notify::new());
let release = Arc::new(tokio::sync::Notify::new());
let next_started = Arc::new(tokio::sync::Notify::new());
let next_release = Arc::new(tokio::sync::Notify::new());
let client = MockClient::sequential(vec![
MockResponse::Gated {
started: started.clone(),
release: release.clone(),
events: simple_text_events(),
},
MockResponse::Gated {
started: next_started.clone(),
release: next_release.clone(),
events: simple_text_events(),
},
]);
let runtime_base = tempfile::tempdir().unwrap();
let cwd = tempfile::tempdir().unwrap();
let store = tempfile::tempdir().unwrap();
let backend = Arc::new(
WorkerRuntimeExecutionBackend::new(MockFactory {
client: client.clone(),
runtime_base: runtime_base.path().to_path_buf(),
cwd: cwd.path().to_path_buf(),
store_dir: store.path().join("sessions"),
worker_metadata_dir: store.path().join("workers"),
observed_cwds: Arc::new(Mutex::new(Vec::new())),
observed_workspace_clients: Arc::new(Mutex::new(Vec::new())),
})
.unwrap(),
);
let runtime =
EmbeddedRuntime::with_execution_backend(RuntimeOptions::default(), backend.clone())
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let detail = runtime
.create_worker(create_request("notify-provenance"))
.unwrap();
let worker_ref = &detail.worker_ref;
let handle = backend.workers.lock().unwrap()[worker_ref].handle.clone();
runtime
.send_input(worker_ref, WorkerInput::user("original request"))
.unwrap();
backend
.run_on_adapter_runtime(async move {
tokio::time::timeout(WAIT, started.notified())
.await
.unwrap();
Ok(())
})
.unwrap();
let account = protocol::AuthenticatedInputSource::Account {
account_id: "account-a".into(),
};
let mut human = WorkerInput::notify("human-notice-body");
human.source = Some(account.clone());
human.submission_request_id = Some("same-client-id".into());
// Simulate an ACK lost above the Controller acceptance boundary. The
// sender retries through Runtime, using the same source, ID and body.
let lost_handle = handle.clone();
backend
.run_on_adapter_runtime(async move {
let mut events = lost_handle.subscribe();
lost_handle
.send(Method::NotifyTracked {
notification_request_id: "same-client-id".into(),
message: "human-notice-body".into(),
source: account,
})
.await
.unwrap();
accepted(&mut events, "same-client-id").await;
Ok(())
})
.unwrap();
let first = runtime.send_input(worker_ref, human.clone()).unwrap();
let retry = runtime.send_input(worker_ref, human.clone()).unwrap();
assert_eq!(first.notification, retry.notification);
assert!(first.submission.is_none());
let mut conflict = human.clone();
conflict.content = "changed-body".into();
assert!(runtime.send_input(worker_ref, conflict).is_err());
let mut other_account = human.clone();
other_account.source = Some(protocol::AuthenticatedInputSource::Account {
account_id: "account-b".into(),
});
other_account.content = "other-account-body".into();
runtime.send_input(worker_ref, other_account).unwrap();
let mut report = human.clone();
report.source = Some(protocol::AuthenticatedInputSource::Worker {
runtime_id: "report-runtime".into(),
worker_id: "report-worker".into(),
});
report.content = "worker-report-body".into();
runtime.send_input(worker_ref, report).unwrap();
let mut system = WorkerInput::notify("system-notice-body");
system.submission_request_id = Some("system-notice-id".into());
runtime.send_input(worker_ref, system).unwrap();
assert_eq!(handle.shared_state.catalog_status(), WorkerStatus::Running);
assert_eq!(
client.call_count.load(Ordering::SeqCst),
1,
"notify must not interrupt the current model stream"
);
let before = handle.committed_entries();
let checkpoint = before
.iter()
.rev()
.find_map(|entry| match entry {
LogEntry::Extension {
domain, payload, ..
} if domain == "worker.pending_activations.v1" => Some(payload),
_ => None,
})
.unwrap();
assert_eq!(
checkpoint["notification_receipts"]
.as_array()
.unwrap()
.len(),
4
);
assert_eq!(
checkpoint["pending_notifications"].as_array().unwrap().len()
+ checkpoint["activating_notification"].as_array().unwrap().len(),
4
);
let sources = checkpoint["notification_receipts"]
.as_array()
.unwrap()
.iter()
.map(|receipt| receipt["source_namespace"].as_str().unwrap())
.collect::<BTreeSet<_>>();
assert_eq!(
sources,
BTreeSet::from([
"account:account-a",
"account:account-b",
"worker:report-runtime:report-worker",
"backend:system-notice-id",
])
);
release.notify_one();
backend
.run_on_adapter_runtime(async move {
tokio::time::timeout(WAIT, next_started.notified())
.await
.unwrap();
Ok(())
})
.unwrap();
let entries = handle.committed_entries();
let history = session_store::collect_state(&entries).annotated_history;
let expected = [
(
"human-notice-body",
LoggedSessionHistoryOrigin::HumanInput {
account_id: "account-a".into(),
},
),
(
"other-account-body",
LoggedSessionHistoryOrigin::HumanInput {
account_id: "account-b".into(),
},
),
(
"worker-report-body",
LoggedSessionHistoryOrigin::WorkerInput {
actor: session_store::LoggedWorkerSubject {
workspace_id: None,
runtime_id: Some("report-runtime".into()),
worker_id: "report-worker".into(),
},
},
),
(
"system-notice-body",
LoggedSessionHistoryOrigin::BackendInstruction {
operation_id: Some("system-notice-id".into()),
},
),
];
let requests = client.captured.lock().unwrap();
let first_items = serde_json::to_string(&requests[0].items).unwrap();
let next_items = serde_json::to_string(&requests[1].items).unwrap();
for (body, origin) in expected {
assert!(!first_items.contains(body));
assert_eq!(next_items.matches(body).count(), 1);
let matching = history
.iter()
.filter(|entry| serde_json::to_string(&entry.item).unwrap().contains(body))
.collect::<Vec<_>>();
assert_eq!(
matching.len(),
1,
"notification must commit exactly once: {body}"
);
assert_eq!(matching[0].metadata.origin, origin);
}
drop(requests);
assert!(!next_items.contains("changed-body"));
// Provenance is annotation, not a grant or promoted Host instruction.
let replay = runtime.send_input(worker_ref, human).unwrap();
assert_eq!(first.notification, replay.notification);
assert_eq!(
serde_json::to_value(handle.committed_entries()).unwrap(),
serde_json::to_value(entries).unwrap(),
"retained receipt replay must not append history"
);
next_release.notify_one();
runtime
.stop_worker(worker_ref, Some("test complete".into()))
.unwrap();
assert!(
runtime
.send_input(worker_ref, WorkerInput::notify("stopped-notice"))
.is_err()
);
}
}
crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs
+123 -6
View File
@@ -90,14 +90,26 @@ mod restore_candidate {
}
async fn notify(handle: &WorkerHandle) {
notify_from_source(handle, None).await;
}
async fn notify_from_source(
handle: &WorkerHandle,
source: Option<protocol::AuthenticatedInputSource>,
) {
let mut events = handle.subscribe();
handle
.send(Method::Notify {
let method = match source {
Some(source) => Method::NotifyTracked {
notification_request_id: NOTICE.into(),
message: NOTICE.into(),
})
.await
.unwrap();
source,
},
None => Method::Notify {
notification_request_id: NOTICE.into(),
message: NOTICE.into(),
},
};
handle.send(method).await.unwrap();
receive_matching(&mut events, "notification acceptance", |event| {
matches!(event, Event::NotificationAccepted { notification_request_id }
if notification_request_id == NOTICE)
@@ -169,6 +181,13 @@ mod restore_candidate {
}
fn with_seed(no_model: bool) -> Self {
Self::with_seed_source(no_model, None)
}
fn with_seed_source(
no_model: bool,
notice_source: Option<protocol::AuthenticatedInputSource>,
) -> Self {
let root = restore_failure_tempdir();
let model = if no_model {
MockClient::new(Vec::new())
@@ -339,7 +358,7 @@ mod restore_candidate {
disposition: protocol::SubmissionDisposition::Queued, ..
}));
receipts.push(serde_json::to_value(receipt).unwrap());
if id == FIRST { notify(&handle).await; }
if id == FIRST { notify_from_source(&handle, notice_source.clone()).await; }
}
assert_eq!(pending(&handle).await.submissions.len(), 2);
shutdown_profile_controller(seed.0.take().unwrap()).await;
@@ -1307,6 +1326,104 @@ mod restore_candidate {
}
}
#[test]
#[serial_test::serial(worker_allocation)]
fn notify_source_and_lost_ack_receipt_survive_stopped_delivery_and_explicit_restore() {
let source = protocol::AuthenticatedInputSource::Account {
account_id: "saved-chat-account".into(),
};
let fixture = Fixture::with_seed_source(false, Some(source.clone()));
let mut retry = WorkerInput::notify(NOTICE);
retry.source = Some(source.clone());
retry.submission_request_id = Some(NOTICE.into());
let before = fixture.canonical_files();
// A direct notify is not a stopped-Worker mailbox or implicit Restore.
assert!(
!fixture
.backend
.dispatch_input(&fixture.request.worker_ref, retry.clone())
.is_accepted()
);
assert!(fixture.backend.workers.lock().unwrap().is_empty());
assert_eq!(fixture.canonical_files(), before);
let checkpoint = queue_checkpoint(&fixture.entries());
assert_eq!(
checkpoint["notification_receipts"][0]["source_namespace"],
source.namespace()
);
fixture.prepare();
fixture
.backend
.activate_restored_worker(fixture.request.operation_id, &fixture.request.worker_ref)
.unwrap();
let handle = fixture.backend.workers.lock().unwrap()[&fixture.request.worker_ref]
.handle
.clone();
fixture
.backend
.run_on_adapter_runtime(async move {
finished(&handle, SECOND).await;
Ok(())
})
.unwrap();
// The seed Controller accepted durably before shutdown; its caller lost
// that receipt. Replay through the concrete adapter must not append or run.
let committed = fixture.canonical_files();
let calls = fixture.model.call_count.load(Ordering::SeqCst);
let accepted = fixture
.backend
.dispatch_input(&fixture.request.worker_ref, retry.clone());
assert!(accepted.is_accepted(), "{accepted:?}");
assert_eq!(
accepted.notification.unwrap().notification_request_id,
NOTICE
);
let mut conflict = retry;
conflict.content = "different-body".into();
let rejected = fixture
.backend
.dispatch_input(&fixture.request.worker_ref, conflict);
assert_eq!(
rejected.outcome,
crate::execution::WorkerExecutionOutcome::Rejected
);
assert_eq!(fixture.canonical_files(), committed);
assert_eq!(fixture.model.call_count.load(Ordering::SeqCst), calls);
let history = session_store::collect_state(&fixture.entries()).annotated_history;
let notices = history
.iter()
.filter(|entry| match &entry.item {
LoggedItem::Message { content, .. } => content.iter().any(|part| {
matches!(part,
LoggedContentPart::Text { text } if text.contains(NOTICE))
}),
_ => false,
})
.collect::<Vec<_>>();
assert_eq!(notices.len(), 1);
assert_eq!(
notices[0].metadata.origin,
LoggedSessionHistoryOrigin::HumanInput {
account_id: "saved-chat-account".into()
}
);
let requests = fixture.model.captured.lock().unwrap();
assert_eq!(
serde_json::to_string(&requests[0].items)
.unwrap()
.matches(NOTICE)
.count(),
1
);
assert_eq!(
serde_json::to_string(&requests[2].items)
.unwrap()
.matches(NOTICE)
.count(),
1
);
}
#[test]
#[serial_test::serial(worker_allocation)]
fn activate_saved_inputs_drains_fifo_once_and_replays_durable_receipts() {
crates/worker-runtime/src/worker_source.rs
+71
View File
@@ -775,6 +775,11 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
client.subjektiv_record_session(workspace_id, request).await,
)
}
WorkspaceServerOperation::ChatWorker(request) => {
generated_workspace_response(
client.chat_worker_operation(workspace_id, request).await,
)
}
WorkspaceServerOperation::SubjektivMemory(request) => {
generated_workspace_response(
client.subjektiv_memory_backend(workspace_id, request).await,
@@ -1130,6 +1135,72 @@ mod tests {
}
}
#[cfg(feature = "http-server")]
#[test]
fn generated_chat_worker_transport_preserves_request_and_signed_worker_source() {
use crate::auth::{RuntimeRequestSourceExpectation, verify_runtime_request_source};
let identity = RuntimeIdentityMaterial::generate("runtime-a").unwrap();
let request = server_api::ChatWorkerRequest {
operation: server_api::ChatWorkerOperation::Send {
request: server_api::ChatSendMessageRequest {
conversation_id: "conversation-a".into(),
operation_id: "send-a".into(),
text: "explicit reply\ncontinued".into(),
reply_to: Some("message-a".into()),
work_ref: Some("work-a".into()),
expected_last_user_message_id: Some("user-message-1".into()),
},
},
};
let reply = serde_json::json!({"kind":"sent","message":{
"id":"message-b","conversation_id":"conversation-a","sequence":2,
"sender":{"kind":"subject","subject_id":"subject-a"},
"text":"explicit reply\ncontinued","reply_to":"message-a","work_ref":"work-a",
"operation_id":"send-a","last_user_message_id":"user-message-1"
}})
.to_string();
let response = format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", reply.len(), reply).into_bytes();
let (url, server) = binary_loopback(response);
let client = RuntimeOwnedWorkspaceClient::new("workspace-a", url, "runtime-a", "worker-a")
.with_runtime_request_source(&identity, "server-a");
let result = client
.execute_server_operation(WorkspaceServerOperation::ChatWorker(request.clone()))
.unwrap();
assert_eq!(result.status, 200);
let (head, body) = server.join().unwrap();
let path = "/api/w/workspace-a/subjektiv/chat/worker";
assert!(head.starts_with(&format!("POST {path} HTTP/1.1\r\n")));
assert_eq!(
serde_json::from_slice::<serde_json::Value>(&body).unwrap(),
serde_json::to_value(request).unwrap()
);
let proof = head
.lines()
.find_map(|line| {
let (name, value) = line.split_once(':')?;
name.eq_ignore_ascii_case(RUNTIME_REQUEST_SOURCE_PROOF_HEADER)
.then(|| value.trim())
})
.unwrap();
let digest = request_body_digest(&body);
verify_runtime_request_source(
proof,
&identity.public_key,
&RuntimeRequestSourceExpectation {
identity_id: "runtime-a",
audience: "server-a",
workspace_id: "workspace-a",
worker_id: Some("worker-a"),
permission: WORKSPACE_REQUEST_PERMISSION,
method: "POST",
path,
body_digest: &digest,
now_unix: unix_now_seconds() as i64,
},
)
.unwrap();
}
#[cfg(feature = "http-server")]
#[test]
fn binary_roundtrip_signs_exact_query_method_worker_and_raw_body() {