supercode-interchange 0.5.45

//! The one conversation record (`docs/ONTOLOGY.md` §2.2): a conversation
//! reached on a surface, routed to a profile, run by a worker, with a
//! lifecycle. A Hermes `sessions` row, an OpenClaw session key and an
//! orchestrator `bindings` row each decode to it ONCE, here; the discovery
//! row's nouns are its projection ([`Binding::nouns`]).

use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

use super::residue::Residue;
use super::surface::{CrossSurface, Recurrence, SurfaceKey, Trigger};
use super::HarnessId;
use crate::session::OrchestrationNouns;

/// Why a binding ended (`docs/ORCHESTRATOR-IR.md` §2.5).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum EndReason {
    /// Idle expiry.
    Idle,
    /// The daily reset boundary.
    Daily,
    /// `/reset` (or the operator verb).
    Reset,
    /// `/new` (or the operator verb).
    New,
    /// Moved to another surface.
    Handoff,
    /// The worker failed.
    Error,
}

impl EndReason {
    /// Parse the wire word; anything else is not an end reason.
    pub fn parse(word: &str) -> Option<Self> {
        Some(match word {
            "idle" => Self::Idle,
            "daily" => Self::Daily,
            "reset" => Self::Reset,
            "new" => Self::New,
            "handoff" => Self::Handoff,
            "error" => Self::Error,
            _ => return None,
        })
    }

    /// The wire word.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Idle => "idle",
            Self::Daily => "daily",
            Self::Reset => "reset",
            Self::New => "new",
            Self::Handoff => "handoff",
            Self::Error => "error",
        }
    }
}

/// The worker session a binding points at.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct Worker {
    /// Which harness runs the conversation.
    pub harness: HarnessId,
    /// The harness's own session id, when one exists yet.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    /// Where that session's transcript can be read (a path or store address).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub locator: Option<String>,
}

/// A conversation moved (or moving) to another surface — Hermes `handoff_*`.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct Handoff {
    /// The destination platform.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub to: Option<String>,
    /// `pending` | `done` | `failed` (the source's own word, verbatim).
    pub state: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    /// The failure, when `state` is `failed`.
    pub error: Option<String>,
}

impl Default for Worker {
    fn default() -> Self {
        Self {
            harness: HarnessId::new(""),
            session_id: None,
            locator: None,
        }
    }
}

/// The conversation record.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct Binding {
    /// The surface; degenerate (all `None`) on a terminal.
    pub key: SurfaceKey,
    /// The profile that owns it (`None` = the home's default).
    #[serde(default)]
    pub profile: Option<String>,
    /// The worker session.
    pub worker: Worker,
    /// Why the conversation exists. The orchestrator's own store keeps no
    /// trigger column (it is derived from the key and recurrence), so the
    /// wire may omit it.
    #[serde(default)]
    pub trigger: Trigger,
    #[serde(default)]
    /// The job this conversation is a fire of.
    pub recurrence: Option<Recurrence>,
    #[serde(default)]
    /// Moved (or moving) to another surface.
    pub handoff: Option<Handoff>,
    /// RFC3339 instants where the source has them.
    #[serde(default)]
    pub started_at: Option<String>,
    #[serde(default)]
    /// Last activity, RFC3339.
    pub last_activity_at: Option<String>,
    #[serde(default)]
    /// End, RFC3339, when ended.
    pub ended_at: Option<String>,
    #[serde(default)]
    /// Why it ended.
    pub end_reason: Option<EndReason>,
    /// Source fields the record does not model, verbatim.
    #[serde(default)]
    pub residue: Residue,
}

impl Default for Binding {
    fn default() -> Self {
        Self {
            key: SurfaceKey::default(),
            profile: None,
            worker: Worker::default(),
            trigger: Trigger::Unknown,
            recurrence: None,
            handoff: None,
            started_at: None,
            last_activity_at: None,
            ended_at: None,
            end_reason: None,
            residue: Residue::default(),
        }
    }
}

impl Binding {
    /// The surface as a discovery row shows it: `None` on a terminal, where the
    /// key carries no key string, platform or chat id.
    pub fn surface(&self) -> Option<SurfaceKey> {
        let k = &self.key;
        if k.key.is_some() || k.platform.is_some() || k.chat_id.is_some() {
            Some(k.clone())
        } else {
            None
        }
    }

