kyyn_core/

plugin.rs

//! The `SourcePlugin` contract — what a source is, to the engine.
//!
//! A plugin's only job is to land raw material in the run directory the
//! engine hands it; the curation agent is the universal adapter, so the
//! engine never parses source payloads. Builtins implement the trait
//! in-process; external plugins speak the same shapes over stdio RON
//! (harness arrives with tap support).

use std::path::PathBuf;

use serde::{Deserialize, Serialize};

/// What a plugin declares about itself.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Describe {
    /// Plugin name (`local-recordings`, `graph`).
    pub name: String,
    /// The link namespace this source's citations carry (`recording`,
    /// `graph`). Recorded into `sources.ron` at install; collision-checked.
    pub link_namespace: String,
    pub fetch_style: FetchStyle,
    /// FALLBACK credentials realm; prefer `config_auth_realm` (derived from
    /// validated configuration — tenant/client — so unrelated tenants never
    /// share a token). `None` = no auth (local sources).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub auth_realm: Option<String>,
    /// Plugin protocol version (ADR 0001). v1 = typed items + engine-owned
    /// checkpoints — the first externally observable wire contract; earlier
    /// in-process trait shapes are implementation history, not versions.
    /// The engine refuses to fetch from a plugin declaring any version other
    /// than [`PROTOCOL`], and the field is REQUIRED on the wire: no plugin
    /// protocol has ever shipped besides the current one, so a missing
    /// declaration is a malformed plugin, not an old one — there is no
    /// legacy to default to.
    pub protocol: u32,
}

/// The plugin contract version this engine speaks (ADR 0001).
pub const PROTOCOL: u32 = 1;

/// How runs of this source are bounded.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum FetchStyle {
    /// Half-open time windows, engine-computed (`[last end − lookback, now)`).
    Windowed,
    /// Current state of something (a tracked file); dedup by version/etag.
    Snapshot,
    /// Ingest whatever completed material is waiting (a local spool).
    Sweep,
}

/// An interactive sign-in challenge (device-code style): show the code,
/// send the human to the URL, poll until done. The `handle` is an opaque
/// continuation for `auth_poll` — the engine holds it server-side.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthChallenge {
    pub verification_url: String,
    pub user_code: String,
    pub expires_in_secs: u64,
    pub handle: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum AuthPollResult {
    Pending,
    /// Signed in — the string names the identity where known.
    Done(String),
    Failed(String),
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum AuthStatus {
    NotRequired,
    /// Signed in — the string names the identity ("tom@…").
    Authenticated(String),
    /// Not signed in — the string says how to fix it.
    NotAuthenticated(String),
}

/// What the engine hands a plugin for AUTH operations. Paths are
/// engine-owned: plugins never invent storage locations.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Context {
    /// The instance's config from `sources.ron`, as canonical RON text.
    pub config: String,
    /// Credential storage for this plugin's auth realm.
    pub secrets_dir: PathBuf,
}

/// One fetch invocation (ADR 0001). The checkpoint is a READ-ONLY snapshot
/// of the last successfully PUBLISHED run's successor — plugins carry no
/// writable durable state, so no crash before publication can suppress a
/// future fetch.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FetchRequest {
    pub config: String,
    pub secrets_dir: PathBuf,
    /// This run's staging directory in the ledger (operation-private).
    pub out_dir: PathBuf,
    pub spec: RunSpec,
    /// Opaque plugin-defined text (RON recommended); engine-persisted.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub checkpoint: Option<String>,
}

/// A fetch's result. The engine publishes the manifest atomically and only
/// then treats `next_checkpoint` as the authoritative cursor.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct FetchResult {
    pub items: Vec<Item>,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub notes: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_checkpoint: Option<String>,
}

/// What the engine asks a fetch to do (matches the plugin's declared style).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum RunSpec {
    /// Half-open `[from, to)`, RFC3339 instants.
    Window {
        from: String,
        to: String,
    },
    Snapshot,
    Sweep,
}

