opensymphony 2.9.3

//! Snapshot and control-plane mapping helpers for the runtime CLI.

use std::{
    collections::{HashSet, VecDeque},
    path::Path,
};

use crate::opensymphony_control::{
    AgentServerStatus, ConversationEvent, DaemonSnapshot, DaemonState, DaemonStatus,
    IssueRuntimeState, IssueSnapshot, MemoryServerStatus, MetricsSnapshot, RecentEvent,
    RecentEventKind, WorkerOutcome,
};
use crate::opensymphony_domain::{
    HarnessInterruptStatus, HealthStatus, IssueIdentifier, IssueStateCategory,
    OrchestratorSnapshot, SchedulerStatus, WorkerOutcomeKind,
};
use crate::opensymphony_openhands::LocalServerSupervisor;
use crate::opensymphony_workflow::ResolvedWorkflow;
use chrono::{DateTime, Utc};

use super::timestamp_to_datetime;

const RECENT_EVENT_LIMIT: usize = 24;

pub(super) fn map_snapshot(
    snapshot: &OrchestratorSnapshot,
    workspace_root: &Path,
    terminal_states: &HashSet<String>,
    agent_server: AgentServerStatus,
    memory_server: MemoryServerStatus,
    recent_events: &VecDeque<RecentEvent>,
) -> DaemonSnapshot {
    let generated_at = timestamp_to_datetime(snapshot.generated_at);
    let last_poll_at = snapshot
        .daemon
        .last_poll_at
        .map(timestamp_to_datetime)
        .unwrap_or(generated_at);
    DaemonSnapshot {
        generated_at,
        daemon: DaemonStatus {
            state: map_daemon_state(snapshot.daemon.health),
            last_poll_at,
            workspace_root: workspace_root.display().to_string(),
            status_line: format!(
                "poll={}ms, running={}, retry_queue={}",
                snapshot.daemon.poll_interval_ms,
                snapshot.daemon.running_issue_count,
                snapshot.daemon.retry_queue_count
            ),
        },
        agent_server,
        memory_server,
        metrics: MetricsSnapshot {
            running_issues: snapshot.daemon.running_issue_count as u32,
            retry_queue_depth: snapshot.daemon.retry_queue_count as u32,
            input_tokens: snapshot.daemon.usage.input_tokens,
            output_tokens: snapshot.daemon.usage.output_tokens,
            cache_read_tokens: snapshot.daemon.usage.cache_read_tokens,
            total_tokens: snapshot.daemon.usage.total_tokens,
            total_cost_micros: snapshot.daemon.usage.estimated_cost_usd_micros.unwrap_or(0),
        },
        issues: snapshot
            .issues
            .iter()
            .map(|issue| map_issue(issue, terminal_states, generated_at))
            .collect(),
        recent_events: recent_events.iter().cloned().collect(),
    }
}

pub(super) fn current_memory_server_status(
    memory_server: Option<&super::super::memory::MemoryServerHandle>,
) -> MemoryServerStatus {
    let Some(memory_server) = memory_server else {
        return MemoryServerStatus::default();
    };
    let reachable = !memory_server.is_finished();
    MemoryServerStatus {
        enabled: true,
        reachable,
        endpoint: Some(memory_server.endpoint().to_string()),
        status_line: if reachable {
            "listening".to_string()
        } else {
            "stopped".to_string()
        },
    }
}

