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.

g1t.sh

git agents cloudflare rust forge mcp

Commit

The work service coordinates work on a repository: sessions of people's own harnesses (start is idempotent on the harness's session id, every update or batch of entries is a heartbeat that renews the session's claims, a session reads as idle after 15 minutes and ends as timed_out after 2 hours), session entries recorded once per key and shown in the pull request's session when the session is on one (session_entries is rebuilt so an entry can belong to a session alone), advisory expiring claims on issues that are never locks (claiming tells you who else is on it and what the issue is blocked by; release, hand-off with a note, and expiry noticed on the next read), a writer for blocked_by on update_issue that refuses other repositories' numbers, the issue itself and cycles, and get_work, which gathers an issue's definition of done, claims, sessions, pull requests with their checks, mergeability, overlaps and queue place, blockers, the issues it blocks, decisions and recent coordination events. It publishes session.started, session.ended, work.claimed, work.released, work.claim_expired, work.handoff_requested, dependency.changed (when blocked_by changes, and when a blocker closes or has its pull request merged) and overlap.detected (once for each side when a push or a new pull request starts an overlap, again only after it cleared), and the contracts and the TypeScript mirror list them.

syntaqxcommitted yesterdayParent85c083aBrowse files

1	+	//! Coordinating work between agents and people: sessions, claims and
2	+	//! dependencies.
3	+	//!
4	+	//! The issue is the unit of work. A session is one run of someone's
5	+	//! own harness (a Claude Code session, a Codex run) on a repository, by a
6	+	//! person's token, optionally attached to an issue, a pull request and a
7	+	//! branch. A claim is an advisory, expiring "I'm working on #N": never
8	+	//! a lock, so two can stand on one issue, and the second claimant is told
9	+	//! about the first. An issue's `blocked_by` names the issues that have to
10	+	//! land first.
11	+	//!
12	+	//! Hooks and telemetry are evidence, never truth: what git and g1t record
13	+	//! is. So sessions tolerate missed heartbeats, reconnects and duplicate
14	+	//! events: starting one is idempotent on the harness's own session id,
15	+	//! entries are idempotent on their `key`, and a session or claim that
16	+	//! went quiet is read as idle, ended or expired when it is next looked at.
17	+
18	+	use serde::{Deserialize, Serialize};
19	+
20	+	use super::{CheckStatus, Issue, Mergeable, NewSessionEntry, Overlap, PullStatus, QueueState, RequiredCheck, SessionEntry, State};
21	+	use crate::credentials::Principal;
22	+	use crate::events::Event;
23	+	use crate::repos::RepoPath;
24	+	use crate::{User, Viewer};
25	+
26	+	/// A session with no heartbeat for this long reads as idle.
27	+	pub const SESSION_IDLE_MS: u64 = 15 * 60 * 1000;
28	+	/// A session with no heartbeat for this long has ended (`timed_out`).
29	+	pub const SESSION_TIMEOUT_MS: u64 = 2 * 60 * 60 * 1000;
30	+	/// How long a claim lasts unless renewed, in minutes.
31	+	pub const CLAIM_DEFAULT_MINUTES: u32 = 30;
32	+	/// The longest a claim may be asked to last, in minutes.
33	+	pub const CLAIM_MAX_MINUTES: u32 = 240;
34	+	/// The most issues one issue can be blocked by.
35	+	pub const MAX_BLOCKED_BY: usize = 50;
36	+	/// The most characters a session's title, a claim's note or a reason keeps.
37	+	pub const MAX_NOTE_CHARS: usize = 500;
38	+	/// The most characters an entry's idempotency key may have.
39	+	pub const MAX_KEY_CHARS: usize = 200;
40	+
41	+	/// The event types about coordinating work, which `get_work` lists as an
42	+	/// issue's recent activity (with the merge queue's, once it sends them).
43	+	pub const COORDINATION_EVENTS: [&str; 11] = [
44	+	"session.started",
45	+	"session.ended",
46	+	"work.claimed",
47	+	"work.released",
48	+	"work.claim_expired",
49	+	"work.handoff_requested",
50	+	"dependency.changed",
51	+	"overlap.detected",
52	+	"integration.failed",
53	+	"integration.passed",
54	+	"decision.recorded",
55	+	];
56	+
57	+	/// The harness a session runs in.
58	+	#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
59	+	#[serde(rename_all = "snake_case")]
60	+	pub enum Harness {
61	+	ClaudeCode,
62	+	Codex,
63	+	Cursor,
64	+	Copilot,
65	+	Other,
66	+	}
67	+
68	+	impl Harness {
69	+	pub const ALL: [Harness; 5] = [Harness::ClaudeCode, Harness::Codex, Harness::Cursor, Harness::Copilot, Harness::Other];
70	+
71	+	pub fn as_str(self) -> &'static str {
72	+	match self {
73	+	Harness::ClaudeCode => "claude_code",
74	+	Harness::Codex => "codex",
75	+	Harness::Cursor => "cursor",
76	+	Harness::Copilot => "copilot",
77	+	Harness::Other => "other",
78	+	}
79	+	}
80	+
81	+	/// From its name, in either case, with `-` for `_`.
82	+	pub fn parse(name: &str) -> Option<Harness> {
83	+	let name = name.trim().to_ascii_lowercase().replace('-', "_");
84	+	Harness::ALL.into_iter().find(\|harness\| harness.as_str() == name)
85	+	}
86	+	}
87	+
88	+	/// Where a session stands.
89	+	#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
90	+	#[serde(rename_all = "snake_case")]
91	+	pub enum SessionState {
92	+	/// Heard from in the last 15 minutes.
93	+	Active,
94	+	/// Not heard from for 15 minutes; it may still come back.
95	+	Idle,
96	+	/// Ended, or not heard from for 2 hours (`end_reason` `timed_out`).
97	+	Ended,
98	+	}
99	+
100	+	impl SessionState {
101	+	pub fn as_str(self) -> &'static str {
102	+	match self {
103	+	SessionState::Active => "active",
104	+	SessionState::Idle => "idle",
105	+	SessionState::Ended => "ended",
106	+	}
107	+	}
108	+
109	+	pub fn parse(name: &str) -> Option<SessionState> {
110	+	match name.trim().to_ascii_lowercase().as_str() {
111	+	"active" => Some(SessionState::Active),
112	+	"idle" => Some(SessionState::Idle),
113	+	"ended" => Some(SessionState::Ended),
114	+	_ => None,
115	+	}
116	+	}
117	+
118	+	/// A session as it reads at `now`: `ended` when it was ended or has
119	+	/// been quiet for [`SESSION_TIMEOUT_MS`], `idle` after
120	+	/// [`SESSION_IDLE_MS`], else `active`.
121	+	pub fn at(ended: bool, last_seen_ms: u64, now_ms: u64) -> SessionState {
122	+	let quiet = now_ms.saturating_sub(last_seen_ms);
123	+	if ended \|\| quiet >= SESSION_TIMEOUT_MS {
124	+	SessionState::Ended
125	+	} else if quiet >= SESSION_IDLE_MS {
126	+	SessionState::Idle
127	+	} else {
128	+	SessionState::Active
129	+	}
130	+	}
131	+	}
132	+
133	+	/// One run of someone's own harness on a repository.
134	+	#[derive(Clone, Debug, Serialize, Deserialize)]
135	+	#[serde(rename_all = "camelCase")]
136	+	pub struct Session {
137	+	/// `ses_…`.
138	+	pub id: String,
139	+	pub repo_id: String,
140	+	/// The issue it works on, if it said.
141	+	pub issue: Option<u32>,
142	+	/// The pull request it works on, if it said.
143	+	pub pull: Option<u32>,
144	+	pub branch: Option<String>,
145	+	/// Whose token started it.
146	+	pub actor: Principal,
147	+	pub harness: Harness,
148	+	/// The harness's own id for the session, which makes starting it again
149	+	/// return this one.
150	+	pub harness_session_id: Option<String>,
151	+	pub title: Option<String>,
152	+	pub state: SessionState,
153	+	/// The commit it last said it was at.
154	+	pub head_commit: Option<String>,
155	+	/// RFC 3339.
156	+	pub started_at: String,
157	+	/// RFC 3339: its last heartbeat or change.
158	+	pub last_seen_at: String,
159	+	/// RFC 3339, once ended.
160	+	pub ended_at: Option<String>,
161	+	/// Why it ended: what the harness said, or `timed_out`.
162	+	pub end_reason: Option<String>,
163	+	}
164	+
165	+	/// Where a claim stands.
166	+	#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
167	+	#[serde(rename_all = "snake_case")]
168	+	pub enum ClaimState {
169	+	Active,
170	+	Released,
171	+	Expired,
172	+	HandedOff,
173	+	}
174	+
175	+	impl ClaimState {
176	+	pub fn as_str(self) -> &'static str {
177	+	match self {
178	+	ClaimState::Active => "active",
179	+	ClaimState::Released => "released",
180	+	ClaimState::Expired => "expired",
181	+	ClaimState::HandedOff => "handed_off",
182	+	}
183	+	}
184	+
185	+	pub fn parse(name: &str) -> Option<ClaimState> {
186	+	match name.trim().to_ascii_lowercase().as_str() {
187	+	"active" => Some(ClaimState::Active),
188	+	"released" => Some(ClaimState::Released),
189	+	"expired" => Some(ClaimState::Expired),
190	+	"handed_off" => Some(ClaimState::HandedOff),
191	+	_ => None,
192	+	}
193	+	}
194	+	}
195	+
196	+	/// An advisory, expiring "I'm working on #N". Never a lock.
197	+	#[derive(Clone, Debug, Serialize, Deserialize)]
198	+	#[serde(rename_all = "camelCase")]
199	+	pub struct Claim {
200	+	/// `clm_…`.
201	+	pub id: String,
202	+	pub repo_id: String,
203	+	/// The issue it is on.
204	+	pub issue: u32,
205	+	/// The session it was made from, whose heartbeats renew it.
206	+	pub session_id: Option<String>,
207	+	pub actor: Principal,
208	+	pub harness: Option<Harness>,
209	+	pub note: Option<String>,
210	+	pub state: ClaimState,
211	+	/// RFC 3339.
212	+	pub created_at: String,
213	+	/// RFC 3339: when it lapses unless renewed.
214	+	pub expires_at: String,
215	+	/// RFC 3339, once it stopped being active.
216	+	pub ended_at: Option<String>,
217	+	pub end_reason: Option<String>,
218	+	/// For a hand-off: the username it was handed to; none for anyone.
219	+	pub handoff_to: Option<String>,
220	+	}
221	+
222	+	/// How long a claim lasts, in minutes: the default when none was asked,
223	+	/// at least one and at most [`CLAIM_MAX_MINUTES`].
224	+	pub fn claim_minutes(asked: Option<u32>) -> u32 {
225	+	asked.unwrap_or(CLAIM_DEFAULT_MINUTES).clamp(1, CLAIM_MAX_MINUTES)
226	+	}
227	+
228	+	/// `blocked_by` as asked for `number`, de-duplicated in the order given,
229	+	/// or why it cannot be: one names `number` itself or is not an issue of
230	+	/// the repository (`issues`), there are too many, or it would close a
231	+	/// cycle through the blockers already recorded (`graph`: each issue's
232	+	/// `blocked_by`).
233	+	pub fn check_blocked_by(
234	+	number: u32,
235	+	asked: &[u32],
236	+	issues: &std::collections::HashSet<u32>,
237	+	graph: &std::collections::HashMap<u32, Vec<u32>>,
238	+	) -> Result<Vec<u32>, String> {
239	+	let mut wanted: Vec<u32> = Vec::new();
240	+	for blocker in asked {
241	+	if !wanted.contains(blocker) {
242	+	wanted.push(*blocker);
243	+	}
244	+	}
245	+	if wanted.len() > MAX_BLOCKED_BY {
246	+	return Err(format!("An issue can be blocked by at most {MAX_BLOCKED_BY} issues."));
247	+	}
248	+	if wanted.contains(&number) {
249	+	return Err(format!("#{number} cannot be blocked by itself."));
250	+	}
251	+	if let Some(missing) = wanted.iter().find(\|blocker\| !issues.contains(blocker)) {
252	+	return Err(format!("#{missing} is not an issue in this repository."));
253	+	}
254	+	// A cycle: some blocker is itself, through what blocks it, blocked by
255	+	// `number`.
256	+	let mut seen = std::collections::HashSet::new();
257	+	let mut stack: Vec<(u32, Vec<u32>)> = wanted.iter().map(\|blocker\| (blocker, vec![number, blocker])).collect();
258	+	while let Some((at, path)) = stack.pop() {
259	+	if !seen.insert(at) {
260	+	continue;
261	+	}
262	+	for next in graph.get(&at).into_iter().flatten() {
263	+	let mut longer = path.clone();
264	+	longer.push(*next);
265	+	if *next == number {
266	+	let chain = longer.iter().map(\|n\| format!("#{n}")).collect::<Vec<_>>().join(" → ");
267	+	return Err(format!("That would make a cycle: {chain}, where each is blocked by the next."));
268	+	}
269	+	stack.push((*next, longer));
270	+	}
271	+	}
272	+	Ok(wanted)
273	+	}
274	+
275	+	/// The items of an issue's "Definition of done" section: each bulleted or
276	+	/// numbered line under the heading, up to the next heading.
277	+	pub fn definition_of_done(body: &str) -> Vec<String> {
278	+	let mut items = Vec::new();
279	+	let mut inside = false;
280	+	for line in body.lines() {
281	+	let trimmed = line.trim();
282	+	if trimmed.starts_with('#') {
283	+	inside = trimmed.trim_start_matches('#').trim().eq_ignore_ascii_case("definition of done");
284	+	continue;
285	+	}
286	+	if !inside {
287	+	continue;
288	+	}
289	+	let item = trimmed
290	+	.strip_prefix("- ")
291	+	.or_else(\|\| trimmed.strip_prefix("* "))
292	+	.or_else(\|\| trimmed.split_once(". ").filter(\|(n, _)\| n.chars().all(\|c\| c.is_ascii_digit()) && !n.is_empty()).map(\|(_, rest)\| rest));
293	+	if let Some(item) = item {
294	+	let item = item.trim_start_matches("[ ] ").trim_start_matches("[x] ").trim_start_matches("[X] ").trim();
295	+	if !item.is_empty() {
296	+	items.push(item.to_owned());
297	+	}
298	+	}
299	+	}
300	+	items
301	+	}
302	+
303	+	/// The pairs of pull requests that overlap now, against those already
304	+	/// announced: which are new (to announce) and which cleared (to forget, so
305	+	/// that they are announced again if they come back).
306	+	pub fn overlap_changes(current: &[u32], announced: &[u32]) -> (Vec<u32>, Vec<u32>) {
307	+	let new = current.iter().filter(\|number\| !announced.contains(number)).copied().collect();
308	+	let cleared = announced.iter().filter(\|number\| !current.contains(number)).copied().collect();
309	+	(new, cleared)
310	+	}
311	+
312	+	// --- Arguments --------------------------------------------------------------
313	+
314	+	/// `start_session`. Returns `Outcome<Session>`: a new one, or the one
315	+	/// already started with the same `harness_session_id` (which then takes
316	+	/// the issue, pull request, branch, title and head given).
317	+	#[derive(Debug, Serialize, Deserialize)]
318	+	pub struct StartSessionArgs {
319	+	pub actor: User,
320	+	pub repo: RepoPath,
321	+	pub harness: Harness,
322	+	#[serde(default)]
323	+	pub harness_session_id: Option<String>,
324	+	#[serde(default)]
325	+	pub issue: Option<u32>,
326	+	#[serde(default)]
327	+	pub pull: Option<u32>,
328	+	#[serde(default)]
329	+	pub branch: Option<String>,
330	+	#[serde(default)]
331	+	pub title: Option<String>,
332	+	#[serde(default)]
333	+	pub head_commit: Option<String>,
334	+	}
335	+
336	+	/// `update_session`: changes whichever fields are given; with none, it is a
337	+	/// heartbeat. Either way it renews the session's active claims. Returns
338	+	/// `Outcome<Session>`.
339	+	#[derive(Debug, Serialize, Deserialize)]
340	+	pub struct UpdateSessionArgs {
341	+	pub actor: User,
342	+	pub repo: RepoPath,
343	+	pub id: String,
344	+	#[serde(default)]
345	+	pub issue: Option<u32>,
346	+	#[serde(default)]
347	+	pub pull: Option<u32>,
348	+	#[serde(default)]
349	+	pub branch: Option<String>,
350	+	#[serde(default)]
351	+	pub head_commit: Option<String>,
352	+	#[serde(default)]
353	+	pub title: Option<String>,
354	+	}
355	+
356	+	/// `end_session`. Returns `Outcome<Session>`; ending one already ended
357	+	/// returns it as it is.
358	+	#[derive(Debug, Serialize, Deserialize)]
359	+	pub struct EndSessionArgs {
360	+	pub actor: User,
361	+	pub repo: RepoPath,
362	+	pub id: String,
363	+	#[serde(default)]
364	+	pub reason: Option<String>,
365	+	}
366	+
367	+	/// `get_work_session`: a session and its entries after `after_seq`.
368	+	/// Returns `Outcome<SessionDetail>`.
369	+	#[derive(Debug, Serialize, Deserialize)]
370	+	pub struct WorkSessionArgs {
371	+	pub viewer: Viewer,
372	+	pub repo: RepoPath,
373	+	pub id: String,
374	+	#[serde(default)]
375	+	pub after_seq: u32,
376	+	}
377	+
378	+	#[derive(Clone, Debug, Serialize, Deserialize)]
379	+	pub struct SessionDetail {
380	+	pub session: Session,
381	+	/// Oldest first, numbered within the session.
382	+	pub entries: Vec<SessionEntry>,
383	+	}
384	+
385	+	/// `list_work_sessions`: a repository's sessions, most recently heard from
386	+	/// first. Returns `Outcome<Vec<Session>>`.
387	+	#[derive(Debug, Serialize, Deserialize)]
388	+	pub struct ListWorkSessionsArgs {
389	+	pub viewer: Viewer,
390	+	pub repo: RepoPath,
391	+	#[serde(default)]
392	+	pub state: Option<SessionState>,
393	+	#[serde(default)]
394	+	pub issue: Option<u32>,
395	+	#[serde(default)]
396	+	pub pull: Option<u32>,
397	+	}
398	+
399	+	/// An entry to record in a session. `key` makes recording it again a no-op.
400	+	#[derive(Clone, Debug, Serialize, Deserialize)]
401	+	pub struct KeyedSessionEntry {
402	+	#[serde(flatten)]
403	+	pub entry: NewSessionEntry,
404	+	#[serde(default)]
405	+	pub key: Option<String>,
406	+	}
407	+
408	+	/// `record_session_entries`. Returns `Outcome<Recorded>`.
409	+	#[derive(Debug, Serialize, Deserialize)]
410	+	pub struct RecordEntriesArgs {
411	+	pub actor: User,
412	+	pub repo: RepoPath,
413	+	pub id: String,
414	+	pub entries: Vec<KeyedSessionEntry>,
415	+	}
416	+
417	+	#[derive(Clone, Debug, Default, Serialize, Deserialize)]
418	+	pub struct Recorded {
419	+	/// Entries newly recorded.
420	+	pub recorded: u32,
421	+	/// Entries skipped because one with the same key was recorded before.
422	+	pub duplicates: u32,
423	+	}
424	+
425	+	/// `claim_issue`. Returns `Outcome<Claimed>`. Claiming an issue you hold
426	+	/// renews your claim (and changes its note or session when given).
427	+	#[derive(Debug, Serialize, Deserialize)]
428	+	pub struct ClaimArgs {
429	+	pub actor: User,
430	+	pub repo: RepoPath,
431	+	pub number: u32,
432	+	#[serde(default)]
433	+	pub session_id: Option<String>,
434	+	#[serde(default)]
435	+	pub note: Option<String>,
436	+	#[serde(default)]
437	+	pub ttl_minutes: Option<u32>,
438	+	}
439	+
440	+	#[derive(Clone, Debug, Serialize, Deserialize)]
441	+	#[serde(rename_all = "camelCase")]
442	+	pub struct Claimed {
443	+	pub claim: Claim,
444	+	/// Other active claims on the issue: someone else is on it too.
445	+	pub others: Vec<Claim>,
446	+	/// What the issue is blocked by, and where each stands.
447	+	pub blocked_by: Vec<Blocker>,
448	+	}
449	+
450	+	/// `release_claim`: gives up your active claim on an issue. Returns
451	+	/// `Outcome<Claim>`.
452	+	#[derive(Debug, Serialize, Deserialize)]
453	+	pub struct ReleaseArgs {
454	+	pub actor: User,
455	+	pub repo: RepoPath,
456	+	pub number: u32,
457	+	#[serde(default)]
458	+	pub reason: Option<String>,
459	+	}
460	+
461	+	/// `hand_off_issue`: ends your active claim on an issue as handed off, to
462	+	/// someone (`to`, a username) or anyone, with a note on where it stands.
463	+	/// Returns `Outcome<Claim>`.
464	+	#[derive(Debug, Serialize, Deserialize)]
465	+	pub struct HandOffArgs {
466	+	pub actor: User,
467	+	pub repo: RepoPath,
468	+	pub number: u32,
469	+	#[serde(default)]
470	+	pub to: Option<String>,
471	+	pub note: String,
472	+	}
473	+
474	+	/// `list_claims`: a repository's claims, newest first; active ones unless
475	+	/// `state` says otherwise. Returns `Outcome<Vec<Claim>>`.
476	+	#[derive(Debug, Serialize, Deserialize)]
477	+	pub struct ListClaimsArgs {
478	+	pub viewer: Viewer,
479	+	pub repo: RepoPath,
480	+	#[serde(default)]
481	+	pub state: Option<ClaimState>,
482	+	#[serde(default)]
483	+	pub issue: Option<u32>,
484	+	}
485	+
486	+	/// `get_work`: everything an agent needs to pick up an issue. Returns
487	+	/// `Outcome<WorkContext>`.
488	+	#[derive(Debug, Serialize, Deserialize)]
489	+	pub struct GetWorkArgs {
490	+	pub viewer: Viewer,
491	+	pub repo: RepoPath,
492	+	pub number: u32,
493	+	}
494	+
495	+	/// An issue in brief.
496	+	#[derive(Clone, Debug, Serialize, Deserialize)]
497	+	pub struct IssueBrief {
498	+	pub number: u32,
499	+	pub title: String,
500	+	pub state: State,
501	+	}
502	+
503	+	/// A pull request in brief.
504	+	#[derive(Clone, Debug, Serialize, Deserialize)]
505	+	pub struct PullBrief {
506	+	pub number: u32,
507	+	pub title: String,
508	+	pub status: PullStatus,
509	+	}
510	+
511	+	/// An issue another is blocked by, and where it stands.
512	+	#[derive(Clone, Debug, Serialize, Deserialize)]
513	+	pub struct Blocker {
514	+	pub number: u32,
515	+	pub title: String,
516	+	pub state: State,
517	+	/// Active claims on it.
518	+	pub claims: Vec<Claim>,
519	+	/// Its pull requests.
520	+	pub pulls: Vec<PullBrief>,
521	+	}
522	+
523	+	/// A pull request's place in the merge queue.
524	+	#[derive(Clone, Debug, Serialize, Deserialize)]
525	+	pub struct QueuePlace {
526	+	pub state: QueueState,
527	+	/// 1 for the front of the queue.
528	+	pub position: u32,
529	+	}
530	+
531	+	/// One of an issue's pull requests, with where it stands.
532	+	#[derive(Clone, Debug, Serialize, Deserialize)]
533	+	#[serde(rename_all = "camelCase")]
534	+	pub struct WorkPull {
535	+	pub number: u32,
536	+	pub title: String,
537	+	pub status: PullStatus,
538	+	pub branch: Option<String>,
539	+	pub head_commit: Option<String>,
540	+	pub author: Principal,
541	+	/// `failed` when the merge queue took it out, until its head moves.
542	+	pub check_status: Option<CheckStatus>,
543	+	/// The checks the branch it merges into requires, as they stand on its
544	+	/// head.
545	+	pub required_checks: Vec<RequiredCheck>,
546	+	pub mergeable: Mergeable,
547	+	/// Other pull requests in progress changing the same files.
548	+	pub overlaps: Vec<Overlap>,
549	+	/// Its place in the merge queue, while it is in it.
550	+	pub queue: Option<QueuePlace>,
551	+	}
552	+
553	+	/// Everything about an issue an agent needs to pick it up: the one call to
554	+	/// make before starting.
555	+	#[derive(Clone, Debug, Serialize, Deserialize)]
556	+	#[serde(rename_all = "camelCase")]
557	+	pub struct WorkContext {
558	+	pub issue: Issue,
559	+	/// The items under its "Definition of done" heading.
560	+	pub definition_of_done: Vec<String>,
561	+	/// The checks every pull request into the default branch must pass.
562	+	pub required_checks: Vec<String>,
563	+	/// Active claims on it.
564	+	pub claims: Vec<Claim>,
565	+	/// Sessions on it or its pull requests, most recently heard from first.
566	+	pub sessions: Vec<Session>,
567	+	pub pulls: Vec<WorkPull>,
568	+	/// What it is blocked by, and where each stands.
569	+	pub blocked_by: Vec<Blocker>,
570	+	/// The issues it blocks.
571	+	pub blocks: Vec<IssueBrief>,
572	+	/// Decision records on it and its pull requests, newest first.
573	+	#[serde(default)]
574	+	pub decisions: Vec<serde_json::Value>,
575	+	/// The latest `integration.failed` event about it, if any.
576	+	#[serde(default)]
577	+	pub last_integration_failure: Option<Event>,
578	+	/// The latest coordination events about it, newest first.
579	+	pub recent: Vec<Event>,
580	+	}
581	+
582	+	// --- Events -----------------------------------------------------------------
583	+
584	+	/// `session.started` and `session.ended`.
585	+	#[derive(Clone, Debug, Serialize, Deserialize)]
586	+	#[serde(rename_all = "camelCase")]
587	+	pub struct SessionEvent {
588	+	pub repo_id: String,
589	+	/// The pull request it works on, else its issue.
590	+	#[serde(default, skip_serializing_if = "Option::is_none")]
591	+	pub number: Option<u32>,
592	+	#[serde(default, skip_serializing_if = "Option::is_none")]
593	+	pub issue: Option<u32>,
594	+	pub session: Session,
595	+	}
596	+
597	+	impl SessionEvent {
598	+	pub fn of(session: &Session) -> SessionEvent {
599	+	SessionEvent {
600	+	repo_id: session.repo_id.clone(),
601	+	number: session.pull.or(session.issue),
602	+	issue: session.issue,
603	+	session: session.clone(),
604	+	}
605	+	}
606	+	}
607	+
608	+	/// `work.claimed`, `work.released`, `work.claim_expired` and
609	+	/// `work.handoff_requested`: `number` and `issue` are the issue claimed.
610	+	#[derive(Clone, Debug, Serialize, Deserialize)]
611	+	#[serde(rename_all = "camelCase")]
612	+	pub struct ClaimEvent {
613	+	pub repo_id: String,
614	+	pub number: u32,
615	+	pub issue: u32,
616	+	pub claim: Claim,
617	+	/// On `work.claimed`: the other active claims on the issue.
618	+	#[serde(default, skip_serializing_if = "Option::is_none")]
619	+	pub others: Option<Vec<Claim>>,
620	+	/// On `work.claimed`, when it takes over a claim handed off.
621	+	#[serde(default, skip_serializing_if = "Option::is_none")]
622	+	pub took_over: Option<String>,
623	+	/// On `work.released`.
624	+	#[serde(default, skip_serializing_if = "Option::is_none")]
625	+	pub reason: Option<String>,
626	+	/// On `work.handoff_requested`: who handed it off.
627	+	#[serde(default, skip_serializing_if = "Option::is_none")]
628	+	pub from: Option<String>,
629	+	/// On `work.handoff_requested`: who it is for; absent for anyone.
630	+	#[serde(default, skip_serializing_if = "Option::is_none")]
631	+	pub to: Option<String>,
632	+	#[serde(default, skip_serializing_if = "Option::is_none")]
633	+	pub note: Option<String>,
634	+	}
635	+
636	+	impl ClaimEvent {
637	+	pub fn of(claim: &Claim) -> ClaimEvent {
638	+	ClaimEvent {
639	+	repo_id: claim.repo_id.clone(),
640	+	number: claim.issue,
641	+	issue: claim.issue,
642	+	claim: claim.clone(),
643	+	others: None,
644	+	took_over: None,
645	+	reason: None,
646	+	from: None,
647	+	to: None,
648	+	note: None,
649	+	}
650	+	}
651	+	}
652	+
653	+	/// `dependency.changed`: an issue's `blocked_by` changed, or one of the
654	+	/// issues it is blocked by closed or had its pull request merged.
655	+	#[derive(Clone, Debug, Default, Serialize, Deserialize)]
656	+	#[serde(rename_all = "camelCase")]
657	+	pub struct DependencyChanged {
658	+	pub repo_id: String,
659	+	/// The issue whose dependencies changed.
660	+	pub number: u32,
661	+	pub issue: u32,
662	+	/// What it is blocked by now.
663	+	pub blocked_by: Vec<u32>,
664	+	/// Added to `blocked_by`.
665	+	pub added: Vec<u32>,
666	+	/// Taken out of `blocked_by`.
667	+	pub removed: Vec<u32>,
668	+	/// Blockers that just closed or had a pull request merged.
669	+	pub cleared: Vec<u32>,
670	+	/// Those of `blocked_by` still open: none means it is unblocked.
671	+	pub remaining: Vec<u32>,
672	+	/// When a merge cleared it: the pull request merged.
673	+	#[serde(default, skip_serializing_if = "Option::is_none")]
674	+	pub merged_pull: Option<u32>,
675	+	}
676	+
677	+	/// `overlap.detected`: a pull request's latest push makes it change files
678	+	/// another open or draft pull request changes. Sent once for each side
679	+	/// when a pair starts overlapping, and again only after it stopped.
680	+	#[derive(Clone, Debug, Serialize, Deserialize)]
681	+	#[serde(rename_all = "camelCase")]
682	+	pub struct OverlapDetected {
683	+	pub repo_id: String,
684	+	/// The pull request this event is about: `pull`.
685	+	pub number: u32,
686	+	pub pull: u32,
687	+	pub other_pull: u32,
688	+	/// The files both change.
689	+	pub files: Vec<String>,
690	+	#[serde(default, skip_serializing_if = "Option::is_none")]
691	+	pub issue: Option<u32>,
692	+	#[serde(default, skip_serializing_if = "Option::is_none")]
693	+	pub other_issue: Option<u32>,
694	+	}
695	+
696	+	#[cfg(test)]
697	+	mod tests {
698	+	use std::collections::{HashMap, HashSet};
699	+
700	+	use super::*;
701	+
702	+	fn issues(numbers: &[u32]) -> HashSet<u32> {
703	+	numbers.iter().copied().collect()
704	+	}
705	+
706	+	#[test]
707	+	fn a_session_goes_idle_then_ends_without_heartbeats() {
708	+	let start = 1_000_000;
709	+	assert_eq!(SessionState::at(false, start, start), SessionState::Active);
710	+	assert_eq!(SessionState::at(false, start, start + SESSION_IDLE_MS - 1), SessionState::Active);
711	+	assert_eq!(SessionState::at(false, start, start + SESSION_IDLE_MS), SessionState::Idle);
712	+	assert_eq!(SessionState::at(false, start, start + SESSION_TIMEOUT_MS), SessionState::Ended);
713	+	assert_eq!(SessionState::at(true, start, start), SessionState::Ended);
714	+	// A clock behind the last heartbeat reads as just heard from.
715	+	assert_eq!(SessionState::at(false, start, start - 5), SessionState::Active);
716	+	}
717	+
718	+	#[test]
719	+	fn claims_last_thirty_minutes_up_to_four_hours() {
720	+	assert_eq!(claim_minutes(None), 30);
721	+	assert_eq!(claim_minutes(Some(0)), 1);
722	+	assert_eq!(claim_minutes(Some(90)), 90);
723	+	assert_eq!(claim_minutes(Some(10_000)), 240);
724	+	}
725	+
726	+	#[test]
727	+	fn harnesses_are_named_one_way() {
728	+	assert_eq!(Harness::parse("claude-code"), Some(Harness::ClaudeCode));
729	+	assert_eq!(Harness::parse("Codex"), Some(Harness::Codex));
730	+	assert_eq!(Harness::parse("vim"), None);
731	+	assert_eq!(serde_json::to_value(Harness::ClaudeCode).unwrap(), "claude_code");
732	+	assert_eq!(serde_json::to_value(ClaimState::HandedOff).unwrap(), "handed_off");
733	+	}
734	+
735	+	#[test]
736	+	fn blocked_by_is_issues_of_the_repository_never_itself() {
737	+	let graph = HashMap::new();
738	+	assert_eq!(check_blocked_by(5, &[3, 4, 3], &issues(&[3, 4, 5]), &graph), Ok(vec![3, 4]));
739	+	assert_eq!(check_blocked_by(5, &[], &issues(&[5]), &graph), Ok(vec![]));
740	+	assert!(check_blocked_by(5, &[5], &issues(&[5]), &graph).unwrap_err().contains("itself"));
741	+	assert!(check_blocked_by(5, &[9], &issues(&[5]), &graph).unwrap_err().contains("#9 is not an issue"));
742	+	let many: Vec<u32> = (100..100 + MAX_BLOCKED_BY as u32 + 1).collect();
743	+	assert!(check_blocked_by(5, &many, &many.iter().copied().collect(), &graph).is_err());
744	+	}
745	+
746	+	#[test]
747	+	fn blocked_by_never_makes_a_cycle() {
748	+	// 3 is blocked by 4, which is blocked by 5.
749	+	let graph: HashMap<u32, Vec<u32>> = [(3, vec![4]), (4, vec![5])].into_iter().collect();
750	+	let all = issues(&[3, 4, 5, 6]);
751	+	let refused = check_blocked_by(5, &[3], &all, &graph).unwrap_err();
752	+	assert!(refused.contains("#5 → #3 → #4 → #5"), "{refused}");
753	+	// A diamond is not a cycle.
754	+	let diamond: HashMap<u32, Vec<u32>> = [(3, vec![5]), (4, vec![5])].into_iter().collect();
755	+	assert_eq!(check_blocked_by(6, &[3, 4], &all, &diamond), Ok(vec![3, 4]));
756	+	// Replacing 5's own blockers is judged without them.
757	+	let graph: HashMap<u32, Vec<u32>> = [(5, vec![3]), (3, vec![4])].into_iter().collect();
758	+	assert_eq!(check_blocked_by(5, &[4], &all, &graph), Ok(vec![4]));
759	+	}
760	+
761	+	#[test]
762	+	fn the_definition_of_done_is_read_from_its_section() {
763	+	let body = "Make it greet.\n\n## Definition of done\n\n- `npm test` passes.\n- [ ] Docs say so\n1. Works offline\n\n## Notes\n- not this";
764	+	assert_eq!(definition_of_done(body), ["`npm test` passes.", "Docs say so", "Works offline"]);
765	+	assert!(definition_of_done("No section").is_empty());
766	+	}
767	+
768	+	#[test]
769	+	fn an_overlap_is_announced_once_until_it_clears() {
770	+	assert_eq!(overlap_changes(&[7, 9], &[]), (vec![7, 9], vec![]));
771	+	// Pushing again with the same overlaps announces nothing.
772	+	assert_eq!(overlap_changes(&[7, 9], &[7, 9]), (vec![], vec![]));
773	+	// 9 stopped overlapping: forgotten, so a later overlap is new again.
774	+	assert_eq!(overlap_changes(&[7], &[7, 9]), (vec![], vec![9]));
775	+	assert_eq!(overlap_changes(&[7, 9], &[7]), (vec![9], vec![]));
776	+	}
777	+
778	+	#[test]
779	+	fn events_name_the_issue_or_pull_request_they_are_about() {
780	+	let session: Session = serde_json::from_value(serde_json::json!({
781	+	"id": "ses_1", "repoId": "rep_1", "issue": 4, "pull": 9, "branch": null,
782	+	"actor": { "id": "usr_1", "username": "ana" }, "harness": "claude_code",
783	+	"harnessSessionId": null, "title": null, "state": "active", "headCommit": null,
784	+	"startedAt": "", "lastSeenAt": "", "endedAt": null, "endReason": null
785	+	}))
786	+	.unwrap();
787	+	let event = serde_json::to_value(SessionEvent::of(&session)).unwrap();
788	+	assert_eq!((event["number"].as_u64(), event["issue"].as_u64()), (Some(9), Some(4)));
789	+	}
790	+	}

