polyc-tui 2026.9.0

//! Fleet data adapter: lists/watches the kube `Conversation` CRD and projects
//! each object into a [`ConversationRow`].
//!
//! Authoritative lifecycle/status comes from
//! `polyc_controller::{Conversation, ConversationStatus}`. The CRD is
//! namespaced (watches are intentionally NOT `Api::all`), so a namespace is
//! always required. Forensics metrics are layered on by the caller via
//! [`enrich_row`].

use anyhow::Result;
use futures::{Stream, StreamExt};
use kube::Client;
use kube::api::{Api, ListParams};
use kube::runtime::{watcher, watcher::Config as WatcherConfig};
use polyc_controller::Conversation;
use serde::Deserialize;

use crate::components::fleet::ConversationRow;

/// Build a kube client from the ambient kubeconfig / in-cluster service account.
///
/// # Errors
/// Returns an error if no usable kube configuration is found.
pub(crate) async fn build_client() -> Result<Client> {
    Ok(Client::try_default().await?)
}

/// List all `Conversation` CRs in `namespace` and project them to rows.
///
/// # Errors
/// Returns an error if the list call fails.
pub(crate) async fn load_fleet(client: &Client, namespace: &str) -> Result<Vec<ConversationRow>> {
    let api: Api<Conversation> = Api::namespaced(client.clone(), namespace);
    let params = ListParams::default();
    let list = api.list(&params).await?;
    Ok(list.items.iter().map(row_from).collect())
}

/// Watch `Conversation` CRs in `namespace`, yielding one row per change.
///
/// Drives `kube::runtime::watcher`; `Apply`/`InitApply` objects are projected
/// via [`row_from`] and surfaced as `Ok` rows. Bookkeeping events (`Init`,
/// `InitDone`, `Delete`) produce no row and are filtered out. The caller
/// forwards each row as `Action::FleetDelta`.
///
/// # Errors
/// The outer `Result` is always `Ok` here (the watcher opens lazily); per-item
/// watch errors are surfaced inline as `Err` stream items.
// `async` is part of the frozen adapter contract (callers `.await` it,
// symmetric with `load_fleet`); the body opens the watcher lazily.
#[allow(clippy::unused_async)]
pub(crate) async fn watch_fleet(
    client: &Client,
    namespace: &str,
) -> Result<impl Stream<Item = Result<ConversationRow>>> {
    let api: Api<Conversation> = Api::namespaced(client.clone(), namespace);
    let stream = watcher(api, WatcherConfig::default()).filter_map(|event| async move {
        match event {
            Ok(watcher::Event::Apply(obj) | watcher::Event::InitApply(obj)) => {
                Some(Ok(row_from(&obj)))
            }
            // Lifecycle bookkeeping and deletes carry no row to render here.
            Ok(watcher::Event::Init | watcher::Event::InitDone | watcher::Event::Delete(_)) => None,
            Err(e) => Some(Err(anyhow::Error::new(e))),
        }
    });
    Ok(stream)
}

/// Project a single `Conversation` CR into a [`ConversationRow`].
///
/// Reads `spec.model`/`spec.principal_ref` and the `status` (`phase/pod_ip`/
/// `harness_ready/closed`). Forensics metric fields are left `None` here and
/// filled by the enrichment poll.
#[must_use]
pub(crate) fn row_from(conv: &Conversation) -> ConversationRow {
    let id = conv.metadata.name.clone().unwrap_or_default();
    let status = conv.status.as_ref();
    ConversationRow {
        id,
        model: conv.spec.model.clone(),
        principal_ref: conv.spec.principal_ref.clone(),
        phase: status.and_then(|s| s.phase.clone()),
        pod_ip: status.and_then(|s| s.pod_ip.clone()),
        harness_ready: status.is_some_and(|s| s.harness_ready),
        closed: status.is_some_and(|s| s.closed),
        committed_turns: None,
        input_tokens: None,
        output_tokens: None,
        pending_approvals: None,
    }
}

// ---------------------------------------------------------------------------
// Forensics enrichment (optional, off-by-default server)
// ---------------------------------------------------------------------------

/// Serde mirror of the forensics `GET /api/conversations/:id` overview body.
#[derive(Debug, Clone, Deserialize)]
pub(crate) struct OverviewResponse {
    /// Number of fully-committed turns.
    pub committed_turns: usize,
    /// Running token totals summed across the journal.
    pub usage: UsageJson,
}

/// Serde mirror of the forensics usage totals block.
#[derive(Debug, Clone, Default, Deserialize)]
pub(crate) struct UsageJson {
    /// Total prompt-side tokens.
    pub input_tokens: u64,
    /// Total completion-side tokens.
    pub output_tokens: u64,
}