fn map_issue(
    issue: &crate::opensymphony_domain::IssueSnapshot,
    terminal_states: &HashSet<String>,
    generated_at: DateTime<Utc>,
) -> IssueSnapshot {
    let runtime_state = match issue.runtime.state {
        SchedulerStatus::Running | SchedulerStatus::Claimed => IssueRuntimeState::Running,
        SchedulerStatus::RetryQueued => IssueRuntimeState::RetryQueued,
        SchedulerStatus::Released => match issue
            .last_worker_outcome
            .as_ref()
            .map(|outcome| outcome.outcome)
        {
            Some(
                WorkerOutcomeKind::Failed
                | WorkerOutcomeKind::TimedOut
                | WorkerOutcomeKind::Stalled,
            ) => IssueRuntimeState::Failed,
            // A released execution with no recorded run never dispatched a
            // worker (e.g. a recovered workspace parked while its issue sits
            // in a non-active tracker state). Key on the *current* tracker
            // state, not the release reason: `release_issue` refreshes the
            // issue state but leaves a stale `TrackerInactive` reason when the
            // execution is already released, so a parked issue later moved
            // straight to a terminal state (Backlog → Done) must still surface
            // as a completion rather than a phantom idle row.
            None if issue.issue.state.category != IssueStateCategory::Terminal => {
                IssueRuntimeState::Idle
            }
            _ => IssueRuntimeState::Completed,
        },
        SchedulerStatus::Unclaimed => IssueRuntimeState::Idle,
    };
    let last_outcome = map_worker_outcome(issue, runtime_state);
    let last_event_at = issue
        .conversation
        .as_ref()
        .and_then(|conversation| conversation.last_event_at)
        .map(timestamp_to_datetime)
        .or_else(|| {
            issue
                .last_worker_outcome
                .as_ref()
                .map(|outcome| timestamp_to_datetime(outcome.finished_at))
        })
        .unwrap_or(generated_at);
    let worker = issue.runtime.worker.as_ref();
    let last_worker_outcome = issue.last_worker_outcome.as_ref();
    let interrupt = issue.runtime.interrupt.as_ref();
    let started_at = issue
        .runtime
        .started_at
        .or_else(|| last_worker_outcome.map(|outcome| outcome.started_at))
        .map(timestamp_to_datetime);
    let finished_at = issue
        .runtime
        .released_at
        .or_else(|| last_worker_outcome.map(|outcome| outcome.finished_at))
        .map(timestamp_to_datetime);
    let runtime_seconds = issue
        .conversation
        .as_ref()
        .map(|conversation| conversation.runtime_seconds)
        .unwrap_or(0)
        .max(runtime_seconds_from_timestamps(
            started_at,
            finished_at,
            generated_at,
            runtime_state,
        ));

    IssueSnapshot {
        identifier: issue.issue.identifier.to_string(),
        title: issue.issue.title.clone(),
        tracker_state: issue.issue.state.name.clone(),
        runtime_state,
        last_outcome,
        last_event_at,
        conversation_id_suffix: issue
            .conversation
            .as_ref()
            .map(|conversation| suffix(conversation.conversation_id.as_str()))
            .unwrap_or_else(|| "-".to_string()),
        // For Codex runs the conversation id *is* the Codex thread id, so
        // carry it in full (not the display suffix) for the debug deep link.
        // Match the same Codex detection the manifest recovery uses (transport
        // target or the Codex runtime contract) so legacy/recovered manifests
        // that only recorded the transport target still expose their thread
        // id — but exclude `--dry-run` route previews, whose synthetic
        // `route-preview-*` conversation id resumes nothing.
        codex_thread_id: issue.conversation.as_ref().and_then(|conversation| {
            let is_codex = conversation.transport_target.as_deref()
                == Some(crate::opensymphony_codex::CODEX_APP_SERVER_KIND)
                || conversation.runtime_contract_version.as_deref()
                    == Some(crate::opensymphony_codex::CODEX_APP_SERVER_CONTRACT);
            let is_route_preview = conversation
                .conversation_id
                .as_str()
                .starts_with(super::backends::ROUTE_PREVIEW_CONVERSATION_PREFIX);
            (is_codex && !is_route_preview)
                .then(|| conversation.conversation_id.as_str().to_string())
        }),
        workspace_path_suffix: issue
            .workspace
            .as_ref()
            .map(|workspace| suffix_path(&workspace.path))
            .unwrap_or_else(|| "-".to_string()),
        branch_name: issue.issue.branch_name.clone(),
        pr_url: issue.issue.pr_url.clone(),
        project_id: issue.issue.project_id.clone(),
        project_slug: issue.issue.project_slug.clone(),
        project_name: issue.issue.project_name.clone(),
        workspace_label: issue
            .workspace
            .as_ref()
            .map(|workspace| suffix_path(&workspace.path))
            .filter(|label| label != "-"),
        retry_count: issue
            .retry
            .as_ref()
            .map(|retry| retry.normal_retry_count)
            .unwrap_or(0),
        claimed_at: issue.runtime.claimed_at.map(timestamp_to_datetime),
        started_at,
        finished_at,
        turn_count: worker
            .map(|worker| worker.turn_count)
            .or_else(|| last_worker_outcome.map(|outcome| outcome.turn_count))
            .unwrap_or(0),
        max_turns: worker.map(|worker| worker.max_turns).unwrap_or(0),
        runtime_seconds,
        blocked: issue.issue.blocked_by.iter().any(|blocker| {
            blocker
                .state
                .as_deref()
                .is_none_or(|state| !is_terminal_state(terminal_states, state))
        }) || (!issue.issue.sub_issues.is_empty()
            && issue
                .issue
                .sub_issues
                .iter()
                .any(|sub_issue| !is_terminal_state(terminal_states, &sub_issue.state))),
        blocked_by: issue
            .issue
            .blocked_by
            .iter()
            .filter_map(|blocker| blocker.identifier.as_ref())
            .map(ToString::to_string)
            .collect(),
        server_base_url: issue
            .conversation
            .as_ref()
            .and_then(|conversation| conversation.server_base_url.clone()),
        transport_target: issue
            .conversation
            .as_ref()
            .and_then(|conversation| conversation.transport_target.clone()),
        http_auth_mode: issue
            .conversation
            .as_ref()
            .and_then(|conversation| conversation.http_auth_mode.clone()),
        websocket_auth_mode: issue
            .conversation
            .as_ref()
            .and_then(|conversation| conversation.websocket_auth_mode.clone()),
        websocket_query_param_name: issue
            .conversation
            .as_ref()
            .and_then(|conversation| conversation.websocket_query_param_name.clone()),
        recent_events: issue
            .conversation
            .as_ref()
            .map(|conversation| {
                conversation
                    .recent_activity
                    .iter()
                    .rev()
                    .map(|activity| ConversationEvent {
                        event_id: activity.event_id.clone(),
                        happened_at: timestamp_to_datetime(activity.happened_at),
                        kind: activity.kind.clone(),
                        summary: activity.summary.clone(),
                        payload: activity.payload.clone(),
                        sequence: activity.sequence,
                    })
                    .collect()
            })
            .unwrap_or_default(),
        modified_files: Vec::new(),
        input_tokens: issue
            .conversation
            .as_ref()
            .map(|conversation| conversation.input_tokens)
            .unwrap_or(0),
        output_tokens: issue
            .conversation
            .as_ref()
            .map(|conversation| conversation.output_tokens)
            .unwrap_or(0),
        cache_read_tokens: issue
            .conversation
            .as_ref()
            .map(|conversation| conversation.cache_read_tokens)
            .unwrap_or(0),
        total_tokens: issue
            .conversation
            .as_ref()
            .map(|conversation| conversation.effective_total_tokens())
            .unwrap_or(0),
        detached: false,
        cancel_requested: matches!(
            interrupt.map(|interrupt| interrupt.status),
            Some(HarnessInterruptStatus::Requested)
        ),
        cancel_acknowledged: matches!(
            interrupt.map(|interrupt| interrupt.status),
            Some(HarnessInterruptStatus::Acknowledged)
        ),
        cancel_failed: matches!(
            interrupt.map(|interrupt| interrupt.status),
            Some(HarnessInterruptStatus::Failed)
        ),
        cancel_timed_out: matches!(
            interrupt.map(|interrupt| interrupt.status),
            Some(HarnessInterruptStatus::TimedOut)
        ),
        cancel_reason: interrupt.map(|interrupt| interrupt.command.reason.as_str().to_string()),
    }
}

