use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::time::SystemTime;

use axum::http::StatusCode;
use axum::Json;

use super::dto::{
    PortalActivity, PortalActivityAudit, PortalActivityAuditLayer, PortalArtifact,
    PortalCheckpoint, PortalChildRun, PortalCostReport, PortalCostSummary, PortalCostTrendPoint,
    PortalExecutionSummary, PortalInsight, PortalPolicySummary, PortalProviderCostBreakdown,
    PortalReplayAssertion, PortalReplaySummary, PortalRunDetail, PortalRunSummary, PortalSpan,
    PortalStage, PortalStageDebug, PortalStats, PortalStorySection, PortalTranscriptStep,
    PortalTransition,
};
use super::errors::{bad_request_error, internal_error};
use super::query::{ErrorResponse, ListRunsQuery};
use super::skill_events::{extract as extract_skill_events, ExtractedEvents};
use super::transcript::{build_story, discover_template_renders, discover_transcript_steps};
use super::util::{
    compact_json, compact_metadata, date_ms, format_duration, humanize_kind, is_completed_status,
    is_failed_status, metadata_pretty_json, metadata_string, owning_stage, preview_text,
    redacted_json_value, redacted_metadata, span_kind_totals, string_array_value, system_time_ms,
};

pub(super) fn scan_runs(run_dir: &Path) -> Result<Vec<PortalRunSummary>, String> {
    let mut files = Vec::new();
    collect_run_files(run_dir, run_dir, &mut files)?;

    let mut runs = Vec::new();
    for (path, modified_at_ms) in files {
        if let Ok(run) = harn_vm::orchestration::load_run_record(&path) {
            if run.type_name != "run_record" || run.id.is_empty() || run.workflow_id.is_empty() {
                continue;
            }
            let relative = path
                .strip_prefix(run_dir)
                .ok()
                .unwrap_or(&path)
                .display()
                .to_string();
            runs.push(build_run_summary(&relative, modified_at_ms, &run));
        }
    }

    runs.sort_by(|a, b| {
        b.started_at
            .cmp(&a.started_at)
            .then_with(|| b.updated_at_ms.cmp(&a.updated_at_ms))
    });
    Ok(runs)
}

pub(super) fn build_cost_report(run_dir: &Path) -> Result<PortalCostReport, String> {
    let mut files = Vec::new();
    collect_run_files(run_dir, run_dir, &mut files)?;

    let mut trend = HashMap::<(String, String), PortalCostTrendPoint>::new();
    let mut provider_breakdown = HashMap::<(String, String), PortalProviderCostBreakdown>::new();
    let mut summary = PortalCostSummary {
        total_cost_usd: 0.0,
        call_count: 0,
        input_tokens: 0,
        output_tokens: 0,
    };

    for (path, _) in files {
        let Ok(run) = harn_vm::orchestration::load_run_record(&path) else {
            continue;
        };
        let pipeline = run
            .workflow_name
            .clone()
            .unwrap_or_else(|| run.workflow_id.clone());
        let date = run
            .started_at
            .split(['T', ' '])
            .next()
            .filter(|value| !value.is_empty())
            .unwrap_or("unknown")
            .to_string();

        for span in run
            .trace_spans
            .iter()
            .filter(|span| span.kind == "llm_call")
        {
            let provider = span
                .metadata
                .get("provider")
                .and_then(|value| value.as_str())
                .unwrap_or("unknown")
                .to_string();
            let model = span
                .metadata
                .get("model")
                .and_then(|value| value.as_str())
                .unwrap_or("unknown")
                .to_string();
            let input_tokens = span
                .metadata
                .get("input_tokens")
                .and_then(|value| value.as_i64())
                .unwrap_or(0);
            let output_tokens = span
                .metadata
                .get("output_tokens")
                .and_then(|value| value.as_i64())
                .unwrap_or(0);
            let cost_usd = span
                .metadata
                .get("cost_usd")
                .and_then(|value| value.as_f64())
                .unwrap_or_else(|| {
                    harn_vm::llm::calculate_cost_for_provider(
                        &provider,
                        &model,
                        input_tokens,
                        output_tokens,
                    )
                });
            if cost_usd <= 0.0 && input_tokens == 0 && output_tokens == 0 {
                continue;
            }

            summary.total_cost_usd += cost_usd;
            summary.call_count += 1;
            summary.input_tokens += input_tokens;
            summary.output_tokens += output_tokens;

            let trend_entry = trend
                .entry((date.clone(), pipeline.clone()))
                .or_insert_with(|| PortalCostTrendPoint {
                    date: date.clone(),
                    pipeline: pipeline.clone(),
                    cost_usd: 0.0,
                    call_count: 0,
                    input_tokens: 0,
                    output_tokens: 0,
                });
            trend_entry.cost_usd += cost_usd;
            trend_entry.call_count += 1;
            trend_entry.input_tokens += input_tokens;
            trend_entry.output_tokens += output_tokens;

            let provider_entry = provider_breakdown
                .entry((provider.clone(), model.clone()))
                .or_insert_with(|| PortalProviderCostBreakdown {
                    provider: provider.clone(),
                    model: model.clone(),
                    cost_usd: 0.0,
                    call_count: 0,
                    input_tokens: 0,
                    output_tokens: 0,
                });
            provider_entry.cost_usd += cost_usd;
            provider_entry.call_count += 1;
            provider_entry.input_tokens += input_tokens;
            provider_entry.output_tokens += output_tokens;
        }
    }

    let mut trend = trend.into_values().collect::<Vec<_>>();
    trend.sort_by(|left, right| {
        left.date
            .cmp(&right.date)
            .then_with(|| left.pipeline.cmp(&right.pipeline))
    });
    let mut provider_breakdown = provider_breakdown.into_values().collect::<Vec<_>>();
    provider_breakdown.sort_by(|left, right| {
        right
            .cost_usd
            .partial_cmp(&left.cost_usd)
            .unwrap_or(std::cmp::Ordering::Equal)
            .then_with(|| left.provider.cmp(&right.provider))
            .then_with(|| left.model.cmp(&right.model))
    });

    Ok(PortalCostReport {
        summary,
        trend,
        provider_breakdown,
    })
}

