Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/services/repos/src/land.rs

180 lines6,556 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Issues and pull requests replace intents and attempts1//! Landing a pull request: moving a repository's branch forward to a commit
Pull requests from branches2//! from a fork, or from another of its own branches.
Rust repos service with shipping; pull requests kept in the model3//!
4//! The Artifacts binding cannot write, so this speaks git's smart HTTP
5//! protocol directly. It asks the fork for a pack holding exactly the
6//! objects the target is missing, then pushes that pack to the target
7//! unchanged. No object is parsed or rebuilt along the way.
8
9use g1t_contracts::repos::GitAccess;
10use worker::js_sys::Uint8Array;
11use worker::{Error, Fetch, Headers, Method, Request, RequestInit, Result};
12
13const ZERO_ID: &str = "0000000000000000000000000000000000000000";
14const FLUSH: &[u8] = b"0000";
15/// Side-band channels: pack data, and fatal errors.
16const PACK_BAND: u8 = 1;
17const ERROR_BAND: u8 = 3;
18
19fn pkt_line(payload: &str) -> Vec<u8> {
20 format!("{:04x}{payload}", payload.len() + 4).into_bytes()
21}
22
23async fn post(access: &GitAccess, service: &str, body: Vec<u8>) -> Result<Vec<u8>> {
24 let headers = Headers::new();
25 headers.set("authorization", &format!("Bearer {}", access.token))?;
26 headers.set("content-type", &format!("application/x-{service}-request"))?;
27 headers.set("accept", &format!("application/x-{service}-result"))?;
28 let mut init = RequestInit::new();
29 init.with_method(Method::Post)
30 .with_headers(headers)
31 .with_body(Some(Uint8Array::from(body.as_slice()).into()));
32 let request = Request::new_with_init(&format!("{}/{service}", access.remote), &init)?;
33 let mut response = Fetch::Request(request).send().await?;
34 let bytes = response.bytes().await?;
35 if response.status_code() != 200 {
36 return Err(Error::RustError(format!(
37 "{service} returned {}: {}",
38 response.status_code(),
39 String::from_utf8_lossy(&bytes)
40 )));
41 }
42 Ok(bytes)
43}
44
45/// The payloads of the pkt-lines in `bytes`, and the offset where they stop.
Pull requests from branches46pub(crate) fn read_pkt_lines(bytes: &[u8]) -> (Vec<&[u8]>, usize) {
Rust repos service with shipping; pull requests kept in the model47 let mut lines = Vec::new();
48 let mut position = 0;
49 while position + 4 <= bytes.len() {
50 let Some(length) = std::str::from_utf8(&bytes[position..position + 4])
51 .ok()
52 .and_then(|hex| usize::from_str_radix(hex, 16).ok())
53 else {
54 break;
55 };
56 if length < 4 {
57 // Flush, delimiter and response-end packets carry no payload.
58 position += 4;
59 continue;
60 }
61 let end = (position + length).min(bytes.len());
62 lines.push(&bytes[position + 4..end]);
63 position = end;
64 }
65 (lines, position)
66}
67
68/// A pack holding everything reachable from `want` that is not reachable
69/// from `have`.
Catching up with main takes seconds when the two sides touched different files70pub(crate) async fn fetch_pack(source: &GitAccess, want: &str, have: Option<&str>) -> Result<Vec<u8>> {
Rust repos service with shipping; pull requests kept in the model71 // Side-band framing puts the pack in its own channel, so its exact bytes
72 // can be recovered. Without it the response ends in a stray flush packet
73 // that a receiver rejects as junk after the pack.
74 let mut body = pkt_line(&format!("want {want} side-band-64k\n"));
75 body.extend_from_slice(FLUSH);
76 if let Some(have) = have {
77 body.extend(pkt_line(&format!("have {have}\n")));
78 }
79 body.extend(pkt_line("done\n"));
80
81 let response = post(source, "git-upload-pack", body).await?;
Agents as a team: lifecycle, merge queue, billing and a new shell82 unpack_sideband(&response)
83}
84
85/// The pack in an upload-pack response that used side-band framing.
86pub(crate) fn unpack_sideband(response: &[u8]) -> Result<Vec<u8>> {
87 let (lines, _) = read_pkt_lines(response);
Rust repos service with shipping; pull requests kept in the model88 let mut pack = Vec::new();
89 for line in lines {
90 match line.first() {
91 Some(&PACK_BAND) => pack.extend_from_slice(&line[1..]),
92 Some(&ERROR_BAND) => {
93 return Err(Error::RustError(format!(
Pull requests from branches94 "the source refused the fetch: {}",
Rust repos service with shipping; pull requests kept in the model95 String::from_utf8_lossy(&line[1..])
96 )));
97 }
98 // ACK and NAK lines, and progress messages.
99 _ => {}
100 }
101 }
102 if !pack.starts_with(b"PACK") {
103 return Err(Error::RustError(format!(
Pull requests from branches104 "the source did not send a pack: {}",
Agents as a team: lifecycle, merge queue, billing and a new shell105 String::from_utf8_lossy(response)
Rust repos service with shipping; pull requests kept in the model106 )));
107 }
108 Ok(pack)
109}
110
111/// Updates `branch` on the target from `old` to `new`, sending `pack`.
112/// `Err(reason)` in the inner result means git refused the update, for
113/// example because the branch is no longer at `old`.
Agents as a team: lifecycle, merge queue, billing and a new shell114pub(crate) async fn push_pack(
Rust repos service with shipping; pull requests kept in the model115 target: &GitAccess,
116 branch: &str,
117 old: Option<&str>,
118 new: &str,
119 pack: Vec<u8>,
120) -> Result<std::result::Result<(), String>> {
Merge queue: tested states are deleted once their entry leaves121 update_ref(target, branch, old, new, Some(pack)).await
122}
123
124/// Removes `branch` from the target, if it is still at `old`.
125pub(crate) async fn delete_ref(
126 target: &GitAccess,
127 branch: &str,
128 old: &str,
129) -> Result<std::result::Result<(), String>> {
130 update_ref(target, branch, Some(old), ZERO_ID, None).await
131}
132
133/// One receive-pack command; a deletion sends no pack.
134async fn update_ref(
135 target: &GitAccess,
136 branch: &str,
137 old: Option<&str>,
138 new: &str,
139 pack: Option<Vec<u8>>,
140) -> Result<std::result::Result<(), String>> {
Rust repos service with shipping; pull requests kept in the model141 let reference = format!("refs/heads/{branch}");
Merge queue: tested states are deleted once their entry leaves142 let sends_pack = pack.is_some();
143 let capabilities = if sends_pack { "report-status" } else { "report-status delete-refs" };
Rust repos service with shipping; pull requests kept in the model144 let mut body = pkt_line(&format!(
Merge queue: tested states are deleted once their entry leaves145 "{} {new} {reference}\0 {capabilities}\n",
Rust repos service with shipping; pull requests kept in the model146 old.unwrap_or(ZERO_ID)
147 ));
148 body.extend_from_slice(FLUSH);
Merge queue: tested states are deleted once their entry leaves149 if let Some(pack) = pack {
150 body.extend(pack);
151 }
Rust repos service with shipping; pull requests kept in the model152
153 let response = post(target, "git-receive-pack", body).await?;
154 let (lines, _) = read_pkt_lines(&response);
155 let lines: Vec<String> = lines
156 .into_iter()
157 .map(|line| String::from_utf8_lossy(line).trim_end().to_owned())
158 .collect();
Merge queue: tested states are deleted once their entry leaves159 // With nothing to unpack a server may not say so.
160 let unpacked = !sends_pack || lines.iter().any(|line| line == "unpack ok");
Rust repos service with shipping; pull requests kept in the model161 let updated = lines.iter().any(|line| *line == format!("ok {reference}"));
162 Ok(if unpacked && updated {
163 Ok(())
164 } else {
165 Err(lines.join("; "))
166 })
167}
168
169/// Moves `branch` on `target` from `old` to `new`, a commit that exists in
170/// `source`. The caller must have checked that `new` descends from `old`.
171pub async fn fast_forward(
172 source: &GitAccess,
173 target: &GitAccess,
174 branch: &str,
175 old: Option<&str>,
176 new: &str,
177) -> Result<std::result::Result<(), String>> {
178 let pack = fetch_pack(source, new, old).await?;
179 push_pack(target, branch, old, new, pack).await
180}