628	628		assignees?: string[];
629	629		/** The number of the milestone to put it in; 0 takes it out. */
630	630		milestone?: number;
	631	+	/** The issues of the same repository that have to land first; replaces the whole set. */
	632	+	blocked_by?: number[];
631	633		};
632	634
633	635		export type OpenPullInput = {
⋯
1052	1054		/** The first day counted, `YYYY-MM-DD`; the last is today (UTC). */
1053	1055		from: string;
1054	1056		};
	1057	+
	1058	+	// --- Coordinating work: sessions, claims and dependencies -----------------
	1059	+	// Mirrors `crates/contracts/src/work/coordination.rs`.
	1060	+
	1061	+	/** The harness a session runs in. */
	1062	+	export type Harness = "claude_code" \| "codex" \| "cursor" \| "copilot" \| "other";
	1063	+
	1064	+	/** Active: heard from in 15 minutes; idle: not since; ended: ended, or quiet for 2 hours. */
	1065	+	export type SessionState = "active" \| "idle" \| "ended";
	1066	+
	1067	+	/** One run of someone's own harness on a repository. */
	1068	+	export type WorkSession = {
	1069	+	/** `ses_…`. */
	1070	+	id: string;
	1071	+	repoId: string;
	1072	+	issue: number \| null;
	1073	+	pull: number \| null;
	1074	+	branch: string \| null;
	1075	+	actor: { id: string; username: string };
	1076	+	harness: Harness;
	1077	+	harnessSessionId: string \| null;
	1078	+	title: string \| null;
	1079	+	state: SessionState;
	1080	+	headCommit: string \| null;
	1081	+	startedAt: string;
	1082	+	lastSeenAt: string;
	1083	+	endedAt: string \| null;
	1084	+	/** What the harness said, or `timed_out`. */
	1085	+	endReason: string \| null;
	1086	+	};
	1087	+
	1088	+	export type ClaimState = "active" \| "released" \| "expired" \| "handed_off";
	1089	+
	1090	+	/** An advisory, expiring "I'm working on #N". Never a lock. */
	1091	+	export type Claim = {
	1092	+	/** `clm_…`. */
	1093	+	id: string;
	1094	+	repoId: string;
	1095	+	issue: number;
	1096	+	sessionId: string \| null;
	1097	+	actor: { id: string; username: string };
	1098	+	harness: Harness \| null;
	1099	+	note: string \| null;
	1100	+	state: ClaimState;
	1101	+	createdAt: string;
	1102	+	expiresAt: string;
	1103	+	endedAt: string \| null;
	1104	+	endReason: string \| null;
	1105	+	/** For a hand-off: who it was handed to; null for anyone. */
	1106	+	handoffTo: string \| null;
	1107	+	};
	1108	+
	1109	+	/** `session.started` and `session.ended`. */
	1110	+	export type SessionEventData = { repoId: string; number?: number; issue?: number; session: WorkSession };
	1111	+
	1112	+	/** `work.claimed`, `work.released`, `work.claim_expired` and `work.handoff_requested`. */
	1113	+	export type ClaimEventData = {
	1114	+	repoId: string;
	1115	+	number: number;
	1116	+	issue: number;
	1117	+	claim: Claim;
	1118	+	others?: Claim[];
	1119	+	tookOver?: string;
	1120	+	reason?: string;
	1121	+	from?: string;
	1122	+	to?: string;
	1123	+	note?: string;
	1124	+	};
	1125	+
	1126	+	/** `dependency.changed`. */
	1127	+	export type DependencyChangedData = {
	1128	+	repoId: string;
	1129	+	number: number;
	1130	+	issue: number;
	1131	+	blockedBy: number[];
	1132	+	added: number[];
	1133	+	removed: number[];
	1134	+	cleared: number[];
	1135	+	remaining: number[];
	1136	+	mergedPull?: number;
	1137	+	};
	1138	+
	1139	+	/** `overlap.detected`, sent once for each side of a new overlap. */
	1140	+	export type OverlapDetectedData = {
	1141	+	repoId: string;
	1142	+	number: number;
	1143	+	pull: number;
	1144	+	otherPull: number;
	1145	+	files: string[];
	1146	+	issue?: number;
	1147	+	otherIssue?: number;
	1148	+	};

