use crate::backends::{ExternalBackend, ExternalId, Status};
use crate::github::http::GhHttp;
use async_trait::async_trait;
use std::sync::Arc;
#[allow(clippy::double_must_use)] #[async_trait]
pub trait CommandStoreOps: Send + Sync {
async fn resolve_task_id(&self, repo: &str, task_id: &str) -> anyhow::Result<Option<i64>>;
async fn set_block_reason(&self, id: i64, reason: Option<&str>) -> anyhow::Result<()>;
async fn reset_counters(&self, id: i64) -> anyhow::Result<()>;
}
#[async_trait]
impl CommandStoreOps for crate::store::TaskStore {
async fn resolve_task_id(&self, repo: &str, task_id: &str) -> anyhow::Result<Option<i64>> {
self.resolve_task_id(repo, task_id).await
}
async fn set_block_reason(&self, id: i64, reason: Option<&str>) -> anyhow::Result<()> {
self.set_block_reason(id, reason).await
}
async fn reset_counters(&self, id: i64) -> anyhow::Result<()> {
self.reset_counters(id).await
}
}
#[derive(Debug, Clone, PartialEq)]
pub enum OwnerCommand {
Retry,
Reroute(Option<String>),
Close,
Block(Option<String>),
Unblock,
Review,
}
impl std::fmt::Display for OwnerCommand {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Retry => write!(f, "/retry"),
Self::Reroute(Some(a)) => write!(f, "/reroute {a}"),
Self::Reroute(None) => write!(f, "/reroute"),
Self::Close => write!(f, "/close"),
Self::Block(Some(r)) => write!(f, "/block {r}"),
Self::Block(None) => write!(f, "/block"),
Self::Unblock => write!(f, "/unblock"),
Self::Review => write!(f, "/review"),
}
}
}
pub fn parse_command(body: &str) -> Option<OwnerCommand> {
let mut fence_state: Option<(char, usize)> = None;
for line in body.lines() {
let trimmed = line.trim();
let fence_ch = if trimmed.starts_with("```") {
Some('`')
} else if trimmed.starts_with("~~~") {
Some('~')
} else {
None
};
if let Some(ch) = fence_ch {
let run_len = trimmed.chars().take_while(|&c| c == ch).count();
match fence_state {
None => fence_state = Some((ch, run_len)),
Some((c, len)) if c == ch && run_len >= len => fence_state = None,
Some(_) => {} }
continue;
}
if fence_state.is_some() {
continue;
}
let command_text = trimmed
.strip_prefix("@orchestrator")
.or_else(|| trimmed.strip_prefix("@orch"))
.map(str::trim_start)
.unwrap_or(trimmed);
if !command_text.starts_with('/') {
continue;
}
let parts: Vec<&str> = command_text.splitn(2, char::is_whitespace).collect();
let cmd = parts[0];
let args = parts
.get(1)
.map(|s| s.trim())
.filter(|s| !s.is_empty())
.map(String::from);
match cmd {
"/retry" => return Some(OwnerCommand::Retry),
"/reroute" => return Some(OwnerCommand::Reroute(args)),
"/close" => return Some(OwnerCommand::Close),
"/block" => return Some(OwnerCommand::Block(args)),
"/unblock" => return Some(OwnerCommand::Unblock),
"/review" => return Some(OwnerCommand::Review),
_ => continue,
}
}
None
}
async fn cmd_reset_counters(store: &Option<Arc<dyn CommandStoreOps>>, repo: &str, task_id: &str) {
let Some(ref s) = store else { return };
match s.resolve_task_id(repo, task_id).await {
Ok(Some(id)) => {
if let Err(e) = s.reset_counters(id).await {
tracing::warn!(task_id, err = %e, "store reset_counters failed");
}
}
Ok(None) => {} Err(e) => {
tracing::warn!(task_id, err = %e, "resolve_task_id failed in store write helper");
}
}
}
pub async fn execute_command(
backend: &Arc<dyn ExternalBackend>,
gh: &GhHttp,
repo: &str,
task_id: &ExternalId,
command: &OwnerCommand,
store: &Option<Arc<dyn CommandStoreOps>>,
task_manager: &Arc<crate::engine::tasks::TaskManager>,
) -> anyhow::Result<String> {
match command {
OwnerCommand::Retry => {
let task = backend.get_task(task_id).await?;
for label in &task.labels {
if label.starts_with("agent:") {
backend.remove_label(task_id, label).await?;
}
}
cmd_reset_counters(store, repo, &task_id.0).await;
task_manager
.update_task_status(task_id, Status::New)
.await?;
Ok("`/retry` — reset attempts, cleared agent, reset to `status:new`".to_string())
}
OwnerCommand::Reroute(agent) => {
let task = backend.get_task(task_id).await?;
for label in &task.labels {
if label.starts_with("agent:") {
backend.remove_label(task_id, label).await?;
}
}
cmd_reset_counters(store, repo, &task_id.0).await;
if let Some(agent_name) = agent {
let label = format!("agent:{agent_name}");
backend.set_labels(task_id, &[label]).await?;
}
task_manager
.update_task_status(task_id, Status::New)
.await?;
match agent {
Some(a) => Ok(format!(
"`/reroute {a}` — cleared agent, reset attempts, forced `agent:{a}`, reset to `status:new`"
)),
None => {
Ok("`/reroute` — cleared agent, reset attempts, reset to `status:new`, will re-route".into())
}
}
}
OwnerCommand::Close => {
task_manager
.update_task_status(task_id, Status::Done)
.await?;
gh.close_issue(repo, &task_id.0).await?;
Ok("`/close` — marked `status:done` and closed issue".to_string())
}
OwnerCommand::Block(reason) => {
if let Some(ref store) = store {
let store_id = store.resolve_task_id(repo, &task_id.0).await?;
if let Some(store_id) = store_id {
store.set_block_reason(store_id, reason.as_deref()).await?;
}
}
task_manager
.update_task_status(task_id, Status::Blocked)
.await?;
match reason {
Some(r) => Ok(format!("`/block` — marked `status:blocked`: {r}")),
None => Ok("`/block` — marked `status:blocked`".to_string()),
}
}
OwnerCommand::Unblock => {
if let Some(ref store) = store {
let store_id = store.resolve_task_id(repo, &task_id.0).await?;
if let Some(store_id) = store_id {
store.set_block_reason(store_id, None).await?;
}
}
task_manager
.update_task_status(task_id, Status::New)
.await?;
Ok("`/unblock` — marked `status:new`, will re-dispatch".to_string())
}
OwnerCommand::Review => {
task_manager
.update_task_status(task_id, Status::NeedsReview)
.await?;
Ok(
"`/review` — set `status:needs_review`, review agent will pick up on next tick"
.to_string(),
)
}
}
}
pub enum CommandOutcome {
Executed { is_pr: bool },
NotOpen { is_pr: bool },
NotCollaborator { is_pr: bool },
FetchFailed,
CollaboratorCheckFailed,
}
#[allow(clippy::too_many_arguments)]
pub async fn validate_and_run_command(
backend: &Arc<dyn ExternalBackend>,
gh: &GhHttp,
repo: &str,
issue_number: &str,
command: &OwnerCommand,
author: &str,
store: &Option<Arc<dyn CommandStoreOps>>,
task_manager: &Arc<crate::engine::tasks::TaskManager>,
) -> CommandOutcome {
let issue_data = match gh.get_issue(repo, issue_number).await {
Ok(data) => data,
Err(e) => {
tracing::warn!(issue = %issue_number, err = %e, "failed to check issue state");
return CommandOutcome::FetchFailed;
}
};
let is_pr = issue_data.pull_request.is_some();
if issue_data.state != "open" {
tracing::debug!(issue = %issue_number, state = %issue_data.state, command = %command, "ignoring slash command on non-open issue");
return CommandOutcome::NotOpen { is_pr };
}
match gh.is_collaborator(repo, author).await {
Ok(true) => {}
Ok(false) => {
tracing::info!(author, command = %command, issue = %issue_number, "ignoring slash command from non-collaborator");
return CommandOutcome::NotCollaborator { is_pr };
}
Err(e) => {
tracing::warn!(author, err = %e, "failed to check collaborator status");
return CommandOutcome::CollaboratorCheckFailed;
}
}
let task_id = ExternalId(issue_number.to_string());
let now = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
match execute_command(backend, gh, repo, &task_id, command, store, task_manager).await {
Ok(msg) => {
tracing::info!(issue = %issue_number, author, command = %command, "executed command");
let confirmation = format!(
"[{now}] {msg} — executed by @{author}{}",
crate::engine::orch_footer()
);
if let Err(e) = backend.post_comment(&task_id, &confirmation).await {
tracing::warn!(issue = %issue_number, err = %e, "failed to post confirmation");
}
}
Err(e) => {
tracing::warn!(issue = %issue_number, command = %command, err = %e, "failed to execute command");
let error_msg = format!(
"[{now}] Failed to execute `{command}`: {e}{}",
crate::engine::orch_footer()
);
if let Err(e2) = backend.post_comment(&task_id, &error_msg).await {
tracing::warn!(issue = %issue_number, err = %e2, "failed to post error comment");
}
}
}
CommandOutcome::Executed { is_pr }
}
#[cfg(test)]
mod tests {
use super::*;
use crate::backends::{ExternalBackend, ExternalId, ExternalTask, Mention, Status};
use crate::store::{NewTask, TaskStore};
use async_trait::async_trait;
use std::sync::Arc;
struct MockBackend {
task: ExternalTask,
fail_remove_label: bool,
}
impl MockBackend {
fn new(id: &str) -> Self {
Self {
task: ExternalTask {
id: ExternalId(id.to_string()),
title: "Task".to_string(),
body: "".to_string(),
state: "open".to_string(),
labels: vec![],
author: "bot".to_string(),
created_at: "2026-01-01T00:00:00Z".to_string(),
updated_at: "2026-01-01T00:00:00Z".to_string(),
url: "".to_string(),
},
fail_remove_label: false,
}
}
fn with_labels(id: &str, labels: Vec<String>) -> Self {
let mut b = Self::new(id);
b.task.labels = labels;
b
}
fn failing_remove_label(id: &str, labels: Vec<String>) -> Self {
let mut b = Self::with_labels(id, labels);
b.fail_remove_label = true;
b
}
}
#[async_trait]
impl ExternalBackend for MockBackend {
fn name(&self) -> &str {
"mock"
}
async fn create_task(
&self,
_title: &str,
_body: &str,
_labels: &[String],
) -> anyhow::Result<ExternalId> {
Ok(self.task.id.clone())
}
async fn get_task(&self, _id: &ExternalId) -> anyhow::Result<ExternalTask> {
Ok(self.task.clone())
}
async fn list_by_status(&self, _status: Status) -> anyhow::Result<Vec<ExternalTask>> {
Ok(vec![])
}
async fn list_routable(&self) -> anyhow::Result<Vec<ExternalTask>> {
Ok(vec![])
}
async fn post_comment(&self, _id: &ExternalId, _body: &str) -> anyhow::Result<()> {
Ok(())
}
async fn set_labels(&self, _id: &ExternalId, _labels: &[String]) -> anyhow::Result<()> {
Ok(())
}
async fn remove_label(&self, _id: &ExternalId, label: &str) -> anyhow::Result<()> {
if self.fail_remove_label {
anyhow::bail!("simulated remove_label failure for {label}");
}
Ok(())
}
async fn get_sub_issues(&self, _id: &ExternalId) -> anyhow::Result<Vec<ExternalId>> {
Ok(vec![])
}
async fn create_sub_task(
&self,
_parent: &ExternalId,
_title: &str,
_body: &str,
_labels: &[String],
) -> anyhow::Result<ExternalId> {
Ok(ExternalId("child".to_string()))
}
async fn ensure_status_label(&self, _label: &str) -> anyhow::Result<()> {
Ok(())
}
async fn has_open_issue_with_title(
&self,
_title: &str,
_label: &str,
) -> anyhow::Result<bool> {
Ok(false)
}
async fn health_check(&self) -> anyhow::Result<()> {
Ok(())
}
async fn is_pr_merged(&self, _branch: &str) -> anyhow::Result<bool> {
Ok(false)
}
async fn get_authenticated_user(&self) -> anyhow::Result<Option<String>> {
Ok(Some("testbot".to_string()))
}
async fn get_mentions(&self, _since: &str) -> anyhow::Result<Vec<Mention>> {
Ok(vec![])
}
async fn update_status(&self, _id: &ExternalId, _status: Status) -> anyhow::Result<()> {
Ok(())
}
}
struct FailingStore;
#[async_trait]
impl CommandStoreOps for FailingStore {
async fn resolve_task_id(
&self,
_repo: &str,
_task_id: &str,
) -> anyhow::Result<Option<i64>> {
anyhow::bail!("simulated resolve_task_id DB failure")
}
async fn set_block_reason(&self, _id: i64, _reason: Option<&str>) -> anyhow::Result<()> {
anyhow::bail!("simulated set_block_reason DB failure")
}
async fn reset_counters(&self, _id: i64) -> anyhow::Result<()> {
Ok(())
}
}
#[tokio::test]
async fn block_command_persists_reason_and_unblock_clears_it() {
let store = Arc::new(TaskStore::open_memory().await.unwrap());
let store_id = store
.create(&NewTask {
external_id: Some("42".to_string()),
repo: "owner/repo".to_string(),
origin: "github".to_string(),
title: "Task".to_string(),
body: "".to_string(),
source: "manual".to_string(),
source_id: "".to_string(),
author: "bot".to_string(),
url: "".to_string(),
labels: vec![],
parent_id: None,
})
.await
.unwrap();
let backend: Arc<dyn ExternalBackend> = Arc::new(MockBackend::new("42"));
let backend_for_call = backend.clone();
let task_manager = Arc::new(crate::engine::tasks::TaskManager::with_store(
backend,
store.clone(),
"owner/repo".to_string(),
));
let gh = crate::github::http::GhHttp::new().unwrap();
let task_id = ExternalId("42".to_string());
execute_command(
&backend_for_call,
&gh,
"owner/repo",
&task_id,
&OwnerCommand::Block(Some("waiting on upstream fix".to_string())),
&Some(Arc::clone(&store) as Arc<dyn CommandStoreOps>),
&task_manager,
)
.await
.unwrap();
let task = store.get(store_id).await.unwrap();
assert_eq!(task.status, crate::store::TaskStatus::Blocked);
assert_eq!(
task.block_reason.as_deref(),
Some("waiting on upstream fix")
);
execute_command(
&backend_for_call,
&gh,
"owner/repo",
&task_id,
&OwnerCommand::Unblock,
&Some(Arc::clone(&store) as Arc<dyn CommandStoreOps>),
&task_manager,
)
.await
.unwrap();
let task = store.get(store_id).await.unwrap();
assert_eq!(task.status, crate::store::TaskStatus::New);
assert!(task.block_reason.is_none());
}
#[test]
fn parse_retry() {
assert_eq!(parse_command("/retry"), Some(OwnerCommand::Retry));
}
#[test]
fn parse_retry_with_trailing_whitespace() {
assert_eq!(parse_command("/retry "), Some(OwnerCommand::Retry));
}
#[test]
fn parse_close() {
assert_eq!(parse_command("/close"), Some(OwnerCommand::Close));
}
#[test]
fn parse_unblock() {
assert_eq!(parse_command("/unblock"), Some(OwnerCommand::Unblock));
}
#[test]
fn parse_review() {
assert_eq!(parse_command("/review"), Some(OwnerCommand::Review));
}
#[test]
fn parse_reroute_no_agent() {
assert_eq!(parse_command("/reroute"), Some(OwnerCommand::Reroute(None)));
}
#[test]
fn parse_reroute_with_agent() {
assert_eq!(
parse_command("/reroute codex"),
Some(OwnerCommand::Reroute(Some("codex".into())))
);
}
#[test]
fn parse_block_no_reason() {
assert_eq!(parse_command("/block"), Some(OwnerCommand::Block(None)));
}
#[test]
fn parse_block_with_reason() {
assert_eq!(
parse_command("/block waiting on upstream fix"),
Some(OwnerCommand::Block(Some("waiting on upstream fix".into())))
);
}
#[test]
fn parse_ignores_non_command_text() {
assert_eq!(parse_command("This is a regular comment"), None);
}
#[test]
fn parse_ignores_unknown_commands() {
assert_eq!(parse_command("/unknown"), None);
}
#[test]
fn parse_command_in_multiline_body() {
let body = "Some context here\n\n/retry\n\nMore text";
assert_eq!(parse_command(body), Some(OwnerCommand::Retry));
}
#[test]
fn parse_first_valid_command_wins() {
let body = "/retry\n/close";
assert_eq!(parse_command(body), Some(OwnerCommand::Retry));
}
#[test]
fn parse_skips_unknown_finds_valid() {
let body = "/unknown\n/close";
assert_eq!(parse_command(body), Some(OwnerCommand::Close));
}
#[test]
fn parse_indented_command() {
assert_eq!(parse_command(" /retry"), Some(OwnerCommand::Retry));
}
#[test]
fn parse_mention_prefixed_command() {
assert_eq!(parse_command("@orch /retry"), Some(OwnerCommand::Retry));
}
#[test]
fn parse_legacy_mention_prefixed_command() {
assert_eq!(
parse_command("@orchestrator /close"),
Some(OwnerCommand::Close)
);
}
#[test]
fn parse_ignores_non_command_after_mention() {
assert_eq!(parse_command("@orch please retry this"), None);
}
#[test]
fn parse_ignores_command_in_code_fence() {
let body = "Some context\n```\n/retry\n```\nMore text";
assert_eq!(parse_command(body), None);
}
#[test]
fn parse_ignores_mention_prefixed_command_in_code_fence() {
let body = "```\n@orch /retry\n```";
assert_eq!(parse_command(body), None);
}
#[test]
fn parse_ignores_command_in_tilde_fence() {
let body = "Some context\n~~~\n/close\n~~~\nMore text";
assert_eq!(parse_command(body), None);
}
#[test]
fn parse_finds_command_after_code_fence() {
let body = "```\n/retry\n```\n/close";
assert_eq!(parse_command(body), Some(OwnerCommand::Close));
}
#[test]
fn parse_ignores_command_in_code_fence_with_lang() {
let body = "```markdown\n/retry\n```";
assert_eq!(parse_command(body), None);
}
#[test]
fn parse_mixed_fence_does_not_close_backtick_with_tilde() {
let body = "```bash\n/retry\n~~~\n/retry\n```";
assert_eq!(parse_command(body), None);
}
#[test]
fn parse_mixed_fence_does_not_close_tilde_with_backtick() {
let body = "~~~\n/close\n```\n/close\n~~~";
assert_eq!(parse_command(body), None);
}
#[test]
fn parse_command_after_mixed_fence_properly_closed() {
let body = "```\n/retry\n~~~\n/retry\n```\n/close";
assert_eq!(parse_command(body), Some(OwnerCommand::Close));
}
#[tokio::test]
async fn retry_propagates_remove_label_error() {
let backend: Arc<dyn ExternalBackend> = Arc::new(MockBackend::failing_remove_label(
"42",
vec!["agent:claude".to_string()],
));
let store = Arc::new(TaskStore::open_memory().await.unwrap());
let task_manager = Arc::new(crate::engine::tasks::TaskManager::with_store(
backend.clone(),
store.clone(),
"owner/repo".to_string(),
));
let gh = crate::github::http::GhHttp::new().unwrap();
let task_id = ExternalId("42".to_string());
let result = execute_command(
&backend,
&gh,
"owner/repo",
&task_id,
&OwnerCommand::Retry,
&Some(store as Arc<dyn CommandStoreOps>),
&task_manager,
)
.await;
assert!(result.is_err(), "/retry must fail when remove_label fails");
assert!(
result.unwrap_err().to_string().contains("remove_label"),
"error must mention remove_label"
);
}
#[tokio::test]
async fn reroute_propagates_remove_label_error() {
let backend: Arc<dyn ExternalBackend> = Arc::new(MockBackend::failing_remove_label(
"42",
vec!["agent:codex".to_string()],
));
let store = Arc::new(TaskStore::open_memory().await.unwrap());
let task_manager = Arc::new(crate::engine::tasks::TaskManager::with_store(
backend.clone(),
store.clone(),
"owner/repo".to_string(),
));
let gh = crate::github::http::GhHttp::new().unwrap();
let task_id = ExternalId("42".to_string());
let result = execute_command(
&backend,
&gh,
"owner/repo",
&task_id,
&OwnerCommand::Reroute(None),
&Some(store as Arc<dyn CommandStoreOps>),
&task_manager,
)
.await;
assert!(
result.is_err(),
"/reroute must fail when remove_label fails"
);
assert!(
result.unwrap_err().to_string().contains("remove_label"),
"error must mention remove_label"
);
}
#[tokio::test]
async fn block_propagates_resolve_task_id_error() {
let backend: Arc<dyn ExternalBackend> = Arc::new(MockBackend::new("42"));
let store = Arc::new(TaskStore::open_memory().await.unwrap());
let task_manager = Arc::new(crate::engine::tasks::TaskManager::with_store(
backend.clone(),
store,
"owner/repo".to_string(),
));
let gh = crate::github::http::GhHttp::new().unwrap();
let task_id = ExternalId("42".to_string());
let failing_store: Arc<dyn CommandStoreOps> = Arc::new(FailingStore);
let result = execute_command(
&backend,
&gh,
"owner/repo",
&task_id,
&OwnerCommand::Block(Some("reason".to_string())),
&Some(failing_store),
&task_manager,
)
.await;
assert!(
result.is_err(),
"/block must fail when resolve_task_id fails"
);
assert!(
result.unwrap_err().to_string().contains("resolve_task_id"),
"error must mention resolve_task_id"
);
}
#[tokio::test]
async fn unblock_propagates_resolve_task_id_error() {
let backend: Arc<dyn ExternalBackend> = Arc::new(MockBackend::new("42"));
let store = Arc::new(TaskStore::open_memory().await.unwrap());
let task_manager = Arc::new(crate::engine::tasks::TaskManager::with_store(
backend.clone(),
store,
"owner/repo".to_string(),
));
let gh = crate::github::http::GhHttp::new().unwrap();
let task_id = ExternalId("42".to_string());
let failing_store: Arc<dyn CommandStoreOps> = Arc::new(FailingStore);
let result = execute_command(
&backend,
&gh,
"owner/repo",
&task_id,
&OwnerCommand::Unblock,
&Some(failing_store),
&task_manager,
)
.await;
assert!(
result.is_err(),
"/unblock must fail when resolve_task_id fails"
);
assert!(
result.unwrap_err().to_string().contains("resolve_task_id"),
"error must mention resolve_task_id"
);
}
#[test]
fn parse_shorter_backtick_does_not_close_longer_backtick_fence() {
let body = "````\n/retry\n```\n/retry\n````";
assert_eq!(parse_command(body), None);
}
#[test]
fn parse_longer_backtick_closes_shorter_backtick_fence() {
let body = "```\n/retry\n````\n/close";
assert_eq!(parse_command(body), Some(OwnerCommand::Close));
}
#[test]
fn parse_shorter_tilde_does_not_close_longer_tilde_fence() {
let body = "~~~~\n/close\n~~~\n/close\n~~~~";
assert_eq!(parse_command(body), None);
}
#[test]
fn parse_longer_tilde_closes_shorter_tilde_fence() {
let body = "~~~\n/close\n~~~~\n/retry";
assert_eq!(parse_command(body), Some(OwnerCommand::Retry));
}
#[test]
fn parse_command_after_longer_backtick_fence_properly_closed() {
let body = "````markdown\n/retry\n```\n/retry\n````\n/close";
assert_eq!(parse_command(body), Some(OwnerCommand::Close));
}
#[test]
fn display_commands() {
assert_eq!(OwnerCommand::Retry.to_string(), "/retry");
assert_eq!(
OwnerCommand::Reroute(Some("codex".into())).to_string(),
"/reroute codex"
);
assert_eq!(OwnerCommand::Reroute(None).to_string(), "/reroute");
assert_eq!(OwnerCommand::Close.to_string(), "/close");
assert_eq!(
OwnerCommand::Block(Some("reason".into())).to_string(),
"/block reason"
);
assert_eq!(OwnerCommand::Block(None).to_string(), "/block");
assert_eq!(OwnerCommand::Unblock.to_string(), "/unblock");
assert_eq!(OwnerCommand::Review.to_string(), "/review");
}
}