fn runtime_seconds_from_timestamps(
    started_at: Option<DateTime<Utc>>,
    finished_at: Option<DateTime<Utc>>,
    generated_at: DateTime<Utc>,
    runtime_state: IssueRuntimeState,
) -> u64 {
    let Some(started_at) = started_at else {
        return 0;
    };
    let end = match runtime_state {
        IssueRuntimeState::Running | IssueRuntimeState::Releasing => generated_at,
        IssueRuntimeState::Completed | IssueRuntimeState::Failed => match finished_at {
            Some(finished_at) => finished_at,
            None => return 0,
        },
        _ => return 0,
    };
    elapsed_seconds(started_at, end)
}

fn elapsed_seconds(started_at: DateTime<Utc>, ended_at: DateTime<Utc>) -> u64 {
    ended_at
        .signed_duration_since(started_at)
        .num_seconds()
        .max(0) as u64
}

fn map_worker_outcome(
    issue: &crate::opensymphony_domain::IssueSnapshot,
    runtime_state: IssueRuntimeState,
) -> WorkerOutcome {
    match runtime_state {
        IssueRuntimeState::Running => WorkerOutcome::Running,
        IssueRuntimeState::Paused => WorkerOutcome::Unknown,
        IssueRuntimeState::RetryQueued => match issue
            .last_worker_outcome
            .as_ref()
            .map(|outcome| outcome.outcome)
        {
            Some(WorkerOutcomeKind::Succeeded) => WorkerOutcome::Continued,
            Some(WorkerOutcomeKind::Cancelled) => WorkerOutcome::Canceled,
            Some(
                WorkerOutcomeKind::Failed
                | WorkerOutcomeKind::TimedOut
                | WorkerOutcomeKind::Stalled
                | WorkerOutcomeKind::Detached
                | WorkerOutcomeKind::CancelFailed,
            ) => WorkerOutcome::Failed,
            None => WorkerOutcome::Continued,
        },
        IssueRuntimeState::Completed => match issue
            .last_worker_outcome
            .as_ref()
            .map(|outcome| outcome.outcome)
        {
            Some(WorkerOutcomeKind::Cancelled) => WorkerOutcome::Canceled,
            Some(
                WorkerOutcomeKind::Failed
                | WorkerOutcomeKind::TimedOut
                | WorkerOutcomeKind::Stalled
                | WorkerOutcomeKind::Detached
                | WorkerOutcomeKind::CancelFailed,
            ) => WorkerOutcome::Failed,
            _ => WorkerOutcome::Completed,
        },
        IssueRuntimeState::Failed => WorkerOutcome::Failed,
        IssueRuntimeState::Idle => WorkerOutcome::Unknown,
        IssueRuntimeState::Releasing => WorkerOutcome::Unknown,
    }
}