1	+	-- Coordinating work (src/coordination/): sessions of people's own harnesses,
2	+	-- advisory claims on issues, and which pairs of pull requests have been
3	+	-- announced as overlapping.
4	+
5	+	-- One run of someone's own harness (a Claude Code session, a Codex run) on
6	+	-- a repository, by a person's token. `state` is stored as active or ended;
7	+	-- idle (15 minutes without a heartbeat) is worked out when read, and two
8	+	-- hours without one ends it as `timed_out`.
9	+	CREATE TABLE sessions (
10	+	id TEXT PRIMARY KEY,
11	+	repo_id TEXT NOT NULL,
12	+	-- The issue and pull request it works on, by number, if it said.
13	+	issue INTEGER,
14	+	pull INTEGER,
15	+	branch TEXT,
16	+	actor_id TEXT NOT NULL,
17	+	actor_name TEXT NOT NULL,
18	+	-- claude_code, codex, cursor, copilot or other.
19	+	harness TEXT NOT NULL,
20	+	-- The harness's own id for it: starting it again returns this row.
21	+	harness_session_id TEXT,
22	+	title TEXT,
23	+	state TEXT NOT NULL DEFAULT 'active',
24	+	head_commit TEXT,
25	+	started_at TEXT NOT NULL,
26	+	last_seen_at TEXT NOT NULL,
27	+	ended_at TEXT,
28	+	end_reason TEXT
29	+	);
30	+	CREATE UNIQUE INDEX sessions_by_harness_id ON sessions (repo_id, actor_id, harness, harness_session_id)
31	+	WHERE harness_session_id IS NOT NULL;
32	+	CREATE INDEX sessions_by_repo ON sessions (repo_id, state, last_seen_at);
33	+	CREATE INDEX sessions_by_issue ON sessions (repo_id, issue) WHERE issue IS NOT NULL;
34	+	CREATE INDEX sessions_by_pull ON sessions (repo_id, pull) WHERE pull IS NOT NULL;
35	+
36	+	-- A session's entries live beside the ones recorded on a pull request, so
37	+	-- `pull_id` may now be empty: an entry belongs to a pull request, a
38	+	-- session, or both. SQLite cannot loosen NOT NULL in place, so the table
39	+	-- is rebuilt with the same columns plus:
40	+	--
41	+	-- session_id the session it was recorded in.
42	+	-- session_seq its place in that session.
43	+	-- key the harness's idempotency key: recording an entry with
44	+	-- a key already recorded in the session does nothing.
45	+	--
46	+	-- An entry of a session that is on a pull request also takes the next
47	+	-- `seq` of that pull request, so it shows in the pull request's session as
48	+	-- the entries recorded on it do. One recorded before the session named
49	+	-- its pull request has `pull_id` NULL and `seq` = `session_seq` until then.
50	+	CREATE TABLE session_entries_next (
51	+	pull_id TEXT REFERENCES pulls (id),
52	+	seq INTEGER NOT NULL,
53	+	kind TEXT NOT NULL,
54	+	text TEXT NOT NULL,
55	+	tool TEXT,
56	+	"commit" TEXT,
57	+	at TEXT NOT NULL,
58	+	session_id TEXT,
59	+	session_seq INTEGER,
60	+	key TEXT,
61	+	CHECK (pull_id IS NOT NULL OR session_id IS NOT NULL)
62	+	);
63	+	INSERT INTO session_entries_next (pull_id, seq, kind, text, tool, "commit", at)
64	+	SELECT pull_id, seq, kind, text, tool, "commit", at FROM session_entries;
65	+	DROP TABLE session_entries;
66	+	ALTER TABLE session_entries_next RENAME TO session_entries;
67	+	CREATE UNIQUE INDEX session_entries_by_pull ON session_entries (pull_id, seq) WHERE pull_id IS NOT NULL;
68	+	CREATE UNIQUE INDEX session_entries_by_session ON session_entries (session_id, session_seq)
69	+	WHERE session_id IS NOT NULL;
70	+	CREATE UNIQUE INDEX session_entries_by_key ON session_entries (session_id, key)
71	+	WHERE session_id IS NOT NULL AND key IS NOT NULL;
72	+
73	+	-- An advisory, expiring "I'm working on #N": never a lock. Active until
74	+	-- released, handed off, or past `expires_at` (noticed when next read, and
75	+	-- then announced as expired). Claiming again, or a heartbeat of the
76	+	-- session it was made from, renews it by `ttl_minutes`.
77	+	CREATE TABLE claims (
78	+	id TEXT PRIMARY KEY,
79	+	repo_id TEXT NOT NULL,
80	+	issue INTEGER NOT NULL,
81	+	session_id TEXT,
82	+	actor_id TEXT NOT NULL,
83	+	actor_name TEXT NOT NULL,
84	+	harness TEXT,
85	+	note TEXT,
86	+	-- active, released, expired or handed_off.
87	+	state TEXT NOT NULL DEFAULT 'active',
88	+	ttl_minutes INTEGER NOT NULL,
89	+	created_at TEXT NOT NULL,
90	+	expires_at TEXT NOT NULL,
91	+	ended_at TEXT,
92	+	end_reason TEXT,
93	+	-- For a hand-off: the username it is for; NULL for anyone.
94	+	handoff_to TEXT
95	+	);
96	+	CREATE INDEX claims_by_issue ON claims (repo_id, issue, state);
97	+	CREATE INDEX claims_by_state ON claims (repo_id, state, expires_at);
98	+	CREATE INDEX claims_by_session ON claims (session_id, state) WHERE session_id IS NOT NULL;
99	+
100	+	-- Pairs of pull requests announced as changing the same files
101	+	-- (`overlap.detected`), lower number first: announced once, and again only
102	+	-- after the pair stopped overlapping and this row went.
103	+	CREATE TABLE overlap_pairs (
104	+	repo_id TEXT NOT NULL,
105	+	low INTEGER NOT NULL,
106	+	high INTEGER NOT NULL,
107	+	-- JSON array of the paths both changed when it was announced.
108	+	files TEXT NOT NULL,
109	+	noticed_at TEXT NOT NULL,
110	+	PRIMARY KEY (repo_id, low, high)
111	+	);

