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/23 14:50:27 UTC
(maka) 01/02: feat(import): read OpenCode snapshots through public SQLite capabilities
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 6e5a26a703538001e497f0e072a8450128dac4f1
Author: Wang <m4...@gmail.com>
AuthorDate: Wed Sep 23 22:30:43 2026 +0800
feat(import): read OpenCode snapshots through public SQLite capabilities
Bound one-shot WAL-aware reads by current Host-path authority, work and output limits. Preserve OpenCode reverts and source part ordering without manufacturing runtime execution.
Generated-by: OpenAI Codex
---
Cargo.lock | 2 +
crates/js-runtime/src/plugin/bridge.rs | 55 ++-
crates/js-runtime/src/plugin/sdk.js | 2 +
crates/plugins/src/filesystem.rs | 1 +
crates/plugins/src/filesystem/database.rs | 134 ++++++
crates/plugins/src/remote.rs | 9 +
crates/runtime-host/Cargo.toml | 2 +
.../runtime-host/src/plugins/javascript/bridge.rs | 5 +
.../src/plugins/javascript/bridge/wire.rs | 26 ++
.../runtime-host/src/server/plugin_remote/views.rs | 13 +
.../src/server/plugin_remote/views/database.rs | 159 +++++++
.../server/plugin_remote/views/database/query.rs | 455 +++++++++++++++++++++
.../runtime-host/tests/fixtures/remote-plugin.mjs | 39 ++
.../tests/integration/plugin_remote.rs | 41 ++
.../tests/integration/plugin_remote/fixture.rs | 33 ++
.../tests/integration/plugin_remote/javascript.rs | 80 +++-
crates/session-import/README.md | 5 +-
crates/session-import/README.zh-CN.md | 5 +-
crates/session-import/src/lib.rs | 1 +
crates/session-import/src/opencode.rs | 375 +++++++++++++++++
crates/session-import/src/opencode/database.rs | 131 ++++++
crates/session-import/src/opencode/wire.rs | 86 ++++
crates/session-import/src/transcript.rs | 2 +
crates/session-import/tests/opencode.rs | 216 ++++++++++
packages/plugin-sdk/README.md | 2 +
packages/plugin-sdk/README.zh-CN.md | 2 +
packages/plugin-sdk/src/database.ts | 44 ++
packages/plugin-sdk/src/host.ts | 4 +
28 files changed, 1898 insertions(+), 31 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index dcd56b58f9..8f842168d7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5079,12 +5079,14 @@ dependencies = [
"maka-runtime",
"maka-sandbox",
"maka-scheduler",
+ "maka-session-import",
"maka-skills",
"maka-tools",
"maka-transport",
"maka-web",
"maka-workhub",
"reqwest",
+ "rusqlite",
"schemars",
"serde",
"serde_json",
diff --git a/crates/js-runtime/src/plugin/bridge.rs b/crates/js-runtime/src/plugin/bridge.rs
index e4b81a4195..c3f80ee13e 100644
--- a/crates/js-runtime/src/plugin/bridge.rs
+++ b/crates/js-runtime/src/plugin/bridge.rs
@@ -96,19 +96,14 @@ async fn op_maka_plugin(
if method.is_empty() || method.len() > 128 {
return Err(JsErrorBox::generic("invalid Host method"));
}
- let size = serde_json::to_vec(&input)
- .map_err(|error| JsErrorBox::generic(error.to_string()))?
- .len();
- if size
- > binding
+ let size = encoded_size(
+ &input,
+ binding
.bridge
.max_input_bytes(&method)
- .min(32 * 1024 * 1024)
- {
- return Err(JsErrorBox::generic(
- "Host call exceeds its input byte limit",
- ));
- }
+ .min(32 * 1024 * 1024),
+ )
+ .map_err(|_| JsErrorBox::generic("Host call exceeds its input byte limit"))?;
let call = bindings
.calls
.clone()
@@ -121,26 +116,44 @@ async fn op_maka_plugin(
.map_err(|_| JsErrorBox::generic("Host input capacity exhausted"))?;
(binding.bridge.clone(), binding.closing.clone(), call, bytes)
};
- let output_limit = bridge.max_output_bytes(&method).min(32 * 1024 * 1024);
+ let output_limit = bridge
+ .max_output_bytes(&method)
+ .min(64 * 1024 * 1024 + 1024);
let output = tokio::select! {
biased;
_ = closing.cancelled() => Err(failed("plugin instance is retired")),
output = bridge.call(method, input) => output,
}
.map_err(|error| JsErrorBox::generic(error.to_string()))?;
- if serde_json::to_vec(&output)
- .map_err(|error| JsErrorBox::generic(error.to_string()))?
- .len()
- > output_limit
- {
- return Err(JsErrorBox::generic(
- "Host result exceeds its output byte limit",
- ));
- }
+ encoded_size(&output, output_limit)
+ .map_err(|_| JsErrorBox::generic("Host result exceeds its output byte limit"))?;
// Even an immediately ready Host operation must return control to the VM
// owner. Otherwise an async JS loop can starve queued revocations/cleanup.
tokio::task::yield_now().await;
Ok(output)
}
+// Count encoded bytes without allocating a second copy of a large Host result.
+fn encoded_size(value: &Value, limit: usize) -> serde_json::Result<usize> {
+ struct Counter {
+ size: usize,
+ limit: usize,
+ }
+ impl std::io::Write for Counter {
+ fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+ if bytes.len() > self.limit.saturating_sub(self.size) {
+ return Err(std::io::Error::other("Host payload byte limit"));
+ }
+ self.size += bytes.len();
+ Ok(bytes.len())
+ }
+ fn flush(&mut self) -> std::io::Result<()> {
+ Ok(())
+ }
+ }
+ let mut counter = Counter { size: 0, limit };
+ serde_json::to_writer(&mut counter, value)?;
+ Ok(counter.size)
+}
+
deno_core::extension!(maka_plugins, ops = [op_maka_plugin]);
diff --git a/crates/js-runtime/src/plugin/sdk.js b/crates/js-runtime/src/plugin/sdk.js
index 09a5ab6fd4..b19b4c8478 100644
--- a/crates/js-runtime/src/plugin/sdk.js
+++ b/crates/js-runtime/src/plugin/sdk.js
@@ -672,6 +672,8 @@
});
return { ...view, files: readDirectory('view', view.files) };
},
+ queryDatabase: (input) =>
+ host('remote.queryDatabase', { authority: call.remoteAuthority, input }),
});
}
if (call?.authority) {
diff --git a/crates/plugins/src/filesystem.rs b/crates/plugins/src/filesystem.rs
index 826b1d1f0f..81a143caa5 100644
--- a/crates/plugins/src/filesystem.rs
+++ b/crates/plugins/src/filesystem.rs
@@ -18,6 +18,7 @@
*/
//! Source-scoped filesystem operations. Host supplies authority, not callers.
+pub mod database;
mod directory;
pub mod entries;
mod read;
diff --git a/crates/plugins/src/filesystem/database.rs b/crates/plugins/src/filesystem/database.rs
new file mode 100644
index 0000000000..72de9dac5c
--- /dev/null
+++ b/crates/plugins/src/filesystem/database.rs
@@ -0,0 +1,134 @@
+/*
+ * 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.
+ */
+
+//! One-shot reads of an explicitly authorized, trusted Host pathname. This is
+//! not a captured ReadDirectory and does not resist malicious same-user path races.
+use serde::{Deserialize, Serialize};
+pub const MAX_BYTES: usize = 64 * 1024 * 1024;
+pub const MAX_ROWS: usize = 250_000;
+
+#[derive(Deserialize, Serialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct Read {
+ pub path: String,
+ pub queries: Vec<Query>,
+}
+#[derive(Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct Query {
+ pub sql: String,
+ #[serde(default)]
+ pub parameters: Vec<Cell>,
+}
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+#[serde(
+ tag = "kind",
+ content = "value",
+ rename_all = "snake_case",
+ deny_unknown_fields
+)]
+pub enum Cell {
+ Null,
+ Integer(#[serde(with = "integer")] i64),
+ Real(f64),
+ Text(String),
+ Blob(#[serde(with = "blob")] Vec<u8>),
+}
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Table {
+ pub columns: Vec<String>,
+ pub rows: Vec<Vec<Cell>>,
+}
+#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum Limit {
+ Rows,
+ Bytes,
+ Work,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+ #[error("database read authority is unavailable")]
+ Denied,
+ #[error("database read cancelled")]
+ Cancelled,
+ #[error("database read capacity is busy")]
+ Busy,
+ #[error("database was not found")]
+ NotFound,
+ #[error("invalid database read: {0}")]
+ Invalid(String),
+ #[error("database read limit exceeded: {0:?}")]
+ Limit(Limit),
+ #[error("database read failed: {0}")]
+ Unavailable(String),
+}
+impl Read {
+ pub fn validate(&self) -> Result<(), Error> {
+ if self.path.is_empty()
+ || self.path.len() > 32 * 1024
+ || self.path.contains('\0')
+ || !std::path::Path::new(&self.path).is_absolute()
+ || self.queries.is_empty()
+ || self.queries.len() > 16
+ || self.queries.iter().any(|query| {
+ query.sql.trim().is_empty()
+ || query.sql.contains('\0')
+ || query.parameters.len() > 128
+ || query
+ .parameters
+ .iter()
+ .any(|value| matches!(value, Cell::Real(number) if !number.is_finite()))
+ })
+ || serde_json::to_vec(self)
+ .map_err(|error| Error::Invalid(error.to_string()))?
+ .len()
+ > 128 * 1024
+ {
+ return Err(Error::Invalid(
+ "invalid path, queries or input budget".into(),
+ ));
+ }
+ Ok(())
+ }
+}
+mod integer {
+ use serde::{Deserialize, Deserializer, Serializer};
+ pub fn serialize<S: Serializer>(value: &i64, serializer: S) -> Result<S::Ok, S::Error> {
+ serializer.serialize_str(&value.to_string())
+ }
+ pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<i64, D::Error> {
+ String::deserialize(deserializer)?
+ .parse()
+ .map_err(serde::de::Error::custom)
+ }
+}
+mod blob {
+ use base64::{Engine, engine::general_purpose::STANDARD};
+ use serde::{Deserialize, Deserializer, Serializer};
+ pub fn serialize<S: Serializer>(value: &[u8], serializer: S) -> Result<S::Ok, S::Error> {
+ serializer.serialize_str(&STANDARD.encode(value))
+ }
+ pub fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Vec<u8>, D::Error> {
+ STANDARD
+ .decode(String::deserialize(deserializer)?)
+ .map_err(serde::de::Error::custom)
+ }
+}
diff --git a/crates/plugins/src/remote.rs b/crates/plugins/src/remote.rs
index afb16bd038..dec4387ae0 100644
--- a/crates/plugins/src/remote.rs
+++ b/crates/plugins/src/remote.rs
@@ -86,6 +86,15 @@ pub trait Views: Send + Sync {
) -> BoxFuture<'_, Result<crate::call::Owned, Error>>;
fn session(&self) -> BoxFuture<'_, Result<SessionView, Error>>;
fn workspace(&self, input: WorkspaceViewInput) -> BoxFuture<'_, Result<SessionView, Error>>;
+ /// Consistent SELECT batch over an explicit trusted Host pathname and its
+ /// SQLite sidecars. Requires HostPaths; never accepts a read-view handle.
+ fn query_database(
+ &self,
+ input: crate::filesystem::database::Read,
+ ) -> BoxFuture<
+ '_,
+ Result<Vec<crate::filesystem::database::Table>, crate::filesystem::database::Error>,
+ >;
}
#[derive(Clone, Debug, thiserror::Error)]
diff --git a/crates/runtime-host/Cargo.toml b/crates/runtime-host/Cargo.toml
index 37612e926b..0ec934a918 100644
--- a/crates/runtime-host/Cargo.toml
+++ b/crates/runtime-host/Cargo.toml
@@ -24,6 +24,7 @@ license.workspace = true
publish.workspace = true
[dependencies]
+rusqlite = { workspace = true, features = ["hooks", "limits"] }
maka-providers.workspace = true
maka-workhub.workspace = true
maka-assistant.workspace = true
@@ -77,6 +78,7 @@ tempfile.workspace = true
[dev-dependencies]
maka-client.workspace = true
+maka-session-import.workspace = true
tokio-rustls = { version = "0.26.5", default-features = false, features = ["aws_lc_rs", "tls12"] }
tempfile.workspace = true
sqlx.workspace = true
diff --git a/crates/runtime-host/src/plugins/javascript/bridge.rs b/crates/runtime-host/src/plugins/javascript/bridge.rs
index 36c47c3199..cd74319334 100644
--- a/crates/runtime-host/src/plugins/javascript/bridge.rs
+++ b/crates/runtime-host/src/plugins/javascript/bridge.rs
@@ -133,6 +133,7 @@ impl Bridge for HostBridge {
fn max_output_bytes(&self, method: &str) -> usize {
match method {
"model.io" => 32 * 1024 * 1024,
+ "remote.queryDatabase" => maka_plugins::filesystem::database::MAX_BYTES + 1024,
// A 1 MiB byte page expands to at most 4 MiB in a JSON array.
"inputs.read" | "view.read" | "pinnedFile.read" => 5 * 1024 * 1024,
_ => 1024 * 1024,
@@ -423,6 +424,10 @@ impl State {
let caller = self.calls.remote(&input.authority)?;
self.session_view(&input.authority, caller.views.workspace(input.input).await?)
}
+ Request::QueryDatabase(input) => {
+ let caller = self.calls.remote(&input.authority)?;
+ encode(caller.views.query_database(input.input).await?)
+ }
Request::CredentialRead(input) => encode(self.credentials.read(input.key).await?),
Request::CredentialWrite(input) => encode(self.credentials.write(input).await?),
Request::TerminalSpawn(input) => {
diff --git a/crates/runtime-host/src/plugins/javascript/bridge/wire.rs b/crates/runtime-host/src/plugins/javascript/bridge/wire.rs
index f592502c59..916e5490c2 100644
--- a/crates/runtime-host/src/plugins/javascript/bridge/wire.rs
+++ b/crates/runtime-host/src/plugins/javascript/bridge/wire.rs
@@ -61,6 +61,8 @@ pub(super) enum Request {
SessionView(Authority),
#[serde(rename = "remote.workspace")]
WorkspaceView(WorkspaceView),
+ #[serde(rename = "remote.queryDatabase")]
+ QueryDatabase(DatabaseRead),
#[serde(rename = "remote.authorize")]
AuthorizeRemote(RemoteAuthorization),
#[serde(rename = "authorization.open")]
@@ -333,6 +335,12 @@ pub(super) struct WorkspaceView {
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
+pub(super) struct DatabaseRead {
+ pub authority: String,
+ pub input: maka_plugins::filesystem::database::Read,
+}
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
pub(super) struct ProcessWrite {
#[serde(flatten)]
pub target: ProcessHandle,
@@ -418,6 +426,7 @@ pub(super) enum Code {
Unavailable,
Busy,
NotFound,
+ LimitExceeded,
}
#[derive(Serialize)]
pub(super) struct Error {
@@ -462,6 +471,23 @@ impl From<maka_plugins::filesystem::ReadError> for Error {
}
}
}
+impl From<maka_plugins::filesystem::database::Error> for Error {
+ fn from(error: maka_plugins::filesystem::database::Error) -> Self {
+ use maka_plugins::filesystem::database::Error as Database;
+ let code = match &error {
+ Database::Denied | Database::Cancelled => Code::Revoked,
+ Database::Busy => Code::Busy,
+ Database::NotFound => Code::NotFound,
+ Database::Invalid(_) => Code::Invalid,
+ Database::Limit(_) => Code::LimitExceeded,
+ Database::Unavailable(_) => Code::Unavailable,
+ };
+ Self {
+ code,
+ message: error.to_string(),
+ }
+ }
+}
impl From<maka_plugins::Error> for Error {
fn from(error: maka_plugins::Error) -> Self {
Self {
diff --git a/crates/runtime-host/src/server/plugin_remote/views.rs b/crates/runtime-host/src/server/plugin_remote/views.rs
index 39c066a41d..93b8b0d1a8 100644
--- a/crates/runtime-host/src/server/plugin_remote/views.rs
+++ b/crates/runtime-host/src/server/plugin_remote/views.rs
@@ -23,6 +23,7 @@ use maka_plugins::remote::{Access, Error, SessionView, Views, WorkspaceViewInput
use std::sync::{Arc, Weak};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;
+mod database;
mod read;
use read::ReadGrant;
@@ -127,6 +128,18 @@ impl SessionViews {
}
}
impl Views for SessionViews {
+ fn query_database(
+ &self,
+ input: maka_plugins::filesystem::database::Read,
+ ) -> BoxFuture<
+ '_,
+ Result<
+ Vec<maka_plugins::filesystem::database::Table>,
+ maka_plugins::filesystem::database::Error,
+ >,
+ > {
+ Box::pin(database::read(self, input))
+ }
fn authorize(
&self,
request: maka_plugins::authorization::Request,
diff --git a/crates/runtime-host/src/server/plugin_remote/views/database.rs b/crates/runtime-host/src/server/plugin_remote/views/database.rs
new file mode 100644
index 0000000000..41442d689c
--- /dev/null
+++ b/crates/runtime-host/src/server/plugin_remote/views/database.rs
@@ -0,0 +1,159 @@
+/*
+ * 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.
+ */
+
+mod query;
+use super::{Access, SessionConfiguration, SessionViews};
+use maka_plugins::{
+ composition::Scope,
+ filesystem::{
+ OpenFile, ReadError, Symlinks,
+ database::{self as api, Error},
+ },
+};
+use std::{
+ path::Path,
+ sync::{Arc, LazyLock},
+};
+use tokio::sync::Semaphore;
+
+static READS: LazyLock<Arc<Semaphore>> = LazyLock::new(|| Arc::new(Semaphore::new(2)));
+
+pub(super) async fn read(views: &SessionViews, input: api::Read) -> Result<Vec<api::Table>, Error> {
+ input.validate()?;
+ if views.access != Access::HostPaths {
+ return Err(Error::Denied);
+ }
+ let lease = views.owner.resource_call().map_err(|_| Error::Denied)?;
+ let stopping = views.owner.stopping().map_err(|_| Error::Denied)?;
+ let host = views.host().await.map_err(remote_error)?;
+ let identity = views.owner.identity().map_err(|_| Error::Denied)?;
+ let path = tokio::fs::canonicalize(&input.path)
+ .await
+ .map_err(io_error)?;
+ // Database reads never expose Host authority, even from an unrestricted Session.
+ if path.starts_with(host.root.canonical_path()) {
+ return Err(Error::Denied);
+ }
+ let scoped = matches!(identity.scope, Scope::Session(_));
+ let workspace = if scoped {
+ let id = views.session_id.as_ref().ok_or(Error::Denied)?;
+ host.log
+ .get_session::<SessionConfiguration>(id)
+ .await
+ .map_err(|error| Error::Unavailable(error.to_string()))?
+ .ok_or(Error::Denied)?
+ .configuration
+ .workspace
+ } else {
+ let parent = path
+ .parent()
+ .ok_or_else(|| Error::Invalid("database has no parent directory".into()))?;
+ let parent = maka_fs_tools::workspace::project::host_path(parent)
+ .map_err(|error| Error::Invalid(error.to_string()))?;
+ crate::server::resolve_workspace_path(parent.to_owned())
+ .await
+ .map_err(|error| Error::Invalid(error.message))?
+ };
+ let host_path = maka_fs_tools::workspace::project::host_path(&path)
+ .map_err(|error| Error::Invalid(error.to_string()))?;
+ let relative = Path::new(host_path)
+ .strip_prefix(Path::new(&workspace.host_cwd))
+ .map_err(|_| Error::Denied)?
+ .components()
+ .map(|part| match part {
+ std::path::Component::Normal(name) => name.to_str().ok_or(Error::Denied),
+ _ => Err(Error::Denied),
+ })
+ .collect::<Result<Vec<_>, _>>()?
+ .join("/");
+ let files = views
+ .files(&workspace, scoped)
+ .await
+ .map_err(remote_error)?;
+ let names = relative.clone();
+ files
+ .with_reader(move |reader| {
+ reader.file_info(OpenFile {
+ path: names.clone(),
+ symlinks: Symlinks::Reject,
+ })?;
+ for suffix in ["-wal", "-shm", "-journal"] {
+ match reader.file_info(OpenFile {
+ path: format!("{names}{suffix}"),
+ symlinks: Symlinks::Reject,
+ }) {
+ Ok(_) => {}
+ Err(ReadError::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => {}
+ Err(error) => return Err(error),
+ }
+ }
+ Ok::<_, ReadError>(())
+ })
+ .await
+ .map_err(read_error)?
+ .map_err(read_error)?;
+ let permit = READS.clone().try_acquire_owned().map_err(|_| Error::Busy)?;
+ let mut ticket = views.resources.reserve().map_err(|_| Error::Cancelled)?;
+ let cancelled = views.cancellation.clone();
+ let result = tokio::task::spawn_blocking(move || {
+ let _lease = lease;
+ let _permit = permit;
+ ticket.start();
+ let result = query::read(&path, input.queries, &cancelled, &stopping);
+ // A rejected read has no unresolved external effect. Close happens
+ // inside query::read, before releasing admission or acknowledging cleanup.
+ ticket.complete(Ok(()));
+ result
+ })
+ .await
+ .map_err(|error| Error::Unavailable(error.to_string()))?;
+ // Retain the original ReadGrant: a changed boundary or revoked credential
+ // cannot be replaced by a newly authorized scope after work completed.
+ files
+ .file_info(OpenFile {
+ path: relative,
+ symlinks: Symlinks::Reject,
+ })
+ .await
+ .map_err(read_error)?;
+ result
+}
+fn io_error(error: std::io::Error) -> Error {
+ match error.kind() {
+ std::io::ErrorKind::NotFound => Error::NotFound,
+ std::io::ErrorKind::PermissionDenied => Error::Denied,
+ _ => Error::Unavailable(error.to_string()),
+ }
+}
+fn read_error(error: ReadError) -> Error {
+ match error {
+ ReadError::Retired => Error::Cancelled,
+ ReadError::Io(error) => io_error(error),
+ ReadError::Invalid(_) => Error::Denied,
+ ReadError::ScanLimit { .. } => Error::Limit(api::Limit::Work),
+ }
+}
+fn remote_error(error: super::Error) -> Error {
+ match error {
+ super::Error::Retired => Error::Denied,
+ super::Error::Cancelled => Error::Cancelled,
+ super::Error::Invalid(message) => Error::Invalid(message),
+ error => Error::Unavailable(error.to_string()),
+ }
+}
diff --git a/crates/runtime-host/src/server/plugin_remote/views/database/query.rs b/crates/runtime-host/src/server/plugin_remote/views/database/query.rs
new file mode 100644
index 0000000000..c6f1a0fe13
--- /dev/null
+++ b/crates/runtime-host/src/server/plugin_remote/views/database/query.rs
@@ -0,0 +1,455 @@
+/*
+ * 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_plugins::filesystem::database::{Cell, Error, Limit, MAX_BYTES, MAX_ROWS, Query, Table};
+use rusqlite::{
+ Connection, OpenFlags,
+ fallible_iterator::FallibleIterator,
+ hooks::{AuthAction, AuthContext, Authorization},
+ limits::Limit as SqlLimit,
+ types::{ToSqlOutput, ValueRef},
+};
+use std::{
+ path::Path,
+ sync::{
+ Arc,
+ atomic::{AtomicBool, Ordering},
+ },
+ time::{Duration, Instant},
+};
+use tokio_util::sync::CancellationToken;
+
+pub(super) fn read(
+ path: &Path,
+ queries: Vec<Query>,
+ cancelled: &CancellationToken,
+ stopping: &CancellationToken,
+) -> Result<Vec<Table>, Error> {
+ if cancelled.is_cancelled() || stopping.is_cancelled() {
+ return Err(Error::Cancelled);
+ }
+ let connection = Connection::open_with_flags(
+ path,
+ OpenFlags::SQLITE_OPEN_READ_ONLY
+ | OpenFlags::SQLITE_OPEN_NO_MUTEX
+ | OpenFlags::SQLITE_OPEN_NOFOLLOW,
+ )
+ .map_err(failed)?;
+ read_connection(connection, queries, cancelled, stopping)
+}
+
+fn read_connection(
+ connection: Connection,
+ queries: Vec<Query>,
+ cancelled: &CancellationToken,
+ stopping: &CancellationToken,
+) -> Result<Vec<Table>, Error> {
+ connection
+ .busy_timeout(Duration::from_millis(250))
+ .map_err(failed)?;
+ connection
+ .pragma_update(None, "query_only", true)
+ .map_err(failed)?;
+ connection
+ .pragma_update(None, "trusted_schema", false)
+ .map_err(failed)?;
+ connection
+ .pragma_update(None, "temp_store", "FILE")
+ .map_err(failed)?;
+ connection
+ .pragma_update(None, "cache_size", -2048)
+ .map_err(failed)?;
+ for (limit, value) in [
+ (SqlLimit::SQLITE_LIMIT_LENGTH, 8 * 1024 * 1024),
+ (SqlLimit::SQLITE_LIMIT_SQL_LENGTH, 128 * 1024),
+ (SqlLimit::SQLITE_LIMIT_COLUMN, 32),
+ (SqlLimit::SQLITE_LIMIT_EXPR_DEPTH, 32),
+ (SqlLimit::SQLITE_LIMIT_FUNCTION_ARG, 16),
+ (SqlLimit::SQLITE_LIMIT_LIKE_PATTERN_LENGTH, 1024),
+ (SqlLimit::SQLITE_LIMIT_COMPOUND_SELECT, 32),
+ (SqlLimit::SQLITE_LIMIT_VARIABLE_NUMBER, 128),
+ (SqlLimit::SQLITE_LIMIT_ATTACHED, 0),
+ ] {
+ connection.set_limit(limit, value).map_err(failed)?;
+ }
+ let deadline = Instant::now() + Duration::from_secs(30);
+ let exhausted = Arc::new(AtomicBool::new(false));
+ let work_exhausted = exhausted.clone();
+ let abort = cancelled.clone();
+ let retired = stopping.clone();
+ let mut work = 0_u64;
+ connection
+ .progress_handler(
+ 1000,
+ Some(move || {
+ work += 1000;
+ if work >= 50_000_000 || Instant::now() >= deadline {
+ work_exhausted.store(true, Ordering::Relaxed);
+ }
+ abort.is_cancelled()
+ || retired.is_cancelled()
+ || work_exhausted.load(Ordering::Relaxed)
+ }),
+ )
+ .map_err(failed)?;
+ connection.execute_batch("BEGIN DEFERRED").map_err(failed)?;
+ connection.authorizer(Some(authorize)).map_err(failed)?;
+ let result = collect(&connection, queries, cancelled, stopping, deadline);
+ // Closing the connection releases the read transaction even after a rejected
+ // statement. Never send user SQL a COMMIT/ROLLBACK or control hook capability.
+ let closed = connection.close().map_err(|(_, error)| failed(error));
+ if cancelled.is_cancelled() || stopping.is_cancelled() {
+ return Err(Error::Cancelled);
+ }
+ if exhausted.load(Ordering::Relaxed) || Instant::now() >= deadline {
+ return Err(Error::Limit(Limit::Work));
+ }
+ closed?;
+ result
+}
+fn authorize(context: AuthContext<'_>) -> Authorization {
+ use Authorization::{Allow, Deny};
+ match context.action {
+ // SQLite also reports statement-local CTE reads without a database name.
+ // ATTACH and user-created temporary tables remain forbidden below.
+ AuthAction::Read { .. } if matches!(context.database_name, None | Some("main")) => Allow,
+ AuthAction::Select | AuthAction::Recursive => Allow,
+ AuthAction::Function { function_name }
+ if [
+ "coalesce",
+ "ifnull",
+ "nullif",
+ "length",
+ "octet_length",
+ "typeof",
+ "lower",
+ "upper",
+ "trim",
+ "ltrim",
+ "rtrim",
+ "substr",
+ "substring",
+ "instr",
+ "like",
+ "glob",
+ "min",
+ "max",
+ "count",
+ "sum",
+ "total",
+ "avg",
+ "json_valid",
+ "json_extract",
+ "json_type",
+ "json_array_length",
+ ]
+ .iter()
+ .any(|name| function_name.eq_ignore_ascii_case(name)) =>
+ {
+ Allow
+ }
+ // Schema inspection is observational; arbitrary PRAGMA is not.
+ AuthAction::Pragma { pragma_name, .. }
+ if [
+ "table_info",
+ "table_xinfo",
+ "table_list",
+ "index_list",
+ "index_info",
+ ]
+ .iter()
+ .any(|name| pragma_name.eq_ignore_ascii_case(name)) =>
+ {
+ Allow
+ }
+ _ => Deny,
+ }
+}
+fn collect(
+ connection: &Connection,
+ queries: Vec<Query>,
+ cancelled: &CancellationToken,
+ stopping: &CancellationToken,
+ deadline: Instant,
+) -> Result<Vec<Table>, Error> {
+ let mut tables = Vec::new();
+ let mut budget = Budget(2);
+ let mut count = 0;
+ for query in queries {
+ if cancelled.is_cancelled() || stopping.is_cancelled() {
+ return Err(Error::Cancelled);
+ }
+ if Instant::now() >= deadline {
+ return Err(Error::Limit(Limit::Work));
+ }
+ // Batch rejects additional statements instead of silently executing a prefix.
+ let mut statements = rusqlite::Batch::new(connection, &query.sql);
+ let mut statement = statements
+ .next()
+ .map_err(failed)?
+ .ok_or_else(|| Error::Invalid("empty statement".into()))?;
+ if statements.next().map_err(failed)?.is_some() {
+ return Err(Error::Invalid(
+ "each query must contain one statement".into(),
+ ));
+ }
+ if !statement.readonly() {
+ return Err(Error::Invalid("statement is not a read-only query".into()));
+ }
+ let columns: Vec<String> = statement
+ .column_names()
+ .iter()
+ .map(|column| (*column).into())
+ .collect();
+ budget.add(&columns)?;
+ budget.reserve(32)?;
+ let parameters = query.parameters.iter().map(Parameter).collect::<Vec<_>>();
+ let mut rows = statement
+ .query(rusqlite::params_from_iter(parameters.iter()))
+ .map_err(failed)?;
+ let mut values = Vec::new();
+ while let Some(row) = rows.next().map_err(failed)? {
+ if cancelled.is_cancelled() || stopping.is_cancelled() {
+ return Err(Error::Cancelled);
+ }
+ if Instant::now() >= deadline {
+ return Err(Error::Limit(Limit::Work));
+ }
+ count += 1;
+ if count > MAX_ROWS {
+ return Err(Error::Limit(Limit::Rows));
+ }
+ budget.reserve(3)?; // Row brackets and separator.
+ let mut value = Vec::with_capacity(columns.len());
+ for index in 0..columns.len() {
+ let field = row.get_ref(index).map_err(failed)?;
+ // Check borrowed payloads before allocating; do not materialize an
+ // entire oversized row before discovering the output limit.
+ if let ValueRef::Text(bytes) | ValueRef::Blob(bytes) = field
+ && bytes.len() > MAX_BYTES.saturating_sub(budget.0)
+ {
+ return Err(Error::Limit(Limit::Bytes));
+ }
+ let cell = match field {
+ ValueRef::Null => Cell::Null,
+ ValueRef::Integer(value) => Cell::Integer(value),
+ ValueRef::Real(value) if value.is_finite() => Cell::Real(value),
+ ValueRef::Real(_) => {
+ return Err(Error::Invalid("non-finite database number".into()));
+ }
+ ValueRef::Text(bytes) => Cell::Text(
+ std::str::from_utf8(bytes)
+ .map_err(|_| Error::Invalid("database text is not UTF-8".into()))?
+ .into(),
+ ),
+ ValueRef::Blob(bytes) => Cell::Blob(bytes.into()),
+ };
+ budget.add(&cell)?;
+ value.push(cell);
+ }
+ values.push(value);
+ }
+ tables.push(Table {
+ columns,
+ rows: values,
+ });
+ }
+ Ok(tables)
+}
+struct Parameter<'a>(&'a Cell);
+impl rusqlite::ToSql for Parameter<'_> {
+ fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
+ Ok(ToSqlOutput::Borrowed(match self.0 {
+ Cell::Null => ValueRef::Null,
+ Cell::Integer(value) => ValueRef::Integer(*value),
+ Cell::Real(value) => ValueRef::Real(*value),
+ Cell::Text(value) => ValueRef::Text(value.as_bytes()),
+ Cell::Blob(value) => ValueRef::Blob(value),
+ }))
+ }
+}
+struct Budget(usize);
+impl Budget {
+ fn reserve(&mut self, bytes: usize) -> Result<(), Error> {
+ if bytes > MAX_BYTES.saturating_sub(self.0) {
+ return Err(Error::Limit(Limit::Bytes));
+ }
+ self.0 += bytes;
+ Ok(())
+ }
+ fn add(&mut self, value: &impl serde::Serialize) -> Result<(), Error> {
+ serde_json::to_writer(&mut *self, value).map_err(|_| Error::Limit(Limit::Bytes))?;
+ self.reserve(1)
+ }
+}
+impl std::io::Write for Budget {
+ fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+ if bytes.len() > MAX_BYTES.saturating_sub(self.0) {
+ return Err(std::io::Error::other("database result byte limit"));
+ }
+ self.0 += bytes.len();
+ Ok(bytes.len())
+ }
+ fn flush(&mut self) -> std::io::Result<()> {
+ Ok(())
+ }
+}
+fn failed(error: rusqlite::Error) -> Error {
+ match error.sqlite_error_code() {
+ Some(
+ rusqlite::ErrorCode::AuthorizationForStatementDenied | rusqlite::ErrorCode::ReadOnly,
+ ) => Error::Invalid("statement is not an allowed database read".into()),
+ Some(rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked) => {
+ Error::Busy
+ }
+ Some(rusqlite::ErrorCode::TooBig) => Error::Limit(Limit::Bytes),
+ _ => Error::Unavailable(error.to_string()),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ fn query(sql: &str) -> Query {
+ Query {
+ sql: sql.into(),
+ parameters: vec![],
+ }
+ }
+ #[test]
+ fn wal_snapshot_covers_the_batch_and_policy_denies_mutation_and_unbounded_work() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory
+ .path()
+ .canonicalize()
+ .unwrap()
+ .join("source.sqlite");
+ let writer = Connection::open(&path).unwrap();
+ writer.execute_batch("PRAGMA journal_mode=WAL; CREATE TABLE source(value INTEGER); INSERT INTO source VALUES(1)").unwrap();
+ let reader = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_ONLY).unwrap();
+ let source = path.clone();
+ // Deterministically commit a real second connection between batch reads.
+ // Override an allowed builtin only in this test; production registers no
+ // plugin-defined SQL functions or special test authorizer exceptions.
+ reader
+ .create_scalar_function(
+ "length",
+ 1,
+ rusqlite::functions::FunctionFlags::SQLITE_UTF8,
+ move |_| {
+ let writer = Connection::open(&source)?;
+ writer.execute("UPDATE source SET value=2", [])?;
+ Ok(0_i64)
+ },
+ )
+ .unwrap();
+ let cancelled = CancellationToken::new();
+ let retired = CancellationToken::new();
+ let tables = read_connection(
+ reader,
+ vec![
+ query("SELECT value FROM source"),
+ query("SELECT length('advance writer')"),
+ query("SELECT value, 9223372036854775807, x'00FF' FROM source"),
+ ],
+ &cancelled,
+ &retired,
+ )
+ .unwrap();
+ assert_eq!(tables[0].rows, [vec![Cell::Integer(1)]]);
+ assert_eq!(
+ tables[2].rows,
+ [vec![
+ Cell::Integer(1),
+ Cell::Integer(i64::MAX),
+ Cell::Blob(vec![0, 255])
+ ]]
+ );
+ assert_eq!(
+ serde_json::to_value(&tables[2]).unwrap()["rows"][0][1]["value"],
+ "9223372036854775807"
+ );
+ assert_eq!(
+ writer
+ .query_row("SELECT value FROM source", [], |row| row.get::<_, i64>(0))
+ .unwrap(),
+ 2
+ );
+ for sql in [
+ "UPDATE source SET value=3",
+ "CREATE TEMP TABLE scratch(x)",
+ "ATTACH DATABASE ':memory:' AS other",
+ "PRAGMA query_only=OFF",
+ "PRAGMA journal_mode=DELETE",
+ "SELECT load_extension('outside')",
+ "SELECT randomblob(8000000)",
+ "SELECT zeroblob(9000000)",
+ "SELECT printf('%0*d',9000000,0)",
+ "SELECT 1; SELECT 2",
+ ] {
+ assert!(
+ read(&path, vec![query(sql)], &cancelled, &retired).is_err(),
+ "{sql}"
+ );
+ }
+ let joined = read(&path, vec![query("WITH RECURSIVE n(x) AS (VALUES(1) UNION ALL SELECT x+1 FROM n WHERE x<35) SELECT value FROM source,n")], &cancelled, &retired).unwrap();
+ assert_eq!(joined[0].rows.len(), 35);
+ writer
+ .execute_batch("CREATE TABLE oversized(payload BLOB)")
+ .unwrap();
+ writer
+ .execute("INSERT INTO oversized VALUES(?1)", [vec![0_u8; 9_000_000]])
+ .unwrap();
+ let oversized = read(
+ &path,
+ vec![query("SELECT payload FROM oversized")],
+ &cancelled,
+ &retired,
+ );
+ assert!(
+ matches!(oversized, Err(Error::Limit(Limit::Bytes))),
+ "{oversized:?}"
+ );
+ assert!(matches!(
+ read(
+ &path,
+ vec![query(
+ "WITH RECURSIVE n(x) AS (VALUES(1) UNION ALL SELECT x+1 FROM n) SELECT sum(x) FROM n"
+ )],
+ &cancelled,
+ &retired
+ ),
+ Err(Error::Limit(Limit::Work))
+ ));
+ cancelled.cancel();
+ assert!(matches!(
+ read(&path, vec![query("SELECT 1")], &cancelled, &retired),
+ Err(Error::Cancelled)
+ ));
+ assert_eq!(
+ writer
+ .query_row("SELECT value FROM source", [], |row| row.get::<_, i64>(0))
+ .unwrap(),
+ 2
+ );
+ writer
+ .execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")
+ .unwrap();
+ }
+}
diff --git a/crates/runtime-host/tests/fixtures/remote-plugin.mjs b/crates/runtime-host/tests/fixtures/remote-plugin.mjs
index 77a56cbc89..009a2300e8 100644
--- a/crates/runtime-host/tests/fixtures/remote-plugin.mjs
+++ b/crates/runtime-host/tests/fixtures/remote-plugin.mjs
@@ -19,6 +19,45 @@
/** @param {import('../../../../packages/plugin-sdk/src/host.js').HostContext} ctx */
export default async function activate(ctx) {
+ /** @type {import('../../../../packages/plugin-sdk/src/host.js').RemoteCaller | undefined} */
+ let previousDatabase;
+ const database = async (
+ /** @type {import('../../../../packages/plugin-sdk/src/host.js').Json} */ input,
+ /** @type {import('../../../../packages/plugin-sdk/src/host.js').RemoteCaller} */ caller,
+ ) => {
+ const request =
+ /** @type {import('../../../../packages/plugin-sdk/src/host.js').DatabaseRead} */ (
+ /** @type {unknown} */ (input)
+ );
+ if (previousDatabase) {
+ try {
+ await previousDatabase.views.queryDatabase(request);
+ throw new Error('completed Remote call retained database authority');
+ } catch (error) {
+ if (error.code !== 'revoked') throw error;
+ }
+ }
+ const tables = await caller.views.queryDatabase(request);
+ previousDatabase = caller;
+ return tables.map((table) => ({ ...table }));
+ };
+ await ctx.remote.method('database', database, { access: 'host_paths' });
+ await ctx.remote.method(
+ 'database-summary',
+ async (input, caller) => {
+ const tables = await database(input, caller);
+ return tables
+ .flatMap((table) => table.rows)
+ .reduce(
+ (size, row) =>
+ size +
+ row.reduce((sum, cell) => sum + (cell.kind === 'text' ? cell.value.length : 0), 0),
+ 0,
+ );
+ },
+ { access: 'host_paths' },
+ );
+ await ctx.remote.method('denied-database', database);
await ctx.remote.method('uncertain', () => {
/** @type {import('../../../../packages/plugin-sdk/src/host.js').RemoteFailure} */
const failure = Object.assign(new Error('publication result needs recovery'), {
diff --git a/crates/runtime-host/tests/integration/plugin_remote.rs b/crates/runtime-host/tests/integration/plugin_remote.rs
index 4ae7fbca3a..5dbd3a30d2 100644
--- a/crates/runtime-host/tests/integration/plugin_remote.rs
+++ b/crates/runtime-host/tests/integration/plugin_remote.rs
@@ -113,6 +113,47 @@ async fn scenario() {
let binding = json!({"client":client,"method":"echo","sessionId":null});
let target = rpc(&mut peer, json!({"kind":"bind","binding":binding})).await["target"].clone();
let document = rpc(&mut peer, json!({"kind":"open_document"})).await["document"].clone();
+ // A native plugin consumes the same Host-path capability as external JS.
+ // Keep the writer open: the selected OpenCode rows live in a real WAL.
+ let database_path = fixture.workspace.join("opencode.sqlite");
+ let database = rusqlite::Connection::open(&database_path).unwrap();
+ database
+ .execute_batch(
+ "PRAGMA journal_mode=WAL;
+ CREATE TABLE session(id TEXT,parent_id TEXT,directory TEXT,title TEXT,revert TEXT);
+ CREATE TABLE message(id TEXT,session_id TEXT,time_created INTEGER,data TEXT);
+ CREATE TABLE part(id TEXT,message_id TEXT,session_id TEXT,time_created INTEGER,data TEXT);
+ INSERT INTO session VALUES('selected',NULL,'/source/project','Selected conversation',NULL);
+ INSERT INTO message VALUES('m','selected',10,'{\"role\":\"user\"}');
+ INSERT INTO part VALUES('p','m','selected',10,'{\"type\":\"text\",\"text\":\"From WAL\"}');
+ INSERT INTO session VALUES('other',NULL,'/source/other','Excluded',NULL);
+ INSERT INTO message VALUES('other-m','other',10,'broken JSON');",
+ )
+ .unwrap();
+ let import_binding = json!({"client":client,"method":"import-history","sessionId":null});
+ let import_target =
+ rpc(&mut peer, json!({"kind":"bind","binding":import_binding})).await["target"].clone();
+ let import_call = json!({"kind":"call","binding":import_binding,"target":import_target,
+ "document":document,"input":{"path":database_path,"session":"selected"}});
+ let imported = rpc(&mut peer, import_call.clone()).await;
+ let transcript: maka_session_import::Transcript =
+ serde_json::from_value(imported["value"].clone()).unwrap();
+ assert_eq!(transcript.title, "Selected conversation");
+ assert_eq!(transcript.records.len(), 1);
+ assert!(
+ matches!(&transcript.records[0].content, maka_runtime::import::Content::User { text } if text == "From WAL")
+ );
+ database
+ .execute(
+ "UPDATE session SET parent_id='parent' WHERE id='selected'",
+ [],
+ )
+ .unwrap();
+ assert_eq!(peer.rpc("plugin.remote", import_call).await["ok"], false);
+ database
+ .execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")
+ .unwrap();
+ drop(database);
let call = json!({"kind":"call","binding":binding,"target":target,"document":document,"input":{"hello":"world"}});
let mut uncertain = call.clone();
uncertain["input"] = json!("uncertain");
diff --git a/crates/runtime-host/tests/integration/plugin_remote/fixture.rs b/crates/runtime-host/tests/integration/plugin_remote/fixture.rs
index dd5a2926c7..2f5c861b85 100644
--- a/crates/runtime-host/tests/integration/plugin_remote/fixture.rs
+++ b/crates/runtime-host/tests/integration/plugin_remote/fixture.rs
@@ -70,6 +70,16 @@ impl Plugin for Example {
)
.unwrap();
} else {
+ staged
+ .insert(
+ key(&identity.package_id, "import-history").unwrap(),
+ Endpoint::new(
+ bundle.content_digest.clone(),
+ Handler::Method(Arc::new(ImportHistory)),
+ )
+ .requiring_host_paths(),
+ )
+ .unwrap();
for (name, handler) in [
(
"echo",
@@ -92,6 +102,29 @@ impl Plugin for Example {
})
}
}
+struct ImportHistory;
+impl Method for ImportHistory {
+ fn call(&self, input: Value, caller: Caller) -> BoxFuture<'static, Result<Value, Error>> {
+ Box::pin(async move {
+ #[derive(serde::Deserialize)]
+ #[serde(deny_unknown_fields)]
+ struct Input {
+ path: String,
+ session: String,
+ }
+ let input: Input =
+ serde_json::from_value(input).map_err(|error| Error::Invalid(error.to_string()))?;
+ let transcript = maka_session_import::opencode::read(
+ caller.views.as_ref(),
+ input.path,
+ &input.session,
+ )
+ .await
+ .map_err(|error| Error::Provider(error.to_string()))?;
+ serde_json::to_value(transcript).map_err(|error| Error::Provider(error.to_string()))
+ })
+ }
+}
impl Method for Adapter {
fn call(&self, input: Value, caller: Caller) -> BoxFuture<'static, Result<Value, Error>> {
if input == "uncertain" {
diff --git a/crates/runtime-host/tests/integration/plugin_remote/javascript.rs b/crates/runtime-host/tests/integration/plugin_remote/javascript.rs
index 1e4e0a1912..bab95d49d3 100644
--- a/crates/runtime-host/tests/integration/plugin_remote/javascript.rs
+++ b/crates/runtime-host/tests/integration/plugin_remote/javascript.rs
@@ -26,12 +26,25 @@ use tokio_util::sync::CancellationToken;
#[tokio::test(flavor = "multi_thread", worker_threads = 3)]
async fn javascript_remote_replaces_exact_registration_and_closes_late_vm_streams() {
- tokio::time::timeout(Duration::from_secs(30), scenario())
- .await
- .unwrap();
+ tokio::time::timeout(Duration::from_secs(45), async {
+ for vm in ["shared", "dedicated"] {
+ scenario(vm).await;
+ }
+ })
+ .await
+ .unwrap();
}
-async fn scenario() {
+async fn scenario(vm: &str) {
let fixture = ClientFixture::new("maka-js-remote-");
+ let database_path = fixture.workspace.join("source.sqlite");
+ let source = rusqlite::Connection::open(&database_path).unwrap();
+ source.execute_batch("PRAGMA journal_mode=WAL; CREATE TABLE source(value INTEGER); INSERT INTO source VALUES(7)").unwrap();
+ source
+ .execute_batch("CREATE TABLE large(payload TEXT)")
+ .unwrap();
+ source
+ .execute("INSERT INTO large VALUES(?1)", ["0".repeat(1_000_000)])
+ .unwrap();
let canonical_workspace = fixture.workspace.canonicalize().unwrap();
let expected_cwd = maka_fs_tools::workspace::project::host_path(&canonical_workspace).unwrap();
let path = fixture.workspace.join("plugin");
@@ -46,13 +59,19 @@ async fn scenario() {
path.join("maka.extension.json"),
serde_json::to_vec(&json!({
"schemaVersion":1,"id":"example.remote",
- "runtime":{"entry":"host.mjs","sdkVersion":1,"vm":"dedicated"},
+ "runtime":{"entry":"host.mjs","sdkVersion":1,"vm":vm},
"client":{"entry":"client.js","sdkVersion":1},
}))
.unwrap(),
)
.unwrap();
- let host = Host::open(fixture.owner()).await.unwrap();
+ let owner = fixture.owner();
+ let denied_path = owner.canonical_path().join("private.sqlite");
+ rusqlite::Connection::open(&denied_path)
+ .unwrap()
+ .execute_batch("CREATE TABLE source(value INTEGER)")
+ .unwrap();
+ let host = Host::open(owner).await.unwrap();
#[cfg(unix)]
let endpoint = fixture.workspace.parent().unwrap().join("js-remote.sock");
#[cfg(windows)]
@@ -120,6 +139,55 @@ async fn scenario() {
.await;
// Business uncertainty does not fence a fully settled provider or revoke
// the caller's document: normal calls below must still succeed.
+ for (method, path, sql, allowed) in [
+ (
+ "database",
+ &database_path,
+ "SELECT value, ? AS precise, x'00ff' AS bytes FROM source",
+ true,
+ ),
+ (
+ "database",
+ &database_path,
+ "SELECT value, ? AS precise, x'00ff' AS bytes FROM source",
+ true,
+ ),
+ ("denied-database", &database_path, "SELECT ?", false),
+ ("database", &denied_path, "SELECT ?", false),
+ (
+ "database",
+ &database_path,
+ "UPDATE source SET value=?",
+ false,
+ ),
+ ] {
+ let (binding, target) = bind(&mut peer, &client, method).await;
+ let result = peer.rpc("plugin.remote", json!({"kind":"call", "binding":binding, "target":target, "document":document,
+ "input":{"path":path,"queries":[{"sql":sql,"parameters":[{"kind":"integer","value":"9223372036854775807"}]}]}})).await;
+ assert_eq!(result["ok"], allowed, "{result}");
+ if allowed {
+ assert_eq!(
+ result["result"]["value"][0]["rows"],
+ json!([[{"kind":"integer","value":"7"},{"kind":"integer","value":"9223372036854775807"},{"kind":"blob","value":"AP8="}]])
+ );
+ }
+ }
+ assert_eq!(
+ source
+ .query_row("SELECT value FROM source", [], |row| row.get::<_, i64>(0))
+ .unwrap(),
+ 7
+ );
+ let (binding, target) = bind(&mut peer, &client, "database-summary").await;
+ let large = peer.rpc("plugin.remote", json!({"kind":"call", "binding":binding, "target":target, "document":document,
+ "input":{"path":database_path,"queries":[{"sql":"WITH RECURSIVE n(x) AS (VALUES(1) UNION ALL SELECT x+1 FROM n WHERE x<35) SELECT payload FROM large,n"}]}})).await;
+ if large["ok"] != true {
+ let status = peer
+ .rpc("plugin.platform.query", json!({"view":"failures"}))
+ .await;
+ panic!("{vm}: {large}; platform: {status}");
+ }
+ assert_eq!(large["result"]["value"], 35_000_000);
for method in ["workspace", "workspace", "denied-workspace"] {
let (binding, target) = bind(&mut peer, &client, method).await;
let result = peer
diff --git a/crates/session-import/README.md b/crates/session-import/README.md
index 9eac7a142a..3d98dc3d49 100644
--- a/crates/session-import/README.md
+++ b/crates/session-import/README.md
@@ -21,15 +21,16 @@
[简体中文](README.zh-CN.md)
-Converts Codex rollouts and Claude Code transcripts into historical records through public plugin read capabilities. The caller supplies a pinned file and the selected source identity; this crate cannot reopen paths, dispatch tools or write canonical events.
+Converts Codex, Claude Code and OpenCode conversations into historical records through public plugin read capabilities. Callers supply the selected source identity and a pinned file or authorized database read view. This crate cannot open ambient paths, dispatch tools or write canonical events.
- Codex uses conversation events and tool response items, applies recorded rollbacks, and excludes provider message mirrors.
- Claude selects rewritten prompts without dropping parallel results or compaction roots. Three digest-checked passes resolve lineage, assemble response fragments and emit history.
+- OpenCode reads the selected root Session, messages and parts in one SQLite snapshot, applying whole-message and partial-message reverts. Invalid cells, duplicate identities, orphan parts and missing completed-tool outcomes fail the import; pending calls remain pending.
- Tool calls and results retain their source order and correlation. Missing outcomes stay missing.
- Filesystem catalogs read bounded summaries (Codex: 512 KiB; Claude: 256 KiB head and tail), apply workspace/archive filters, and paginate by modification time and relative path. Query-bound cursors resume after the last delivered entry, including when the 48 KiB wire budget cuts a page short.
- Source workspace/model values are observations, not execution configuration. No credentials or provider options enter the result.
- Corrupt interior JSON, inconsistent source identity and changing multi-pass input fail the import. Only an unfinished final JSON write is omitted and identified in the fingerprint.
-Reads are bounded: 2 GiB per source prefix, 64 MiB per JSONL line, one million source lines and 65,536 structural JSON tokens per line. Retained payloads are checked before decoding; encoded history is limited to the Runtime's 6 MiB / 7,500-record budget. Claude additionally bounds its lineage index and response fragments. Oversized input is rejected, not truncated.
+JSONL reads are bounded: 2 GiB per source prefix, 64 MiB per line, one million lines and 65,536 structural JSON tokens per record. OpenCode snapshots also obey the public database row, byte and work limits. Retained payloads are checked before decoding; encoded history is limited to the Runtime's 6 MiB / 7,500-record budget. Claude additionally bounds its lineage index and response fragments. Oversized input is rejected, not truncated.
The returned transcript is not a published Session. Publication and retry receipts belong to the public Session import capability.
diff --git a/crates/session-import/README.zh-CN.md b/crates/session-import/README.zh-CN.md
index 52386e4eaf..d417ed0442 100644
--- a/crates/session-import/README.zh-CN.md
+++ b/crates/session-import/README.zh-CN.md
@@ -21,15 +21,16 @@
[English](README.md)
-通过公共插件读取能力，将 Codex rollout 和 Claude Code transcript 转换为历史记录。调用方提供固定文件句柄和所选来源身份；本 crate 不重新打开路径、不派发工具，也不写入规范事件。
+通过公共插件读取能力，将 Codex、Claude Code 和 OpenCode 对话转换为历史记录。调用方提供所选来源身份，以及固定文件句柄或已授权的数据库读取视图；本 crate 不自行打开环境路径、不派发工具，也不写入规范事件。
- Codex 使用对话事件与工具响应记录，应用来源回滚，排除重复的 provider 消息镜像。
- Claude 选择修改后的用户提问，但保留并行工具结果及压缩前历史。三次校验摘要的读取依次解析分支、合并回复分片、生成历史。
+- OpenCode 在同一个 SQLite 快照中读取所选根 Session、消息和片段，应用整条消息或部分片段的撤回标记。字段类型错误、重复身份、孤立片段或已完成工具缺少结果时拒绝导入；未完成调用保持未完成。
- 工具调用和结果保留来源顺序与关联；缺失的结果仍然缺失。
- 文件目录查询只读有界摘要（Codex：512 KiB；Claude：头尾各 256 KiB），按工作目录和归档条件筛选，以修改时间和相对路径分页。游标绑定查询，即使页面因 48 KiB 传输预算提前结束，也从最后实际交付的记录之后继续。
- 来源工作目录和模型名称仅是观察信息，不是执行配置；凭据和 provider options 不进入结果。
- 中间 JSON 损坏、来源身份不符或多次读取内容变化时拒绝导入。只有末尾未完成的 JSON 写入会被省略，并记录在文件指纹中。
-读取有界：来源前缀最多 2 GiB，单行 64 MiB，来源行数一百万，每行 JSON 结构标记 65,536 个。保留载荷在解码前检查容量；编码后的历史受 Runtime 的 6 MiB／7,500 条记录限制。Claude 另有限制分支索引和回复分片的预算。超限时拒绝，不截断历史。
+JSONL 读取有界：来源前缀最多 2 GiB，单行 64 MiB，行数一百万，每条记录的 JSON 结构标记 65,536 个。OpenCode 快照同时受公共数据库的行数、字节数和工作量限制。保留载荷在解码前检查容量；编码后的历史受 Runtime 的 6 MiB／7,500 条记录限制。Claude 另有限制分支索引和回复分片的预算。超限时拒绝，不截断历史。
返回的 transcript 尚未发布为 Session；发布和重试回执由公共 Session 导入能力负责。
diff --git a/crates/session-import/src/lib.rs b/crates/session-import/src/lib.rs
index b209ca87cb..508bb6fffb 100644
--- a/crates/session-import/src/lib.rs
+++ b/crates/session-import/src/lib.rs
@@ -23,6 +23,7 @@ pub mod catalog;
pub mod claude;
pub mod codex;
mod jsonl;
+pub mod opencode;
mod transcript;
pub use transcript::{Error, Fingerprint, Transcript};
diff --git a/crates/session-import/src/opencode.rs b/crates/session-import/src/opencode.rs
new file mode 100644
index 0000000000..379910a959
--- /dev/null
+++ b/crates/session-import/src/opencode.rs
@@ -0,0 +1,375 @@
+/*
+ * 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.
+ */
+
+//! OpenCode's selected-session snapshot. The database reader must obtain these
+//! rows in one read transaction; conversion never opens an ambient source path.
+mod database;
+mod wire;
+use crate::{
+ Error, Fingerprint, Transcript, jsonl,
+ transcript::{Records, identity, source_cwd, title},
+};
+pub use database::read;
+use maka_runtime::import::{Content, Record, Source};
+use serde::{Deserialize, Serialize};
+use serde_json::{Value, value::RawValue};
+use sha2::{Digest, Sha256};
+use std::collections::{BTreeMap, BTreeSet};
+use wire::{Part as ContentPart, Role, Status};
+
+pub const MAX_RAW_BYTES: u64 = 64 * 1024 * 1024;
+pub const MAX_ROWS: usize = 250_000;
+
+/// Raw SQL cells, not a provider export or a second canonical message model.
+pub struct Snapshot {
+ pub session: Session,
+ pub messages: Vec<Message>,
+ pub parts: Vec<Part>,
+}
+#[derive(Serialize)]
+pub struct Session {
+ pub id: String,
+ pub parent_id: Option<String>,
+ pub directory: Option<String>,
+ pub title: Option<String>,
+ pub revert: Option<Revert>,
+}
+#[derive(Deserialize, Serialize)]
+pub struct Revert {
+ #[serde(rename = "messageID")]
+ pub message_id: String,
+ #[serde(rename = "partID")]
+ pub part_id: Option<String>,
+}
+#[derive(Serialize)]
+pub struct Message {
+ pub id: String,
+ pub created_at: Option<u64>,
+ pub data: Box<RawValue>,
+}
+#[derive(Serialize)]
+pub struct Part {
+ pub id: String,
+ pub message_id: String,
+ pub created_at: Option<u64>,
+ pub data: Box<RawValue>,
+}
+
+pub fn convert(mut snapshot: Snapshot, expected_session_id: &str) -> Result<Transcript, Error> {
+ identity(expected_session_id)?;
+ if snapshot.session.id != expected_session_id {
+ return Err(Error::Invalid(
+ "OpenCode source does not match the selected session",
+ ));
+ }
+ if snapshot
+ .session
+ .parent_id
+ .as_ref()
+ .is_some_and(|id| !id.is_empty())
+ {
+ return Err(Error::Invalid("OpenCode source is a child session"));
+ }
+ if snapshot.messages.len().saturating_add(snapshot.parts.len()) > MAX_ROWS {
+ return Err(Error::Limit {
+ kind: "source_rows",
+ max: MAX_ROWS as u64,
+ });
+ }
+ let mut fingerprint = Evidence::default();
+ fingerprint.record(&snapshot.session)?;
+ let mut message_ids = BTreeSet::new();
+ for message in &snapshot.messages {
+ identity(&message.id)?;
+ timestamp(message.created_at)?;
+ if !message_ids.insert(message.id.as_str()) {
+ return Err(Error::Invalid("duplicate OpenCode message identity"));
+ }
+ fingerprint.record(message)?;
+ }
+ let mut part_ids = BTreeSet::new();
+ for part in &snapshot.parts {
+ identity(&part.id)?;
+ identity(&part.message_id)?;
+ timestamp(part.created_at)?;
+ if !part_ids.insert(part.id.as_str()) || !message_ids.contains(part.message_id.as_str()) {
+ return Err(Error::Invalid("duplicate or orphaned OpenCode part"));
+ }
+ fingerprint.record(part)?;
+ }
+ drop(message_ids);
+ drop(part_ids);
+ snapshot.messages.sort_by(|a, b| {
+ a.created_at
+ .cmp(&b.created_at)
+ .then_with(|| a.id.cmp(&b.id))
+ });
+ snapshot.parts.sort_by(|a, b| {
+ a.message_id
+ .cmp(&b.message_id)
+ .then_with(|| a.id.cmp(&b.id))
+ });
+ if let Some(revert) = &snapshot.session.revert {
+ identity(&revert.message_id)?;
+ let index = snapshot
+ .messages
+ .iter()
+ .position(|message| message.id == revert.message_id)
+ .ok_or(Error::Invalid("OpenCode revert message is missing"))?;
+ if let Some(part_id) = &revert.part_id {
+ identity(part_id)?;
+ let boundary = snapshot
+ .parts
+ .iter()
+ .position(|part| part.id == *part_id && part.message_id == revert.message_id)
+ .ok_or(Error::Invalid("OpenCode revert part is missing"))?;
+ // A partial-message revert retains only parts preceding its boundary.
+ let mut at = 0;
+ snapshot.parts.retain(|part| {
+ let keep = part.message_id != revert.message_id || at < boundary;
+ at += 1;
+ keep
+ });
+ snapshot.messages.truncate(index + 1);
+ } else {
+ snapshot.messages.truncate(index);
+ }
+ }
+ let mut parts = BTreeMap::<&str, Vec<(usize, &Part)>>::new();
+ for (index, part) in snapshot.parts.iter().enumerate() {
+ parts
+ .entry(&part.message_id)
+ .or_default()
+ .push((index, part));
+ }
+ let mut records = Records::default();
+ let mut turn = None;
+ for (message_index, message) in snapshot.messages.iter().enumerate() {
+ let data: wire::Message = jsonl::decode(&message.data, message_index as u64 + 1)?;
+ if matches!(data.role, Role::User) || turn.is_none() {
+ turn = Some(message.id.clone());
+ }
+ let turn = turn.as_ref().expect("message opens a source turn");
+ let ts = data
+ .time
+ .as_ref()
+ .and_then(|time| time.created.as_ref())
+ .and_then(|time| time.millis())
+ .or(message.created_at);
+ if let Some(model) = &data.model {
+ identity(model)?;
+ }
+ for (part_index, part) in parts.remove(message.id.as_str()).unwrap_or_default() {
+ let content: ContentPart = jsonl::decode(&part.data, part_index as u64 + 1)?;
+ let record_id = format!("part:{part_index}");
+ let mut append = |suffix: &str, content| {
+ records.push(Record {
+ source_message_id: format!("{record_id}:{suffix}"),
+ source_turn_id: turn.clone(),
+ timestamp: ts,
+ content,
+ })
+ };
+ match content {
+ ContentPart::Text { text, synthetic } => {
+ if text.is_empty() {
+ continue;
+ }
+ append(
+ "text",
+ match data.role {
+ Role::User if synthetic => Content::Note { text },
+ Role::User => Content::User { text },
+ Role::Assistant => Content::Assistant {
+ text,
+ model: data.model.clone(),
+ thinking: None,
+ },
+ },
+ )?;
+ }
+ ContentPart::Reasoning { text } if matches!(data.role, Role::Assistant) => {
+ if !text.is_empty() {
+ append(
+ "thinking",
+ Content::Assistant {
+ text: String::new(),
+ model: data.model.clone(),
+ thinking: Some(text),
+ },
+ )?;
+ }
+ }
+ ContentPart::Tool {
+ call_id,
+ tool,
+ state,
+ } if matches!(data.role, Role::Assistant) => {
+ identity(&call_id)?;
+ identity(&tool)?;
+ // A source may reuse callID in another message. The durable
+ // part identity, normalized to its snapshot position, owns pairing.
+ append(
+ "call",
+ Content::ToolCall {
+ call_id: record_id.clone(),
+ name: tool,
+ input: state.input,
+ },
+ )?;
+ let output = match state.status {
+ Status::Completed => Some((
+ state
+ .output
+ .ok_or(Error::Invalid("completed OpenCode tool has no output"))?,
+ false,
+ )),
+ Status::Error => Some((
+ state
+ .error
+ .ok_or(Error::Invalid("failed OpenCode tool has no error"))?,
+ true,
+ )),
+ Status::Pending | Status::Running => None,
+ };
+ if let Some((output, is_error)) = output {
+ append(
+ "result",
+ Content::ToolResult {
+ call_id: record_id.clone(),
+ output: Value::String(output),
+ is_error,
+ },
+ )?;
+ }
+ }
+ ContentPart::File { filename, mime } => {
+ let label = filename.as_deref().or(mime.as_deref()).unwrap_or("file");
+ let text = format!("[Attachment: {label}]");
+ append(
+ "attachment",
+ if matches!(data.role, Role::User) {
+ Content::User { text }
+ } else {
+ Content::Note { text }
+ },
+ )?;
+ }
+ ContentPart::Compaction => append(
+ "compaction",
+ Content::Note {
+ text: "Source context was compacted.".into(),
+ },
+ )?,
+ _ => {}
+ }
+ }
+ if let Some(error) = data.error {
+ records.push(Record {
+ source_message_id: format!("message:{message_index}:error"),
+ source_turn_id: turn.clone(),
+ timestamp: ts,
+ content: Content::Note {
+ text: format!("Source error: {}", error.name),
+ },
+ })?;
+ }
+ if let Some(finish) = data.finish {
+ records.push(Record {
+ source_message_id: format!("message:{message_index}:finish"),
+ source_turn_id: turn.clone(),
+ timestamp: ts,
+ content: Content::Note {
+ text: format!("Source finish: {finish}"),
+ },
+ })?;
+ }
+ }
+ let records = records.finish()?;
+ let name = snapshot
+ .session
+ .title
+ .as_deref()
+ .map(title)
+ .filter(|name| !name.is_empty())
+ .or_else(|| {
+ records.iter().find_map(|record| match &record.content {
+ Content::User { text } if !text.trim().is_empty() => Some(title(text)),
+ _ => None,
+ })
+ })
+ .unwrap_or_else(|| expected_session_id.into());
+ Ok(Transcript {
+ source: Source {
+ adapter: "opencode".into(),
+ session_id: expected_session_id.into(),
+ },
+ cwd: source_cwd(snapshot.session.directory),
+ title: name,
+ records,
+ fingerprint: Fingerprint {
+ bytes: fingerprint.bytes,
+ sha256: format!("{:x}", fingerprint.hash.finalize()),
+ incomplete_tail: false,
+ },
+ })
+}
+fn timestamp(value: Option<u64>) -> Result<(), Error> {
+ if value.is_some_and(|value| value > 9_007_199_254_740_991) {
+ Err(Error::Invalid("invalid OpenCode timestamp"))
+ } else {
+ Ok(())
+ }
+}
+
+/// Framed JSON rows retain the exact selected source bytes, without allocating
+/// a second transcript-sized serialization just to calculate its evidence.
+#[derive(Default)]
+struct Evidence {
+ bytes: u64,
+ hash: Sha256,
+}
+impl Evidence {
+ fn record(&mut self, value: &impl Serialize) -> Result<(), Error> {
+ use std::io::Write;
+ serde_json::to_writer(&mut *self, value).map_err(|_| Error::Limit {
+ kind: "source_bytes",
+ max: MAX_RAW_BYTES,
+ })?;
+ self.write_all(b"\n").map_err(|_| Error::Limit {
+ kind: "source_bytes",
+ max: MAX_RAW_BYTES,
+ })
+ }
+}
+impl std::io::Write for Evidence {
+ fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+ if bytes.len() as u64 > MAX_RAW_BYTES - self.bytes {
+ return Err(std::io::Error::other(
+ "source snapshot exceeds its byte limit",
+ ));
+ }
+ self.bytes += bytes.len() as u64;
+ self.hash.update(bytes);
+ Ok(bytes.len())
+ }
+ fn flush(&mut self) -> std::io::Result<()> {
+ Ok(())
+ }
+}
diff --git a/crates/session-import/src/opencode/database.rs b/crates/session-import/src/opencode/database.rs
new file mode 100644
index 0000000000..a2e390c92f
--- /dev/null
+++ b/crates/session-import/src/opencode/database.rs
@@ -0,0 +1,131 @@
+/*
+ * 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::{Message, Part, Session, Snapshot};
+use crate::{Error, Transcript, transcript::identity};
+use maka_plugins::{
+ filesystem::database::{Cell, Query, Read, Table},
+ remote::Views,
+};
+use serde_json::value::RawValue;
+
+/// Reads only the selected session through the caller's public Host-path authority.
+/// All three tables come from one WAL-aware snapshot, with no source mutation.
+pub async fn read(views: &dyn Views, path: String, session_id: &str) -> Result<Transcript, Error> {
+ identity(session_id)?;
+ let tables = views.query_database(Read {
+ path,
+ queries: [
+ "SELECT id,parent_id,directory,title,revert FROM session WHERE id=?1",
+ "SELECT id,time_created,data FROM message WHERE session_id=?1 ORDER BY time_created,id",
+ "SELECT id,message_id,time_created,data FROM part WHERE session_id=?1 ORDER BY message_id,id",
+ ].into_iter().map(|sql| Query {
+ sql: sql.into(), parameters: vec![Cell::Text(session_id.into())],
+ }).collect(),
+ }).await?;
+ super::convert(decode(tables)?, session_id)
+}
+
+fn decode(tables: Vec<Table>) -> Result<Snapshot, Error> {
+ let [session, messages, parts]: [Table; 3] = tables
+ .try_into()
+ .map_err(|_| Error::Invalid("incomplete OpenCode snapshot"))?;
+ let [session]: [Vec<Cell>; 1] = rows(
+ session,
+ &["id", "parent_id", "directory", "title", "revert"],
+ )?
+ .try_into()
+ .map_err(|_| Error::Invalid("OpenCode session missing or ambiguous"))?;
+ let [id, parent_id, directory, title, revert] = fields(session)?;
+ let session = Session {
+ id: text(id)?,
+ parent_id: optional_text(parent_id)?,
+ directory: optional_text(directory)?,
+ title: optional_text(title)?,
+ revert: optional_text(revert)?
+ .map(|value| {
+ serde_json::from_str(&value)
+ .map_err(|_| Error::Invalid("invalid OpenCode revert marker"))
+ })
+ .transpose()?,
+ };
+ let messages = rows(messages, &["id", "time_created", "data"])?
+ .into_iter()
+ .map(|row| {
+ let [id, created_at, data] = fields(row)?;
+ Ok(Message {
+ id: text(id)?,
+ created_at: timestamp(created_at)?,
+ data: json(data)?,
+ })
+ })
+ .collect::<Result<_, Error>>()?;
+ let parts = rows(parts, &["id", "message_id", "time_created", "data"])?
+ .into_iter()
+ .map(|row| {
+ let [id, message_id, created_at, data] = fields(row)?;
+ Ok(Part {
+ id: text(id)?,
+ message_id: text(message_id)?,
+ created_at: timestamp(created_at)?,
+ data: json(data)?,
+ })
+ })
+ .collect::<Result<_, Error>>()?;
+ Ok(Snapshot {
+ session,
+ messages,
+ parts,
+ })
+}
+
+fn rows(table: Table, columns: &[&str]) -> Result<Vec<Vec<Cell>>, Error> {
+ if table.columns != columns {
+ return Err(Error::Invalid("unexpected OpenCode columns"));
+ }
+ Ok(table.rows)
+}
+fn fields<const N: usize>(row: Vec<Cell>) -> Result<[Cell; N], Error> {
+ row.try_into()
+ .map_err(|_| Error::Invalid("invalid OpenCode row width"))
+}
+fn text(cell: Cell) -> Result<String, Error> {
+ match cell {
+ Cell::Text(value) => Ok(value),
+ _ => Err(Error::Invalid("OpenCode field is not text")),
+ }
+}
+fn optional_text(cell: Cell) -> Result<Option<String>, Error> {
+ match cell {
+ Cell::Null => Ok(None),
+ cell => text(cell).map(Some),
+ }
+}
+fn timestamp(cell: Cell) -> Result<Option<u64>, Error> {
+ match cell {
+ Cell::Null => Ok(None),
+ Cell::Integer(value) if (0..=9_007_199_254_740_991).contains(&value) => {
+ Ok(Some(value as u64))
+ }
+ _ => Err(Error::Invalid("invalid OpenCode timestamp")),
+ }
+}
+fn json(cell: Cell) -> Result<Box<RawValue>, Error> {
+ RawValue::from_string(text(cell)?).map_err(|_| Error::Invalid("invalid OpenCode JSON"))
+}
diff --git a/crates/session-import/src/opencode/wire.rs b/crates/session-import/src/opencode/wire.rs
new file mode 100644
index 0000000000..74e2f2bd66
--- /dev/null
+++ b/crates/session-import/src/opencode/wire.rs
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::transcript::Timestamp;
+use serde::Deserialize;
+use serde_json::Value;
+
+#[derive(Deserialize)]
+pub(super) struct Message {
+ pub role: Role,
+ #[serde(rename = "modelID")]
+ pub model: Option<String>,
+ pub time: Option<Time>,
+ pub finish: Option<String>,
+ pub error: Option<Failure>,
+}
+#[derive(Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub(super) enum Role {
+ User,
+ Assistant,
+}
+#[derive(Deserialize)]
+pub(super) struct Time {
+ pub created: Option<Timestamp>,
+}
+#[derive(Deserialize)]
+pub(super) struct Failure {
+ pub name: String,
+}
+#[derive(Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub(super) enum Part {
+ Text {
+ text: String,
+ #[serde(default)]
+ synthetic: bool,
+ },
+ Reasoning {
+ text: String,
+ },
+ Tool {
+ #[serde(rename = "callID")]
+ call_id: String,
+ tool: String,
+ state: State,
+ },
+ File {
+ filename: Option<String>,
+ mime: Option<String>,
+ },
+ Compaction,
+ #[serde(other)]
+ Other,
+}
+#[derive(Deserialize)]
+pub(super) struct State {
+ pub status: Status,
+ pub input: Option<Value>,
+ pub output: Option<String>,
+ pub error: Option<String>,
+}
+#[derive(Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub(super) enum Status {
+ Pending,
+ Running,
+ Completed,
+ Error,
+}
diff --git a/crates/session-import/src/transcript.rs b/crates/session-import/src/transcript.rs
index 33c8a45294..c3e187743f 100644
--- a/crates/session-import/src/transcript.rs
+++ b/crates/session-import/src/transcript.rs
@@ -25,6 +25,8 @@ use serde::{Deserialize, Serialize};
pub enum Error {
#[error(transparent)]
Read(#[from] ReadError),
+ #[error(transparent)]
+ Database(#[from] maka_plugins::filesystem::database::Error),
#[error("invalid source record at line {line}: {source}")]
Decode {
line: u64,
diff --git a/crates/session-import/tests/opencode.rs b/crates/session-import/tests/opencode.rs
new file mode 100644
index 0000000000..2646aa2ab4
--- /dev/null
+++ b/crates/session-import/tests/opencode.rs
@@ -0,0 +1,216 @@
+/*
+ * 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_runtime::import::Content;
+use maka_session_import::{
+ Error,
+ opencode::{self, Message, Part, Session, Snapshot},
+};
+use serde_json::{Value, json, value::to_raw_value};
+
+fn message(id: &str, at: u64, data: Value) -> Message {
+ Message {
+ id: id.into(),
+ created_at: Some(at),
+ data: to_raw_value(&data).unwrap(),
+ }
+}
+fn part(id: &str, message: &str, at: u64, data: Value) -> Part {
+ Part {
+ id: format!("{at:03}-{id}"),
+ message_id: message.into(),
+ created_at: Some(at),
+ data: to_raw_value(&data).unwrap(),
+ }
+}
+fn snapshot() -> Snapshot {
+ Snapshot {
+ session: Session {
+ id: "s".into(),
+ parent_id: None,
+ directory: Some("/source/project".into()),
+ title: None,
+ revert: None,
+ },
+ messages: vec![
+ message(
+ "answer",
+ 20,
+ json!({"role":"assistant","modelID":"source-model","finish":"tool-calls"}),
+ ),
+ message("prompt", 10, json!({"role":"user"})),
+ ],
+ parts: vec![
+ part(
+ "last",
+ "answer",
+ 24,
+ json!({"type":"tool","callID":"reused","tool":"shell","state":{"status":"running","input":{"command":"still pending"}}}),
+ ),
+ part(
+ "prompt-text",
+ "prompt",
+ 10,
+ json!({"type":"text","text":"Read the file"}),
+ ),
+ part(
+ "answer-text",
+ "answer",
+ 20,
+ json!({"type":"text","text":"Before thinking"}),
+ ),
+ part(
+ "answer-thinking",
+ "answer",
+ 21,
+ json!({"type":"reasoning","text":"Then thinking"}),
+ ),
+ part(
+ "first",
+ "answer",
+ 22,
+ json!({"type":"tool","callID":"reused","tool":"Read","state":{"status":"completed","input":{"path":"a"},"output":"one\ntwo\n"}}),
+ ),
+ part(
+ "failure",
+ "answer",
+ 23,
+ json!({"type":"tool","callID":"reused","tool":"Read","state":{"status":"error","input":{"path":"b"},"error":"source failure"}}),
+ ),
+ ],
+ }
+}
+#[test]
+fn opencode_preserves_part_order_tool_pairing_and_missing_results_without_native_execution() {
+ let mut source = snapshot();
+ // Source part IDs, not insertion timestamps, determine displayed order.
+ for part in &mut source.parts {
+ part.created_at = part.created_at.map(|time| 100 - time);
+ }
+ let transcript = opencode::convert(source, "s").unwrap();
+ assert_eq!(transcript.title, "Read the file");
+ assert_eq!(transcript.cwd.as_deref(), Some("/source/project"));
+ assert!(!transcript.fingerprint.incomplete_tail);
+ let records = transcript.records;
+ assert!(
+ records
+ .iter()
+ .all(|record| record.source_turn_id == "prompt")
+ );
+ assert!(matches!(&records[0].content, Content::User { text } if text == "Read the file"));
+ assert!(
+ matches!(&records[1].content, Content::Assistant { text, thinking: None, model: Some(model) } if text == "Before thinking" && model == "source-model")
+ );
+ assert!(
+ matches!(&records[2].content, Content::Assistant { text, thinking: Some(thinking), .. } if text.is_empty() && thinking == "Then thinking")
+ );
+ let calls: Vec<_> = records
+ .iter()
+ .filter_map(|record| match &record.content {
+ Content::ToolCall { call_id, .. } => Some(call_id),
+ _ => None,
+ })
+ .collect();
+ let results: Vec<_> = records
+ .iter()
+ .filter_map(|record| match &record.content {
+ Content::ToolResult {
+ call_id,
+ output,
+ is_error,
+ } => Some((call_id, output, *is_error)),
+ _ => None,
+ })
+ .collect();
+ assert_eq!(calls.len(), 3);
+ assert_eq!(
+ results,
+ [
+ (calls[0], &json!("one\ntwo\n"), false),
+ (calls[1], &json!("source failure"), true)
+ ]
+ );
+ assert_ne!(calls[0], calls[1]);
+ assert_ne!(calls[1], calls[2]);
+ assert!(
+ matches!(&records.last().unwrap().content, Content::Note { text } if text == "Source finish: tool-calls")
+ );
+ let mut partial = snapshot();
+ partial.session.revert = Some(opencode::Revert {
+ message_id: "answer".into(),
+ part_id: Some("022-first".into()),
+ });
+ for part in &mut partial.parts {
+ part.created_at = part.created_at.map(|time| 100 - time);
+ }
+ let partial = opencode::convert(partial, "s").unwrap();
+ assert!(!partial.records.iter().any(|record| matches!(
+ record.content,
+ Content::ToolCall { .. } | Content::ToolResult { .. }
+ )));
+ assert!(
+ partial
+ .records
+ .iter()
+ .any(|record| matches!(&record.content,
+ Content::Assistant { text, .. } if text == "Before thinking"))
+ );
+ let mut reverted = snapshot();
+ reverted.session.revert = Some(opencode::Revert {
+ message_id: "answer".into(),
+ part_id: None,
+ });
+ let reverted = opencode::convert(reverted, "s").unwrap();
+ assert_eq!(reverted.records.len(), 1);
+ assert!(matches!(reverted.records[0].content, Content::User { .. }));
+}
+#[test]
+fn opencode_refuses_ambiguous_or_corrupt_snapshots_before_publishing_any_history() {
+ let mut unknown_revert = snapshot();
+ unknown_revert.session.revert = Some(opencode::Revert {
+ message_id: "missing".into(),
+ part_id: None,
+ });
+ assert!(opencode::convert(unknown_revert, "s").is_err());
+ let mut invalid_role = snapshot();
+ invalid_role.messages[0].data = to_raw_value(&json!({"role":"asssistant"})).unwrap();
+ assert!(opencode::convert(invalid_role, "s").is_err());
+ assert!(opencode::convert(snapshot(), "wrong").is_err());
+ let mut child = snapshot();
+ child.session.parent_id = Some("parent".into());
+ assert!(opencode::convert(child, "s").is_err());
+ let mut orphan = snapshot();
+ orphan.parts[0].message_id = "missing".into();
+ assert!(opencode::convert(orphan, "s").is_err());
+ let mut broken = snapshot();
+ broken.parts[0].data = to_raw_value(
+ &json!({"type":"tool","tool":"Read","callID":"c","state":{"status":"completed"}}),
+ )
+ .unwrap();
+ assert!(opencode::convert(broken, "s").is_err());
+ let mut dense = snapshot();
+ dense.parts[0].data = to_raw_value(&json!({"type":"tool","tool":"Read","callID":"c","state":{"status":"running","input":vec![0;70_000]}})).unwrap();
+ assert!(matches!(
+ opencode::convert(dense, "s"),
+ Err(Error::Limit {
+ kind: "record_complexity",
+ ..
+ })
+ ));
+}
diff --git a/packages/plugin-sdk/README.md b/packages/plugin-sdk/README.md
index e1ebf7ee75..24e3f8a830 100644
--- a/packages/plugin-sdk/README.md
+++ b/packages/plugin-sdk/README.md
@@ -66,6 +66,8 @@ A settled but uncertain effect is recorded as `unknown` and returned explicitly.
Remote Session/workspace views include `files` with the same interface. Reads recheck current credentials and workspace binding, belong to the Remote call's resource settlement, and expire with that callback. A serialized workspace path is only observation, not authority.
+`caller.views.queryDatabase({ path, queries })` reads an explicit trusted Host SQLite path through a `host_paths` Remote endpoint. The batch shares one read transaction, honors WAL and closes before settling. Host state is excluded; current credentials and Session scope still apply. This is not a captured-directory guarantee against malicious same-user path replacement. Results preserve integers as decimal strings and blobs as base64. Queries allow read-oriented builtins and schema inspe [...]
+
`ctx.tools.bind(definitions, capture)` freezes one implementation and optional context for a tool group at each logical model step. The capture callback receives a read-only workspace view; returning `null` omits the group. Returned calls use the frozen closure, not a recaptured implementation. Closing the registration withdraws the whole group. `alwaysVisible` advertises a tool without a prior search.
Capture receives non-secret `model` facts: the selected ID, effective capabilities and supported provider-tool protocol. A binding may return `providerTools: { Research: { id: 'openai.web_search', args: {} } }` for its own registered names. These tools execute inside the primary model request, not through a local handler; a provider-only binding omits `invoke`. Descriptors, local closures and supporting context freeze together through physical retries. Provider tools cannot settle a Host [...]
diff --git a/packages/plugin-sdk/README.zh-CN.md b/packages/plugin-sdk/README.zh-CN.md
index 2fa54a9112..757938f706 100644
--- a/packages/plugin-sdk/README.zh-CN.md
+++ b/packages/plugin-sdk/README.zh-CN.md
@@ -64,6 +64,8 @@ Prompt 回调通过 `call.workspace`、输入准备通过 `request.workspace`
Remote 的 Session／工作区视图通过 `files` 提供相同接口。读取会重新检查当前凭据与工作区绑定，纳入 Remote 调用的资源结算，并随回调结束失效。序列化的工作区路径仅用于观察，不是访问授权。
+`caller.views.queryDatabase({ path, queries })` 通过 `host_paths` Remote 端点读取明确的受信任 Host SQLite 路径。整批查询共用一个读取事务，正确读取 WAL，关闭后才完成结算。Host 数据目录不可访问，当前凭据与 Session 作用域仍有效；不保证抵御同用户恶意并发替换路径。整数以十进制字符串、二进制以 base64 无损传递。只允许读取类内置函数与 schema 查询，不允许修改或加载扩展；上限为 16 条语句、250,000 行、64 MiB 编码结果，工作量与超时采用协作检查，不是 OS 隔离。
+
`ctx.tools.bind(definitions, capture)` 在每个逻辑模型步骤为一组工具冻结实现及可选上下文。capture 回调获得工作区只读视图，返回 `null` 表示本步不提供该组工具。模型返回的调用使用已冻结的闭包，不重新采样实现；关闭注册会一起撤下整组工具。`alwaysVisible` 可让工具无需搜索即对模型可见。
capture 收到不含秘密的 `model`：选定模型 ID、生效的能力和可用的 provider-tool 协议。绑定可为自己注册的名称返回 `providerTools: { Research: { id: 'openai.web_search', args: {} } }`，工具直接由主模型请求执行，不调用本地处理器；仅含此类工具的绑定无需 `invoke`。描述、闭包和上下文一起冻结，物理重试不重新采样。供应商工具不能结束 Host Turn，也不能嵌入 Code Mode。
diff --git a/packages/plugin-sdk/src/database.ts b/packages/plugin-sdk/src/database.ts
new file mode 100644
index 0000000000..dae137692a
--- /dev/null
+++ b/packages/plugin-sdk/src/database.ts
@@ -0,0 +1,44 @@
+/*
+ * 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.
+ */
+
+/** SQL integers are decimal strings and blobs are base64; neither loses precision at the JS boundary. */
+export type DatabaseCell =
+ | { kind: 'null' }
+ | { kind: 'integer'; value: string }
+ | { kind: 'real'; value: number }
+ | { kind: 'text'; value: string }
+ | { kind: 'blob'; value: string };
+export interface DatabaseRead {
+ /** Explicit trusted Host pathname. Requires a host_paths Remote endpoint.
+ * Source data is read-only; normal SQLite WAL/SHM read coordination is allowed.
+ * Not a captured-directory capability against malicious same-user path replacement.
+ */
+ path: string;
+ /** One SELECT per query, one read transaction for the entire batch.
+ * Schema-inspection PRAGMAs are allowed; mutation, ATTACH and extension loading are not.
+ * At most 16 queries, 250,000 rows and 64 MiB encoded result. VM work and a
+ * 30-second deadline are checked cooperatively, not an OS resource sandbox.
+ * Functions are limited to read-oriented scalar, aggregate and JSON inspection builtins.
+ */
+ queries: readonly { sql: string; parameters?: readonly DatabaseCell[] }[];
+}
+export interface DatabaseTable {
+ columns: readonly string[];
+ rows: readonly (readonly DatabaseCell[])[];
+}
diff --git a/packages/plugin-sdk/src/host.ts b/packages/plugin-sdk/src/host.ts
index e23273a6d5..42ee629561 100644
--- a/packages/plugin-sdk/src/host.ts
+++ b/packages/plugin-sdk/src/host.ts
@@ -31,6 +31,7 @@ export type * from './terminal.js';
export type * from './credentials.js';
export type * from './http.js';
export type * from './filesystem.js';
+export type * from './database.js';
export type * from './llm.js';
export type * from './models.js';
export type * from './providers.js';
@@ -81,6 +82,9 @@ export interface RemoteCaller {
): Promise<T>;
session(): Promise<SessionView>;
workspace(input: WorkspaceViewInput): Promise<SessionView>;
+ queryDatabase(
+ input: import('./database.js').DatabaseRead,
+ ): Promise<readonly import('./database.js').DatabaseTable[]>;
};
}
export interface WorkspaceViewInput {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org