services/work/src/coordination/workspace.rs · flagon-io/g1t

.prepare("SELECT repo_id, path FROM repo_workspaces WHERE workspace = ?1 AND (?2 IS NULL OR lower(path) = lower(?1 || '/' || ?2)) ORDER BY path LIMIT ?3")

245

.bind(&[workspace.into(), only.map_or(JsValue::NULL, JsValue::from), MAX_REPOS.into()])?

246

.all()

247

.await?

248

.results::<Placed>()?;

249

if placed.is_empty() {

250

return Ok(Vec::new());

251

}

252

let ids: Vec<String> = placed.iter().map(|row| row.repo_id.clone()).collect();

253

let mut readable: Vec<Repo> = g1t_kit::call(&self.repos, "readable", &ReadableArgs { ids, viewer: viewer.clone() }).await?;

254

readable.retain(|repo| repo.namespace.eq_ignore_ascii_case(workspace));

255

readable.sort_by(|a, b| a.name.cmp(&b.name));

Ok(readable)

}

pub(crate) async fn list_workspace_sessions(&self, a: ListWorkspaceSessionsArgs) -> Result<Outcome<WorkspaceSessions>> {

260

let workspace = a.workspace.trim().to_lowercase();

261

if workspace.is_empty() {

262

return Ok(Outcome::fail(FailureCode::Invalid, "Name the workspace by its slug."));

263

}

264

let only = a.repo.as_deref().map(str::trim).filter(|name| !name.is_empty()).map(|name| name.rsplit('/').next().unwrap_or(name).to_owned());

265

let repos = self.readable_in(&workspace, &a.viewer, only.as_deref()).await?;

266

if repos.is_empty() && only.is_some() {

267

return Ok(Outcome::fail(FailureCode::NotFound, "Repository not found."));

268

}

269

let ids: Vec<String> = repos.iter().map(|repo| repo.id.clone()).collect();

270

let path_of: HashMap<&str, String> = repos.iter().map(|repo| (repo.id.as_str(), format!("{}/{}", repo.namespace, repo.name))).collect();

271

let member = a.viewer.as_ref().is_some_and(|viewer| viewer.is_member(&workspace));

272

let limit = a.limit.unwrap_or(DEFAULT_WORKSPACE_SESSIONS).clamp(1, MAX_WORKSPACE_SESSIONS);

273

let now = now_ms();

274

let person = a.person.as_deref().map(|person| person.trim().trim_start_matches('@')).filter(|person| !person.is_empty());

275

// A work item by number, in this workspace.

276

let item_id = match a.item {

277

Some(number) => match crate::items::by_number(self, &workspace, number).await? {

278

Some(held) => Some(held.item.id),

279

None => return Ok(Outcome::Ok(WorkspaceSessions { repos: sessions_repos(&repos), ..WorkspaceSessions::default() })),

},

None => None,

};

// Harness and hosted sessions.

285

let from_table = !matches!(a.kind, Some(SessionKind::Agent | SessionKind::Recorded));

286

let mut sessions: Vec<Session> = if from_table && !ids.is_empty() {

287

self.db

288

.prepare(workspace_sessions_sql(a.shown))

289

.bind(&[

290

serde_json::to_string(&ids)?.into(),

291

rfc3339(now.saturating_sub(SESSION_TIMEOUT_MS)).into(),

292

u32::from(false).into(),

293

a.harness.map_or(JsValue::NULL, |harness| harness.as_str().into()),

294

person.map_or(JsValue::NULL, JsValue::from),

295

item_id.as_deref().map_or(JsValue::NULL, JsValue::from),

296

limit.into(),

297

a.kind.map_or(JsValue::NULL, |kind| kind.as_str().into()),

])?

.all()

.await?

.results::<SessionRow>()?

302

.into_iter()

303

.map(|row| row.at(now))

.collect()

} else {

Vec::new()

};

self.fill_subagents(&mut sessions).await?;

309

self.fill_first_prompts(&mut sessions).await?;

310

shown::sessions(&mut sessions, &a.viewer, &workspace);

311

let mut trees: HashMap<String, Vec<Session>> = HashMap::new();

312

if a.subagents {

313

for session in sessions.iter().filter(|session| session.subagents > 0).take(TREES) {

314

let mut children = self.subagents_of(&session.id, now).await?;

315

count_subagents(&mut children);

316

shown::sessions(&mut children, &a.viewer, &workspace);

317

trees.insert(session.id.clone(), children);

318

}

319

}

320

let table: Vec<WorkspaceSession> = sessions

321

.into_iter()

322

.filter_map(|session| {

323

let path = path_of.get(session.repo_id.as_str())?.clone();

324

let mut listed = from_session(session, &path, now);

325

listed.children = trees.remove(&listed.id).unwrap_or_default();

Some(listed)

})

.collect();

// g1t's agent: the workspace's business, as its fleet is.

331

let wants = |kind: SessionKind| a.kind.is_none_or(|wanted| wanted == kind) && a.harness.is_none();

332

let runs: Vec<WorkspaceSession> = if member && wants(SessionKind::Agent) {

333

self.agent_runs_in(&workspace, &ids, a.shown == SessionsShown::Live, member, limit)

334

.await?

335

.into_iter()

336

.map(|(repo_id, run)| from_run(run, &repo_id))

.collect()

} else {

Vec::new()

};

let recorded: Vec<(WorkspaceSession, bool)> = if wants(SessionKind::Recorded) {

342

self.recorded_in(&ids, limit)

.await?

.into_iter()

.filter_map(|row| {

let path = path_of.get(row.repo_id.as_str())?.clone();

347

let has_runs = row.has_runs != 0;

348

Some((from_recorded(row, &path, now), has_runs))

})

.collect()

} else {

Vec::new()

};

let mut runs = runs;

self.fill_pulls(&mut runs).await?;

356

let wanted = Wanted { shown: a.shown, kind: a.kind, harness: a.harness, person, item: a.item };

357

let mut listed = merged(table, runs, recorded, &wanted, limit as usize);

358

self.fill_pulls(&mut listed).await?;

359

if member {

360

self.fill_item_titles(&workspace, &mut listed).await?;

361

}

362

363

let (queue, landed) = if a.sessions_only { (Vec::new(), Vec::new()) } else { (self.queues_in(&repos, &a.viewer).await?, self.landed_in(&repos).await?) };

364

Ok(Outcome::Ok(WorkspaceSessions { sessions: listed, queue, landed, repos: sessions_repos(&repos) }))

365

}