1	+	//! Claims: an advisory, expiring "I'm working on #N".
2	+	//!
3	+	//! A claim is never a lock: anyone may claim an issue someone else has
4	+	//! claimed, and is told who else is on it. It lapses unless renewed, by
5	+	//! claiming again or by a heartbeat of the session it was made from, and
6	+	//! ends when released, handed off, or when its issue closes.
7	+
8	+	use g1t_contracts::identity::UsernameArgs;
9	+	use g1t_contracts::time::rfc3339;
10	+	use g1t_contracts::work::*;
11	+	use g1t_contracts::{FailureCode, Outcome, User, Viewer, new_id};
12	+	use g1t_kit::now_ms;
13	+	use worker::Result;
14	+	use worker::wasm_bindgen::JsValue;
15	+
16	+	use super::{CLAIM_COLUMNS, ClaimRow, after_minutes, tidy, unwrap_or_fail};
17	+	use crate::Work;
18	+
19	+	/// The most claims one list returns.
20	+	const LIST_LIMIT: u32 = 100;
21	+
22	+	fn text_or_null(value: &Option<String>) -> JsValue {
23	+	value.as_deref().map_or(JsValue::NULL, JsValue::from)
24	+	}
25	+
26	+	impl Work {
27	+	/// The caller's active claim on an issue, if they hold one.
28	+	async fn own_claim(&self, repo_id: &str, issue: u32, actor: &User) -> Result<Option<Claim>> {
29	+	Ok(self.active_claims(repo_id, issue).await?.into_iter().find(\|claim\| claim.actor.id == actor.id))
30	+	}
31	+
32	+	pub(crate) async fn claim_issue(&self, a: ClaimArgs) -> Result<Outcome<Claimed>> {
33	+	let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?);
34	+	let Some(issue) = self.issue(&repo.id, a.number).await? else {
35	+	return Ok(Outcome::fail(FailureCode::NotFound, "Issue not found."));
36	+	};
37	+	if issue.state == State::Closed {
38	+	return Ok(Outcome::fail(FailureCode::Conflict, format!("#{} is closed.", issue.number)));
39	+	}
40	+	// The session it is made from, whose heartbeats renew it.
41	+	let session = match a.session_id.as_deref().map(str::trim).filter(\|id\| !id.is_empty()) {
42	+	Some(id) => match self.session_row(&repo.id, id).await? {
43	+	Some(row) if row.actor_id == a.actor.id => Some(row.at(now_ms())),
44	+	_ => return Ok(Outcome::fail(FailureCode::Invalid, "session_id is not one of your sessions in this repository.")),
45	+	},
46	+	None => None,
47	+	};
48	+	if session.as_ref().is_some_and(\|session\| session.state == SessionState::Ended) {
49	+	return Ok(Outcome::fail(FailureCode::Invalid, "That session has ended. Start a new one."));
50	+	}
51	+	let minutes = claim_minutes(a.ttl_minutes);
52	+	let note = tidy(a.note.as_deref(), MAX_NOTE_CHARS);
53	+	let session_id = session.as_ref().map(\|session\| session.id.clone());
54	+	let harness = session.as_ref().map(\|session\| session.harness.as_str().to_owned());
55	+	let now = now_ms();
56	+	let expires_at = after_minutes(now, minutes);
57	+	let standing = self.active_claims(&repo.id, issue.number).await?;
58	+	let (mine, others): (Vec<Claim>, Vec<Claim>) = standing.into_iter().partition(\|claim\| claim.actor.id == a.actor.id);
59	+	let blocked_by = self.blockers(&repo.id, &issue.blocked_by).await?;
60	+	// Claiming what you hold renews it.
61	+	if let Some(held) = mine.into_iter().next() {
62	+	let renewed = self
63	+	.db
64	+	.prepare(format!(
65	+	"UPDATE claims
66	+	SET expires_at = ?, ttl_minutes = ?, note = COALESCE(?, note),
67	+	session_id = COALESCE(?, session_id), harness = COALESCE(?, harness)
68	+	WHERE id = ? AND state = 'active'
69	+	RETURNING {CLAIM_COLUMNS}"
70	+	))
71	+	.bind(&[
72	+	expires_at.as_str().into(),
73	+	minutes.into(),
74	+	text_or_null(&note),
75	+	text_or_null(&session_id),
76	+	text_or_null(&harness),
77	+	held.id.as_str().into(),
78	+	])?
79	+	.first::<ClaimRow>(None)
80	+	.await?;
81	+	if let Some(renewed) = renewed {
82	+	return Ok(Outcome::Ok(Claimed { claim: renewed.into(), others, blocked_by }));
83	+	}
84	+	}
85	+	// Taking over what was handed off to you, or to anyone.
86	+	let handed: Option<ClaimRow> = self
87	+	.db
88	+	.prepare(format!(
89	+	"SELECT {CLAIM_COLUMNS} FROM claims
90	+	WHERE repo_id = ? AND issue = ? AND state <> 'active' ORDER BY ended_at DESC, id DESC LIMIT 1"
91	+	))
92	+	.bind(&[repo.id.as_str().into(), issue.number.into()])?
93	+	.first::<ClaimRow>(None)
94	+	.await?;
95	+	let took_over = handed
96	+	.map(Claim::from)
97	+	.filter(\|last\| last.state == ClaimState::HandedOff)
98	+	.filter(\|last\| last.handoff_to.as_deref().is_none_or(\|to\| to == a.actor.username))
99	+	.map(\|last\| last.id);
100	+	let at = rfc3339(now);
101	+	let created = self
102	+	.db
103	+	.prepare(format!(
104	+	"INSERT INTO claims
105	+	(id, repo_id, issue, session_id, actor_id, actor_name, harness, note, state, ttl_minutes,
106	+	created_at, expires_at)
107	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?)
108	+	RETURNING {CLAIM_COLUMNS}"
109	+	))
110	+	.bind(&[
111	+	new_id("clm", now).into(),
112	+	repo.id.as_str().into(),
113	+	issue.number.into(),
114	+	text_or_null(&session_id),
115	+	a.actor.id.as_str().into(),
116	+	a.actor.username.as_str().into(),
117	+	text_or_null(&harness),
118	+	text_or_null(&note),
119	+	minutes.into(),
120	+	at.as_str().into(),
121	+	expires_at.as_str().into(),
122	+	])?
123	+	.first::<ClaimRow>(None)
124	+	.await?;
125	+	let Some(created) = created else {
126	+	return Err(worker::Error::RustError("the claim was not recorded".into()));
127	+	};
128	+	let claim = Claim::from(created);
129	+	self.publish(
130	+	"work.claimed",
131	+	&repo.id,
132	+	&a.actor,
133	+	ClaimEvent { others: Some(others.clone()), took_over, ..ClaimEvent::of(&claim) },
134	+	)
135	+	.await?;
136	+	Ok(Outcome::Ok(Claimed { claim, others, blocked_by }))
137	+	}
138	+
139	+	/// Ends the caller's active claim on an issue as `state`, or refuses
140	+	/// when they hold none.
141	+	async fn end_own_claim(
142	+	&self,
143	+	actor: &User,
144	+	repo: &g1t_contracts::repos::RepoPath,
145	+	number: u32,
146	+	state: ClaimState,
147	+	reason: &str,
148	+	handoff: Option<(&Option<String>, &str)>,
149	+	) -> Result<Outcome<Claim>> {
150	+	let repo = unwrap_or_fail!(self.participating_repo(repo, actor).await?);
151	+	let Some(held) = self.own_claim(&repo.id, number, actor).await? else {
152	+	return Ok(Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}.")));
153	+	};
154	+	let at = rfc3339(now_ms());
155	+	let (handoff_to, note) = match handoff {
156	+	Some((to, note)) => (text_or_null(to), JsValue::from(note)),
157	+	None => (JsValue::NULL, JsValue::NULL),
158	+	};
159	+	let ended = self
160	+	.db
161	+	.prepare(format!(
162	+	"UPDATE claims
163	+	SET state = ?, ended_at = ?, end_reason = ?, handoff_to = ?, note = COALESCE(?, note)
164	+	WHERE id = ? AND state = 'active'
165	+	RETURNING {CLAIM_COLUMNS}"
166	+	))
167	+	.bind(&[state.as_str().into(), at.into(), reason.into(), handoff_to, note, held.id.as_str().into()])?
168	+	.first::<ClaimRow>(None)
169	+	.await?;
170	+	Ok(match ended {
171	+	Some(row) => Outcome::Ok(row.into()),
172	+	None => Outcome::fail(FailureCode::NotFound, format!("You hold no claim on #{number}.")),
173	+	})
174	+	}
175	+
176	+	pub(crate) async fn release_claim(&self, a: ReleaseArgs) -> Result<Outcome<Claim>> {
177	+	let reason = tidy(a.reason.as_deref(), MAX_NOTE_CHARS).unwrap_or_else(\|\| "released".to_owned());
178	+	let claim = unwrap_or_fail!(
179	+	self.end_own_claim(&a.actor, &a.repo, a.number, ClaimState::Released, &reason, None).await?
180	+	);
181	+	self.publish(
182	+	"work.released",
183	+	&claim.repo_id.clone(),
184	+	&a.actor,
185	+	ClaimEvent { reason: Some(reason), ..ClaimEvent::of(&claim) },
186	+	)
187	+	.await?;
188	+	Ok(Outcome::Ok(claim))
189	+	}
190	+
191	+	pub(crate) async fn hand_off_issue(&self, a: HandOffArgs) -> Result<Outcome<Claim>> {
192	+	let Some(note) = tidy(Some(&a.note), MAX_NOTE_CHARS) else {
193	+	return Ok(Outcome::fail(FailureCode::Invalid, "Say where the work stands in note, for whoever picks it up."));
194	+	};
195	+	let to = tidy(a.to.as_deref().map(\|to\| to.trim().trim_start_matches('@')), 100).map(\|to\| to.to_lowercase());
196	+	if let Some(name) = &to {
197	+	let account: Viewer =
198	+	g1t_kit::call(&self.identity, "user_by_username", &UsernameArgs { username: name.clone() }).await?;
199	+	if account.is_none() {
200	+	return Ok(Outcome::fail(FailureCode::Invalid, format!("There is no account named {name}.")));
201	+	}
202	+	}
203	+	let claim = unwrap_or_fail!(
204	+	self.end_own_claim(&a.actor, &a.repo, a.number, ClaimState::HandedOff, "handed_off", Some((&to, &note)))
205	+	.await?
206	+	);
207	+	self.publish(
208	+	"work.handoff_requested",
209	+	&claim.repo_id.clone(),
210	+	&a.actor,
211	+	ClaimEvent { from: Some(a.actor.username.clone()), to, note: Some(note), ..ClaimEvent::of(&claim) },
212	+	)
213	+	.await?;
214	+	Ok(Outcome::Ok(claim))
215	+	}
216	+
217	+	pub(crate) async fn list_claims(&self, a: ListClaimsArgs) -> Result<Outcome<Vec<Claim>>> {
218	+	let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?);
219	+	let state = a.state.unwrap_or(ClaimState::Active);
220	+	let rows = self
221	+	.db
222	+	.prepare(format!(
223	+	"SELECT {CLAIM_COLUMNS} FROM claims
224	+	WHERE repo_id = ?1 AND state = ?2 AND (?3 IS NULL OR issue = ?3)
225	+	ORDER BY created_at DESC, id DESC LIMIT ?4"
226	+	))
227	+	.bind(&[
228	+	repo.id.as_str().into(),
229	+	state.as_str().into(),
230	+	a.issue.filter(\|number\| *number > 0).map_or(JsValue::NULL, JsValue::from),
231	+	LIST_LIMIT.into(),
232	+	])?
233	+	.all()
234	+	.await?
235	+	.results::<ClaimRow>()?;
236	+	Ok(Outcome::Ok(rows.into_iter().map(Claim::from).collect()))
237	+	}
238	+
239	+	/// An issue closed: its active claims end as released (`issue_closed`),
240	+	/// each announced.
241	+	pub(crate) async fn release_claims_on_close(&self, repo_id: &str, issue: u32, actor: Option<&User>) -> Result<()> {
242	+	let at = rfc3339(now_ms());
243	+	let released = self
244	+	.db
245	+	.prepare(format!(
246	+	"UPDATE claims SET state = 'released', ended_at = ?, end_reason = 'issue_closed'
247	+	WHERE repo_id = ? AND issue = ? AND state = 'active'
248	+	RETURNING {CLAIM_COLUMNS}"
249	+	))
250	+	.bind(&[at.into(), repo_id.into(), issue.into()])?
251	+	.all()
252	+	.await?
253	+	.results::<ClaimRow>()?;
254	+	for row in released {
255	+	let claim = Claim::from(row);
256	+	let data = ClaimEvent { reason: Some("issue_closed".to_owned()), ..ClaimEvent::of(&claim) };
257	+	match actor {
258	+	Some(actor) => self.publish("work.released", repo_id, actor, data).await?,
259	+	None => self.publish_as("work.released", repo_id, None, data).await?,
260	+	}
261	+	}
262	+	Ok(())
263	+	}
264	+	}

