1use crate::db::DbPool;
9use crate::error::StorageResult;
10use chrono::{DateTime, Utc};
11use luft_core::contract::backend::AgentStatus;
12use luft_core::contract::event::{AgentEvent, ProgressDelta, RunStatus};
13use luft_core::contract::ids::{AgentId, PhaseId, RunId};
14use serde_json::Value as Json;
15use sqlx::Row;
16use std::sync::Arc;
17
18#[derive(Clone)]
22pub struct EventWriter {
23 pool: DbPool,
24}
25
26impl EventWriter {
27 pub fn new(pool: DbPool) -> Self {
28 Self { pool }
29 }
30
31 pub fn pool(&self) -> &DbPool {
32 &self.pool
33 }
34
35 pub async fn write_event(&self, event: &AgentEvent) -> StorageResult<()> {
39 match event {
40 AgentEvent::RunStarted { run_id, task, ts } => {
41 self.write_run_started(*run_id, task, *ts).await?;
42 }
43 AgentEvent::PhaseStarted {
44 run_id,
45 phase_id,
46 label,
47 planned,
48 description,
49 role,
50 ..
51 } => {
52 self.write_phase_started(
53 *run_id,
54 *phase_id,
55 label,
56 *planned,
57 description.as_deref(),
58 role.as_deref(),
59 )
60 .await?;
61 }
62 AgentEvent::AgentStarted {
63 run_id,
64 phase_id,
65 agent_id,
66 prompt_preview,
67 model,
68 description: _,
69 role: _,
70 name: _,
71 agent_seq: _,
72 ts: _,
73 } => {
74 self.write_agent_started(
75 *run_id,
76 *phase_id,
77 *agent_id,
78 prompt_preview,
79 model.as_deref(),
80 )
81 .await?;
82 }
83 AgentEvent::AgentProgress {
84 run_id,
85 agent_id,
86 delta,
87 } => {
88 self.write_delta(*run_id, *agent_id, delta).await?;
89 }
90 AgentEvent::AgentDone {
91 run_id,
92 agent_id,
93 status,
94 tokens,
95 elapsed_ms,
96 name: _,
97 agent_seq: _,
98 output: _,
99 findings: _,
100 prompt: _,
101 retry_count,
102 ts: _,
103 } => {
104 self.write_agent_done(
105 *run_id,
106 *agent_id,
107 status,
108 *tokens,
109 *elapsed_ms,
110 *retry_count,
111 )
112 .await?;
113 }
114 AgentEvent::PhaseDone {
115 run_id,
116 phase_id,
117 ok,
118 failed,
119 ..
120 } => {
121 self.write_phase_done(*run_id, *phase_id, *ok, *failed)
122 .await?;
123 }
124 AgentEvent::RunDone {
125 run_id,
126 status,
127 total_tokens,
128 report,
129 ..
130 } => {
131 self.write_run_done(*run_id, status, *total_tokens, report)
132 .await?;
133 }
134 AgentEvent::Log { .. } => {
135 }
137 AgentEvent::SignalReceived { .. } => {
138 }
141 AgentEvent::BudgetSet { .. } => {
142 }
144 AgentEvent::ReportEmitted {
145 run_id,
146 phase_id,
147 report,
148 } => {
149 self.write_report_emitted(*run_id, *phase_id, report)
150 .await?;
151 }
152 AgentEvent::ParallelStarted {
153 run_id,
154 phase_id,
155 span_id,
156 count,
157 } => {
158 self.write_span_started(
159 *run_id,
160 *phase_id,
161 *span_id,
162 "parallel",
163 Some(*count as i64),
164 None,
165 None,
166 None,
167 )
168 .await?;
169 }
170 AgentEvent::ParallelDone {
171 run_id,
172 phase_id: _,
173 span_id,
174 ok,
175 failed,
176 results,
177 elapsed_ms,
178 } => {
179 self.write_span_done(
180 *run_id,
181 *span_id,
182 *ok as i64,
183 *failed as i64,
184 Some(results),
185 None,
186 None,
187 *elapsed_ms,
188 )
189 .await?;
190 }
191 AgentEvent::WorkflowStarted {
192 run_id,
193 span_id,
194 path,
195 args,
196 } => {
197 self.write_span_started(
198 *run_id,
199 0,
200 *span_id,
201 "workflow",
202 None,
203 None,
204 Some(path),
205 Some(args),
206 )
207 .await?;
208 }
209 AgentEvent::WorkflowDone {
210 run_id,
211 span_id,
212 path,
213 report,
214 elapsed_ms,
215 error,
216 } => {
217 self.write_span_done(
218 *run_id,
219 *span_id,
220 if error.is_none() { 1 } else { 0 },
221 if error.is_some() { 1 } else { 0 },
222 Some(report),
223 error.clone(),
224 Some(path),
225 *elapsed_ms,
226 )
227 .await?;
228 }
229 AgentEvent::ConvergeStarted {
230 run_id,
231 phase_id,
232 span_id,
233 items,
234 max_rounds,
235 } => {
236 self.write_span_started(
237 *run_id,
238 *phase_id,
239 *span_id,
240 "converge",
241 Some(*items as i64),
242 Some(*max_rounds as i64),
243 None,
244 None,
245 )
246 .await?;
247 }
248 AgentEvent::ConvergeDone {
249 run_id,
250 phase_id: _,
251 span_id,
252 rounds,
253 converged,
254 surviving: _,
255 result,
256 elapsed_ms,
257 error,
258 } => {
259 self.write_converge_done(
260 *run_id,
261 *span_id,
262 *rounds as i64,
263 *converged,
264 Some(result),
265 error.clone(),
266 *elapsed_ms,
267 )
268 .await?;
269 }
270 AgentEvent::PipelineStarted {
271 run_id,
272 total_stages,
273 items,
274 } => {
275 self.write_span_started(
276 *run_id,
277 0,
278 0,
279 "pipeline",
280 Some(*items as i64),
281 Some(*total_stages as i64),
282 None,
283 None,
284 )
285 .await?;
286 }
287 AgentEvent::PipelineStageStarted {
288 run_id,
289 stage_index,
290 label,
291 agents_in_stage,
292 } => {
293 self.write_span_started(
294 *run_id,
295 0,
296 *stage_index as u64 + 1,
297 "pipeline_stage",
298 Some(*agents_in_stage as i64),
299 None,
300 Some(label),
301 None,
302 )
303 .await?;
304 }
305 AgentEvent::PipelineItemDone {
306 run_id: _,
307 stage_index: _,
308 item_index: _,
309 status,
310 tokens,
311 elapsed_ms,
312 } => {
313 tracing::trace!(
315 ?status,
316 ?tokens,
317 elapsed_ms,
318 "pipeline item done (skipping item-level write)"
319 );
320 }
321 AgentEvent::PipelineDone {
322 run_id,
323 stages_completed,
324 total_ok,
325 total_failed,
326 } => {
327 self.write_span_done(
328 *run_id,
329 0,
330 *total_ok as i64,
331 *total_failed as i64,
332 None,
333 None,
334 None,
335 0,
336 )
337 .await?;
338 tracing::trace!(stages_completed, "pipeline done");
339 }
340 AgentEvent::AcpRaw { .. } => {}
343 AgentEvent::PhaseSpanStarted { .. }
346 | AgentEvent::PhaseSpanDone { .. }
347 | AgentEvent::PlanPreview { .. }
348 | AgentEvent::SchemaRetry { .. } => {}
349 }
350
351 self.append_audit(event).await?;
353 Ok(())
354 }
355
356 async fn write_run_started(
361 &self,
362 run_id: RunId,
363 task: &str,
364 ts: DateTime<Utc>,
365 ) -> StorageResult<()> {
366 sqlx::query(
367 "INSERT OR IGNORE INTO runs (run_id, task, status, started_ts)
368 VALUES (?, ?, 'running', ?)",
369 )
370 .bind(run_id)
371 .bind(task)
372 .bind(ts.to_rfc3339())
373 .execute(&self.pool)
374 .await?;
375 Ok(())
376 }
377
378 async fn write_phase_started(
379 &self,
380 run_id: RunId,
381 phase_id: PhaseId,
382 label: &str,
383 planned: usize,
384 description: Option<&str>,
385 role: Option<&str>,
386 ) -> StorageResult<()> {
387 sqlx::query(
388 "INSERT INTO phases (run_id, phase_id, label, planned, description, role, started_ts)
389 VALUES (?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
390 ON CONFLICT(run_id, phase_id) DO UPDATE SET
391 label = excluded.label,
392 planned = excluded.planned,
393 description = COALESCE(excluded.description, phases.description),
394 role = COALESCE(excluded.role, phases.role),
395 started_ts = COALESCE(phases.started_ts, excluded.started_ts)",
396 )
397 .bind(run_id)
398 .bind(phase_id as i64)
399 .bind(label)
400 .bind(planned as i64)
401 .bind(description)
402 .bind(role)
403 .execute(&self.pool)
404 .await?;
405 Ok(())
406 }
407
408 async fn write_agent_started(
409 &self,
410 run_id: RunId,
411 phase_id: PhaseId,
412 agent_id: AgentId,
413 prompt_preview: &str,
414 model: Option<&str>,
415 ) -> StorageResult<()> {
416 sqlx::query(
417 "INSERT INTO runs (run_id, task, status, started_ts)
418 VALUES (?, '', 'running', strftime('%Y-%m-%dT%H:%M:%fZ','now'))
419 ON CONFLICT(run_id) DO NOTHING",
420 )
421 .bind(run_id)
422 .execute(&self.pool)
423 .await?;
424
425 sqlx::query(
426 "INSERT INTO agents (run_id, agent_id, phase_id, model, status,
427 prompt_preview, started_ts)
428 VALUES (?, ?, ?, ?, 'running', ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
429 ON CONFLICT(run_id, agent_id) DO UPDATE SET
430 phase_id = excluded.phase_id,
431 model = excluded.model,
432 prompt_preview = excluded.prompt_preview,
433 status = 'running',
434 started_ts = excluded.started_ts",
435 )
436 .bind(run_id)
437 .bind(agent_id)
438 .bind(phase_id as i64)
439 .bind(model)
440 .bind(prompt_preview)
441 .execute(&self.pool)
442 .await?;
443 Ok(())
444 }
445
446 async fn write_delta(
447 &self,
448 run_id: RunId,
449 agent_id: AgentId,
450 delta: &ProgressDelta,
451 ) -> StorageResult<()> {
452 let now = Utc::now().to_rfc3339();
453 let phase_id: Option<i64> =
454 sqlx::query_scalar("SELECT phase_id FROM agents WHERE run_id = ? AND agent_id = ?")
455 .bind(run_id)
456 .bind(agent_id)
457 .fetch_optional(&self.pool)
458 .await?
459 .flatten();
460
461 match delta {
462 ProgressDelta::Message { text } => {
463 sqlx::query(
464 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind, role, text)
465 VALUES (?, ?, ?, ?, 'message', 'assistant', ?)",
466 )
467 .bind(run_id)
468 .bind(agent_id)
469 .bind(phase_id)
470 .bind(now)
471 .bind(text)
472 .execute(&self.pool)
473 .await?;
474 }
475 ProgressDelta::ToolCall { name, summary } => {
476 sqlx::query(
477 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind, name, text)
478 VALUES (?, ?, ?, ?, 'tool_call', ?, ?)",
479 )
480 .bind(run_id)
481 .bind(agent_id)
482 .bind(phase_id)
483 .bind(now)
484 .bind(name)
485 .bind(summary)
486 .execute(&self.pool)
487 .await?;
488 }
489 ProgressDelta::FileEdit { path } => {
490 sqlx::query(
491 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind, file_path)
492 VALUES (?, ?, ?, ?, 'file_edit', ?)",
493 )
494 .bind(run_id)
495 .bind(agent_id)
496 .bind(phase_id)
497 .bind(now)
498 .bind(path.to_string_lossy().to_string())
499 .execute(&self.pool)
500 .await?;
501 }
502 ProgressDelta::Tokens { usage } => {
503 sqlx::query(
504 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind,
505 input_tokens, output_tokens,
506 cache_read_tokens, cache_write_tokens)
507 VALUES (?, ?, ?, ?, 'tokens', ?, ?, ?, ?)",
508 )
509 .bind(run_id)
510 .bind(agent_id)
511 .bind(phase_id)
512 .bind(now)
513 .bind(usage.input as i64)
514 .bind(usage.output as i64)
515 .bind(usage.cache_read as i64)
516 .bind(usage.cache_write as i64)
517 .execute(&self.pool)
518 .await?;
519 }
520 }
521 Ok(())
522 }
523
524 async fn write_agent_done(
525 &self,
526 run_id: RunId,
527 agent_id: AgentId,
528 status: &AgentStatus,
529 tokens: luft_core::contract::ids::TokenUsage,
530 elapsed_ms: u64,
531 retry_count: u32,
532 ) -> StorageResult<()> {
533 sqlx::query(
534 "UPDATE agents
535 SET status = ?,
536 input_tokens = ?,
537 output_tokens = ?,
538 cache_read_tokens = ?,
539 cache_write_tokens = ?,
540 done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
541 elapsed_ms = ?,
542 retry_count = ?
543 WHERE run_id = ? AND agent_id = ?",
544 )
545 .bind(agent_status_str(status.clone()))
546 .bind(tokens.input as i64)
547 .bind(tokens.output as i64)
548 .bind(tokens.cache_read as i64)
549 .bind(tokens.cache_write as i64)
550 .bind(elapsed_ms as i64)
551 .bind(retry_count as i64)
552 .bind(run_id)
553 .bind(agent_id)
554 .execute(&self.pool)
555 .await?;
556 Ok(())
557 }
558
559 async fn write_phase_done(
560 &self,
561 run_id: RunId,
562 phase_id: PhaseId,
563 ok: usize,
564 failed: usize,
565 ) -> StorageResult<()> {
566 sqlx::query(
567 "UPDATE phases
568 SET ok = ?, failed = ?, done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now')
569 WHERE run_id = ? AND phase_id = ?",
570 )
571 .bind(ok as i64)
572 .bind(failed as i64)
573 .bind(run_id)
574 .bind(phase_id as i64)
575 .execute(&self.pool)
576 .await?;
577 Ok(())
578 }
579
580 async fn write_run_done(
581 &self,
582 run_id: RunId,
583 status: &RunStatus,
584 total_tokens: luft_core::contract::ids::TokenUsage,
585 report: &Json,
586 ) -> StorageResult<()> {
587 sqlx::query(
588 "UPDATE runs
589 SET status = ?,
590 finished_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
591 input_tokens = ?,
592 output_tokens = ?,
593 cache_read_tokens = ?,
594 cache_write_tokens = ?,
595 report = ?
596 WHERE run_id = ?",
597 )
598 .bind(run_status_str(*status))
599 .bind(total_tokens.input as i64)
600 .bind(total_tokens.output as i64)
601 .bind(total_tokens.cache_read as i64)
602 .bind(total_tokens.cache_write as i64)
603 .bind(serde_json::to_string(report)?)
604 .bind(run_id)
605 .execute(&self.pool)
606 .await?;
607 Ok(())
608 }
609
610 async fn write_report_emitted(
611 &self,
612 run_id: RunId,
613 phase_id: PhaseId,
614 report: &Json,
615 ) -> StorageResult<()> {
616 sqlx::query("UPDATE runs SET report = ? WHERE run_id = ?")
617 .bind(serde_json::to_string(report)?)
618 .bind(run_id)
619 .execute(&self.pool)
620 .await?;
621
622 sqlx::query(
623 "UPDATE phases SET done_ts = COALESCE(done_ts, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
624 WHERE run_id = ? AND phase_id = ?",
625 )
626 .bind(run_id)
627 .bind(phase_id as i64)
628 .execute(&self.pool)
629 .await?;
630 Ok(())
631 }
632
633 #[allow(clippy::too_many_arguments)]
634 async fn write_span_started(
635 &self,
636 run_id: RunId,
637 phase_id: PhaseId,
638 span_id: u64,
639 kind: &str,
640 items: Option<i64>,
641 max_rounds: Option<i64>,
642 path: Option<&str>,
643 _args: Option<&Json>,
644 ) -> StorageResult<()> {
645 sqlx::query(
646 "INSERT INTO spans (run_id, span_id, kind, phase_id, items, max_rounds,
647 path, started_ts)
648 VALUES (?, ?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
649 ON CONFLICT(run_id, span_id) DO UPDATE SET
650 kind = excluded.kind,
651 items = COALESCE(excluded.items, spans.items),
652 max_rounds = COALESCE(excluded.max_rounds, spans.max_rounds),
653 path = COALESCE(excluded.path, spans.path),
654 started_ts = COALESCE(spans.started_ts, excluded.started_ts)",
655 )
656 .bind(run_id)
657 .bind(span_id as i64)
658 .bind(kind)
659 .bind(phase_id as i64)
660 .bind(items)
661 .bind(max_rounds)
662 .bind(path)
663 .execute(&self.pool)
664 .await?;
665 Ok(())
666 }
667
668 #[allow(clippy::too_many_arguments)]
669 async fn write_span_done(
670 &self,
671 run_id: RunId,
672 span_id: u64,
673 ok: i64,
674 failed: i64,
675 result: Option<&Json>,
676 error: Option<String>,
677 _path: Option<&str>,
678 elapsed_ms: u64,
679 ) -> StorageResult<()> {
680 sqlx::query(
681 "UPDATE spans
682 SET ok = ?, failed = ?,
683 result = COALESCE(?, result),
684 error = ?,
685 done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
686 elapsed_ms = ?
687 WHERE run_id = ? AND span_id = ?",
688 )
689 .bind(ok)
690 .bind(failed)
691 .bind(result.map(serde_json::to_string).transpose()?)
692 .bind(error)
693 .bind(elapsed_ms as i64)
694 .bind(run_id)
695 .bind(span_id as i64)
696 .execute(&self.pool)
697 .await?;
698 Ok(())
699 }
700
701 #[allow(clippy::too_many_arguments)]
702 async fn write_converge_done(
703 &self,
704 run_id: RunId,
705 span_id: u64,
706 rounds: i64,
707 converged: bool,
708 result: Option<&Json>,
709 error: Option<String>,
710 elapsed_ms: u64,
711 ) -> StorageResult<()> {
712 sqlx::query(
713 "UPDATE spans
714 SET rounds = ?, converged = ?,
715 result = COALESCE(?, result),
716 error = ?,
717 done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
718 elapsed_ms = ?
719 WHERE run_id = ? AND span_id = ?",
720 )
721 .bind(rounds)
722 .bind(converged as i64)
723 .bind(result.map(serde_json::to_string).transpose()?)
724 .bind(error)
725 .bind(elapsed_ms as i64)
726 .bind(run_id)
727 .bind(span_id as i64)
728 .execute(&self.pool)
729 .await?;
730 Ok(())
731 }
732
733 async fn append_audit(&self, event: &AgentEvent) -> StorageResult<()> {
734 let (run_id, type_name) = audit_metadata(event);
735 if run_id.is_none() {
736 return Ok(());
737 }
738 if type_name == "acp_raw" {
740 return Ok(());
741 }
742 let payload = serde_json::to_string(event)?;
743 sqlx::query("INSERT INTO events (run_id, type, payload) VALUES (?, ?, ?)")
744 .bind(run_id.unwrap())
745 .bind(type_name)
746 .bind(payload)
747 .execute(&self.pool)
748 .await?;
749 Ok(())
750 }
751}
752
753fn audit_metadata(event: &AgentEvent) -> (Option<RunId>, &'static str) {
758 match event {
759 AgentEvent::RunStarted { run_id, .. } => (Some(*run_id), "run_started"),
760 AgentEvent::PhaseStarted { run_id, .. } => (Some(*run_id), "phase_started"),
761 AgentEvent::AgentStarted { run_id, .. } => (Some(*run_id), "agent_started"),
762 AgentEvent::AgentProgress { run_id, .. } => (Some(*run_id), "agent_progress"),
763 AgentEvent::AcpRaw { run_id, .. } => (Some(*run_id), "acp_raw"),
764 AgentEvent::AgentDone { run_id, .. } => (Some(*run_id), "agent_done"),
765 AgentEvent::PhaseDone { run_id, .. } => (Some(*run_id), "phase_done"),
766 AgentEvent::RunDone { run_id, .. } => (Some(*run_id), "run_done"),
767 AgentEvent::Log { run_id, .. } => (Some(*run_id), "log"),
768 AgentEvent::BudgetSet { run_id, .. } => (Some(*run_id), "budget_set"),
769 AgentEvent::ReportEmitted { run_id, .. } => (Some(*run_id), "report_emitted"),
770 AgentEvent::ParallelStarted { run_id, .. } => (Some(*run_id), "parallel_started"),
771 AgentEvent::ParallelDone { run_id, .. } => (Some(*run_id), "parallel_done"),
772 AgentEvent::WorkflowStarted { run_id, .. } => (Some(*run_id), "workflow_started"),
773 AgentEvent::WorkflowDone { run_id, .. } => (Some(*run_id), "workflow_done"),
774 AgentEvent::ConvergeStarted { run_id, .. } => (Some(*run_id), "converge_started"),
775 AgentEvent::ConvergeDone { run_id, .. } => (Some(*run_id), "converge_done"),
776 AgentEvent::PipelineStarted { run_id, .. } => (Some(*run_id), "pipeline_started"),
777 AgentEvent::PipelineStageStarted { run_id, .. } => {
778 (Some(*run_id), "pipeline_stage_started")
779 }
780 AgentEvent::PipelineItemDone { run_id, .. } => (Some(*run_id), "pipeline_item_done"),
781 AgentEvent::PipelineDone { run_id, .. } => (Some(*run_id), "pipeline_done"),
782 AgentEvent::PhaseSpanStarted { run_id, .. } => (Some(*run_id), "phase_span_started"),
783 AgentEvent::PhaseSpanDone { run_id, .. } => (Some(*run_id), "phase_span_done"),
784 AgentEvent::PlanPreview { run_id, .. } => (Some(*run_id), "plan_preview"),
785 AgentEvent::SignalReceived { run_id, .. } => (*run_id, "signal_received"),
786 AgentEvent::SchemaRetry { run_id, .. } => (Some(*run_id), "schema_retry"),
787 }
788}
789
790fn agent_status_str(s: AgentStatus) -> &'static str {
791 match s {
792 AgentStatus::Ok => "ok",
793 AgentStatus::Error => "error",
794 AgentStatus::Cancelled => "cancelled",
795 AgentStatus::TimedOut => "timed_out",
796 }
797}
798
799fn run_status_str(s: RunStatus) -> &'static str {
800 match s {
801 RunStatus::Completed => "completed",
802 RunStatus::Failed => "failed",
803 RunStatus::Cancelled => "cancelled",
804 RunStatus::Partial => "partial",
805 }
806}
807
808#[allow(dead_code)]
813pub(crate) async fn fetch_event_count(pool: &DbPool, run_id: RunId) -> StorageResult<i64> {
814 let row = sqlx::query("SELECT COUNT(*) AS c FROM events WHERE run_id = ?")
815 .bind(run_id)
816 .fetch_one(pool)
817 .await?;
818 Ok(row.try_get::<i64, _>("c")?)
819}
820
821#[allow(dead_code)]
822pub(crate) async fn fetch_turn_count(
823 pool: &DbPool,
824 run_id: RunId,
825 kind: &str,
826) -> StorageResult<i64> {
827 let row = sqlx::query("SELECT COUNT(*) AS c FROM turns WHERE run_id = ? AND kind = ?")
828 .bind(run_id)
829 .bind(kind)
830 .fetch_one(pool)
831 .await?;
832 Ok(row.try_get::<i64, _>("c")?)
833}
834
835pub type SharedWriter = Arc<EventWriter>;
837
838#[cfg(test)]
843mod tests {
844 use super::*;
845 use crate::db::open_db;
846 use luft_core::contract::event::LogLevel;
847 use luft_core::contract::ids::TokenUsage;
848 use std::path::PathBuf;
849 use tempfile::tempdir;
850
851 async fn setup() -> (tempfile::TempDir, EventWriter) {
852 let dir = tempdir().unwrap();
853 let pool = open_db(&dir.path().join("test.db")).await.unwrap();
854 (dir, EventWriter::new(pool))
855 }
856
857 #[tokio::test]
858 async fn run_lifecycle() {
859 let (_dir, w) = setup().await;
860 let run_id = uuid::Uuid::now_v7();
861 let ts = Utc::now();
862
863 w.write_event(&AgentEvent::RunStarted {
864 run_id,
865 task: "hello".into(),
866 ts,
867 })
868 .await
869 .unwrap();
870
871 let row = sqlx::query("SELECT task, status FROM runs WHERE run_id = ?")
872 .bind(run_id)
873 .fetch_one(w.pool())
874 .await
875 .unwrap();
876 let task: String = row.try_get("task").unwrap();
877 let status: String = row.try_get("status").unwrap();
878 assert_eq!(task, "hello");
879 assert_eq!(status, "running");
880
881 w.write_event(&AgentEvent::RunDone {
882 run_id,
883 status: RunStatus::Completed,
884 total_tokens: TokenUsage {
885 input: 100,
886 output: 50,
887 cache_read: 0,
888 cache_write: 0,
889 },
890 report: serde_json::json!({"result": "ok"}),
891 ts: chrono::Utc::now(),
892 })
893 .await
894 .unwrap();
895
896 let row = sqlx::query("SELECT status, finished_ts FROM runs WHERE run_id = ?")
897 .bind(run_id)
898 .fetch_one(w.pool())
899 .await
900 .unwrap();
901 let status: String = row.try_get("status").unwrap();
902 let finished_ts: Option<String> = row.try_get("finished_ts").unwrap();
903 assert_eq!(status, "completed");
904 assert!(finished_ts.is_some());
905 }
906
907 #[tokio::test]
908 async fn agent_started_creates_agent_row() {
909 let (_dir, w) = setup().await;
910 let run_id = uuid::Uuid::now_v7();
911 let agent_id = uuid::Uuid::now_v7();
912
913 w.write_event(&AgentEvent::AgentStarted {
914 run_id,
915 phase_id: 1,
916 agent_id,
917 prompt_preview: "do something".into(),
918 model: Some("claude-sonnet-4".into()),
919 description: None,
920 role: None,
921 name: None,
922 agent_seq: 0,
923 ts: Default::default(),
924 })
925 .await
926 .unwrap();
927
928 let row = sqlx::query(
929 "SELECT phase_id, model, status, prompt_preview FROM agents WHERE run_id = ? AND agent_id = ?",
930 )
931 .bind(run_id)
932 .bind(agent_id)
933 .fetch_one(w.pool())
934 .await
935 .unwrap();
936
937 let phase_id: i64 = row.try_get("phase_id").unwrap();
938 let model: String = row.try_get("model").unwrap();
939 let status: String = row.try_get("status").unwrap();
940 assert_eq!(phase_id, 1);
941 assert_eq!(model, "claude-sonnet-4");
942 assert_eq!(status, "running");
943 }
944
945 #[tokio::test]
946 async fn progress_delta_message_creates_turn() {
947 let (_dir, w) = setup().await;
948 let run_id = uuid::Uuid::now_v7();
949 let agent_id = uuid::Uuid::now_v7();
950
951 w.write_event(&AgentEvent::AgentStarted {
953 run_id,
954 phase_id: 2,
955 agent_id,
956 prompt_preview: "p".into(),
957 model: None,
958 description: None,
959 role: None,
960 name: None,
961 agent_seq: 0,
962 ts: Default::default(),
963 })
964 .await
965 .unwrap();
966
967 w.write_event(&AgentEvent::AgentProgress {
968 run_id,
969 agent_id,
970 delta: ProgressDelta::Message {
971 text: "thinking...".into(),
972 },
973 })
974 .await
975 .unwrap();
976
977 let count = fetch_turn_count(w.pool(), run_id, "message").await.unwrap();
978 assert_eq!(count, 1);
979
980 let row = sqlx::query("SELECT text, role, phase_id FROM turns WHERE run_id = ? LIMIT 1")
981 .bind(run_id)
982 .fetch_one(w.pool())
983 .await
984 .unwrap();
985 let text: String = row.try_get("text").unwrap();
986 let role: String = row.try_get("role").unwrap();
987 let phase_id: i64 = row.try_get("phase_id").unwrap();
988 assert_eq!(text, "thinking...");
989 assert_eq!(role, "assistant");
990 assert_eq!(phase_id, 2);
991 }
992
993 #[tokio::test]
994 async fn progress_delta_tool_call_and_file_edit() {
995 let (_dir, w) = setup().await;
996 let run_id = uuid::Uuid::now_v7();
997 let agent_id = uuid::Uuid::now_v7();
998
999 w.write_event(&AgentEvent::AgentStarted {
1000 run_id,
1001 phase_id: 0,
1002 agent_id,
1003 prompt_preview: "".into(),
1004 model: None,
1005 description: None,
1006 role: None,
1007 name: None,
1008 agent_seq: 0,
1009 ts: Default::default(),
1010 })
1011 .await
1012 .unwrap();
1013
1014 w.write_event(&AgentEvent::AgentProgress {
1015 run_id,
1016 agent_id,
1017 delta: ProgressDelta::ToolCall {
1018 name: "ReadFile".into(),
1019 summary: "read".into(),
1020 },
1021 })
1022 .await
1023 .unwrap();
1024
1025 w.write_event(&AgentEvent::AgentProgress {
1026 run_id,
1027 agent_id,
1028 delta: ProgressDelta::FileEdit {
1029 path: PathBuf::from("src/main.rs"),
1030 },
1031 })
1032 .await
1033 .unwrap();
1034
1035 assert_eq!(
1036 fetch_turn_count(w.pool(), run_id, "tool_call")
1037 .await
1038 .unwrap(),
1039 1
1040 );
1041 assert_eq!(
1042 fetch_turn_count(w.pool(), run_id, "file_edit")
1043 .await
1044 .unwrap(),
1045 1
1046 );
1047 }
1048
1049 #[tokio::test]
1050 async fn progress_delta_tokens_writes_agents_row() {
1051 let (_dir, w) = setup().await;
1052 let run_id = uuid::Uuid::now_v7();
1053 let agent_id = uuid::Uuid::now_v7();
1054
1055 w.write_event(&AgentEvent::AgentStarted {
1056 run_id,
1057 phase_id: 0,
1058 agent_id,
1059 prompt_preview: "".into(),
1060 model: None,
1061 description: None,
1062 role: None,
1063 name: None,
1064 agent_seq: 0,
1065 ts: Default::default(),
1066 })
1067 .await
1068 .unwrap();
1069
1070 w.write_event(&AgentEvent::AgentProgress {
1071 run_id,
1072 agent_id,
1073 delta: ProgressDelta::Tokens {
1074 usage: TokenUsage {
1075 input: 10,
1076 output: 5,
1077 cache_read: 0,
1078 cache_write: 0,
1079 },
1080 },
1081 })
1082 .await
1083 .unwrap();
1084
1085 assert_eq!(
1086 fetch_turn_count(w.pool(), run_id, "tokens").await.unwrap(),
1087 1
1088 );
1089 }
1090
1091 #[tokio::test]
1092 async fn agent_done_updates_status_and_tokens() {
1093 let (_dir, w) = setup().await;
1094 let run_id = uuid::Uuid::now_v7();
1095 let agent_id = uuid::Uuid::now_v7();
1096
1097 w.write_event(&AgentEvent::AgentStarted {
1098 run_id,
1099 phase_id: 0,
1100 agent_id,
1101 prompt_preview: "".into(),
1102 model: None,
1103 description: None,
1104 role: None,
1105 name: None,
1106 agent_seq: 0,
1107 ts: Default::default(),
1108 })
1109 .await
1110 .unwrap();
1111
1112 w.write_event(&AgentEvent::AgentDone {
1113 run_id,
1114 agent_id,
1115 status: AgentStatus::Ok,
1116 tokens: TokenUsage {
1117 input: 200,
1118 output: 80,
1119 cache_read: 0,
1120 cache_write: 0,
1121 },
1122 elapsed_ms: 1234,
1123 name: None,
1124 agent_seq: 0,
1125 output: serde_json::Value::Null,
1126 findings: Vec::new(),
1127 prompt: String::new(),
1128 retry_count: 0,
1129 ts: Default::default(),
1130 })
1131 .await
1132 .unwrap();
1133
1134 let row = sqlx::query(
1135 "SELECT status, input_tokens, output_tokens, elapsed_ms FROM agents WHERE run_id = ? AND agent_id = ?",
1136 )
1137 .bind(run_id)
1138 .bind(agent_id)
1139 .fetch_one(w.pool())
1140 .await
1141 .unwrap();
1142
1143 let status: String = row.try_get("status").unwrap();
1144 let input: i64 = row.try_get("input_tokens").unwrap();
1145 let output: i64 = row.try_get("output_tokens").unwrap();
1146 let elapsed: i64 = row.try_get("elapsed_ms").unwrap();
1147 assert_eq!(status, "ok");
1148 assert_eq!(input, 200);
1149 assert_eq!(output, 80);
1150 assert_eq!(elapsed, 1234);
1151 }
1152
1153 #[tokio::test]
1154 async fn phase_started_and_done() {
1155 let (_dir, w) = setup().await;
1156 let run_id = uuid::Uuid::now_v7();
1157
1158 w.write_event(&AgentEvent::RunStarted {
1159 run_id,
1160 task: "t".into(),
1161 ts: Utc::now(),
1162 })
1163 .await
1164 .unwrap();
1165
1166 w.write_event(&AgentEvent::PhaseStarted {
1167 run_id,
1168 phase_id: 1,
1169 label: "explore".into(),
1170 planned: 3,
1171 parent_span_id: None,
1172 description: None,
1173 role: None,
1174 ts: chrono::Utc::now(),
1175 })
1176 .await
1177 .unwrap();
1178
1179 w.write_event(&AgentEvent::PhaseDone {
1180 run_id,
1181 phase_id: 1,
1182 ok: 2,
1183 failed: 1,
1184 ts: chrono::Utc::now(),
1185 })
1186 .await
1187 .unwrap();
1188
1189 let row = sqlx::query(
1190 "SELECT label, planned, ok, failed FROM phases WHERE run_id = ? AND phase_id = ?",
1191 )
1192 .bind(run_id)
1193 .bind(1i64)
1194 .fetch_one(w.pool())
1195 .await
1196 .unwrap();
1197 let label: String = row.try_get("label").unwrap();
1198 let planned: i64 = row.try_get("planned").unwrap();
1199 let ok: i64 = row.try_get("ok").unwrap();
1200 let failed: i64 = row.try_get("failed").unwrap();
1201 assert_eq!(label, "explore");
1202 assert_eq!(planned, 3);
1203 assert_eq!(ok, 2);
1204 assert_eq!(failed, 1);
1205 }
1206
1207 #[tokio::test]
1208 async fn spans_track_orchestration() {
1209 let (_dir, w) = setup().await;
1210 let run_id = uuid::Uuid::now_v7();
1211
1212 w.write_event(&AgentEvent::RunStarted {
1213 run_id,
1214 task: "t".into(),
1215 ts: Utc::now(),
1216 })
1217 .await
1218 .unwrap();
1219
1220 w.write_event(&AgentEvent::ParallelStarted {
1221 run_id,
1222 phase_id: 1,
1223 span_id: 7,
1224 count: 4,
1225 })
1226 .await
1227 .unwrap();
1228
1229 w.write_event(&AgentEvent::ParallelDone {
1230 run_id,
1231 phase_id: 1,
1232 span_id: 7,
1233 ok: 3,
1234 failed: 1,
1235 results: serde_json::json!([1, 2, 3]),
1236 elapsed_ms: 999,
1237 })
1238 .await
1239 .unwrap();
1240
1241 let row = sqlx::query(
1242 "SELECT kind, items, ok, failed, elapsed_ms FROM spans WHERE run_id = ? AND span_id = ?",
1243 )
1244 .bind(run_id)
1245 .bind(7i64)
1246 .fetch_one(w.pool())
1247 .await
1248 .unwrap();
1249 let kind: String = row.try_get("kind").unwrap();
1250 let items: i64 = row.try_get("items").unwrap();
1251 let ok: i64 = row.try_get("ok").unwrap();
1252 let failed: i64 = row.try_get("failed").unwrap();
1253 assert_eq!(kind, "parallel");
1254 assert_eq!(items, 4);
1255 assert_eq!(ok, 3);
1256 assert_eq!(failed, 1);
1257 }
1258
1259 #[tokio::test]
1260 async fn audit_log_captures_all_events() {
1261 let (_dir, w) = setup().await;
1262 let run_id = uuid::Uuid::now_v7();
1263
1264 w.write_event(&AgentEvent::RunStarted {
1265 run_id,
1266 task: "t".into(),
1267 ts: Utc::now(),
1268 })
1269 .await
1270 .unwrap();
1271
1272 w.write_event(&AgentEvent::Log {
1273 run_id,
1274 agent_id: None,
1275 level: LogLevel::Info,
1276 msg: "hi".into(),
1277 })
1278 .await
1279 .unwrap();
1280
1281 let count = fetch_event_count(w.pool(), run_id).await.unwrap();
1283 assert!(count >= 1);
1284 }
1285
1286 #[tokio::test]
1287 async fn cascade_delete_removes_all() {
1288 let (_dir, w) = setup().await;
1289 let run_id = uuid::Uuid::now_v7();
1290
1291 w.write_event(&AgentEvent::RunStarted {
1292 run_id,
1293 task: "t".into(),
1294 ts: Utc::now(),
1295 })
1296 .await
1297 .unwrap();
1298
1299 let agent_id = uuid::Uuid::now_v7();
1300 w.write_event(&AgentEvent::AgentStarted {
1301 run_id,
1302 phase_id: 0,
1303 agent_id,
1304 prompt_preview: "".into(),
1305 model: None,
1306 description: None,
1307 role: None,
1308 name: None,
1309 agent_seq: 0,
1310 ts: Default::default(),
1311 })
1312 .await
1313 .unwrap();
1314
1315 w.write_event(&AgentEvent::AgentProgress {
1316 run_id,
1317 agent_id,
1318 delta: ProgressDelta::Message { text: "hi".into() },
1319 })
1320 .await
1321 .unwrap();
1322
1323 sqlx::query("DELETE FROM runs WHERE run_id = ?")
1324 .bind(run_id)
1325 .execute(w.pool())
1326 .await
1327 .unwrap();
1328
1329 let turn_count = fetch_turn_count(w.pool(), run_id, "message").await.unwrap();
1330 let agent_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM agents WHERE run_id = ?")
1331 .bind(run_id)
1332 .fetch_one(w.pool())
1333 .await
1334 .unwrap();
1335 assert_eq!(turn_count, 0);
1336 assert_eq!(agent_count, 0);
1337 }
1338
1339 #[tokio::test]
1340 async fn acp_raw_is_skipped() {
1341 let (_dir, w) = setup().await;
1342 let run_id = uuid::Uuid::now_v7();
1343 let agent_id = uuid::Uuid::now_v7();
1344
1345 w.write_event(&AgentEvent::AcpRaw {
1346 run_id,
1347 agent_id,
1348 kind: "agent_message_chunk".into(),
1349 raw: serde_json::json!({"text": "chunk"}),
1350 })
1351 .await
1352 .unwrap();
1353
1354 let count = fetch_event_count(w.pool(), run_id).await.unwrap();
1355 assert_eq!(count, 0);
1356 }
1357}