366

367

/// Each session's pull request's title and status, and, for one with

368

/// no work item of its own, the pull request's.

369

async fn fill_pulls(&self, sessions: &mut [WorkspaceSession]) -> Result<()> {

370

let pairs: Vec<serde_json::Value> = sessions

371

.iter()

372

.filter(|session| session.pull_title.is_none() || session.item.is_none())

373

.filter_map(|session| session.pull.map(|number| serde_json::json!({ "r": session.repo_id, "n": number })))

374

.collect();

375

if pairs.is_empty() {

376

return Ok(());

377

}

378

let facts = self.db.prepare(PULL_FACTS).bind(&[serde_json::to_string(&pairs)?.into()])?.all().await?.results::<PullFact>()?;

379

for session in sessions.iter_mut() {

380

let Some(fact) = session.pull.and_then(|number| facts.iter().find(|fact| fact.repo_id == session.repo_id && fact.number == number)) else {

381

continue;

382

};

383

session.pull_title.get_or_insert_with(|| fact.title.clone());

384

session.pull_status.get_or_insert(fact.status);

385

if session.item.is_none() {

386

session.item = fact.item;

387

}

388

if session.title.is_none() && session.kind == SessionKind::Agent {

389

session.title = Some(fact.title.clone());

}

}

Ok(())

}

/// Each session's work item's title, for a member of the workspace.

396