1	+	//! `get_work`: everything about an issue an agent needs to pick it up, in
2	+	//! one call. What it is for and what done means, who is on it, the pull
3	+	//! requests already made for it and where each stands, what it waits on
4	+	//! and what waits on it, the decisions made about it, and its latest
5	+	//! coordination events.
6	+
7	+	use futures_util::future::{try_join, try_join_all, try_join3};
8	+	use g1t_contracts::events::{Event, ListArgs as ListEventsArgs};
9	+	use g1t_contracts::repos::Repo;
10	+	use g1t_contracts::work::*;
11	+	use g1t_contracts::{FailureCode, Outcome};
12	+	use g1t_kit::now_ms;
13	+	use serde::Deserialize;
14	+	use worker::Result;
15	+
16	+	use super::{SESSION_COLUMNS, SessionRow, unwrap_or_fail};
17	+	use crate::Work;
18	+	use crate::rows::{PULL_COLUMNS, PullRow};
19	+
20	+	/// How many sessions, decisions and events `get_work` gives at most.
21	+	const SESSIONS: u32 = 20;
22	+	const DECISIONS: u32 = 20;
23	+	const EVENTS: u32 = 20;
24	+
25	+	#[derive(Deserialize)]
26	+	struct PlaceRow {
27	+	state: QueueState,
28	+	position: u32,
29	+	}
30	+
31	+	impl Work {
32	+	pub(crate) async fn get_work(&self, a: GetWorkArgs) -> Result<Outcome<WorkContext>> {
33	+	let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?);
34	+	let Some(issue) = self.issue(&repo.id, a.number).await? else {
35	+	return Ok(Outcome::fail(FailureCode::NotFound, "Issue not found."));
36	+	};
37	+	let pulls: Vec<Pull> = self
38	+	.db
39	+	.prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE issue_id = ? ORDER BY number"))
40	+	.bind(&[issue.id.as_str().into()])?
41	+	.all()
42	+	.await?
43	+	.results::<PullRow>()?
44	+	.into_iter()
45	+	.map(Pull::from)
46	+	.collect();
47	+	let pull_numbers: Vec<u32> = pulls.iter().map(\|pull\| pull.number).collect();
48	+	let mut numbers = vec![issue.number];
49	+	numbers.extend(&pull_numbers);
50	+
51	+	let people = try_join3(
52	+	self.active_claims(&repo.id, issue.number),
53	+	self.work_sessions(&repo.id, issue.number, &pull_numbers),
54	+	self.default_branch_settings(&repo),
55	+	);
56	+	let dependencies = try_join(self.blockers(&repo.id, &issue.blocked_by), self.blocked_issues(&repo.id, issue.number));
57	+	let history = try_join3(
58	+	self.coordination_events(&repo.id, &numbers, COORDINATION_EVENTS.to_vec(), EVENTS),
59	+	self.coordination_events(&repo.id, &numbers, vec!["integration.failed"], 1),
60	+	self.decisions_on(&repo.id, issue.number, &pull_numbers),
61	+	);
62	+	let work_pulls = try_join_all(pulls.iter().map(\|pull\| self.work_pull(&repo, pull)));
63	+	let (((claims, sessions, settings), (blocked_by, blocks)), ((recent, failures, decisions), pulls)) =
64	+	try_join(try_join(people, dependencies), try_join(history, work_pulls)).await?;
65	+	Ok(Outcome::Ok(WorkContext {
66	+	definition_of_done: definition_of_done(&issue.body),
67	+	required_checks: settings.required_checks,
68	+	issue,
69	+	claims,
70	+	sessions,
71	+	pulls,
72	+	blocked_by,
73	+	blocks,
74	+	decisions,
75	+	last_integration_failure: failures.into_iter().next(),
76	+	recent,
77	+	}))
78	+	}
79	+
80	+	/// Sessions on an issue or its pull requests, most recently heard from
81	+	/// first.
82	+	async fn work_sessions(&self, repo_id: &str, issue: u32, pulls: &[u32]) -> Result<Vec<Session>> {
83	+	let now = now_ms();
84	+	Ok(self
85	+	.db
86	+	.prepare(format!(
87	+	"SELECT {SESSION_COLUMNS} FROM sessions
88	+	WHERE repo_id = ? AND (issue = ? OR pull IN (SELECT value FROM json_each(?)))
89	+	ORDER BY last_seen_at DESC LIMIT ?"
90	+	))
91	+	.bind(&[repo_id.into(), issue.into(), serde_json::to_string(pulls)?.into(), SESSIONS.into()])?
92	+	.all()
93	+	.await?
94	+	.results::<SessionRow>()?
95	+	.into_iter()
96	+	.map(\|row\| row.at(now))
97	+	.collect())
98	+	}
99	+
100	+	/// The latest events of `types` about `numbers`, newest first.
101	+	async fn coordination_events(&self, repo_id: &str, numbers: &[u32], types: Vec<&str>, limit: u32) -> Result<Vec<Event>> {
102	+	g1t_kit::call(
103	+	&self.events,
104	+	"list",
105	+	&ListEventsArgs {
106	+	repo_id: Some(repo_id.to_owned()),
107	+	types: types.into_iter().map(str::to_owned).collect(),
108	+	numbers: numbers.to_vec(),
109	+	limit: Some(limit),
110	+	..ListEventsArgs::default()
111	+	},
112	+	)
113	+	.await
114	+	}
115	+
116	+	/// Decision records on an issue and its pull requests, newest first,
117	+	/// as stored: none until the repository has any. Each `evidence` is
118	+	/// given as JSON.
119	+	async fn decisions_on(&self, repo_id: &str, issue: u32, pulls: &[u32]) -> Result<Vec<serde_json::Value>> {
120	+	let statement = self
121	+	.db
122	+	.prepare(
123	+	"SELECT * FROM decisions
124	+	WHERE repo_id = ? AND (issue = ? OR pull IN (SELECT value FROM json_each(?)))
125	+	ORDER BY created_at DESC LIMIT ?",
126	+	)
127	+	.bind(&[repo_id.into(), issue.into(), serde_json::to_string(pulls)?.into(), DECISIONS.into()])?;
128	+	// Decision records are kept by the merge queue's side of
129	+	// coordination; until they are, there are none to give.
130	+	let rows: Vec<serde_json::Value> = match statement.all().await {
131	+	Ok(result) => result.results()?,
132	+	Err(_) => return Ok(Vec::new()),
133	+	};
134	+	Ok(rows
135	+	.into_iter()
136	+	.map(\|mut row\| {
137	+	if let Some(evidence) = row.get("evidence").and_then(serde_json::Value::as_str)
138	+	&& let Ok(parsed) = serde_json::from_str::<serde_json::Value>(evidence)
139	+	{
140	+	row["evidence"] = parsed;
141	+	}
142	+	row
143	+	})
144	+	.collect())
145	+	}
146	+
147	+	/// One of the issue's pull requests and where it stands: its required
148	+	/// checks, whether it merges, what it overlaps and its place in the
149	+	/// merge queue, for those still in progress.
150	+	async fn work_pull(&self, repo: &Repo, pull: &Pull) -> Result<WorkPull> {
151	+	let mut pull = pull.clone();
152	+	crate::fill_base(&mut pull, repo);
153	+	let brief = \|pull: &Pull\| WorkPull {
154	+	number: pull.number,
155	+	title: pull.title.clone(),
156	+	status: pull.status,
157	+	branch: pull.branch.clone(),
158	+	head_commit: pull.head_commit.clone(),
159	+	author: pull.owner().into(),
160	+	check_status: pull.check_status,
161	+	required_checks: Vec::new(),
162	+	mergeable: Mergeable::Unknown,
163	+	overlaps: Vec::new(),
164	+	queue: None,
165	+	};
166	+	if !pull.status.is_active() {
167	+	return Ok(brief(&pull));
168	+	}
169	+	let required = async {
170	+	let settings = self.settings_on(repo, &pull).await?;
171	+	if settings.required_checks.is_empty() {
172	+	return Ok(Vec::new());
173	+	}
174	+	let statuses = self.statuses(&repo.id, pull.head_commit.as_deref()).await?;
175	+	let checked = self.with_job_checks(&repo.id, pull.head_commit.as_deref(), statuses).await?;
176	+	Ok::<_, worker::Error>(required_checks(&settings.required_checks, &checked))
177	+	};
178	+	let place_query = self
179	+	.db
180	+	.prepare(
181	+	"SELECT mine.state AS state,
182	+	(SELECT count(*) FROM queue_entries ahead
183	+	WHERE ahead.repo_id = mine.repo_id AND ahead.state IN ('waiting', 'testing', 'passed')
184	+	AND (ahead.created_at < mine.created_at
185	+	OR (ahead.created_at = mine.created_at AND ahead.id < mine.id))) + 1 AS position
186	+	FROM queue_entries mine
187	+	WHERE mine.pull_id = ? AND mine.state IN ('waiting', 'testing', 'passed') LIMIT 1",
188	+	)
189	+	.bind(&[pull.id.as_str().into()])?;
190	+	let place = place_query.first::<PlaceRow>(None);
191	+	let ((required_checks, (mergeable, _)), (overlaps, place)) =
192	+	try_join(try_join(required, self.mergeability(&pull)), try_join(self.overlaps(&pull), place)).await?;
193	+	Ok(WorkPull {
194	+	required_checks,
195	+	mergeable,
196	+	overlaps,
197	+	queue: place.map(\|place\| QueuePlace { state: place.state, position: place.position }),
198	+	..brief(&pull)
199	+	})
200	+	}
201	+	}