/// Serde mirror of the forensics `GET /api/conversations/:id/approvals` body.
#[derive(Debug, Clone, Deserialize)]
pub(crate) struct ApprovalsCountResponse {
    /// Paired approval entries in journal order.
    pub approvals: Vec<ApprovalCountEntry>,
}

/// Minimal serde mirror of one approval entry — enough to count pending ones.
#[derive(Debug, Clone, Deserialize)]
pub(crate) struct ApprovalCountEntry {
    /// The matched response, if one exists in the journal. `None` => pending.
    pub response: Option<serde_json::Value>,
}

/// Poll the forensics server for a single conversation's metric enrichment and
/// fold it into `row` in place. Best-effort: any failure leaves metric fields
/// as-is (they stay `None`) and is reported via the returned `Result` so the
/// caller can choose to ignore or log it.
///
/// # Errors
/// Returns an error if any HTTP request fails or a body cannot be decoded.
pub(crate) async fn enrich_row(forensics_base_url: &str, row: &mut ConversationRow) -> Result<()> {
    let metrics = fetch_metrics(forensics_base_url, &row.id).await?;
    row.committed_turns = Some(metrics.committed_turns);
    row.input_tokens = Some(metrics.input_tokens);
    row.output_tokens = Some(metrics.output_tokens);
    row.pending_approvals = Some(metrics.pending_approvals);
    Ok(())
}

/// Forensics-derived metrics for one conversation.
#[derive(Debug, Clone, Default)]
pub(crate) struct ConversationMetrics {
    /// Number of fully-committed turns.
    pub committed_turns: usize,
    /// Total prompt-side tokens.
    pub input_tokens: u64,
    /// Total completion-side tokens.
    pub output_tokens: u64,
    /// Count of approvals with no recorded response (still pending).
    pub pending_approvals: usize,
}

/// Fetch the overview + approvals projections for `conversation_id` from the
/// forensics server and reduce them to a [`ConversationMetrics`].
///
/// # Errors
/// Returns an error if either HTTP request fails or a body cannot be decoded.
pub(crate) async fn fetch_metrics(
    forensics_base_url: &str,
    conversation_id: &str,
) -> Result<ConversationMetrics> {
    let client = crate::data::http_client();
    let base = forensics_base_url.trim_end_matches('/');

    let overview: OverviewResponse = client
        .get(format!("{base}/api/conversations/{conversation_id}"))
        .send()
        .await?
        .error_for_status()?
        .json()
        .await?;

    let approvals: ApprovalsCountResponse = client
        .get(format!(
            "{base}/api/conversations/{conversation_id}/approvals"
        ))
        .send()
        .await?
        .error_for_status()?
        .json()
        .await?;

    let pending_approvals = approvals
        .approvals
        .iter()
        .filter(|a| a.response.is_none())
        .count();

    Ok(ConversationMetrics {
        committed_turns: overview.committed_turns,
        input_tokens: overview.usage.input_tokens,
        output_tokens: overview.usage.output_tokens,
        pending_approvals,
    })
}

/// Serde mirror of the forensics `GET /api/conversations` index.
#[derive(Debug, Clone, Deserialize)]
pub(crate) struct ConversationListResponse {
    /// Every conversation id the event log knows about.
    pub conversation_ids: Vec<String>,
}

/// List every conversation id recorded in the forensics event log.
///
/// In shared-harness mode the control plane serves turns WITHOUT creating a
/// kube `Conversation` CR, so the event log is the only fleet-visible record of
/// that activity. The caller merges these ids with the kube-sourced rows so
/// such conversations still appear in the fleet (see [`forensics_row`]).
///
/// # Errors
/// Returns an error if the request fails or the body cannot be decoded.
pub(crate) async fn list_conversations(forensics_base_url: &str) -> Result<Vec<String>> {
    let base = forensics_base_url.trim_end_matches('/');
    let resp: ConversationListResponse = crate::data::http_client()
        .get(format!("{base}/api/conversations"))
        .send()
        .await?
        .error_for_status()?
        .json()
        .await?;
    Ok(resp.conversation_ids)
}

/// Build a minimal [`ConversationRow`] for a conversation that exists only in
/// the forensics event log (no kube `Conversation` CR backing it).
///
/// Lifecycle fields (`pod_ip`/`harness_ready`/`closed`) are kube concepts and
/// stay empty; `phase` is marked `event-log` so the row is visibly distinct
/// from a controller-managed one. Metric fields are filled by [`enrich_row`].
#[must_use]
pub(crate) fn forensics_row(id: &str) -> ConversationRow {
    ConversationRow {
        id: id.to_owned(),
        model: String::new(),
        principal_ref: String::new(),
        phase: Some("event-log".to_owned()),
        pod_ip: None,
        harness_ready: false,
        closed: false,
        committed_turns: None,
        input_tokens: None,
        output_tokens: None,
        pending_approvals: None,
    }
}