onepipeline 0.38.0

//! The merged event stream's envelope.
//!
//! `onepipeline` merges the three libraries' streams into one, so it both
//! *relays* envelopes produced by `oneagentgraph` and `onevcs` and *emits* its
//! own. The envelope, its labels, sources and phases, the artifact reference and
//! the kind are `onemessagebus-agent`'s, over the `onemessagebus` core, and are
//! re-exported here at the paths this crate has always published them at — the
//! same types both siblings re-export, so a relayed envelope is one value on
//! every side of a relay. `onemessagebus`'s `docs/contract.md` is the one source
//! of that shape; `docs/contract.md` here keeps a marked copy of the text, and
//! `tests/contract.rs` drives it through these re-exports.
//!
//! What stays this crate's is its vocabulary: the closed set of kinds it emits
//! ([`PipelineKind`]), and the payload each carries, every one a registered bus
//! message in the registry this crate constructs.

use std::sync::OnceLock;

use onemessagebus::{Read, Registry};
use onemessagebus_agent::registry::{EVENT_ENVELOPE_FAMILY, EVENT_ENVELOPE_READS};

pub use onemessagebus::{Kind as EventKind, MAX_PAYLOAD_TEXT_BYTES};
pub use onemessagebus_agent::event::{ArtifactRef, Envelope, Labels, Phase, Source};

/// The envelope version this crate stamps on everything it writes.
///
/// **2** since the journal's record shapes moved: at `1` an `edit-committed` may
/// be an accepted command that changed nothing; at `2` it means something changed,
/// an accepted command that did not is [`PipelineKind::CommandAccepted`], and both
/// kinds carry `operation_kinds` so a reader keys on what happened without
/// deserializing the command. Entry 65 of `docs/contract-divergences.md` proposes
/// the move.
///
/// The number is the agent profile's for the `pipeline` source, and the newest
/// version its registry reads of `agent.event-envelope`: a test beside this holds
/// all three to one another. A **relayed** envelope keeps its producer's own
/// number, exactly as it keeps that producer's `stream`, `seq`, `source` and kind.
pub const ENVELOPE_VERSION: u32 = EVENT_ENVELOPE_READS[0];

/// Every envelope version this build reads, newest first.
///
/// The agent profile's read-set for `agent.event-envelope`, which is what its
/// registry registers and what the fold asks that registry of a record. Version
/// `1` is read whole: nothing was removed from the envelope or from a record's
/// payload, so a `1` folds exactly as it always did.
pub const ENVELOPE_VERSIONS_READ: &[u32] = EVENT_ENVELOPE_READS;

/// Whether this build knows the envelope schema a record was written at.
///
/// Asked of **this library's own** records and never of a relayed one: a
/// sibling's version is that library's own vocabulary, and judging it by this
/// crate's table would refuse a producer for moving at its own pace.
///
/// Answered by the registry's [`Registry::read_at`] for `agent.event-envelope`:
/// a version it reads at is one in the read set the registry declares, and that
/// set is taken from the registry once per process rather than rebuilt for every
/// record folded — the registry derives it by walking every document it holds,
/// and a fold asks this of each record of a run. A
/// reader that meets `false` has met a record a *newer* build wrote, whose kinds
/// or payload may mean something this build would read wrongly. What the fold
/// does with that is report rather than guess — it marks the run as one it could
/// not read whole, and a driver says so before it converges.
#[must_use]
pub(crate) fn written_at_a_known_version(envelope: &Envelope) -> bool {
    static READ_AT: OnceLock<Vec<u32>> = OnceLock::new();
    READ_AT
        .get_or_init(|| {
            registry()
                .read_set(EVENT_ENVELOPE_FAMILY)
                .into_iter()
                .filter(|version| {
                    matches!(
                        registry().read_at(EVENT_ENVELOPE_FAMILY, *version),
                        Read::At(_)
                    )
                })
                .collect()
        })
        .contains(&envelope.v)
}

/// The registry this crate constructs: the agent profile's, with every payload
/// this crate emits registered beside the envelope that carries it.
///
/// Built once per process, on first use; `main` asks for it before anything
/// else so a build whose own payload documents do not register fails at start
/// rather than on the first record it folds.
pub(crate) fn registry() -> &'static Registry {
    static REGISTRY: OnceLock<Registry> = OnceLock::new();
    REGISTRY.get_or_init(crate::payload::registry)
}

