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