1use std::collections::{BTreeMap, BTreeSet};
5
6use async_trait::async_trait;
7use chrono::{DateTime, Utc};
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10
11pub const SESSION_EVENT_FORMAT_VERSION: u32 = 1;
12
13#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
14pub struct SessionEvent {
15 pub session_id: String,
16 pub seq: u64,
17 pub occurred_at: DateTime<Utc>,
18 pub event: Event,
19}
20
21impl SessionEvent {
22 pub fn pending(session_id: impl Into<String>, event: Event) -> Self {
23 Self {
24 session_id: session_id.into(),
25 seq: 0,
26 occurred_at: Utc::now(),
27 event,
28 }
29 }
30
31 pub fn format_version(&self) -> u32 {
32 self.event.format_version()
33 }
34
35 pub fn event_type(&self) -> &str {
36 self.event.event_type()
37 }
38
39 pub fn ignorable(&self) -> bool {
40 self.event.ignorable()
41 }
42}
43
44#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
45#[serde(tag = "type", rename_all = "snake_case")]
46pub enum Event {
47 SessionCreated {
48 profile_revision_id: String,
49 },
50 SessionForked {
51 parent_session_id: String,
52 parent_seq: u64,
53 },
54 SessionDeleted {
55 reason: String,
56 },
57 InputQueued {
58 input_id: String,
59 run_id: String,
60 mode: DeliveryMode,
61 content: Vec<ContentBlock>,
62 explicit_skill: Option<String>,
63 },
64 InputClaimed {
65 input_id: String,
66 run_id: String,
67 },
68 InputCancelled {
69 input_id: String,
70 run_id: String,
71 error_code: String,
72 },
73 RunStarted {
74 run_id: String,
75 input_id: String,
76 },
77 RunWaiting {
78 run_id: String,
79 interaction_id: String,
80 },
81 RunResumed {
82 run_id: String,
83 interaction_id: String,
84 },
85 RunFinished {
86 run_id: String,
87 status: RunStatus,
88 error_code: Option<String>,
89 },
90 TurnStarted {
91 run_id: String,
92 turn: u32,
93 },
94 TurnFinished {
95 run_id: String,
96 turn: u32,
97 },
98 StepStarted {
99 run_id: String,
100 step: u32,
101 },
102 StepFinished {
103 run_id: String,
104 step: u32,
105 },
106 UserMessage {
107 run_id: String,
108 content: Vec<ContentBlock>,
109 },
110 AssistantDelta {
111 run_id: String,
112 step: u32,
113 attempt: u32,
114 content: String,
115 },
116 AssistantMessage {
117 run_id: String,
118 step: u32,
119 attempt: u32,
120 content: Vec<ContentBlock>,
121 },
122 AssistantToolCalls {
123 run_id: String,
124 step: u32,
125 content: Option<String>,
126 calls: Vec<RecordedToolCall>,
127 },
128 ToolCall {
129 run_id: String,
130 step: u32,
131 call_id: String,
132 tool: String,
133 arguments: Value,
134 },
135 ToolAuthorization {
136 run_id: String,
137 step: u32,
138 call_id: String,
139 status: ToolAuthorizationStatus,
140 reason: Option<String>,
141 },
142 ToolExecutionStarted {
143 run_id: String,
144 step: u32,
145 call_id: String,
146 },
147 ToolResult {
148 run_id: String,
149 step: u32,
150 call_id: String,
151 result: Value,
152 is_error: bool,
153 },
154 UsageRecorded {
155 run_id: String,
156 operation_id: String,
157 prompt_tokens: u64,
158 completion_tokens: u64,
159 #[serde(default)]
160 cost_units: u64,
161 },
162 RetryScheduled {
163 run_id: String,
164 attempt: u32,
165 delay_ms: u64,
166 reason: String,
167 },
168 ModelRequestPrepared {
169 run_id: String,
170 step: u32,
171 attempt: u32,
172 #[serde(default)]
173 provider_attempt_id: String,
174 #[serde(default)]
175 operation_id: String,
176 #[serde(default)]
177 reserved_prompt_tokens: u64,
178 #[serde(default)]
179 reserved_completion_tokens: u64,
180 request: Value,
181 prompt_sections: Value,
182 },
183 ContextInjected {
184 run_id: String,
185 step: u32,
186 contribution_id: String,
187 source: String,
188 version: String,
189 authority: String,
190 form: String,
191 content: Vec<ContentBlock>,
192 },
193 ModelAttemptFailed {
194 run_id: String,
195 step: u32,
196 attempt: u32,
197 error: String,
198 retryable: bool,
199 },
200 CompactionStarted {
201 run_id: String,
202 compaction_id: String,
203 source_through_seq: u64,
204 },
205 ToolResultsPruned {
206 run_id: String,
207 call_ids: Vec<String>,
208 },
209 SummaryReplaced {
210 run_id: String,
211 through_seq: u64,
212 summary: String,
213 compactor: String,
214 model: String,
215 },
216 CompactionFinished {
217 run_id: String,
218 compaction_id: String,
219 status: String,
220 error: Option<String>,
221 },
222 InteractionRequested {
223 run_id: String,
224 interaction_id: String,
225 kind: InteractionKind,
226 payload: Value,
227 },
228 InteractionResolved {
229 run_id: String,
230 interaction_id: String,
231 resolution: InteractionResolution,
232 payload: Value,
233 },
234 ChildSessionLinked {
235 run_id: String,
236 child_session_id: String,
237 provider: String,
238 },
239 Extension {
240 run_id: String,
241 plugin_id: String,
242 event_type: String,
243 payload: Value,
244 },
245 #[serde(skip)]
246 Opaque {
247 format_version: u32,
248 event_type: String,
249 ignorable: bool,
250 payload: Value,
251 },
252}
253
254impl Event {
255 pub fn format_version(&self) -> u32 {
256 match self {
257 Self::Opaque { format_version, .. } => *format_version,
258 _ => SESSION_EVENT_FORMAT_VERSION,
259 }
260 }
261
262 pub fn event_type(&self) -> &str {
263 match self {
264 Self::SessionCreated { .. } => "session_created",
265 Self::SessionForked { .. } => "session_forked",
266 Self::SessionDeleted { .. } => "session_deleted",
267 Self::InputQueued { .. } => "input_queued",
268 Self::InputClaimed { .. } => "input_claimed",
269 Self::InputCancelled { .. } => "input_cancelled",
270 Self::RunStarted { .. } => "run_started",
271 Self::RunWaiting { .. } => "run_waiting",
272 Self::RunResumed { .. } => "run_resumed",
273 Self::RunFinished { .. } => "run_finished",
274 Self::TurnStarted { .. } => "turn_started",
275 Self::TurnFinished { .. } => "turn_finished",
276 Self::StepStarted { .. } => "step_started",
277 Self::StepFinished { .. } => "step_finished",
278 Self::UserMessage { .. } => "user_message",
279 Self::AssistantDelta { .. } => "assistant_delta",
280 Self::AssistantMessage { .. } => "assistant_message",
281 Self::AssistantToolCalls { .. } => "assistant_tool_calls",
282 Self::ToolCall { .. } => "tool_call",
283 Self::ToolAuthorization { .. } => "tool_authorization",
284 Self::ToolExecutionStarted { .. } => "tool_execution_started",
285 Self::ToolResult { .. } => "tool_result",
286 Self::UsageRecorded { .. } => "usage_recorded",
287 Self::RetryScheduled { .. } => "retry_scheduled",
288 Self::ModelRequestPrepared { .. } => "model_request_prepared",
289 Self::ContextInjected { .. } => "context_injected",
290 Self::ModelAttemptFailed { .. } => "model_attempt_failed",
291 Self::CompactionStarted { .. } => "compaction_started",
292 Self::ToolResultsPruned { .. } => "tool_results_pruned",
293 Self::SummaryReplaced { .. } => "summary_replaced",
294 Self::CompactionFinished { .. } => "compaction_finished",
295 Self::InteractionRequested { .. } => "interaction_requested",
296 Self::InteractionResolved { .. } => "interaction_resolved",
297 Self::ChildSessionLinked { .. } => "child_session_linked",
298 Self::Extension { .. } => "extension",
299 Self::Opaque { event_type, .. } => event_type,
300 }
301 }
302
303 pub fn ignorable(&self) -> bool {
304 match self {
305 Self::Opaque { ignorable, .. } => *ignorable,
306 _ => false,
307 }
308 }
309}
310
311#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
312#[serde(rename_all = "snake_case")]
313pub enum DeliveryMode {
314 Followup,
315 Steer,
316 Inject,
317}
318
319#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
320#[serde(rename_all = "snake_case")]
321pub enum RunStatus {
322 Completed,
323 Failed,
324 Cancelled,
325 MaxStepsReached,
326}
327
328impl RunStatus {
329 pub const fn as_str(self) -> &'static str {
330 match self {
331 Self::Completed => "completed",
332 Self::Failed => "failed",
333 Self::Cancelled => "cancelled",
334 Self::MaxStepsReached => "max_steps_reached",
335 }
336 }
337}
338
339#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
340#[serde(rename_all = "snake_case")]
341pub enum InteractionKind {
342 Action,
343 UserQuestion,
344}
345
346impl InteractionKind {
347 pub const fn as_str(self) -> &'static str {
348 match self {
349 Self::Action => "action",
350 Self::UserQuestion => "user_question",
351 }
352 }
353}
354
355#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
356#[serde(rename_all = "snake_case")]
357pub enum InteractionResolution {
358 Confirmed,
359 Rejected,
360 Answered,
361}
362
363impl InteractionResolution {
364 pub const fn as_str(self) -> &'static str {
365 match self {
366 Self::Confirmed => "confirmed",
367 Self::Rejected => "rejected",
368 Self::Answered => "answered",
369 }
370 }
371}
372
373#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
374#[serde(rename_all = "snake_case")]
375pub enum ToolAuthorizationStatus {
376 Allowed,
377 Waiting,
378 Denied,
379}
380
381#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
382pub struct RecordedToolCall {
383 pub call_id: String,
384 pub tool: String,
385 pub arguments: Value,
386}
387
388impl ToolAuthorizationStatus {
389 pub const fn as_str(self) -> &'static str {
390 match self {
391 Self::Allowed => "allowed",
392 Self::Waiting => "waiting",
393 Self::Denied => "denied",
394 }
395 }
396}
397
398#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
399#[serde(tag = "type", rename_all = "snake_case")]
400pub enum ContentBlock {
401 Text {
402 text: String,
403 },
404 Resource {
405 resource_id: String,
406 media_type: String,
407 },
408 Data {
409 slot: String,
410 value: Value,
411 },
412 Citation {
413 resource_id: String,
414 label: String,
415 uri: String,
416 excerpt: Option<String>,
417 },
418}
419
420#[derive(Debug, Clone, Default, PartialEq)]
421pub struct SessionProjection {
422 pub session_id: String,
423 pub profile_revision_id: String,
424 pub deleted: bool,
425 pub last_seq: u64,
426 pub active_run_id: Option<String>,
427 pub waiting_interaction_id: Option<String>,
428 pub messages: Vec<ProjectedMessage>,
429 pub injected_context: Vec<ProjectedContext>,
430 pub run_status: BTreeMap<String, String>,
431 pub open_tool_calls: BTreeMap<String, OpenToolCall>,
432 pub started_tool_calls: BTreeSet<String>,
433 pub queued_inputs: BTreeMap<String, (String, DeliveryMode)>,
434 pub claimed_inputs: BTreeMap<String, String>,
435 pub open_turn: Option<u32>,
436 pub open_steps: BTreeSet<u32>,
437 pub next_step: u32,
438 pub open_compaction: Option<(String, u64)>,
439 pub summary: Option<String>,
440 usage_operations: BTreeMap<(String, String), (u64, u64, u64)>,
441 pending_provider_attempts: BTreeMap<(String, String), (u64, u64)>,
442 seen_tool_calls: BTreeSet<String>,
443}
444
445#[derive(Debug, Clone, PartialEq)]
446pub struct ProjectedContext {
447 pub run_id: String,
448 pub step: u32,
449 pub contribution_id: String,
450 pub source: String,
451 pub version: String,
452 pub authority: String,
453 pub form: String,
454 pub content: Vec<ContentBlock>,
455}
456
457#[derive(Debug, Clone, PartialEq)]
458pub struct ProjectedMessage {
459 pub role: &'static str,
460 pub run_id: String,
461 pub content: Vec<ContentBlock>,
462}
463
464#[derive(Debug, Clone, PartialEq)]
465pub struct OpenToolCall {
466 pub run_id: String,
467 pub step: u32,
468 pub tool: String,
469 pub arguments: Value,
470 pub source_event_seq: u64,
471}
472
473impl SessionProjection {
474 pub fn replay(events: &[SessionEvent]) -> Result<Self, EventError> {
475 let mut projection = Self::default();
476 for event in events {
477 projection.apply(event)?;
478 }
479 Ok(projection)
480 }
481
482 pub fn apply(&mut self, envelope: &SessionEvent) -> Result<(), EventError> {
483 if envelope.seq != self.last_seq + 1 {
484 return Err(EventError::Sequence {
485 expected: self.last_seq + 1,
486 actual: envelope.seq,
487 });
488 }
489 if self.session_id.is_empty() {
490 self.session_id = envelope.session_id.clone();
491 }
492 if self.session_id != envelope.session_id {
493 return Err(EventError::SessionMismatch);
494 }
495 if envelope.format_version() != SESSION_EVENT_FORMAT_VERSION {
496 if envelope.ignorable() {
497 self.last_seq = envelope.seq;
498 return Ok(());
499 }
500 return Err(EventError::UnsupportedFormat(envelope.format_version()));
501 }
502 if self.deleted {
503 return Err(EventError::SessionClosed);
504 }
505 match &envelope.event {
506 Event::SessionCreated {
507 profile_revision_id,
508 } => {
509 if envelope.seq != 1 || !self.profile_revision_id.is_empty() {
510 return Err(EventError::DuplicateSession);
511 }
512 self.profile_revision_id = profile_revision_id.clone();
513 }
514 Event::SessionDeleted { .. } => {
515 if let Some(run_id) = self.active_run_id.take() {
516 self.run_status.insert(run_id, "cancelled".into());
517 }
518 for (_, (run_id, _)) in std::mem::take(&mut self.queued_inputs) {
519 self.run_status.insert(run_id, "cancelled".into());
520 }
521 self.waiting_interaction_id = None;
522 self.open_turn = None;
523 self.open_steps.clear();
524 self.open_tool_calls.clear();
525 self.started_tool_calls.clear();
526 self.open_compaction = None;
527 self.deleted = true;
528 }
529 Event::RunStarted { run_id, input_id } => {
530 if self.active_run_id.is_some() {
531 return Err(EventError::ConcurrentRun);
532 }
533 if self.claimed_inputs.get(input_id).map(String::as_str) != Some(run_id) {
534 return Err(EventError::UnclaimedInput(input_id.clone()));
535 }
536 self.active_run_id = Some(run_id.clone());
537 self.next_step = 1;
538 self.run_status.insert(run_id.clone(), "running".into());
539 }
540 Event::RunWaiting {
541 run_id,
542 interaction_id,
543 } => {
544 self.require_active(run_id)?;
545 self.waiting_interaction_id = Some(interaction_id.clone());
546 self.run_status
547 .insert(run_id.clone(), "waiting_for_input".into());
548 }
549 Event::RunResumed {
550 run_id,
551 interaction_id,
552 } => {
553 self.require_active(run_id)?;
554 if self.waiting_interaction_id.as_deref() != Some(interaction_id) {
555 return Err(EventError::InteractionMismatch);
556 }
557 self.waiting_interaction_id = None;
558 self.run_status.insert(run_id.clone(), "running".into());
559 }
560 Event::RunFinished { run_id, status, .. } => {
561 self.require_active(run_id)?;
562 if !self.open_tool_calls.is_empty()
563 || !self.open_steps.is_empty()
564 || self.open_turn.is_some()
565 || self.open_compaction.is_some()
566 || self
567 .queued_inputs
568 .values()
569 .any(|(target_run_id, _)| target_run_id == run_id)
570 {
571 return Err(EventError::OpenLifecycle);
572 }
573 self.run_status
574 .insert(run_id.clone(), status.as_str().into());
575 self.active_run_id = None;
576 self.waiting_interaction_id = None;
577 }
578 Event::UserMessage { run_id, content } => self.messages.push(ProjectedMessage {
579 role: "user",
580 run_id: run_id.clone(),
581 content: content.clone(),
582 }),
583 Event::AssistantMessage {
584 run_id, content, ..
585 } => self.messages.push(ProjectedMessage {
586 role: "assistant",
587 run_id: run_id.clone(),
588 content: content.clone(),
589 }),
590 Event::ContextInjected {
591 run_id,
592 step,
593 contribution_id,
594 source,
595 version,
596 authority,
597 form,
598 content,
599 } => {
600 self.require_active(run_id)?;
601 if !self.open_steps.contains(step) {
602 return Err(EventError::LifecycleMismatch);
603 }
604 self.injected_context.push(ProjectedContext {
605 run_id: run_id.clone(),
606 step: *step,
607 contribution_id: contribution_id.clone(),
608 source: source.clone(),
609 version: version.clone(),
610 authority: authority.clone(),
611 form: form.clone(),
612 content: content.clone(),
613 });
614 }
615 Event::InputQueued {
616 input_id,
617 run_id,
618 mode,
619 ..
620 } => {
621 if *mode != DeliveryMode::Followup {
622 self.require_active(run_id)?;
623 }
624 if self
625 .queued_inputs
626 .insert(input_id.clone(), (run_id.clone(), *mode))
627 .is_some()
628 {
629 return Err(EventError::DuplicateInput(input_id.clone()));
630 }
631 }
632 Event::InputClaimed { input_id, run_id } => {
633 if self
634 .queued_inputs
635 .remove(input_id)
636 .map(|value| value.0)
637 .as_deref()
638 != Some(run_id)
639 || self
640 .claimed_inputs
641 .insert(input_id.clone(), run_id.clone())
642 .is_some()
643 {
644 return Err(EventError::UnqueuedInput(input_id.clone()));
645 }
646 }
647 Event::InputCancelled {
648 input_id,
649 run_id,
650 error_code,
651 } => {
652 if self
653 .queued_inputs
654 .remove(input_id)
655 .map(|value| value.0)
656 .as_deref()
657 != Some(run_id)
658 {
659 return Err(EventError::UnqueuedInput(input_id.clone()));
660 }
661 self.run_status.insert(
662 run_id.clone(),
663 if error_code == "cancelled" {
664 "cancelled"
665 } else {
666 "failed"
667 }
668 .into(),
669 );
670 }
671 Event::TurnStarted { run_id, turn } => {
672 self.require_active(run_id)?;
673 if self.open_turn.replace(*turn).is_some() {
674 return Err(EventError::ConcurrentTurn);
675 }
676 }
677 Event::TurnFinished { run_id, turn } => {
678 self.require_active(run_id)?;
679 if self.open_turn != Some(*turn)
680 || !self.open_steps.is_empty()
681 || !self.open_tool_calls.is_empty()
682 {
683 return Err(EventError::LifecycleMismatch);
684 }
685 self.open_turn = None;
686 }
687 Event::StepStarted { run_id, step } => {
688 self.require_active(run_id)?;
689 if self.open_turn.is_none()
690 || !self.open_steps.is_empty()
691 || *step != self.next_step
692 || !self.open_steps.insert(*step)
693 {
694 return Err(EventError::ConcurrentStep);
695 }
696 }
697 Event::StepFinished { run_id, step } => {
698 self.require_active(run_id)?;
699 if self
700 .open_tool_calls
701 .values()
702 .any(|call| call.run_id == *run_id && call.step == *step)
703 || !self.open_steps.remove(step)
704 {
705 return Err(EventError::LifecycleMismatch);
706 }
707 self.next_step = step.saturating_add(1);
708 }
709 Event::ToolCall {
710 run_id,
711 step,
712 call_id,
713 tool,
714 arguments,
715 } => {
716 self.require_active(run_id)?;
717 if !self.open_steps.contains(step) {
718 return Err(EventError::LifecycleMismatch);
719 }
720 if !self.seen_tool_calls.insert(call_id.clone())
721 || self
722 .open_tool_calls
723 .insert(
724 call_id.clone(),
725 OpenToolCall {
726 run_id: run_id.clone(),
727 step: *step,
728 tool: tool.clone(),
729 arguments: arguments.clone(),
730 source_event_seq: envelope.seq,
731 },
732 )
733 .is_some()
734 {
735 return Err(EventError::DuplicateToolCall(call_id.clone()));
736 }
737 }
738 Event::ToolResult {
739 run_id,
740 step,
741 call_id,
742 ..
743 } => {
744 self.require_active(run_id)?;
745 if !self.open_steps.contains(step) {
746 return Err(EventError::LifecycleMismatch);
747 }
748 let Some(call) = self.open_tool_calls.get(call_id) else {
749 return Err(EventError::OrphanToolResult(call_id.clone()));
750 };
751 if call.run_id != *run_id || call.step != *step {
752 return Err(EventError::ToolResultMismatch(call_id.clone()));
753 }
754 self.open_tool_calls.remove(call_id);
755 self.started_tool_calls.remove(call_id);
756 }
757 Event::ToolExecutionStarted {
758 run_id,
759 step,
760 call_id,
761 } => {
762 self.require_active(run_id)?;
763 let Some(call) = self.open_tool_calls.get(call_id) else {
764 return Err(EventError::OrphanToolResult(call_id.clone()));
765 };
766 if call.run_id != *run_id
767 || call.step != *step
768 || !self.started_tool_calls.insert(call_id.clone())
769 {
770 return Err(EventError::ToolResultMismatch(call_id.clone()));
771 }
772 }
773 Event::ModelRequestPrepared {
774 run_id,
775 operation_id,
776 reserved_prompt_tokens,
777 reserved_completion_tokens,
778 ..
779 } if !operation_id.is_empty() => {
780 self.require_active(run_id)?;
781 let key = (run_id.clone(), operation_id.clone());
782 if !self.usage_operations.contains_key(&key) {
783 let reservation = (*reserved_prompt_tokens, *reserved_completion_tokens);
784 match self.pending_provider_attempts.get(&key) {
785 Some(existing) if *existing != reservation => {
786 return Err(EventError::UsageConflict(operation_id.clone()));
787 }
788 Some(_) => {}
789 None => {
790 self.pending_provider_attempts.insert(key, reservation);
791 }
792 }
793 }
794 }
795 Event::UsageRecorded {
796 run_id,
797 operation_id,
798 prompt_tokens,
799 completion_tokens,
800 cost_units,
801 } => {
802 self.require_active(run_id)?;
803 let key = (run_id.clone(), operation_id.clone());
804 let usage = (*prompt_tokens, *completion_tokens, *cost_units);
805 match self.usage_operations.get(&key) {
806 Some(existing) if *existing != usage => {
807 return Err(EventError::UsageConflict(operation_id.clone()));
808 }
809 Some(_) => {}
810 None => {
811 self.pending_provider_attempts.remove(&key);
812 self.usage_operations.insert(key, usage);
813 }
814 }
815 }
816 Event::CompactionStarted {
817 run_id,
818 compaction_id,
819 source_through_seq,
820 } => {
821 self.require_active(run_id)?;
822 if self.open_compaction.is_some() {
823 return Err(EventError::ConcurrentCompaction);
824 }
825 self.open_compaction = Some((compaction_id.clone(), *source_through_seq));
826 }
827 Event::CompactionFinished {
828 run_id,
829 compaction_id,
830 ..
831 } => {
832 self.require_active(run_id)?;
833 if self.open_compaction.as_ref().map(|value| value.0.as_str())
834 != Some(compaction_id.as_str())
835 {
836 return Err(EventError::CompactionMismatch);
837 }
838 self.open_compaction = None;
839 }
840 Event::SummaryReplaced { summary, .. } => self.summary = Some(summary.clone()),
841 Event::Opaque {
842 event_type,
843 ignorable: false,
844 ..
845 } => return Err(EventError::UnknownRequired(event_type.clone())),
846 _ => {}
847 }
848 self.last_seq = envelope.seq;
849 Ok(())
850 }
851
852 fn require_active(&self, run_id: &str) -> Result<(), EventError> {
853 if self.active_run_id.as_deref() == Some(run_id) {
854 Ok(())
855 } else {
856 Err(EventError::RunMismatch)
857 }
858 }
859
860 pub fn usage_for(&self, run_id: &str) -> (u64, u64) {
861 self.usage_operations
862 .iter()
863 .filter(|((recorded_run_id, _), _)| recorded_run_id == run_id)
864 .fold((0, 0), |total, (_, usage)| {
865 (total.0 + usage.0, total.1 + usage.1)
866 })
867 }
868
869 pub fn billable_units_for(&self, run_id: &str) -> u64 {
870 let recorded = self
871 .usage_operations
872 .iter()
873 .filter(|((recorded_run_id, _), _)| recorded_run_id == run_id)
874 .map(|(_, usage)| usage.0 + usage.1 + usage.2)
875 .sum::<u64>();
876 recorded
877 + self
878 .pending_provider_attempts
879 .iter()
880 .filter(|((recorded_run_id, _), _)| recorded_run_id == run_id)
881 .map(|(_, usage)| usage.0 + usage.1)
882 .sum::<u64>()
883 }
884}
885
886#[derive(Debug, thiserror::Error, PartialEq, Eq)]
887pub enum EventError {
888 #[error("unsupported session event format version {0}")]
889 UnsupportedFormat(u32),
890 #[error("unknown required session event type {0}")]
891 UnknownRequired(String),
892 #[error("event sequence mismatch: expected {expected}, got {actual}")]
893 Sequence { expected: u64, actual: u64 },
894 #[error("event belongs to another session")]
895 SessionMismatch,
896 #[error("session creation must be the first and only creation event")]
897 DuplicateSession,
898 #[error("session already has an active run")]
899 ConcurrentRun,
900 #[error("session is closed")]
901 SessionClosed,
902 #[error("event does not match the active run")]
903 RunMismatch,
904 #[error("interaction does not match the waiting run")]
905 InteractionMismatch,
906 #[error("run cannot finish with an open turn, step or tool call")]
907 OpenLifecycle,
908 #[error("input was queued twice: {0}")]
909 DuplicateInput(String),
910 #[error("input was claimed before it was queued: {0}")]
911 UnqueuedInput(String),
912 #[error("run started from an unclaimed input: {0}")]
913 UnclaimedInput(String),
914 #[error("session already has an active turn")]
915 ConcurrentTurn,
916 #[error("turn already has this active step")]
917 ConcurrentStep,
918 #[error("turn or step lifecycle does not pair")]
919 LifecycleMismatch,
920 #[error("duplicate tool call {0}")]
921 DuplicateToolCall(String),
922 #[error("tool result has no matching call {0}")]
923 OrphanToolResult(String),
924 #[error("tool result does not match the call run and step: {0}")]
925 ToolResultMismatch(String),
926 #[error("usage operation was recorded with different totals: {0}")]
927 UsageConflict(String),
928 #[error("session already has an active compaction")]
929 ConcurrentCompaction,
930 #[error("compaction lifecycle does not pair")]
931 CompactionMismatch,
932 #[error("event store conflict: {0}")]
933 Conflict(String),
934 #[error("event store unavailable: {0}")]
935 Unavailable(String),
936}
937
938#[async_trait]
939pub trait SessionEventStore: Send + Sync {
940 async fn append(
941 &self,
942 tenant_id: &str,
943 session_id: &str,
944 expected_seq: u64,
945 events: Vec<Event>,
946 ) -> Result<Vec<SessionEvent>, EventError>;
947 async fn load(
948 &self,
949 tenant_id: &str,
950 session_id: &str,
951 after_seq: u64,
952 ) -> Result<Vec<SessionEvent>, EventError>;
953}
954
955pub fn text(value: impl Into<String>) -> Vec<ContentBlock> {
956 vec![ContentBlock::Text { text: value.into() }]
957}
958
959pub fn recovery_events(projection: &SessionProjection) -> Vec<Event> {
960 failure_events(projection, "worker_restarted")
961}
962
963pub fn failure_events(projection: &SessionProjection, error_code: &str) -> Vec<Event> {
964 termination_events(projection, RunStatus::Failed, error_code)
965}
966
967pub fn cancel_events(projection: &SessionProjection) -> Vec<Event> {
968 termination_events(projection, RunStatus::Cancelled, "cancelled")
969}
970
971pub fn session_deletion_events(projection: &SessionProjection, reason: &str) -> Vec<Event> {
972 let active = projection.active_run_id.as_deref();
973 let mut events = cancel_events(projection);
974 events.extend(
975 projection
976 .queued_inputs
977 .iter()
978 .filter(|(_, (run_id, _))| Some(run_id.as_str()) != active)
979 .map(|(input_id, (run_id, _))| Event::InputCancelled {
980 input_id: input_id.clone(),
981 run_id: run_id.clone(),
982 error_code: "cancelled".into(),
983 }),
984 );
985 events.push(Event::SessionDeleted {
986 reason: reason.into(),
987 });
988 events
989}
990
991fn termination_events(
992 projection: &SessionProjection,
993 status: RunStatus,
994 error_code: &str,
995) -> Vec<Event> {
996 let Some(run_id) = &projection.active_run_id else {
997 return Vec::new();
998 };
999 let mut events = projection
1000 .queued_inputs
1001 .iter()
1002 .filter(|(_, (target_run_id, _))| target_run_id == run_id)
1003 .map(|(input_id, _)| Event::InputCancelled {
1004 input_id: input_id.clone(),
1005 run_id: run_id.clone(),
1006 error_code: error_code.into(),
1007 })
1008 .collect::<Vec<_>>();
1009 events.extend(
1010 projection
1011 .open_tool_calls
1012 .iter()
1013 .map(|(call_id, call)| Event::ToolResult {
1014 run_id: run_id.clone(),
1015 step: call.step,
1016 call_id: call_id.clone(),
1017 result: termination_result(error_code),
1018 is_error: true,
1019 }),
1020 );
1021 if let Some((compaction_id, _)) = &projection.open_compaction {
1022 events.push(Event::CompactionFinished {
1023 run_id: run_id.clone(),
1024 compaction_id: compaction_id.clone(),
1025 status: "failed".into(),
1026 error: Some(error_code.into()),
1027 });
1028 }
1029 events.extend(
1030 projection
1031 .open_steps
1032 .iter()
1033 .map(|step| Event::StepFinished {
1034 run_id: run_id.clone(),
1035 step: *step,
1036 }),
1037 );
1038 if let Some(turn) = projection.open_turn {
1039 events.push(Event::TurnFinished {
1040 run_id: run_id.clone(),
1041 turn,
1042 });
1043 }
1044 events.push(Event::RunFinished {
1045 run_id: run_id.clone(),
1046 status,
1047 error_code: Some(error_code.into()),
1048 });
1049 events
1050}
1051
1052fn termination_result(error_code: &str) -> Value {
1053 if error_code == "tool_outcome_unknown" || error_code == "worker_restarted" {
1054 serde_json::json!({
1055 "error": error_code,
1056 "guidance": "The tool outcome is unknown. Verify external state before retrying any operation with side effects; ask the user when verification is unavailable."
1057 })
1058 } else {
1059 serde_json::json!({"error":error_code})
1060 }
1061}
1062
1063#[cfg(test)]
1064mod tests {
1065 use super::*;
1066
1067 fn event(seq: u64, event: Event) -> SessionEvent {
1068 SessionEvent {
1069 session_id: "s".into(),
1070 seq,
1071 occurred_at: Utc::now(),
1072 event,
1073 }
1074 }
1075
1076 #[test]
1077 fn replay_enforces_single_run_and_tool_pairs() {
1078 let events = vec![
1079 event(
1080 1,
1081 Event::SessionCreated {
1082 profile_revision_id: "p1".into(),
1083 },
1084 ),
1085 event(
1086 2,
1087 Event::InputQueued {
1088 input_id: "i1".into(),
1089 run_id: "r1".into(),
1090 mode: DeliveryMode::Followup,
1091 content: text("hi"),
1092 explicit_skill: None,
1093 },
1094 ),
1095 event(
1096 3,
1097 Event::InputClaimed {
1098 input_id: "i1".into(),
1099 run_id: "r1".into(),
1100 },
1101 ),
1102 event(
1103 4,
1104 Event::RunStarted {
1105 run_id: "r1".into(),
1106 input_id: "i1".into(),
1107 },
1108 ),
1109 event(
1110 5,
1111 Event::TurnStarted {
1112 run_id: "r1".into(),
1113 turn: 1,
1114 },
1115 ),
1116 event(
1117 6,
1118 Event::StepStarted {
1119 run_id: "r1".into(),
1120 step: 1,
1121 },
1122 ),
1123 event(
1124 7,
1125 Event::ToolCall {
1126 run_id: "r1".into(),
1127 step: 1,
1128 call_id: "c1".into(),
1129 tool: "echo".into(),
1130 arguments: serde_json::json!({"x":1}),
1131 },
1132 ),
1133 event(
1134 8,
1135 Event::ToolResult {
1136 run_id: "r1".into(),
1137 step: 1,
1138 call_id: "c1".into(),
1139 result: serde_json::json!({"x":1}),
1140 is_error: false,
1141 },
1142 ),
1143 event(
1144 9,
1145 Event::StepFinished {
1146 run_id: "r1".into(),
1147 step: 1,
1148 },
1149 ),
1150 event(
1151 10,
1152 Event::TurnFinished {
1153 run_id: "r1".into(),
1154 turn: 1,
1155 },
1156 ),
1157 event(
1158 11,
1159 Event::RunFinished {
1160 run_id: "r1".into(),
1161 status: RunStatus::Completed,
1162 error_code: None,
1163 },
1164 ),
1165 ];
1166 let projection = SessionProjection::replay(&events).unwrap();
1167 assert_eq!(projection.last_seq, 11);
1168 assert!(projection.active_run_id.is_none());
1169 }
1170
1171 #[test]
1172 fn replay_rejects_orphan_tool_result() {
1173 let events = vec![
1174 event(
1175 1,
1176 Event::SessionCreated {
1177 profile_revision_id: "p1".into(),
1178 },
1179 ),
1180 event(
1181 2,
1182 Event::InputQueued {
1183 input_id: "i1".into(),
1184 run_id: "r1".into(),
1185 mode: DeliveryMode::Followup,
1186 content: text("hi"),
1187 explicit_skill: None,
1188 },
1189 ),
1190 event(
1191 3,
1192 Event::InputClaimed {
1193 input_id: "i1".into(),
1194 run_id: "r1".into(),
1195 },
1196 ),
1197 event(
1198 4,
1199 Event::RunStarted {
1200 run_id: "r1".into(),
1201 input_id: "i1".into(),
1202 },
1203 ),
1204 event(
1205 5,
1206 Event::TurnStarted {
1207 run_id: "r1".into(),
1208 turn: 1,
1209 },
1210 ),
1211 event(
1212 6,
1213 Event::StepStarted {
1214 run_id: "r1".into(),
1215 step: 1,
1216 },
1217 ),
1218 event(
1219 7,
1220 Event::ToolResult {
1221 run_id: "r1".into(),
1222 step: 1,
1223 call_id: "missing".into(),
1224 result: Value::Null,
1225 is_error: true,
1226 },
1227 ),
1228 ];
1229 assert_eq!(
1230 SessionProjection::replay(&events).unwrap_err(),
1231 EventError::OrphanToolResult("missing".into())
1232 );
1233 }
1234
1235 #[test]
1236 fn queued_input_failure_is_not_projected_as_cancellation() {
1237 let events = vec![
1238 event(
1239 1,
1240 Event::SessionCreated {
1241 profile_revision_id: "p1".into(),
1242 },
1243 ),
1244 event(
1245 2,
1246 Event::InputQueued {
1247 input_id: "i1".into(),
1248 run_id: "r1".into(),
1249 mode: DeliveryMode::Followup,
1250 content: text("hi"),
1251 explicit_skill: None,
1252 },
1253 ),
1254 event(
1255 3,
1256 Event::InputCancelled {
1257 input_id: "i1".into(),
1258 run_id: "r1".into(),
1259 error_code: "profile_not_found".into(),
1260 },
1261 ),
1262 ];
1263 let projection = SessionProjection::replay(&events).unwrap();
1264 assert_eq!(
1265 projection.run_status.get("r1").map(String::as_str),
1266 Some("failed")
1267 );
1268 }
1269
1270 #[test]
1271 fn session_deletion_closes_active_and_queued_runs_before_tombstone() {
1272 let mut events = vec![
1273 event(
1274 1,
1275 Event::SessionCreated {
1276 profile_revision_id: "p1".into(),
1277 },
1278 ),
1279 event(
1280 2,
1281 Event::InputQueued {
1282 input_id: "i1".into(),
1283 run_id: "r1".into(),
1284 mode: DeliveryMode::Followup,
1285 content: text("start"),
1286 explicit_skill: None,
1287 },
1288 ),
1289 event(
1290 3,
1291 Event::InputClaimed {
1292 input_id: "i1".into(),
1293 run_id: "r1".into(),
1294 },
1295 ),
1296 event(
1297 4,
1298 Event::RunStarted {
1299 run_id: "r1".into(),
1300 input_id: "i1".into(),
1301 },
1302 ),
1303 event(
1304 5,
1305 Event::TurnStarted {
1306 run_id: "r1".into(),
1307 turn: 1,
1308 },
1309 ),
1310 event(
1311 6,
1312 Event::InputQueued {
1313 input_id: "i2".into(),
1314 run_id: "r2".into(),
1315 mode: DeliveryMode::Followup,
1316 content: text("later"),
1317 explicit_skill: None,
1318 },
1319 ),
1320 ];
1321 let projection = SessionProjection::replay(&events).unwrap();
1322 for event_value in session_deletion_events(&projection, "api_deleted") {
1323 let seq = events.len() as u64 + 1;
1324 events.push(event(seq, event_value));
1325 }
1326 let deleted = SessionProjection::replay(&events).unwrap();
1327 assert!(deleted.deleted);
1328 assert_eq!(
1329 deleted.run_status.get("r1").map(String::as_str),
1330 Some("cancelled")
1331 );
1332 assert_eq!(
1333 deleted.run_status.get("r2").map(String::as_str),
1334 Some("cancelled")
1335 );
1336 assert!(events.iter().any(|event| matches!(
1337 &event.event,
1338 Event::RunFinished { run_id, status: RunStatus::Cancelled, .. } if run_id == "r1"
1339 )));
1340 assert!(matches!(
1341 events.last().map(|event| &event.event),
1342 Some(Event::SessionDeleted { .. })
1343 ));
1344 }
1345
1346 #[test]
1347 fn usage_operations_are_idempotent_and_conflicts_fail_replay() {
1348 let mut events = vec![
1349 event(
1350 1,
1351 Event::SessionCreated {
1352 profile_revision_id: "p1".into(),
1353 },
1354 ),
1355 event(
1356 2,
1357 Event::InputQueued {
1358 input_id: "i1".into(),
1359 run_id: "r1".into(),
1360 mode: DeliveryMode::Followup,
1361 content: text("hi"),
1362 explicit_skill: None,
1363 },
1364 ),
1365 event(
1366 3,
1367 Event::InputClaimed {
1368 input_id: "i1".into(),
1369 run_id: "r1".into(),
1370 },
1371 ),
1372 event(
1373 4,
1374 Event::RunStarted {
1375 run_id: "r1".into(),
1376 input_id: "i1".into(),
1377 },
1378 ),
1379 event(
1380 5,
1381 Event::UsageRecorded {
1382 run_id: "r1".into(),
1383 operation_id: "model:1:attempt:1".into(),
1384 prompt_tokens: 7,
1385 completion_tokens: 3,
1386 cost_units: 5,
1387 },
1388 ),
1389 event(
1390 6,
1391 Event::UsageRecorded {
1392 run_id: "r1".into(),
1393 operation_id: "model:1:attempt:1".into(),
1394 prompt_tokens: 7,
1395 completion_tokens: 3,
1396 cost_units: 5,
1397 },
1398 ),
1399 ];
1400 assert_eq!(
1401 SessionProjection::replay(&events).unwrap().usage_for("r1"),
1402 (7, 3)
1403 );
1404 assert_eq!(
1405 SessionProjection::replay(&events)
1406 .unwrap()
1407 .billable_units_for("r1"),
1408 15
1409 );
1410
1411 events.push(event(
1412 7,
1413 Event::UsageRecorded {
1414 run_id: "r1".into(),
1415 operation_id: "model:1:attempt:1".into(),
1416 prompt_tokens: 8,
1417 completion_tokens: 3,
1418 cost_units: 0,
1419 },
1420 ));
1421 assert_eq!(
1422 SessionProjection::replay(&events).unwrap_err(),
1423 EventError::UsageConflict("model:1:attempt:1".into())
1424 );
1425 }
1426
1427 #[test]
1428 fn unresolved_provider_attempt_is_conservatively_billable_and_reconcilable() {
1429 let mut events = vec![
1430 event(
1431 1,
1432 Event::SessionCreated {
1433 profile_revision_id: "p1".into(),
1434 },
1435 ),
1436 event(
1437 2,
1438 Event::InputQueued {
1439 input_id: "i1".into(),
1440 run_id: "r1".into(),
1441 mode: DeliveryMode::Followup,
1442 content: text("hi"),
1443 explicit_skill: None,
1444 },
1445 ),
1446 event(
1447 3,
1448 Event::InputClaimed {
1449 input_id: "i1".into(),
1450 run_id: "r1".into(),
1451 },
1452 ),
1453 event(
1454 4,
1455 Event::RunStarted {
1456 run_id: "r1".into(),
1457 input_id: "i1".into(),
1458 },
1459 ),
1460 event(
1461 5,
1462 Event::ModelRequestPrepared {
1463 run_id: "r1".into(),
1464 step: 1,
1465 attempt: 1,
1466 provider_attempt_id: "r1:model:1:attempt:1".into(),
1467 operation_id: "model:1:attempt:1".into(),
1468 reserved_prompt_tokens: 7,
1469 reserved_completion_tokens: 11,
1470 request: Value::Null,
1471 prompt_sections: Value::Null,
1472 },
1473 ),
1474 ];
1475 assert_eq!(
1476 SessionProjection::replay(&events)
1477 .unwrap()
1478 .billable_units_for("r1"),
1479 18
1480 );
1481 events.push(event(
1482 6,
1483 Event::UsageRecorded {
1484 run_id: "r1".into(),
1485 operation_id: "model:1:attempt:1".into(),
1486 prompt_tokens: 6,
1487 completion_tokens: 2,
1488 cost_units: 0,
1489 },
1490 ));
1491 assert_eq!(
1492 SessionProjection::replay(&events)
1493 .unwrap()
1494 .billable_units_for("r1"),
1495 8
1496 );
1497 }
1498
1499 fn open_step_events() -> Vec<SessionEvent> {
1500 vec![
1501 event(
1502 1,
1503 Event::SessionCreated {
1504 profile_revision_id: "p1".into(),
1505 },
1506 ),
1507 event(
1508 2,
1509 Event::InputQueued {
1510 input_id: "i1".into(),
1511 run_id: "r1".into(),
1512 mode: DeliveryMode::Followup,
1513 content: text("hi"),
1514 explicit_skill: None,
1515 },
1516 ),
1517 event(
1518 3,
1519 Event::InputClaimed {
1520 input_id: "i1".into(),
1521 run_id: "r1".into(),
1522 },
1523 ),
1524 event(
1525 4,
1526 Event::RunStarted {
1527 run_id: "r1".into(),
1528 input_id: "i1".into(),
1529 },
1530 ),
1531 event(
1532 5,
1533 Event::TurnStarted {
1534 run_id: "r1".into(),
1535 turn: 1,
1536 },
1537 ),
1538 event(
1539 6,
1540 Event::StepStarted {
1541 run_id: "r1".into(),
1542 step: 1,
1543 },
1544 ),
1545 ]
1546 }
1547
1548 #[test]
1549 fn replay_rejects_overlapping_or_out_of_order_steps() {
1550 let mut overlapping = open_step_events();
1551 overlapping.push(event(
1552 7,
1553 Event::StepStarted {
1554 run_id: "r1".into(),
1555 step: 2,
1556 },
1557 ));
1558 assert_eq!(
1559 SessionProjection::replay(&overlapping).unwrap_err(),
1560 EventError::ConcurrentStep
1561 );
1562
1563 let mut skipped = open_step_events();
1564 skipped[5] = event(
1565 6,
1566 Event::StepStarted {
1567 run_id: "r1".into(),
1568 step: 2,
1569 },
1570 );
1571 assert_eq!(
1572 SessionProjection::replay(&skipped).unwrap_err(),
1573 EventError::ConcurrentStep
1574 );
1575 }
1576
1577 #[test]
1578 fn replay_rejects_cross_step_results_and_dangling_calls() {
1579 let mut events = open_step_events();
1580 events.push(event(
1581 7,
1582 Event::ToolCall {
1583 run_id: "r1".into(),
1584 step: 1,
1585 call_id: "c1".into(),
1586 tool: "echo".into(),
1587 arguments: Value::Null,
1588 },
1589 ));
1590 events.push(event(
1591 8,
1592 Event::ToolResult {
1593 run_id: "r1".into(),
1594 step: 2,
1595 call_id: "c1".into(),
1596 result: Value::Null,
1597 is_error: false,
1598 },
1599 ));
1600 assert_eq!(
1601 SessionProjection::replay(&events).unwrap_err(),
1602 EventError::LifecycleMismatch
1603 );
1604
1605 let mut dangling = open_step_events();
1606 dangling.push(event(
1607 7,
1608 Event::ToolCall {
1609 run_id: "r1".into(),
1610 step: 1,
1611 call_id: "c1".into(),
1612 tool: "echo".into(),
1613 arguments: Value::Null,
1614 },
1615 ));
1616 dangling.push(event(
1617 8,
1618 Event::StepFinished {
1619 run_id: "r1".into(),
1620 step: 1,
1621 },
1622 ));
1623 assert_eq!(
1624 SessionProjection::replay(&dangling).unwrap_err(),
1625 EventError::LifecycleMismatch
1626 );
1627 }
1628
1629 #[test]
1630 fn replay_rejects_reused_tool_call_ids() {
1631 let mut events = open_step_events();
1632 events.extend([
1633 event(
1634 7,
1635 Event::ToolCall {
1636 run_id: "r1".into(),
1637 step: 1,
1638 call_id: "c1".into(),
1639 tool: "echo".into(),
1640 arguments: Value::Null,
1641 },
1642 ),
1643 event(
1644 8,
1645 Event::ToolResult {
1646 run_id: "r1".into(),
1647 step: 1,
1648 call_id: "c1".into(),
1649 result: Value::Null,
1650 is_error: false,
1651 },
1652 ),
1653 event(
1654 9,
1655 Event::StepFinished {
1656 run_id: "r1".into(),
1657 step: 1,
1658 },
1659 ),
1660 event(
1661 10,
1662 Event::StepStarted {
1663 run_id: "r1".into(),
1664 step: 2,
1665 },
1666 ),
1667 event(
1668 11,
1669 Event::ToolCall {
1670 run_id: "r1".into(),
1671 step: 2,
1672 call_id: "c1".into(),
1673 tool: "echo".into(),
1674 arguments: Value::Null,
1675 },
1676 ),
1677 ]);
1678 assert_eq!(
1679 SessionProjection::replay(&events).unwrap_err(),
1680 EventError::DuplicateToolCall("c1".into())
1681 );
1682 }
1683
1684 #[test]
1685 fn replay_keeps_injected_context_without_changing_run_state() {
1686 let mut events = open_step_events();
1687 events.push(event(
1688 7,
1689 Event::ContextInjected {
1690 run_id: "r1".into(),
1691 step: 1,
1692 contribution_id: "memory:1".into(),
1693 source: "memory".into(),
1694 version: "v1".into(),
1695 authority: "tenant".into(),
1696 form: "message".into(),
1697 content: text("tenant context"),
1698 },
1699 ));
1700
1701 let projection = SessionProjection::replay(&events).unwrap();
1702 assert_eq!(projection.active_run_id.as_deref(), Some("r1"));
1703 assert_eq!(projection.open_steps, BTreeSet::from([1]));
1704 assert_eq!(
1705 projection.injected_context,
1706 vec![ProjectedContext {
1707 run_id: "r1".into(),
1708 step: 1,
1709 contribution_id: "memory:1".into(),
1710 source: "memory".into(),
1711 version: "v1".into(),
1712 authority: "tenant".into(),
1713 form: "message".into(),
1714 content: text("tenant context"),
1715 }]
1716 );
1717 }
1718
1719 #[test]
1720 fn replay_skips_unknown_ignorable_formats_and_rejects_required_events() {
1721 let optional = event(
1722 1,
1723 Event::Opaque {
1724 format_version: SESSION_EVENT_FORMAT_VERSION,
1725 event_type: "future_optional".into(),
1726 ignorable: true,
1727 payload: serde_json::json!({"answer": 42}),
1728 },
1729 );
1730 let projection = SessionProjection::replay(&[optional]).unwrap();
1731 assert_eq!(projection.last_seq, 1);
1732
1733 let required = event(
1734 1,
1735 Event::Opaque {
1736 format_version: SESSION_EVENT_FORMAT_VERSION,
1737 event_type: "future_required".into(),
1738 ignorable: false,
1739 payload: Value::Null,
1740 },
1741 );
1742 assert_eq!(
1743 SessionProjection::replay(&[required]).unwrap_err(),
1744 EventError::UnknownRequired("future_required".into())
1745 );
1746
1747 let unsupported = event(
1748 1,
1749 Event::Opaque {
1750 format_version: SESSION_EVENT_FORMAT_VERSION + 1,
1751 event_type: "future_optional".into(),
1752 ignorable: true,
1753 payload: Value::Null,
1754 },
1755 );
1756 assert_eq!(
1757 SessionProjection::replay(&[unsupported]).unwrap().last_seq,
1758 1
1759 );
1760
1761 let required_new_format = event(
1762 1,
1763 Event::Opaque {
1764 format_version: SESSION_EVENT_FORMAT_VERSION + 1,
1765 event_type: "future_required".into(),
1766 ignorable: false,
1767 payload: Value::Null,
1768 },
1769 );
1770 assert_eq!(
1771 SessionProjection::replay(&[required_new_format]).unwrap_err(),
1772 EventError::UnsupportedFormat(SESSION_EVENT_FORMAT_VERSION + 1)
1773 );
1774 }
1775}