/// One PROVIDER item — a message, an event, a chat message, a recording
/// session, a file version. Identity is `source instance + kind + id`
/// (ADR 0001); run ids, filenames and bundle names are never identity.
/// Serde defaults keep legacy aggregate manifests parseable.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Item {
    /// Stable provider id — engine-owned cross-run dedup keys on this.
    pub id: String,
    /// Provider kind: `email`, `event`, `chat-message`, `transcript`,
    /// `recording`, `file`… Empty = legacy aggregate-manifest item.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub kind: String,
    /// Provider version (an eTag) when the provider has one.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub version: Option<String>,
    /// sha256 (hex) of the item's canonical PRIMARY content — the locator-
    /// selected record's compact JSON when a locator names one in a JSON
    /// bundle, the first file's bytes otherwise. The version of last resort
    /// and the primary evidence-integrity anchor. Empty = legacy.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub content_hash: String,
    /// Files written, relative to the run dir.
    pub files: Vec<String>,
    /// sha256 (hex) of each SECONDARY file — every entry of `files` beyond
    /// the first — keyed by its run-relative path (ADR 0006). Extends the
    /// primary anchor's integrity discipline to transcripts, attendance
    /// reports, attachments: the engine refuses a fetch whose digests don't
    /// match the bytes, and they join the item-version fingerprint (see
    /// [`Item::version_fingerprint`]). REQUIRED for multi-file items; empty
    /// on single-file items and on manifests that predate per-file hashes,
    /// whose secondary files read as unverified until re-fetched.
    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
    pub file_hashes: std::collections::BTreeMap<String, String>,
    /// Where the item lives WITHIN a bundle file (e.g. the record's
    /// provider id in a JSON array) — bundles are storage, not identity.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub locator: Option<String>,
    /// Item metadata (a recording's start instant, a subject line…).
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub meta: String,
}

impl Item {
    /// The item-VERSION fingerprint — what "same version" means for cross-
    /// run dedup and evaluation keys. An aggregate of the primary anchor
    /// (`content_hash`, a LOGICAL record digest for bundle items — never the
    /// physical bundle file, which would bump every sibling record when one
    /// changes) and the secondary-file digests in canonical (sorted) path
    /// order, so a plugin reordering its files fabricates nothing. A changed
    /// transcript is therefore a NEW version (new curation work) even while
    /// the primary record stands. With no secondary digests the fingerprint
    /// IS the bare `content_hash`, keeping single-file items' versions and
    /// evaluation keys exactly as they were before per-file hashes.
    pub fn version_fingerprint(&self) -> String {
        version_fingerprint(&self.content_hash, &self.file_hashes)
    }
}

/// See [`Item::version_fingerprint`] — shared with engine projections that
/// store the components rather than the item.
pub fn version_fingerprint(
    content_hash: &str,
    file_hashes: &std::collections::BTreeMap<String, String>,
) -> String {
    if file_hashes.is_empty() {
        return content_hash.to_string();
    }
    use sha2::Digest;
    let mut h = sha2::Sha256::new();
    h.update(content_hash.as_bytes());
    for (path, digest) in file_hashes {
        h.update([0u8]);
        h.update(path.as_bytes());
        h.update([0u8]);
        h.update(digest.as_bytes());
    }
    format!("{:x}", h.finalize())
}

/// The contract (ADR 0001). Builtins implement it in-process; the stdio
/// harness for external plugins wraps the same shapes.
pub trait SourcePlugin {
    fn describe(&self) -> Describe;
    /// Pure, offline config validation — run at PROPOSAL validation so a
    /// malformed sources.ron cannot be accepted (SOL finding 35).
    fn validate_config(&self, config: &str) -> Result<(), String>;
    /// The auth realm this CONFIGURATION requires (tenant/client-derived —
    /// SOL finding 33). `None` falls back to `Describe::auth_realm`.
    fn config_auth_realm(&self, _config: &str) -> Result<Option<String>, String> {
        Ok(None)
    }
    fn auth_status(&self, ctx: &Context) -> Result<AuthStatus, String>;
    fn authenticate(&self, ctx: &Context) -> Result<(), String>;
    /// Begin an interactive sign-in (device-code style). Default: not
    /// supported — surfaces degrade to "run the CLI" with a live status
    /// poll, so plugins without an interactive flow still work in the web.
    fn auth_begin(&self, _ctx: &Context) -> Result<AuthChallenge, String> {
        Err("interactive sign-in is not supported for this source — use the CLI".into())
    }
    /// One poll of a pending interactive sign-in.
    fn auth_poll(&self, _ctx: &Context, _handle: &str) -> Result<AuthPollResult, String> {
        Err("interactive sign-in is not supported for this source".into())
    }
    fn fetch(&self, req: &FetchRequest) -> Result<FetchResult, String>;
}

