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