async fn fill_item_titles(&self, workspace: &str, sessions: &mut [WorkspaceSession]) -> Result<()> {

397

let mut numbers: Vec<u32> = sessions.iter().filter_map(|session| session.item).collect();

398

numbers.sort_unstable();

399

numbers.dedup();

400

if numbers.is_empty() {

401

return Ok(());

402

}

403

let titles: HashMap<u32, String> = self

404

.db

405

.prepare("SELECT number, title FROM work_items WHERE workspace = ?1 AND number IN (SELECT value FROM json_each(?2))")

406

.bind(&[workspace.into(), serde_json::to_string(&numbers)?.into()])?

407

.all()

408

.await?

409

.results::<Titled>()?

410

.into_iter()

411

.map(|row| (row.number, row.title))

412

.collect();

413

for session in sessions {

414

session.item_title = session.item.and_then(|number| titles.get(&number).cloned());

}

Ok(())

}

/// The merge queues of `repos` with something waiting, busiest first.

420

async fn queues_in(&self, repos: &[Repo], viewer: &Viewer) -> Result<Vec<RepoQueue>> {

421

if repos.is_empty() {

422

return Ok(Vec::new());

423

}

424

let ids: Vec<&str> = repos.iter().map(|repo| repo.id.as_str()).collect();

425

let busy = self

426

.db

427

.prepare(QUEUED_REPOS)

428

.bind(&[serde_json::to_string(&ids)?.into(), (QUEUES as u32).into()])?

429

.all()

430

.await?

431

.results::<Value>()?;

432

let mut queues = Vec::new();

433

for id in busy {

434

let Some(repo) = repos.iter().find(|repo| repo.id == id.value) else { continue };

435

let path = RepoPath { namespace: repo.namespace.clone(), name: repo.name.clone() };

436

if let Outcome::Ok(queue) = self.queue(QueueArgs { repo: path, viewer: viewer.clone() }).await? {

437

queues.push(RepoQueue { repo: format!("{}/{}", repo.namespace, repo.name), default_branch: repo.default_branch.clone(), queue });

}

}

Ok(queues)

}

/// The pull requests of `repos` merged most recently.

444

async fn landed_in(&self, repos: &[Repo]) -> Result<Vec<RepoPull>> {

445

if repos.is_empty() {

446

return Ok(Vec::new());

447

}

448

let ids: Vec<&str> = repos.iter().map(|repo| repo.id.as_str()).collect();

let pulls = self

.db

.prepare(format!(

"SELECT {PULL_COLUMNS} FROM pulls WHERE repo_id IN (SELECT value FROM json_each(?1)) AND status = 'merged' AND merged_at IS NOT NULL

453

ORDER BY merged_at DESC LIMIT ?2"

454

))

455

.bind(&[serde_json::to_string(&ids)?.into(), LANDED.into()])?

456

.all()

457

.await?

458

.results::<PullRow>()?;

Ok(pulls

.into_iter()

.map(Pull::from)

.filter_map(|mut pull| {

463

let repo = repos.iter().find(|repo| repo.id == pull.repo_id)?;

464

crate::fill_base(&mut pull, repo);

465

Some(RepoPull { repo: format!("{}/{}", repo.namespace, repo.name), pull })

})

.collect())

}

/// One harness or hosted session of a workspace by its id, read as its

471

/// repository's own list reads it.

472

pub(crate) async fn get_workspace_session(&self, a: GetWorkspaceSessionArgs) -> Result<Outcome<WorkspaceSessionDetail>> {

473

let workspace = a.workspace.trim().to_lowercase();

let placed = self

.db

.prepare(

"SELECT s.repo_id, rw.path FROM sessions s JOIN repo_workspaces rw ON rw.repo_id = s.repo_id

478

WHERE s.id = ?1 AND rw.workspace = ?2",

479

)

480

.bind(&[a.id.trim().into(), workspace.as_str().into()])?

481

.first::<Placed>(None)

482

.await?;

483

let Some(placed) = placed else {

484

return Ok(Outcome::fail(FailureCode::NotFound, "Session not found."));

485

};

486

let Some((namespace, name)) = placed.path.split_once('/') else {

487

return Ok(Outcome::fail(FailureCode::NotFound, "Session not found."));

488

};

489

let repo = RepoPath { namespace: namespace.to_owned(), name: name.to_owned() };

490

let args = WorkSessionArgs { viewer: a.viewer, repo, id: a.id.trim().to_owned(), after_seq: a.after_seq };

491

Ok(match self.get_work_session(args).await? {

492

Outcome::Ok(detail) => Outcome::Ok(WorkspaceSessionDetail { repo: placed.path, detail }),

493

Outcome::Fail(failure) => Outcome::Fail(failure),

})

}

}