1	+	//! What an issue is blocked by: the issues of the same repository that have
2	+	//! to land first (`issues.blocked_by`, a JSON array of numbers).
3	+	//!
4	+	//! `update_issue` replaces the set, refusing an issue that is not one of
5	+	//! the repository's, the issue itself, or a set that would make a cycle.
6	+	//! Each change, and each blocker that closes or has its pull request
7	+	//! merged, is a `dependency.changed` event on the issue it blocks.
8	+
9	+	use std::collections::{HashMap, HashSet};
10	+
11	+	use g1t_contracts::work::*;
12	+	use g1t_contracts::{FailureCode, Outcome, User};
13	+	use serde::Deserialize;
14	+	use worker::Result;
15	+	use worker::wasm_bindgen::JsValue;
16	+
17	+	use super::{CLAIM_COLUMNS, ClaimRow};
18	+	use crate::Work;
19	+
20	+	#[derive(Deserialize)]
21	+	struct EdgeRow {
22	+	number: u32,
23	+	blocked_by: String,
24	+	}
25	+
26	+	#[derive(Deserialize)]
27	+	struct BriefRow {
28	+	number: u32,
29	+	title: String,
30	+	state: State,
31	+	}
32	+
33	+	#[derive(Deserialize)]
34	+	struct PullBriefRow {
35	+	number: u32,
36	+	title: String,
37	+	status: PullStatus,
38	+	issue_number: u32,
39	+	}
40	+
41	+	fn numbers(json: &str) -> Vec<u32> {
42	+	serde_json::from_str(json).unwrap_or_default()
43	+	}
44	+
45	+	impl Work {
46	+	/// `asked` as `number`'s new `blocked_by`, or why it cannot be.
47	+	pub(crate) async fn checked_blocked_by(&self, repo_id: &str, number: u32, asked: &[u32]) -> Result<Outcome<Vec<u32>>> {
48	+	let wanted = serde_json::to_string(asked)?;
49	+	let results = self
50	+	.db
51	+	.batch(vec![
52	+	self.db
53	+	.prepare(
54	+	"SELECT number, title, state FROM issues
55	+	WHERE repo_id = ? AND number IN (SELECT value FROM json_each(?))",
56	+	)
57	+	.bind(&[repo_id.into(), wanted.into()])?,
58	+	self.db
59	+	.prepare("SELECT number, blocked_by FROM issues WHERE repo_id = ? AND blocked_by <> '[]'")
60	+	.bind(&[repo_id.into()])?,
61	+	])
62	+	.await?;
63	+	let mut results = results.into_iter();
64	+	let found: Vec<BriefRow> = match results.next() {
65	+	Some(result) => result.results()?,
66	+	None => Vec::new(),
67	+	};
68	+	let edges: Vec<EdgeRow> = match results.next() {
69	+	Some(result) => result.results()?,
70	+	None => Vec::new(),
71	+	};
72	+	let issues: HashSet<u32> = found.iter().map(\|row\| row.number).collect();
73	+	let graph: HashMap<u32, Vec<u32>> = edges.into_iter().map(\|row\| (row.number, numbers(&row.blocked_by))).collect();
74	+	Ok(match check_blocked_by(number, asked, &issues, &graph) {
75	+	Ok(set) => Outcome::Ok(set),
76	+	Err(message) => Outcome::fail(FailureCode::Invalid, message),
77	+	})
78	+	}
79	+
80	+	/// Announces a change to `issue`'s `blocked_by` (from `before`), and
81	+	/// notes it in the issue's conversation.
82	+	pub(crate) async fn blocked_by_changed(&self, issue: &Issue, before: &[u32], actor: &User) -> Result<()> {
83	+	let added: Vec<u32> = issue.blocked_by.iter().filter(\|n\| !before.contains(n)).copied().collect();
84	+	let removed: Vec<u32> = before.iter().filter(\|n\| !issue.blocked_by.contains(n)).copied().collect();
85	+	if added.is_empty() && removed.is_empty() {
86	+	return Ok(());
87	+	}
88	+	let remaining = self.still_open(&issue.repo_id, &issue.blocked_by, None).await?;
89	+	self.publish(
90	+	"dependency.changed",
91	+	&issue.repo_id,
92	+	actor,
93	+	DependencyChanged {
94	+	repo_id: issue.repo_id.clone(),
95	+	number: issue.number,
96	+	issue: issue.number,
97	+	blocked_by: issue.blocked_by.clone(),
98	+	added: added.clone(),
99	+	removed: removed.clone(),
100	+	cleared: Vec::new(),
101	+	remaining,
102	+	merged_pull: None,
103	+	},
104	+	)
105	+	.await?;
106	+	let list = \|numbers: &[u32]\| numbers.iter().map(\|n\| format!("#{n}")).collect::<Vec<_>>().join(", ");
107	+	let who = (actor.id.as_str(), actor.username.as_str());
108	+	if !added.is_empty() {
109	+	self.note(&issue.repo_id, issue.number, who, &format!("marked this as blocked by {}", list(&added))).await?;
110	+	}
111	+	if !removed.is_empty() {
112	+	self.note(&issue.repo_id, issue.number, who, &format!("marked this as no longer blocked by {}", list(&removed)))
113	+	.await?;
114	+	}
115	+	Ok(())
116	+	}
117	+
118	+	/// Those of `numbers` whose issue is still open, leaving out `cleared`
119	+	/// (a blocker whose pull request just merged).
120	+	async fn still_open(&self, repo_id: &str, numbers: &[u32], cleared: Option<u32>) -> Result<Vec<u32>> {
121	+	if numbers.is_empty() {
122	+	return Ok(Vec::new());
123	+	}
124	+	let open: Vec<BriefRow> = self
125	+	.db
126	+	.prepare(
127	+	"SELECT number, title, state FROM issues
128	+	WHERE repo_id = ? AND state = 'open' AND number IN (SELECT value FROM json_each(?))",
129	+	)
130	+	.bind(&[repo_id.into(), serde_json::to_string(numbers)?.into()])?
131	+	.all()
132	+	.await?
133	+	.results()?;
134	+	Ok(numbers
135	+	.iter()
136	+	.filter(\|number\| Some(number) != cleared && open.iter().any(\|row\| row.number == number))
137	+	.copied()
138	+	.collect())
139	+	}
140	+
141	+	/// An issue closed (`closed`) or had a pull request merged for it
142	+	/// (`merged_pull`): its claims end if it closed, and each open issue it
143	+	/// blocks hears that it cleared. Never fails what caused it: a problem
144	+	/// is logged.
145	+	pub(crate) async fn issue_resolved(&self, repo_id: &str, number: u32, actor: &User, closed: bool, merged_pull: Option<u32>) {
146	+	let resolved = async {
147	+	if closed {
148	+	self.release_claims_on_close(repo_id, number, Some(actor)).await?;
149	+	}
150	+	let dependents: Vec<EdgeRow> = self
151	+	.db
152	+	.prepare(
153	+	"SELECT number, blocked_by FROM issues
154	+	WHERE repo_id = ? AND state = 'open'
155	+	AND EXISTS (SELECT 1 FROM json_each(issues.blocked_by) WHERE json_each.value = ?)",
156	+	)
157	+	.bind(&[repo_id.into(), number.into()])?
158	+	.all()
159	+	.await?
160	+	.results()?;
161	+	for dependent in dependents {
162	+	let blocked_by = numbers(&dependent.blocked_by);
163	+	let remaining = self.still_open(repo_id, &blocked_by, Some(number)).await?;
164	+	self.publish(
165	+	"dependency.changed",
166	+	repo_id,
167	+	actor,
168	+	DependencyChanged {
169	+	repo_id: repo_id.to_owned(),
170	+	number: dependent.number,
171	+	issue: dependent.number,
172	+	blocked_by,
173	+	cleared: vec![number],
174	+	remaining,
175	+	merged_pull,
176	+	..DependencyChanged::default()
177	+	},
178	+	)
179	+	.await?;
180	+	}
181	+	Ok::<(), worker::Error>(())
182	+	};
183	+	if let Err(error) = resolved.await {
184	+	worker::console_error!("#{number} resolved, but its dependents were not told: {error}");
185	+	}
186	+	}
187	+
188	+	/// The issues `numbers` names, each with its active claims and its pull
189	+	/// requests, in the order given.
190	+	pub(crate) async fn blockers(&self, repo_id: &str, numbers: &[u32]) -> Result<Vec<Blocker>> {
191	+	if numbers.is_empty() {
192	+	return Ok(Vec::new());
193	+	}
194	+	let list = serde_json::to_string(numbers)?;
195	+	let bind = \|sql: String\| -> Result<worker::D1PreparedStatement> {
196	+	self.db.prepare(sql).bind(&[repo_id.into(), JsValue::from(list.as_str())])
197	+	};
198	+	let results = self
199	+	.db
200	+	.batch(vec![
201	+	bind("SELECT number, title, state FROM issues
202	+	WHERE repo_id = ? AND number IN (SELECT value FROM json_each(?))"
203	+	.to_owned())?,
204	+	bind(format!(
205	+	"SELECT {CLAIM_COLUMNS} FROM claims
206	+	WHERE repo_id = ? AND state = 'active' AND issue IN (SELECT value FROM json_each(?))
207	+	ORDER BY created_at, id"
208	+	))?,
209	+	bind("SELECT number, title, status, issue_number FROM pulls
210	+	WHERE repo_id = ? AND issue_number IN (SELECT value FROM json_each(?))
211	+	ORDER BY number"
212	+	.to_owned())?,
213	+	])
214	+	.await?;
215	+	let mut results = results.into_iter();
216	+	let mut next = \|\| results.next();
217	+	let issues: Vec<BriefRow> = match next() {
218	+	Some(result) => result.results()?,
219	+	None => Vec::new(),
220	+	};
221	+	let claims: Vec<Claim> = match next() {
222	+	Some(result) => result.results::<ClaimRow>()?.into_iter().map(Claim::from).collect(),
223	+	None => Vec::new(),
224	+	};
225	+	let pulls: Vec<PullBriefRow> = match next() {
226	+	Some(result) => result.results()?,
227	+	None => Vec::new(),
228	+	};
229	+	Ok(numbers
230	+	.iter()
231	+	.filter_map(\|number\| {
232	+	let issue = issues.iter().find(\|issue\| issue.number == *number)?;
233	+	Some(Blocker {
234	+	number: issue.number,
235	+	title: issue.title.clone(),
236	+	state: issue.state,
237	+	claims: claims.iter().filter(\|claim\| claim.issue == *number).cloned().collect(),
238	+	pulls: pulls
239	+	.iter()
240	+	.filter(\|pull\| pull.issue_number == *number)
241	+	.map(\|pull\| PullBrief { number: pull.number, title: pull.title.clone(), status: pull.status })
242	+	.collect(),
243	+	})
244	+	})
245	+	.collect())
246	+	}
247	+
248	+	/// The issues that name `number` in their `blocked_by`, by number.
249	+	pub(crate) async fn blocked_issues(&self, repo_id: &str, number: u32) -> Result<Vec<IssueBrief>> {
250	+	Ok(self
251	+	.db
252	+	.prepare(
253	+	"SELECT number, title, state FROM issues
254	+	WHERE repo_id = ? AND EXISTS (SELECT 1 FROM json_each(issues.blocked_by) WHERE json_each.value = ?)
255	+	ORDER BY number LIMIT 100",
256	+	)
257	+	.bind(&[repo_id.into(), number.into()])?
258	+	.all()
259	+	.await?
260	+	.results::<BriefRow>()?
261	+	.into_iter()
262	+	.map(\|row\| IssueBrief { number: row.number, title: row.title, state: row.state })
263	+	.collect())
264	+	}
265	+	}