pub(super) fn filter_and_sort_runs(
    runs: Vec<PortalRunSummary>,
    query: &ListRunsQuery,
) -> Vec<PortalRunSummary> {
    let search = query
        .q
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(|value| value.to_ascii_lowercase());
    let workflow = query
        .workflow
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string);
    let status = query.status.as_deref().unwrap_or("all");
    let sort = query.sort.as_deref().unwrap_or("newest");
    let skill_filter = query
        .skill
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string);

    let mut filtered = runs
        .into_iter()
        .filter(|run| match workflow.as_deref() {
            Some(name) => run.workflow_name == name,
            None => true,
        })
        .filter(|run| matches_run_status(run, status))
        .filter(|run| match skill_filter.as_deref() {
            Some(skill) => run.skills.iter().any(|s| s == skill),
            None => true,
        })
        .filter(|run| match search.as_deref() {
            Some(needle) => matches_run_search(run, needle),
            None => true,
        })
        .collect::<Vec<_>>();

    filtered.sort_by(|left, right| match sort {
        "duration" => right
            .duration_ms
            .unwrap_or_default()
            .cmp(&left.duration_ms.unwrap_or_default())
            .then_with(|| right.started_at.cmp(&left.started_at)),
        "oldest" => left
            .started_at
            .cmp(&right.started_at)
            .then_with(|| left.updated_at_ms.cmp(&right.updated_at_ms)),
        _ => right
            .started_at
            .cmp(&left.started_at)
            .then_with(|| right.updated_at_ms.cmp(&left.updated_at_ms)),
    });
    filtered
}

fn matches_run_status(run: &PortalRunSummary, status: &str) -> bool {
    match status {
        "failed" => is_failed_status(&run.status),
        "completed" => is_completed_status(&run.status),
        "active" => !is_failed_status(&run.status) && !is_completed_status(&run.status),
        _ => true,
    }
}

fn matches_run_search(run: &PortalRunSummary, needle: &str) -> bool {
    let models = run.models.join(" ");
    format!(
        "{} {} {} {} {} {}",
        run.workflow_name,
        run.status,
        run.path,
        run.last_stage_node_id.as_deref().unwrap_or_default(),
        run.failure_summary.as_deref().unwrap_or_default(),
        models
    )
    .to_ascii_lowercase()
    .contains(needle)
}

fn trace_span_end_ms(span: &harn_vm::orchestration::RunTraceSpanRecord) -> u64 {
    span.start_ms.saturating_add(span.duration_ms)
}