/// Every event kind this library emits, and exactly those.
///
/// Enumerated because they are `onepipeline`'s own vocabulary: a kind this crate
/// writes cannot be a typo, and a reader folds a closed set rather than matching
/// strings. The kinds a *sibling* produces stay [`EventKind`]'s wire string —
/// this crate relays those unchanged, and an enum there would reject a kind a
/// newer sibling already emits. `docs/contract.md` lists exactly these.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
#[non_exhaustive]
pub enum PipelineKind {
    /// The run was launched.
    RunStarted,
    /// A launch deliberately proceeded beside live repository holders.
    ConcurrentAcknowledged,
    /// Every dependency of a node has settled `done`, so it may dispatch now.
    NodeReady,
    /// A node's dispatch was started.
    NodeDispatched,
    /// A node reached a terminal status.
    NodeSettled,
    /// A live edit was accepted, and committing it is what made the change it
    /// records.
    ///
    /// Emitted only where at least one operation the command compiled to
    /// **changed** something a reader folding this record moves —
    /// `Operation::commits_a_change` decides that, exhaustively over the
    /// operations. That is the desired graph *and* the node record derived beside
    /// it, because those are one durable document to such a reader: an
    /// attestation, a park and a settlement from evidence move a node's recorded
    /// state without moving the graph, and are read back off this record by name.
    /// "Something changed" is what a reader has always taken this kind to mean,
    /// and it is exactly what it still means. The kinds of the operations it
    /// committed ride on the record as `operation_kinds`, so a reader keys on
    /// what happened without parsing the command that produced it.
    EditCommitted,
    /// A command was accepted and committed nothing a reader folds.
    ///
    /// The other half of the split above: a `finding` and a `complete` are
    /// accepted, are answered `applied`, and change nothing — each is a
    /// **report**, whose own record is a planner surface and a
    /// `completion-requested` respectively — so journalling them as committed
    /// edits made one kind carry two meanings.
    CommandAccepted,
    /// A live edit was refused, with the reason its submitter was told.
    EditRejected,
    /// A surface was *sent*. Delivery is a separate fact.
    PlannerSurfaceQueued,
    /// A surface was *consumed* by the planner. This is what restarts the check-in clock.
    PlannerSurfaced,
    /// The planner answered a consumed surface.
    PlannerReplied,
    /// A human action was attested.
    HumanAttested,
    /// A fresh driver was attached to an intact ledger.
    DriverAdopted,
    /// The run was ended by `stop`.
    RunStopped,
    /// An in-flight dispatch recorded nothing past the stall threshold.
    QuietWorker,
    /// The loop is not running a node it has not settled, and this is why.
    ///
    /// Written when a hold **begins**, again when what the node is held by
    /// **changes**, and never on a pass where it is held by what it was held by
    /// before. `reasons` carries one entry per reason holding it at once, so a
    /// node behind three running nodes and a node whose dependency has not
    /// settled and a node that is both are three answers a reader tells apart
    /// without joining another record.
    NodeHeld,
    /// That hold cleared, carrying the reasons that were holding it.
    NodeUnheld,
    /// A blocking surface began holding a subtree of dependents back.
    DecisionPending,
    /// That surface was cleared, and the subtree it held was released.
    DecisionCleared,
    /// A cross-DAG edge resolved, with how far its upstream had got when it did.
    CrossDagSatisfied,
    /// A cross-DAG upstream advanced after its consumer recorded it.
    UpstreamModified,
    /// The planner requested completion, independently of graph mutation.
    CompletionRequested,
    /// A node is held under `published` adoption, waiting on releases.
    ///
    /// Raised when the wait begins and again on its own interval, so a wait
    /// nobody has ended cannot go silent. Each awaited release names its style,
    /// so a wait on a machine and a wait on a person are tellable apart from the
    /// payload as well as from the surface beside it.
    ReleaseWait,
    /// One release a node was waiting on has happened.
    ReleaseArrived,
    /// A fast-adoption node was told the releases it was waiting on arrived, and
    /// whether the note reached a running turn or its next dispatch.
    ReleaseAdopted,
    /// One mechanically checkable acceptance criterion was compared against the
    /// branch its node settled on.
    ///
    /// Emitted for every criterion this build could parse into "this named file
    /// holds this literal", carrying the answer — `match`, `mismatch`, or
    /// `unread`, the check declining to answer a file it could not read. A
    /// criterion it could not parse is not recorded at all: the check says
    /// nothing about prose it has no business ruling on.
    CriterionChecked,
    /// A drafting dispatch ran for a change request's body and produced none.
    ///
    /// Only where one was *configured and attempted*: a launch that named no
    /// pr-author graph, and a node that carried its own `body`, both spend no
    /// dispatch and neither is a failure to report. The payload's `ending` says
    /// which of the three it was, because they need three different fixes.
    BodyNotDrafted,
    /// A party of a node's conversation was **shown** a manager's note.
    ///
    /// Written when the relayed stream shows the presentation happening — the
    /// worker's turn opening on the note, the supervisor's turn opening after
    /// it — and never at delivery, where the conversation has only said where
    /// it *will* route the note. `note-delivered` records what was confirmed
    /// the moment the conversation acknowledged the note and what it routed
    /// onward; this is the record for each routed presentation that then
    /// happened, so a conversation interrupted between the two leaves no claim
    /// that the second party saw anything. Carries `party`, the `turn` the
    /// presentation opened, and the note.
    NoteShown,
    /// A run-end hook is about to be started, and this is the run's marker that
    /// one has fired.
    ///
    /// Journaled before the command starts, so a run carrying one fires neither
    /// hook again whichever process looks next. Carries `hook`, `command`, and
    /// `reason`.
    RunHookFired,
    /// A run-end hook ended, and how: `hook`, `exit`, `ending`, and the `log` its
    /// output was kept in.
    RunHookFinished,
    /// A driver let go of a run paused on a decision, so no run-end hook fired.
    ///
    /// Carries the `settlement` it let go at, which is `awaiting-planner`: the run
    /// has not ended, and the driver that adopts it judges again.
    RunHookWithheld,
}