fn sessions_repos(repos: &[Repo]) -> Vec<SessionsRepo> {

499

repos

500

.iter()

501

.map(|repo| SessionsRepo { id: repo.id.clone(), path: format!("{}/{}", repo.namespace, repo.name), default_branch: repo.default_branch.clone() })

.collect()

}

#[cfg(test)]

mod tests {

use super::*;

use crate::runs::RECORDED_IN;

509

use rusqlite::{Connection, params};

510

511

const NOW: u64 = 1_791_800_000_000;

512

const MINUTE: u64 = 60_000;

513

514

fn at(ago_minutes: u64) -> String {

515

rfc3339(NOW - ago_minutes * MINUTE)

516

}

517

518

/// Two repositories of acme and one of another workspace, a work item,

519

/// and pull requests in each.

520

fn database() -> Connection {

521

let db = crate::migrated::database();

522

db.execute_batch(

523

"INSERT INTO repo_workspaces (repo_id, workspace, path, private) VALUES

524

('rep_web', 'acme', 'acme/web', 1), ('rep_api', 'acme', 'acme/api', 1), ('rep_x', 'other', 'other/x', 0);

525

INSERT INTO work_items (id, workspace, number, title, body, author_id, author_name, created_at, updated_at)

526

VALUES ('wki_4', 'acme', 4, 'Greet people', '', 'usr_1', 'ana', '', '');

527

INSERT INTO pulls (id, repo_id, number, title, agent, runtime, author_id, author_name, created_at, updated_at, item_id) VALUES

528

('pr_web_9', 'rep_web', 9, 'Say hello', 'claude-code', 'external', 'usr_1', 'ana', '', '', 'wki_4'),

529

('pr_api_7', 'rep_api', 7, 'Recorded only', 'cursor', 'external', 'usr_2', 'bo', '', '', NULL),

530

('pr_api_8', 'rep_api', 8, 'By g1t', 'g1t', 'external', 'usr_2', 'bo', '', '', NULL),

531

('pr_x_1', 'rep_x', 1, 'Elsewhere', 'codex', 'external', 'usr_3', 'cy', '', '', NULL);",

)

.unwrap();

db

}

#[allow(clippy::too_many_arguments)]

538

fn session(db: &Connection, id: &str, repo: &str, item: Option<&str>, pull: Option<u32>, actor: &str, harness: &str, state: &str, last: u64, parent: Option<&str>) {

539

db.execute(

540

"INSERT INTO sessions (id, repo_id, item_id, pull, branch, actor_id, actor_name, harness, harness_session_id,

541

title, state, head_commit, started_at, last_seen_at, parent_id)

542

VALUES (?1, ?2, ?3, ?4, 'main', ?5, ?5, ?6, ?1, NULL, ?7, NULL, ?8, ?8, ?9)",

543

params![id, repo, item, pull, actor, harness, state, at(last), parent],

)

.unwrap();

}

#[allow(clippy::too_many_arguments)]

549

fn listed(db: &Connection, shown: SessionsShown, repos: &[&str], harness: Option<&str>, person: Option<&str>, item: Option<&str>, kind: Option<&str>) -> Vec<String> {

550

let sql = workspace_sessions_sql(shown);

551

let mut statement = db.prepare(&sql).unwrap();

552

statement

553

.query_map(

554

params![serde_json::to_string(repos).unwrap(), rfc3339(NOW - SESSION_TIMEOUT_MS), 0, harness, person, item, 100, kind],

555

|row| row.get::<_, String>("id"),

)

.unwrap()

.map(Result::unwrap)

.collect()

}

fn seeded() -> Connection {

563

let db = database();

564

session(&db, "ses_cc", "rep_web", Some("wki_4"), Some(9), "ana", "claude_code", "active", 1, None);

565

session(&db, "ses_sub", "rep_web", Some("wki_4"), None, "ana", "claude_code", "active", 1, Some("ses_cc"));

566

session(&db, "ses_codex", "rep_api", None, None, "bo", "codex", "active", 20, None);

567

session(&db, "ses_cursor", "rep_api", None, None, "bo", "cursor", "active", 45, None);

568

session(&db, "ses_hosted", "rep_api", None, None, "ana", "claude_managed_agents", "active", 3, None);

569

session(&db, "ses_done", "rep_web", None, None, "cy", "other", "ended", 90, None);

570

session(&db, "ses_x", "rep_x", None, None, "cy", "codex", "active", 1, None);

db

}