pub(super) fn current_agent_server_status(
    supervisor: &mut Option<LocalServerSupervisor>,
    base_url: &str,
) -> AgentServerStatus {
    if let Some(supervisor) = supervisor.as_mut()
        && let Ok(status) = supervisor.status()
    {
        return AgentServerStatus {
            reachable: matches!(
                status.state,
                crate::opensymphony_openhands::ServerState::Ready
            ),
            base_url: status.base_url,
            conversation_count: 0,
            status_line: format!("{:?}", status.state).to_ascii_lowercase(),
        };
    }

    AgentServerStatus {
        reachable: true,
        base_url: base_url.to_string(),
        conversation_count: 0,
        status_line: "reachable".to_string(),
    }
}

pub(super) fn push_recent_event(
    recent_events: &mut VecDeque<RecentEvent>,
    kind: RecentEventKind,
    issue_identifier: Option<IssueIdentifier>,
    summary: String,
    happened_at: DateTime<Utc>,
) {
    recent_events.push_front(RecentEvent {
        happened_at,
        issue_identifier: issue_identifier.map(|identifier| identifier.to_string()),
        kind,
        summary,
    });
    while recent_events.len() > RECENT_EVENT_LIMIT {
        let _ = recent_events.pop_back();
    }
}

pub(super) fn terminal_state_set(workflow: &ResolvedWorkflow) -> HashSet<String> {
    workflow
        .config
        .tracker
        .terminal_states
        .iter()
        .map(|state| state.trim().to_ascii_lowercase())
        .collect()
}

