You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/09/22 13:35:04 UTC
(maka) branch feat/runtime-host-rust updated: fix(runtime): absorb message recovery and Chat image fixes
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch feat/runtime-host-rust
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/feat/runtime-host-rust by this push:
new d2c48826bd fix(runtime): absorb message recovery and Chat image fixes
d2c48826bd is described below
commit d2c48826bd3da3e7223ce398bc752e249fd12093
Author: Wang <m4...@gmail.com>
AuthorDate: Tue Sep 22 21:34:57 2026 +0800
fix(runtime): absorb message recovery and Chat image fixes
Generated-by: Codex
---
.../src/main/__tests__/session-local.test.ts | 57 +++++-
.../src/main/runtime-host-session-observer.ts | 8 +-
apps/desktop/src/main/session-local-service.ts | 42 ++++-
.../workbar/tools/side-chat/use-quote-companion.ts | 12 +-
crates/event-log/src/message_resolution.rs | 12 +-
crates/event-log/tests/message_queue.rs | 3 +
crates/model/src/sdk.rs | 10 +-
crates/model/src/sdk/chat.rs | 66 +++++++
crates/model/tests/integration/provider_stream.rs | 58 ++++++
crates/protocol/src/lib.rs | 2 +-
crates/protocol/src/message/output.rs | 6 +-
crates/protocol/tests/fixtures/operations.mjs | 9 +
crates/protocol/tests/message_contract.rs | 3 +-
crates/protocol/tests/operation_source_contract.rs | 43 +++++
crates/runtime-host/src/execution.rs | 2 +
crates/runtime-host/src/execution/message.rs | 3 +
.../src/execution/message/submissions.rs | 127 +++++++++++++
.../runtime-host/src/execution/plugins/messages.rs | 3 +
.../src/plugins/javascript/provider.rs | 6 +-
crates/runtime-host/src/server/messages.rs | 147 +++++++++++----
packages/runtime-host/src/protocol/index.ts | 3 +-
packages/runtime-host/src/protocol/message.ts | 7 +-
.../runtime-host/src/protocol/model-provider.ts | 199 +++++++++++++++++++++
packages/runtime-host/src/protocol/operations.ts | 3 +
tests/fixtures/client-message-queue.mjs | 2 +
tests/fixtures/client-message-submit.mjs | 12 +-
26 files changed, 786 insertions(+), 59 deletions(-)
diff --git a/apps/desktop/src/main/__tests__/session-local.test.ts b/apps/desktop/src/main/__tests__/session-local.test.ts
index c895e63285..80a111ce6c 100644
--- a/apps/desktop/src/main/__tests__/session-local.test.ts
+++ b/apps/desktop/src/main/__tests__/session-local.test.ts
@@ -30,7 +30,7 @@ import {
RuntimeHostOperationError,
RuntimeHostRequestInterruptedError,
} from '@maka/runtime-host/client';
-import type { TurnMessageSubmitInput, TurnMessageSubmitResult } from '@maka/runtime-host/protocol';
+import type { TurnMessageExecutionResolution, TurnMessageSubmitInput, TurnMessageSubmitResult } from '@maka/runtime-host/protocol';
import { DesktopSessionLocalStore, type LocalMessageIntent } from '../session-local-store.js';
import {
DesktopSessionLocalService,
@@ -85,6 +85,9 @@ async function database(t: TestContext, now?: () => number) {
function client(hostEpoch: string): NonNullable<DesktopSessionLocalTarget['client']> {
return {
hostEpoch,
+ async queryMessageExecutions() {
+ return { resolutions: [] };
+ },
async getSession() {
return null;
},
@@ -401,7 +404,7 @@ test('an unknown Host outcome blocks later local sends until the original messag
assert.deepEqual(calls, ['message-1', 'other-session', 'message-1', 'message-2']);
});
-test('lost ACK recovery never changes epoch or ID and does not block another Session', async (t) => {
+test('lost ACK keeps unknown identities blocked until the new epoch positively resolves them', async (t) => {
const { store, beforeClose } = await database(t);
const calls: TurnMessageSubmitInput[] = [];
let target: DesktopSessionLocalTarget = {
@@ -447,10 +450,58 @@ test('lost ACK recovery never changes epoch or ID and does not block another Ses
await waitFor(() => store.get('authority', 'message-2')?.state === 'accepted');
assert.deepEqual(
calls.filter((call) => call.messageId === 'message-1').map((call) => call.originHostEpoch),
- ['epoch-1', 'epoch-1'],
+ ['epoch-1'],
);
assert.equal(store.get('authority', 'message-1')?.state, 'unknown');
assert.equal(calls.find((call) => call.messageId === 'message-2')?.originHostEpoch, 'epoch-2');
+ store.enqueue('authority', intent('message-3'));
+ target = { ...target, client: {
+ ...client('epoch-2'),
+ async queryMessageExecutions() {
+ return { resolutions: [{ messageId: 'message-1', state: 'not_admitted' }] };
+ },
+ } };
+ service.wake();
+ await waitFor(() => store.get('authority', 'message-3')?.state === 'accepted');
+ assert.equal(store.get('authority', 'message-1')?.state, 'failed');
+ assert.equal(store.get('authority', 'message-1')?.intent.originHostEpoch, 'epoch-1');
+ assert.equal(service.listMessages(target, 'session-1')[0]?.canCancel, true);
+ assert.equal(calls.filter((call) => call.messageId === 'message-1').length, 1);
+});
+
+test('reopened outbox resolves durable ownership without re-submitting or rewriting the dispatch epoch', async (t) => {
+ const resolutions: TurnMessageExecutionResolution[] = [
+ { messageId: 'message-1', state: 'owned', turnId: 'original-turn', runId: 'original-run' },
+ { messageId: 'message-1', state: 'pending' },
+ { messageId: 'message-1', state: 'cancelled' },
+ ];
+ for (const resolution of resolutions) await t.test(resolution.state, async (t) => {
+ const db = await database(t);
+ db.store.enqueue('authority', intent());
+ const record = db.store.get('authority', 'message-1')!;
+ db.store.update({ ...record, state: 'unknown', intent: { ...record.intent, originHostEpoch: 'old' } });
+ const store = db.reopen();
+ const calls: string[] = [];
+ const target: DesktopSessionLocalTarget = {
+ partition: 'authority', profileId: 'profile', scope: { hostId: 'root', targetEpoch: 'target' },
+ client: { ...client('new'), async queryMessageExecutions(input) {
+ assert.deepEqual(input, { sessionId: 'session-1', messageIds: ['message-1'] });
+ return { resolutions: [resolution] };
+ } },
+ submit: async (input) => { calls.push(input.messageId); return accepted; },
+ };
+ const service = new DesktopSessionLocalService(store, {
+ targets: () => [target], changed() {}, onError: (error) => assert.fail(String(error)),
+ });
+ db.beforeClose.push(() => service.close());
+ store.enqueue('authority', intent('next'));
+ service.wake();
+ await waitFor(() => store.get('authority', 'next')?.state === 'accepted');
+ assert.deepEqual(calls, ['next']);
+ assert.equal(store.get('authority', 'message-1')?.state, resolution.state === 'cancelled' ? 'failed' : 'accepted');
+ assert.equal(store.get('authority', 'message-1')?.intent.originHostEpoch, 'old');
+ if (resolution.state === 'owned') assert.equal(service.listMessages(target, 'session-1')[0]?.turnId, 'original-turn');
+ });
});
test('a removed authority cannot be repopulated by an in-flight admission', async (t) => {
diff --git a/apps/desktop/src/main/runtime-host-session-observer.ts b/apps/desktop/src/main/runtime-host-session-observer.ts
index 43bb48d23d..38ad0c82b0 100644
--- a/apps/desktop/src/main/runtime-host-session-observer.ts
+++ b/apps/desktop/src/main/runtime-host-session-observer.ts
@@ -872,13 +872,19 @@ export class RuntimeHostSessionObserver {
const turnId = resolution.state === 'owned'
? resolution.turnId : (next.rootTurn ?? previous.rootTurn)?.turnId;
if (!turnId) continue;
+ const outcome = resolution.state === 'owned'
+ ? 'admitted' as const
+ : resolution.state === 'cancelled' || resolution.state === 'not_admitted'
+ ? 'retracted' as const
+ : undefined;
+ if (!outcome) continue;
this.#broadcast(state.sessionId, {
type: 'message_admission',
id: `host-message-resolution:${next.queue.hostEpoch}:${next.queue.queueRevision}:${resolution.messageId}`,
turnId,
ts: this.#now(),
messageId: resolution.messageId,
- outcome: resolution.state === 'owned' ? 'admitted' : 'retracted',
+ outcome,
});
}
} catch {
diff --git a/apps/desktop/src/main/session-local-service.ts b/apps/desktop/src/main/session-local-service.ts
index 598f03c468..dfc8651b5f 100644
--- a/apps/desktop/src/main/session-local-service.ts
+++ b/apps/desktop/src/main/session-local-service.ts
@@ -63,7 +63,7 @@ export interface DesktopSessionLocalTarget {
readonly profileId: string;
readonly client?: Pick<
DesktopRuntimeHostClient,
- 'hostEpoch' | 'createSession' | 'getSession' | 'listSessions' | 'ingestAttachment'
+ 'hostEpoch' | 'createSession' | 'getSession' | 'listSessions' | 'ingestAttachment' | 'queryMessageExecutions'
>;
readonly submit?: (input: TurnMessageSubmitInput) => Promise<TurnMessageSubmitResult>;
}
@@ -383,6 +383,46 @@ export class DesktopSessionLocalService {
const stillOwned = () =>
this.#current(target) && this.store.get(record.partition, record.messageId) !== undefined;
try {
+ if (record.intent.originHostEpoch !== undefined && record.intent.originHostEpoch !== client.hostEpoch) {
+ const { resolutions } = await client.queryMessageExecutions({
+ sessionId: record.sessionId,
+ messageIds: [record.messageId],
+ }).catch((error: unknown) => {
+ if (error instanceof RuntimeHostOperationError && error.code === 'unauthorized') throw error;
+ throw new Error('Host has not resolved the original message', { cause: error });
+ });
+ if (!stillOwned()) return;
+ const resolution = resolutions.find((entry) => entry.messageId === record.messageId);
+ if (!resolution) {
+ // Omission is unknown, never permission to resend an accepted effect.
+ throw new Error('Host has not resolved the original message');
+ }
+ const current = this.store.get(record.partition, record.messageId)!;
+ switch (resolution.state) {
+ case 'owned':
+ this.store.update({ ...current, state: 'accepted', error: undefined,
+ result: { disposition: 'turn_started', turnId: resolution.turnId, preparation: [] } });
+ break;
+ case 'pending':
+ this.store.update({ ...current, state: 'accepted', error: undefined,
+ result: { disposition: 'followup', preparation: [] } });
+ break;
+ case 'cancelled':
+ case 'not_admitted':
+ this.store.update({ ...current, state: 'failed', result: undefined,
+ error: resolution.state === 'cancelled'
+ ? 'The Host cancelled this message; the local copy is retained.'
+ : 'The Host never admitted this message; the local copy is retained.' });
+ break;
+ }
+ const timer = this.#retries.get(key);
+ if (timer) clearTimeout(timer);
+ this.#retries.delete(key);
+ this.#probed.delete(key);
+ this.#catalogFresh.delete(target.partition);
+ this.deps.changed(target.scope, record.sessionId);
+ return;
+ }
const creation = this.store.creation(target.partition, record.sessionId);
if (creation) {
// session.create already has a durable request fingerprint. Replaying
diff --git a/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts b/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts
index 15e1c57d39..20fc98a1e4 100644
--- a/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts
+++ b/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts
@@ -671,13 +671,13 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan
try {
const { resolutions } = await sideChat.queryMessageExecutions(forkId, [...messageIds]);
if (!mountedRef.current || companionIdRef.current !== forkId) return;
- const cancelled = new Set<string>();
+ const retired = new Set<string>();
const unprovenOwnedTurnIds = new Set<string>();
let ownershipChanged = false;
for (const resolution of resolutions) {
const pending = pendingAdmissionRef.current;
- if (resolution.state === 'cancelled') {
- cancelled.add(resolution.messageId);
+ if (resolution.state === 'cancelled' || resolution.state === 'not_admitted') {
+ retired.add(resolution.messageId);
if (pending?.messageId === resolution.messageId) releaseAdmission(pending);
} else if (resolution.state === 'owned') {
bindPendingMessageTurn(resolution.messageId, resolution.turnId);
@@ -697,7 +697,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan
setHasContent(true);
setOwnTurnTick((tick) => tick + 1);
}
- for (const messageId of cancelled) pendingUserMessagesRef.current.delete(messageId);
+ for (const messageId of retired) pendingUserMessagesRef.current.delete(messageId);
if (unprovenOwnedTurnIds.size > 0) {
const recovered = await Promise.allSettled(
[...unprovenOwnedTurnIds].map((turnId) =>
@@ -711,10 +711,10 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan
}
}
reconcilePendingUserMessages();
- if (cancelled.size > 0) {
+ if (retired.size > 0) {
setMessageQueue((current) => ({
...current,
- entries: current.entries.filter((entry) => !cancelled.has(entry.messageId)),
+ entries: current.entries.filter((entry) => !retired.has(entry.messageId)),
}));
}
} catch {
diff --git a/crates/event-log/src/message_resolution.rs b/crates/event-log/src/message_resolution.rs
index 08273fab38..6e704a5239 100644
--- a/crates/event-log/src/message_resolution.rs
+++ b/crates/event-log/src/message_resolution.rs
@@ -28,6 +28,11 @@ pub use owner::MessageExecution;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum MessageResolution {
+ /// No durable message evidence in this existing Session's SQL snapshot.
+ /// Host must also exclude in-flight preparation before asserting non-admission.
+ Absent {
+ message_id: String,
+ },
Pending {
message_id: String,
},
@@ -61,6 +66,9 @@ impl EventLog {
self.connection.run(move |connection| Box::pin(async move {
let mut tx = connection.begin().await?;
let mut resolutions = Vec::new();
+ let exists: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM session_control WHERE id = ?)")
+ .bind(&session).fetch_one(&mut *tx).await?;
+ if !exists { return Ok(resolutions); }
for message_id in messages {
let owner: Option<String> = sqlx::query_scalar(
"SELECT json_extract(e.event_json, '$.invocation')
@@ -81,8 +89,8 @@ impl EventLog {
let pending: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM message_admissions WHERE session_id = ? AND message_id = ?)"
).bind(&session).bind(&message_id).fetch_one(&mut *tx).await?;
- if !pending { continue; }
- MessageResolution::Pending { message_id }
+ if pending { MessageResolution::Pending { message_id } }
+ else { MessageResolution::Absent { message_id } }
};
resolutions.push(resolution);
}
diff --git a/crates/event-log/tests/message_queue.rs b/crates/event-log/tests/message_queue.rs
index f732ebd807..d6958a2c30 100644
--- a/crates/event-log/tests/message_queue.rs
+++ b/crates/event-log/tests/message_queue.rs
@@ -278,6 +278,9 @@ async fn queue_edits_cancel_or_deliver_once_with_atomic_revision_and_original_ow
MessageResolution::Cancelled {
message_id: "three".into()
},
+ MessageResolution::Absent {
+ message_id: "absent".into()
+ },
]
);
assert!(
diff --git a/crates/model/src/sdk.rs b/crates/model/src/sdk.rs
index eea36bb943..a8874b760a 100644
--- a/crates/model/src/sdk.rs
+++ b/crates/model/src/sdk.rs
@@ -28,6 +28,8 @@ use serde_json::Value;
use std::sync::Arc;
use tokio::sync::mpsc;
+mod chat;
+
fn serialize_tools<S: serde::Serializer>(
tools: &[ToolDefinition],
serializer: S,
@@ -54,7 +56,7 @@ fn serialize_tools<S: serde::Serializer>(
pub(super) async fn stream(
runtime: TrustedRuntime,
- request: ModelRequest,
+ mut request: ModelRequest,
context: Context,
) -> Result<(), ModelError> {
let Context {
@@ -65,6 +67,12 @@ pub(super) async fn stream(
} = context;
#[derive(Serialize)]
struct Tools<'a>(#[serde(serialize_with = "serialize_tools")] &'a [ToolDefinition]);
+ if matches!(
+ request.provider.kind,
+ crate::ProviderKind::OpenaiChat | crate::ProviderKind::OpenaiCompatible { .. }
+ ) {
+ request.prompt = chat::project(request.prompt);
+ }
let tools = serde_json::to_value(Tools(&request.tools))
.map_err(|error| ModelError::Adapter(error.to_string()))?;
let mut request =
diff --git a/crates/model/src/sdk/chat.rs b/crates/model/src/sdk/chat.rs
new file mode 100644
index 0000000000..cae78b09d7
--- /dev/null
+++ b/crates/model/src/sdk/chat.rs
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::prompt::{ContentPart, Message, ToolOutput};
+
+/// Chat has text-only tool messages. Keep the entire result group together and
+/// carry its images in the following user message, never between parallel results.
+pub(super) fn project(messages: Vec<Message>) -> Vec<Message> {
+ let mut projected = Vec::with_capacity(messages.len());
+ let mut images = Vec::new();
+ for mut message in messages {
+ if let Message::Tool { content, .. } = &mut message {
+ for result in content {
+ let ToolOutput::Content(parts) = &mut result.output else {
+ continue;
+ };
+ let mut labelled = false;
+ for part in parts {
+ if matches!(part, ContentPart::File { media_type, .. } if media_type.starts_with("image/"))
+ {
+ if !labelled {
+ images.push(ContentPart::text(format!(
+ "Image from tool {} ({}):",
+ result.tool_name, result.tool_call_id
+ )));
+ labelled = true;
+ }
+ images.push(std::mem::replace(
+ part,
+ ContentPart::text("Image supplied below."),
+ ));
+ }
+ }
+ }
+ } else if !images.is_empty() {
+ projected.push(Message::User {
+ content: std::mem::take(&mut images),
+ provider_options: None,
+ });
+ }
+ projected.push(message);
+ }
+ if !images.is_empty() {
+ projected.push(Message::User {
+ content: images,
+ provider_options: None,
+ });
+ }
+ projected
+}
diff --git a/crates/model/tests/integration/provider_stream.rs b/crates/model/tests/integration/provider_stream.rs
index 9ee1ffd3fc..13b0719fed 100644
--- a/crates/model/tests/integration/provider_stream.rs
+++ b/crates/model/tests/integration/provider_stream.rs
@@ -212,6 +212,64 @@ assert_eq!(step.finish_reason, maka_runtime::model::ModelFinishReason::ToolCalls
}).await.expect("provider stream must make bounded progress");
}
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn chat_tool_images_follow_the_complete_parallel_result_group_on_the_wire() {
+ use maka_model::prompt::{AssistantPart, ContentPart, FileData, Message, ToolOutput};
+
+ tokio::time::timeout(Duration::from_secs(30), async {
+ let executor = ModelExecutor::new(1, Duration::from_secs(10)).unwrap();
+ for kind in [ProviderKind::OpenaiChat, ProviderKind::OpenaiCompatible { name: "fixture".into() }] {
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let base = format!("http://{}/v1", listener.local_addr().unwrap());
+ let server = tokio::spawn(async move {
+ let (mut socket, _) = listener.accept().await.unwrap();
+ let wire = read_request(&mut socket).await;
+ let (first, last) = fixtures(ProviderKind::OpenaiChat, "seen");
+ let body = first + &last;
+ socket.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap();
+ serde_json::from_str::<Value>(wire.split_once("\r\n\r\n").unwrap().1).unwrap()
+ });
+ let append_user = matches!(kind, ProviderKind::OpenaiChat);
+ let mut input = request(kind, base);
+ input.prompt.push(Message::Assistant {
+ content: ["first", "second"].into_iter().map(|id| AssistantPart::ToolCall {
+ tool_call_id: id.into(), tool_name: "echo".into(), input: json!({}),
+ provider_executed: None, provider_options: None,
+ }).collect(),
+ provider_options: None,
+ });
+ for id in ["first", "second"] {
+ input.prompt.push(Message::tool(id, "echo", ToolOutput::Content(vec![
+ ContentPart::text(format!("result-{id}")),
+ ContentPart::File { data: FileData::Data("aW1hZ2U=".into()), media_type: "image/png".into(), provider_options: None },
+ ])));
+ }
+ if append_user { input.prompt.push(Message::user("compare these images")); }
+ let mut stream = executor.stream(input, CancellationToken::new()).await.unwrap();
+ while let Some(event) = stream.next().await { event.unwrap(); }
+ let body = server.await.unwrap();
+ let messages = body["messages"].as_array().unwrap();
+ let mut roles = vec!["user", "assistant", "tool", "tool", "user"];
+ if append_user { roles.push("user"); }
+ assert_eq!(messages.iter().map(|message| message["role"].as_str().unwrap()).collect::<Vec<_>>(), roles);
+ for (index, id) in [(2, "first"), (3, "second")] {
+ assert_eq!(messages[index]["tool_call_id"], id);
+ let text = messages[index]["content"].as_str().unwrap();
+ assert!(text.contains(&format!("result-{id}")));
+ assert!(!text.contains("aW1hZ2U="));
+ assert!(!text.contains("image/png"));
+ }
+ let images = messages[4]["content"].as_array().unwrap();
+ assert_eq!(images.len(), 4);
+ for (index, id) in [(0, "first"), (2, "second")] {
+ assert_eq!(images[index]["text"], format!("Image from tool echo ({id}):"));
+ assert_eq!(images[index + 1]["type"], "image_url");
+ assert_eq!(images[index + 1]["image_url"]["url"], "data:image/png;base64,aW1hZ2U=");
+ }
+ }
+ }).await.expect("Chat image projection must make bounded progress");
+}
+
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn malformed_provider_tool_identity_cannot_complete_a_model_step() {
tokio::time::timeout(Duration::from_secs(30), async {
diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs
index 04416e2d50..2d23790e97 100644
--- a/crates/protocol/src/lib.rs
+++ b/crates/protocol/src/lib.rs
@@ -58,7 +58,7 @@ use serde::Serialize;
use serde_json::Value;
pub const PROTOCOL_VERSION: u64 = 0;
-pub const COMPATIBILITY_EPOCH: u64 = 167;
+pub const COMPATIBILITY_EPOCH: u64 = 177;
pub const COMPOSITION_ID: &str = "maka.interactive";
pub const MAX_MESSAGE_BYTES: usize = 768 * 1024;
pub const MAX_IN_FLIGHT_DOMAIN_REQUESTS: usize = 64;
diff --git a/crates/protocol/src/message/output.rs b/crates/protocol/src/message/output.rs
index d64042ce58..0ca93e44ed 100644
--- a/crates/protocol/src/message/output.rs
+++ b/crates/protocol/src/message/output.rs
@@ -85,6 +85,9 @@ pub struct ExecutionQueryResult {
deny_unknown_fields
)]
pub enum ExecutionResolution {
+ NotAdmitted {
+ message_id: String,
+ },
Pending {
message_id: String,
},
@@ -159,7 +162,8 @@ pub fn decode_output(operation: Operation, value: &Value) -> Result<Output> {
let mut seen = HashSet::new();
for resolution in &output.resolutions {
let message = match resolution {
- ExecutionResolution::Pending { message_id }
+ ExecutionResolution::NotAdmitted { message_id }
+ | ExecutionResolution::Pending { message_id }
| ExecutionResolution::Cancelled { message_id } => message_id,
ExecutionResolution::Owned {
message_id,
diff --git a/crates/protocol/tests/fixtures/operations.mjs b/crates/protocol/tests/fixtures/operations.mjs
index 4289178968..830bb1405e 100644
--- a/crates/protocol/tests/fixtures/operations.mjs
+++ b/crates/protocol/tests/fixtures/operations.mjs
@@ -44,6 +44,15 @@ assert.deepEqual(
'Rust Host operation contracts differ from current source',
);
assert.equal(actual.epoch, RUNTIME_HOST_COMPATIBILITY_EPOCH);
+const providerCatalog = HOST_OPERATION_SPECS['model.provider.catalog.query'];
+for (const page of actual.providerPages) {
+ assert.deepEqual(providerCatalog.decodeOutput(page), page);
+}
+const page = actual.providerPages[0];
+assert.throws(() => providerCatalog.decodeOutput({ ...page, next: 'unrelated' }));
+assert.throws(() =>
+ providerCatalog.decodeOutput({ ...page, entries: [page.entries[0], page.entries[0]] }),
+);
for (const { operation, input, decoded } of actual.targets) {
const decode = () => HOST_OPERATION_SPECS[operation].decodeInput(input);
if (decoded === null) assert.throws(decode, JSON.stringify({ operation, input }));
diff --git a/crates/protocol/tests/message_contract.rs b/crates/protocol/tests/message_contract.rs
index 510077644f..493f4b19b1 100644
--- a/crates/protocol/tests/message_contract.rs
+++ b/crates/protocol/tests/message_contract.rs
@@ -73,7 +73,7 @@ fn original_message_codec_agrees_on_identity_intent_queue_states_and_bounded_pro
(TurnMessageQuery, json!({"cancelledMessageIds":["m"]})),
(
TurnMessageExecutionQuery,
- json!({"resolutions":[{"state":"pending","messageId":"a"},{"state":"cancelled","messageId":"b"},{"state":"owned","messageId":"c","turnId":"t","runId":"r"}]}),
+ json!({"resolutions":[{"state":"pending","messageId":"a"},{"state":"cancelled","messageId":"b"},{"state":"owned","messageId":"c","turnId":"t","runId":"r"},{"state":"not_admitted","messageId":"d"}]}),
),
(
QueueRetract,
@@ -109,6 +109,7 @@ fn original_message_codec_agrees_on_identity_intent_queue_states_and_bounded_pro
for resolution in [
json!({"state":"unknown","messageId":"m"}),
json!({"state":"pending","messageId":"m","turnId":"t"}),
+ json!({"state":"not_admitted","messageId":"m","runId":"r"}),
json!({"state":"owned","messageId":"m","turnId":"t"}),
] {
add(
diff --git a/crates/protocol/tests/operation_source_contract.rs b/crates/protocol/tests/operation_source_contract.rs
index 54602db05d..bbf3135c13 100644
--- a/crates/protocol/tests/operation_source_contract.rs
+++ b/crates/protocol/tests/operation_source_contract.rs
@@ -66,6 +66,7 @@ fn operation_inventory_and_execution_targets_match_current_typescript() {
"epoch": maka_protocol::COMPATIBILITY_EPOCH,
"operations": operations,
"targets": execution_targets(),
+ "providerPages": provider_pages(),
}))
.unwrap(),
)
@@ -83,6 +84,48 @@ fn operation_inventory_and_execution_targets_match_current_typescript() {
assert_eq!(report["operationCount"], Operation::ALL.len());
}
+fn provider_pages() -> Vec<Value> {
+ use maka_plugins::provider::{
+ Descriptor, Identity,
+ authentication::Method,
+ catalog::{Entry, Page},
+ };
+ let descriptor = Descriptor {
+ label: "模型账户".into(),
+ configuration_schema: json!({"type":"object"}),
+ configuration_defaults: json!({}),
+ authentication: vec![Method {
+ id: "key".into(),
+ label: "API key".into(),
+ input_schema: json!({"type":"object"}),
+ interactive: false,
+ }],
+ discovery: true,
+ };
+ descriptor.validate().unwrap();
+ let identity = Identity {
+ package_id: "example.provider".into(),
+ entry_id: "example.provider".into(),
+ scope: maka_runtime::scope::Scope::Profile,
+ name: "provider".into(),
+ };
+ identity.validate().unwrap();
+ [
+ Page::Page {
+ revision: 7,
+ entries: vec![Entry {
+ identity,
+ descriptor,
+ }],
+ next: Some("provider".into()),
+ },
+ Page::RevisionChanged { revision: 8 },
+ ]
+ .into_iter()
+ .map(|page| serde_json::to_value(page).unwrap())
+ .collect()
+}
+
fn execution_targets() -> Vec<Value> {
let mut cases = Vec::new();
for target in [
diff --git a/crates/runtime-host/src/execution.rs b/crates/runtime-host/src/execution.rs
index 6579cf10aa..037573c066 100644
--- a/crates/runtime-host/src/execution.rs
+++ b/crates/runtime-host/src/execution.rs
@@ -77,6 +77,7 @@ pub(crate) struct Executions {
// never the model, effect, or human approval lifetime.
interactions: Arc<crate::server::interactions::Interactions>,
active: Mutex<HashMap<String, ActiveRun>>,
+ submissions: message::Submissions,
plugin_processes: Arc<Mutex<HashMap<String, usize>>>,
workers: TaskTracker,
handoff_wake: tokio::sync::Notify,
@@ -117,6 +118,7 @@ impl Executions {
.with_catalog(plugin_catalog.clone());
Ok(Self {
plugin_catalog,
+ submissions: Default::default(),
plugin_calls,
oauth: crate::oauth::Authority::new(workers.clone(), shutdown.clone()),
controllers: Default::default(),
diff --git a/crates/runtime-host/src/execution/message.rs b/crates/runtime-host/src/execution/message.rs
index 55013d188a..3a8076778b 100644
--- a/crates/runtime-host/src/execution/message.rs
+++ b/crates/runtime-host/src/execution/message.rs
@@ -30,6 +30,8 @@ use maka_runtime::{
use uuid::Uuid;
mod queued;
+mod submissions;
+pub(super) use submissions::Submissions;
impl Executions {
pub(crate) async fn submit(
@@ -39,6 +41,7 @@ impl Executions {
root_id: &str,
epoch: &str,
) -> Result<SubmitResult> {
+ let _submission = self.submissions.track(&input.session_id, &input.message_id);
self.ordinary_session(&input.session_id).await?;
let mut prepared = None;
let mut queued: Option<(
diff --git a/crates/runtime-host/src/execution/message/submissions.rs b/crates/runtime-host/src/execution/message/submissions.rs
new file mode 100644
index 0000000000..69fddd4ff1
--- /dev/null
+++ b/crates/runtime-host/src/execution/message/submissions.rs
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use super::super::{Executions, Result};
+use maka_event_log::message_resolution::MessageResolution;
+use maka_protocol::message::{ExecutionResolution, QueryInput};
+use std::{collections::HashMap, sync::Mutex};
+
+/// Input preparation releases the admission gate. Keep its identities visible
+/// until the submit future settles, including concurrent retries of one message.
+#[derive(Default)]
+pub(in crate::execution) struct Submissions(Mutex<HashMap<(String, String), usize>>);
+
+impl Submissions {
+ pub(in crate::execution) fn track<'a>(
+ &'a self,
+ session: &str,
+ message: &str,
+ ) -> Submission<'a> {
+ let key = (session.to_owned(), message.to_owned());
+ *self.0.lock().unwrap().entry(key.clone()).or_default() += 1;
+ Submission {
+ submissions: self,
+ key,
+ }
+ }
+
+ fn contains(&self, session: &str, message: &str) -> bool {
+ self.0
+ .lock()
+ .unwrap()
+ .contains_key(&(session.to_owned(), message.to_owned()))
+ }
+}
+
+pub(in crate::execution) struct Submission<'a> {
+ submissions: &'a Submissions,
+ key: (String, String),
+}
+
+impl Drop for Submission<'_> {
+ fn drop(&mut self) {
+ let mut entries = self.submissions.0.lock().unwrap();
+ let count = entries.get_mut(&self.key).expect("live submission");
+ *count -= 1;
+ if *count == 0 {
+ entries.remove(&self.key);
+ }
+ }
+}
+
+impl Executions {
+ pub(crate) async fn query_message_executions(
+ &self,
+ input: QueryInput,
+ ) -> Result<Vec<ExecutionResolution>> {
+ // Plugin submissions transfer this same gate to their durable worker.
+ // SQL absence alone cannot exclude a not-yet-committed admission.
+ let _gate = self.lock_admission().await;
+ let stored = self
+ .log
+ .message_resolutions(&input.session_id, &input.message_ids)
+ .await
+ .map_err(|error| self.message_storage_error(error))?;
+ Ok(stored
+ .into_iter()
+ .filter_map(|resolution| match resolution {
+ MessageResolution::Absent { message_id } => {
+ (!self.submissions.contains(&input.session_id, &message_id))
+ .then_some(ExecutionResolution::NotAdmitted { message_id })
+ }
+ MessageResolution::Pending { message_id } => {
+ Some(ExecutionResolution::Pending { message_id })
+ }
+ MessageResolution::Cancelled { message_id } => {
+ Some(ExecutionResolution::Cancelled { message_id })
+ }
+ MessageResolution::Owned {
+ message_id,
+ invocation,
+ } => Some(ExecutionResolution::Owned {
+ message_id,
+ turn_id: invocation.turn_id,
+ run_id: invocation.run_id,
+ }),
+ })
+ .collect())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::Submissions;
+
+ #[tokio::test]
+ async fn cancelling_one_duplicate_keeps_the_other_preparation_visible() {
+ let submissions = Submissions::default();
+ let original = submissions.track("session", "message");
+ let duplicate = async {
+ let _duplicate = submissions.track("session", "message");
+ std::future::pending::<()>().await;
+ };
+ let mut duplicate = Box::pin(duplicate);
+ assert!(futures_util::poll!(&mut duplicate).is_pending());
+ drop(original);
+ assert!(submissions.contains("session", "message"));
+ assert!(!submissions.contains("other-session", "message"));
+ drop(duplicate);
+ assert!(!submissions.contains("session", "message"));
+ }
+}
diff --git a/crates/runtime-host/src/execution/plugins/messages.rs b/crates/runtime-host/src/execution/plugins/messages.rs
index 17ff92ecea..e2416c90f8 100644
--- a/crates/runtime-host/src/execution/plugins/messages.rs
+++ b/crates/runtime-host/src/execution/plugins/messages.rs
@@ -55,6 +55,9 @@ impl BoundCommands {
.digest()
.map_err(|e| Error::Invalid(e.to_string()))?;
let host = self.executions()?;
+ let _submission = host
+ .submissions
+ .track(&request.invocation.session_id, &request.message_id);
let mut prepared = None;
loop {
let gate = host.interactions.own_admission().await;
diff --git a/crates/runtime-host/src/plugins/javascript/provider.rs b/crates/runtime-host/src/plugins/javascript/provider.rs
index 61460db604..9164c86b95 100644
--- a/crates/runtime-host/src/plugins/javascript/provider.rs
+++ b/crates/runtime-host/src/plugins/javascript/provider.rs
@@ -47,7 +47,7 @@ pub(super) struct JavaScript {
rename_all_fields = "camelCase"
)]
enum Operation {
- Resolve(Resolve),
+ Resolve(Box<Resolve>),
Authorize {
connection: Connection,
credential: Option<Credential>,
@@ -135,7 +135,9 @@ impl JavaScript {
impl Provider for JavaScript {
fn resolve(&self, request: Resolve) -> BoxFuture<'_, Result<Model, Error>> {
Box::pin(async move {
- let model: Model = self.invoke(Operation::Resolve(request), None).await?;
+ let model: Model = self
+ .invoke(Operation::Resolve(Box::new(request)), None)
+ .await?;
model.validate()?;
Ok(model)
})
diff --git a/crates/runtime-host/src/server/messages.rs b/crates/runtime-host/src/server/messages.rs
index a973a72fbc..3b2fd00e26 100644
--- a/crates/runtime-host/src/server/messages.rs
+++ b/crates/runtime-host/src/server/messages.rs
@@ -71,11 +71,16 @@ pub(super) async fn execute(
.submit(*input, connection_id, host.root_id(), &host.epoch)
.await
.map(Output::Submit),
- Input::Query(input) | Input::ExecutionQuery(input) => host
+ Input::ExecutionQuery(input) => host
+ .executions
+ .query_message_executions(input)
+ .await
+ .map(|resolutions| Output::Executions(ExecutionQueryResult { resolutions })),
+ Input::Query(input) => host
.log
.message_resolutions(&input.session_id, &input.message_ids)
.await
- .map(|resolutions| query_result(operation, resolutions))
+ .map(query_result)
.map_err(|error| storage_error(host, error)),
input => mutations::execute(host, connection_id, operation, input).await,
};
@@ -88,39 +93,16 @@ pub(super) async fn execute(
Err(error) => Outcome::failure(error),
})
}
-fn query_result(operation: Operation, resolutions: Vec<MessageResolution>) -> Output {
- if operation == Operation::TurnMessageQuery {
- return Output::Query(QueryResult {
- cancelled_message_ids: resolutions
- .into_iter()
- .filter_map(|r| {
- if let MessageResolution::Cancelled { message_id } = r {
- Some(message_id)
- } else {
- None
- }
- })
- .collect(),
- });
- }
- Output::Executions(ExecutionQueryResult {
- resolutions: resolutions
+fn query_result(resolutions: Vec<MessageResolution>) -> Output {
+ Output::Query(QueryResult {
+ cancelled_message_ids: resolutions
.into_iter()
- .map(|r| match r {
- MessageResolution::Pending { message_id } => {
- ExecutionResolution::Pending { message_id }
+ .filter_map(|r| {
+ if let MessageResolution::Cancelled { message_id } = r {
+ Some(message_id)
+ } else {
+ None
}
- MessageResolution::Cancelled { message_id } => {
- ExecutionResolution::Cancelled { message_id }
- }
- MessageResolution::Owned {
- message_id,
- invocation,
- } => ExecutionResolution::Owned {
- message_id,
- turn_id: invocation.turn_id,
- run_id: invocation.run_id,
- },
})
.collect(),
})
@@ -134,6 +116,7 @@ fn failure(code: Code, message: &str) -> OperationError {
message: message.chars().take(1024).collect(),
}
}
+
fn fingerprint(input: &Input) -> Result<String, OperationError> {
Ok(format!(
"sha256:{:x}",
@@ -143,3 +126,101 @@ fn fingerprint(input: &Input) -> Result<String, OperationError> {
)
))
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use maka_event_log::{
+ message_admissions::PendingMessageAdmission,
+ root::{RootNamespaces, RootOwner},
+ };
+ use maka_runtime::{
+ event::Invocation,
+ input::DeliveredMessage,
+ message::{MessageDisposition, RootSourceMessage},
+ };
+
+ #[tokio::test]
+ async fn non_admission_query_waits_for_the_shared_writer_and_observes_its_commit() {
+ let directory = tempfile::tempdir().unwrap();
+ let namespaces = RootNamespaces {
+ ownership: directory.path().join("owners"),
+ control: directory.path().join("control"),
+ };
+ let host =
+ Host::open(RootOwner::create(&directory.path().join("root"), &namespaces).unwrap())
+ .await
+ .unwrap();
+ let cwd = directory
+ .path()
+ .canonicalize()
+ .unwrap()
+ .to_string_lossy()
+ .into_owned();
+ let configuration = crate::session::PreparedSession::new(serde_json::from_value(serde_json::json!({
+ "sessionId":"session", "workspace":{"kind":"host_path","path":cwd}, "executorId":"fixture"
+ })).unwrap()).unwrap().bind(
+ maka_protocol::session::WorkspaceProjection {
+ target: maka_protocol::session::WorkspaceTarget::HostPath { path: cwd.clone() },
+ host_cwd: cwd,
+ },
+ crate::session::SessionTarget::Executor { executor_id: "fixture".to_owned().try_into().unwrap() },
+ maka_protocol::session::SandboxMode::ReadOnly,
+ );
+ host.log
+ .create_session("session", "create", &configuration, 1)
+ .await
+ .unwrap();
+ let query = QueryInput {
+ session_id: "session".into(),
+ message_ids: vec!["message".into(), "absent".into()],
+ };
+ let gate = host.executions.lock_admission().await;
+ let mut reader = Box::pin(host.executions.query_message_executions(query));
+ assert!(futures_util::poll!(&mut reader).is_pending());
+ let content: maka_runtime::input::MessageInput = "queued input".into();
+ host.log
+ .admit_message(PendingMessageAdmission {
+ invocation: Invocation {
+ session_id: "session".into(),
+ turn_id: "turn".into(),
+ run_id: "run".into(),
+ invocation_id: "invocation".into(),
+ },
+ steering_invocation: None,
+ source: RootSourceMessage {
+ message: DeliveredMessage {
+ message_id: "message".into(),
+ submitted_content_digest: content.content_digest().unwrap(),
+ content,
+ },
+ submitted_placement: Placement::NextTurn,
+ disposition: MessageDisposition::TurnStarted,
+ submitted_intent: None,
+ },
+ required_tools: Default::default(),
+ admitted_at: 2,
+ })
+ .await
+ .unwrap();
+ drop(gate);
+ assert_eq!(
+ reader.await.unwrap(),
+ vec![
+ ExecutionResolution::Pending {
+ message_id: "message".into()
+ },
+ ExecutionResolution::NotAdmitted {
+ message_id: "absent".into()
+ },
+ ]
+ );
+ host.draining.cancel();
+ host.plugin_tasks.close();
+ host.plugin_tasks.wait().await;
+ host.executions.shutdown().await;
+ host.shells.shutdown().await;
+ host.log.shutdown().await.unwrap();
+ host.configuration.shutdown().await.unwrap();
+ }
+}
diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts
index 2ba88cd103..9c9a9c0270 100644
--- a/packages/runtime-host/src/protocol/index.ts
+++ b/packages/runtime-host/src/protocol/index.ts
@@ -109,7 +109,8 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const;
export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const;
// Increment when the same protocol version no longer guarantees safe Client-Host
// interoperability. Mismatches are rejected before domain commands are admitted.
-export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 167 as const;
+// Rust preview: positive message non-admission; distinct from main's epochs through 176.
+export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 177 as const;
export const RUNTIME_HOST_MAX_MESSAGE_BYTES = 768 * 1024;
export const RUNTIME_HOST_MAX_IN_FLIGHT_DOMAIN_REQUESTS = 64;
export const INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID = 'maka.interactive' as const;
diff --git a/packages/runtime-host/src/protocol/message.ts b/packages/runtime-host/src/protocol/message.ts
index c2ce9ca48f..4b85ce0bbe 100644
--- a/packages/runtime-host/src/protocol/message.ts
+++ b/packages/runtime-host/src/protocol/message.ts
@@ -138,6 +138,7 @@ export interface TurnMessageExecutionQueryResult {
}
export type TurnMessageExecutionResolution =
+ | { readonly messageId: string; readonly state: 'not_admitted' }
| { readonly messageId: string; readonly state: 'pending' }
| { readonly messageId: string; readonly state: 'cancelled' }
| {
@@ -396,14 +397,14 @@ function decodeTurnMessageExecutionQueryResult(value: unknown): TurnMessageExecu
state: 'pending',
};
}
- if (resolution.state === 'cancelled') {
- assertExactKeys(resolution, 'turn.message.execution.query cancelled resolution', [
+ if (resolution.state === 'cancelled' || resolution.state === 'not_admitted') {
+ assertExactKeys(resolution, 'turn.message.execution.query terminal resolution', [
'messageId',
'state',
]);
return {
messageId: requireEntityId(resolution.messageId, 'messageId'),
- state: 'cancelled',
+ state: resolution.state,
};
}
if (resolution.state === 'owned') {
diff --git a/packages/runtime-host/src/protocol/model-provider.ts b/packages/runtime-host/src/protocol/model-provider.ts
new file mode 100644
index 0000000000..52e7942bbc
--- /dev/null
+++ b/packages/runtime-host/src/protocol/model-provider.ts
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import type { Json, ProviderDescriptor, ProviderIdentity } from '@maka-agent/plugin-sdk/host';
+import {
+ requireCount,
+ requireEncodedByteLimit,
+ requireExactRecord,
+ requireRecord,
+ requireShapedRecord,
+ requireUtf8String,
+} from './codec.js';
+import { invalidProtocolFrame } from './errors.js';
+import { defineOperation } from './operation-spec.js';
+
+export interface ModelProviderCatalogQuery {
+ scope?: 'profile' | 'desktop-ui' | `session:${string}`;
+ after?: string | null;
+ revision?: number | null;
+}
+export type ModelProviderCatalogResult =
+ | { kind: 'revision_changed'; revision: number }
+ | {
+ kind: 'page';
+ revision: number;
+ entries: { identity: ProviderIdentity; descriptor: ProviderDescriptor }[];
+ next: string | null;
+ };
+
+function identifier(value: unknown, max = 256): string {
+ const text = requireUtf8String(value, 'provider identifier', max);
+ if (!text || /[\p{White_Space}\p{Cc}]/u.test(text))
+ throw invalidProtocolFrame('Invalid provider identifier');
+ return text;
+}
+function scope(value: unknown): NonNullable<ModelProviderCatalogQuery['scope']> {
+ if (value === 'profile' || value === 'desktop-ui') return value;
+ if (typeof value === 'string' && value.startsWith('session:')) {
+ return `session:${identifier(value.slice(8))}`;
+ }
+ throw invalidProtocolFrame('Invalid provider scope');
+}
+function flag(value: unknown): boolean {
+ if (typeof value !== 'boolean') throw invalidProtocolFrame('Invalid provider flag');
+ return value;
+}
+function label(value: unknown, controls = false): string {
+ const text = requireUtf8String(value, 'provider label', 256);
+ if (!/[^\p{White_Space}]/u.test(text) || (!controls && /[\p{Cc}]/u.test(text)))
+ throw invalidProtocolFrame('Invalid provider label');
+ return text;
+}
+function jsonObject(value: unknown): { [key: string]: Json } {
+ return requireRecord(value, 'provider JSON object') as { [key: string]: Json };
+}
+function descriptor(value: unknown): ProviderDescriptor {
+ requireEncodedByteLimit(value, 'provider descriptor', 64 * 1024);
+ const row = requireExactRecord(value, 'provider descriptor', [
+ 'label',
+ 'configurationSchema',
+ 'configurationDefaults',
+ 'authentication',
+ 'discovery',
+ ]);
+ if (!Array.isArray(row.authentication) || row.authentication.length > 16) {
+ throw invalidProtocolFrame('Invalid provider authentication methods');
+ }
+ const ids = new Set<string>();
+ const authentication = row.authentication.map((value) => {
+ const method = requireExactRecord(value, 'provider authentication method', [
+ 'id',
+ 'label',
+ 'inputSchema',
+ 'interactive',
+ ]);
+ const id = identifier(method.id);
+ if (ids.has(id)) throw invalidProtocolFrame('Duplicate provider authentication method');
+ ids.add(id);
+ return {
+ id,
+ label: label(method.label, true),
+ inputSchema: jsonObject(method.inputSchema),
+ interactive: flag(method.interactive),
+ };
+ });
+ return {
+ label: label(row.label),
+ configurationSchema: jsonObject(row.configurationSchema),
+ configurationDefaults: jsonObject(row.configurationDefaults),
+ authentication,
+ discovery: flag(row.discovery),
+ };
+}
+function identity(value: unknown): ProviderIdentity {
+ const row = requireExactRecord(value, 'provider identity', [
+ 'packageId',
+ 'entryId',
+ 'scope',
+ 'name',
+ ]);
+ const owner = (value: unknown) => {
+ const name = identifier(value, 128);
+ if (!/^[a-z][a-z0-9]*(?:[._:-][a-z0-9]+)*$/.test(name))
+ throw invalidProtocolFrame('Invalid provider owner');
+ return name;
+ };
+ const selected = scope(row.scope);
+ if (selected === 'desktop-ui') throw invalidProtocolFrame('Client scope cannot provide models');
+ return {
+ packageId: owner(row.packageId),
+ entryId: owner(row.entryId),
+ scope: selected,
+ name: identifier(row.name),
+ };
+}
+
+export const MODEL_PROVIDER_OPERATION_SPECS = {
+ 'model.provider.catalog.query': defineOperation<
+ ModelProviderCatalogQuery,
+ ModelProviderCatalogResult,
+ | 'host_not_ready'
+ | 'host_draining'
+ | 'invalid_request'
+ | 'operation_unavailable'
+ | 'internal_failure'
+ | 'persistence_failed'
+ >({
+ mode: 'query',
+ availability: 'ready',
+ errors: [
+ 'host_not_ready',
+ 'host_draining',
+ 'invalid_request',
+ 'operation_unavailable',
+ 'internal_failure',
+ 'persistence_failed',
+ ],
+ decodeInput(value) {
+ const row = requireShapedRecord(
+ value,
+ 'provider catalog query',
+ [],
+ ['scope', 'after', 'revision'],
+ );
+ return {
+ ...(row.scope === undefined ? {} : { scope: scope(row.scope) }),
+ ...(row.after === undefined
+ ? {}
+ : { after: row.after === null ? null : identifier(row.after) }),
+ ...(row.revision === undefined
+ ? {}
+ : {
+ revision:
+ row.revision === null ? null : requireCount(row.revision, 'provider revision'),
+ }),
+ };
+ },
+ decodeOutput(value) {
+ requireEncodedByteLimit(value, 'provider catalog page', 512 * 1024 + 4096);
+ const row = requireRecord(value, 'provider catalog page');
+ const revision = requireCount(row.revision, 'provider revision');
+ if (row.kind === 'revision_changed') {
+ requireExactRecord(row, 'provider revision change', ['kind', 'revision']);
+ return { kind: 'revision_changed', revision };
+ }
+ requireExactRecord(row, 'provider page', ['kind', 'revision', 'entries', 'next']);
+ if (row.kind !== 'page' || !Array.isArray(row.entries) || row.entries.length > 32)
+ throw invalidProtocolFrame('Invalid provider page');
+ const names = new Set<string>();
+ const entries = row.entries.map((value) => {
+ const entry = requireExactRecord(value, 'provider entry', ['identity', 'descriptor']);
+ const selected = identity(entry.identity);
+ if (names.has(selected.name)) throw invalidProtocolFrame('Duplicate provider name');
+ names.add(selected.name);
+ return { identity: selected, descriptor: descriptor(entry.descriptor) };
+ });
+ const next = row.next === null ? null : identifier(row.next);
+ if (next !== null && next !== entries.at(-1)?.identity.name)
+ throw invalidProtocolFrame('Invalid provider cursor');
+ return { kind: 'page', revision, entries, next };
+ },
+ }),
+} as const;
diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts
index f581a50c60..a4085a55e0 100644
--- a/packages/runtime-host/src/protocol/operations.ts
+++ b/packages/runtime-host/src/protocol/operations.ts
@@ -23,6 +23,8 @@ import { AGENT_GRAPH_OPERATION_SPECS } from './agent-graph.js';
import { requireExactRecord, requireId, requireRecord, requireString } from './codec.js';
import { CONNECTION_EFFECT_OPERATION_SPECS } from './connection-effects.js';
import { CONFIGURATION_OPERATION_SPECS } from './configuration.js';
+import { MODEL_PROVIDER_OPERATION_SPECS } from './model-provider.js';
+export type { ModelProviderCatalogQuery, ModelProviderCatalogResult } from './model-provider.js';
import { DEEP_RESEARCH_OPERATION_SPECS } from './deep-research.js';
import { DAILY_REVIEW_OPERATION_SPECS } from './daily-review.js';
import { CONTEXT_OPERATION_SPECS } from './context.js';
@@ -197,6 +199,7 @@ export * from './usage-pricing.js';
export * from './workspace.js';
export const HOST_OPERATION_SPECS = composeOperationSpecMaps(
+ MODEL_PROVIDER_OPERATION_SPECS,
HOST_BOOTSTRAP_OPERATION_SPECS,
HOST_RESOURCE_OPERATION_SPECS,
PEER_MESH_OPERATION_SPECS,
diff --git a/tests/fixtures/client-message-queue.mjs b/tests/fixtures/client-message-queue.mjs
index 35cc40176a..9bd8d64e8b 100644
--- a/tests/fixtures/client-message-queue.mjs
+++ b/tests/fixtures/client-message-queue.mjs
@@ -88,6 +88,7 @@ export async function verifyMessageQueue(connection, workspace, reopened, openCl
assert.deepEqual((await query()).resolutions, [
previous.rootProof,
...['one', 'two', 'steer'].map((messageId) => ({ messageId, state: 'cancelled' })),
+ { messageId: 'absent', state: 'not_admitted' },
]);
console.log('message-queue-reopened');
return;
@@ -103,6 +104,7 @@ export async function verifyMessageQueue(connection, workspace, reopened, openCl
assert.deepEqual((await query()).resolutions, [
rootProof,
...['one', 'two', 'steer'].map((messageId) => ({ messageId, state: 'pending' })),
+ { messageId: 'absent', state: 'not_admitted' },
]);
assert.deepEqual(await invoke('turn.message.query', { sessionId, messageIds: ids }), {
cancelledMessageIds: [],
diff --git a/tests/fixtures/client-message-submit.mjs b/tests/fixtures/client-message-submit.mjs
index 551772eeb8..3dc5e062f7 100644
--- a/tests/fixtures/client-message-submit.mjs
+++ b/tests/fixtures/client-message-submit.mjs
@@ -325,7 +325,9 @@ export async function verifyMessageSubmit(connection, workspace, reopened, openC
}),
(error) => error.code === 'operation_unavailable',
);
- assert.deepEqual(await resolutions([first.messageId]), { resolutions: [] });
+ assert.deepEqual(await resolutions([first.messageId]), {
+ resolutions: [{ messageId: first.messageId, state: 'not_admitted' }],
+ });
assert.equal(model.requests.length, 0);
const blocked = await submit({
...first,
@@ -336,7 +338,9 @@ export async function verifyMessageSubmit(connection, workspace, reopened, openC
});
assert.equal(blocked.disposition, 'blocked');
assert.equal(blocked.preparation[0].receipt.failed[0].reason, 'disabled');
- assert.deepEqual(await resolutions(['disabled-skill']), { resolutions: [] });
+ assert.deepEqual(await resolutions(['disabled-skill']), {
+ resolutions: [{ messageId: 'disabled-skill', state: 'not_admitted' }],
+ });
assert.equal(model.requests.length, 0);
const sibling = await openClient();
@@ -439,7 +443,9 @@ export async function verifyMessageSubmit(connection, workspace, reopened, openC
});
assert.equal(incompatible.disposition, 'blocked');
assert.equal(incompatible.preparation[0].receipt.failed[0].reason, 'host_incompatible');
- assert.deepEqual(await resolutions(['wrong-run-tools']), { resolutions: [] });
+ assert.deepEqual(await resolutions(['wrong-run-tools']), {
+ resolutions: [{ messageId: 'wrong-run-tools', state: 'not_admitted' }],
+ });
const followup = {
...second,
messageId: 'followup-user-id',
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org