#[test]

fn only_the_repositories_looked_in_are_listed_and_subagents_stay_under_their_parent() {

576

let db = seeded();

577

let all = listed(&db, SessionsShown::Recent, &["rep_web", "rep_api"], None, None, None, None);

578

assert_eq!(all, ["ses_cc", "ses_hosted", "ses_codex", "ses_cursor", "ses_done"]);

579

// A repository the viewer may not read is never looked in.

580

let web = listed(&db, SessionsShown::Recent, &["rep_web"], None, None, None, None);

581

assert_eq!(web, ["ses_cc", "ses_done"]);

582

assert!(!all.contains(&"ses_x".to_owned()));

}

#[test]

fn live_is_what_is_going_and_ended_takes_those_gone_quiet_too() {

587

let db = seeded();

588

let repos = ["rep_web", "rep_api"];

589

// Idle (20 minutes) is still live; quiet past the timeout (45) is not.

590

assert_eq!(listed(&db, SessionsShown::Live, &repos, None, None, None, None), ["ses_cc", "ses_hosted", "ses_codex"]);

591

assert_eq!(listed(&db, SessionsShown::Ended, &repos, None, None, None, None), ["ses_cursor", "ses_done"]);

}

#[test]

fn sessions_filter_by_harness_person_kind_and_work_item() {

596

let db = seeded();

597

let repos = ["rep_web", "rep_api"];

598

assert_eq!(listed(&db, SessionsShown::Recent, &repos, Some("codex"), None, None, None), ["ses_codex"]);

599

assert_eq!(listed(&db, SessionsShown::Recent, &repos, None, Some("BO"), None, None), ["ses_codex", "ses_cursor"]);

600

assert_eq!(listed(&db, SessionsShown::Recent, &repos, None, None, None, Some("hosted")), ["ses_hosted"]);

601

assert_eq!(listed(&db, SessionsShown::Recent, &repos, None, None, None, Some("harness")), ["ses_cc", "ses_codex", "ses_cursor", "ses_done"]);

602

assert_eq!(listed(&db, SessionsShown::Recent, &repos, None, None, Some("wki_4"), None), ["ses_cc"]);

603

// Tied to the work item another way: a link.

604

db.execute(super::super::links::INSERT_LINK, params!["ses_codex", "rep_api", "item", 0, "ref", at(1), "wki_4"]).unwrap();

605

assert_eq!(listed(&db, SessionsShown::Recent, &repos, None, None, Some("wki_4"), None), ["ses_cc", "ses_codex"]);

606

}

607

608

fn recorded(db: &Connection, repos: &[&str]) -> Vec<RecordedRow> {

609

let mut statement = db.prepare(RECORDED_IN).unwrap();

610

let rows = statement

611

.query_map(params![serde_json::to_string(repos).unwrap(), 50], |row| {

612

Ok(RecordedRow {

613

pull_id: row.get("pull_id")?,

614

repo_id: row.get("repo_id")?,

615

number: row.get("number")?,

616

title: row.get("title")?,

617

status: PullStatus::Open,

618

agent: row.get("agent")?,

619

author_name: row.get("author_name")?,

620

item: row.get("item")?,

621

started_at: row.get("started_at")?,

622

last_at: row.get("last_at")?,

623

has_runs: row.get("has_runs")?,

})

})

.unwrap();

rows.map(Result::unwrap).collect()

628

}

629

630

fn run(number: u32, status: &str) -> AgentRun {

631

serde_json::from_value(serde_json::json!({

632

"id": "run_1", "repo": { "namespace": "acme", "name": "api" }, "number": number, "title": null, "kind": "implement",

633

"agent": "g1t", "model": null, "status": status, "step": "Running tests", "stepCount": 3, "startedBy": null,

634

"error": null, "costUsd": null, "turns": null, "createdAt": at(10), "startedAt": at(10), "finishedAt": null, "updatedAt": at(1)

}))

.unwrap()

}