fn is_terminal_state(terminal_states: &HashSet<String>, state: &str) -> bool {
    terminal_states.contains(&state.trim().to_ascii_lowercase())
}

fn map_daemon_state(health: HealthStatus) -> DaemonState {
    match health {
        HealthStatus::Unknown | HealthStatus::Starting => DaemonState::Starting,
        HealthStatus::Healthy => DaemonState::Ready,
        HealthStatus::Degraded | HealthStatus::Failed => DaemonState::Degraded,
    }
}

fn suffix(value: &str) -> String {
    if value.len() <= 8 {
        value.to_string()
    } else {
        value[value.len() - 8..].to_string()
    }
}

fn suffix_path(path: &Path) -> String {
    path.file_name()
        .map(|name| name.to_string_lossy().into_owned())
        .unwrap_or_else(|| path.display().to_string())
}

#[cfg(test)]
mod tests {
    use std::{collections::BTreeMap, path::PathBuf};

    use crate::opensymphony_domain::{
        BlockerRef, ComponentHealthSnapshot, ConversationId, ConversationMetadata, DaemonSnapshot,
        HealthStatus, IssueId, IssueIdentifier, IssueRef, IssueSnapshot as DomainIssueSnapshot,
        IssueState, IssueStateCategory, NormalizedIssue, OrchestratorSnapshot,
        RuntimeStateSnapshot, RuntimeStreamState, RuntimeUsageTotals, SchedulerStatus, TimestampMs,
        WorkerAttemptSnapshot, WorkerId, WorkspaceKey, WorkspaceRecord,
    };
    use serde_json::json;

    use super::{map_snapshot, terminal_state_set, timestamp_to_datetime};

    fn must<T, E: std::fmt::Display>(result: Result<T, E>) -> T {
        match result {
            Ok(value) => value,
            Err(error) => panic!("{error}"),
        }
    }

    fn ts(value: u64) -> TimestampMs {
        TimestampMs::new(value)
    }

    fn resolved_workflow_for_tests() -> crate::opensymphony_workflow::ResolvedWorkflow {
        let workflow = crate::opensymphony_workflow::WorkflowDefinition::parse(
            r#"---
tracker:
  kind: linear
  project_slug: sample-project
  active_states:
    - In Progress
  terminal_states:
    - Done
---
{{ issue.identifier }}
"#,
        )
        .expect("workflow should parse");

        workflow
            .resolve(
                std::path::Path::new("/tmp"),
                &BTreeMap::from([("LINEAR_API_KEY".to_owned(), "linear-token".to_owned())]),
            )
            .expect("workflow should resolve")
    }

