Skip to main content

onevcs_testing/
repository.rs

1//! The repository side: one implementation of [`Vcs`] over either store.
2//!
3//! What it is not: git. It answers the questions the interface asks, records what
4//! it was asked, and emits the events the real implementation emits. What it
5//! cannot do is tell you whether a tree is dirty or whether a merge conflicts,
6//! because there is no tree — a journey that needs those drives the real `Git`.
7//!
8//! Publishing is the one operation that reaches past the repository: the host side
9//! of it is *performed*, against the [`Hosting`] the publication was handed, and
10//! the repository side of it is neither performed nor claimed. What that leaves
11//! out is written where each piece is left out.
12
13use std::path::{Path, PathBuf};
14
15use serde_json::{json, Map, Value};
16
17use std::collections::BTreeSet;
18
19use onevcs::rules::{Approvals, Drafts};
20use onevcs::{
21    ChangeRequest, ChangeSpec, Check, CheckState, DraftReason, Error, EventKind, FailureKind,
22    HeldBy, Holding, Hosting, Identity, Landed, Lifecycle, MergeOutcome, MergePolicy,
23    PreservedBranch, Provenance, Publication, PublishOutcome, PublishRequest, Recoverable,
24    RemoteHost, Result, Scope, Session, SessionRecord, SessionRequest, SessionToken, Sha, Vcs,
25};
26
27use crate::events::{self, Emission};
28use crate::remote::DEFAULT_HOST;
29use crate::state::{self, VcsState};
30use crate::store::{FileStore, MemoryStore, Store};
31
32/// The base a session is cut from when the request names none.
33///
34/// The real implementation asks the origin for its default branch, and this
35/// provider has no origin to ask.
36pub const DEFAULT_BASE: &str = "main";
37
38/// The policy a publication takes when nothing was seeded and nothing requested.
39///
40/// The policy the contract's own `default:` names, which is what the real
41/// implementation resolves to for a registry with no rules file.
42pub const DEFAULT_PUBLICATION: MergePolicy = MergePolicy::ChangeOpen;
43
44/// The approvals a publication takes when nothing was seeded: the contract's own
45/// `default:`, as [`DEFAULT_PUBLICATION`] is.
46pub const DEFAULT_APPROVALS: Approvals = Approvals::Required;
47
48/// The repository side of a run, over whichever store holds its state.
49///
50/// The two flavours below are this one behaviour with a different store under it,
51/// so neither can learn something the other does not know.
52#[derive(Debug)]
53pub struct Repository<T> {
54    store: T,
55    root: PathBuf,
56    trees: Trees,
57}
58
59/// What a session's worktree path means.
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61enum Trees {
62    /// The path is named and nothing is created there: the in-memory provider
63    /// touches no filesystem beyond the event stream.
64    Named,
65    /// The directory is created, so a journey has somewhere to write work.
66    Created,
67}
68
69/// A repository provider that keeps its state in this process: no disk, no
70/// visibility to a second process, and the fastest of the two.
71///
72/// Its sessions name a worktree under the system temporary directory and **do not
73/// create it** — nothing here touches the filesystem except the event stream, which
74/// is the record a journey reads.
75pub type MemoryVcs = Repository<MemoryStore<VcsState>>;
76
77/// A repository provider that keeps its state in one JSON document, so several
78/// `onevcs` invocations see one another's effects.
79///
80/// Its sessions name a worktree beside that document **and create it**, so a
81/// journey that writes a file into a session's tree has somewhere to write it.
82pub type FileVcs = Repository<FileStore<VcsState>>;
83
84impl MemoryVcs {
85    /// A repository provider knowing nothing.
86    pub fn new() -> Self {
87        Self::seeded(VcsState::default())
88    }
89
90    /// A repository provider that starts from a scenario.
91    pub fn seeded(state: VcsState) -> Self {
92        Self {
93            store: MemoryStore::new(state),
94            root: std::env::temp_dir().join("onevcs-testing-memory"),
95            trees: Trees::Named,
96        }
97    }
98
99    /// Everything it knows.
100    pub fn state(&self) -> VcsState {
101        self.store
102            .snapshot()
103            .expect("an in-memory store always answers")
104    }
105}
106
107impl Default for MemoryVcs {
108    fn default() -> Self {
109        Self::new()
110    }
111}
112
113impl FileVcs {
114    /// A repository provider keeping its state at `path`: whatever is already
115    /// there, or nothing.
116    ///
117    /// Attaching rather than replacing, so a second provider over the same path
118    /// picks up what the first one left — which is what a journey driving several
119    /// invocations reaches for this flavour to get.
120    pub fn create(path: impl Into<PathBuf>) -> Result<Self> {
121        Self::over(FileStore::attach(path, &VcsState::default())?)
122    }
123
124    /// A repository provider that starts from a scenario, keeping its state at
125    /// `path` and replacing whatever was there.
126    pub fn seeded(path: impl Into<PathBuf>, state: VcsState) -> Result<Self> {
127        Self::over(FileStore::replace(path, &state)?)
128    }
129
130    fn over(store: FileStore<VcsState>) -> Result<Self> {
131        let root = store
132            .path()
133            .parent()
134            .filter(|parent| !parent.as_os_str().is_empty())
135            .unwrap_or_else(|| Path::new("."))
136            .join("worktrees");
137        Ok(Self {
138            store,
139            root,
140            trees: Trees::Created,
141        })
142    }
143
144    /// Everything it knows, read back out of its document.
145    pub fn state(&self) -> Result<VcsState> {
146        self.store.snapshot()
147    }
148}
149
150impl<T: Store<VcsState>> Vcs for Repository<T> {
151    fn resolve_identity(&self, origin_or_path: &str) -> Result<Identity> {
152        self.store.with(|state| {
153            state::identity_of(state, origin_or_path)
154                .cloned()
155                .ok_or_else(|| Error::Invalid {
156                    reason: format!(
157                        "{origin_or_path:?} does not name a repository this provider knows; {}",
158                        state::known(state)
159                    ),
160                })
161        })
162    }
163
164    fn open_session(&self, req: SessionRequest) -> Result<Session> {
165        let root = self.root.clone();
166        let (session, emission) = self.store.with(|state| {
167            let identity = state::identity_of(state, &req.repo)
168                .cloned()
169                .ok_or_else(|| Error::Invalid {
170                    reason: format!(
171                        "{:?} does not name a repository this provider knows; {}",
172                        req.repo,
173                        state::known(state)
174                    ),
175                })?;
176            // Consecutive and predictable, so a journey can name the session it is
177            // about to open — the one thing a digest-shaped token takes away.
178            let token = SessionToken(format!("s-testing-{}", state.sessions.len() + 1));
179            let run_root = root.join(&token.0);
180            // Both names are checked before they are recorded, because both go on to
181            // spell a ref for whoever holds the session — and a provider that
182            // accepted a name git refuses would let a journey pass where the real
183            // run stops.
184            let base = req.base.clone().unwrap_or_else(|| DEFAULT_BASE.to_owned());
185            state::named_branch(&base, "the base")?;
186            for (key, value) in &req.labels {
187                state::label_pair(key, value)?;
188            }
189            // Never a conflict: a provider has no base to merge into a continued
190            // branch, so every session it opens opens clean — which is also why its
191            // `session-opened` below never carries the `conflict` key the real one
192            // writes over a merge it left in progress, and why `refuse_conflicts`
193            // has nothing here to refuse.
194            let session = Session {
195                worktree: run_root.join("worktree"),
196                branch: state::requested_branch(&req, &token)?,
197                base,
198                token: token.clone(),
199                conflict: None,
200            };
201            state.sessions.push(session.clone());
202            state
203                .session_identities
204                .insert(token.clone(), identity.origin.clone());
205            if !req.labels.is_empty() {
206                state
207                    .session_labels
208                    .insert(token.clone(), req.labels.clone());
209            }
210            let emission = Emission {
211                stream: token.0.clone(),
212                identity: Some(identity.origin.clone()),
213                kind: EventKind::SessionOpened,
214                payload: object(json!({
215                    "token": token.0,
216                    "identity": identity.origin,
217                    "branch": session.branch,
218                    "base": session.base,
219                    "worktree": session.worktree.display().to_string(),
220                    // Synthetic, and named anyway: a consumer reading this event
221                    // reads the same keys whichever implementation produced it.
222                    "clone": run_root.join("clone").display().to_string(),
223                    "execution_checkout": run_root.join("checkout").display().to_string(),
224                    "publication_checkout": run_root.join("checkout").display().to_string(),
225                    // A run root of its own, which is the one placement a provider
226                    // that keeps no pool can make — and the one every real session
227                    // gets on a host that configures none.
228                    "placement": {"kind": "run-root"},
229                })),
230            };
231            Ok((session, emission))
232        })?;
233        if self.trees == Trees::Created {
234            std::fs::create_dir_all(&session.worktree).map_err(|e| Error::Invalid {
235                reason: format!("cannot create {}: {e}", session.worktree.display()),
236            })?;
237        }
238        events::emit(&emission);
239        Ok(session)
240    }
241
242    fn adopt_session(&self, token: SessionToken) -> Result<Session> {
243        self.store.with(|state| {
244            state::session_of(state, &token)
245                .cloned()
246                .ok_or_else(|| Error::Invalid {
247                    reason: format!(
248                        "no session {:?} is open; `onevcs session open` prints a token",
249                        token.0
250                    ),
251                })
252        })
253    }
254
255    fn preserve(&self, s: &Session, provenance: Provenance) -> Result<PreservedBranch> {
256        let (branch, emission) = self.store.with(|state| {
257            let identity = state::identity_for(state, &s.token)?;
258            let branch = PreservedBranch {
259                branch: s.branch.clone(),
260                base: s.base.clone(),
261                provenance,
262                change_url: None,
263                change_base: None,
264            };
265            let row = Recoverable {
266                identity: identity.clone(),
267                branch: branch.clone(),
268                checkout: s.worktree.clone(),
269                stopped_because: format!("session {} was left open", s.token.0),
270                recover_command: recover_command(&s.branch, &s.worktree, provenance),
271                // Preserved work this provider was handed is work nobody published:
272                // there is no base here whose history could record a landing, and no
273                // tree to compare content against, so the one answer it can give is
274                // the one it knows.
275                landed: Landed::No,
276                // Both answered where the row is *read* rather than frozen in here:
277                // whether a session still holds its branch is a fact about the session
278                // now, and there is no tree here to count a diff's lines against.
279                held_by: None,
280                net_negative: None,
281                // The session that preserved it is the session that answers for it,
282                // and what it was opened with is what the row carries.
283                session: Some(s.token.clone()),
284                labels: state
285                    .session_labels
286                    .get(&s.token)
287                    .cloned()
288                    .unwrap_or_default(),
289                // A preservation onto an origin is a thing only a real repository can
290                // do — there is no git here and no remote to push to — so this provider
291                // answers what a scenario wrote down and nothing else. A hand-written
292                // state may still seed one, exactly as it seeds a hold.
293                on_origin: None,
294                // Whether a branch may be retired is decided from every copy of it this
295                // host holds, its base on the origin, and this host's records — none of
296                // which a provider has — so this provider classifies nothing, and a row
297                // it answers carries no classification rather than an invented one.
298                retirement: None,
299            };
300            // Preserving the same branch twice replaces its row rather than listing
301            // it twice, which is what `recoverable` does across the checkouts a
302            // branch is reachable from.
303            state.preserved.retain(|kept| {
304                kept.identity != row.identity || kept.branch.branch != row.branch.branch
305            });
306            state.preserved.push(row);
307            let emission = Emission {
308                stream: s.token.0.clone(),
309                // No identity label, because the real implementation carries none
310                // here: the label is stamped where a session is opened, and work is
311                // preserved against a stream a later process opened fresh. Claiming
312                // it would be drift in the direction that looks like more
313                // information.
314                identity: None,
315                kind: EventKind::CommitPreserved,
316                payload: object(json!({
317                    "branch": s.branch,
318                    "sha": events::stable_sha(&[&s.token.0, &s.branch, spell(provenance)]),
319                    "provenance": spell(provenance),
320                })),
321            };
322            Ok((branch, emission))
323        })?;
324        events::emit(&emission);
325        Ok(branch)
326    }
327
328    fn session(&self, token: &SessionToken) -> Result<SessionRecord> {
329        self.store.with(|state| {
330            let session = state::session_of(state, token)
331                .cloned()
332                .ok_or_else(|| unknown_session(token))?;
333            let identity = state::identity_for(state, token)?;
334            // Read off what was preserved rather than remembered separately: the
335            // real implementation reads the branch, so a session whose work was
336            // preserved behind an incomplete-step marker answers the same here.
337            let provenance = state
338                .preserved
339                .iter()
340                .find(|row| row.identity == identity && row.branch.branch == session.branch)
341                .map_or(Provenance::Complete, |row| row.branch.provenance);
342            Ok(SessionRecord {
343                lifecycle: if state.closed_sessions.contains(token) {
344                    Lifecycle::Closed
345                } else {
346                    Lifecycle::Open
347                },
348                session,
349                identity,
350                provenance,
351                // A retry link is a fact about run clones under a state root, and
352                // this provider keeps none: its sessions are records in a store, so
353                // there is no second copy of a branch for one to tell apart.
354                retried_by: None,
355            })
356        })
357    }
358
359    fn close_session(&self, token: &SessionToken) -> Result<Session> {
360        self.store.with(|state| {
361            let session = state::session_of(state, token)
362                .cloned()
363                .ok_or_else(|| unknown_session(token))?;
364            let emission = Emission {
365                stream: token.0.clone(),
366                // No identity label, because the real implementation carries none
367                // here: closing opens the stream fresh, and the label is stamped
368                // where a session is opened.
369                identity: None,
370                kind: EventKind::SessionClosed,
371                payload: object(json!({"token": token.0, "branch": session.branch})),
372            };
373            // Match the real provider's observable ordering: a follower reads the
374            // stream before consulting this state, so the terminator must exist
375            // before `Closed` can be returned to that concurrent reader.
376            events::emit(&emission);
377            state.closed_sessions.insert(token.clone());
378            Ok(session)
379        })
380    }
381
382    /// Publish a session's branch, as far as a provider honestly can.
383    ///
384    /// The host side is performed rather than described: a change request is really
385    /// opened against the [`Hosting`] this was handed, really adopted when one is
386    /// already open, and really merged under the policy — so the six host methods
387    /// are exercised and what the host recorded is what a journey reads back.
388    ///
389    /// The repository side is not, and none of it is claimed. There is no origin to
390    /// fetch from, no tree to build in, no push, and no lock to queue behind, so no
391    /// `fetch`, `push`, `lock-wait`, `lock-acquired`, or `merge-queued` event is
392    /// emitted. What is emitted is what was decided: the change that was opened, and
393    /// the merge that landed.
394    fn publish(
395        &self,
396        token: &SessionToken,
397        request: &PublishRequest,
398        hosting: &dyn Hosting,
399    ) -> Result<Publication> {
400        let (publication, emissions) = self.store.with(|state| {
401            let session = state::session_of(state, token)
402                .cloned()
403                .ok_or_else(|| unknown_session(token))?;
404            let identity = state::identity_for(state, token)?;
405            // The publication rule the crate next door states, applied rather than
406            // restated: a reason that would not render as itself is refused here
407            // exactly where the real implementation refuses it, before any host is
408            // asked and before anything is recorded.
409            if let Some(reason) = &request.draft {
410                reason.checked()?;
411            }
412            let resolved = state.policy.unwrap_or(DEFAULT_PUBLICATION);
413            let policy = match request.policy {
414                Some(requested) => resolved.narrow(requested)?,
415                None => resolved,
416            };
417            let published = |outcome, emissions| {
418                (
419                    Publication {
420                        session: token.clone(),
421                        branch: session.branch.clone(),
422                        policy,
423                        outcome,
424                    },
425                    emissions,
426                )
427            };
428            // A session that has already landed has nothing the base does not carry,
429            // which is what the real implementation reports for the same reason. One
430            // whose change request is merely open or queued has *not* landed, and
431            // publishing it again adopts that change rather than opening a second —
432            // so it falls through to the host, as it does there.
433            if state.publications.iter().any(|earlier| {
434                earlier.session == *token && matches!(earlier.outcome, PublishOutcome::Merged(_))
435            }) {
436                let (publication, emissions) =
437                    published(PublishOutcome::NothingToPublish, Vec::new());
438                state.publications.push(publication.clone());
439                return Ok((publication, emissions));
440            }
441
442            let (outcome, emissions) = if policy == MergePolicy::LocalDirect {
443                // The same refusal the real publication makes at the same boundary: a
444                // local-direct publication opens no change request, so there is
445                // nothing to draft and the work would land carrying the very pin the
446                // draft exists to hold back.
447                match &request.draft {
448                    Some(reason) => (
449                        failed(&Error::Invalid {
450                            reason: format!(
451                                "{branch:?} was asked to publish as {asked}, and this identity \
452                                 publishes with local-direct, which opens no change request at \
453                                 all",
454                                branch = session.branch,
455                                asked = asked_for(reason),
456                            ),
457                        }),
458                        Vec::new(),
459                    ),
460                    None => record_local_landing(&identity, &session, token),
461                }
462            } else {
463                match slug(&identity) {
464                    Some(slug) => {
465                        // Recorded as it happens, as next door: a publication that
466                        // fails after opening, drafting or lifting keeps that record.
467                        let mut emissions = Vec::new();
468                        let outcome = match publish_as_change(
469                            hosting,
470                            &Publishing {
471                                slug: &slug,
472                                identity: &identity,
473                                session: &session,
474                                policy,
475                                approvals: state.approvals.unwrap_or(DEFAULT_APPROVALS),
476                                drafts: state.drafts.unwrap_or_default(),
477                                request,
478                                token,
479                            },
480                            &mut emissions,
481                        ) {
482                            Ok(outcome) => outcome,
483                            // Once a publication has started, what stops it is an outcome
484                            // rather than a refusal — the same split the real
485                            // implementation keeps, so a caller reads one shape.
486                            Err(error) => failed(&error),
487                        };
488                        (outcome, emissions)
489                    }
490                    None => (refusal(&identity), Vec::new()),
491                }
492            };
493            let (publication, emissions) = published(outcome, emissions);
494            state.publications.push(publication.clone());
495            Ok((publication, emissions))
496        })?;
497        for emission in &emissions {
498            events::emit(emission);
499        }
500        Ok(publication)
501    }
502
503    /// The same rows [`Vcs::recoverable`] answers with.
504    ///
505    /// This provider is *handed* its preserved work, and what it is handed is work
506    /// nobody published: there is no base here whose history could record a landing
507    /// and no tree to compare content against, so it withholds nothing and the wider
508    /// question has the same answer as the narrower one.
509    fn preserved(&self, scope: Scope) -> Result<Vec<Recoverable>> {
510        self.recoverable(scope)
511    }
512
513    fn recoverable(&self, scope: Scope) -> Result<Vec<Recoverable>> {
514        self.store.with(|state| {
515            let wanted = match &scope {
516                Scope::All => None,
517                Scope::Repo(repo) => Some(
518                    state::identity_of(state, repo)
519                        .map(|identity| identity.origin.clone())
520                        .ok_or_else(|| Error::Invalid {
521                            reason: format!(
522                                "{repo:?} does not name a repository this provider knows; {}",
523                                state::known(state)
524                            ),
525                        })?,
526                ),
527            };
528            // Newest first, as the real implementation reports them.
529            Ok(state
530                .preserved
531                .iter()
532                .rev()
533                .filter(|row| wanted.as_ref().is_none_or(|key| *key == row.identity))
534                .cloned()
535                .map(|row| held(state, row))
536                .collect())
537        })
538    }
539}
540
541/// Say which live session still holds a row's branch, when one of this provider's
542/// does.
543///
544/// Read here rather than written when the work was preserved, because closing the
545/// session is what ends the hold and a row that had frozen the answer would go on
546/// saying somebody is in there. What makes it *live* is the same thing that makes it
547/// live for the real implementation in-process: the session is open, and the process
548/// that opened it is this one. A row a scenario seeded the answer into keeps it — a
549/// hand-written state says what it means to say.
550fn held(state: &VcsState, row: Recoverable) -> Recoverable {
551    if row.held_by.is_some() {
552        return row;
553    }
554    let holder = state.sessions.iter().find(|session| {
555        session.branch == row.branch.branch
556            && state.session_identities.get(&session.token) == Some(&row.identity)
557            && !state.closed_sessions.contains(&session.token)
558    });
559    Recoverable {
560        held_by: holder.map(|session| HeldBy {
561            token: Some(session.token.clone()),
562            worktree: session.worktree.clone(),
563            holding: Holding::OwnerRunning,
564        }),
565        ..row
566    }
567}
568
569/// The refusal a session this provider never opened meets.
570fn unknown_session(token: &SessionToken) -> Error {
571    Error::Invalid {
572        reason: format!(
573            "no session {:?} is open; `onevcs session open` prints a token",
574            token.0
575        ),
576    }
577}
578
579/// The `owner/name` slug an identity key spells, when it is a GitHub one.
580///
581/// The host is checked rather than assumed, exactly as the real implementation
582/// checks it: a GitLab origin has the same three segments, and a provider that
583/// published one anyway would let a journey pass where the real run answers that
584/// nobody has implemented that host.
585fn slug(identity: &str) -> Option<String> {
586    let mut parts = identity.split('/');
587    let (host, owner, name) = (parts.next()?, parts.next()?, parts.next()?);
588    if parts.next().is_some() || host != DEFAULT_HOST || owner.is_empty() || name.is_empty() {
589        return None;
590    }
591    Some(format!("{owner}/{name}"))
592}
593
594/// Record that a `local-direct` publication landed — record, and nothing more.
595///
596/// Landing one is entirely repository-side work, a squash built detached and
597/// pushed, which this provider does not perform and must not be named as if it
598/// did. What it does is decide the outcome and emit the completion that says so.
599fn record_local_landing(
600    identity: &str,
601    session: &Session,
602    token: &SessionToken,
603) -> (PublishOutcome, Vec<Emission>) {
604    let sha = events::stable_sha(&["publish", &token.0, &session.branch]);
605    let emission = Emission {
606        stream: token.0.clone(),
607        identity: Some(identity.to_owned()),
608        kind: EventKind::MergeCompleted,
609        // Every record of a landing says when the base received it and at which
610        // commit, as `onevcs`'s own does: here, the moment this provider decided it.
611        payload: object(json!({
612            "identity": identity,
613            "sha": sha,
614            "base": session.base,
615            "landed_at": events::timestamp(),
616            "landing": sha,
617        })),
618    };
619    (PublishOutcome::Merged(Sha(sha)), vec![emission])
620}
621
622/// Everything one provider publication knows about itself.
623struct Publishing<'a> {
624    slug: &'a str,
625    identity: &'a str,
626    session: &'a Session,
627    policy: MergePolicy,
628    approvals: Approvals,
629    drafts: Drafts,
630    request: &'a PublishRequest,
631    token: &'a SessionToken,
632}
633
634impl Publishing<'_> {
635    /// One event on the session's own stream, under this identity.
636    fn emission(&self, kind: EventKind, payload: Value) -> Emission {
637        Emission {
638            stream: self.token.0.clone(),
639            identity: Some(self.identity.to_owned()),
640            kind,
641            payload: object(payload),
642        }
643    }
644}
645
646/// Publish as a change request: open the session's change on the host, or adopt
647/// the one it already holds, watch its required checks, and then do with it what
648/// the policy asks.
649///
650/// **The draft lifecycle, as next door.** A publication carrying no reason opens its
651/// change request as a draft while the checks run — or, adopting a draft, records it
652/// as awaiting them rather than lifting it on the spot — and once they are green lifts
653/// it before asking for any merge, or keeps it under `change-open` with approvals
654/// required. `drafts: {disabled: true}` opens ready and lifts an adopted draft at once.
655///
656/// **What this provider cannot do is wait.** It has no clock, so each phase of the
657/// watch is **one reading** of the host: a reading that has not settled is the bound
658/// elapsing (`checks-unsettled`), and a draft on which no required check has run is
659/// the grace window elapsing — it is lifted early, and the next reading is the host's
660/// answer after the lift, which is where [`HostState::checks_after_lift`](crate::HostState::checks_after_lift)
661/// comes in. A consumer drives each row of the lifecycle by seeding the reading it
662/// wants rather than by timing one.
663fn publish_as_change(
664    hosting: &dyn Hosting,
665    publishing: &Publishing<'_>,
666    emissions: &mut Vec<Emission>,
667) -> Result<PublishOutcome> {
668    let Publishing {
669        slug,
670        session,
671        policy,
672        request,
673        ..
674    } = *publishing;
675    let host = hosting.for_repo(slug)?;
676    // Who the host believes is calling travels with the change, as it does in the
677    // real publication and for the same reason.
678    let author = host.authenticated_user()?;
679    let lifecycle = !publishing.drafts.disabled.unwrap_or(false);
680    // Checked before anything is opened, as the real publication checks it before
681    // anything is pushed.
682    if lifecycle {
683        grace_seconds()?;
684    }
685    let existing = host.find_changes(&session.branch, &session.base)?;
686    // Whether the host already held it, which is what decides whether there can be a
687    // draft to lift — and, as next door, keeps a publication that opens its own change
688    // request from asking this host anything extra.
689    let adopted = !existing.is_empty();
690    let change = match existing.into_iter().next() {
691        Some(change) => change,
692        None => host.open_change(ChangeSpec {
693            head: session.branch.clone(),
694            base: session.base.clone(),
695            // A requested title has been checked by the conversion that built it, so
696            // this provider cannot accept one the real publication would refuse. The
697            // real implementation takes the subject from the branch's commits when no
698            // title was requested, and a provider has no commits to read — so an
699            // unrequested title names the branch instead.
700            title: request
701                .title
702                .as_deref()
703                .map_or_else(|| format!("Publish {}", session.branch), str::to_owned),
704            // Verbatim, and nothing when there is none — the real publication
705            // composes no body either, so a provider that composed one would be a
706            // consumer's suite proving a change request nobody opens.
707            body: request.body.clone(),
708            // The reason travels to the host and no further, as it does there: what
709            // the host does with it is open the change as a draft.
710            draft: request.draft.clone(),
711            draft_awaiting_checks: lifecycle && request.draft.is_none(),
712        })?,
713    };
714    emissions.push(publishing.emission(
715        EventKind::ChangeOpened,
716        json!({
717            "url": change.url.to_string(),
718            "host": "github",
719            "id": change.id.0,
720            "base": change.base,
721            "author": author,
722        }),
723    ));
724    if let Some(reason) = &request.draft {
725        // The same question the real publication asks, for the same reason: a host
726        // that ignored the request, or a change request already open for review,
727        // would otherwise be reported as held back while it can land.
728        if !host.is_draft(&change)? {
729            return Err(Error::Invalid {
730                reason: format!(
731                    "{url} is open for review on the host, and this publication asked for \
732                     {asked}. A change that is open can land, so reporting it as a draft would \
733                     say the work is held back when nothing is holding it",
734                    url = change.url,
735                    asked = asked_for(reason),
736                ),
737            });
738        }
739        // The publication record, and the only place the reason is written: the real
740        // implementation writes nothing of it into the change request, and a provider
741        // that did would be a consumer's suite proving a body nobody renders. The
742        // payload is the reason's own serialized form — its `kind` and every field of
743        // that kind — under the change request it holds, exactly as next door.
744        let mut drafted = object(json!({
745            "url": change.url.to_string(),
746            "id": change.id.0,
747            "base": change.base,
748        }));
749        if let serde_json::Value::Object(fields) =
750            serde_json::to_value(reason).expect("a draft reason serializes")
751        {
752            drafted.extend(fields);
753        }
754        emissions.push(publishing.emission(EventKind::ChangeDrafted, Value::Object(drafted)));
755        // Under every policy: a draft is unmergeable in that state, so nothing below
756        // asks this host to merge it.
757        return Ok(PublishOutcome::ChangeDraft(change.url.clone()));
758    }
759    // Three answers about whether the host holds it as a draft, told apart as next
760    // door: a host that was never taught to draft one is holding nothing, and a host
761    // that *could not say* is a refusal rather than a change nobody is holding.
762    // Asked of a change this publication just opened only under the lifecycle, which
763    // asked for a draft: without it, one opened moments ago without a reason is one
764    // nobody drafted, and a host written against the earlier surface is asked nothing.
765    let is_draft = match if lifecycle || adopted {
766        host.is_draft(&change)
767    } else {
768        Ok(false)
769    } {
770        Ok(draft) => draft,
771        Err(Error::NotImplemented { .. }) => false,
772        Err(unreadable) => return Err(unreadable),
773    };
774    let drafted = if lifecycle {
775        if is_draft {
776            emissions.push(publishing.emission(
777                EventKind::ChangeDrafted,
778                json!({
779                    "url": change.url.to_string(),
780                    "id": change.id.0,
781                    "base": change.base,
782                    "kind": "awaiting-checks",
783                }),
784            ));
785        }
786        is_draft
787    } else {
788        // Without the lifecycle, publishing without a reason lifts an adopted draft
789        // on the spot, as it always did; one opened moments ago is one nobody drafted.
790        if adopted && is_draft {
791            lift(host.as_ref(), publishing, &change, emissions)?;
792        }
793        false
794    };
795
796    // Every change policy watches — except that `change-auto` on a change nobody
797    // drafted arms the host's own merge and leaves the checks to it, as next door.
798    // Next door's merge watch still refuses a required check that already concluded
799    // red rather than arm a merge over it, so this reading does too; one still
800    // running arms it, and the host's hold is what the outcome reports.
801    let still_draft = if drafted || policy != MergePolicy::ChangeAuto {
802        watch(host.as_ref(), publishing, &change, drafted, emissions)?
803    } else {
804        let answered = host.change_checks(&change)?;
805        let checks: Vec<&Check> = answered.checks.iter().collect();
806        if let Some(failed) = red(&checks, &standings(&checks, &Declared::Unknown)) {
807            return Err(failed);
808        }
809        false
810    };
811    if still_draft {
812        if policy == MergePolicy::ChangeOpen && publishing.approvals == Approvals::Required {
813            emissions.push(publishing.emission(
814                EventKind::DraftKeptForReview,
815                json!({
816                    "url": change.url.to_string(),
817                    "id": change.id.0,
818                    "base": change.base,
819                }),
820            ));
821            return Ok(PublishOutcome::ChangeReviewDraft(change.url.clone()));
822        }
823        // Lifted before any merge is asked for: this host, like GitHub, will neither
824        // merge a draft nor arm its own merge on one.
825        lift(host.as_ref(), publishing, &change, emissions)?;
826    }
827
828    if policy == MergePolicy::ChangeOpen {
829        return Ok(PublishOutcome::ChangeOpen(change.url.clone()));
830    }
831    Ok(match host.merge(&change, policy)? {
832        MergeOutcome::Merged(sha) => {
833            // When the host says the base received it, as `onevcs`'s own record of a
834            // landing carries it. Where the host cannot say there is no landing commit
835            // in this world to read a time from either, so the record says nobody can
836            // time it rather than claim the moment this provider saw the merge.
837            let landed_at = host
838                .merge_time(&change)
839                .ok()
840                .flatten()
841                .and_then(|spelled| events::moment(&spelled));
842            emissions.push(publishing.emission(
843                EventKind::ChangeMerged,
844                json!({
845                    "url": change.url.to_string(),
846                    "sha": sha.0,
847                    "landed_at": landed_at,
848                    "landing": sha.0,
849                }),
850            ));
851            emissions.push(publishing.emission(
852                EventKind::MergeCompleted,
853                json!({
854                    "identity": publishing.identity,
855                    "sha": sha.0,
856                    "landed_at": landed_at,
857                    "landing": sha.0,
858                }),
859            ));
860            PublishOutcome::Merged(sha)
861        }
862        MergeOutcome::Queued => PublishOutcome::Queued(change.url.clone()),
863        MergeOutcome::Open => PublishOutcome::ChangeOpen(change.url.clone()),
864    })
865}
866
867/// Take a change out of its draft on the host, and record the lift.
868fn lift(
869    host: &dyn RemoteHost,
870    publishing: &Publishing<'_>,
871    change: &ChangeRequest,
872    emissions: &mut Vec<Emission>,
873) -> Result<()> {
874    host.ready_for_review(change)?;
875    emissions.push(publishing.emission(
876        EventKind::DraftLifted,
877        json!({"url": change.url.to_string(), "id": change.id.0}),
878    ));
879    Ok(())
880}
881
882/// Which checks a merge requires, as the host said: nothing, these names, or unknown.
883enum Declared {
884    Nothing,
885    Names(BTreeSet<String>),
886    Unknown,
887}
888
889/// Each required check's name and the state its entries add up to — `None` where the
890/// host reported none — ranked as next door: red, then running, then passed, then no
891/// verdict, then skipped.
892fn standings(checks: &[&Check], declared: &Declared) -> Vec<(String, Option<CheckState>)> {
893    let names: BTreeSet<String> = match declared {
894        Declared::Names(names) => names.clone(),
895        Declared::Nothing => BTreeSet::new(),
896        Declared::Unknown => checks
897            .iter()
898            .filter(|check| check.required)
899            .map(|check| check.name.clone())
900            .collect(),
901    };
902    let rank = |state: CheckState| match state {
903        CheckState::Failed => 0,
904        CheckState::Pending => 1,
905        CheckState::Passed => 2,
906        CheckState::NoVerdict => 3,
907        CheckState::Skipped => 4,
908    };
909    names
910        .into_iter()
911        .map(|name| {
912            let state = checks
913                .iter()
914                .filter(|check| check.name == name)
915                .map(|check| check.state())
916                .min_by_key(|state| rank(*state));
917            (name, state)
918        })
919        .collect()
920}
921
922/// The refusal for the first required check that concluded red.
923fn red(checks: &[&Check], standing: &[(String, Option<CheckState>)]) -> Option<Error> {
924    let (name, _) = standing
925        .iter()
926        .find(|(_, state)| *state == Some(CheckState::Failed))?;
927    let check = checks
928        .iter()
929        .find(|check| &check.name == name && check.state() == CheckState::Failed)?;
930    Some(Error::ChecksFailed {
931        reason: format!(
932            "required check {:?} concluded {}.",
933            check.name,
934            check
935                .conclusion
936                .as_deref()
937                .unwrap_or("without a conclusion")
938        ),
939    })
940}
941
942/// Whether every required check lets the change through: passed, or skipped.
943fn through(standing: &[(String, Option<CheckState>)]) -> bool {
944    standing
945        .iter()
946        .all(|(_, state)| matches!(state, Some(CheckState::Passed | CheckState::Skipped)))
947}
948
949/// The skipped required checks, by name.
950fn skipped(standing: &[(String, Option<CheckState>)]) -> Vec<String> {
951    standing
952        .iter()
953        .filter(|(_, state)| *state == Some(CheckState::Skipped))
954        .map(|(name, _)| name.clone())
955        .collect()
956}
957
958/// The watch, one reading per phase: `Ok(true)` is green on a draft still standing,
959/// `Ok(false)` green on a ready change. See [`publish_as_change`] for what a reading
960/// stands for.
961fn watch(
962    host: &dyn RemoteHost,
963    publishing: &Publishing<'_>,
964    change: &ChangeRequest,
965    drafted: bool,
966    emissions: &mut Vec<Emission>,
967) -> Result<bool> {
968    // `unread`: the requirement was read from the host's own per-check marking because
969    // its declaration could not be read, which the record says, as next door.
970    let settled = |emissions: &mut Vec<Emission>, skipped: Vec<String>, unread: Option<&str>| {
971        let verdict = if skipped.is_empty() {
972            "passed"
973        } else {
974            "passed-with-skipped"
975        };
976        let mut payload = json!({
977            "url": change.url.to_string(),
978            "id": change.id.0,
979            "head": change.head_sha.0,
980            "verdict": verdict,
981            "skipped": skipped,
982        });
983        if let Some(because) = unread {
984            eprintln!(
985                "onevcs: warning: the required checks on {} were read from the host's own \
986                 per-check marking, because the declaration could not be read: {because}",
987                change.url
988            );
989            payload["requirement"] = json!({"read_from": "host-marking", "because": because});
990        }
991        emissions.push(publishing.emission(EventKind::ChecksSettled, payload));
992    };
993    let unsettled = |standing: &[(String, Option<CheckState>)]| Error::ChecksUnsettled {
994        reason: format!(
995            "the host had not settled its required checks on {}; still unsettled: {}",
996            change.url,
997            standing
998                .iter()
999                .filter(|(_, state)| !matches!(state, Some(CheckState::Passed)))
1000                .map(|(name, _)| format!("{name:?}"))
1001                .collect::<Vec<_>>()
1002                .join(", ")
1003        ),
1004    };
1005    let answered = host.change_checks(change)?;
1006    let complete = answered.complete();
1007    let checks: Vec<&Check> = answered.checks.iter().collect();
1008
1009    if !drafted {
1010        let standing = standings(&checks, &Declared::Unknown);
1011        if let Some(failed) = red(&checks, &standing) {
1012            return Err(failed);
1013        }
1014        if !through(&standing) {
1015            return Err(unsettled(&standing));
1016        }
1017        settled(emissions, skipped(&standing), None);
1018        return Ok(false);
1019    }
1020
1021    // The draft: what the host requires, and a skipped check has not run.
1022    let (mut declared, mut unread) = match host.required_checks_on(&change.base) {
1023        Ok(answer) if answer.checks.is_empty() && answer.complete() => (Declared::Nothing, None),
1024        Ok(answer) if !answer.checks.is_empty() => (Declared::Names(answer.checks), None),
1025        Ok(_) => (
1026            Declared::Unknown,
1027            Some(format!(
1028                "the host answered only in part about which checks a merge into {} requires",
1029                change.base
1030            )),
1031        ),
1032        Err(refused) => (
1033            Declared::Unknown,
1034            Some(format!(
1035                "the host would not say which checks a merge into {} requires: {refused}",
1036                change.base
1037            )),
1038        ),
1039    };
1040    if matches!(declared, Declared::Unknown)
1041        && complete
1042        && !checks.is_empty()
1043        && checks.iter().all(|check| !check.required)
1044    {
1045        declared = Declared::Nothing;
1046        unread = None;
1047    }
1048    if matches!(declared, Declared::Nothing) {
1049        settled(emissions, Vec::new(), None);
1050        return Ok(true);
1051    }
1052    let unread = unread.as_deref();
1053    let standing = standings(&checks, &declared);
1054    if let Some(failed) = red(&checks, &standing) {
1055        return Err(failed);
1056    }
1057    if !standing.is_empty()
1058        && standing
1059            .iter()
1060            .all(|(_, state)| *state == Some(CheckState::Passed))
1061    {
1062        settled(emissions, Vec::new(), unread);
1063        return Ok(true);
1064    }
1065    let not_run: Vec<String> = standing
1066        .iter()
1067        .filter(|(_, state)| matches!(state, None | Some(CheckState::Skipped)))
1068        .map(|(name, _)| name.clone())
1069        .collect();
1070    let running = standing
1071        .iter()
1072        .any(|(_, state)| matches!(state, Some(CheckState::Pending | CheckState::NoVerdict)));
1073    let unseen = matches!(declared, Declared::Unknown) && standing.is_empty();
1074    // The grace window, elapsed: where the host would not say what it requires and
1075    // checks ran on the draft with none skipped, its own marking is the answer.
1076    if unseen
1077        && !checks.is_empty()
1078        && checks
1079            .iter()
1080            .all(|check| check.state() != CheckState::Skipped)
1081    {
1082        settled(emissions, Vec::new(), unread);
1083        return Ok(true);
1084    }
1085    if running || (not_run.is_empty() && !unseen) {
1086        return Err(unsettled(&standing));
1087    }
1088
1089    // The grace window elapsed with nothing required run: lift early, say so, and read
1090    // the host again as a ready change whose checks must run after the lift.
1091    let snapshot: Vec<Check> = answered.checks.clone();
1092    lift(host, publishing, change, emissions)?;
1093    let warned = publishing.drafts.warn_on_early_lift.unwrap_or(true);
1094    let grace = grace_seconds()?;
1095    if warned {
1096        eprintln!(
1097            "onevcs: warning: {} was lifted out of its draft before its required checks ran on \
1098             it ({}): none had run within {grace}s, which is what a workflow that skips drafts \
1099             looks like",
1100            change.url,
1101            not_run.join(", ")
1102        );
1103    }
1104    emissions.push(publishing.emission(
1105        EventKind::DraftLiftedEarly,
1106        json!({
1107            "url": change.url.to_string(),
1108            "id": change.id.0,
1109            "base": change.base,
1110            "awaited": not_run,
1111            "grace_seconds": grace,
1112            "warned": warned,
1113        }),
1114    ));
1115    let after = host.change_checks(change)?.checks;
1116    let from_the_draft = |check: &Check| !ran_after_the_lift(check, &snapshot);
1117    let counted: Vec<&Check> = after
1118        .iter()
1119        .filter(|check| !(from_the_draft(check) && check.state() == CheckState::Skipped))
1120        .collect();
1121    let standing = standings(&counted, &declared);
1122    if let Some(failed) = red(&counted, &standing) {
1123        return Err(failed);
1124    }
1125    let rerun = after.iter().any(|check| {
1126        !from_the_draft(check)
1127            && (standing.is_empty() || standing.iter().any(|(name, _)| *name == check.name))
1128    });
1129    if !rerun {
1130        return Err(Error::ChecksUnsettled {
1131            reason: format!(
1132                "the required checks on {} did not re-run after its draft was lifted: no run \
1133                 the host attached after the lift appeared, and the runs from while it was a \
1134                 draft were skipped, which is not a verdict. The likely cause is a workflow that \
1135                 does not trigger on `ready_for_review`",
1136                change.url
1137            ),
1138        });
1139    }
1140    if !through(&standing) {
1141        return Err(unsettled(&standing));
1142    }
1143    settled(emissions, skipped(&standing), unread);
1144    Ok(false)
1145}
1146
1147/// Whether `check` is a run the host attached after the lift rather than the draft's
1148/// run of it in `snapshot`, decided as the real publication decides it: by the run's
1149/// own identity, never by its conclusion or its status. A check the draft never
1150/// reported is new, and one it did report is new only where it reports a `started_at`
1151/// no draft-era run of that check reported, running or settled — so a consumer seeds a
1152/// re-run by giving it a start of its own.
1153fn ran_after_the_lift(check: &Check, snapshot: &[Check]) -> bool {
1154    let earlier: Vec<&Check> = snapshot
1155        .iter()
1156        .filter(|seen| seen.name == check.name)
1157        .collect();
1158    if earlier.is_empty() {
1159        return true;
1160    }
1161    check.started_at.as_ref().is_some_and(|started| {
1162        earlier
1163            .iter()
1164            .all(|seen| seen.started_at.as_ref() != Some(started))
1165    })
1166}
1167
1168/// The default grace window of the real publication, which is `onevcs`'s own and
1169/// private to it. A copy, so it is gated: `lifecycle.rs`'s
1170/// `the_grace_window_an_early_lift_records_is_the_one_onevcs_defaults_to` reads the
1171/// constant out of `onevcs`'s source and holds this to it.
1172const DEFAULT_DRAFT_GRACE_SECONDS: f64 = 120.0;
1173
1174/// The knob a real publication waits its grace window by.
1175const DRAFT_GRACE_ENV: &str = "ONEVCS_DRAFT_CHECKS_GRACE_SECONDS";
1176
1177/// The grace window a real publication would have waited out, as the operator set it
1178/// or the default — recorded on `draft-lifted-early` so the payload reads as the real
1179/// one does, though nothing here waits.
1180///
1181/// Validated as the real publication validates it, and refused by name where it is
1182/// not a finite number of seconds above zero: a provider that quietly took the default
1183/// would let a consumer's suite pass on a setting the real one refuses.
1184fn grace_seconds() -> Result<f64> {
1185    let Some(raw) = std::env::var_os(DRAFT_GRACE_ENV) else {
1186        return Ok(DEFAULT_DRAFT_GRACE_SECONDS);
1187    };
1188    let raw = raw.to_string_lossy().into_owned();
1189    let seconds: f64 = raw.trim().parse().map_err(|_| Error::Invalid {
1190        reason: format!("{DRAFT_GRACE_ENV} must be a number of seconds, not {raw:?}"),
1191    })?;
1192    if !seconds.is_finite() || seconds <= 0.0 {
1193        return Err(Error::Invalid {
1194            reason: format!(
1195                "{DRAFT_GRACE_ENV} must be a finite number of seconds above zero, not {raw:?}"
1196            ),
1197        });
1198    }
1199    Ok(seconds)
1200}
1201
1202/// The draft a publication asked for, as the two refusals next door spell it: "a
1203/// draft awaiting <repository> <target>", or "a draft held by the session that opened
1204/// it (<why>)". A restatement, so the refusals here read as the real ones do.
1205fn asked_for(reason: &DraftReason) -> String {
1206    match reason {
1207        DraftReason::AwaitingRelease {
1208            awaiting, target, ..
1209        } => format!("a draft awaiting {awaiting} {target}"),
1210        DraftReason::Held { because } => {
1211            format!("a draft held by the session that opened it ({because})")
1212        }
1213    }
1214}
1215
1216/// What a publication answers for an identity no change request can be opened
1217/// against.
1218///
1219/// Two failures rather than one, as the real implementation keeps them: an
1220/// identity that is not hosted at all is asking for the wrong policy, while a
1221/// hosted one on a host this build does not speak for is asking for an
1222/// implementation that has not arrived.
1223fn refusal(identity: &str) -> PublishOutcome {
1224    failed(&if identity.split('/').count() == 3 {
1225        Error::NotImplemented {
1226            operation: "RemoteHost for a host other than github.com",
1227        }
1228    } else {
1229        Error::Invalid {
1230            reason: format!(
1231                "identity {identity:?} is not a hosted repository, so it cannot publish a \
1232                 change request; a local identity publishes with local-direct"
1233            ),
1234        }
1235    })
1236}
1237
1238/// One failure, as the outcome a publication that started and did not land is.
1239fn failed(error: &Error) -> PublishOutcome {
1240    PublishOutcome::Failed {
1241        // Through the crate's own mapping, so the kind a caller branches on is the
1242        // one the real implementation would report for the same failure.
1243        kind: FailureKind::of(error),
1244        reason: error.to_string(),
1245        // A provider has no execution checkout, so there is nowhere a branch could
1246        // have been handed back to and nothing to report about one.
1247        retained: None,
1248    }
1249}
1250
1251/// The argv that lands a preserved branch, as `recoverable` reports it.
1252///
1253/// Both verbs take the branch by name and the repository by path, the way the real
1254/// implementation reports them: the provenance decides which one, and neither
1255/// depends on the directory a reader happens to be standing in.
1256// llmlint: ignore[names_match_behavior] named for the public
1257// `Recoverable::recover_command` field it fills, which `onevcs` publishes and this
1258// crate must answer identically; a provider whose helper were named otherwise would
1259// read as filling something else.
1260fn recover_command(branch: &str, checkout: &Path, provenance: Provenance) -> Vec<String> {
1261    let verb = match provenance {
1262        Provenance::IncompleteStep => "recover",
1263        Provenance::Complete => "publish-branch",
1264    };
1265    vec![
1266        "onevcs".to_owned(),
1267        verb.to_owned(),
1268        branch.to_owned(),
1269        "--repo".to_owned(),
1270        checkout.display().to_string(),
1271    ]
1272}
1273
1274/// How a provenance kind is spelled in an event payload.
1275fn spell(provenance: Provenance) -> &'static str {
1276    match provenance {
1277        Provenance::Complete => "complete",
1278        Provenance::IncompleteStep => "incomplete-step",
1279    }
1280}
1281
1282fn object(value: Value) -> Map<String, Value> {
1283    value.as_object().cloned().unwrap_or_default()
1284}