// --- the tap harness wire (ADR 0005) ----------------------------------------

/// One request to a tap binary: `<binary> --plugin <name>` with this as RON
/// on stdin, one [`PluginResponse`] as RON on stdout. Mirrors the schema
/// protocol: a fresh process per call, no session state — checkpoints and
/// ledgers are engine-owned, so plugins have nothing to keep alive.
#[derive(Debug, Serialize, Deserialize)]
pub enum PluginRequest {
    Describe,
    ValidateConfig { config: String },
    ConfigAuthRealm { config: String },
    AuthStatus { ctx: Context },
    Authenticate { ctx: Context },
    AuthBegin { ctx: Context },
    AuthPoll { ctx: Context, handle: String },
    Fetch { req: FetchRequest },
}

#[derive(Debug, Serialize, Deserialize)]
pub enum PluginResponse {
    Describe(Describe),
    Ok,
    Realm(Option<String>),
    Status(AuthStatus),
    Challenge(AuthChallenge),
    Poll(AuthPollResult),
    Fetched(FetchResult),
    /// Any verb's failure — the message the trait method returned.
    Error(String),
}

/// The whole main() of a tap binary: parse `--plugin <name>`, resolve it
/// through the tap's own table, serve one request. Progress goes to stderr
/// (the harness forwards it to the host's progress sink).
pub fn tap_main(select: impl Fn(&str) -> Option<Box<dyn SourcePlugin>>) {
    use std::io::Read;
    let respond = |r: &PluginResponse| match crate::ronfmt::to_ron(r) {
        Ok(text) => print!("{text}"),
        Err(e) => {
            eprintln!("serializing response: {e}");
            std::process::exit(2);
        }
    };
    let args: Vec<String> = std::env::args().collect();
    let name = match args.iter().position(|a| a == "--plugin") {
        Some(i) if i + 1 < args.len() => args[i + 1].clone(),
        _ => {
            respond(&PluginResponse::Error("usage: --plugin <name>".into()));
            std::process::exit(2);
        }
    };
    let Some(plugin) = select(&name) else {
        respond(&PluginResponse::Error(format!(
            "this tap does not serve a plugin named '{name}'"
        )));
        std::process::exit(2);
    };
    let mut input = String::new();
    if let Err(e) = std::io::stdin().read_to_string(&mut input) {
        respond(&PluginResponse::Error(format!("reading stdin: {e}")));
        std::process::exit(2);
    }
    let request = match ron::from_str::<PluginRequest>(&input) {
        Ok(r) => r,
        Err(e) => {
            respond(&PluginResponse::Error(format!("parsing request: {e}")));
            std::process::exit(2);
        }
    };
    let response = match request {
        PluginRequest::Describe => PluginResponse::Describe(plugin.describe()),
        PluginRequest::ValidateConfig { config } => match plugin.validate_config(&config) {
            Ok(()) => PluginResponse::Ok,
            Err(e) => PluginResponse::Error(e),
        },
        PluginRequest::ConfigAuthRealm { config } => match plugin.config_auth_realm(&config) {
            Ok(r) => PluginResponse::Realm(r),
            Err(e) => PluginResponse::Error(e),
        },
        PluginRequest::AuthStatus { ctx } => match plugin.auth_status(&ctx) {
            Ok(s) => PluginResponse::Status(s),
            Err(e) => PluginResponse::Error(e),
        },
        PluginRequest::Authenticate { ctx } => match plugin.authenticate(&ctx) {
            Ok(()) => PluginResponse::Ok,
            Err(e) => PluginResponse::Error(e),
        },
        PluginRequest::AuthBegin { ctx } => match plugin.auth_begin(&ctx) {
            Ok(c) => PluginResponse::Challenge(c),
            Err(e) => PluginResponse::Error(e),
        },
        PluginRequest::AuthPoll { ctx, handle } => match plugin.auth_poll(&ctx, &handle) {
            Ok(p) => PluginResponse::Poll(p),
            Err(e) => PluginResponse::Error(e),
        },
        PluginRequest::Fetch { req } => match plugin.fetch(&req) {
            Ok(r) => PluginResponse::Fetched(r),
            Err(e) => PluginResponse::Error(e),
        },
    };
    respond(&response);
}