1646	1646		"g1t-rules",
1647	1647		"getrandom 0.2.17",
1648	1648		"hex",
	1649	+	"rusqlite",
1649	1650		"serde",
1650	1651		"serde_json",
1651	1652		"sha2 0.10.9",

11	11		use crate::repos::{CompareArgs, RepoPath};
12	12		use crate::{User, Viewer};
13	13
	14	+	// Sessions, claims and dependencies: work/coordination.rs.
	15	+	mod coordination;
	16	+	pub use coordination::*;
	17	+
14	18		/// The labels a new repository starts with, and that "Add the default
15	19		/// labels" adds to one that is missing some: `(name, color, description)`.
16	20		/// Colors are six hex digits, without `#`.
⋯
1454	1458		/// milestone. Needs the Triage role.
1455	1459		#[serde(default)]
1456	1460		pub milestone: Option<u32>,
	1461	+	/// The numbers of the issues of the same repository that have to land
	1462	+	/// first; replaces the whole set. Refused when one is not an issue, is
	1463	+	/// the issue itself, or would make a cycle.
	1464	+	#[serde(default)]
	1465	+	pub blocked_by: Option<Vec<u32>>,
1457	1466		}
1458	1467
1459	1468		/// `close_issue` and `reopen_issue`. Each returns `Outcome<Issue>`.