    #[test]
    fn map_snapshot_preserves_recent_events_and_run_metrics() {
        let recent_activity = (0..12)
            .map(
                |index| crate::opensymphony_domain::ConversationActivityEvent {
                    event_id: format!("evt-{index}"),
                    happened_at: ts(1_000 + index),
                    kind: "ActionEvent".to_owned(),
                    summary: format!("summary {index}"),
                    payload: (index == 0).then(|| json!({"command": "npm test"})),
                    sequence: index,
                },
            )
            .collect();

        let snapshot = OrchestratorSnapshot::new(
            ts(2_000),
            DaemonSnapshot::new(
                HealthStatus::Healthy,
                1_000,
                4,
                Some(ts(2_000)),
                ComponentHealthSnapshot::default(),
                RuntimeUsageTotals::default(),
            ),
            vec![DomainIssueSnapshot {
                issue: NormalizedIssue {
                    id: must(IssueId::new("lin_352")),
                    identifier: must(IssueIdentifier::new("COE-352")),
                    title: "Render media pipeline".to_owned(),
                    description: None,
                    priority: None,
                    state: IssueState {
                        id: None,
                        name: "In Progress".to_owned(),
                        category: IssueStateCategory::Active,
                    },
                    branch_name: None,
                    pr_url: None,
                    url: None,
                    labels: Vec::new(),
                    project_id: Some("proj-open".to_owned()),
                    project_slug: Some("opensymphony-bootstrap".to_owned()),
                    project_name: Some("OpenSymphony".to_owned()),
                    parent_id: None,
                    blocked_by: Vec::<BlockerRef>::new(),
                    sub_issues: Vec::<IssueRef>::new(),
                    created_at: None,
                    updated_at: None,
                },
                runtime: RuntimeStateSnapshot {
                    state: SchedulerStatus::Running,
                    claimed_at: Some(ts(900)),
                    started_at: Some(ts(1_000)),
                    released_at: None,
                    release_reason: None,
                    worker: Some(WorkerAttemptSnapshot {
                        worker_id: must(WorkerId::new("worker-352")),
                        attempt: None,
                        normal_retry_count: 0,
                        turn_count: 3,
                        max_turns: 8,
                    }),
                    last_event_at: Some(ts(1_011)),
                    stalled_at: None,
                    interrupt: None,
                },
                workspace: Some(WorkspaceRecord {
                    path: PathBuf::from("/tmp/workspaces/COE-352"),
                    workspace_key: must(WorkspaceKey::new("COE-352")),
                    created_now: false,
                    created_at: None,
                    updated_at: None,
                    last_seen_tracker_refresh_at: None,
                }),
                conversation: Some(ConversationMetadata {
                    conversation_id: must(ConversationId::new("conv_352")),
                    server_base_url: Some("http://127.0.0.1:3000".to_owned()),
                    transport_target: Some("loopback".to_owned()),
                    http_auth_mode: Some("none".to_owned()),
                    websocket_auth_mode: Some("none".to_owned()),
                    websocket_query_param_name: None,
                    fresh_conversation: false,
                    runtime_contract_version: Some("openhands-sdk-agent-server-v1".to_owned()),
                    stream_state: RuntimeStreamState::Ready,
                    last_event_id: Some("evt-11".to_owned()),
                    last_event_kind: Some("ActionEvent".to_owned()),
                    last_event_at: Some(ts(1_011)),
                    last_event_summary: Some("summary 11".to_owned()),
                    recent_activity,
                    input_tokens: 0,
                    output_tokens: 0,
                    cache_read_tokens: 0,
                    total_tokens: 0,
                    runtime_seconds: 0,
                    next_activity_sequence: 0,
                }),
                retry: None,
                last_worker_outcome: None,
                recent_worker_outcomes: Vec::new(),
            }],
        );

        let mapped = map_snapshot(
            &snapshot,
            PathBuf::from("/tmp/workspaces").as_path(),
            &terminal_state_set(&resolved_workflow_for_tests()),
            crate::opensymphony_control::AgentServerStatus {
                reachable: true,
                base_url: "http://127.0.0.1:3000".to_owned(),
                conversation_count: 1,
                status_line: "healthy".to_owned(),
            },
            crate::opensymphony_control::MemoryServerStatus::default(),
            &std::collections::VecDeque::new(),
        );

        assert_eq!(mapped.issues[0].recent_events.len(), 12);
        assert_eq!(mapped.issues[0].recent_events[0].summary, "summary 11");
        assert_eq!(mapped.issues[0].recent_events[11].summary, "summary 0");
        assert_eq!(
            mapped.issues[0].recent_events[11].payload,
            Some(json!({"command": "npm test"}))
        );
        assert_eq!(
            mapped.issues[0].claimed_at,
            Some(timestamp_to_datetime(ts(900)))
        );
        assert_eq!(
            mapped.issues[0].started_at,
            Some(timestamp_to_datetime(ts(1_000)))
        );
        assert_eq!(mapped.issues[0].turn_count, 3);
        assert_eq!(mapped.issues[0].max_turns, 8);
        assert_eq!(mapped.issues[0].runtime_seconds, 1);
        assert_eq!(mapped.issues[0].project_id.as_deref(), Some("proj-open"));
        assert_eq!(
            mapped.issues[0].project_slug.as_deref(),
            Some("opensymphony-bootstrap")
        );
        assert_eq!(
            mapped.issues[0].project_name.as_deref(),
            Some("OpenSymphony")
        );
        assert_eq!(mapped.issues[0].workspace_label.as_deref(), Some("COE-352"));
    }

