1use std::collections::VecDeque;
2use std::path::{Path, PathBuf};
3use std::sync::Arc;
4use std::time::Duration;
5
6use tokio::io::{AsyncSeekExt, AsyncWriteExt, SeekFrom};
7use tokio::sync::{mpsc, oneshot};
8
9use crate::event::{Event, EventEnvelope};
10use crate::index::{AnchorIndex, ProjectEventInsert};
11use crate::redact::Redactor;
12
13pub struct EventWriter {
14 thread: Option<std::thread::JoinHandle<()>>,
15 tx: mpsc::UnboundedSender<EventEnvelope>,
16 flush_tx: mpsc::UnboundedSender<oneshot::Sender<()>>,
17 stop_tx: Option<oneshot::Sender<()>>,
18 events_path: PathBuf,
19}
20
21impl EventWriter {
22 pub fn spawn(session_dir: impl AsRef<Path>) -> std::io::Result<Self> {
23 Self::spawn_full(session_dir, None, None, None)
24 }
25
26 pub fn spawn_with(
27 session_dir: impl AsRef<Path>,
28 redactor: Option<Arc<Redactor>>,
29 ) -> std::io::Result<Self> {
30 Self::spawn_full(session_dir, redactor, None, None)
31 }
32
33 pub fn spawn_full(
36 session_dir: impl AsRef<Path>,
37 redactor: Option<Arc<Redactor>>,
38 project_index: Option<Arc<AnchorIndex>>,
39 session_id: Option<String>,
40 ) -> std::io::Result<Self> {
41 let session_dir = session_dir.as_ref().to_path_buf();
42 let events_path = session_dir.join("events.jsonl");
43 std::fs::create_dir_all(&session_dir)?;
44 let (tx, rx) = mpsc::unbounded_channel::<EventEnvelope>();
45 let (flush_tx, flush_rx) = mpsc::unbounded_channel::<oneshot::Sender<()>>();
46 let (stop_tx, stop_rx) = oneshot::channel::<()>();
47 let file_path = events_path.clone();
48 let thread = std::thread::Builder::new()
49 .name("atman-event-writer".into())
50 .spawn(move || {
51 let rt = match tokio::runtime::Builder::new_current_thread()
52 .enable_all()
53 .build()
54 {
55 Ok(rt) => rt,
56 Err(e) => {
57 crate::notify!(error, "event writer rt init failed: {e}");
58 return;
59 }
60 };
61 rt.block_on(async move {
62 if let Err(e) = writer_loop(
63 rx,
64 flush_rx,
65 stop_rx,
66 &file_path,
67 project_index,
68 session_id,
69 redactor,
70 )
71 .await
72 {
73 crate::notify!(error, "event writer failed: {e}");
74 }
75 });
76 })?;
77 Ok(Self {
78 thread: Some(thread),
79 tx,
80 flush_tx,
81 stop_tx: Some(stop_tx),
82 events_path,
83 })
84 }
85
86 pub async fn flush(&self) {
87 let (tx, rx) = oneshot::channel::<()>();
88 if self.flush_tx.send(tx).is_err() {
89 return;
90 }
91 let _ = rx.await;
92 }
93
94 pub fn sender(&self) -> mpsc::UnboundedSender<EventEnvelope> {
95 self.tx.clone()
96 }
97
98 pub fn events_path(&self) -> &Path {
99 &self.events_path
100 }
101
102 pub async fn shutdown(mut self) {
103 if let Some(stop_tx) = self.stop_tx.take() {
104 let _ = stop_tx.send(());
105 }
106 if let Some(thread) = self.thread.take() {
107 let _ = tokio::task::spawn_blocking(move || {
108 let _ = thread.join();
109 })
110 .await;
111 }
112 }
113}
114
115impl Drop for EventWriter {
116 fn drop(&mut self) {
117 if let Some(stop_tx) = self.stop_tx.take() {
118 let _ = stop_tx.send(());
119 }
120 if let Some(thread) = self.thread.take() {
121 let _ = thread.join();
122 }
123 }
124}
125
126const MAX_BUFFERED: usize = 10_000;
127const RECOVERY_INTERVAL: Duration = Duration::from_secs(2);
128
129struct DegradedBuffer {
130 events: VecDeque<EventEnvelope>,
131 error: Option<String>,
132 dropped: u64,
133 max_buffered: usize,
134}
135
136impl DegradedBuffer {
137 fn new(max_buffered: usize) -> Self {
138 Self {
139 events: VecDeque::new(),
140 error: None,
141 dropped: 0,
142 max_buffered,
143 }
144 }
145
146 fn is_degraded(&self) -> bool {
147 self.error.is_some()
148 }
149
150 fn degrade(&mut self, event: EventEnvelope, error: impl Into<String>) {
151 self.error = Some(error.into());
152 self.buffer(event);
153 }
154
155 fn buffer(&mut self, event: EventEnvelope) {
156 if self.events.len() == self.max_buffered {
157 self.events.pop_front();
158 self.dropped += 1;
159 }
160 self.events.push_back(event);
161 }
162
163 fn recover(&mut self) {
164 self.error = None;
165 }
166}
167
168async fn writer_loop(
169 mut rx: mpsc::UnboundedReceiver<EventEnvelope>,
170 mut flush_rx: mpsc::UnboundedReceiver<oneshot::Sender<()>>,
171 mut stop_rx: oneshot::Receiver<()>,
172 path: &Path,
173 project_index: Option<Arc<AnchorIndex>>,
174 session_id: Option<String>,
175 redactor: Option<Arc<Redactor>>,
176) -> std::io::Result<()> {
177 #[allow(clippy::suspicious_open_options)]
179 let mut file = tokio::fs::OpenOptions::new()
180 .create(true)
181 .read(true)
182 .write(true)
183 .open(path)
184 .await?;
185 let mut offset = file.seek(SeekFrom::End(0)).await?;
186 let session_dir = path.parent().unwrap_or(path);
187 let indexer = project_index.zip(session_id);
188 let mut degraded = DegradedBuffer::new(MAX_BUFFERED);
189 let mut recovery_tick = tokio::time::interval(RECOVERY_INTERVAL);
190
191 loop {
192 tokio::select! {
193 biased;
194 _ = &mut stop_rx => {
195 while let Ok(event) = rx.try_recv() {
196 handle_event(
197 &mut file,
198 &mut offset,
199 event,
200 &mut degraded,
201 indexer.as_ref(),
202 redactor.as_deref(),
203 session_dir,
204 ).await;
205 }
206 while let Ok(waiter) = flush_rx.try_recv() {
207 let _ = waiter.send(());
208 }
209 break;
210 }
211 _ = recovery_tick.tick() => {
212 retry_degraded_on_tick(
213 &mut file,
214 &mut offset,
215 &mut degraded,
216 indexer.as_ref(),
217 redactor.as_deref(),
218 session_dir,
219 ).await;
220 }
221 maybe_event = rx.recv() => {
222 match maybe_event {
223 Some(event) => {
224 handle_event(
225 &mut file,
226 &mut offset,
227 event,
228 &mut degraded,
229 indexer.as_ref(),
230 redactor.as_deref(),
231 session_dir,
232 ).await;
233 }
234 None => break,
235 }
236 }
237 maybe_flush = flush_rx.recv() => {
238 match maybe_flush {
239 Some(waiter) => {
240 while let Ok(event) = rx.try_recv() {
241 handle_event(
242 &mut file,
243 &mut offset,
244 event,
245 &mut degraded,
246 indexer.as_ref(),
247 redactor.as_deref(),
248 session_dir,
249 ).await;
250 }
251 retry_buffered(
252 &mut file,
253 &mut offset,
254 &mut degraded,
255 indexer.as_ref(),
256 redactor.as_deref(),
257 session_dir,
258 ).await;
259 if let Err(e) = file.sync_data().await {
260 crate::notify!(error, "event writer flush failed: {e}");
261 }
262 let _ = waiter.send(());
263 }
264 None => break,
265 }
266 }
267 }
268 }
269 retry_buffered(
270 &mut file,
271 &mut offset,
272 &mut degraded,
273 indexer.as_ref(),
274 redactor.as_deref(),
275 session_dir,
276 )
277 .await;
278 if !degraded.events.is_empty() {
279 crate::notify!(
280 error,
281 "event writer stopped with {} buffered event(s) not persisted (dropped {}): {}",
282 degraded.events.len(),
283 degraded.dropped,
284 degraded.error.as_deref().unwrap_or("write failed")
285 );
286 }
287 if let Err(e) = file.sync_data().await {
288 crate::notify!(error, "event writer final sync failed: {e}");
289 }
290 Ok(())
291}
292
293async fn handle_event(
294 file: &mut tokio::fs::File,
295 offset: &mut u64,
296 event: EventEnvelope,
297 degraded: &mut DegradedBuffer,
298 indexer: Option<&(Arc<AnchorIndex>, String)>,
299 redactor: Option<&Redactor>,
300 session_dir: &Path,
301) {
302 if degraded.is_degraded() {
303 degraded.buffer(event);
304 return;
305 }
306 if let Err(e) = write_event(file, offset, &event, indexer, redactor, session_dir).await {
307 crate::notify!(
308 error,
309 "event writer write failed (seq={}): {e}; buffering events in memory",
310 event.seq
311 );
312 degraded.degrade(event, e.to_string());
313 }
314}
315
316async fn retry_degraded_on_tick(
317 file: &mut tokio::fs::File,
318 offset: &mut u64,
319 degraded: &mut DegradedBuffer,
320 indexer: Option<&(Arc<AnchorIndex>, String)>,
321 redactor: Option<&Redactor>,
322 session_dir: &Path,
323) {
324 if !degraded.is_degraded() || degraded.events.is_empty() {
325 return;
326 }
327
328 retry_buffered(file, offset, degraded, indexer, redactor, session_dir).await;
329 if !degraded.is_degraded() {
330 crate::notify!(info, location = Status, "事件写入已恢复");
331 }
332}
333
334async fn retry_buffered(
335 file: &mut tokio::fs::File,
336 offset: &mut u64,
337 degraded: &mut DegradedBuffer,
338 indexer: Option<&(Arc<AnchorIndex>, String)>,
339 redactor: Option<&Redactor>,
340 session_dir: &Path,
341) {
342 while let Some(event) = degraded.events.pop_front() {
343 if let Err(e) = write_event(file, offset, &event, indexer, redactor, session_dir).await {
344 crate::notify!(
345 error,
346 "event writer retry failed (seq={}): {e}; {} event(s) remain buffered",
347 event.seq,
348 degraded.events.len() + 1
349 );
350 degraded.events.push_front(event);
351 degraded.error = Some(e.to_string());
352 return;
353 }
354 }
355 degraded.recover();
356}
357
358async fn write_event(
359 file: &mut tokio::fs::File,
360 offset: &mut u64,
361 envelope: &EventEnvelope,
362 indexer: Option<&(Arc<AnchorIndex>, String)>,
363 redactor: Option<&Redactor>,
364 session_dir: &Path,
365) -> std::io::Result<()> {
366 let line = serialize_event(envelope, redactor);
367 let start = *offset;
368 file.seek(SeekFrom::Start(start)).await?;
369 if let Err(e) = file.write_all(line.as_bytes()).await {
370 *offset = start;
371 return Err(e);
372 }
373 if let Err(e) = file.write_all(b"\n").await {
374 *offset = start;
375 return Err(e);
376 }
377 let end = file.stream_position().await?;
378 if let Err(e) = file.sync_data().await {
379 *offset = start;
380 return Err(e);
381 }
382 *offset = end;
383 if let Err(error) =
384 crate::session_meta::SessionStats::record_persisted_event(session_dir, start, end, envelope)
385 {
386 crate::notify!(
387 warn,
388 location = Log,
389 stack = merge_count("session_stats.persist_failed", 60_000),
390 "session stats update failed (seq={}): {error}",
391 envelope.seq
392 );
393 }
394 if let Some((idx, sid)) = indexer
395 && let Err(e) = insert_project_row(idx, sid, envelope, &line)
396 {
397 crate::notify!(
398 warn,
399 location = Log,
400 stack = merge_count("project_index.insert_failed", 60_000),
401 "project index insert failed (seq={}): {e}",
402 envelope.seq
403 );
404 }
405 Ok(())
406}
407
408fn serialize_event(envelope: &EventEnvelope, redactor: Option<&Redactor>) -> String {
409 let Some(r) = redactor else {
410 return serde_json::to_string(envelope).unwrap_or_else(|e| {
411 format!(
412 "{{\"type\":\"encode_error\",\"error\":{:?}}}",
413 e.to_string()
414 )
415 });
416 };
417 let mut value = match serde_json::to_value(envelope) {
418 Ok(v) => v,
419 Err(e) => {
420 return format!(
421 "{{\"type\":\"encode_error\",\"error\":{:?}}}",
422 e.to_string()
423 );
424 }
425 };
426 r.redact_json(&mut value);
427 serde_json::to_string(&value).unwrap_or_else(|e| {
428 format!(
429 "{{\"type\":\"encode_error\",\"error\":{:?}}}",
430 e.to_string()
431 )
432 })
433}
434
435fn insert_project_row(
436 index: &AnchorIndex,
437 session_id: &str,
438 envelope: &EventEnvelope,
439 payload_json: &str,
440) -> rusqlite::Result<()> {
441 let event = &envelope.event;
442 let ts = extract_ts(envelope);
443 let kind = event_kind(event);
444 let (turn_id, flow_run_id) = extract_anchors(event);
445 let text = extract_text_content(event).unwrap_or_default();
446 index.insert_project_event_raw(ProjectEventInsert {
447 session_id,
448 seq: envelope.seq as i64,
449 ts: &ts,
450 kind,
451 turn_id: turn_id.as_deref(),
452 flow_run_id: flow_run_id.as_deref(),
453 text_content: &text,
454 payload_json,
455 })?;
456 Ok(())
457}
458
459pub(crate) fn extract_ts(envelope: &EventEnvelope) -> String {
460 envelope.ts.to_rfc3339()
461}
462
463pub(crate) fn event_kind(event: &Event) -> &'static str {
464 match event {
465 Event::FlowStart { .. } => "flow_start",
466 Event::FlowEnd { .. } => "flow_end",
467 Event::WorkspaceLifecycle { .. } => "workspace_lifecycle",
468 Event::LlmCall { .. } => "llm_call",
469 Event::TurnStart { .. } => "turn_start",
470 Event::TurnEnd { .. } => "turn_end",
471 Event::UserMsg { .. } => "user_msg",
472 Event::AssistantMsg { .. } => "assistant_msg",
473 Event::ToolResultMsg { .. } => "tool_result_msg",
474 Event::ToolResultMetrics { .. } => "tool_result_metrics",
475 Event::DiffPreview { .. } => "diff_preview",
476 Event::CompactionSummary { .. } => "compaction_summary",
477 Event::SystemMsg { .. } => "system_msg",
478 Event::UserInject { .. } => "user_inject",
479 Event::ContentFilterHit { .. } => "content_filter_hit",
480 Event::ContextCompact { .. } => "context_compact",
481 Event::Checkpoint { .. } => "checkpoint",
482 Event::ContextTruncated { .. } => "context_truncated",
483 Event::WatchWarn { .. } => "watch_warn",
484 Event::PendingPrompt { .. } => "pending_prompt",
485 Event::PromptResolved { .. } => "prompt_resolved",
486 Event::LlmPartialCall { .. } => "llm_partial_call",
487 Event::FlowGraph { .. } => "flow_graph",
488 Event::FlowNodeStart { .. } => "flow_node_start",
489 Event::FlowNodeEnd { .. } => "flow_node_end",
490 Event::ToolNode { .. } => "tool_node",
491 Event::AttachmentDegraded { .. } => "attachment_degraded",
492 Event::ToolPendingApproval { .. } => "tool_pending_approval",
493 Event::ToolApproved { .. } => "tool_approved",
494 Event::ToolDenied { .. } => "tool_denied",
495 Event::PermissionRequestCreated { .. } => "permission_request_created",
496 Event::PermissionRequestTargeted { .. } => "permission_request_targeted",
497 Event::PermissionRequestDeferred { .. } => "permission_request_deferred",
498 Event::PermissionRequestApproved { .. } => "permission_request_approved",
499 Event::PermissionRequestDenied { .. } => "permission_request_denied",
500 Event::PermissionRequestCancelled { .. } => "permission_request_cancelled",
501 Event::PermissionGroupCreated { .. } => "permission_group_created",
502 Event::PermissionGroupUpdated { .. } => "permission_group_updated",
503 Event::PermissionGroupResolved { .. } => "permission_group_resolved",
504 Event::PermissionGrantCreated { .. } => "permission_grant_created",
505 Event::PermissionGrantExpired { .. } => "permission_grant_expired",
506 Event::UnrestrictedExecution { .. } => "unrestricted_execution",
507 Event::TerminalFinalState { .. } => "terminal_final_state",
508 Event::MermaidDiagram { .. } => "mermaid_diagram",
509 }
510}
511
512pub(crate) fn extract_anchors(event: &Event) -> (Option<String>, Option<String>) {
513 match event {
514 Event::FlowStart { run_id, .. }
515 | Event::FlowEnd { run_id, .. }
516 | Event::WorkspaceLifecycle { run_id, .. } => (None, Some(run_id.0.to_string())),
517 Event::TurnStart { turn_id, .. } | Event::TurnEnd { turn_id, .. } => {
518 (Some(turn_id.0.to_string()), None)
519 }
520 Event::UserInject { turn_id, .. } => (Some(turn_id.0.to_string()), None),
521 Event::UserMsg {
522 turn_id,
523 flow_run_id,
524 ..
525 }
526 | Event::AssistantMsg {
527 turn_id,
528 flow_run_id,
529 ..
530 }
531 | Event::ToolResultMsg {
532 turn_id,
533 flow_run_id,
534 ..
535 }
536 | Event::ToolResultMetrics {
537 turn_id,
538 flow_run_id,
539 ..
540 }
541 | Event::SystemMsg {
542 turn_id,
543 flow_run_id,
544 ..
545 } => (
546 Some(turn_id.0.to_string()),
547 flow_run_id.as_ref().map(|r| r.0.to_string()),
548 ),
549 Event::DiffPreview {
550 turn_id,
551 flow_run_id,
552 ..
553 } => (
554 turn_id.as_ref().map(|t| t.0.to_string()),
555 flow_run_id.as_ref().map(|r| r.0.to_string()),
556 ),
557 Event::ContentFilterHit {
558 turn_id,
559 flow_run_id,
560 ..
561 }
562 | Event::ContextTruncated {
563 turn_id,
564 flow_run_id,
565 ..
566 }
567 | Event::WatchWarn {
568 turn_id,
569 flow_run_id,
570 ..
571 } => (
572 turn_id.as_ref().map(|t| t.0.to_string()),
573 flow_run_id.as_ref().map(|r| r.0.to_string()),
574 ),
575 Event::LlmPartialCall {
576 turn_id,
577 flow_run_id,
578 ..
579 } => (
580 turn_id.as_ref().map(|t| t.0.to_string()),
581 flow_run_id.as_ref().map(|r| r.0.to_string()),
582 ),
583 Event::FlowGraph { run_id, .. }
584 | Event::FlowNodeStart { run_id, .. }
585 | Event::FlowNodeEnd { run_id, .. }
586 | Event::ToolNode { run_id, .. }
587 | Event::ToolPendingApproval { run_id, .. }
588 | Event::ToolApproved { run_id, .. }
589 | Event::ToolDenied { run_id, .. } => (None, Some(run_id.0.to_string())),
590 Event::PermissionRequestCreated { payload }
591 | Event::PermissionRequestTargeted { payload }
592 | Event::PermissionRequestDeferred { payload }
593 | Event::PermissionRequestApproved { payload }
594 | Event::PermissionRequestDenied { payload }
595 | Event::PermissionRequestCancelled { payload }
596 | Event::UnrestrictedExecution { payload } => {
597 (None, Some(payload.requesting_run_id.0.to_string()))
598 }
599 Event::PermissionGroupCreated { payload }
600 | Event::PermissionGroupUpdated { payload }
601 | Event::PermissionGroupResolved { payload } => {
602 let anchor = match &payload.owner {
603 crate::permission_audit::PermissionGroupAuditOwner::Flow { run_id } => {
604 run_id.to_string()
605 }
606 crate::permission_audit::PermissionGroupAuditOwner::User { session_id } => {
607 format!("user:{session_id}")
608 }
609 crate::permission_audit::PermissionGroupAuditOwner::System => "system".into(),
610 };
611 (None, Some(anchor))
612 }
613 Event::PermissionGrantCreated { payload } | Event::PermissionGrantExpired { payload } => {
614 (None, Some(payload.requesting_run_id.0.to_string()))
615 }
616 Event::AttachmentDegraded {
617 turn_id,
618 flow_run_id,
619 ..
620 } => (
621 turn_id.as_ref().map(|t| t.0.to_string()),
622 flow_run_id.as_ref().map(|r| r.0.to_string()),
623 ),
624 Event::CompactionSummary { flow_run_id, .. }
625 | Event::ContextCompact { flow_run_id, .. }
626 | Event::Checkpoint { flow_run_id, .. } => (
627 None,
628 flow_run_id.as_ref().map(|run_id| run_id.0.to_string()),
629 ),
630 Event::LlmCall { .. }
631 | Event::PendingPrompt { .. }
632 | Event::PromptResolved { .. }
633 | Event::TerminalFinalState { .. }
634 | Event::MermaidDiagram { .. } => (None, None),
635 }
636}
637
638pub(crate) fn extract_text_content(event: &Event) -> Option<String> {
639 match event {
640 Event::UserMsg { message, .. }
641 | Event::AssistantMsg { message, .. }
642 | Event::ToolResultMsg { message, .. }
643 | Event::SystemMsg { message, .. } => Some(message.text_concat()),
644 Event::WatchWarn { message, .. } => Some(message.clone()),
645 Event::CompactionSummary { summary, .. } => Some(summary.clone()),
646 Event::AttachmentDegraded {
647 file_basename,
648 reason,
649 ..
650 } => Some(format!("{file_basename} {reason}")),
651 _ => None,
652 }
653}
654
655#[cfg(test)]
656mod tests {
657 use super::*;
658 use crate::event::{Event, FlowRunId, FlowStatus};
659 use tempfile::TempDir;
660
661 fn flow_start(seq: u64) -> EventEnvelope {
662 EventEnvelope::new(
663 seq,
664 Event::FlowStart {
665 run_id: FlowRunId::now(),
666 flow_name: format!("flow_{seq}"),
667 parent_run_id: None,
668 parent_node_id: None,
669 spawned: false,
670 },
671 )
672 }
673
674 #[test]
675 fn every_owned_message_role_keeps_its_flow_anchor() {
676 let turn_id = crate::event::TurnId::now();
677 let flow_run_id = FlowRunId::now();
678 let user = crate::message::Message::user_text(turn_id.clone(), "delegated");
679 let system = crate::message::Message::system_text(turn_id.clone(), "handoff");
680
681 for event in [
682 Event::UserMsg {
683 turn_id: turn_id.clone(),
684 flow_run_id: Some(flow_run_id.clone()),
685 message: user,
686 },
687 Event::SystemMsg {
688 turn_id: turn_id.clone(),
689 flow_run_id: Some(flow_run_id.clone()),
690 message: system,
691 },
692 ] {
693 assert_eq!(
694 extract_anchors(&event),
695 (Some(turn_id.to_string()), Some(flow_run_id.to_string()))
696 );
697 }
698 }
699
700 #[test]
701 fn every_owned_compaction_event_keeps_its_flow_anchor() {
702 let flow_run_id = FlowRunId::now();
703 for event in [
704 Event::CompactionSummary {
705 session_id: "session".into(),
706 flow_run_id: Some(flow_run_id.clone()),
707 range_start: 0,
708 range_end: 1,
709 compacted_count: 2,
710 before_tokens: 100,
711 after_tokens: 10,
712 summary: "summary".into(),
713 },
714 Event::ContextCompact {
715 session_id: "session".into(),
716 flow_run_id: Some(flow_run_id.clone()),
717 before_tokens: 100,
718 after_tokens: 10,
719 compacted_range_start: 0,
720 compacted_range_end: 1,
721 summary_text: Some("summary".into()),
722 replacement_msg_seq: None,
723 },
724 Event::Checkpoint {
725 session_id: "session".into(),
726 flow_run_id: Some(flow_run_id.clone()),
727 messages: Vec::new(),
728 window_tokens: 10,
729 },
730 ] {
731 assert_eq!(
732 extract_anchors(&event),
733 (None, Some(flow_run_id.to_string()))
734 );
735 }
736 }
737
738 #[test]
739 fn degraded_buffer_keeps_later_events_and_drops_oldest_at_capacity() {
740 let mut degraded = DegradedBuffer::new(2);
741
742 degraded.degrade(flow_start(1), "disk full");
743 degraded.buffer(flow_start(2));
744
745 assert!(degraded.is_degraded());
746 assert_eq!(
747 degraded
748 .events
749 .iter()
750 .map(|event| event.seq)
751 .collect::<Vec<_>>(),
752 vec![1, 2]
753 );
754 assert_eq!(degraded.dropped, 0);
755
756 degraded.buffer(flow_start(3));
757
758 assert_eq!(
759 degraded
760 .events
761 .iter()
762 .map(|event| event.seq)
763 .collect::<Vec<_>>(),
764 vec![2, 3]
765 );
766 assert_eq!(degraded.dropped, 1);
767 assert_eq!(degraded.error.as_deref(), Some("disk full"));
768 }
769
770 #[tokio::test]
771 async fn recovery_tick_retries_buffered_events() {
772 let dir = TempDir::new().unwrap();
773 let path = dir.path().join("events.jsonl");
774 tokio::fs::write(&path, b"").await.unwrap();
775 let mut file = tokio::fs::File::open(&path).await.unwrap();
776 let mut offset = 0;
777 let mut degraded = DegradedBuffer::new(MAX_BUFFERED);
778
779 handle_event(
780 &mut file,
781 &mut offset,
782 flow_start(1),
783 &mut degraded,
784 None,
785 None,
786 dir.path(),
787 )
788 .await;
789
790 assert!(degraded.is_degraded());
791 assert_eq!(degraded.events.len(), 1);
792
793 drop(file);
794 #[allow(clippy::suspicious_open_options)]
795 let mut file = tokio::fs::OpenOptions::new()
796 .read(true)
797 .write(true)
798 .open(&path)
799 .await
800 .unwrap();
801 let mut offset = 0;
802 retry_degraded_on_tick(
803 &mut file,
804 &mut offset,
805 &mut degraded,
806 None,
807 None,
808 dir.path(),
809 )
810 .await;
811
812 assert!(!degraded.is_degraded());
813 assert!(degraded.events.is_empty());
814 let contents = tokio::fs::read_to_string(path).await.unwrap();
815 let event: serde_json::Value = serde_json::from_str(contents.trim()).unwrap();
816 assert_eq!(event["seq"], 1);
817 assert_eq!(event["type"], "flow_start");
818 }
819
820 #[tokio::test]
821 async fn write_event_overwrites_partial_line() {
822 let dir = TempDir::new().unwrap();
823 let path = dir.path().join("events.jsonl");
824 let mut file = tokio::fs::OpenOptions::new()
825 .create(true)
826 .truncate(false)
827 .read(true)
828 .write(true)
829 .open(&path)
830 .await
831 .unwrap();
832 let event = flow_start(1);
833 let line = serialize_event(&event, None);
834 file.write_all(&line.as_bytes()[..line.len() / 2])
835 .await
836 .unwrap();
837 let mut offset = 0;
838
839 write_event(&mut file, &mut offset, &event, None, None, dir.path())
840 .await
841 .unwrap();
842
843 let contents = tokio::fs::read_to_string(path).await.unwrap();
844 assert_eq!(contents.lines().count(), 1);
845 let _: serde_json::Value = serde_json::from_str(contents.trim()).unwrap();
846 }
847
848 #[tokio::test]
849 async fn writer_appends_events_as_jsonl() {
850 let dir = TempDir::new().unwrap();
851 let writer = EventWriter::spawn(dir.path()).unwrap();
852 let tx = writer.sender();
853 for i in 0..5 {
854 tx.send(EventEnvelope::new(
855 i as u64,
856 Event::FlowStart {
857 run_id: FlowRunId::now(),
858 flow_name: format!("flow_{i}"),
859 parent_run_id: None,
860 parent_node_id: None,
861 spawned: false,
862 },
863 ))
864 .unwrap();
865 }
866 drop(tx);
867 writer.shutdown().await;
868 let path = dir.path().join("events.jsonl");
869 let contents = tokio::fs::read_to_string(&path).await.unwrap();
870 let lines: Vec<_> = contents.lines().collect();
871 assert_eq!(lines.len(), 5);
872 for line in lines {
873 let v: serde_json::Value = serde_json::from_str(line).unwrap();
874 assert_eq!(v["type"], "flow_start");
875 assert!(v["run_id"].is_string());
876 assert!(v["flow_name"].is_string());
877 }
878 let stats = crate::session_meta::SessionStats::load_or_rebuild(dir.path()).unwrap();
879 assert_eq!(stats.event_count, 5);
880 assert_eq!(stats.message_count, 0);
881 assert_eq!(stats.event_bytes, std::fs::metadata(path).unwrap().len());
882 }
883
884 #[tokio::test]
885 async fn writer_writes_to_project_index_with_session_id() {
886 let session_dir = TempDir::new().unwrap();
887 let project_dir = TempDir::new().unwrap();
888 let idx = Arc::new(AnchorIndex::open_project(project_dir.path()).unwrap());
889 let writer = EventWriter::spawn_full(
890 session_dir.path(),
891 None,
892 Some(idx.clone()),
893 Some("sess-x".into()),
894 )
895 .unwrap();
896 let tx = writer.sender();
897 for i in 0..3 {
898 tx.send(EventEnvelope::new(
899 (i + 1) as u64,
900 Event::FlowStart {
901 run_id: FlowRunId::now(),
902 flow_name: format!("flow_{i}"),
903 parent_run_id: None,
904 parent_node_id: None,
905 spawned: false,
906 },
907 ))
908 .unwrap();
909 }
910 drop(tx);
911 writer.shutdown().await;
912
913 let jsonl_lines = tokio::fs::read_to_string(session_dir.path().join("events.jsonl"))
914 .await
915 .unwrap()
916 .lines()
917 .count();
918 assert_eq!(jsonl_lines, 3);
919
920 let conn = idx.conn();
921 let count: i64 = conn
922 .query_row(
923 "SELECT COUNT(*) FROM events WHERE session_id = ?",
924 rusqlite::params!["sess-x"],
925 |r| r.get(0),
926 )
927 .unwrap();
928 assert_eq!(count, 3);
929 }
930
931 #[tokio::test]
932 async fn writer_indexes_user_msg_text_for_project_fts() {
933 use crate::event::TurnId;
934 use crate::message::Message;
935
936 let session_dir = TempDir::new().unwrap();
937 let project_dir = TempDir::new().unwrap();
938 let idx = Arc::new(AnchorIndex::open_project(project_dir.path()).unwrap());
939 let writer = EventWriter::spawn_full(
940 session_dir.path(),
941 None,
942 Some(idx.clone()),
943 Some("sess-x".into()),
944 )
945 .unwrap();
946 let tid = TurnId::now();
947 writer
948 .sender()
949 .send(EventEnvelope::new(
950 1,
951 Event::UserMsg {
952 turn_id: tid.clone(),
953 flow_run_id: None,
954 message: Message::user_text(tid, "sqlite fts full text search"),
955 },
956 ))
957 .unwrap();
958 writer.shutdown().await;
959
960 let hits = idx
961 .fts_search_project_events("sqlite", Some("sess-x"), 10)
962 .unwrap();
963 assert_eq!(hits.len(), 1);
964 assert_eq!(hits[0].session_id, "sess-x");
965 assert_eq!(hits[0].seq, 1);
966 let stats = crate::session_meta::SessionStats::load_or_rebuild(session_dir.path()).unwrap();
967 assert_eq!(stats.event_count, 1);
968 assert_eq!(stats.user_message_count, 1);
969 assert_eq!(stats.message_count, 1);
970 }
971
972 #[tokio::test]
973 async fn writer_serializes_flow_end_with_status() {
974 let dir = TempDir::new().unwrap();
975 let writer = EventWriter::spawn(dir.path()).unwrap();
976 writer
977 .sender()
978 .send(EventEnvelope::new(
979 0,
980 Event::FlowEnd {
981 run_id: FlowRunId::now(),
982 flow_name: "t".into(),
983 status: FlowStatus::Errored {
984 message: "boom".into(),
985 },
986 },
987 ))
988 .unwrap();
989 writer.shutdown().await;
990 let contents = tokio::fs::read_to_string(dir.path().join("events.jsonl"))
991 .await
992 .unwrap();
993 let v: serde_json::Value = serde_json::from_str(contents.trim()).unwrap();
994 assert_eq!(v["type"], "flow_end");
995 assert_eq!(v["status"]["kind"], "errored");
996 }
997}