fn collect_run_files(
    root: &Path,
    current: &Path,
    out: &mut Vec<(PathBuf, u128)>,
) -> Result<(), String> {
    let entries = fs::read_dir(current)
        .map_err(|error| format!("failed to read {}: {error}", current.display()))?;
    for entry in entries {
        let entry =
            entry.map_err(|error| format!("failed to iterate {}: {error}", current.display()))?;
        let path = entry.path();
        let metadata = fs::symlink_metadata(&path)
            .map_err(|error| format!("failed to stat {}: {error}", path.display()))?;
        if metadata.file_type().is_symlink() {
            continue;
        }
        if metadata.is_dir() {
            collect_run_files(root, &path, out)?;
            continue;
        }
        if path.extension().and_then(|ext| ext.to_str()) != Some("json") {
            continue;
        }
        let modified_at_ms = Some(metadata)
            .and_then(|meta| meta.modified().ok())
            .and_then(system_time_ms)
            .unwrap_or_else(|| system_time_ms(SystemTime::now()).unwrap_or(0));
        if path.starts_with(root) {
            out.push((path, modified_at_ms));
        }
    }
    Ok(())
}

pub(super) fn resolve_run_path(
    root: &Path,
    relative: &str,
) -> Result<PathBuf, (StatusCode, Json<ErrorResponse>)> {
    if relative.trim().is_empty() {
        return Err(bad_request_error("run path is required"));
    }
    let relative_path = Path::new(relative);
    if relative_path.is_absolute() {
        return Err(bad_request_error(
            "run path must be relative to the configured run directory",
        ));
    }
    if relative_path.components().any(|component| {
        matches!(
            component,
            std::path::Component::ParentDir
                | std::path::Component::RootDir
                | std::path::Component::Prefix(_)
        )
    }) {
        return Err(bad_request_error(
            "run path must stay inside the configured run directory",
        ));
    }
    let joined = root.join(relative_path);
    if joined.exists() {
        let canonical_root = root
            .canonicalize()
            .map_err(|error| internal_error(format!("failed to resolve run directory: {error}")))?;
        let canonical_joined = joined
            .canonicalize()
            .map_err(|error| internal_error(format!("failed to resolve run path: {error}")))?;
        if !canonical_joined.starts_with(&canonical_root) {
            return Err(bad_request_error(
                "run path must stay inside the configured run directory",
            ));
        }
    }
    Ok(joined)
}

pub(super) fn build_run_summary(
    path: &str,
    updated_at_ms: u128,
    run: &harn_vm::orchestration::RunRecord,
) -> PortalRunSummary {
    let usage = run.usage.clone().unwrap_or_default();
    let (last_stage_node_id, failure_summary) = latest_stage_summary(run);
    let extracted = extract_skill_events(run);
    PortalRunSummary {
        path: path.to_string(),
        id: run.id.clone(),
        workflow_name: run
            .workflow_name
            .clone()
            .unwrap_or_else(|| run.workflow_id.clone()),
        status: run.status.clone(),
        last_stage_node_id,
        failure_summary,
        started_at: run.started_at.clone(),
        finished_at: run.finished_at.clone(),
        duration_ms: run_duration_ms(run),
        stage_count: run.stages.len(),
        child_run_count: run.child_runs.len(),
        call_count: usage.call_count,
        input_tokens: usage.input_tokens,
        output_tokens: usage.output_tokens,
        models: usage.models,
        updated_at_ms,
        skills: extracted.active_skills,
    }
}

fn latest_stage_summary(
    run: &harn_vm::orchestration::RunRecord,
) -> (Option<String>, Option<String>) {
    let last_stage = run.stages.last();
    let last_stage_node_id = last_stage.map(|stage| stage.node_id.clone());
    let failure_summary = run
        .stages
        .iter()
        .rev()
        .find(|stage| is_failed_status(&stage.status) || is_failed_status(&stage.outcome))
        .map(|stage| {
            let error = stage.metadata.get("error").map(compact_json).or_else(|| {
                stage
                    .attempts
                    .iter()
                    .rev()
                    .find_map(|attempt| attempt.error.clone())
            });
            match error {
                Some(error) if !error.is_empty() => format!("{} failed: {}", stage.node_id, error),
                _ => format!("{} failed with {}", stage.node_id, stage.outcome),
            }
        });
    (last_stage_node_id, failure_summary)
}

pub(super) fn summarize_runs(runs: &[PortalRunSummary]) -> PortalStats {
    let total_runs = runs.len();
    let completed_runs = runs
        .iter()
        .filter(|run| is_completed_status(&run.status))
        .count();
    let active_runs = runs
        .iter()
        .filter(|run| !is_completed_status(&run.status) && !is_failed_status(&run.status))
        .count();
    let failed_runs = runs
        .iter()
        .filter(|run| is_failed_status(&run.status))
        .count();
    let durations: Vec<u64> = runs.iter().filter_map(|run| run.duration_ms).collect();
    let avg_duration_ms = if durations.is_empty() {
        0
    } else {
        let total = durations
            .iter()
            .map(|duration| u128::from(*duration))
            .sum::<u128>();
        u64::try_from(total / durations.len() as u128).unwrap_or(u64::MAX)
    };
    PortalStats {
        total_runs,
        completed_runs,
        active_runs,
        failed_runs,
        avg_duration_ms,
    }
}

