1use std::collections::BTreeMap;
2use std::path::Path;
3use std::sync::Arc;
4
5use crate::bus::SessionBus;
6pub use crate::store_id::{store_id_path, StoreId, StoreIdLock};
7use crate::wal::{EventBody, GenParams, PurgeMode, RebaseEdit, Wal, WalRecord};
8use crate::DaemonError;
9
10#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
11pub struct LedgerEntry {
12 pub name: String,
13 pub arguments: String,
14 pub state: u8,
15 pub cancel_requested: bool,
16 pub lease_deadline_unix_ms: Option<u64>,
17 pub late_effects: u32,
18}
19
20#[derive(Debug, Clone, PartialEq)]
21pub struct CommittedEvent {
22 pub event_id: u64,
23 pub epoch: u64,
24 pub ts_unix_ms: u64,
25 pub body: EventBody,
26}
27
28#[derive(Debug)]
29pub struct SessionState {
30 pub id: u64,
31 pub parent: Option<u64>,
32 pub base: u64,
33 pub params: GenParams,
34 pub epoch: u64,
35 pub tokens: Vec<u32>,
36 pub events: Vec<CommittedEvent>,
37 pub last_finish: u32,
38 pub open_tool_calls: BTreeMap<u64, (String, String)>,
39 pub ledger: BTreeMap<u64, LedgerEntry>,
40 pub open_permissions: BTreeMap<u64, (u64, String)>,
41 pub behavior_fp: Option<u64>,
42 pub meta_version: u64,
43 pub title: Option<String>,
44 pub archived: bool,
45 pub generation: u64,
46 pub purged: bool,
47 pub tombstone_ledger: BTreeMap<u64, String>,
48 pub rerooted: bool,
49 pub qos_class: u8,
50 pub batch_invariant: bool,
51 pub pin_deadline_unix_ms: u64,
52}
53
54impl SessionState {
55 pub fn next_event_id(&self) -> u64 {
56 self.events.len() as u64
57 }
58}
59
60pub struct SessionStore {
61 wal: Wal,
62 store_id: StoreId,
63 _held: Option<StoreIdLock>,
64 sessions: BTreeMap<u64, SessionState>,
65 next_session: u64,
66 bus: Arc<SessionBus>,
67}
68
69impl SessionStore {
70 pub fn ephemeral() -> SessionStore {
71 SessionStore {
72 wal: Wal::ephemeral(),
73 store_id: StoreId::mint(),
74 _held: None,
75 sessions: BTreeMap::new(),
76 next_session: 1,
77 bus: SessionBus::new(),
78 }
79 }
80
81 pub fn open(path: &Path) -> Result<SessionStore, DaemonError> {
82 let mut sessions: BTreeMap<u64, SessionState> = BTreeMap::new();
83 let id_path = store_id_path(path);
84 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
86 let id_lock = loop {
87 match StoreId::try_lock(&id_path)? {
88 Some(lock) => break Some(lock),
89 None if std::time::Instant::now() < deadline => std::thread::sleep(std::time::Duration::from_millis(20)),
90 None => break None,
91 }
92 };
93 let Some(id_lock) = id_lock else {
94 return Err(DaemonError::Io(std::io::Error::new(
95 std::io::ErrorKind::WouldBlock,
96 format!(
97 "the session log {} is already open in another session store (another \
98 server on the same sessions directory, or the same model loaded twice); \
99 stop that one first",
100 path.display()
101 ),
102 )));
103 };
104 let fresh_wal = match std::fs::metadata(path) {
105 Ok(m) => m.len() == 0,
106 Err(e) if e.kind() == std::io::ErrorKind::NotFound => true,
107 Err(e) => return Err(e.into()),
108 };
109 let store_id = StoreId::open(&id_path, fresh_wal, &id_lock)?;
110 let wal = Wal::open(path, |r| replay(&mut sessions, r))?;
111 let mut next_session = 1;
112 for id in sessions.keys() {
113 next_session = next_session.max(id + 1);
114 }
115 let mut store = SessionStore {
116 wal,
117 store_id,
118 _held: Some(id_lock),
119 sessions,
120 next_session,
121 bus: SessionBus::new(),
122 };
123 let ids: Vec<u64> = store
124 .sessions
125 .values()
126 .filter(|s| !s.purged)
127 .map(|s| s.id)
128 .collect();
129 for id in ids {
130 let new_epoch = store.sessions[&id].epoch + 1;
131 let record = WalRecord {
132 session: id,
133 event_id: store.sessions[&id].next_event_id(),
134 epoch: new_epoch,
135 ts_unix_ms: store.stamp(),
136 body: EventBody::EpochBump,
137 };
138 store.wal.append(&record)?;
139 replay(&mut store.sessions, record);
140 }
141 Ok(store)
142 }
143
144 pub fn create(
145 &mut self,
146 parent: Option<u64>,
147 params: GenParams,
148 ) -> Result<u64, DaemonError> {
149 if let Some(p) = parent {
150 self.session(p)?;
151 }
152 let id = self.next_session;
153 let record = WalRecord {
154 session: id,
155 event_id: 0,
156 epoch: 1,
157 ts_unix_ms: self.stamp(),
158 body: EventBody::Created { parent, params },
159 };
160 self.wal.append(&record)?;
161 self.next_session += 1;
162 replay(&mut self.sessions, record);
163 Ok(id)
164 }
165
166 pub fn fork(
167 &mut self,
168 parent: u64,
169 fork_at: u64,
170 params: Option<GenParams>,
171 ) -> Result<u64, DaemonError> {
172 let p = self.session(parent)?;
173 let max = p.next_event_id();
174 if fork_at == 0 || fork_at > max {
175 return Err(DaemonError::ForkPoint {
176 session: parent,
177 fork_at,
178 max,
179 });
180 }
181 let params = params.unwrap_or(p.params);
182 let id = self.next_session;
183 let record = WalRecord {
184 session: id,
185 event_id: fork_at,
186 epoch: 1,
187 ts_unix_ms: self.stamp(),
188 body: EventBody::Forked {
189 parent,
190 fork_at,
191 params,
192 },
193 };
194 self.wal.append(&record)?;
195 self.next_session += 1;
196 replay(&mut self.sessions, record);
197 Ok(id)
198 }
199
200 pub fn append(
201 &mut self,
202 session: u64,
203 text: Option<String>,
204 span: Vec<u32>,
205 ) -> Result<CommittedEvent, DaemonError> {
206 self.commit(session, EventBody::Appended { text, span })
207 }
208
209 pub fn append_message(
210 &mut self,
211 session: u64,
212 role: u32,
213 text: String,
214 span: Vec<u32>,
215 ) -> Result<CommittedEvent, DaemonError> {
216 self.commit(session, EventBody::Message { role, text, span })
217 }
218
219 pub fn commit_generation_prompt(
220 &mut self,
221 session: u64,
222 span: Vec<u32>,
223 ) -> Result<CommittedEvent, DaemonError> {
224 self.commit(session, EventBody::GenerationPrompt { span })
225 }
226
227 pub fn commit_tool_use(
228 &mut self,
229 session: u64,
230 name: String,
231 arguments: String,
232 ) -> Result<CommittedEvent, DaemonError> {
233 self.commit(session, EventBody::ToolUse { name, arguments })
234 }
235
236 pub fn commit_tool_result(
237 &mut self,
238 session: u64,
239 call_id: u64,
240 content: String,
241 span: Vec<u32>,
242 ) -> Result<CommittedEvent, DaemonError> {
243 let s = self.session(session)?;
244 if !s.open_tool_calls.contains_key(&call_id) {
245 return Err(DaemonError::UnknownToolCall(call_id));
246 }
247 self.commit(
248 session,
249 EventBody::ToolResult {
250 call_id,
251 content,
252 span,
253 },
254 )
255 }
256
257 pub fn commit_generated(
258 &mut self,
259 session: u64,
260 span: Vec<u32>,
261 text: String,
262 channel: u32,
263 finish: u32,
264 ) -> Result<CommittedEvent, DaemonError> {
265 self.commit(
266 session,
267 EventBody::Generated {
268 span,
269 text,
270 channel,
271 finish,
272 },
273 )
274 }
275
276 pub fn commit_generation_fingerprint(
277 &mut self,
278 session: u64,
279 digest: u64,
280 ) -> Result<Option<CommittedEvent>, DaemonError> {
281 if self.session(session)?.behavior_fp == Some(digest) {
282 return Ok(None);
283 }
284 self.commit(session, EventBody::GenerationFingerprint { digest })
285 .map(Some)
286 }
287
288 pub fn commit_tool_parse_failure(
289 &mut self,
290 session: u64,
291 raw: String,
292 ) -> Result<CommittedEvent, DaemonError> {
293 self.commit(session, EventBody::ToolParseFailure { raw })
294 }
295
296 fn commit(&mut self, session: u64, body: EventBody) -> Result<CommittedEvent, DaemonError> {
297 let s = self.session(session)?;
298 let record = WalRecord {
299 session,
300 event_id: s.next_event_id(),
301 epoch: s.epoch,
302 ts_unix_ms: self.stamp(),
303 body,
304 };
305 self.wal.append(&record)?;
306 replay(&mut self.sessions, record.clone());
307 let ev = CommittedEvent {
308 event_id: record.event_id,
309 epoch: record.epoch,
310 ts_unix_ms: record.ts_unix_ms,
311 body: record.body,
312 };
313 self.bus.publish_committed(session, &ev);
314 Ok(ev)
315 }
316
317 pub fn store_id(&self) -> StoreId {
318 self.store_id
319 }
320
321 pub fn wal_version(&self) -> u8 {
322 self.wal.version()
323 }
324
325 fn stamp(&self) -> u64 {
326 if self.wal.version() == 2 {
327 crate::wal::now_unix_ms()
328 } else {
329 0
330 }
331 }
332
333 pub fn bus(&self) -> Arc<SessionBus> {
334 Arc::clone(&self.bus)
335 }
336
337 pub fn session(&self, id: u64) -> Result<&SessionState, DaemonError> {
338 let s = self.sessions.get(&id).ok_or(DaemonError::UnknownSession(id))?;
339 if s.purged {
340 return Err(DaemonError::Purged(id));
341 }
342 Ok(s)
343 }
344
345 pub fn tombstone(&self, id: u64) -> Option<&SessionState> {
346 self.sessions.get(&id).filter(|s| s.purged)
347 }
348
349 pub fn read(&self, id: u64, cursor: u64) -> Result<&[CommittedEvent], DaemonError> {
350 let s = self.session(id)?;
351 let start = (cursor as usize).min(s.events.len());
352 Ok(&s.events[start..])
353 }
354
355 pub fn drop_session(&mut self, id: u64) -> Result<(), DaemonError> {
356 self.sessions
357 .remove(&id)
358 .map(|_| ())
359 .ok_or(DaemonError::UnknownSession(id))
360 }
361
362 pub fn session_ids(&self) -> Vec<u64> {
363 self.sessions
364 .values()
365 .filter(|s| !s.purged)
366 .map(|s| s.id)
367 .collect()
368 }
369
370 pub fn set_meta(
371 &mut self,
372 session: u64,
373 expected_version: u64,
374 title: Option<String>,
375 archived: Option<bool>,
376 ) -> Result<u64, DaemonError> {
377 let s = self.session(session)?;
378 if s.meta_version != expected_version {
379 return Err(DaemonError::MetaConflict {
380 session,
381 expected: expected_version,
382 actual: s.meta_version,
383 });
384 }
385 let version = expected_version + 1;
386 self.commit(
387 session,
388 EventBody::MetaUpdated {
389 version,
390 title,
391 archived,
392 },
393 )?;
394 Ok(version)
395 }
396
397 pub fn set_pin(&mut self, session: u64, deadline_unix_ms: u64) -> Result<CommittedEvent, DaemonError> {
398 self.session(session)?;
399 self.commit(session, EventBody::Pinned { deadline_unix_ms })
400 }
401
402 pub fn set_qos(
403 &mut self,
404 session: u64,
405 class: u8,
406 batch_invariant: bool,
407 ) -> Result<CommittedEvent, DaemonError> {
408 if class > crate::qos::BACKGROUND_AGENT {
409 return Err(DaemonError::Protocol("unknown QoS class"));
410 }
411 self.commit(
412 session,
413 EventBody::QosSet {
414 class,
415 batch_invariant,
416 },
417 )
418 }
419
420 pub fn rebase_branch(
421 &mut self,
422 parent: u64,
423 fork_at: u64,
424 params: Option<GenParams>,
425 edits: Vec<RebaseEdit>,
426 ) -> Result<u64, DaemonError> {
427 let p = self.session(parent)?;
428 let max = p.next_event_id();
429 if fork_at == 0 || fork_at > max {
430 return Err(DaemonError::ForkPoint {
431 session: parent,
432 fork_at,
433 max,
434 });
435 }
436 let params = params.unwrap_or(p.params);
437 let id = self.next_session;
438 let record = WalRecord {
439 session: id,
440 event_id: fork_at,
441 epoch: 1,
442 ts_unix_ms: self.stamp(),
443 body: EventBody::Rebased {
444 parent,
445 fork_at,
446 params,
447 edits,
448 },
449 };
450 self.wal.append(&record)?;
451 self.next_session += 1;
452 replay(&mut self.sessions, record);
453 Ok(id)
454 }
455
456 pub fn commit_block(
457 &mut self,
458 session: u64,
459 role: u32,
460 kind: u32,
461 payload: String,
462 span: Vec<u32>,
463 visibility_version: u32,
464 ) -> Result<CommittedEvent, DaemonError> {
465 self.commit(
466 session,
467 EventBody::Block {
468 role,
469 kind,
470 payload,
471 span,
472 visibility_version,
473 },
474 )
475 }
476
477 pub fn commit_permission_request(
478 &mut self,
479 session: u64,
480 call_id: u64,
481 text: String,
482 ) -> Result<CommittedEvent, DaemonError> {
483 let s = self.session(session)?;
484 if call_id != 0 && !s.open_tool_calls.contains_key(&call_id) {
485 return Err(DaemonError::UnknownToolCall(call_id));
486 }
487 self.commit(session, EventBody::PermissionRequest { call_id, text })
488 }
489
490 pub fn commit_permission_response(
491 &mut self,
492 session: u64,
493 request_id: u64,
494 granted: bool,
495 ) -> Result<CommittedEvent, DaemonError> {
496 let s = self.session(session)?;
497 if !s.open_permissions.contains_key(&request_id) {
498 return Err(DaemonError::UnknownPermission(request_id));
499 }
500 self.commit(
501 session,
502 EventBody::PermissionResponse {
503 request_id,
504 granted,
505 },
506 )
507 }
508
509 pub fn commit_tool_cancel_request(
510 &mut self,
511 session: u64,
512 call_id: u64,
513 ) -> Result<CommittedEvent, DaemonError> {
514 let s = self.session(session)?;
515 if !s.open_tool_calls.contains_key(&call_id) {
516 return Err(DaemonError::UnknownToolCall(call_id));
517 }
518 self.commit(session, EventBody::ToolCancelRequested { call_id })
519 }
520
521 pub fn commit_tool_outcome(
522 &mut self,
523 session: u64,
524 call_id: u64,
525 outcome: u8,
526 note: String,
527 ) -> Result<CommittedEvent, DaemonError> {
528 let s = self.session(session)?;
529 if !s.open_tool_calls.contains_key(&call_id) {
530 return Err(DaemonError::UnknownToolCall(call_id));
531 }
532 if outcome != crate::wal::tool_outcome::FAILED && outcome != crate::wal::tool_outcome::CANCELLED {
533 return Err(DaemonError::Protocol("unknown tool outcome"));
534 }
535 self.commit(
536 session,
537 EventBody::ToolOutcome {
538 call_id,
539 outcome,
540 note,
541 },
542 )
543 }
544
545 pub fn commit_tool_reconciliation(
546 &mut self,
547 session: u64,
548 call_id: u64,
549 note: String,
550 ) -> Result<CommittedEvent, DaemonError> {
551 let s = self.session(session)?;
552 if !s.ledger.contains_key(&call_id) {
553 return Err(DaemonError::UnknownToolCall(call_id));
554 }
555 self.commit(session, EventBody::ToolReconciliation { call_id, note })
556 }
557
558 pub fn commit_tool_lease(
559 &mut self,
560 session: u64,
561 call_id: u64,
562 deadline_unix_ms: u64,
563 ) -> Result<CommittedEvent, DaemonError> {
564 let s = self.session(session)?;
565 if !s.open_tool_calls.contains_key(&call_id) {
566 return Err(DaemonError::UnknownToolCall(call_id));
567 }
568 self.commit(
569 session,
570 EventBody::ToolLease {
571 call_id,
572 deadline_unix_ms,
573 },
574 )
575 }
576
577 pub fn expire_tool_call(
578 &mut self,
579 session: u64,
580 call_id: u64,
581 reason: &str,
582 ) -> Result<CommittedEvent, DaemonError> {
583 let s = self.session(session)?;
584 if !s.open_tool_calls.contains_key(&call_id) {
585 return Err(DaemonError::UnknownToolCall(call_id));
586 }
587 self.commit(
588 session,
589 EventBody::ToolExpired {
590 call_id,
591 reason: reason.to_string(),
592 },
593 )
594 }
595
596 pub fn children_of(&self, session: u64) -> Vec<u64> {
597 self.sessions
598 .values()
599 .filter(|c| c.parent == Some(session) && !c.purged)
600 .map(|c| c.id)
601 .collect()
602 }
603
604 pub fn descendants_of(&self, session: u64) -> Vec<u64> {
605 let mut out = Vec::new();
606 let mut stack = self.children_of(session);
607 while let Some(c) = stack.pop() {
608 out.push(c);
609 stack.extend(self.children_of(c));
610 }
611 out
612 }
613
614 pub fn purge(
615 &mut self,
616 session: u64,
617 expected_generation: u64,
618 mode: PurgeMode,
619 ) -> Result<Vec<u64>, DaemonError> {
620 let s = self.session(session)?;
621 if s.generation != expected_generation {
622 return Err(DaemonError::GenerationConflict {
623 session,
624 expected: expected_generation,
625 actual: s.generation,
626 });
627 }
628 if !s.open_tool_calls.is_empty() {
629 return Err(DaemonError::Protocol("purge requires a settled tool ledger"));
630 }
631 let mut purged = Vec::new();
632 match mode {
633 PurgeMode::Reroot => {
634 for c in self.children_of(session) {
635 let child = self.session(c)?;
636 let prefix: Vec<(u64, EventBody)> = child.events[..child.base as usize]
637 .iter()
638 .map(|e| (e.event_id, e.body.clone()))
639 .collect();
640 self.commit(
641 c,
642 EventBody::Rerooted {
643 purged_parent: session,
644 prefix,
645 },
646 )?;
647 }
648 }
649 PurgeMode::Cascade => {
650 let mut desc = self.descendants_of(session);
651 desc.reverse();
652 for d in desc {
653 let g = self.session(d)?.generation;
654 self.commit(
655 d,
656 EventBody::Purged {
657 generation: g + 1,
658 descendants: PurgeMode::Cascade,
659 },
660 )?;
661 purged.push(d);
662 }
663 }
664 }
665 self.commit(
666 session,
667 EventBody::Purged {
668 generation: expected_generation + 1,
669 descendants: mode,
670 },
671 )?;
672 purged.push(session);
673 Ok(purged)
674 }
675}
676
677pub fn ledger_state_name(state: u8) -> &'static str {
678 use crate::wal::ledger_state::*;
679 match state {
680 OPEN => "open",
681 CANCEL_REQUESTED => "cancel_requested",
682 SUCCEEDED => "closed",
683 FAILED => "failed",
684 CANCELLED => "cancelled",
685 EXPIRED => "expired",
686 CANCELLED_WITH_LATE_EFFECT => "cancelled_with_late_effect",
687 _ => "unknown",
688 }
689}
690
691fn replay(sessions: &mut BTreeMap<u64, SessionState>, r: WalRecord) {
692 match &r.body {
693 EventBody::Created { parent, params } => {
694 sessions.insert(
695 r.session,
696 SessionState {
697 id: r.session,
698 parent: *parent,
699 base: 0,
700 params: *params,
701 epoch: r.epoch,
702 tokens: Vec::new(),
703 events: Vec::new(),
704 last_finish: 0,
705 open_tool_calls: BTreeMap::new(),
706 ledger: BTreeMap::new(),
707 open_permissions: BTreeMap::new(),
708 behavior_fp: None,
709 meta_version: 0,
710 title: None,
711 archived: false,
712 generation: 0,
713 purged: false,
714 tombstone_ledger: BTreeMap::new(),
715 rerooted: false,
716 qos_class: crate::qos::FOREGROUND_AGENT,
717 batch_invariant: false,
718 pin_deadline_unix_ms: 0,
719 },
720 );
721 }
722 EventBody::Forked {
723 parent,
724 fork_at,
725 params,
726 }
727 | EventBody::Rebased {
728 parent,
729 fork_at,
730 params,
731 ..
732 } => {
733 let inherited: Vec<CommittedEvent> = sessions
734 .get(parent)
735 .map(|p| p.events[..(*fork_at as usize).min(p.events.len())].to_vec())
736 .unwrap_or_default();
737 let (qos_class, batch_invariant) = sessions
738 .get(parent)
739 .map(|p| (p.qos_class, p.batch_invariant))
740 .unwrap_or((crate::qos::FOREGROUND_AGENT, false));
741 let mut tokens = Vec::new();
742 let mut last_finish = 0;
743 let mut behavior_fp = None;
744 for e in &inherited {
745 match &e.body {
746 EventBody::Appended { span, .. }
747 | EventBody::Message { span, .. }
748 | EventBody::GenerationPrompt { span }
749 | EventBody::ToolResult { span, .. } => tokens.extend_from_slice(span),
750 EventBody::Generated { span, finish, .. } => {
751 tokens.extend_from_slice(span);
752 last_finish = *finish;
753 }
754 EventBody::GenerationFingerprint { digest } => behavior_fp = Some(*digest),
755 _ => {}
756 }
757 }
758 sessions.insert(
759 r.session,
760 SessionState {
761 id: r.session,
762 parent: Some(*parent),
763 base: *fork_at,
764 params: *params,
765 epoch: r.epoch,
766 tokens,
767 events: inherited,
768 last_finish,
769 open_tool_calls: BTreeMap::new(),
770 ledger: BTreeMap::new(),
771 open_permissions: BTreeMap::new(),
772 behavior_fp,
773 meta_version: 0,
774 title: None,
775 archived: false,
776 generation: 0,
777 purged: false,
778 tombstone_ledger: BTreeMap::new(),
779 rerooted: false,
780 qos_class,
781 batch_invariant,
782 pin_deadline_unix_ms: 0,
783 },
784 );
785 }
786 EventBody::QosSet {
787 class,
788 batch_invariant,
789 } => {
790 if let Some(s) = sessions.get_mut(&r.session) {
791 s.qos_class = *class;
792 s.batch_invariant = *batch_invariant;
793 }
794 }
795 EventBody::MetaUpdated {
796 version,
797 title,
798 archived,
799 } => {
800 if let Some(s) = sessions.get_mut(&r.session) {
801 s.meta_version = *version;
802 if let Some(t) = title {
803 s.title = Some(t.clone());
804 }
805 if let Some(a) = archived {
806 s.archived = *a;
807 }
808 }
809 }
810 EventBody::ToolExpired { call_id, .. } => {
811 if let Some(s) = sessions.get_mut(&r.session) {
812 s.open_tool_calls.remove(call_id);
813 if let Some(e) = s.ledger.get_mut(call_id) {
814 e.state = crate::wal::ledger_state::EXPIRED;
815 }
816 }
817 }
818 EventBody::Rerooted { prefix, .. } => {
819 if let Some(s) = sessions.get_mut(&r.session) {
820 if (s.events.len() as u64) < s.base {
821 let mut evs: Vec<CommittedEvent> = prefix
822 .iter()
823 .map(|(id, body)| CommittedEvent {
824 event_id: *id,
825 epoch: 0,
826 ts_unix_ms: 0,
827 body: body.clone(),
828 })
829 .collect();
830 evs.extend(std::mem::take(&mut s.events));
831 let mut tokens = Vec::new();
832 for e in &evs {
833 match &e.body {
834 EventBody::Appended { span, .. }
835 | EventBody::Message { span, .. }
836 | EventBody::GenerationPrompt { span }
837 | EventBody::ToolResult { span, .. }
838 | EventBody::Generated { span, .. } => tokens.extend_from_slice(span),
839 _ => {}
840 }
841 }
842 s.events = evs;
843 s.tokens = tokens;
844 }
845 }
846 }
847 EventBody::Purged {
848 generation,
849 descendants,
850 } => {
851 if let Some(s) = sessions.get_mut(&r.session) {
852 s.generation = *generation;
853 s.purged = true;
854 let mut ledger = BTreeMap::new();
855 for (id, e) in &s.ledger {
856 ledger.insert(*id, ledger_state_name(e.state).to_string());
857 }
858 s.tombstone_ledger = ledger;
859 s.events.clear();
860 s.tokens.clear();
861 s.open_tool_calls.clear();
862 s.title = None;
863 }
864 if *descendants == PurgeMode::Reroot {
865 for c in sessions.values_mut() {
866 if c.parent == Some(r.session) && !c.purged {
867 c.rerooted = true;
868 }
869 }
870 }
871 return;
872 }
873 EventBody::Appended { span, .. }
874 | EventBody::Message { span, .. }
875 | EventBody::GenerationPrompt { span }
876 | EventBody::ToolResult { span, .. }
877 | EventBody::Block { span, .. }
878 | EventBody::Generated { span, .. } => {
879 if let Some(s) = sessions.get_mut(&r.session) {
880 s.tokens.extend_from_slice(span);
881 s.epoch = s.epoch.max(r.epoch);
882 if let EventBody::Generated { finish, .. } = &r.body {
883 s.last_finish = *finish;
884 }
885 if let EventBody::ToolResult { call_id, .. } = &r.body {
886 s.open_tool_calls.remove(call_id);
887 if let Some(e) = s.ledger.get_mut(call_id) {
888 e.state = crate::wal::ledger_state::SUCCEEDED;
889 }
890 }
891 }
892 }
893 EventBody::ToolUse { name, arguments } => {
894 if let Some(s) = sessions.get_mut(&r.session) {
895 s.epoch = s.epoch.max(r.epoch);
896 s.open_tool_calls
897 .insert(r.event_id, (name.clone(), arguments.clone()));
898 s.ledger.insert(
899 r.event_id,
900 LedgerEntry {
901 name: name.clone(),
902 arguments: arguments.clone(),
903 state: crate::wal::ledger_state::OPEN,
904 cancel_requested: false,
905 lease_deadline_unix_ms: None,
906 late_effects: 0,
907 },
908 );
909 }
910 }
911 EventBody::PermissionRequest { call_id, text } => {
912 if let Some(s) = sessions.get_mut(&r.session) {
913 s.open_permissions
914 .insert(r.event_id, (*call_id, text.clone()));
915 }
916 }
917 EventBody::PermissionResponse { request_id, .. } => {
918 if let Some(s) = sessions.get_mut(&r.session) {
919 s.open_permissions.remove(request_id);
920 }
921 }
922 EventBody::ToolCancelRequested { call_id } => {
923 if let Some(s) = sessions.get_mut(&r.session) {
924 if let Some(e) = s.ledger.get_mut(call_id) {
925 e.cancel_requested = true;
926 if e.state == crate::wal::ledger_state::OPEN {
927 e.state = crate::wal::ledger_state::CANCEL_REQUESTED;
928 }
929 }
930 }
931 }
932 EventBody::ToolOutcome { call_id, outcome, .. } => {
933 if let Some(s) = sessions.get_mut(&r.session) {
934 s.open_tool_calls.remove(call_id);
935 if let Some(e) = s.ledger.get_mut(call_id) {
936 e.state = if *outcome == crate::wal::tool_outcome::CANCELLED {
937 crate::wal::ledger_state::CANCELLED
938 } else {
939 crate::wal::ledger_state::FAILED
940 };
941 }
942 }
943 }
944 EventBody::ToolReconciliation { call_id, .. } => {
945 if let Some(s) = sessions.get_mut(&r.session) {
946 if let Some(e) = s.ledger.get_mut(call_id) {
947 e.late_effects += 1;
948 if e.state == crate::wal::ledger_state::CANCELLED
949 || e.state == crate::wal::ledger_state::CANCELLED_WITH_LATE_EFFECT
950 {
951 e.state = crate::wal::ledger_state::CANCELLED_WITH_LATE_EFFECT;
952 }
953 }
954 }
955 }
956 EventBody::ToolLease {
957 call_id,
958 deadline_unix_ms,
959 } => {
960 if let Some(s) = sessions.get_mut(&r.session) {
961 if let Some(e) = s.ledger.get_mut(call_id) {
962 e.lease_deadline_unix_ms = Some(*deadline_unix_ms);
963 }
964 }
965 }
966 EventBody::Pinned { deadline_unix_ms } => {
967 if let Some(s) = sessions.get_mut(&r.session) {
968 s.pin_deadline_unix_ms = *deadline_unix_ms;
969 }
970 }
971 EventBody::EpochBump => {
972 if let Some(s) = sessions.get_mut(&r.session) {
973 s.epoch = s.epoch.max(r.epoch);
974 }
975 }
976 EventBody::GenerationFingerprint { digest } => {
977 if let Some(s) = sessions.get_mut(&r.session) {
978 s.behavior_fp = Some(*digest);
979 }
980 }
981 EventBody::ToolParseFailure { .. } => {}
982 }
983 if let Some(s) = sessions.get_mut(&r.session) {
984 if s.purged {
985 return;
986 }
987 s.events.push(CommittedEvent {
988 event_id: r.event_id,
989 epoch: r.epoch,
990 ts_unix_ms: r.ts_unix_ms,
991 body: r.body,
992 });
993 }
994}