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