pub(super) fn build_run_detail(
    run_dir: &Path,
    relative_path: &str,
    run: &harn_vm::orchestration::RunRecord,
) -> PortalRunDetail {
    let summary = build_run_summary(relative_path, 0, run);
    let spans = build_spans(run);
    let stages = build_stages(run);
    let audit_index = build_tool_call_audit_index(run);
    let activities = build_activities(&spans, &stages, &audit_index);
    let transitions = build_transitions(run);
    let checkpoints = build_checkpoints(run);
    let artifacts = build_artifacts(run);
    let transcript_steps = discover_transcript_steps(run_dir, relative_path).unwrap_or_default();
    let template_renders = discover_template_renders(run_dir, relative_path).unwrap_or_default();
    let story = build_story(run);
    let child_runs = run
        .child_runs
        .iter()
        .map(|child| PortalChildRun {
            worker_name: child.worker_name.clone(),
            status: child.status.clone(),
            started_at: child.started_at.clone(),
            finished_at: child.finished_at.clone(),
            run_id: child.run_id.clone(),
            run_path: child.run_path.as_ref().and_then(|path| {
                PathBuf::from(path)
                    .strip_prefix(run_dir)
                    .ok()
                    .map(|value| value.display().to_string())
                    .or_else(|| Some(path.clone()))
            }),
            task: child.task.clone(),
        })
        .collect::<Vec<_>>();
    let insights = build_insights(run, &summary, &stages, &spans, &story, &transcript_steps);
    let ExtractedEvents {
        skill_timeline,
        skill_matches,
        tool_loads,
        active_skills,
    } = extract_skill_events(run);

    PortalRunDetail {
        summary,
        task: run.task.clone(),
        workflow_id: run.workflow_id.clone(),
        parent_run_id: run.parent_run_id.clone(),
        root_run_id: run.root_run_id.clone(),
        policy_summary: build_policy_summary(run),
        replay_summary: build_replay_summary(run.replay_fixture.as_ref()),
        execution: run.execution.clone(),
        insights,
        stages,
        spans,
        activities,
        transitions,
        checkpoints,
        artifacts,
        execution_summary: build_execution_summary(run.execution.as_ref()),
        transcript_steps,
        template_renders,
        story,
        child_runs,
        observability: run.observability.clone().unwrap_or_else(|| {
            let run_path = run_dir.join(relative_path);
            harn_vm::orchestration::derive_run_observability(run, Some(&run_path))
        }),
        skill_timeline,
        skill_match_events: skill_matches,
        tool_load_events: tool_loads,
        active_skills,
    }
}

