refactor: remove Matrix bot, make agent-fleet platform-agnostic API service
- Remove src/integrations/matrix/ (bot connection, command parsing, notification formatting)
- Remove matrix-sdk dependency from Cargo.toml
- Remove MatrixConfig from config.rs and [matrix] from config.example.toml
- Add GET /api/v1/tasks (list with status/agent_id filter)
- Add POST /api/v1/tasks/{task_id}/retry (Failed/AgentLost → Assigned)
- Add EventStore::list_tasks() with parameterized query
- 29/29 tests pass
Platform integration (Telegram, Matrix, Feishu) is Agent-side responsibility.
agent-fleet is now a pure HTTP API orchestration engine.
This commit is contained in:
parent
6efca09018
commit
1bc7580ecc
15 changed files with 435 additions and 2367 deletions
197
src/api.rs
197
src/api.rs
|
|
@ -379,6 +379,58 @@ fn parse_task_source(source: &str) -> Option<(String, u64)> {
|
|||
Some((repo.to_string(), issue_number))
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub struct ListTasksQuery {
|
||||
pub status: Option<String>,
|
||||
pub agent_id: Option<String>,
|
||||
}
|
||||
|
||||
pub async fn list_tasks(
|
||||
State(state): State<AppState>,
|
||||
Query(query): Query<ListTasksQuery>,
|
||||
) -> Result<Json<Vec<Task>>, ApiError> {
|
||||
let store = state.store.clone();
|
||||
|
||||
tokio::task::spawn_blocking(move || -> Result<Json<Vec<Task>>, ApiError> {
|
||||
let store = store.lock().map_err(|e| ApiError::Poisoned(e.to_string()))?;
|
||||
let tasks = store.list_tasks(query.status.as_deref(), query.agent_id.as_deref())?;
|
||||
Ok(Json(tasks))
|
||||
})
|
||||
.await?
|
||||
}
|
||||
|
||||
pub async fn retry_task(
|
||||
State(state): State<AppState>,
|
||||
axum::extract::Path(task_id): axum::extract::Path<String>,
|
||||
) -> Result<Json<Task>, ApiError> {
|
||||
let store = state.store.clone();
|
||||
let sm = StateMachine::new(store.clone());
|
||||
|
||||
let task_id_for_check = task_id.clone();
|
||||
let current = tokio::task::spawn_blocking(move || -> Result<Option<Task>, ApiError> {
|
||||
let store = store.lock().map_err(|e| ApiError::Poisoned(e.to_string()))?;
|
||||
Ok(store.read_task(&task_id_for_check)?)
|
||||
})
|
||||
.await??;
|
||||
|
||||
let task = current.ok_or_else(|| ApiError::NotFound(format!("task {}", task_id)))?;
|
||||
|
||||
if !matches!(task.status, TaskStatus::Failed | TaskStatus::AgentLost) {
|
||||
return Err(ApiError::BadRequest(format!(
|
||||
"task {} is not retryable (current status: {})",
|
||||
task.task_id,
|
||||
task.status.as_str()
|
||||
)));
|
||||
}
|
||||
|
||||
let updated = sm
|
||||
.transition(&task_id, TaskStatus::Assigned, None, "retry")
|
||||
.await
|
||||
.map_err(|e| ApiError::BadRequest(e.to_string()))?;
|
||||
|
||||
Ok(Json(updated))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
|
@ -734,4 +786,149 @@ mod tests {
|
|||
};
|
||||
assert_eq!(task.status, TaskStatus::Running);
|
||||
}
|
||||
|
||||
// ─── Task API tests ─────────────────────────────────────────
|
||||
|
||||
fn sample_task_variant(task_id: &str, status: TaskStatus, agent_id: Option<&str>) -> Task {
|
||||
Task {
|
||||
task_id: task_id.to_string(),
|
||||
source: format!("forgejo:org/repo#{task_id}"),
|
||||
task_type: "code".into(),
|
||||
priority: Priority::High,
|
||||
status,
|
||||
assigned_agent_id: agent_id.map(String::from),
|
||||
requirements: "do something".into(),
|
||||
labels: vec!["agent:code".into(), "priority:high".into()],
|
||||
created_at: Utc::now(),
|
||||
assigned_at: None,
|
||||
started_at: None,
|
||||
completed_at: None,
|
||||
retry_count: 0,
|
||||
max_retries: 2,
|
||||
timeout_seconds: 1800,
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_tasks_returns_all_tasks() {
|
||||
let (_dir, state) = test_store();
|
||||
{
|
||||
let store = state.store.lock().unwrap();
|
||||
store.insert_task(&sample_task_variant("task-1", TaskStatus::Created, None)).unwrap();
|
||||
store.insert_task(&sample_task_variant("task-2", TaskStatus::Running, Some("worker-01"))).unwrap();
|
||||
}
|
||||
|
||||
let tasks = list_tasks(
|
||||
State(state),
|
||||
Query(ListTasksQuery { status: None, agent_id: None }),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(tasks.0.len(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_tasks_filters_by_status() {
|
||||
let (_dir, state) = test_store();
|
||||
{
|
||||
let store = state.store.lock().unwrap();
|
||||
store.insert_task(&sample_task_variant("task-1", TaskStatus::Created, None)).unwrap();
|
||||
store.insert_task(&sample_task_variant("task-2", TaskStatus::Running, Some("worker-01"))).unwrap();
|
||||
}
|
||||
|
||||
let tasks = list_tasks(
|
||||
State(state),
|
||||
Query(ListTasksQuery { status: Some("running".into()), agent_id: None }),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(tasks.0.len(), 1);
|
||||
assert_eq!(tasks.0[0].task_id, "task-2");
|
||||
assert_eq!(tasks.0[0].status, TaskStatus::Running);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_tasks_filters_by_agent() {
|
||||
let (_dir, state) = test_store();
|
||||
{
|
||||
let store = state.store.lock().unwrap();
|
||||
store.insert_task(&sample_task_variant("task-1", TaskStatus::Running, Some("worker-01"))).unwrap();
|
||||
store.insert_task(&sample_task_variant("task-2", TaskStatus::Running, Some("worker-02"))).unwrap();
|
||||
}
|
||||
|
||||
let tasks = list_tasks(
|
||||
State(state),
|
||||
Query(ListTasksQuery { status: None, agent_id: Some("worker-01".into()) }),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(tasks.0.len(), 1);
|
||||
assert_eq!(tasks.0[0].task_id, "task-1");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn retry_task_succeeds_for_failed_task() {
|
||||
let (_dir, state) = test_store();
|
||||
{
|
||||
let store = state.store.lock().unwrap();
|
||||
store.insert_task(&sample_task_variant("task-failed", TaskStatus::Failed, Some("worker-01"))).unwrap();
|
||||
}
|
||||
|
||||
let updated = retry_task(State(state.clone()), axum::extract::Path("task-failed".to_string()))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(updated.0.status, TaskStatus::Assigned);
|
||||
|
||||
// Verify in DB
|
||||
let task = {
|
||||
let store = state.store.lock().unwrap();
|
||||
store.read_task("task-failed").unwrap().unwrap()
|
||||
};
|
||||
assert_eq!(task.status, TaskStatus::Assigned);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn retry_task_succeeds_for_agent_lost_task() {
|
||||
let (_dir, state) = test_store();
|
||||
{
|
||||
let store = state.store.lock().unwrap();
|
||||
store.insert_task(&sample_task_variant("task-lost", TaskStatus::AgentLost, Some("worker-01"))).unwrap();
|
||||
}
|
||||
|
||||
let updated = retry_task(State(state.clone()), axum::extract::Path("task-lost".to_string()))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(updated.0.status, TaskStatus::Assigned);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn retry_task_rejects_non_retryable_status() {
|
||||
let (_dir, state) = test_store();
|
||||
{
|
||||
let store = state.store.lock().unwrap();
|
||||
store.insert_task(&sample_task_variant("task-running", TaskStatus::Running, Some("worker-01"))).unwrap();
|
||||
}
|
||||
|
||||
let err = retry_task(State(state.clone()), axum::extract::Path("task-running".to_string()))
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(matches!(err, ApiError::BadRequest(_)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn retry_task_returns_not_found_for_missing_task() {
|
||||
let (_dir, state) = test_store();
|
||||
|
||||
let err = retry_task(State(state), axum::extract::Path("nonexistent".to_string()))
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(matches!(err, ApiError::NotFound(_)));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,7 +6,6 @@ use crate::adapters::AdapterInstanceConfig;
|
|||
pub struct Config {
|
||||
pub server: ServerConfig,
|
||||
pub forgejo: ForgejoConfig,
|
||||
pub matrix: MatrixConfig,
|
||||
pub orchestrator: OrchestratorConfig,
|
||||
#[serde(default)]
|
||||
pub adapters: Vec<AdapterInstanceConfig>,
|
||||
|
|
@ -25,14 +24,6 @@ pub struct ForgejoConfig {
|
|||
pub webhook_secret: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct MatrixConfig {
|
||||
pub homeserver_url: String,
|
||||
pub user_id: String,
|
||||
pub access_token: String,
|
||||
pub room_id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct OrchestratorConfig {
|
||||
pub db_path: String,
|
||||
|
|
@ -54,12 +45,6 @@ impl Default for Config {
|
|||
token: String::new(),
|
||||
webhook_secret: String::new(),
|
||||
},
|
||||
matrix: MatrixConfig {
|
||||
homeserver_url: "https://matrix.0x08.org".into(),
|
||||
user_id: "@jeeves:0x08.org".into(),
|
||||
access_token: String::new(),
|
||||
room_id: String::new(),
|
||||
},
|
||||
orchestrator: OrchestratorConfig {
|
||||
db_path: "data/agent-fleet.db".into(),
|
||||
heartbeat_interval_secs: 60,
|
||||
|
|
|
|||
|
|
@ -275,6 +275,36 @@ impl EventStore {
|
|||
.collect::<SqlResult<Vec<_>>>()
|
||||
}
|
||||
|
||||
pub fn list_tasks(
|
||||
&self,
|
||||
status: Option<&str>,
|
||||
agent_id: Option<&str>,
|
||||
) -> SqlResult<Vec<Task>> {
|
||||
let mut sql = String::from(
|
||||
"SELECT task_id, source, task_type, priority, status, assigned_agent_id,
|
||||
requirements, labels, created_at, assigned_at, started_at, completed_at,
|
||||
retry_count, max_retries, timeout_seconds
|
||||
FROM tasks WHERE 1=1",
|
||||
);
|
||||
let mut param_values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
|
||||
|
||||
if let Some(s) = status {
|
||||
sql.push_str(" AND status = ?");
|
||||
param_values.push(Box::new(s.to_string()));
|
||||
}
|
||||
if let Some(a) = agent_id {
|
||||
sql.push_str(" AND assigned_agent_id = ?");
|
||||
param_values.push(Box::new(a.to_string()));
|
||||
}
|
||||
sql.push_str(" ORDER BY created_at DESC");
|
||||
|
||||
let params: Vec<&dyn rusqlite::types::ToSql> = param_values.iter().map(|p| p.as_ref()).collect();
|
||||
|
||||
let mut stmt = self.conn.prepare(&sql)?;
|
||||
stmt.query_map(params.as_slice(), Self::row_to_task)?
|
||||
.collect::<SqlResult<Vec<_>>>()
|
||||
}
|
||||
|
||||
// ─── Task/event write operations ─────────────────────────────
|
||||
|
||||
pub fn insert_task(&self, task: &Task) -> SqlResult<()> {
|
||||
|
|
|
|||
|
|
@ -1,531 +0,0 @@
|
|||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use matrix_sdk::authentication::matrix::MatrixSession;
|
||||
use matrix_sdk::config::SyncSettings;
|
||||
use matrix_sdk::ruma::events::room::message::{
|
||||
MessageType, OriginalSyncRoomMessageEvent, RoomMessageEventContent,
|
||||
};
|
||||
use matrix_sdk::ruma::events::relation::Thread;
|
||||
use matrix_sdk::ruma::OwnedEventId;
|
||||
use matrix_sdk::ruma::{OwnedDeviceId, OwnedRoomId, OwnedUserId, UserId};
|
||||
use matrix_sdk::{Client, Room};
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
use crate::config::MatrixConfig;
|
||||
use crate::core::event_store::EventStore;
|
||||
use crate::core::models::{Agent, TaskStatus};
|
||||
use crate::core::state_machine::StateMachine;
|
||||
|
||||
/// The bot context — shared state for all handlers.
|
||||
#[derive(Clone)]
|
||||
pub struct BotContext {
|
||||
pub store: Arc<Mutex<EventStore>>,
|
||||
pub sm: Arc<StateMachine>,
|
||||
pub config: MatrixConfig,
|
||||
/// Maps agent_id → root event_id for per-agent threads.
|
||||
pub agent_threads: Arc<RwLock<HashMap<String, OwnedEventId>>>,
|
||||
}
|
||||
|
||||
// ─── Command parsing ────────────────────────────────────────────
|
||||
|
||||
/// A parsed Matrix command.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum BotCommand {
|
||||
FleetStatus,
|
||||
Assign { agent_id: String, issue_ref: String },
|
||||
Retry { issue_ref: String },
|
||||
Unknown(String),
|
||||
}
|
||||
|
||||
/// Parse a Matrix message body into a BotCommand.
|
||||
/// Only returns Some for known command prefixes: fleet, assign, retry.
|
||||
pub fn parse_command(body: &str) -> Option<BotCommand> {
|
||||
let body = body.trim();
|
||||
// Accept both "/fleet status" and "fleet status" (in case the / is stripped)
|
||||
let body = body.strip_prefix('/').unwrap_or(body);
|
||||
|
||||
let mut parts = body.splitn(4, ' ');
|
||||
let first = parts.next()?.to_lowercase();
|
||||
match first.as_str() {
|
||||
"fleet" => {
|
||||
let sub = parts.next()?.to_lowercase();
|
||||
match sub.as_str() {
|
||||
"status" => Some(BotCommand::FleetStatus),
|
||||
_ => Some(BotCommand::Unknown(format!("fleet {sub}"))),
|
||||
}
|
||||
}
|
||||
"assign" => {
|
||||
let agent_id = parts.next()?.to_string();
|
||||
let issue_ref = parts.next()?.to_string();
|
||||
Some(BotCommand::Assign { agent_id, issue_ref })
|
||||
}
|
||||
"retry" => {
|
||||
let issue_ref = parts.next()?.to_string();
|
||||
Some(BotCommand::Retry { issue_ref })
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Notification formatting ────────────────────────────────────
|
||||
|
||||
/// Format a task-assigned notification.
|
||||
///
|
||||
/// Spec: `📋 #42 → worker-03 [code:typescript]`
|
||||
pub fn format_task_assigned(task_id: &str, agent_id: &str, task_type: &str) -> String {
|
||||
format!("📋 {task_id} → {agent_id} [{task_type}]")
|
||||
}
|
||||
|
||||
/// Format a task-completed notification.
|
||||
///
|
||||
/// Spec: `✅ #42 completed by worker-03 — PR #15 — "修复登录验证 bug"`
|
||||
pub fn format_task_completed(task_id: &str, agent_id: &str, summary: &str, artifact_hint: Option<&str>) -> String {
|
||||
match artifact_hint {
|
||||
Some(hint) => format!("✅ {task_id} completed by {agent_id} — {hint} — \"{summary}\""),
|
||||
None => format!("✅ {task_id} completed by {agent_id} — \"{summary}\""),
|
||||
}
|
||||
}
|
||||
|
||||
/// Format a task-failed notification.
|
||||
///
|
||||
/// Spec: `❌ #42 failed — worker-03 — "构建超时"`
|
||||
pub fn format_task_failed(task_id: &str, agent_id: &str, error: &str) -> String {
|
||||
format!("❌ {task_id} failed — {agent_id} — \"{error}\"")
|
||||
}
|
||||
|
||||
/// Format an agent-offline alert.
|
||||
///
|
||||
/// Spec: `⚠️ worker-03 offline — 2 running tasks affected`
|
||||
pub fn format_agent_offline(agent_id: &str, affected_tasks: usize) -> String {
|
||||
format!("⚠️ {agent_id} offline — {affected_tasks} running tasks affected")
|
||||
}
|
||||
|
||||
/// Format the fleet status table as plain text.
|
||||
pub fn format_fleet_status(agents: &[Agent]) -> String {
|
||||
if agents.is_empty() {
|
||||
return "No agents registered.".to_string();
|
||||
}
|
||||
|
||||
let mut lines = vec!["Agent ID | Type | Status | Tasks | Capabilities".to_string()];
|
||||
lines.push("-----------------+--------------+---------+-------+---------------------------".to_string());
|
||||
|
||||
for agent in agents {
|
||||
let caps = agent.capabilities.join(", ");
|
||||
let caps_display = if caps.len() > 27 {
|
||||
format!("{}…", &caps[..24])
|
||||
} else {
|
||||
caps
|
||||
};
|
||||
lines.push(format!(
|
||||
"{:<16} | {:<12} | {:<7} | {:<5} | {}",
|
||||
truncate_str(&agent.agent_id, 16),
|
||||
truncate_str(agent.agent_type.as_str(), 12),
|
||||
agent.status.as_str(),
|
||||
agent.current_tasks,
|
||||
caps_display,
|
||||
));
|
||||
}
|
||||
|
||||
lines.join("\n")
|
||||
}
|
||||
|
||||
fn truncate_str(s: &str, max_len: usize) -> String {
|
||||
if s.len() <= max_len {
|
||||
s.to_string()
|
||||
} else {
|
||||
format!("{}…", &s[..max_len - 1])
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Bot lifecycle ──────────────────────────────────────────────
|
||||
|
||||
/// Build a Matrix client from config.
|
||||
pub async fn build_client(config: &MatrixConfig) -> Result<Client, Box<dyn std::error::Error + Send + Sync>> {
|
||||
let client = Client::builder()
|
||||
.homeserver_url(&config.homeserver_url)
|
||||
.build()
|
||||
.await?;
|
||||
|
||||
let user_id = UserId::parse(&config.user_id)?;
|
||||
let device_id: OwnedDeviceId = "AGENTFLEET".into();
|
||||
|
||||
let session = MatrixSession {
|
||||
meta: matrix_sdk::SessionMeta {
|
||||
user_id,
|
||||
device_id,
|
||||
},
|
||||
tokens: matrix_sdk::authentication::matrix::MatrixSessionTokens {
|
||||
access_token: config.access_token.clone(),
|
||||
refresh_token: None,
|
||||
},
|
||||
};
|
||||
|
||||
client.restore_session(session).await?;
|
||||
Ok(client)
|
||||
}
|
||||
|
||||
/// Start the Matrix bot — spawns a background sync loop.
|
||||
pub async fn start_bot(
|
||||
config: MatrixConfig,
|
||||
store: Arc<Mutex<EventStore>>,
|
||||
sm: Arc<StateMachine>,
|
||||
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
let room_id = OwnedRoomId::try_from(config.room_id.as_str())?;
|
||||
let client = build_client(&config).await?;
|
||||
|
||||
// Join the room
|
||||
let _ = client.join_room_by_id(&room_id).await?;
|
||||
|
||||
let ctx = BotContext {
|
||||
store,
|
||||
sm,
|
||||
config,
|
||||
agent_threads: Arc::new(RwLock::new(HashMap::new())),
|
||||
};
|
||||
|
||||
// Register the message handler
|
||||
client.add_event_handler(move |ev: OriginalSyncRoomMessageEvent, room: Room| {
|
||||
let ctx = ctx.clone();
|
||||
async move {
|
||||
handle_message(ev, room, ctx).await;
|
||||
}
|
||||
});
|
||||
|
||||
// Spawn the sync loop in background
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
match client.sync(SyncSettings::default()).await {
|
||||
Ok(_) => {}
|
||||
Err(e) => {
|
||||
tracing::error!("Matrix sync error: {e}");
|
||||
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
tracing::info!("Matrix bot started, monitoring room {}", room_id);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Handle an incoming Matrix room message.
|
||||
async fn handle_message(
|
||||
ev: OriginalSyncRoomMessageEvent,
|
||||
room: Room,
|
||||
ctx: BotContext,
|
||||
) {
|
||||
// Only handle text messages
|
||||
let text = match &ev.content.msgtype {
|
||||
MessageType::Text(text) => text.body.clone(),
|
||||
_ => return,
|
||||
};
|
||||
|
||||
// Skip our own messages
|
||||
if ev.sender == ctx.config.user_id.parse::<OwnedUserId>().unwrap_or_else(|_| ev.sender.clone()) {
|
||||
return;
|
||||
}
|
||||
|
||||
let Some(cmd) = parse_command(&text) else {
|
||||
return;
|
||||
};
|
||||
|
||||
tracing::debug!(?cmd, "received matrix command");
|
||||
|
||||
match cmd {
|
||||
BotCommand::FleetStatus => {
|
||||
let agents = {
|
||||
let store = ctx.store.lock().unwrap();
|
||||
store.list_agents(None, None).unwrap_or_default()
|
||||
};
|
||||
let table = format_fleet_status(&agents);
|
||||
send_plain(&room, &table).await;
|
||||
}
|
||||
BotCommand::Assign { agent_id, issue_ref } => {
|
||||
let result = {
|
||||
let store = ctx.store.lock().unwrap();
|
||||
let task = store.read_task(&issue_ref);
|
||||
let agent = store.find_agent_by_id(&agent_id);
|
||||
(task, agent)
|
||||
};
|
||||
|
||||
match result {
|
||||
(Ok(Some(_task)), Ok(Some(_agent))) => {
|
||||
match ctx.sm.transition(&issue_ref, TaskStatus::Assigned, Some(&agent_id), "manual assign via matrix").await {
|
||||
Ok(updated) => {
|
||||
let msg = format_task_assigned(&updated.task_id, &agent_id, &updated.task_type);
|
||||
send_plain(&room, &msg).await;
|
||||
}
|
||||
Err(e) => {
|
||||
send_plain(&room, &format!("❌ Failed to assign: {e}")).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
(Ok(None), _) => {
|
||||
send_plain(&room, &format!("Task not found: {issue_ref}")).await;
|
||||
}
|
||||
(_, Ok(None)) => {
|
||||
send_plain(&room, &format!("Agent not found: {agent_id}")).await;
|
||||
}
|
||||
(Err(e), _) | (_, Err(e)) => {
|
||||
send_plain(&room, &format!("Database error: {e}")).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
BotCommand::Retry { issue_ref } => {
|
||||
let task_status = {
|
||||
let store = ctx.store.lock().unwrap();
|
||||
store.read_task(&issue_ref).ok().flatten()
|
||||
};
|
||||
|
||||
let Some(task) = task_status else {
|
||||
send_plain(&room, &format!("Task not found: {issue_ref}")).await;
|
||||
return;
|
||||
};
|
||||
|
||||
if !matches!(task.status, TaskStatus::Failed | TaskStatus::AgentLost) {
|
||||
send_plain(&room, &format!("Task {issue_ref} is not in a retryable state (current: {})", task.status.as_str())).await;
|
||||
return;
|
||||
}
|
||||
|
||||
// Reset to created (re-enqueue)
|
||||
match ctx.sm.transition(&issue_ref, TaskStatus::Assigned, None, "retry via matrix").await {
|
||||
Ok(_) => {
|
||||
send_plain(&room, &format!("🔄 Task {issue_ref} re-queued for assignment")).await;
|
||||
}
|
||||
Err(e) => {
|
||||
send_plain(&room, &format!("❌ Failed to retry: {e}")).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
BotCommand::Unknown(cmd) => {
|
||||
send_plain(&room, &format!("Unknown command: `{cmd}`. Available: /fleet status, /assign <agent> <issue>, /retry <issue>")).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a plain text message to a room.
|
||||
async fn send_plain(room: &Room, body: &str) {
|
||||
let content = RoomMessageEventContent::text_plain(body);
|
||||
if let Err(e) = room.send(content).await {
|
||||
tracing::error!("Failed to send Matrix message: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a message as a thread reply (per-agent thread).
|
||||
pub async fn send_thread_message(
|
||||
room: &Room,
|
||||
thread_root_event_id: &ruma::OwnedEventId,
|
||||
body: &str,
|
||||
) {
|
||||
let mut content = RoomMessageEventContent::text_plain(body);
|
||||
content.relates_to = Some(matrix_sdk::ruma::events::room::message::Relation::Thread(
|
||||
Thread::without_fallback(thread_root_event_id.clone()),
|
||||
));
|
||||
if let Err(e) = room.send(content).await {
|
||||
tracing::error!("Failed to send thread message: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
/// Post a notification for a task event to the Matrix room.
|
||||
/// This is a helper meant to be called from the API layer.
|
||||
pub async fn notify_task_event(
|
||||
room: &Room,
|
||||
event_type: &str,
|
||||
task_id: &str,
|
||||
agent_id: &str,
|
||||
task_type: &str,
|
||||
summary: Option<&str>,
|
||||
error: Option<&str>,
|
||||
artifact_hint: Option<&str>,
|
||||
) {
|
||||
let msg = match event_type {
|
||||
"task.assigned" => format_task_assigned(task_id, agent_id, task_type),
|
||||
"task.completed" => format_task_completed(
|
||||
task_id,
|
||||
agent_id,
|
||||
summary.unwrap_or(""),
|
||||
artifact_hint,
|
||||
),
|
||||
"task.failed" => format_task_failed(
|
||||
task_id,
|
||||
agent_id,
|
||||
error.unwrap_or("unknown error"),
|
||||
),
|
||||
_ => return,
|
||||
};
|
||||
send_plain(room, &msg).await;
|
||||
}
|
||||
|
||||
/// Post an agent-offline alert to the Matrix room.
|
||||
pub async fn notify_agent_offline(room: &Room, agent_id: &str, affected_tasks: usize) {
|
||||
let msg = format_agent_offline(agent_id, affected_tasks);
|
||||
send_plain(room, &msg).await;
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::core::models::{Agent, AgentType, AgentStatus};
|
||||
use chrono::Utc;
|
||||
use std::collections::HashMap;
|
||||
|
||||
// ─── Command parsing tests ──────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn parse_fleet_status() {
|
||||
assert_eq!(parse_command("/fleet status"), Some(BotCommand::FleetStatus));
|
||||
assert_eq!(parse_command("fleet status"), Some(BotCommand::FleetStatus));
|
||||
assert_eq!(parse_command("/FLEET STATUS"), Some(BotCommand::FleetStatus));
|
||||
assert_eq!(parse_command(" /fleet status "), Some(BotCommand::FleetStatus));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_assign_command() {
|
||||
assert_eq!(
|
||||
parse_command("/assign worker-03 org/repo#42"),
|
||||
Some(BotCommand::Assign {
|
||||
agent_id: "worker-03".into(),
|
||||
issue_ref: "org/repo#42".into(),
|
||||
})
|
||||
);
|
||||
assert_eq!(
|
||||
parse_command("assign worker-03 org/repo#42"),
|
||||
Some(BotCommand::Assign {
|
||||
agent_id: "worker-03".into(),
|
||||
issue_ref: "org/repo#42".into(),
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_retry_command() {
|
||||
assert_eq!(
|
||||
parse_command("/retry org/repo#42"),
|
||||
Some(BotCommand::Retry {
|
||||
issue_ref: "org/repo#42".into(),
|
||||
})
|
||||
);
|
||||
assert_eq!(
|
||||
parse_command("retry #42"),
|
||||
Some(BotCommand::Retry {
|
||||
issue_ref: "#42".into(),
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_unknown_command() {
|
||||
// Non-fleet/assign/retry prefixes return None (not a bot command)
|
||||
assert_eq!(parse_command("/deploy prod"), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_non_command_returns_none() {
|
||||
assert_eq!(parse_command("hello world"), None);
|
||||
assert_eq!(parse_command(""), None);
|
||||
assert_eq!(parse_command("just chatting"), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_fleet_subcommand_unknown() {
|
||||
assert_eq!(
|
||||
parse_command("/fleet deploy"),
|
||||
Some(BotCommand::Unknown("fleet deploy".into()))
|
||||
);
|
||||
}
|
||||
|
||||
// ─── Notification formatting tests ──────────────────────────
|
||||
|
||||
#[test]
|
||||
fn task_assigned_format() {
|
||||
let msg = format_task_assigned("org/repo#42", "worker-03", "code:typescript");
|
||||
assert_eq!(msg, "📋 org/repo#42 → worker-03 [code:typescript]");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn task_completed_format_with_artifact() {
|
||||
let msg = format_task_completed(
|
||||
"org/repo#42",
|
||||
"worker-03",
|
||||
"修复登录验证 bug",
|
||||
Some("PR #15"),
|
||||
);
|
||||
assert_eq!(
|
||||
msg,
|
||||
"✅ org/repo#42 completed by worker-03 — PR #15 — \"修复登录验证 bug\""
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn task_completed_format_without_artifact() {
|
||||
let msg = format_task_completed(
|
||||
"org/repo#42",
|
||||
"worker-03",
|
||||
"fixed the thing",
|
||||
None,
|
||||
);
|
||||
assert_eq!(msg, "✅ org/repo#42 completed by worker-03 — \"fixed the thing\"");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn task_failed_format() {
|
||||
let msg = format_task_failed("org/repo#42", "worker-03", "构建超时");
|
||||
assert_eq!(msg, "❌ org/repo#42 failed — worker-03 — \"构建超时\"");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn agent_offline_format() {
|
||||
let msg = format_agent_offline("worker-03", 2);
|
||||
assert_eq!(msg, "⚠️ worker-03 offline — 2 running tasks affected");
|
||||
}
|
||||
|
||||
// ─── Fleet status table tests ───────────────────────────────
|
||||
|
||||
fn sample_agent(id: &str, status: AgentStatus, tasks: u32) -> Agent {
|
||||
Agent {
|
||||
agent_id: id.to_string(),
|
||||
agent_type: AgentType::CodexCli,
|
||||
hostname: "host-01".into(),
|
||||
capabilities: vec!["code:rust".into(), "review".into()],
|
||||
max_concurrency: 2,
|
||||
current_tasks: tasks,
|
||||
status,
|
||||
last_heartbeat_at: Utc::now(),
|
||||
registered_at: Utc::now(),
|
||||
metadata: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fleet_status_empty() {
|
||||
let table = format_fleet_status(&[]);
|
||||
assert_eq!(table, "No agents registered.");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fleet_status_with_agents() {
|
||||
let agents = vec![
|
||||
sample_agent("worker-01", AgentStatus::Online, 1),
|
||||
sample_agent("worker-02", AgentStatus::Offline, 0),
|
||||
];
|
||||
let table = format_fleet_status(&agents);
|
||||
assert!(table.contains("worker-01"));
|
||||
assert!(table.contains("worker-02"));
|
||||
assert!(table.contains("online"));
|
||||
assert!(table.contains("offline"));
|
||||
assert!(table.contains("code:rust"));
|
||||
assert!(table.contains("review"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fleet_status_table_has_header() {
|
||||
let agents = vec![sample_agent("w1", AgentStatus::Online, 0)];
|
||||
let table = format_fleet_status(&agents);
|
||||
assert!(table.contains("Agent ID"));
|
||||
assert!(table.contains("Type"));
|
||||
assert!(table.contains("Status"));
|
||||
assert!(table.contains("Tasks"));
|
||||
assert!(table.contains("Capabilities"));
|
||||
}
|
||||
}
|
||||
|
|
@ -1,2 +1 @@
|
|||
pub mod forgejo;
|
||||
pub mod matrix;
|
||||
|
|
|
|||
21
src/main.rs
21
src/main.rs
|
|
@ -87,10 +87,15 @@ async fn main() {
|
|||
|
||||
let app = axum::Router::new()
|
||||
.route("/healthz", axum::routing::get(|| async { "ok" }))
|
||||
// Agent registry
|
||||
.route("/api/v1/agents/register", axum::routing::post(api::register_agent))
|
||||
.route("/api/v1/agents/heartbeat", axum::routing::post(api::heartbeat))
|
||||
.route("/api/v1/agents/deregister", axum::routing::post(api::deregister))
|
||||
.route("/api/v1/agents", axum::routing::get(api::list_agents))
|
||||
// Task management
|
||||
.route("/api/v1/tasks", axum::routing::get(api::list_tasks))
|
||||
.route("/api/v1/tasks/{task_id}/retry", axum::routing::post(api::retry_task))
|
||||
// Receipts & webhooks
|
||||
.route("/api/v1/receipts", axum::routing::post(api::submit_receipt))
|
||||
.route(
|
||||
"/api/v1/webhooks/forgejo",
|
||||
|
|
@ -106,21 +111,5 @@ async fn main() {
|
|||
.expect("failed to bind");
|
||||
|
||||
tracing::info!("listening on {}", listener.local_addr().unwrap());
|
||||
|
||||
// Start Matrix bot
|
||||
if !config.matrix.access_token.is_empty() && !config.matrix.room_id.is_empty() {
|
||||
let matrix_cfg = config.matrix.clone();
|
||||
let matrix_store = store.clone();
|
||||
let matrix_sm = state_machine.clone();
|
||||
tokio::spawn(async move {
|
||||
match crate::integrations::matrix::start_bot(matrix_cfg, matrix_store, matrix_sm).await {
|
||||
Ok(_) => tracing::info!("Matrix bot stopped"),
|
||||
Err(e) => tracing::error!("Matrix bot error: {e}"),
|
||||
}
|
||||
});
|
||||
} else {
|
||||
tracing::info!("Matrix bot disabled (no access_token or room_id configured)");
|
||||
}
|
||||
|
||||
axum::serve(listener, app).await.expect("server error");
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue