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

update: 残存 Session 識別子の Segment 化（review follow-up）

Browse Source 
レビュー指摘の通り、次の session-grouping-introduce で新 SessionId が
入る前に名称衝突を避けるため取り残しを掃除。

- PodError::Session{Empty,ScopeMissing} → Segment{Empty,ScopeMissing}
- ScopeLockError::SessionConflict → SegmentConflict
- Pod.session_state / SegmentState.set_session_id 系
- source_session_id / prev_session_id / ensure_session_head / short_session
- pod_cli の "Session ID:" 表示
- fs_store の sessions ローカル変数
This commit is contained in:
Hare
2026-05-20 05:17:49 +09:00
parent 2d23673393
commit c2b55a498b
14 changed files with 135 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/pod-registry/src/error.rs
+1 -1
View File
@@ -30,7 +30,7 @@ pub enum ScopeLockError {
"session {segment_id} is already held by pod `{pod_name}` at {}",
.socket.display()
)]
SessionConflict {
SegmentConflict {
segment_id: SegmentId,
pod_name: String,
socket: PathBuf,
crates/pod-registry/src/lifecycle.rs
+3 -3
View File
@@ -131,7 +131,7 @@ pub fn update_segment(pod_name: &str, new_segment_id: SegmentId) -> Result<(), S
let mut guard = LockFileGuard::open(&lock_path)?;
if let Some(other) = guard.data().find_by_segment(new_segment_id) {
if other.pod_name != pod_name {
return Err(ScopeLockError::SessionConflict {
return Err(ScopeLockError::SegmentConflict {
segment_id: new_segment_id,
pod_name: other.pod_name.clone(),
socket: other.socket.clone(),
@@ -320,7 +320,7 @@ mod tests {
// `a` cannot adopt b's live session id.
let err = update_segment("a", s_b).unwrap_err();
match err {
ScopeLockError::SessionConflict {
ScopeLockError::SegmentConflict {
pod_name,
segment_id,
..
@@ -328,7 +328,7 @@ mod tests {
assert_eq!(pod_name, "b");
assert_eq!(segment_id, s_b);
}
other => panic!("expected SessionConflict, got {other:?}"),
other => panic!("expected SegmentConflict, got {other:?}"),
}
}
}
crates/pod-registry/src/mutate.rs
+4 -4
View File
@@ -63,7 +63,7 @@ pub fn register_pod_with_deny(
return Err(ScopeLockError::DuplicatePodName(pod_name));
}
if let Some(existing) = guard.data().find_by_segment(segment_id) {
return Err(ScopeLockError::SessionConflict {
return Err(ScopeLockError::SegmentConflict {
segment_id,
pod_name: existing.pod_name.clone(),
socket: existing.socket.clone(),
@@ -588,7 +588,7 @@ mod tests {
)
.unwrap();
// Second registration tries to grab the same segment_id under
// a different pod_name. Without the SessionConflict check both
// a different pod_name. Without the SegmentConflict check both
// would succeed and race on the same jsonl.
let err = register_pod(
&mut g,
@@ -600,7 +600,7 @@ mod tests {
)
.unwrap_err();
match err {
ScopeLockError::SessionConflict {
ScopeLockError::SegmentConflict {
segment_id,
pod_name,
..
@@ -608,7 +608,7 @@ mod tests {
assert_eq!(segment_id, shared_session);
assert_eq!(pod_name, "first");
}
other => panic!("expected SessionConflict, got {other:?}"),
other => panic!("expected SegmentConflict, got {other:?}"),
}
}
}
crates/pod/examples/pod_cli.rs
+1 -1
View File
@@ -76,7 +76,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
}
// 6. Session ID for potential restore
println!("\nSession ID: {}", pod.segment_id());
println!("\nSegment ID: {}", pod.segment_id());
Ok(())
}
crates/pod/src/main.rs
+2 -2
View File
@@ -185,8 +185,8 @@ async fn main() -> ExitCode {
return ExitCode::FAILURE;
}
}
} else if let Some(source_session_id) = cli.session {
match Pod::restore_from_manifest(source_session_id, manifest, store, loader).await {
} else if let Some(source_segment_id) = cli.session {
match Pod::restore_from_manifest(source_segment_id, manifest, store, loader).await {
Ok(p) => p,
Err(e) => {
eprintln!("error: failed to restore pod: {e}");
crates/pod/src/pod.rs
+32 -32
View File
@@ -51,7 +51,7 @@ use tokio::task::JoinHandle;
/// 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`.
/// concurrent writers in `ensure_segment_head`.
pub struct SegmentState {
segment_id: ArcSwap<SegmentId>,
entries_written: AtomicUsize,
@@ -69,7 +69,7 @@ impl SegmentState {
**self.segment_id.load()
}
pub fn set_session_id(&self, id: SegmentId) {
pub fn set_segment_id(&self, id: SegmentId) {
self.segment_id.store(Arc::new(id));
}
@@ -163,8 +163,8 @@ pub struct Pod<C: LlmClient, St: Store> {
store: St,
/// Shared session pointer. Source of truth for the Pod's current
/// `segment_id` and append tally. `self.segment_id()` is a thin
/// wrapper over `session_state.segment_id()`.
session_state: Arc<SegmentState>,
/// wrapper over `segment_state.segment_id()`.
segment_state: Arc<SegmentState>,
/// Absolute working directory of the Pod.
pwd: PathBuf,
/// Shared, atomically-swappable view of the Pod's resolved scope.
@@ -338,7 +338,7 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Pod<C, St> {
manifest: self.manifest.clone(),
worker: Some(worker),
store: self.store.clone(),
session_state: self.session_state.clone(),
segment_state: self.segment_state.clone(),
pwd: self.pwd.clone(),
scope: self.scope.clone(),
hook_builder: HookRegistryBuilder::new(),
@@ -382,7 +382,7 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Pod<C, St> {
pub fn log_writer_handle(&self) -> LogWriterHandle<St> {
LogWriterHandle {
store: self.store.clone(),
state: self.session_state.clone(),
state: self.segment_state.clone(),
sink: self.sink.clone(),
}
}
@@ -469,7 +469,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
pwd: PathBuf,
scope: Scope,
) -> Result<Self, PodError> {
// Segment creation is deferred to `ensure_session_head` at first
// Segment creation is deferred to `ensure_segment_head` at first
// run so a later-installed system-prompt template (see
// `set_system_prompt_template`) can be captured by `SegmentStart`.
let segment_id = session_store::new_segment_id();
@@ -478,7 +478,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
manifest,
worker: Some(worker),
store,
session_state: SegmentState::new(segment_id, 0),
segment_state: SegmentState::new(segment_id, 0),
pwd,
scope: SharedScope::new(scope),
hook_builder: HookRegistryBuilder::new(),
@@ -545,7 +545,7 @@ 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 segment_id(&self) -> SegmentId {
self.session_state.segment_id()
self.segment_state.segment_id()
}
/// The Pod's manifest.
@@ -604,7 +604,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// can restore the narrowed scope instead of reclaiming delegated
/// writes.
pub fn persist_scope_snapshot(&mut self) -> Result<(), StoreError> {
if self.session_state.entries_written() == 0 {
if self.segment_state.entries_written() == 0 {
return Ok(());
}
let snapshot = {
@@ -627,9 +627,9 @@ 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 segment_id = self.session_state.segment_id();
let segment_id = self.segment_state.segment_id();
self.store.append(segment_id, &entry)?;
self.session_state.increment_entries();
self.segment_state.increment_entries();
self.sink.publish(entry);
Ok(())
}
@@ -1143,7 +1143,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
self.ensure_interceptor_installed();
self.ensure_system_prompt_materialized()?;
self.cleanup_finished_memory_task();
self.ensure_session_head()?;
self.ensure_segment_head()?;
if self.should_pre_run_compact() {
self.join_memory_task().await;
}
@@ -1616,10 +1616,10 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
/// `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> {
fn ensure_segment_head(&mut self) -> Result<(), PodError> {
let w = self.worker.as_ref().unwrap();
let prev_session_id = self.session_state.segment_id();
let entries_written = self.session_state.entries_written();
let prev_segment_id = self.segment_state.segment_id();
let entries_written = self.segment_state.entries_written();
if entries_written == 0 {
let initial = LogEntry::SegmentStart {
ts: segment_log::now_millis(),
@@ -1636,7 +1636,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// Check store count + auto-fork if it drifted.
let store_count = self
.store
.read_entry_count(prev_session_id)
.read_entry_count(prev_segment_id)
.map_err(PodError::from)?;
if store_count == entries_written {
return Ok(());
@@ -1656,8 +1656,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
self.store
.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.segment_state.set_segment_id(fork_id);
self.segment_state.set_entries_written(1);
self.sink.reset_with_initial(entry);
if self.scope_allocation.is_some() {
pod_registry::update_segment(&self.manifest.pod.name, fork_id)?;
@@ -2145,7 +2145,7 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
// the broadcast sink so existing subscribers see the new
// `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 old_session_id = self.segment_state.segment_id();
let source_turn_count = self.worker.as_ref().unwrap().turn_count();
let w = self.worker.as_ref().unwrap();
let entry = LogEntry::SegmentStart {
@@ -2160,8 +2160,8 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
}),
};
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);
self.segment_state.set_segment_id(new_segment_id);
self.segment_state.set_entries_written(1);
let session_start = entry;
// Broadcast the SegmentStart through the sink. This atomically
// resets the mirror to `[SegmentStart]` so any subscriber
@@ -2435,12 +2435,12 @@ impl<C: LlmClient, St: Store> Pod<C, St> {
extract::ExtractedPayload::default()
});
let source_session_id = self.session_state.segment_id();
let source_segment_id = self.segment_state.segment_id();
let staging_id = if payload.is_empty() {
String::new()
} else {
let source = memory::schema::SourceRef {
segment_id: source_session_id.to_string(),
segment_id: source_segment_id.to_string(),
range: [start_entry as u64, end_entry as u64],
};
let (id, _) = extract::write_staging(&layout, source, payload)
@@ -2736,7 +2736,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
let skill_shadows = std::mem::take(&mut common.skill_shadows);
// Segment creation is deferred to the first run (see
// `ensure_session_head`) so the SegmentStart entry can capture
// `ensure_segment_head`) so the SegmentStart entry can capture
// the rendered system prompt, not the raw template source. The
// segment_id is allocated here so the pod-registry registration
// can record it from the start.
@@ -2765,7 +2765,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
manifest,
worker: Some(worker),
store,
session_state: SegmentState::new(segment_id, 0),
segment_state: SegmentState::new(segment_id, 0),
pwd: common.pwd,
scope: SharedScope::new(common.scope),
hook_builder: HookRegistryBuilder::new(),
@@ -2835,7 +2835,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
manifest,
worker: Some(worker),
store,
session_state: SegmentState::new(segment_id, 0),
segment_state: SegmentState::new(segment_id, 0),
pwd: common.pwd,
scope: SharedScope::new(common.scope),
hook_builder: HookRegistryBuilder::new(),
@@ -2903,13 +2903,13 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
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 { segment_id });
return Err(PodError::SegmentEmpty { segment_id });
}
let mirror_entries: Vec<LogEntry> = raw_entries.clone();
let scope_snapshot = state
.pod_scope
.clone()
.ok_or(PodError::SessionScopeMissing { segment_id })?;
.ok_or(PodError::SegmentScopeMissing { segment_id })?;
let mut common = prepare_pod_common_with_scope(
&manifest,
@@ -2974,7 +2974,7 @@ impl<St: Store> Pod<Box<dyn LlmClient>, St> {
manifest,
worker: Some(worker),
store,
session_state: SegmentState::new(segment_id, state.entries_count),
segment_state: SegmentState::new(segment_id, state.entries_count),
pwd: common.pwd,
scope: SharedScope::new(common.scope),
hook_builder: HookRegistryBuilder::new(),
@@ -3235,12 +3235,12 @@ pub enum PodError {
WorkflowResolve(#[from] WorkflowResolveError),
#[error("session {segment_id} has no entries to restore")]
SessionEmpty { segment_id: SegmentId },
SegmentEmpty { segment_id: SegmentId },
#[error(
"session {segment_id} has no persisted scope snapshot; refusing resume without explicit scope"
)]
SessionScopeMissing { segment_id: SegmentId },
SegmentScopeMissing { segment_id: SegmentId },
}
/// Bundle of resources that every high-level Pod constructor needs:
crates/pod/src/spawn/tool.rs
+1 -1
View File
@@ -482,7 +482,7 @@ fn pod_registry_err_to_tool(e: ScopeLockError) -> ToolError {
| ScopeLockError::WriteConflict { .. }
| ScopeLockError::DuplicatePodName(_)
| ScopeLockError::UnknownPod(_)
| ScopeLockError::SessionConflict { .. } => ToolError::InvalidArgument(e.to_string()),
| ScopeLockError::SegmentConflict { .. } => ToolError::InvalidArgument(e.to_string()),
ScopeLockError::Io(_) => ToolError::ExecutionFailed(e.to_string()),
}
}
crates/pod/tests/restore_test.rs
+5 -5
View File
@@ -64,7 +64,7 @@ async fn restore_from_manifest_rejects_empty_session_log() {
// Pre-create an empty `<id>.jsonl` so `read_all` succeeds with no
// entries. `collect_state` returns `entries_count = 0`, which
// `restore_from_manifest` rejects with `SessionEmpty` *before* it
// `restore_from_manifest` rejects with `SegmentEmpty` *before* it
// gets as far as building the LLM client — so the test does not
// need credentials or a runtime sandbox.
let id: SegmentId = session_store::new_segment_id();
@@ -75,8 +75,8 @@ 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 { segment_id }) => assert_eq!(segment_id, id),
Err(other) => panic!("expected SessionEmpty, got {other:?}"),
Err(PodError::SegmentEmpty { segment_id }) => assert_eq!(segment_id, id),
Err(other) => panic!("expected SegmentEmpty, got {other:?}"),
Ok(_) => panic!("expected empty session log to fail"),
}
}
@@ -101,8 +101,8 @@ async fn restore_from_manifest_rejects_session_without_scope_snapshot() {
Pod::restore_from_manifest(id, manifest, store, pod::PromptLoader::builtins_only()).await;
match result {
Err(PodError::SessionScopeMissing { segment_id }) => assert_eq!(segment_id, id),
Err(other) => panic!("expected SessionScopeMissing, got {other:?}"),
Err(PodError::SegmentScopeMissing { segment_id }) => assert_eq!(segment_id, id),
Err(other) => panic!("expected SegmentScopeMissing, got {other:?}"),
Ok(_) => panic!("expected missing scope snapshot to fail"),
}
}
crates/session-store/src/fs_store.rs
+4 -4
View File
@@ -80,7 +80,7 @@ impl Store for FsStore {
}
fn list_segments(&self) -> Result<Vec<SegmentId>, StoreError> {
let mut sessions = Vec::new();
let mut segments = Vec::new();
for entry in fs::read_dir(&self.root)? {
let entry = entry?;
let path = entry.path();
@@ -89,13 +89,13 @@ impl Store for FsStore {
if name.ends_with(".jsonl") && !name.ends_with(".trace.jsonl") {
let stem = name.trim_end_matches(".jsonl");
if let Ok(id) = stem.parse::<SegmentId>() {
sessions.push(id);
segments.push(id);
}
}
}
// UUID v7: lexicographic sort = chronological sort, newest first
sessions.sort_by(|a, b| b.cmp(a));
Ok(sessions)
segments.sort_by(|a, b| b.cmp(a));
Ok(segments)
}
fn create_segment(&self, id: SegmentId, entries: &[LogEntry]) -> Result<(), StoreError> {
crates/session-store/src/segment.rs
+2 -2
View File
@@ -60,7 +60,7 @@ pub fn create_segment_with_id(
pub fn create_compacted_segment(
store: &impl Store,
state: SegmentStartState<'_>,
source_session_id: SegmentId,
source_segment_id: SegmentId,
source_turn_count: usize,
) -> Result<SegmentId, StoreError> {
let segment_id = crate::new_segment_id();
@@ -71,7 +71,7 @@ pub fn create_compacted_segment(
history: to_logged(state.history),
forked_from: None,
compacted_from: Some(SegmentOrigin {
segment_id: source_session_id,
segment_id: source_segment_id,
at_turn_index: source_turn_count,
}),
};
crates/tui/src/picker.rs
+3 -3
View File
@@ -72,7 +72,7 @@ struct Row {
preview: String,
/// `Some(pod_name)` when a live Pod currently holds an allocation
/// for this session in `pods.json`. Picking such a row launches
/// `pod --session <UUID>` which will fail with `SessionConflict` —
/// `pod --session <UUID>` which will fail with `SegmentConflict` —
/// the badge warns the user up-front.
live_pod: Option<String>,
}
@@ -300,7 +300,7 @@ fn row_line(row: &Row, selected: bool) -> Line<'_> {
};
let mut spans = vec![
Span::raw(marker),
Span::styled(short_session(row.id), id_style),
Span::styled(short_segment(row.id), id_style),
Span::raw(" "),
];
if let Some(ref pod_name) = row.live_pod {
@@ -313,7 +313,7 @@ fn row_line(row: &Row, selected: bool) -> Line<'_> {
Line::from(spans)
}
fn short_session(id: SegmentId) -> String {
fn short_segment(id: SegmentId) -> String {
let s = id.to_string();
s.chars().take(8).collect()
}
crates/tui/src/spawn.rs
+2 -2
View File
@@ -445,7 +445,7 @@ fn draw_form(f: &mut Frame<'_>, form: &Form) {
.split(area);
let title_text = match form.resume_from {
Some(id) => format!("resume pod session: {}", short_session(id)),
Some(id) => format!("resume pod session: {}", short_segment(id)),
None => "spawn pod".to_string(),
};
let title = Paragraph::new(Line::from(vec![Span::styled(
@@ -473,7 +473,7 @@ fn draw_form(f: &mut Frame<'_>, form: &Form) {
/// First 8 hex digits of a UUID — short enough to skim, long enough
/// to disambiguate inside a 10-row picker.
pub(crate) fn short_session(id: SegmentId) -> String {
pub(crate) fn short_segment(id: SegmentId) -> String {
let s = id.to_string();
s.chars().take(8).collect()
}