1use std::collections::{BTreeMap, BTreeSet};
2
3use serde::{Deserialize, Serialize};
4use serde_json::Value;
5use sha2::{Digest, Sha256};
6
7use crate::event_log::{AnyEventLog, EventId, EventLog, LogError};
8use crate::provenance::event_record_hash_from_headers;
9use crate::redact::{current_policy, RedactionPolicy};
10
11use super::super::ArtifactRecord;
12use super::{
13 LlmUsageRecord, RunCheckpointRecord, RunChildRecord, RunHitlQuestionRecord, RunRecord,
14 RunStageRecord, RunTraceSpanRecord,
15};
16
17mod visible_transcript;
18
19use visible_transcript::public_assistant_transcript_text;
20
21pub const RUN_VIEW_SCHEMA: &str = "harn.run_view.v1";
22pub const SESSION_VIEW_SCHEMA: &str = "harn.session_view.v1";
23pub const RUN_VIEW_SCHEMA_VERSION: u32 = 1;
24pub const SESSION_VIEW_SCHEMA_VERSION: u32 = 1;
25pub const SESSION_VIEW_QUERY_METHOD: &str = "harn.session_view.query";
26
27const TEXT_LIMIT: usize = 16 * 1024;
28const PREVIEW_LIMIT: usize = 1200;
29
30#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
31#[serde(default)]
32pub struct ViewProducer {
33 pub name: String,
34 pub version: String,
35}
36
37impl Default for ViewProducer {
38 fn default() -> Self {
39 Self {
40 name: "harn".to_string(),
41 version: env!("CARGO_PKG_VERSION").to_string(),
42 }
43 }
44}
45
46#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
47#[serde(default)]
48pub struct ProjectionInfo {
49 pub projection_id: String,
50 pub projection_hash: Option<String>,
51 pub prefix_hash: Option<String>,
52 pub last_event_id: Option<EventId>,
53}
54
55#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
56#[serde(default)]
57pub struct RunView {
58 pub schema: String,
59 pub schema_version: u32,
60 pub producer: ViewProducer,
61 pub run: RunViewRun,
62 pub projection: ProjectionInfo,
63 pub visible_text: Option<String>,
64 pub transcript: TranscriptSummary,
65 pub usage: RunViewUsage,
66 pub providers: Vec<RunViewProvider>,
67 pub stages: Vec<RunViewStage>,
68 pub artifacts: Vec<RunViewArtifact>,
69 pub checkpoints: Vec<RunViewCheckpoint>,
70 pub pending: RunViewPendingState,
71 pub failure: Option<RunViewFailure>,
72 pub metadata: RunViewMetadata,
73}
74
75#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
76#[serde(default)]
77pub struct RunViewRun {
78 pub run_id: String,
79 pub session_id: Option<String>,
80 pub parent_run_id: Option<String>,
81 pub root_run_id: Option<String>,
82 pub parent_session_id: Option<String>,
83 pub child_runs: Vec<RunViewChild>,
84 pub run_path: Option<String>,
85 pub status: String,
86 pub workflow_id: String,
87 pub workflow_name: Option<String>,
88 pub task: String,
89 pub started_at: String,
90 pub finished_at: Option<String>,
91 pub duration_ms: Option<u64>,
92}
93
94#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
95#[serde(default)]
96pub struct RunViewChild {
97 pub worker_id: String,
98 pub worker_name: String,
99 pub run_id: Option<String>,
100 pub session_id: Option<String>,
101 pub parent_session_id: Option<String>,
102 pub run_path: Option<String>,
103 pub status: String,
104 pub task: String,
105}
106
107#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
108#[serde(default)]
109pub struct RunViewUsage {
110 pub input_tokens: i64,
111 pub output_tokens: i64,
112 pub total_duration_ms: i64,
113 pub call_count: i64,
114 pub total_cost: f64,
115 pub models: Vec<String>,
116}
117
118impl From<&LlmUsageRecord> for RunViewUsage {
119 fn from(value: &LlmUsageRecord) -> Self {
120 Self {
121 input_tokens: value.input_tokens,
122 output_tokens: value.output_tokens,
123 total_duration_ms: value.total_duration_ms,
124 call_count: value.call_count,
125 total_cost: value.total_cost,
126 models: value.models.clone(),
127 }
128 }
129}
130
131impl RunViewUsage {
132 fn add_usage(&mut self, usage: &RunViewUsage) {
133 self.input_tokens += usage.input_tokens;
134 self.output_tokens += usage.output_tokens;
135 self.total_duration_ms += usage.total_duration_ms;
136 self.call_count += usage.call_count;
137 self.total_cost += usage.total_cost;
138 for model in &usage.models {
139 if !model.is_empty() && !self.models.contains(model) {
140 self.models.push(model.clone());
141 }
142 }
143 }
144}
145
146#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
147#[serde(default)]
148pub struct RunViewProvider {
149 pub provider: String,
150 pub model: String,
151 pub call_count: i64,
152 pub input_tokens: i64,
153 pub output_tokens: i64,
154 pub cost_usd: f64,
155}
156
157#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
158#[serde(default)]
159pub struct RunViewStage {
160 pub id: String,
161 pub node_id: String,
162 pub kind: String,
163 pub status: String,
164 pub outcome: String,
165 pub branch: Option<String>,
166 pub started_at: String,
167 pub finished_at: Option<String>,
168 pub duration_ms: Option<u64>,
169 pub visible_text: Option<String>,
170 pub usage: RunViewUsage,
171 pub provider: Option<String>,
172 pub model: Option<String>,
173 pub artifact_refs: Vec<String>,
174 pub attempt_count: usize,
175 pub error: Option<String>,
176}
177
178#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
179#[serde(default)]
180pub struct RunViewArtifact {
181 pub id: String,
182 pub kind: String,
183 pub title: Option<String>,
184 pub source: Option<String>,
185 pub stage: Option<String>,
186 pub estimated_tokens: Option<usize>,
187 pub lineage: Vec<String>,
188 pub preview: Option<String>,
189}
190
191#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
192#[serde(default)]
193pub struct RunViewCheckpoint {
194 pub id: String,
195 pub reason: String,
196 pub ready_count: usize,
197 pub completed_count: usize,
198 pub last_stage_id: Option<String>,
199 pub persisted_at: String,
200}
201
202#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
203#[serde(default)]
204pub struct RunViewPendingState {
205 pub nodes: Vec<String>,
206 pub approvals: Vec<RunViewApproval>,
207 pub auth: Vec<RunViewAuth>,
208}
209
210#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
211#[serde(default)]
212pub struct RunViewApproval {
213 pub request_id: String,
214 pub prompt: String,
215 pub agent: String,
216 pub trace_id: Option<String>,
217 pub asked_at: String,
218}
219
220#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
221#[serde(default)]
222pub struct RunViewAuth {
223 pub provider: Option<String>,
224 pub server: Option<String>,
225 pub scope: Option<String>,
226 pub stage_id: Option<String>,
227 pub message: Option<String>,
228}
229
230#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
231#[serde(default)]
232pub struct RunViewFailure {
233 pub stage_id: Option<String>,
234 pub node_id: Option<String>,
235 pub status: String,
236 pub outcome: String,
237 pub message: Option<String>,
238}
239
240#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
241#[serde(default)]
242pub struct TranscriptSummary {
243 pub present: bool,
244 pub message_count: usize,
245 pub event_count: usize,
246 pub summary: Option<String>,
247 pub source: Option<String>,
248}
249
250#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
251#[serde(default)]
252pub struct RunViewMetadata {
253 pub record_type: String,
254 pub stage_count: usize,
255 pub transition_count: usize,
256 pub artifact_count: usize,
257 pub checkpoint_count: usize,
258 pub child_run_count: usize,
259 pub observability_present: bool,
260 pub planner_round_count: usize,
261 pub tool_recording_count: usize,
262 pub replay_fixture_id: Option<String>,
263 pub execution: Option<super::RunExecutionRecord>,
264}
265
266#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
267#[serde(default)]
268pub struct SessionView {
269 pub schema: String,
270 pub schema_version: u32,
271 pub producer: ViewProducer,
272 pub session: SessionViewSession,
273 pub projection: ProjectionInfo,
274 pub runs: Vec<RunView>,
275 pub history: Vec<SessionViewHistoryItem>,
276 pub usage: RunViewUsage,
277 pub pending: RunViewPendingState,
278 pub failure: Option<RunViewFailure>,
279 pub metadata: SessionViewMetadata,
280}
281
282#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
283#[serde(default)]
284pub struct SessionViewSession {
285 pub session_id: Option<String>,
286 pub parent_session_id: Option<String>,
287 pub root_session_id: Option<String>,
288 pub status: String,
289 pub run_count: usize,
290 pub started_at: Option<String>,
291 pub updated_at: Option<String>,
292 pub last_event_id: Option<EventId>,
293 pub chain_root_hash: Option<String>,
294}
295
296#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
297#[serde(default)]
298pub struct SessionViewHistoryItem {
299 pub run_id: String,
300 pub run_path: Option<String>,
301 pub session_id: Option<String>,
302 pub status: String,
303 pub started_at: Option<String>,
304 pub finished_at: Option<String>,
305 pub last_event_id: Option<EventId>,
306 pub visible_text: Option<String>,
307}
308
309#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
310#[serde(default)]
311pub struct SessionViewMetadata {
312 pub record_count: usize,
313 pub event_count: usize,
314 pub has_event_log: bool,
315}
316
317#[derive(Clone, Debug, Default)]
318pub struct RunViewOptions {
319 pub producer: ViewProducer,
320 pub run_path: Option<String>,
321 pub last_event_id: Option<EventId>,
322 pub prefix_hash: Option<String>,
323}
324
325#[derive(Clone, Debug, Default)]
326pub struct SessionViewOptions {
327 pub producer: ViewProducer,
328 pub session_id: Option<String>,
329 pub parent_session_id: Option<String>,
330 pub root_session_id: Option<String>,
331 pub status: Option<String>,
332 pub started_at: Option<String>,
333 pub updated_at: Option<String>,
334 pub last_event_id: Option<EventId>,
335 pub chain_root_hash: Option<String>,
336 pub event_count: usize,
337 pub has_event_log: bool,
338}
339
340#[derive(Debug)]
341pub enum RunViewError {
342 EventLog(LogError),
343}
344
345impl std::fmt::Display for RunViewError {
346 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
347 match self {
348 Self::EventLog(error) => error.fmt(f),
349 }
350 }
351}
352
353impl std::error::Error for RunViewError {}
354
355impl From<LogError> for RunViewError {
356 fn from(error: LogError) -> Self {
357 Self::EventLog(error)
358 }
359}
360
361pub fn build_run_view(run: &RunRecord) -> RunView {
362 build_run_view_with_options(run, RunViewOptions::default())
363}
364
365pub fn build_run_view_with_path(run: &RunRecord, run_path: Option<impl Into<String>>) -> RunView {
366 build_run_view_with_options(
367 run,
368 RunViewOptions {
369 run_path: run_path.map(Into::into),
370 ..RunViewOptions::default()
371 },
372 )
373}
374
375pub async fn build_run_view_with_event_log(
376 run: &RunRecord,
377 run_path: Option<impl Into<String>>,
378 log: Option<&AnyEventLog>,
379) -> Result<RunView, RunViewError> {
380 let mut options = RunViewOptions {
381 run_path: run_path.map(Into::into),
382 ..RunViewOptions::default()
383 };
384 if let Some(log) = log {
385 if let Some(session_id) = infer_run_session_id(run) {
386 let (last_event_id, prefix_hash) = read_session_tip(log, &session_id).await?;
387 options.last_event_id = last_event_id;
388 options.prefix_hash = prefix_hash;
389 }
390 }
391 Ok(build_run_view_with_options(run, options))
392}
393
394pub fn build_run_view_with_options(run: &RunRecord, options: RunViewOptions) -> RunView {
395 let policy = current_policy();
396 let session_id = infer_run_session_id(run);
397 let parent_session_id = infer_parent_session_id(run);
398 let stages = run
399 .stages
400 .iter()
401 .map(|stage| build_stage_view(stage, &policy))
402 .collect::<Vec<_>>();
403 let visible_text = bounded_join(
404 run.stages
405 .iter()
406 .filter_map(|stage| stage.visible_text.as_deref())
407 .map(|text| redact_bounded(text, &policy, TEXT_LIMIT)),
408 TEXT_LIMIT,
409 )
410 .or_else(|| public_assistant_transcript_text(run.transcript.as_ref(), &policy));
411 let usage = run
412 .usage
413 .as_ref()
414 .map(RunViewUsage::from)
415 .unwrap_or_else(|| usage_from_stages(&stages));
416 let mut view = RunView {
417 schema: RUN_VIEW_SCHEMA.to_string(),
418 schema_version: RUN_VIEW_SCHEMA_VERSION,
419 producer: options.producer.clone(),
420 run: RunViewRun {
421 run_id: run.id.clone(),
422 session_id,
423 parent_run_id: run.parent_run_id.clone(),
424 root_run_id: run.root_run_id.clone(),
425 parent_session_id,
426 child_runs: run
427 .child_runs
428 .iter()
429 .map(|child| build_child_view(child, &policy))
430 .collect(),
431 run_path: options
432 .run_path
433 .clone()
434 .or_else(|| run.persisted_path.clone()),
435 status: run.status.clone(),
436 workflow_id: run.workflow_id.clone(),
437 workflow_name: run.workflow_name.clone(),
438 task: redact_bounded(&run.task, &policy, TEXT_LIMIT),
439 started_at: run.started_at.clone(),
440 finished_at: run.finished_at.clone(),
441 duration_ms: run_duration_ms(run),
442 },
443 projection: ProjectionInfo {
444 projection_id: String::new(),
445 projection_hash: None,
446 prefix_hash: options.prefix_hash,
447 last_event_id: options.last_event_id,
448 },
449 visible_text,
450 transcript: transcript_summary_for_run(run, &policy),
451 usage,
452 providers: provider_summary(run),
453 stages,
454 artifacts: run
455 .artifacts
456 .iter()
457 .map(|artifact| build_artifact_view(artifact, &policy))
458 .collect(),
459 checkpoints: run.checkpoints.iter().map(build_checkpoint_view).collect(),
460 pending: RunViewPendingState {
461 nodes: run.pending_nodes.clone(),
462 approvals: run
463 .hitl_questions
464 .iter()
465 .map(|question| build_approval_view(question, &policy))
466 .collect(),
467 auth: pending_auth(run, &policy),
468 },
469 failure: failure_summary(run, &policy),
470 metadata: RunViewMetadata {
471 record_type: run.type_name.clone(),
472 stage_count: run.stages.len(),
473 transition_count: run.transitions.len(),
474 artifact_count: run.artifacts.len(),
475 checkpoint_count: run.checkpoints.len(),
476 child_run_count: run.child_runs.len(),
477 observability_present: run.observability.is_some(),
478 planner_round_count: run
479 .observability
480 .as_ref()
481 .map(|observability| observability.planner_rounds.len())
482 .unwrap_or_default(),
483 tool_recording_count: run.tool_recordings.len(),
484 replay_fixture_id: run
485 .replay_fixture
486 .as_ref()
487 .map(|fixture| fixture.id.clone()),
488 execution: run.execution.clone(),
489 },
490 };
491 finalize_run_projection(&mut view);
492 view
493}
494
495pub fn build_session_view_from_run_views(
496 runs: Vec<RunView>,
497 options: SessionViewOptions,
498) -> SessionView {
499 let session_id = options
500 .session_id
501 .clone()
502 .or_else(|| runs.iter().find_map(|run| run.run.session_id.clone()));
503 let mut usage = RunViewUsage::default();
504 let mut pending = RunViewPendingState::default();
505 let mut failure = None;
506 let mut started_at = options.started_at.clone();
507 let mut updated_at = options.updated_at.clone();
508 let history = runs
509 .iter()
510 .map(|run| {
511 usage.add_usage(&run.usage);
512 pending.nodes.extend(run.pending.nodes.clone());
513 pending.approvals.extend(run.pending.approvals.clone());
514 pending.auth.extend(run.pending.auth.clone());
515 if failure.is_none() {
516 failure = run.failure.clone();
517 }
518 if !run.run.started_at.is_empty() {
519 started_at = min_opt_string(started_at.take(), Some(run.run.started_at.clone()));
520 updated_at = max_opt_string(updated_at.take(), Some(run.run.started_at.clone()));
521 }
522 updated_at = max_opt_string(updated_at.take(), run.run.finished_at.clone());
523 SessionViewHistoryItem {
524 run_id: run.run.run_id.clone(),
525 run_path: run.run.run_path.clone(),
526 session_id: run.run.session_id.clone(),
527 status: run.run.status.clone(),
528 started_at: non_empty_string(&run.run.started_at),
529 finished_at: run.run.finished_at.clone(),
530 last_event_id: run.projection.last_event_id,
531 visible_text: run.visible_text.clone(),
532 }
533 })
534 .collect::<Vec<_>>();
535 let status = options
536 .status
537 .clone()
538 .unwrap_or_else(|| aggregate_session_status(&runs));
539 let last_event_id = options.last_event_id.or_else(|| {
540 runs.iter()
541 .filter_map(|run| run.projection.last_event_id)
542 .max()
543 });
544 let chain_root_hash = options.chain_root_hash.clone().or_else(|| {
545 runs.iter()
546 .rev()
547 .find_map(|run| run.projection.prefix_hash.clone())
548 });
549 let mut view = SessionView {
550 schema: SESSION_VIEW_SCHEMA.to_string(),
551 schema_version: SESSION_VIEW_SCHEMA_VERSION,
552 producer: options.producer.clone(),
553 session: SessionViewSession {
554 session_id,
555 parent_session_id: options.parent_session_id.clone().or_else(|| {
556 runs.iter()
557 .find_map(|run| run.run.parent_session_id.clone())
558 }),
559 root_session_id: options.root_session_id.clone(),
560 status,
561 run_count: runs.len(),
562 started_at,
563 updated_at,
564 last_event_id,
565 chain_root_hash,
566 },
567 projection: ProjectionInfo {
568 projection_id: String::new(),
569 projection_hash: None,
570 prefix_hash: None,
571 last_event_id,
572 },
573 runs,
574 history,
575 usage,
576 pending: dedupe_pending(pending),
577 failure,
578 metadata: SessionViewMetadata {
579 record_count: 0,
580 event_count: options.event_count,
581 has_event_log: options.has_event_log,
582 },
583 };
584 view.metadata.record_count = view.runs.len();
585 view.projection.prefix_hash = view.session.chain_root_hash.clone();
586 finalize_session_projection(&mut view);
587 view
588}
589
590pub async fn build_session_view_from_run_records(
591 runs: Vec<(&RunRecord, Option<String>)>,
592 session_id: Option<String>,
593 log: Option<&AnyEventLog>,
594) -> Result<SessionView, RunViewError> {
595 let mut views = Vec::new();
596 for (run, path) in runs {
597 views.push(build_run_view_with_event_log(run, path, log).await?);
598 }
599 let mut options = SessionViewOptions {
600 session_id,
601 has_event_log: log.is_some(),
602 ..SessionViewOptions::default()
603 };
604 if let (Some(log), Some(session_id)) = (log, options.session_id.as_deref()) {
605 let (last_event_id, chain_root_hash) = read_session_tip(log, session_id).await?;
606 options.last_event_id = last_event_id;
607 options.chain_root_hash = chain_root_hash;
608 }
609 Ok(build_session_view_from_run_views(views, options))
610}
611
612pub async fn build_empty_session_view(
613 session_id: Option<String>,
614 log: Option<&AnyEventLog>,
615) -> Result<SessionView, RunViewError> {
616 let mut options = SessionViewOptions {
617 session_id: session_id.clone(),
618 has_event_log: log.is_some(),
619 ..SessionViewOptions::default()
620 };
621 if let (Some(log), Some(session_id)) = (log, session_id.as_deref()) {
622 let (last_event_id, chain_root_hash) = read_session_tip(log, session_id).await?;
623 options.last_event_id = last_event_id;
624 options.chain_root_hash = chain_root_hash;
625 }
626 Ok(build_session_view_from_run_views(Vec::new(), options))
627}
628
629async fn read_session_tip(
630 log: &AnyEventLog,
631 session_id: &str,
632) -> Result<(Option<EventId>, Option<String>), LogError> {
633 let topic = crate::session_timeline::agent_events_topic(session_id);
634 let Some(latest) = log.latest(&topic).await? else {
635 return Ok((None, None));
636 };
637 let from = latest.checked_sub(1);
638 let events = log.read_range(&topic, from, 1).await?;
639 let prefix_hash = events
640 .into_iter()
641 .find(|(event_id, _)| *event_id == latest)
642 .and_then(|(event_id, event)| {
643 event_record_hash_from_headers(topic.as_str(), event_id, &event).ok()
644 });
645 Ok((Some(latest), prefix_hash))
646}
647
648fn build_child_view(child: &RunChildRecord, policy: &RedactionPolicy) -> RunViewChild {
649 RunViewChild {
650 worker_id: child.worker_id.clone(),
651 worker_name: child.worker_name.clone(),
652 run_id: child.run_id.clone(),
653 session_id: child.session_id.clone(),
654 parent_session_id: child.parent_session_id.clone(),
655 run_path: child.run_path.clone(),
656 status: child.status.clone(),
657 task: redact_bounded(&child.task, policy, TEXT_LIMIT),
658 }
659}
660
661fn build_stage_view(stage: &RunStageRecord, policy: &RedactionPolicy) -> RunViewStage {
662 let usage = stage
663 .usage
664 .as_ref()
665 .map(RunViewUsage::from)
666 .unwrap_or_default();
667 let artifact_refs = stage
668 .produced_artifact_ids
669 .iter()
670 .chain(stage.artifacts.iter().map(|artifact| &artifact.id))
671 .filter(|id| !id.is_empty())
672 .cloned()
673 .collect::<BTreeSet<_>>()
674 .into_iter()
675 .collect();
676 RunViewStage {
677 id: stage.id.clone(),
678 node_id: stage.node_id.clone(),
679 kind: stage.kind.clone(),
680 status: stage.status.clone(),
681 outcome: stage.outcome.clone(),
682 branch: stage.branch.clone(),
683 started_at: stage.started_at.clone(),
684 finished_at: stage.finished_at.clone(),
685 duration_ms: stage_duration_ms(stage),
686 visible_text: stage
687 .visible_text
688 .as_deref()
689 .map(|text| redact_bounded(text, policy, TEXT_LIMIT)),
690 usage,
691 provider: metadata_string_any(&stage.metadata, &["provider"])
692 .or_else(|| metadata_path_string(&stage.metadata, &["model_policy", "provider"])),
693 model: metadata_string_any(&stage.metadata, &["model"])
694 .or_else(|| metadata_path_string(&stage.metadata, &["model_policy", "model"])),
695 artifact_refs,
696 attempt_count: stage.attempts.len(),
697 error: stage_error(stage, policy),
698 }
699}
700
701fn build_artifact_view(artifact: &ArtifactRecord, policy: &RedactionPolicy) -> RunViewArtifact {
702 RunViewArtifact {
703 id: artifact.id.clone(),
704 kind: artifact.kind.clone(),
705 title: artifact.title.clone(),
706 source: artifact.source.clone(),
707 stage: artifact.stage.clone(),
708 estimated_tokens: artifact.estimated_tokens,
709 lineage: artifact.lineage.clone(),
710 preview: artifact
711 .text
712 .as_deref()
713 .map(|text| redact_bounded(text, policy, PREVIEW_LIMIT))
714 .or_else(|| {
715 artifact
716 .data
717 .as_ref()
718 .map(|data| redact_json_preview(data, policy))
719 }),
720 }
721}
722
723fn build_checkpoint_view(checkpoint: &RunCheckpointRecord) -> RunViewCheckpoint {
724 RunViewCheckpoint {
725 id: checkpoint.id.clone(),
726 reason: checkpoint.reason.clone(),
727 ready_count: checkpoint.ready_nodes.len(),
728 completed_count: checkpoint.completed_nodes.len(),
729 last_stage_id: checkpoint.last_stage_id.clone(),
730 persisted_at: checkpoint.persisted_at.clone(),
731 }
732}
733
734fn build_approval_view(
735 question: &RunHitlQuestionRecord,
736 policy: &RedactionPolicy,
737) -> RunViewApproval {
738 RunViewApproval {
739 request_id: question.request_id.clone(),
740 prompt: redact_bounded(&question.prompt, policy, PREVIEW_LIMIT),
741 agent: question.agent.clone(),
742 trace_id: question.trace_id.clone(),
743 asked_at: question.asked_at.clone(),
744 }
745}
746
747fn provider_summary(run: &RunRecord) -> Vec<RunViewProvider> {
748 let mut providers = BTreeMap::<(String, String), RunViewProvider>::new();
749 for span in run
750 .trace_spans
751 .iter()
752 .filter(|span| span.kind == "llm_call")
753 {
754 let provider = span
755 .metadata
756 .get("provider")
757 .and_then(Value::as_str)
758 .unwrap_or("unknown")
759 .to_string();
760 let model = span
761 .metadata
762 .get("model")
763 .and_then(Value::as_str)
764 .unwrap_or("unknown")
765 .to_string();
766 let input_tokens = metadata_i64(&span.metadata, "input_tokens");
767 let output_tokens = metadata_i64(&span.metadata, "output_tokens");
768 let cost_usd = span
769 .metadata
770 .get("cost_usd")
771 .and_then(Value::as_f64)
772 .unwrap_or_else(|| {
773 crate::llm::calculate_cost_for_provider(
774 &provider,
775 &model,
776 input_tokens,
777 output_tokens,
778 )
779 });
780 let entry = providers
781 .entry((provider.clone(), model.clone()))
782 .or_insert_with(|| RunViewProvider {
783 provider,
784 model,
785 ..RunViewProvider::default()
786 });
787 entry.call_count += 1;
788 entry.input_tokens += input_tokens;
789 entry.output_tokens += output_tokens;
790 entry.cost_usd += cost_usd;
791 }
792 if providers.is_empty() {
793 if let Some(usage) = &run.usage {
794 for model in &usage.models {
795 if model.is_empty() {
796 continue;
797 }
798 providers.insert(
799 ("unknown".to_string(), model.clone()),
800 RunViewProvider {
801 provider: "unknown".to_string(),
802 model: model.clone(),
803 call_count: usage.call_count,
804 input_tokens: usage.input_tokens,
805 output_tokens: usage.output_tokens,
806 cost_usd: usage.total_cost,
807 },
808 );
809 }
810 }
811 }
812 providers.into_values().collect()
813}
814
815fn transcript_summary_for_run(run: &RunRecord, policy: &RedactionPolicy) -> TranscriptSummary {
816 if let Some(transcript) = run.transcript.as_ref() {
817 return transcript_summary(Some(transcript), policy);
818 }
819 transcript_summary_from_stages(&run.stages, policy)
820}
821
822fn transcript_summary_from_stages(
823 stages: &[RunStageRecord],
824 policy: &RedactionPolicy,
825) -> TranscriptSummary {
826 let mut out = TranscriptSummary::default();
827 let mut summaries = Vec::new();
828 for stage in stages {
829 let Some(transcript) = stage.transcript.as_ref() else {
830 continue;
831 };
832 out.present = true;
833 out.message_count += count_array_field(transcript, "messages");
834 out.event_count += count_array_field(transcript, "events");
835 if let Some(summary) = transcript_summary(Some(transcript), policy).summary {
836 let label = non_empty_string(&stage.node_id)
837 .or_else(|| non_empty_string(&stage.id))
838 .unwrap_or_else(|| "stage".to_string());
839 summaries.push(format!("{label}: {summary}"));
840 }
841 }
842 if out.present {
843 out.summary = bounded_join(summaries, PREVIEW_LIMIT);
844 out.source = Some("stages".to_string());
845 }
846 out
847}
848
849fn transcript_summary(value: Option<&Value>, policy: &RedactionPolicy) -> TranscriptSummary {
850 let Some(value) = value else {
851 return TranscriptSummary::default();
852 };
853 TranscriptSummary {
854 present: true,
855 message_count: count_array_field(value, "messages"),
856 event_count: count_array_field(value, "events"),
857 summary: value
858 .get("summary")
859 .and_then(Value::as_str)
860 .map(|text| redact_bounded(text, policy, PREVIEW_LIMIT))
861 .or_else(|| {
862 value
863 .get("summary")
864 .map(|value| redact_json_preview(value, policy))
865 }),
866 source: value
867 .get("source")
868 .and_then(Value::as_str)
869 .map(str::to_string),
870 }
871}
872
873fn pending_auth(run: &RunRecord, policy: &RedactionPolicy) -> Vec<RunViewAuth> {
874 let mut auth = Vec::new();
875 collect_auth_from_metadata(None, &run.metadata, &mut auth, policy);
876 for stage in &run.stages {
877 collect_auth_from_metadata(Some(&stage.id), &stage.metadata, &mut auth, policy);
878 }
879 auth
880}
881
882fn collect_auth_from_metadata(
883 stage_id: Option<&str>,
884 metadata: &BTreeMap<String, Value>,
885 out: &mut Vec<RunViewAuth>,
886 policy: &RedactionPolicy,
887) {
888 for key in ["pending_auth", "auth_required", "mcp_auth_required"] {
889 let Some(value) = metadata.get(key) else {
890 continue;
891 };
892 match value {
893 Value::Array(items) => {
894 for item in items {
895 out.push(auth_from_value(stage_id, item, policy));
896 }
897 }
898 Value::Object(_) => out.push(auth_from_value(stage_id, value, policy)),
899 Value::Bool(true) => out.push(RunViewAuth {
900 stage_id: stage_id.map(str::to_string),
901 ..RunViewAuth::default()
902 }),
903 Value::String(message) => out.push(RunViewAuth {
904 stage_id: stage_id.map(str::to_string),
905 message: Some(redact_bounded(message, policy, PREVIEW_LIMIT)),
906 ..RunViewAuth::default()
907 }),
908 _ => {}
909 }
910 }
911}
912
913fn auth_from_value(stage_id: Option<&str>, value: &Value, policy: &RedactionPolicy) -> RunViewAuth {
914 let object = value.as_object();
915 let field = |name: &str| {
916 object
917 .and_then(|object| object.get(name))
918 .and_then(Value::as_str)
919 .map(|text| redact_bounded(text, policy, PREVIEW_LIMIT))
920 };
921 RunViewAuth {
922 provider: field("provider"),
923 server: field("server").or_else(|| field("server_name")),
924 scope: field("scope"),
925 stage_id: stage_id.map(str::to_string).or_else(|| field("stage_id")),
926 message: field("message").or_else(|| Some(redact_json_preview(value, policy))),
927 }
928}
929
930fn failure_summary(run: &RunRecord, policy: &RedactionPolicy) -> Option<RunViewFailure> {
931 run.stages
932 .iter()
933 .rev()
934 .find(|stage| failed_status(&stage.status) || failed_status(&stage.outcome))
935 .map(|stage| RunViewFailure {
936 stage_id: Some(stage.id.clone()),
937 node_id: Some(stage.node_id.clone()),
938 status: stage.status.clone(),
939 outcome: stage.outcome.clone(),
940 message: stage_error(stage, policy)
941 .or_else(|| Some(format!("{} failed with {}", stage.node_id, stage.outcome))),
942 })
943 .or_else(|| {
944 failed_status(&run.status).then(|| RunViewFailure {
945 status: run.status.clone(),
946 outcome: run.status.clone(),
947 ..RunViewFailure::default()
948 })
949 })
950}
951
952fn stage_error(stage: &RunStageRecord, policy: &RedactionPolicy) -> Option<String> {
953 stage
954 .metadata
955 .get("error")
956 .map(|value| redact_json_preview(value, policy))
957 .or_else(|| {
958 stage
959 .attempts
960 .iter()
961 .rev()
962 .find_map(|attempt| attempt.error.as_deref())
963 .map(|error| redact_bounded(error, policy, PREVIEW_LIMIT))
964 })
965}
966
967fn failed_status(value: &str) -> bool {
968 matches!(
969 value,
970 "failed" | "error" | "errored" | "cancelled" | "canceled" | "timeout" | "timed_out"
971 )
972}
973
974fn usage_from_stages(stages: &[RunViewStage]) -> RunViewUsage {
975 let mut usage = RunViewUsage::default();
976 for stage in stages {
977 usage.add_usage(&stage.usage);
978 }
979 usage
980}
981
982fn run_duration_ms(run: &RunRecord) -> Option<u64> {
983 let from_usage = run
984 .usage
985 .as_ref()
986 .and_then(|usage| u64::try_from(usage.total_duration_ms).ok())
987 .filter(|duration| *duration > 0);
988 let from_spans = run
989 .trace_spans
990 .iter()
991 .map(trace_span_end_ms)
992 .max()
993 .filter(|duration| *duration > 0);
994 let from_timestamps = run
995 .finished_at
996 .as_deref()
997 .and_then(|finished| timestamp_delta_ms(&run.started_at, finished));
998 from_timestamps.or(from_spans).or(from_usage)
999}
1000
1001fn stage_duration_ms(stage: &RunStageRecord) -> Option<u64> {
1002 stage
1003 .usage
1004 .as_ref()
1005 .and_then(|usage| u64::try_from(usage.total_duration_ms).ok())
1006 .filter(|duration| *duration > 0)
1007 .or_else(|| {
1008 stage
1009 .finished_at
1010 .as_deref()
1011 .and_then(|finished| timestamp_delta_ms(&stage.started_at, finished))
1012 })
1013}
1014
1015use super::time::timestamp_delta_ms;
1016
1017fn trace_span_end_ms(span: &RunTraceSpanRecord) -> u64 {
1018 span.start_ms.saturating_add(span.duration_ms)
1019}
1020
1021fn infer_run_session_id(run: &RunRecord) -> Option<String> {
1022 metadata_string_any(&run.metadata, &["session_id", "agent_session_id"])
1023 .or_else(|| metadata_path_string(&run.metadata, &["model_policy", "session_id"]))
1024 .or_else(|| metadata_path_string(&run.metadata, &["audit", "session_id"]))
1025 .or_else(|| {
1026 run.child_runs
1027 .iter()
1028 .find_map(|child| child.session_id.clone())
1029 })
1030 .or_else(|| {
1031 run.stages.iter().find_map(|stage| {
1032 metadata_string_any(&stage.metadata, &["session_id", "agent_session_id"])
1033 .or_else(|| {
1034 metadata_path_string(&stage.metadata, &["model_policy", "session_id"])
1035 })
1036 .or_else(|| metadata_path_string(&stage.metadata, &["audit", "session_id"]))
1037 .or_else(|| {
1038 metadata_path_string(&stage.metadata, &["worker", "audit", "session_id"])
1039 })
1040 })
1041 })
1042 .or_else(|| {
1043 run.trace_spans.iter().find_map(|span| {
1044 metadata_string_any(&span.metadata, &["session_id", "agent_session_id"])
1045 })
1046 })
1047}
1048
1049fn infer_parent_session_id(run: &RunRecord) -> Option<String> {
1050 metadata_string_any(&run.metadata, &["parent_session_id"])
1051 .or_else(|| metadata_path_string(&run.metadata, &["audit", "parent_session_id"]))
1052 .or_else(|| {
1053 run.child_runs
1054 .iter()
1055 .find_map(|child| child.parent_session_id.clone())
1056 })
1057 .or_else(|| {
1058 run.stages.iter().find_map(|stage| {
1059 metadata_string_any(&stage.metadata, &["parent_session_id"])
1060 .or_else(|| {
1061 metadata_path_string(&stage.metadata, &["audit", "parent_session_id"])
1062 })
1063 .or_else(|| {
1064 metadata_path_string(
1065 &stage.metadata,
1066 &["worker", "audit", "parent_session_id"],
1067 )
1068 })
1069 })
1070 })
1071}
1072
1073fn metadata_string_any(metadata: &BTreeMap<String, Value>, keys: &[&str]) -> Option<String> {
1074 keys.iter()
1075 .find_map(|key| metadata.get(*key).and_then(Value::as_str))
1076 .filter(|value| !value.is_empty())
1077 .map(str::to_string)
1078}
1079
1080fn metadata_path_string(metadata: &BTreeMap<String, Value>, path: &[&str]) -> Option<String> {
1081 let mut value = metadata.get(*path.first()?)?;
1082 for key in &path[1..] {
1083 value = value.get(*key)?;
1084 }
1085 value
1086 .as_str()
1087 .filter(|value| !value.is_empty())
1088 .map(str::to_string)
1089}
1090
1091fn metadata_i64(metadata: &BTreeMap<String, Value>, key: &str) -> i64 {
1092 metadata
1093 .get(key)
1094 .and_then(Value::as_i64)
1095 .or_else(|| {
1096 metadata
1097 .get(key)
1098 .and_then(Value::as_u64)
1099 .and_then(|value| i64::try_from(value).ok())
1100 })
1101 .unwrap_or_default()
1102}
1103
1104fn count_array_field(value: &Value, field: &str) -> usize {
1105 value
1106 .get(field)
1107 .and_then(Value::as_array)
1108 .map(Vec::len)
1109 .unwrap_or_default()
1110}
1111
1112fn redact_json_preview(value: &Value, policy: &RedactionPolicy) -> String {
1113 let mut value = value.clone();
1114 policy.redact_json_in_place(&mut value);
1115 bounded_text(
1116 &serde_json::to_string(&value).unwrap_or_default(),
1117 PREVIEW_LIMIT,
1118 )
1119}
1120
1121fn redact_bounded(text: &str, policy: &RedactionPolicy, limit: usize) -> String {
1122 let redacted = policy.redact_string(text);
1123 bounded_text(redacted.as_ref(), limit)
1124}
1125
1126fn bounded_text(text: &str, limit: usize) -> String {
1127 if text.len() <= limit {
1128 return text.to_string();
1129 }
1130 let boundary = text
1131 .char_indices()
1132 .map(|(index, _)| index)
1133 .take_while(|index| *index <= limit)
1134 .last()
1135 .unwrap_or(0);
1136 format!("{}...", &text[..boundary])
1137}
1138
1139fn bounded_join(values: impl IntoIterator<Item = String>, limit: usize) -> Option<String> {
1140 let mut out = String::new();
1141 for value in values {
1142 if value.is_empty() {
1143 continue;
1144 }
1145 if !out.is_empty() {
1146 out.push_str("\n\n");
1147 }
1148 out.push_str(&value);
1149 if out.len() > limit {
1150 return Some(bounded_text(&out, limit));
1151 }
1152 }
1153 non_empty_string(&out)
1154}
1155
1156fn non_empty_string(value: &str) -> Option<String> {
1157 (!value.is_empty()).then(|| value.to_string())
1158}
1159
1160fn min_opt_string(left: Option<String>, right: Option<String>) -> Option<String> {
1161 match (left, right) {
1162 (Some(left), Some(right)) => Some(left.min(right)),
1163 (Some(left), None) => Some(left),
1164 (None, Some(right)) => Some(right),
1165 (None, None) => None,
1166 }
1167}
1168
1169fn max_opt_string(left: Option<String>, right: Option<String>) -> Option<String> {
1170 match (left, right) {
1171 (Some(left), Some(right)) => Some(left.max(right)),
1172 (Some(left), None) => Some(left),
1173 (None, Some(right)) => Some(right),
1174 (None, None) => None,
1175 }
1176}
1177
1178fn aggregate_session_status(runs: &[RunView]) -> String {
1179 if runs.is_empty() {
1180 return "unknown".to_string();
1181 }
1182 if runs
1183 .iter()
1184 .any(|run| failed_status(&run.run.status) || run.failure.is_some())
1185 {
1186 return "failed".to_string();
1187 }
1188 if runs.iter().all(|run| {
1189 matches!(
1190 run.run.status.as_str(),
1191 "completed" | "succeeded" | "success" | "ok"
1192 )
1193 }) {
1194 return "completed".to_string();
1195 }
1196 "active".to_string()
1197}
1198
1199fn dedupe_pending(mut pending: RunViewPendingState) -> RunViewPendingState {
1200 let mut nodes = BTreeSet::new();
1201 pending.nodes.retain(|node| nodes.insert(node.clone()));
1202 let mut approvals = BTreeSet::new();
1203 pending
1204 .approvals
1205 .retain(|approval| approvals.insert(approval.request_id.clone()));
1206 let mut auth_seen = BTreeSet::new();
1207 pending.auth.retain(|item| {
1208 auth_seen.insert((
1209 item.provider.clone(),
1210 item.server.clone(),
1211 item.scope.clone(),
1212 item.stage_id.clone(),
1213 ))
1214 });
1215 pending
1216}
1217
1218fn finalize_run_projection(view: &mut RunView) {
1219 if let Some(hash) = projection_hash(RUN_VIEW_SCHEMA, view) {
1220 view.projection.projection_id =
1221 format!("run_view:{}:{}", view.run.run_id, hash_suffix(&hash));
1222 view.projection.projection_hash = Some(hash);
1223 } else {
1224 view.projection.projection_id = format!("run_view:{}", view.run.run_id);
1225 }
1226}
1227
1228fn finalize_session_projection(view: &mut SessionView) {
1229 let id = view
1230 .session
1231 .session_id
1232 .clone()
1233 .unwrap_or_else(|| "unknown".to_string());
1234 if let Some(hash) = projection_hash(SESSION_VIEW_SCHEMA, view) {
1235 view.projection.projection_id = format!("session_view:{id}:{}", hash_suffix(&hash));
1236 view.projection.projection_hash = Some(hash);
1237 } else {
1238 view.projection.projection_id = format!("session_view:{id}");
1239 }
1240}
1241
1242fn projection_hash<T: Serialize>(schema: &str, value: &T) -> Option<String> {
1243 let mut value = serde_json::to_value(value).ok()?;
1244 if let Some(projection) = value
1245 .as_object_mut()
1246 .and_then(|object| object.get_mut("projection"))
1247 .and_then(Value::as_object_mut)
1248 {
1249 projection.remove("projection_id");
1250 projection.remove("projection_hash");
1251 }
1252 let bytes = serde_json::to_vec(&value).ok()?;
1253 let mut hasher = Sha256::new();
1254 hasher.update(schema.as_bytes());
1255 hasher.update([0]);
1256 hasher.update(bytes);
1257 Some(format!("sha256:{}", hex::encode(hasher.finalize())))
1258}
1259
1260fn hash_suffix(hash: &str) -> String {
1261 hash.strip_prefix("sha256:")
1262 .unwrap_or(hash)
1263 .chars()
1264 .take(12)
1265 .collect()
1266}
1267
1268#[cfg(test)]
1269mod tests {
1270 use serde_json::json;
1271
1272 use super::*;
1273
1274 fn sample_run() -> RunRecord {
1275 RunRecord {
1276 type_name: "run_record".to_string(),
1277 id: "run_1".to_string(),
1278 workflow_id: "wf".to_string(),
1279 workflow_name: Some("Workflow".to_string()),
1280 task: "do work".to_string(),
1281 status: "completed".to_string(),
1282 started_at: "2026-01-01T00:00:00Z".to_string(),
1283 finished_at: Some("2026-01-01T00:00:02Z".to_string()),
1284 stages: vec![RunStageRecord {
1285 id: "stage_1".to_string(),
1286 node_id: "plan".to_string(),
1287 kind: "llm".to_string(),
1288 status: "completed".to_string(),
1289 outcome: "ok".to_string(),
1290 started_at: "2026-01-01T00:00:00Z".to_string(),
1291 finished_at: Some("2026-01-01T00:00:01Z".to_string()),
1292 visible_text: Some("done".to_string()),
1293 usage: Some(LlmUsageRecord {
1294 input_tokens: 10,
1295 output_tokens: 5,
1296 total_duration_ms: 1000,
1297 call_count: 1,
1298 total_cost: 0.01,
1299 models: vec!["model-a".to_string()],
1300 }),
1301 metadata: BTreeMap::from([
1302 ("session_id".to_string(), json!("session_1")),
1303 ("provider".to_string(), json!("test")),
1304 ("model".to_string(), json!("model-a")),
1305 ]),
1306 ..RunStageRecord::default()
1307 }],
1308 trace_spans: vec![RunTraceSpanRecord {
1309 kind: "llm_call".to_string(),
1310 metadata: BTreeMap::from([
1311 ("provider".to_string(), json!("test")),
1312 ("model".to_string(), json!("model-a")),
1313 ("input_tokens".to_string(), json!(10)),
1314 ("output_tokens".to_string(), json!(5)),
1315 ("cost_usd".to_string(), json!(0.01)),
1316 ]),
1317 ..RunTraceSpanRecord::default()
1318 }],
1319 transcript: Some(json!({
1320 "source": "inline",
1321 "summary": "short",
1322 "messages": [{"role": "assistant"}],
1323 "events": [{"kind": "output"}]
1324 })),
1325 ..RunRecord::default()
1326 }
1327 }
1328
1329 #[test]
1330 fn build_run_view_projects_stable_public_fields() {
1331 let view = build_run_view_with_path(&sample_run(), Some("runs/run_1.json"));
1332 assert_eq!(view.schema, RUN_VIEW_SCHEMA);
1333 assert_eq!(view.schema_version, RUN_VIEW_SCHEMA_VERSION);
1334 assert_eq!(view.run.run_id, "run_1");
1335 assert_eq!(view.run.session_id.as_deref(), Some("session_1"));
1336 assert_eq!(view.run.run_path.as_deref(), Some("runs/run_1.json"));
1337 assert_eq!(view.run.duration_ms, Some(2000));
1338 assert_eq!(view.visible_text.as_deref(), Some("done"));
1339 assert_eq!(view.transcript.message_count, 1);
1340 assert_eq!(view.usage.input_tokens, 10);
1341 assert_eq!(view.providers.len(), 1);
1342 assert!(view.projection.projection_id.starts_with("run_view:run_1:"));
1343 assert!(view.projection.projection_hash.is_some());
1344 }
1345
1346 #[test]
1347 fn build_run_view_falls_back_to_public_assistant_transcript_blocks() {
1348 let secret = "sk-proj-test-abcdefghijklmnopqrstuvwxyz123456";
1349 let run = RunRecord {
1350 type_name: "run_record".to_string(),
1351 id: "transcript_only".to_string(),
1352 status: "completed".to_string(),
1353 transcript: Some(json!({
1354 "events": [
1355 {
1356 "kind": "message",
1357 "role": "user",
1358 "visibility": "public",
1359 "blocks": [{
1360 "type": "text",
1361 "text": "user prompt must not become output",
1362 "visibility": "public"
1363 }]
1364 },
1365 {
1366 "kind": "message",
1367 "role": "assistant",
1368 "visibility": "public",
1369 "text": "event-level text is not an authority",
1370 "blocks": [
1371 {"type": "output_text", "text": "visible answer ", "visibility": "public"},
1372 {"type": "text", "text": format!("with {secret}"), "visibility": "public"},
1373 {"type": "reasoning", "text": "public reasoning stays private", "visibility": "public"},
1374 {"type": "output_text", "text": "private output", "visibility": "private"}
1375 ]
1376 },
1377 {
1378 "kind": "message",
1379 "role": "assistant",
1380 "visibility": "private",
1381 "blocks": [{
1382 "type": "output_text",
1383 "text": "private event output",
1384 "visibility": "public"
1385 }]
1386 }
1387 ]
1388 })),
1389 ..RunRecord::default()
1390 };
1391
1392 let view = build_run_view(&run);
1393 let visible = view.visible_text.expect("public assistant output");
1394 assert!(visible.starts_with("visible answer with "));
1395 assert!(visible.contains("<redacted:openai_key:"));
1396 assert!(!visible.contains(secret));
1397 assert!(!visible.contains("user prompt"));
1398 assert!(!visible.contains("event-level text"));
1399 assert!(!visible.contains("reasoning"));
1400 assert!(!visible.contains("private output"));
1401 assert!(!visible.contains("private event output"));
1402 }
1403
1404 #[test]
1405 fn build_run_view_tolerates_sparse_legacy_records() {
1406 let run = RunRecord {
1407 type_name: "run_record".to_string(),
1408 id: "legacy".to_string(),
1409 status: "failed".to_string(),
1410 ..RunRecord::default()
1411 };
1412 let view = build_run_view(&run);
1413 assert_eq!(view.run.run_id, "legacy");
1414 assert_eq!(view.run.session_id, None);
1415 assert!(!view.transcript.present);
1416 assert_eq!(
1417 view.failure.as_ref().map(|failure| failure.status.as_str()),
1418 Some("failed")
1419 );
1420 }
1421
1422 #[test]
1423 fn build_session_view_aggregates_runs() {
1424 let run = build_run_view(&sample_run());
1425 let view = build_session_view_from_run_views(
1426 vec![run],
1427 SessionViewOptions {
1428 session_id: Some("session_1".to_string()),
1429 last_event_id: Some(7),
1430 chain_root_hash: Some("sha256:abc".to_string()),
1431 ..SessionViewOptions::default()
1432 },
1433 );
1434 assert_eq!(view.schema, SESSION_VIEW_SCHEMA);
1435 assert_eq!(view.session.session_id.as_deref(), Some("session_1"));
1436 assert_eq!(view.session.last_event_id, Some(7));
1437 assert_eq!(view.session.chain_root_hash.as_deref(), Some("sha256:abc"));
1438 assert_eq!(view.history.len(), 1);
1439 assert_eq!(view.usage.call_count, 1);
1440 assert!(view
1441 .projection
1442 .projection_id
1443 .starts_with("session_view:session_1:"));
1444 }
1445
1446 #[test]
1447 fn build_run_view_summarizes_stage_only_transcripts() {
1448 let mut run = sample_run();
1449 run.transcript = None;
1450 run.stages[0].transcript = Some(json!({
1451 "summary": "stage transcript only",
1452 "messages": [{"role": "assistant"}, {"role": "tool"}],
1453 "events": [{"kind": "tool_result"}]
1454 }));
1455
1456 let view = build_run_view(&run);
1457 assert!(view.transcript.present);
1458 assert_eq!(view.transcript.source.as_deref(), Some("stages"));
1459 assert_eq!(view.transcript.message_count, 2);
1460 assert_eq!(view.transcript.event_count, 1);
1461 assert_eq!(
1462 view.transcript.summary.as_deref(),
1463 Some("plan: stage transcript only")
1464 );
1465 }
1466
1467 #[test]
1468 fn build_run_view_redacts_child_tasks_and_approvals() {
1469 let mut run = sample_run();
1470 run.child_runs.push(RunChildRecord {
1471 worker_id: "worker_1".to_string(),
1472 worker_name: "worker".to_string(),
1473 task: "inspect AKIAABCDEFGHIJKLMNOP".to_string(),
1474 ..RunChildRecord::default()
1475 });
1476 run.hitl_questions.push(RunHitlQuestionRecord {
1477 request_id: "approval_1".to_string(),
1478 prompt: "approve AKIAABCDEFGHIJKLMNOP".to_string(),
1479 ..RunHitlQuestionRecord::default()
1480 });
1481
1482 let view = build_run_view(&run);
1483 assert!(!view.run.child_runs[0].task.contains("AKIAABCDEFGHIJKLMNOP"));
1484 assert!(!view.pending.approvals[0]
1485 .prompt
1486 .contains("AKIAABCDEFGHIJKLMNOP"));
1487 }
1488}