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: 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        payload: object(json!({"identity": identity, "sha": sha, "base": session.base})),
610    };
611    (PublishOutcome::Merged(Sha(sha)), vec![emission])
612}
613
614/// Everything one provider publication knows about itself.
615struct Publishing<'a> {
616    slug: &'a str,
617    identity: &'a str,
618    session: &'a Session,
619    policy: MergePolicy,
620    approvals: Approvals,
621    drafts: Drafts,
622    request: &'a PublishRequest,
623    token: &'a SessionToken,
624}
625
626impl Publishing<'_> {
627    /// One event on the session's own stream, under this identity.
628    fn emission(&self, kind: EventKind, payload: Value) -> Emission {
629        Emission {
630            stream: self.token.0.clone(),
631            identity: Some(self.identity.to_owned()),
632            kind,
633            payload: object(payload),
634        }
635    }
636}
637
638/// Publish as a change request: open the session's change on the host, or adopt
639/// the one it already holds, watch its required checks, and then do with it what
640/// the policy asks.
641///
642/// **The draft lifecycle, as next door.** A publication carrying no reason opens its
643/// change request as a draft while the checks run — or, adopting a draft, records it
644/// as awaiting them rather than lifting it on the spot — and once they are green lifts
645/// it before asking for any merge, or keeps it under `change-open` with approvals
646/// required. `drafts: {disabled: true}` opens ready and lifts an adopted draft at once.
647///
648/// **What this provider cannot do is wait.** It has no clock, so each phase of the
649/// watch is **one reading** of the host: a reading that has not settled is the bound
650/// elapsing (`checks-unsettled`), and a draft on which no required check has run is
651/// the grace window elapsing — it is lifted early, and the next reading is the host's
652/// answer after the lift, which is where [`HostState::checks_after_lift`](crate::HostState::checks_after_lift)
653/// comes in. A consumer drives each row of the lifecycle by seeding the reading it
654/// wants rather than by timing one.
655fn publish_as_change(
656    hosting: &dyn Hosting,
657    publishing: &Publishing<'_>,
658    emissions: &mut Vec<Emission>,
659) -> Result<PublishOutcome> {
660    let Publishing {
661        slug,
662        session,
663        policy,
664        request,
665        ..
666    } = *publishing;
667    let host = hosting.for_repo(slug)?;
668    // Who the host believes is calling travels with the change, as it does in the
669    // real publication and for the same reason.
670    let author = host.authenticated_user()?;
671    let lifecycle = !publishing.drafts.disabled.unwrap_or(false);
672    // Checked before anything is opened, as the real publication checks it before
673    // anything is pushed.
674    if lifecycle {
675        grace_seconds()?;
676    }
677    let existing = host.find_changes(&session.branch, &session.base)?;
678    // Whether the host already held it, which is what decides whether there can be a
679    // draft to lift — and, as next door, keeps a publication that opens its own change
680    // request from asking this host anything extra.
681    let adopted = !existing.is_empty();
682    let change = match existing.into_iter().next() {
683        Some(change) => change,
684        None => host.open_change(ChangeSpec {
685            head: session.branch.clone(),
686            base: session.base.clone(),
687            // A requested title has been checked by the conversion that built it, so
688            // this provider cannot accept one the real publication would refuse. The
689            // real implementation takes the subject from the branch's commits when no
690            // title was requested, and a provider has no commits to read — so an
691            // unrequested title names the branch instead.
692            title: request
693                .title
694                .as_deref()
695                .map_or_else(|| format!("Publish {}", session.branch), str::to_owned),
696            // Verbatim, and nothing when there is none — the real publication
697            // composes no body either, so a provider that composed one would be a
698            // consumer's suite proving a change request nobody opens.
699            body: request.body.clone(),
700            // The reason travels to the host and no further, as it does there: what
701            // the host does with it is open the change as a draft.
702            draft: request.draft.clone(),
703            draft_awaiting_checks: lifecycle && request.draft.is_none(),
704        })?,
705    };
706    emissions.push(publishing.emission(
707        EventKind::ChangeOpened,
708        json!({
709            "url": change.url.to_string(),
710            "host": "github",
711            "id": change.id.0,
712            "base": change.base,
713            "author": author,
714        }),
715    ));
716    if let Some(reason) = &request.draft {
717        // The same question the real publication asks, for the same reason: a host
718        // that ignored the request, or a change request already open for review,
719        // would otherwise be reported as held back while it can land.
720        if !host.is_draft(&change)? {
721            return Err(Error::Invalid {
722                reason: format!(
723                    "{url} is open for review on the host, and this publication asked for \
724                     {asked}. A change that is open can land, so reporting it as a draft would \
725                     say the work is held back when nothing is holding it",
726                    url = change.url,
727                    asked = asked_for(reason),
728                ),
729            });
730        }
731        // The publication record, and the only place the reason is written: the real
732        // implementation writes nothing of it into the change request, and a provider
733        // that did would be a consumer's suite proving a body nobody renders. The
734        // payload is the reason's own serialized form — its `kind` and every field of
735        // that kind — under the change request it holds, exactly as next door.
736        let mut drafted = object(json!({
737            "url": change.url.to_string(),
738            "id": change.id.0,
739            "base": change.base,
740        }));
741        if let serde_json::Value::Object(fields) =
742            serde_json::to_value(reason).expect("a draft reason serializes")
743        {
744            drafted.extend(fields);
745        }
746        emissions.push(publishing.emission(EventKind::ChangeDrafted, Value::Object(drafted)));
747        // Under every policy: a draft is unmergeable in that state, so nothing below
748        // asks this host to merge it.
749        return Ok(PublishOutcome::ChangeDraft(change.url.clone()));
750    }
751    // Three answers about whether the host holds it as a draft, told apart as next
752    // door: a host that was never taught to draft one is holding nothing, and a host
753    // that *could not say* is a refusal rather than a change nobody is holding.
754    // Asked of a change this publication just opened only under the lifecycle, which
755    // asked for a draft: without it, one opened moments ago without a reason is one
756    // nobody drafted, and a host written against the earlier surface is asked nothing.
757    let is_draft = match if lifecycle || adopted {
758        host.is_draft(&change)
759    } else {
760        Ok(false)
761    } {
762        Ok(draft) => draft,
763        Err(Error::NotImplemented { .. }) => false,
764        Err(unreadable) => return Err(unreadable),
765    };
766    let drafted = if lifecycle {
767        if is_draft {
768            emissions.push(publishing.emission(
769                EventKind::ChangeDrafted,
770                json!({
771                    "url": change.url.to_string(),
772                    "id": change.id.0,
773                    "base": change.base,
774                    "kind": "awaiting-checks",
775                }),
776            ));
777        }
778        is_draft
779    } else {
780        // Without the lifecycle, publishing without a reason lifts an adopted draft
781        // on the spot, as it always did; one opened moments ago is one nobody drafted.
782        if adopted && is_draft {
783            lift(host.as_ref(), publishing, &change, emissions)?;
784        }
785        false
786    };
787
788    // Every change policy watches — except that `change-auto` on a change nobody
789    // drafted arms the host's own merge and leaves the checks to it, as next door.
790    // Next door's merge watch still refuses a required check that already concluded
791    // red rather than arm a merge over it, so this reading does too; one still
792    // running arms it, and the host's hold is what the outcome reports.
793    let still_draft = if drafted || policy != MergePolicy::ChangeAuto {
794        watch(host.as_ref(), publishing, &change, drafted, emissions)?
795    } else {
796        let answered = host.change_checks(&change)?;
797        let checks: Vec<&Check> = answered.checks.iter().collect();
798        if let Some(failed) = red(&checks, &standings(&checks, &Declared::Unknown)) {
799            return Err(failed);
800        }
801        false
802    };
803    if still_draft {
804        if policy == MergePolicy::ChangeOpen && publishing.approvals == Approvals::Required {
805            emissions.push(publishing.emission(
806                EventKind::DraftKeptForReview,
807                json!({
808                    "url": change.url.to_string(),
809                    "id": change.id.0,
810                    "base": change.base,
811                }),
812            ));
813            return Ok(PublishOutcome::ChangeReviewDraft(change.url.clone()));
814        }
815        // Lifted before any merge is asked for: this host, like GitHub, will neither
816        // merge a draft nor arm its own merge on one.
817        lift(host.as_ref(), publishing, &change, emissions)?;
818    }
819
820    if policy == MergePolicy::ChangeOpen {
821        return Ok(PublishOutcome::ChangeOpen(change.url.clone()));
822    }
823    Ok(match host.merge(&change, policy)? {
824        MergeOutcome::Merged(sha) => {
825            emissions.push(publishing.emission(
826                EventKind::ChangeMerged,
827                json!({"url": change.url.to_string(), "sha": sha.0}),
828            ));
829            emissions.push(publishing.emission(
830                EventKind::MergeCompleted,
831                json!({"identity": publishing.identity, "sha": sha.0}),
832            ));
833            PublishOutcome::Merged(sha)
834        }
835        MergeOutcome::Queued => PublishOutcome::Queued(change.url.clone()),
836        MergeOutcome::Open => PublishOutcome::ChangeOpen(change.url.clone()),
837    })
838}
839
840/// Take a change out of its draft on the host, and record the lift.
841fn lift(
842    host: &dyn RemoteHost,
843    publishing: &Publishing<'_>,
844    change: &ChangeRequest,
845    emissions: &mut Vec<Emission>,
846) -> Result<()> {
847    host.ready_for_review(change)?;
848    emissions.push(publishing.emission(
849        EventKind::DraftLifted,
850        json!({"url": change.url.to_string(), "id": change.id.0}),
851    ));
852    Ok(())
853}
854
855/// Which checks a merge requires, as the host said: nothing, these names, or unknown.
856enum Declared {
857    Nothing,
858    Names(BTreeSet<String>),
859    Unknown,
860}
861
862/// Each required check's name and the state its entries add up to — `None` where the
863/// host reported none — ranked as next door: red, then running, then passed, then no
864/// verdict, then skipped.
865fn standings(checks: &[&Check], declared: &Declared) -> Vec<(String, Option<CheckState>)> {
866    let names: BTreeSet<String> = match declared {
867        Declared::Names(names) => names.clone(),
868        Declared::Nothing => BTreeSet::new(),
869        Declared::Unknown => checks
870            .iter()
871            .filter(|check| check.required)
872            .map(|check| check.name.clone())
873            .collect(),
874    };
875    let rank = |state: CheckState| match state {
876        CheckState::Failed => 0,
877        CheckState::Pending => 1,
878        CheckState::Passed => 2,
879        CheckState::NoVerdict => 3,
880        CheckState::Skipped => 4,
881    };
882    names
883        .into_iter()
884        .map(|name| {
885            let state = checks
886                .iter()
887                .filter(|check| check.name == name)
888                .map(|check| check.state())
889                .min_by_key(|state| rank(*state));
890            (name, state)
891        })
892        .collect()
893}
894
895/// The refusal for the first required check that concluded red.
896fn red(checks: &[&Check], standing: &[(String, Option<CheckState>)]) -> Option<Error> {
897    let (name, _) = standing
898        .iter()
899        .find(|(_, state)| *state == Some(CheckState::Failed))?;
900    let check = checks
901        .iter()
902        .find(|check| &check.name == name && check.state() == CheckState::Failed)?;
903    Some(Error::ChecksFailed {
904        reason: format!(
905            "required check {:?} concluded {}.",
906            check.name,
907            check
908                .conclusion
909                .as_deref()
910                .unwrap_or("without a conclusion")
911        ),
912    })
913}
914
915/// Whether every required check lets the change through: passed, or skipped.
916fn through(standing: &[(String, Option<CheckState>)]) -> bool {
917    standing
918        .iter()
919        .all(|(_, state)| matches!(state, Some(CheckState::Passed | CheckState::Skipped)))
920}
921
922/// The skipped required checks, by name.
923fn skipped(standing: &[(String, Option<CheckState>)]) -> Vec<String> {
924    standing
925        .iter()
926        .filter(|(_, state)| *state == Some(CheckState::Skipped))
927        .map(|(name, _)| name.clone())
928        .collect()
929}
930
931/// The watch, one reading per phase: `Ok(true)` is green on a draft still standing,
932/// `Ok(false)` green on a ready change. See [`publish_as_change`] for what a reading
933/// stands for.
934fn watch(
935    host: &dyn RemoteHost,
936    publishing: &Publishing<'_>,
937    change: &ChangeRequest,
938    drafted: bool,
939    emissions: &mut Vec<Emission>,
940) -> Result<bool> {
941    // `unread`: the requirement was read from the host's own per-check marking because
942    // its declaration could not be read, which the record says, as next door.
943    let settled = |emissions: &mut Vec<Emission>, skipped: Vec<String>, unread: Option<&str>| {
944        let verdict = if skipped.is_empty() {
945            "passed"
946        } else {
947            "passed-with-skipped"
948        };
949        let mut payload = json!({
950            "url": change.url.to_string(),
951            "id": change.id.0,
952            "head": change.head_sha.0,
953            "verdict": verdict,
954            "skipped": skipped,
955        });
956        if let Some(because) = unread {
957            eprintln!(
958                "onevcs: warning: the required checks on {} were read from the host's own \
959                 per-check marking, because the declaration could not be read: {because}",
960                change.url
961            );
962            payload["requirement"] = json!({"read_from": "host-marking", "because": because});
963        }
964        emissions.push(publishing.emission(EventKind::ChecksSettled, payload));
965    };
966    let unsettled = |standing: &[(String, Option<CheckState>)]| Error::ChecksUnsettled {
967        reason: format!(
968            "the host had not settled its required checks on {}; still unsettled: {}",
969            change.url,
970            standing
971                .iter()
972                .filter(|(_, state)| !matches!(state, Some(CheckState::Passed)))
973                .map(|(name, _)| format!("{name:?}"))
974                .collect::<Vec<_>>()
975                .join(", ")
976        ),
977    };
978    let answered = host.change_checks(change)?;
979    let complete = answered.complete();
980    let checks: Vec<&Check> = answered.checks.iter().collect();
981
982    if !drafted {
983        let standing = standings(&checks, &Declared::Unknown);
984        if let Some(failed) = red(&checks, &standing) {
985            return Err(failed);
986        }
987        if !through(&standing) {
988            return Err(unsettled(&standing));
989        }
990        settled(emissions, skipped(&standing), None);
991        return Ok(false);
992    }
993
994    // The draft: what the host requires, and a skipped check has not run.
995    let (mut declared, mut unread) = match host.required_checks_on(&change.base) {
996        Ok(answer) if answer.checks.is_empty() && answer.complete() => (Declared::Nothing, None),
997        Ok(answer) if !answer.checks.is_empty() => (Declared::Names(answer.checks), None),
998        Ok(_) => (
999            Declared::Unknown,
1000            Some(format!(
1001                "the host answered only in part about which checks a merge into {} requires",
1002                change.base
1003            )),
1004        ),
1005        Err(refused) => (
1006            Declared::Unknown,
1007            Some(format!(
1008                "the host would not say which checks a merge into {} requires: {refused}",
1009                change.base
1010            )),
1011        ),
1012    };
1013    if matches!(declared, Declared::Unknown)
1014        && complete
1015        && !checks.is_empty()
1016        && checks.iter().all(|check| !check.required)
1017    {
1018        declared = Declared::Nothing;
1019        unread = None;
1020    }
1021    if matches!(declared, Declared::Nothing) {
1022        settled(emissions, Vec::new(), None);
1023        return Ok(true);
1024    }
1025    let unread = unread.as_deref();
1026    let standing = standings(&checks, &declared);
1027    if let Some(failed) = red(&checks, &standing) {
1028        return Err(failed);
1029    }
1030    if !standing.is_empty()
1031        && standing
1032            .iter()
1033            .all(|(_, state)| *state == Some(CheckState::Passed))
1034    {
1035        settled(emissions, Vec::new(), unread);
1036        return Ok(true);
1037    }
1038    let not_run: Vec<String> = standing
1039        .iter()
1040        .filter(|(_, state)| matches!(state, None | Some(CheckState::Skipped)))
1041        .map(|(name, _)| name.clone())
1042        .collect();
1043    let running = standing
1044        .iter()
1045        .any(|(_, state)| matches!(state, Some(CheckState::Pending | CheckState::NoVerdict)));
1046    let unseen = matches!(declared, Declared::Unknown) && standing.is_empty();
1047    // The grace window, elapsed: where the host would not say what it requires and
1048    // checks ran on the draft with none skipped, its own marking is the answer.
1049    if unseen
1050        && !checks.is_empty()
1051        && checks
1052            .iter()
1053            .all(|check| check.state() != CheckState::Skipped)
1054    {
1055        settled(emissions, Vec::new(), unread);
1056        return Ok(true);
1057    }
1058    if running || (not_run.is_empty() && !unseen) {
1059        return Err(unsettled(&standing));
1060    }
1061
1062    // The grace window elapsed with nothing required run: lift early, say so, and read
1063    // the host again as a ready change whose checks must run after the lift.
1064    let snapshot: Vec<Check> = answered.checks.clone();
1065    lift(host, publishing, change, emissions)?;
1066    let warned = publishing.drafts.warn_on_early_lift.unwrap_or(true);
1067    let grace = grace_seconds()?;
1068    if warned {
1069        eprintln!(
1070            "onevcs: warning: {} was lifted out of its draft before its required checks ran on \
1071             it ({}): none had run within {grace}s, which is what a workflow that skips drafts \
1072             looks like",
1073            change.url,
1074            not_run.join(", ")
1075        );
1076    }
1077    emissions.push(publishing.emission(
1078        EventKind::DraftLiftedEarly,
1079        json!({
1080            "url": change.url.to_string(),
1081            "id": change.id.0,
1082            "base": change.base,
1083            "awaited": not_run,
1084            "grace_seconds": grace,
1085            "warned": warned,
1086        }),
1087    ));
1088    let after = host.change_checks(change)?.checks;
1089    let from_the_draft = |check: &Check| !ran_after_the_lift(check, &snapshot);
1090    let counted: Vec<&Check> = after
1091        .iter()
1092        .filter(|check| !(from_the_draft(check) && check.state() == CheckState::Skipped))
1093        .collect();
1094    let standing = standings(&counted, &declared);
1095    if let Some(failed) = red(&counted, &standing) {
1096        return Err(failed);
1097    }
1098    let rerun = after.iter().any(|check| {
1099        !from_the_draft(check)
1100            && (standing.is_empty() || standing.iter().any(|(name, _)| *name == check.name))
1101    });
1102    if !rerun {
1103        return Err(Error::ChecksUnsettled {
1104            reason: format!(
1105                "the required checks on {} did not re-run after its draft was lifted: no run \
1106                 the host attached after the lift appeared, and the runs from while it was a \
1107                 draft were skipped, which is not a verdict. The likely cause is a workflow that \
1108                 does not trigger on `ready_for_review`",
1109                change.url
1110            ),
1111        });
1112    }
1113    if !through(&standing) {
1114        return Err(unsettled(&standing));
1115    }
1116    settled(emissions, skipped(&standing), unread);
1117    Ok(false)
1118}
1119
1120/// Whether `check` is a run the host attached after the lift rather than the draft's
1121/// run of it in `snapshot`, decided as the real publication decides it: by the run's
1122/// own identity, never by its conclusion or its status. A check the draft never
1123/// reported is new, and one it did report is new only where it reports a `started_at`
1124/// no draft-era run of that check reported, running or settled — so a consumer seeds a
1125/// re-run by giving it a start of its own.
1126fn ran_after_the_lift(check: &Check, snapshot: &[Check]) -> bool {
1127    let earlier: Vec<&Check> = snapshot
1128        .iter()
1129        .filter(|seen| seen.name == check.name)
1130        .collect();
1131    if earlier.is_empty() {
1132        return true;
1133    }
1134    check.started_at.as_ref().is_some_and(|started| {
1135        earlier
1136            .iter()
1137            .all(|seen| seen.started_at.as_ref() != Some(started))
1138    })
1139}
1140
1141/// The default grace window of the real publication, which is `onevcs`'s own and
1142/// private to it. A copy, so it is gated: `lifecycle.rs`'s
1143/// `the_grace_window_an_early_lift_records_is_the_one_onevcs_defaults_to` reads the
1144/// constant out of `onevcs`'s source and holds this to it.
1145const DEFAULT_DRAFT_GRACE_SECONDS: f64 = 120.0;
1146
1147/// The knob a real publication waits its grace window by.
1148const DRAFT_GRACE_ENV: &str = "ONEVCS_DRAFT_CHECKS_GRACE_SECONDS";
1149
1150/// The grace window a real publication would have waited out, as the operator set it
1151/// or the default — recorded on `draft-lifted-early` so the payload reads as the real
1152/// one does, though nothing here waits.
1153///
1154/// Validated as the real publication validates it, and refused by name where it is
1155/// not a finite number of seconds above zero: a provider that quietly took the default
1156/// would let a consumer's suite pass on a setting the real one refuses.
1157fn grace_seconds() -> Result<f64> {
1158    let Some(raw) = std::env::var_os(DRAFT_GRACE_ENV) else {
1159        return Ok(DEFAULT_DRAFT_GRACE_SECONDS);
1160    };
1161    let raw = raw.to_string_lossy().into_owned();
1162    let seconds: f64 = raw.trim().parse().map_err(|_| Error::Invalid {
1163        reason: format!("{DRAFT_GRACE_ENV} must be a number of seconds, not {raw:?}"),
1164    })?;
1165    if !seconds.is_finite() || seconds <= 0.0 {
1166        return Err(Error::Invalid {
1167            reason: format!(
1168                "{DRAFT_GRACE_ENV} must be a finite number of seconds above zero, not {raw:?}"
1169            ),
1170        });
1171    }
1172    Ok(seconds)
1173}
1174
1175/// The draft a publication asked for, as the two refusals next door spell it: "a
1176/// draft awaiting <repository> <target>", or "a draft held by the session that opened
1177/// it (<why>)". A restatement, so the refusals here read as the real ones do.
1178fn asked_for(reason: &DraftReason) -> String {
1179    match reason {
1180        DraftReason::AwaitingRelease {
1181            awaiting, target, ..
1182        } => format!("a draft awaiting {awaiting} {target}"),
1183        DraftReason::Held { because } => {
1184            format!("a draft held by the session that opened it ({because})")
1185        }
1186    }
1187}
1188
1189/// What a publication answers for an identity no change request can be opened
1190/// against.
1191///
1192/// Two failures rather than one, as the real implementation keeps them: an
1193/// identity that is not hosted at all is asking for the wrong policy, while a
1194/// hosted one on a host this build does not speak for is asking for an
1195/// implementation that has not arrived.
1196fn refusal(identity: &str) -> PublishOutcome {
1197    failed(&if identity.split('/').count() == 3 {
1198        Error::NotImplemented {
1199            operation: "RemoteHost for a host other than github.com",
1200        }
1201    } else {
1202        Error::Invalid {
1203            reason: format!(
1204                "identity {identity:?} is not a hosted repository, so it cannot publish a \
1205                 change request; a local identity publishes with local-direct"
1206            ),
1207        }
1208    })
1209}
1210
1211/// One failure, as the outcome a publication that started and did not land is.
1212fn failed(error: &Error) -> PublishOutcome {
1213    PublishOutcome::Failed {
1214        // Through the crate's own mapping, so the kind a caller branches on is the
1215        // one the real implementation would report for the same failure.
1216        kind: FailureKind::of(error),
1217        reason: error.to_string(),
1218        // A provider has no execution checkout, so there is nowhere a branch could
1219        // have been handed back to and nothing to report about one.
1220        retained: None,
1221    }
1222}
1223
1224/// The argv that lands a preserved branch, as `recoverable` reports it.
1225///
1226/// Both verbs take the branch by name and the repository by path, the way the real
1227/// implementation reports them: the provenance decides which one, and neither
1228/// depends on the directory a reader happens to be standing in.
1229// llmlint: ignore[names_match_behavior] named for the public
1230// `Recoverable::recover_command` field it fills, which `onevcs` publishes and this
1231// crate must answer identically; a provider whose helper were named otherwise would
1232// read as filling something else.
1233fn recover_command(branch: &str, checkout: &Path, provenance: Provenance) -> Vec<String> {
1234    let verb = match provenance {
1235        Provenance::IncompleteStep => "recover",
1236        Provenance::Complete => "publish-branch",
1237    };
1238    vec![
1239        "onevcs".to_owned(),
1240        verb.to_owned(),
1241        branch.to_owned(),
1242        "--repo".to_owned(),
1243        checkout.display().to_string(),
1244    ]
1245}
1246
1247/// How a provenance kind is spelled in an event payload.
1248fn spell(provenance: Provenance) -> &'static str {
1249    match provenance {
1250        Provenance::Complete => "complete",
1251        Provenance::IncompleteStep => "incomplete-step",
1252    }
1253}
1254
1255fn object(value: Value) -> Map<String, Value> {
1256    value.as_object().cloned().unwrap_or_default()
1257}