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 16:59:46 UTC
(maka) branch feat/runtime-host-rust updated: refactor(plugins): persist session management ownership
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 b948c62c0c refactor(plugins): persist session management ownership
b948c62c0c is described below
commit b948c62c0c3266327c77233bc8cd116b763299e9
Author: Wang <m4...@gmail.com>
AuthorDate: Sun Sep 20 00:59:37 2026 +0800
refactor(plugins): persist session management ownership
Replace WorkHub identity checks in ordinary Session controls with durable package/scope reservations. Preserve create identity, reject namespace reassignment, and keep execution grants distinct from management authority after plugin disable and Host restart.
Generated-by: OpenAI Codex
---
.../event-log/migrations/0027_session_managers.sql | 26 +++++
crates/event-log/src/schema.rs | 2 +-
crates/event-log/src/sessions.rs | 10 ++
crates/event-log/src/sessions/manager.rs | 95 ++++++++++++++++
crates/event-log/tests/migrations.rs | 12 +-
crates/event-log/tests/read_state.rs | 2 +-
crates/event-log/tests/session_manager.rs | 111 +++++++++++++++++++
crates/event-log/tests/shell_runs.rs | 2 +-
crates/runtime-host/src/execution.rs | 10 +-
crates/runtime-host/src/execution/admission.rs | 2 +-
crates/runtime-host/src/execution/message.rs | 2 +-
crates/runtime-host/src/execution/plugins.rs | 9 ++
crates/runtime-host/src/execution/plugins/root.rs | 9 ++
crates/runtime-host/src/execution/resume.rs | 5 +-
crates/runtime-host/src/plugins.rs | 2 +
crates/runtime-host/src/plugins/workhub.rs | 7 ++
crates/runtime-host/src/server.rs | 9 ++
crates/runtime-host/src/server/sessions.rs | 12 +-
.../src/server/sessions/configuration.rs | 8 +-
crates/runtime-host/src/server/sessions/create.rs | 12 +-
.../runtime-host/src/server/sessions/mutation.rs | 14 +--
.../runtime-host/src/server/sessions/workspace.rs | 8 +-
crates/runtime-host/src/session.rs | 23 ++++
.../runtime-host/tests/integration/input_plugin.rs | 121 ++++++++++++++++++++-
.../tests/integration/plugin_commands.rs | 2 +-
docs/rust-parity.md | 7 +-
docs/rust-parity.zh-CN.md | 6 +-
27 files changed, 468 insertions(+), 60 deletions(-)
diff --git a/crates/event-log/migrations/0027_session_managers.sql b/crates/event-log/migrations/0027_session_managers.sql
new file mode 100644
index 0000000000..fe1dd4f909
--- /dev/null
+++ b/crates/event-log/migrations/0027_session_managers.sql
@@ -0,0 +1,26 @@
+/*
+ * 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 TABLE session_managers (
+ session_id TEXT PRIMARY KEY NOT NULL,
+ package_id TEXT NOT NULL,
+ scope_id TEXT NOT NULL,
+ fingerprint TEXT NOT NULL
+);
+PRAGMA user_version = 27;
diff --git a/crates/event-log/src/schema.rs b/crates/event-log/src/schema.rs
index daa9c906fc..55078896d1 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..=26)))
+ || (application_id == 0x4d414b52 && matches!(version, 0..=27)))
{
return Err(StoreError::UnsupportedDatabase);
}
diff --git a/crates/event-log/src/sessions.rs b/crates/event-log/src/sessions.rs
index d0c4bd60a5..d10ebb9e89 100644
--- a/crates/event-log/src/sessions.rs
+++ b/crates/event-log/src/sessions.rs
@@ -21,7 +21,9 @@
mod activity;
mod execution;
+mod manager;
mod metadata;
+pub use manager::ManagedSession;
pub(crate) mod read_state;
pub(crate) use activity::{initialize_execution, project_execution, register_functions};
pub(crate) use execution::advance_revision;
@@ -220,6 +222,14 @@ pub(crate) async fn insert(
if configuration.len() > MAX_CONFIGURATION_BYTES {
return Err(invalid("session configuration exceeds 64 KiB"));
}
+ let reserved: Option<String> =
+ sqlx::query_scalar("SELECT fingerprint FROM session_managers WHERE session_id = ?")
+ .bind(id)
+ .fetch_optional(&mut *tx)
+ .await?;
+ if reserved.is_some_and(|reserved| reserved != fingerprint) {
+ return Err(StoreError::SessionConflict);
+ }
let inserted = sqlx::query(
"INSERT INTO session_control SELECT ?, ?, 1, ?, ?, 0, ?
WHERE NOT EXISTS (SELECT 1 FROM session_control WHERE id = ?)",
diff --git a/crates/event-log/src/sessions/manager.rs b/crates/event-log/src/sessions/manager.rs
new file mode 100644
index 0000000000..38f947d226
--- /dev/null
+++ b/crates/event-log/src/sessions/manager.rs
@@ -0,0 +1,95 @@
+/*
+ * 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::validate_id;
+use crate::{EventLog, StoreError};
+use maka_plugins::{composition::Scope, storage::Namespace};
+use sqlx::{Connection, Row};
+
+/// A trusted Host declaration, not a caller-provided authorization credential.
+#[derive(Clone, Debug)]
+pub struct ManagedSession {
+ pub session_id: String,
+ pub manager: Namespace,
+ /// Only this create receipt may materialize the reserved identity.
+ pub fingerprint: String,
+}
+
+impl EventLog {
+ /// Reserve before admitting clients. This does not adopt or rewrite existing
+ /// data: the manager must still prove its exact creation receipt before use.
+ /// A malformed existing Session must not open an ordinary execution bypass.
+ pub async fn reserve_managed_session(&self, claim: &ManagedSession) -> Result<(), StoreError> {
+ self.validate_root()?;
+ validate_id(&claim.session_id)?;
+ if claim.manager.scope() == &Scope::DesktopUi {
+ return Err(StoreError::InvalidTransition(
+ "Desktop UI cannot manage Host Sessions".into(),
+ ));
+ }
+ if claim.fingerprint.is_empty() || claim.fingerprint.len() > 512 {
+ return Err(StoreError::InvalidTransition(
+ "invalid session fingerprint length".into(),
+ ));
+ }
+ let claim = claim.clone();
+ let scope = String::from(claim.manager.scope().clone());
+ self.connection.run(move |connection| Box::pin(async move {
+ let mut tx = connection.begin_with("BEGIN IMMEDIATE").await?;
+ if let Some(row) = sqlx::query("SELECT package_id, scope_id, fingerprint FROM session_managers WHERE session_id = ?")
+ .bind(&claim.session_id).fetch_optional(&mut *tx).await? {
+ if row.get::<String, _>("package_id") != claim.manager.package()
+ || row.get::<String, _>("scope_id") != scope
+ || row.get::<String, _>("fingerprint") != claim.fingerprint {
+ return Err(StoreError::SessionConflict);
+ }
+ return Ok(());
+ }
+ sqlx::query("INSERT INTO session_managers VALUES (?, ?, ?, ?)")
+ .bind(claim.session_id).bind(claim.manager.package()).bind(scope)
+ .bind(claim.fingerprint).execute(&mut *tx).await?;
+ tx.commit().await.map_err(StoreError::CommitUnknown)
+ })).await
+ }
+
+ pub async fn session_manager(&self, session_id: &str) -> Result<Option<Namespace>, StoreError> {
+ self.validate_root()?;
+ validate_id(session_id)?;
+ let session_id = session_id.to_owned();
+ self.connection
+ .run(move |connection| {
+ Box::pin(async move {
+ sqlx::query(
+ "SELECT package_id, scope_id FROM session_managers WHERE session_id = ?",
+ )
+ .bind(session_id)
+ .fetch_optional(connection)
+ .await?
+ .map(|row| {
+ let scope = Scope::try_from(row.get::<String, _>("scope_id"))
+ .map_err(|error| StoreError::InvalidTransition(error.to_string()))?;
+ Namespace::new(row.get::<String, _>("package_id"), scope)
+ .map_err(|error| StoreError::InvalidTransition(error.to_string()))
+ })
+ .transpose()
+ })
+ })
+ .await
+ }
+}
diff --git a/crates/event-log/tests/migrations.rs b/crates/event-log/tests/migrations.rs
index c13286c2de..ef2e56c449 100644
--- a/crates/event-log/tests/migrations.rs
+++ b/crates/event-log/tests/migrations.rs
@@ -114,7 +114,7 @@ async fn embedded_migration_adopts_only_rust_schema_and_reopens_without_rewritin
.collect::<Vec<_>>(),
vec![
1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24,
- 25, 26
+ 25, 26, 27
]
);
assert!(checksums.iter().all(|(_, checksum)| checksum.len() == 48));
@@ -150,7 +150,7 @@ async fn embedded_migration_adopts_only_rust_schema_and_reopens_without_rewritin
// table also exercises idempotent backfill after interrupted adoption.
connection
.execute_batch(
- "DROP TABLE model_request_compositions; DROP TABLE request_compositions; DROP TABLE graph_wakes; DROP TABLE graph_intents; DROP TABLE graph_updates; DROP TABLE graph_epochs; DROP TABLE plugin_execution_receipts; DROP TABLE plugin_data; DROP TABLE plugin_packages; DROP TABLE plugin_package_files;
+ "DROP TABLE session_managers; DROP TABLE model_request_compositions; DROP TABLE request_compositions; DROP TABLE graph_wakes; DROP TABLE graph_intents; DROP TABLE graph_updates; DROP TABLE graph_epochs; DROP TABLE plugin_execution_receipts; DROP TABLE plugin_data; DROP TABLE plugin_packages; DROP TABLE plugin_package_files;
DROP TABLE plugin_package_blobs; DROP TABLE plugin_composition;
DROP VIEW workhub_corrections; DROP VIEW workhub_assignments;
DROP VIEW workhub_stops; ALTER TABLE legacy_workhub_stops RENAME TO workhub_stops;
@@ -205,7 +205,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(),
- 26
+ 27
);
}
@@ -274,7 +274,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(),
- 26
+ 27
);
}
@@ -336,7 +336,7 @@ async fn interrupted_initial_migration_remains_openable_under_the_rust_applicati
connection
.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
.unwrap(),
- 26
+ 27
);
assert_eq!(
connection
@@ -346,6 +346,6 @@ async fn interrupted_initial_migration_remains_openable_under_the_rust_applicati
|row| row.get::<_, i64>(0)
)
.unwrap(),
- 26
+ 27
);
}
diff --git a/crates/event-log/tests/read_state.rs b/crates/event-log/tests/read_state.rs
index 67048dd917..644e829088 100644
--- a/crates/event-log/tests/read_state.rs
+++ b/crates/event-log/tests/read_state.rs
@@ -261,7 +261,7 @@ async fn migration_backfills_only_existing_sessions_and_never_replays_over_ackno
log.close().await.unwrap();
// Legacy ledger adoption must also preserve an already durable acknowledgement.
source
- .execute_batch("DROP TABLE model_request_compositions; DROP TABLE request_compositions; DROP TABLE graph_wakes; DROP TABLE graph_intents; DROP TABLE graph_updates; DROP TABLE graph_epochs; DROP TABLE plugin_execution_receipts; DROP TABLE plugin_data; DROP TABLE plugin_packages; DROP TABLE plugin_package_files; DROP TABLE plugin_package_blobs; DROP TABLE plugin_composition; DROP VIEW workhub_corrections; DROP VIEW workhub_assignments; DROP VIEW workhub_stops; ALTER TABLE legacy_wo [...]
+ .execute_batch("DROP TABLE session_managers; DROP TABLE model_request_compositions; DROP TABLE request_compositions; DROP TABLE graph_wakes; DROP TABLE graph_intents; DROP TABLE graph_updates; DROP TABLE graph_epochs; DROP TABLE plugin_execution_receipts; DROP TABLE plugin_data; DROP TABLE plugin_packages; DROP TABLE plugin_package_files; DROP TABLE plugin_package_blobs; DROP TABLE plugin_composition; DROP VIEW workhub_corrections; DROP VIEW workhub_assignments; DROP VIEW workhub [...]
.unwrap();
let log = EventLog::open(&path).await.unwrap();
assert_eq!(
diff --git a/crates/event-log/tests/session_manager.rs b/crates/event-log/tests/session_manager.rs
new file mode 100644
index 0000000000..b18fc44f4a
--- /dev/null
+++ b/crates/event-log/tests/session_manager.rs
@@ -0,0 +1,111 @@
+/*
+ * 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 maka_event_log::{EventLog, StoreError, sessions::ManagedSession};
+use maka_plugins::{composition::Scope, storage::Namespace};
+use serde_json::{Value, json};
+
+#[tokio::test]
+async fn manager_reservations_survive_restart_without_reassigning_or_adopting_data() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("events.sqlite");
+ let claim = ManagedSession {
+ session_id: "example-inbox".into(),
+ manager: Namespace::new("example.workflow", Scope::Profile).unwrap(),
+ fingerprint: "owned-creation".into(),
+ };
+ let log = EventLog::open(&path).await.unwrap();
+ log.reserve_managed_session(&claim).await.unwrap();
+ log.reserve_managed_session(&claim).await.unwrap();
+ assert!(matches!(
+ log.create_session(&claim.session_id, "ordinary-create", &json!({}), 1)
+ .await,
+ Err(StoreError::SessionConflict)
+ ));
+ let record = log
+ .create_session(
+ &claim.session_id,
+ &claim.fingerprint,
+ &json!({"name":"inbox"}),
+ 2,
+ )
+ .await
+ .unwrap();
+ let mut wrong_owner = claim.clone();
+ wrong_owner.manager = Namespace::new("other.workflow", Scope::Profile).unwrap();
+ assert!(matches!(
+ log.reserve_managed_session(&wrong_owner).await,
+ Err(StoreError::SessionConflict)
+ ));
+ let mut wrong_receipt = claim.clone();
+ wrong_receipt.fingerprint = "replacement".into();
+ assert!(matches!(
+ log.reserve_managed_session(&wrong_receipt).await,
+ Err(StoreError::SessionConflict)
+ ));
+ let mut wrong_scope = claim.clone();
+ wrong_scope.manager =
+ Namespace::new("example.workflow", Scope::Session("other".into())).unwrap();
+ assert!(matches!(
+ log.reserve_managed_session(&wrong_scope).await,
+ Err(StoreError::SessionConflict)
+ ));
+
+ // A reservation closes ordinary access even when legacy data is malformed,
+ // but does not grant the manager a matching create receipt or rewrite data.
+ let legacy = ManagedSession {
+ session_id: "legacy".into(),
+ ..claim.clone()
+ };
+ log.create_session(
+ "legacy",
+ "different-creation",
+ &json!({"name":"unrelated"}),
+ 3,
+ )
+ .await
+ .unwrap();
+ log.reserve_managed_session(&legacy).await.unwrap();
+ assert!(matches!(
+ log.probe_session_create::<Value>("legacy", &legacy.fingerprint)
+ .await,
+ Err(StoreError::SessionConflict)
+ ));
+ assert_eq!(
+ log.get_session::<Value>("legacy")
+ .await
+ .unwrap()
+ .unwrap()
+ .configuration["name"],
+ "unrelated"
+ );
+ log.close().await.unwrap();
+
+ let log = EventLog::open(&path).await.unwrap();
+ assert_eq!(
+ log.session_manager(&claim.session_id).await.unwrap(),
+ Some(claim.manager.clone())
+ );
+ assert_eq!(
+ log.get_session::<Value>(&claim.session_id).await.unwrap(),
+ Some(record)
+ );
+ assert_eq!(log.session_manager("ordinary").await.unwrap(), None);
+ log.close().await.unwrap();
+}
diff --git a/crates/event-log/tests/shell_runs.rs b/crates/event-log/tests/shell_runs.rs
index 3cf5022bf6..1efd123132 100644
--- a/crates/event-log/tests/shell_runs.rs
+++ b/crates/event-log/tests/shell_runs.rs
@@ -278,7 +278,7 @@ async fn migration_and_recovery_commit_are_atomic_under_real_sqlite_faults() {
.unwrap();
// A prior Rust schema, not an old Maka/user database.
sqlx::raw_sql(
- "DROP TABLE model_request_compositions; DROP TABLE request_compositions; DROP TABLE graph_wakes; DROP TABLE graph_intents; DROP TABLE graph_updates; DROP TABLE graph_epochs; DROP TABLE plugin_execution_receipts; DROP TABLE plugin_data; DROP TABLE plugin_packages; DROP TABLE plugin_package_files; DROP TABLE plugin_package_blobs; DROP TABLE plugin_composition; DROP VIEW workhub_corrections; DROP VIEW workhub_assignments; DROP VIEW workhub_stops; ALTER TABLE legacy_workhub_stops REN [...]
+ "DROP TABLE session_managers; DROP TABLE model_request_compositions; DROP TABLE request_compositions; DROP TABLE graph_wakes; DROP TABLE graph_intents; DROP TABLE graph_updates; DROP TABLE graph_epochs; DROP TABLE plugin_execution_receipts; DROP TABLE plugin_data; DROP TABLE plugin_packages; DROP TABLE plugin_package_files; DROP TABLE plugin_package_blobs; DROP TABLE plugin_composition; DROP VIEW workhub_corrections; DROP VIEW workhub_assignments; DROP VIEW workhub_stops; ALTER T [...]
PRAGMA user_version = 5;",
)
.execute(&mut observer)
diff --git a/crates/runtime-host/src/execution.rs b/crates/runtime-host/src/execution.rs
index d1ab30b12f..5e9ac56727 100644
--- a/crates/runtime-host/src/execution.rs
+++ b/crates/runtime-host/src/execution.rs
@@ -298,14 +298,10 @@ fn execution_error(error: RunError) -> OperationError {
};
failure(code, &error.to_string())
}
-fn ordinary_session(session: &str) -> Result<()> {
- if session == maka_runtime::workhub::COORDINATION_SESSION_ID {
- return Err(failure(
- Code::OperationConflict,
- "Use the WorkHub execution entry point",
- ));
+impl Executions {
+ async fn ordinary_session(&self, session: &str) -> Result<()> {
+ crate::session::require_unmanaged(&self.log, session, Code::OperationConflict).await
}
- Ok(())
}
fn internal(error: impl std::fmt::Display) -> OperationError {
diff --git a/crates/runtime-host/src/execution/admission.rs b/crates/runtime-host/src/execution/admission.rs
index e2eaf4ebb9..38a245a9b5 100644
--- a/crates/runtime-host/src/execution/admission.rs
+++ b/crates/runtime-host/src/execution/admission.rs
@@ -29,7 +29,7 @@ impl Executions {
connection_id: Uuid,
root_id: &str,
) -> Result<TurnStartResult> {
- super::ordinary_session(&input.session_id)?;
+ self.ordinary_session(&input.session_id).await?;
let fingerprint = format!(
"sha256:{:x}",
Sha256::digest(serde_json::to_vec(&input).map_err(internal)?)
diff --git a/crates/runtime-host/src/execution/message.rs b/crates/runtime-host/src/execution/message.rs
index bb21a74e85..79bf03e36c 100644
--- a/crates/runtime-host/src/execution/message.rs
+++ b/crates/runtime-host/src/execution/message.rs
@@ -39,7 +39,7 @@ impl Executions {
root_id: &str,
epoch: &str,
) -> Result<SubmitResult> {
- super::ordinary_session(&input.session_id)?;
+ self.ordinary_session(&input.session_id).await?;
let mut prepared = None;
let mut queued: Option<(
maka_runtime::event::Invocation,
diff --git a/crates/runtime-host/src/execution/plugins.rs b/crates/runtime-host/src/execution/plugins.rs
index d469a55f49..7286e46444 100644
--- a/crates/runtime-host/src/execution/plugins.rs
+++ b/crates/runtime-host/src/execution/plugins.rs
@@ -234,6 +234,15 @@ impl BoundCommands {
async fn authorize(&self, host: &Executions, session: &str) -> Result<(), Error> {
self.authorize_origin(host).await?;
+ if host
+ .log
+ .session_manager(session)
+ .await
+ .map_err(storage)?
+ .is_some_and(|manager| manager != self.namespace)
+ {
+ return Err(Error::Denied);
+ }
let grant = self
.grants
.lock()
diff --git a/crates/runtime-host/src/execution/plugins/root.rs b/crates/runtime-host/src/execution/plugins/root.rs
index 5eae287e09..eb245b2ab8 100644
--- a/crates/runtime-host/src/execution/plugins/root.rs
+++ b/crates/runtime-host/src/execution/plugins/root.rs
@@ -36,6 +36,15 @@ impl BoundCommands {
else {
return Ok(());
};
+ if host
+ .log
+ .session_manager(&source.session_id)
+ .await
+ .map_err(storage)?
+ .is_some_and(|manager| manager != self.namespace)
+ {
+ return Err(Error::Denied);
+ }
let current = host
.log
.get_session::<SessionConfiguration>(&source.session_id)
diff --git a/crates/runtime-host/src/execution/resume.rs b/crates/runtime-host/src/execution/resume.rs
index ed8309cd0a..0a05b85587 100644
--- a/crates/runtime-host/src/execution/resume.rs
+++ b/crates/runtime-host/src/execution/resume.rs
@@ -45,7 +45,8 @@ impl Executions {
input: TurnResumeQueryInput,
connection: Uuid,
) -> Result<TurnResumePlan> {
- super::ordinary_session(&input.session_id)
+ self.ordinary_session(&input.session_id)
+ .await
.map_err(|error| failure(Code::OperationUnavailable, &error.message))?;
let session = self.resume_session(&input.session_id).await?;
if self
@@ -98,7 +99,7 @@ impl Executions {
input: TurnResumeStartInput,
connection: Uuid,
) -> Result<TurnResumeStartResult> {
- super::ordinary_session(&input.session_id)?;
+ self.ordinary_session(&input.session_id).await?;
let mut prepared = None;
loop {
let admission = self.lock_admission().await;
diff --git a/crates/runtime-host/src/plugins.rs b/crates/runtime-host/src/plugins.rs
index a0087763ea..137d5990f6 100644
--- a/crates/runtime-host/src/plugins.rs
+++ b/crates/runtime-host/src/plugins.rs
@@ -56,6 +56,8 @@ pub struct Setup {
pub builtins: Definitions,
pub layers: BTreeMap<String, Vec<Operation>>,
pub loader: Option<Arc<dyn PackageLoader>>,
+ /// Persistent Session ownership declared by linked plugins before admission.
+ pub managed_sessions: Vec<maka_event_log::sessions::ManagedSession>,
}
pub(crate) struct ExternalLoader(pub Option<Arc<dyn PackageLoader>>);
diff --git a/crates/runtime-host/src/plugins/workhub.rs b/crates/runtime-host/src/plugins/workhub.rs
index 69b32b8b0b..30e8694eba 100644
--- a/crates/runtime-host/src/plugins/workhub.rs
+++ b/crates/runtime-host/src/plugins/workhub.rs
@@ -79,6 +79,13 @@ pub(crate) fn install(
}
catalog.host_only::<Control>()?;
catalog.reserve_for::<Control>(ID, ID)?;
+ setup
+ .managed_sessions
+ .push(maka_event_log::sessions::ManagedSession {
+ session_id: maka_runtime::workhub::COORDINATION_SESSION_ID.into(),
+ manager: maka_plugins::storage::Namespace::new(ID, Scope::Profile)?,
+ fingerprint: coordinator::fingerprint(),
+ });
setup.builtins.insert(
ID.into(),
Arc::new(Definition {
diff --git a/crates/runtime-host/src/server.rs b/crates/runtime-host/src/server.rs
index 4ad9cb5074..a626f1db8a 100644
--- a/crates/runtime-host/src/server.rs
+++ b/crates/runtime-host/src/server.rs
@@ -290,6 +290,15 @@ impl Host {
root.root_id().into(),
)?));
}
+ for claim in &setup.managed_sessions {
+ if !setup.builtins.contains_key(claim.manager.package()) {
+ return Err(maka_plugins::Error::Invalid(
+ "Session manager is not a linked plugin".into(),
+ )
+ .into());
+ }
+ log.reserve_managed_session(claim).await?;
+ }
let data_root = root.canonical_path().to_owned();
let data = tokio::task::spawn_blocking(move || {
maka_plugins::storage::Directories::open(&data_root)
diff --git a/crates/runtime-host/src/server/sessions.rs b/crates/runtime-host/src/server/sessions.rs
index f0fab09e0c..765bf8aeae 100644
--- a/crates/runtime-host/src/server/sessions.rs
+++ b/crates/runtime-host/src/server/sessions.rs
@@ -128,12 +128,12 @@ pub(super) async fn execute(
.map(Output::Query),
Operation::SessionLifecycleSet => {
let input = decode_session_lifecycle_set_input(value).map_err(invalid)?;
- if input.session_id == maka_runtime::workhub::COORDINATION_SESSION_ID {
- return Err(failure(
- OperationErrorCode::OperationConflict,
- "WorkHub lifecycle requires WorkHub authority",
- ));
- }
+ crate::session::require_unmanaged(
+ log,
+ &input.session_id,
+ OperationErrorCode::OperationConflict,
+ )
+ .await?;
let _admission = host.executions.lock_admission().await;
if input.state == SessionLifecycleState::Archived
&& host
diff --git a/crates/runtime-host/src/server/sessions/configuration.rs b/crates/runtime-host/src/server/sessions/configuration.rs
index dca3e30620..bfd2fca651 100644
--- a/crates/runtime-host/src/server/sessions/configuration.rs
+++ b/crates/runtime-host/src/server/sessions/configuration.rs
@@ -26,12 +26,8 @@ use serde_json::Value;
pub(super) async fn update(host: &Host, value: &Value) -> Result<SessionUpdateResult> {
let input = decode_session_configuration_update_input(value).map_err(invalid)?;
- if input.session_id == "maka_workhub_coordination" {
- return Err(failure(
- Code::OperationConflict,
- "WorkHub configuration requires WorkHub authority",
- ));
- }
+ crate::session::require_unmanaged(&host.log, &input.session_id, Code::OperationConflict)
+ .await?;
apply(host, input).await
}
diff --git a/crates/runtime-host/src/server/sessions/create.rs b/crates/runtime-host/src/server/sessions/create.rs
index 32723762e4..c6ed4e3a30 100644
--- a/crates/runtime-host/src/server/sessions/create.rs
+++ b/crates/runtime-host/src/server/sessions/create.rs
@@ -29,12 +29,12 @@ pub(super) async fn create(
input: SessionCreateInput,
) -> Result<SessionCatalogItem> {
let log = &host.log;
- if input.session_id == "maka_workhub_coordination" {
- return Err(failure(
- OperationErrorCode::OperationConflict,
- "Session identity is reserved for WorkHub coordination",
- ));
- }
+ crate::session::require_unmanaged(
+ log,
+ &input.session_id,
+ OperationErrorCode::OperationConflict,
+ )
+ .await?;
let thinking = input.thinking_level;
let prepared = PreparedSession::new(input).map_err(invalid)?;
let fingerprint = prepared.fingerprint();
diff --git a/crates/runtime-host/src/server/sessions/mutation.rs b/crates/runtime-host/src/server/sessions/mutation.rs
index 18a93d7d25..a3131093b5 100644
--- a/crates/runtime-host/src/server/sessions/mutation.rs
+++ b/crates/runtime-host/src/server/sessions/mutation.rs
@@ -26,12 +26,7 @@ use serde_json::Value;
pub(super) async fn metadata(log: &EventLog, value: &Value) -> Result<SessionUpdateResult> {
let input = decode_session_metadata_update_input(value).map_err(invalid)?;
- if input.session_id == "maka_workhub_coordination" {
- return Err(failure(
- Code::OperationUnavailable,
- "WorkHub metadata requires WorkHub authority",
- ));
- }
+ crate::session::require_unmanaged(log, &input.session_id, Code::OperationUnavailable).await?;
let patch = input.patch.clone();
let mutation = log
.update_session_metadata(
@@ -56,12 +51,7 @@ pub(in crate::server) use crate::session::mutation_projection as result;
pub(super) async fn read_marker(log: &EventLog, value: &Value) -> Result<SessionCatalogItem> {
let input = decode_session_read_marker_set_input(value).map_err(invalid)?;
- if input.session_id == "maka_workhub_coordination" {
- return Err(failure(
- Code::OperationConflict,
- "WorkHub read state requires WorkHub authority",
- ));
- }
+ crate::session::require_unmanaged(log, &input.session_id, Code::OperationConflict).await?;
let record = log
.set_session_read_marker(&input.session_id, &input.read_through_message_id)
.await
diff --git a/crates/runtime-host/src/server/sessions/workspace.rs b/crates/runtime-host/src/server/sessions/workspace.rs
index 06729fb48d..86ed6d4429 100644
--- a/crates/runtime-host/src/server/sessions/workspace.rs
+++ b/crates/runtime-host/src/server/sessions/workspace.rs
@@ -26,12 +26,8 @@ use std::path::{Component, Path, PathBuf};
pub(super) async fn relocate(host: &Host, value: &Value) -> Result<SessionUpdateResult> {
let input = decode_session_workspace_relocate_input(value).map_err(invalid)?;
- if input.session_id == "maka_workhub_coordination" {
- return Err(failure(
- Code::OperationConflict,
- "WorkHub workspace requires WorkHub authority",
- ));
- }
+ crate::session::require_unmanaged(&host.log, &input.session_id, Code::OperationConflict)
+ .await?;
let _admission = host.executions.lock_admission().await;
if host.draining.is_cancelled() {
return Err(failure(Code::HostDraining, "Host is draining"));
diff --git a/crates/runtime-host/src/session.rs b/crates/runtime-host/src/session.rs
index 38338f3cb4..b627072388 100644
--- a/crates/runtime-host/src/session.rs
+++ b/crates/runtime-host/src/session.rs
@@ -28,6 +28,29 @@ pub(crate) use projection::mutation_projection;
pub use metadata::apply_metadata_patch;
+/// Persistent manager ownership survives unload, disabled entries and restart.
+pub(crate) async fn require_unmanaged(
+ log: &maka_event_log::EventLog,
+ session_id: &str,
+ code: maka_protocol::OperationErrorCode,
+) -> std::result::Result<(), maka_protocol::OperationError> {
+ if log
+ .session_manager(session_id)
+ .await
+ .map_err(|error| maka_protocol::OperationError {
+ code: maka_protocol::OperationErrorCode::PersistenceFailed,
+ message: error.to_string(),
+ })?
+ .is_some()
+ {
+ return Err(maka_protocol::OperationError {
+ code,
+ message: "Session operation requires its manager".into(),
+ });
+ }
+ Ok(())
+}
+
use maka_event_log::sessions::SessionRecord;
use maka_protocol::session::*;
use maka_protocol::{ProtocolError, Result};
diff --git a/crates/runtime-host/tests/integration/input_plugin.rs b/crates/runtime-host/tests/integration/input_plugin.rs
index db49268f45..8ddfd5e1de 100644
--- a/crates/runtime-host/tests/integration/input_plugin.rs
+++ b/crates/runtime-host/tests/integration/input_plugin.rs
@@ -113,7 +113,126 @@ fn setup(business: &Business) -> Setup {
]))
.unwrap(),
)]),
- loader: None,
+ ..Default::default()
+ }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn custom_session_manager_blocks_ordinary_access_after_disable_and_restart() {
+ let fixture = ClientFixture::new("maka-managed-plugin-");
+ for reopened in [false, true] {
+ let mut plugins = setup(&Business::default());
+ if !reopened {
+ plugins
+ .managed_sessions
+ .push(maka_event_log::sessions::ManagedSession {
+ session_id: "example-inbox".into(),
+ manager: maka_plugins::storage::Namespace::new(
+ "example",
+ maka_plugins::composition::Scope::Profile,
+ )
+ .unwrap(),
+ fingerprint: "example-owned-creation".into(),
+ });
+ }
+ let host = Host::open_with_options(
+ fixture.owner(),
+ None,
+ HostOptions {
+ plugins,
+ ..Default::default()
+ },
+ )
+ .await
+ .unwrap();
+ #[cfg(unix)]
+ let endpoint = fixture.workspace.parent().unwrap().join("managed.sock");
+ #[cfg(windows)]
+ let endpoint =
+ std::path::PathBuf::from(format!(r"\\.\pipe\maka-managed-{}", uuid::Uuid::new_v4()));
+ let stop = CancellationToken::new();
+ let cleanup = stop.clone().drop_guard();
+ let server = tokio::spawn(
+ LocalListener::bind(&endpoint)
+ .unwrap()
+ .serve(host.clone(), stop),
+ );
+ let mut peer = Peer::new(host.clone(), "managed-plugin").await;
+ super::skills_plugin::converged(&mut peer).await;
+ if !reopened {
+ let disabled = peer.rpc("plugin.composition.apply", json!({
+ "operations":[{"type":"update","entryId":"example","patch":{"disabled":true}}]
+ })).await;
+ assert_eq!(disabled["ok"], true, "{disabled}");
+ super::skills_plugin::converged(&mut peer).await;
+ }
+ for (operation, input) in [
+ (
+ "session.create",
+ json!({
+ "sessionId":"example-inbox", "workspace":{"kind":"host_path","path":fixture.workspace},
+ "modelTarget":{"kind":"default"}
+ }),
+ ),
+ (
+ "turn.start",
+ json!({"sessionId":"example-inbox","turnId":"bypass","content":{"text":"ordinary"}}),
+ ),
+ (
+ "session.lifecycle.set",
+ json!({"sessionId":"example-inbox","state":"archived"}),
+ ),
+ ] {
+ let result = peer.rpc(operation, input).await;
+ assert_eq!(
+ result["error"]["code"], "operation_conflict",
+ "{operation}: {result}"
+ );
+ assert_eq!(
+ result["error"]["message"],
+ "Session operation requires its manager"
+ );
+ }
+ // Even an explicitly restored execution grant is not management authority.
+ let fiber = maka_plugins::fiber::Fiber::new(
+ "other",
+ "other",
+ maka_plugins::composition::Scope::Profile,
+ )
+ .unwrap();
+ fiber.begin_loading().unwrap();
+ fiber.ready().unwrap();
+ fiber.publish().unwrap();
+ let commands = host
+ .restore_plugin_execution(
+ fiber.context(),
+ vec![maka_plugins::execution::SessionBoundary {
+ session_id: "example-inbox".into(),
+ boundary_revision: 0,
+ permission_mode: maka_runtime::execution::PermissionMode::Bypass,
+ cwd: fixture.workspace.to_string_lossy().into_owned(),
+ }],
+ )
+ .unwrap();
+ assert!(matches!(
+ commands
+ .submit(maka_plugins::execution::Submit {
+ operation_id: "bypass".into(),
+ session_id: "example-inbox".into(),
+ content: "ordinary".into(),
+ orchestration_mode: None,
+ })
+ .await,
+ Err(maka_plugins::execution::CommandError::Denied)
+ ));
+ fiber
+ .shutdown(tokio::time::Instant::now() + Duration::from_secs(1))
+ .await
+ .unwrap();
+ peer.close().await;
+ drop(cleanup);
+ server.await.unwrap().unwrap();
+ drop(host);
}
}
diff --git a/crates/runtime-host/tests/integration/plugin_commands.rs b/crates/runtime-host/tests/integration/plugin_commands.rs
index 2a845213f2..b34259982e 100644
--- a/crates/runtime-host/tests/integration/plugin_commands.rs
+++ b/crates/runtime-host/tests/integration/plugin_commands.rs
@@ -464,7 +464,7 @@ fn native_plugin() -> maka_runtime_host::plugins::Setup {
.unwrap(),
],
)]),
- loader: None,
+ ..Default::default()
}
}
diff --git a/docs/rust-parity.md b/docs/rust-parity.md
index 2a88ef8a0f..1846d268bb 100644
--- a/docs/rust-parity.md
+++ b/docs/rust-parity.md
@@ -57,13 +57,18 @@ Client changes are allowed when they remove an obsolete path, as with Graph.
| Domain / current coupling | Target and completion condition |
| --- | --- |
| **Skills:** `maka.skills` owns discovery, input preparation, per-step tool/context snapshots, governance, preference CAS, preview, import and workspace/user publication. Published Client Contributions own Session/new-workspace pickers, management and draft suggestions. | Desktop supplies target-bound Slots, generic Remote transport and authorized native file actions. The old scanner, importer, controller and Skills IPC/preload facade are removed. Host retains thin external protocol ada [...]
-| **WorkHub:** `maka.workhub` owns coordinator configuration, answer composition, native `workhub_tasks`, routing/selection/correction/Stop/Resume workflows and recovery policy. Its Client resolves main and floating-window Session bindings through Remote; backend retirement withdraws the Client and rejects stale calls. Desktop supplies window control and workspace/preferences, not task orchestration. | Migrate the full conversation UI and feedback; replace private Session-management rule [...]
+| **WorkHub:** `maka.workhub` owns coordinator configuration, answer composition, native `workhub_tasks`, routing/selection/correction/Stop/Resume workflows and recovery policy. Its Client resolves main and floating-window Session bindings through Remote; backend retirement withdraws the Client and rejects stale calls. Desktop supplies window control and workspace/preferences, not task orchestration. | Migrate the full conversation UI and feedback consumer; replace the private execution [...]
| **Default assistant behavior:** `maka.assistant` publishes the default behavior, persona, personalization and workspace instructions. | Sources are frozen per logical model step; disabling the plugin removes its persona. Explicit Session/child instructions remain independent of replaceable prompt Contributions. Execution/compaction invariants remain Host-owned. |
| **Graph/Swarm and Scheduler:** already built-in plugins; behavior selection uses an open typed identity. Graph uses atomic activate/stop/retire-idle commands and read-only preferences, without `Executions`, configuration writers or Host locks. | Preserve existing orchestration and wakeup behavior. Reuse these narrow commands where semantics match; review Scheduler against the same ownership rule. Typed domain repositories remain legitimate. |
| **Code Mode:** mode selection, nested dispatch and history projection cross several crates. | After the first domain migrations, move its user-facing tool and mode policy where a real contribution boundary helps. Keep V8 ownership, nested-call permissions, dispatch/settlement and canonical history in the runtime. Do not invent a universal executor hook merely to move `exec`. |
| **Files and Shell tools:** registrations are assembled in Host over existing filesystem/process owners. | Tool definitions and assembly can become built-in contributions. Resource ownership, write coordination and PTY cancellation remain Host services. Defer this structural migration until it removes concrete coupling; a plugin per tool adds no value. |
| **Client Capability / MCP, providers and transport** | Keep their current resource/authority boundaries. Desktop-owned MCP does not move into Host, and model vendors do not each require a plugin. Complete their functional gaps independently of structural migrations. |
+Linked plugins declare managed Sessions through `Setup::managed_sessions`. Host persists the
+package/scope owner and creation identity before serving clients. Ordinary mutation/execution
+and other plugins' execution grants cannot bypass that owner, including after disable or restart.
+This reservation does not validate or adopt existing Session data.
+
“Whole domain” means one business implementation and lifecycle, not moving every related type or
table. Canonical Skill receipts and WorkHub execution facts may remain typed runtime contracts.
Reuse SQLx migrations and domain stores; changing ownership does not require changing disk format,
diff --git a/docs/rust-parity.zh-CN.md b/docs/rust-parity.zh-CN.md
index 1906334c64..4893d88c7c 100644
--- a/docs/rust-parity.zh-CN.md
+++ b/docs/rust-parity.zh-CN.md
@@ -50,13 +50,17 @@
| 领域／当前耦合 | 目标与完成条件 |
| --- | --- |
| **Skills：**`maka.skills` 拥有发现、输入准备、每步工具／上下文快照、治理、偏好 CAS、预览、导入及 workspace／user 发布；已发布的 Client Contribution 拥有 Session／新工作区选择器、管理页与草稿建议。 | Desktop 提供目标绑定的 Slot、通用 Remote 传输及授权的原生文件操作；旧扫描、导入、控制器和 Skills IPC／preload 门面已删除。Host 仅保留薄外部协议适配、准入与不可变回执，不再解析 Skill。 |
-| **WorkHub：**`maka.workhub` 拥有协调会话配置、回答组合、原生 `workhub_tasks`、路由／选择／纠正／Stop／Resume 流程和恢复策略。Client 通过 Remote 解析主窗口与浮动窗口 Session 绑定；后端退休撤下 Client 并拒绝旧代调用。Desktop 提供窗口控制与工作区／偏好，不再编排任务。 | 迁移完整会话 UI 和反馈，替换私有 Session 管理规则并补齐协调会话 steering／followup。保留精确消息定位、原子准入／取消／回执，以及 Host 对已接受操作的结算。 |
+| **WorkHub：**`maka.workhub` 拥有协调会话配置、回答组合、原生 `workhub_tasks`、路由／选择／纠正／Stop／Resume 流程和恢复策略。Client 通过 Remote 解析主窗口与浮动窗口 Session 绑定；后端退休撤下 Client 并拒绝旧代调用。Desktop 提供窗口控制与工作区／偏好，不再编排任务。 | 迁移完整会话 UI 和反馈消费者，替换私有执行 profile 并补齐协调会话 steering／followup。保留精确消息定位、原子准入／取消／回执，以及 Host 对已接受操作的结算。 |
| **默认助手行为：**`maka.assistant` 发布默认 behavior、persona、个性化和工作区指令。 | 每逻辑模型步骤冻结来源；停用后不保留隐藏 persona。显式 Session／子任务指令独立于可替换的提示词 Contribution；执行／压缩不变量仍由 Host 维护。 |
| **Graph／Swarm 与 Scheduler：**已经是内置插件，behavior 按开放的类型化身份选择。Graph 使用原子激活／停止／空闲退休命令及只读偏好，不接收 `Executions`、配置写入器或 Host 锁。 | 保持已有编排与唤醒行为；语义相同时复用这些窄命令，按同一归属规则复核 Scheduler。类型化领域 repository 可以保留。 |
| **Code Mode：**模式选择、嵌套派发及历史投影跨越多个 crate。 | 首批领域迁移后，在有实际 Contribution 边界收益时迁移面向用户的工具和模式策略；V8 所有权、嵌套调用权限、派发／结算及规范历史保留 runtime。不为搬迁 `exec` 发明万能执行 hook。 |
| **文件与 Shell 工具：**Host 基于现有文件／进程 owner 装配注册。 | 工具定义与装配可以成为内置 Contribution；资源所有权、写入协调和 PTY 取消仍归 Host。结构迁移等能消除具体耦合时再做，不为每个工具建插件。 |
| **Client Capability／MCP、provider 与传输** | 保持当前资源与权威边界；Desktop MCP 不搬进 Host，模型厂商不强制逐个插件化。功能缺口独立于结构迁移完成。 |
+链接的原生插件通过 `Setup::managed_sessions` 声明托管 Session。Host 在接受客户端前持久化
+package/scope owner 与创建身份；停用或重启后，普通修改／执行和其他插件的执行授权仍不能绕过
+该 owner。这项保留不验证或接管既有 Session 数据。
+
“完整领域”指单一业务实现和生命周期，不是搬走所有相关类型和表。
规范 Skill 回执和 WorkHub 执行事实可以继续是 runtime 的类型化契约。
沿用 SQLx migrations 和领域存储；归属变化不要求改变磁盘格式、全部搬入 KV 或一个插件一个 crate。
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org