Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

feat: freeze Host Journal context across compaction

Browse Source
This commit is contained in:
Hare
2026-10-17 12:19:37 +09:00
parent 10d200d9ba
commit 956aeed6b2
18 changed files with 357 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/session-store/src/history.rs
+2
View File
@@ -69,6 +69,8 @@ pub enum LoggedSessionHistoryOrigin {
},
/// Host-produced opening context summary; not auto-read or TaskStore material.
CompactionSummary,
/// Frozen Host Journal reference, never independent observation evidence.
JournalContext,
DerivedSummary,
LegacyUnknown,
}
crates/session-store/src/public_material.rs
+3 -1
View File
@@ -829,7 +829,9 @@ fn boundary_reference(
fn canonical_body(entry: &LoggedHistoryEntry) -> Option<CanonicalBody<'_>> {
if matches!(
entry.metadata.origin,
LoggedSessionHistoryOrigin::DerivedSummary | LoggedSessionHistoryOrigin::CompactionSummary
LoggedSessionHistoryOrigin::DerivedSummary
| LoggedSessionHistoryOrigin::CompactionSummary
| LoggedSessionHistoryOrigin::JournalContext
) {
return None;
}
crates/session-store/src/public_material/tests.rs
+14 -1
View File
@@ -543,6 +543,14 @@ fn reasoning_proves_rollback_safety_but_private_and_derived_items_are_not_materi
let segment = crate::new_segment_id();
let mut derived = message("derived", LoggedRole::User, "summary");
derived.metadata.origin = LoggedSessionHistoryOrigin::DerivedSummary;
let mut journal = message("journal", LoggedRole::User, "frozen Journal reference");
journal.metadata.origin = LoggedSessionHistoryOrigin::JournalContext;
let mut summary = message(
"compaction-summary",
LoggedRole::User,
"complementary summary",
);
summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary;
let reasoning = entry(
"reasoning",
LoggedItem::Reasoning {
@@ -559,7 +567,12 @@ fn reasoning_proves_rollback_safety_but_private_and_derived_items_are_not_materi
&[
seed(
session,
vec![derived, message("system", LoggedRole::System, "secret")],
vec![
derived,
journal,
summary,
message("system", LoggedRole::System, "secret"),
],
),
LogEntry::AnnotatedSystemItem {
ts: 1,
crates/session-store/src/public_snapshot.rs
+1
View File
@@ -1889,6 +1889,7 @@ fn provenance(origin: &LoggedSessionHistoryOrigin) -> SessionEntryProvenance {
LoggedSessionHistoryOrigin::ModelOutput { .. } => SessionEntryProvenance::ModelOutput,
LoggedSessionHistoryOrigin::ToolOutput { .. } => SessionEntryProvenance::ToolOutput,
LoggedSessionHistoryOrigin::CompactionSummary
| LoggedSessionHistoryOrigin::JournalContext
| LoggedSessionHistoryOrigin::DerivedSummary => SessionEntryProvenance::DerivedSummary,
LoggedSessionHistoryOrigin::LegacyUnknown => SessionEntryProvenance::LegacyUnknown,
}
crates/standalone/src/subjektiv.rs
+38
View File
@@ -840,6 +840,44 @@ impl SubjektivHost for LocalSubjectHost {
.journal_startup_snapshot(&reader, predecessor.as_deref())
.map_err(journal::journal_error)
}
fn journal_compaction_granted(&self) -> bool {
self.live.load(Ordering::Acquire)
}
fn journal_compaction_context(
&self,
context: &SubjektivHostContext,
) -> Result<subjektiv::journal::startup::JournalStartupSnapshot, SubjektivHostError> {
self.body(context)?;
let attribution = self
.catalog
.store
.session_attribution(&context.session_id)
.map_err(domain_error)?
.filter(|attr| {
attr.subject_id == self.binding.subject_id
&& attr.runtime_id == self.binding.scope_id
&& attr.worker_id == self.worker_id.to_string()
})
.ok_or_else(|| {
SubjektivHostError::InvalidArgument(
"foreign or unattributed compaction Session".into(),
)
})?;
let host = subjektiv::journal::JournalHostContext::validated_session(
&self.catalog.store,
attribution,
)
.map_err(journal::journal_error)?;
if host.writer_binding_id() != Some(self.writer.binding_id.as_str()) {
return Err(SubjektivHostError::InvalidArgument(
"compaction writer lifetime mismatch".into(),
));
}
self.catalog
.store
.journal_compaction_snapshot(&host)
.map_err(journal::journal_error)
}
fn journal_recording_granted(&self) -> bool {
self.live.load(Ordering::Acquire)
}
crates/subjektiv/src/journal/startup.rs
+21 -2
View File
@@ -71,11 +71,30 @@ impl SubjektivStore {
&self,
host: &JournalHostContext,
previous_session: Option<&str>,
) -> JournalResult<JournalStartupSnapshot> {
self.context_snapshot(host, previous_session, false)
}
pub fn journal_compaction_snapshot(
&self,
host: &JournalHostContext,
) -> JournalResult<JournalStartupSnapshot> {
let session = host.attribution.as_ref().ok_or_else(|| {
JournalError::PermissionDenied("compaction requires a Host-bound body Session".into())
})?;
self.context_snapshot(host, Some(&session.session_id), true)
}
fn context_snapshot(
&self,
host: &JournalHostContext,
preferred_session: Option<&str>,
require_current_writer: bool,
) -> JournalResult<JournalStartupSnapshot> {
self.database.try_transaction(|tx| {
super::store::check_host(self, tx, host, false)?;
super::store::check_host(self, tx, host, require_current_writer)?;
let mut ids = Vec::new();
if let Some(session) = previous_session {
if let Some(session) = preferred_session {
let previous: Option<(String, String)> = tx.query_row(
"SELECT id,subject_id FROM journals WHERE source_session_id=?1",
[session], |r| Ok((r.get(0)?, r.get(1)?)),
crates/subjektiv/src/journal/transport.rs
+6
View File
@@ -21,6 +21,12 @@ pub enum JournalRecordingReason {
WorkerRestored,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct JournalCompactionRequest {
pub session_id: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct JournalRecordingRequest {
crates/worker/src/feature/builtin/journal.rs
+60
View File
@@ -3,6 +3,12 @@
//! Hooks request Host reconciliation, never infer source ranges from in-memory history.
//! Hosts retain exact operation inputs/receipts and validate live Job authority on every
//! operation. Successful operations are never rolled back by Job finalization.
//!
//! Compaction freezes a read-only Host snapshot independently of admission/Job progress.
//! Its rendered reference is shared by mandatory compactor input and the replacement seed;
//! no restore/turn hook refreshes it. Previous references are only historical compactor
//! material so fallback can recover information intentionally omitted from an earlier
//! complementary summary, without accumulating stale references in the next segment.
use crate::feature::{
FeatureDescriptor, FeatureHookPoint, FeatureInstallContext, FeatureInstallError, FeatureModule,
@@ -79,6 +85,60 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
}
}
/// Read once without waiting for recording. The returned body is immutable mandatory
/// compactor input and the exact replacement-segment reference, including fallback status.
/// Selection and authorization belong to the Host, not the core compaction engine.
pub(crate) fn freeze_compaction_journal(
connection: std::io::Result<Option<SubjektivHostConnection>>,
prompts: &crate::prompt::catalog::PromptCatalog,
) -> Result<String, crate::prompt::catalog::CatalogError> {
let workspace_id = connection.as_ref().ok().and_then(|connection| {
connection
.as_ref()
.and_then(|connection| connection.host.workspace_id())
});
let read = match connection {
Ok(Some(connection)) if connection.host.journal_compaction_granted() => {
Some(connection.journal_compaction_context().map_err(|_| ()))
}
Ok(_) => None,
Err(_) => Some(Err(())),
};
let (status, json) = match read {
None => ("disabled", String::new()),
Some(Ok(snapshot)) => {
let status = if snapshot.journals.is_empty() {
"absent"
} else {
"available"
};
match serde_json::to_string(&snapshot) {
Ok(json) if json.len() <= subjektiv::journal::startup::STARTUP_MAX_BYTES => {
(status, json)
}
_ => ("unavailable", String::new()),
}
}
Some(Err(())) => ("unavailable", String::new()),
};
if status == "unavailable" {
tracing::warn!("Journal compaction snapshot unavailable; continuing without Journal");
}
use sha2::{Digest, Sha256};
let snapshot_sha256: Option<String> = (!json.is_empty()).then(|| {
Sha256::digest(json.as_bytes())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
});
prompts.subjektiv_journal_compaction(
status,
&json,
snapshot_sha256.as_deref(),
workspace_id.as_deref(),
)
}
#[derive(Clone)]
pub struct JournalRecordingFeature {
host: SubjektivHostConnection,
crates/worker/src/prompt/catalog.rs
+22
View File
@@ -103,6 +103,7 @@ pub enum WorkerPrompt {
ResidentMemoryRestoreSection,
SubjektivResidentContext,
SubjektivJournalStartup,
SubjektivJournalCompaction,
WorkerOrchestrationGuidanceSection,
SubWorkerSpawnToolDescription,
}
@@ -128,6 +129,7 @@ impl WorkerPrompt {
Self::ResidentMemorySummarySection => "internal.resident_memory_summary_section",
Self::ResidentMemoryRestoreSection => "internal.resident_memory_restore_section",
Self::SubjektivJournalStartup => "internal.subjektiv_journal_startup",
Self::SubjektivJournalCompaction => "internal.subjektiv_journal_compaction",
Self::SubjektivResidentContext => "internal.subjektiv_resident_context",
Self::WorkerOrchestrationGuidanceSection => {
"internal.worker_orchestration_guidance_section"
@@ -154,6 +156,7 @@ impl WorkerPrompt {
WorkerPrompt::ResidentMemoryRestoreSection,
WorkerPrompt::SubjektivResidentContext,
WorkerPrompt::SubjektivJournalStartup,
WorkerPrompt::SubjektivJournalCompaction,
WorkerPrompt::WorkerOrchestrationGuidanceSection,
WorkerPrompt::SubWorkerSpawnToolDescription,
];
@@ -543,6 +546,25 @@ impl PromptCatalog {
},
)
}
pub(crate) fn subjektiv_journal_compaction(
&self,
status: &str,
snapshot_json: &str,
snapshot_sha256: Option<&str>,
workspace_id: Option<&str>,
) -> Result<String, CatalogError> {
self.render(
WorkerPrompt::SubjektivJournalCompaction,
minijinja::context! {
status => status,
snapshot_json => snapshot_json,
snapshot_sha256 => snapshot_sha256,
workspace_id => workspace_id,
source_digest => self.projection.source_digest,
projection_digest => self.projection.catalog_digest,
},
)
}
pub fn subjektiv_journal_startup(&self, snapshot_json: &str) -> Result<String, CatalogError> {
self.render(
WorkerPrompt::SubjektivJournalStartup,
crates/worker/src/session_capture.rs
+3 -3
View File
@@ -736,9 +736,9 @@ fn public_provenance(origin: &WorkerHistoryProvenance) -> SessionEntryProvenance
}
WorkerHistoryProvenance::ModelOutput { .. } => SessionEntryProvenance::ModelOutput,
WorkerHistoryProvenance::ToolOutput { .. } => SessionEntryProvenance::ToolOutput,
WorkerHistoryProvenance::CompactionSummary | WorkerHistoryProvenance::DerivedSummary => {
SessionEntryProvenance::DerivedSummary
}
WorkerHistoryProvenance::CompactionSummary
| WorkerHistoryProvenance::DerivedSummary
| WorkerHistoryProvenance::JournalContext => SessionEntryProvenance::DerivedSummary,
WorkerHistoryProvenance::LegacyUnknown => SessionEntryProvenance::LegacyUnknown,
}
}
crates/worker/src/session_history.rs
+11 -4
View File
@@ -156,10 +156,17 @@ pub(crate) fn restore_history_entries(
}
*target = from_logged(entry);
}
LogEntry::AnnotatedSystemItem { entry, .. } => history.push(HistoryEntry::new(
entry.item.to_history_item(),
entry.metadata.clone(),
)),
LogEntry::AnnotatedSystemItem { entry, .. } => {
history.push(HistoryEntry::new(entry.item.to_history_item(), {
let mut metadata = entry.metadata.clone();
// Typed startup records predate the dedicated Journal origin.
// Classify only this trusted event, never a plaintext marker.
if matches!(entry.item, session_store::SystemItem::JournalStartup { .. }) {
metadata.origin = WorkerHistoryProvenance::JournalContext;
}
metadata
}))
}
_ => {}
}
}
crates/worker/src/subjektiv.rs
+47
View File
@@ -80,6 +80,17 @@ pub trait SubjektivHost: Send + Sync + 'static {
"Journal startup not granted".into(),
))
}
fn journal_compaction_granted(&self) -> bool {
false
}
fn journal_compaction_context(
&self,
_context: &SubjektivHostContext,
) -> Result<JournalStartupSnapshot, SubjektivHostError> {
Err(SubjektivHostError::Unavailable(
"Journal compaction not granted".into(),
))
}
/// Recording admission is separate from both Memory and Journal Job operation grants.
fn journal_recording_granted(&self) -> bool {
false
@@ -267,6 +278,29 @@ impl SubjektivHost for BackendSubjektivHost {
.map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?,
)
}
fn journal_compaction_granted(&self) -> bool {
true
}
fn journal_compaction_context(
&self,
context: &SubjektivHostContext,
) -> Result<JournalStartupSnapshot, SubjektivHostError> {
decode(
self.client
.execute_with_timeout(
WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!("/api/w/{}/subjektiv/journal/compaction", self.workspace_id),
serde_json::to_string(&transport::JournalCompactionRequest {
session_id: context.session_id.clone(),
})
.map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?,
),
Duration::from_secs(5),
)
.map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?,
)
}
fn journal_recording_granted(&self) -> bool {
// This adapter exists only for explicitly enabled, authenticated Host connections.
// The Backend still verifies the main Worker/Subject binding on every request.
@@ -488,6 +522,19 @@ impl SubjektivHostConnection {
}
self.host.journal_startup_context(&self.context)
}
pub fn journal_compaction_granted(&self) -> bool {
self.host.journal_compaction_granted()
}
pub fn journal_compaction_context(&self) -> Result<JournalStartupSnapshot, SubjektivHostError> {
if !self.host.journal_compaction_granted() {
return Err(SubjektivHostError::Unavailable(
"Journal compaction not granted".into(),
));
}
self.host.journal_compaction_context(&self.context)
}
pub fn new(
host: Arc<dyn SubjektivHost>,
context: SubjektivHostContext,
crates/worker/src/worker.rs
+58 -4
View File
@@ -6918,6 +6918,13 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// within `retained_tokens`. Item-granular, turn boundaries ignored.
let cut = self.split_for_retained(retained_tokens);
// Freeze one Host-owned read before starting the compactor. Background recording
// remains independent; never re-read when building the replacement segment.
let journal_context = crate::feature::builtin::journal::freeze_compaction_journal(
self.subjektiv_host_connection(),
&self.prompts.load_full(),
)
.map_err(WorkerError::PromptCatalog)?;
let history_entries = self.session.history().entries();
let retain_from = cut.index.min(history_entries.len());
let initial_summary = initial_compaction_summary(history_entries)?;
@@ -6926,7 +6933,13 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// it as well would duplicate inherited context on every compact.
let retained_history_entries = history_entries[retain_from..]
.iter()
.filter(|entry| entry.annotation.origin != WorkerHistoryProvenance::CompactionSummary)
.filter(|entry| {
!matches!(
entry.annotation.origin,
WorkerHistoryProvenance::CompactionSummary
| WorkerHistoryProvenance::JournalContext
)
})
.cloned()
.collect::<Vec<_>>();
let mut retained_items = retained_history_entries
@@ -6943,7 +6956,13 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// never needs durable image bodies; relevant images are nominated with add_reference.
let mut entries_to_summarise = history_entries[..retain_from]
.iter()
.filter(|entry| entry.annotation.origin != WorkerHistoryProvenance::CompactionSummary)
.filter(|entry| {
!matches!(
entry.annotation.origin,
WorkerHistoryProvenance::CompactionSummary
| WorkerHistoryProvenance::JournalContext
)
})
.cloned()
.collect::<Vec<_>>();
project_compactor_capture_attachments(&mut entries_to_summarise);
@@ -7013,7 +7032,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// The full Host summary is mandatory input, separate from the bounded
// prefix overview and current TaskStore snapshot.
let task_snapshot_text = self.task_feature.snapshot_text();
let summary_input = build_summary_input(
let mut summary_input = build_summary_input(
&items_to_summarise,
initial_summary.as_deref(),
&default_refs,
@@ -7025,6 +7044,28 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
summary_target_tokens,
},
);
// Previous references are historical source material, not additional latest
// Journals. In particular, a failed/empty read must not silently lose context
// that the previous summary intentionally delegated to its injected Journal.
// Carry them only into the compactor, never alongside the new reference seed.
for entry in history_entries
.iter()
.filter(|entry| entry.annotation.origin == WorkerHistoryProvenance::JournalContext)
{
if let Some(body) = entry.item.as_text() {
summary_input.text.push_str(
"\n\n## Previous Journal reference (historical, not automatically retained)\n",
);
summary_input.text.push_str(body);
summary_input
.text
.push_str("\n## End previous Journal reference\n");
}
}
// Mandatory context, outside overview truncation. The internal Worker's normal
// request budget checks include this exact body and the full previous summary.
summary_input.text.push_str("\n\n");
summary_input.text.push_str(&journal_context);
if summary_input.warning_exceeded {
self.alert(
AlertLevel::Warn,
@@ -7340,6 +7381,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
crate::session_history::COMPACTION_SUMMARY_HEADER
));
compact_introduced_system_messages.push(summary_message.clone());
let journal_message = Item::system_message(journal_context);
compact_introduced_system_messages.push(journal_message.clone());
compact_introduced_system_messages.extend(auto_read_messages.iter().cloned());
if let Some(msg) = reference_message.as_ref() {
compact_introduced_system_messages.push(msg.clone());
@@ -7352,6 +7395,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
compact_introduced_system_messages.push(task_snapshot_message.clone());
new_history.push(summary_message);
new_history.push(journal_message);
new_history.extend(auto_read_messages);
if let Some(msg) = reference_message {
new_history.push(msg);
@@ -7407,6 +7451,16 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
}),
),
)
} else if index == 1 {
HistoryEntry::new(
item,
new_history_metadata(
WorkerHistoryProvenance::JournalContext,
Some(SessionHistoryDerivation {
sources: derived_sources.clone(),
}),
),
)
} else if let Some((offset, original)) = original_entries[original_cursor..]
.iter()
.enumerate()
@@ -16732,7 +16786,7 @@ permission = "write"
assert_eq!(replay[0].item.as_text(), Some("PRIOR_JOURNAL_REFERENCE"));
assert_eq!(
replay[0].annotation.origin,
WorkerHistoryProvenance::DerivedSummary
WorkerHistoryProvenance::JournalContext
);
let view =
crate::session_capture::SessionCapture::from_history_entries("startup", replay);
crates/workspace-server/src/server.rs
+1
View File
@@ -13148,6 +13148,7 @@ fn build_inner_router(api: WorkspaceApi) -> Router {
post(scoped_submit_backend_job_result),
)
.route("/api/w/{workspace_id}/subjektiv/journal/release", post(subjektiv_journal::release_route))
.route("/api/w/{workspace_id}/subjektiv/journal/compaction", post(subjektiv_journal::compaction_route))
.route("/api/w/{workspace_id}/subjektiv/journal/startup", get(subjektiv_journal::startup_route))
.route("/api/w/{workspace_id}/subjektiv/journal/recording", post(subjektiv_journal::recording_route))
.route("/api/w/{workspace_id}/subjektiv/journal/memory", post(subjektiv_memory_journal::route))
crates/workspace-server/src/server/subjektiv_journal.rs
+51
View File
@@ -598,6 +598,57 @@ pub(super) async fn startup_route(
.await
}
pub(super) async fn compaction_route(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
source: Option<axum::Extension<crate::worker_source::VerifiedRuntimeRequestSource>>,
Json(input): Json<JournalCompactionRequest>,
) -> ApiResult<Json<journal::startup::JournalStartupSnapshot>> {
let worker = verified_worker(source)?;
api.store
.db_executor()
.run(move || {
validate_workspace_scope(&api, &path.workspace_id)?;
let lease = api
.store
.require_current_worker_singleton_owner(&path.workspace_id, &worker)?
.ok_or_else(|| {
Error::WorkspacePermissionDenied(
"Journal compaction requires the current main Subject Worker".into(),
)
})?;
let (subject, role) = subjektiv_scope_from_singleton_key(&lease.key)?;
if role != SubjektivWorkerAuthority::Subject {
return Err(Error::WorkspacePermissionDenied(
"Journal compaction is not a Job capability".into(),
)
.into());
}
let store = store(&api)?;
let attribution = store
.session_attribution(&input.session_id)
.map_err(|error| Error::Store(error.to_string()))?
.filter(|attr| {
attr.subject_id == subject
&& attr.runtime_id == worker.runtime_id
&& attr.worker_id == worker.worker_id
})
.ok_or_else(|| {
Error::WorkspacePermissionDenied(
"foreign or unattributed compaction Session".into(),
)
})?;
let host =
JournalHostContext::validated_session(&store, attribution).map_err(domain)?;
store
.journal_compaction_snapshot(&host)
.map(Json)
.map_err(domain)
.map_err(ApiError::from)
})
.await
}
#[derive(Clone, Debug, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub(super) struct SubjectReleaseInput {