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

update: SessionId / SessionStart / SessionOrigin 等を Segment 系名称へ

Browse Source 
- Type/Function/Variantを Segment* 系へ統一
  - SessionId/SessionStart/SessionOrigin/SessionStartState/SessionState/SessionLogSink/SessionLockInfo
  - new_session_id / session_id / create_session* / list_sessions / lookup_session / update_session / find_by_session
  - protocol Event::SessionRotated → SegmentRotated、CompactDone.new_session_id → new_segment_id
- Module: session_log → segment_log / session → segment (file mv 含む)
  pod 側の session_log_sink → segment_log_sink も同様
- crate 名 (session-store)、CLI flag (--session)、ResumeWithSession (CLI tied) は据え置き
- session-tests/session_metrics_test 等の Store impl も追従
This commit is contained in:
Hare
2026-05-20 05:06:04 +09:00
parent de549812ab
commit 22f5d02385
55 changed files with 611 additions and 610 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/pod/examples/pod_cli.rs
+2 -2
View File
@@ -54,7 +54,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let mut pod = Pod::from_manifest_toml(&toml, store).await?;
let manifest: &PodManifest = pod.manifest();
println!("Pod: {}", manifest.pod.name);
println!("Session: {}", pod.session_id());
println!("Session: {}", pod.segment_id());
// 4. Run a prompt
let result = pod.run_text("What is the capital of France?").await?;
@@ -76,7 +76,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
}
// 6. Session ID for potential restore
println!("\nSession ID: {}", pod.session_id());
println!("\nSession ID: {}", pod.segment_id());
Ok(())
}
crates/pod/src/controller.rs
+6 -6
View File
@@ -11,7 +11,7 @@ use crate::ipc::notify_buffer::NotifyBuffer;
use crate::ipc::server::SocketServer;
use crate::pod::{Pod, PodError, PodRunResult, SystemItemCommitter};
use crate::runtime::dir::RuntimeDir;
use crate::session_log_sink::SessionLogSink;
use crate::segment_log_sink::SegmentLogSink;
use crate::shared_state::PodSharedState;
use crate::spawn::comm_tools::{
list_pods_tool, read_pod_output_tool, send_to_pod_tool, stop_pod_tool,
@@ -33,10 +33,10 @@ pub struct PodHandle {
pub shared_state: Arc<PodSharedState>,
pub runtime_dir: Arc<RuntimeDir>,
pub alerter: Alerter,
/// Session-log mirror + broadcast handle. The IPC server snapshots
/// Segment-log mirror + broadcast handle. The IPC server snapshots
/// it on every new connection (Event::Snapshot) and forwards
/// subsequent commits (Event::Entry) on the receiver.
pub sink: SessionLogSink,
pub sink: SegmentLogSink,
}
impl PodHandle {
@@ -214,7 +214,7 @@ impl PodController {
let greeting = build_greeting(&pod);
let shared_state = Arc::new(PodSharedState::new(
pod.manifest().pod.name.clone(),
pod.session_id(),
pod.segment_id(),
manifest_toml.clone(),
greeting,
));
@@ -430,7 +430,7 @@ where
let scope_handle = pod.scope().clone();
let pwd = pod.pwd().to_path_buf();
let task_store = pod.task_store();
let session_id_for_usage = pod.session_id().to_string();
let session_id_for_usage = pod.segment_id().to_string();
let scope_change_sink = pod.scope_change_sink();
let memory_config = pod.manifest().memory.clone();
let spawner_name = pod.manifest().pod.name.clone();
@@ -992,7 +992,7 @@ mod tests {
let (cancel_tx, cancel_rx) = mpsc::channel::<()>(1);
let shared_state = Arc::new(PodSharedState::new(
"child-pod".to_string(),
session_store::new_session_id(),
session_store::new_segment_id(),
String::new(),
protocol::Greeting {
pod_name: "child-pod".to_string(),
crates/pod/src/ipc/server.rs
+3 -3
View File
@@ -105,10 +105,10 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: PodHandle) {
match entry {
Ok(entry) => {
let outbound = match entry {
session_store::LogEntry::SessionStart { .. } => {
session_store::LogEntry::SegmentStart { .. } => {
let value = serde_json::to_value(&entry)
.expect("LogEntry is Serialize");
Some(Event::SessionRotated { entry: value })
Some(Event::SegmentRotated { entry: value })
}
session_store::LogEntry::SystemItem { item, .. } => {
let value = serde_json::to_value(&item)
@@ -119,7 +119,7 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: PodHandle) {
Some(Event::InvokeStart { kind: trigger })
}
other => {
// `SessionLogSink::is_live_relevant` keeps
// `SegmentLogSink::is_live_relevant` keeps
// non-live-relevant variants off the
// broadcast lane; reaching here means the
// two are out of sync and we silently
crates/pod/src/lib.rs
+2 -2
View File
@@ -5,7 +5,7 @@ pub mod hook;
pub mod ipc;
pub mod prompt;
pub mod runtime;
pub mod session_log_sink;
pub mod segment_log_sink;
pub mod shared_state;
pub mod spawn;
pub mod workflow;
@@ -31,5 +31,5 @@ pub use prompt::system::{SystemPromptContext, SystemPromptError, SystemPromptTem
pub use protocol::{ErrorCode, Event, Method, PodStatus, TurnResult};
pub use provider::{ProviderError, build_client};
pub use runtime::dir::RuntimeDir;
pub use session_log_sink::SessionLogSink;
pub use segment_log_sink::SegmentLogSink;
pub use shared_state::PodSharedState;
crates/pod/src/main.rs
+2 -2
View File
@@ -5,7 +5,7 @@ use std::process::ExitCode;
use clap::Parser;
use manifest::{PodManifest, paths};
use pod::{Pod, PodController, PodFactory, PromptLoader};
use session_store::{FsStore, SessionId};
use session_store::{FsStore, SegmentId};
const USER_MANIFEST_ENV: &str = "INSOMNIA_USER_MANIFEST";
@@ -53,7 +53,7 @@ struct Cli {
/// Mutually exclusive with `--adopt` (spawned children always start
/// fresh).
#[arg(long, value_name = "UUID", conflicts_with = "adopt")]
session: Option<SessionId>,
session: Option<SegmentId>,
}
fn resolve_manifest(cli: &Cli) -> Result<(PodManifest, PromptLoader), String> {
crates/pod/src/pod.rs
+124 -124
View File
@@ -9,11 +9,11 @@ use llm_worker::llm_client::client::LlmClient;
use llm_worker::state::Mutable;
use llm_worker::{ToolOutputLimits, UsageRecord, Worker, WorkerError, WorkerResult};
use session_store::{
LogEntry, PodScopeSnapshot, SessionId, Store, StoreError, SystemItem, session_log, to_logged,
LogEntry, PodScopeSnapshot, SegmentId, Store, StoreError, SystemItem, segment_log, to_logged,
};
use tracing::{info, warn};
use crate::session_log_sink::SessionLogSink;
use crate::segment_log_sink::SegmentLogSink;
use manifest::{
Permission, PodManifest, PodManifestConfig, ResolveError, Scope, ScopeConfig, ScopeError,
@@ -44,33 +44,33 @@ use tokio::task::JoinHandle;
/// Lock-free shared session pointer.
///
/// Holds the current `(session_id, entries_written)` pair so that the
/// Holds the current `(segment_id, entries_written)` pair so that the
/// Pod and every `LogWriterHandle` clone see a consistent view through
/// `Arc`-shared lock-free reads. `session_id` is wrapped in `ArcSwap`
/// `Arc`-shared lock-free reads. `segment_id` is wrapped in `ArcSwap`
/// so fork (a rare, run-start-only event) can atomically swap it
/// without taking a mutex on the append hot path. `entries_written` is
/// an `AtomicUsize` bumped on every successful append; the writer's
/// tally is compared against the store's on-disk count to detect
/// concurrent writers in `ensure_session_head`.
pub struct SessionState {
session_id: ArcSwap<SessionId>,
pub struct SegmentState {
segment_id: ArcSwap<SegmentId>,
entries_written: AtomicUsize,
}
impl SessionState {
pub fn new(session_id: SessionId, entries_written: usize) -> Arc<Self> {
impl SegmentState {
pub fn new(segment_id: SegmentId, entries_written: usize) -> Arc<Self> {
Arc::new(Self {
session_id: ArcSwap::from_pointee(session_id),
segment_id: ArcSwap::from_pointee(segment_id),
entries_written: AtomicUsize::new(entries_written),
})
}
pub fn session_id(&self) -> SessionId {
**self.session_id.load()
pub fn segment_id(&self) -> SegmentId {
**self.segment_id.load()
}
pub fn set_session_id(&self, id: SessionId) {
self.session_id.store(Arc::new(id));
pub fn set_session_id(&self, id: SegmentId) {
self.segment_id.store(Arc::new(id));
}
pub fn entries_written(&self) -> usize {
@@ -94,8 +94,8 @@ impl SessionState {
#[derive(Clone)]
pub struct LogWriterHandle<St: Clone> {
pub store: St,
pub state: Arc<SessionState>,
pub sink: SessionLogSink,
pub state: Arc<SegmentState>,
pub sink: SegmentLogSink,
}
impl<St> LogWriterHandle<St>
@@ -107,8 +107,8 @@ where
/// writes for `< PIPE_BUF` lines, so no user-space serialization is
/// needed across appenders.
pub fn append_entry(&self, entry: LogEntry) -> Result<(), StoreError> {
let session_id = self.state.session_id();
self.store.append(session_id, &entry)?;
let segment_id = self.state.segment_id();
self.store.append(segment_id, &entry)?;
self.state.increment_entries();
self.sink.publish(entry);
Ok(())
@@ -128,7 +128,7 @@ where
{
fn commit_system_item(&self, item: SystemItem) {
let entry = LogEntry::SystemItem {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
item,
};
if let Err(err) = self.append_entry(entry) {
@@ -162,9 +162,9 @@ pub struct Pod<C: LlmClient, St: Store> {
worker: Option<Worker<C, Mutable>>,
store: St,
/// Shared session pointer. Source of truth for the Pod's current
/// `session_id` and append tally. `self.session_id()` is a thin
/// wrapper over `session_state.session_id()`.
session_state: Arc<SessionState>,
/// `segment_id` and append tally. `self.segment_id()` is a thin
/// wrapper over `session_state.segment_id()`.
session_state: Arc<SegmentState>,
/// Absolute working directory of the Pod.
pwd: PathBuf,
/// Shared, atomically-swappable view of the Pod's resolved scope.
@@ -284,7 +284,7 @@ pub struct Pod<C: LlmClient, St: Store> {
memory_task: Option<JoinHandle<()>>,
/// Typed user submissions in submit order. K-th entry corresponds to
/// the K-th `Item::user_message` in `worker.history()` (modulo seed
/// history loaded via `SessionStart.history`, whose original segments
/// history loaded via `SegmentStart.history`, whose original segments
/// are not preserved). Populated from log on `restore_from_manifest`,
/// appended after `save_user_input` on each `run`. Pre-`Event::Snapshot`
/// this fed `PodSharedState.user_segments`; the new wire format
@@ -295,7 +295,7 @@ pub struct Pod<C: LlmClient, St: Store> {
/// every successful `session_store::append_entry` write so connected
/// clients see a `(snapshot, live)` stream consistent with what's
/// on disk.
sink: SessionLogSink,
sink: SegmentLogSink,
/// `true` once `wire_history_persistence` has installed the
/// `Worker::on_history_append` callback that commits each appended
/// item as a singular `LogEntry::AssistantItem` / `ToolResult`
@@ -369,7 +369,7 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Pod<C, St> {
// The memory-task clone never appends to the session log
// (it only reads `worker.history()`), so a fresh sink is
// fine — nothing observes its broadcast.
sink: SessionLogSink::new(),
sink: SegmentLogSink::new(),
history_persistence_wired: false,
log_writer: None,
}
@@ -422,7 +422,7 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Pod<C, St> {
) {
return;
}
let entry = session_store::classify_history_item(item, session_log::now_millis());
let entry = session_store::classify_history_item(item, segment_log::now_millis());
if let Err(err) = writer.append_entry(entry) {
warn!(error = %err, "history append commit failed; dropping");
}
@@ -469,16 +469,16 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
pwd: PathBuf,
scope: Scope,
) -> Result<Self, PodError> {
// Session creation is deferred to `ensure_session_head` at first
// Segment creation is deferred to `ensure_session_head` at first
// run so a later-installed system-prompt template (see
// `set_system_prompt_template`) can be captured by `SessionStart`.
let session_id = session_store::new_session_id();
// `set_system_prompt_template`) can be captured by `SegmentStart`.
let segment_id = session_store::new_segment_id();
let prompts = PromptCatalog::builtins_only()?;
let mut pod = Self {
manifest,
worker: Some(worker),
store,
session_state: SessionState::new(session_id, 0),
session_state: SegmentState::new(segment_id, 0),
pwd,
scope: SharedScope::new(scope),
hook_builder: HookRegistryBuilder::new(),
@@ -506,7 +506,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
extract_pointer: Arc::new(Mutex::new(None)),
memory_task: None,
user_segments: Vec::new(),
sink: SessionLogSink::new(),
sink: SegmentLogSink::new(),
history_persistence_wired: false,
log_writer: None,
};
@@ -544,8 +544,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// The session ID used for persistence. Read lock-free from the
/// shared session pointer so fork-time swaps are observed immediately.
pub fn session_id(&self) -> SessionId {
self.session_state.session_id()
pub fn segment_id(&self) -> SegmentId {
self.session_state.segment_id()
}
/// The Pod's manifest.
@@ -616,7 +616,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
};
let payload = serde_json::to_value(&snapshot).expect("PodScopeSnapshot is Serialize");
self.commit_entry(LogEntry::Extension {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
domain: session_store::POD_SCOPE_EXTENSION_DOMAIN.into(),
payload,
})
@@ -627,8 +627,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// concurrent appenders — the kernel orders `O_APPEND` writes for
/// lines smaller than `PIPE_BUF`.
pub(crate) fn commit_entry(&self, entry: LogEntry) -> Result<(), StoreError> {
let session_id = self.session_state.session_id();
self.store.append(session_id, &entry)?;
let segment_id = self.session_state.segment_id();
self.store.append(segment_id, &entry)?;
self.session_state.increment_entries();
self.sink.publish(entry);
Ok(())
@@ -637,7 +637,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// Cloneable sink handle. Exposed to the controller so the IPC
/// layer can `subscribe_with_snapshot` and stream entries to
/// clients without consulting any other state.
pub fn sink(&self) -> SessionLogSink {
pub fn sink(&self) -> SegmentLogSink {
self.sink.clone()
}
@@ -661,7 +661,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
if let Some(snapshot) = snapshot {
let payload = serde_json::to_value(&snapshot).expect("PodScopeSnapshot is Serialize");
self.commit_entry(LogEntry::Extension {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
domain: session_store::POD_SCOPE_EXTENSION_DOMAIN.into(),
payload,
})?;
@@ -716,7 +716,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// Snapshot of the typed user segments tracked alongside worker
/// history. The K-th entry corresponds to the K-th `Item::user_message`
/// derived from `LogEntry::UserInput` entries (post-compaction); seed
/// history loaded via `SessionStart.history` does not contribute,
/// history loaded via `SegmentStart.history` does not contribute,
/// which is acceptable because the original segments are unrecoverable.
pub fn user_segments(&self) -> &[Vec<Segment>] {
&self.user_segments
@@ -829,7 +829,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
fn try_record_metric(&mut self, metric: &session_metrics::Metric) {
let payload = serde_json::to_value(metric).expect("Metric is Serialize");
let entry = LogEntry::Extension {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
domain: session_metrics::DOMAIN.into(),
payload,
};
@@ -1188,7 +1188,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// IDLE → active marker. Commits first so the next UserInput entry
// is contained inside this Invoke range. See `tickets/invoke-turn-llmcall-semantics.md`.
self.commit_entry(LogEntry::Invoke {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
trigger: protocol::InvokeKind::UserSend,
})?;
@@ -1196,7 +1196,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// pushes its flattened copy into history. save_delta deliberately
// skips the resulting `is_user_message()` item to avoid double-write.
self.commit_entry(LogEntry::UserInput {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
segments: input.clone(),
})
?;
@@ -1376,7 +1376,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
return;
};
if let Err(err) =
memory::append_use_event(layout, self.session_id().to_string(), source, records)
memory::append_use_event(layout, self.segment_id().to_string(), source, records)
{
warn!(error = %err, "failed to append memory usage event");
}
@@ -1387,7 +1387,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
return;
};
if let Err(err) =
memory::append_resident_exposure_event(layout, self.session_id().to_string(), records)
memory::append_resident_exposure_event(layout, self.segment_id().to_string(), records)
{
warn!(error = %err, "failed to append resident exposure event");
}
@@ -1578,7 +1578,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// drain. The trailing SystemItem entries (drained by the
// PodInterceptor) carry the actual payload.
self.commit_entry(LogEntry::Invoke {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
trigger: kind,
})?;
@@ -1612,17 +1612,17 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
///
/// On the first call for a Pod built via `from_manifest`, the session
/// has not been written to the store yet — this is when we append the
/// initial `SessionStart` entry, carrying the system prompt that
/// initial `SegmentStart` entry, carrying the system prompt that
/// `ensure_system_prompt_materialized` has just rendered. Subsequent
/// calls fall through to entry-count comparison, which auto-forks
/// when another writer has appended behind our back.
fn ensure_session_head(&mut self) -> Result<(), PodError> {
let w = self.worker.as_ref().unwrap();
let prev_session_id = self.session_state.session_id();
let prev_session_id = self.session_state.segment_id();
let entries_written = self.session_state.entries_written();
if entries_written == 0 {
let initial = LogEntry::SessionStart {
ts: session_log::now_millis(),
let initial = LogEntry::SegmentStart {
ts: segment_log::now_millis(),
system_prompt: w.get_system_prompt().map(String::from),
config: w.request_config().clone(),
history: to_logged(w.history()),
@@ -1642,11 +1642,11 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
return Ok(());
}
// Fork: mint a fresh session and switch to it. The new
// SessionStart entry replaces the mirror and is broadcast
// SegmentStart entry replaces the mirror and is broadcast
// through the sink so existing subscribers reset their view.
let fork_id = session_store::new_session_id();
let entry = LogEntry::SessionStart {
ts: session_log::now_millis(),
let fork_id = session_store::new_segment_id();
let entry = LogEntry::SegmentStart {
ts: segment_log::now_millis(),
system_prompt: w.get_system_prompt().map(String::from),
config: w.request_config().clone(),
history: to_logged(w.history()),
@@ -1654,13 +1654,13 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
compacted_from: None,
};
self.store
.create_session(fork_id, &[entry.clone()])
.create_segment(fork_id, &[entry.clone()])
.map_err(PodError::from)?;
self.session_state.set_session_id(fork_id);
self.session_state.set_entries_written(1);
self.sink.reset_with_initial(entry);
if self.scope_allocation.is_some() {
pod_registry::update_session(&self.manifest.pod.name, fork_id)?;
pod_registry::update_segment(&self.manifest.pod.name, fork_id)?;
}
Ok(())
}
@@ -1717,12 +1717,12 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
self.send_event(Event::CompactStart);
match self.compact(retained).await {
Ok(new_session_id) => {
Ok(new_segment_id) => {
info!(
new_session_id = %new_session_id,
new_segment_id = %new_segment_id,
"Compaction succeeded, resuming execution"
);
self.send_event(Event::CompactDone { new_session_id });
self.send_event(Event::CompactDone { new_segment_id });
if let Some(ref state) = self.compact_state {
state.record_compact_success();
}
@@ -1767,12 +1767,12 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
let retained = state.retained_tokens();
self.send_event(Event::CompactStart);
match self.compact(retained).await {
Ok(new_session_id) => {
Ok(new_segment_id) => {
info!(
new_session_id = %new_session_id,
new_segment_id = %new_segment_id,
"Proactive pre-run compaction succeeded"
);
self.send_event(Event::CompactDone { new_session_id });
self.send_event(Event::CompactDone { new_segment_id });
state.record_compact_success();
}
Err(e) => {
@@ -1814,7 +1814,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
.iter()
.cloned()
.collect();
let ts = session_log::now_millis();
let ts = segment_log::now_millis();
for item in &new_items {
if item.is_user_message() {
continue;
@@ -1837,7 +1837,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
let turn_count = self.worker.as_ref().unwrap().turn_count();
self.commit_entry(LogEntry::TurnEnd {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
turn_count,
})
?;
@@ -1874,7 +1874,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
correlation_id,
} = recorded;
self.commit_entry(LogEntry::LlmUsage {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
history_len: record.history_len,
input_total_tokens: record.input_total_tokens,
cache_read_tokens: record.cache_read_tokens,
@@ -1900,7 +1900,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
match result {
Ok(r) => {
self.commit_entry(LogEntry::RunCompleted {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
interrupted,
result: r.clone(),
})
@@ -1908,7 +1908,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
}
Err(e) => {
self.commit_entry(LogEntry::RunErrored {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
interrupted,
message: e.to_string(),
})
@@ -1928,7 +1928,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// - a clone of the main LlmClient via `clone_boxed()`.
///
/// Returns the new session ID.
pub async fn compact(&mut self, retained_tokens: u64) -> Result<SessionId, PodError> {
pub async fn compact(&mut self, retained_tokens: u64) -> Result<SegmentId, PodError> {
use crate::compact::worker::{
CompactWorkerContext, CompactWorkerInterceptor, add_reference_tool,
mark_read_required_tool, write_summary_tool,
@@ -2001,7 +2001,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
.compact_system()
.map_err(PodError::PromptCatalog)?;
let mut summary_worker = Worker::new(summary_client).system_prompt(summary_system_prompt);
summary_worker.set_cache_key(Some(self.session_id().to_string()));
summary_worker.set_cache_key(Some(self.segment_id().to_string()));
// Occupancy-based input-token meter + interceptor. The tracker pairs
// each pre-request history length with the following UsageEvent, then
@@ -2140,41 +2140,41 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
task_snapshot_text.clone(),
));
// Build the SessionStart entry for the new compacted session,
// Build the SegmentStart entry for the new compacted session,
// then atomically rotate to it: create on disk, swap head, reset
// the broadcast sink so existing subscribers see the new
// `SessionStart { compacted_from }` and reset their view.
let new_session_id = session_store::new_session_id();
let old_session_id = self.session_state.session_id();
// `SegmentStart { compacted_from }` and reset their view.
let new_segment_id = session_store::new_segment_id();
let old_session_id = self.session_state.segment_id();
let source_turn_count = self.worker.as_ref().unwrap().turn_count();
let w = self.worker.as_ref().unwrap();
let entry = LogEntry::SessionStart {
ts: session_log::now_millis(),
let entry = LogEntry::SegmentStart {
ts: segment_log::now_millis(),
system_prompt: w.get_system_prompt().map(String::from),
config: w.request_config().clone(),
history: to_logged(&new_history),
forked_from: None,
compacted_from: Some(session_store::SessionOrigin {
session_id: old_session_id,
compacted_from: Some(session_store::SegmentOrigin {
segment_id: old_session_id,
at_turn_index: source_turn_count,
}),
};
self.store.create_session(new_session_id, &[entry.clone()])?;
self.session_state.set_session_id(new_session_id);
self.store.create_segment(new_segment_id, &[entry.clone()])?;
self.session_state.set_session_id(new_segment_id);
self.session_state.set_entries_written(1);
let session_start = entry;
// Broadcast the SessionStart through the sink. This atomically
// resets the mirror to `[SessionStart]` so any subscriber
// Broadcast the SegmentStart through the sink. This atomically
// resets the mirror to `[SegmentStart]` so any subscriber
// querying after this point sees the post-compaction prefix.
self.sink.reset_with_initial(session_start);
// Keep pods.json pointing at the live session_id. Without this
// a concurrent `restore_from_manifest(new_session_id)` would
// Keep pods.json pointing at the live segment_id. Without this
// a concurrent `restore_from_manifest(new_segment_id)` would
// see no live writer and grab the session this Pod just moved
// into, causing two writers to race on the same jsonl. Skipped
// when no allocation is installed (e.g. compact under
// `Pod::new` in tests).
if self.scope_allocation.is_some() {
pod_registry::update_session(&self.manifest.pod.name, new_session_id)?;
pod_registry::update_segment(&self.manifest.pod.name, new_segment_id)?;
}
// Align user_segments with the post-compaction history. Items
// before `retain_from` (now folded into the summary) lose their
@@ -2188,8 +2188,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
self.worker.as_mut().unwrap().set_history(new_history);
// Compaction-introduced system messages are part of the new
// SessionStart's history (broadcast above) — clients derive
// their blocks from `SessionStart.history`. No per-item
// SegmentStart's history (broadcast above) — clients derive
// their blocks from `SegmentStart.history`. No per-item
// broadcast is required.
let _ = &compact_introduced_system_messages;
let worker = self.worker.as_mut().unwrap();
@@ -2198,9 +2198,9 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// compact layout guarantees history[0] is the summary.
worker.set_cache_anchor(Some(0));
// Re-key the OpenAI Responses prompt cache namespace to the new
// session_id so post-compact turns share a key with extract /
// segment_id so post-compact turns share a key with extract /
// consolidate workers running in the same session.
worker.set_cache_key(Some(new_session_id.to_string()));
worker.set_cache_key(Some(new_segment_id.to_string()));
self.usage_history
.lock()
.expect("usage_history poisoned")
@@ -2219,7 +2219,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
.lock()
.expect("extract_pointer poisoned") = None;
Ok(new_session_id)
Ok(new_segment_id)
}
/// Build the LlmClient for the compactor Worker.
@@ -2367,7 +2367,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// Read the session log to get the current entry count. This is
// the boundary for the source.range end_entry. Called once per
// extract, on a small local file.
let entries_now = self.store.read_all(self.session_id())?.len();
let entries_now = self.store.read_all(self.segment_id())?.len();
if entries_now == 0 {
return Ok(ExtractDecision::Skipped);
}
@@ -2399,7 +2399,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
.memory_extract_system(memory_language)
.map_err(PodError::PromptCatalog)?;
let mut extract_worker = Worker::new(client).system_prompt(extract_system_prompt);
extract_worker.set_cache_key(Some(self.session_id().to_string()));
extract_worker.set_cache_key(Some(self.segment_id().to_string()));
// Occupancy-based input-token meter + interceptor. The tracker pairs
// each pre-request history length with the following UsageEvent, then
@@ -2435,12 +2435,12 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
extract::ExtractedPayload::default()
});
let source_session_id = self.session_state.session_id();
let source_session_id = self.session_state.segment_id();
let staging_id = if payload.is_empty() {
String::new()
} else {
let source = memory::schema::SourceRef {
session_id: source_session_id.to_string(),
segment_id: source_session_id.to_string(),
range: [start_entry as u64, end_entry as u64],
};
let (id, _) = extract::write_staging(&layout, source, payload)
@@ -2456,7 +2456,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
let payload_value = serde_json::to_value(&pointer_payload)
.expect("ExtractPointerPayload is always JSON-serializable");
self.commit_entry(LogEntry::Extension {
ts: session_log::now_millis(),
ts: segment_log::now_millis(),
domain: extract::EXTRACT_DOMAIN.into(),
payload: payload_value,
})?;
@@ -2598,7 +2598,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
}
};
let mut worker = Worker::new(client).system_prompt(consolidation_system_prompt);
worker.set_cache_key(Some(self.session_id().to_string()));
worker.set_cache_key(Some(self.segment_id().to_string()));
// Memory tools are self-contained — they bypass ScopedFs and write
// directly under the workspace via WorkspaceLayout. Resident
@@ -2610,7 +2610,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
let query_cfg = memory::tool::QueryConfig::from(memory_cfg);
worker.register_tool(memory::tool::read_tool_with_usage(
layout.clone(),
self.session_id().to_string(),
self.segment_id().to_string(),
));
worker.register_tool(memory::tool::write_tool(layout.clone()));
worker.register_tool(memory::tool::edit_tool(layout.clone()));
@@ -2735,12 +2735,12 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
let mut common = prepare_pod_common(&manifest, &loader, /* parse_template */ true)?;
let skill_shadows = std::mem::take(&mut common.skill_shadows);
// Session creation is deferred to the first run (see
// `ensure_session_head`) so the SessionStart entry can capture
// Segment creation is deferred to the first run (see
// `ensure_session_head`) so the SegmentStart entry can capture
// the rendered system prompt, not the raw template source. The
// session_id is allocated here so the pod-registry registration
// segment_id is allocated here so the pod-registry registration
// can record it from the start.
let session_id = session_store::new_session_id();
let segment_id = session_store::new_segment_id();
// Register this Pod in the machine-wide pod-registry
// before building anything else, so a spawn that conflicts on
@@ -2754,18 +2754,18 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
std::process::id(),
socket_path,
common.scope.allow_rules(),
session_id,
segment_id,
)?;
let mut worker = Worker::new(common.client);
apply_worker_manifest(&mut worker, &manifest.worker);
worker.set_cache_key(Some(session_id.to_string()));
worker.set_cache_key(Some(segment_id.to_string()));
let mut pod = Self {
manifest,
worker: Some(worker),
store,
session_state: SessionState::new(session_id, 0),
session_state: SegmentState::new(segment_id, 0),
pwd: common.pwd,
scope: SharedScope::new(common.scope),
hook_builder: HookRegistryBuilder::new(),
@@ -2793,7 +2793,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
extract_pointer: Arc::new(Mutex::new(None)),
memory_task: None,
user_segments: Vec::new(),
sink: SessionLogSink::new(),
sink: SegmentLogSink::new(),
history_persistence_wired: false,
log_writer: None,
};
@@ -2820,22 +2820,22 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
let mut common = prepare_pod_common(&manifest, &loader, /* parse_template */ true)?;
let skill_shadows = std::mem::take(&mut common.skill_shadows);
let session_id = session_store::new_session_id();
let segment_id = session_store::new_segment_id();
let scope_allocation = pod_registry::adopt_allocation(
manifest.pod.name.clone(),
std::process::id(),
session_id,
segment_id,
)?;
let mut worker = Worker::new(common.client);
apply_worker_manifest(&mut worker, &manifest.worker);
worker.set_cache_key(Some(session_id.to_string()));
worker.set_cache_key(Some(segment_id.to_string()));
let mut pod = Self {
manifest,
worker: Some(worker),
store,
session_state: SessionState::new(session_id, 0),
session_state: SegmentState::new(segment_id, 0),
pwd: common.pwd,
scope: SharedScope::new(common.scope),
hook_builder: HookRegistryBuilder::new(),
@@ -2863,7 +2863,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
extract_pointer: Arc::new(Mutex::new(None)),
memory_task: None,
user_segments: Vec::new(),
sink: SessionLogSink::new(),
sink: SegmentLogSink::new(),
history_persistence_wired: false,
log_writer: None,
};
@@ -2878,13 +2878,13 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
/// Resolves the manifest cascade exactly like [`Self::from_manifest`]
/// (pwd / scope / pod-registry / client / prompt catalog), seeds a
/// fresh Worker from the source session's `RestoredState`, and
/// reuses the same `session_id` so subsequent turns append to the
/// reuses the same `segment_id` so subsequent turns append to the
/// source jsonl as a continuation of the same conversation.
///
/// Concurrent writers are prevented by the pod-registry:
/// the registration carries `session_id`, and this constructor
/// refuses to start when `pod_registry::lookup_session` already finds
/// a live Pod writing to `session_id`. So there is no need to fork —
/// the registration carries `segment_id`, and this constructor
/// refuses to start when `pod_registry::lookup_segment` already finds
/// a live Pod writing to `segment_id`. So there is no need to fork —
/// resume is "the same session, a different process owning it".
///
/// `system_prompt` is replayed verbatim from the session log —
@@ -2892,7 +2892,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
/// session keeps a stable cache prefix even when the manifest's
/// instruction template would render differently today.
pub async fn restore_from_manifest(
session_id: SessionId,
segment_id: SegmentId,
manifest: PodManifest,
store: St,
loader: PromptLoader,
@@ -2900,16 +2900,16 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
// Read raw entries once so we can both reconstruct state and
// seed the broadcast sink's mirror with the same prefix that
// sits on disk.
let raw_entries = store.read_all(session_id)?;
let raw_entries = store.read_all(segment_id)?;
let state = session_store::collect_state(&raw_entries);
if state.entries_count == 0 {
return Err(PodError::SessionEmpty { session_id });
return Err(PodError::SessionEmpty { segment_id });
}
let mirror_entries: Vec<LogEntry> = raw_entries.clone();
let scope_snapshot = state
.pod_scope
.clone()
.ok_or(PodError::SessionScopeMissing { session_id })?;
.ok_or(PodError::SessionScopeMissing { segment_id })?;
let mut common = prepare_pod_common_with_scope(
&manifest,
@@ -2923,7 +2923,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
let skill_shadows = std::mem::take(&mut common.skill_shadows);
// Atomic: register_pod inside install_top_level rejects when
// another live allocation already holds `session_id`. Wrapping
// another live allocation already holds `segment_id`. Wrapping
// the lookup + install inside a single `LockFileGuard` is what
// makes "no two live Pods write to the same session log"
// actually structural rather than a hopeful pre-check.
@@ -2937,14 +2937,14 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
socket_path,
common.scope.allow_rules(),
common.scope.deny_rules(),
session_id,
segment_id,
)?;
// Build the worker and apply the manifest defaults first, then
// overwrite the pieces the session log is authoritative for.
let mut worker = Worker::new(common.client);
apply_worker_manifest(&mut worker, &manifest.worker);
worker.set_cache_key(Some(session_id.to_string()));
worker.set_cache_key(Some(segment_id.to_string()));
if let Some(ref prompt) = state.system_prompt {
worker.set_system_prompt(prompt);
}
@@ -2974,7 +2974,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
manifest,
worker: Some(worker),
store,
session_state: SessionState::new(session_id, state.entries_count),
session_state: SegmentState::new(segment_id, state.entries_count),
pwd: common.pwd,
scope: SharedScope::new(common.scope),
hook_builder: HookRegistryBuilder::new(),
@@ -3007,7 +3007,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
// Seed the mirror with the entries we just replayed so a
// late-attaching client sees the full prefix without an
// extra round trip.
sink: SessionLogSink::with_initial(mirror_entries),
sink: SegmentLogSink::with_initial(mirror_entries),
history_persistence_wired: false,
log_writer: None,
};
@@ -3234,13 +3234,13 @@ pub enum PodError {
#[error("workflow invocation failed: {0}")]
WorkflowResolve(#[from] WorkflowResolveError),
#[error("session {session_id} has no entries to restore")]
SessionEmpty { session_id: SessionId },
#[error("session {segment_id} has no entries to restore")]
SessionEmpty { segment_id: SegmentId },
#[error(
"session {session_id} has no persisted scope snapshot; refusing resume without explicit scope"
"session {segment_id} has no persisted scope snapshot; refusing resume without explicit scope"
)]
SessionScopeMissing { session_id: SessionId },
SessionScopeMissing { segment_id: SegmentId },
}
/// Bundle of resources that every high-level Pod constructor needs:
crates/pod/src/runtime/dir.rs
+1 -1
View File
@@ -131,7 +131,7 @@ mod tests {
fn test_state() -> PodSharedState {
PodSharedState::new(
"test-pod".into(),
session_store::new_session_id(),
session_store::new_segment_id(),
"[pod]\nname = \"test-pod\"".into(),
protocol::Greeting {
pod_name: "test-pod".into(),
crates/pod/src/session_log_sink.rs → crates/pod/src/segment_log_sink.rs
+24 -24
View File
@@ -10,11 +10,11 @@
//! Atomicity contract:
//!
//! 1. Pod writes the entry to disk via the `Store`.
//! 2. Pod calls [`SessionLogSink::publish`] which acquires the mirror
//! 2. Pod calls [`SegmentLogSink::publish`] which acquires the mirror
//! mutex, pushes the entry, and fires `broadcast::send` — all under
//! the same critical section.
//!
//! [`SessionLogSink::subscribe_with_snapshot`] takes the same mutex,
//! [`SegmentLogSink::subscribe_with_snapshot`] takes the same mutex,
//! so the `(snapshot, receiver)` pair returned to a connecting client
//! splits the entry sequence cleanly: every entry shows up in exactly
//! one of `snapshot` or on `receiver`.
@@ -39,24 +39,24 @@ const BROADCAST_CAPACITY: usize = 256;
/// for read-only `subscribe_with_snapshot` access and keeps one for
/// its own write path.
#[derive(Clone)]
pub struct SessionLogSink {
pub struct SegmentLogSink {
inner: Arc<SinkInner>,
}
struct SinkInner {
/// Full session log mirror in commit order. Reset on session swap
/// (compaction / fork) via [`SessionLogSink::reset_with_initial`].
/// (compaction / fork) via [`SegmentLogSink::reset_with_initial`].
mirror: StdMutex<Vec<LogEntry>>,
/// Broadcast channel for live entry updates. The same `Sender`
/// survives session swaps so existing subscribers keep their
/// receiver — they observe the swap as a freshly broadcast
/// `LogEntry::SessionStart` and reset their view accordingly.
/// `LogEntry::SegmentStart` and reset their view accordingly.
broadcast_tx: broadcast::Sender<LogEntry>,
}
impl SessionLogSink {
impl SegmentLogSink {
/// Create a fresh sink with an empty mirror. Used before any entry
/// has been written (deferred SessionStart) or as a placeholder in
/// has been written (deferred SegmentStart) or as a placeholder in
/// tests.
pub fn new() -> Self {
let (broadcast_tx, _) = broadcast::channel(BROADCAST_CAPACITY);
@@ -89,7 +89,7 @@ impl SessionLogSink {
///
/// Live broadcast fires only for entries that the streaming-event
/// lane does not cover:
/// - `LogEntry::SessionStart` → `Event::SessionRotated` on the wire.
/// - `LogEntry::SegmentStart` → `Event::SegmentRotated` on the wire.
/// - `LogEntry::SystemItem` → `Event::SystemItem`.
/// - `LogEntry::Invoke` → `Event::InvokeStart`.
/// Everything else (AssistantItem, ToolResult, UserInput, TurnEnd,
@@ -119,7 +119,7 @@ impl SessionLogSink {
fn is_live_relevant(entry: &LogEntry) -> bool {
matches!(
entry,
LogEntry::SessionStart { .. }
LogEntry::SegmentStart { .. }
| LogEntry::SystemItem { .. }
| LogEntry::Invoke { .. }
)
@@ -127,12 +127,12 @@ impl SessionLogSink {
/// Atomically swap the mirror to `[initial]` and broadcast the new
/// session-start entry. Used during compaction / fork: the new
/// `LogEntry::SessionStart` is the first entry of the replacement
/// `LogEntry::SegmentStart` is the first entry of the replacement
/// session, and existing subscribers transition by replaying it
/// like any other live entry.
///
/// Existing snapshot prefixes seen by old subscribers stay valid
/// for the prior session; the new `SessionStart` on the broadcast
/// for the prior session; the new `SegmentStart` on the broadcast
/// is the signal to reset their derived view.
pub fn reset_with_initial(&self, initial: LogEntry) {
let mut mirror = self
@@ -188,7 +188,7 @@ impl SessionLogSink {
}
}
impl Default for SessionLogSink {
impl Default for SegmentLogSink {
fn default() -> Self {
Self::new()
}
@@ -198,10 +198,10 @@ impl Default for SessionLogSink {
mod tests {
use super::*;
use llm_worker::llm_client::RequestConfig;
use session_store::session_log::now_millis;
use session_store::segment_log::now_millis;
fn session_start() -> LogEntry {
LogEntry::SessionStart {
LogEntry::SegmentStart {
ts: now_millis(),
system_prompt: None,
config: RequestConfig::default(),
@@ -220,13 +220,13 @@ mod tests {
#[test]
fn publish_then_subscribe_returns_history_in_snapshot() {
let sink = SessionLogSink::new();
let sink = SegmentLogSink::new();
sink.publish(session_start());
sink.publish(turn_end(1));
let (snapshot, mut rx) = sink.subscribe_with_snapshot();
assert_eq!(snapshot.len(), 2);
assert!(matches!(snapshot[0], LogEntry::SessionStart { .. }));
assert!(matches!(snapshot[0], LogEntry::SegmentStart { .. }));
assert!(matches!(
snapshot[1],
LogEntry::TurnEnd { turn_count: 1, .. }
@@ -246,7 +246,7 @@ mod tests {
#[test]
fn subscribe_then_publish_delivers_only_live_relevant_entries() {
let sink = SessionLogSink::new();
let sink = SegmentLogSink::new();
sink.publish(session_start());
let (snapshot, mut rx) = sink.subscribe_with_snapshot();
@@ -270,7 +270,7 @@ mod tests {
#[test]
fn snapshot_and_live_never_overlap() {
let sink = SessionLogSink::new();
let sink = SegmentLogSink::new();
sink.publish(session_start());
let (snapshot, mut rx) = sink.subscribe_with_snapshot();
sink.publish(notification_entry("post-snapshot"));
@@ -285,7 +285,7 @@ mod tests {
#[test]
fn reset_with_initial_clears_and_broadcasts() {
let sink = SessionLogSink::new();
let sink = SegmentLogSink::new();
sink.publish(session_start());
sink.publish(turn_end(1));
@@ -293,18 +293,18 @@ mod tests {
sink.reset_with_initial(session_start());
match rx.try_recv() {
Ok(LogEntry::SessionStart { .. }) => {}
other => panic!("expected SessionStart broadcast, got {other:?}"),
Ok(LogEntry::SegmentStart { .. }) => {}
other => panic!("expected SegmentStart broadcast, got {other:?}"),
}
let (post_snapshot, _) = sink.subscribe_with_snapshot();
assert_eq!(post_snapshot.len(), 1);
assert!(matches!(post_snapshot[0], LogEntry::SessionStart { .. }));
assert!(matches!(post_snapshot[0], LogEntry::SegmentStart { .. }));
}
#[test]
fn replace_silent_does_not_broadcast() {
let sink = SessionLogSink::new();
let sink = SegmentLogSink::new();
sink.publish(session_start());
let (_pre_snapshot, mut rx) = sink.subscribe_with_snapshot();
@@ -318,7 +318,7 @@ mod tests {
#[test]
fn with_initial_seeds_the_mirror() {
let sink = SessionLogSink::with_initial(vec![session_start(), turn_end(1)]);
let sink = SegmentLogSink::with_initial(vec![session_start(), turn_end(1)]);
let (snapshot, _) = sink.subscribe_with_snapshot();
assert_eq!(snapshot.len(), 2);
}
crates/pod/src/shared_state.rs
+7 -7
View File
@@ -2,7 +2,7 @@ use std::sync::{OnceLock, RwLock};
use protocol::PodStatus;
use serde_json::json;
use session_store::SessionId;
use session_store::SegmentId;
use crate::fs_view::PodFsView;
@@ -28,7 +28,7 @@ pub struct KnowledgeCandidate {
/// greeting, and completion lookup hubs.
pub struct PodSharedState {
pub pod_name: String,
pub session_id: SessionId,
pub segment_id: SegmentId,
pub manifest_toml: String,
pub greeting: protocol::Greeting,
pub status: RwLock<PodStatus>,
@@ -46,13 +46,13 @@ pub struct PodSharedState {
impl PodSharedState {
pub fn new(
pod_name: String,
session_id: SessionId,
segment_id: SegmentId,
manifest_toml: String,
greeting: protocol::Greeting,
) -> Self {
Self {
pod_name,
session_id,
segment_id,
manifest_toml,
greeting,
status: RwLock::new(PodStatus::Idle),
@@ -123,7 +123,7 @@ impl PodSharedState {
let status = self.get_status();
json!({
"state": status,
"session_id": self.session_id.to_string(),
"segment_id": self.segment_id.to_string(),
"pod_name": self.pod_name,
})
.to_string()
@@ -137,7 +137,7 @@ mod tests {
fn test_state() -> PodSharedState {
PodSharedState::new(
"test-pod".into(),
session_store::new_session_id(),
session_store::new_segment_id(),
"[pod]\nname = \"test-pod\"".into(),
test_greeting(),
)
@@ -176,7 +176,7 @@ mod tests {
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
assert_eq!(parsed["state"], "idle");
assert_eq!(parsed["pod_name"], "test-pod");
assert!(parsed["session_id"].is_string());
assert!(parsed["segment_id"].is_string());
}
#[test]
crates/pod/src/spawn/comm_tools.rs
+2 -2
View File
@@ -430,13 +430,13 @@ fn extract_assistant_text(entries: &[serde_json::Value]) -> String {
for value in entries {
// The wire payload is the JSON form of `session_store::LogEntry`.
// Walk Assistant items inside each entry that can carry them:
// post-compaction `SessionStart.history` (seed) and per-LLM-call
// post-compaction `SegmentStart.history` (seed) and per-LLM-call
// `AssistantItems` deltas.
let Ok(entry) = serde_json::from_value::<LogEntry>(value.clone()) else {
continue;
};
let logged_items = match entry {
LogEntry::SessionStart { history, .. } => history,
LogEntry::SegmentStart { history, .. } => history,
LogEntry::AssistantItems { items, .. } => items,
_ => continue,
};
crates/pod/tests/compact_events_test.rs
+12 -12
View File
@@ -178,7 +178,7 @@ fn drain(rx: &mut broadcast::Receiver<Event>) -> Vec<Event> {
}
/// Collect every system-message text that the post-compaction
/// `SessionStart.history` carries, by reading the sink mirror directly.
/// `SegmentStart.history` carries, by reading the sink mirror directly.
fn system_texts_in_sink_session_start(
pod: &pod::Pod<
impl llm_worker::llm_client::client::LlmClient + Clone + 'static,
@@ -187,7 +187,7 @@ fn system_texts_in_sink_session_start(
) -> Vec<String> {
let (entries, _rx) = pod.sink().subscribe_with_snapshot();
for entry in entries.into_iter().rev() {
if let session_store::LogEntry::SessionStart { history, .. } = entry {
if let session_store::LogEntry::SegmentStart { history, .. } = entry {
return history
.into_iter()
.filter_map(|logged| {
@@ -229,7 +229,7 @@ async fn compact_emits_session_start_carrying_summary_and_task_snapshot() {
pod.compact(10_000).await.unwrap();
let system_texts = system_texts_in_sink_session_start(&pod);
// The post-compaction `SessionStart.history` carries the new system
// The post-compaction `SegmentStart.history` carries the new system
// messages introduced by the compactor. Clients re-seed their view
// from this entry alone, so it is the load-bearing payload.
assert!(
@@ -289,11 +289,11 @@ async fn pre_run_compact_success_broadcasts_start_and_done() {
// CompactDone carries the new session id.
let new_id_in_event = events.iter().find_map(|e| match e {
Event::CompactDone { new_session_id } => Some(*new_session_id),
Event::CompactDone { new_segment_id } => Some(*new_segment_id),
_ => None,
});
assert!(new_id_in_event.is_some(), "CompactDone missing");
assert_eq!(new_id_in_event.unwrap(), pod.session_id());
assert_eq!(new_id_in_event.unwrap(), pod.segment_id());
}
#[tokio::test]
@@ -345,10 +345,10 @@ async fn mid_turn_compact_success_broadcasts_start_and_done() {
);
let new_id_in_event = events.iter().find_map(|e| match e {
Event::CompactDone { new_session_id } => Some(*new_session_id),
Event::CompactDone { new_segment_id } => Some(*new_segment_id),
_ => None,
});
assert_eq!(new_id_in_event, Some(pod.session_id()));
assert_eq!(new_id_in_event, Some(pod.segment_id()));
}
/// Regression: `Pod::compact()` must reset the in-memory
@@ -520,7 +520,7 @@ async fn pre_run_compact_failure_broadcasts_start_and_failed() {
// ---------------------------------------------------------------------------
// Detached post-run memory jobs (`spawn_post_run_memory_jobs` /
// `wait_for_memory_jobs`). Covers the detach round-trip and the structural
// invariant that the cloned memory-task Pod shares `SessionState` with the
// invariant that the cloned memory-task Pod shares `SegmentState` with the
// source Pod, so that `save_extension` from the background extract does not
// leave the next turn's `save_user_input` looking at a stale session pointer.
@@ -570,7 +570,7 @@ async fn spawn_and_wait_drives_extract_to_completion() {
#[tokio::test]
async fn detached_extract_does_not_fork_session_log() {
// Source pod and the cloned memory-task pod share `SessionState` via
// Source pod and the cloned memory-task pod share `SegmentState` via
// `Arc<_>`. The detached extract advances the entry tally through
// `save_extension`; the next `run` must see that same tally so
// `ensure_head_or_fork` does not spawn a new session.
@@ -583,18 +583,18 @@ async fn detached_extract_does_not_fork_session_log() {
let mut pod = make_pod_with_manifest(EXTRACT_NO_COMPACT_MANIFEST, client).await;
pod.run_text("first").await.unwrap();
let session_before = pod.session_id();
let session_before = pod.segment_id();
pod.spawn_post_run_memory_jobs();
pod.wait_for_memory_jobs().await;
pod.run_text("second").await.unwrap();
let session_after = pod.session_id();
let session_after = pod.segment_id();
assert_eq!(
session_before, session_after,
"detached extract's save_extension and the next turn's save_user_input \
must share the entry tally through SessionState — a fork here means the \
must share the entry tally through SegmentState — a fork here means the \
clone carried its own counter"
);
}
crates/pod/tests/consolidation_test.rs
+1 -1
View File
@@ -172,7 +172,7 @@ fn write_n_staging(layout: &WorkspaceLayout, n: usize) -> Vec<uuid::Uuid> {
let (id, _) = write_staging(
layout,
SourceRef {
session_id: format!("s-{i}"),
segment_id: format!("s-{i}"),
range: [i as u64, i as u64],
},
ExtractedPayload::default(),
crates/pod/tests/controller_test.rs
+1 -1
View File
@@ -22,7 +22,7 @@ fn history_from_sink(handle: &PodHandle) -> Vec<Item> {
let mut items = Vec::new();
for entry in entries {
match entry {
LogEntry::SessionStart { history, .. } => {
LogEntry::SegmentStart { history, .. } => {
items.extend(history.into_iter().map(Item::from));
}
LogEntry::UserInput { segments, .. } => {
crates/pod/tests/pod_comm_tools_test.rs
+1 -1
View File
@@ -349,7 +349,7 @@ async fn stop_pod_sends_shutdown_and_releases_scope() {
permission: Permission::Write,
recursive: true,
}],
session_store::new_session_id(),
session_store::new_segment_id(),
)
.unwrap();
pod_registry::delegate_scope(
crates/pod/tests/pod_events_test.rs
+1 -1
View File
@@ -358,7 +358,7 @@ async fn shutdown_releases_scope_allocation_when_present() {
std::process::id(),
"/tmp/kid.sock".into(),
vec![],
session_store::new_session_id(),
session_store::new_segment_id(),
)
.unwrap();
std::mem::forget(guard);
crates/pod/tests/restore_test.rs
+8 -8
View File
@@ -8,7 +8,7 @@
use std::sync::{LazyLock, Mutex};
use pod::{Pod, PodError};
use session_store::{FsStore, SessionId, StoreError};
use session_store::{FsStore, SegmentId, StoreError};
const MINIMAL_MANIFEST_TOML: &str = r#"
[pod]
@@ -42,7 +42,7 @@ async fn restore_from_manifest_rejects_unknown_session() {
// A freshly-minted id with no jsonl file at all → store returns
// NotFound, which `Pod::restore_from_manifest` surfaces verbatim
// as `PodError::Store`.
let unknown = session_store::new_session_id();
let unknown = session_store::new_segment_id();
let result =
Pod::restore_from_manifest(unknown, manifest, store, pod::PromptLoader::builtins_only())
.await;
@@ -67,7 +67,7 @@ async fn restore_from_manifest_rejects_empty_session_log() {
// `restore_from_manifest` rejects with `SessionEmpty` *before* it
// gets as far as building the LLM client — so the test does not
// need credentials or a runtime sandbox.
let id: SessionId = session_store::new_session_id();
let id: SegmentId = session_store::new_segment_id();
let path = store_tmp.path().join(format!("{id}.jsonl"));
std::fs::write(&path, b"").unwrap();
@@ -75,7 +75,7 @@ async fn restore_from_manifest_rejects_empty_session_log() {
Pod::restore_from_manifest(id, manifest, store, pod::PromptLoader::builtins_only()).await;
match result {
Err(PodError::SessionEmpty { session_id }) => assert_eq!(session_id, id),
Err(PodError::SessionEmpty { segment_id }) => assert_eq!(segment_id, id),
Err(other) => panic!("expected SessionEmpty, got {other:?}"),
Ok(_) => panic!("expected empty session log to fail"),
}
@@ -89,19 +89,19 @@ async fn restore_from_manifest_rejects_session_without_scope_snapshot() {
let store = FsStore::new(store_tmp.path()).unwrap();
let manifest = pod::PodManifest::from_toml(MINIMAL_MANIFEST_TOML).unwrap();
let id = session_store::new_session_id();
let state = session_store::SessionStartState {
let id = session_store::new_segment_id();
let state = session_store::SegmentStartState {
system_prompt: None,
config: &Default::default(),
history: &[],
};
session_store::create_session_with_id(&store, id, state).unwrap();
session_store::create_segment_with_id(&store, id, state).unwrap();
let result =
Pod::restore_from_manifest(id, manifest, store, pod::PromptLoader::builtins_only()).await;
match result {
Err(PodError::SessionScopeMissing { session_id }) => assert_eq!(session_id, id),
Err(PodError::SessionScopeMissing { segment_id }) => assert_eq!(segment_id, id),
Err(other) => panic!("expected SessionScopeMissing, got {other:?}"),
Ok(_) => panic!("expected missing scope snapshot to fail"),
}
crates/pod/tests/session_metrics_test.rs
+18 -18
View File
@@ -26,7 +26,7 @@ use llm_worker::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEve
use llm_worker::llm_client::{ClientError, LlmClient, Request};
use llm_worker::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
use session_metrics::{DOMAIN, Metric, metrics_from_extensions};
use session_store::{FsStore, LogEntry, SessionId, Store, StoreError, TraceEntry};
use session_store::{FsStore, LogEntry, SegmentId, Store, StoreError, TraceEntry};
use pod::{Pod, PodManifest};
@@ -200,7 +200,7 @@ async fn prune_metrics_emit_skip_then_fire_with_post_request_join() {
text_response_with_cache("done", 1234, 50),
]);
let (mut pod, _store_tmp, _pwd_tmp) = make_pod(manifest_toml(1, 1), client, "big_tool").await;
let session_id = pod.session_id();
let segment_id = pod.segment_id();
// Cloning the store handle to read the session log back after the
// runs complete — the Pod retains its own copy.
let store = pod.store().clone();
@@ -208,7 +208,7 @@ async fn prune_metrics_emit_skip_then_fire_with_post_request_join() {
pod.run_text("first").await.unwrap();
pod.run_text("second").await.unwrap();
let state = session_store::restore(&store, session_id).unwrap();
let state = session_store::restore(&store, segment_id).unwrap();
let metrics = metrics_from_extensions(&state.extensions);
// Run 1 has 2 LLM iterations (tool loop), each evaluates prune with
@@ -288,13 +288,13 @@ async fn prune_metrics_record_below_min_savings_skip() {
]);
let (mut pod, _store_tmp, _pwd_tmp) =
make_pod(manifest_toml(1, u64::MAX), client, "big_tool").await;
let session_id = pod.session_id();
let segment_id = pod.segment_id();
let store = pod.store().clone();
pod.run_text("first").await.unwrap();
pod.run_text("second").await.unwrap();
let state = session_store::restore(&store, session_id).unwrap();
let state = session_store::restore(&store, segment_id).unwrap();
let metrics = metrics_from_extensions(&state.extensions);
let below = metrics
.iter()
@@ -327,7 +327,7 @@ struct MetricFailingStore {
}
impl Store for MetricFailingStore {
fn append(&self, id: SessionId, entry: &LogEntry) -> Result<(), StoreError> {
fn append(&self, id: SegmentId, entry: &LogEntry) -> Result<(), StoreError> {
if let LogEntry::Extension { domain, .. } = entry {
if domain == DOMAIN {
return Err(StoreError::Io(std::io::Error::other("synthetic failure")));
@@ -335,22 +335,22 @@ impl Store for MetricFailingStore {
}
self.inner.append(id, entry)
}
fn read_all(&self, id: SessionId) -> Result<Vec<LogEntry>, StoreError> {
fn read_all(&self, id: SegmentId) -> Result<Vec<LogEntry>, StoreError> {
self.inner.read_all(id)
}
fn list_sessions(&self) -> Result<Vec<SessionId>, StoreError> {
self.inner.list_sessions()
fn list_segments(&self) -> Result<Vec<SegmentId>, StoreError> {
self.inner.list_segments()
}
fn create_session(&self, id: SessionId, entries: &[LogEntry]) -> Result<(), StoreError> {
self.inner.create_session(id, entries)
fn create_segment(&self, id: SegmentId, entries: &[LogEntry]) -> Result<(), StoreError> {
self.inner.create_segment(id, entries)
}
fn exists(&self, id: SessionId) -> Result<bool, StoreError> {
fn exists(&self, id: SegmentId) -> Result<bool, StoreError> {
self.inner.exists(id)
}
fn read_entry_count(&self, id: SessionId) -> Result<usize, StoreError> {
fn read_entry_count(&self, id: SegmentId) -> Result<usize, StoreError> {
self.inner.read_entry_count(id)
}
fn append_trace(&self, id: SessionId, entry: &TraceEntry) -> Result<(), StoreError> {
fn append_trace(&self, id: SegmentId, entry: &TraceEntry) -> Result<(), StoreError> {
self.inner.append_trace(id, entry)
}
}
@@ -386,12 +386,12 @@ async fn metric_write_failure_emits_warn_alert_and_does_not_abort_run() {
let alerter = pod::Alerter::new(tx);
pod.attach_alerter(alerter);
let session_id = pod.session_id();
let segment_id = pod.segment_id();
// Run completes successfully despite metric failure.
pod.run_text("hello").await.unwrap();
// No metrics ended up in the log (writes were rejected).
let state = session_store::restore(&store, session_id).unwrap();
let state = session_store::restore(&store, segment_id).unwrap();
let metrics = metrics_from_extensions(&state.extensions);
assert!(metrics.is_empty(), "metrics must drop on write failure");
@@ -446,10 +446,10 @@ permission = "write"
let mut pod = Pod::new(manifest, worker, store.clone(), pwd, scope)
.await
.unwrap();
let session_id = pod.session_id();
let segment_id = pod.segment_id();
pod.run_text("hello").await.unwrap();
let state = session_store::restore(&store, session_id).unwrap();
let state = session_store::restore(&store, segment_id).unwrap();
let metrics = metrics_from_extensions(&state.extensions);
assert!(
metrics.is_empty(),
crates/pod/tests/spawn_pod_test.rs
+1 -1
View File
@@ -73,7 +73,7 @@ async fn setup_spawner(
permission: Permission::Write,
recursive: true,
}],
session_store::new_session_id(),
session_store::new_segment_id(),
)
.unwrap();
// Leak the guard — the spawner allocation needs to outlive the
crates/pod/tests/system_prompt_template_test.rs
+3 -3
View File
@@ -182,16 +182,16 @@ async fn session_start_state_captures_rendered_prompt() {
.unwrap();
pod.run_text("hi").await.unwrap();
let entries = pod.store().read_all(pod.session_id()).unwrap();
let entries = pod.store().read_all(pod.segment_id()).unwrap();
let first = entries.first().expect("at least one entry");
match first {
LogEntry::SessionStart { system_prompt, .. } => {
LogEntry::SegmentStart { system_prompt, .. } => {
let sp = system_prompt.as_deref().expect("system prompt set");
assert!(sp.starts_with("hello cwd="));
assert!(sp.contains(&pwd.display().to_string()));
assert!(sp.contains("## Working boundaries"));
}
other => panic!("expected SessionStart as first entry, got {other:?}"),
other => panic!("expected SegmentStart as first entry, got {other:?}"),
}
}