#[test]

fn a_pull_request_recorded_directly_is_a_session_and_one_moved_from_a_harness_is_not() {

641

let db = seeded();

642

db.execute_batch(&format!(

643

"INSERT INTO session_entries (pull_id, seq, kind, text, at) VALUES ('pr_api_7', 1, 'prompt', 'Fix it', '{a}'), ('pr_api_7', 2, 'message', 'Done', '{b}');

644

INSERT INTO session_entries (pull_id, seq, kind, text, at, session_id, session_seq) VALUES ('pr_web_9', 1, 'message', 'Moved', '{b}', 'ses_cc', 1);

645

INSERT INTO session_entries (pull_id, seq, kind, text, at) VALUES ('pr_x_1', 1, 'message', 'Elsewhere', '{b}');

646

INSERT INTO session_entries (pull_id, seq, kind, text, at) VALUES ('pr_api_8', 1, 'message', 'g1t', '{c}');

647

INSERT INTO agent_runs (id, workspace, repo_id, repo, number, pull_id, kind, agent, status, sandbox, token_hash, created_at, updated_at)

648

VALUES ('run_1', 'acme', 'rep_api', 'acme/api', 8, 'pr_api_8', 'implement', 'g1t', 'running', 'sb', 'h', '{a}', '{b}');",

a = at(10),

b = at(2),

c = at(5)

))

.unwrap();

let rows = recorded(&db, &["rep_web", "rep_api"]);

655

let numbers: Vec<u32> = rows.iter().map(|row| row.number).collect();

656

assert_eq!(numbers, [7, 8]);

657

assert_eq!(rows[0].started_at, at(10));

658

assert_eq!(rows[0].has_runs, 0);

659

assert_eq!(rows[1].has_runs, 1);

660

// Together: the run of g1t's agent stands for its pull request, and

661

// the recorded one with nothing behind it is a session of its own.

662

let recorded: Vec<(WorkspaceSession, bool)> = rows

663

.into_iter()

664

.map(|row| {

665

let runs = row.has_runs != 0;

666

(from_recorded(row, "acme/api", NOW), runs)

667

})

668

.collect();

669

let wanted = Wanted { shown: SessionsShown::Recent, kind: None, harness: None, person: None, item: None };

670

let all = merged(Vec::new(), vec![from_run(run(8, "running"), "rep_api")], recorded, &wanted, 10);

671

let kinds: Vec<(SessionKind, &str)> = all.iter().map(|session| (session.kind, session.id.as_str())).collect();

672

assert_eq!(kinds, [(SessionKind::Agent, "run_1"), (SessionKind::Recorded, "pr_api_7")]);

673

assert_eq!(all[0].actor.as_deref(), Some("g1t"));

674

assert_eq!(all[0].state, SessionState::Active);

675

assert_eq!(all[1].state, SessionState::Active);

676

let live_by_bo = Wanted { shown: SessionsShown::Live, kind: Some(SessionKind::Recorded), harness: None, person: Some("bo"), item: None };

677

assert_eq!(merged(Vec::new(), Vec::new(), vec![(all[1].clone(), false)], &live_by_bo, 10).len(), 1);

678

let by_harness = Wanted { shown: SessionsShown::Recent, kind: None, harness: Some(Harness::Codex), person: None, item: None };

679

assert!(merged(Vec::new(), Vec::new(), vec![(all[1].clone(), false)], &by_harness, 10).is_empty());

680

// A finished run is not live.

681

let live = Wanted { shown: SessionsShown::Live, kind: None, harness: None, person: None, item: None };

682

assert!(merged(Vec::new(), vec![from_run(run(8, "succeeded"), "rep_api")], Vec::new(), &live, 10).is_empty());

}

#[test]