impl PipelineKind {
    /// The kind as it appears on the wire.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::RunStarted => "run-started",
            Self::ConcurrentAcknowledged => "concurrent-acknowledged",
            Self::NodeReady => "node-ready",
            Self::NodeDispatched => "node-dispatched",
            Self::NodeSettled => "node-settled",
            Self::EditCommitted => "edit-committed",
            Self::CommandAccepted => "command-accepted",
            Self::EditRejected => "edit-rejected",
            Self::PlannerSurfaceQueued => "planner-surface-queued",
            Self::PlannerSurfaced => "planner-surfaced",
            Self::PlannerReplied => "planner-replied",
            Self::HumanAttested => "human-attested",
            Self::DriverAdopted => "driver-adopted",
            Self::RunStopped => "run-stopped",
            Self::QuietWorker => "quiet-worker",
            Self::NodeHeld => "node-held",
            Self::NodeUnheld => "node-unheld",
            Self::DecisionPending => "decision-pending",
            Self::DecisionCleared => "decision-cleared",
            Self::CrossDagSatisfied => "cross-dag-satisfied",
            Self::UpstreamModified => "upstream-modified",
            Self::CompletionRequested => "completion-requested",
            Self::ReleaseWait => "release-wait",
            Self::ReleaseArrived => "release-arrived",
            Self::ReleaseAdopted => "release-adopted",
            Self::CriterionChecked => "criterion-checked",
            Self::BodyNotDrafted => "body-not-drafted",
            Self::NoteShown => "note-shown",
            Self::RunHookFired => "run-hook-fired",
            Self::RunHookFinished => "run-hook-finished",
            Self::RunHookWithheld => "run-hook-withheld",
        }
    }

    /// The kind an envelope carries, when it is one of this library's own.
    ///
    /// `None` for anything else, which is every kind a sibling produced: the
    /// merged store holds all three vocabularies and only this one is closed.
    pub fn from_wire(kind: &EventKind) -> Option<Self> {
        PIPELINE_KINDS
            .iter()
            .copied()
            .find(|candidate| candidate.as_str() == kind.0)
    }
}

impl std::fmt::Display for PipelineKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

impl From<PipelineKind> for EventKind {
    fn from(kind: PipelineKind) -> Self {
        Self(kind.as_str().to_string())
    }
}