fn build_insights(
    run: &harn_vm::orchestration::RunRecord,
    summary: &PortalRunSummary,
    stages: &[PortalStage],
    spans: &[PortalSpan],
    story: &[PortalStorySection],
    transcript_steps: &[PortalTranscriptStep],
) -> Vec<PortalInsight> {
    let slowest_stage = stages
        .iter()
        .filter_map(|stage| stage.duration_ms.map(|duration| (stage, duration)))
        .max_by_key(|(_, duration)| *duration);
    let noisiest_kind = span_kind_totals(spans)
        .into_iter()
        .max_by_key(|(_, duration)| *duration);
    let top_story = story.first();
    let heaviest_turn = transcript_steps
        .iter()
        .max_by_key(|step| (step.total_messages, step.input_tokens.unwrap_or_default()));
    vec![
        PortalInsight {
            label: "Run shape".to_string(),
            value: format!(
                "{} stages • {} child runs",
                summary.stage_count, summary.child_run_count
            ),
            detail: format!(
                "status={} • {} transcript sections",
                run.status,
                story.len()
            ),
        },
        PortalInsight {
            label: "Slowest stage".to_string(),
            value: slowest_stage
                .map(|(stage, duration)| {
                    format!("{} ({})", stage.node_id, format_duration(duration))
                })
                .unwrap_or_else(|| "No timed stages".to_string()),
            detail: slowest_stage
                .map(|(stage, _)| {
                    format!(
                        "{} attempts • {} artifacts",
                        stage.attempt_count, stage.artifact_count
                    )
                })
                .unwrap_or_else(|| "Stage timing metadata was not available".to_string()),
        },
        PortalInsight {
            label: "Where time went".to_string(),
            value: noisiest_kind
                .as_ref()
                .map(|(kind, duration)| {
                    format!("{} ({})", humanize_kind(kind), format_duration(*duration))
                })
                .unwrap_or_else(|| "No trace spans".to_string()),
            detail: format!("{} spans captured", spans.len()),
        },
        PortalInsight {
            label: "First thing to read".to_string(),
            value: top_story
                .map(|section| section.title.clone())
                .unwrap_or_else(|| "No transcript sections".to_string()),
            detail: top_story
                .map(|section| section.preview.clone())
                .unwrap_or_else(|| {
                    "This run has trace data but no captured visible transcript".to_string()
                }),
        },
        PortalInsight {
            label: "Heaviest model turn".to_string(),
            value: heaviest_turn
                .map(|step| format!("Step {} ({})", step.call_index, step.total_messages))
                .unwrap_or_else(|| "No saved model transcript".to_string()),
            detail: heaviest_turn
                .map(|step| {
                    format!(
                        "kept {} • added {} • {}",
                        step.kept_messages, step.added_messages, step.summary
                    )
                })
                .unwrap_or_else(|| {
                    "Enable HARN_LLM_TRANSCRIPT_DIR to persist full model-turn detail".to_string()
                }),
        },
    ]
}

fn build_stages(run: &harn_vm::orchestration::RunRecord) -> Vec<PortalStage> {
    run.stages
        .iter()
        .map(|stage| PortalStage {
            id: stage.id.clone(),
            node_id: stage.node_id.clone(),
            kind: stage.kind.clone(),
            status: stage.status.clone(),
            outcome: stage.outcome.clone(),
            branch: stage.branch.clone(),
            started_at: stage.started_at.clone(),
            finished_at: stage.finished_at.clone(),
            duration_ms: stage_duration_ms(stage),
            artifact_count: stage.artifacts.len(),
            attempt_count: stage.attempts.len(),
            verification_summary: stage.verification.as_ref().map(compact_json),
            debug: build_stage_debug(stage),
        })
        .collect()
}

fn build_spans(run: &harn_vm::orchestration::RunRecord) -> Vec<PortalSpan> {
    let mut spans = run.trace_spans.clone();
    spans.sort_by(|a, b| {
        a.start_ms
            .cmp(&b.start_ms)
            .then_with(|| b.duration_ms.cmp(&a.duration_ms))
    });
    let by_id: HashMap<u64, harn_vm::orchestration::RunTraceSpanRecord> = spans
        .iter()
        .map(|span| (span.span_id, span.clone()))
        .collect();
    let mut lane_ends: Vec<u64> = Vec::new();

    spans
        .into_iter()
        .map(|span| {
            let depth = span_depth(&span, &by_id);
            let lane = match lane_ends.iter().position(|end| *end <= span.start_ms) {
                Some(index) => {
                    lane_ends[index] = trace_span_end_ms(&span);
                    index
                }
                None => {
                    lane_ends.push(trace_span_end_ms(&span));
                    lane_ends.len() - 1
                }
            };

            PortalSpan {
                span_id: span.span_id,
                parent_id: span.parent_id,
                kind: span.kind.clone(),
                name: span.name.clone(),
                start_ms: span.start_ms,
                duration_ms: span.duration_ms,
                end_ms: trace_span_end_ms(&span),
                label: span_label(&span),
                lane,
                depth,
                metadata: redacted_metadata(&span.metadata),
            }
        })
        .collect()
}

fn build_activities(
    spans: &[PortalSpan],
    stages: &[PortalStage],
    audit_index: &HashMap<String, PortalActivityAudit>,
) -> Vec<PortalActivity> {
    spans
        .iter()
        .filter(|span| span.kind != "pipeline")
        .map(|span| {
            let call_id = span
                .metadata
                .get("call_id")
                .and_then(|value| value.as_str())
                .map(str::to_string);
            let audit = call_id
                .as_deref()
                .and_then(|id| audit_index.get(id).cloned());
            PortalActivity {
                label: span.label.clone(),
                kind: span.kind.clone(),
                started_offset_ms: span.start_ms,
                duration_ms: span.duration_ms,
                stage_node_id: owning_stage(span, stages).map(|stage| stage.node_id.clone()),
                call_id,
                summary: compact_metadata(&span.metadata),
                audit,
            }
        })
        .collect()
}

