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/19 19:33:01 UTC
(maka) branch feat/runtime-host-rust updated: fix(queue): retain original submission proof after editing
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 f922ea9e48 fix(queue): retain original submission proof after editing
f922ea9e48 is described below
commit f922ea9e48c36c722b9248642b88d13af6248ab9
Author: Wang <m4...@gmail.com>
AuthorDate: Sun Sep 20 03:32:56 2026 +0800
fix(queue): retain original submission proof after editing
Generated-by: OpenAI Codex
---
crates/event-log/src/message_queue.rs | 1 -
crates/event-log/src/message_queue/edit.rs | 4 +-
crates/event-log/tests/message_queue.rs | 1 -
crates/event-log/tests/message_submit_receipts.rs | 6 +-
.../runtime-host/src/server/messages/capacity.rs | 3 -
.../runtime-host/src/server/messages/mutations.rs | 3 -
crates/runtime-host/src/server/messages/update.rs | 5 --
tests/fixtures/client-workhub-queue.mjs | 64 ++++++++++++++++++++--
8 files changed, 65 insertions(+), 22 deletions(-)
diff --git a/crates/event-log/src/message_queue.rs b/crates/event-log/src/message_queue.rs
index 90cf92e92a..1e1c0a04b6 100644
--- a/crates/event-log/src/message_queue.rs
+++ b/crates/event-log/src/message_queue.rs
@@ -49,7 +49,6 @@ pub enum QueueEdit {
Update {
message_id: String,
content: Box<MessageInput>,
- submitted_content_digest: String,
skill_invocation: maka_runtime::skills::SkillInvocationResult,
required_tools: std::collections::BTreeSet<String>,
},
diff --git a/crates/event-log/src/message_queue/edit.rs b/crates/event-log/src/message_queue/edit.rs
index 35be7be931..fbabcf7c14 100644
--- a/crates/event-log/src/message_queue/edit.rs
+++ b/crates/event-log/src/message_queue/edit.rs
@@ -93,14 +93,14 @@ pub(crate) async fn apply(
QueueEdit::Update {
message_id,
content,
- submitted_content_digest,
skill_invocation,
required_tools,
} => {
let index = queued(&queue, &message_id)?;
let entry = &mut queue.entries[index];
entry.source.message.content = *content;
- entry.source.message.submitted_content_digest = submitted_content_digest;
+ // Edits change delivery, not the original submission identity used
+ // to prove admission after an Epoch change.
entry.source.skill_invocation = skill_invocation;
entry.required_tools = required_tools;
entry.source.validate().map_err(invalid)?;
diff --git a/crates/event-log/tests/message_queue.rs b/crates/event-log/tests/message_queue.rs
index f306180c09..28175288fd 100644
--- a/crates/event-log/tests/message_queue.rs
+++ b/crates/event-log/tests/message_queue.rs
@@ -148,7 +148,6 @@ async fn queue_edits_cancel_or_deliver_once_with_atomic_revision_and_original_ow
QueueEdit::Update {
message_id: "two".into(),
content: Box::new("edited 😀".into()),
- submitted_content_digest: format!("sha256:{}", "d".repeat(64)),
skill_invocation: skills.clone(),
required_tools: ["Read".into()].into(),
},
diff --git a/crates/event-log/tests/message_submit_receipts.rs b/crates/event-log/tests/message_submit_receipts.rs
index d2337aa0a8..826c87da96 100644
--- a/crates/event-log/tests/message_submit_receipts.rs
+++ b/crates/event-log/tests/message_submit_receipts.rs
@@ -83,7 +83,6 @@ async fn submit_receipt_is_atomic_and_survives_edit_delivery_cancellation_and_ep
QueueEdit::Update {
message_id: "queued".into(),
content: Box::new("edited".into()),
- submitted_content_digest: format!("sha256:{}", "d".repeat(64)),
skill_invocation: Default::default(),
required_tools: Default::default(),
},
@@ -116,6 +115,11 @@ async fn submit_receipt_is_atomic_and_survives_edit_delivery_cancellation_and_ep
.unwrap()
.unwrap();
assert_eq!(edited.source.message.content.text, "edited");
+ assert_eq!(
+ edited.source.message.submitted_content_digest,
+ pending.source.message.submitted_content_digest,
+ "queue edits preserve canonical proof of the original submission"
+ );
append(
&log,
&owner,
diff --git a/crates/runtime-host/src/server/messages/capacity.rs b/crates/runtime-host/src/server/messages/capacity.rs
index 25790491ed..07f2a9d4c1 100644
--- a/crates/runtime-host/src/server/messages/capacity.rs
+++ b/crates/runtime-host/src/server/messages/capacity.rs
@@ -44,7 +44,6 @@ pub(super) fn check(
QueueEdit::Update {
message_id,
content,
- submitted_content_digest,
skill_invocation,
required_tools,
} => {
@@ -54,7 +53,6 @@ pub(super) fn check(
.find(|e| &e.source.message.message_id == message_id)
.expect("validated entry");
entry.source.message.content = (**content).clone();
- entry.source.message.submitted_content_digest = submitted_content_digest.clone();
entry.source.skill_invocation = skill_invocation.clone();
entry.required_tools = required_tools.clone();
}
@@ -295,7 +293,6 @@ mod tests {
let edit = QueueEdit::Update {
message_id: "queued".into(),
content: Box::new("x".repeat(bytes).into()),
- submitted_content_digest: format!("sha256:{}", "b".repeat(64)),
skill_invocation: Default::default(),
required_tools: Default::default(),
};
diff --git a/crates/runtime-host/src/server/messages/mutations.rs b/crates/runtime-host/src/server/messages/mutations.rs
index 4e73541e5a..b9a0fab1d5 100644
--- a/crates/runtime-host/src/server/messages/mutations.rs
+++ b/crates/runtime-host/src/server/messages/mutations.rs
@@ -158,8 +158,6 @@ pub(super) async fn execute(
let mut source = entry.source.clone();
let required_tools;
source.message.content = super::update::content(source.message.content, &i.text)?;
- source.message.submitted_content_digest =
- super::update::digest(&source.message.content)?;
source.skill_invocation = Default::default();
if let Some(references) = &mut source.message.content.inline_references {
references.retain(|reference| {
@@ -226,7 +224,6 @@ pub(super) async fn execute(
QueueEdit::Update {
required_tools,
message_id: i.entry_id.clone(),
- submitted_content_digest: source.message.submitted_content_digest,
content: Box::new(source.message.content),
skill_invocation: source.skill_invocation,
}
diff --git a/crates/runtime-host/src/server/messages/update.rs b/crates/runtime-host/src/server/messages/update.rs
index c6602dd843..d2a2fe28d0 100644
--- a/crates/runtime-host/src/server/messages/update.rs
+++ b/crates/runtime-host/src/server/messages/update.rs
@@ -20,11 +20,6 @@
use super::{Code, OperationError, failure};
use maka_runtime::input::MessageInput;
-pub(super) fn digest(content: &MessageInput) -> Result<String, OperationError> {
- content
- .content_digest()
- .map_err(|e| failure(Code::InternalFailure, &e.to_string()))
-}
pub(super) fn content(
mut content: MessageInput,
text: &str,
diff --git a/tests/fixtures/client-workhub-queue.mjs b/tests/fixtures/client-workhub-queue.mjs
index b0c71d9b39..39a24c630f 100644
--- a/tests/fixtures/client-workhub-queue.mjs
+++ b/tests/fixtures/client-workhub-queue.mjs
@@ -160,6 +160,55 @@ export async function verifyWorkhubQueue(connection) {
enqueue({ ...input, content: { text: 'CHANGED' } }),
(error) => error.code === 'operation_conflict',
);
+ // Retraction/order control accepted work; editing must prepare new content
+ // against an active behavior and promotion must retain the frozen policy.
+ await enqueue({ ...input, messageId: 'promoted', content: { text: 'PROMOTED' } });
+ await enqueue({ ...input, messageId: 'discarded', content: { text: 'DISCARDED' } });
+ await toggleWorkhub(connection, true);
+ const scope = { sessionId, originHostEpoch: connection.hostEpoch };
+ const reordered = await request('queue.entries.reorder', {
+ ...scope,
+ reorderId: 'reorder',
+ entryIds: ['discarded', 'followup', 'promoted'],
+ });
+ const edit = {
+ ...scope,
+ updateId: 'edit',
+ entryId: 'followup',
+ expectedQueueRevision: reordered.queueRevision,
+ text: 'EDITED_FOLLOWUP',
+ };
+ await assert.rejects(
+ request('queue.entry.update', edit),
+ (error) => error.code === 'operation_unavailable',
+ );
+ await toggleWorkhub(connection, false);
+ await request('queue.entry.update', edit);
+ await request('queue.entry.promote', { ...scope, promoteId: 'promote', entryId: 'promoted' });
+ await toggleWorkhub(connection, true);
+ const retract = { ...scope, retractId: 'discard', entryId: 'discarded' };
+ assert.deepEqual(
+ await request('queue.entry.retract', retract),
+ await request('queue.entry.retract', retract),
+ );
+ assert.deepEqual(
+ (
+ await request('turn.message.execution.query', {
+ sessionId,
+ messageIds: ['discarded'],
+ })
+ ).resolutions,
+ [{ messageId: 'discarded', state: 'cancelled' }],
+ );
+ await toggleWorkhub(connection, false);
+ remote = await workhubRemote(connection);
+ remotes.push(remote);
+ const replay = remote.method('enqueue');
+ assert.deepEqual(
+ await replay(input),
+ receipt,
+ 'editing cannot change the original admission receipt',
+ );
release.resolve();
const deadline = Date.now() + 8000;
let resolutions;
@@ -167,10 +216,10 @@ export async function verifyWorkhubQueue(connection) {
if (failure) throw failure;
const page = await request('turn.message.execution.query', {
sessionId,
- messageIds: ['steering', 'followup'],
+ messageIds: ['steering', 'promoted', 'followup'],
});
resolutions = page.resolutions;
- if (resolutions.length === 2 && resolutions.every((item) => item.state === 'owned')) {
+ if (resolutions.length === 3 && resolutions.every((item) => item.state === 'owned')) {
const turns = await Promise.all(
resolutions.map((item) => request('turn.query', { sessionId, turnId: item.turnId })),
);
@@ -180,21 +229,24 @@ export async function verifyWorkhubQueue(connection) {
await delay(10);
}
assert.equal(resolutions.find((item) => item.messageId === 'steering').turnId, turnId);
+ assert.equal(resolutions.find((item) => item.messageId === 'promoted').turnId, turnId);
assert.notEqual(resolutions.find((item) => item.messageId === 'followup').turnId, turnId);
assert.equal(requests.length, 3);
assert(JSON.stringify(requests[1].messages).includes('STEERING'));
- assert(JSON.stringify(requests[2].messages).includes('FOLLOWUP'));
+ assert(JSON.stringify(requests[1].messages).includes('PROMOTED'));
+ assert(JSON.stringify(requests[2].messages).includes('EDITED_FOLLOWUP'));
+ assert(!JSON.stringify(requests).includes('DISCARDED'));
await assert.rejects(
- enqueue({ ...input, messageId: 'late' }),
+ replay({ ...input, messageId: 'late' }),
(error) => error.code === 'operation_conflict',
);
assert.equal(
- (await enqueue(input)).disposition,
+ (await replay(input)).disposition,
'followup',
'exact replay precedes active-Turn checks',
);
await toggleWorkhub(connection, true);
- await assert.rejects(enqueue({ ...input, messageId: 'retired' }));
+ await assert.rejects(replay({ ...input, messageId: 'retired' }));
await toggleWorkhub(connection, false);
remote = await workhubRemote(connection);
remotes.push(remote);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org