/// Every kind, for the lookup above and for the contract's own list.
pub const PIPELINE_KINDS: &[PipelineKind] = &[
    PipelineKind::RunStarted,
    PipelineKind::ConcurrentAcknowledged,
    PipelineKind::NodeReady,
    PipelineKind::NodeDispatched,
    PipelineKind::NodeSettled,
    PipelineKind::EditCommitted,
    PipelineKind::CommandAccepted,
    PipelineKind::EditRejected,
    PipelineKind::PlannerSurfaceQueued,
    PipelineKind::PlannerSurfaced,
    PipelineKind::PlannerReplied,
    PipelineKind::HumanAttested,
    PipelineKind::DriverAdopted,
    PipelineKind::RunStopped,
    PipelineKind::QuietWorker,
    PipelineKind::NodeHeld,
    PipelineKind::NodeUnheld,
    PipelineKind::DecisionPending,
    PipelineKind::DecisionCleared,
    PipelineKind::CrossDagSatisfied,
    PipelineKind::UpstreamModified,
    PipelineKind::CompletionRequested,
    PipelineKind::ReleaseWait,
    PipelineKind::ReleaseArrived,
    PipelineKind::ReleaseAdopted,
    PipelineKind::CriterionChecked,
    PipelineKind::BodyNotDrafted,
    PipelineKind::NoteShown,
    PipelineKind::RunHookFired,
    PipelineKind::RunHookFinished,
    PipelineKind::RunHookWithheld,
];