/// Build a lookup table from `tool_call_id` to the joined audit/receipt
/// payload by scanning every transcript on the run (run-level plus
/// per-stage). One pass per run, O(1) lookup per activity span — keeps
/// `build_activities` linear in the span count even when the audit
/// stream is large.
fn build_tool_call_audit_index(
    run: &harn_vm::orchestration::RunRecord,
) -> HashMap<String, PortalActivityAudit> {
    let mut index: HashMap<String, PortalActivityAudit> = HashMap::new();
    if let Some(transcript) = &run.transcript {
        collect_tool_call_audit_events(&redacted_json_value(transcript), &mut index);
    }
    for stage in &run.stages {
        if let Some(transcript) = &stage.transcript {
            collect_tool_call_audit_events(&redacted_json_value(transcript), &mut index);
        }
    }
    index
}

fn collect_tool_call_audit_events(
    transcript: &serde_json::Value,
    index: &mut HashMap<String, PortalActivityAudit>,
) {
    let Some(events) = transcript.get("events").and_then(|value| value.as_array()) else {
        return;
    };
    for event in events {
        let kind = event
            .get("kind")
            .and_then(|value| value.as_str())
            .unwrap_or_default();
        if kind != "tool_call_audit" {
            continue;
        }
        let metadata = event.get("metadata");
        let Some(call_id) = metadata
            .and_then(|m| m.get("tool_call_id"))
            .and_then(|value| value.as_str())
            .filter(|value| !value.is_empty())
        else {
            continue;
        };
        let entry = portal_audit_from_event_metadata(metadata.unwrap_or(&serde_json::Value::Null));
        // Latest audit for a given `tool_call_id` wins — middleware can
        // emit multiple events in sequence (e.g. consent prompt update
        // followed by the final outcome) and the most recent payload
        // carries the terminal status + receipt.
        index.insert(call_id.to_string(), entry);
    }
}