    fn released_issue_snapshot(
        state_name: &str,
        category: IssueStateCategory,
        release_reason: crate::opensymphony_domain::ReleaseReason,
    ) -> DomainIssueSnapshot {
        DomainIssueSnapshot {
            issue: NormalizedIssue {
                id: must(IssueId::new("lin_532")),
                identifier: must(IssueIdentifier::new("COE-532")),
                title: "Symbol identity container chain".to_owned(),
                description: None,
                priority: None,
                state: IssueState {
                    id: None,
                    name: state_name.to_owned(),
                    category,
                },
                branch_name: None,
                pr_url: None,
                url: None,
                labels: Vec::new(),
                project_id: None,
                project_slug: None,
                project_name: None,
                parent_id: None,
                blocked_by: Vec::<BlockerRef>::new(),
                sub_issues: Vec::<IssueRef>::new(),
                created_at: None,
                updated_at: None,
            },
            runtime: RuntimeStateSnapshot {
                state: SchedulerStatus::Released,
                claimed_at: None,
                started_at: None,
                released_at: Some(ts(1_500)),
                release_reason: Some(release_reason),
                worker: None,
                last_event_at: None,
                stalled_at: None,
                interrupt: None,
            },
            workspace: None,
            conversation: None,
            retry: None,
            last_worker_outcome: None,
            recent_worker_outcomes: Vec::new(),
        }
    }

    fn map_single_issue(issue: DomainIssueSnapshot) -> crate::opensymphony_control::IssueSnapshot {
        let snapshot = OrchestratorSnapshot::new(
            ts(2_000),
            DaemonSnapshot::new(
                HealthStatus::Healthy,
                1_000,
                4,
                Some(ts(2_000)),
                ComponentHealthSnapshot::default(),
                RuntimeUsageTotals::default(),
            ),
            vec![issue],
        );
        let mut mapped = map_snapshot(
            &snapshot,
            PathBuf::from("/tmp/workspaces").as_path(),
            &terminal_state_set(&resolved_workflow_for_tests()),
            crate::opensymphony_control::AgentServerStatus {
                reachable: true,
                base_url: "http://127.0.0.1:3000".to_owned(),
                conversation_count: 0,
                status_line: "healthy".to_owned(),
            },
            crate::opensymphony_control::MemoryServerStatus::default(),
            &std::collections::VecDeque::new(),
        );
        mapped.issues.remove(0)
    }

    fn map_single_issue_runtime_state(
        issue: DomainIssueSnapshot,
    ) -> crate::opensymphony_control::IssueRuntimeState {
        map_single_issue(issue).runtime_state
    }

    fn codex_conversation(thread_id: &str) -> ConversationMetadata {
        ConversationMetadata {
            conversation_id: must(ConversationId::new(thread_id.to_owned())),
            server_base_url: None,
            transport_target: Some("codex_app_server".to_owned()),
            http_auth_mode: None,
            websocket_auth_mode: None,
            websocket_query_param_name: None,
            fresh_conversation: false,
            runtime_contract_version: Some("codex-app-server-json-rpc-v2".to_owned()),
            stream_state: RuntimeStreamState::Ready,
            last_event_id: None,
            last_event_kind: None,
            last_event_at: None,
            last_event_summary: None,
            recent_activity: Vec::new(),
            input_tokens: 0,
            output_tokens: 0,
            cache_read_tokens: 0,
            total_tokens: 0,
            runtime_seconds: 0,
            next_activity_sequence: 0,
        }
    }

    fn running_issue_with_conversation(conversation: ConversationMetadata) -> DomainIssueSnapshot {
        let mut issue = released_issue_snapshot(
            "In Progress",
            IssueStateCategory::Active,
            crate::opensymphony_domain::ReleaseReason::Completed,
        );
        issue.conversation = Some(conversation);
        issue
    }

