use std::collections::BTreeMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use onevcs::{
DraftReason, EventStream, Lifecycle, MergePolicy, Providers, Publication, PublishOutcome,
PublishRequest, Session, SessionRequest, SessionToken, Subject,
};
use serde::{Deserialize, Serialize};
use crate::error::{Error, Result};
use crate::event::Envelope;
use crate::filter::EventFilter;
const ONEVCS: &str = "onevcs";
fn sibling(message: impl Into<String>) -> Error {
Error::Sibling {
tool: ONEVCS,
message: message.into(),
}
}
fn refusal(error: onevcs::Error) -> Error {
sibling(error.to_string())
}
fn providers() -> Providers<'static> {
Providers::real()
}
const SESSION_OPEN_CONFLICT: &str = "the base conflicts with this branch";
pub fn session_open(request: &SessionRequest) -> Result<Session> {
providers()
.vcs
.open_session(request.clone())
.map_err(session_refusal)
}
fn session_refusal(error: onevcs::Error) -> Error {
match &error {
onevcs::Error::SyncConflict { .. } => sibling(format!("{SESSION_OPEN_CONFLICT}: {error}")),
_ => refusal(error),
}
}
pub(crate) fn session_open_conflicted(error: &Error) -> bool {
matches!(
error,
Error::Sibling { tool, message }
if *tool == ONEVCS && message.starts_with(SESSION_OPEN_CONFLICT)
)
}
pub fn publish(
token: &SessionToken,
policy: Option<MergePolicy>,
title: Option<&str>,
body: Option<&str>,
draft: Option<&DraftReason>,
) -> Result<Publication> {
let title = title
.map(|title| title.parse::<Subject>().map_err(sibling))
.transpose()?;
if let Some(reason) = draft {
reason.checked().map_err(refusal)?;
}
onevcs::publish(
&providers(),
token,
&PublishRequest {
policy,
title,
body: body.map(str::to_owned),
draft: draft.cloned(),
},
)
.map_err(refusal)
}
pub fn session_change(token: &SessionToken) -> Result<Option<onevcs::SessionChange>> {
onevcs::session_change(&providers(), token).map_err(refusal)
}
pub fn describe_change(
token: &SessionToken,
title: Option<&str>,
body: &str,
) -> Result<onevcs::SessionChange> {
let title = title
.map(|title| title.parse::<Subject>().map_err(sibling))
.transpose()?;
onevcs::describe_change(
&providers(),
token,
&onevcs::ChangeDescription {
title,
body: body.to_owned(),
},
)
.map_err(refusal)
}
pub const DRAFTED: &str = "change-draft";
pub fn outcome_of(outcome: &PublishOutcome) -> &'static str {
match outcome {
PublishOutcome::Merged(_) => "merged",
PublishOutcome::ChangeOpen(_) => "change-open",
PublishOutcome::ChangeDraft(_) => DRAFTED,
PublishOutcome::Queued(_) => "queued",
PublishOutcome::NothingToPublish => "no-changes",
PublishOutcome::Failed { kind, .. } => failure_of(*kind).outcome(),
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Preserving {
ChecksFailed,
ChecksUnsettled,
PushRejected,
SyncConflict,
}
impl Preserving {
#[must_use]
pub fn outcome(self) -> &'static str {
match self {
Self::ChecksFailed => "checks-failed",
Self::ChecksUnsettled => "checks-unsettled",
Self::PushRejected => "push-rejected",
Self::SyncConflict => "sync-conflict",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Failure {
Preserving(Preserving),
Unread,
Terminal,
}
impl Failure {
pub const RESIDUAL: &'static str = "publication-failed";
pub const UNREAD: &'static str = "pushed-unverified";
#[must_use]
pub fn outcome(self) -> &'static str {
match self {
Self::Preserving(preserving) => preserving.outcome(),
Self::Unread => Self::UNREAD,
Self::Terminal => Self::RESIDUAL,
}
}
}
pub fn failure_of(kind: onevcs::FailureKind) -> Failure {
use onevcs::FailureKind;
match kind {
FailureKind::ChecksFailed => Failure::Preserving(Preserving::ChecksFailed),
FailureKind::ChecksUnsettled => Failure::Preserving(Preserving::ChecksUnsettled),
FailureKind::PushRejected => Failure::Preserving(Preserving::PushRejected),
FailureKind::SyncConflict => Failure::Preserving(Preserving::SyncConflict),
FailureKind::PushedUnverified => Failure::Unread,
FailureKind::Gate | FailureKind::Invalid | FailureKind::NotImplemented => Failure::Terminal,
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Evidence {
pub kind: crate::event::EventKind,
pub id: crate::event::ArtifactId,
}
pub fn evidence_in(token: &SessionToken) -> Vec<Evidence> {
let mut evidence: Vec<Evidence> = Vec::new();
for envelope in events(token, None) {
for artifact in envelope.artifacts {
let found = Evidence {
kind: envelope.kind.clone(),
id: artifact.id,
};
if !evidence.iter().any(|kept| kept.id == found.id) {
evidence.push(found);
}
}
}
evidence
}
pub fn landing_of(outcome: &PublishOutcome) -> Option<crate::graph::Landing> {
use crate::graph::Landing;
match outcome {
PublishOutcome::Merged(_) => Some(Landing::Landed),
PublishOutcome::ChangeOpen(_)
| PublishOutcome::ChangeDraft(_)
| PublishOutcome::Queued(_) => Some(Landing::Unlanded),
PublishOutcome::NothingToPublish | PublishOutcome::Failed { .. } => None,
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum LandingRead {
Answered(onevcs::Landed),
Refused {
because: String,
},
}
pub(crate) fn landing_now(branch: &str, repo: Option<&str>) -> LandingRead {
crate::rendercost::landing_read_taken(branch, repo);
let answered = read_of(onevcs::landing_status(branch, repo));
crate::rendercost::repository_asked(repo);
answered
}
fn read_of(answered: onevcs::Result<onevcs::Landed>) -> LandingRead {
match answered {
Ok(landed) => LandingRead::Answered(landed),
Err(refused) => LandingRead::Refused {
because: crate::views::one_line(&format!("this host could not decide it: {refused}")),
},
}
}
pub(crate) fn proved_landed(branch: &str, repo: Option<&str>) -> bool {
matches!(
landing_now(branch, repo),
LandingRead::Answered(landed) if landed.is_landed()
)
}
#[derive(Debug, Default)]
pub(crate) struct UnreadMergePaths {
asking: BTreeMap<crate::graph::NodeRef, Asking>,
}
const HOLD_ASKS: std::num::NonZeroU32 = match std::num::NonZeroU32::new(12) {
Some(asks) => asks,
None => unreachable!(),
};
#[derive(Debug, Clone, Copy)]
enum Asking {
Unanswered(std::num::NonZeroU32),
Landed(std::num::NonZeroU32),
}
impl Asking {
fn asks(self) -> std::num::NonZeroU32 {
match self {
Self::Unanswered(asks) | Self::Landed(asks) => asks,
}
}
}
impl UnreadMergePaths {
pub(crate) fn every(&self) -> Duration {
crate::engine::merge_path_backoff()
}
pub(crate) fn watching(
&self,
state: &crate::projection::RunState,
statuses: &BTreeMap<String, crate::graph::NodeStatus>,
) -> Vec<crate::graph::NodeRef> {
let budget = HOLD_ASKS.get();
state
.graph
.iter()
.filter_map(crate::graph::NodeRef::of)
.filter(|node| {
statuses.get(node.as_str()) == Some(&crate::graph::NodeStatus::Failed)
&& state.outcomes.get(node.as_str()).map(String::as_str)
== Some(Failure::UNREAD)
&& match self.asking.get(node) {
None => true,
Some(Asking::Unanswered(asks)) => asks.get() < budget,
Some(Asking::Landed(_)) => false,
}
})
.filter(|node| {
state
.graph
.dependents_of(node.as_str())
.iter()
.any(|dependent| {
statuses.get(dependent) == Some(&crate::graph::NodeStatus::Blocked)
})
})
.collect()
}
pub(crate) fn apply(&self, state: &mut crate::projection::RunState) -> bool {
let mut restored = false;
for (node, _) in self
.asking
.iter()
.filter(|(_, asking)| matches!(asking, Asking::Landed(_)))
{
let already = state
.landings
.insert(node.as_str().to_owned(), crate::graph::Landing::Landed);
restored |= already != Some(crate::graph::Landing::Landed);
}
restored
}
pub(crate) fn read_again(
&mut self,
state: &mut crate::projection::RunState,
watching: &[crate::graph::NodeRef],
) -> bool {
let mut lifted = false;
for node in watching {
let id = node.as_str();
let asks = self
.asking
.get(node)
.map_or(std::num::NonZeroU32::MIN, |asking| {
asking.asks().saturating_add(1)
});
self.asking.insert(node.clone(), Asking::Unanswered(asks));
let Some(branch) = state.branches.get(id).cloned() else {
continue;
};
let repo = state.graph.get(id).and_then(|node| node.repo.clone());
if proved_landed(&branch, repo.as_deref()) {
self.asking.insert(node.clone(), Asking::Landed(asks));
lifted = true;
}
}
let _ = self.apply(state);
lifted
}
}
pub fn change_url(outcome: &PublishOutcome) -> Option<String> {
match outcome {
PublishOutcome::ChangeOpen(url)
| PublishOutcome::ChangeDraft(url)
| PublishOutcome::Queued(url) => Some(url.to_string()),
_ => None,
}
}
pub fn working_session(token: &SessionToken) -> Option<Session> {
onevcs::session(&providers(), token)
.map(|record| record.session)
.map_err(|error| {
eprintln!(
"onepipeline: cannot read session {}'s record: {error}",
token.0
);
error
})
.ok()
}
pub fn session_close(token: &SessionToken) -> Result<Session> {
onevcs::close_session(&providers(), token).map_err(refusal)
}
pub fn branch_head_in(token: &SessionToken) -> Option<String> {
match session_tip(token) {
SessionTip::At(commit) => Some(commit.as_str().to_owned()),
SessionTip::Unmoved | SessionTip::Unknown => None,
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LevelBranch {
pub branch: String,
pub base: String,
pub wrote: Wrote,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Wrote {
Nothing,
ACommitTheBaseCarries,
}
fn reflog_entry_wrote_a_commit(subject: &str) -> bool {
let subject = subject.trim();
if subject.starts_with("commit")
|| subject.starts_with("cherry-pick")
|| subject.starts_with("am:")
{
return true;
}
if let Some(rest) = subject.strip_prefix("rebase") {
return !["(start)", "(finish)", "(abort)"]
.iter()
.any(|phase| rest.trim_start().starts_with(phase));
}
(subject.starts_with("merge") || subject.starts_with("pull"))
&& !subject.ends_with("Fast-forward")
}
pub fn wait_out_the_second(began: std::time::SystemTime) {
let Ok(since) = began.duration_since(std::time::UNIX_EPOCH) else {
return;
};
let Some(next) =
std::time::UNIX_EPOCH.checked_add(Duration::from_secs(since.as_secs().saturating_add(1)))
else {
return;
};
while let Ok(remaining) = next.duration_since(std::time::SystemTime::now()) {
if remaining.is_zero() {
break;
}
std::thread::sleep(remaining);
}
}
pub fn level_with_base(
worktree: &std::path::Path,
base: &str,
began: std::time::SystemTime,
) -> Option<LevelBranch> {
let git = |args: &[&str]| -> Option<String> {
let output = std::process::Command::new("git")
.args(args)
.current_dir(worktree)
.stdin(std::process::Stdio::null())
.output()
.map_err(|error| {
eprintln!(
"onepipeline: cannot run `git {}` in {}: {error}",
args.join(" "),
worktree.display()
);
})
.ok()?;
if !output.status.success() {
eprintln!(
"onepipeline: `git {}` in {} exited {}: {}",
args.join(" "),
worktree.display(),
output.status.code().unwrap_or(-1),
crate::views::one_line(&String::from_utf8_lossy(&output.stderr))
);
return None;
}
Some(String::from_utf8_lossy(&output.stdout).trim().to_owned())
};
if !git(&["status", "--porcelain"])?.is_empty() {
return None;
}
let remote = format!("origin/{base}");
let carried = git(&[
"for-each-ref",
"--format=%(refname)",
&format!("refs/remotes/{remote}"),
])?;
let compared = if carried.is_empty() {
base.to_owned()
} else {
remote
};
let ahead = git(&["rev-list", "--count", &format!("{compared}..HEAD")])?;
if ahead.parse::<u64>().ok()? != 0 {
return None;
}
let began = began
.duration_since(std::time::UNIX_EPOCH)
.map(|since| since.as_secs())
.unwrap_or(0);
let written: Vec<String> = git(&["log", "-g", "--date=unix", "--format=%gd %H %gs", "HEAD"])?
.lines()
.filter_map(|line| {
let (stamp, rest) = line.split_once(' ')?;
let (sha, subject) = rest.split_once(' ')?;
let stamped: u64 = stamp
.strip_prefix("HEAD@{")?
.strip_suffix('}')?
.parse()
.ok()?;
(stamped > began && reflog_entry_wrote_a_commit(subject)).then(|| sha.to_owned())
})
.collect();
let mut wrote = Wrote::Nothing;
for sha in &written {
match std::process::Command::new("git")
.args(["merge-base", "--is-ancestor", sha, "HEAD"])
.current_dir(worktree)
.stdin(std::process::Stdio::null())
.output()
{
Ok(answered) if answered.status.success() => {
wrote = Wrote::ACommitTheBaseCarries;
break;
}
Ok(answered) if answered.status.code() == Some(1) => {}
_ => return None,
}
}
let branch = git(&["rev-parse", "--abbrev-ref", "HEAD"])?;
Some(LevelBranch {
branch,
base: compared,
wrote,
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Commit(String);
impl Commit {
pub(crate) fn of(sha: &str) -> Option<Self> {
usable(sha).map(Self)
}
pub(crate) fn as_str(&self) -> &str {
&self.0
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SessionTip {
At(Commit),
Unmoved,
Unknown,
}
pub fn session_tip(token: &SessionToken) -> SessionTip {
let Some(mut stream) = opened(token, None) else {
return SessionTip::Unknown;
};
let batch = match stream.read() {
Ok(batch) => batch,
Err(error) => {
eprintln!(
"onepipeline: cannot read session {}'s events: {error}",
token.0
);
return SessionTip::Unknown;
}
};
let preserved = kind_of(onevcs::EventKind::CommitPreserved);
let Some(envelope) = batch
.into_iter()
.rev()
.map(relayed)
.find(|envelope| envelope.kind == preserved)
else {
return SessionTip::Unmoved;
};
envelope
.payload
.get("sha")
.and_then(|sha| sha.as_str())
.and_then(Commit::of)
.map_or(SessionTip::Unknown, SessionTip::At)
}
pub fn change_opened_in(token: &SessionToken) -> Option<String> {
let opened = kind_of(onevcs::EventKind::ChangeOpened);
events(token, None)
.iter()
.rev()
.find(|envelope| envelope.kind == opened)
.and_then(|envelope| envelope.payload.get("url"))
.and_then(|url| url.as_str())
.and_then(|url| onevcs::Url::parse(url.trim()).ok())
.map(|url| url.to_string())
}
pub fn change_drafted_in(token: &SessionToken) -> bool {
let drafted = kind_of(onevcs::EventKind::ChangeDrafted);
events(token, None)
.iter()
.any(|envelope| envelope.kind == drafted)
}
pub fn holders_of(repo: &str) -> std::result::Result<Vec<onevcs::SessionHolder>, String> {
onevcs::session_holders(repo).map_err(|error| {
eprintln!("onepipeline: cannot read the session holders of {repo}: {error}");
error.to_string()
})
}
fn opened(token: &SessionToken, filter: Option<&EventFilter>) -> Option<EventStream> {
let filter = match filter.map(sibling_filter).transpose() {
Ok(filter) => filter.unwrap_or_default(),
Err(error) => {
eprintln!(
"onepipeline: cannot follow session {}'s events: {error}",
token.0
);
return None;
}
};
match EventStream::open_filtered(token, filter) {
Ok(stream) => Some(stream),
Err(error) => {
eprintln!(
"onepipeline: cannot read session {}'s events: {error}",
token.0
);
None
}
}
}
fn next_batch(stream: &mut EventStream, token: &SessionToken) -> Vec<Envelope> {
match stream.read() {
Ok(events) => events.into_iter().map(relayed).collect(),
Err(error) => {
eprintln!(
"onepipeline: cannot read session {}'s events: {error}",
token.0
);
Vec::new()
}
}
}
pub fn events(token: &SessionToken, filter: Option<&EventFilter>) -> Vec<Envelope> {
let Some(mut stream) = opened(token, filter) else {
return Vec::new();
};
next_batch(&mut stream, token)
}
fn sibling_filter(filter: &EventFilter) -> Result<onevcs::EventFilter> {
let document = serde_json::to_string(filter).map_err(|error| Error::Sibling {
tool: "onevcs",
message: format!("rendering the event filter: {error}"),
})?;
serde_json::from_str(&document).map_err(|error| Error::Sibling {
tool: "onevcs",
message: format!("`onevcs` refused the event filter: {error}"),
})
}
fn relayed(envelope: onevcs::Envelope) -> Envelope {
Envelope {
v: envelope.v,
ts: envelope.ts,
stream: envelope.stream,
seq: envelope.seq,
source: source_of(envelope.source),
kind: kind_of(envelope.kind),
phase: Some(phase_of(envelope.phase)),
labels: labels_of(envelope.labels),
payload: envelope.payload,
artifacts: envelope
.artifacts
.into_iter()
.map(|artifact| crate::event::ArtifactRef {
id: crate::event::ArtifactId(artifact.id.0),
kind: artifact.kind,
bytes: artifact.bytes,
})
.collect(),
}
}
fn phase_of(phase: onevcs::Phase) -> crate::event::Phase {
match phase {
onevcs::Phase::Development => crate::event::Phase::Development,
onevcs::Phase::Integrate => crate::event::Phase::Integrate,
onevcs::Phase::Review => crate::event::Phase::Review,
onevcs::Phase::Release => crate::event::Phase::Release,
}
}
fn phase_of_kind(kind: onevcs::EventKind) -> Option<crate::event::Phase> {
onevcs::Phase::of(kind).map(phase_of)
}
fn source_of(source: onevcs::Source) -> crate::event::Source {
match source {
onevcs::Source::Agentgraph => crate::event::Source::Agentgraph,
onevcs::Source::Vcs => crate::event::Source::Vcs,
onevcs::Source::Pipeline => crate::event::Source::Pipeline,
}
}
fn kind_of(kind: onevcs::EventKind) -> crate::event::EventKind {
let wire = serde_json::to_value(kind)
.ok()
.and_then(|value| value.as_str().map(str::to_string))
.unwrap_or_else(|| format!("{kind:?}"));
crate::event::EventKind(wire)
}
fn labels_of(labels: onevcs::Labels) -> crate::event::Labels {
let mut extra = labels.extra;
if let Some(member) = labels.member {
extra.insert("member".to_owned(), serde_json::json!(member));
}
crate::event::Labels {
run_id: labels.run_id,
round: labels.round,
node: labels.node,
step: labels.step,
persona: labels.persona,
extra,
}
}
const FOLLOW_POLL: Duration = Duration::from_millis(20);
pub fn follow(
token: &SessionToken,
filter: Option<&EventFilter>,
sink: Box<dyn Fn(Envelope) + Send>,
) -> Option<Follower> {
let mut stream = opened(token, filter)?;
let progress: Arc<Mutex<Watermarks>> = Arc::new(Mutex::new(Watermarks::default()));
let reached = Arc::clone(&progress);
let stop = Arc::new(AtomicBool::new(false));
let stopping = Arc::clone(&stop);
let followed = token.clone();
let reader = std::thread::Builder::new()
.name(format!("onevcs-{}-events", token.0))
.spawn(move || loop {
for envelope in next_batch(&mut stream, &followed) {
if let Ok(mut reached) = reached.lock() {
reached.reached(&envelope);
}
sink(envelope);
}
if stopping.load(Ordering::SeqCst) || settled(&followed) {
return;
}
std::thread::sleep(FOLLOW_POLL);
});
match reader {
Ok(reader) => Some(Follower {
reader: Some(reader),
stop,
progress,
}),
Err(error) => {
eprintln!(
"onepipeline: cannot follow session {}'s events: {error}",
token.0
);
None
}
}
}
fn settled(session: &SessionToken) -> bool {
onevcs::session(&providers(), session)
.map(|record| record.lifecycle == Lifecycle::Closed)
.unwrap_or(true)
}
#[derive(Debug, Default, Clone)]
pub struct Watermarks {
reached: BTreeMap<String, u64>,
}
impl Watermarks {
pub fn of_relayed(envelopes: &[Envelope]) -> Self {
let mut marks = Self::default();
for envelope in envelopes {
marks.reached(envelope);
}
marks
}
pub fn reached(&mut self, envelope: &Envelope) {
self.reached
.entry(envelope.stream.clone())
.and_modify(|seq| *seq = (*seq).max(envelope.seq))
.or_insert(envelope.seq);
}
#[must_use]
pub fn beyond(&self, envelope: &Envelope) -> bool {
self.reached
.get(&envelope.stream)
.is_none_or(|seq| envelope.seq > *seq)
}
}
#[must_use]
pub fn is_terminator(envelope: &Envelope) -> bool {
envelope.kind == kind_of(onevcs::EventKind::SessionClosed)
}
#[derive(Debug)]
pub struct Follower {
reader: Option<std::thread::JoinHandle<()>>,
stop: Arc<AtomicBool>,
progress: Arc<Mutex<Watermarks>>,
}
impl Drop for Follower {
fn drop(&mut self) {
self.stop.store(true, Ordering::SeqCst);
if let Some(reader) = self.reader.take() {
let _ = reader.join();
}
}
}
impl Follower {
pub fn finish(mut self) -> Watermarks {
self.stop.store(true, Ordering::SeqCst);
if let Some(reader) = self.reader.take() {
let _ = reader.join();
}
let mut reached = self
.progress
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
std::mem::take(&mut *reached)
}
}
pub fn session_opened_event(session: &Session, labels: &crate::event::Labels) -> Envelope {
Envelope {
v: crate::event::ENVELOPE_VERSION,
ts: crate::sys::now_rfc3339(),
stream: format!("onevcs-{}", session.token.0),
seq: 0,
source: crate::event::Source::Vcs,
kind: kind_of(onevcs::EventKind::SessionOpened),
phase: phase_of_kind(onevcs::EventKind::SessionOpened),
labels: labels.clone(),
payload: crate::journal::payload(&[
("token", serde_json::json!(session.token.0)),
("branch", serde_json::json!(session.branch)),
("base", serde_json::json!(session.base)),
("worktree", serde_json::json!(session.worktree)),
]),
artifacts: Vec::new(),
}
}
pub fn published_event(published: &Publication, labels: &crate::event::Labels) -> Envelope {
Envelope {
v: crate::event::ENVELOPE_VERSION,
ts: crate::sys::now_rfc3339(),
stream: format!("onevcs-{}", published.branch),
seq: 1,
source: crate::event::Source::Vcs,
kind: crate::event::EventKind("published".into()),
phase: None,
labels: labels.clone(),
payload: crate::journal::payload(&[
("branch", serde_json::json!(published.branch)),
("policy", serde_json::json!(published.policy)),
("outcome", serde_json::json!(outcome_of(&published.outcome))),
("url", serde_json::json!(change_url(&published.outcome))),
(
"landing",
serde_json::json!(landing_of(&published.outcome).map(crate::graph::Landing::as_str)),
),
]),
artifacts: Vec::new(),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DispatchSession {
token: SessionToken,
branch: BranchName,
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct SessionAsWritten {
token: String,
branch: String,
}
impl Serialize for DispatchSession {
fn serialize<S: serde::Serializer>(&self, writer: S) -> std::result::Result<S::Ok, S::Error> {
SessionAsWritten {
token: self.token.0.clone(),
branch: self.branch.0.clone(),
}
.serialize(writer)
}
}
impl<'de> Deserialize<'de> for DispatchSession {
fn deserialize<D: serde::Deserializer<'de>>(reader: D) -> std::result::Result<Self, D::Error> {
let written = SessionAsWritten::deserialize(reader)?;
let token = token_of(&written.token).ok_or_else(|| {
serde::de::Error::custom(format!("'{}' is no session handle", written.token))
})?;
let branch = BranchName::checked(&written.branch).ok_or_else(|| {
serde::de::Error::custom(format!("'{}' is no branch name", written.branch))
})?;
Ok(Self { token, branch })
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BranchName(String);
impl BranchName {
pub fn checked(value: &str) -> Option<Self> {
usable(value).map(Self)
}
pub fn as_str(&self) -> &str {
&self.0
}
}
impl std::fmt::Display for BranchName {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.0)
}
}
impl DispatchSession {
pub fn read_from(envelope: &Envelope) -> Option<Self> {
let session: Session =
serde_json::from_value(serde_json::Value::Object(envelope.payload.clone())).ok()?;
let token = token_of(&session.token.0)?;
if !wrote(&envelope.stream, &token) {
return None;
}
Some(Self {
token,
branch: BranchName::checked(&session.branch)?,
})
}
pub fn token(&self) -> &SessionToken {
&self.token
}
pub fn branch(&self) -> &BranchName {
&self.branch
}
}
fn wrote(stream: &str, token: &SessionToken) -> bool {
stream == token.0 || stream == format!("onevcs-{}", token.0)
}
fn token_of(value: &str) -> Option<SessionToken> {
let value = usable(value)?;
if value.contains(['/', '\\']) || value.trim_matches('.').is_empty() {
return None;
}
Some(SessionToken(value))
}
pub(crate) fn usable(value: &str) -> Option<String> {
if value.is_empty() || value.len() >= crate::event::MAX_PAYLOAD_TEXT_BYTES {
return None;
}
if value.chars().any(|c| c.is_whitespace() || c.is_control()) {
return None;
}
Some(value.to_owned())
}
pub fn is_session_opened(kind: &crate::event::EventKind) -> bool {
*kind == session_opened_kind()
}
pub(crate) fn session_opened_kind() -> crate::event::EventKind {
kind_of(onevcs::EventKind::SessionOpened)
}
pub fn is_merge_completed(kind: &crate::event::EventKind) -> bool {
*kind == kind_of(onevcs::EventKind::MergeCompleted)
}
pub fn landing_commit_of(event: &crate::event::Envelope) -> Option<String> {
if event.source != crate::event::Source::Vcs || !is_merge_completed(&event.kind) {
return None;
}
usable(event.payload.get("sha")?.as_str()?)
}
pub fn request_for(node: &crate::plan::Node) -> Option<SessionRequest> {
Some(SessionRequest {
repo: node.repo.clone()?,
branch: node.branch.clone(),
base: node.base_branch.clone(),
execution_checkout: node.execution_checkout.clone(),
})
}
#[cfg(test)]
pub(crate) fn scratch_home_held() -> std::sync::MutexGuard<'static, ()> {
static IN_USE: Mutex<()> = Mutex::new(());
IN_USE
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
#[cfg(test)]
mod tests {
use std::collections::BTreeSet;
use super::*;
use crate::plan::Node;
const EVERY_KIND: &[onevcs::FailureKind] = &[
onevcs::FailureKind::Gate,
onevcs::FailureKind::Invalid,
onevcs::FailureKind::SyncConflict,
onevcs::FailureKind::NotImplemented,
onevcs::FailureKind::ChecksFailed,
onevcs::FailureKind::ChecksUnsettled,
onevcs::FailureKind::PushRejected,
onevcs::FailureKind::PushedUnverified,
];
const EVERY_PRESERVING: &[Preserving] = &[
Preserving::ChecksFailed,
Preserving::ChecksUnsettled,
Preserving::PushRejected,
Preserving::SyncConflict,
];
#[test]
fn the_publication_failure_words_and_the_contract_are_one_vocabulary() {
let contract = std::fs::read_to_string(
std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("docs/contract.md"),
)
.expect("the contract ships");
for kind in EVERY_KIND {
let word = failure_of(*kind).outcome();
assert!(
contract.contains(&format!("`{word}`")),
"docs/contract.md does not name the `{word}` outcome this crate settles on"
);
}
let clause = contract
.split_once("under a word of its own:")
.expect("the contract lists the words a failed publication settles on")
.1
.split_once("is the **residual**")
.expect("the clause ends where the residual is named")
.0;
let listed: BTreeSet<&str> = clause.split('`').skip(1).step_by(2).collect();
let vocabulary: BTreeSet<&str> = EVERY_PRESERVING
.iter()
.map(|preserving| preserving.outcome())
.chain([Failure::UNREAD, Failure::RESIDUAL])
.collect();
let produced: BTreeSet<&str> = EVERY_KIND
.iter()
.map(|kind| failure_of(*kind).outcome())
.collect();
assert_eq!(
produced, vocabulary,
"the words `failure_of` settles on are not the vocabulary `Preserving` closes"
);
assert_eq!(
listed, vocabulary,
"the contract's publication-failure words are not the ones this crate settles on"
);
}
#[test]
fn the_readmes_publication_failure_summary_is_the_vocabulary_this_crate_settles_on() {
let raw = std::fs::read_to_string(
std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("README.md"),
)
.expect("the README ships");
let readme = raw.split_whitespace().collect::<Vec<_>>().join(" ");
for kind in EVERY_KIND {
let word = failure_of(*kind).outcome();
assert!(
readme.contains(&format!("`{word}`")),
"the README does not name the `{word}` outcome this crate settles on"
);
}
let clause = readme
.split_once("settle under a word of their own")
.expect("the README names the failures that settle under a word of their own")
.1
.split_once("The first four leave the rejected tree")
.expect("that clause ends where the README says what those failures share")
.0;
let listed: BTreeSet<&str> = clause.split('`').skip(1).step_by(2).collect();
let routed: BTreeSet<&str> = EVERY_PRESERVING
.iter()
.map(|preserving| preserving.outcome())
.chain(std::iter::once(Failure::UNREAD))
.collect();
assert_eq!(
listed, routed,
"the README's named failures are not the ones this crate gives a word of its own"
);
assert!(
readme.contains(&format!("`{}` is answered differently", Failure::UNREAD)),
"the README does not say that `{}` is recovered by a re-read",
Failure::UNREAD
);
assert!(
readme.contains(crate::engine::MERGE_PATH_READS_ENV),
"the README does not name what bounds that re-read"
);
assert!(
readme.contains(&format!("Everything else settles `{}`", Failure::RESIDUAL)),
"the README does not name `{}` as what everything else settles on",
Failure::RESIDUAL
);
}
#[test]
fn each_kind_is_on_the_side_of_the_line_the_contract_puts_it() {
let terminal = [
onevcs::FailureKind::Invalid,
onevcs::FailureKind::NotImplemented,
onevcs::FailureKind::Gate,
];
for kind in terminal {
assert_eq!(
failure_of(kind),
Failure::Terminal,
"{kind:?} is retried, and asking again would reproduce the diagnosis"
);
}
assert_eq!(
failure_of(onevcs::FailureKind::PushedUnverified),
Failure::Unread,
"a push that reached the remote is answered by re-dispatching the agent"
);
let preserving: BTreeSet<&str> = EVERY_KIND
.iter()
.filter(|kind| !terminal.contains(kind))
.filter(|kind| **kind != onevcs::FailureKind::PushedUnverified)
.map(|kind| match failure_of(*kind) {
Failure::Preserving(preserving) => preserving.outcome(),
Failure::Unread | Failure::Terminal => {
panic!("{kind:?} is not the tree being rejected, so nothing re-dispatches it")
}
})
.collect();
assert_eq!(
preserving,
BTreeSet::from([
"checks-failed",
"checks-unsettled",
"push-rejected",
"sync-conflict"
]),
"the failures a further attempt can answer are not the four the contract names"
);
assert_eq!(Failure::Terminal.outcome(), Failure::RESIDUAL);
assert_eq!(Failure::Unread.outcome(), "pushed-unverified");
}
fn ours(token: &str, branch: &str) -> Envelope {
session_opened_event(
&Session {
token: SessionToken(token.to_owned()),
worktree: std::path::PathBuf::from("/tmp/worktree"),
branch: branch.to_owned(),
base: "main".to_owned(),
},
&crate::event::Labels::default(),
)
}
#[test]
fn a_session_record_is_read_only_where_every_value_it_names_is_usable() {
let read = DispatchSession::read_from(&ours("s-abc", "onevcs/s-abc"))
.expect("a whole session record is read");
assert_eq!(read.token(), &SessionToken("s-abc".into()));
assert_eq!(read.branch().as_str(), "onevcs/s-abc");
let mut theirs = ours("s-abc", "onevcs/s-abc");
theirs.stream = "s-abc".to_owned();
for (key, value) in [
("identity", serde_json::json!("github.com/owner/service")),
("clone", serde_json::json!("/tmp/runs/s-abc/clone")),
("execution_checkout", serde_json::json!("/tmp/service")),
("publication_checkout", serde_json::json!("/tmp/service")),
("reused", serde_json::json!(true)),
] {
theirs.payload.insert(key.to_owned(), value);
}
assert_eq!(
DispatchSession::read_from(&theirs).as_ref(),
Some(&read),
"the sibling's own record of a session it opened was not read"
);
let without = |key: &str| {
let mut event = ours("s-abc", "onevcs/s-abc");
event.payload.remove(key);
event
};
let mut elsewhere = ours("s-elsewhere", "onevcs/s-elsewhere");
elsewhere.stream = "onevcs-s-abc".to_owned();
for (why, event) in [
("a record naming no branch at all", without("branch")),
("a record naming no token at all", without("token")),
("a record about another session entirely", elsewhere),
] {
assert_eq!(
DispatchSession::read_from(&event),
None,
"{why} was read as a session a manager can be sent to"
);
}
for (why, value) in unusable() {
assert_eq!(
DispatchSession::read_from(&ours("s-abc", &value)),
None,
"a branch that is {why} was read as one a manager can be sent to"
);
}
let hops = [
("a directory hop", "..".to_owned()),
("carrying a path separator", "onevcs/../x".to_owned()),
];
for (why, value) in unusable().into_iter().chain(hops) {
let mut record = ours(&value, "onevcs/s-abc");
record.stream = value.clone();
assert_eq!(
DispatchSession::read_from(&record),
None,
"a token that is {why} was read as one a session answers to"
);
}
}
fn unusable() -> Vec<(&'static str, String)> {
vec![
("empty", String::new()),
(
"a line of its own",
"onevcs/x\n audit running".to_owned(),
),
("carrying a space", "onevcs/ x".to_owned()),
("carrying a control character", "onevcs/x\u{7}".to_owned()),
(
"as long as the bound a producer cuts text at",
"b".repeat(crate::event::MAX_PAYLOAD_TEXT_BYTES),
),
]
}
#[test]
fn a_lifecycle_node_asks_for_the_session_its_fields_describe() {
let node = Node {
id: "service".into(),
repo: Some("owner/repo".into()),
branch: Some("feature".into()),
base_branch: Some("main".into()),
execution_checkout: Some("primary".into()),
persona: Some("engineer".into()),
task: Some("## What\nship".into()),
..Node::default()
};
let request = request_for(&node).expect("a lifecycle node asks for a session");
assert_eq!(request.repo, "owner/repo");
assert_eq!(request.branch.as_deref(), Some("feature"));
assert_eq!(request.base.as_deref(), Some("main"));
assert_eq!(request.execution_checkout.as_deref(), Some("primary"));
}
#[test]
fn a_direct_agent_node_asks_for_no_session() {
let node = Node {
id: "build".into(),
persona: Some("engineer".into()),
task: Some("## What\ndo it".into()),
..Node::default()
};
assert!(request_for(&node).is_none());
}
#[test]
fn every_ending_a_publication_has_settles_the_node_under_its_own_name() {
let sha = onevcs::Sha("abc".into());
let url: onevcs::Url = "https://example.invalid/pull/7".parse().expect("a URL");
assert_eq!(outcome_of(&PublishOutcome::Merged(sha)), "merged");
assert_eq!(
outcome_of(&PublishOutcome::ChangeOpen(url.clone())),
"change-open"
);
assert_eq!(
outcome_of(&PublishOutcome::ChangeDraft(url.clone())),
"change-draft"
);
assert_eq!(outcome_of(&PublishOutcome::Queued(url)), "queued");
assert_eq!(outcome_of(&PublishOutcome::NothingToPublish), "no-changes");
let failed = |kind| {
outcome_of(&PublishOutcome::Failed {
kind,
reason: "the publication said no".into(),
retained: None,
})
};
assert_eq!(failed(onevcs::FailureKind::Gate), "publication-failed");
assert_eq!(failed(onevcs::FailureKind::Invalid), "publication-failed");
assert_eq!(
failed(onevcs::FailureKind::NotImplemented),
"publication-failed"
);
assert_eq!(failed(onevcs::FailureKind::ChecksFailed), "checks-failed");
assert_eq!(
failed(onevcs::FailureKind::ChecksUnsettled),
"checks-unsettled"
);
assert_eq!(failed(onevcs::FailureKind::PushRejected), "push-rejected");
assert_eq!(failed(onevcs::FailureKind::SyncConflict), "sync-conflict");
assert_eq!(
failed(onevcs::FailureKind::PushedUnverified),
"pushed-unverified"
);
}
#[test]
fn only_a_change_observed_on_its_base_is_called_landed() {
use crate::graph::Landing;
let url: onevcs::Url = "https://example.invalid/pull/7".parse().expect("a URL");
assert_eq!(
landing_of(&PublishOutcome::Merged(onevcs::Sha("abc".into()))),
Some(Landing::Landed)
);
assert_eq!(
landing_of(&PublishOutcome::ChangeOpen(url.clone())),
Some(Landing::Unlanded)
);
assert_eq!(
landing_of(&PublishOutcome::ChangeDraft(url.clone())),
Some(Landing::Unlanded)
);
assert_eq!(
landing_of(&PublishOutcome::Queued(url)),
Some(Landing::Unlanded)
);
assert_eq!(landing_of(&PublishOutcome::NothingToPublish), None);
assert_eq!(
landing_of(&PublishOutcome::Failed {
kind: onevcs::FailureKind::PushRejected,
reason: "the merge path refused the publishing push".into(),
retained: None,
}),
None
);
}
#[test]
fn a_change_request_is_where_a_human_reads_it_and_a_local_merge_names_none() {
let url: onevcs::Url = "https://example.invalid/pull/7".parse().expect("a URL");
assert_eq!(
change_url(&PublishOutcome::ChangeOpen(url.clone())).as_deref(),
Some("https://example.invalid/pull/7")
);
assert_eq!(
change_url(&PublishOutcome::Queued(url)).as_deref(),
Some("https://example.invalid/pull/7")
);
assert_eq!(
change_url(&PublishOutcome::Merged(onevcs::Sha("abc".into()))),
None
);
assert_eq!(change_url(&PublishOutcome::NothingToPublish), None);
}
#[test]
fn a_publication_records_what_the_sibling_said_it_did() {
let url: onevcs::Url = "https://example.invalid/pull/7".parse().expect("a URL");
let published = Publication {
session: SessionToken("s-1".into()),
branch: "onepipeline/service".into(),
policy: MergePolicy::ChangeOpen,
outcome: PublishOutcome::ChangeOpen(url),
};
let event = published_event(&published, &crate::event::Labels::default());
assert_eq!(event.stream, "onevcs-onepipeline/service");
assert_eq!(event.payload["branch"], "onepipeline/service");
assert_eq!(event.payload["policy"], "change-open");
assert_eq!(event.payload["outcome"], "change-open");
assert_eq!(event.payload["url"], "https://example.invalid/pull/7");
assert_eq!(event.payload["landing"], "unlanded");
let merged = Publication {
outcome: PublishOutcome::Merged(onevcs::Sha("abc".into())),
..published
};
let event = published_event(&merged, &crate::event::Labels::default());
assert_eq!(event.payload["landing"], "landed");
let empty = Publication {
outcome: PublishOutcome::NothingToPublish,
..merged
};
let event = published_event(&empty, &crate::event::Labels::default());
assert_eq!(event.payload["landing"], serde_json::Value::Null);
}
#[test]
fn only_the_refusal_this_module_composed_reads_as_a_session_open_conflict() {
let refused = session_refusal(onevcs::Error::SyncConflict {
reason: "both sides changed README.md".into(),
});
assert!(
session_open_conflicted(&refused),
"the refusal this module composes for the conflict was not read as one: {refused}"
);
let invalid = session_refusal(onevcs::Error::Invalid {
reason: "no such base".into(),
});
assert!(
!session_open_conflicted(&invalid),
"a refusal no part of this is was read as the conflict: {invalid}"
);
assert!(!session_open_conflicted(&Error::Sibling {
tool: ONEVCS,
message: format!("the agent reported that {SESSION_OPEN_CONFLICT}"),
}));
assert!(!session_open_conflicted(&Error::Sibling {
tool: "oneagentgraph",
message: format!("{SESSION_OPEN_CONFLICT}: sync conflict"),
}));
assert!(!session_open_conflicted(&Error::Invalid(format!(
"{SESSION_OPEN_CONFLICT}: sync conflict"
))));
}
fn a_cut_worktree(name: &str) -> (std::path::PathBuf, std::path::PathBuf) {
let root =
std::env::temp_dir().join(format!("onepipeline-level-{name}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir_all(&root).expect("a scratch root");
let base = root.join("base");
let clone = root.join("clone");
let worktree = root.join("worktree");
git_in(&root, &["init", "-q", "--initial-branch=main", "base"]);
std::fs::write(base.join("README.md"), "seed\n").expect("the seed file");
git_in(&base, &["add", "-A"]);
git_in(&base, &["commit", "-q", "-m", "chore: seed"]);
git_in(
&base,
&["config", "receive.denyCurrentBranch", "updateInstead"],
);
git_in(
&root,
&["clone", "-q", "--shared", "--no-checkout", "base", "clone"],
);
git_in(
&clone,
&[
"worktree",
"add",
"-q",
"-b",
"work",
&worktree.to_string_lossy(),
"origin/main",
],
);
(base, worktree)
}
fn git_in(dir: &std::path::Path, args: &[&str]) {
let ran = std::process::Command::new("git")
.args(args)
.current_dir(dir)
.env("GIT_AUTHOR_NAME", "test")
.env("GIT_AUTHOR_EMAIL", "test@example.invalid")
.env("GIT_COMMITTER_NAME", "test")
.env("GIT_COMMITTER_EMAIL", "test@example.invalid")
.output()
.expect("git runs");
assert!(
ran.status.success(),
"`git {}` in {}: {}",
args.join(" "),
dir.display(),
String::from_utf8_lossy(&ran.stderr)
);
}
fn commit_in(worktree: &std::path::Path, name: &str) {
std::fs::write(worktree.join(name), format!("{name}\n")).expect("the file");
git_in(worktree, &["add", "-A"]);
git_in(worktree, &["commit", "-q", "-m", &format!("feat: {name}")]);
}
#[test]
fn a_level_branch_is_told_by_a_commit_written_here_and_not_by_movement_alone() {
use std::time::{Duration, SystemTime, UNIX_EPOCH};
let level =
|worktree: &std::path::Path| super::level_with_base(worktree, "main", UNIX_EPOCH);
let wrote = |worktree: &std::path::Path| level(worktree).map(|read| read.wrote);
let land = |worktree: &std::path::Path| {
git_in(worktree, &["push", "-q", "origin", "HEAD:refs/heads/main"]);
git_in(worktree, &["fetch", "-q", "origin"]);
};
let (_, fresh) = a_cut_worktree("fresh");
assert_eq!(
level(&fresh),
Some(LevelBranch {
branch: "work".into(),
base: "origin/main".into(),
wrote: Wrote::Nothing,
})
);
for (name, catch_up) in [
("ff", vec!["merge", "-q", "--ff-only", "origin/main"]),
("reset", vec!["reset", "-q", "--hard", "origin/main"]),
] {
let (base, worktree) = a_cut_worktree(name);
commit_in(&base, "landed-by-somebody-else.md");
git_in(&worktree, &["fetch", "-q", "origin"]);
git_in(&worktree, &catch_up);
assert_eq!(
wrote(&worktree),
Some(Wrote::Nothing),
"{name}: a worker that moved onto a commit the base already had was read as \
having committed it"
);
}
let (_, landed) = a_cut_worktree("landed");
commit_in(&landed, "mine.md");
land(&landed);
assert_eq!(wrote(&landed), Some(Wrote::ACommitTheBaseCarries), "commit");
let (base, picked) = a_cut_worktree("picked");
git_in(&base, &["checkout", "-q", "-b", "side"]);
commit_in(&base, "elsewhere.md");
git_in(&base, &["checkout", "-q", "main"]);
git_in(&picked, &["fetch", "-q", "origin"]);
git_in(&picked, &["cherry-pick", "-x", "origin/side"]);
assert_eq!(level(&picked), None, "a cherry-picked commit read as level");
land(&picked);
assert_eq!(
wrote(&picked),
Some(Wrote::ACommitTheBaseCarries),
"cherry-pick"
);
let (base, merged) = a_cut_worktree("merged");
commit_in(&merged, "mine.md");
commit_in(&base, "theirs.md");
git_in(&merged, &["fetch", "-q", "origin"]);
git_in(
&merged,
&["merge", "-q", "--no-ff", "-m", "merge", "origin/main"],
);
land(&merged);
assert_eq!(wrote(&merged), Some(Wrote::ACommitTheBaseCarries), "merge");
let (base, rebased) = a_cut_worktree("rebased");
commit_in(&rebased, "mine.md");
commit_in(&base, "theirs.md");
git_in(&rebased, &["fetch", "-q", "origin"]);
git_in(&rebased, &["rebase", "-q", "origin/main"]);
land(&rebased);
assert_eq!(
wrote(&rebased),
Some(Wrote::ACommitTheBaseCarries),
"rebase"
);
let (base, applied) = a_cut_worktree("applied");
git_in(&base, &["checkout", "-q", "-b", "side"]);
commit_in(&base, "patch.md");
let patch = std::process::Command::new("git")
.args(["format-patch", "-1", "--stdout", "side"])
.current_dir(&base)
.output()
.expect("git runs");
assert!(patch.status.success());
git_in(&base, &["checkout", "-q", "main"]);
std::fs::write(applied.join("../patch.mbox"), &patch.stdout).expect("the patch");
git_in(&applied, &["am", "-q", "../patch.mbox"]);
assert_eq!(level(&applied), None, "an applied patch read as level");
land(&applied);
assert_eq!(wrote(&applied), Some(Wrote::ACommitTheBaseCarries), "am");
let (_, undone) = a_cut_worktree("undone");
commit_in(&undone, "undone.md");
git_in(&undone, &["reset", "-q", "--hard", "origin/main"]);
assert_eq!(
wrote(&undone),
Some(Wrote::Nothing),
"a commit reset off the branch was read as one the base carries"
);
let (_, resumed) = a_cut_worktree("resumed");
std::fs::write(resumed.join("earlier.md"), "earlier\n").expect("the file");
git_in(&resumed, &["add", "-A"]);
let earlier = SystemTime::now() - Duration::from_secs(86_400);
let stamp = format!(
"@{} +0000",
earlier
.duration_since(UNIX_EPOCH)
.expect("after the epoch")
.as_secs()
);
let dated = std::process::Command::new("git")
.args(["commit", "-q", "-m", "feat: earlier"])
.current_dir(&resumed)
.env("GIT_AUTHOR_NAME", "test")
.env("GIT_AUTHOR_EMAIL", "test@example.invalid")
.env("GIT_COMMITTER_NAME", "test")
.env("GIT_COMMITTER_EMAIL", "test@example.invalid")
.env("GIT_AUTHOR_DATE", &stamp)
.env("GIT_COMMITTER_DATE", &stamp)
.status()
.expect("git runs");
assert!(dated.success());
land(&resumed);
assert_eq!(
wrote(&resumed),
Some(Wrote::ACommitTheBaseCarries),
"read as the dispatch that wrote it, the commit counts"
);
let begun = |began: SystemTime| {
super::level_with_base(&resumed, "main", began).map(|read| read.wrote)
};
assert_eq!(
begun(earlier - Duration::from_secs(1)),
Some(Wrote::ACommitTheBaseCarries),
"a commit stamped the second after this dispatch began was not read as its own"
);
assert_eq!(
begun(earlier),
Some(Wrote::Nothing),
"a commit stamped in the second this dispatch began was read as this one's"
);
assert_eq!(
begun(earlier + Duration::from_secs(3_600)),
Some(Wrote::Nothing),
"a commit an earlier dispatch wrote in this worktree was read as this one's"
);
let (_, ahead) = a_cut_worktree("ahead");
commit_in(&ahead, "ahead.md");
assert_eq!(
level(&ahead),
None,
"a branch ahead of its base read as level"
);
let (_, dirty) = a_cut_worktree("dirty");
std::fs::write(dirty.join("dirty.md"), "uncommitted\n").expect("the file");
assert_eq!(level(&dirty), None, "a dirty worktree read as level");
for name in [
"fresh", "ff", "reset", "landed", "picked", "merged", "rebased", "applied", "undone",
"resumed", "ahead", "dirty",
] {
let _ = std::fs::remove_dir_all(
std::env::temp_dir()
.join(format!("onepipeline-level-{name}-{}", std::process::id())),
);
}
}
#[test]
fn the_second_a_dispatch_began_in_is_over_before_its_first_session_opens() {
use std::time::{Duration, SystemTime, UNIX_EPOCH};
let seconds = |at: SystemTime| {
at.duration_since(UNIX_EPOCH)
.expect("after the epoch")
.as_secs()
};
let began = SystemTime::now();
super::wait_out_the_second(began);
assert!(
seconds(SystemTime::now()) > seconds(began),
"the wait returned inside the second it was asked to leave"
);
let long_ago = SystemTime::now() - Duration::from_secs(60);
let asked = std::time::Instant::now();
super::wait_out_the_second(long_ago);
assert!(
asked.elapsed() < Duration::from_secs(1),
"a second already past was waited for"
);
super::wait_out_the_second(UNIX_EPOCH - Duration::from_secs(1));
}
#[test]
fn a_reflog_entry_that_wrote_a_commit_is_told_from_one_that_only_moved_head() {
for wrote in [
"commit: feat: one",
"commit (amend): feat: one",
"commit (initial): chore: seed",
"commit (merge): Merge branch 'x'",
"cherry-pick: feat: one",
"am: feat: one",
"rebase (pick): feat: one",
"rebase (continue): feat: one",
"merge origin/main: Merge made by the 'ort' strategy.",
"pull: Merge made by the 'ort' strategy.",
] {
assert!(super::reflog_entry_wrote_a_commit(wrote), "{wrote}");
}
for moved in [
"",
"reset: moving to origin/main",
"reset: moving to HEAD",
"checkout: moving from main to work",
"merge origin/main: Fast-forward",
"pull: Fast-forward",
"rebase (start): checkout origin/main",
"rebase (finish): returning to refs/heads/work",
"rebase (abort): returning to refs/heads/work",
] {
assert!(!super::reflog_entry_wrote_a_commit(moved), "{moved:?}");
}
}
#[test]
fn a_session_tip_tells_a_branch_that_did_not_move_from_one_nothing_could_read() {
let _home = super::scratch_home_held();
let root = std::env::temp_dir().join(format!("onepipeline-tip-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir_all(root.join("streams")).expect("a scratch state root");
std::env::set_var(onevcs_home(), &root);
let record = |token: &str, seq: u64, kind: &str, payload: serde_json::Value| {
serde_json::json!({
"v": 1,
"ts": "2026-01-01T00:00:00.000Z",
"stream": token,
"seq": seq,
"source": "vcs",
"kind": kind,
"labels": {},
"payload": payload,
"artifacts": [],
})
.to_string()
};
let write = |token: &str, body: String| {
std::fs::write(root.join("streams").join(format!("{token}.ndjson")), body)
.expect("the stream is written");
};
let opened = |token: &str| record(token, 1, "session-opened", serde_json::json!({}));
let committed = |token: &str, sha: &str| {
record(
token,
2,
"commit-preserved",
serde_json::json!({"branch": "b", "sha": sha}),
)
};
let at = "s-tip-committed";
write(at, format!("{}\n{}\n", opened(at), committed(at, "c0ffee")));
assert_eq!(
session_tip(&SessionToken(at.into())),
SessionTip::At(Commit::of("c0ffee").expect("a commit this crate carries"))
);
let unmoved = "s-tip-nothing";
write(unmoved, format!("{}\n", opened(unmoved)));
assert_eq!(
session_tip(&SessionToken(unmoved.into())),
SessionTip::Unmoved
);
let torn = "s-tip-torn";
let whole = committed(torn, "decaf");
write(torn, format!("{}\n{}", opened(torn), &whole[..20]));
assert_eq!(
session_tip(&SessionToken(torn.into())),
SessionTip::Unknown,
"a batch the reader refused was read as a session that committed nothing"
);
assert_eq!(
session_tip(&SessionToken("s-tip-neverwritten".into())),
SessionTip::Unknown
);
let forged = "s-tip-forged";
write(
forged,
format!(
"{}\n{}\n",
opened(forged),
committed(forged, "c0ffee\u{7}bad")
),
);
assert_eq!(
session_tip(&SessionToken(forged.into())),
SessionTip::Unknown,
"a commit that would forge a row was read as a session that committed nothing"
);
assert_eq!(
branch_head_in(&SessionToken(at.into())),
Some("c0ffee".to_string())
);
assert_eq!(branch_head_in(&SessionToken(unmoved.into())), None);
assert_eq!(branch_head_in(&SessionToken(torn.into())), None);
let _ = std::fs::remove_dir_all(&root);
}
#[test]
fn a_session_stream_that_is_not_whole_is_read_for_what_it_holds() {
let _home = super::scratch_home_held();
let root = std::env::temp_dir().join(format!("onepipeline-stream-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
std::fs::create_dir_all(root.join("streams")).expect("a scratch state root");
std::env::set_var(onevcs_home(), &root);
let record = |token: &str, seq: u64, kind: &str| {
serde_json::json!({
"v": 1,
"ts": "2026-01-01T00:00:00.000Z",
"stream": token,
"seq": seq,
"source": "vcs",
"kind": kind,
"labels": {},
"payload": {},
"artifacts": [],
})
.to_string()
};
let write = |token: &str, body: String| {
std::fs::write(root.join("streams").join(format!("{token}.ndjson")), body)
.expect("the stream is written");
};
let seqs = |envelopes: &[Envelope]| envelopes.iter().map(|e| e.seq).collect::<Vec<_>>();
let torn = "s-unterminated";
let followed = SessionToken(torn.to_owned());
write(
torn,
format!(
"{}\n{}",
record(torn, 1, "session-opened"),
record(torn, 2, "push")
),
);
let mut stream = opened(&followed, None).expect("the stream opens");
assert_eq!(
seqs(&next_batch(&mut stream, &followed)),
vec![1, 2],
"a record whose newline was still unwritten was lost"
);
write(
torn,
format!(
"{}\n{}\n{}\n",
record(torn, 1, "session-opened"),
record(torn, 2, "push"),
record(torn, 3, "session-closed")
),
);
assert_eq!(
seqs(&next_batch(&mut stream, &followed)),
vec![3],
"the terminator arriving handed a record back a second time"
);
let cut = "s-cutmidline";
let whole = record(cut, 1, "session-opened");
let partial = record(cut, 2, "push");
write(cut, format!("{whole}\n{}", &partial[..20]));
assert!(
events(&SessionToken(cut.to_owned()), None).is_empty(),
"the typed reader now hands back the whole records before a torn one; \
narrow this assertion to the torn record alone"
);
assert!(events(&SessionToken("s-neverwritten".into()), None).is_empty());
let _ = std::fs::remove_dir_all(&root);
}
fn onevcs_home() -> &'static str {
"ONEVCS_HOME"
}
#[test]
fn the_marks_a_later_reader_starts_from_are_folded_from_what_the_store_holds() {
let wrote = |stream: &str, seq: u64| Envelope {
v: crate::event::ENVELOPE_VERSION,
ts: "2026-01-01T00:00:00.000Z".into(),
stream: stream.to_owned(),
seq,
source: crate::event::Source::Vcs,
kind: crate::event::EventKind("release-observed".into()),
phase: Some(crate::event::Phase::Release),
labels: crate::event::Labels::default(),
payload: serde_json::Map::new(),
artifacts: Vec::new(),
};
let held = vec![wrote("s-1", 1), wrote("s-1", 2), wrote("releases-abc", 7)];
let marks = Watermarks::of_relayed(&held);
for envelope in &held {
assert!(
!marks.beyond(envelope),
"a record the store already holds would be relayed again: {envelope:?}"
);
}
assert!(marks.beyond(&wrote("s-1", 3)));
assert!(marks.beyond(&wrote("releases-abc", 8)));
assert!(!marks.beyond(&wrote("releases-abc", 6)));
assert!(marks.beyond(&wrote("releases-def", 0)));
}
#[test]
fn a_relayed_envelope_keeps_the_kind_and_attribution_its_producer_wrote() {
let mut labels = onevcs::Labels {
member: Some("worker".into()),
..onevcs::Labels::default()
};
labels
.extra
.insert("session".into(), serde_json::json!("s-1"));
let envelope = relayed(onevcs::Envelope {
v: 1,
ts: "2026-01-01T00:00:00.000Z".into(),
stream: "s-1".into(),
seq: 4,
source: onevcs::Source::Vcs,
kind: onevcs::EventKind::ChangeOpened,
phase: onevcs::Phase::Review,
labels,
payload: serde_json::Map::new(),
artifacts: vec![onevcs::ArtifactRef {
id: onevcs::ArtifactId("a-1".into()),
kind: "log".into(),
bytes: 12,
}],
});
assert_eq!(envelope.kind.0, "change-opened");
assert_eq!(envelope.source, crate::event::Source::Vcs);
assert_eq!(envelope.seq, 4);
assert_eq!(envelope.labels.extra["member"], "worker");
assert_eq!(envelope.labels.extra["session"], "s-1");
assert_eq!(envelope.artifacts[0].id.0, "a-1");
}
#[test]
fn a_read_that_got_no_answer_says_what_refused_rather_than_that_nothing_landed() {
let refused = read_of(Err(onevcs::Error::Invalid {
reason: "no such repository\nand a second line".into(),
}));
assert_eq!(
refused,
LandingRead::Refused {
because: "this host could not decide it: invalid input: no such repository and \
a second line"
.into()
}
);
assert!(!proved_landed_from(&refused), "a refusal read as a landing");
let landed = read_of(Ok(onevcs::Landed::Yes {
evidence: onevcs::LandingEvidence::RecordedLanding {
commit: onevcs::Sha("abc1234".into()),
},
}));
assert!(matches!(
&landed,
LandingRead::Answered(onevcs::Landed::Yes { .. })
));
assert!(proved_landed_from(&landed));
}
fn proved_landed_from(read: &LandingRead) -> bool {
matches!(read, LandingRead::Answered(landed) if landed.is_landed())
}
}