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

fix: preserve Console Notify ACK identity and request ID limits

Browse Source
This commit is contained in:
Hare
2026-10-30 00:00:05 +09:00
parent 87a1b6f1eb
commit c215c72873
14 changed files with 383 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/protocol/src/lib.rs
+52
View File
@@ -103,6 +103,33 @@ impl AuthenticatedInputSource {
}
}
/// Server-owned receipt partition, independent of client-visible ACK identity.
/// Console Notify keeps its original bounded ID without sharing HTTP/chat receipts.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NotificationReceiptScope {
#[default]
Source,
BrowserConsole,
}
impl NotificationReceiptScope {
pub fn namespace(self, source_namespace: String) -> String {
match self {
Self::Source => source_namespace,
Self::BrowserConsole => format!("browser-console:{source_namespace}"),
}
}
}
// Like provenance, a receipt partition is rebound by the authenticated transport.
fn ignore_wire_notification_scope<'de, D: serde::Deserializer<'de>>(
deserializer: D,
) -> Result<NotificationReceiptScope, D::Error> {
let _ = serde::de::IgnoredAny::deserialize(deserializer)?;
Ok(NotificationReceiptScope::Source)
}
/// Caller-owned identity for one state-changing Worker command.
///
/// Commands operate on the controller state at admission time, not a caller's
@@ -311,6 +338,8 @@ pub enum Method {
NotifyTracked {
notification_request_id: String,
message: String,
#[serde(default, deserialize_with = "ignore_wire_notification_scope")]
receipt_scope: NotificationReceiptScope,
#[serde(default, deserialize_with = "ignore_wire_input_source")]
source: AuthenticatedInputSource,
},
@@ -2242,6 +2271,29 @@ mod tests {
);
}
#[test]
fn notification_receipt_scope_is_not_selected_by_wire_input() {
for scope in [
serde_json::json!("browser_console"),
serde_json::json!({"forged":true}),
serde_json::Value::Null,
] {
let method: Method = serde_json::from_value(serde_json::json!({
"method":"notify_tracked", "params":{
"notification_request_id":"original", "message":"advisory",
"receipt_scope":scope,
"source":{"kind":"account","account_id":"forged"}
}
}))
.unwrap();
assert!(matches!(method, Method::NotifyTracked {
receipt_scope: NotificationReceiptScope::Source,
source: AuthenticatedInputSource::UntrustedWire,
notification_request_id, ..
} if notification_request_id == "original"));
}
}
#[test]
fn notify_rejects_typed_intent_instead_of_silently_discarding_it() {
for method in ["notify", "notify_tracked"] {
crates/standalone/src/host.rs
+2
View File
@@ -706,6 +706,7 @@ fn authorize_local_input(method: protocol::Method) -> protocol::Method {
message,
..
} => Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id,
message,
source: AuthenticatedInputSource::LocalOperator,
@@ -952,6 +953,7 @@ mod local_input_tests {
message: "hello".into(),
},
Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: "tracked-notify".into(),
message: "hello".into(),
source,
crates/worker-runtime/src/http_server.rs
+13 -1
View File
@@ -1415,6 +1415,14 @@ fn authorize_runtime_protocol_method(
message,
..
} => protocol::Method::NotifyTracked {
receipt_scope: if matches!(
transport_source,
Some(protocol::AuthenticatedInputSource::Account { .. })
) {
protocol::NotificationReceiptScope::BrowserConsole
} else {
protocol::NotificationReceiptScope::Source
},
source: transport_source.cloned().unwrap_or_else(|| {
protocol::AuthenticatedInputSource::Backend {
operation_id: notification_request_id.clone(),
@@ -2985,6 +2993,7 @@ mod tests {
.unwrap()
.expect("account source header must resolve");
let wire = serde_json::to_string(&protocol::Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: "notification-1".into(),
message: "hello".into(),
source: protocol::AuthenticatedInputSource::Account {
@@ -2997,9 +3006,11 @@ mod tests {
assert!(matches!(
authorize_runtime_protocol_method(decoded, Some(&source)),
protocol::Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::BrowserConsole,
notification_request_id,
source: protocol::AuthenticatedInputSource::Account { account_id },
..
} if account_id == "account-1"
} if account_id == "account-1" && notification_request_id == "notification-1"
));
}
@@ -5310,6 +5321,7 @@ mod ws_tests {
let (mut stream, _) = connect_async(request).await.unwrap();
let _ = next_frame(&mut stream).await;
let notification = protocol::Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: "notification-1".into(),
message: "hello".into(),
source: protocol::AuthenticatedInputSource::Account {
crates/worker-runtime/src/worker_backend.rs
+1
View File
@@ -6540,6 +6540,7 @@ where
WorkerExecutionOperation::Input,
worker,
Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: notification_request_id.clone(),
message: input.content,
source: input.source.unwrap_or_else(|| {
crates/worker-runtime/src/worker_backend_notify_transport_tests.rs
+1
View File
@@ -91,6 +91,7 @@ mod notify_transport {
let mut events = lost_handle.subscribe();
lost_handle
.send(Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: "same-client-id".into(),
message: "human-notice-body".into(),
source: account,
crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs
+1
View File
@@ -100,6 +100,7 @@ mod restore_candidate {
let mut events = handle.subscribe();
let method = match source {
Some(source) => Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: NOTICE.into(),
message: NOTICE.into(),
source,
crates/worker/src/controller.rs
+9
View File
@@ -546,6 +546,7 @@ fn durable_parent_notification_target<St: Store + Clone + Send + Sync + 'static>
) -> crate::spawn::tool::ParentNotificationTarget {
crate::spawn::tool::ParentNotificationTarget::Durable(Arc::new(move |method| {
let Method::NotifyTracked {
receipt_scope,
notification_request_id,
message,
source,
@@ -554,6 +555,7 @@ fn durable_parent_notification_target<St: Store + Clone + Send + Sync + 'static>
return;
};
let (source_namespace, provenance) = resolved_input_source(&pending_submissions, &source);
let source_namespace = receipt_scope.namespace(source_namespace);
match pending_submissions.accept_notification_from_source(
notification_request_id,
message,
@@ -2268,12 +2270,14 @@ where
None
}
Method::NotifyTracked {
receipt_scope,
notification_request_id,
message,
source,
} => {
let (source_namespace, provenance) =
resolved_input_source(pending_submissions, &source);
let source_namespace = receipt_scope.namespace(source_namespace);
durably_accept_notification(
notification_request_id,
message,
@@ -2749,12 +2753,14 @@ async fn controller_loop<C, St>(
}
Method::NotifyTracked {
receipt_scope,
notification_request_id,
message,
source,
} => {
let (source_namespace, provenance) =
resolved_input_source(&pending_submissions, &source);
let source_namespace = receipt_scope.namespace(source_namespace);
durably_accept_notification(
notification_request_id,
message,
@@ -3671,12 +3677,14 @@ where
});
}
Some(Method::NotifyTracked {
receipt_scope,
notification_request_id,
message,
source,
}) => {
let (source_namespace, provenance) =
resolved_input_source(pending_submissions, &source);
let source_namespace = receipt_scope.namespace(source_namespace);
if durably_accept_notification(
notification_request_id,
message,
@@ -4824,6 +4832,7 @@ mod tests {
let pending = PendingSubmissionHandle::for_test(dir.path());
let (event_tx, mut event_rx) = broadcast::channel(4);
let method = Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: "completion-1".into(),
message: "subworker completed".into(),
source: protocol::AuthenticatedInputSource::UntrustedWire,
crates/worker/src/spawn/tool.rs
+1
View File
@@ -272,6 +272,7 @@ impl ParentNotificationTarget {
pub(crate) fn notify(&self, child_session_id: String, message: String) {
let method = Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: protocol::new_submission_request_id(),
message,
source: protocol::AuthenticatedInputSource::SubWorker {
crates/worker/tests/compact_events_test.rs
+1
View File
@@ -1935,6 +1935,7 @@ async fn manual_compact_cancel_clears_progress_before_returning_idle() {
let completion_request_id = "subworker-completion-during-compact";
handle
.send(Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: completion_request_id.into(),
message: "subworker completed".into(),
source: protocol::AuthenticatedInputSource::UntrustedWire,
crates/worker/tests/feature_invocation_test.rs
+1
View File
@@ -601,6 +601,7 @@ fn notify_rejects_typed_side_effect_message_payloads() {
message: "/prepare(first)".into(),
},
protocol::Method::NotifyTracked {
receipt_scope: protocol::NotificationReceiptScope::Source,
notification_request_id: "n-1".into(),
message: "/prepare(first)".into(),
source: protocol::AuthenticatedInputSource::UntrustedWire,
crates/workspace-server/src/server.rs
+2 -1
View File
@@ -34788,7 +34788,8 @@ pub(crate) fn authorize_browser_worker_method(
notification_request_id,
message,
} => Ok(protocol::Method::NotifyTracked {
notification_request_id: format!("browser-ws-notify:{notification_request_id}"),
receipt_scope: protocol::NotificationReceiptScope::BrowserConsole,
notification_request_id,
message,
source: source.clone(),
}),
crates/workspace-server/src/server/tests/chat_tests.rs
+240 -1
View File
@@ -777,7 +777,10 @@ fn publish_chat_fixture_profile(api: &WorkspaceApi) {
}
impl ConnectedChat {
async fn new(model: ChatModel, app_layer: impl FnOnce(Router) -> Router) -> Self {
async fn new(
model: impl agen::llm_client::LlmClient + Clone + 'static,
app_layer: impl FnOnce(Router) -> Router,
) -> Self {
const RUNTIME: &str = "connected-chat";
let dir = tempfile::tempdir().unwrap();
let backend_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -1270,3 +1273,239 @@ async fn chat_public_receive_busy_messages_commit_before_next_model_input_and_id
);
assert!(!public.to_string().contains("INTERNAL_ASSISTANT"));
}
#[derive(Clone)]
struct NotifyObservationModel(tokio::sync::mpsc::Sender<String>);
#[async_trait::async_trait]
impl agen::llm_client::LlmClient for NotifyObservationModel {
async fn stream(
&self,
request: agen::llm_client::Request,
) -> std::result::Result<agen::llm_client::ResponseStream, agen::llm_client::ClientError> {
use agen::llm_client::event::{Event, StopReason};
self.0.send(format!("{:?}", request.items)).await.unwrap();
Ok(Box::pin(futures::stream::iter([
Ok(Event::text_block_start(0)),
Ok(Event::text_delta(0, "internal only")),
Ok(Event::text_block_stop(0, Some(StopReason::EndTurn))),
])))
}
fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
Box::new(self.clone())
}
}
/// Real authenticated Backend WS -> signed Runtime WS -> controller -> SQLite,
/// followed by the production Console admission tracker (Deno, the normal test
/// runner prerequisite). Only the model is scripted; no browser rendering claim.
#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
async fn browser_notify_preserves_ack_id_limits_retry_and_console_next_draft() {
let (tx, mut inputs) = tokio::sync::mpsc::channel(32);
let f = ConnectedChat::new(NotifyObservationModel(tx), |router| router).await;
let subject = open_subjektiv_store(&f.api)
.unwrap()
.create_subject(crate::subjektiv::SubjectRole::new("notify-console").unwrap())
.unwrap()
.id;
let operations = [
"session.record",
"memory.resident.read",
"memory.resident.supply",
"journal.startup.read",
"journal.startup.supply",
"chat.read",
"chat.send",
];
f.post(&format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/subjects/{subject}/permissions"),json!({"operation_id":"notify-console-policy","expected_change_id":null,"execution":operations.into_iter().map(|op|subject_permission(&subject,op)).collect::<Vec<_>>(),"delegation":[]})).await;
let conversation = f
.post(
&conversations_path(),
json!({"operation_id":"notify-console-conversation","subject_id":subject}),
)
.await;
let c = conversation["id"].as_str().unwrap();
let created = f.post(&format!("/api/w/{TEST_WORKSPACE_ID}/workers"),json!({"runtime_id":"connected-chat","display_name":"Notify Console","profile":"project:connected-chat","initial_submit":[],"workdir_attachments":[],"feature_connections":{"subjektiv":{"subject_id":subject}}})).await;
let worker = created["worker_id"].as_str().unwrap();
let chat = f
.post(
&messages_path(c),
json!({"operation_id":"notify-console-chat","text":"HTTP_CHAT_RECEIPT"}),
)
.await;
assert_eq!(chat["delivery"]["state"], "accepted");
// Deliberately reuse the exact Runtime ID of the accepted HTTP chat input.
// Its receipt must not reject/consume a different Console message.
let canonical = serde_json::to_vec(&(
format!(
"account:{}",
conversation["participant_account_id"].as_str().unwrap()
),
format!("chat-message:{}", chat["message"]["id"].as_str().unwrap()),
))
.unwrap();
let digest = sha2::Sha256::digest(canonical)
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>();
let collision_id = format!("runtime-input:notify:{digest}");
let url = format!(
"{}/api/w/{TEST_WORKSPACE_ID}/runtimes/connected-chat/workers/{worker}/protocol/ws",
f.url.replacen("http://", "ws://", 1)
);
let mut request = url.into_client_request().unwrap();
request.headers_mut().insert(
"authorization",
format!("Bearer {}", f.owner).parse().unwrap(),
);
let (mut ws, _) = connect_async(request).await.unwrap();
let cases = [
(collision_id, "WS_ISOLATED_FROM_HTTP", true),
("a".repeat(128), "WS_ASCII_LIMIT", true),
("é".repeat(64), "WS_UTF8_LIMIT", true),
("b".repeat(129), "WS_ASCII_OVER_LIMIT", false),
(format!("{}x", "é".repeat(64)), "WS_UTF8_OVER_LIMIT", false),
("console-next-id".into(), "WS_NEXT_DRAFT", true),
];
let mut exchanges = Vec::new();
for (id, text, accepted) in &cases {
let method = protocol::Method::Notify {
notification_request_id: id.clone(),
message: (*text).into(),
};
let ack = console_notify_ack(&mut ws, &method).await;
assert_eq!(
matches!(ack, protocol::Event::NotificationAccepted { .. }),
*accepted,
"{ack:?}"
);
let mut delivered_ack = ack;
if *accepted {
// Lose the first ACK at the Console boundary, resend the exact wire
// input, and deliver the real retry ACK to the actual tracker below.
delivered_ack = console_notify_ack(&mut ws, &method).await;
assert!(matches!(
delivered_ack,
protocol::Event::NotificationAccepted { .. }
));
}
exchanges.push(
json!({"method":method,"event":delivered_ack,"accepted":accepted,"retry":accepted}),
);
if *accepted {
let changed = protocol::Method::Notify {
notification_request_id: id.clone(),
message: format!("CHANGED_{text}"),
};
let rejected = console_notify_ack(&mut ws, &changed).await;
assert!(
matches!(rejected, protocol::Event::NotificationRejected { .. }),
"{rejected:?}"
);
exchanges
.push(json!({"method":changed,"event":rejected,"accepted":false,"retry":false}));
}
}
let mut console = tokio::process::Command::new("deno");
console
.kill_on_drop(true)
.args(["run", "--quiet"])
.arg(
Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../tests/support/console_notify_admissions.ts"),
)
.arg(serde_json::to_string(&exchanges).unwrap());
let output = tokio::time::timeout(std::time::Duration::from_secs(30), console.output())
.await
.expect("Console admission tracker deadline")
.expect("Deno (normal test runner prerequisite)");
assert!(
output.status.success(),
"production Console admission tracker failed: {}{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
tokio::time::timeout(std::time::Duration::from_secs(30), async {
while !inputs
.recv()
.await
.expect("real model input")
.contains("WS_NEXT_DRAFT")
{}
})
.await
.expect("accepted Console input must reach next model request");
let history = session_store::read_retained_session_snapshot(
&f._dir.path().join("runtime/workers").join(worker),
&format!("worker-runtime-{worker}"),
2 * 1024 * 1024,
)
.unwrap();
let notifications = history
.snapshot
.entries
.iter()
.filter_map(|entry| match &entry.data {
protocol::SessionSnapshotEntryData::SystemItem {
item_kind, content, ..
} if item_kind == "notification" => Some(content.as_str()),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
for (_, text, accepted) in cases {
assert_eq!(
notifications.matches(text).count(),
usize::from(accepted),
"one history input per accepted ID, none for rejection: {notifications}"
);
assert!(!notifications.contains(&format!("CHANGED_{text}")));
}
assert_eq!(notifications.matches("HTTP_CHAT_RECEIPT").count(), 1);
}
async fn console_notify_ack(
ws: &mut tokio_tungstenite::WebSocketStream<
tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
>,
method: &protocol::Method,
) -> protocol::Event {
let protocol::Method::Notify {
notification_request_id: expected,
..
} = method
else {
panic!("Console Notify required")
};
ws.send(TungsteniteMessage::Text(
protocol::stream::encode_method(method).unwrap().into(),
))
.await
.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(30), async {
loop {
let frame = ws.next().await.expect("real protocol WS frame").unwrap();
if let TungsteniteMessage::Text(text) = frame {
let event = protocol::stream::decode_event(&text).unwrap();
match &event {
protocol::Event::NotificationAccepted {
notification_request_id,
}
| protocol::Event::NotificationRejected {
notification_request_id,
..
} => {
assert_eq!(
notification_request_id, expected,
"client-visible ACK must preserve original request identity"
);
return event;
}
protocol::Event::Error { .. } => panic!("unexpected protocol error: {event:?}"),
_ => {}
}
}
}
})
.await
.expect("Console Notify ACK deadline")
}
tests/support/console_notify_admissions.ts
+58
View File
@@ -0,0 +1,58 @@
import { ComposerAdmissions } from "../../web/workspace/src/lib/workspace/console/composer-delivery.ts";
import type {
Event,
Method,
} from "../../web/workspace/src/lib/generated/protocol.ts";
// Rustの実WS結合試験で得たACKを既存Consoleの受付管理へ渡す。
// モデル・ブラウザ描画は対象外。Denoは通常テストrunnerと同じ前提。
type Exchange = {
method: Extract<Method, { method: "notify" }>;
event: Event;
accepted: boolean;
retry: boolean;
};
const exchanges: Exchange[] = JSON.parse(Deno.args[0]);
const admissions = new ComposerAdmissions();
for (const exchange of exchanges) {
const { method, event, accepted, retry } = exchange;
const snapshot = {
document: method.params.message,
content: method.params.message,
segments: [{ kind: "text" as const, content: method.params.message }],
pastes: [],
textPastes: [],
};
const record = admissions.begin("real-worker", method, snapshot, "notify");
if (JSON.stringify(record.method) !== JSON.stringify(method)) {
throw new Error("次のdraftが前の受付に置き換わった");
}
if (retry) {
admissions.disconnected("real-worker");
if (admissions.get("real-worker")?.status !== "unknown") {
throw new Error("ACK消失が拒否または成功になった");
}
if (
JSON.stringify(admissions.retry("real-worker")?.method) !==
JSON.stringify(method)
) {
throw new Error("再送で入力またはIDが変わった");
}
}
const result = admissions.acknowledge("real-worker", event);
if (!result || result.accepted !== accepted) {
throw new Error(
`実WSのACKが元のIDへ相関しない: ${JSON.stringify(exchange)}`,
);
}
if (
accepted
? admissions.get("real-worker") !== null
: admissions.get("real-worker")?.status !== "rejected"
) {
throw new Error("ACK後の受付状態が不正");
}
}
console.log(
`Console accepted/rejected/retry/next-draft: ${exchanges.length} real WS exchanges`,
);
web/workspace/src/lib/generated/protocol-validator.ts
+1 -1
View File
File diff suppressed because one or more lines are too long