From 7be6643c1168531ad59ff774426a938fb0cdc389 Mon Sep 17 00:00:00 2001 From: Dan Nicolau Date: Thu, 6 Aug 2026 16:11:04 -0400 Subject: [PATCH] feat: authorize GitHub AI engagement --- .donkeyspace/policy.yml | 5 + .env.example | 5 +- README.md | 4 +- crates/donkeyspace-api/src/main.rs | 838 +++++++++++++++--- crates/donkeyspace-core/src/fake_agent.rs | 5 +- .../donkeyspace-core/src/github_workflow.rs | 7 +- crates/donkeyspace-core/src/lib.rs | 5 +- crates/donkeyspace-core/src/policy.rs | 310 ++++++- crates/donkeyspace-db/src/lib.rs | 176 +++- crates/donkeyspace-github/src/lib.rs | 64 +- crates/donkeyspace-worker/src/main.rs | 88 +- crates/donkeyspace-worker/src/plugin_flow.rs | 24 +- docs/architecture.md | 2 +- docs/github-workflow.md | 9 + docs/plugin-interface.md | 5 + docs/policy.example.yml | 13 + docs/policy.md | 46 +- docs/policy.plugin.example.yml | 11 + migrations/0001_init.sql | 39 + 19 files changed, 1453 insertions(+), 203 deletions(-) diff --git a/.donkeyspace/policy.yml b/.donkeyspace/policy.yml index 2628ff1..aa32b45 100644 --- a/.donkeyspace/policy.yml +++ b/.donkeyspace/policy.yml @@ -13,6 +13,11 @@ workflow: allow_labels: - "ai" + engagement: + default: + allow: + - type: token_owner + agents: triage: enabled: true diff --git a/.env.example b/.env.example index a1f7a1c..54fb14f 100644 --- a/.env.example +++ b/.env.example @@ -15,8 +15,9 @@ DONKEYSPACE_API_PROXY_TARGET=http://localhost:8080 # Use the same value in the GitHub webhook configuration. DONKEYSPACE_WEBHOOK_SECRET= -# Required for private repository checkout, branch pushes, PR creation, -# creating missing donkeyspace labels, and applying pending GitHub labels/comments. +# Required by the secure default engagement policy and for private repository +# checkout, branch pushes, PR creation, labels, and comments. Invalid configured +# tokens make the API fail at startup. DONKEYSPACE_GITHUB_TOKEN= # Optional webhook-free GitHub ingestion. Comma-separate owner/repository names. diff --git a/README.md b/README.md index 1ca2a2e..2f411fb 100644 --- a/README.md +++ b/README.md @@ -70,7 +70,9 @@ If your credentials live elsewhere, pass them explicitly: docker compose --env-file /path/to/secrets.env up -d --force-recreate worker ``` -Without the token, GitHub writes remain pending and private-repo checkout fails. +Without the token, GitHub writes remain pending, private-repo checkout fails, +and the secure default engagement policy denies new AI work. The API validates +a configured token by resolving its authenticated user at startup. When the token is configured, the worker ensures every configured workflow, allow, and block label exists in GitHub repositories already seen by donkeyspace webhooks. diff --git a/crates/donkeyspace-api/src/main.rs b/crates/donkeyspace-api/src/main.rs index 5773515..405cae5 100644 --- a/crates/donkeyspace-api/src/main.rs +++ b/crates/donkeyspace-api/src/main.rs @@ -7,22 +7,32 @@ use axum::{ routing::{get, post}, }; use donkeyspace_core::{ - AgentRole, LabelState, PluginManifest, Policy, WorkflowState, normalize_workflow_labels, + AgentRole, EngagementGate, EngagementSelector, LabelState, PluginManifest, Policy, + WorkflowState, normalize_workflow_labels, }; use donkeyspace_db::{ - DbConfig, JobRecord, PgPool, PullRequestInput, RepositoryInput, WorkflowItemInput, - acquire_job_lease, active_job_exists_for_workflow_item, apply_migrations, connect, create_job, - create_retry_job, get_job, get_workflow_item_by_issue_number, get_workflow_item_state, - latest_workflow_job_input, list_job_command_results, list_job_outbound_actions, - list_job_transitions, list_jobs, list_open_managed_pull_requests_for_base, - list_recent_outbound_actions, record_state_transition, record_webhook_delivery, - repair_job_exists_for_pr_base, resume_latest_paused_job, reviewer_job_exists_for_pr_head, - upsert_pull_request, upsert_repository, upsert_workflow_item, + DbConfig, EngagementDecisionInput, JobRecord, PgPool, PullRequestInput, RepositoryInput, + WorkflowItemInput, acquire_job_lease, active_job_exists_for_workflow_item, apply_migrations, + connect, create_job, create_retry_job, get_job, get_workflow_item_by_issue_number, + get_workflow_item_state, github_managed_resource_exists, latest_workflow_job_input, + list_job_command_results, list_job_outbound_actions, list_job_transitions, list_jobs, + list_open_managed_pull_requests_for_base, list_recent_engagement_decisions, + list_recent_outbound_actions, pending_outbound_comment_exists, record_engagement_decision, + record_state_transition, record_webhook_delivery, repair_job_exists_for_pr_base, + resume_latest_paused_job, reviewer_job_exists_for_pr_head, upsert_pull_request, + upsert_repository, upsert_workflow_item, }; use donkeyspace_github::GitHubClient; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; -use std::{env, fs, net::SocketAddr, sync::Arc, time::Duration}; +use std::{ + collections::HashMap, + env, fs, + net::SocketAddr, + sync::Arc, + time::{Duration, Instant}, +}; +use tokio::sync::Mutex; use tower_http::trace::TraceLayer; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; use uuid::Uuid; @@ -32,6 +42,9 @@ struct AppState { webhook_secret: Option, pool: Option, policy: Policy, + github: Option, + github_token_owner: Option, + verification_cache: Arc>>, } #[derive(Debug, Serialize)] @@ -77,10 +90,24 @@ async fn main() -> Result<(), Box> { None }; + let github = env::var("DONKEYSPACE_GITHUB_TOKEN") + .ok() + .filter(|value| !value.trim().is_empty()) + .map(GitHubClient::new) + .transpose()?; + let github_token_owner = if let Some(client) = &github { + Some(client.authenticated_login().await?) + } else { + None + }; + let state = Arc::new(AppState { webhook_secret: env::var("DONKEYSPACE_WEBHOOK_SECRET").ok(), pool, policy, + github, + github_token_owner, + verification_cache: Arc::new(Mutex::new(HashMap::new())), }); start_github_poller(state.clone())?; @@ -89,6 +116,7 @@ async fn main() -> Result<(), Box> { .route("/healthz", get(healthz)) .route("/api/runs", get(api_runs)) .route("/api/outbound-actions", get(api_outbound_actions)) + .route("/api/engagement-decisions", get(api_engagement_decisions)) .route("/api/runs/{id}", get(api_run)) .route("/api/runs/{id}/transitions", get(api_run_transitions)) .route("/api/runs/{id}/lease", post(api_lease_run)) @@ -115,11 +143,9 @@ fn start_github_poller(state: Arc) -> Result<(), Box) -> Result<(), Box ( "issues", @@ -260,6 +291,7 @@ fn github_poll_event_to_ingress( "repository": repository, "issue": source.get("issue")?, "label": source.get("label").cloned().unwrap_or(Value::Null), + "sender": sender, }), ), "IssueCommentEvent" => ( @@ -269,6 +301,7 @@ fn github_poll_event_to_ingress( "repository": repository, "issue": source.get("issue")?, "comment": source.get("comment")?, + "sender": sender, }), ), "PullRequestEvent" => ( @@ -415,6 +448,28 @@ async fn api_outbound_actions(State(state): State>) -> impl IntoRe } } +async fn api_engagement_decisions(State(state): State>) -> impl IntoResponse { + let Some(pool) = &state.pool else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + Json(ApiError::new("database is not configured")), + ) + .into_response(); + }; + + match list_recent_engagement_decisions(pool, 100).await { + Ok(decisions) => Json(decisions).into_response(), + Err(error) => { + tracing::error!(%error, "failed to list engagement decisions"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ApiError::new("failed to list engagement decisions")), + ) + .into_response() + } + } +} + async fn api_run_transitions( State(state): State>, Path(id): Path, @@ -572,7 +627,7 @@ async fn github_webhook( return StatusCode::ACCEPTED; }; - match persist_github_webhook(pool, &state.policy, event, delivery, &body).await { + match persist_github_webhook(pool, &state, event, delivery, &body).await { Ok(WebhookPersistOutcome::Ignored) => StatusCode::ACCEPTED, Ok(WebhookPersistOutcome::Duplicate) => StatusCode::OK, Ok(WebhookPersistOutcome::Queued(job)) => { @@ -624,21 +679,23 @@ fn lifecycle_start_role(policy: &Policy) -> Result, Box Result> { match event { "issues" | "issue_comment" => { - persist_issue_webhook(pool, policy, event, delivery, body).await + persist_issue_webhook(pool, state, event, delivery, body).await + } + "pull_request" => { + persist_pull_request_webhook(pool, &state.policy, event, delivery, body).await } - "pull_request" => persist_pull_request_webhook(pool, policy, event, delivery, body).await, - "push" => persist_push_webhook(pool, policy, event, delivery, body).await, + "push" => persist_push_webhook(pool, &state.policy, event, delivery, body).await, _ => { let payload: Value = serde_json::from_slice(body)?; let inserted = record_webhook_delivery(pool, None, delivery, event, &payload).await?; - Ok(if inserted { + Ok(if inserted.is_some() { WebhookPersistOutcome::Ignored } else { WebhookPersistOutcome::Duplicate @@ -649,35 +706,36 @@ async fn persist_github_webhook( async fn persist_issue_webhook( pool: &PgPool, - policy: &Policy, + app_state: &AppState, event: &str, delivery: &str, body: &[u8], ) -> Result> { + let policy = &app_state.policy; let payload: GitHubIssueWebhook = serde_json::from_slice(body)?; - let payload_value: Value = serde_json::from_slice(body)?; + let mut payload_value: Value = serde_json::from_slice(body)?; let repository_id = upsert_repository( pool, &RepositoryInput { provider: "github".to_string(), - owner: payload.repository.owner.login, - name: payload.repository.name, - default_branch: payload.repository.default_branch, + owner: payload.repository.owner.login.clone(), + name: payload.repository.name.clone(), + default_branch: payload.repository.default_branch.clone(), }, ) .await?; let inserted = record_webhook_delivery(pool, Some(repository_id), delivery, event, &payload_value).await?; - if !inserted { + let Some(webhook_delivery_id) = inserted else { return Ok(WebhookPersistOutcome::Duplicate); - } + }; let labels = payload .issue .labels - .into_iter() - .map(|label| label.name) + .iter() + .map(|label| label.name.clone()) .collect::>(); let label_state = normalize_workflow_labels(&labels, &policy.workflow.state_labels); let label_state_name = match &label_state { @@ -725,26 +783,6 @@ async fn persist_issue_webhook( return Ok(WebhookPersistOutcome::Ignored); } - if is_projected_work_item(&payload.issue.body) { - tracing::info!( - issue_number = payload.issue.number, - "projected plugin work-item issue did not queue an independent lifecycle" - ); - return Ok(WebhookPersistOutcome::Ignored); - } - - let automation_decision = policy.automation_decision_for_labels(&labels); - if !automation_decision.is_allowed() { - tracing::info!( - event, - action = payload.action, - issue_number = payload.issue.number, - reason = automation_decision.reason(), - "policy did not allow agent work" - ); - return Ok(WebhookPersistOutcome::Ignored); - } - if !should_queue_triage( event, &payload.action, @@ -764,6 +802,125 @@ async fn persist_issue_webhook( return Ok(WebhookPersistOutcome::Ignored); } + let gate = engagement_gate(event, current_state.as_deref()) + .ok_or("queueable github event has no engagement gate")?; + let managed_resource = if let Some(comment) = &payload.comment { + let registered = match comment.id { + Some(comment_id) => { + github_managed_resource_exists( + pool, + repository_id, + "issue_comment", + &comment_id.to_string(), + ) + .await? + } + None => false, + }; + let pending = match payload_value + .pointer("/comment/body") + .and_then(Value::as_str) + { + Some(body) => pending_outbound_comment_exists(pool, workflow_item_id, body).await?, + None => false, + }; + registered || pending + } else { + is_projected_work_item(&payload.issue.body) + || github_managed_resource_exists( + pool, + repository_id, + "issue", + &payload.issue.id.to_string(), + ) + .await? + }; + if managed_resource { + record_engagement_decision( + pool, + &EngagementDecisionInput { + webhook_delivery_id, + workflow_item_id: Some(workflow_item_id), + gate: gate.as_str().into(), + disposition: "system_generated".into(), + actor: payload + .sender + .as_ref() + .and_then(|actor| serde_json::to_value(actor).ok()), + matched_selector: None, + reason: "donkeyspace-managed github resource cannot trigger agent work".into(), + }, + ) + .await?; + return Ok(WebhookPersistOutcome::Ignored); + } + + let automation_decision = policy.automation_decision_for_labels(&labels); + if !automation_decision.is_allowed() { + record_engagement_decision( + pool, + &EngagementDecisionInput { + webhook_delivery_id, + workflow_item_id: Some(workflow_item_id), + gate: gate.as_str().into(), + disposition: "denied".into(), + actor: payload + .sender + .as_ref() + .and_then(|actor| serde_json::to_value(actor).ok()), + matched_selector: None, + reason: automation_decision.reason(), + }, + ) + .await?; + return Ok(WebhookPersistOutcome::Ignored); + } + + let authorization = authorize_engagement(app_state, gate, &labels, &payload).await; + let audit = record_engagement_decision( + pool, + &EngagementDecisionInput { + webhook_delivery_id, + workflow_item_id: Some(workflow_item_id), + gate: gate.as_str().into(), + disposition: if authorization.allowed { + "allowed" + } else { + "denied" + } + .into(), + actor: payload + .sender + .as_ref() + .and_then(|actor| serde_json::to_value(actor).ok()), + matched_selector: authorization.matched_selector.clone(), + reason: authorization.reason.clone(), + }, + ) + .await?; + if !authorization.allowed { + tracing::info!( + event, + action = payload.action, + issue_number = payload.issue.number, + reason = authorization.reason, + "engagement authorization denied agent work" + ); + return Ok(WebhookPersistOutcome::Ignored); + } + if let Value::Object(map) = &mut payload_value { + map.insert( + "donkeyspace_engagement".into(), + json!({ + "decision_id": audit.id, + "gate": gate.as_str(), + "actor": payload.sender, + "matched_selector": authorization.matched_selector, + "reason": authorization.reason, + }), + ); + } + if active_job_exists_for_workflow_item(pool, workflow_item_id).await? { tracing::info!( event, @@ -774,13 +931,7 @@ async fn persist_issue_webhook( return Ok(WebhookPersistOutcome::Ignored); } - if event == "issue_comment" - && current_state.as_deref() == Some(WorkflowState::NeedsHuman.as_str()) - && payload - .comment - .as_ref() - .map(is_human_comment) - .unwrap_or(false) + if gate == EngagementGate::NeedsHumanResume && let Some(job) = resume_latest_paused_job(pool, workflow_item_id, &payload_value).await? { record_state_transition( @@ -789,7 +940,10 @@ async fn persist_issue_webhook( Some(job.id), current_state.as_deref(), "lifecycle_resumed", - "resumed paused plugin lifecycle from human comment", + &format!( + "resumed paused plugin lifecycle from authorized github event; engagement decision {}", + audit.id + ), ) .await?; return Ok(WebhookPersistOutcome::Queued(job)); @@ -804,7 +958,10 @@ async fn persist_issue_webhook( Some(job.id), current_state.as_deref(), &format!("{initial_role}_queued"), - &format!("queued {initial_role} job from github webhook"), + &format!( + "queued {initial_role} job from github webhook; engagement decision {}", + audit.id + ), ) .await?; @@ -837,7 +994,7 @@ async fn persist_pull_request_webhook( let inserted = record_webhook_delivery(pool, Some(repository_id), delivery, event, &payload_value).await?; - if !inserted { + if inserted.is_none() { return Ok(WebhookPersistOutcome::Duplicate); } @@ -968,7 +1125,7 @@ async fn persist_push_webhook( let inserted = record_webhook_delivery(pool, Some(repository_id), delivery, event, &payload_value).await?; - if !inserted { + if inserted.is_none() { return Ok(WebhookPersistOutcome::Duplicate); } if policy.lifecycle.plugin.is_some() { @@ -1084,12 +1241,282 @@ fn should_queue_triage( state, "needs_info" | "needs_human" | "blocked" ) - ) && comment.map(is_human_comment).unwrap_or(false) + ) && comment.is_some() } _ => false, } } +fn engagement_gate(event: &str, current_state: Option<&str>) -> Option { + match current_state { + Some("needs_info") => Some(EngagementGate::NeedsInfoResume), + Some("blocked") => Some(EngagementGate::BlockedResume), + Some("needs_human") => Some(EngagementGate::NeedsHumanResume), + _ if event == "issues" => Some(EngagementGate::Initial), + _ => None, + } +} + +#[derive(Debug)] +struct AuthorizationDecision { + allowed: bool, + reason: String, + matched_selector: Option, +} + +async fn authorize_engagement( + state: &AppState, + gate: EngagementGate, + labels: &[String], + payload: &GitHubIssueWebhook, +) -> AuthorizationDecision { + let Some(actor) = payload.sender.as_ref() else { + return AuthorizationDecision { + allowed: false, + reason: "github event is missing sender identity".into(), + matched_selector: None, + }; + }; + if actor.login.trim().is_empty() { + return AuthorizationDecision { + allowed: false, + reason: "github event sender identity has no login".into(), + matched_selector: None, + }; + } + if payload + .comment + .as_ref() + .is_some_and(|comment| comment.id.is_none()) + { + return AuthorizationDecision { + allowed: false, + reason: "github comment event is missing comment identity".into(), + matched_selector: None, + }; + } + let rule = state.policy.workflow.engagement.rule(gate); + let missing_labels = rule + .required_labels + .iter() + .filter(|required| !labels.iter().any(|label| label == *required)) + .cloned() + .collect::>(); + if !missing_labels.is_empty() { + return AuthorizationDecision { + allowed: false, + reason: format!( + "missing required engagement labels: {}", + missing_labels.join(", ") + ), + matched_selector: None, + }; + } + + let (content_actor, content_association) = match payload.comment.as_ref() { + Some(comment) => (comment.user.as_ref(), comment.author_association.as_deref()), + None => ( + payload.issue.user.as_ref(), + payload.issue.author_association.as_deref(), + ), + }; + let author_association = if content_actor + .is_some_and(|content_actor| actor.login.eq_ignore_ascii_case(&content_actor.login)) + { + content_association + } else { + None + }; + let performed_app = payload + .comment + .as_ref() + .and_then(|comment| comment.performed_via_github_app.as_ref()) + .or(payload.issue.performed_via_github_app.as_ref()); + let mut failures = Vec::new(); + + for selector in &rule.allow { + let result: Result = match selector { + EngagementSelector::TokenOwner => Ok(state + .github_token_owner + .as_ref() + .map(|login| login.eq_ignore_ascii_case(&actor.login)) + .unwrap_or(false)), + EngagementSelector::AnyUser => Ok(actor.kind.as_deref() == Some("User")), + EngagementSelector::User { login } => Ok( + actor.kind.as_deref() == Some("User") && actor.login.eq_ignore_ascii_case(login) + ), + EngagementSelector::IssueAuthor => Ok(payload + .issue + .user + .as_ref() + .is_some_and(|author| actor.login.eq_ignore_ascii_case(&author.login))), + EngagementSelector::RepositoryOwner => Ok(payload.repository.owner.kind.as_deref() + != Some("Organization") + && actor + .login + .eq_ignore_ascii_case(&payload.repository.owner.login)), + EngagementSelector::RepositoryOrganizationMember => { + if payload.repository.owner.kind.as_deref() != Some("Organization") { + Ok(false) + } else { + verify_organization_member(state, &payload.repository.owner.login, &actor.login) + .await + } + } + EngagementSelector::OrganizationMember { organization } => { + verify_organization_member(state, organization, &actor.login).await + } + EngagementSelector::TeamMember { + organization, + team_slug, + } => verify_team_member(state, organization, team_slug, &actor.login).await, + EngagementSelector::AuthorAssociation { association } => { + Ok(author_association == Some(association.as_str())) + } + EngagementSelector::CollaboratorPermission { minimum } => { + verify_collaborator_permission( + state, + &payload.repository.owner.login, + &payload.repository.name, + &actor.login, + ) + .await + .map(|actual| permission_rank(&actual) >= permission_rank(minimum)) + } + EngagementSelector::Bot { login } => { + Ok(actor.kind.as_deref() == Some("Bot") && actor.login.eq_ignore_ascii_case(login)) + } + EngagementSelector::GitHubApp { id, slug } => Ok(performed_app + .map(|app| { + id.map(|expected| app.id == expected).unwrap_or(false) + || slug + .as_ref() + .map(|expected| app.slug.eq_ignore_ascii_case(expected)) + .unwrap_or(false) + }) + .unwrap_or(false)), + }; + + match result { + Ok(true) => { + return AuthorizationDecision { + allowed: true, + reason: format!("actor matched engagement selector `{selector:?}`"), + matched_selector: serde_json::to_value(selector).ok(), + }; + } + Ok(false) => failures.push(format!("`{selector:?}` did not match")), + Err(error) => failures.push(format!("`{selector:?}` could not be verified: {error}")), + } + } + + AuthorizationDecision { + allowed: false, + reason: if failures.is_empty() { + "engagement rule has no allowed identities".into() + } else { + failures.join("; ") + }, + matched_selector: None, + } +} + +async fn verify_organization_member( + state: &AppState, + organization: &str, + actor: &str, +) -> Result { + let key = format!("org:{organization}:{actor}").to_ascii_lowercase(); + if let Some(value) = verification_cache_get(state, &key).await { + return Ok(value == "true"); + } + let result = match &state.github { + Some(client) => client + .organization_member(organization, actor) + .await + .map_err(|error| error.to_string()), + None => Err("github credentials are unavailable".into()), + }?; + verification_cache_put(state, key, result.to_string()).await; + Ok(result) +} + +async fn verify_team_member( + state: &AppState, + organization: &str, + team_slug: &str, + actor: &str, +) -> Result { + let key = format!("team:{organization}:{team_slug}:{actor}").to_ascii_lowercase(); + if let Some(value) = verification_cache_get(state, &key).await { + return Ok(value == "true"); + } + let result = match &state.github { + Some(client) => client + .team_member(organization, team_slug, actor) + .await + .map_err(|error| error.to_string()), + None => Err("github credentials are unavailable".into()), + }?; + verification_cache_put(state, key, result.to_string()).await; + Ok(result) +} + +async fn verify_collaborator_permission( + state: &AppState, + owner: &str, + repo: &str, + actor: &str, +) -> Result { + let key = format!("permission:{owner}:{repo}:{actor}").to_ascii_lowercase(); + if let Some(value) = verification_cache_get(state, &key).await { + return Ok(value); + } + let result = match &state.github { + Some(client) => client + .collaborator_permission(owner, repo, actor) + .await + .map_err(|error| error.to_string()), + None => Err("github credentials are unavailable".into()), + }?; + verification_cache_put(state, key, result.clone()).await; + Ok(result) +} + +async fn verification_cache_get(state: &AppState, key: &str) -> Option { + let cache = state.verification_cache.lock().await; + cache.get(key).and_then(|(created_at, value)| { + (created_at.elapsed() < Duration::from_secs(300)).then(|| value.clone()) + }) +} + +async fn verification_cache_put(state: &AppState, key: String, value: String) { + let mut cache = state.verification_cache.lock().await; + if cache.len() >= 1_024 { + cache.retain(|_, (created_at, _)| created_at.elapsed() < Duration::from_secs(300)); + if cache.len() >= 1_024 + && let Some(oldest) = cache + .iter() + .min_by_key(|(_, (created_at, _))| *created_at) + .map(|(key, _)| key.clone()) + { + cache.remove(&oldest); + } + } + cache.insert(key, (Instant::now(), value)); +} + +fn permission_rank(permission: &str) -> u8 { + match permission { + "admin" => 5, + "maintain" => 4, + "write" | "push" => 3, + "triage" => 2, + "read" | "pull" => 1, + _ => 0, + } +} + fn should_queue_reviewer( action: &str, pr_state: &str, @@ -1120,14 +1547,6 @@ fn can_retry_job(job: &JobRecord) -> bool { ) } -fn is_human_comment(comment: &GitHubComment) -> bool { - !comment_is_from_donkeyspace(comment) -} - -fn comment_is_from_donkeyspace(comment: &GitHubComment) -> bool { - comment.body.trim_start().starts_with("donkeyspace ") -} - fn pull_request_is_managed(pull_request: &GitHubPullRequest) -> bool { pull_request.head.ref_name.starts_with("donkeyspace/issue-") || pull_request @@ -1181,6 +1600,8 @@ struct GitHubIssueWebhook { comment: Option, #[serde(default)] label: Option, + #[serde(default)] + sender: Option, } #[derive(Debug, Deserialize)] @@ -1208,6 +1629,8 @@ struct GitHubRepository { #[derive(Debug, Deserialize)] struct GitHubOwner { login: String, + #[serde(rename = "type", default)] + kind: Option, } #[derive(Debug, Deserialize)] @@ -1218,6 +1641,12 @@ struct GitHubIssue { #[serde(default)] body: String, labels: Vec, + #[serde(default)] + user: Option, + #[serde(default)] + author_association: Option, + #[serde(default)] + performed_via_github_app: Option, } #[derive(Debug, Deserialize)] @@ -1227,7 +1656,30 @@ struct GitHubLabel { #[derive(Debug, Deserialize)] struct GitHubComment { - body: String, + #[serde(default)] + id: Option, + #[serde(default)] + user: Option, + #[serde(default)] + author_association: Option, + #[serde(default)] + performed_via_github_app: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +struct GitHubActor { + #[serde(default)] + login: String, + #[serde(default)] + id: Option, + #[serde(rename = "type", default)] + kind: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +struct GitHubAppIdentity { + id: u64, + slug: String, } #[derive(Debug, Deserialize)] @@ -1281,15 +1733,206 @@ impl ApiError { #[cfg(test)] mod tests { use super::{ - GitHubComment, JobRecord, can_retry_job, comment_is_from_donkeyspace, - extract_linked_issue_number, github_poll_event_to_ingress, is_projected_work_item, - issue_number_from_donkeyspace_branch, parse_polled_repositories, should_queue_reviewer, - should_queue_triage, + AppState, GitHubActor, GitHubAppIdentity, GitHubComment, GitHubIssue, GitHubIssueWebhook, + GitHubLabel, GitHubOwner, GitHubRepository, JobRecord, authorize_engagement, can_retry_job, + engagement_gate, extract_linked_issue_number, github_poll_event_to_ingress, + is_projected_work_item, issue_number_from_donkeyspace_branch, parse_polled_repositories, + permission_rank, should_queue_reviewer, should_queue_triage, }; use chrono::{DateTime, Utc}; + use donkeyspace_core::{EngagementGate, EngagementSelector, Policy}; use serde_json::json; + use std::{collections::HashMap, sync::Arc}; + use tokio::sync::Mutex; use uuid::Uuid; + fn comment() -> GitHubComment { + GitHubComment { + id: Some(1), + user: Some(GitHubActor { + login: "human".into(), + id: Some(1), + kind: Some("User".into()), + }), + author_association: Some("MEMBER".into()), + performed_via_github_app: None, + } + } + + fn engagement_state(selectors: Vec) -> AppState { + let mut policy = + Policy::from_yaml(include_str!("../../../docs/policy.example.yml")).unwrap(); + policy.workflow.engagement.default.allow = selectors; + policy.workflow.engagement.initial = None; + AppState { + webhook_secret: None, + pool: None, + policy, + github: None, + github_token_owner: Some("maintainer".into()), + verification_cache: Arc::new(Mutex::new(HashMap::new())), + } + } + + fn engagement_payload(sender: GitHubActor) -> GitHubIssueWebhook { + GitHubIssueWebhook { + action: "opened".into(), + repository: GitHubRepository { + name: "repo".into(), + default_branch: "main".into(), + owner: GitHubOwner { + login: "acme".into(), + kind: Some("Organization".into()), + }, + }, + issue: GitHubIssue { + id: 7, + number: 7, + state: "open".into(), + body: "work".into(), + labels: vec![GitHubLabel { name: "ai".into() }], + user: Some(GitHubActor { + login: "author".into(), + id: Some(2), + kind: Some("User".into()), + }), + author_association: Some("NONE".into()), + performed_via_github_app: None, + }, + comment: None, + label: None, + sender: Some(sender), + } + } + + #[tokio::test] + async fn secure_default_allows_only_authenticated_token_owner() { + let state = engagement_state(vec![EngagementSelector::TokenOwner]); + let allowed = authorize_engagement( + &state, + EngagementGate::Initial, + &["ai".into()], + &engagement_payload(GitHubActor { + login: "maintainer".into(), + id: Some(1), + kind: Some("User".into()), + }), + ) + .await; + let denied = authorize_engagement( + &state, + EngagementGate::Initial, + &["ai".into()], + &engagement_payload(GitHubActor { + login: "outsider".into(), + id: Some(3), + kind: Some("User".into()), + }), + ) + .await; + let mut missing_actor = engagement_payload(GitHubActor { + login: "unused".into(), + id: None, + kind: None, + }); + missing_actor.sender = None; + let missing_actor = authorize_engagement( + &state, + EngagementGate::Initial, + &["ai".into()], + &missing_actor, + ) + .await; + + assert!(allowed.allowed); + assert!(!denied.allowed); + assert!(!missing_actor.allowed); + assert!(missing_actor.reason.contains("missing sender")); + } + + #[tokio::test] + async fn github_app_selector_uses_verified_app_metadata() { + let state = engagement_state(vec![EngagementSelector::GitHubApp { + id: None, + slug: Some("trusted-app".into()), + }]); + let mut payload = engagement_payload(GitHubActor { + login: "trusted-app[bot]".into(), + id: Some(4), + kind: Some("Bot".into()), + }); + payload.issue.performed_via_github_app = Some(GitHubAppIdentity { + id: 10, + slug: "trusted-app".into(), + }); + + assert!( + authorize_engagement(&state, EngagementGate::Initial, &["ai".into()], &payload) + .await + .allowed + ); + } + + #[tokio::test] + async fn selectors_requiring_content_authors_fail_closed_when_they_are_missing() { + let issue_author_state = engagement_state(vec![EngagementSelector::IssueAuthor]); + let mut issue_payload = engagement_payload(GitHubActor { + login: "author".into(), + id: Some(2), + kind: Some("User".into()), + }); + issue_payload.issue.user = None; + assert!( + !authorize_engagement( + &issue_author_state, + EngagementGate::Initial, + &["ai".into()], + &issue_payload, + ) + .await + .allowed + ); + + let association_state = engagement_state(vec![EngagementSelector::AuthorAssociation { + association: "OWNER".into(), + }]); + let mut comment_payload = engagement_payload(GitHubActor { + login: "commenter".into(), + id: Some(3), + kind: Some("User".into()), + }); + comment_payload.comment = Some(GitHubComment { + id: Some(9), + user: None, + author_association: Some("OWNER".into()), + performed_via_github_app: None, + }); + assert!( + !authorize_engagement( + &association_state, + EngagementGate::NeedsInfoResume, + &["ai".into()], + &comment_payload, + ) + .await + .allowed + ); + } + + #[test] + fn state_and_permission_ordering_are_explicit() { + assert_eq!( + engagement_gate("issue_comment", Some("needs_human")), + Some(EngagementGate::NeedsHumanResume) + ); + assert_eq!( + engagement_gate("issues", Some("blocked")), + Some(EngagementGate::BlockedResume) + ); + assert!(permission_rank("maintain") > permission_rank("write")); + assert!(permission_rank("triage") > permission_rank("read")); + } + #[test] fn parses_polled_repository_list() { let repositories = parse_polled_repositories("acme/rtl, acme/dv").unwrap(); @@ -1310,6 +1953,7 @@ mod tests { let event = json!({ "id": "12345", "type": "IssuesEvent", + "actor": {"login": "alice", "id": 1, "type": "User"}, "payload": { "action": "opened", "issue": {"id": 7, "number": 3, "state": "open", "labels": []} @@ -1321,6 +1965,7 @@ mod tests { assert_eq!(ingress.delivery_id, "github-poll:acme/rtl:12345"); assert_eq!(ingress.payload["repository"]["default_branch"], "main"); assert_eq!(ingress.payload["issue"]["number"], 3); + assert_eq!(ingress.payload["sender"]["login"], "alice"); } #[test] @@ -1333,6 +1978,7 @@ mod tests { let event = json!({ "id": "12346", "type": "IssuesEvent", + "actor": {"login": "alice", "id": 1, "type": "User"}, "payload": { "action": "labeled", "issue": {"id": 7, "number": 3, "state": "open", "labels": [{"name": "ai"}]}, @@ -1354,6 +2000,7 @@ mod tests { let event = json!({ "id": "67890", "type": "PushEvent", + "actor": {"login": "alice", "id": 1, "type": "User"}, "payload": {"ref": "refs/heads/main", "head": "abc123"} }); @@ -1442,9 +2089,7 @@ mod tests { "created", "open", Some("needs_info"), - Some(&GitHubComment { - body: "Here are the reproduction steps.".to_string(), - }), + Some(&comment()), )); } @@ -1455,9 +2100,7 @@ mod tests { "created", "open", Some("blocked"), - Some(&GitHubComment { - body: "I added the missing detail.".to_string(), - }), + Some(&comment()), )); } @@ -1468,9 +2111,7 @@ mod tests { "created", "open", Some("needs_human"), - Some(&GitHubComment { - body: "N+2 is acceptable and makes sense.".to_string(), - }), + Some(&comment()), )); } @@ -1481,9 +2122,7 @@ mod tests { "edited", "open", Some("blocked"), - Some(&GitHubComment { - body: "Updated with more details.".to_string(), - }), + Some(&comment()), )); } @@ -1494,9 +2133,7 @@ mod tests { "created", "open", Some("ready"), - Some(&GitHubComment { - body: "Looks good.".to_string(), - }), + Some(&comment()), )); } @@ -1512,25 +2149,16 @@ mod tests { } #[test] - fn donkeyspace_comment_does_not_queue_triage() { - assert!(!queue_triage( + fn comment_body_prefix_is_not_used_for_trigger_classification() { + assert!(queue_triage( "issue_comment", "created", "open", Some("blocked"), - Some(&GitHubComment { - body: "donkeyspace triage needs clarification before this issue can move to implementation.".to_string(), - }), + Some(&comment()), )); } - #[test] - fn detects_donkeyspace_generated_comment_after_whitespace() { - assert!(comment_is_from_donkeyspace(&GitHubComment { - body: "\n donkeyspace marked this issue ready for agent implementation.".to_string(), - })); - } - #[test] fn projected_work_item_marker_prevents_recursive_lifecycle() { assert!(is_projected_work_item( diff --git a/crates/donkeyspace-core/src/fake_agent.rs b/crates/donkeyspace-core/src/fake_agent.rs index 9f3785e..1e86570 100644 --- a/crates/donkeyspace-core/src/fake_agent.rs +++ b/crates/donkeyspace-core/src/fake_agent.rs @@ -12,10 +12,9 @@ pub fn fake_triage_issue(input: &Value) -> RunResult { .and_then(Value::as_str) .unwrap_or_default() .trim(); - let latest_human_comment = input + let latest_comment = input .pointer("/comment/body") .and_then(Value::as_str) - .filter(|comment| !comment.trim_start().starts_with("donkeyspace ")) .unwrap_or_default() .trim(); let repository_context = input @@ -38,7 +37,7 @@ pub fn fake_triage_issue(input: &Value) -> RunResult { }; } - if title.is_empty() || meaningful_word_count(&format!("{body}\n{latest_human_comment}")) < 8 { + if title.is_empty() || meaningful_word_count(&format!("{body}\n{latest_comment}")) < 8 { return RunResult { outcome: Outcome::NeedsInfo, summary: "The issue needs clearer acceptance criteria before implementation." diff --git a/crates/donkeyspace-core/src/github_workflow.rs b/crates/donkeyspace-core/src/github_workflow.rs index fb9d31f..f01fd37 100644 --- a/crates/donkeyspace-core/src/github_workflow.rs +++ b/crates/donkeyspace-core/src/github_workflow.rs @@ -81,7 +81,7 @@ pub fn triage_github_issue_actions( } pub fn triage_comment_body(result: &RunResult, target_state: WorkflowState) -> Option { - match result.outcome { + let body = match result.outcome { Outcome::NeedsInfo => { let questions = result .questions @@ -130,7 +130,9 @@ pub fn triage_comment_body(result: &RunResult, target_state: WorkflowState) -> O )) } _ => None, - } + }?; + + Some(format!("{body}\n\n")) } fn workflow_label_text(state: WorkflowState) -> &'static str { @@ -268,6 +270,7 @@ mod tests { assert!(body.contains("- How should this be verified?")); assert!(body.contains("Current state: `ai:needs-info`")); + assert!(body.contains("")); } #[test] diff --git a/crates/donkeyspace-core/src/lib.rs b/crates/donkeyspace-core/src/lib.rs index 71a2d6b..746cf07 100644 --- a/crates/donkeyspace-core/src/lib.rs +++ b/crates/donkeyspace-core/src/lib.rs @@ -14,8 +14,9 @@ pub use plugin::{ PluginValidator, PluginWorkItem, PluginWorkItemRegistry, }; pub use policy::{ - AgentConfig, AgentRoleConfig, AutomationDecision, LifecyclePolicy, PluginFlowSelection, Policy, - PolicyError, StageAccessOverride, TaskAccessOverride, + AgentConfig, AgentRoleConfig, AutomationDecision, EngagementGate, EngagementPolicy, + EngagementRule, EngagementSelector, LifecyclePolicy, PluginFlowSelection, Policy, PolicyError, + StageAccessOverride, TaskAccessOverride, }; pub use run_result::{ AgentHandoff, Confidence, Outcome, PluginStageResult, PluginTaskResult, Risk, RunResult, diff --git a/crates/donkeyspace-core/src/policy.rs b/crates/donkeyspace-core/src/policy.rs index dbe712d..79e2c83 100644 --- a/crates/donkeyspace-core/src/policy.rs +++ b/crates/donkeyspace-core/src/policy.rs @@ -58,6 +58,7 @@ impl Policy { "lifecycle plugin flow cannot be empty".to_string(), )); } + policy.workflow.engagement.validate()?; Ok(policy) } @@ -70,6 +71,197 @@ impl Policy { } } +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +pub struct EngagementPolicy { + #[serde(default = "EngagementRule::token_owner")] + pub default: EngagementRule, + #[serde(default)] + pub initial: Option, + #[serde(default)] + pub needs_info_resume: Option, + #[serde(default)] + pub blocked_resume: Option, + #[serde(default)] + pub needs_human_resume: Option, +} + +impl Default for EngagementPolicy { + fn default() -> Self { + Self { + default: EngagementRule::token_owner(), + initial: None, + needs_info_resume: None, + blocked_resume: None, + needs_human_resume: None, + } + } +} + +impl EngagementPolicy { + pub fn rule(&self, gate: EngagementGate) -> &EngagementRule { + match gate { + EngagementGate::Initial => self.initial.as_ref(), + EngagementGate::NeedsInfoResume => self.needs_info_resume.as_ref(), + EngagementGate::BlockedResume => self.blocked_resume.as_ref(), + EngagementGate::NeedsHumanResume => self.needs_human_resume.as_ref(), + } + .unwrap_or(&self.default) + } + + fn validate(&self) -> Result<(), PolicyError> { + for rule in [ + Some(&self.default), + self.initial.as_ref(), + self.needs_info_resume.as_ref(), + self.blocked_resume.as_ref(), + self.needs_human_resume.as_ref(), + ] + .into_iter() + .flatten() + { + for selector in &rule.allow { + selector.validate()?; + } + } + Ok(()) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EngagementGate { + Initial, + NeedsInfoResume, + BlockedResume, + NeedsHumanResume, +} + +impl EngagementGate { + pub fn as_str(self) -> &'static str { + match self { + Self::Initial => "initial", + Self::NeedsInfoResume => "needs_info_resume", + Self::BlockedResume => "blocked_resume", + Self::NeedsHumanResume => "needs_human_resume", + } + } +} + +#[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq, Eq)] +pub struct EngagementRule { + #[serde(default)] + pub required_labels: Vec, + #[serde(default)] + pub allow: Vec, +} + +impl EngagementRule { + fn token_owner() -> Self { + Self { + required_labels: Vec::new(), + allow: vec![EngagementSelector::TokenOwner], + } + } +} + +#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)] +#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)] +pub enum EngagementSelector { + TokenOwner, + AnyUser, + User { + login: String, + }, + IssueAuthor, + RepositoryOwner, + RepositoryOrganizationMember, + OrganizationMember { + organization: String, + }, + TeamMember { + organization: String, + team_slug: String, + }, + AuthorAssociation { + association: String, + }, + CollaboratorPermission { + minimum: String, + }, + Bot { + login: String, + }, + #[serde(rename = "github_app")] + GitHubApp { + id: Option, + slug: Option, + }, +} + +impl EngagementSelector { + fn validate(&self) -> Result<(), PolicyError> { + let nonempty = |name: &str, value: &str| { + if value.trim().is_empty() { + Err(PolicyError::Invalid(format!( + "engagement selector `{name}` cannot be empty" + ))) + } else { + Ok(()) + } + }; + match self { + Self::User { login } | Self::Bot { login } => nonempty("login", login), + Self::OrganizationMember { organization } => nonempty("organization", organization), + Self::TeamMember { + organization, + team_slug, + } => { + nonempty("organization", organization)?; + nonempty("team_slug", team_slug) + } + Self::AuthorAssociation { association } => { + const ASSOCIATIONS: &[&str] = &[ + "COLLABORATOR", + "CONTRIBUTOR", + "FIRST_TIMER", + "FIRST_TIME_CONTRIBUTOR", + "MANNEQUIN", + "MEMBER", + "NONE", + "OWNER", + ]; + if ASSOCIATIONS.contains(&association.as_str()) { + Ok(()) + } else { + Err(PolicyError::Invalid(format!( + "unknown GitHub author association `{association}`" + ))) + } + } + Self::CollaboratorPermission { minimum } => { + if ["read", "triage", "write", "maintain", "admin"].contains(&minimum.as_str()) { + Ok(()) + } else { + Err(PolicyError::Invalid(format!( + "unknown GitHub collaborator permission `{minimum}`" + ))) + } + } + Self::GitHubApp { id, slug } => match (id, slug) { + (Some(_), None) => Ok(()), + (None, Some(slug)) => nonempty("slug", slug), + _ => Err(PolicyError::Invalid( + "github_app selector requires exactly one of `id` or `slug`".into(), + )), + }, + Self::TokenOwner + | Self::AnyUser + | Self::IssueAuthor + | Self::RepositoryOwner + | Self::RepositoryOrganizationMember => Ok(()), + } + } +} + #[derive(Debug, Clone, Default, Deserialize, Serialize, PartialEq, Eq)] pub struct LifecyclePolicy { /// An optional plugin flow that replaces the built-in @@ -85,6 +277,8 @@ pub struct WorkflowPolicy { pub block_labels: Vec, #[serde(default)] pub allow_labels: Vec, + #[serde(default)] + pub engagement: EngagementPolicy, } impl WorkflowPolicy { @@ -217,17 +411,16 @@ impl RiskPolicy { Some("policy routes high-risk work to human review".to_string()) } else if self.route_unknown_to_human && result.risk == Risk::Unknown { Some("policy routes unknown-risk work to human review".to_string()) - } else if let Some(pattern) = self.human_review_paths.iter().find(|pattern| { - result - .changed_files - .iter() - .any(|path| path_matches(pattern, path)) - }) { - Some(format!( - "policy requires human review for `{pattern}` changes" - )) } else { - None + self.human_review_paths + .iter() + .find(|pattern| { + result + .changed_files + .iter() + .any(|path| path_matches(pattern, path)) + }) + .map(|pattern| format!("policy requires human review for `{pattern}` changes")) }; if let Some(reason) = reason { @@ -284,7 +477,7 @@ pub struct DashboardPolicy { #[cfg(test)] mod tests { - use super::Policy; + use super::{EngagementGate, EngagementRule, EngagementSelector, Policy}; use crate::{Confidence, Outcome, Risk, RunResult}; #[test] @@ -297,6 +490,101 @@ mod tests { assert_eq!(policy.workflow.state_labels["ready"], "ai:ready"); } + #[test] + fn omitted_engagement_defaults_every_gate_to_token_owner() { + let policy = Policy::from_yaml( + r#" +version: 1 +workflow: { state_labels: {} } +checks: {} +risk: + default: unknown + agent_classification: true + route_unknown_to_human: true + route_high_to_human: true +automation: + max_concurrent_jobs: 1 + retry_failed_jobs: false + auto_merge: false +"#, + ) + .unwrap(); + + for gate in [ + EngagementGate::Initial, + EngagementGate::NeedsInfoResume, + EngagementGate::BlockedResume, + EngagementGate::NeedsHumanResume, + ] { + assert_eq!( + policy.workflow.engagement.rule(gate).allow, + vec![EngagementSelector::TokenOwner] + ); + } + } + + #[test] + fn engagement_gate_override_and_full_selectors_parse() { + let mut yaml = include_str!("../../../docs/policy.example.yml").to_string(); + yaml = yaml.replace( + " needs_human_resume:\n", + " blocked_resume:\n allow: []\n needs_human_resume:\n", + ); + let policy = Policy::from_yaml(&yaml).unwrap(); + + assert!( + policy + .workflow + .engagement + .rule(EngagementGate::BlockedResume) + .allow + .is_empty() + ); + assert!(matches!( + policy + .workflow + .engagement + .rule(EngagementGate::NeedsHumanResume) + .allow[1], + EngagementSelector::CollaboratorPermission { .. } + )); + } + + #[test] + fn all_engagement_selector_shapes_parse() { + let rule: EngagementRule = serde_yaml::from_str( + r#" +allow: + - { type: token_owner } + - { type: any_user } + - { type: user, login: alice } + - { type: issue_author } + - { type: repository_owner } + - { type: repository_organization_member } + - { type: organization_member, organization: acme } + - { type: team_member, organization: acme, team_slug: maintainers } + - { type: author_association, association: OWNER } + - { type: collaborator_permission, minimum: maintain } + - { type: bot, login: "dependabot[bot]" } + - { type: github_app, id: 123 } + - { type: github_app, slug: dependabot } +"#, + ) + .unwrap(); + + assert_eq!(rule.allow.len(), 13); + for selector in &rule.allow { + selector.validate().unwrap(); + } + } + + #[test] + fn invalid_engagement_selector_fails_policy_loading() { + let yaml = include_str!("../../../docs/policy.example.yml") + .replace("minimum: \"write\"", "minimum: \"superuser\""); + assert!(Policy::from_yaml(&yaml).is_err()); + } + #[test] fn repair_agent_defaults_disabled_for_older_policy_files() { let policy = Policy::from_yaml( diff --git a/crates/donkeyspace-db/src/lib.rs b/crates/donkeyspace-db/src/lib.rs index 57e13b7..f26caf7 100644 --- a/crates/donkeyspace-db/src/lib.rs +++ b/crates/donkeyspace-db/src/lib.rs @@ -194,10 +194,35 @@ pub struct OutboundActionRecord { pub status: String, pub payload: Value, pub last_error: Option, + pub provider_resource_id: Option, pub created_at: DateTime, pub updated_at: DateTime, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EngagementDecisionInput { + pub webhook_delivery_id: i64, + pub workflow_item_id: Option, + pub gate: String, + pub disposition: String, + pub actor: Option, + pub matched_selector: Option, + pub reason: String, +} + +#[derive(Debug, Clone, FromRow, Serialize, Deserialize)] +pub struct EngagementDecisionRecord { + pub id: i64, + pub webhook_delivery_id: i64, + pub workflow_item_id: Option, + pub gate: String, + pub disposition: String, + pub actor: Option, + pub matched_selector: Option, + pub reason: String, + pub created_at: DateTime, +} + #[derive(Debug, Clone, FromRow, Serialize, Deserialize)] pub struct ManagedPullRequestRecord { pub workflow_item_id: i64, @@ -528,7 +553,7 @@ pub async fn record_webhook_delivery( delivery_id: &str, event_name: &str, payload: &Value, -) -> Result { +) -> Result, DbError> { let inserted = sqlx::query_scalar::<_, i64>( r#" INSERT INTO webhook_deliveries (repository_id, delivery_id, event_name, payload) @@ -544,7 +569,120 @@ pub async fn record_webhook_delivery( .fetch_optional(pool) .await?; - Ok(inserted.is_some()) + Ok(inserted) +} + +pub async fn record_engagement_decision( + pool: &PgPool, + input: &EngagementDecisionInput, +) -> Result { + Ok(sqlx::query_as::<_, EngagementDecisionRecord>( + r#" + INSERT INTO engagement_decisions ( + webhook_delivery_id, workflow_item_id, gate, disposition, + actor, matched_selector, reason + ) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (webhook_delivery_id) DO UPDATE SET + workflow_item_id = EXCLUDED.workflow_item_id, + gate = EXCLUDED.gate, + disposition = EXCLUDED.disposition, + actor = EXCLUDED.actor, + matched_selector = EXCLUDED.matched_selector, + reason = EXCLUDED.reason + RETURNING * + "#, + ) + .bind(input.webhook_delivery_id) + .bind(input.workflow_item_id) + .bind(&input.gate) + .bind(&input.disposition) + .bind(&input.actor) + .bind(&input.matched_selector) + .bind(&input.reason) + .fetch_one(pool) + .await?) +} + +pub async fn list_recent_engagement_decisions( + pool: &PgPool, + limit: i64, +) -> Result, DbError> { + Ok(sqlx::query_as::<_, EngagementDecisionRecord>( + "SELECT * FROM engagement_decisions ORDER BY created_at DESC LIMIT $1", + ) + .bind(limit) + .fetch_all(pool) + .await?) +} + +pub async fn record_github_managed_resource_for_workflow_item( + pool: &PgPool, + workflow_item_id: i64, + resource_kind: &str, + provider_id: &str, + metadata: &Value, +) -> Result<(), DbError> { + sqlx::query( + r#" + INSERT INTO github_managed_resources ( + repository_id, workflow_item_id, resource_kind, provider_id, metadata + ) + SELECT repository_id, id, $2, $3, $4 + FROM workflow_items WHERE id = $1 + ON CONFLICT (repository_id, resource_kind, provider_id) DO NOTHING + "#, + ) + .bind(workflow_item_id) + .bind(resource_kind) + .bind(provider_id) + .bind(metadata) + .execute(pool) + .await?; + Ok(()) +} + +pub async fn github_managed_resource_exists( + pool: &PgPool, + repository_id: i64, + resource_kind: &str, + provider_id: &str, +) -> Result { + Ok(sqlx::query_scalar::<_, bool>( + r#" + SELECT EXISTS ( + SELECT 1 FROM github_managed_resources + WHERE repository_id = $1 AND resource_kind = $2 AND provider_id = $3 + ) + "#, + ) + .bind(repository_id) + .bind(resource_kind) + .bind(provider_id) + .fetch_one(pool) + .await?) +} + +pub async fn pending_outbound_comment_exists( + pool: &PgPool, + workflow_item_id: i64, + body: &str, +) -> Result { + Ok(sqlx::query_scalar::<_, bool>( + r#" + SELECT EXISTS ( + SELECT 1 FROM outbound_actions + WHERE workflow_item_id = $1 + AND action_type = 'issue.create_comment' + AND status = 'pending' + AND payload ->> 'body' = $2 + ) + "#, + ) + .bind(workflow_item_id) + .bind(body) + .fetch_one(pool) + .await?) } pub async fn create_job( @@ -935,20 +1073,50 @@ pub async fn list_pending_outbound_actions( Ok(actions) } -pub async fn mark_outbound_action_completed(pool: &PgPool, id: i64) -> Result<(), DbError> { +pub async fn mark_outbound_action_completed( + pool: &PgPool, + id: i64, + provider_resource_id: Option<&str>, +) -> Result<(), DbError> { + let mut transaction = pool.begin().await?; sqlx::query( r#" UPDATE outbound_actions SET status = 'completed', last_error = NULL, + provider_resource_id = $2, updated_at = now() WHERE id = $1 "#, ) .bind(id) - .execute(pool) + .bind(provider_resource_id) + .execute(&mut *transaction) .await?; + if let Some(provider_id) = provider_resource_id { + sqlx::query( + r#" + INSERT INTO github_managed_resources ( + repository_id, workflow_item_id, outbound_action_id, + resource_kind, provider_id, metadata + ) + SELECT workflow_items.repository_id, outbound_actions.workflow_item_id, + outbound_actions.id, 'issue_comment', $2, '{}'::jsonb + FROM outbound_actions + JOIN workflow_items ON workflow_items.id = outbound_actions.workflow_item_id + WHERE outbound_actions.id = $1 + ON CONFLICT (repository_id, resource_kind, provider_id) DO NOTHING + "#, + ) + .bind(id) + .bind(provider_id) + .execute(&mut *transaction) + .await?; + } + + transaction.commit().await?; + Ok(()) } diff --git a/crates/donkeyspace-github/src/lib.rs b/crates/donkeyspace-github/src/lib.rs index 96d1cf4..8130f89 100644 --- a/crates/donkeyspace-github/src/lib.rs +++ b/crates/donkeyspace-github/src/lib.rs @@ -67,6 +67,12 @@ pub struct GitHubWorkItem { pub depends_on: Vec, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GitHubProjectedIssue { + pub id: i64, + pub number: i64, +} + impl GitHubClient { pub fn new(token: impl Into) -> Result { Ok(Self { @@ -131,12 +137,60 @@ impl GitHubClient { repo: &str, issue_number: i64, body: &str, - ) -> Result<(), GitHubClientError> { - self.client + ) -> Result { + let comment = self + .client .issues(owner, repo) .create_comment(issue_number as u64, body) .await?; - Ok(()) + Ok(comment.id.to_string()) + } + + pub async fn authenticated_login(&self) -> Result { + Ok(self.client.current().user().await?.login) + } + + pub async fn collaborator_permission( + &self, + owner: &str, + repo: &str, + username: &str, + ) -> Result { + let permission = self + .client + .repos(owner, repo) + .get_contributor_permission(username) + .send() + .await?; + Ok(permission.role_name) + } + + pub async fn organization_member( + &self, + organization: &str, + username: &str, + ) -> Result { + Ok(self + .client + .orgs(organization) + .check_membership(username) + .await?) + } + + pub async fn team_member( + &self, + organization: &str, + team_slug: &str, + username: &str, + ) -> Result { + let membership: Value = self + .client + .get( + format!("/orgs/{organization}/teams/{team_slug}/memberships/{username}"), + None::<&()>, + ) + .await?; + Ok(membership.get("state").and_then(Value::as_str) == Some("active")) } pub async fn project_work_items( @@ -145,7 +199,7 @@ impl GitHubClient { repo: &str, parent_issue_number: i64, work_items: &[GitHubWorkItem], - ) -> Result, GitHubClientError> { + ) -> Result, GitHubClientError> { let mut projected = BTreeMap::::new(); for item in work_items { let issue: Value = self @@ -192,7 +246,7 @@ impl GitHubClient { } Ok(projected .into_iter() - .map(|(id, (_, number))| (id, number)) + .map(|(key, (id, number))| (key, GitHubProjectedIssue { id, number })) .collect()) } diff --git a/crates/donkeyspace-worker/src/main.rs b/crates/donkeyspace-worker/src/main.rs index 2927356..2a7882d 100644 --- a/crates/donkeyspace-worker/src/main.rs +++ b/crates/donkeyspace-worker/src/main.rs @@ -450,26 +450,6 @@ async fn execute_job( match running_job.role.as_str() { "triage" => { - if input_is_donkeyspace_comment(&running_job.input) { - let result_value = json!({ - "outcome": "blocked", - "summary": "Ignored donkeyspace-generated comment webhook.", - "confidence": "high", - "risk": "unknown", - "questions": [], - "tests": [], - "changed_files": [], - "human_review_reason": null, - "blocked_reason": "donkeyspace-generated comments do not trigger triage", - }); - complete_job(pool, running_job.id, &result_value).await?; - tracing::info!( - job_id = %running_job.id, - "ignored donkeyspace-generated comment job" - ); - return Ok(()); - } - let repository_context = match build_repository_context( &running_job.input, running_job.id, @@ -1103,7 +1083,7 @@ async fn execute_developer_job( "owner": owner, "repo": repo, "issue_number": issue_number, - "body": format!("donkeyspace implementation lifecycle opened a pull request: {pull_request_url}"), + "body": format!("donkeyspace implementation lifecycle opened a pull request: {pull_request_url}\n\n"), }), }, ) @@ -1270,7 +1250,10 @@ async fn execute_reviewer_job( "owner": repository_owner(&running_job.input)?, "repo": repository_name(&running_job.input)?, "issue_number": pull_request_number(&running_job.input).unwrap_or_else(|| issue_number(&running_job.input).unwrap_or(0)), - "body": reviewer_comment_body(&result, running_job.id, &repository_context), + "body": format!( + "{}\n\n", + reviewer_comment_body(&result, running_job.id, &repository_context) + ), }), }, ) @@ -2556,7 +2539,10 @@ async fn create_repair_comment_action( "owner": repository_owner(&running_job.input)?, "repo": repository_name(&running_job.input)?, "issue_number": pull_request_number(&running_job.input).unwrap_or_else(|| issue_number(&running_job.input).unwrap_or(0)), - "body": repair_comment_body(result, running_job.id), + "body": format!( + "{}\n\n", + repair_comment_body(result, running_job.id) + ), }), }, ) @@ -2944,15 +2930,6 @@ fn latest_comment(input: &Value) -> Option { .map(ToString::to_string) } -fn input_is_donkeyspace_comment(input: &serde_json::Value) -> bool { - input.pointer("/action").and_then(serde_json::Value::as_str) == Some("created") - && input - .pointer("/comment/body") - .and_then(serde_json::Value::as_str) - .map(|body| body.trim_start().starts_with("donkeyspace ")) - .unwrap_or(false) -} - fn input_issue_is_closed(input: &serde_json::Value) -> bool { input .pointer("/issue/state") @@ -2972,8 +2949,9 @@ async fn process_outbound_actions( ) -> Result<(), Box> { for action in list_pending_outbound_actions(pool, 20).await? { match execute_outbound_action(client, &action).await { - Ok(()) => { - mark_outbound_action_completed(pool, action.id).await?; + Ok(provider_resource_id) => { + mark_outbound_action_completed(pool, action.id, provider_resource_id.as_deref()) + .await?; tracing::info!( action_id = action.id, action_type = action.action_type, @@ -2999,8 +2977,8 @@ async fn process_outbound_actions( async fn execute_outbound_action( client: &GitHubClient, action: &OutboundActionRecord, -) -> Result<(), Box> { - match action.action_type.as_str() { +) -> Result, Box> { + let provider_resource_id = match action.action_type.as_str() { "issue.add_label" => { let payload: AddLabelPayload = serde_json::from_value(action.payload.clone())?; client @@ -3011,6 +2989,7 @@ async fn execute_outbound_action( &payload.label, ) .await?; + None } "issue.remove_labels" => { let payload: RemoveLabelsPayload = serde_json::from_value(action.payload.clone())?; @@ -3019,10 +2998,11 @@ async fn execute_outbound_action( .remove_issue_label(&payload.owner, &payload.repo, payload.issue_number, &label) .await?; } + None } "issue.create_comment" => { let payload: CreateCommentPayload = serde_json::from_value(action.payload.clone())?; - client + let comment_id = client .create_issue_comment( &payload.owner, &payload.repo, @@ -3030,13 +3010,14 @@ async fn execute_outbound_action( &payload.body, ) .await?; + Some(comment_id) } unsupported => { return Err(format!("unsupported outbound action type: {unsupported}").into()); } - } + }; - Ok(()) + Ok(provider_resource_id) } #[derive(Debug, Deserialize)] @@ -3066,36 +3047,15 @@ struct CreateCommentPayload { #[cfg(test)] mod tests { use super::{ - agent_run_input, command_summary, conventional_commit_title, input_is_donkeyspace_comment, - input_issue_is_closed, merge_refused_unrelated_histories, non_empty_string, - normalize_reviewer_result, parse_porcelain_status, policy_managed_labels, - required_check_failure_summary, reviewer_changed_files, reviewer_comment_body, - token_usage_exceeded_triage_result, + agent_run_input, command_summary, conventional_commit_title, input_issue_is_closed, + merge_refused_unrelated_histories, non_empty_string, normalize_reviewer_result, + parse_porcelain_status, policy_managed_labels, required_check_failure_summary, + reviewer_changed_files, reviewer_comment_body, token_usage_exceeded_triage_result, }; use donkeyspace_core::{Confidence, Outcome, Policy, Risk, RunResult, TestResult, TestStatus}; use serde_json::json; use uuid::Uuid; - #[test] - fn detects_generated_comment_job() { - assert!(input_is_donkeyspace_comment(&json!({ - "action": "created", - "comment": { - "body": "donkeyspace triage needs clarification before this issue can move to implementation." - } - }))); - } - - #[test] - fn human_comment_job_is_not_generated() { - assert!(!input_is_donkeyspace_comment(&json!({ - "action": "created", - "comment": { - "body": "Here are the reproduction steps." - } - }))); - } - #[test] fn closed_issue_input_is_not_eligible_for_agent_work() { assert!(input_issue_is_closed(&json!({ diff --git a/crates/donkeyspace-worker/src/plugin_flow.rs b/crates/donkeyspace-worker/src/plugin_flow.rs index 6ea1a04..6e845c2 100644 --- a/crates/donkeyspace-worker/src/plugin_flow.rs +++ b/crates/donkeyspace-worker/src/plugin_flow.rs @@ -5,7 +5,8 @@ use donkeyspace_core::{ TestResult, TestStatus, }; use donkeyspace_db::{ - JobRecord, PgPool, complete_job, create_waiting_job, fail_job, start_waiting_job, + JobRecord, PgPool, complete_job, create_waiting_job, fail_job, + record_github_managed_resource_for_workflow_item, start_waiting_job, }; use donkeyspace_github::{GitHubClient, GitHubWorkItem}; use futures::future::join_all; @@ -462,7 +463,26 @@ async fn run_work_item_lifecycle( .project_work_items(owner, repo, parent_issue_number, &work_items) .await { - Ok(issues) => projected_issues = issues, + Ok(issues) => { + if let Some(tracking) = &tracking + && let Some(workflow_item_id) = tracking.coordinator.workflow_item_id + { + for (work_item, issue) in &issues { + record_github_managed_resource_for_workflow_item( + tracking.pool, + workflow_item_id, + "issue", + &issue.id.to_string(), + &json!({"work_item": work_item, "issue_number": issue.number}), + ) + .await?; + } + } + projected_issues = issues + .into_iter() + .map(|(work_item, issue)| (work_item, issue.number)) + .collect(); + } Err(error) => tracing::warn!(%error, "github work-item projection failed"), } } diff --git a/docs/architecture.md b/docs/architecture.md index 93e294c..ca6dbb9 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -109,7 +109,7 @@ Compose-managed `codex-home` volume for local use. - No GitHub App authentication; local operation uses a token. - No token accounting, retention policy, or systematic secret redaction. - The dashboard does not yet show GitHub links, transitions, policy snapshots, - or policy decisions, although some of that data is available from the API. + or engagement decisions, although decision records are available from the API. - The Compose dashboard uses the Vite development server rather than production static assets served by the API. - Test coverage is primarily unit-level; PostgreSQL, live GitHub, and complete diff --git a/docs/github-workflow.md b/docs/github-workflow.md index 9209096..f02f8f8 100644 --- a/docs/github-workflow.md +++ b/docs/github-workflow.md @@ -17,6 +17,10 @@ Only one workflow label should be active on an issue at a time. ## Event Triggers +Before a trigger creates or resumes a job, Donkeyspace evaluates the engagement +rule for that event and workflow state. Denied events remain recorded for audit +but do not reach the built-in lifecycle or a lifecycle plugin. + The current implementation handles these GitHub events: - `issues.opened`: schedule triage unless blocked by policy. @@ -89,6 +93,11 @@ conflict comment. Database job leases prevent two workers from claiming the same queued job. Webhook delivery IDs and PR head/base checks suppress duplicate scheduling. +Donkeyspace-created comment IDs and plugin-projected issue IDs are stored, so +their events are suppressed without trusting a user-controlled comment prefix. +Generated comments also carry a `` marker for +traceability, but the marker is never sufficient to classify an event as +system-generated. ## Comment Formats diff --git a/docs/plugin-interface.md b/docs/plugin-interface.md index 9b450d3..7d317da 100644 --- a/docs/plugin-interface.md +++ b/docs/plugin-interface.md @@ -316,6 +316,11 @@ preserved. A human reply requeues the same coordinator UUID, reuses the checkout and projected block issues, and restarts only the target task and its downstream dependents. Successful parallel siblings remain complete. +The reply must first pass the policy's `needs_human_resume` engagement rule; +denied replies leave the coordinator paused. Projected issue IDs are registered +as Donkeyspace-managed resources so their webhook or polling events cannot +start an independent lifecycle. + ## Filesystem isolation Every attempt receives a separate physical workspace containing only declared diff --git a/docs/policy.example.yml b/docs/policy.example.yml index 6308253..ad67c42 100644 --- a/docs/policy.example.yml +++ b/docs/policy.example.yml @@ -16,6 +16,19 @@ workflow: allow_labels: - "ai" + # Omit engagement to use this same token-owner-only rule by default. + engagement: + default: + allow: + - type: token_owner + needs_human_resume: + required_labels: + - "ai" + allow: + - type: token_owner + - type: collaborator_permission + minimum: "write" + agents: triage: enabled: true diff --git a/docs/policy.md b/docs/policy.md index fb37661..aba94b8 100644 --- a/docs/policy.md +++ b/docs/policy.md @@ -29,6 +29,50 @@ workflow: Block labels win over allow labels. For example, an issue with both `ai` and `ai:disabled` will not queue triage. +## Engagement Authorization + +`workflow.engagement` controls which GitHub actors may start or resume AI work. Separate +rules cover initial issue events, clarification after `ai:needs-info`, blocked +work, and `ai:needs-human` resumption, including plugin checkpoints. + +When `engagement` is omitted, every gate allows only the user authenticated by +`DONKEYSPACE_GITHUB_TOKEN`. If the token is absent, that default denies all +engagement. A configured but invalid token makes the API fail at startup. + +```yaml +workflow: + engagement: + default: + required_labels: ["ai"] + allow: + - type: token_owner + - type: collaborator_permission + minimum: write + needs_human_resume: + allow: + - type: token_owner + - type: team_member + organization: example + team_slug: maintainers +``` + +An omitted gate inherits `default`. Required labels are all required; identity +selectors are alternatives. An explicit empty `allow` list denies everyone. +Supported selectors are `token_owner`, `any_user`, `user`, `issue_author`, +`repository_owner`, `repository_organization_member`, `organization_member`, +`team_member`, `author_association`, `collaborator_permission`, `bot`, and +`github_app`. GitHub App selectors accept exactly one numeric `id` or `slug`. + +API-backed selectors require adequate PAT repository and organization +visibility. Missing actor metadata, insufficient scope, lookup errors, and +unknown permissions fail closed. Decisions are available from +`GET /api/engagement-decisions`. + +For public repositories, prefer the token owner, explicit maintainers/teams, +or collaborators with at least `write`; `any_user` and `issue_author` allow +arbitrary public issue content to reach the agent. Private repositories may use +`read` collaborator access when repository access is the trust boundary. + ## Agents `agents` controls which roles can run and which command Donkeyspace invokes inside the prepared workspace. @@ -94,7 +138,7 @@ Failed jobs can be retried manually through `POST /api/runs/{id}/retry` or the dashboard when `dashboard.allow_retry` is true. Only failed jobs are eligible; results with `blocked` or `needs_human` outcomes must be resolved by a person instead. For a paused lifecycle plugin, a human comment on the parent issue -resumes the saved coordinator checkpoint. A retry creates a new job linked +resumes the checkpoint only after `needs_human_resume` authorizes the actor. A retry creates a new job linked through `retry_of_job_id`. ## Dashboard diff --git a/docs/policy.plugin.example.yml b/docs/policy.plugin.example.yml index 88f4ea6..0ef7bc9 100644 --- a/docs/policy.plugin.example.yml +++ b/docs/policy.plugin.example.yml @@ -11,6 +11,17 @@ workflow: block_labels: ["ai:disabled"] allow_labels: ["ai"] + engagement: + default: + allow: + - type: token_owner + needs_human_resume: + required_labels: ["ai"] + allow: + - type: token_owner + - type: collaborator_permission + minimum: "write" + lifecycle: plugin: manifest_path: /plugins/example/donkeyspace-plugin.yml diff --git a/migrations/0001_init.sql b/migrations/0001_init.sql index 14709c3..0c3aca8 100644 --- a/migrations/0001_init.sql +++ b/migrations/0001_init.sql @@ -42,6 +42,33 @@ CREATE TABLE IF NOT EXISTS webhook_deliveries ( received_at TIMESTAMPTZ NOT NULL DEFAULT now() ); +CREATE TABLE IF NOT EXISTS engagement_decisions ( + id BIGSERIAL PRIMARY KEY, + webhook_delivery_id BIGINT NOT NULL UNIQUE REFERENCES webhook_deliveries(id), + workflow_item_id BIGINT REFERENCES workflow_items(id), + gate TEXT NOT NULL, + disposition TEXT NOT NULL, + actor JSONB, + matched_selector JSONB, + reason TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS engagement_decisions_workflow_item_id_idx + ON engagement_decisions(workflow_item_id); + +CREATE TABLE IF NOT EXISTS github_managed_resources ( + id BIGSERIAL PRIMARY KEY, + repository_id BIGINT NOT NULL REFERENCES repositories(id), + workflow_item_id BIGINT REFERENCES workflow_items(id), + outbound_action_id BIGINT, + resource_kind TEXT NOT NULL, + provider_id TEXT NOT NULL, + metadata JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (repository_id, resource_kind, provider_id) +); + CREATE TABLE IF NOT EXISTS jobs ( id UUID PRIMARY KEY, workflow_item_id BIGINT REFERENCES workflow_items(id), @@ -127,6 +154,18 @@ CREATE TABLE IF NOT EXISTS outbound_actions ( ); ALTER TABLE outbound_actions ADD COLUMN IF NOT EXISTS last_error TEXT; +ALTER TABLE outbound_actions ADD COLUMN IF NOT EXISTS provider_resource_id TEXT; +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'github_managed_resources_outbound_action_id_fkey' + ) THEN + ALTER TABLE github_managed_resources + ADD CONSTRAINT github_managed_resources_outbound_action_id_fkey + FOREIGN KEY (outbound_action_id) REFERENCES outbound_actions(id); + END IF; +END $$; CREATE INDEX IF NOT EXISTS outbound_actions_status_idx ON outbound_actions(status); CREATE INDEX IF NOT EXISTS outbound_actions_job_id_idx ON outbound_actions(job_id);