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 11:00:32 UTC
(maka) branch feat/runtime-host-rust updated: feat(event-log): reclaim retired conversation material without losing accounting
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 8aa3667fa0 feat(event-log): reclaim retired conversation material without losing accounting
8aa3667fa0 is described below
commit 8aa3667fa03fe93b2378c386be0ab2d2eb75c492
Author: Wang <m4...@gmail.com>
AuthorDate: Wed Sep 23 18:57:33 2026 +0800
feat(event-log): reclaim retired conversation material without losing accounting
Generated-by: Codex
---
crates/event-log/migrations/0001_runtime.sql | 88 +++++----
crates/event-log/src/lib.rs | 2 +
crates/event-log/src/prefix.rs | 5 +
crates/event-log/src/schema.rs | 7 +
crates/event-log/src/sessions.rs | 2 +
crates/event-log/src/sessions/material.rs | 186 ++++++++++++++++++
crates/event-log/tests/archive.rs | 4 +
crates/event-log/tests/session_copy.rs | 5 +
crates/event-log/tests/session_material.rs | 223 ++++++++++++++++++++++
crates/event-log/tests/tool_usage.rs | 39 ++++
crates/runtime-host/src/execution/plugins/root.rs | 1 +
crates/runtime-host/src/execution/removal.rs | 17 ++
crates/runtime-host/src/server/sessions/copy.rs | 1 +
docs/rust-parity.md | 2 +-
docs/rust-parity.zh-CN.md | 2 +-
docs/rust-runtime.md | 8 +-
docs/rust-runtime.zh-CN.md | 6 +-
17 files changed, 560 insertions(+), 38 deletions(-)
diff --git a/crates/event-log/migrations/0001_runtime.sql b/crates/event-log/migrations/0001_runtime.sql
index 59f0e88898..82f189fb33 100644
--- a/crates/event-log/migrations/0001_runtime.sql
+++ b/crates/event-log/migrations/0001_runtime.sql
@@ -31,37 +31,45 @@ CREATE TABLE event_log (
CASE WHEN kind IN ('model_requested', 'model_completed', 'model_interrupted',
'invocation_ended', 'auxiliary_model_started', 'auxiliary_model_settled',
'tool_dispatched', 'tool_settled', 'tool_rejected')
- THEN COALESCE(json_extract(event_json, '$.recorded_at.secs_since_epoch'),
- json_extract(event_json, '$.started_at.secs_since_epoch'),
- json_extract(event_json, '$.completed_at.secs_since_epoch')) * 1000.0
- + COALESCE(json_extract(event_json, '$.recorded_at.nanos_since_epoch'),
- json_extract(event_json, '$.started_at.nanos_since_epoch'),
- json_extract(event_json, '$.completed_at.nanos_since_epoch')) / 1000000.0
+ THEN COALESCE(json_extract(COALESCE(event_json, retained_json), '$.recorded_at.secs_since_epoch'),
+ json_extract(COALESCE(event_json, retained_json), '$.started_at.secs_since_epoch'),
+ json_extract(COALESCE(event_json, retained_json), '$.completed_at.secs_since_epoch')) * 1000.0
+ + COALESCE(json_extract(COALESCE(event_json, retained_json), '$.recorded_at.nanos_since_epoch'),
+ json_extract(COALESCE(event_json, retained_json), '$.started_at.nanos_since_epoch'),
+ json_extract(COALESCE(event_json, retained_json), '$.completed_at.nanos_since_epoch')) / 1000000.0
END
) STORED,
accounting_usage TEXT GENERATED ALWAYS AS (
CASE
- WHEN kind = 'model_completed' THEN json_extract(event_json, '$.fact.output.usage')
- WHEN kind = 'model_observed' AND json_extract(event_json, '$.fact.event.kind') = 'finished'
- THEN json_extract(event_json, '$.fact.event.data.usage')
- WHEN kind = 'auxiliary_model_usage' THEN json_extract(event_json, '$.usage')
+ WHEN kind = 'model_completed' THEN json_extract(COALESCE(event_json, retained_json), '$.fact.output.usage')
+ WHEN kind = 'model_observed' AND json_extract(COALESCE(event_json, retained_json), '$.fact.event.kind') = 'finished'
+ THEN json_extract(COALESCE(event_json, retained_json), '$.fact.event.data.usage')
+ WHEN kind = 'auxiliary_model_usage' THEN json_extract(COALESCE(event_json, retained_json), '$.usage')
END
) STORED,
invocation_model TEXT GENERATED ALWAYS AS (
- CASE WHEN kind = 'invocation_opened' THEN json_extract(event_json, '$.fact.configuration.model') END
+ CASE WHEN kind = 'invocation_opened' THEN json_extract(COALESCE(event_json, retained_json), '$.fact.configuration.model') END
) STORED,
accounting_tool TEXT GENERATED ALWAYS AS (
CASE
WHEN kind IN ('tool_dispatched', 'tool_rejected') THEN json_object(
- 'invocation', json_extract(event_json, '$.invocation'),
- 'call', json_extract(event_json, '$.fact.call'),
- 'name', json_extract(event_json, '$.fact.name'),
- 'rejection', json_extract(event_json, '$.fact.reason.kind'))
+ 'invocation', json_extract(COALESCE(event_json, retained_json), '$.invocation'),
+ 'call', json_extract(COALESCE(event_json, retained_json), '$.fact.call'),
+ 'name', json_extract(COALESCE(event_json, retained_json), '$.fact.name'),
+ 'rejection', json_extract(COALESCE(event_json, retained_json), '$.fact.reason.kind'))
WHEN kind = 'tool_settled' THEN json_object(
- 'outcome', json_extract(event_json, '$.fact.outcome.kind'))
+ 'outcome', json_extract(COALESCE(event_json, retained_json), '$.fact.outcome.kind'))
END
) STORED,
- event_json TEXT NOT NULL,
+ event_json TEXT,
+ -- A collected body is absent, never a synthetic replayable RuntimeEvent.
+ retained_json TEXT CHECK(retained_json IS NULL OR json_valid(retained_json)),
+ body_digest TEXT,
+ event_session TEXT GENERATED ALWAYS AS (
+ json_extract(COALESCE(event_json, retained_json), '$.invocation.session_id')
+ ) STORED,
+ CHECK((event_json IS NOT NULL AND retained_json IS NULL AND body_digest IS NULL)
+ OR (event_json IS NULL AND retained_json IS NOT NULL AND body_digest IS NOT NULL)),
CHECK(invocation_id IS NOT NULL OR operation_id IS NULL)
);
@@ -254,7 +262,18 @@ CREATE TABLE "tool_result_payloads" (
);
CREATE VIEW runtime_events AS
- SELECT * FROM event_log WHERE invocation_id IS NOT NULL;
+ SELECT * FROM event_log WHERE invocation_id IS NOT NULL AND event_json IS NOT NULL;
+
+-- Accounting can read compact retained proof, never conversation history.
+CREATE VIEW accounting_events AS
+ SELECT sequence, event_id, invocation_id, kind, operation_id,
+ accounting_at, accounting_usage, invocation_model, accounting_tool,
+ COALESCE(event_json, retained_json) AS event_json
+ FROM event_log;
+
+CREATE INDEX event_material_owner ON event_log(event_session, sequence)
+ WHERE event_json IS NOT NULL;
+CREATE INDEX event_proof_owner ON event_log(event_session, sequence);
-- Copies retain original facts, never execution authority. Flatten membership
-- at creation so descendants do not query mutable parent Session metadata.
@@ -290,6 +309,11 @@ CREATE TABLE session_revision_sources (
PRIMARY KEY(session_id, sequence)
);
+CREATE INDEX history_source_owners ON session_history_members(sequence, session_id);
+CREATE INDEX archive_source_owners ON session_history_members(archive_sequence, session_id)
+ WHERE archive_sequence IS NOT NULL;
+CREATE INDEX revision_source_owners ON session_revision_sources(sequence, session_id);
+
CREATE VIEW session_history_events AS
SELECT json_extract(event_json, '$.invocation.session_id') AS owner_session_id,
0 AS inherited, NULL AS archive_sequence, e.* FROM runtime_events e
@@ -343,8 +367,8 @@ CREATE TABLE model_valuations (
);
CREATE INDEX model_usage_observation ON event_log(
- invocation_id, json_extract(event_json, '$.fact.step_id'), sequence DESC
-) WHERE kind = 'model_observed' AND json_extract(event_json, '$.fact.event.kind') = 'finished';
+ invocation_id, json_extract(COALESCE(event_json, retained_json), '$.fact.step_id'), sequence DESC
+) WHERE kind = 'model_observed' AND json_extract(COALESCE(event_json, retained_json), '$.fact.event.kind') = 'finished';
-- Count physical admissions, not history memberships or accepted responses.
-- A rejected response can still have provider-reported usage. No mutable
@@ -370,19 +394,19 @@ WITH attempts AS (
SELECT observed.accounting_usage
FROM event_log observed INDEXED BY model_usage_observation
WHERE observed.kind = 'model_observed'
- AND json_extract(observed.event_json, '$.fact.event.kind') = 'finished'
+ AND json_extract(COALESCE(observed.event_json, observed.retained_json), '$.fact.event.kind') = 'finished'
AND observed.invocation_id = request.invocation_id
- AND json_extract(observed.event_json, '$.fact.step_id') = request.operation_id
+ AND json_extract(COALESCE(observed.event_json, observed.retained_json), '$.fact.step_id') = request.operation_id
ORDER BY observed.sequence DESC LIMIT 1
), '{}') AS usage
- FROM runtime_events request
- JOIN runtime_events opening ON opening.invocation_id = request.invocation_id
+ FROM accounting_events request
+ JOIN accounting_events opening ON opening.invocation_id = request.invocation_id
AND opening.kind = 'invocation_opened'
- LEFT JOIN runtime_events completed ON completed.operation_id = request.operation_id
+ LEFT JOIN accounting_events completed ON completed.operation_id = request.operation_id
AND completed.kind = 'model_completed'
- LEFT JOIN runtime_events interrupted ON interrupted.operation_id = request.operation_id
+ LEFT JOIN accounting_events interrupted ON interrupted.operation_id = request.operation_id
AND interrupted.kind = 'model_interrupted'
- LEFT JOIN runtime_events terminal ON terminal.invocation_id = request.invocation_id
+ LEFT JOIN accounting_events terminal ON terminal.invocation_id = request.invocation_id
AND terminal.kind = 'invocation_ended'
WHERE request.kind = 'model_requested'
)
@@ -414,7 +438,7 @@ FROM event_log started
LEFT JOIN event_log settled ON settled.kind = 'auxiliary_model_settled'
AND json_extract(settled.event_json, '$.request_id') = started.event_id
LEFT JOIN event_log observed ON observed.kind = 'auxiliary_model_usage'
- AND json_extract(observed.event_json, '$.request_id') = started.event_id
+ AND json_extract(COALESCE(observed.event_json, observed.retained_json), '$.request_id') = started.event_id
WHERE started.kind = 'auxiliary_model_started';
CREATE VIEW model_usage AS
@@ -444,12 +468,12 @@ SELECT request.sequence, request.event_id,
'outcome', CASE json_extract(settled.accounting_tool, '$.outcome')
WHEN 'succeeded' THEN 'success' WHEN 'failed' THEN 'error' ELSE 'unknown' END)
END AS result
-FROM runtime_events request
-JOIN runtime_events opening ON opening.invocation_id = request.invocation_id
+FROM accounting_events request
+JOIN accounting_events opening ON opening.invocation_id = request.invocation_id
AND opening.kind = 'invocation_opened'
-LEFT JOIN runtime_events settled ON settled.invocation_id = request.invocation_id
+LEFT JOIN accounting_events settled ON settled.invocation_id = request.invocation_id
AND settled.operation_id = request.operation_id AND settled.kind = 'tool_settled'
-LEFT JOIN runtime_events terminal ON terminal.invocation_id = request.invocation_id
+LEFT JOIN accounting_events terminal ON terminal.invocation_id = request.invocation_id
AND terminal.kind = 'invocation_ended'
WHERE request.kind IN ('tool_dispatched', 'tool_rejected');
diff --git a/crates/event-log/src/lib.rs b/crates/event-log/src/lib.rs
index 19b72162d2..65f2254b5d 100644
--- a/crates/event-log/src/lib.rs
+++ b/crates/event-log/src/lib.rs
@@ -100,6 +100,8 @@ pub enum StoreError {
InvalidTransition(String),
#[error("prefix exceeds the requested event or byte limit")]
PrefixTooLarge,
+ #[error("conversation material in this scope was collected after Session removal")]
+ MaterialCollected,
#[error("derived transcript identity conflicts with its committed bytes")]
TranscriptConflict,
#[error(transparent)]
diff --git a/crates/event-log/src/prefix.rs b/crates/event-log/src/prefix.rs
index d30f95d4b2..f5be7642d3 100644
--- a/crates/event-log/src/prefix.rs
+++ b/crates/event-log/src/prefix.rs
@@ -49,6 +49,11 @@ impl EventLog {
self.connection.run(move |connection| Box::pin(async move {
let mut transaction = connection.begin().await?;
let selection = Selection::resolve(&mut transaction, &scope).await?;
+ let collected: bool = sqlx::query_scalar(
+ "SELECT EXISTS(SELECT 1 FROM event_log WHERE invocation_id IS NOT NULL
+ AND event_json IS NULL AND (?1 IS NULL OR event_session=?1))"
+ ).bind(&selection.session).fetch_one(&mut *transaction).await?;
+ if collected { return Err(StoreError::MaterialCollected); }
let lineage = Selection::predicate("runtime_events", "?2");
let source = if selection.session.is_some() { "session_history_events" } else { "runtime_events" };
let session_filter = if selection.session.is_some() {
diff --git a/crates/event-log/src/schema.rs b/crates/event-log/src/schema.rs
index 69f27ab066..e4998c193d 100644
--- a/crates/event-log/src/schema.rs
+++ b/crates/event-log/src/schema.rs
@@ -60,6 +60,13 @@ pub(crate) async fn initialize_connection(
return Err(StoreError::UnsupportedDatabase);
}
}
+ // A fresh file needs pointer-map pages before SQLx creates its first table.
+ // Reclamation can then return space in small steps without a full VACUUM.
+ if tables == 0 {
+ sqlx::query("PRAGMA auto_vacuum = INCREMENTAL")
+ .execute(&mut *connection)
+ .await?;
+ }
// Establish Rust identity before SQLx creates its migration ledger, so a
// crash before migration 1 is distinguishable from an unrelated database.
if application_id == 0 {
diff --git a/crates/event-log/src/sessions.rs b/crates/event-log/src/sessions.rs
index 6bbde2cd48..ca7a64a0e1 100644
--- a/crates/event-log/src/sessions.rs
+++ b/crates/event-log/src/sessions.rs
@@ -25,6 +25,8 @@ pub(crate) use admission::retain;
pub(crate) mod copy;
pub use copy::{AbandonRevision, SessionCopy, SessionCopyResult};
mod execution;
+mod material;
+pub use material::MaterialCollection;
mod metadata;
mod origin;
mod processes;
diff --git a/crates/event-log/src/sessions/material.rs b/crates/event-log/src/sessions/material.rs
new file mode 100644
index 0000000000..bc457a0c2f
--- /dev/null
+++ b/crates/event-log/src/sessions/material.rs
@@ -0,0 +1,186 @@
+/*
+ * 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::{EventLog, StoreError};
+use maka_runtime::tool_call::{RejectionKind, ToolRejection};
+use maka_runtime::{
+ artifact::content_digest,
+ event::{Fact, InvocationOutcome, RuntimeEvent, ToolOutcome},
+ model::ModelEvent,
+};
+use serde_json::{Value, json};
+use sqlx::Connection;
+
+// Walk only reverse references of one retired Session, including archive writers.
+// UNION deduplicates diamonds; active owners short-circuit the search.
+const RETAINED: &str = "
+WITH RECURSIVE owners(id) AS (
+ VALUES(?1)
+ UNION
+ SELECT h.session_id FROM owners o
+ JOIN event_log e ON e.event_session=o.id
+ JOIN session_history_members h ON h.sequence=e.sequence
+ UNION
+ SELECT h.session_id FROM owners o
+ JOIN event_log e ON e.event_session=o.id
+ JOIN session_history_members h ON h.archive_sequence=e.sequence
+ UNION
+ SELECT h.session_id FROM owners o
+ JOIN event_log e ON e.event_session=o.id
+ JOIN session_revision_sources h ON h.sequence=e.sequence
+)
+SELECT EXISTS(SELECT 1 FROM owners JOIN session_control ON session_control.id=owners.id)";
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum MaterialCollection {
+ Done,
+ Retained(String),
+ Collected,
+}
+
+impl EventLog {
+ /// One Session and at most eight bodies per writer job. Advance the scan
+ /// past Retained Sessions; retry the same cursor after Collected.
+ /// Proof, payload retirement and body removal share the canonical writer.
+ pub async fn collect_session_material(
+ &self,
+ after: Option<&str>,
+ ) -> Result<MaterialCollection, StoreError> {
+ self.validate_root()?;
+ let after = after.map(str::to_owned);
+ self.connection.run(move |connection| Box::pin(async move {
+ let mut tx = connection.begin_with("BEGIN IMMEDIATE").await?;
+ let session: Option<String> = sqlx::query_scalar(
+ "SELECT session_id FROM session_retirements r WHERE remove_session=1 AND completed=1
+ AND (?1 IS NULL OR session_id>?1) AND EXISTS (SELECT 1 FROM event_log e
+ WHERE e.event_session=r.session_id AND e.event_json IS NOT NULL)
+ ORDER BY session_id LIMIT 1"
+ ).bind(after).fetch_optional(&mut *tx).await?;
+ let Some(session) = session else {
+ tx.rollback().await?;
+ let free: i64 = sqlx::query_scalar("PRAGMA freelist_count").fetch_one(&mut *connection).await?;
+ if free == 0 { return Ok(MaterialCollection::Done); }
+ let incremental: i64 = sqlx::query_scalar("PRAGMA auto_vacuum").fetch_one(&mut *connection).await?;
+ if incremental != 2 { return Err(StoreError::UnsupportedDatabase); }
+ sqlx::query("PRAGMA incremental_vacuum(128)").execute(connection).await?;
+ return Ok(MaterialCollection::Collected);
+ };
+ // A previously collected body proves this owner set was already
+ // released. All owners were removed, so none can publish new copies.
+ let released: bool = sqlx::query_scalar(
+ "SELECT EXISTS(SELECT 1 FROM event_log WHERE event_session=? AND event_json IS NULL)"
+ ).bind(&session).fetch_one(&mut *tx).await?;
+ if !released && sqlx::query_scalar::<_, bool>(RETAINED)
+ .bind(&session).fetch_one(&mut *tx).await?
+ {
+ return Ok(MaterialCollection::Retained(session));
+ }
+ let rows: Vec<(String, String)> = sqlx::query_as(
+ "SELECT event_id, event_json FROM event_log WHERE event_session=?
+ AND event_json IS NOT NULL ORDER BY sequence LIMIT 8"
+ ).bind(session).fetch_all(&mut *tx).await?;
+ for (id, body) in rows {
+ let event: RuntimeEvent = serde_json::from_str(&body)?;
+ let proof = retained(&event);
+ sqlx::query("DELETE FROM tool_result_payloads WHERE event_id=?")
+ .bind(&id).execute(&mut *tx).await?;
+ let surface: Option<String> = sqlx::query_scalar(
+ "DELETE FROM model_request_compositions WHERE event_id=? RETURNING digest"
+ ).bind(&id).fetch_optional(&mut *tx).await?;
+ if let Some(surface) = surface {
+ sqlx::query("DELETE FROM request_compositions WHERE digest=?1
+ AND NOT EXISTS (SELECT 1 FROM model_request_compositions WHERE digest=?1)")
+ .bind(surface).execute(&mut *tx).await?;
+ }
+ sqlx::query("UPDATE event_log SET retained_json=?, body_digest=?, event_json=NULL WHERE event_id=?")
+ .bind(serde_json::to_string(&proof)?).bind(content_digest(body.as_bytes()))
+ .bind(&id).execute(&mut *tx).await?;
+ }
+ tx.commit().await.map_err(StoreError::CommitUnknown)?;
+ Ok(MaterialCollection::Collected)
+ })).await
+ }
+}
+
+/// Minimal original fields used by accounting and terminal proof. This is not
+/// a RuntimeEvent and must never enter replay, input projection or recovery.
+fn retained(event: &RuntimeEvent) -> Value {
+ let fact = match &event.fact {
+ Fact::InvocationOpened { configuration, .. } => json!({
+ "configuration": configuration.as_ref().map(|config| json!({"model": config.model}))
+ }),
+ Fact::ModelRequested {
+ step_id,
+ model_id,
+ purpose,
+ ..
+ } => json!({"step_id":step_id,"model_id":model_id,"purpose":purpose}),
+ Fact::ModelCompleted { step_id, output } => {
+ json!({"step_id":step_id,"output":{"usage":output.usage}})
+ }
+ Fact::ModelInterrupted { step_id, status } => json!({"step_id":step_id,"status":status}),
+ Fact::ModelObserved {
+ step_id,
+ event: ModelEvent::Finished { usage, .. },
+ } => json!({"step_id":step_id,"event":{"kind":"finished","data":{"usage":usage}}}),
+ Fact::ToolDispatched { call, name, .. } => json!({"call":call,"name":name}),
+ Fact::ToolRejected {
+ call, name, reason, ..
+ } => {
+ let kind = match reason {
+ ToolRejection::Unavailable => RejectionKind::Unavailable,
+ ToolRejection::InvalidInput { .. } => RejectionKind::InvalidInput,
+ ToolRejection::PolicyDenied { .. } => RejectionKind::PolicyDenied,
+ ToolRejection::PreparationFailed { .. } => RejectionKind::PreparationFailed,
+ ToolRejection::ExclusiveConflict => RejectionKind::ExclusiveConflict,
+ ToolRejection::Cancelled => RejectionKind::Cancelled,
+ };
+ json!({"call":call,"name":name,"reason":{"kind":kind}})
+ }
+ Fact::ToolSettled { outcome, .. } => json!({"outcome":{"kind":match outcome {
+ ToolOutcome::Succeeded { .. } => "succeeded",
+ ToolOutcome::Failed { .. } => "failed",
+ ToolOutcome::Unknown { .. } => "unknown",
+ }}}),
+ Fact::InvocationEnded { outcome } => {
+ let outcome = match outcome {
+ InvocationOutcome::Completed => json!({"kind":"completed"}),
+ InvocationOutcome::Failed { class, .. } => json!({"kind":"failed","class":class}),
+ InvocationOutcome::Cancelled { source } => {
+ json!({"kind":"cancelled","source":source})
+ }
+ InvocationOutcome::HandoffPaused { pause } => {
+ json!({"kind":"handoff_paused","pause":{"intent":pause.intent}})
+ }
+ InvocationOutcome::ContextCompactFinished { outcome } => {
+ json!({"kind":"context_compact_finished","outcome":outcome})
+ }
+ };
+ json!({"outcome":outcome})
+ }
+ Fact::ExecutorStarted { .. }
+ | Fact::ExecutorObserved { .. }
+ | Fact::ExecutorCompleted { .. }
+ | Fact::MessageSteered { .. }
+ | Fact::ContextCheckpointRecorded { .. }
+ | Fact::ToolResultArchived { .. }
+ | Fact::ModelObserved { .. } => json!({}),
+ };
+ json!({"id":event.id,"invocation":event.invocation,"recorded_at":event.recorded_at,"fact":fact})
+}
diff --git a/crates/event-log/tests/archive.rs b/crates/event-log/tests/archive.rs
index 25709b694a..95c1e12459 100644
--- a/crates/event-log/tests/archive.rs
+++ b/crates/event-log/tests/archive.rs
@@ -784,6 +784,10 @@ async fn inherited_results_prune_and_compact_without_changing_siblings_or_frozen
log.finish_session_retirement("branch").await.unwrap(),
maka_event_log::sessions::SessionRetirement::Removed
);
+ assert_eq!(
+ log.collect_session_material(None).await.unwrap(),
+ maka_event_log::sessions::MaterialCollection::Retained("branch".into())
+ );
assert_eq!(
log.read_model_context("nested", None, 100, 65536)
.await
diff --git a/crates/event-log/tests/session_copy.rs b/crates/event-log/tests/session_copy.rs
index af1e298244..12fbc17bee 100644
--- a/crates/event-log/tests/session_copy.rs
+++ b/crates/event-log/tests/session_copy.rs
@@ -565,6 +565,11 @@ async fn copies_own_history_and_files_without_replaying_execution_across_retries
log.finish_session_retirement("source").await.unwrap(),
maka_event_log::sessions::SessionRetirement::Removed
);
+ assert_eq!(
+ log.collect_session_material(None).await.unwrap(),
+ maka_event_log::sessions::MaterialCollection::Retained("source".into()),
+ "live descendants retain source proof"
+ );
log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
assert_eq!(
diff --git a/crates/event-log/tests/session_material.rs b/crates/event-log/tests/session_material.rs
new file mode 100644
index 0000000000..b27096266e
--- /dev/null
+++ b/crates/event-log/tests/session_material.rs
@@ -0,0 +1,223 @@
+/*
+ * 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::{SessionCopy, SessionRetirement},
+};
+use maka_runtime::{
+ event::{Fact, InvocationInput, InvocationOutcome, LogScope},
+ model::{ModelEvent, ModelFinishReason, ModelUsage},
+ session::CopyPurpose,
+};
+use serde_json::{Value, json};
+
+#[path = "usage/fixture.rs"]
+mod fixture;
+use fixture::{event, query, request};
+
+#[tokio::test]
+async fn collection_waits_for_the_last_history_owner_and_preserves_frozen_accounting() {
+ let directory = tempfile::tempdir().unwrap();
+ let path = directory.path().join("material.sqlite");
+ let log = EventLog::open(&path).await.unwrap();
+ log.create_session("source", "create", &json!({}), 1)
+ .await
+ .unwrap();
+ let opening = event(
+ Fact::InvocationOpened {
+ configuration: None,
+ input: InvocationInput::Message {
+ content: "private-question".repeat(10000).into(),
+ source_messages: vec![],
+ request_fingerprint: None,
+ },
+ },
+ 0,
+ );
+ log.append(&opening).await.unwrap();
+ log.append(&request("observed", 1000)).await.unwrap();
+ let mut frozen = query();
+ frozen.through = Some(log.model_attempts(query(), 0, 100).await.unwrap().through);
+ for i in 0..12 {
+ log.append(&event(
+ Fact::ModelObserved {
+ step_id: "observed".into(),
+ event: ModelEvent::PartDelta {
+ id: "part".into(),
+ text: "private-response".repeat(1000),
+ provider_options: None,
+ },
+ },
+ 1100 + i,
+ ))
+ .await
+ .unwrap();
+ }
+ log.append(&event(
+ Fact::ModelObserved {
+ step_id: "observed".into(),
+ event: ModelEvent::Finished {
+ reason: ModelFinishReason::Stop,
+ usage: ModelUsage {
+ input_tokens: Some(100),
+ output_tokens: Some(12),
+ ..Default::default()
+ },
+ provider_options: None,
+ },
+ },
+ 1200,
+ ))
+ .await
+ .unwrap();
+ log.append(&event(
+ Fact::ModelInterrupted {
+ step_id: "observed".into(),
+ status: maka_runtime::event::ModelInterruption::Failed,
+ },
+ 1250,
+ ))
+ .await
+ .unwrap();
+ // Provider usage is factual even when no response was accepted.
+ log.append(&event(
+ Fact::InvocationEnded {
+ outcome: InvocationOutcome::Failed {
+ class: "connection_lost".into(),
+ message: None,
+ },
+ },
+ 1300,
+ ))
+ .await
+ .unwrap();
+ let before = log.usage_summary(query()).await.unwrap().summary;
+ let frozen_before = log.usage_summary(frozen.clone()).await.unwrap().summary;
+ let revision = log
+ .get_session::<Value>("source")
+ .await
+ .unwrap()
+ .unwrap()
+ .revision;
+ log.copy_session(
+ SessionCopy {
+ source_session_id: "source".into(),
+ target_session_id: "copy".into(),
+ expected_source_revision: revision,
+ purpose: CopyPurpose::Branch {
+ turn_id: None,
+ side_conversation: false,
+ },
+ },
+ &json!({}),
+ 2,
+ )
+ .await
+ .unwrap();
+ let inherited = log
+ .scoped_prefix(LogScope::Session { id: "copy".into() }, 100, 1_000_000)
+ .await
+ .unwrap()
+ .digest;
+ retire(&log, "source").await;
+ assert_eq!(
+ log.collect_session_material(None).await.unwrap(),
+ maka_event_log::sessions::MaterialCollection::Retained("source".into())
+ );
+ assert_eq!(
+ log.scoped_prefix(LogScope::Session { id: "copy".into() }, 100, 1_000_000)
+ .await
+ .unwrap()
+ .digest,
+ inherited
+ );
+ retire(&log, "copy").await;
+ let inspect = rusqlite::Connection::open(&path).unwrap();
+ let pages_before: i64 = inspect
+ .query_row("PRAGMA page_count", [], |row| row.get(0))
+ .unwrap();
+ drop(inspect);
+ assert_eq!(
+ log.collect_session_material(None).await.unwrap(),
+ maka_event_log::sessions::MaterialCollection::Collected
+ );
+ assert_eq!(
+ log.usage_summary(query()).await.unwrap().summary,
+ before,
+ "partial collection keeps accounting exact"
+ );
+ log.close().await.unwrap();
+ let log = EventLog::open(&path).await.unwrap();
+ while log.collect_session_material(None).await.unwrap()
+ == maka_event_log::sessions::MaterialCollection::Collected
+ {}
+ assert_eq!(log.usage_summary(query()).await.unwrap().summary, before);
+ assert_eq!(
+ log.usage_summary(frozen).await.unwrap().summary,
+ frozen_before
+ );
+ assert!(matches!(
+ log.prefix(100, 1_000_000).await,
+ Err(StoreError::MaterialCollected)
+ ));
+ assert!(log.unfinished_invocations(10).await.unwrap().is_empty());
+ assert!(
+ log.append(&opening).await.is_err(),
+ "collection cannot resurrect a retired invocation"
+ );
+ assert!(log.session_copy_receipt("copy").await.unwrap().is_some());
+ let inspect = rusqlite::Connection::open(&path).unwrap();
+ let (bodies, leaked, digest): (i64, i64, String) = inspect
+ .query_row(
+ "SELECT (SELECT COUNT(*) FROM event_log WHERE event_json IS NOT NULL),
+ (SELECT COUNT(*) FROM event_log WHERE retained_json LIKE '%private-%'),
+ (SELECT body_digest FROM event_log WHERE event_id=?1)",
+ [&opening.event().id],
+ |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
+ )
+ .unwrap();
+ assert_eq!((bodies, leaked), (0, 0));
+ let pages_after: i64 = inspect
+ .query_row("PRAGMA page_count", [], |row| row.get(0))
+ .unwrap();
+ assert!(
+ pages_after < pages_before,
+ "collection returns database pages, not only reusable slots"
+ );
+ assert_eq!(
+ digest,
+ maka_runtime::artifact::content_digest(&serde_json::to_vec(opening.event()).unwrap())
+ );
+ log.close().await.unwrap();
+}
+
+async fn retire(log: &EventLog, session: &str) {
+ let revision = log
+ .get_session::<Value>(session)
+ .await
+ .unwrap()
+ .unwrap()
+ .revision;
+ log.begin_session_removal(session, revision).await.unwrap();
+ assert_eq!(
+ log.finish_session_retirement(session).await.unwrap(),
+ SessionRetirement::Removed
+ );
+}
diff --git a/crates/event-log/tests/tool_usage.rs b/crates/event-log/tests/tool_usage.rs
index 73193a5cfd..40f3f0fe44 100644
--- a/crates/event-log/tests/tool_usage.rs
+++ b/crates/event-log/tests/tool_usage.rs
@@ -35,6 +35,9 @@ async fn tool_accounting_preserves_dispatch_truth_and_snapshot_without_private_b
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("usage.sqlite");
let log = EventLog::open(&path).await.unwrap();
+ log.create_session("source", "source", &json!({}), 1)
+ .await
+ .unwrap();
log.append(&event(
Fact::InvocationOpened {
configuration: None,
@@ -306,6 +309,42 @@ async fn tool_accounting_preserves_dispatch_truth_and_snapshot_without_private_b
);
log.close().await.unwrap();
let log = EventLog::open(&path).await.unwrap();
+ let revision = log
+ .get_session::<serde_json::Value>("source")
+ .await
+ .unwrap()
+ .unwrap()
+ .revision;
+ log.begin_session_removal("source", revision).await.unwrap();
+ log.finish_session_retirement("source").await.unwrap();
+ assert_eq!(
+ log.collect_session_material(None).await.unwrap(),
+ maka_event_log::sessions::MaterialCollection::Collected
+ );
+ log.close().await.unwrap();
+ let log = EventLog::open(&path).await.unwrap();
+ while log.collect_session_material(None).await.unwrap()
+ == maka_event_log::sessions::MaterialCollection::Collected
+ {}
+ assert!(matches!(
+ log.prefix(100, 1_000_000).await,
+ Err(maka_event_log::StoreError::MaterialCollected)
+ ));
+ let inspect = rusqlite::Connection::open(&path).unwrap();
+ let (bodies, payloads, leaked): (i64, i64, i64) = inspect
+ .query_row(
+ "SELECT (SELECT COUNT(*) FROM event_log WHERE event_json IS NOT NULL),
+ (SELECT COUNT(*) FROM tool_result_payloads),
+ (SELECT COUNT(*) FROM event_log WHERE retained_json LIKE '%private-tool-data%')",
+ [],
+ |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
+ )
+ .unwrap();
+ assert_eq!((bodies, payloads, leaked), (0, 0, 0));
+ assert_eq!(
+ log.usage_summary(query()).await.unwrap().summary.tools,
+ totals
+ );
assert_eq!(
log.tool_attempts(query(), 0, 100).await.unwrap().attempts,
page.attempts
diff --git a/crates/runtime-host/src/execution/plugins/root.rs b/crates/runtime-host/src/execution/plugins/root.rs
index d37f9413f3..1b28f7f4c3 100644
--- a/crates/runtime-host/src/execution/plugins/root.rs
+++ b/crates/runtime-host/src/execution/plugins/root.rs
@@ -246,6 +246,7 @@ impl BoundCommands {
let result = match result {
Ok(maka_event_log::sessions::AbandonRevision::Abandoned) => {
grants.lock().unwrap().remove(&id);
+ worker.request_removal_recovery();
worker.publish_session_change(&id).await;
Ok(RevisionDisposition::Abandoned)
}
diff --git a/crates/runtime-host/src/execution/removal.rs b/crates/runtime-host/src/execution/removal.rs
index 941d45f787..fc55c59c9d 100644
--- a/crates/runtime-host/src/execution/removal.rs
+++ b/crates/runtime-host/src/execution/removal.rs
@@ -150,6 +150,23 @@ impl Executions {
.await
.map_err(internal)?;
if page.is_empty() {
+ // The same recovery owner drains small material batches. Copied
+ // history pins may defer collection until their last owner retires.
+ let mut cursor = None;
+ while self.accepting() {
+ use maka_event_log::sessions::MaterialCollection;
+ match self
+ .log
+ .collect_session_material(cursor.as_deref())
+ .await
+ .map_err(internal)?
+ {
+ MaterialCollection::Done => break,
+ MaterialCollection::Retained(session) => cursor = Some(session),
+ MaterialCollection::Collected => {}
+ }
+ tokio::task::yield_now().await;
+ }
return Ok(pending);
}
for session in page {
diff --git a/crates/runtime-host/src/server/sessions/copy.rs b/crates/runtime-host/src/server/sessions/copy.rs
index c9225f56a7..e9acfccc44 100644
--- a/crates/runtime-host/src/server/sessions/copy.rs
+++ b/crates/runtime-host/src/server/sessions/copy.rs
@@ -203,6 +203,7 @@ pub(super) async fn abandon(host: &Host, input: AbandonInput) -> Result<AbandonO
.map_err(stored)?
{
AbandonRevision::Abandoned => {
+ host.executions.request_removal_recovery();
host.executions
.publish_session_change(&input.target_session_id)
.await;
diff --git a/docs/rust-parity.md b/docs/rust-parity.md
index 9817e26f62..b90288ce78 100644
--- a/docs/rust-parity.md
+++ b/docs/rust-parity.md
@@ -92,7 +92,7 @@ implementations; wire adapters may retain existing client vocabulary.
| Usage / Pricing | Physical Agent and auxiliary SDK accounting, frozen valuation, public Rust/JS scoped mixed model/tool activity paging, and shared native/plugin pricing query/CAS edits are implemented. Consistent summaries include complete bounded breakdowns and missing-data coverage; The Insights plugin provides settings reports, filtering, pagination, persisted views and rate editing; Session Inspector uses the public Session-bound Client slot. | **Plugin + Host.** Reports and rebui [...]
| Background health | BackgroundTaskHealth process and endpoint checks. | **Plugin + Host.** Health interpretation and Tool in a plugin; Host exposes authorized resource observations and bounded probes. A stored PID is not resource ownership. |
| Session lineage | Public Rust/JS history copying and source reads, native branch/revision create/abandon, independent inherited pruning, Desktop canonical-input editing and complete durable drafts are implemented. Regeneration is removed; edits use revision and resend. | **Host.** One transactional lineage and workspace authority; plugins request changes through commands, not private lineage storage. |
-| Session lifecycle | Native and public plugin removal/preview/receipt APIs, atomic family retirement and queued cancellation, restart cleanup and shared-worktree ownership are implemented. Unreferenced canonical-material reclamation and shared Session queries remain. | **Host.** Preserve referenced history and accepted receipts; unknown process cleanup must not delete its workspace. Shared queries require real collaboration authorization. |
+| Session lifecycle | Native and public plugin removal/preview/receipt APIs, atomic family retirement and queued cancellation, restart cleanup and shared-worktree ownership are implemented. Unreferenced conversation bodies, tool payloads and request surfaces are reclaimed in restartable batches while accounting, receipts and referenced history survive. Shared Session queries remain. | **Host.** Preserve referenced history and accepted receipts; unknown process cleanup must not delete its [...]
| Session transfer | Bundle export/import and unified external catalog/source/import for Codex, Claude Code and OpenCode. | **Plugin + Host.** Source parsing/discovery can be adapters; Host owns bounded canonical import, identities, artifacts, provenance and atomic publication. Native bundle format remains a Host contract. Never import source event bytes as trusted execution authority. |
| Runtime policy | Shell/external-agent consumers and ordinary named tool profiles. | **Split.** Shell launch policy and capability ceilings stay in Host. External-agent settings belong to their domain. Profiles contribute definitions; Host applies the intersection at admission and per-step capture. Storing settings alone is insufficient. |
| Access / collaboration | Credential rotation prepare/revoke; principal revoke; collaboration access, invitation, grant revoke, principal rename/revoke; Turn-request create/query/decide/acknowledge/withdraw. | **Host.** Reuse credential and durable admission authority. Plugins may provide workflows/UI but cannot decide grants, bypass revocation or own canonical accepted Turn requests. |
diff --git a/docs/rust-parity.zh-CN.md b/docs/rust-parity.zh-CN.md
index bd753ebba6..d7ff69338b 100644
--- a/docs/rust-parity.zh-CN.md
+++ b/docs/rust-parity.zh-CN.md
@@ -82,7 +82,7 @@ runtime 契约不能反向依赖插件实现，协议适配层可以保留现有
| Usage／Pricing | 已实现 Agent 与辅助 SDK 的物理计量、冻结估价、公共 Rust／JS 作用域模型／工具混合活动分页，以及原生／插件共享的报价查询与 CAS 修改。一致快照汇总包含有界完整分组与缺失数据覆盖率；Insights 插件已提供设置报表、筛选、分页、视图持久化和报价编辑；Session Inspector 已使用公共 Session 作用域 Client 插槽。 | **插件 + Host。** 报表和可重建投影可归 Insights 领域；Host 不依赖插件存活来记录用量，并提供一致快照。缺失用量不能视为零。 |
| 后台健康 | BackgroundTaskHealth 的进程和端点检查。 | **插件 + Host。** 插件解释健康状态并提供工具；Host 提供授权资源观察和有界探测。保存 PID 不等于拥有进程。 |
| Session 谱系 | 已实现公共 Rust／JS 历史复制和原始输入读取、原生分支／修订创建与放弃、继承历史独立裁剪、Desktop 规范输入编辑及完整持久草稿。已移除 regenerate，修改请求使用编辑重发。 | **Host。** 谱系及工作区只有一个事务权威；插件请求命令，不在私有存储中重做谱系。 |
-| Session 生命周期 | 已实现原生与公共插件删除／预览／回执接口、原子家族退休及队列取消、重启清理和共享 worktree 所有权。无引用规范材料回收及 shared 查询仍待完成。 | **Host。** 保留被引用历史及已接受回执；进程清理不确定时不得删除工作目录。shared 查询需要真实协作授权。 |
+| Session 生命周期 | 已实现原生与公共插件删除／预览／回执接口、原子家族退休及队列取消、重启清理和共享 worktree 所有权。无引用对话正文、工具载荷与请求表面已支持可恢复的分批回收，保留计量、回执及被引用历史；shared 查询仍待完成。 | **Host。** 保留被引用历史及已接受回执；进程清理不确定时不得删除工作目录。shared 查询需要真实协作授权。 |
| Session 迁入迁出 | bundle 导入／导出；Codex、Claude Code、OpenCode 的统一外部 catalog/source/import。 | **插件 + Host。** 来源解析／发现可以作为 adapter；Host 拥有有界规范导入、身份、附件、来源记录和原子发布。原生 bundle 格式仍是 Host 合同，不能把外部事件字节直接作为可信执行权威。 |
| Runtime policy | shell／external-agent 消费，普通 named tool profiles。 | **拆分。** shell 启动策略和能力上限留在 Host；外部 agent 设置由对应领域消费。Profile 提供定义，Host 在准入和每步捕获时取能力交集。只保存设置不算完成。 |
| 接入／协作 | credential rotation prepare/revoke、principal revoke；collaboration access、邀请、grant revoke、principal rename/revoke；Turn-request create/query/decide/acknowledge/withdraw。 | **Host。** 复用凭据和持久化准入权威。插件可以提供流程／UI，但不能决定授权、绕过撤销或拥有规范的已接受 Turn request。 |
diff --git a/docs/rust-runtime.md b/docs/rust-runtime.md
index fded74bb80..4c6a95d917 100644
--- a/docs/rust-runtime.md
+++ b/docs/rust-runtime.md
@@ -350,8 +350,12 @@ over every directly removed member; history access alone is insufficient.
Host drains accepted executions and processes before reclaiming owned worktrees, and
keeps a shared checkout until its last owner retires. Unproven process cleanup leaves
-the workspace intact without blocking unrelated Sessions. Canonical facts and inherited
-archive proofs remain stored; removal is not secure erasure of historical data.
+the workspace intact without blocking unrelated Sessions. Once no surviving history owner
+needs a removed Session, background batches release its event bodies, tool payloads and
+unshared request surfaces. Original body digests, accounting and terminal proof, identities
+and accepted receipts remain. Inherited archive/checkpoint proofs stay pinned transitively;
+collected bodies cannot be replayed. Incremental vacuum returns unused pages in bounded
+steps; removal is not secure erasure of database/WAL backups.
## Conversation recall
diff --git a/docs/rust-runtime.zh-CN.md b/docs/rust-runtime.zh-CN.md
index 028be30a77..3efe9432a9 100644
--- a/docs/rust-runtime.zh-CN.md
+++ b/docs/rust-runtime.zh-CN.md
@@ -264,8 +264,10 @@ Desktop 可预览并附加同一 Host 中另一会话的已提交文本快照。
必须覆盖全部直接删除成员的权限；历史读取授权不等于删除授权。
Host 等待已接受执行和进程清理后才回收托管 worktree，共享目录保留到最后一个所有者退出。
-不能确认进程清理时保留目录，不阻塞无关会话。规范事实与继承归档证明仍保存在日志中；
-删除会话不等于安全擦除历史数据。
+不能确认进程清理时保留目录，不阻塞无关会话。没有存活历史所有者引用已删除会话后，
+后台分批释放事件正文、工具载荷及无共享引用的请求表面，保留原始正文摘要、计量和终态证明、
+身份及已接受回执。继承的归档／checkpoint 证明沿引用关系保留；回收后的正文不能重放。
+增量 vacuum 分批归还空闲页面；删除不等于安全擦除数据库／WAL 的备份。
## 会话回忆
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org