/// The id of a stored artifact.
///
/// This crate's own rather than the bus's, as it is `onevcs`'s: `onemessagebus`
/// 0.4.0 carries an artifact's id as a plain string on [`ArtifactRef`], and a
/// publication's failure evidence names a typed one.
#[derive(
    Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
)]
#[serde(transparent)]
pub struct ArtifactId(pub String);

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    /// The envelope this build writes, as a document.
    const GOLDEN: &str = include_str!("../tests/golden/envelope-v2.json");

    /// One this build did **not** write, at the version before the bump.
    const GOLDEN_BEFORE: &str = include_str!("../tests/golden/envelope-v1.json");

    /// The shape this build stamps its own records with, held to a committed
    /// document — through the re-exported types, byte for byte.
    ///
    /// A runs root outlives the build that wrote into it and is read by things
    /// outside this repository, so the envelope is a published document rather
    /// than an internal struct: a field renamed, an optional one becoming an
    /// explicit null, or the version moving without anyone deciding to move it
    /// are all things a consumer finds out about by breaking.
    #[test]
    fn the_envelope_this_build_writes_is_the_committed_golden() {
        let golden: serde_json::Value = serde_json::from_str(GOLDEN).expect("the golden is JSON");
        assert_eq!(
            golden["v"],
            json!(ENVELOPE_VERSION),
            "the golden is not at the version this build writes. Bump ENVELOPE_VERSION and \
             add tests/golden/envelope-v<n>.json together, keeping the older file as the \
             read-compatibility reference"
        );

        let envelope: Envelope = serde_json::from_str(GOLDEN).expect("it parses");
        assert_eq!(envelope.v, ENVELOPE_VERSION);
        assert_eq!(envelope.source, Source::Pipeline);
        assert_eq!(envelope.kind, EventKind("edit-committed".into()));
        assert!(written_at_a_known_version(&envelope));
        assert_eq!(
            serde_json::to_string_pretty(&envelope).expect("it serializes"),
            GOLDEN.trim_end(),
            "the envelope changed shape. Bump ENVELOPE_VERSION, add the golden for the new \
             one, and keep this file as what the older version looked like"
        );
    }

    /// And the one before it still reads, whole, and comes back out unchanged.
    ///
    /// The version bump is a statement about what a **new** record promises, not
    /// a line drawn under the old ones: a v1 journal is the ordinary contents of
    /// a runs root, and this build folds it. It is also not restamped — reading a
    /// record does not make it this build's — so a store round-trips as its
    /// writer wrote it.
    #[test]
    fn the_envelope_version_before_this_one_is_still_read_and_never_restamped() {
        let before: serde_json::Value =
            serde_json::from_str(GOLDEN_BEFORE).expect("the golden is JSON");
        assert_eq!(before["v"], json!(1));
        assert!(
            ENVELOPE_VERSIONS_READ.contains(&1),
            "this build no longer reads the version its committed fixture is written at"
        );

        let envelope: Envelope = serde_json::from_str(GOLDEN_BEFORE).expect("it parses");
        assert!(written_at_a_known_version(&envelope));
        assert_eq!(envelope.v, 1, "a version this build read was rewritten");
        assert_eq!(
            serde_json::to_string_pretty(&envelope).expect("it serializes"),
            GOLDEN_BEFORE.trim_end()
        );

        // And a version nothing has published is not read, which is what makes
        // the set above a statement rather than a comment.
        let ahead: Envelope = serde_json::from_value(json!({
            "v": ENVELOPE_VERSION + 1,
            "ts": "2026-09-09T04:00:00.000Z",
            "stream": "onepipeline-7f3a",
            "seq": 43,
            "source": "pipeline",
            "kind": "edit-committed",
            "labels": {},
            "payload": {},
            "artifacts": []
        }))
        .expect("a newer build's record still parses structurally");
        assert!(!written_at_a_known_version(&ahead));
    }

    /// The version this build writes and the versions it reads are the agent
    /// profile's registry's answer, not a second table kept here.
    ///
    /// Three statements of one number — the constant a writer stamps, the
    /// profile's write version for the `pipeline` source, and the newest version
    /// the registry this crate constructs reads — so a bus release that moved one
    /// without the others fails here rather than in a store.
    #[test]
    fn the_registry_answers_the_envelope_versions_this_build_writes_and_reads() {
        let registry = registry();
        assert_eq!(
            registry.read_set(EVENT_ENVELOPE_FAMILY),
            ENVELOPE_VERSIONS_READ.to_vec()
        );
        assert_eq!(
            registry.writes(EVENT_ENVELOPE_FAMILY),
            Some(ENVELOPE_VERSION)
        );
        assert_eq!(Source::Pipeline.write_version(), ENVELOPE_VERSION);
        for version in ENVELOPE_VERSIONS_READ {
            assert_eq!(
                registry.read_at(EVENT_ENVELOPE_FAMILY, *version),
                Read::At(ENVELOPE_VERSION)
            );
        }
        match registry.read_at(EVENT_ENVELOPE_FAMILY, ENVELOPE_VERSION + 1) {
            Read::Unknown(unknown) => {
                assert_eq!(unknown.declared, ENVELOPE_VERSION + 1);
                assert_eq!(unknown.read_set, ENVELOPE_VERSIONS_READ.to_vec());
            }
            Read::At(at) => panic!("a version nothing published was read at {at}"),
        }
    }

    /// The optional fields are optional in both directions: absent stays absent
    /// on the wire, and present survives the trip.
    ///
    /// `phase` is the one an envelope declares, and it is the field a store
    /// written before there was a phase depends on: a build that serialized it as
    /// an explicit null would rewrite every such record the first time it read
    /// one back.
    #[test]
    fn an_envelopes_optional_fields_round_trip_and_are_omitted_when_empty() {
        let bare = json!({
            "v": ENVELOPE_VERSION,
            "ts": "2026-09-09T04:00:00.000Z",
            "stream": "onepipeline-7f3a",
            "seq": 1,
            "source": "pipeline",
            "kind": "node-ready",
            "labels": {},
            "payload": {},
            "artifacts": []
        });
        let envelope: Envelope = serde_json::from_value(bare.clone()).expect("it parses");
        assert_eq!(envelope.dimensions.phase, None);
        assert_eq!(serde_json::to_value(&envelope).expect("serializes"), bare);

        let mut with = bare.clone();
        with["phase"] = json!("release");
        let envelope: Envelope = serde_json::from_value(with.clone()).expect("it parses");
        assert_eq!(envelope.dimensions.phase, Some(Phase::Release));
        assert_eq!(serde_json::to_value(&envelope).expect("serializes"), with);

        // The three defaulted containers are the same promise: a record that
        // omitted them reads, and comes back out omitting nothing it carried.
        let minimal = json!({
            "v": ENVELOPE_VERSION,
            "ts": "2026-09-09T04:00:00.000Z",
            "stream": "onepipeline-7f3a",
            "seq": 2,
            "source": "pipeline",
            "kind": "node-ready"
        });
        let envelope: Envelope = serde_json::from_value(minimal).expect("it parses");
        assert!(envelope.payload.is_empty() && envelope.artifacts.is_empty());
        assert_eq!(envelope.labels, Labels::default());
    }
}