    /// The discovery row's nouns. `workspace` is left for the row's own cwd
    /// rule (`SessionMeta::workspace`), the one derivation that is not this
    /// record's to make.
    pub fn nouns(&self) -> OrchestrationNouns {
        OrchestrationNouns {
            trigger: Some(self.trigger),
            surface: self.surface(),
            profile: self.profile.clone(),
            recurrence: self.recurrence.clone(),
            cross_surface: self.handoff.as_ref().map(|h| CrossSurface {
                state: h.state.clone(),
                platform: h.to.clone(),
                error: h.error.clone(),
            }),
            workspace: None,
        }
    }
}

// ------------------------------------------------------------- Hermes rows

/// One Hermes `sessions` row, the columns a binding is made of. Empty strings
/// are read as absent by the decoder.
#[derive(Debug, Clone, Default)]
pub struct HermesSessionRow {
    /// `sessions.id`.
    pub id: String,
    /// `sessions.source`: `cli` | `tui` | `acp` | `api_server` | `cron` | `webhook` | a platform.
    pub source: Option<String>,
    /// `delegate` for a subagent child (Hermes lineage), else anything.
    pub lineage_kind: Option<String>,
    /// `sessions.session_key`, the gateway conversation key.
    pub session_key: Option<String>,
    /// `sessions.chat_id`.
    pub chat_id: Option<String>,
    /// `sessions.chat_type`.
    pub chat_type: Option<String>,
    /// `sessions.thread_id`.
    pub thread_id: Option<String>,
    /// `sessions.user_id` (the participant).
    pub user_id: Option<String>,
    /// `sessions.profile_name`, the routed profile.
    pub profile_name: Option<String>,
    /// `sessions.handoff_state`.
    pub handoff_state: Option<String>,
    /// `sessions.handoff_platform`.
    pub handoff_platform: Option<String>,
    /// `sessions.handoff_error`.
    pub handoff_error: Option<String>,
    /// Epoch seconds, as the store keeps them.
    pub started_at: Option<f64>,
    /// Epoch seconds when the session ended, if it has.
    pub ended_at: Option<f64>,
    /// `sessions.end_reason`, the store's own word.
    pub end_reason: Option<String>,
}

/// Hermes `sessions.source` → trigger. Cron fires are tagged `cron`; the CLI,
/// TUI and ACP adapter are human surfaces; `api_server` is the HTTP API;
/// `webhook` is inbound; every other value is a messaging platform.
pub fn hermes_trigger_for_source(source: &str) -> Trigger {
    match source {
        "" => Trigger::Unknown,
        "cron" => Trigger::Cron,
        "webhook" => Trigger::Webhook,
        "cli" | "tui" | "acp" | "console" => Trigger::Human,
        "api_server" | "api" => Trigger::Api,
        "kanban" => Trigger::Task,
        _ => Trigger::Channel,
    }
}

/// Hermes cron fire session ids are minted as `cron_<job_id>_<YYYYMMDD_HHMMSS>`
/// (`cron/scheduler.py`); recover the job id.
pub fn hermes_cron_job_id(session_id: &str) -> Option<String> {
    let rest = session_id.strip_prefix("cron_")?;
    let (job, stamp) = rest.rsplit_once('_')?;
    let (job, date) = job.rsplit_once('_')?;
    let ok = date.len() == 8
        && stamp.len() == 6
        && date.chars().all(|c| c.is_ascii_digit())
        && stamp.chars().all(|c| c.is_ascii_digit());
    if ok && !job.is_empty() {
        Some(job.to_string())
    } else {
        None
    }
}

/// Parse a Hermes gateway session key
/// (`agent:<profile|main>:<platform>:<chat_type>[:<chat_id>][:<thread_id>][:<participant>]`).
/// Returns the surface and the profile namespace (`None` for `main`).
pub fn parse_hermes_session_key(key: &str) -> Option<(SurfaceKey, Option<String>)> {
    let parts: Vec<&str> = key.split(':').collect();
    if parts.len() < 4 || parts[0] != "agent" {
        return None;
    }
    let profile = match parts[1] {
        "" | "main" | "default" => None,
        p => Some(p.to_string()),
    };
    let surface = SurfaceKey {
        key: Some(key.to_string()),
        platform: Some(parts[2].to_string()),
        kind: Some(parts[3].to_string()),
        chat_id: parts.get(4).map(|s| s.to_string()),
        thread_id: parts.get(5).map(|s| s.to_string()),
        participant_id: parts.get(6).map(|s| s.to_string()),
    };
    Some((surface, profile))
}

