diff --git a/.github/workflows/sdlc-orchestrator.yml b/.github/workflows/sdlc-orchestrator.yml index 643f576..3707315 100644 --- a/.github/workflows/sdlc-orchestrator.yml +++ b/.github/workflows/sdlc-orchestrator.yml @@ -1,6 +1,9 @@ name: Tutti SDLC Orchestrator on: + schedule: + # Sweep stale claims every 30 minutes + - cron: "*/30 * * * *" workflow_dispatch: inputs: mode: @@ -17,7 +20,27 @@ on: default: "agent-ops" jobs: + # Periodic sweep of stale issue claim leases + sweep-stale-claims: + if: ${{ github.event_name == 'schedule' }} + runs-on: ubuntu-latest + permissions: + issues: write + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Setup Rust + uses: dtolnay/rust-toolchain@stable + + - name: Sweep stale claims + env: + GH_TOKEN: ${{ github.token }} + run: | + cargo run --quiet -- issue-claim sweep + orchestrate: + if: ${{ github.event_name == 'workflow_dispatch' }} runs-on: ubuntu-latest concurrency: group: sdlc-orchestrator-${{ github.repository }} @@ -53,12 +76,27 @@ jobs: - name: Preflight tests run: cargo test -q - - name: Run smoke workflow + - name: Run smoke workflow with claim lifecycle if: ${{ inputs.mode == 'smoke' }} env: GH_TOKEN: ${{ github.token }} ISSUE_LABEL: ${{ inputs.issue_label }} + GITHUB_RUN_ID: ${{ github.run_id }} run: | + STATE_FILE=".tutti/state/auto/selected_issue.json" + cleanup() { + if [ -f "$STATE_FILE" ]; then + cargo run --quiet -- issue-claim release --state "$STATE_FILE" \ + --reason "workflow finished for run $GITHUB_RUN_ID" || true + fi + } + trap cleanup EXIT + + # Start heartbeat in background (every 5 min) + ( while true; do sleep 300; [ -f "$STATE_FILE" ] && cargo run --quiet -- issue-claim heartbeat --state "$STATE_FILE" 2>/dev/null || true; done ) & + HEARTBEAT_PID=$! + trap "kill $HEARTBEAT_PID 2>/dev/null; cleanup" EXIT + cargo run --quiet -- run sdlc-smoke --strict - name: Verify codex CLI is available @@ -66,10 +104,25 @@ jobs: run: | command -v codex >/dev/null || { echo "codex CLI not found on runner"; exit 1; } - - name: Run full auto workflow + - name: Run full auto workflow with claim lifecycle if: ${{ inputs.mode == 'auto' }} env: GH_TOKEN: ${{ github.token }} ISSUE_LABEL: ${{ inputs.issue_label }} + GITHUB_RUN_ID: ${{ github.run_id }} run: | + STATE_FILE=".tutti/state/auto/selected_issue.json" + cleanup() { + if [ -f "$STATE_FILE" ]; then + cargo run --quiet -- issue-claim release --state "$STATE_FILE" \ + --reason "workflow finished for run $GITHUB_RUN_ID" || true + fi + } + trap cleanup EXIT + + # Start heartbeat in background (every 5 min) + ( while true; do sleep 300; [ -f "$STATE_FILE" ] && cargo run --quiet -- issue-claim heartbeat --state "$STATE_FILE" 2>/dev/null || true; done ) & + HEARTBEAT_PID=$! + trap "kill $HEARTBEAT_PID 2>/dev/null; cleanup" EXIT + cargo run --quiet -- run sdlc-auto --strict diff --git a/docs/examples/tutti-codex-sdlc.toml b/docs/examples/tutti-codex-sdlc.toml index d396f4d..6f88f2c 100644 --- a/docs/examples/tutti-codex-sdlc.toml +++ b/docs/examples/tutti-codex-sdlc.toml @@ -60,7 +60,7 @@ description = "Preflight test run before unattended automation" [[workflow.step]] id = "select_issue" type = "command" -run = "scripts/automation/select_issue.sh .tutti/state/auto/selected_issue.json \"${ISSUE_LABEL:-agent-ops}\"" +run = "cargo run --quiet -- issue-claim acquire --output .tutti/state/auto/selected_issue.json --label \"${ISSUE_LABEL:-agent-ops}\" --lease-ttl-secs 1800" fail_mode = "closed" [[workflow.step]] @@ -110,7 +110,7 @@ agent = "docs-release" [[workflow.step]] id = "select_issue" type = "command" -run = "scripts/automation/select_issue.sh .tutti/state/auto/selected_issue.json \"${ISSUE_LABEL:-agent-ops}\"" +run = "cargo run --quiet -- issue-claim acquire --output .tutti/state/auto/selected_issue.json --label \"${ISSUE_LABEL:-agent-ops}\" --lease-ttl-secs 1800" fail_mode = "closed" [[workflow.step]] diff --git a/src/cli/issue_claim.rs b/src/cli/issue_claim.rs new file mode 100644 index 0000000..5dd8a5f --- /dev/null +++ b/src/cli/issue_claim.rs @@ -0,0 +1,887 @@ +use crate::error::{Result, TuttiError}; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use std::path::Path; +use std::process::Command; + +const CLAIM_MARKER_START: &str = ""; +const DEFAULT_LEASE_TTL_SECS: u64 = 1800; // 30 minutes +const MAX_EVENTS: usize = 20; + +// --------------------------------------------------------------------------- +// Data types +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ClaimStatus { + Active, + Released, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ClaimEvent { + pub timestamp: DateTime, + pub kind: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub detail: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ClaimRecord { + pub schema_version: u32, + pub run_id: String, + pub claimed_at: DateTime, + pub lease_ttl_secs: u64, + pub last_heartbeat_at: DateTime, + pub status: ClaimStatus, + #[serde(skip_serializing_if = "Option::is_none")] + pub released_at: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub release_reason: Option, + pub events: Vec, +} + +impl ClaimRecord { + fn new(run_id: &str, lease_ttl_secs: u64) -> Self { + let now = Utc::now(); + Self { + schema_version: 1, + run_id: run_id.to_string(), + claimed_at: now, + lease_ttl_secs, + last_heartbeat_at: now, + status: ClaimStatus::Active, + released_at: None, + release_reason: None, + events: vec![ClaimEvent { + timestamp: now, + kind: "Acquired".to_string(), + detail: None, + }], + } + } + + fn expires_at(&self) -> DateTime { + { + let secs = i64::try_from(self.lease_ttl_secs).unwrap_or(i64::MAX); + self.last_heartbeat_at + chrono::Duration::seconds(secs) + } + } + + fn is_expired(&self) -> bool { + Utc::now() > self.expires_at() + } + + fn push_event(&mut self, kind: &str, detail: Option) { + self.events.push(ClaimEvent { + timestamp: Utc::now(), + kind: kind.to_string(), + detail, + }); + if self.events.len() > MAX_EVENTS { + self.events.remove(0); + } + } +} + +/// Output written to selected_issue.json (extends the old format). +#[derive(Debug, Serialize, Deserialize)] +pub struct SelectedIssueOutput { + pub issue_number: u64, + pub title: String, + pub url: String, + pub labels: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub author: Option, + pub created_at: String, + pub body: String, + // claim metadata + pub run_id: String, + pub claimed_at: DateTime, + pub lease_ttl_secs: u64, + pub last_heartbeat_at: DateTime, + pub claim_comment_id: u64, + pub claim_status: ClaimStatus, +} + +// --------------------------------------------------------------------------- +// GitHub helpers +// --------------------------------------------------------------------------- + +fn gh_repo() -> Result { + if let Ok(repo) = std::env::var("GITHUB_REPOSITORY") { + return Ok(repo); + } + let out = Command::new("gh") + .args([ + "repo", + "view", + "--json", + "nameWithOwner", + "-q", + ".nameWithOwner", + ]) + .output()?; + if !out.status.success() { + return Err(TuttiError::IssueClaim( + "failed to detect repository via gh cli".into(), + )); + } + Ok(String::from_utf8_lossy(&out.stdout).trim().to_string()) +} + +fn gh_run_id() -> String { + std::env::var("GITHUB_RUN_ID").unwrap_or_else(|_| format!("local-{}", std::process::id())) +} + +/// List open issues for a label, returning raw JSON array. +fn gh_list_issues(repo: &str, label: &str) -> Result> { + let out = Command::new("gh") + .args([ + "issue", + "list", + "--repo", + repo, + "--state", + "open", + "--label", + label, + "--limit", + "100", + "--json", + "number,title,url,labels,author,createdAt", + ]) + .output()?; + if !out.status.success() { + return Err(TuttiError::IssueClaim(format!( + "gh issue list failed: {}", + String::from_utf8_lossy(&out.stderr) + ))); + } + let items: Vec = serde_json::from_slice(&out.stdout)?; + Ok(items) +} + +/// Fetch full issue body. +fn gh_issue_body(repo: &str, number: u64) -> Result { + let out = Command::new("gh") + .args([ + "issue", + "view", + &number.to_string(), + "--repo", + repo, + "--json", + "body", + "-q", + ".body", + ]) + .output()?; + if !out.status.success() { + return Err(TuttiError::IssueClaim(format!( + "gh issue view failed: {}", + String::from_utf8_lossy(&out.stderr) + ))); + } + Ok(String::from_utf8_lossy(&out.stdout).trim().to_string()) +} + +/// Add a label to an issue. +fn gh_add_label(repo: &str, number: u64, label: &str) -> Result<()> { + let out = Command::new("gh") + .args([ + "issue", + "edit", + &number.to_string(), + "--repo", + repo, + "--add-label", + label, + ]) + .output()?; + if !out.status.success() { + return Err(TuttiError::IssueClaim(format!( + "gh add label failed: {}", + String::from_utf8_lossy(&out.stderr) + ))); + } + Ok(()) +} + +/// Remove a label from an issue. +fn gh_remove_label(repo: &str, number: u64, label: &str) -> Result<()> { + let out = Command::new("gh") + .args([ + "issue", + "edit", + &number.to_string(), + "--repo", + repo, + "--remove-label", + label, + ]) + .output()?; + if !out.status.success() { + let stderr = String::from_utf8_lossy(&out.stderr).to_lowercase(); + // Idempotent case: label already absent. + if stderr.contains("not found") + || stderr.contains("does not exist") + || (stderr.contains("label") && stderr.contains("not") && stderr.contains("issue")) + { + return Ok(()); + } + return Err(TuttiError::IssueClaim(format!( + "gh remove label failed: {}", + String::from_utf8_lossy(&out.stderr) + ))); + } + Ok(()) +} + +fn is_trusted_claim_comment(comment: &serde_json::Value, expected_run_id: Option<&str>) -> bool { + let body = match comment["body"].as_str() { + Some(b) => b, + None => return false, + }; + let record = match decode_claim_comment(body) { + Some(r) => r, + None => return false, + }; + + if let Some(run_id) = expected_run_id + && record.run_id != run_id + { + return false; + } + + let login = comment["user"]["login"].as_str().unwrap_or(""); + let user_type = comment["user"]["type"].as_str().unwrap_or(""); + + if std::env::var("GITHUB_ACTIONS").is_ok() { + let actor = std::env::var("GITHUB_ACTOR").unwrap_or_default(); + return user_type == "Bot" + || login == "github-actions[bot]" + || (!actor.is_empty() && login == actor); + } + + true +} + +/// Create a comment on an issue, return the comment ID. +fn gh_create_comment(repo: &str, number: u64, body: &str) -> Result { + let out = Command::new("gh") + .args([ + "issue", + "comment", + &number.to_string(), + "--repo", + repo, + "--body", + body, + ]) + .output()?; + if !out.status.success() { + return Err(TuttiError::IssueClaim(format!( + "gh issue comment failed: {}", + String::from_utf8_lossy(&out.stderr) + ))); + } + // gh issue comment outputs the URL of the created comment + // e.g. https://github.com/owner/repo/issues/29#issuecomment-123456 + let url = String::from_utf8_lossy(&out.stdout).trim().to_string(); + // Extract comment ID from URL + if let Some(id_str) = url.rsplit("issuecomment-").next() + && let Ok(id) = id_str.trim().parse::() + { + return Ok(id); + } + // Fallback: fetch comments and find ours + let expected_run_id = decode_claim_comment(body).map(|r| r.run_id); + let comments = gh_list_comments(repo, number)?; + for c in comments.iter().rev() { + if is_trusted_claim_comment(c, expected_run_id.as_deref()) + && let Some(id) = c["id"].as_u64() + { + return Ok(id); + } + } + Err(TuttiError::IssueClaim( + "could not determine created comment ID".into(), + )) +} + +/// Update an existing comment. +fn gh_update_comment(repo: &str, comment_id: u64, body: &str) -> Result<()> { + let out = Command::new("gh") + .args([ + "api", + &format!("repos/{repo}/issues/comments/{comment_id}"), + "-X", + "PATCH", + "-f", + &format!("body={body}"), + ]) + .output()?; + if !out.status.success() { + return Err(TuttiError::IssueClaim(format!( + "gh api comment update failed: {}", + String::from_utf8_lossy(&out.stderr) + ))); + } + Ok(()) +} + +/// List comments on an issue. +fn gh_list_comments(repo: &str, number: u64) -> Result> { + let out = Command::new("gh") + .args([ + "api", + &format!("repos/{repo}/issues/{number}/comments"), + "--paginate", + "--slurp", + ]) + .output()?; + if !out.status.success() { + return Err(TuttiError::IssueClaim(format!( + "gh api comments failed: {}", + String::from_utf8_lossy(&out.stderr) + ))); + } + let pages: Vec> = serde_json::from_slice(&out.stdout)?; + let comments = pages.into_iter().flatten().collect(); + Ok(comments) +} + +/// List issues with a specific label (for sweep). +fn gh_issues_with_label(repo: &str, label: &str) -> Result> { + let out = Command::new("gh") + .args([ + "issue", "list", "--repo", repo, "--state", "open", "--label", label, "--limit", "100", + "--json", "number", + ]) + .output()?; + if !out.status.success() { + return Err(TuttiError::IssueClaim(format!( + "gh issue list for sweep failed: {}", + String::from_utf8_lossy(&out.stderr) + ))); + } + let items: Vec = serde_json::from_slice(&out.stdout)?; + Ok(items.iter().filter_map(|v| v["number"].as_u64()).collect()) +} + +// --------------------------------------------------------------------------- +// Claim comment encoding +// --------------------------------------------------------------------------- + +fn encode_claim_comment(record: &ClaimRecord) -> Result { + let json = serde_json::to_string(record)?; + let status_emoji = match record.status { + ClaimStatus::Active => "🔒", + ClaimStatus::Released => "🔓", + }; + Ok(format!( + "{CLAIM_MARKER_START}{json}{CLAIM_MARKER_END}\n\n\ + {status_emoji} **Claim** — run `{}` | status: `{:?}` | \ + expires: {} | heartbeat: {}", + record.run_id, + record.status, + record.expires_at().format("%Y-%m-%dT%H:%M:%SZ"), + record.last_heartbeat_at.format("%Y-%m-%dT%H:%M:%SZ"), + )) +} + +fn decode_claim_comment(body: &str) -> Option { + let start = body.find(CLAIM_MARKER_START)?; + let json_start = start + CLAIM_MARKER_START.len(); + let end = body[json_start..].find(CLAIM_MARKER_END)?; + let json_str = &body[json_start..json_start + end]; + serde_json::from_str(json_str).ok() +} + +// --------------------------------------------------------------------------- +// Claim operations +// --------------------------------------------------------------------------- + +/// Find all claim comments on an issue, returning (comment_id, record) pairs. +fn find_claim_comments(repo: &str, issue_number: u64) -> Result> { + let comments = gh_list_comments(repo, issue_number)?; + let mut claims = Vec::new(); + for c in &comments { + if is_trusted_claim_comment(c, None) + && let Some(body) = c["body"].as_str() + && let Some(record) = decode_claim_comment(body) + && let Some(id) = c["id"].as_u64() + { + claims.push((id, record)); + } + } + Ok(claims) +} + +/// Release stale (expired) claims on an issue. +fn release_stale_claims(repo: &str, issue_number: u64) -> Result { + let claims = find_claim_comments(repo, issue_number)?; + let mut released = 0u32; + for (comment_id, mut record) in claims { + if record.status == ClaimStatus::Active && record.is_expired() { + record.status = ClaimStatus::Released; + record.released_at = Some(Utc::now()); + record.release_reason = Some("lease expired (stale)".into()); + record.push_event("Released", Some("lease expired".into())); + let body = encode_claim_comment(&record)?; + gh_update_comment(repo, comment_id, &body)?; + released += 1; + eprintln!( + "released stale claim on #{} (run {}, expired {})", + issue_number, + record.run_id, + record.expires_at().format("%Y-%m-%dT%H:%M:%SZ") + ); + } + } + Ok(released) +} + +/// Find the winning active claim (earliest claimed_at, then lowest comment_id). +fn winner_active_claim(claims: &[(u64, ClaimRecord)]) -> Option<(u64, &ClaimRecord)> { + claims + .iter() + .filter(|(_, r)| r.status == ClaimStatus::Active && !r.is_expired()) + // Use server-ordered comment id to avoid cross-runner clock skew. + .min_by_key(|(cid, _)| *cid) + .map(|(cid, r)| (*cid, r)) +} + +// --------------------------------------------------------------------------- +// Public subcommands +// --------------------------------------------------------------------------- + +/// `tt issue-claim acquire` +pub fn acquire(output_path: &Path, label: &str, lease_ttl_secs: Option) -> Result<()> { + let repo = gh_repo()?; + let run_id = gh_run_id(); + let ttl = lease_ttl_secs.unwrap_or(DEFAULT_LEASE_TTL_SECS); + + let issues = gh_list_issues(&repo, label)?; + + // Filter out already-claimed issues and sort by creation date + let mut candidates: Vec<&serde_json::Value> = issues + .iter() + .filter(|i| { + let labels = i["labels"] + .as_array() + .map(|arr| { + arr.iter() + .filter_map(|l| l["name"].as_str()) + .collect::>() + }) + .unwrap_or_default(); + !labels.contains(&"automation-claimed") + }) + .collect(); + + candidates.sort_by_key(|i| i["createdAt"].as_str().unwrap_or("").to_string()); + + if candidates.is_empty() { + // Also try issues with automation-claimed but whose claims are all stale + let claimed_issues: Vec<&serde_json::Value> = issues + .iter() + .filter(|i| { + let labels = i["labels"] + .as_array() + .map(|arr| { + arr.iter() + .filter_map(|l| l["name"].as_str()) + .collect::>() + }) + .unwrap_or_default(); + labels.contains(&"automation-claimed") + }) + .collect(); + + for issue in &claimed_issues { + let number = issue["number"].as_u64().unwrap_or(0); + if number == 0 { + continue; + } + release_stale_claims(&repo, number)?; + let claims = find_claim_comments(&repo, number)?; + if winner_active_claim(&claims).is_none() { + // All claims released — remove label and treat as candidate + gh_remove_label(&repo, number, "automation-claimed")?; + candidates.push(issue); + } + } + candidates.sort_by_key(|i| i["createdAt"].as_str().unwrap_or("").to_string()); + } + + if candidates.is_empty() { + return Err(TuttiError::IssueClaim(format!( + "no unclaimed open issues found for label '{label}'" + ))); + } + + for issue in candidates { + let number = match issue["number"].as_u64() { + Some(n) => n, + None => continue, + }; + + // Create claim record and post as comment + let record = ClaimRecord::new(&run_id, ttl); + let comment_body = encode_claim_comment(&record)?; + + // Add label first to prevent races + gh_add_label(&repo, number, "automation-claimed")?; + + let comment_id = match gh_create_comment(&repo, number, &comment_body) { + Ok(id) => id, + Err(err) => { + // Best-effort cleanup to avoid stranded labels when comment creation fails. + let _ = gh_remove_label(&repo, number, "automation-claimed"); + return Err(err); + } + }; + + // Verify we won the race (check all claims, find winner) + let claims = find_claim_comments(&repo, number)?; + if let Some((winner_id, _)) = winner_active_claim(&claims) + && winner_id != comment_id + { + // We lost the race — release our claim and continue to next candidate. + let mut our_record = record.clone(); + our_record.status = ClaimStatus::Released; + our_record.released_at = Some(Utc::now()); + our_record.release_reason = Some("lost claim race".into()); + our_record.push_event("Released", Some("lost claim race".into())); + let body = encode_claim_comment(&our_record)?; + gh_update_comment(&repo, comment_id, &body)?; + continue; + } + + // Fetch issue body + let body = gh_issue_body(&repo, number)?; + + // Build output + let labels: Vec = issue["labels"] + .as_array() + .map(|arr| { + arr.iter() + .filter_map(|l| l["name"].as_str().map(|s| s.to_string())) + .collect() + }) + .unwrap_or_default(); + + let output = SelectedIssueOutput { + issue_number: number, + title: issue["title"].as_str().unwrap_or("").to_string(), + url: issue["url"].as_str().unwrap_or("").to_string(), + labels, + author: issue["author"]["login"].as_str().map(|s| s.to_string()), + created_at: issue["createdAt"].as_str().unwrap_or("").to_string(), + body, + run_id: run_id.clone(), + claimed_at: record.claimed_at, + lease_ttl_secs: ttl, + last_heartbeat_at: record.last_heartbeat_at, + claim_comment_id: comment_id, + claim_status: ClaimStatus::Active, + }; + + // Write output + if let Some(parent) = output_path.parent() { + std::fs::create_dir_all(parent)?; + } + let tmp_path = output_path.with_extension("json.tmp"); + let json = serde_json::to_string_pretty(&output)?; + std::fs::write(&tmp_path, &json)?; + std::fs::rename(&tmp_path, output_path)?; + + eprintln!( + "claimed issue #{} (comment {}, ttl {}s, run {})", + number, comment_id, ttl, run_id + ); + println!("{}", output_path.display()); + return Ok(()); + } + + Err(TuttiError::IssueClaim(format!( + "could not claim any candidate issue for label '{label}'" + ))) +} + +/// `tt issue-claim heartbeat` +pub fn heartbeat(state_path: &Path) -> Result<()> { + let repo = gh_repo()?; + let data = std::fs::read_to_string(state_path) + .map_err(|e| TuttiError::IssueClaim(format!("cannot read state file: {e}")))?; + let output: SelectedIssueOutput = serde_json::from_str(&data)?; + + let comment_id = output.claim_comment_id; + let issue_number = output.issue_number; + + // Fetch current claim from comment + let comments = gh_list_comments(&repo, issue_number)?; + let mut found = false; + for c in &comments { + if c["id"].as_u64() == Some(comment_id) { + if let Some(body) = c["body"].as_str() + && let Some(mut record) = decode_claim_comment(body) + { + if record.status != ClaimStatus::Active { + return Err(TuttiError::IssueClaim( + "claim is no longer active — cannot renew".into(), + )); + } + record.last_heartbeat_at = Utc::now(); + record.push_event("Renewed", None); + let new_body = encode_claim_comment(&record)?; + gh_update_comment(&repo, comment_id, &new_body)?; + found = true; + + // Also update local state + let mut updated_output: serde_json::Value = serde_json::from_str(&data)?; + updated_output["last_heartbeat_at"] = + serde_json::Value::String(record.last_heartbeat_at.to_rfc3339()); + let json = serde_json::to_string_pretty(&updated_output)?; + let tmp_path = state_path.with_extension("json.tmp"); + std::fs::write(&tmp_path, &json)?; + std::fs::rename(&tmp_path, state_path)?; + + eprintln!( + "heartbeat renewed for #{} (expires {})", + issue_number, + record.expires_at().format("%Y-%m-%dT%H:%M:%SZ") + ); + } + break; + } + } + if !found { + return Err(TuttiError::IssueClaim(format!( + "claim comment {comment_id} not found on issue #{issue_number}" + ))); + } + Ok(()) +} + +/// `tt issue-claim release` +pub fn release(state_path: &Path, reason: Option<&str>) -> Result<()> { + let repo = gh_repo()?; + let data = std::fs::read_to_string(state_path) + .map_err(|e| TuttiError::IssueClaim(format!("cannot read state file: {e}")))?; + let output: SelectedIssueOutput = serde_json::from_str(&data)?; + + let comment_id = output.claim_comment_id; + let issue_number = output.issue_number; + let release_reason = reason.unwrap_or("workflow completed"); + + // Fetch and update claim comment + let comments = gh_list_comments(&repo, issue_number)?; + let mut released = false; + for c in &comments { + if c["id"].as_u64() == Some(comment_id) { + if let Some(body) = c["body"].as_str() + && let Some(mut record) = decode_claim_comment(body) + { + if record.status == ClaimStatus::Released { + eprintln!("claim already released for #{}", issue_number); + released = true; + } else { + record.status = ClaimStatus::Released; + record.released_at = Some(Utc::now()); + record.release_reason = Some(release_reason.to_string()); + record.push_event("Released", Some(release_reason.to_string())); + let new_body = encode_claim_comment(&record)?; + gh_update_comment(&repo, comment_id, &new_body)?; + released = true; + } + } + break; + } + } + + if !released { + eprintln!("warning: claim comment {} not found", comment_id); + } + + // Check if there are remaining active claims; if not, remove the label + let remaining_claims = find_claim_comments(&repo, issue_number)?; + if winner_active_claim(&remaining_claims).is_none() { + gh_remove_label(&repo, issue_number, "automation-claimed")?; + eprintln!("removed automation-claimed label from #{}", issue_number); + } + + // Update local state + let mut updated: serde_json::Value = serde_json::from_str(&data)?; + updated["claim_status"] = serde_json::Value::String("released".into()); + let json = serde_json::to_string_pretty(&updated)?; + std::fs::write(state_path, json)?; + + eprintln!( + "released claim on #{} (reason: {})", + issue_number, release_reason + ); + Ok(()) +} + +/// `tt issue-claim sweep` +pub fn sweep() -> Result<()> { + let repo = gh_repo()?; + let issues = gh_issues_with_label(&repo, "automation-claimed")?; + + if issues.is_empty() { + eprintln!("no issues with automation-claimed label"); + return Ok(()); + } + + let mut total_released = 0u32; + let mut labels_removed = 0u32; + + for number in issues { + let released = release_stale_claims(&repo, number)?; + total_released += released; + + // If no active claims remain, remove the label + let claims = find_claim_comments(&repo, number)?; + if winner_active_claim(&claims).is_none() { + gh_remove_label(&repo, number, "automation-claimed")?; + labels_removed += 1; + eprintln!("removed automation-claimed label from #{}", number); + } + } + + eprintln!( + "sweep complete: {} stale claims released, {} labels removed", + total_released, labels_removed + ); + Ok(()) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_claim_record_new() { + let record = ClaimRecord::new("run-123", 1800); + assert_eq!(record.run_id, "run-123"); + assert_eq!(record.lease_ttl_secs, 1800); + assert_eq!(record.status, ClaimStatus::Active); + assert!(record.released_at.is_none()); + assert_eq!(record.events.len(), 1); + assert_eq!(record.events[0].kind, "Acquired"); + } + + #[test] + fn test_claim_record_expiry() { + let mut record = ClaimRecord::new("run-456", 0); // 0s TTL = immediately expired + // Need to set last_heartbeat_at to the past + record.last_heartbeat_at = Utc::now() - chrono::Duration::seconds(1); + assert!(record.is_expired()); + } + + #[test] + fn test_claim_record_not_expired() { + let record = ClaimRecord::new("run-789", 3600); + assert!(!record.is_expired()); + } + + #[test] + fn test_encode_decode_roundtrip() { + let record = ClaimRecord::new("run-abc", 1800); + let encoded = encode_claim_comment(&record).expect("encode should succeed"); + let decoded = decode_claim_comment(&encoded).expect("should decode"); + assert_eq!(decoded.run_id, "run-abc"); + assert_eq!(decoded.lease_ttl_secs, 1800); + assert_eq!(decoded.status, ClaimStatus::Active); + } + + #[test] + fn test_decode_no_marker() { + assert!(decode_claim_comment("just a regular comment").is_none()); + } + + #[test] + fn test_push_event_caps_at_max() { + let mut record = ClaimRecord::new("run-xyz", 1800); + for i in 0..25 { + record.push_event("Renewed", Some(format!("tick {i}"))); + } + assert_eq!(record.events.len(), MAX_EVENTS); + } + + #[test] + fn test_winner_active_claim_empty() { + let claims: Vec<(u64, ClaimRecord)> = vec![]; + assert!(winner_active_claim(&claims).is_none()); + } + + #[test] + fn test_winner_active_claim_picks_earliest() { + let mut r1 = ClaimRecord::new("run-a", 3600); + r1.claimed_at = Utc::now() - chrono::Duration::seconds(100); + r1.last_heartbeat_at = Utc::now(); + let mut r2 = ClaimRecord::new("run-b", 3600); + r2.claimed_at = Utc::now() - chrono::Duration::seconds(50); + r2.last_heartbeat_at = Utc::now(); + + let claims = vec![(10, r1), (20, r2)]; + let (winner_id, winner) = winner_active_claim(&claims).unwrap(); + assert_eq!(winner_id, 10); + assert_eq!(winner.run_id, "run-a"); + } + + #[test] + fn test_winner_skips_released() { + let mut r1 = ClaimRecord::new("run-released", 3600); + r1.status = ClaimStatus::Released; + let r2 = ClaimRecord::new("run-active", 3600); + + let claims = vec![(10, r1), (20, r2)]; + let (winner_id, winner) = winner_active_claim(&claims).unwrap(); + assert_eq!(winner_id, 20); + assert_eq!(winner.run_id, "run-active"); + } + + #[test] + fn test_winner_skips_expired() { + let mut r1 = ClaimRecord::new("run-expired", 0); + r1.last_heartbeat_at = Utc::now() - chrono::Duration::seconds(10); + let r2 = ClaimRecord::new("run-fresh", 3600); + + let claims = vec![(10, r1), (20, r2)]; + let (winner_id, winner) = winner_active_claim(&claims).unwrap(); + assert_eq!(winner_id, 20); + assert_eq!(winner.run_id, "run-fresh"); + } + + #[test] + fn test_selected_issue_output_serializes() { + let output = SelectedIssueOutput { + issue_number: 29, + title: "test issue".into(), + url: "https://example.com".into(), + labels: vec!["bug".into()], + author: Some("user".into()), + created_at: "2026-03-15T00:00:00Z".into(), + body: "issue body".into(), + run_id: "run-1".into(), + claimed_at: Utc::now(), + lease_ttl_secs: 1800, + last_heartbeat_at: Utc::now(), + claim_comment_id: 12345, + claim_status: ClaimStatus::Active, + }; + let json = serde_json::to_string(&output).unwrap(); + assert!(json.contains("claim_comment_id")); + assert!(json.contains("run_id")); + } +} diff --git a/src/cli/mod.rs b/src/cli/mod.rs index eb97266..a7a2065 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -9,6 +9,7 @@ pub mod down; pub mod handoff; pub mod health; pub mod init; +pub mod issue_claim; pub mod land; pub mod logs; pub mod peek; @@ -360,6 +361,12 @@ pub enum Commands { #[command(subcommand)] command: Option, }, + + /// Manage issue claim leases for automation + IssueClaim { + #[command(subcommand)] + command: IssueClaimSubcommand, + }, } #[derive(Subcommand)] @@ -405,6 +412,42 @@ pub enum PermissionsSubcommand { }, } +#[derive(Subcommand)] +pub enum IssueClaimSubcommand { + /// Acquire an exclusive lease on the next unclaimed issue + Acquire { + /// Output JSON path for selected issue state + #[arg(long, default_value = ".tutti/state/auto/selected_issue.json")] + output: std::path::PathBuf, + + /// GitHub issue label to filter + #[arg(long, default_value = "agent-ops")] + label: String, + + /// Lease time-to-live in seconds (default: 1800 = 30 min) + #[arg(long)] + lease_ttl_secs: Option, + }, + /// Renew an active lease (call periodically during workflow) + Heartbeat { + /// Path to selected issue state JSON + #[arg(long, default_value = ".tutti/state/auto/selected_issue.json")] + state: std::path::PathBuf, + }, + /// Release a claim when workflow completes or fails + Release { + /// Path to selected issue state JSON + #[arg(long, default_value = ".tutti/state/auto/selected_issue.json")] + state: std::path::PathBuf, + + /// Reason for releasing + #[arg(long)] + reason: Option, + }, + /// Sweep and release all stale (expired) claims + Sweep, +} + #[derive(Subcommand)] pub enum HandoffSubcommand { /// Generate a handoff packet for an agent diff --git a/src/error.rs b/src/error.rs index 001d34d..8c0402b 100644 --- a/src/error.rs +++ b/src/error.rs @@ -49,6 +49,9 @@ pub enum TuttiError { #[error("{0}")] Json(#[from] serde_json::Error), + + #[error("issue claim error: {0}")] + IssueClaim(String), } pub type Result = std::result::Result; diff --git a/src/main.rs b/src/main.rs index 5552025..1335e16 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,7 +13,7 @@ mod usage; mod worktree; use clap::Parser; -use cli::{Cli, Commands, WorkspacesSubcommand}; +use cli::{Cli, Commands, IssueClaimSubcommand, WorkspacesSubcommand}; use std::process; fn main() { @@ -144,6 +144,19 @@ fn main() { Some(WorkspacesSubcommand::Status) => cli::workspaces::status(), None => cli::workspaces::list(), }, + Commands::IssueClaim { command } => match command { + IssueClaimSubcommand::Acquire { + ref output, + ref label, + lease_ttl_secs, + } => cli::issue_claim::acquire(output, label, lease_ttl_secs), + IssueClaimSubcommand::Heartbeat { ref state } => cli::issue_claim::heartbeat(state), + IssueClaimSubcommand::Release { + ref state, + ref reason, + } => cli::issue_claim::release(state, reason.as_deref()), + IssueClaimSubcommand::Sweep => cli::issue_claim::sweep(), + }, }; if let Err(e) = result {