fn every_kind_reads_as_one_shape_and_a_quiet_session_reads_as_ended() {

687

let session: Session = serde_json::from_value(serde_json::json!({

688

"id": "ses_q", "repoId": "rep_api", "item": null, "pull": 7, "branch": "fix", "actor": { "id": "usr_2", "username": "bo" },

689

"harness": "claude_managed_agents", "harnessSessionId": null, "title": "Fix it", "state": "idle", "headCommit": null,

690

"startedAt": at(60), "lastSeenAt": at(40), "endedAt": null, "endReason": null, "subagents": 2

691

}))

692

.unwrap();

693

let listed = from_session(session, "acme/api", NOW);

694

assert_eq!(listed.kind, SessionKind::Hosted);

695

assert_eq!(listed.state, SessionState::Ended);

696

assert_eq!(listed.session.as_ref().unwrap().end_reason.as_deref(), Some("timed_out"));

697

assert_eq!(listed.subagents, 2);

698

assert_eq!(listed.repo, "acme/api");

699

// A recorded one on the pull request this session is on is this session.

700

let recorded = RecordedRow {

701

pull_id: "pr_api_7".into(),

702

repo_id: "rep_api".into(),

703

number: 7,

704

title: "Fix it".into(),

705

status: PullStatus::Open,

706

agent: "cursor".into(),

707

author_name: "bo".into(),

item: None,

started_at: at(60),

last_at: at(50),

has_runs: 0,

};

let wanted = Wanted { shown: SessionsShown::Recent, kind: None, harness: None, person: None, item: None };

714

let all = merged(vec![listed], Vec::new(), vec![(from_recorded(recorded, "acme/api", NOW), false)], &wanted, 10);

715

assert_eq!(all.len(), 1);

716

assert_eq!(all[0].id, "ses_q");

}

#[test]

fn a_session_linked_to_a_work_item_is_listed_for_it() {

721

// On no work item of its own, but linked to #12 (a claim, a trailer).

722

let session: Session = serde_json::from_value(serde_json::json!({

723

"id": "ses_l", "repoId": "rep_api", "item": null, "pull": null, "branch": null, "actor": { "id": "usr_2", "username": "bo" },

724

"harness": "codex", "harnessSessionId": null, "title": null, "state": "active", "headCommit": null,

725

"startedAt": at(5), "lastSeenAt": at(1), "endedAt": null, "endReason": null, "subagents": 0

726

}))

727

.unwrap();

728

let listed = from_session(session, "acme/api", NOW);

729

let for_item = Wanted { shown: SessionsShown::Recent, kind: None, harness: None, person: None, item: Some(12) };

730

assert_eq!(merged(vec![listed], Vec::new(), Vec::new(), &for_item, 10).len(), 1);

}

#[test]

fn pull_facts_and_busy_queues_are_read_across_repositories() {

735

let db = seeded();

736

let pairs = serde_json::json!([{ "r": "rep_web", "n": 9 }, { "r": "rep_api", "n": 7 }, { "r": "rep_api", "n": 99 }]).to_string();

737

let mut statement = db.prepare(PULL_FACTS).unwrap();

738

let mut facts: Vec<(String, u32, String, Option<u32>)> = statement

739

.query_map(params![pairs], |row| Ok((row.get("repo_id")?, row.get("number")?, row.get("title")?, row.get("item")?)))

.unwrap()

.map(Result::unwrap)

.collect();

facts.sort();

assert_eq!(facts, [("rep_api".into(), 7, "Recorded only".into(), None), ("rep_web".into(), 9, "Say hello".into(), Some(4))]);

745

db.execute_batch(

746

"INSERT INTO queue_entries (id, repo_id, pull_id, number, state, enqueued_by, created_at) VALUES

747

('q1', 'rep_api', 'pr_api_7', 7, 'waiting', '{}', '1'), ('q2', 'rep_api', 'pr_api_8', 8, 'testing', '{}', '2'),

748

('q3', 'rep_web', 'pr_web_9', 9, 'landed', '{}', '3'), ('q4', 'rep_x', 'pr_x_1', 1, 'waiting', '{}', '4');",

749

)

750

.unwrap();

751

let mut statement = db.prepare(QUEUED_REPOS).unwrap();

752

let busy: Vec<String> =

753

statement.query_map(params![r#"["rep_web","rep_api"]"#, 12], |row| row.get::<_, String>("value")).unwrap().map(Result::unwrap).collect();

754

assert_eq!(busy, ["rep_api"]);

755

}

756

}