/// Render a surface as Hermes's `build_session_key` form, which
/// [`parse_hermes_session_key`] reads back unchanged.
pub fn render_hermes_session_key(profile: &str, key: &SurfaceKey) -> String {
    let mut parts = vec![
        "agent".to_string(),
        if profile.is_empty() {
            "main".to_string()
        } else {
            profile.to_string()
        },
        key.platform.clone().unwrap_or_default(),
        key.kind.clone().unwrap_or_default(),
    ];
    parts.extend(
        [
            key.chat_id.clone(),
            key.thread_id.clone(),
            key.participant_id.clone(),
        ]
        .into_iter()
        .flatten(),
    );
    parts.join(":")
}

fn epoch_to_rfc3339(seconds: f64) -> String {
    let millis = (seconds * 1000.0).round() as i64;
    let secs = millis.div_euclid(1000);
    let sub = millis.rem_euclid(1000) as u32;
    // civil-from-days (Howard Hinnant), enough for a timestamp string
    let days = secs.div_euclid(86_400);
    let sod = secs.rem_euclid(86_400);
    let z = days + 719_468;
    let era = z.div_euclid(146_097);
    let doe = z - era * 146_097;
    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    format!(
        "{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}.{sub:03}Z",
        sod / 3600,
        (sod % 3600) / 60,
        sod % 60
    )
}

impl Binding {
    /// Decode a Hermes `sessions` row. Always yields a record: a terminal
    /// session is a binding with a degenerate key. The columns win over the
    /// parsed key when both are present (ORC-8 finding: `api_server`
    /// conversations carry the surface only in `session_key`).
    pub fn from_hermes_row(row: &HermesSessionRow, locator: Option<&str>) -> Self {
        let nonempty = |v: &Option<String>| v.clone().filter(|s| !s.is_empty());
        let source = nonempty(&row.source).unwrap_or_default();
        let mut trigger = hermes_trigger_for_source(&source);
        if row.lineage_kind.as_deref() == Some("delegate") {
            trigger = Trigger::Parent;
        }
        let mut recurrence = None;
        if let Some(job_id) = hermes_cron_job_id(&row.id) {
            recurrence = Some(Recurrence {
                job_id,
                kind: "cron".into(),
            });
            trigger = Trigger::Cron;
        }
        let mut profile = None;
        let mut key = nonempty(&row.session_key)
            .and_then(|k| parse_hermes_session_key(&k))
            .map(|(surface, key_profile)| {
                profile = key_profile;
                surface
            })
            .unwrap_or_default();
        if key.key.is_none() {
            key.key = nonempty(&row.session_key);
        }
        if let Some(v) = nonempty(&row.chat_id) {
            key.chat_id = Some(v);
        }
        if let Some(v) = nonempty(&row.chat_type) {
            key.kind = Some(v);
        }
        if let Some(v) = nonempty(&row.thread_id) {
            key.thread_id = Some(v);
        }
        if let Some(v) = nonempty(&row.user_id) {
            key.participant_id = Some(v);
        }
        if key.platform.is_none() && trigger == Trigger::Channel {
            key.platform = Some(source.clone());
        }
        if let Some(p) = nonempty(&row.profile_name) {
            profile = Some(p);
        }
        let handoff = nonempty(&row.handoff_state).map(|state| Handoff {
            to: nonempty(&row.handoff_platform),
            state,
            error: nonempty(&row.handoff_error),
        });
        let mut residue = Residue::default();
        let end_reason = match nonempty(&row.end_reason) {
            Some(word) => match EndReason::parse(&word) {
                Some(r) => Some(r),
                None => {
                    residue.keep("end_reason", serde_json::Value::String(word));
                    None
                }
            },
            None => None,
        };
        Self {
            key,
            profile,
            worker: Worker {
                harness: HarnessId::new(HarnessId::HERMES),
                session_id: Some(row.id.clone()),
                locator: locator.map(str::to_string),
            },
            trigger,
            recurrence,
            handoff,
            started_at: row.started_at.map(epoch_to_rfc3339),
            last_activity_at: row.ended_at.or(row.started_at).map(epoch_to_rfc3339),
            ended_at: row.ended_at.map(epoch_to_rfc3339),
            end_reason,
            residue,
        }
    }
}

