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::AcpRequest { .. } => {}
346 | AgentEvent::PlanPreview { .. }
349 | AgentEvent::SchemaRetry { .. } => {}
350 }
351
352 self.append_audit(event).await?;
354 Ok(())
355 }
356
357 async fn write_run_started(
362 &self,
363 run_id: RunId,
364 task: &str,
365 ts: DateTime<Utc>,
366 ) -> StorageResult<()> {
367 sqlx::query(
368 "INSERT OR IGNORE INTO runs (run_id, task, status, started_ts)
369 VALUES (?, ?, 'running', ?)",
370 )
371 .bind(run_id)
372 .bind(task)
373 .bind(ts.to_rfc3339())
374 .execute(&self.pool)
375 .await?;
376 Ok(())
377 }
378
379 async fn write_phase_started(
380 &self,
381 run_id: RunId,
382 phase_id: PhaseId,
383 label: &str,
384 planned: usize,
385 description: Option<&str>,
386 role: Option<&str>,
387 ) -> StorageResult<()> {
388 sqlx::query(
389 "INSERT INTO phases (run_id, phase_id, label, planned, description, role, started_ts)
390 VALUES (?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
391 ON CONFLICT(run_id, phase_id) DO UPDATE SET
392 label = excluded.label,
393 planned = excluded.planned,
394 description = COALESCE(excluded.description, phases.description),
395 role = COALESCE(excluded.role, phases.role),
396 started_ts = COALESCE(phases.started_ts, excluded.started_ts)",
397 )
398 .bind(run_id)
399 .bind(phase_id as i64)
400 .bind(label)
401 .bind(planned as i64)
402 .bind(description)
403 .bind(role)
404 .execute(&self.pool)
405 .await?;
406 Ok(())
407 }
408
409 async fn write_agent_started(
410 &self,
411 run_id: RunId,
412 phase_id: PhaseId,
413 agent_id: AgentId,
414 prompt_preview: &str,
415 model: Option<&str>,
416 ) -> StorageResult<()> {
417 sqlx::query(
418 "INSERT INTO runs (run_id, task, status, started_ts)
419 VALUES (?, '', 'running', strftime('%Y-%m-%dT%H:%M:%fZ','now'))
420 ON CONFLICT(run_id) DO NOTHING",
421 )
422 .bind(run_id)
423 .execute(&self.pool)
424 .await?;
425
426 sqlx::query(
427 "INSERT INTO agents (run_id, agent_id, phase_id, model, status,
428 prompt_preview, started_ts)
429 VALUES (?, ?, ?, ?, 'running', ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
430 ON CONFLICT(run_id, agent_id) DO UPDATE SET
431 phase_id = excluded.phase_id,
432 model = excluded.model,
433 prompt_preview = excluded.prompt_preview,
434 status = 'running',
435 started_ts = excluded.started_ts",
436 )
437 .bind(run_id)
438 .bind(agent_id)
439 .bind(phase_id as i64)
440 .bind(model)
441 .bind(prompt_preview)
442 .execute(&self.pool)
443 .await?;
444 Ok(())
445 }
446
447 async fn write_delta(
448 &self,
449 run_id: RunId,
450 agent_id: AgentId,
451 delta: &ProgressDelta,
452 ) -> StorageResult<()> {
453 let now = Utc::now().to_rfc3339();
454 let phase_id: Option<i64> =
455 sqlx::query_scalar("SELECT phase_id FROM agents WHERE run_id = ? AND agent_id = ?")
456 .bind(run_id)
457 .bind(agent_id)
458 .fetch_optional(&self.pool)
459 .await?
460 .flatten();
461
462 match delta {
463 ProgressDelta::Message { text } => {
464 sqlx::query(
465 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind, role, text)
466 VALUES (?, ?, ?, ?, 'message', 'assistant', ?)",
467 )
468 .bind(run_id)
469 .bind(agent_id)
470 .bind(phase_id)
471 .bind(now)
472 .bind(text)
473 .execute(&self.pool)
474 .await?;
475 }
476 ProgressDelta::ToolCall { name, summary } => {
477 sqlx::query(
478 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind, name, text)
479 VALUES (?, ?, ?, ?, 'tool_call', ?, ?)",
480 )
481 .bind(run_id)
482 .bind(agent_id)
483 .bind(phase_id)
484 .bind(now)
485 .bind(name)
486 .bind(summary)
487 .execute(&self.pool)
488 .await?;
489 }
490 ProgressDelta::FileEdit { path } => {
491 sqlx::query(
492 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind, file_path)
493 VALUES (?, ?, ?, ?, 'file_edit', ?)",
494 )
495 .bind(run_id)
496 .bind(agent_id)
497 .bind(phase_id)
498 .bind(now)
499 .bind(path.to_string_lossy().to_string())
500 .execute(&self.pool)
501 .await?;
502 }
503 ProgressDelta::Tokens { usage } => {
504 sqlx::query(
505 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind,
506 input_tokens, output_tokens,
507 cache_read_tokens, cache_write_tokens)
508 VALUES (?, ?, ?, ?, 'tokens', ?, ?, ?, ?)",
509 )
510 .bind(run_id)
511 .bind(agent_id)
512 .bind(phase_id)
513 .bind(now)
514 .bind(usage.input as i64)
515 .bind(usage.output as i64)
516 .bind(usage.cache_read as i64)
517 .bind(usage.cache_write as i64)
518 .execute(&self.pool)
519 .await?;
520 }
521 }
522 Ok(())
523 }
524
525 async fn write_agent_done(
526 &self,
527 run_id: RunId,
528 agent_id: AgentId,
529 status: &AgentStatus,
530 tokens: luft_core::contract::ids::TokenUsage,
531 elapsed_ms: u64,
532 retry_count: u32,
533 ) -> StorageResult<()> {
534 sqlx::query(
535 "UPDATE agents
536 SET status = ?,
537 input_tokens = ?,
538 output_tokens = ?,
539 cache_read_tokens = ?,
540 cache_write_tokens = ?,
541 done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
542 elapsed_ms = ?,
543 retry_count = ?
544 WHERE run_id = ? AND agent_id = ?",
545 )
546 .bind(agent_status_str(status.clone()))
547 .bind(tokens.input as i64)
548 .bind(tokens.output as i64)
549 .bind(tokens.cache_read as i64)
550 .bind(tokens.cache_write as i64)
551 .bind(elapsed_ms as i64)
552 .bind(retry_count as i64)
553 .bind(run_id)
554 .bind(agent_id)
555 .execute(&self.pool)
556 .await?;
557 Ok(())
558 }
559
560 async fn write_phase_done(
561 &self,
562 run_id: RunId,
563 phase_id: PhaseId,
564 ok: usize,
565 failed: usize,
566 ) -> StorageResult<()> {
567 sqlx::query(
568 "UPDATE phases
569 SET ok = ?, failed = ?, done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now')
570 WHERE run_id = ? AND phase_id = ?",
571 )
572 .bind(ok as i64)
573 .bind(failed as i64)
574 .bind(run_id)
575 .bind(phase_id as i64)
576 .execute(&self.pool)
577 .await?;
578 Ok(())
579 }
580
581 async fn write_run_done(
582 &self,
583 run_id: RunId,
584 status: &RunStatus,
585 total_tokens: luft_core::contract::ids::TokenUsage,
586 report: &Json,
587 ) -> StorageResult<()> {
588 sqlx::query(
589 "UPDATE runs
590 SET status = ?,
591 finished_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
592 input_tokens = ?,
593 output_tokens = ?,
594 cache_read_tokens = ?,
595 cache_write_tokens = ?,
596 report = ?
597 WHERE run_id = ?",
598 )
599 .bind(run_status_str(*status))
600 .bind(total_tokens.input as i64)
601 .bind(total_tokens.output as i64)
602 .bind(total_tokens.cache_read as i64)
603 .bind(total_tokens.cache_write as i64)
604 .bind(serde_json::to_string(report)?)
605 .bind(run_id)
606 .execute(&self.pool)
607 .await?;
608 Ok(())
609 }
610
611 async fn write_report_emitted(
612 &self,
613 run_id: RunId,
614 phase_id: PhaseId,
615 report: &Json,
616 ) -> StorageResult<()> {
617 sqlx::query("UPDATE runs SET report = ? WHERE run_id = ?")
618 .bind(serde_json::to_string(report)?)
619 .bind(run_id)
620 .execute(&self.pool)
621 .await?;
622
623 sqlx::query(
624 "UPDATE phases SET done_ts = COALESCE(done_ts, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
625 WHERE run_id = ? AND phase_id = ?",
626 )
627 .bind(run_id)
628 .bind(phase_id as i64)
629 .execute(&self.pool)
630 .await?;
631 Ok(())
632 }
633
634 #[allow(clippy::too_many_arguments)]
635 async fn write_span_started(
636 &self,
637 run_id: RunId,
638 phase_id: PhaseId,
639 span_id: u64,
640 kind: &str,
641 items: Option<i64>,
642 max_rounds: Option<i64>,
643 path: Option<&str>,
644 _args: Option<&Json>,
645 ) -> StorageResult<()> {
646 sqlx::query(
647 "INSERT INTO spans (run_id, span_id, kind, phase_id, items, max_rounds,
648 path, started_ts)
649 VALUES (?, ?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
650 ON CONFLICT(run_id, span_id) DO UPDATE SET
651 kind = excluded.kind,
652 items = COALESCE(excluded.items, spans.items),
653 max_rounds = COALESCE(excluded.max_rounds, spans.max_rounds),
654 path = COALESCE(excluded.path, spans.path),
655 started_ts = COALESCE(spans.started_ts, excluded.started_ts)",
656 )
657 .bind(run_id)
658 .bind(span_id as i64)
659 .bind(kind)
660 .bind(phase_id as i64)
661 .bind(items)
662 .bind(max_rounds)
663 .bind(path)
664 .execute(&self.pool)
665 .await?;
666 Ok(())
667 }
668
669 #[allow(clippy::too_many_arguments)]
670 async fn write_span_done(
671 &self,
672 run_id: RunId,
673 span_id: u64,
674 ok: i64,
675 failed: i64,
676 result: Option<&Json>,
677 error: Option<String>,
678 _path: Option<&str>,
679 elapsed_ms: u64,
680 ) -> StorageResult<()> {
681 sqlx::query(
682 "UPDATE spans
683 SET ok = ?, failed = ?,
684 result = COALESCE(?, result),
685 error = ?,
686 done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
687 elapsed_ms = ?
688 WHERE run_id = ? AND span_id = ?",
689 )
690 .bind(ok)
691 .bind(failed)
692 .bind(result.map(serde_json::to_string).transpose()?)
693 .bind(error)
694 .bind(elapsed_ms as i64)
695 .bind(run_id)
696 .bind(span_id as i64)
697 .execute(&self.pool)
698 .await?;
699 Ok(())
700 }
701
702 #[allow(clippy::too_many_arguments)]
703 async fn write_converge_done(
704 &self,
705 run_id: RunId,
706 span_id: u64,
707 rounds: i64,
708 converged: bool,
709 result: Option<&Json>,
710 error: Option<String>,
711 elapsed_ms: u64,
712 ) -> StorageResult<()> {
713 sqlx::query(
714 "UPDATE spans
715 SET rounds = ?, converged = ?,
716 result = COALESCE(?, result),
717 error = ?,
718 done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
719 elapsed_ms = ?
720 WHERE run_id = ? AND span_id = ?",
721 )
722 .bind(rounds)
723 .bind(converged as i64)
724 .bind(result.map(serde_json::to_string).transpose()?)
725 .bind(error)
726 .bind(elapsed_ms as i64)
727 .bind(run_id)
728 .bind(span_id as i64)
729 .execute(&self.pool)
730 .await?;
731 Ok(())
732 }
733
734 async fn append_audit(&self, event: &AgentEvent) -> StorageResult<()> {
735 let (run_id, type_name) = audit_metadata(event);
736 if run_id.is_none() {
737 return Ok(());
738 }
739 if type_name == "acp_raw" {
741 return Ok(());
742 }
743 let payload = serde_json::to_string(event)?;
744 sqlx::query("INSERT INTO events (run_id, type, payload) VALUES (?, ?, ?)")
745 .bind(run_id.unwrap())
746 .bind(type_name)
747 .bind(payload)
748 .execute(&self.pool)
749 .await?;
750 Ok(())
751 }
752}
753
754fn audit_metadata(event: &AgentEvent) -> (Option<RunId>, &'static str) {
759 match event {
760 AgentEvent::RunStarted { run_id, .. } => (Some(*run_id), "run_started"),
761 AgentEvent::PhaseStarted { run_id, .. } => (Some(*run_id), "phase_started"),
762 AgentEvent::AgentStarted { run_id, .. } => (Some(*run_id), "agent_started"),
763 AgentEvent::AgentProgress { run_id, .. } => (Some(*run_id), "agent_progress"),
764 AgentEvent::AcpRaw { run_id, .. } => (Some(*run_id), "acp_raw"),
765 AgentEvent::AcpRequest { run_id, .. } => (Some(*run_id), "acp_request"),
766 AgentEvent::AgentDone { run_id, .. } => (Some(*run_id), "agent_done"),
767 AgentEvent::PhaseDone { run_id, .. } => (Some(*run_id), "phase_done"),
768 AgentEvent::RunDone { run_id, .. } => (Some(*run_id), "run_done"),
769 AgentEvent::Log { run_id, .. } => (Some(*run_id), "log"),
770 AgentEvent::BudgetSet { run_id, .. } => (Some(*run_id), "budget_set"),
771 AgentEvent::ReportEmitted { run_id, .. } => (Some(*run_id), "report_emitted"),
772 AgentEvent::ParallelStarted { run_id, .. } => (Some(*run_id), "parallel_started"),
773 AgentEvent::ParallelDone { run_id, .. } => (Some(*run_id), "parallel_done"),
774 AgentEvent::WorkflowStarted { run_id, .. } => (Some(*run_id), "workflow_started"),
775 AgentEvent::WorkflowDone { run_id, .. } => (Some(*run_id), "workflow_done"),
776 AgentEvent::ConvergeStarted { run_id, .. } => (Some(*run_id), "converge_started"),
777 AgentEvent::ConvergeDone { run_id, .. } => (Some(*run_id), "converge_done"),
778 AgentEvent::PipelineStarted { run_id, .. } => (Some(*run_id), "pipeline_started"),
779 AgentEvent::PipelineStageStarted { run_id, .. } => {
780 (Some(*run_id), "pipeline_stage_started")
781 }
782 AgentEvent::PipelineItemDone { run_id, .. } => (Some(*run_id), "pipeline_item_done"),
783 AgentEvent::PipelineDone { run_id, .. } => (Some(*run_id), "pipeline_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 description: None,
1172 role: None,
1173 ts: chrono::Utc::now(),
1174 })
1175 .await
1176 .unwrap();
1177
1178 w.write_event(&AgentEvent::PhaseDone {
1179 run_id,
1180 phase_id: 1,
1181 ok: 2,
1182 failed: 1,
1183 ts: chrono::Utc::now(),
1184 })
1185 .await
1186 .unwrap();
1187
1188 let row = sqlx::query(
1189 "SELECT label, planned, ok, failed FROM phases WHERE run_id = ? AND phase_id = ?",
1190 )
1191 .bind(run_id)
1192 .bind(1i64)
1193 .fetch_one(w.pool())
1194 .await
1195 .unwrap();
1196 let label: String = row.try_get("label").unwrap();
1197 let planned: i64 = row.try_get("planned").unwrap();
1198 let ok: i64 = row.try_get("ok").unwrap();
1199 let failed: i64 = row.try_get("failed").unwrap();
1200 assert_eq!(label, "explore");
1201 assert_eq!(planned, 3);
1202 assert_eq!(ok, 2);
1203 assert_eq!(failed, 1);
1204 }
1205
1206 #[tokio::test]
1207 async fn spans_track_orchestration() {
1208 let (_dir, w) = setup().await;
1209 let run_id = uuid::Uuid::now_v7();
1210
1211 w.write_event(&AgentEvent::RunStarted {
1212 run_id,
1213 task: "t".into(),
1214 ts: Utc::now(),
1215 })
1216 .await
1217 .unwrap();
1218
1219 w.write_event(&AgentEvent::ParallelStarted {
1220 run_id,
1221 phase_id: 1,
1222 span_id: 7,
1223 count: 4,
1224 })
1225 .await
1226 .unwrap();
1227
1228 w.write_event(&AgentEvent::ParallelDone {
1229 run_id,
1230 phase_id: 1,
1231 span_id: 7,
1232 ok: 3,
1233 failed: 1,
1234 results: serde_json::json!([1, 2, 3]),
1235 elapsed_ms: 999,
1236 })
1237 .await
1238 .unwrap();
1239
1240 let row = sqlx::query(
1241 "SELECT kind, items, ok, failed, elapsed_ms FROM spans WHERE run_id = ? AND span_id = ?",
1242 )
1243 .bind(run_id)
1244 .bind(7i64)
1245 .fetch_one(w.pool())
1246 .await
1247 .unwrap();
1248 let kind: String = row.try_get("kind").unwrap();
1249 let items: i64 = row.try_get("items").unwrap();
1250 let ok: i64 = row.try_get("ok").unwrap();
1251 let failed: i64 = row.try_get("failed").unwrap();
1252 assert_eq!(kind, "parallel");
1253 assert_eq!(items, 4);
1254 assert_eq!(ok, 3);
1255 assert_eq!(failed, 1);
1256 }
1257
1258 #[tokio::test]
1259 async fn audit_log_captures_all_events() {
1260 let (_dir, w) = setup().await;
1261 let run_id = uuid::Uuid::now_v7();
1262
1263 w.write_event(&AgentEvent::RunStarted {
1264 run_id,
1265 task: "t".into(),
1266 ts: Utc::now(),
1267 })
1268 .await
1269 .unwrap();
1270
1271 w.write_event(&AgentEvent::Log {
1272 run_id,
1273 agent_id: None,
1274 level: LogLevel::Info,
1275 msg: "hi".into(),
1276 })
1277 .await
1278 .unwrap();
1279
1280 let count = fetch_event_count(w.pool(), run_id).await.unwrap();
1282 assert!(count >= 1);
1283 }
1284
1285 #[tokio::test]
1286 async fn cascade_delete_removes_all() {
1287 let (_dir, w) = setup().await;
1288 let run_id = uuid::Uuid::now_v7();
1289
1290 w.write_event(&AgentEvent::RunStarted {
1291 run_id,
1292 task: "t".into(),
1293 ts: Utc::now(),
1294 })
1295 .await
1296 .unwrap();
1297
1298 let agent_id = uuid::Uuid::now_v7();
1299 w.write_event(&AgentEvent::AgentStarted {
1300 run_id,
1301 phase_id: 0,
1302 agent_id,
1303 prompt_preview: "".into(),
1304 model: None,
1305 description: None,
1306 role: None,
1307 name: None,
1308 agent_seq: 0,
1309 ts: Default::default(),
1310 })
1311 .await
1312 .unwrap();
1313
1314 w.write_event(&AgentEvent::AgentProgress {
1315 run_id,
1316 agent_id,
1317 delta: ProgressDelta::Message { text: "hi".into() },
1318 })
1319 .await
1320 .unwrap();
1321
1322 sqlx::query("DELETE FROM runs WHERE run_id = ?")
1323 .bind(run_id)
1324 .execute(w.pool())
1325 .await
1326 .unwrap();
1327
1328 let turn_count = fetch_turn_count(w.pool(), run_id, "message").await.unwrap();
1329 let agent_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM agents WHERE run_id = ?")
1330 .bind(run_id)
1331 .fetch_one(w.pool())
1332 .await
1333 .unwrap();
1334 assert_eq!(turn_count, 0);
1335 assert_eq!(agent_count, 0);
1336 }
1337
1338 #[tokio::test]
1339 async fn acp_raw_is_skipped() {
1340 let (_dir, w) = setup().await;
1341 let run_id = uuid::Uuid::now_v7();
1342 let agent_id = uuid::Uuid::now_v7();
1343
1344 w.write_event(&AgentEvent::AcpRaw {
1345 run_id,
1346 agent_id,
1347 kind: "agent_message_chunk".into(),
1348 raw: serde_json::json!({"text": "chunk"}),
1349 })
1350 .await
1351 .unwrap();
1352
1353 let count = fetch_event_count(w.pool(), run_id).await.unwrap();
1354 assert_eq!(count, 0);
1355 }
1356}