    #[test]
    fn codex_conversation_records_full_thread_id() {
        // For a Codex run the conversation id is the Codex thread id: the
        // snapshot must carry it in full (not the display suffix) so the
        // desktop can open codex://threads/<id>.
        let issue = map_single_issue(running_issue_with_conversation(codex_conversation(
            "019f3979-3aa3-71f3-86b1-18e92c71fbc9",
        )));
        assert_eq!(
            issue.codex_thread_id.as_deref(),
            Some("019f3979-3aa3-71f3-86b1-18e92c71fbc9")
        );
    }

    #[test]
    fn openhands_conversation_records_no_codex_thread_id() {
        let mut conversation = codex_conversation("ignored");
        conversation.transport_target = Some("loopback".to_owned());
        conversation.runtime_contract_version = Some("openhands-sdk-agent-server-v1".to_owned());
        let issue = map_single_issue(running_issue_with_conversation(conversation));
        assert_eq!(issue.codex_thread_id, None);
    }

    #[test]
    fn dry_run_route_preview_records_no_codex_thread_id() {
        // A `--dry-run` route preview to Codex targets the codex_app_server
        // transport but carries the routing contract and a synthetic
        // `route-preview-*` id that resumes nothing — it must not surface as
        // a Codex thread.
        let mut conversation = codex_conversation("route-preview-worker-1");
        conversation.transport_target =
            Some(crate::opensymphony_codex::CODEX_APP_SERVER_KIND.to_owned());
        conversation.runtime_contract_version = Some("opensymphony-routing-alpha-v1".to_owned());
        let issue = map_single_issue(running_issue_with_conversation(conversation));
        assert_eq!(issue.codex_thread_id, None);
    }

    #[test]
    fn legacy_codex_manifest_without_contract_still_records_thread_id() {
        // A run recovered from an older manifest may record only the Codex
        // transport target (no runtime contract). It is still a real Codex
        // thread, so its full id must survive for the debug deep link.
        let mut conversation = codex_conversation("019f3979-3aa3-71f3-86b1-18e92c71fbc9");
        conversation.transport_target =
            Some(crate::opensymphony_codex::CODEX_APP_SERVER_KIND.to_owned());
        conversation.runtime_contract_version = None;
        let issue = map_single_issue(running_issue_with_conversation(conversation));
        assert_eq!(
            issue.codex_thread_id.as_deref(),
            Some("019f3979-3aa3-71f3-86b1-18e92c71fbc9")
        );
    }

    #[test]
    fn tracker_inactive_release_without_a_run_maps_to_idle_not_completed() {
        // A recovered workspace whose issue sits in Backlog is parked, not
        // completed — Completed here surfaced phantom rows in the desktop
        // Completed pane and blocked the Current pane from showing the issue
        // after it moved back to an active state.
        let state = map_single_issue_runtime_state(released_issue_snapshot(
            "Backlog",
            IssueStateCategory::NonActive,
            crate::opensymphony_domain::ReleaseReason::TrackerInactive,
        ));
        assert_eq!(state, crate::opensymphony_control::IssueRuntimeState::Idle);
    }

    #[test]
    fn tracker_terminal_release_without_a_run_still_maps_to_completed() {
        // An issue moved straight to Done in the tracker was completed
        // externally; the tracker state is authoritative.
        let state = map_single_issue_runtime_state(released_issue_snapshot(
            "Done",
            IssueStateCategory::Terminal,
            crate::opensymphony_domain::ReleaseReason::TrackerTerminal,
        ));
        assert_eq!(
            state,
            crate::opensymphony_control::IssueRuntimeState::Completed
        );
    }

    #[test]
    fn stale_tracker_inactive_reason_with_terminal_state_still_maps_to_completed() {
        // A parked issue (released `TrackerInactive`) moved straight to Done
        // keeps its stale release reason because `release_issue` does not
        // re-release an already-released execution. The current terminal
        // tracker state must still win so the completion reaches the
        // Completed pane and dashboard instead of vanishing as idle.
        let state = map_single_issue_runtime_state(released_issue_snapshot(
            "Done",
            IssueStateCategory::Terminal,
            crate::opensymphony_domain::ReleaseReason::TrackerInactive,
        ));
        assert_eq!(
            state,
            crate::opensymphony_control::IssueRuntimeState::Completed
        );
    }
}