/// The Hermes `sessions.source` word for a binding (inverse of
/// [`hermes_trigger_for_source`]): a channel conversation's source is its
/// platform; the rest are the words Hermes's own doors mint.
pub fn hermes_source_for_binding(binding: &Binding) -> String {
    match binding.trigger {
        Trigger::Cron => "cron".into(),
        Trigger::Webhook => "webhook".into(),
        Trigger::Api => "api_server".into(),
        Trigger::Task => "kanban".into(),
        Trigger::Human => "cli".into(),
        Trigger::Parent => "delegate".into(),
        Trigger::Heartbeat => "heartbeat".into(),
        Trigger::Channel | Trigger::Unknown => {
            binding.key.platform.clone().unwrap_or_else(|| "cli".into())
        }
    }
}

// ----------------------------------------------------------- OpenClaw keys

/// Render an OpenClaw gateway session key for a binding under an agent
/// (inverse of [`parse_openclaw_session_key`]): a cron fire is `cron:<jobId>`,
/// a conversation is the agent shape.
pub fn render_openclaw_session_key(agent: &str, binding: &Binding) -> String {
    let key = &binding.key;
    if let Some(k) = &key.key {
        return k.clone();
    }
    if let Some(r) = &binding.recurrence {
        return format!("cron:{}", r.job_id);
    }
    if key.kind.as_deref() == Some("main") || key.platform.is_none() {
        return format!("agent:{agent}:main");
    }
    let mut out = format!(
        "agent:{agent}:{}:{}:{}",
        key.platform.clone().unwrap_or_default(),
        key.kind.clone().unwrap_or_else(|| "dm".into()),
        key.chat_id.clone().unwrap_or_default()
    );
    if let Some(t) = &key.thread_id {
        out.push_str(&format!(":thread:{t}"));
    }
    out
}

/// Parse an OpenClaw gateway session key. Shapes (`docs/channels/channel-routing.md`,
/// `docs/automation/cron-jobs.md`, `docs/cli/acp.md` upstream):
/// `agent:<id>:main`, `agent:<id>:<channel>:<group|channel>:<cid>[:thread|topic:<tid>]`,
/// `cron:<jobId>`, `hook:<name>:<id>`, `acp-bridge:<uuid>`.
pub fn parse_openclaw_session_key(
    key: &str,
) -> Option<(Option<String>, SurfaceKey, Trigger, Option<Recurrence>)> {
    let parts: Vec<&str> = key.split(':').collect();
    match parts.first().copied() {
        Some("agent") if parts.len() >= 3 => {
            let agent = Some(parts[1].to_string());
            if parts[2] == "main" {
                let surface = SurfaceKey {
                    key: Some(key.to_string()),
                    kind: Some("main".to_string()),
                    ..SurfaceKey::default()
                };
                return Some((agent, surface, Trigger::Unknown, None));
            }
            if parts.len() < 5 {
                return None;
            }
            let thread_id = match (parts.get(5), parts.get(6)) {
                (Some(&"thread"), Some(t)) | (Some(&"topic"), Some(t)) => Some(t.to_string()),
                _ => None,
            };
            let surface = SurfaceKey {
                key: Some(key.to_string()),
                platform: Some(parts[2].to_string()),
                kind: Some(parts[3].to_string()),
                chat_id: Some(parts[4].to_string()),
                thread_id,
                participant_id: None,
            };
            Some((agent, surface, Trigger::Channel, None))
        }
        Some("cron") if parts.len() >= 2 => Some((
            None,
            SurfaceKey {
                key: Some(key.to_string()),
                ..SurfaceKey::default()
            },
            Trigger::Cron,
            Some(Recurrence {
                job_id: parts[1..].join(":"),
                kind: "cron".into(),
            }),
        )),
        Some("hook") if parts.len() >= 2 => Some((
            None,
            SurfaceKey {
                key: Some(key.to_string()),
                ..SurfaceKey::default()
            },
            Trigger::Webhook,
            None,
        )),
        Some("acp-bridge") => Some((
            None,
            SurfaceKey {
                key: Some(key.to_string()),
                platform: Some("acp".into()),
                ..SurfaceKey::default()
            },
            Trigger::Api,
            None,
        )),
        _ => None,
    }
}

