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::PlanPreview { .. }
346 | AgentEvent::SchemaRetry { .. } => {}
347 }
348
349 self.append_audit(event).await?;
351 Ok(())
352 }
353
354 async fn write_run_started(
359 &self,
360 run_id: RunId,
361 task: &str,
362 ts: DateTime<Utc>,
363 ) -> StorageResult<()> {
364 sqlx::query(
365 "INSERT OR IGNORE INTO runs (run_id, task, status, started_ts)
366 VALUES (?, ?, 'running', ?)",
367 )
368 .bind(run_id)
369 .bind(task)
370 .bind(ts.to_rfc3339())
371 .execute(&self.pool)
372 .await?;
373 Ok(())
374 }
375
376 async fn write_phase_started(
377 &self,
378 run_id: RunId,
379 phase_id: PhaseId,
380 label: &str,
381 planned: usize,
382 description: Option<&str>,
383 role: Option<&str>,
384 ) -> StorageResult<()> {
385 sqlx::query(
386 "INSERT INTO phases (run_id, phase_id, label, planned, description, role, started_ts)
387 VALUES (?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
388 ON CONFLICT(run_id, phase_id) DO UPDATE SET
389 label = excluded.label,
390 planned = excluded.planned,
391 description = COALESCE(excluded.description, phases.description),
392 role = COALESCE(excluded.role, phases.role),
393 started_ts = COALESCE(phases.started_ts, excluded.started_ts)",
394 )
395 .bind(run_id)
396 .bind(phase_id as i64)
397 .bind(label)
398 .bind(planned as i64)
399 .bind(description)
400 .bind(role)
401 .execute(&self.pool)
402 .await?;
403 Ok(())
404 }
405
406 async fn write_agent_started(
407 &self,
408 run_id: RunId,
409 phase_id: PhaseId,
410 agent_id: AgentId,
411 prompt_preview: &str,
412 model: Option<&str>,
413 ) -> StorageResult<()> {
414 sqlx::query(
415 "INSERT INTO runs (run_id, task, status, started_ts)
416 VALUES (?, '', 'running', strftime('%Y-%m-%dT%H:%M:%fZ','now'))
417 ON CONFLICT(run_id) DO NOTHING",
418 )
419 .bind(run_id)
420 .execute(&self.pool)
421 .await?;
422
423 sqlx::query(
424 "INSERT INTO agents (run_id, agent_id, phase_id, model, status,
425 prompt_preview, started_ts)
426 VALUES (?, ?, ?, ?, 'running', ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
427 ON CONFLICT(run_id, agent_id) DO UPDATE SET
428 phase_id = excluded.phase_id,
429 model = excluded.model,
430 prompt_preview = excluded.prompt_preview,
431 status = 'running',
432 started_ts = excluded.started_ts",
433 )
434 .bind(run_id)
435 .bind(agent_id)
436 .bind(phase_id as i64)
437 .bind(model)
438 .bind(prompt_preview)
439 .execute(&self.pool)
440 .await?;
441 Ok(())
442 }
443
444 async fn write_delta(
445 &self,
446 run_id: RunId,
447 agent_id: AgentId,
448 delta: &ProgressDelta,
449 ) -> StorageResult<()> {
450 let now = Utc::now().to_rfc3339();
451 let phase_id: Option<i64> =
452 sqlx::query_scalar("SELECT phase_id FROM agents WHERE run_id = ? AND agent_id = ?")
453 .bind(run_id)
454 .bind(agent_id)
455 .fetch_optional(&self.pool)
456 .await?
457 .flatten();
458
459 match delta {
460 ProgressDelta::Message { text } => {
461 sqlx::query(
462 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind, role, text)
463 VALUES (?, ?, ?, ?, 'message', 'assistant', ?)",
464 )
465 .bind(run_id)
466 .bind(agent_id)
467 .bind(phase_id)
468 .bind(now)
469 .bind(text)
470 .execute(&self.pool)
471 .await?;
472 }
473 ProgressDelta::ToolCall { name, summary } => {
474 sqlx::query(
475 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind, name, text)
476 VALUES (?, ?, ?, ?, 'tool_call', ?, ?)",
477 )
478 .bind(run_id)
479 .bind(agent_id)
480 .bind(phase_id)
481 .bind(now)
482 .bind(name)
483 .bind(summary)
484 .execute(&self.pool)
485 .await?;
486 }
487 ProgressDelta::FileEdit { path } => {
488 sqlx::query(
489 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind, file_path)
490 VALUES (?, ?, ?, ?, 'file_edit', ?)",
491 )
492 .bind(run_id)
493 .bind(agent_id)
494 .bind(phase_id)
495 .bind(now)
496 .bind(path.to_string_lossy().to_string())
497 .execute(&self.pool)
498 .await?;
499 }
500 ProgressDelta::Tokens { usage } => {
501 sqlx::query(
502 "INSERT INTO turns (run_id, agent_id, phase_id, ts, kind,
503 input_tokens, output_tokens,
504 cache_read_tokens, cache_write_tokens)
505 VALUES (?, ?, ?, ?, 'tokens', ?, ?, ?, ?)",
506 )
507 .bind(run_id)
508 .bind(agent_id)
509 .bind(phase_id)
510 .bind(now)
511 .bind(usage.input as i64)
512 .bind(usage.output as i64)
513 .bind(usage.cache_read as i64)
514 .bind(usage.cache_write as i64)
515 .execute(&self.pool)
516 .await?;
517 }
518 }
519 Ok(())
520 }
521
522 async fn write_agent_done(
523 &self,
524 run_id: RunId,
525 agent_id: AgentId,
526 status: &AgentStatus,
527 tokens: luft_core::contract::ids::TokenUsage,
528 elapsed_ms: u64,
529 retry_count: u32,
530 ) -> StorageResult<()> {
531 sqlx::query(
532 "UPDATE agents
533 SET status = ?,
534 input_tokens = ?,
535 output_tokens = ?,
536 cache_read_tokens = ?,
537 cache_write_tokens = ?,
538 done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
539 elapsed_ms = ?,
540 retry_count = ?
541 WHERE run_id = ? AND agent_id = ?",
542 )
543 .bind(agent_status_str(status.clone()))
544 .bind(tokens.input as i64)
545 .bind(tokens.output as i64)
546 .bind(tokens.cache_read as i64)
547 .bind(tokens.cache_write as i64)
548 .bind(elapsed_ms as i64)
549 .bind(retry_count as i64)
550 .bind(run_id)
551 .bind(agent_id)
552 .execute(&self.pool)
553 .await?;
554 Ok(())
555 }
556
557 async fn write_phase_done(
558 &self,
559 run_id: RunId,
560 phase_id: PhaseId,
561 ok: usize,
562 failed: usize,
563 ) -> StorageResult<()> {
564 sqlx::query(
565 "UPDATE phases
566 SET ok = ?, failed = ?, done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now')
567 WHERE run_id = ? AND phase_id = ?",
568 )
569 .bind(ok as i64)
570 .bind(failed as i64)
571 .bind(run_id)
572 .bind(phase_id as i64)
573 .execute(&self.pool)
574 .await?;
575 Ok(())
576 }
577
578 async fn write_run_done(
579 &self,
580 run_id: RunId,
581 status: &RunStatus,
582 total_tokens: luft_core::contract::ids::TokenUsage,
583 report: &Json,
584 ) -> StorageResult<()> {
585 sqlx::query(
586 "UPDATE runs
587 SET status = ?,
588 finished_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
589 input_tokens = ?,
590 output_tokens = ?,
591 cache_read_tokens = ?,
592 cache_write_tokens = ?,
593 report = ?
594 WHERE run_id = ?",
595 )
596 .bind(run_status_str(*status))
597 .bind(total_tokens.input as i64)
598 .bind(total_tokens.output as i64)
599 .bind(total_tokens.cache_read as i64)
600 .bind(total_tokens.cache_write as i64)
601 .bind(serde_json::to_string(report)?)
602 .bind(run_id)
603 .execute(&self.pool)
604 .await?;
605 Ok(())
606 }
607
608 async fn write_report_emitted(
609 &self,
610 run_id: RunId,
611 phase_id: PhaseId,
612 report: &Json,
613 ) -> StorageResult<()> {
614 sqlx::query("UPDATE runs SET report = ? WHERE run_id = ?")
615 .bind(serde_json::to_string(report)?)
616 .bind(run_id)
617 .execute(&self.pool)
618 .await?;
619
620 sqlx::query(
621 "UPDATE phases SET done_ts = COALESCE(done_ts, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
622 WHERE run_id = ? AND phase_id = ?",
623 )
624 .bind(run_id)
625 .bind(phase_id as i64)
626 .execute(&self.pool)
627 .await?;
628 Ok(())
629 }
630
631 #[allow(clippy::too_many_arguments)]
632 async fn write_span_started(
633 &self,
634 run_id: RunId,
635 phase_id: PhaseId,
636 span_id: u64,
637 kind: &str,
638 items: Option<i64>,
639 max_rounds: Option<i64>,
640 path: Option<&str>,
641 _args: Option<&Json>,
642 ) -> StorageResult<()> {
643 sqlx::query(
644 "INSERT INTO spans (run_id, span_id, kind, phase_id, items, max_rounds,
645 path, started_ts)
646 VALUES (?, ?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
647 ON CONFLICT(run_id, span_id) DO UPDATE SET
648 kind = excluded.kind,
649 items = COALESCE(excluded.items, spans.items),
650 max_rounds = COALESCE(excluded.max_rounds, spans.max_rounds),
651 path = COALESCE(excluded.path, spans.path),
652 started_ts = COALESCE(spans.started_ts, excluded.started_ts)",
653 )
654 .bind(run_id)
655 .bind(span_id as i64)
656 .bind(kind)
657 .bind(phase_id as i64)
658 .bind(items)
659 .bind(max_rounds)
660 .bind(path)
661 .execute(&self.pool)
662 .await?;
663 Ok(())
664 }
665
666 #[allow(clippy::too_many_arguments)]
667 async fn write_span_done(
668 &self,
669 run_id: RunId,
670 span_id: u64,
671 ok: i64,
672 failed: i64,
673 result: Option<&Json>,
674 error: Option<String>,
675 _path: Option<&str>,
676 elapsed_ms: u64,
677 ) -> StorageResult<()> {
678 sqlx::query(
679 "UPDATE spans
680 SET ok = ?, failed = ?,
681 result = COALESCE(?, result),
682 error = ?,
683 done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
684 elapsed_ms = ?
685 WHERE run_id = ? AND span_id = ?",
686 )
687 .bind(ok)
688 .bind(failed)
689 .bind(result.map(serde_json::to_string).transpose()?)
690 .bind(error)
691 .bind(elapsed_ms as i64)
692 .bind(run_id)
693 .bind(span_id as i64)
694 .execute(&self.pool)
695 .await?;
696 Ok(())
697 }
698
699 #[allow(clippy::too_many_arguments)]
700 async fn write_converge_done(
701 &self,
702 run_id: RunId,
703 span_id: u64,
704 rounds: i64,
705 converged: bool,
706 result: Option<&Json>,
707 error: Option<String>,
708 elapsed_ms: u64,
709 ) -> StorageResult<()> {
710 sqlx::query(
711 "UPDATE spans
712 SET rounds = ?, converged = ?,
713 result = COALESCE(?, result),
714 error = ?,
715 done_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
716 elapsed_ms = ?
717 WHERE run_id = ? AND span_id = ?",
718 )
719 .bind(rounds)
720 .bind(converged as i64)
721 .bind(result.map(serde_json::to_string).transpose()?)
722 .bind(error)
723 .bind(elapsed_ms as i64)
724 .bind(run_id)
725 .bind(span_id as i64)
726 .execute(&self.pool)
727 .await?;
728 Ok(())
729 }
730
731 async fn append_audit(&self, event: &AgentEvent) -> StorageResult<()> {
732 let (run_id, type_name) = audit_metadata(event);
733 if run_id.is_none() {
734 return Ok(());
735 }
736 if type_name == "acp_raw" {
738 return Ok(());
739 }
740 let payload = serde_json::to_string(event)?;
741 sqlx::query("INSERT INTO events (run_id, type, payload) VALUES (?, ?, ?)")
742 .bind(run_id.unwrap())
743 .bind(type_name)
744 .bind(payload)
745 .execute(&self.pool)
746 .await?;
747 Ok(())
748 }
749}
750
751fn audit_metadata(event: &AgentEvent) -> (Option<RunId>, &'static str) {
756 match event {
757 AgentEvent::RunStarted { run_id, .. } => (Some(*run_id), "run_started"),
758 AgentEvent::PhaseStarted { run_id, .. } => (Some(*run_id), "phase_started"),
759 AgentEvent::AgentStarted { run_id, .. } => (Some(*run_id), "agent_started"),
760 AgentEvent::AgentProgress { run_id, .. } => (Some(*run_id), "agent_progress"),
761 AgentEvent::AcpRaw { run_id, .. } => (Some(*run_id), "acp_raw"),
762 AgentEvent::AgentDone { run_id, .. } => (Some(*run_id), "agent_done"),
763 AgentEvent::PhaseDone { run_id, .. } => (Some(*run_id), "phase_done"),
764 AgentEvent::RunDone { run_id, .. } => (Some(*run_id), "run_done"),
765 AgentEvent::Log { run_id, .. } => (Some(*run_id), "log"),
766 AgentEvent::BudgetSet { run_id, .. } => (Some(*run_id), "budget_set"),
767 AgentEvent::ReportEmitted { run_id, .. } => (Some(*run_id), "report_emitted"),
768 AgentEvent::ParallelStarted { run_id, .. } => (Some(*run_id), "parallel_started"),
769 AgentEvent::ParallelDone { run_id, .. } => (Some(*run_id), "parallel_done"),
770 AgentEvent::WorkflowStarted { run_id, .. } => (Some(*run_id), "workflow_started"),
771 AgentEvent::WorkflowDone { run_id, .. } => (Some(*run_id), "workflow_done"),
772 AgentEvent::ConvergeStarted { run_id, .. } => (Some(*run_id), "converge_started"),
773 AgentEvent::ConvergeDone { run_id, .. } => (Some(*run_id), "converge_done"),
774 AgentEvent::PipelineStarted { run_id, .. } => (Some(*run_id), "pipeline_started"),
775 AgentEvent::PipelineStageStarted { run_id, .. } => {
776 (Some(*run_id), "pipeline_stage_started")
777 }
778 AgentEvent::PipelineItemDone { run_id, .. } => (Some(*run_id), "pipeline_item_done"),
779 AgentEvent::PipelineDone { run_id, .. } => (Some(*run_id), "pipeline_done"),
780 AgentEvent::PlanPreview { run_id, .. } => (Some(*run_id), "plan_preview"),
781 AgentEvent::SignalReceived { run_id, .. } => (*run_id, "signal_received"),
782 AgentEvent::SchemaRetry { run_id, .. } => (Some(*run_id), "schema_retry"),
783 }
784}
785
786fn agent_status_str(s: AgentStatus) -> &'static str {
787 match s {
788 AgentStatus::Ok => "ok",
789 AgentStatus::Error => "error",
790 AgentStatus::Cancelled => "cancelled",
791 AgentStatus::TimedOut => "timed_out",
792 }
793}
794
795fn run_status_str(s: RunStatus) -> &'static str {
796 match s {
797 RunStatus::Completed => "completed",
798 RunStatus::Failed => "failed",
799 RunStatus::Cancelled => "cancelled",
800 RunStatus::Partial => "partial",
801 }
802}
803
804#[allow(dead_code)]
809pub(crate) async fn fetch_event_count(pool: &DbPool, run_id: RunId) -> StorageResult<i64> {
810 let row = sqlx::query("SELECT COUNT(*) AS c FROM events WHERE run_id = ?")
811 .bind(run_id)
812 .fetch_one(pool)
813 .await?;
814 Ok(row.try_get::<i64, _>("c")?)
815}
816
817#[allow(dead_code)]
818pub(crate) async fn fetch_turn_count(
819 pool: &DbPool,
820 run_id: RunId,
821 kind: &str,
822) -> StorageResult<i64> {
823 let row = sqlx::query("SELECT COUNT(*) AS c FROM turns WHERE run_id = ? AND kind = ?")
824 .bind(run_id)
825 .bind(kind)
826 .fetch_one(pool)
827 .await?;
828 Ok(row.try_get::<i64, _>("c")?)
829}
830
831pub type SharedWriter = Arc<EventWriter>;
833
834#[cfg(test)]
839mod tests {
840 use super::*;
841 use crate::db::open_db;
842 use luft_core::contract::event::LogLevel;
843 use luft_core::contract::ids::TokenUsage;
844 use std::path::PathBuf;
845 use tempfile::tempdir;
846
847 async fn setup() -> (tempfile::TempDir, EventWriter) {
848 let dir = tempdir().unwrap();
849 let pool = open_db(&dir.path().join("test.db")).await.unwrap();
850 (dir, EventWriter::new(pool))
851 }
852
853 #[tokio::test]
854 async fn run_lifecycle() {
855 let (_dir, w) = setup().await;
856 let run_id = uuid::Uuid::now_v7();
857 let ts = Utc::now();
858
859 w.write_event(&AgentEvent::RunStarted {
860 run_id,
861 task: "hello".into(),
862 ts,
863 })
864 .await
865 .unwrap();
866
867 let row = sqlx::query("SELECT task, status FROM runs WHERE run_id = ?")
868 .bind(run_id)
869 .fetch_one(w.pool())
870 .await
871 .unwrap();
872 let task: String = row.try_get("task").unwrap();
873 let status: String = row.try_get("status").unwrap();
874 assert_eq!(task, "hello");
875 assert_eq!(status, "running");
876
877 w.write_event(&AgentEvent::RunDone {
878 run_id,
879 status: RunStatus::Completed,
880 total_tokens: TokenUsage {
881 input: 100,
882 output: 50,
883 cache_read: 0,
884 cache_write: 0,
885 },
886 report: serde_json::json!({"result": "ok"}),
887 ts: chrono::Utc::now(),
888 })
889 .await
890 .unwrap();
891
892 let row = sqlx::query("SELECT status, finished_ts FROM runs WHERE run_id = ?")
893 .bind(run_id)
894 .fetch_one(w.pool())
895 .await
896 .unwrap();
897 let status: String = row.try_get("status").unwrap();
898 let finished_ts: Option<String> = row.try_get("finished_ts").unwrap();
899 assert_eq!(status, "completed");
900 assert!(finished_ts.is_some());
901 }
902
903 #[tokio::test]
904 async fn agent_started_creates_agent_row() {
905 let (_dir, w) = setup().await;
906 let run_id = uuid::Uuid::now_v7();
907 let agent_id = uuid::Uuid::now_v7();
908
909 w.write_event(&AgentEvent::AgentStarted {
910 run_id,
911 phase_id: 1,
912 agent_id,
913 prompt_preview: "do something".into(),
914 model: Some("claude-sonnet-4".into()),
915 description: None,
916 role: None,
917 name: None,
918 agent_seq: 0,
919 ts: Default::default(),
920 })
921 .await
922 .unwrap();
923
924 let row = sqlx::query(
925 "SELECT phase_id, model, status, prompt_preview FROM agents WHERE run_id = ? AND agent_id = ?",
926 )
927 .bind(run_id)
928 .bind(agent_id)
929 .fetch_one(w.pool())
930 .await
931 .unwrap();
932
933 let phase_id: i64 = row.try_get("phase_id").unwrap();
934 let model: String = row.try_get("model").unwrap();
935 let status: String = row.try_get("status").unwrap();
936 assert_eq!(phase_id, 1);
937 assert_eq!(model, "claude-sonnet-4");
938 assert_eq!(status, "running");
939 }
940
941 #[tokio::test]
942 async fn progress_delta_message_creates_turn() {
943 let (_dir, w) = setup().await;
944 let run_id = uuid::Uuid::now_v7();
945 let agent_id = uuid::Uuid::now_v7();
946
947 w.write_event(&AgentEvent::AgentStarted {
949 run_id,
950 phase_id: 2,
951 agent_id,
952 prompt_preview: "p".into(),
953 model: None,
954 description: None,
955 role: None,
956 name: None,
957 agent_seq: 0,
958 ts: Default::default(),
959 })
960 .await
961 .unwrap();
962
963 w.write_event(&AgentEvent::AgentProgress {
964 run_id,
965 agent_id,
966 delta: ProgressDelta::Message {
967 text: "thinking...".into(),
968 },
969 })
970 .await
971 .unwrap();
972
973 let count = fetch_turn_count(w.pool(), run_id, "message").await.unwrap();
974 assert_eq!(count, 1);
975
976 let row = sqlx::query("SELECT text, role, phase_id FROM turns WHERE run_id = ? LIMIT 1")
977 .bind(run_id)
978 .fetch_one(w.pool())
979 .await
980 .unwrap();
981 let text: String = row.try_get("text").unwrap();
982 let role: String = row.try_get("role").unwrap();
983 let phase_id: i64 = row.try_get("phase_id").unwrap();
984 assert_eq!(text, "thinking...");
985 assert_eq!(role, "assistant");
986 assert_eq!(phase_id, 2);
987 }
988
989 #[tokio::test]
990 async fn progress_delta_tool_call_and_file_edit() {
991 let (_dir, w) = setup().await;
992 let run_id = uuid::Uuid::now_v7();
993 let agent_id = uuid::Uuid::now_v7();
994
995 w.write_event(&AgentEvent::AgentStarted {
996 run_id,
997 phase_id: 0,
998 agent_id,
999 prompt_preview: "".into(),
1000 model: None,
1001 description: None,
1002 role: None,
1003 name: None,
1004 agent_seq: 0,
1005 ts: Default::default(),
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 ts: Default::default(),
1062 })
1063 .await
1064 .unwrap();
1065
1066 w.write_event(&AgentEvent::AgentProgress {
1067 run_id,
1068 agent_id,
1069 delta: ProgressDelta::Tokens {
1070 usage: TokenUsage {
1071 input: 10,
1072 output: 5,
1073 cache_read: 0,
1074 cache_write: 0,
1075 },
1076 },
1077 })
1078 .await
1079 .unwrap();
1080
1081 assert_eq!(
1082 fetch_turn_count(w.pool(), run_id, "tokens").await.unwrap(),
1083 1
1084 );
1085 }
1086
1087 #[tokio::test]
1088 async fn agent_done_updates_status_and_tokens() {
1089 let (_dir, w) = setup().await;
1090 let run_id = uuid::Uuid::now_v7();
1091 let agent_id = uuid::Uuid::now_v7();
1092
1093 w.write_event(&AgentEvent::AgentStarted {
1094 run_id,
1095 phase_id: 0,
1096 agent_id,
1097 prompt_preview: "".into(),
1098 model: None,
1099 description: None,
1100 role: None,
1101 name: None,
1102 agent_seq: 0,
1103 ts: Default::default(),
1104 })
1105 .await
1106 .unwrap();
1107
1108 w.write_event(&AgentEvent::AgentDone {
1109 run_id,
1110 agent_id,
1111 status: AgentStatus::Ok,
1112 tokens: TokenUsage {
1113 input: 200,
1114 output: 80,
1115 cache_read: 0,
1116 cache_write: 0,
1117 },
1118 elapsed_ms: 1234,
1119 name: None,
1120 agent_seq: 0,
1121 output: serde_json::Value::Null,
1122 findings: Vec::new(),
1123 prompt: String::new(),
1124 retry_count: 0,
1125 ts: Default::default(),
1126 })
1127 .await
1128 .unwrap();
1129
1130 let row = sqlx::query(
1131 "SELECT status, input_tokens, output_tokens, elapsed_ms FROM agents WHERE run_id = ? AND agent_id = ?",
1132 )
1133 .bind(run_id)
1134 .bind(agent_id)
1135 .fetch_one(w.pool())
1136 .await
1137 .unwrap();
1138
1139 let status: String = row.try_get("status").unwrap();
1140 let input: i64 = row.try_get("input_tokens").unwrap();
1141 let output: i64 = row.try_get("output_tokens").unwrap();
1142 let elapsed: i64 = row.try_get("elapsed_ms").unwrap();
1143 assert_eq!(status, "ok");
1144 assert_eq!(input, 200);
1145 assert_eq!(output, 80);
1146 assert_eq!(elapsed, 1234);
1147 }
1148
1149 #[tokio::test]
1150 async fn phase_started_and_done() {
1151 let (_dir, w) = setup().await;
1152 let run_id = uuid::Uuid::now_v7();
1153
1154 w.write_event(&AgentEvent::RunStarted {
1155 run_id,
1156 task: "t".into(),
1157 ts: Utc::now(),
1158 })
1159 .await
1160 .unwrap();
1161
1162 w.write_event(&AgentEvent::PhaseStarted {
1163 run_id,
1164 phase_id: 1,
1165 label: "explore".into(),
1166 planned: 3,
1167 description: None,
1168 role: None,
1169 ts: chrono::Utc::now(),
1170 })
1171 .await
1172 .unwrap();
1173
1174 w.write_event(&AgentEvent::PhaseDone {
1175 run_id,
1176 phase_id: 1,
1177 ok: 2,
1178 failed: 1,
1179 ts: chrono::Utc::now(),
1180 })
1181 .await
1182 .unwrap();
1183
1184 let row = sqlx::query(
1185 "SELECT label, planned, ok, failed FROM phases WHERE run_id = ? AND phase_id = ?",
1186 )
1187 .bind(run_id)
1188 .bind(1i64)
1189 .fetch_one(w.pool())
1190 .await
1191 .unwrap();
1192 let label: String = row.try_get("label").unwrap();
1193 let planned: i64 = row.try_get("planned").unwrap();
1194 let ok: i64 = row.try_get("ok").unwrap();
1195 let failed: i64 = row.try_get("failed").unwrap();
1196 assert_eq!(label, "explore");
1197 assert_eq!(planned, 3);
1198 assert_eq!(ok, 2);
1199 assert_eq!(failed, 1);
1200 }
1201
1202 #[tokio::test]
1203 async fn spans_track_orchestration() {
1204 let (_dir, w) = setup().await;
1205 let run_id = uuid::Uuid::now_v7();
1206
1207 w.write_event(&AgentEvent::RunStarted {
1208 run_id,
1209 task: "t".into(),
1210 ts: Utc::now(),
1211 })
1212 .await
1213 .unwrap();
1214
1215 w.write_event(&AgentEvent::ParallelStarted {
1216 run_id,
1217 phase_id: 1,
1218 span_id: 7,
1219 count: 4,
1220 })
1221 .await
1222 .unwrap();
1223
1224 w.write_event(&AgentEvent::ParallelDone {
1225 run_id,
1226 phase_id: 1,
1227 span_id: 7,
1228 ok: 3,
1229 failed: 1,
1230 results: serde_json::json!([1, 2, 3]),
1231 elapsed_ms: 999,
1232 })
1233 .await
1234 .unwrap();
1235
1236 let row = sqlx::query(
1237 "SELECT kind, items, ok, failed, elapsed_ms FROM spans WHERE run_id = ? AND span_id = ?",
1238 )
1239 .bind(run_id)
1240 .bind(7i64)
1241 .fetch_one(w.pool())
1242 .await
1243 .unwrap();
1244 let kind: String = row.try_get("kind").unwrap();
1245 let items: i64 = row.try_get("items").unwrap();
1246 let ok: i64 = row.try_get("ok").unwrap();
1247 let failed: i64 = row.try_get("failed").unwrap();
1248 assert_eq!(kind, "parallel");
1249 assert_eq!(items, 4);
1250 assert_eq!(ok, 3);
1251 assert_eq!(failed, 1);
1252 }
1253
1254 #[tokio::test]
1255 async fn audit_log_captures_all_events() {
1256 let (_dir, w) = setup().await;
1257 let run_id = uuid::Uuid::now_v7();
1258
1259 w.write_event(&AgentEvent::RunStarted {
1260 run_id,
1261 task: "t".into(),
1262 ts: Utc::now(),
1263 })
1264 .await
1265 .unwrap();
1266
1267 w.write_event(&AgentEvent::Log {
1268 run_id,
1269 agent_id: None,
1270 level: LogLevel::Info,
1271 msg: "hi".into(),
1272 })
1273 .await
1274 .unwrap();
1275
1276 let count = fetch_event_count(w.pool(), run_id).await.unwrap();
1278 assert!(count >= 1);
1279 }
1280
1281 #[tokio::test]
1282 async fn cascade_delete_removes_all() {
1283 let (_dir, w) = setup().await;
1284 let run_id = uuid::Uuid::now_v7();
1285
1286 w.write_event(&AgentEvent::RunStarted {
1287 run_id,
1288 task: "t".into(),
1289 ts: Utc::now(),
1290 })
1291 .await
1292 .unwrap();
1293
1294 let agent_id = uuid::Uuid::now_v7();
1295 w.write_event(&AgentEvent::AgentStarted {
1296 run_id,
1297 phase_id: 0,
1298 agent_id,
1299 prompt_preview: "".into(),
1300 model: None,
1301 description: None,
1302 role: None,
1303 name: None,
1304 agent_seq: 0,
1305 ts: Default::default(),
1306 })
1307 .await
1308 .unwrap();
1309
1310 w.write_event(&AgentEvent::AgentProgress {
1311 run_id,
1312 agent_id,
1313 delta: ProgressDelta::Message { text: "hi".into() },
1314 })
1315 .await
1316 .unwrap();
1317
1318 sqlx::query("DELETE FROM runs WHERE run_id = ?")
1319 .bind(run_id)
1320 .execute(w.pool())
1321 .await
1322 .unwrap();
1323
1324 let turn_count = fetch_turn_count(w.pool(), run_id, "message").await.unwrap();
1325 let agent_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM agents WHERE run_id = ?")
1326 .bind(run_id)
1327 .fetch_one(w.pool())
1328 .await
1329 .unwrap();
1330 assert_eq!(turn_count, 0);
1331 assert_eq!(agent_count, 0);
1332 }
1333
1334 #[tokio::test]
1335 async fn acp_raw_is_skipped() {
1336 let (_dir, w) = setup().await;
1337 let run_id = uuid::Uuid::now_v7();
1338 let agent_id = uuid::Uuid::now_v7();
1339
1340 w.write_event(&AgentEvent::AcpRaw {
1341 run_id,
1342 agent_id,
1343 kind: "agent_message_chunk".into(),
1344 raw: serde_json::json!({"text": "chunk"}),
1345 })
1346 .await
1347 .unwrap();
1348
1349 let count = fetch_event_count(w.pool(), run_id).await.unwrap();
1350 assert_eq!(count, 0);
1351 }
1352}