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/14 16:09:42 UTC
(maka) 10/20: feat(runtime-host): project canonical WorkHub assignments
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
commit 7c23e4697d0f6df69f9627bb20d5ac5f21cc0d4f
Author: Wang <m4...@gmail.com>
AuthorDate: Mon Sep 14 20:27:53 2026 +0800
feat(runtime-host): project canonical WorkHub assignments
Generated-by: OpenAI Codex
---
.../migrations/0021_workhub_assignment_target.sql | 25 +++++
crates/event-log/src/observation/delivery.rs | 5 +-
crates/event-log/src/schema.rs | 2 +-
crates/event-log/src/transcript.rs | 2 +-
crates/event-log/src/transcript/evidence.rs | 23 ++++-
crates/event-log/src/workhub.rs | 39 ++++++-
crates/event-log/src/workhub/candidates.rs | 23 ++++-
crates/event-log/tests/migrations.rs | 10 +-
crates/event-log/tests/workhub.rs | 4 +-
crates/event-log/tests/workhub_attachments.rs | 1 +
crates/event-log/tests/workhub_candidates.rs | 4 +-
crates/event-log/tests/workhub_creation.rs | 43 +++++++-
crates/event-log/tests/workhub_resume.rs | 1 +
crates/event-log/tests/workhub_revision.rs | 41 +++++++-
crates/event-log/tests/workhub_stop.rs | 1 +
crates/presentation/src/lib.rs | 21 +++-
crates/presentation/src/message.rs | 2 +-
crates/presentation/src/workhub.rs | 109 ++++++++++++++++++-
crates/protocol/src/session_types.rs | 13 +--
crates/protocol/src/workhub/action.rs | 21 +---
.../src/server/subscriptions/delivery/events.rs | 1 +
.../src/server/subscriptions/streams.rs | 1 +
crates/runtime-host/src/server/workhub/action.rs | 1 +
.../src/server/workhub/action/target.rs | 33 +++++-
.../runtime-host/src/server/workhub/candidates.rs | 9 +-
crates/runtime/src/execution.rs | 13 +++
crates/runtime/src/workhub.rs | 9 ++
crates/runtime/src/workhub/description.rs | 102 ++++++++++++++++++
docs/rust-runtime.md | 4 +-
docs/rust-runtime.zh-CN.md | 3 +-
tests/fixtures/client-workhub-delegation.mjs | 115 +++++++++++++++------
tests/fixtures/client-workhub-routing.mjs | 4 +-
32 files changed, 583 insertions(+), 102 deletions(-)
diff --git a/crates/event-log/migrations/0021_workhub_assignment_target.sql b/crates/event-log/migrations/0021_workhub_assignment_target.sql
new file mode 100644
index 0000000000..688732401a
--- /dev/null
+++ b/crates/event-log/migrations/0021_workhub_assignment_target.sql
@@ -0,0 +1,25 @@
+/*
+ * 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.
+ */
+
+
+CREATE INDEX workhub_assignment_target ON event_log(
+ json_extract(event_json, '$.fact.delegation.target.session_id'), sequence DESC
+) WHERE kind = 'workhub_delegated';
+
+PRAGMA user_version = 21;
diff --git a/crates/event-log/src/observation/delivery.rs b/crates/event-log/src/observation/delivery.rs
index 35bdee3c73..c901c71f9c 100644
--- a/crates/event-log/src/observation/delivery.rs
+++ b/crates/event-log/src/observation/delivery.rs
@@ -38,6 +38,7 @@ pub struct StoreStreamEvent {
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum StreamFact {
InvocationOpened,
+ WorkhubDelegated,
MessageSteered,
PartStarted {
step_id: String,
@@ -235,7 +236,7 @@ SELECT json_object(
WHEN kind = 'tool_settled' THEN
json_object('kind', kind, 'operation_id', operation_id,
'outcome', json_extract(event_json, '$.fact.outcome.kind'))
- WHEN kind IN ('invocation_opened', 'message_steered') THEN json_object('kind', kind)
+ WHEN kind IN ('invocation_opened', 'message_steered', 'workhub_delegated') THEN json_object('kind', kind)
WHEN kind = 'invocation_ended' THEN json_object('kind', kind,
'failed', json(CASE WHEN json_extract(event_json, '$.fact.outcome.kind') = 'failed' THEN 'true' ELSE 'false' END))
WHEN kind IN ('model_completed', 'model_interrupted') THEN
@@ -273,7 +274,7 @@ AND (kind NOT IN ('model_observed', 'model_completed', 'model_interrupted') OR
AND json_extract(opening.event_json, '$.fact.input.kind') IN ('message', 'continuation')
AND COALESCE(json_extract(request.event_json, '$.fact.purpose'), 'main') = 'main'))
AND (kind IN ('invocation_opened', 'message_steered', 'invocation_ended', 'model_completed', 'model_interrupted',
- 'tool_dispatched', 'tool_rejected', 'tool_settled')
+ 'tool_dispatched', 'tool_rejected', 'tool_settled', 'workhub_delegated')
OR (kind = 'model_observed'
AND json_extract(event_json, '$.fact.event.kind') IN
('part_started', 'part_delta', 'part_finished')))
diff --git a/crates/event-log/src/schema.rs b/crates/event-log/src/schema.rs
index 0442d5bfa3..50f1554528 100644
--- a/crates/event-log/src/schema.rs
+++ b/crates/event-log/src/schema.rs
@@ -38,7 +38,7 @@ pub(crate) async fn initialize_connection(
.fetch_one(&mut *connection)
.await?;
if !((application_id == 0 && version == 0 && tables == 0)
- || (application_id == 0x4d414b52 && matches!(version, 0..=20)))
+ || (application_id == 0x4d414b52 && matches!(version, 0..=21)))
{
return Err(StoreError::UnsupportedDatabase);
}
diff --git a/crates/event-log/src/transcript.rs b/crates/event-log/src/transcript.rs
index 9af9a4d722..21d7467765 100644
--- a/crates/event-log/src/transcript.rs
+++ b/crates/event-log/src/transcript.rs
@@ -131,7 +131,7 @@ impl EventLog {
AND COALESCE(json_extract(request.event_json, '$.fact.purpose'), 'main') = 'main'))
AND kind IN ('invocation_opened', 'message_steered', 'model_completed',
'model_interrupted', 'invocation_ended',
- 'tool_dispatched', 'tool_rejected', 'tool_settled')
+ 'tool_dispatched', 'tool_rejected', 'tool_settled', 'workhub_delegated')
UNION ALL
SELECT sequence, NULL, NULL FROM session_events
WHERE json_extract(event_json, '$.session_id') = ?1
diff --git a/crates/event-log/src/transcript/evidence.rs b/crates/event-log/src/transcript/evidence.rs
index a964939b01..aaf1df3c8f 100644
--- a/crates/event-log/src/transcript/evidence.rs
+++ b/crates/event-log/src/transcript/evidence.rs
@@ -49,15 +49,30 @@ pub(super) async fn selected(
boundary: bool,
) -> Result<Vec<StoredEvent>, StoreError> {
if boundary {
- let tool: bool = sqlx::query_scalar(
- "SELECT kind IN ('tool_dispatched', 'tool_rejected', 'tool_settled')
- FROM runtime_events WHERE sequence = ? AND invocation_id = ?",
+ let kind: String = sqlx::query_scalar(
+ "SELECT kind FROM runtime_events WHERE sequence = ? AND invocation_id = ?",
)
.bind(through)
.bind(invocation)
.fetch_one(&mut *tx)
.await?;
- if tool {
+ if kind == "workhub_delegated" {
+ return read(
+ tx,
+ sqlx::query(
+ "SELECT sequence, length(CAST(event_json AS BLOB)) FROM runtime_events
+ WHERE invocation_id = ?1 AND sequence <= ?2
+ AND (kind = 'invocation_opened' OR sequence = ?2) ORDER BY sequence",
+ )
+ .bind(invocation)
+ .bind(through),
+ )
+ .await;
+ }
+ if matches!(
+ kind.as_str(),
+ "tool_dispatched" | "tool_rejected" | "tool_settled"
+ ) {
return super::tools::selected(tx, invocation, through).await;
}
}
diff --git a/crates/event-log/src/workhub.rs b/crates/event-log/src/workhub.rs
index 62b27991b8..f9822b6ead 100644
--- a/crates/event-log/src/workhub.rs
+++ b/crates/event-log/src/workhub.rs
@@ -31,7 +31,7 @@ mod control;
mod create;
mod resume;
pub mod stop;
-pub use candidates::activity_at;
+pub use candidates::{Candidate, activity_at};
/// The pending target and the action fact share append's transaction. Exact
/// event replay returns before this hook, so it never recreates consumed work.
@@ -119,13 +119,48 @@ pub(crate) async fn apply(
} => (
configuration_digest.clone(),
maka_runtime::artifact::content_digest(
- configuration.ok_or(StoreError::PrefixTooLarge)?.as_bytes(),
+ configuration
+ .as_ref()
+ .ok_or(StoreError::PrefixTooLarge)?
+ .as_bytes(),
),
),
};
if expected != actual {
return Err(StoreError::RevisionConflict { expected, actual });
}
+ if let Some(description) = &delegation.description {
+ #[derive(serde::Deserialize)]
+ struct DescriptionEvidence {
+ name: String,
+ model: Option<maka_runtime::execution::ModelBinding>,
+ permission_mode: Option<maka_runtime::execution::PermissionMode>,
+ }
+ let evidence: DescriptionEvidence =
+ serde_json::from_str(configuration.as_ref().ok_or(StoreError::PrefixTooLarge)?)?;
+ if evidence.name != description.name() {
+ return Err(invalid("WorkHub target description changed"));
+ }
+ if let maka_runtime::workhub::DelegationDescription::Created { spec, .. } = description
+ && let Some(defaults) = &spec.defaults
+ {
+ if defaults
+ .permission_mode
+ .is_some_and(|mode| Some(mode) != evidence.permission_mode)
+ {
+ return Err(invalid("WorkHub creation permission choice changed"));
+ }
+ if let Some(choice) = &defaults.model
+ && evidence.model.as_ref().is_none_or(|model| {
+ model.connection_id != choice.llm_connection_id
+ || model.connection_slug != choice.llm_connection_slug
+ || model.model != choice.model
+ })
+ {
+ return Err(invalid("WorkHub creation model choice changed"));
+ }
+ }
+ }
let owner = delegation
.delivery
.is_steering()
diff --git a/crates/event-log/src/workhub/candidates.rs b/crates/event-log/src/workhub/candidates.rs
index f7ebeeca39..404bf6fe7b 100644
--- a/crates/event-log/src/workhub/candidates.rs
+++ b/crates/event-log/src/workhub/candidates.rs
@@ -25,6 +25,11 @@ use maka_runtime::event::Invocation;
use serde::de::DeserializeOwned;
use sqlx::Connection;
+pub struct Candidate<T> {
+ pub session: SessionRecord<T>,
+ pub latest_delegation_action_id: Option<String>,
+}
+
impl EventLog {
/// Revalidate an offered target independently of the current display window.
pub async fn workhub_candidate<T, F>(
@@ -60,7 +65,7 @@ impl EventLog {
pub async fn workhub_candidates<T, F>(
&self,
eligible: F,
- ) -> Result<Vec<SessionRecord<T>>, StoreError>
+ ) -> Result<Vec<Candidate<T>>, StoreError>
where
T: DeserializeOwned + Send + Sync + 'static,
F: Fn(&SessionRecord<T>) -> bool + Send + 'static,
@@ -100,8 +105,22 @@ impl EventLog {
}
cursor = ids.last().unwrap().clone();
}
+ let mut result = Vec::with_capacity(candidates.len());
+ for session in candidates {
+ let latest_delegation_action_id = sqlx::query_scalar(
+ "SELECT json_extract(assignment.event_json, '$.fact.delegation.action_id')
+ FROM runtime_events assignment
+ WHERE assignment.kind = 'workhub_delegated'
+ AND json_extract(assignment.event_json, '$.fact.delegation.target.session_id') = ?
+ AND NOT EXISTS (SELECT 1 FROM workhub_stops stop
+ WHERE stop.delegation_action_id = json_extract(assignment.event_json, '$.fact.delegation.action_id')
+ AND json_extract(stop.resolution_json, '$.outcome') != 'not_owned')
+ ORDER BY assignment.sequence DESC LIMIT 1",
+ ).bind(&session.id).fetch_optional(&mut *tx).await?;
+ result.push(Candidate { session, latest_delegation_action_id });
+ }
tx.commit().await?;
- Ok(candidates)
+ Ok(result)
})
})
.await
diff --git a/crates/event-log/tests/migrations.rs b/crates/event-log/tests/migrations.rs
index 22e54cbba6..35dca59d50 100644
--- a/crates/event-log/tests/migrations.rs
+++ b/crates/event-log/tests/migrations.rs
@@ -113,7 +113,7 @@ async fn embedded_migration_adopts_only_rust_schema_and_reopens_without_rewritin
.map(|(version, _)| *version)
.collect::<Vec<_>>(),
vec![
- 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20
+ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21
]
);
assert!(checksums.iter().all(|(_, checksum)| checksum.len() == 48));
@@ -201,7 +201,7 @@ async fn embedded_migration_adopts_only_rust_schema_and_reopens_without_rewritin
.query_row("SELECT count(*) FROM _sqlx_migrations", [], |row| row
.get::<_, i64>(0))
.unwrap(),
- 20
+ 21
);
}
@@ -270,7 +270,7 @@ async fn mismatched_and_unknown_migrations_fail_closed_without_touching_committe
.query_row("SELECT count(*) FROM _sqlx_migrations", [], |row| row
.get::<_, i64>(0))
.unwrap(),
- 20
+ 21
);
}
@@ -332,7 +332,7 @@ async fn interrupted_initial_migration_remains_openable_under_the_rust_applicati
connection
.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
.unwrap(),
- 20
+ 21
);
assert_eq!(
connection
@@ -342,6 +342,6 @@ async fn interrupted_initial_migration_remains_openable_under_the_rust_applicati
|row| row.get::<_, i64>(0)
)
.unwrap(),
- 20
+ 21
);
}
diff --git a/crates/event-log/tests/workhub.rs b/crates/event-log/tests/workhub.rs
index 414ad7f962..32a052371a 100644
--- a/crates/event-log/tests/workhub.rs
+++ b/crates/event-log/tests/workhub.rs
@@ -82,6 +82,7 @@ async fn steering_keeps_delivery_ownership_across_waiting_shells_and_terminal_ha
.unwrap();
let delegation = Delegation {
kind: Default::default(),
+ description: None,
delivery: DelegationDelivery::Steering {
configuration_digest: basis.configuration_digest.clone(),
},
@@ -131,7 +132,7 @@ async fn steering_keeps_delivery_ownership_across_waiting_shells_and_terminal_ha
.await
.unwrap()
.iter()
- .all(|record| record.id != "target")
+ .all(|candidate| candidate.session.id != "target")
);
assert!(log.pending_messages("target").await.unwrap().is_empty());
log.commit_interaction_outcome(
@@ -303,6 +304,7 @@ async fn delegation_is_atomic_and_replay_does_not_reassign_or_requeue() {
let target = invocation("target", "delegated");
let delegation = Delegation {
kind: Default::default(),
+ description: None,
delivery: Default::default(),
action_id: "action".into(),
request_fingerprint: content_digest(b"bound proposal"),
diff --git a/crates/event-log/tests/workhub_attachments.rs b/crates/event-log/tests/workhub_attachments.rs
index f236206bb9..7e5149c52b 100644
--- a/crates/event-log/tests/workhub_attachments.rs
+++ b/crates/event-log/tests/workhub_attachments.rs
@@ -101,6 +101,7 @@ async fn delegation_copies_attachments_atomically_and_replays_without_the_delete
.unwrap();
let delegation = Delegation {
kind: Default::default(),
+ description: None,
delivery: Default::default(),
action_id: "copy-action".into(),
request_fingerprint: content_digest(b"copy request"),
diff --git a/crates/event-log/tests/workhub_candidates.rs b/crates/event-log/tests/workhub_candidates.rs
index 14f36001d4..979a5acf6b 100644
--- a/crates/event-log/tests/workhub_candidates.rs
+++ b/crates/event-log/tests/workhub_candidates.rs
@@ -109,12 +109,12 @@ async fn candidates_rank_all_pages_by_canonical_activity_after_filtering_unsafe_
assert_eq!(
candidates
.iter()
- .map(|record| record.id.clone())
+ .map(|candidate| candidate.session.id.clone())
.collect::<Vec<_>>(),
expected
);
assert_eq!(
- maka_event_log::workhub::activity_at(&candidates[0]),
+ maka_event_log::workhub::activity_at(&candidates[0].session),
237_000
);
log.close().await.unwrap();
diff --git a/crates/event-log/tests/workhub_creation.rs b/crates/event-log/tests/workhub_creation.rs
index 79623928ae..6f40177dd3 100644
--- a/crates/event-log/tests/workhub_creation.rs
+++ b/crates/event-log/tests/workhub_creation.rs
@@ -21,8 +21,12 @@ use maka_event_log::EventLog;
use maka_runtime::{
artifact::content_digest,
event::{EventWrite, Fact, Invocation, InvocationOutcome, RuntimeEvent},
+ execution::{PermissionMode, WorkspaceTarget},
input::InvocationInput,
- workhub::{COORDINATION_SESSION_ID, Delegation, DelegationKind, created_session_id},
+ workhub::{
+ COORDINATION_SESSION_ID, CreateDefaults, CreateModel, CreateSpec, Delegation,
+ DelegationDescription, DelegationKind, created_session_id,
+ },
};
use serde_json::{Value, json};
@@ -61,6 +65,23 @@ async fn workhub_creation_rolls_back_with_its_action_and_replay_preserves_later_
delegation: Box::new(Delegation {
action_id: "create-action".into(),
kind: DelegationKind::Created,
+ description: Some(DelegationDescription::Created {
+ name: "new task".into(),
+ spec: CreateSpec {
+ title: " new task ".into(),
+ workspace: WorkspaceTarget::HostPath {
+ path: "/workspace/.".into(),
+ },
+ defaults: Some(CreateDefaults {
+ permission_mode: Some(PermissionMode::Explore),
+ model: Some(CreateModel {
+ llm_connection_id: "connection".into(),
+ llm_connection_slug: "fixture".into(),
+ model: "fixture-model".into(),
+ }),
+ }),
+ },
+ }),
delivery: Default::default(),
request_fingerprint: content_digest(b"new request"),
source_message_event_id: opening.event().id.clone(),
@@ -76,7 +97,21 @@ async fn workhub_creation_rolls_back_with_its_action_and_replay_preserves_later_
},
))
.unwrap();
- let config = json!({"name": "new task"});
+ let config = json!({"name": "new task", "permission_mode": "explore",
+ "model": {"connection_id": "connection", "connection_slug": "fixture", "model": "fixture-model"}});
+ for (field, value) in [
+ ("permission_mode", json!("bypass")),
+ (
+ "model",
+ json!({"connection_id": "other", "connection_slug": "fixture", "model": "fixture-model"}),
+ ),
+ ] {
+ let mut changed = config.clone();
+ changed[field] = value;
+ assert!(log.create_workhub_session(&action, &changed).await.is_err());
+ assert!(log.get_session::<Value>(&target).await.unwrap().is_none());
+ assert!(log.workhub_action("create-action").await.unwrap().is_none());
+ }
let before = log
.list_sessions::<Value>(None, None, 32)
.await
@@ -155,7 +190,9 @@ async fn workhub_creation_rolls_back_with_its_action_and_replay_preserves_later_
);
let record = log.get_session::<Value>(&target).await.unwrap().unwrap();
assert_eq!(record.revision, 2);
- assert_eq!(record.configuration, json!({"name": "renamed"}));
+ let mut renamed = config.clone();
+ renamed["name"] = json!("renamed");
+ assert_eq!(record.configuration, renamed);
assert_eq!(log.pending_messages(&target).await.unwrap().len(), 1);
log.close().await.unwrap();
}
diff --git a/crates/event-log/tests/workhub_resume.rs b/crates/event-log/tests/workhub_resume.rs
index 1ab6f73031..ad982c3076 100644
--- a/crates/event-log/tests/workhub_resume.rs
+++ b/crates/event-log/tests/workhub_resume.rs
@@ -76,6 +76,7 @@ async fn linked_resume_is_atomic_and_follows_its_message_not_other_delegations_o
.revision;
let delegation = Delegation {
kind: Default::default(),
+ description: None,
delivery: Default::default(),
action_id: id.into(),
request_fingerprint: content_digest(id.as_bytes()),
diff --git a/crates/event-log/tests/workhub_revision.rs b/crates/event-log/tests/workhub_revision.rs
index f321c99291..bf32714732 100644
--- a/crates/event-log/tests/workhub_revision.rs
+++ b/crates/event-log/tests/workhub_revision.rs
@@ -22,7 +22,7 @@ use maka_runtime::{
artifact::content_digest,
event::{CommitError, EventWrite, Fact, Invocation, RuntimeEvent},
input::InvocationInput,
- workhub::{COORDINATION_SESSION_ID, Delegation},
+ workhub::{COORDINATION_SESSION_ID, Delegation, DelegationDescription},
};
use serde_json::{Value, json};
@@ -65,6 +65,9 @@ async fn target_metadata_change_invalidates_uncommitted_delegation_but_not_its_d
assert_eq!(boundary.opening_event_id, source.event().id);
let mut delegation = Delegation {
kind: Default::default(),
+ description: Some(DelegationDescription::Existing {
+ name: "original".into(),
+ }),
delivery: Default::default(),
action_id: "action".into(),
request_fingerprint: content_digest(b"request"),
@@ -101,6 +104,13 @@ async fn target_metadata_change_invalidates_uncommitted_delegation_but_not_its_d
assert!(log.pending_messages("target").await.unwrap().is_empty());
assert!(log.workhub_action("action").await.unwrap().is_none());
delegation.target_revision = 2;
+ assert!(
+ log.append(&write(delegation.clone())).await.is_err(),
+ "a fresh revision cannot authorize a false frozen name"
+ );
+ delegation.description = Some(DelegationDescription::Existing {
+ name: "changed after candidate selection".into(),
+ });
let action = write(delegation);
let receipt = log.append(&action).await.unwrap();
log.update_session_metadata("target", 2, |config: &mut Value| {
@@ -111,5 +121,34 @@ async fn target_metadata_change_invalidates_uncommitted_delegation_but_not_its_d
.unwrap();
assert_eq!(log.append(&action).await.unwrap(), receipt);
assert_eq!(log.pending_messages("target").await.unwrap().len(), 1);
+ assert!(
+ log.prepare_transcript(COORDINATION_SESSION_ID, receipt, 32)
+ .await
+ .unwrap()
+ );
+ let rows = log
+ .transcript_headers(
+ COORDINATION_SESSION_ID,
+ &maka_event_log::transcript::TranscriptRead {
+ through: maka_presentation::watermark(receipt).unwrap(),
+ position: 0,
+ direction: maka_event_log::transcript::TranscriptDirection::Newer,
+ limit: 16,
+ },
+ )
+ .await
+ .unwrap();
+ let row = rows.last().unwrap();
+ let bytes = log
+ .transcript_fragment(COORDINATION_SESSION_ID, row.sequence, 0, row.total_bytes)
+ .await
+ .unwrap();
+ let record: Value = serde_json::from_slice(&bytes).unwrap();
+ assert_eq!(record["kind"], "delegation_assigned");
+ assert_eq!(
+ record["targetSessionName"],
+ "changed after candidate selection"
+ );
+ assert_eq!(record["delegationId"], action.event().id);
log.close().await.unwrap();
}
diff --git a/crates/event-log/tests/workhub_stop.rs b/crates/event-log/tests/workhub_stop.rs
index 8418f29944..a96cef9ea7 100644
--- a/crates/event-log/tests/workhub_stop.rs
+++ b/crates/event-log/tests/workhub_stop.rs
@@ -52,6 +52,7 @@ async fn stop_claim_cancellation_and_recovery_keep_the_original_message_and_exac
let mut target = opening("target", None);
let delegation = Delegation {
kind: Default::default(),
+ description: None,
delivery: Default::default(),
action_id: "delegation".into(),
request_fingerprint: content_digest(b"delegation"),
diff --git a/crates/presentation/src/lib.rs b/crates/presentation/src/lib.rs
index fefa70d38e..8ba5e3bd4b 100644
--- a/crates/presentation/src/lib.rs
+++ b/crates/presentation/src/lib.rs
@@ -75,6 +75,7 @@ pub struct InvocationView {
max_text_bytes: usize,
tools: tools::Tools,
summary_step: Option<String>,
+ workhub_source: Option<(String, maka_runtime::input::MessageInput)>,
}
impl InvocationView {
@@ -88,6 +89,7 @@ impl InvocationView {
max_text_bytes,
tools: tools::Tools::new(max_text_bytes),
summary_step: None,
+ workhub_source: None,
})
}
@@ -150,6 +152,10 @@ impl InvocationView {
..
} = input
{
+ if event.invocation.session_id == maka_runtime::workhub::COORDINATION_SESSION_ID
+ {
+ self.workhub_source = Some((event.id.clone(), content.clone()));
+ }
messages.push(user::project(
if source_messages.len() == 1 {
&source_messages[0].message.message_id
@@ -278,10 +284,23 @@ impl InvocationView {
| Fact::ToolRejected { .. } => {
messages.extend(self.tools.boundary(event, ts, resolved)?);
}
+ Fact::WorkhubDelegated { delegation } => {
+ let (opening, source) =
+ self.workhub_source
+ .as_ref()
+ .ok_or(ProjectionError::Invalid(
+ "WorkHub assignment has no user source",
+ ))?;
+ if *opening != delegation.source_message_event_id {
+ return Err(ProjectionError::Invalid(
+ "WorkHub assignment source changed",
+ ));
+ }
+ messages.extend(workhub::assigned(event, delegation, source)?);
+ }
Fact::InvocationOpened { .. } => unreachable!(),
Fact::ContextCheckpointRecorded { .. }
| Fact::ToolResultArchived { .. }
- | Fact::WorkhubDelegated { .. }
| Fact::WorkhubResumeObserved { .. } => {}
}
}
diff --git a/crates/presentation/src/message.rs b/crates/presentation/src/message.rs
index 6c4f605e31..44b602c309 100644
--- a/crates/presentation/src/message.rs
+++ b/crates/presentation/src/message.rs
@@ -85,7 +85,7 @@ pub struct Message {
pub enum Content {
WorkhubCoordination {
#[serde(flatten)]
- record: crate::workhub::StopMessage,
+ record: crate::workhub::CoordinationRecord,
},
User {
text: String,
diff --git a/crates/presentation/src/workhub.rs b/crates/presentation/src/workhub.rs
index c11ded982e..15fccc2960 100644
--- a/crates/presentation/src/workhub.rs
+++ b/crates/presentation/src/workhub.rs
@@ -24,6 +24,111 @@ use maka_runtime::{
};
use serde::Serialize;
+#[derive(Clone, Debug, PartialEq, Serialize)]
+#[serde(untagged)]
+pub enum CoordinationRecord {
+ Stop(Box<StopMessage>),
+ Assigned(Box<AssignmentMessage>),
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AssignmentMessage {
+ schema_version: u8,
+ kind: AssignmentKind,
+ action_id: String,
+ action_fingerprint: String,
+ coordination_turn_id: String,
+ delegation_id: String,
+ target_session_id: String,
+ target_turn_id: String,
+ target_message_id: String,
+ target_session_name: String,
+ disposition: Disposition,
+ user_text: String,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ attachments: Option<Vec<maka_runtime::attachment::AttachmentRef>>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ target_attachments: Option<Vec<maka_runtime::attachment::AttachmentRef>>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ delegation_text: Option<String>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ create: Option<maka_runtime::workhub::CreateSpec>,
+ #[serde(skip_serializing_if = "std::ops::Not::not")]
+ steered: bool,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+enum AssignmentKind {
+ DelegationAssigned,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+enum Disposition {
+ DelegateExisting,
+ CreateNew,
+}
+
+pub(crate) fn assigned(
+ event: &maka_runtime::event::RuntimeEvent,
+ delegation: &maka_runtime::workhub::Delegation,
+ source: &maka_runtime::input::MessageInput,
+) -> Result<Option<Message>, ProjectionError> {
+ use maka_runtime::workhub::DelegationDescription;
+ delegation
+ .validate(&event.invocation)
+ .map_err(ProjectionError::Invalid)?;
+ // Older facts have no canonical display description. Never substitute
+ // today's mutable Session metadata for missing historical evidence.
+ let Some(description) = &delegation.description else {
+ return Ok(None);
+ };
+ let (disposition, create) = match description {
+ DelegationDescription::Existing { .. } => (Disposition::DelegateExisting, None),
+ DelegationDescription::Created { spec, .. } => (Disposition::CreateNew, Some(spec.clone())),
+ };
+ let target_attachments = source
+ .attachments
+ .as_ref()
+ .map(|items| {
+ items
+ .iter()
+ .map(|item| delegation.attachment(item))
+ .collect::<Result<Vec<_>, _>>()
+ })
+ .transpose()
+ .map_err(ProjectionError::Invalid)?;
+ Ok(Some(Message {
+ id: event.id.clone(),
+ turn_id: event.invocation.turn_id.clone(),
+ ts: crate::message::timestamp(event)?,
+ content: Content::WorkhubCoordination {
+ record: CoordinationRecord::Assigned(Box::new(AssignmentMessage {
+ schema_version: 1,
+ kind: AssignmentKind::DelegationAssigned,
+ action_id: delegation.action_id.clone(),
+ action_fingerprint: delegation.request_fingerprint.clone(),
+ coordination_turn_id: event.invocation.turn_id.clone(),
+ delegation_id: event.id.clone(),
+ target_session_id: delegation.target.session_id.clone(),
+ target_turn_id: delegation.target.turn_id.clone(),
+ target_message_id: delegation.target_message_id(),
+ target_session_name: description.name().into(),
+ disposition,
+ user_text: source.text.clone(),
+ attachments: source.attachments.clone(),
+ target_attachments,
+ delegation_text: (delegation.delegation_text != source.text)
+ .then(|| delegation.delegation_text.clone()),
+ create,
+ steered: delegation.delivery.is_steering(),
+ })),
+ },
+ }))
+}
+
#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StopMessage {
@@ -105,7 +210,7 @@ pub fn stop(
turn_id: event.turn_id.clone(),
ts: crate::message::capture_time(event.recorded_at)?,
content: Content::WorkhubCoordination {
- record: StopMessage {
+ record: CoordinationRecord::Stop(Box::new(StopMessage {
schema_version: 3,
action_id: intent.request.action_id.clone(),
action_fingerprint: intent.request.request_fingerprint.clone(),
@@ -114,7 +219,7 @@ pub fn stop(
stops_delegation_id: delegation_id.into(),
target_session_id: intent.request.target_session_id.clone(),
detail,
- },
+ })),
},
},
})
diff --git a/crates/protocol/src/session_types.rs b/crates/protocol/src/session_types.rs
index a4e7318d9d..e34e41ba7d 100644
--- a/crates/protocol/src/session_types.rs
+++ b/crates/protocol/src/session_types.rs
@@ -18,7 +18,7 @@
*/
pub use maka_runtime::execution::{
- CollaborationMode, OrchestrationMode, PermissionMode, ThinkingLevel,
+ CollaborationMode, OrchestrationMode, PermissionMode, ThinkingLevel, WorkspaceTarget,
};
use serde::{Deserialize, Serialize};
@@ -37,17 +37,6 @@ wire_enum!(RevisionState { Preparing=>"preparing", Committed=>"committed" });
wire_enum!(SessionLifecycleState { Active=>"active", Archived=>"archived" });
wire_enum!(UnsupportedReason { NotWireRepresentable=>"not_wire_representable" });
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-#[serde(
- tag = "kind",
- rename_all = "snake_case",
- rename_all_fields = "camelCase",
- deny_unknown_fields
-)]
-pub enum WorkspaceTarget {
- Project { project_id: String },
- HostPath { path: String },
-}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct WorkspaceProjection {
diff --git a/crates/protocol/src/workhub/action.rs b/crates/protocol/src/workhub/action.rs
index 9a6856f02e..b2ba12fe70 100644
--- a/crates/protocol/src/workhub/action.rs
+++ b/crates/protocol/src/workhub/action.rs
@@ -18,10 +18,8 @@
*/
use super::candidates::{text, workspace};
-use crate::{
- ProtocolError, Result,
- session::{PermissionMode, WorkspaceTarget},
-};
+use crate::{ProtocolError, Result, session::WorkspaceTarget};
+pub use maka_runtime::workhub::CreateDefaults;
use serde::{Deserialize, Serialize};
use serde_json::Value;
@@ -88,21 +86,6 @@ pub struct LinkedTarget {
pub struct CreateContext {
pub workspace: WorkspaceTarget,
}
-#[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase", deny_unknown_fields)]
-pub struct CreateDefaults {
- #[serde(skip_serializing_if = "Option::is_none")]
- pub model: Option<CreateModel>,
- #[serde(skip_serializing_if = "Option::is_none")]
- pub permission_mode: Option<PermissionMode>,
-}
-#[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(rename_all = "camelCase", deny_unknown_fields)]
-pub struct CreateModel {
- pub llm_connection_id: String,
- pub llm_connection_slug: String,
- pub model: String,
-}
#[derive(Debug, Serialize, Deserialize)]
#[serde(
tag = "disposition",
diff --git a/crates/runtime-host/src/server/subscriptions/delivery/events.rs b/crates/runtime-host/src/server/subscriptions/delivery/events.rs
index 70a9915480..97dbd1c572 100644
--- a/crates/runtime-host/src/server/subscriptions/delivery/events.rs
+++ b/crates/runtime-host/src/server/subscriptions/delivery/events.rs
@@ -74,6 +74,7 @@ impl Delivery {
if matches!(
stored.fact,
StreamFact::InvocationOpened
+ | StreamFact::WorkhubDelegated
| StreamFact::MessageSteered
| StreamFact::StepEnded { .. }
| StreamFact::InvocationEnded { .. }
diff --git a/crates/runtime-host/src/server/subscriptions/streams.rs b/crates/runtime-host/src/server/subscriptions/streams.rs
index 449d21a4de..47c30289a9 100644
--- a/crates/runtime-host/src/server/subscriptions/streams.rs
+++ b/crates/runtime-host/src/server/subscriptions/streams.rs
@@ -137,6 +137,7 @@ impl Streams {
self.finish_matching(|_| true, interrupted, &mut deltas);
}
StreamFact::InvocationOpened
+ | StreamFact::WorkhubDelegated
| StreamFact::MessageSteered
| StreamFact::ToolDispatched { .. }
| StreamFact::ToolRejected { .. }
diff --git a/crates/runtime-host/src/server/workhub/action.rs b/crates/runtime-host/src/server/workhub/action.rs
index ec13301cf6..898a8af7c5 100644
--- a/crates/runtime-host/src/server/workhub/action.rs
+++ b/crates/runtime-host/src/server/workhub/action.rs
@@ -165,6 +165,7 @@ async fn admit(
};
let delegation = Delegation {
kind: target.kind(),
+ description: Some(target.description()),
delivery,
action_id: input.action_id,
request_fingerprint: fingerprint,
diff --git a/crates/runtime-host/src/server/workhub/action/target.rs b/crates/runtime-host/src/server/workhub/action/target.rs
index b6f44dc5ee..52fecac9cd 100644
--- a/crates/runtime-host/src/server/workhub/action/target.rs
+++ b/crates/runtime-host/src/server/workhub/action/target.rs
@@ -23,20 +23,37 @@ use maka_protocol::{
session::*,
workhub::{ActInput, Proposal, RoutingProposal},
};
-use maka_runtime::workhub::{DelegationKind, created_session_id};
+use maka_runtime::workhub::{
+ CreateSpec, DelegationDescription, DelegationKind, created_session_id,
+};
pub(super) enum Target {
Existing {
id: String,
+ name: String,
revision: u64,
configuration_digest: String,
},
Created {
id: String,
- configuration: SessionConfiguration,
+ configuration: Box<SessionConfiguration>,
+ spec: CreateSpec,
},
}
impl Target {
+ pub(super) fn description(&self) -> DelegationDescription {
+ match self {
+ Self::Existing { name, .. } => DelegationDescription::Existing { name: name.clone() },
+ Self::Created {
+ spec,
+ configuration,
+ ..
+ } => DelegationDescription::Created {
+ name: configuration.name.clone(),
+ spec: spec.clone(),
+ },
+ }
+ }
pub(super) fn id(&self) -> &str {
match self {
Self::Existing { id, .. } | Self::Created { id, .. } => id,
@@ -89,6 +106,7 @@ pub(super) async fn prepare(
})?;
return Ok(Target::Existing {
id: record.id,
+ name: record.configuration.name,
revision: record.revision,
configuration_digest: record.configuration_digest,
});
@@ -108,6 +126,7 @@ pub(super) async fn prepare(
.find_map(|(candidate, record)| {
(&candidate.candidate_ref == candidate_ref).then_some(Target::Existing {
id: record.id,
+ name: record.configuration.name,
revision: record.revision,
configuration_digest: record.configuration_digest,
})
@@ -163,7 +182,15 @@ pub(super) async fn prepare(
.await
.map_err(context_error)?;
super::super::super::projects::record_usage(host, &configuration.workspace).await?;
- Ok(Target::Created { id, configuration })
+ Ok(Target::Created {
+ id,
+ configuration: Box::new(configuration),
+ spec: CreateSpec {
+ title: title.clone(),
+ workspace: context.workspace.clone(),
+ defaults: input.new_work_defaults.clone(),
+ },
+ })
}
Proposal::Linked(_) => Err(failure(
Code::OperationUnavailable,
diff --git a/crates/runtime-host/src/server/workhub/candidates.rs b/crates/runtime-host/src/server/workhub/candidates.rs
index e49617ed12..d4814f13c5 100644
--- a/crates/runtime-host/src/server/workhub/candidates.rs
+++ b/crates/runtime-host/src/server/workhub/candidates.rs
@@ -54,7 +54,8 @@ pub(super) async fn query(host: &Arc<Host>) -> Result<Candidates, OperationError
.map_err(sessions::stored)?;
let mut candidates = records
.iter()
- .map(|record| {
+ .map(|candidate| {
+ let record = &candidate.session;
let updated_at = maka_event_log::workhub::activity_at(record);
let projection = sessions::projection::project(record.clone());
Candidate {
@@ -64,10 +65,14 @@ pub(super) async fn query(host: &Arc<Host>) -> Result<Candidates, OperationError
workspace: projection.workspace,
state: projection.status,
updated_at,
- latest_delegation_action_id: None,
+ latest_delegation_action_id: candidate.latest_delegation_action_id.clone(),
}
})
.collect::<Vec<_>>();
+ let records = records
+ .into_iter()
+ .map(|candidate| candidate.session)
+ .collect::<Vec<_>>();
let surface = candidates
.iter()
.zip(records.iter().map(|record| &record.configuration))
diff --git a/crates/runtime/src/execution.rs b/crates/runtime/src/execution.rs
index 2439e6868e..72268cc85a 100644
--- a/crates/runtime/src/execution.rs
+++ b/crates/runtime/src/execution.rs
@@ -109,6 +109,19 @@ pub struct InvocationConfiguration {
pub thinking_level: Option<ThinkingLevel>,
}
+/// User-selected workspace locator, before Host path resolution.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(
+ tag = "kind",
+ rename_all = "snake_case",
+ rename_all_fields = "camelCase",
+ deny_unknown_fields
+)]
+pub enum WorkspaceTarget {
+ Project { project_id: String },
+ HostPath { path: String },
+}
+
/// Intrinsic workspace identity is distinct from its current filesystem location.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(try_from = "String", into = "String")]
diff --git a/crates/runtime/src/workhub.rs b/crates/runtime/src/workhub.rs
index 395615998b..e80071ebb7 100644
--- a/crates/runtime/src/workhub.rs
+++ b/crates/runtime/src/workhub.rs
@@ -26,6 +26,9 @@ use crate::{
};
use serde::{Deserialize, Serialize};
+mod description;
+pub use description::{CreateDefaults, CreateModel, CreateSpec, DelegationDescription};
+
pub const COORDINATION_SESSION_ID: &str = "maka_workhub_coordination";
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -165,6 +168,9 @@ pub struct Delegation {
/// because the existing worker legitimately advances this revision.
pub target_revision: u64,
pub delegation_text: String,
+ /// Absent only in older facts that did not capture their display evidence.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub description: Option<DelegationDescription>,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
@@ -213,6 +219,9 @@ pub fn stop_abort_source(action_id: &str) -> String {
impl Delegation {
pub fn validate(&self, coordinator: &Invocation) -> Result<(), &'static str> {
use crate::interaction::entity_id;
+ if let Some(description) = &self.description {
+ description.validate(self.kind)?;
+ }
if coordinator.session_id != COORDINATION_SESSION_ID
|| self.target.session_id == COORDINATION_SESSION_ID
{
diff --git a/crates/runtime/src/workhub/description.rs b/crates/runtime/src/workhub/description.rs
new file mode 100644
index 0000000000..97d549aa8d
--- /dev/null
+++ b/crates/runtime/src/workhub/description.rs
@@ -0,0 +1,102 @@
+/*
+ * 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::DelegationKind;
+use crate::execution::{PermissionMode, WorkspaceTarget};
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub enum DelegationDescription {
+ Existing { name: String },
+ Created { name: String, spec: CreateSpec },
+}
+
+/// Original user choices, not defaults resolved from later Host configuration.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct CreateSpec {
+ pub title: String,
+ pub workspace: WorkspaceTarget,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub defaults: Option<CreateDefaults>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct CreateDefaults {
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub model: Option<CreateModel>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub permission_mode: Option<PermissionMode>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct CreateModel {
+ pub llm_connection_id: String,
+ pub llm_connection_slug: String,
+ pub model: String,
+}
+
+impl DelegationDescription {
+ pub fn name(&self) -> &str {
+ match self {
+ Self::Existing { name } | Self::Created { name, .. } => name,
+ }
+ }
+
+ pub(super) fn validate(&self, kind: DelegationKind) -> Result<(), &'static str> {
+ if self.name().trim().is_empty() || self.name().len() > 4096 {
+ return Err("invalid WorkHub target description");
+ }
+ match (self, kind) {
+ (Self::Existing { .. }, DelegationKind::Existing) => Ok(()),
+ (Self::Created { spec, .. }, DelegationKind::Created) => {
+ if spec.title.trim().is_empty() || spec.title.len() > 4096 {
+ return Err("invalid WorkHub creation title");
+ }
+ let locator = match &spec.workspace {
+ WorkspaceTarget::Project { project_id } => project_id,
+ WorkspaceTarget::HostPath { path } => path,
+ };
+ if locator.trim().is_empty() || locator.contains('\0') || locator.len() > 32768 {
+ return Err("invalid WorkHub creation workspace");
+ }
+ if let Some(model) = spec
+ .defaults
+ .as_ref()
+ .and_then(|defaults| defaults.model.as_ref())
+ {
+ for field in [
+ &model.llm_connection_id,
+ &model.llm_connection_slug,
+ &model.model,
+ ] {
+ if field.trim().is_empty() || field.encode_utf16().count() > 512 {
+ return Err("invalid WorkHub creation model");
+ }
+ }
+ }
+ Ok(())
+ }
+ _ => Err("WorkHub description does not match its disposition"),
+ }
+ }
+}
diff --git a/docs/rust-runtime.md b/docs/rust-runtime.md
index 4786ec6f45..1e735622f5 100644
--- a/docs/rust-runtime.md
+++ b/docs/rust-runtime.md
@@ -165,7 +165,9 @@ Interactive target selection preserves the original offer and answer across retr
Linked stop and resume follow the delegated message's exact execution lineage,
with durable action replay. Stop requests and resolutions are Session-owned log facts;
recovery can publish a resolution after its coordinator Run ends without modifying that Run.
-Correction and complete WorkHub UI projection remain incomplete.
+Canonical assignment records preserve original creation choices and target-owned attachments;
+candidate queries expose the latest active association. Correction and full WorkHub Desktop
+acceptance remain incomplete.
Skill mutations, update previews and their transaction recovery remain incomplete.
Session branching, removal, import/export and recap, along with additional runtime
diff --git a/docs/rust-runtime.zh-CN.md b/docs/rust-runtime.zh-CN.md
index 5ac58a84f6..8dd1db71aa 100644
--- a/docs/rust-runtime.zh-CN.md
+++ b/docs/rust-runtime.zh-CN.md
@@ -134,7 +134,8 @@ Grep 差分测试需要 PATH 中有 `rg`；runtime 本身不依赖该可执行
WorkHub 已支持专属会话解析、查询、模型配置、受限 Desktop 工具与附件读取，以及候选发现和向已有或新建会话委派任务。
委派时原子转移附件；运行中目标通过普通持久消息队列接收 steering；交互式目标选择在重试时复用原候选表单与答案。
关联停止与恢复按委派消息追踪精确执行链，并持久化动作重放结果。停止请求和结果是 Session 所有的日志事实；
-恢复可在协调 Run 结束后发布结果，不修改已结束的 Run。修正及完整 WorkHub UI 投影仍未完成。
+恢复可在协调 Run 结束后发布结果，不修改已结束的 Run。结构化委派记录保留原始创建选项和目标附件，
+候选查询提供最新有效关联；修正及完整 WorkHub Desktop 验收仍未完成。
Skills 变更、更新预览及其事务恢复仍未完成。
会话分支、删除、导入导出、回顾及其它 runtime policy 设置也未完成。
diff --git a/tests/fixtures/client-workhub-delegation.mjs b/tests/fixtures/client-workhub-delegation.mjs
index efdaefb3fe..2c7e18948e 100644
--- a/tests/fixtures/client-workhub-delegation.mjs
+++ b/tests/fixtures/client-workhub-delegation.mjs
@@ -34,6 +34,33 @@ import { resumeDelegation } from './client-workhub-resume.mjs';
const sessionId = 'maka_workhub_coordination';
const turnId = 'delegate-request';
+async function coordinationRecord(observer, sequence, kind, actionId) {
+ for (;;) {
+ const frame = await observer.waitFor(
+ (frame) => frame.kind === 'subscription.transcript_advanced' && frame.sequence > sequence,
+ );
+ sequence = frame.sequence;
+ // loadTranscript is a frozen bootstrap; live reads use the announced fence.
+ const page = await observer.subscription.loadTranscriptPage({
+ source: 'durable',
+ direction: 'older',
+ throughSequence: frame.throughSequence,
+ cursor: null,
+ anchorSequence: null,
+ maxBytes: 48 * 1024,
+ });
+ const decoded = await observer.subscription.decodeTranscriptPage(page, decodeStoredMessage);
+ assert.equal(decoded.nextCursor, null);
+ const row = decoded.messages.find(
+ ({ message }) =>
+ message.type === 'workhub_coordination' &&
+ message.kind === kind &&
+ message.actionId === actionId,
+ );
+ if (row) return row.message;
+ }
+}
+
export async function verifyWorkhubDelegation(connection, workspace, reopened, mode = 'existing') {
const createNew = mode === 'created';
const selectTarget = mode === 'selected';
@@ -95,6 +122,7 @@ export async function verifyWorkhubDelegation(connection, workspace, reopened, m
calls = 0,
input,
receipt,
+ assignmentRow,
attachment,
stopInput,
stopReceipt,
@@ -301,11 +329,35 @@ export async function verifyWorkhubDelegation(connection, workspace, reopened, m
act({ ...input, turnId: 'not-active' }),
(error) => error.code === 'operation_conflict',
);
+ const beforeAssignment = sourceObserver.frames.at(-1)?.sequence ?? 0;
receipt = selectTarget
? await chooseTarget(request, act, sourceObserver, input, workspace)
: await act(input);
assert.equal(receipt.disposition, createNew ? 'create_new' : 'delegate_existing');
assert.equal(receipt.targetSessionId, targetSessionId);
+ const assignment = await coordinationRecord(
+ sourceObserver,
+ beforeAssignment,
+ 'delegation_assigned',
+ input.actionId,
+ );
+ assert.equal(assignment.delegationId, assignment.id);
+ assignmentRow = assignment;
+ assert.equal(assignment.targetSessionId, targetSessionId);
+ assert.equal(assignment.targetTurnId, receipt.targetTurnId);
+ assert.equal(assignment.disposition, receipt.disposition);
+ assert.equal(assignment.userText, 'Please implement the requested task');
+ assert.deepEqual(assignment.attachments, [attachment]);
+ assert.equal(assignment.targetAttachments.length, 1);
+ assert.equal(assignment.targetAttachments[0].ref.sessionId, targetSessionId);
+ assert.equal(assignment.steered, steerTarget ? true : undefined);
+ if (createNew)
+ assert.deepEqual(assignment.create, {
+ title: input.proposal.title,
+ workspace: input.create.workspace,
+ defaults: input.newWorkDefaults,
+ });
+ else assert.equal(assignment.create, undefined);
if (steerTarget) {
assert.equal(receipt.steered, true);
assert.equal(receipt.targetTurnId, 'already-running');
@@ -334,41 +386,16 @@ export async function verifyWorkhubDelegation(connection, workspace, reopened, m
expects: { targetSessionId },
},
};
- let sourceSequence = sourceObserver.frames.at(-1)?.sequence ?? 0;
+ const sourceSequence = sourceObserver.frames.at(-1)?.sequence ?? 0;
stopReceipt = await act(stopInput);
// The coordinator stays inside this tool call: no subsequent
// Invocation boundary or reconnect can hide a missed control wakeup.
- for (;;) {
- const frame = await sourceObserver.waitFor(
- (frame) =>
- frame.kind === 'subscription.transcript_advanced' &&
- frame.sequence > sourceSequence,
- );
- sourceSequence = frame.sequence;
- // loadTranscript is the frozen bootstrap, not a live refresh.
- const page = await sourceObserver.subscription.loadTranscriptPage({
- source: 'durable',
- direction: 'older',
- throughSequence: frame.throughSequence,
- cursor: null,
- anchorSequence: null,
- maxBytes: 48 * 1024,
- });
- const decoded = await sourceObserver.subscription.decodeTranscriptPage(
- page,
- decodeStoredMessage,
- );
- assert.equal(decoded.nextCursor, null);
- if (
- decoded.messages.some(
- ({ message: row }) =>
- row.type === 'workhub_coordination' &&
- row.kind === 'delegation_stop_resolved' &&
- row.actionId === stopInput.actionId,
- )
- )
- break;
- }
+ await coordinationRecord(
+ sourceObserver,
+ sourceSequence,
+ 'delegation_stop_resolved',
+ stopInput.actionId,
+ );
assert.deepEqual(stopReceipt, {
disposition: 'stop_work',
outcome: steerTarget ? 'not_owned' : 'stop_delivered',
@@ -469,13 +496,33 @@ export async function verifyWorkhubDelegation(connection, workspace, reopened, m
);
}
await targetObserver.close();
+ const beforeRename = await querySession(request, targetSessionId);
+ await request('session.metadata.update', {
+ sessionId: targetSessionId,
+ expectedRevision: beforeRename.revision,
+ patch: { name: 'Renamed after assignment' },
+ });
targetObserver = await watchSession(connection, targetSessionId, { kind: 'tail', maxBytes: 2 });
const rows = await targetObserver.subscription.loadTranscript(decodeStoredMessage);
- if (stopInput) {
+ {
await sourceObserver.close();
sourceObserver = await watchSession(connection, sessionId, { kind: 'tail', maxBytes: 2 });
coordinationRows = await sourceObserver.subscription.loadTranscript(decodeStoredMessage);
- const control = coordinationRows.filter((row) => row.type === 'workhub_coordination');
+ const assignments = coordinationRows.filter(
+ (row) => row.type === 'workhub_coordination' && row.kind === 'delegation_assigned',
+ );
+ assert.deepEqual(assignments, [assignmentRow]);
+ const candidates = await request('workhub.coordination.candidates', {});
+ assert.equal(
+ candidates.candidates.find((candidate) => candidate.sessionId === targetSessionId)
+ ?.latestDelegationActionId,
+ stopTarget ? undefined : input.actionId,
+ );
+ }
+ if (stopInput) {
+ const control = coordinationRows.filter(
+ (row) => row.type === 'workhub_coordination' && row.kind !== 'delegation_assigned',
+ );
assert.equal(control.length, 2);
assert.equal(control[0].kind, 'delegation_stop_requested');
assert.equal(control[1].kind, 'delegation_stop_resolved');
diff --git a/tests/fixtures/client-workhub-routing.mjs b/tests/fixtures/client-workhub-routing.mjs
index cb67840fdd..d657eb7856 100644
--- a/tests/fixtures/client-workhub-routing.mjs
+++ b/tests/fixtures/client-workhub-routing.mjs
@@ -52,8 +52,8 @@ export async function prepareRouting({
if (createNew) {
const input = {
...base,
- proposal: { disposition: 'create_new', title: 'Created task' },
- create: { workspace: { kind: 'host_path', path: workspace } },
+ proposal: { disposition: 'create_new', title: ' Created task ' },
+ create: { workspace: { kind: 'host_path', path: workspace + '/.' } },
newWorkDefaults: {
model: {
llmConnectionId: model.connectionId,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org