impl Binding {
    /// Decode an OpenClaw session key. `None` when the key has no known shape
    /// (nothing is claimed, as before). `agent_from_path` is the profile the
    /// file's own `agents/<id>/` directory names, used when the key names none.
    pub fn from_openclaw_key(
        key: &str,
        agent_from_path: Option<&str>,
        session_id: Option<&str>,
        locator: Option<&str>,
    ) -> Option<Self> {
        let (agent, surface, trigger, recurrence) = parse_openclaw_session_key(key)?;
        Some(Self {
            key: surface,
            profile: agent.or_else(|| agent_from_path.map(str::to_string)),
            worker: Worker {
                harness: HarnessId::new(HarnessId::OPENCLAW),
                session_id: session_id.map(str::to_string),
                locator: locator.map(str::to_string),
            },
            trigger,
            recurrence,
            ..Self::default()
        })
    }
}

// ------------------------------------------------------ orchestrator rows

/// One row of an orchestrator profile's `bindings` table, as read.
#[derive(Debug, Clone, Default)]
pub struct OrchestratorBindingRow {
    /// Surface platform.
    pub platform: String,
    /// Surface chat type (`dm` | `group` | `channel` | `thread`).
    pub chat_type: String,
    /// Surface chat id.
    pub chat_id: Option<String>,
    /// Surface thread id.
    pub thread_id: Option<String>,
    /// Surface participant id.
    pub participant_id: Option<String>,
    /// The worker harness.
    pub worker_harness: String,
    /// The worker session id; `None` until the worker has reported one.
    pub worker_session_id: Option<String>,
    /// Where the worker transcript can be read.
    pub worker_locator: Option<String>,
    /// RFC3339 start.
    pub started_at: Option<String>,
    /// RFC3339 last activity.
    pub last_activity_at: Option<String>,
    /// RFC3339 end, if ended.
    pub ended_at: Option<String>,
    /// Why it ended, the store's own word.
    pub end_reason: Option<String>,
    /// Handoff destination platform.
    pub handoff_to: Option<String>,
    /// Handoff state.
    pub handoff_state: Option<String>,
    /// Handoff error.
    pub handoff_error: Option<String>,
    /// The job a fire binding belongs to.
    pub recurrence_job_id: Option<String>,
}

impl Binding {
    /// Decode an orchestrator `bindings` row under `profile`. The key string is
    /// the orchestrator's own rendering (`docs/ORCHESTRATOR-IR.md` §2.3), which
    /// is Hermes's form, so [`parse_hermes_session_key`] reads it back unchanged.
    pub fn from_orchestrator_row(profile: &str, row: &OrchestratorBindingRow) -> Self {
        let mut key = SurfaceKey {
            key: None,
            platform: Some(row.platform.clone()),
            kind: Some(row.chat_type.clone()),
            chat_id: row.chat_id.clone(),
            thread_id: row.thread_id.clone(),
            participant_id: row.participant_id.clone(),
        };
        key.key = Some(render_hermes_session_key(profile, &key));
        let trigger = if row.recurrence_job_id.is_some() {
            Trigger::Cron
        } else if row.platform == "webhook" {
            Trigger::Webhook
        } else {
            Trigger::Channel
        };
        let mut residue = Residue::default();
        let end_reason = match row.end_reason.as_deref() {
            Some(word) => match EndReason::parse(word) {
                Some(r) => Some(r),
                None => {
                    residue.keep("end_reason", serde_json::Value::String(word.to_string()));
                    None
                }
            },
            None => None,
        };
        Self {
            key,
            profile: Some(profile.to_string()),
            worker: Worker {
                harness: HarnessId::new(&row.worker_harness),
                session_id: row.worker_session_id.clone().filter(|s| !s.is_empty()),
                locator: row.worker_locator.clone(),
            },
            trigger,
            recurrence: row.recurrence_job_id.clone().map(|job_id| Recurrence {
                job_id,
                kind: "cron".into(),
            }),
            handoff: row.handoff_state.clone().map(|state| Handoff {
                to: row.handoff_to.clone(),
                state,
                error: row.handoff_error.clone(),
            }),
            started_at: row.started_at.clone(),
            last_activity_at: row.last_activity_at.clone(),
            ended_at: row.ended_at.clone(),
            end_reason,
            residue,
        }
    }
}