rebranding herald => steep
ci/woodpecker/push/tests Pipeline was successful

This commit is contained in:
qpismont committed 2026-09-29 18:36:01 +00:00
1 parent 9f03384bc9
commit f5f0d3999e
30 files changed
+96 -93

No files matched your search

+28
View File
@@ -0,0 +1,28 @@
[package]
name = "steep-server"
version = "1.2.0"
edition = "2024"
[dependencies]
reqwest = { workspace = true }
tokio = { workspace = true }
tokio-util = { workspace = true }
futures-util = { workspace = true }
serde_json = { workspace = true }
serde = { workspace = true }
sentry = { workspace = true }
sentry-anyhow = { workspace = true }
dotenvy = { workspace = true }
tower = { workspace = true }
tower-http = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
axum = { workspace = true }
anyhow = { workspace = true }
thiserror = { workspace = true }
ring = { workspace = true }
hex = { workspace = true }
metrics = { workspace = true }
metrics-exporter-prometheus = { workspace = true }
devcontainer-rs = { path = "../devcontainer-rs" }
tempfile = "3"
+147
View File
@@ -0,0 +1,147 @@
use axum::body::{Bytes, to_bytes};
use axum::extract::{FromRef, FromRequest, State};
use axum::http::Request;
use axum::response::IntoResponse;
use axum::routing::{get, post};
use axum::{Json, Router};
use reqwest::StatusCode;
use ring::hmac;
use sentry::integrations::tower::{NewSentryLayer, SentryHttpLayer};
use serde_json::Value;
use tower::ServiceBuilder;
use tower_http::trace::TraceLayer;
use tracing::{info, instrument};
use tokio_util::sync::CancellationToken;
use crate::consts::{GITEA_EVENT_TYPE_HEADER_NAME, GITEA_SIG_HEADER_NAME, MAX_WEBHOOK_BODY_SIZE};
use crate::errors::AppError;
use crate::gitea::WebhookType;
use crate::metrics;
use crate::state::AppState;
pub async fn start(app_state: AppState, shutdown: CancellationToken) -> anyhow::Result<()> {
let http_port = app_state.config.http_port;
let app = Router::new()
.route("/", get(root))
.route("/webhook", post(webhook))
.layer(
ServiceBuilder::new()
.layer(NewSentryLayer::<Request<_>>::new_from_top())
.layer(SentryHttpLayer::new())
.layer(TraceLayer::new_for_http()),
)
.with_state(app_state);
let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{}", http_port)).await?;
info!("Listening API on port {}", http_port);
axum::serve(listener, app)
.with_graceful_shutdown(async move { shutdown.cancelled().await })
.await
.map_err(anyhow::Error::from)
.inspect(|_| info!("API shutting down complete"))?;
Ok(())
}
async fn root() -> &'static str {
"Hi, I'm Steep :)"
}
#[instrument(skip(app_state), fields(webhook_type), err)]
async fn webhook(
State(app_state): State<AppState>,
WebhookExtract(wb): WebhookExtract,
) -> Result<impl IntoResponse, AppError> {
tracing::Span::current().record("webhook_type", tracing::field::debug(&wb));
let event_type = wb.event_type_str();
metrics::webhook_received(event_type);
let event_id = wb.event_id();
if app_state.bot.check_and_mark(event_id).await {
metrics::webhook_duplicate(event_type);
return Err(AppError::AlreadyProcessedErr);
}
if app_state.bot_tx.try_send(wb).is_err() {
app_state.bot.unmark(event_id).await;
metrics::webhook_channel_full(event_type);
return Err(AppError::ChannelFullErr);
}
Ok((StatusCode::CREATED, "Task started"))
}
pub struct WebhookExtract(pub WebhookType);
impl<S> FromRequest<S> for WebhookExtract
where
AppState: FromRef<S>,
S: Send + Sync,
{
type Rejection = AppError;
#[instrument(skip(req, state), err)]
async fn from_request(req: axum::extract::Request, state: &S) -> Result<Self, Self::Rejection> {
let app_state = AppState::from_ref(state);
let headers = req.headers();
let sig_header = extract_header(GITEA_SIG_HEADER_NAME, headers)?;
let type_header = extract_header(GITEA_EVENT_TYPE_HEADER_NAME, headers)?;
let body_bytes = read_body(req.into_body()).await?;
verify_signature(
app_state.config.webhook_secret.as_bytes(),
&sig_header,
&body_bytes,
)?;
let body_str = String::from_utf8_lossy(&body_bytes).into_owned();
sentry::configure_scope(|scope| {
scope.add_event_processor(move |mut event| {
let mut request = event.request.take().unwrap_or_default();
request.data = Some(body_str.clone());
event.request = Some(request);
Some(event)
});
});
let webhook = parse_webhook(&type_header, &app_state.bot.name(), &body_bytes)?;
Ok(WebhookExtract(webhook))
}
}
fn extract_header(key: &str, headers: &axum::http::HeaderMap) -> Result<String, AppError> {
let value = headers
.get(key)
.ok_or(AppError::WebHookMissingHeaderErr(key.into()))?
.to_str()
.map_err(anyhow::Error::from)?;
Ok(value.to_owned())
}
async fn read_body(body: axum::body::Body) -> Result<Bytes, AppError> {
to_bytes(body, MAX_WEBHOOK_BODY_SIZE)
.await
.map_err(anyhow::Error::from)
.map_err(AppError::from)
}
fn parse_webhook(header: &str, bot_name: &str, body_bytes: &[u8]) -> Result<WebhookType, AppError> {
let Json(value) =
Json::<Value>::from_bytes(body_bytes).map_err(|_| AppError::MalformedJsonErr)?;
WebhookType::from_event(header, bot_name, value)
}
fn verify_signature(secret_key: &[u8], sig_header: &str, body: &[u8]) -> Result<(), AppError> {
let sig_header_decoded =
hex::decode(sig_header).map_err(|_| AppError::WebHookSigHeaderInvalidErr)?;
let key = hmac::Key::new(hmac::HMAC_SHA256, secret_key);
hmac::verify(&key, body, &sig_header_decoded).map_err(|_| AppError::WebHookSigHeaderInvalidErr)
}
+320
View File
@@ -0,0 +1,320 @@
use crate::{
gitea::{GiteaAPI, WebhookType},
metrics,
open_router::OpenRouterClient,
sandbox::SandboxConfig,
};
use serde::{Deserialize, Deserializer};
use std::{collections::HashSet, sync::Arc};
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken;
use tracing::{error, info, instrument};
#[derive(Deserialize, Debug)]
pub struct ReviewResult {
pub reviews: Vec<ReviewItem>,
pub comment: String,
pub cost: Option<f64>,
}
#[derive(Deserialize, Debug)]
pub struct ReviewItem {
pub filename: String,
pub line: Option<u64>,
#[serde(default, deserialize_with = "deserialize_side")]
pub side: Option<ReviewSide>,
#[serde(default, deserialize_with = "deserialize_severity")]
pub severity: Option<ReviewSeverity>,
pub message: String,
}
/// Which version of the file a review comment is anchored on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReviewSide {
/// The line was added by the pull request: `line` is a line number of the
/// new version of the file.
Added,
/// The line was removed by the pull request: `line` is a line number of the
/// old version of the file.
Removed,
}
impl ReviewSide {
/// Reads the side the model asked for, tolerating casing and synonyms.
fn parse(raw: &str) -> Option<Self> {
match raw.trim().to_ascii_lowercase().as_str() {
"added" | "add" | "new" | "right" => Some(Self::Added),
"removed" | "remove" | "deleted" | "delete" | "old" | "left" => Some(Self::Removed),
_ => None,
}
}
}
/// What kind of problem a review reports.
///
/// The categories are the ones the prompt asks for; a review whose severity is
/// missing or unreadable falls back to [`ReviewSeverity::Maintainability`], the
/// least alarming bucket, instead of failing the whole review.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReviewSeverity {
/// Wrong behaviour.
Bug,
/// A vulnerability.
Security,
/// A resource problem.
Performance,
/// Everything else: readability, structure, naming, tests.
Maintainability,
}
impl ReviewSeverity {
/// Reads the severity the model asked for, tolerating casing and synonyms.
fn parse(raw: &str) -> Option<Self> {
match raw.trim().to_ascii_lowercase().as_str() {
"bug" | "bugs" | "correctness" | "error" => Some(Self::Bug),
"security" | "vulnerability" => Some(Self::Security),
"performance" | "perf" => Some(Self::Performance),
"maintainability" | "quality" | "style" => Some(Self::Maintainability),
_ => None,
}
}
/// Lower-case label used in the review markdown.
pub fn label(self) -> &'static str {
match self {
Self::Bug => "bug",
Self::Security => "security",
Self::Performance => "performance",
Self::Maintainability => "maintainability",
}
}
}
/// Reads the side the model asked for. An unreadable value is ignored instead of
/// failing the whole review: the side is then resolved from the changed lines.
fn deserialize_side<'de, D>(deserializer: D) -> Result<Option<ReviewSide>, D::Error>
where
D: Deserializer<'de>,
{
let raw = Option::<serde_json::Value>::deserialize(deserializer)?;
Ok(raw
.as_ref()
.and_then(serde_json::Value::as_str)
.and_then(ReviewSide::parse))
}
/// Reads the severity the model asked for, ignoring an unreadable value: the
/// review is then reported as [`ReviewSeverity::Maintainability`].
fn deserialize_severity<'de, D>(deserializer: D) -> Result<Option<ReviewSeverity>, D::Error>
where
D: Deserializer<'de>,
{
let raw = Option::<serde_json::Value>::deserialize(deserializer)?;
Ok(raw
.as_ref()
.and_then(serde_json::Value::as_str)
.and_then(ReviewSeverity::parse))
}
#[derive(Clone)]
pub struct Bot {
bot_name: String,
gitea_api: GiteaAPI,
open_router_client: OpenRouterClient,
max_concurrent: usize,
open_router_model: String,
sandbox: SandboxConfig,
actions_handled: Arc<Mutex<HashSet<u64>>>,
}
impl Bot {
pub fn new(
bot_name: String,
gitea_api: GiteaAPI,
open_router_client: OpenRouterClient,
max_concurrent: usize,
open_router_model: String,
sandbox: SandboxConfig,
) -> Self {
Self {
bot_name,
gitea_api,
open_router_client,
max_concurrent,
open_router_model,
sandbox,
actions_handled: Arc::new(Mutex::new(HashSet::new())),
}
}
pub fn name(&self) -> String {
self.bot_name.clone()
}
pub async fn start(
&self,
mut rx: tokio::sync::mpsc::Receiver<WebhookType>,
shutdown: CancellationToken,
) -> anyhow::Result<()> {
info!("Bot started");
let sem = Arc::new(tokio::sync::Semaphore::new(self.max_concurrent));
let mut tasks = tokio::task::JoinSet::new();
loop {
let wb = tokio::select! {
biased;
_ = shutdown.cancelled() => break,
msg = rx.recv() => match msg {
Some(wb) => wb,
None => break,
},
};
while let Some(res) = tasks.try_join_next() {
if let Err(e) = res {
error!("Task panicked: {e}");
}
}
info!(queued = rx.len(), active = tasks.len(), "Webhook received");
let permit = sem.clone().acquire_owned().await?;
let self_clone = self.clone();
metrics::increment_task_active();
tasks.spawn(async move {
self_clone.exec(wb).await;
drop(permit);
metrics::decrement_task_active();
});
}
tasks.join_all().await;
info!("Bot shutting down complete");
Ok(())
}
#[instrument(skip(self, webhook), fields(repo, pr))]
pub async fn exec(&self, webhook: WebhookType) {
let event_type_str = webhook.event_type_str();
match &webhook {
WebhookType::Review(p) => {
tracing::Span::current().record("repo", &p.repository.full_name);
tracing::Span::current().record("pr", p.pull_request.number);
}
};
let tools = crate::sandbox::tools::for_webhook(&webhook);
let exec_result = match webhook {
WebhookType::Review(review_payload) => crate::bot_actions::review::exec_review(
&self.gitea_api,
&self.open_router_client,
&self.open_router_model,
&self.sandbox,
tools,
review_payload,
),
}
.await;
match exec_result {
Ok(_) => {
metrics::task_completed(event_type_str);
info!("Task completed");
}
Err(err) => {
metrics::task_failed(event_type_str);
error!(%err, "Task error");
sentry_anyhow::capture_anyhow(&err);
}
}
}
pub async fn check_and_mark(&self, event_id: u64) -> bool {
let mut action_handled_lock = self.actions_handled.lock().await;
!action_handled_lock.insert(event_id)
}
pub async fn unmark(&self, event_id: u64) {
let mut action_handled_lock = self.actions_handled.lock().await;
action_handled_lock.remove(&event_id);
}
}
#[cfg(test)]
mod tests {
use super::*;
fn make_actions_handled() -> Arc<Mutex<HashSet<u64>>> {
Arc::new(Mutex::new(HashSet::new()))
}
async fn check_and_mark(actions_handled: &Arc<Mutex<HashSet<u64>>>, event_id: u64) -> bool {
let mut lock = actions_handled.lock().await;
!lock.insert(event_id)
}
async fn unmark(actions_handled: &Arc<Mutex<HashSet<u64>>>, event_id: u64) {
let mut lock = actions_handled.lock().await;
lock.remove(&event_id);
}
#[tokio::test]
async fn test_check_and_mark_first_call_returns_false() {
let actions_handled = make_actions_handled();
assert!(!check_and_mark(&actions_handled, 1).await);
}
#[tokio::test]
async fn test_check_and_mark_second_call_returns_true() {
let actions_handled = make_actions_handled();
check_and_mark(&actions_handled, 1).await;
assert!(check_and_mark(&actions_handled, 1).await);
}
#[tokio::test]
async fn test_check_and_mark_different_ids_are_independent() {
let actions_handled = make_actions_handled();
check_and_mark(&actions_handled, 1).await;
assert!(!check_and_mark(&actions_handled, 2).await);
}
#[tokio::test]
async fn test_unmark_allows_reprocessing() {
let actions_handled = make_actions_handled();
check_and_mark(&actions_handled, 1).await;
unmark(&actions_handled, 1).await;
assert!(!check_and_mark(&actions_handled, 1).await);
}
#[tokio::test]
async fn test_unmark_nonexistent_id_is_noop() {
let actions_handled = make_actions_handled();
unmark(&actions_handled, 99).await;
assert!(!check_and_mark(&actions_handled, 99).await);
}
#[tokio::test]
async fn test_check_and_mark_concurrent_same_id() {
let actions_handled = make_actions_handled();
let actions_handled2 = Arc::clone(&actions_handled);
let t1 = tokio::spawn(async move { check_and_mark(&actions_handled, 42).await });
let t2 = tokio::spawn(async move { check_and_mark(&actions_handled2, 42).await });
let (r1, r2) = tokio::join!(t1, t2);
let results = [r1.unwrap(), r2.unwrap()];
// exactement un seul des deux doit retourner false (non traité)
assert_eq!(results.iter().filter(|&&r| !r).count(), 1);
// l'autre doit retourner true (déjà traité)
assert_eq!(results.iter().filter(|&&r| r).count(), 1);
}
}
@@ -0,0 +1 @@
pub mod review;
File diff suppressed because it is too large. Load diff
+82
View File
@@ -0,0 +1,82 @@
pub const GITEA_SIG_HEADER_NAME: &str = "x-gitea-signature";
pub const GITEA_EVENT_TYPE_HEADER_NAME: &str = "x-gitea-event-type";
pub const MAX_WEBHOOK_BODY_SIZE: usize = 1024 * 1024; // 1 MiB
pub const MAX_DIFF_SIZE: usize = 1024 * 1024; // 1 MiB
pub const BOT_PROCESS_MSG: &str = "
Review in progress with the model \"{model}\"...
";
pub const SANDBOX_SYSTEM_PROMPT: &str = "
You are a senior software engineer reviewing a pull request inside an
isolated sandbox.
The repository is checked out in your working directory. Explore it with the
provided read-only tools (ls, file_size, read_file, grep, find); paths are
relative to the repository root. Every line you read is prefixed with its
absolute line number: that is the number you must cite to anchor a comment.
Read before you assert: a review anchored on a line you did not read is
worthless, so never comment on code you have not seen. Gather enough context
to be confident, but do not re-read what you already have.
When you answer, send the requested JSON only: a raw JSON object, no markdown
code fence, nothing before or after it.
";
pub const REVIEW_PROMPT: &str = "
You are a senior software engineer reviewing a pull request.
Judge the changes for correctness, security, resource use and
maintainability. Report real problems; do not invent issues, and do not
report pure formatting a formatter would fix.
Be exhaustive: one review per distinct issue, and cover every changed file
that has something to report. Do not stop at the first few findings, and do
not merge several issues into a single review.
This is the pull request subject: \"{subject}\"
This is the user comment: \"{comment}\"
If the user asks about something precise, address that first.
The pull request changes these files and lines:
{changes}
The code is not provided. Read the files you need with the available tools
before answering.
`added` line numbers refer to the new version of the file, which is what your
working directory contains now. `removed` line numbers refer to the old
version, which is not in your working directory: you cannot read a removed
line, only the code around where it used to be.
Every review must anchor on one of the listed line numbers:
- filename: the full path exactly as listed above,
- line: one of the numbers listed for that file,
- side: \"added\" for a number from the `added` list, \"removed\" for one from
the `removed` list,
- severity: exactly one of \"bug\", \"security\", \"performance\" or
\"maintainability\" — a bug is wrong behaviour, security a vulnerability,
performance a resource problem, maintainability everything else.
Answer in french, with the raw json object only: no markdown code fence, no
text before or after it. All fields are mandatory. The `comment` field must
hold a short summary that lists the issues, and also what the pull request
does well: the author should get the compliments too.
{
\"reviews\": [
{
\"filename\": \"\",
\"line\": 0,
\"side\": \"\",
\"severity\": \"\",
\"message\": \"\"
}
],
\"comment\": \"\"
}
";
+88
View File
@@ -0,0 +1,88 @@
use anyhow::anyhow;
#[derive(Clone)]
pub struct EnvConfig {
pub http_port: u16,
pub webhook_secret: String,
pub open_router_api_key: String,
pub open_router_model: String,
pub open_router_timeout: u64,
pub bot_max_concurrent: usize,
pub gitea_url: String,
pub gitea_token: String,
pub gitea_timeout: u64,
pub metrics_bind_addr: Option<String>,
pub sandbox_max_iterations: usize,
}
pub fn load_config() -> anyhow::Result<EnvConfig> {
let http_port = try_get_env("HTTP_PORT")?.parse()?;
let webhook_secret = try_get_env("WEBHOOK_SIG_HEADER_SECRET")?;
let open_router_api_key = try_get_env("OPEN_ROUTER_API_KEY")?;
let open_router_model = try_get_env("OPEN_ROUTER_MODEL")?;
let open_router_timeout = try_get_env("OPEN_ROUTER_TIMEOUT")?.parse()?;
let bot_max_concurrent = try_get_env("BOT_MAX_CONCURRENT")?.parse()?;
let gitea_url = try_get_env("GITEA_URL")?;
let gitea_token = try_get_env("GITEA_TOKEN")?;
let gitea_timeout = try_get_env("GITEA_TIMEOUT")?.parse()?;
let metrics_bind_addr = std::env::var("METRICS_BIND_ADDR").ok();
let sandbox_max_iterations = std::env::var("SANDBOX_MAX_ITERATIONS")
.ok()
.and_then(|value| value.parse().ok())
.unwrap_or(8);
Ok(EnvConfig {
http_port,
webhook_secret,
open_router_api_key,
open_router_model,
open_router_timeout,
bot_max_concurrent,
gitea_url,
gitea_token,
gitea_timeout,
metrics_bind_addr,
sandbox_max_iterations,
})
}
pub fn try_get_env(key: &str) -> anyhow::Result<String> {
let env_value = std::env::var(key).map_err(|e| anyhow::anyhow!("{}: {}", key, e))?;
if env_value.trim().is_empty() {
return Err(anyhow!(format!("env var {} is empty", key)));
}
Ok(env_value)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_try_get_env_returns_value() {
unsafe { std::env::set_var("TEST_ENV_PRESENT", "hello") };
assert_eq!(try_get_env("TEST_ENV_PRESENT").unwrap(), "hello");
}
#[test]
fn test_try_get_env_missing_var_returns_error() {
unsafe { std::env::remove_var("TEST_ENV_MISSING") };
assert!(try_get_env("TEST_ENV_MISSING").is_err());
}
#[test]
fn test_try_get_env_empty_value_returns_error() {
unsafe { std::env::set_var("TEST_ENV_EMPTY", "") };
let err = try_get_env("TEST_ENV_EMPTY").unwrap_err();
assert!(err.to_string().contains("TEST_ENV_EMPTY"));
}
#[test]
fn test_try_get_env_whitespace_only_returns_error() {
unsafe { std::env::set_var("TEST_ENV_WHITESPACE", " ") };
let err = try_get_env("TEST_ENV_WHITESPACE").unwrap_err();
assert!(err.to_string().contains("TEST_ENV_WHITESPACE"));
}
}
+82
View File
@@ -0,0 +1,82 @@
use anyhow::anyhow;
use axum::response::IntoResponse;
use reqwest::StatusCode;
#[derive(thiserror::Error, Debug)]
pub enum AppError {
#[error("Unauthorized user id")]
UnauthorizedUserErr,
#[error("Unknow gitea event")]
UnknownEventErr,
#[error("Malformed Json")]
MalformedJsonErr,
#[error("WebHook header not found")]
WebHookMissingHeaderErr(String),
#[error("WebHook sig header is invalid")]
WebHookSigHeaderInvalidErr,
#[error("WebHook have bad action")]
InvalidActionErr,
#[error("Channel full")]
ChannelFullErr,
#[error("Already processed")]
AlreadyProcessedErr,
#[error(transparent)]
BadJsonStructErr(#[from] serde_json::Error),
#[error(transparent)]
Other(#[from] anyhow::Error),
}
impl IntoResponse for AppError {
fn into_response(self) -> axum::response::Response {
match self {
AppError::InvalidActionErr => (
StatusCode::UNPROCESSABLE_ENTITY,
"WebHook have bad action".to_string(),
),
AppError::UnknownEventErr => {
(StatusCode::BAD_REQUEST, "Unknow gitea event".to_string())
}
AppError::UnauthorizedUserErr => (
StatusCode::UNAUTHORIZED,
"Unauthorized user name".to_string(),
),
AppError::MalformedJsonErr => (StatusCode::BAD_REQUEST, "Malformed Json".to_string()),
AppError::BadJsonStructErr(err) => (
StatusCode::BAD_REQUEST,
format!("Json not contains mandatory fields: {}", err),
),
AppError::WebHookMissingHeaderErr(h) => {
(StatusCode::BAD_REQUEST, format!("header {} is missing", h))
}
AppError::WebHookSigHeaderInvalidErr => (
StatusCode::UNAUTHORIZED,
"WebHook sig header is invalid".to_string(),
),
AppError::AlreadyProcessedErr => (StatusCode::OK, "Already processed".to_string()),
AppError::ChannelFullErr => {
sentry_anyhow::capture_anyhow(&anyhow!("Max concurrent tasks reached"));
(
StatusCode::SERVICE_UNAVAILABLE,
"Max concurrent tasks reached".to_string(),
)
}
AppError::Other(err) => {
sentry_anyhow::capture_anyhow(&err);
(
StatusCode::INTERNAL_SERVER_ERROR,
"Internal server error".to_string(),
)
}
}
.into_response()
}
}
+571
View File
@@ -0,0 +1,571 @@
use std::time::Duration;
use futures_util::stream::TryStreamExt;
use serde::Deserialize;
use serde_json::{Value, json};
use tokio::io::AsyncReadExt;
use tokio_util::io::StreamReader;
use tracing::{instrument, warn};
use crate::{
bot::{ReviewResult, ReviewSeverity, ReviewSide},
consts::MAX_DIFF_SIZE,
errors::AppError,
};
/// Page size requested when listing the files of a pull request.
const FILE_PAGE_SIZE: u64 = 50;
/// Maximum number of pages fetched for a pull request file list.
const MAX_FILE_PAGES: u64 = 10;
#[derive(Clone)]
pub struct GiteaAPI {
base_url: String,
token: String,
client: reqwest::Client,
}
impl GiteaAPI {
pub fn new(base_url: &str, token: &str, timeout: u64) -> anyhow::Result<Self> {
let mut default_headers = reqwest::header::HeaderMap::new();
default_headers.insert(
reqwest::header::HeaderName::from_static("authorization"),
reqwest::header::HeaderValue::from_str(&format!("Bearer {}", token))?,
);
Ok(Self {
base_url: String::from(base_url),
token: String::from(token),
client: reqwest::Client::builder()
.timeout(Duration::from_secs(timeout))
.default_headers(default_headers)
.build()?,
})
}
/// API token used to authenticate against Gitea.
pub fn token(&self) -> &str {
&self.token
}
#[instrument(skip(self))]
pub async fn get_authorized_user(&self) -> anyhow::Result<User> {
let url = format!("{}/api/v1/user", self.base_url);
let res = self.client.get(url).send().await?;
if !res.status().is_success() {
return Err(anyhow::anyhow!(
"Failed to get authorized user: {}",
res.status()
));
}
res.json::<User>().await.map_err(anyhow::Error::from)
}
#[instrument(skip(self))]
pub async fn comment(
&self,
body: &str,
full_name: &str,
index: u64,
) -> anyhow::Result<Comment> {
let url = format!(
"{}/api/v1/repos/{}/issues/{}/comments",
self.base_url, full_name, index
);
let res = self
.client
.post(url)
.json(&json!({
"body": body
}))
.send()
.await?;
if !res.status().is_success() {
return Err(anyhow::anyhow!("Failed to comment: {}", res.status()));
}
res.json::<Comment>().await.map_err(anyhow::Error::from)
}
#[instrument(skip(self))]
pub async fn edit_comment(
&self,
body: &str,
full_name: &str,
comment_id: u64,
) -> anyhow::Result<()> {
let url = format!(
"{}/api/v1/repos/{}/issues/comments/{}",
self.base_url, full_name, comment_id
);
let res = self
.client
.patch(url)
.json(&json!({
"body": body
}))
.send()
.await?;
if !res.status().is_success() {
return Err(anyhow::anyhow!("Failed to comment: {}", res.status()));
}
Ok(())
}
#[instrument(skip(self))]
pub async fn delete_comment(&self, full_name: &str, comment_id: u64) -> anyhow::Result<()> {
let url = format!(
"{}/api/v1/repos/{}/issues/comments/{}",
self.base_url, full_name, comment_id
);
let res = self.client.delete(url).send().await?;
if !res.status().is_success() {
return Err(anyhow::anyhow!(
"Failed to delete comment: {}",
res.status()
));
}
Ok(())
}
/// Raw unified diff of a pull request.
///
/// The API endpoint is used rather than the `diff_url` carried by the
/// webhook: that one points at a web route, which is session authenticated
/// and therefore does not serve private repositories to an API token.
#[instrument(skip(self))]
pub async fn pull_request_diff(&self, full_name: &str, index: u64) -> anyhow::Result<String> {
let url = format!(
"{}/api/v1/repos/{}/pulls/{}.diff",
self.base_url, full_name, index
);
let res = self.client.get(url).send().await?;
if !res.status().is_success() {
return Err(anyhow::anyhow!(
"Failed to download pull request diff: {}",
res.status()
));
}
read_capped(res).await
}
/// Files changed by a pull request, with their exact path and status.
#[instrument(skip(self))]
pub async fn pull_request_files(
&self,
full_name: &str,
index: u64,
) -> anyhow::Result<Vec<PullRequestFile>> {
let mut files = Vec::new();
for page in 1..=MAX_FILE_PAGES {
let url = format!(
"{}/api/v1/repos/{}/pulls/{}/files?limit={FILE_PAGE_SIZE}&page={page}",
self.base_url, full_name, index
);
let res = self.client.get(url).send().await?;
if !res.status().is_success() {
return Err(anyhow::anyhow!(
"Failed to list pull request files: {}",
res.status()
));
}
let page_files = res.json::<Vec<PullRequestFile>>().await?;
// The instance may cap the page size below the requested one, so a
// short page is not the end of the list: an empty one is.
if page_files.is_empty() {
return Ok(files);
}
files.extend(page_files);
}
warn!(files = files.len(), "Pull request file list was truncated");
Ok(files)
}
#[instrument(skip(self, review_result))]
pub async fn post_pull_request_review(
&self,
review_result: &ReviewResult,
final_comment: &str,
full_name: &str,
index: u64,
) -> anyhow::Result<()> {
let url = format!(
"{}/api/v1/repos/{}/pulls/{}/reviews",
self.base_url, full_name, index
);
let comments = review_result
.reviews
.iter()
.filter_map(|review| {
let line = review.line?;
let path = review.filename.clone();
let severity = review
.severity
.unwrap_or(ReviewSeverity::Maintainability)
.label();
let body = format!("**[{severity}]** {}", review.message);
// A line removed by the pull request only exists in the old
// version of the file, so it is anchored with `old_position`.
Some(match review.side {
Some(ReviewSide::Removed) => json!({
"path": path,
"old_position": line,
"body": body
}),
_ => json!({
"path": path,
"new_position": line,
"body": body
}),
})
})
.collect::<Vec<_>>();
let res = self
.client
.post(url)
.json(&json!({
"event": "COMMENT",
"body": final_comment,
"comments": comments
}))
.send()
.await?;
if !res.status().is_success() {
return Err(anyhow::anyhow!("Failed to post review: {}", res.status()));
}
Ok(())
}
}
/// Reads a response body, refusing to buffer more than [`MAX_DIFF_SIZE`].
async fn read_capped(response: reqwest::Response) -> anyhow::Result<String> {
let stream = response.bytes_stream().map_err(std::io::Error::other);
let mut buf = Vec::with_capacity(MAX_DIFF_SIZE);
StreamReader::new(stream)
.take((MAX_DIFF_SIZE + 1) as u64)
.read_to_end(&mut buf)
.await?;
if buf.len() > MAX_DIFF_SIZE {
anyhow::bail!("Pull request diff exceeds the maximum allowed size of 1 MiB");
}
Ok(String::from_utf8_lossy(&buf).into_owned())
}
#[derive(Debug)]
pub enum WebhookType {
Review(ReviewPayload),
}
impl WebhookType {
pub fn event_id(&self) -> u64 {
match self {
WebhookType::Review(payload) => payload.comment.id,
}
}
pub fn event_type_str(&self) -> &'static str {
match self {
WebhookType::Review(_) => "review",
}
}
}
#[derive(Deserialize, Debug)]
pub struct ReviewPayload {
pub action: String,
pub pull_request: PullRequest,
pub repository: Repository,
pub comment: Comment,
}
#[derive(Deserialize, Debug)]
pub struct PullRequest {
pub number: u64,
pub title: String,
}
#[derive(Deserialize, Debug)]
pub struct Comment {
pub id: u64,
pub body: String,
}
#[derive(Deserialize, Debug)]
pub struct User {
pub login: String,
}
#[derive(Deserialize, Debug)]
pub struct Repository {
pub full_name: String,
pub clone_url: String,
}
/// A file changed by a pull request, as reported by the API.
#[derive(Deserialize, Debug)]
pub struct PullRequestFile {
/// Path of the file in the new version of the repository.
pub filename: String,
/// Previous path, for a renamed file.
#[serde(default)]
pub previous_filename: Option<String>,
/// `added`, `modified`, `deleted`, `renamed`…
pub status: String,
}
impl WebhookType {
pub fn from_event(event: &str, bot_name: &str, json: Value) -> Result<Self, AppError> {
let wb = match event {
"pull_request_comment" => Ok(WebhookType::Review(serde_json::from_value(json)?)),
_ => Err(AppError::UnknownEventErr),
}?;
let action = match &wb {
WebhookType::Review(review_payload) => &review_payload.action,
};
if action != "created" {
return Err(AppError::InvalidActionErr);
}
let pr_body = match &wb {
WebhookType::Review(review_payload) => &review_payload.comment.body,
};
if !pr_body.starts_with(&format!("@{}", bot_name)) {
return Err(AppError::UnauthorizedUserErr);
}
Ok(wb)
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn test_from_event_valid_pull_request_comment() {
let json = json!({
"action": "created",
"pull_request": {
"id": 42,
"diff_url": "https://mydiff.fr",
"number": 1,
"title": "My PR"
},
"repository": {
"full_name": "owner/repo",
"clone_url": "https://github.com/owner/repo.git"
},
"comment": {
"id": 7,
"body": "@test_bot LGTM",
"user": {
"id": 100,
"login": "test_user"
}
}
});
let result = WebhookType::from_event("pull_request_comment", "test_bot", json);
assert!(result.is_ok());
match result.unwrap() {
WebhookType::Review(payload) => {
assert_eq!(payload.action, "created");
assert_eq!(payload.comment.id, 7);
assert_eq!(payload.comment.body, "@test_bot LGTM");
}
}
}
#[test]
fn test_from_event_unknown_event() {
let json = json!({});
let result = WebhookType::from_event("push", "test_bot", json);
assert!(result.is_err());
match result.unwrap_err() {
AppError::UnknownEventErr => {}
_ => panic!("expected UnknownEventErr"),
}
}
#[test]
fn test_from_event_malformed_json() {
let json = json!({
"action": "created"
// pull_request and comment are missing
});
let result = WebhookType::from_event("pull_request_comment", "test_bot", json);
assert!(result.is_err());
match result.unwrap_err() {
AppError::BadJsonStructErr(_) => {}
_ => panic!("expected BadJsonStructErr"),
}
}
#[test]
fn test_from_event_rejects_non_created_action() {
let json = json!({
"action": "edited",
"pull_request": {
"id": 1,
"diff_url": "https://mydiff.fr",
"number": 1,
"title": "My PR"
},
"repository": {
"full_name": "owner/repo",
"clone_url": "https://github.com/owner/repo.git"
},
"comment": {
"id": 1,
"body": "@test_bot body",
"user": {
"id": 1,
"login": "test_user"
}
}
});
let result = WebhookType::from_event("pull_request_comment", "test_bot", json);
assert!(result.is_err());
match result.unwrap_err() {
AppError::InvalidActionErr => {}
_ => panic!("expected InvalidActionErr"),
}
}
#[test]
fn test_deserialize_review_payload() {
let json = json!({
"action": "created",
"pull_request": {
"id": 99,
"diff_url": "https://mydiff.fr",
"number": 5,
"title": "My PR"
},
"repository": {
"full_name": "owner/repo",
"clone_url": "https://github.com/owner/repo.git"
},
"comment": {
"id": 12,
"body": "Needs work",
"user": {
"id": 200,
"login": "test_user"
}
}
});
let payload: ReviewPayload = serde_json::from_value(json).unwrap();
assert_eq!(payload.action, "created");
assert_eq!(payload.comment.id, 12);
assert_eq!(payload.comment.body, "Needs work");
assert_eq!(
payload.repository.clone_url,
"https://github.com/owner/repo.git"
);
}
#[test]
fn test_from_event_empty_json() {
let result = WebhookType::from_event("pull_request_comment", "test_bot", json!({}));
assert!(result.is_err());
assert!(matches!(result.unwrap_err(), AppError::BadJsonStructErr(_)));
}
#[test]
fn test_from_event_rejects_wrong_bot_name() {
let json = json!({
"action": "created",
"pull_request": {
"id": 1,
"diff_url": "https://mydiff.fr",
"number": 1,
"title": "My PR"
},
"repository": {
"full_name": "owner/repo",
"clone_url": "https://github.com/owner/repo.git"
},
"comment": {
"id": 1,
"body": "@other_bot do something",
"user": {
"id": 1,
"login": "test_user"
}
}
});
let result = WebhookType::from_event("pull_request_comment", "test_bot", json);
assert!(matches!(result.unwrap_err(), AppError::UnauthorizedUserErr));
}
#[test]
fn test_from_event_rejects_no_bot_prefix() {
let json = json!({
"action": "created",
"pull_request": {
"id": 1,
"diff_url": "https://mydiff.fr",
"number": 1,
"title": "My PR"
},
"repository": {
"full_name": "owner/repo",
"clone_url": "https://github.com/owner/repo.git"
},
"comment": {
"id": 1,
"body": "just a comment without bot mention",
"user": {
"id": 1,
"login": "test_user"
}
}
});
let result = WebhookType::from_event("pull_request_comment", "test_bot", json);
assert!(matches!(result.unwrap_err(), AppError::UnauthorizedUserErr));
}
}
+157
View File
@@ -0,0 +1,157 @@
use std::time::Duration;
use crate::{
bot::Bot,
gitea::{GiteaAPI, WebhookType},
open_router::OpenRouterClient,
sandbox::SandboxConfig,
state::AppState,
};
use dotenvy::dotenv;
use tokio::signal::unix::{SignalKind, signal};
use tokio_util::sync::CancellationToken;
use tracing::{info, warn};
use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt};
mod api;
mod bot;
mod bot_actions;
mod consts;
mod env;
mod errors;
mod gitea;
mod metrics;
mod open_router;
mod sandbox;
mod state;
mod text;
/// Délai laissé aux reviews en cours et aux serveurs pour s'arrêter proprement.
///
/// Sans lui, une review prise dans une sandbox récalcitrante garderait le processus
/// en vie jusqu'à ce que le superviseur le tue.
const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(60);
fn main() -> anyhow::Result<()> {
dotenv().ok();
tracing_subscriber::registry()
.with(fmt::layer())
.with(
EnvFilter::try_from_default_env() // lit RUST_LOG depuis l'env
.unwrap_or_else(|_| EnvFilter::new("info")),
)
.init();
let _sentry_guard = if let Ok(sentry_dsn) = env::try_get_env("SENTRY_DSN") {
info!("Initialize sentry");
Some(sentry::init(
sentry::ClientOptions::new()
.dsn(&sentry_dsn)
.maybe_release(sentry::release_name!())
.send_default_pii(true),
))
} else {
warn!("SENTRY_DSN not set, sentry will not be initialized");
None
};
tokio::runtime::Runtime::new()?.block_on(run())
}
async fn run() -> anyhow::Result<()> {
let config = env::load_config()?;
if let Some(metric_bind_addr) = &config.metrics_bind_addr {
metrics::install(metric_bind_addr)?;
}
let gitea_api = GiteaAPI::new(&config.gitea_url, &config.gitea_token, config.gitea_timeout)?;
let gitea_user = gitea_api.get_authorized_user().await?;
info!(
port = config.http_port,
model = %config.open_router_model,
gitea_url = %config.gitea_url,
bot_name = %gitea_user.login,
"Starting Steep"
);
let open_router_client = OpenRouterClient::new(
&config.open_router_api_key,
&config.open_router_model,
config.open_router_timeout,
)?;
let shutdown = CancellationToken::new();
let sandbox = SandboxConfig {
runtime: devcontainer_rs::ContainerRuntime::connect()?,
max_iterations: config.sandbox_max_iterations,
};
if !sandbox.runtime.available().await {
warn!(
endpoint = sandbox.runtime.endpoint(),
"Container daemon is not reachable, every review will fail"
);
}
let bot = Bot::new(
gitea_user.login,
gitea_api,
open_router_client,
config.bot_max_concurrent,
config.open_router_model.clone(),
sandbox,
);
let (tx, rx) = tokio::sync::mpsc::channel::<WebhookType>(config.bot_max_concurrent * 2);
let app_state = AppState {
bot_tx: tx,
bot: bot.clone(),
config,
};
let signal = async {
let mut sigterm = signal(SignalKind::terminate())?;
let mut sigint = signal(SignalKind::interrupt())?;
tokio::select! {
_ = sigterm.recv() => info!("Received SIGTERM"),
_ = sigint.recv() => info!("Received SIGINT"),
}
info!("Shutting down...");
shutdown.cancel();
anyhow::Ok(())
};
let shutdown_deadline = async {
shutdown.cancelled().await;
tokio::time::sleep(SHUTDOWN_TIMEOUT).await;
};
tokio::select! {
result = async {
tokio::try_join!(
bot.start(rx, shutdown.clone()),
api::start(app_state, shutdown.clone()),
signal
)
} => {
result?;
}
() = shutdown_deadline => {
warn!(
timeout = ?SHUTDOWN_TIMEOUT,
"Shutdown did not finish in time, exiting anyway"
);
}
}
info!("Shutdown complete");
Ok(())
}
+87
View File
@@ -0,0 +1,87 @@
use std::{net::SocketAddr, str::FromStr};
use metrics::{Unit, counter, describe_counter, describe_gauge, gauge};
pub fn webhook_received(event_type: &str) {
counter!("steep_webhooks_received_total", "event_type" => event_type.to_string()).increment(1);
}
pub fn webhook_duplicate(event_type: &str) {
counter!("steep_webhooks_duplicate_total", "event_type" => event_type.to_string()).increment(1);
}
pub fn webhook_channel_full(event_type: &str) {
counter!("steep_webhooks_channel_full_total", "event_type" => event_type.to_string())
.increment(1);
}
pub fn increment_task_active() {
gauge!("steep_bot_tasks_active").increment(1.0);
}
pub fn decrement_task_active() {
gauge!("steep_bot_tasks_active").decrement(1.0);
}
pub fn task_completed(event_type: &str) {
counter!("steep_bot_tasks_completed_total", "event_type" => event_type.to_string())
.increment(1);
}
pub fn task_failed(event_type: &str) {
counter!("steep_bot_tasks_failed_total", "event_type" => event_type.to_string()).increment(1);
}
pub fn openrouter_cost_usd(cost: f64) {
counter!("steep_openrouter_cost_cents_total").increment((cost * 100.0).round() as u64);
}
pub fn describe() {
describe_counter!(
"steep_webhooks_received_total",
Unit::Count,
"Total webhooks received"
);
describe_counter!(
"steep_webhooks_duplicate_total",
Unit::Count,
"Webhooks rejected as duplicates"
);
describe_counter!(
"steep_webhooks_channel_full_total",
Unit::Count,
"Webhooks dropped because the bot channel was full"
);
describe_gauge!(
"steep_bot_tasks_active",
Unit::Count,
"Bot tasks currently in progress"
);
describe_counter!(
"steep_bot_tasks_completed_total",
Unit::Count,
"Bot tasks completed successfully"
);
describe_counter!(
"steep_bot_tasks_failed_total",
Unit::Count,
"Bot tasks that failed"
);
describe_counter!(
"steep_openrouter_cost_cents_total",
Unit::Count,
"Total OpenRouter cost in cents (divide by 100 for USD)"
);
}
pub fn install(bind_addr: &str) -> anyhow::Result<()> {
describe();
let builder = metrics_exporter_prometheus::PrometheusBuilder::new();
builder
.with_http_listener(SocketAddr::from_str(bind_addr)?)
.install()?;
tracing::info!(bind_addr, "Prometheus metrics exporter installed");
Ok(())
}
+473
View File
@@ -0,0 +1,473 @@
//! Minimal OpenRouter chat-completions client.
//!
//! Steep only needs a non-streaming `POST /chat/completions` with optional
//! tool calling, so the wire types are implemented in-tree instead of pulling a
//! third-party SDK (and its own `reqwest` version) into the workspace.
//!
//! Only the response fields Steep consumes are modelled: `content`,
//! `tool_calls` and `usage.cost`. Unknown fields are ignored.
use std::time::Duration;
use anyhow::Context;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tracing::instrument;
/// OpenRouter API root, version prefix included.
const BASE_URL: &str = "https://openrouter.ai/api/v1";
/// What the model is allowed to do on a given turn.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ToolChoice {
/// The model decides whether to call a tool.
Auto,
/// The model must answer, without calling any tool. The tools stay declared,
/// so the conversation remains the same shape as on the other turns.
None,
}
impl ToolChoice {
fn as_str(self) -> &'static str {
match self {
Self::Auto => "auto",
Self::None => "none",
}
}
}
/// Result of a completion that may contain tool calls.
pub struct ToolChatResult {
pub message: Option<String>,
pub tool_calls: Vec<ToolCall>,
pub cost: Option<f64>,
}
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum Role {
System,
User,
Assistant,
Tool,
}
/// A single turn of the conversation.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Message {
pub role: Role,
#[serde(skip_serializing_if = "Option::is_none")]
pub content: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tool_calls: Option<Vec<ToolCall>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tool_call_id: Option<String>,
}
impl Message {
pub fn new(role: Role, content: impl Into<String>) -> Self {
Self {
role,
content: Some(content.into()),
tool_calls: None,
tool_call_id: None,
}
}
/// Assistant turn requesting tool calls. Models often answer with tool
/// calls but no text, in which case the content is sent as `null`.
pub fn assistant_with_tool_calls(
content: impl Into<String>,
tool_calls: Vec<ToolCall>,
) -> Self {
let content = content.into();
Self {
role: Role::Assistant,
content: (!content.is_empty()).then_some(content),
tool_calls: Some(tool_calls),
tool_call_id: None,
}
}
/// Result of a tool call, linked to the request by `tool_call_id`.
pub fn tool_response(tool_call_id: impl Into<String>, content: impl Into<String>) -> Self {
Self {
role: Role::Tool,
content: Some(content.into()),
tool_calls: None,
tool_call_id: Some(tool_call_id.into()),
}
}
}
/// A tool the model may call.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Tool {
#[serde(rename = "type", default = "function_type")]
pub kind: String,
pub function: FunctionDefinition,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct FunctionDefinition {
pub name: String,
pub description: String,
/// JSON schema describing the accepted arguments.
pub parameters: Value,
}
impl Tool {
pub fn new(name: impl Into<String>, description: impl Into<String>, parameters: Value) -> Self {
Self {
kind: function_type(),
function: FunctionDefinition {
name: name.into(),
description: description.into(),
parameters,
},
}
}
}
/// A tool call requested by the model.
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct ToolCall {
pub id: String,
#[serde(rename = "type", default = "function_type")]
pub kind: String,
pub function: FunctionCall,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct FunctionCall {
pub name: String,
/// Arguments as a JSON-encoded string. Kept verbatim so that echoing the
/// call back into the conversation does not re-encode or corrupt it.
pub arguments: String,
}
impl ToolCall {
pub fn id(&self) -> &str {
&self.id
}
pub fn name(&self) -> &str {
&self.function.name
}
pub fn arguments_json(&self) -> &str {
&self.function.arguments
}
}
fn function_type() -> String {
String::from("function")
}
#[derive(Serialize)]
struct ChatRequest<'a> {
model: &'a str,
messages: &'a [Message],
reasoning: Reasoning,
tools: &'a [Tool],
tool_choice: &'a str,
}
/// OpenRouter reasoning config; Steep always asks for reasoning.
#[derive(Serialize)]
struct Reasoning {
enabled: bool,
}
#[derive(Deserialize)]
struct ChatResponse {
#[serde(default)]
choices: Vec<Choice>,
#[serde(default)]
usage: Option<Usage>,
}
#[derive(Deserialize)]
struct Choice {
message: ResponseMessage,
}
#[derive(Deserialize)]
struct ResponseMessage {
#[serde(default)]
content: Option<Content>,
#[serde(default)]
tool_calls: Option<Vec<ToolCall>>,
}
/// Message content is a plain string for most models and a list of typed parts
/// for others; both are collapsed to text.
#[derive(Deserialize)]
#[serde(untagged)]
enum Content {
Text(String),
Parts(Vec<ContentPart>),
}
#[derive(Deserialize)]
struct ContentPart {
#[serde(default)]
text: Option<String>,
}
impl Content {
fn into_text(self) -> String {
match self {
Self::Text(text) => text,
Self::Parts(parts) => parts
.into_iter()
.filter_map(|part| part.text)
.collect::<Vec<_>>()
.join(""),
}
}
}
#[derive(Deserialize)]
struct Usage {
#[serde(default)]
cost: Option<f64>,
}
/// Error payload returned by OpenRouter for a failed request.
#[derive(Deserialize)]
struct ErrorResponse {
error: ErrorDetail,
}
#[derive(Deserialize)]
struct ErrorDetail {
message: String,
}
#[derive(Clone)]
pub struct OpenRouterClient {
client: reqwest::Client,
api_key: String,
model: String,
}
impl OpenRouterClient {
pub fn new(token: &str, model: &str, timeout: u64) -> anyhow::Result<Self> {
Ok(Self {
client: reqwest::Client::builder()
.timeout(Duration::from_secs(timeout))
.build()?,
api_key: String::from(token),
model: String::from(model),
})
}
/// Sends a conversation with tool definitions and returns either a final
/// message or the tool calls requested by the model.
///
/// The conversation is borrowed rather than taken by value: the caller keeps
/// appending to it between iterations, and a copy per iteration would be pure
/// waste.
#[instrument(skip(self, messages, tools), err)]
pub async fn chat_with_tools(
&self,
messages: &[Message],
tools: &[Tool],
tool_choice: ToolChoice,
) -> anyhow::Result<ToolChatResult> {
let response = self.complete(messages, tools, tool_choice).await?;
let cost = response.usage.and_then(|usage| usage.cost);
let message = response
.choices
.into_iter()
.next()
.ok_or_else(|| anyhow::anyhow!("No choices in response"))?
.message;
Ok(ToolChatResult {
message: message.content.map(Content::into_text),
tool_calls: message.tool_calls.unwrap_or_default(),
cost,
})
}
async fn complete(
&self,
messages: &[Message],
tools: &[Tool],
tool_choice: ToolChoice,
) -> anyhow::Result<ChatResponse> {
let request = ChatRequest {
model: &self.model,
messages,
reasoning: Reasoning { enabled: true },
tools,
tool_choice: tool_choice.as_str(),
};
let response = self
.client
.post(format!("{BASE_URL}/chat/completions"))
.bearer_auth(&self.api_key)
.json(&request)
.send()
.await
.context("failed to reach OpenRouter")?;
let status = response.status();
if !status.is_success() {
let body = response.text().await.unwrap_or_default();
anyhow::bail!("OpenRouter returned {status}: {}", error_message(&body));
}
response
.json::<ChatResponse>()
.await
.context("invalid OpenRouter response")
}
}
/// Extracts the message from an OpenRouter error body, falling back to the raw
/// body when it is not the expected JSON shape.
fn error_message(body: &str) -> String {
serde_json::from_str::<ErrorResponse>(body)
.map(|response| response.error.message)
.unwrap_or_else(|_| body.trim().to_string())
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
/// Builds a tool call the way the API returns one, so the fixture also
/// covers deserialization.
fn tool_call(id: &str, name: &str, arguments: &str) -> ToolCall {
serde_json::from_value(json!({
"id": id,
"type": "function",
"function": { "name": name, "arguments": arguments }
}))
.unwrap()
}
#[test]
fn tool_call_arguments_are_kept_verbatim() {
let call = tool_call("call_1", "grep", r#"{"pattern":"fn main"}"#);
let message = Message::assistant_with_tool_calls("", vec![call]);
let serialized = serde_json::to_value(&message).unwrap();
assert_eq!(serialized["content"], Value::Null);
assert_eq!(serialized["tool_calls"][0]["type"], "function");
assert_eq!(serialized["tool_calls"][0]["function"]["name"], "grep");
assert_eq!(
serialized["tool_calls"][0]["function"]["arguments"],
r#"{"pattern":"fn main"}"#
);
}
#[test]
fn tool_response_carries_role_and_tool_call_id() {
let message = Message::tool_response("call_1", "src/main.rs");
let serialized = serde_json::to_value(&message).unwrap();
assert_eq!(serialized["role"], "tool");
assert_eq!(serialized["tool_call_id"], "call_1");
assert_eq!(serialized["content"], "src/main.rs");
}
#[test]
fn request_serializes_tool_choice_and_reasoning() {
let messages = [Message::new(Role::User, "hi")];
let tools = [Tool::new("ls", "List files", json!({"type": "object"}))];
let request = serde_json::to_value(ChatRequest {
model: "some/model",
messages: &messages,
reasoning: Reasoning { enabled: true },
tools: &tools,
tool_choice: ToolChoice::Auto.as_str(),
})
.unwrap();
assert_eq!(request["model"], "some/model");
assert_eq!(request["reasoning"]["enabled"], true);
assert_eq!(request["tool_choice"], "auto");
assert_eq!(request["tools"][0]["function"]["name"], "ls");
assert_eq!(request["messages"][0]["role"], "user");
}
#[test]
fn a_forced_answer_turn_disables_the_tools() {
assert_eq!(ToolChoice::None.as_str(), "none");
assert_eq!(ToolChoice::Auto.as_str(), "auto");
}
#[test]
fn response_parses_tool_calls_and_cost() {
let response: ChatResponse = serde_json::from_value(json!({
"choices": [{
"message": {
"role": "assistant",
"content": null,
"tool_calls": [{
"id": "call_1",
"type": "function",
"function": { "name": "ls", "arguments": "{\"path\":\"src\"}" }
}]
}
}],
"usage": { "cost": 0.0021 }
}))
.unwrap();
assert_eq!(response.usage.and_then(|usage| usage.cost), Some(0.0021));
let message = response.choices.into_iter().next().unwrap().message;
assert!(message.content.is_none());
let calls = message.tool_calls.unwrap();
assert_eq!(calls[0].name(), "ls");
assert_eq!(calls[0].id(), "call_1");
assert_eq!(calls[0].arguments_json(), r#"{"path":"src"}"#);
}
#[test]
fn response_parses_parts_content() {
let response: ChatResponse = serde_json::from_value(json!({
"choices": [{
"message": {
"role": "assistant",
"content": [
{ "type": "text", "text": "hello " },
{ "type": "text", "text": "world" }
]
}
}]
}))
.unwrap();
assert!(response.usage.is_none());
let message = response.choices.into_iter().next().unwrap().message;
assert_eq!(message.content.unwrap().into_text(), "hello world");
}
#[test]
fn error_message_prefers_api_message() {
let body = r#"{"error":{"message":"No auth credentials found","code":401}}"#;
assert_eq!(error_message(body), "No auth credentials found");
}
#[test]
fn error_message_falls_back_to_raw_body() {
assert_eq!(
error_message(" <html>bad gateway</html> "),
"<html>bad gateway</html>"
);
}
}
+316
View File
@@ -0,0 +1,316 @@
//! Tool-calling loop driving the model against a [`Sandbox`].
//!
//! The loop sends the conversation and the available tools to OpenRouter,
//! executes any requested tool call inside the sandbox and feeds the results
//! back, until the model produces a final message or the iteration budget is
//! exhausted.
use std::str::FromStr;
use std::time::Instant;
use anyhow::Context;
use serde_json::Value;
use tracing::{info, warn};
use crate::{
open_router::{Message, OpenRouterClient, Role, Tool, ToolCall, ToolChoice},
sandbox::{Sandbox, tools},
text::{excerpt, truncate_bytes},
};
/// Number of characters of the arguments of a tool call kept in the logs.
const MAX_LOGGED_ARGUMENTS: usize = 200;
/// Number of characters of a tool result kept in the logs.
///
/// Sizes alone make a broken tool invisible: a command that fails returns a short
/// `error: …` instead of the expected content, which is exactly what this excerpt
/// makes obvious.
const MAX_LOGGED_OUTPUT: usize = 120;
/// Maximum number of bytes of a tool result handed to the model.
///
/// A result stays in the conversation and is re-sent on every following
/// iteration, so an unbounded one (a whole file, a match on every line) inflates
/// the context and the cost of the whole rest of the run — and can overflow the
/// model context window outright.
const MAX_TOOL_OUTPUT_BYTES: usize = 32 * 1024;
/// Asked of the model when its answer does not satisfy the caller's contract.
const RETRY_PROMPT: &str = "
Your answer is not valid: {error}
Answer again with the requested format only: a raw json object, without any
markdown code fence and without any text before or after it.
";
/// Asked of the model on the last turn, in place of one more exploration turn.
///
/// Without it, a model still exploring on its last turn would ask for another
/// tool, and the run would end on the iteration budget with no answer at all.
const FINAL_PROMPT: &str = "
You are out of turns: this is your last one. Answer now, with the requested
format, from what you have already gathered, without calling any tool.
";
/// Final output of an agent run.
pub struct AgentResult<A> {
/// Answer of the model, parsed into the type the caller asked for.
pub answer: A,
pub cost: Option<f64>,
pub iterations: usize,
}
/// Runs the tool-calling loop until the model answers or `max_iterations` is
/// reached.
///
/// `tool_definitions` is the set of tools the model may call; it is selected by
/// the caller based on the webhook action (see [`tools::for_webhook`]).
///
/// The answer of the model is parsed into `A`, which is how the caller states the
/// format it expects: while [`FromStr`] rejects the answer, the error is sent back
/// to the model so that it can correct itself, which costs an iteration. The
/// parsing happens here because that is where the conversation and the remaining
/// budget are at hand.
pub async fn run<A>(
open_router: &OpenRouterClient,
sandbox: &Sandbox,
tool_definitions: Vec<Tool>,
system_prompt: &str,
user_prompt: &str,
max_iterations: usize,
) -> anyhow::Result<AgentResult<A>>
where
A: FromStr,
A::Err: std::fmt::Display,
{
let mut messages = vec![
Message::new(Role::System, system_prompt),
Message::new(Role::User, user_prompt),
];
info!(
tools = %tool_definitions
.iter()
.map(|tool| tool.function.name.as_str())
.collect::<Vec<_>>()
.join(", "),
max_iterations,
"Starting tool-calling loop"
);
let mut total_cost = 0.0_f64;
let mut has_cost = false;
let mut last_rejection: Option<String> = None;
for iteration in 1..=max_iterations {
// Le dernier tour n'est plus un tour d'exploration : le modèle doit rendre
// sa réponse, avec ce qu'il a déjà vu.
let last = iteration == max_iterations;
if last {
info!(iteration, "Last turn: asking for the final answer");
messages.push(Message::new(Role::User, FINAL_PROMPT));
}
let started = Instant::now();
let response = open_router
.chat_with_tools(
&messages,
&tool_definitions,
if last {
ToolChoice::None
} else {
ToolChoice::Auto
},
)
.await?;
if let Some(cost) = response.cost {
total_cost += cost;
has_cost = true;
}
info!(
iteration,
max_iterations,
last_turn = last,
tool_calls = response.tool_calls.len(),
answer_bytes = response.message.as_deref().map_or(0, str::len),
elapsed_ms = started.elapsed().as_millis() as u64,
"Model answered"
);
// Au dernier tour, les outils sont refusés : même si le fournisseur les
// renvoie malgré tout, on ne les exécute pas et on tente la réponse.
if last || response.tool_calls.is_empty() {
let answer = response.message.unwrap_or_default();
match answer.parse::<A>() {
Ok(parsed) => {
return Ok(AgentResult {
answer: parsed,
cost: has_cost.then_some(total_cost),
iterations: iteration,
});
}
Err(err) => {
warn!(iteration, %err, "Model answer was rejected, asking for another one");
// Keep the rejected answer in the history, so that the model
// sees what it has to fix.
messages.push(Message::new(Role::Assistant, answer));
messages.push(Message::new(
Role::User,
RETRY_PROMPT.replace("{error}", &err.to_string()),
));
last_rejection = Some(err.to_string());
continue;
}
}
}
messages.push(Message::assistant_with_tool_calls(
response.message.unwrap_or_default(),
response.tool_calls.clone(),
));
for call in &response.tool_calls {
let started = Instant::now();
let (content, truncated) = execute(sandbox, call).await;
info!(
iteration,
tool = call.name(),
arguments = %excerpt(call.arguments_json(), MAX_LOGGED_ARGUMENTS),
output_bytes = content.len(),
truncated,
// Les sauts de ligne casseraient la lisibilité d'une ligne de log.
output = %excerpt(&content, MAX_LOGGED_OUTPUT).replace('\n', " "),
elapsed_ms = started.elapsed().as_millis() as u64,
"Tool call finished"
);
messages.push(Message::tool_response(call.id(), content));
}
}
warn!(
max_iterations,
"Agent used its last turn without a valid answer"
);
match last_rejection {
Some(error) => anyhow::bail!(
"agent exceeded the maximum number of iterations ({max_iterations}), \
last answer rejected: {error}"
),
None => anyhow::bail!("agent exceeded the maximum number of iterations ({max_iterations})"),
}
}
/// Runs a tool call and returns its result for the model, along with whether the
/// result had to be truncated.
async fn execute(sandbox: &Sandbox, call: &ToolCall) -> (String, bool) {
let args = match parse_args(call) {
Ok(args) => args,
Err(err) => return cap_output(format!("error: {err}")),
};
match tools::dispatch(sandbox, call.name(), &args).await {
Ok(output) => cap_output(output),
Err(err) => cap_output(format!("error: {err}")),
}
}
/// Bounds [`MAX_TOOL_OUTPUT_BYTES`] of a tool result, telling the model what was
/// dropped so that it can narrow its request.
fn cap_output(mut output: String) -> (String, bool) {
let total = output.len();
if !truncate_bytes(&mut output, MAX_TOOL_OUTPUT_BYTES) {
return (output, false);
}
let kept = output.len();
output.push_str(&format!(
"\n… output truncated: {total} bytes in total, the first {kept} are shown. \
Narrow the request (path, pattern or line range) to see the rest."
));
(output, true)
}
fn parse_args(call: &ToolCall) -> anyhow::Result<Value> {
let raw = call.arguments_json().trim();
if raw.is_empty() {
return Ok(Value::Object(serde_json::Map::new()));
}
serde_json::from_str(raw)
.with_context(|| format!("invalid arguments for tool `{}`", call.name()))
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
/// Builds a tool call the way the API returns one, so the fixture also
/// covers deserialization.
fn tool_call(name: &str, arguments: &str) -> ToolCall {
serde_json::from_value(json!({
"id": "call_1",
"type": "function",
"function": { "name": name, "arguments": arguments }
}))
.unwrap()
}
#[test]
fn parse_args_accepts_empty_arguments() {
let call = tool_call("ls", "");
assert_eq!(
parse_args(&call).unwrap(),
Value::Object(serde_json::Map::new())
);
}
#[test]
fn parse_args_parses_json_object() {
let call = tool_call("ls", r#"{"path":"src"}"#);
assert_eq!(parse_args(&call).unwrap()["path"], "src");
}
#[test]
fn parse_args_rejects_invalid_json() {
let call = tool_call("ls", "not json");
assert!(parse_args(&call).is_err());
}
#[test]
fn a_small_tool_output_is_kept_as_is() {
let (content, truncated) = cap_output(String::from("src/main.rs"));
assert_eq!(content, "src/main.rs");
assert!(!truncated);
}
#[test]
fn a_large_tool_output_is_truncated_and_told_to_the_model() {
let (content, truncated) = cap_output("a".repeat(MAX_TOOL_OUTPUT_BYTES + 1));
assert!(truncated);
assert!(content.starts_with(&"a".repeat(MAX_TOOL_OUTPUT_BYTES)));
assert!(content.contains("output truncated"));
}
#[test]
fn a_truncated_tool_output_stays_valid_utf8() {
let (content, truncated) = cap_output("é".repeat(MAX_TOOL_OUTPUT_BYTES));
assert!(truncated);
assert!(content.starts_with('é'));
}
}
+292
View File
@@ -0,0 +1,292 @@
//! Le [`Sandbox`] : clone d'une pull request exécuté dans un devcontainer
//! éphémère, et sa configuration.
use std::{
path::{Path, PathBuf},
process::Stdio,
};
use anyhow::Context;
use devcontainer_rs::{Container, ContainerRuntime, ExecOutput};
use tempfile::TempDir;
use tracing::{info, instrument};
/// Devcontainer locations recognized within a repository, in priority order.
const DEVCONTAINER_PATHS: [&str; 2] = [".devcontainer/devcontainer.json", ".devcontainer.json"];
/// Sandbox-related runtime configuration.
#[derive(Clone)]
pub struct SandboxConfig {
/// Client du daemon de containers qui exécute la sandbox.
pub runtime: ContainerRuntime,
/// Maximum number of tool-calling iterations per agent run.
pub max_iterations: usize,
}
/// A cloned repository running inside an ephemeral devcontainer.
pub struct Sandbox {
// Owns the temporary directory; dropping it cleans up the clone.
_workspace: TempDir,
/// Path of the clone, as copied into the container.
repo_dir: PathBuf,
container: Container,
}
impl Sandbox {
/// Clones the pull request head, builds the devcontainer and starts it.
///
/// The clone is PR-aware: it fetches `refs/pull/<number>/head`, which works
/// for both same-repository and forked pull requests.
#[instrument(skip(runtime, token), fields(pr = pull_request_number))]
pub async fn create(
runtime: &ContainerRuntime,
clone_url: &str,
token: &str,
pull_request_number: u64,
) -> anyhow::Result<Self> {
let workspace = tempfile::tempdir().context("failed to create sandbox workspace")?;
let repo_dir = workspace.path().join("repo");
clone_pull_request(clone_url, token, pull_request_number, &repo_dir).await?;
make_readable(&repo_dir).await?;
let devcontainer = match find_devcontainer(&repo_dir) {
Some(path) => devcontainer_rs::parse(&path).await?,
None => {
info!(
"no devcontainer found in the pull request, falling back to the default image"
);
devcontainer_rs::DevContainer::default()
}
};
let container = devcontainer.up(runtime, &repo_dir).await?;
let sandbox = Self {
_workspace: workspace,
repo_dir,
container,
};
if let Err(err) = sandbox.check_workspace().await {
let _ = sandbox.container.remove().await;
return Err(err);
}
Ok(sandbox)
}
/// Vérifie que le clone est bien visible dans le container.
///
/// Sans ce contrôle, un workspace vide — par exemple un daemon qui n'a pas pu
/// recevoir le clone — fait échouer chaque outil ; le modèle enchaîne alors les
/// appels ratés jusqu'au budget d'itérations, sans jamais pouvoir reviewer quoi
/// que ce soit.
async fn check_workspace(&self) -> anyhow::Result<()> {
let workspace_folder = self.workspace_folder();
let probe = self
.container
.exec(&["ls", "-A", "--", workspace_folder])
.await
.with_context(|| format!("failed to list `{workspace_folder}` in the sandbox"))?;
if !probe.success() {
let details = self.diagnose_workspace().await;
anyhow::bail!(
"the sandbox cannot list `{workspace_folder}`: {} ({details})",
probe.stderr.trim()
);
}
if probe.stdout.trim().is_empty() {
anyhow::bail!(
"the sandbox workspace `{workspace_folder}` is empty: the container daemon does not see the clone"
);
}
info!(
clone = %self.repo_dir.display(),
workspace = %workspace_folder,
entries = probe.stdout.lines().count(),
"Sandbox workspace is readable"
);
Ok(())
}
/// Rassemble de quoi expliquer un refus d'accès au workspace.
///
/// Un `EACCES` a deux causes possibles, indistinguables dans le message de
/// `ls` : les permissions du clone, ou un confinement du noyau qui bloque
/// l'accès. L'identité de l'utilisateur d'exec et les permissions du point de
/// montage permettent de trancher.
async fn diagnose_workspace(&self) -> String {
let mut details = Vec::new();
if let Ok(output) = self.container.exec(&["id"]).await {
details.push(output.stdout.trim().to_string());
}
if let Ok(output) = self
.container
.exec(&["ls", "-ld", "--", self.workspace_folder()])
.await
&& output.success()
{
details.push(output.stdout.trim().to_string());
}
details.join(", ")
}
/// Executes a command in the container as an argv vector (no shell).
pub async fn exec(&self, cmd: &[&str]) -> anyhow::Result<ExecOutput> {
Ok(self.container.exec(cmd).await?)
}
/// Path of the repository inside the container.
pub fn workspace_folder(&self) -> &str {
self.container.workspace_folder()
}
/// Stops and removes the container. The temporary clone is removed on drop.
pub async fn cleanup(self) -> anyhow::Result<()> {
self.container.remove().await?;
Ok(())
}
}
fn find_devcontainer(repo_dir: &Path) -> Option<PathBuf> {
DEVCONTAINER_PATHS
.iter()
.map(|relative| repo_dir.join(relative))
.find(|candidate| candidate.is_file())
}
async fn clone_pull_request(
repo_url: &str,
token: &str,
pull_request_number: u64,
dest: &Path,
) -> anyhow::Result<()> {
let dest = dest.display().to_string();
run_git(
token,
&[
"clone".to_string(),
"--depth".to_string(),
"1".to_string(),
repo_url.to_string(),
dest.clone(),
],
)
.await?;
run_git(
token,
&[
"-C".to_string(),
dest.clone(),
"fetch".to_string(),
"--depth".to_string(),
"1".to_string(),
"origin".to_string(),
format!("refs/pull/{pull_request_number}/head"),
],
)
.await?;
run_git(
token,
&[
"-C".to_string(),
dest,
"checkout".to_string(),
"FETCH_HEAD".to_string(),
],
)
.await?;
Ok(())
}
/// Rend le clone lisible par les utilisateurs du container sandbox.
///
/// Le container peut ne pas avoir les mêmes uid que Steep (podman rootless
/// mappe les uid à travers des plages subuid), et l'umask de l'opérateur peut être
/// restrictif : sans cela, les outils de la sandbox échouent en `Permission
/// denied` sur des fichiers que Steep vient de cloner lui-même.
async fn make_readable(repo_dir: &Path) -> anyhow::Result<()> {
let output = tokio::process::Command::new("chmod")
.args(["-R", "a+rX"])
.arg(repo_dir)
.stdin(Stdio::null())
.output()
.await
.context("failed to spawn chmod")?;
if !output.status.success() {
anyhow::bail!(
"chmod failed on `{}`: {}",
repo_dir.display(),
String::from_utf8_lossy(&output.stderr).trim()
);
}
Ok(())
}
/// Runs git with the token injected through `http.extraHeader`, keeping the
/// secret out of the process arguments.
async fn run_git(token: &str, args: &[String]) -> anyhow::Result<()> {
let output = tokio::process::Command::new("git")
.args(args)
.env("GIT_CONFIG_COUNT", "1")
.env("GIT_CONFIG_KEY_0", "http.extraHeader")
.env(
"GIT_CONFIG_VALUE_0",
format!("Authorization: token {token}"),
)
.env("GIT_TERMINAL_PROMPT", "0")
.stdin(Stdio::null())
.output()
.await
.context("failed to spawn git")?;
if !output.status.success() {
anyhow::bail!(
"git {} failed: {}",
args.join(" "),
String::from_utf8_lossy(&output.stderr).trim()
);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn find_devcontainer_prefers_dot_devcontainer_dir() {
let dir = tempfile::tempdir().unwrap();
let nested = dir.path().join(".devcontainer");
std::fs::create_dir(&nested).unwrap();
std::fs::write(nested.join("devcontainer.json"), "{}").unwrap();
std::fs::write(dir.path().join(".devcontainer.json"), "{}").unwrap();
assert_eq!(
find_devcontainer(dir.path()),
Some(nested.join("devcontainer.json"))
);
}
#[test]
fn find_devcontainer_returns_none_when_absent() {
let dir = tempfile::tempdir().unwrap();
assert_eq!(find_devcontainer(dir.path()), None);
}
}
+13
View File
@@ -0,0 +1,13 @@
//! Sandboxed tool execution for the AI bot.
//!
//! A [`Sandbox`] clones a pull request into a temporary directory, builds and
//! starts its devcontainer (via `devcontainer-rs`) and exposes command
//! execution inside the resulting container. The [`tools`] module maps model
//! tool calls to commands run in that container, and [`agent`] drives the
//! tool-calling loop against OpenRouter.
pub mod agent;
mod instance;
pub mod tools;
pub use instance::{Sandbox, SandboxConfig};
+303
View File
@@ -0,0 +1,303 @@
//! Tool definitions exposed to the model and their execution inside a
//! [`Sandbox`].
//!
//! Every tool is read-only and confined to the repository workspace: paths are
//! resolved relative to the container workspace folder and rejected if they
//! escape it. Commands are executed as argv vectors (never through a shell),
//! so tool arguments cannot be used for shell injection.
use std::path::Path;
use anyhow::{Context, bail};
use devcontainer_rs::{ExecOutput, normalize};
use serde_json::{Value, json};
use super::Sandbox;
use crate::{gitea::WebhookType, open_router::Tool};
/// Tools available to the model for a given webhook action.
///
/// The match is exhaustive on [`WebhookType`], so adding a new action forces a
/// decision here about which tools that action may use.
pub fn for_webhook(webhook: &WebhookType) -> Vec<Tool> {
match webhook {
WebhookType::Review(_) => review_tools(),
}
}
/// Read-only tools used to explore a repository during a review.
fn review_tools() -> Vec<Tool> {
vec![
Tool::new(
"ls",
"List the entries of a directory inside the repository.",
json!({
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Directory path relative to the repository root. Defaults to the repository root."
}
}
}),
),
Tool::new(
"file_size",
"Get the size of a file inside the repository in bytes.",
json!({
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "File path relative to the repository root."
}
}
}),
),
Tool::new(
"read_file",
"Read the content of a text file inside the repository. Every line is \
prefixed with its absolute line number, even when only a range is read.",
json!({
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "File path relative to the repository root."
},
"start_line": {
"type": "integer",
"description": "First line to read (1-based, inclusive). Defaults to the first line."
},
"end_line": {
"type": "integer",
"description": "Last line to read (1-based, inclusive). Defaults to the last line."
}
},
"required": ["path"]
}),
),
Tool::new(
"grep",
"Search for a regular expression across the repository files.",
json!({
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "Extended regular expression to search for."
},
"path": {
"type": "string",
"description": "File or directory to search in, relative to the repository root. Defaults to the repository root."
}
},
"required": ["pattern"]
}),
),
Tool::new(
"find",
"Find files by name pattern inside the repository.",
json!({
"type": "object",
"properties": {
"pattern": {
"type": "string",
"description": "Glob pattern matched against file names, e.g. `*.rs`."
},
"path": {
"type": "string",
"description": "Directory to search in, relative to the repository root. Defaults to the repository root."
}
},
"required": ["pattern"]
}),
),
]
}
/// Executes a tool call and returns its textual result.
///
/// Errors are returned as `Err` so the caller can decide whether to surface
/// them to the model or abort.
pub async fn dispatch(sandbox: &Sandbox, name: &str, args: &Value) -> anyhow::Result<String> {
match name {
"ls" => ls(sandbox, args).await,
"read_file" => read_file(sandbox, args).await,
"file_size" => file_size(sandbox, args).await,
"grep" => grep(sandbox, args).await,
"find" => find(sandbox, args).await,
other => bail!("unknown tool `{other}`"),
}
}
async fn file_size(sandbox: &Sandbox, args: &Value) -> anyhow::Result<String> {
let path = resolve(sandbox, required_str(args, "path")?)?;
let size = sandbox.exec(&["du", "-b", "--", &path]).await?;
into_stdout(size)
}
async fn ls(sandbox: &Sandbox, args: &Value) -> anyhow::Result<String> {
let path = resolve(sandbox, optional_str(args, "path").unwrap_or("."))?;
let output = sandbox.exec(&["ls", "-la", "--", &path]).await?;
into_stdout(output)
}
async fn read_file(sandbox: &Sandbox, args: &Value) -> anyhow::Result<String> {
let path = resolve(sandbox, required_str(args, "path")?)?;
let start = args.get("start_line").and_then(Value::as_u64);
let end = args.get("end_line").and_then(Value::as_u64);
let (first_line, output) = if start.is_none() && end.is_none() {
(1, sandbox.exec(&["cat", "--", &path]).await?)
} else {
let start = start.unwrap_or(1);
let end = end
.map(|line| line.to_string())
.unwrap_or_else(|| "$".to_string());
let range = format!("{start},{end}p");
(
start,
sandbox.exec(&["sed", "-n", &range, "--", &path]).await?,
)
};
Ok(number_lines(&into_stdout(output)?, first_line))
}
/// Préfixe chaque ligne par son numéro.
///
/// Le modèle doit citer une ligne précise pour ancrer son commentaire : sans
/// numéros, il les compte lui-même et se décale de quelques lignes, ce qui place le
/// commentaire à côté du code visé.
fn number_lines(content: &str, first_line: u64) -> String {
content
.lines()
.enumerate()
.map(|(offset, line)| format!("{}:{line}", first_line + offset as u64))
.collect::<Vec<_>>()
.join("\n")
}
async fn grep(sandbox: &Sandbox, args: &Value) -> anyhow::Result<String> {
let pattern = required_str(args, "pattern")?;
let path = resolve(sandbox, optional_str(args, "path").unwrap_or("."))?;
let output = sandbox
.exec(&["grep", "-rn", "-E", "--", pattern, &path])
.await?;
// grep exits with 1 when there is no match, which is not an error.
match output.status {
0 | 1 => Ok(output.stdout),
_ => bail!("grep failed: {}", output.stderr.trim()),
}
}
async fn find(sandbox: &Sandbox, args: &Value) -> anyhow::Result<String> {
let pattern = required_str(args, "pattern")?;
let path = resolve(sandbox, optional_str(args, "path").unwrap_or("."))?;
let output = sandbox
.exec(&["find", &path, "-type", "f", "-name", pattern])
.await?;
into_stdout(output)
}
/// Resolves a tool path against the container workspace folder, rejecting any
/// path that escapes it.
fn resolve(sandbox: &Sandbox, path: &str) -> anyhow::Result<String> {
let workspace = sandbox.workspace_folder();
let candidate = Path::new(workspace).join(path);
let normalized =
normalize(&candidate).with_context(|| format!("path `{path}` escapes the workspace"))?;
if !normalized.starts_with(workspace) {
bail!("path `{path}` escapes the workspace");
}
Ok(normalized.display().to_string())
}
fn required_str<'a>(args: &'a Value, key: &str) -> anyhow::Result<&'a str> {
args.get(key)
.and_then(Value::as_str)
.with_context(|| format!("`{key}` is required"))
}
fn optional_str<'a>(args: &'a Value, key: &str) -> Option<&'a str> {
args.get(key).and_then(Value::as_str)
}
fn into_stdout(output: ExecOutput) -> anyhow::Result<String> {
if output.success() {
Ok(output.stdout)
} else {
bail!(
"command failed (status {}): {}",
output.status,
output.stderr.trim()
)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::gitea::{Comment, PullRequest, Repository, ReviewPayload};
fn review_webhook() -> WebhookType {
WebhookType::Review(ReviewPayload {
action: "created".to_string(),
pull_request: PullRequest {
number: 1,
title: "My PR".to_string(),
},
repository: Repository {
full_name: "owner/repo".to_string(),
clone_url: "https://github.com/owner/repo.git".to_string(),
},
comment: Comment {
id: 1,
body: "@bot review".to_string(),
},
})
}
#[test]
fn review_webhook_exposes_read_only_tools() {
let names: Vec<String> = for_webhook(&review_webhook())
.into_iter()
.map(|tool| tool.function.name)
.collect();
assert_eq!(names, vec!["ls", "file_size", "read_file", "grep", "find"]);
}
#[test]
fn required_str_reports_missing_key() {
let err = required_str(&json!({}), "path").unwrap_err();
assert!(err.to_string().contains("path"));
}
#[test]
fn lines_are_numbered_from_the_first_one() {
assert_eq!(number_lines("a\nb\n", 1), "1:a\n2:b");
}
#[test]
fn a_range_keeps_the_absolute_line_numbers() {
// Un extrait lu à partir de la ligne 12 doit garder la numérotation du
// fichier : sinon le modèle citerait des lignes décalées.
assert_eq!(number_lines("x\ny", 12), "12:x\n13:y");
}
#[test]
fn an_empty_read_stays_empty() {
assert_eq!(number_lines("", 1), "");
}
}
+8
View File
@@ -0,0 +1,8 @@
use crate::{bot::Bot, env::EnvConfig, gitea::WebhookType};
#[derive(Clone)]
pub struct AppState {
pub bot_tx: tokio::sync::mpsc::Sender<WebhookType>,
pub bot: Bot,
pub config: EnvConfig,
}
+81
View File
@@ -0,0 +1,81 @@
//! Small text helpers shared by the modules that log what the model answered and
//! bound what the tools return to it.
/// First `limit` characters of `text`, for logs.
///
/// Cuts on character boundaries, so the excerpt stays valid UTF-8, and marks a
/// truncation with an ellipsis.
pub fn excerpt(text: &str, limit: usize) -> String {
let mut chars = text.chars();
let excerpt = chars.by_ref().take(limit).collect::<String>();
if chars.next().is_some() {
return format!("{excerpt}…");
}
excerpt
}
/// Truncates `text` in place to at most `limit` bytes, cutting on a character
/// boundary so the result stays valid UTF-8.
///
/// Returns `true` when something was dropped.
pub fn truncate_bytes(text: &mut String, limit: usize) -> bool {
if text.len() <= limit {
return false;
}
let mut end = limit;
while !text.is_char_boundary(end) {
end -= 1;
}
text.truncate(end);
true
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_short_text_is_kept_as_is() {
assert_eq!(excerpt("abc", 3), "abc");
}
#[test]
fn a_long_text_is_truncated_and_marked() {
assert_eq!(excerpt("abcdef", 3), "abc…");
}
#[test]
fn truncation_cuts_on_character_boundaries() {
assert_eq!(excerpt("ééé", 2), "éé…");
}
#[test]
fn a_short_text_is_not_truncated() {
let mut text = String::from("abc");
assert!(!truncate_bytes(&mut text, 3));
assert_eq!(text, "abc");
}
#[test]
fn a_long_text_is_truncated_within_the_limit() {
let mut text = String::from("abcdef");
assert!(truncate_bytes(&mut text, 4));
assert_eq!(text, "abcd");
}
#[test]
fn byte_truncation_never_splits_a_character() {
let mut text = String::from("ééé");
// The limit falls in the middle of the second `é`: it is dropped.
assert!(truncate_bytes(&mut text, 3));
assert_eq!(text, "é");
}
}