1103	1103		("edit_comment", Scope::IssuesWrite),
1104	1104		("delete_comment", Scope::IssuesWrite),
1105	1105		("import_issue", Scope::IssuesWrite),
	1106	+	// Coordinating work on issues: claims, and the context to pick one up.
	1107	+	("list_claims", Scope::IssuesRead),
	1108	+	("get_work", Scope::IssuesRead),
	1109	+	("claim_issue", Scope::IssuesWrite),
	1110	+	("release_claim", Scope::IssuesWrite),
	1111	+	("hand_off_issue", Scope::IssuesWrite),
1106	1112		// Pull requests.
1107	1113		("list_pull_requests", Scope::PullRequestsRead),
1108	1114		("get_pull_request", Scope::PullRequestsRead),
⋯
1120	1126		("merge_pull_request", Scope::PullRequestsWrite),
1121	1127		("request_reviewers", Scope::PullRequestsWrite),
1122	1128		("remove_requested_reviewers", Scope::PullRequestsWrite),
	1129	+	// Harness sessions, which record how pull requests are made.
	1130	+	("get_session", Scope::PullRequestsRead),
	1131	+	("list_sessions", Scope::PullRequestsRead),
	1132	+	("start_session", Scope::PullRequestsWrite),
	1133	+	("update_session", Scope::PullRequestsWrite),
	1134	+	("end_session", Scope::PullRequestsWrite),
	1135	+	("record_session_entries", Scope::PullRequestsWrite),
1123	1136		// Workflows.
1124	1137		("list_workflows", Scope::WorkflowsRead),
1125	1138		("list_workflow_runs", Scope::WorkflowsRead),

15	15		use crate::{User, Viewer};
16	16
17	17		/// Every event a webhook can be sent, in the order people are shown them.
18		−	pub const EVENT_TYPES: [&str; 107] = [
	18	+	pub const EVENT_TYPES: [&str; 115] = [
19	19		"git.push",
20	20		"git.deleted",
21	21		"branch.renamed",
⋯
99	99		"deployment_status.created",
100	100		"queue.changed",
101	101		"session.appended",
	102	+	"session.started",
	103	+	"session.ended",
	104	+	"work.claimed",
	105	+	"work.released",
	106	+	"work.claim_expired",
	107	+	"work.handoff_requested",
	108	+	"dependency.changed",
	109	+	"overlap.detected",
102	110		"package.published",
103	111		"package.version_deleted",
104	112		"package.deleted",

12	12		import type { CheckRunEventData, CheckSuiteEventData, StatusEventData } from "./checks";
13	13		import type { DeploymentStatus, RepoDeployment } from "./deployments";
14	14		import type { TeamRole, TeamVisibility } from "./teams";
15		−	import type { AgentRef, Verdict } from "./work";
	15	+	import type { AgentRef, ClaimEventData, DependencyChangedData, OverlapDetectedData, SessionEventData, Verdict } from "./work";
16	16
17	17		/** What every `package.` event names. /
18	18		export type PackageEventData = {
⋯
392	392		};
393	393		};
394	394		"session.appended": { pullId: string; repoId: string; number: number; count: number };
	395	+	/** A harness session started on the repository (not on an idempotent re-start). */
	396	+	"session.started": SessionEventData;
	397	+	/** A session ended, or timed out after 2 hours without a heartbeat. */
	398	+	"session.ended": SessionEventData;
	399	+	/** Someone claimed an issue; `others` are the other active claims on it. */
	400	+	"work.claimed": ClaimEventData;
	401	+	"work.released": ClaimEventData;
	402	+	"work.claim_expired": ClaimEventData;
	403	+	"work.handoff_requested": ClaimEventData;
	404	+	/** An issue's blockers changed, or one of them closed or had its pull request merged. */
	405	+	"dependency.changed": DependencyChangedData;
	406	+	/** A push made a pull request change files another open one changes. */
	407	+	"overlap.detected": OverlapDetectedData;
395	408		/**
396	409		* A build of a project finished, for production or one pull request's
397	410		* preview (`number`). `path` is the deployment's page on the site;

472	472		["edit_comment", "issues:write"],
473	473		["delete_comment", "issues:write"],
474	474		["import_issue", "issues:write"],
	475	+	["list_claims", "issues:read"],
	476	+	["get_work", "issues:read"],
	477	+	["claim_issue", "issues:write"],
	478	+	["release_claim", "issues:write"],
	479	+	["hand_off_issue", "issues:write"],
475	480		["list_pull_requests", "pull_requests:read"],
476	481		["get_pull_request", "pull_requests:read"],
477	482		["get_pull_request_changes", "pull_requests:read"],
⋯
488	493		["merge_pull_request", "pull_requests:write"],
489	494		["request_reviewers", "pull_requests:write"],
490	495		["remove_requested_reviewers", "pull_requests:write"],
	496	+	["get_session", "pull_requests:read"],
	497	+	["list_sessions", "pull_requests:read"],
	498	+	["start_session", "pull_requests:write"],
	499	+	["update_session", "pull_requests:write"],
	500	+	["end_session", "pull_requests:write"],
	501	+	["record_session_entries", "pull_requests:write"],
491	502		["list_workflows", "workflows:read"],
492	503		["list_workflow_runs", "workflows:read"],
493	504		["get_workflow_run", "workflows:read"],

21	21		hex = "0.4"
22	22		sha2 = "0.10"
23	23
	24	+	[dev-dependencies]
	25	+	rusqlite = { version = "0.32", features = ["bundled"] }
	26	+
24	27		# wasm-opt at -O1: about the same gzipped size as -O in a tenth of the
25	28		# time.
26	29		[package.metadata.wasm-pack.profile.release]

flagon-io/g1t

Comments