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