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/10/20 07:49:52 UTC
(maka) branch rust-preview updated: fix(agent): remove unsupported queue and delivery quotas
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch rust-preview
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/rust-preview by this push:
new 13eda4e69a fix(agent): remove unsupported queue and delivery quotas
13eda4e69a is described below
commit 13eda4e69aa64b1a86f1d0e4e6e3094aa7d4ef7d
Author: Wang <m4...@gmail.com>
AuthorDate: Tue Oct 20 15:49:42 2026 +0800
fix(agent): remove unsupported queue and delivery quotas
Accept complete pending queues and preserve full subscription and mutation
progress values through Host, UI and IPC. Keep schemas, exact numbers,
identity, atomic revisions, backpressure and cancellation semantics.
Generated-by: Codex
---
packages/plugin-sdk/src/contracts.ts | 3 -
packages/plugin-sdk/src/generated/runtime.ts | 4 -
platform/event-log/README.md | 2 +-
platform/event-log/README.zh-CN.md | 2 +-
platform/event-log/src/message_admissions.rs | 7 -
.../event-log/src/message_admissions/insert.rs | 19 +-
platform/event-log/src/message_queue/edit.rs | 11 +-
platform/event-log/tests/message_queue.rs | 198 +++++++++++++++++++++
.../tests/application_services/subscriptions.rs | 21 ++-
platform/host-runtime/README.md | 2 +-
platform/host-runtime/README.zh-CN.md | 2 +-
.../host-runtime/src/application/subscriptions.rs | 3 +-
platform/plugin-core/src/json.rs | 5 -
platform/transport/tests/private.rs | 113 +++++++++++-
platform/transport/tests/public_host.rs | 113 ++++++++++++
platform/ui-js/tests/runtime/reconciliation.rs | 4 -
platform/ui-js/tests/runtime/subscriptions.rs | 123 +++++++++----
platform/ui-runtime/src/actions/mutations.rs | 26 ++-
platform/ui-runtime/src/actions/subscriptions.rs | 19 +-
platform/ui-runtime/src/tree.rs | 5 -
plugins/agent/api/src/contract.rs | 3 +-
plugins/agent/native/src/owner/tests/queue.rs | 119 +++++++++++++
plugins/agent/native/src/owner/tests/terminal.rs | 8 -
plugins/agent/native/src/service.rs | 2 +-
.../workspace-processes/native/src/terminal.rs | 4 +
.../native/src/terminal/display.rs | 10 +-
.../native/src/terminal/output.rs | 4 +-
products/maka/cli/notices.json | 4 +-
28 files changed, 690 insertions(+), 146 deletions(-)
diff --git a/packages/plugin-sdk/src/contracts.ts b/packages/plugin-sdk/src/contracts.ts
index 683fd7c621..05c71afe9e 100644
--- a/packages/plugin-sdk/src/contracts.ts
+++ b/packages/plugin-sdk/src/contracts.ts
@@ -10490,7 +10490,6 @@ export const contracts = {
type: 'string',
},
minItems: 1,
- maxItems: 64,
},
kind: {
type: 'string',
@@ -19681,7 +19680,6 @@ export const contracts = {
items: {
$ref: '#/$defs/AgentQueueEntry',
},
- maxItems: 64,
},
},
additionalProperties: false,
@@ -25957,7 +25955,6 @@ export const contracts = {
items: {
$ref: '#/$defs/AgentQueueEntry',
},
- maxItems: 64,
},
},
additionalProperties: false,
diff --git a/packages/plugin-sdk/src/generated/runtime.ts b/packages/plugin-sdk/src/generated/runtime.ts
index 62798894d5..4ea178c097 100644
--- a/packages/plugin-sdk/src/generated/runtime.ts
+++ b/packages/plugin-sdk/src/generated/runtime.ts
@@ -4666,7 +4666,6 @@ export type AgentQueueAction =
| {
/**
* @minItems 1
- * @maxItems 64
*/
messageIds: [string, ...string[]];
kind: 'reorder';
@@ -10107,9 +10106,6 @@ export interface AgentHistoryPrompt {
}
export interface AgentQueue {
revision: number;
- /**
- * @maxItems 64
- */
entries: AgentQueueEntry[];
}
export interface AgentQueueEntry {
diff --git a/platform/event-log/README.md b/platform/event-log/README.md
index c33c0fd52e..d61b7fb752 100644
--- a/platform/event-log/README.md
+++ b/platform/event-log/README.md
@@ -25,7 +25,7 @@ Owns the SQLite event store and State Root lease. It persists application source
Original operation contracts are stored once by digest. Operation requests and receipt references carry that identity, so journal reads, caller history and UI delivery do not repeatedly copy service schemas. Reads verify the exact original request fingerprint; admission still validates the complete contract and resource requirements.
-Message delivery retains prepared content, the editable original and the original submission digest. Its internal records can exceed the public request envelope because they preserve these separate facts; writes and exact message reads do not reapply that envelope to the encoded record. Queue aggregate budgets and caller delivery budgets remain separate. Stop and whole-queue retraction retain complete admitted entries in their transactional receipts; receipt encoding adds its own envelop [...]
+Message delivery retains prepared content, the editable original and the original submission digest. Its internal records can exceed the public request envelope because they preserve these separate facts; writes and exact message reads do not reapply that envelope to the encoded record. Pending queues have no fixed message-count or aggregate-byte quota. Stop and whole-queue retraction retain complete admitted entries in their transactional receipts, without a separate receipt-size quota. [...]
Source messages, executor tool calls and generated tool results share the Session's visible-message identity namespace. Admission checks both claim orders against canonical facts, including results synthesized for pending tools at executor completion or invocation termination. Only actual result identities are occupied; a matching suffix alone does not reserve an identity.
diff --git a/platform/event-log/README.zh-CN.md b/platform/event-log/README.zh-CN.md
index 04dfafdaf1..72abaa24e0 100644
--- a/platform/event-log/README.zh-CN.md
+++ b/platform/event-log/README.zh-CN.md
@@ -25,7 +25,7 @@
原始操作契约按摘要只存储一次。操作请求和回执引用只携带该身份，日志读取、调用方历史与 UI 传递不再逐次复制服务 schema。读取仍核对原始请求指纹，执行准入仍校验完整契约和资源要求。
-消息交付保留准备后的内容、可编辑原文和最初提交的摘要。内部记录保存这些不同事实，编码后可能大于公开请求；写入与精确消息读取不再对整条编码记录重复套用请求大小限制。队列总量预算和调用方传输预算分别管理。停止和整队撤回在同一事务的回执中保留完整获准条目；回执编码增加包装开销，不重复套用待处理内容的字节预算。精确重试和重开后的读取返回同一份原始回执。通过 `editable_message` 按 Session、Turn 和消息身份读取一条已拥有的来源。
+消息交付保留准备后的内容、可编辑原文和最初提交的摘要。内部记录保存这些不同事实，编码后可能大于公开请求；写入与精确消息读取不再对整条编码记录重复套用请求大小限制。待处理队列不设固定条数或总字节额度。停止和整队撤回在同一事务的回执中保留完整获准条目，不另设回执大小额度。精确重试和重开后的读取返回同一份原始回执。通过 `editable_message` 按 Session、Turn 和消息身份读取一条已拥有的来源。
来源消息、执行器工具调用与生成的工具结果共享 Session 的可见消息身份空间。无论哪一方先声明，准入都依据规范事实检查冲突，包括执行器完成或调用终止时为待完成工具补出的结果。只有实际结果身份被占用；仅后缀相同不会预留身份。
diff --git a/platform/event-log/src/message_admissions.rs b/platform/event-log/src/message_admissions.rs
index 656fd5ae79..2c7224ccca 100644
--- a/platform/event-log/src/message_admissions.rs
+++ b/platform/event-log/src/message_admissions.rs
@@ -31,7 +31,6 @@ pub(crate) mod insert;
mod receipts;
pub use receipts::MessageSubmitReceipt;
-const MAX_PENDING_BYTES: i64 = 8 * 1024 * 1024;
/// Dispatcher-owned, non-secret prerequisites for accepted work. The store
/// preserves this value; only the named dispatcher interprets its typed basis.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -200,12 +199,6 @@ pub(super) async fn pending(
tx: &mut SqliteConnection,
session: &str,
) -> Result<Vec<PendingMessageAdmission>, StoreError> {
- let (count, bytes): (i64, i64) = sqlx::query_as(
- "SELECT COUNT(*), COALESCE(SUM(length(CAST(record_json AS BLOB))),0) FROM message_admissions WHERE session_id = ?"
- ).bind(session).fetch_one(&mut *tx).await?;
- if count > 64 || bytes > MAX_PENDING_BYTES {
- return Err(StoreError::PrefixTooLarge);
- }
let rows: Vec<(String, String)> = sqlx::query_as(
"SELECT message_id, record_json FROM message_admissions WHERE session_id = ? ORDER BY position, message_id"
).bind(session).fetch_all(tx).await?;
diff --git a/platform/event-log/src/message_admissions/insert.rs b/platform/event-log/src/message_admissions/insert.rs
index fad3859c5a..eec8e7c00b 100644
--- a/platform/event-log/src/message_admissions/insert.rs
+++ b/platform/event-log/src/message_admissions/insert.rs
@@ -17,7 +17,7 @@
* under the License.
*/
-use super::{MAX_PENDING_BYTES, PendingMessageAdmission, invalid, read};
+use super::{PendingMessageAdmission, invalid, read};
use crate::StoreError;
use maka_agent_facts::event::Invocation;
use maka_agent_facts::message::MessageDisposition;
@@ -112,17 +112,18 @@ pub(crate) async fn insert(
if other_owner {
return Err(invalid("pending Session belongs to another dispatcher"));
}
- let (count, bytes, position): (i64, i64, i64) = sqlx::query_as(
- "SELECT COUNT(*), COALESCE(SUM(length(CAST(record_json AS BLOB))),0), COALESCE(MAX(position),0)
- FROM message_admissions WHERE session_id = ?"
- ).bind(session).fetch_one(&mut *tx).await?;
- if count >= 64 || bytes + encoded.len() as i64 > MAX_PENDING_BYTES {
- return Err(StoreError::PrefixTooLarge);
- }
+ let (pending, position): (bool, i64) = sqlx::query_as(
+ "SELECT EXISTS(SELECT 1 FROM message_admissions WHERE session_id = ?1),
+ COALESCE(MAX(position),0)
+ FROM message_admissions WHERE session_id = ?1",
+ )
+ .bind(session)
+ .fetch_one(&mut *tx)
+ .await?;
let position = position
.checked_add(1)
.ok_or_else(|| invalid("message queue position exhausted"))?;
- if admission.source.disposition == MessageDisposition::TurnStarted && count != 0 {
+ if admission.source.disposition == MessageDisposition::TurnStarted && pending {
return Err(StoreError::SessionBusy);
}
sqlx::query("INSERT INTO message_admissions (session_id, message_id, position, record_json) VALUES (?, ?, ?, ?)")
diff --git a/platform/event-log/src/message_queue/edit.rs b/platform/event-log/src/message_queue/edit.rs
index 9266a6870b..976dd19fc6 100644
--- a/platform/event-log/src/message_queue/edit.rs
+++ b/platform/event-log/src/message_queue/edit.rs
@@ -106,9 +106,7 @@ pub(crate) async fn apply(
entry.source.validate().map_err(invalid)?;
}
QueueEdit::Reorder { message_ids } => {
- if message_ids.len() > 64
- || message_ids.iter().collect::<HashSet<_>>().len() != message_ids.len()
- {
+ if message_ids.iter().collect::<HashSet<_>>().len() != message_ids.len() {
return Err(invalid("invalid message reorder identities"));
}
let lane = message_ids
@@ -152,13 +150,6 @@ pub(crate) async fn apply(
}
let changed = queue.entries != original;
if changed {
- let bytes = queue.entries.iter().try_fold(0usize, |sum, entry| {
- let bytes = serde_json::to_vec(entry)?.len();
- Ok::<_, StoreError>(sum.saturating_add(bytes))
- })?;
- if bytes > 8 * 1024 * 1024 {
- return Err(StoreError::PrefixTooLarge);
- }
for entry in &retracted {
sqlx::query("INSERT INTO message_cancellations(session_id, message_id, cancellation_id) VALUES (?, ?, ?)")
.bind(session).bind(&entry.source.message.message_id)
diff --git a/platform/event-log/tests/message_queue.rs b/platform/event-log/tests/message_queue.rs
index 0df789cf2b..bef45dd2c2 100644
--- a/platform/event-log/tests/message_queue.rs
+++ b/platform/event-log/tests/message_queue.rs
@@ -645,3 +645,201 @@ async fn pending_dispatchers_remain_disjoint_across_reopen_and_receipts_bind_exe
assert!(log.admit_queued_message("epoch", 0, entry).await.is_ok());
log.shutdown().await.unwrap();
}
+
+#[tokio::test]
+async fn pending_queue_preserves_more_than_sixty_four_messages_through_reorder_and_retraction() {
+ let temp = tempfile::tempdir().unwrap();
+ let path = temp.path().join("many-pending.sqlite");
+ let log = EventLog::open(&path).await.unwrap();
+ log.create_session("session", "create", &json!({}), 1)
+ .await
+ .unwrap();
+ let owner = invocation("many");
+ append(&log, &owner, opening()).await;
+ for index in 0..65 {
+ let id = format!("pending-{index:02}");
+ log.admit_message(admission(&owner, &id, Disposition::Followup))
+ .await
+ .unwrap();
+ }
+ let queue = log.message_queue("session").await.unwrap();
+ assert_eq!(queue.revision, 65);
+ assert_eq!(queue.entries.len(), 65);
+ assert_eq!(
+ log.pending_messages("session").await.unwrap(),
+ queue.entries
+ );
+ for (index, entry) in queue.entries.iter().enumerate() {
+ assert_eq!(
+ entry.source.message.message_id,
+ format!("pending-{index:02}")
+ );
+ assert_eq!(
+ entry.source.unprepared_content.text,
+ format!("pending-{index:02}")
+ );
+ }
+ let ids: Vec<_> = (0..65)
+ .rev()
+ .map(|index| format!("pending-{index:02}"))
+ .collect();
+ let reordered = log
+ .edit_message_queue(
+ "session",
+ 65,
+ QueueEdit::Reorder {
+ message_ids: ids.clone(),
+ },
+ command("many-reorder", Kind::Reorder),
+ )
+ .await
+ .unwrap();
+ assert_eq!(reordered.revision, 66);
+ let queue = log.message_queue("session").await.unwrap();
+ assert_eq!(
+ queue
+ .entries
+ .iter()
+ .map(|entry| entry.source.message.message_id.clone())
+ .collect::<Vec<_>>(),
+ ids
+ );
+ let mut duplicates = ids.clone();
+ duplicates[64] = duplicates[0].clone();
+ assert!(
+ log.edit_message_queue(
+ "session",
+ 66,
+ QueueEdit::Reorder {
+ message_ids: duplicates
+ },
+ command("many-duplicate", Kind::Reorder),
+ )
+ .await
+ .is_err()
+ );
+ assert_eq!(log.message_queue("session").await.unwrap(), queue);
+ let edit = QueueEdit::RetractAll {
+ cancellation_id: "many-cancel".into(),
+ };
+ let cancel = command("many-retract", Kind::RetractAll);
+ let receipt = log
+ .edit_message_queue("session", 66, edit.clone(), cancel.clone())
+ .await
+ .unwrap();
+ assert_eq!(receipt.revision, 67);
+ assert_eq!(receipt.retracted, queue.entries);
+ assert_eq!(
+ log.edit_message_queue("session", 66, edit, cancel.clone())
+ .await
+ .unwrap(),
+ receipt
+ );
+ for id in ids {
+ assert!(log.message_cancelled("session", &id).await.unwrap());
+ }
+ assert!(log.pending_messages("session").await.unwrap().is_empty());
+ log.close().await.unwrap();
+ let log = EventLog::open(&path).await.unwrap();
+ assert_eq!(
+ log.queue_command_receipt("session", &cancel).await.unwrap(),
+ Some(receipt)
+ );
+ log.close().await.unwrap();
+}
+
+#[tokio::test]
+async fn pending_queue_preserves_more_than_eight_mib_through_admission_edit_and_receipt_restart() {
+ let temp = tempfile::tempdir().unwrap();
+ let path = temp.path().join("large-pending.sqlite");
+ let log = EventLog::open(&path).await.unwrap();
+ log.create_session("session", "create", &json!({}), 1)
+ .await
+ .unwrap();
+ let owner = invocation("large");
+ append(&log, &owner, opening()).await;
+ for index in 0..7 {
+ let mut pending = admission(&owner, &format!("large-{index}"), Disposition::Followup);
+ let text = char::from(b'a' + index).to_string().repeat(600 * 1024);
+ pending.source.message.content.text = text.clone();
+ pending.source.unprepared_content.text = text;
+ log.admit_message(pending).await.unwrap();
+ }
+ let queue = log.message_queue("session").await.unwrap();
+ assert_eq!(queue.revision, 7);
+ assert_eq!(queue.entries.len(), 7);
+ assert!(
+ queue
+ .entries
+ .iter()
+ .map(|entry| serde_json::to_vec(entry).unwrap().len())
+ .sum::<usize>()
+ > 8 * 1024 * 1024
+ );
+ assert_eq!(
+ log.pending_messages("session").await.unwrap(),
+ queue.entries
+ );
+ for (index, entry) in queue.entries.iter().enumerate() {
+ let text = char::from(b'a' + index as u8)
+ .to_string()
+ .repeat(600 * 1024);
+ assert_eq!(entry.source.message.message_id, format!("large-{index}"));
+ assert_eq!(entry.source.message.content.text, text);
+ assert_eq!(entry.source.unprepared_content.text, text);
+ }
+ let mut content = queue.entries[0].source.message.content.clone();
+ let mut unprepared = queue.entries[0].source.unprepared_content.clone();
+ content.text = "edited-large-source".repeat(30_000);
+ unprepared.text = content.text.clone();
+ let update = QueueEdit::Update {
+ message_id: "large-0".into(),
+ content: Box::new(content.clone()),
+ unprepared_content: Box::new(unprepared.clone()),
+ required_tools: Default::default(),
+ };
+ let updated = log
+ .edit_message_queue("session", 7, update, command("large-update", Kind::Update))
+ .await
+ .unwrap();
+ assert_eq!(updated.revision, 8);
+ let queue = log.message_queue("session").await.unwrap();
+ assert_eq!(queue.entries[0].source.message.content, content);
+ assert_eq!(queue.entries[0].source.unprepared_content, unprepared);
+ assert_eq!(
+ queue.entries[0].source.message.submitted_content_digest,
+ format!("sha256:{}", "c".repeat(64))
+ );
+ let edit = QueueEdit::RetractAll {
+ cancellation_id: "large-cancel".into(),
+ };
+ let cancel = command("large-retract", Kind::RetractAll);
+ let receipt = log
+ .edit_message_queue("session", 8, edit.clone(), cancel.clone())
+ .await
+ .unwrap();
+ assert_eq!(receipt.revision, 9);
+ assert_eq!(receipt.retracted, queue.entries);
+ assert!(serde_json::to_vec(&receipt).unwrap().len() > 8 * 1024 * 1024);
+ assert_eq!(
+ log.edit_message_queue("session", 8, edit, cancel.clone())
+ .await
+ .unwrap(),
+ receipt
+ );
+ for index in 0..7 {
+ assert!(
+ log.message_cancelled("session", &format!("large-{index}"))
+ .await
+ .unwrap()
+ );
+ }
+ assert!(log.pending_messages("session").await.unwrap().is_empty());
+ log.close().await.unwrap();
+ let log = EventLog::open(&path).await.unwrap();
+ assert_eq!(
+ log.queue_command_receipt("session", &cancel).await.unwrap(),
+ Some(receipt)
+ );
+ log.close().await.unwrap();
+}
diff --git a/platform/host-js/tests/application_services/subscriptions.rs b/platform/host-js/tests/application_services/subscriptions.rs
index c6d2b2c1e2..31643f61fe 100644
--- a/platform/host-js/tests/application_services/subscriptions.rs
+++ b/platform/host-js/tests/application_services/subscriptions.rs
@@ -80,6 +80,10 @@ async fn native_stream(
(app, client, binding, issuer)
}
+fn wide_item(index: usize) -> Value {
+ json!(format!("{index}:{}:尾", "中🙂\"\\\n\0".repeat(8192)))
+}
+
#[tokio::test]
async fn javascript_streams_ordered_items_and_final_result_in_both_realms() {
tokio::time::timeout(Duration::from_secs(20), async {
@@ -101,11 +105,14 @@ async fn javascript_streams_ordered_items_and_final_result_in_both_realms() {
app.ready().await.unwrap();
let client=app.client();
let binding=client.bind(client.activation(&id(2)).unwrap(),"echo".into()).await.unwrap().remove(0);
- let mut stream=client.subscribe(binding,"echo".into(),json!("item"),scope(&issuer,true),Attachments::new().into()).await.unwrap();
- for i in 0..35 { assert_eq!(stream.recv().await,Some(json!(format!("item:{i}")))); }
+ let mut stream=client.subscribe(binding,"echo".into(),wide_item(0),scope(&issuer,true),Attachments::new().into()).await.unwrap();
+ for i in 0..35 {
+ let expected = json!(format!("{}:{i}", wide_item(0).as_str().unwrap()));
+ assert!(stream.recv().await == Some(expected), "original JavaScript item {i} changed");
+ }
assert_eq!(stream.recv().await,None);
let completion=stream.completion().await;
- assert_eq!(completion.result.unwrap(),json!("item:done"));
+ assert!(completion.result.unwrap() == json!(format!("{}:done", wide_item(0).as_str().unwrap())), "original JavaScript terminal result changed");
completion.cleanup.unwrap();
app.shutdown().await.unwrap();
}
@@ -248,12 +255,12 @@ async fn retained_context_cannot_publish_after_producer_end_or_block_eof() {
}
#[tokio::test]
-async fn item_schema_json_budget_and_unpolled_pending_emits_are_bounded() {
+async fn large_items_preserve_schema_and_unpolled_pending_emit_bounds() {
tokio::time::timeout(Duration::from_secs(15), async {
let (app, client, binding, issuer) = native_stream(|_, context| {
Box::pin(async move {
assert!(context.emit(json!(false)).await.is_err());
- assert!(context.emit(json!("x".repeat(64 * 1024))).await.is_err());
+ context.emit(wide_item(0)).await?;
let mut pending = Vec::new();
for i in 0..16 {
pending.push(context.emit(json!(format!("item:{i}"))));
@@ -282,6 +289,10 @@ async fn item_schema_json_budget_and_unpolled_pending_emits_are_bounded() {
)
.await
.unwrap();
+ assert!(
+ stream.recv().await == Some(wide_item(0)),
+ "original native item changed"
+ );
for i in 0..16 {
assert_eq!(stream.recv().await, Some(json!(format!("item:{i}"))));
}
diff --git a/platform/host-runtime/README.md b/platform/host-runtime/README.md
index fc096150af..aae978fb69 100644
--- a/platform/host-runtime/README.md
+++ b/platform/host-runtime/README.md
@@ -25,7 +25,7 @@ Runs checked service compositions and owns call admission, activation and retire
A committed source is desired configuration. It becomes effective only after its instances converge. Updates preserve unchanged activations and their captured bindings; replacements wait for accepted calls and native resources to settle. Mutation receipts distinguish business results from cleanup, so a lost reply must be resolved through its original receipt.
-Retiring either party of a subscription requests cancellation of its original invocation. The Host still waits for the handler and its native resources to settle before closing the provider. Accepted queries and mutations retain their original completion and cleanup semantics.
+Retiring either party of a subscription requests cancellation of its original invocation. The Host still waits for the handler and its native resources to settle before closing the provider. Accepted queries and mutations retain their original completion and cleanup semantics. Subscription and mutation-progress delivery use queue backpressure and consumer demand without a separate item-size quota. JavaScript bridge and IPC envelope requirements still apply at their respective boundaries.
Complete source documents, including frozen remote contracts and package references, have a shared 2 MiB budget for admission, persistence and management. Read them in pages under one source revision; replace large documents through immutable resources.
diff --git a/platform/host-runtime/README.zh-CN.md b/platform/host-runtime/README.zh-CN.md
index e269733916..9f1946ec15 100644
--- a/platform/host-runtime/README.zh-CN.md
+++ b/platform/host-runtime/README.zh-CN.md
@@ -25,7 +25,7 @@
提交的配置表示期望状态，实例完成收敛后才成为有效状态。更新保留未改变的激活及其绑定；替换必须等待已接受调用和原生资源完成清理。操作回执区分业务结果与清理状态，丢失响应后应查询原回执。
-订阅的提供方或调用方退役时，Host 请求取消原订阅调用，并在原处理程序及其原生资源完成清理后关闭提供方。已受理的查询和变更仍保留原有完成与清理语义。
+订阅的提供方或调用方退役时，Host 请求取消原订阅调用，并在原处理程序及其原生资源完成清理后关闭提供方。已受理的查询和变更仍保留原有完成与清理语义。订阅和变更进度通过队列背压和消费者需求控制交付，不另设单项大小额度。JavaScript 桥和 IPC 信封要求仍在各自边界生效。
完整配置文档包含冻结的远端契约和包引用，准入、持久化与管理共用 2 MiB 预算。读取时固定配置版本并逐页获取；较大文档通过不可变资源替换。
diff --git a/platform/host-runtime/src/application/subscriptions.rs b/platform/host-runtime/src/application/subscriptions.rs
index fade9be073..2b08dd5ee9 100644
--- a/platform/host-runtime/src/application/subscriptions.rs
+++ b/platform/host-runtime/src/application/subscriptions.rs
@@ -98,7 +98,7 @@ impl Emitter {
}
pub(super) fn emit(&self, value: Value) -> BoxFuture<'static, Result<()>> {
- // Reserve before serialization/validation, including unpolled futures.
+ // Reserve before validation, including unpolled futures.
let reserved = self
.0
.pending
@@ -109,7 +109,6 @@ impl Emitter {
if self.0.closed.is_cancelled() || self.0.stop.is_cancelled() {
return Err("Host subscription is closed".into());
}
- check_json_limit(&value, maka_plugin_core::json::SUBSCRIPTION_ITEM_BYTES)?;
self.0.output.validate(&value).map_err(|e| e.to_string())?;
Ok(permit)
});
diff --git a/platform/plugin-core/src/json.rs b/platform/plugin-core/src/json.rs
index c118107813..6286548dbc 100644
--- a/platform/plugin-core/src/json.rs
+++ b/platform/plugin-core/src/json.rs
@@ -20,11 +20,6 @@
//! Bounded JSON envelopes shared by native and JavaScript plugin boundaries.
use serde_json::Value;
-/// Maximum encoded subscription Item. The Host's 16-slot accepted queue and
-/// 16 pending emit reservations each retain at most 1 MiB of encoded values.
-/// This bounds outstanding delivery, never the stream's lifetime output.
-pub const SUBSCRIPTION_ITEM_BYTES: usize = 64 * 1024;
-
/// Reject integer rounding before a value crosses an application boundary.
pub fn check_delivery(value: &Value, limit: usize) -> Result<usize, String> {
let size = value_size(value, limit)?;
diff --git a/platform/transport/tests/private.rs b/platform/transport/tests/private.rs
index da2e932617..72cdfeda2f 100644
--- a/platform/transport/tests/private.rs
+++ b/platform/transport/tests/private.rs
@@ -54,6 +54,78 @@ struct Fixture {
server: Arc<Server>,
peer: Arc<maka_transport::transport::Peers>,
}
+// This progress fixture uses the same contract at the caller, gateway and provider.
+fn wide_progress_contract() -> maka_plugin_core::contract::Contract {
+ let mut contract = support::business();
+ let maka_plugin_core::contract::Interface::Service { methods } = &mut contract.interface else {
+ unreachable!()
+ };
+ methods.get_mut("write").unwrap().progress = Some(serde_json::json!({
+ "type":"object","properties":{"step":{"type":"string"}},
+ "required":["step"],"additionalProperties":false
+ }));
+ contract
+}
+fn wide_progress() -> serde_json::Value {
+ serde_json::json!({"step":format!("原始:{}:尾", "中🙂\"\\\n\0".repeat(8192))})
+}
+fn replace_progress_contract(
+ input: &mut maka_plugin_core::composition::Input,
+ packages: &mut [maka_packages::Package],
+) {
+ let contract = wide_progress_contract();
+ for package in packages {
+ let original = package.application_ref();
+ let mut declaration = package
+ .manifest()
+ .application
+ .as_ref()
+ .unwrap()
+ .declaration()
+ .clone();
+ let mut changed = false;
+ for entry in declaration.entries.values_mut() {
+ for import in entry.imports.values_mut() {
+ if import.contract.name == contract.name {
+ import.contract = contract.clone();
+ changed = true;
+ }
+ }
+ for export in entry.exports.values_mut() {
+ if export.contract.name == contract.name {
+ export.contract = contract.clone();
+ changed = true;
+ }
+ }
+ }
+ if changed {
+ let mut files = package.files().clone();
+ if package.manifest().id == "local.ui" {
+ // Keep the original progress in state, without rendering its entire
+ // payload into the unrelated terminal Text node.
+ let source = String::from_utf8(files.remove("index.mjs").unwrap()).unwrap();
+ let display =
+ "JSON.stringify(state.outcome?.reply?.receipt??state.outcome??'Ready')";
+ assert!(source.contains(display));
+ files.insert("index.mjs".into(), source.replace(display,
+ "(state.outcome?.status==='updated'?'Progress received':JSON.stringify(state.outcome?.reply?.receipt??state.outcome??'Ready'))").into_bytes());
+ }
+ files.insert(
+ maka_packages::MANIFEST_FILE.into(),
+ serde_json::to_vec(&serde_json::json!({
+ "schemaVersion":1,"id":package.manifest().id,"application":declaration
+ }))
+ .unwrap(),
+ );
+ *package = maka_packages::Package::new(files).unwrap();
+ for node in input.instances.values_mut() {
+ if node.package == original.clone().into() {
+ node.package = package.application_ref().into();
+ }
+ }
+ }
+ }
+}
impl Fixture {
async fn new() -> Self {
Self::with_gate(None).await
@@ -76,7 +148,16 @@ impl Fixture {
let store = Arc::new(EventLog::for_root(owner).await.unwrap());
let owner = Arc::new(RootOwner::create(&caller_root, &namespaces).unwrap());
let caller = Arc::new(EventLog::for_root(owner).await.unwrap());
- let (input, packages, observations) = support::ui(PresentationTarget::Tui);
+ let wide_progress = gate.as_ref().is_some_and(|gate| gate.progress);
+ let (mut input, mut packages, mut observations) = support::ui(PresentationTarget::Tui);
+ if wide_progress {
+ replace_progress_contract(&mut input, &mut packages);
+ for observation in &mut observations {
+ if observation.declaration.contract.name == support::business().name {
+ observation.declaration.contract = wide_progress_contract();
+ }
+ }
+ }
let source = caller
.commit_application(Uuid::new_v4(), 0, input, observations, packages)
.await
@@ -117,7 +198,11 @@ impl Fixture {
exports: Ok(BTreeMap::from([(
"Counter".into(),
maka_plugin_core::runtime::Published {
- contract: support::business(),
+ contract: if wide_progress {
+ wide_progress_contract()
+ } else {
+ support::business()
+ },
value: gate.clone()
as Arc<dyn maka_host_runtime::application::Handler>,
},
@@ -129,6 +214,9 @@ impl Fixture {
)
.unwrap();
}
+ if wide_progress {
+ replace_progress_contract(&mut input, &mut packages);
+ }
store
.commit_application(deployment, 0, input, vec![], packages)
.await
@@ -417,7 +505,7 @@ impl maka_host_runtime::application::Handler for Gate {
Box::pin(async move {
if method == "write" {
if progress {
- context.emit(serde_json::json!({"step":1})).await?;
+ context.emit(wide_progress()).await?;
}
stop.cancelled().await;
cancelled.notify_one();
@@ -2096,20 +2184,31 @@ async fn checked_ui_mutation_progress_cancel_waits_for_original_completion_and_r
let mut task = ui.start_task(&root).unwrap().unwrap();
loop {
let completed = task.future.await;
+ let progress = matches!(
+ &completed,
+ maka_ui_runtime::actions::TaskCompletion::MutationProgress { .. }
+ );
task = ui
.finish_task(task.id, completed)
.unwrap()
.expect("operation still running");
- ui.deliver_action_results(&root, |_, _, _| Ok(false))
+ let delivered = ui
+ .deliver_action_results(&root, |_, _, _| Ok(false))
.unwrap();
+ if progress {
+ assert_eq!(
+ delivered, 1,
+ "original mutation progress must enter the component"
+ );
+ }
ui.wait_idle().await.unwrap();
if ui.frame(&root).unwrap().state["outcome"]["status"] == "updated" {
break;
}
}
- assert_eq!(
- ui.frame(&root).unwrap().state["outcome"]["value"],
- serde_json::json!({"step":1})
+ assert!(
+ ui.frame(&root).unwrap().state["outcome"]["value"] == wide_progress(),
+ "original Unicode mutation progress changed"
);
assert_eq!(ui.accepted_action_count(), 1);
let id = task.id;
diff --git a/platform/transport/tests/public_host.rs b/platform/transport/tests/public_host.rs
index 035ff0d1bf..14df96f82f 100644
--- a/platform/transport/tests/public_host.rs
+++ b/platform/transport/tests/public_host.rs
@@ -113,6 +113,119 @@ async fn public_ipc_persists_original_gateway_intent_and_retires_original_owner(
}).await.unwrap();
}
+struct IpcSubscription {
+ value: Value,
+}
+impl maka_host_runtime::application::Handler for IpcSubscription {
+ fn call(
+ &self,
+ _: String,
+ _: Value,
+ context: maka_host_runtime::application::Context,
+ ) -> futures_util::future::BoxFuture<'static, Result<Value, String>> {
+ let value = self.value.clone();
+ Box::pin(async move {
+ context.emit(value).await?;
+ Ok(Value::Null)
+ })
+ }
+}
+impl maka_host_runtime::application::NativeOwner for IpcSubscription {
+ fn prepare_retirement(
+ &self,
+ ) -> Option<Box<dyn maka_host_runtime::application::NativeRetirement>> {
+ Some(Box::new(()))
+ }
+ fn close(&self) -> futures_util::future::BoxFuture<'static, Result<(), String>> {
+ Box::pin(async { Ok(()) })
+ }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn public_ipc_pulls_large_subscription_original_and_confirms_cleanup() {
+ tokio::time::timeout(Duration::from_secs(20), async {
+ let expected = serde_json::json!({"text": "中🙂\"\\\n\0".repeat(8192), "tail": "尾"});
+ let encoded = serde_json::to_vec(&expected).unwrap();
+ assert!(encoded.len() > 64 * 1024);
+ assert!(encoded.len() < 1024 * 1024);
+ let directory = tempfile::tempdir().unwrap();
+ maka_event_log::root::private_directory(directory.path()).unwrap();
+ let namespaces = RootNamespaces {
+ ownership: directory.path().join("owners"),
+ control: directory.path().join("control"),
+ };
+ let root = directory.path().join("root");
+ let store = Arc::new(EventLog::for_root(Arc::new(RootOwner::create(&root, &namespaces).unwrap())).await.unwrap());
+ let deployment = Uuid::new_v4();
+ let (mut input, mut packages) = support::host();
+ let mut contract = support::business();
+ let maka_plugin_core::contract::Interface::Service { methods } = &mut contract.interface else { unreachable!() };
+ methods.get_mut("read").unwrap().mode = maka_plugin_core::contract::MethodMode::Subscription;
+ methods.get_mut("read").unwrap().output = serde_json::json!({"type":["object","null"], "properties":{"text":{"type":"string"},"tail":{"type":"string"}},"required":["text","tail"],"additionalProperties":false});
+ for package in &mut packages {
+ let name = package.manifest().id.clone();
+ if name != "local.business" && name != "local.gateway" { continue; }
+ let mut declaration = package.manifest().application.as_ref().unwrap().declaration().clone();
+ let entry = declaration.entries.get_mut(&maka_plugin_core::package::EntryTarget::Host).unwrap();
+ if name == "local.business" {
+ entry.implementation = maka_plugin_core::package::Implementation::Native { implementation: "public.subscription".into() };
+ entry.exports.get_mut("Counter").unwrap().contract = contract.clone();
+ } else {
+ entry.imports.get_mut("business").unwrap().contract = contract.clone();
+ }
+ *package = support::counter::package(&name, declaration, None);
+ input.instances.get_mut(if name == "local.business" { "business" } else { "gateway" }).unwrap().package = package.application_ref().into();
+ }
+ let provider = Arc::new(IpcSubscription { value: expected.clone() });
+ let mut native = NativeRegistry::default();
+ native.insert("public.subscription".into(), Arc::new(move |_, _| {
+ let provider = provider.clone();
+ let contract = contract.clone();
+ Box::pin(async move {
+ maka_host_runtime::application::NativeActivation {
+ exports: Ok(BTreeMap::from([("Counter".into(), maka_plugin_core::runtime::Published { contract, value: provider.clone() as Arc<dyn maka_host_runtime::application::Handler> })])),
+ owner: provider,
+ }
+ })
+ })).unwrap();
+ store.commit_application(deployment, 0, input, vec![], packages).await.unwrap();
+ let host = Host::open(store.clone(), deployment, native, Some(maka_host_js::factory(Default::default())), BTreeMap::from([
+ ("business".into(), HostEndpoint { consumer: support::identity(13), import: "business".into(), grant: Grant { methods: ["read".into()].into(), recovery: false } }),
+ ]), vec![]).await.unwrap();
+ #[cfg(unix)]
+ let address = directory.path().join("host.sock");
+ #[cfg(windows)]
+ let address = std::path::PathBuf::from(format!(r"\\.\pipe\maka-public-test-{}", Uuid::new_v4()));
+ let listener = LocalListener::bind(&address).unwrap();
+ let serving = tokio::spawn({ let host = host.clone(); async move { host.serve_local(listener, CancellationToken::new()).await } });
+ let mut client = Client::connect(&address, Some(host.epoch())).await.unwrap();
+ let mut call = client.call("business", "read".into(), Value::Null, Default::default()).await.unwrap();
+ call.start().await.unwrap();
+ let subscribed = call.receive().await.unwrap();
+ let item = call.pull().await.unwrap();
+ match &item {
+ Reply::Item { value } => eprintln!("IPC subscription item encoded_bytes={} original_equal={}", serde_json::to_vec(value).unwrap().len(), value == &expected),
+ Reply::Finished { result, cleanup } => eprintln!("IPC subscription finished before item: error={:?} cleanup={cleanup:?}", result.as_ref().err()),
+ _ => eprintln!("IPC subscription returned an unexpected reply before item"),
+ }
+ let completion = if matches!(item, Reply::Item { .. }) { Some(call.pull().await.unwrap()) } else { None };
+ call.close().await.unwrap();
+ let cleanup_confirmed = call.cleanup_confirmed();
+ client.close().await.unwrap();
+ assert!(Client::retire(&address, host.epoch()).await.unwrap());
+ serving.await.unwrap().unwrap();
+ drop((call, client, host, store));
+ drop(RootOwner::open(&root, &namespaces).unwrap());
+ assert!(matches!(subscribed, Reply::Subscribed));
+ match item {
+ Reply::Item { value } => assert!(value == expected, "large subscription original differs after IPC"),
+ _ => panic!("large subscription original did not cross IPC; see short reply metadata above"),
+ }
+ assert!(matches!(completion, Some(Reply::Finished { result: Ok(Value::Null), cleanup: Ok(()) })));
+ assert!(cleanup_confirmed);
+ }).await.unwrap();
+}
+
struct Blocked {
entered: Arc<tokio::sync::Semaphore>,
cancelled: Arc<tokio::sync::Semaphore>,
diff --git a/platform/ui-js/tests/runtime/reconciliation.rs b/platform/ui-js/tests/runtime/reconciliation.rs
index 1dd0077a84..9ffc579607 100644
--- a/platform/ui-js/tests/runtime/reconciliation.rs
+++ b/platform/ui-js/tests/runtime/reconciliation.rs
@@ -154,10 +154,6 @@ fn committed_ui_addition_retains_original_mount_frame_and_then_replacement_waits
.unwrap();
app.advance_deployment().unwrap();
assert!(app.is_retiring(&root));
- assert_eq!(
- app.live_source_revisions(),
- std::collections::BTreeSet::from([1, 3])
- );
assert_eq!(app.frame(&root).unwrap().state, state);
assert_eq!(app.deployment_status().effective, Some(next));
assert!(app.deployment_status().converging);
diff --git a/platform/ui-js/tests/runtime/subscriptions.rs b/platform/ui-js/tests/runtime/subscriptions.rs
index fb0d05344d..59ba8baeb7 100644
--- a/platform/ui-js/tests/runtime/subscriptions.rs
+++ b/platform/ui-js/tests/runtime/subscriptions.rs
@@ -44,7 +44,7 @@ use std::{
};
use tokio_util::sync::CancellationToken;
-fn service_contract() -> Contract {
+fn service_contract(mode: Mode) -> Contract {
Contract {
name: "test.Stream".into(),
version: 1,
@@ -57,7 +57,11 @@ fn service_contract() -> Contract {
progress: None,
mode: MethodMode::Subscription,
input: json!({}),
- output: json!({"type":"integer"}),
+ output: if matches!(mode, Mode::WideItem | Mode::WideJavaScript) {
+ json!({"type":"string"})
+ } else {
+ json!({"type":"integer"})
+ },
resources: BTreeMap::new(),
},
),
@@ -116,7 +120,11 @@ fn sequence(index: usize, stop: CancellationToken, mode: Mode) -> StreamFuture {
Box::pin(async move {
if stop.is_cancelled() || index == 3 {
return StreamCompletion::Finished(ServiceCompletion {
- result: Ok(json!(3)),
+ result: Ok(if matches!(mode, Mode::WideItem | Mode::WideJavaScript) {
+ json!("done")
+ } else {
+ json!(3)
+ }),
cleanup: if matches!(mode, Mode::FailedCleanup) {
Err("actual cleanup unknown".into())
} else {
@@ -127,7 +135,7 @@ fn sequence(index: usize, stop: CancellationToken, mode: Mode) -> StreamFuture {
StreamCompletion::Item {
value: match mode {
Mode::BadItem => json!("wrong schema"),
- Mode::WideItem => json!("x".repeat(64 * 1024)),
+ Mode::WideItem | Mode::WideJavaScript => wide_item(index),
_ => json!(index),
},
next: sequence(index + 1, stop, mode),
@@ -143,8 +151,12 @@ enum Mode {
Panic,
BadItem,
WideItem,
+ WideJavaScript,
FailedCleanup,
}
+fn wide_item(index: usize) -> Value {
+ json!(format!("{index}:{}:尾", "中🙂\"\\\n\0".repeat(8192)))
+}
struct Parent(Mode);
impl NativeComponent<()> for Parent {
fn call(
@@ -389,17 +401,25 @@ fn fixture_data(
.declaration()
.clone();
let entry = declaration.entries.get_mut(&EntryTarget::Tui).unwrap();
- entry.scene = None;
- entry.implementation = Implementation::Native {
- implementation: "test.parent".into(),
- };
+ let javascript = matches!(mode, Mode::WideJavaScript);
+ if javascript {
+ entry.implementation = Implementation::JavaScript {
+ module: "index.mjs".into(),
+ realm: maka_plugin_core::package::Realm::Dedicated,
+ };
+ } else {
+ entry.scene = None;
+ entry.implementation = Implementation::Native {
+ implementation: "test.parent".into(),
+ };
+ }
let parent_contract = entry.exports["App"].contract.clone();
let child_contract = component_contract(
"test.Child",
BTreeMap::from([("change".into(), json!({"type":"integer"}))]),
);
for (name, contract) in [
- ("stream", service_contract()),
+ ("stream", service_contract(mode)),
("child", child_contract.clone()),
] {
entry.imports.insert(
@@ -410,10 +430,15 @@ fn fixture_data(
},
);
}
- let package = support::package("native.parent", declaration, None);
+ let source = javascript.then(|| format!(r#"
+ export function activate() {{ return {{App:{{contract:{},kind:'component',component:['app']}}}}; }}
+ function frame(state) {{ return {{actions:[],events:[],subscriptions:[{{key:'feed0',binding:'stream',member:0,method:'read',input:0,resources:{{}}}}],frame:{{revision:state.revision,state,surface:{{kind:'terminal',binding:'surface'}},view:{{kind:'text',key:'root',text:'stream',color:16777215}}}}}}; }}
+ export const app = {{mount(){{return frame({{revision:0}});}},event(state,event){{return frame({{revision:state.revision+1,feedback:event.event}});}},update(state){{return frame({{...state,revision:state.revision+1}});}}}};
+ "#, serde_json::to_string(&parent_contract).unwrap()));
+ let package = support::package("native.parent", declaration, source);
input.instances.get_mut("app").unwrap().package = package.application_ref().into();
packages.push(package);
- let service = native_package("native.stream", "test.stream", service_contract());
+ let service = native_package("native.stream", "test.stream", service_contract(mode));
let child = native_package("native.child", "test.child", child_contract.clone());
for (name, key, package) in [("stream", 3, &service), ("child", 4, &child)] {
input
@@ -458,7 +483,7 @@ fn fixture_data(
Ok(BTreeMap::from([(
"Export".into(),
NativeExport {
- contract: service_contract(),
+ contract: service_contract(mode),
value: NativeCapability::Service(Arc::new(Sequence(observed.clone(), mode))),
},
)]))
@@ -858,38 +883,68 @@ fn mutation_factory_panic_cancels_stream_waiting_ack_and_preserves_failed_prepar
}
#[test]
-fn invalid_native_items_stop_before_feedback_and_complete_original_source() {
- for (mode, message) in [(Mode::BadItem, "integer"), (Mode::WideItem, "64 KiB")] {
+fn large_native_items_reach_native_and_javascript_components_exactly() {
+ for mode in [Mode::WideItem, Mode::WideJavaScript] {
let timer = timer();
let (mut app, starts) = fixture(mode, &timer);
let root = app.mount_launch(json!({})).unwrap();
- let (id, item) = run(&timer, start(&mut app, &root));
- let successor = app.finish_task(id, item).unwrap().unwrap();
- assert_eq!(
- app.deliver_action_results(&root, |_, _, _| Ok(false))
- .unwrap(),
- 0,
- "invalid item must never enter the component"
- );
- let (_, actual) = run(&timer, successor);
- assert!(matches!(
- &actual,
- TaskCompletion::Subscription(StreamCompletion::Finished(_))
- ));
- assert!(app.finish_task(id, actual).unwrap().is_none());
+ let mut task = start(&mut app, &root);
+ for index in 0..3 {
+ let (id, item) = run(&timer, task);
+ task = app.finish_task(id, item).unwrap().unwrap();
+ assert_eq!(
+ app.deliver_action_results(&root, |_, _, _| Ok(false))
+ .unwrap(),
+ 1
+ );
+ assert!(
+ app.frame(&root).unwrap().state["feedback"]
+ == json!({"status":"updated","value":wide_item(index)}),
+ "original component item {index} changed"
+ );
+ }
+ finish(&timer, &mut app, &root, task);
assert_eq!(
- app.deliver_action_results(&root, |_, _, _| Ok(false))
- .unwrap(),
- 1
+ app.frame(&root).unwrap().state["feedback"],
+ json!({"status":"succeeded","value":"done"})
);
- let feedback = &app.frame(&root).unwrap().state["feedback"];
- assert_eq!(feedback["status"], "failed");
- assert!(feedback["reason"].as_str().unwrap().contains(message));
assert_eq!(starts.load(Ordering::SeqCst), 1);
assert_eq!(app.accepted_action_count(), 0);
close(&mut app, &[root]);
}
}
+
+#[test]
+fn invalid_native_items_stop_before_feedback_and_complete_original_source() {
+ let timer = timer();
+ let (mut app, starts) = fixture(Mode::BadItem, &timer);
+ let root = app.mount_launch(json!({})).unwrap();
+ let (id, item) = run(&timer, start(&mut app, &root));
+ let successor = app.finish_task(id, item).unwrap().unwrap();
+ assert_eq!(
+ app.deliver_action_results(&root, |_, _, _| Ok(false))
+ .unwrap(),
+ 0,
+ "invalid item must never enter the component"
+ );
+ let (_, actual) = run(&timer, successor);
+ assert!(matches!(
+ &actual,
+ TaskCompletion::Subscription(StreamCompletion::Finished(_))
+ ));
+ assert!(app.finish_task(id, actual).unwrap().is_none());
+ assert_eq!(
+ app.deliver_action_results(&root, |_, _, _| Ok(false))
+ .unwrap(),
+ 1
+ );
+ let feedback = &app.frame(&root).unwrap().state["feedback"];
+ assert_eq!(feedback["status"], "failed");
+ assert!(feedback["reason"].as_str().unwrap().contains("integer"));
+ assert_eq!(starts.load(Ordering::SeqCst), 1);
+ assert_eq!(app.accepted_action_count(), 0);
+ close(&mut app, &[root]);
+}
#[test]
fn completed_stream_with_failed_cleanup_preserves_work_until_actual_confirmation() {
let timer = timer();
diff --git a/platform/ui-runtime/src/actions/mutations.rs b/platform/ui-runtime/src/actions/mutations.rs
index 6e30538ff3..2104292b69 100644
--- a/platform/ui-runtime/src/actions/mutations.rs
+++ b/platform/ui-runtime/src/actions/mutations.rs
@@ -413,20 +413,18 @@ impl<V: ?Sized + 'static> Application<V> {
.as_ref()
.filter(|_| !running.completed)
.ok_or_else(|| Error::Invalid("progress requires an active mutation".into()))?;
- let valid = tree::serialized_bytes(&value).and_then(|bytes| {
- if bytes > 64 * 1024 {
- return Err(Error::Invalid("mutation progress exceeds 64 KiB".into()));
- }
- let Export::Service { methods, .. } = running.execution.value() else {
- unreachable!()
- };
- methods[&running.method]
- .progress
- .as_ref()
- .ok_or_else(|| Error::Invalid("mutation has no progress declaration".into()))?
- .validate(&value)
- .map_err(|e| Error::Invalid(e.to_string()))
- });
+ let Export::Service { methods, .. } = running.execution.value() else {
+ unreachable!()
+ };
+ let valid = methods[&running.method]
+ .progress
+ .as_ref()
+ .ok_or_else(|| Error::Invalid("mutation has no progress declaration".into()))
+ .and_then(|progress| {
+ progress
+ .validate(&value)
+ .map_err(|e| Error::Invalid(e.to_string()))
+ });
let consumer = running.consumer.clone();
let key = running.key.clone();
let wanted = self.update_wanted(&consumer, &key, id);
diff --git a/platform/ui-runtime/src/actions/subscriptions.rs b/platform/ui-runtime/src/actions/subscriptions.rs
index 14655522a0..7b1998ac90 100644
--- a/platform/ui-runtime/src/actions/subscriptions.rs
+++ b/platform/ui-runtime/src/actions/subscriptions.rs
@@ -138,18 +138,13 @@ impl<V: ?Sized + 'static> Application<V> {
));
}
let wanted = self.subscription_wanted(&running.consumer, &running.key, id);
- let valid = tree::serialized_bytes(&value).and_then(|bytes| {
- if bytes > maka_plugin_core::json::SUBSCRIPTION_ITEM_BYTES {
- return Err(Error::Invalid("subscription item exceeds 64 KiB".into()));
- }
- let Export::Service { methods, .. } = running.execution.value() else {
- unreachable!()
- };
- methods[&running.method]
- .output
- .validate(&value)
- .map_err(|e| Error::Invalid(e.to_string()))
- });
+ let Export::Service { methods, .. } = running.execution.value() else {
+ unreachable!()
+ };
+ let valid = methods[&running.method]
+ .output
+ .validate(&value)
+ .map_err(|e| Error::Invalid(e.to_string()));
let consumer = running.consumer.clone();
let key = running.key.clone();
if let Err(error) = valid {
diff --git a/platform/ui-runtime/src/tree.rs b/platform/ui-runtime/src/tree.rs
index db4376c49d..048ec48ff0 100644
--- a/platform/ui-runtime/src/tree.rs
+++ b/platform/ui-runtime/src/tree.rs
@@ -23,11 +23,6 @@ use std::{
time::Instant,
};
-/// Measure stream payloads without allocating a serialized copy.
-pub(super) fn serialized_bytes(value: &Value) -> Result<usize> {
- maka_plugin_core::json::value_size(value, usize::MAX).map_err(Error::Invalid)
-}
-
#[derive(Clone)]
pub(super) struct Child {
binding: String,
diff --git a/plugins/agent/api/src/contract.rs b/plugins/agent/api/src/contract.rs
index 5e75595e1d..d82aa2bb76 100644
--- a/plugins/agent/api/src/contract.rs
+++ b/plugins/agent/api/src/contract.rs
@@ -969,7 +969,6 @@ pub enum DeliveryPage {
#[schemars(rename = "AgentQueue")]
pub struct Queue {
pub revision: u64,
- #[schemars(length(max = 64))]
pub entries: Vec<QueueEntry>,
}
#[derive(Serialize, JsonSchema, Deserialize)]
@@ -1057,7 +1056,7 @@ pub enum QueueAction {
message: String,
},
Reorder {
- #[schemars(length(min = 1, max = 64))]
+ #[schemars(length(min = 1))]
message_ids: Vec<String>,
},
}
diff --git a/plugins/agent/native/src/owner/tests/queue.rs b/plugins/agent/native/src/owner/tests/queue.rs
index 8a369645e7..b3e17a904a 100644
--- a/plugins/agent/native/src/owner/tests/queue.rs
+++ b/plugins/agent/native/src/owner/tests/queue.rs
@@ -19,6 +19,125 @@
use super::*;
+// Keep the provider request held until Stop. Record behavioral failures before
+// cleanup so the old admission refusal cannot leave a live worker behind.
+async fn unrestricted_queue_regression(count: usize, large: bool) {
+ tokio::time::timeout(Duration::from_secs(180), async {
+ let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let fixture = Fixture::new(&format!("http://{}/v1", listener.local_addr().unwrap()), None).await;
+ let session = fixture.create("Unrestricted pending queue").await;
+ let started = fixture.send(&session).await.unwrap();
+ let (mut held, _) = listener.accept().await.unwrap();
+ request(&mut held).await;
+ let selection = fixture.selection().await;
+ let token = fixture.owner.runs.lock().await.controls[&session].cancellation.clone();
+ let mut originals = Vec::<(String, String)>::new();
+ let observed: Result<(), String> = async {
+ for index in 0..count {
+ let message = if large {
+ char::from(b'a' + index as u8).to_string().repeat(600 * 1024)
+ } else {
+ format!("{}-{index}", "🦀".repeat(80))
+ };
+ let value = fixture.mutation(2, "send", json!({"sessionId":session,"message":message,"model":selection}), Uuid::new_v4()).await?;
+ if value["kind"] != "queued" { return Err(format!("send {} did not queue: {value}", index + 1)); }
+ let id = value["messageId"].as_str().ok_or("missing queued message ID")?.to_owned();
+ originals.push((id, message));
+ }
+ let queue = fixture.query("queue", json!({"sessionId":session})).await?;
+ let ids: Vec<_> = originals.iter().map(|(id, _)| id.clone()).collect();
+ let actual: Vec<_> = queue["entries"].as_array().ok_or("missing entries")?.iter().map(|row| row["messageId"].clone()).collect();
+ if actual != ids.iter().map(|id| json!(id)).collect::<Vec<_>>() { return Err("queue omitted or reordered accepted inputs".into()); }
+ for (id, message) in &originals {
+ let read = fixture.query("queueMessage", json!({"sessionId":session,"messageId":id,"expectedRevision":queue["revision"]})).await?;
+ if read != json!({"kind":"ready","revision":queue["revision"],"message":message,"hasAttachments":false}) { return Err(format!("original body changed for {id}")); }
+ }
+ let durable = fixture.store.message_queue(&session).await.map_err(|e| e.to_string())?;
+ if large && durable.entries.iter().map(|entry| serde_json::to_vec(entry).unwrap().len()).sum::<usize>() <= 8 * 1024 * 1024 { return Err("workload did not cross the former aggregate byte limit".into()); }
+ for (entry, (id, message)) in durable.entries.iter().zip(&originals) {
+ if entry.source.message.message_id != *id || entry.source.unprepared_content.text != *message || entry.source.message.content.text != *message { return Err("durable original input changed".into()); }
+ }
+ let through = fixture.query("observe", json!({"sessionId":session})).await?["through"].clone();
+ let mut watch = fixture.app.client().subscribe(fixture.binding(2).await, "watch".into(), json!({"sessionId":session,"after":through}), scope(&fixture.issuer, true), Attachments::new().into()).await.map_err(|e| e.to_string())?;
+ let first = watch.recv().await.ok_or("watch closed without delivering the queue")?;
+ if first != json!({"kind":"queue","queue":queue}) { return Err("watch did not deliver the complete queue".into()); }
+ if !large && serde_json::to_vec(&first).unwrap().len() <= 64 * 1024 { return Err("watch workload did not cross the former delivery byte limit".into()); }
+ drop(watch);
+ let replacement = if large { "z".repeat(600 * 1024) } else { "edited original input".into() };
+ let update = json!({"sessionId":session,"expectedRevision":queue["revision"],"edit":{"kind":"update","messageId":ids[0],"message":replacement}});
+ let operation = Uuid::new_v4();
+ let changed = fixture.mutation(2, "editQueue", update.clone(), operation).await?;
+ if changed["kind"] != "committed" || fixture.mutation(2, "editQueue", update.clone(), operation).await? != changed { return Err("update did not commit idempotently".into()); }
+ if fixture.mutation(2, "editQueue", update, Uuid::new_v4()).await?["kind"] != "revision_conflict" { return Err("stale edit crossed its revision fence".into()); }
+ let read = fixture.query("queueMessage", json!({"sessionId":session,"messageId":ids[0],"expectedRevision":changed["revision"]})).await?;
+ if read != json!({"kind":"ready","revision":changed["revision"],"message":replacement,"hasAttachments":false}) { return Err("edited body was truncated or changed".into()); }
+ let foreign = fixture.create("Other queue owner").await;
+ let foreign_queue = fixture.query("queue", json!({"sessionId":foreign})).await?;
+ if fixture.query("queueMessage", json!({"sessionId":foreign,"messageId":ids[0],"expectedRevision":foreign_queue["revision"]})).await? != json!({"kind":"missing"}) { return Err("queue body crossed Session ownership".into()); }
+ let reversed: Vec<_> = ids.iter().rev().cloned().collect();
+ let reordered = fixture.mutation(2, "editQueue", json!({"sessionId":session,"expectedRevision":changed["revision"],"edit":{"kind":"reorder","messageIds":reversed}}), Uuid::new_v4()).await?;
+ if reordered["kind"] != "committed" { return Err(format!("full lane reorder rejected: {reordered}")); }
+ let queue = fixture.query("queue", json!({"sessionId":session})).await?;
+ let actual: Vec<_> = queue["entries"].as_array().ok_or("missing reordered entries")?.iter().map(|row| row["messageId"].clone()).collect();
+ if actual != reversed.iter().map(|id| json!(id)).collect::<Vec<_>>() { return Err("full lane reorder lost identities or order".into()); }
+ let retracted = fixture.mutation(2, "editQueue", json!({"sessionId":session,"expectedRevision":reordered["revision"],"edit":{"kind":"retract","messageId":reversed[0]}}), Uuid::new_v4()).await?;
+ if retracted["kind"] != "committed" || fixture.query("queueMessage", json!({"sessionId":session,"messageId":reversed[0],"expectedRevision":retracted["revision"]})).await? != json!({"kind":"missing"}) { return Err("retraction did not remove its target".into()); }
+ if large {
+ // Restore seven large bodies so Stop and its recovered receipt
+ // also cross the former aggregate limit after retraction.
+ let message = "replacement".repeat(600 * 1024 / "replacement".len());
+ let value = fixture.mutation(2, "send", json!({"sessionId":session,"message":message,"model":selection}), Uuid::new_v4()).await?;
+ if value["kind"] != "queued" { return Err("replacement did not queue".into()); }
+ originals.push((value["messageId"].as_str().ok_or("missing replacement ID")?.to_owned(), message));
+ }
+ Ok(())
+ }.await;
+ let stop_input = json!({"sessionId":session,"invocationId":started["invocationId"]});
+ let operation = Uuid::new_v4();
+ let stopped = fixture.mutation(2, "stop", stop_input.clone(), operation).await;
+ let replayed = fixture.mutation(2, "stop", stop_input.clone(), operation).await;
+ let cancelled_by_stop = token.is_cancelled();
+ if stopped.is_err() {
+ // Baseline cleanup uses individual public retractions, whose receipts
+ // contain no queue-sized payload, then a fresh Stop operation.
+ for (id, _) in &originals {
+ let queue = fixture.query("queue", json!({"sessionId":session})).await.unwrap();
+ fixture.mutation(2, "editQueue", json!({"sessionId":session,"expectedRevision":queue["revision"],"edit":{"kind":"retract","messageId":id}}), Uuid::new_v4()).await.unwrap();
+ }
+ fixture.mutation(2, "stop", stop_input, Uuid::new_v4()).await.unwrap();
+ }
+ fixture.drained().await;
+ let remaining = fixture.store.message_queue(&session).await.unwrap();
+ let mut cancelled = Vec::new();
+ let mut canonical = Vec::new();
+ for (id, _) in &originals {
+ cancelled.push(fixture.store.message_cancelled(&session, id).await.unwrap());
+ canonical.push(fixture.store.root_message(&session, id).await.unwrap().is_none());
+ }
+ fixture.app.shutdown().await.unwrap();
+ fixture.config.shutdown().await.unwrap();
+ assert!(observed.is_ok(), "{observed:?}");
+ let admitted = count + usize::from(large);
+ assert_eq!(originals.len(), admitted);
+ assert_eq!(stopped, Ok(json!({"requested":true})));
+ assert_eq!(replayed, stopped);
+ assert!(cancelled_by_stop);
+ assert!(remaining.entries.is_empty());
+ assert_eq!(cancelled, vec![true; admitted]);
+ assert_eq!(canonical, vec![true; admitted]);
+ }).await.unwrap();
+}
+
+#[tokio::test]
+async fn queue_accepts_more_than_sixty_four_and_delivers_a_complete_large_watch() {
+ unrestricted_queue_regression(256, false).await;
+}
+
+#[tokio::test]
+async fn queue_accepts_more_than_eight_mib_and_preserves_full_original_messages() {
+ unrestricted_queue_regression(7, true).await;
+}
+
#[tokio::test]
async fn queued_followup_uses_original_worker_and_a_new_invocation() {
tokio::time::timeout(Duration::from_secs(30), async {
diff --git a/plugins/agent/native/src/owner/tests/terminal.rs b/plugins/agent/native/src/owner/tests/terminal.rs
index 8ef87eef29..16d64c8cf9 100644
--- a/plugins/agent/native/src/owner/tests/terminal.rs
+++ b/plugins/agent/native/src/owner/tests/terminal.rs
@@ -322,10 +322,6 @@ fn output_request(target: &Value, after: u64) -> Value {
}
fn output_event(value: Value) -> Value {
assert_eq!(value["kind"], "completed", "{value}");
- assert!(
- serde_json::to_vec(&value).unwrap().len()
- <= maka_plugin_core::json::SUBSCRIPTION_ITEM_BYTES
- );
value["value"].clone()
}
async fn output_read(f: &Fixture, target: &Value, after: u64) -> Value {
@@ -658,10 +654,6 @@ async fn terminal_styled_display_survives_retirement_cancel_and_host_reopen_beyo
let mut stream = subscribe(f, owner, "readDisplay", target.clone()).await;
let mut assembly = None;
while let Some(item) = stream.recv().await {
- assert!(
- serde_json::to_vec(&item).unwrap().len()
- <= maka_plugin_core::json::SUBSCRIPTION_ITEM_BYTES
- );
assert_eq!(item["kind"], "completed", "{item}");
match serde_json::from_value::<Part>(item["value"].clone()).unwrap() {
Part::Begin {
diff --git a/plugins/agent/native/src/service.rs b/plugins/agent/native/src/service.rs
index 755669a208..36f8261c27 100644
--- a/plugins/agent/native/src/service.rs
+++ b/plugins/agent/native/src/service.rs
@@ -763,7 +763,7 @@ impl Handler for Agent {
let queue = owner.queue(&input.session_id).await?;
queue_revision = Some(queue.revision);
let update = encode(DeliveryPage::Queue { queue })?;
- maka_plugin_core::json::check_delivery(&update, 48 * 1024)
+ maka_plugin_core::json::exact_numbers(&update)
.map_err(|_| "agent_queue_unavailable")?;
context.emit(update).await?;
}
diff --git a/plugins/workspace/workspace-processes/native/src/terminal.rs b/plugins/workspace/workspace-processes/native/src/terminal.rs
index 0cc7585eff..f84e43aa05 100644
--- a/plugins/workspace/workspace-processes/native/src/terminal.rs
+++ b/plugins/workspace/workspace-processes/native/src/terminal.rs
@@ -38,6 +38,10 @@ use serde_json::Value;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
+// Terminal delivery work batches continue until all output has been delivered.
+// This is not a Host item contract or a limit on the total retained output.
+const DELIVERY_CHUNK_BYTES: usize = 64 * 1024;
+
pub(super) struct Service(pub(super) Arc<Processes>);
impl Handler for Service {
fn call(
diff --git a/plugins/workspace/workspace-processes/native/src/terminal/display.rs b/plugins/workspace/workspace-processes/native/src/terminal/display.rs
index b4791c524b..4a6cea447a 100644
--- a/plugins/workspace/workspace-processes/native/src/terminal/display.rs
+++ b/plugins/workspace/workspace-processes/native/src/terminal/display.rs
@@ -22,7 +22,6 @@ use maka_agent_facts::{
shell_run::ShellOutput,
terminal::display::{Buffer, Part, Row, RowPart},
};
-use maka_plugin_core::json::SUBSCRIPTION_ITEM_BYTES;
use std::collections::VecDeque;
impl Processes {
@@ -66,11 +65,10 @@ impl Processes {
(Buffer::LastAlternate, alternate),
] {
for (row, runs) in rows.iter().enumerate() {
- for part in fragments(buffer, row as u32, runs, SUBSCRIPTION_ITEM_BYTES - overhead)?
- {
+ for part in fragments(buffer, row as u32, runs, DELIVERY_CHUNK_BYTES - overhead)? {
let length = serde_json::to_vec(&part).map_err(failed)?.len();
let separator = usize::from(!parts.is_empty());
- if bytes + separator + length > SUBSCRIPTION_ITEM_BYTES {
+ if bytes + separator + length > DELIVERY_CHUNK_BYTES {
self.display_emit(
&target,
Part::Rows {
@@ -212,7 +210,7 @@ mod tests {
Buffer::Screen,
row as u32,
runs,
- SUBSCRIPTION_ITEM_BYTES - overhead,
+ DELIVERY_CHUNK_BYTES - overhead,
)
.unwrap()
{
@@ -224,7 +222,7 @@ mod tests {
.unwrap(),
)
.unwrap();
- assert!(encoded.len() <= SUBSCRIPTION_ITEM_BYTES);
+ assert!(encoded.len() <= DELIVERY_CHUNK_BYTES);
assembly.rows(parts).unwrap();
count += 1;
}
diff --git a/plugins/workspace/workspace-processes/native/src/terminal/output.rs b/plugins/workspace/workspace-processes/native/src/terminal/output.rs
index e4564e722b..82fdd9636e 100644
--- a/plugins/workspace/workspace-processes/native/src/terminal/output.rs
+++ b/plugins/workspace/workspace-processes/native/src/terminal/output.rs
@@ -90,7 +90,7 @@ impl Processes {
}
}
-/// Fill the existing Item envelope, including base64 expansion. The maximum
+/// Fill a terminal delivery chunk, including base64 expansion. The maximum
/// cursor and u16 geometry reserve their full decimal widths for every resize.
/// Derive overhead from the real serialized variant, not a second byte policy.
fn output_bytes() -> Result<usize, ToolError> {
@@ -105,7 +105,7 @@ fn output_bytes() -> Result<usize, ToolError> {
envelope["value"][field] = serde_json::json!({"cols": u16::MAX, "rows": u16::MAX});
}
let overhead = serde_json::to_vec(&envelope).map_err(failed)?.len();
- Ok((maka_plugin_core::json::SUBSCRIPTION_ITEM_BYTES - overhead) / 4 * 3)
+ Ok((DELIVERY_CHUNK_BYTES - overhead) / 4 * 3)
}
fn project(event: PtyStreamEvent) -> OutputEvent {
diff --git a/products/maka/cli/notices.json b/products/maka/cli/notices.json
index b49fddcb07..049d07bf61 100644
--- a/products/maka/cli/notices.json
+++ b/products/maka/cli/notices.json
@@ -5,8 +5,8 @@
"package-lock.json": "a5af15aca21fabe0e1bdd48784167df4dc81cee17777d54861dcbd641168a4eb",
"package.json": "38baf3beeb5876849ca16e31868d59bfaee769f260c7555b4a8a1847b92e28a1",
"packages/plugin-sdk/src/configurations.ts": "e7bb3e7245adc43f27a3834c2522b2645540ae5a6a1300a706842abd4a4da852",
- "packages/plugin-sdk/src/contracts.ts": "ee1d7719f8c9a1468cb6c814c4ef55ed6fcfc1ca40f38ac902005cabcff34bdd",
- "packages/plugin-sdk/src/generated/runtime.ts": "446e6bebc603fa14c60b34d089415d5a552c8189923d6fe4fe77abd5cec1b3cd",
+ "packages/plugin-sdk/src/contracts.ts": "b0d1f51939f05ff65026caeb58b02933fa4a014fa05c24e609da8e08904b1462",
+ "packages/plugin-sdk/src/generated/runtime.ts": "ad6374f8ccd098a662cec66d4f74a36cd7e00b1b40b0f77342f7e022e66b357b",
"packages/plugin-sdk/src/index.ts": "e33226c48bc54fbf377a604ead3b28ea448f50c287582804dab1e6e1059a6f71",
"packages/plugin-sdk/src/native.ts": "6a826e012f5b7887c88ea295d5826b29939ae2ff0ff722db7b4124e37fc49749",
"packages/plugin-sdk/src/ui.ts": "d669d2c5c70fc61aae3c0582f586950bf9cea863ee72b48af673e57c30c1b0b9",
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org