fn portal_audit_from_event_metadata(metadata: &serde_json::Value) -> PortalActivityAudit {
    let receipt = metadata.get("receipt");
    let audit = metadata
        .get("audit")
        .cloned()
        .unwrap_or(serde_json::Value::Null);
    let reason = audit
        .get("summary")
        .and_then(|value| value.as_str())
        .map(str::to_string)
        .or_else(|| {
            receipt
                .and_then(|r| r.get("reason"))
                .and_then(|value| value.as_str())
                .map(str::to_string)
        })
        .filter(|value| !value.is_empty());
    let kind = audit
        .get("kind")
        .and_then(|value| value.as_str())
        .map(str::to_string)
        .or_else(|| {
            receipt
                .and_then(|r| r.get("kind"))
                .and_then(|value| value.as_str())
                .map(str::to_string)
        })
        .filter(|value| !value.is_empty());
    let status = receipt
        .and_then(|r| r.get("status"))
        .and_then(|value| value.as_str())
        .map(str::to_string)
        .unwrap_or_else(|| "unknown".to_string());
    let layers = audit
        .get("layers")
        .and_then(|value| value.as_array())
        .map(|items| {
            items
                .iter()
                .filter_map(portal_audit_layer_from_value)
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();
    let receipt_uri = audit
        .get("receipt_uri")
        .and_then(|value| value.as_str())
        .or_else(|| {
            receipt
                .and_then(|r| r.get("audit"))
                .and_then(|value| value.get("receipt_uri"))
                .and_then(|value| value.as_str())
        })
        .map(str::to_string)
        .filter(|value| !value.is_empty());

    PortalActivityAudit {
        reason,
        kind,
        status,
        layers,
        receipt_uri,
    }
}

fn portal_audit_layer_from_value(value: &serde_json::Value) -> Option<PortalActivityAuditLayer> {
    let name = value
        .get("name")
        .and_then(|v| v.as_str())
        .map(str::to_string)
        .filter(|v| !v.is_empty())?;
    let status = value
        .get("status")
        .and_then(|v| v.as_str())
        .map(str::to_string)
        .unwrap_or_else(|| "unknown".to_string());
    let metadata = value
        .as_object()
        .map(|obj| {
            let mut clone = serde_json::Map::new();
            for (key, val) in obj {
                if key == "name" || key == "status" {
                    continue;
                }
                clone.insert(key.clone(), val.clone());
            }
            clone
        })
        .filter(|map| !map.is_empty())
        .map(serde_json::Value::Object);
    Some(PortalActivityAuditLayer {
        name,
        status,
        metadata,
    })
}

fn build_transitions(run: &harn_vm::orchestration::RunRecord) -> Vec<PortalTransition> {
    run.transitions
        .iter()
        .map(|transition| PortalTransition {
            from_node_id: transition.from_node_id.clone(),
            to_node_id: transition.to_node_id.clone(),
            branch: transition.branch.clone(),
            consumed_count: transition.consumed_artifact_ids.len(),
            produced_count: transition.produced_artifact_ids.len(),
        })
        .collect()
}

fn build_checkpoints(run: &harn_vm::orchestration::RunRecord) -> Vec<PortalCheckpoint> {
    run.checkpoints
        .iter()
        .map(|checkpoint| PortalCheckpoint {
            reason: checkpoint.reason.clone(),
            ready_count: checkpoint.ready_nodes.len(),
            completed_count: checkpoint.completed_nodes.len(),
            last_stage_id: checkpoint.last_stage_id.clone(),
        })
        .collect()
}

fn build_artifacts(run: &harn_vm::orchestration::RunRecord) -> Vec<PortalArtifact> {
    run.artifacts
        .iter()
        .map(|artifact| PortalArtifact {
            id: artifact.id.clone(),
            kind: artifact.kind.clone(),
            title: artifact
                .title
                .clone()
                .unwrap_or_else(|| artifact.kind.clone()),
            source: artifact.source.clone(),
            stage: artifact.stage.clone(),
            estimated_tokens: artifact.estimated_tokens,
            lineage_count: artifact.lineage.len(),
            preview: artifact
                .text
                .as_deref()
                .map(preview_text)
                .or_else(|| artifact.data.as_ref().map(compact_json))
                .unwrap_or_else(|| "No text body persisted".to_string()),
        })
        .collect()
}

fn build_execution_summary(
    execution: Option<&harn_vm::orchestration::RunExecutionRecord>,
) -> Option<PortalExecutionSummary> {
    execution.map(|execution| PortalExecutionSummary {
        cwd: execution.cwd.clone(),
        repo_path: execution.repo_path.clone(),
        worktree_path: execution.worktree_path.clone(),
        branch: execution.branch.clone(),
        adapter: execution.adapter.clone(),
    })
}

pub(super) fn build_policy_summary(run: &harn_vm::orchestration::RunRecord) -> PortalPolicySummary {
    let validation = run.metadata.get("validation").cloned().and_then(|value| {
        serde_json::from_value::<harn_vm::orchestration::WorkflowValidationReport>(value).ok()
    });
    let capabilities = run
        .policy
        .capabilities
        .iter()
        .flat_map(|(capability, ops)| {
            if ops.is_empty() {
                vec![capability.clone()]
            } else {
                ops.iter()
                    .map(|op| format!("{capability}.{op}"))
                    .collect::<Vec<_>>()
            }
        })
        .collect::<Vec<_>>();
    let tool_arg_constraints = run
        .policy
        .tool_arg_constraints
        .iter()
        .map(|constraint| {
            if constraint.arg_patterns.is_empty() {
                constraint.tool.clone()
            } else {
                format!(
                    "{} → {}",
                    constraint.tool,
                    constraint.arg_patterns.join(", ")
                )
            }
        })
        .collect::<Vec<_>>();

    PortalPolicySummary {
        tools: run.policy.tools.clone(),
        capabilities,
        workspace_roots: run.policy.workspace_roots.clone(),
        side_effect_level: run.policy.side_effect_level.clone(),
        recursion_limit: run.policy.recursion_limit,
        tool_arg_constraints,
        validation_valid: validation.as_ref().map(|report| report.valid),
        validation_errors: validation
            .as_ref()
            .map(|report| report.errors.clone())
            .unwrap_or_default(),
        validation_warnings: validation
            .as_ref()
            .map(|report| report.warnings.clone())
            .unwrap_or_default(),
        reachable_nodes: validation
            .as_ref()
            .map(|report| report.reachable_nodes.clone())
            .unwrap_or_default(),
    }
}

pub(super) fn build_replay_summary(
    replay: Option<&harn_vm::orchestration::ReplayFixture>,
) -> Option<PortalReplaySummary> {
    replay.map(|fixture| PortalReplaySummary {
        fixture_id: fixture.id.clone(),
        source_run_id: fixture.source_run_id.clone(),
        created_at: fixture.created_at.clone(),
        expected_status: fixture.expected_status.clone(),
        stage_assertions: fixture
            .stage_assertions
            .iter()
            .map(|stage| PortalReplayAssertion {
                node_id: stage.node_id.clone(),
                expected_status: stage.expected_status.clone(),
                expected_outcome: stage.expected_outcome.clone(),
                expected_branch: stage.expected_branch.clone(),
                required_artifact_kinds: stage.required_artifact_kinds.clone(),
                visible_text_contains: stage.visible_text_contains.clone(),
            })
            .collect(),
    })
}

fn build_stage_debug(stage: &harn_vm::orchestration::RunStageRecord) -> PortalStageDebug {
    let usage = stage.usage.clone().unwrap_or_default();
    PortalStageDebug {
        call_count: usage.call_count,
        input_tokens: usage.input_tokens,
        output_tokens: usage.output_tokens,
        consumed_artifact_ids: stage.consumed_artifact_ids.clone(),
        produced_artifact_ids: stage.produced_artifact_ids.clone(),
        selected_artifact_ids: string_array_value(&stage.metadata, "selected_artifact_ids"),
        worker_id: metadata_string(&stage.metadata, "worker_id"),
        error: metadata_pretty_json(&stage.metadata, "error"),
        model_policy: metadata_pretty_json(&stage.metadata, "model_policy"),
        auto_compact: metadata_pretty_json(&stage.metadata, "auto_compact"),
        output_visibility: metadata_string(&stage.metadata, "output_visibility"),
        context_policy: metadata_pretty_json(&stage.metadata, "context_policy"),
        retry_policy: metadata_pretty_json(&stage.metadata, "retry_policy"),
        capability_policy: metadata_pretty_json(&stage.metadata, "effective_capability_policy"),
        input_contract: metadata_pretty_json(&stage.metadata, "input_contract"),
        output_contract: metadata_pretty_json(&stage.metadata, "output_contract"),
        prompt: metadata_pretty_json(&stage.metadata, "prompt"),
        system_prompt: metadata_pretty_json(&stage.metadata, "system_prompt"),
        rendered_context: metadata_pretty_json(&stage.metadata, "rendered_context"),
    }
}

fn stage_duration_ms(stage: &harn_vm::orchestration::RunStageRecord) -> Option<u64> {
    if let Some(finished) = &stage.finished_at {
        let start = date_ms(&stage.started_at)?;
        let finish = date_ms(finished)?;
        return finish.checked_sub(start);
    }
    stage.usage.as_ref().and_then(|usage| {
        let duration = usage.total_duration_ms.max(0) as u64;
        if duration > 0 {
            Some(duration)
        } else {
            None
        }
    })
}

fn run_duration_ms(run: &harn_vm::orchestration::RunRecord) -> Option<u64> {
    if let Some(usage) = &run.usage {
        let duration = usage.total_duration_ms.max(0) as u64;
        if duration > 0 {
            return Some(duration);
        }
    }
    if let Some(max_end) = run.trace_spans.iter().map(trace_span_end_ms).max() {
        if max_end > 0 {
            return Some(max_end);
        }
    }
    let stage_total = run
        .stages
        .iter()
        .filter_map(stage_duration_ms)
        .fold(0u64, u64::saturating_add);
    if stage_total > 0 {
        return Some(stage_total);
    }
    if let Some(finished) = &run.finished_at {
        let start = date_ms(&run.started_at)?;
        let finish = date_ms(finished)?;
        return finish.checked_sub(start);
    }
    None
}

fn span_depth(
    span: &harn_vm::orchestration::RunTraceSpanRecord,
    by_id: &HashMap<u64, harn_vm::orchestration::RunTraceSpanRecord>,
) -> usize {
    let mut depth = 0usize;
    let mut cursor = span.parent_id.and_then(|id| by_id.get(&id));
    let mut seen = std::collections::HashSet::new();
    while let Some(parent) = cursor {
        if !seen.insert(parent.span_id) {
            break;
        }
        depth += 1;
        cursor = parent.parent_id.and_then(|id| by_id.get(&id));
    }
    depth
}

fn span_label(span: &harn_vm::orchestration::RunTraceSpanRecord) -> String {
    match span.kind.as_str() {
        "llm_call" => span
            .metadata
            .get("model")
            .and_then(|value| value.as_str())
            .map(|model| format!("Model • {model}"))
            .unwrap_or_else(|| "Model call".to_string()),
        "tool_call" => span
            .metadata
            .get("tool_name")
            .and_then(|value| value.as_str())
            .map(|tool| format!("Tool • {tool}"))
            .unwrap_or_else(|| format!("Tool • {}", span.name)),
        _ => span.name.replace('_', " "),
    }
}
