Menu
AkurAI-Build
publicLatest change 6844bb20842ff478454dfe4a5610878b6a47d65a - Format secure collaboration changes by Ólafur Búi Ólafsson
use std::{
fs,
io::{self, BufRead, Write},
path::{Path, PathBuf},
process::Command,
thread,
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use anyhow::{Context, Result, bail, ensure};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use crate::{
config,
db::{Database, RepositoryQuery, RunQuery},
git_process, hosted_git, release,
runner::Runner,
tree::RepoTree,
};
const PROTOCOL_VERSION: &str = "2025-06-18";
/// Longest a tool call may block the single-threaded MCP loop. MCP clients
/// time out requests at ~30s, and every queued request waits behind the
/// current one, so a longer server-side wait can never be observed — it only
/// starves unrelated calls. Long-running work is executed by the serve
/// daemon's worker loop; wait-style tools poll the database and return the
/// latest run state when this budget elapses.
const MAX_BLOCKING_WAIT: Duration = Duration::from_secs(25);
/// Fixed context every tool call runs against: the encrypted database, the
/// data root used to resolve hosted mirrors/workspaces/artifacts, and
/// whether native (non-Docker) job execution is permitted.
struct Ctx {
database: Database,
root: PathBuf,
allow_native: bool,
tree: RepoTree,
actor: String,
}
impl Ctx {
fn runner(&self) -> Result<Runner> {
Runner::new(self.database.clone(), self.root.clone(), self.allow_native)
}
}
/// Dispatch one JSON-RPC request against long-lived server state (database,
/// data root, native-execution flag, repo tree). Used by the HTTP `/mcp`
/// bridge in `server.rs` so the HTTP and stdio (`serve`, above) transports
/// share the exact same tool implementations — no duplicated dispatch logic.
pub fn handle_request(
database: Database,
root: PathBuf,
allow_native: bool,
tree: RepoTree,
request: &Value,
) -> Option<Value> {
handle_request_as(database, root, allow_native, tree, "static-bearer", request)
}
/// Dispatch a request with the authenticated principal that initiated it.
/// Stdio callers use the fixed `stdio` actor; HTTP callers should pass the
/// session subject or their static automation principal.
pub fn handle_request_as(
database: Database,
root: PathBuf,
allow_native: bool,
tree: RepoTree,
actor: &str,
request: &Value,
) -> Option<Value> {
let context = Ctx {
database,
root,
allow_native,
tree,
actor: actor.to_owned(),
};
handle(&context, request)
}
pub fn serve(database: Database, root: PathBuf, allow_native: bool) -> Result<()> {
let tree = RepoTree::new(&root)?;
let context = Ctx {
database,
root,
allow_native,
tree,
actor: "stdio".to_owned(),
};
let stdin = io::stdin();
let mut stdout = io::stdout().lock();
for line in stdin.lock().lines() {
let line = line.context("read MCP request")?;
if line.trim().is_empty() {
continue;
}
let response = match serde_json::from_str::<Value>(&line) {
Ok(request) => handle(&context, &request),
Err(error) => Some(json_rpc_error(
Value::Null,
-32700,
format!("invalid JSON: {error}"),
)),
};
if let Some(response) = response {
serde_json::to_writer(&mut stdout, &response)?;
stdout.write_all(b"\n")?;
stdout.flush()?;
}
}
// Runs are executed exclusively by the worker daemon; wait-style tools
// only poll the database, so there is nothing to join on shutdown.
Ok(())
}
fn handle(context: &Ctx, request: &Value) -> Option<Value> {
let id = request.get("id")?.clone();
let method = request.get("method").and_then(Value::as_str);
let Some(method) = method else {
return Some(json_rpc_error(id, -32600, "missing method"));
};
let response = match method {
"initialize" => json_rpc_result(
id,
json!({
"protocolVersion": PROTOCOL_VERSION,
"capabilities": {"tools": {"listChanged": false}},
"serverInfo": {"name": "akurai-build", "version": env!("CARGO_PKG_VERSION")},
"instructions": "Full read/write access to AkurAI Build repositories, pipeline runs, artifacts, protected-environment promotions, and Titan workers. Mutating tools (repo_add, repo_host, repo_sync, repo_rename, repo_remove, repo_visibility, run_queue, run_retry, run_promote) change production CI/CD state; repo_remove and run_promote are consequential."
}),
),
"ping" => json_rpc_result(id, json!({})),
"tools/list" => json_rpc_result(id, json!({"tools": tools()})),
"tools/call" => {
let params = request.get("params").cloned().unwrap_or_else(|| json!({}));
let name = params.get("name").and_then(Value::as_str);
let arguments = params
.get("arguments")
.cloned()
.unwrap_or_else(|| json!({}));
match name {
Some(name) => match call_tool(context, name, arguments) {
Ok(data) => json_rpc_result(id, tool_result(data, false)),
Err(error) => json_rpc_result(id, tool_result(json!(error.to_string()), true)),
},
None => json_rpc_error(id, -32602, "tool name is required"),
}
}
_ => json_rpc_error(id, -32601, format!("unknown method: {method}")),
};
Some(response)
}
fn call_tool(context: &Ctx, name: &str, arguments: Value) -> Result<Value> {
let descriptor = audit_descriptor(context, name, &arguments);
let result = dispatch_tool(context, name, arguments);
if let Some((operation, repository, environment, target)) = descriptor {
let outcome = if result.is_ok() {
"succeeded"
} else {
"failed"
};
let details = json!({"tool": name});
let audit = context.database.record_audit_event(
&context.actor,
&operation,
repository.as_deref(),
environment.as_deref(),
target.as_deref(),
outcome,
&details,
);
if result.is_ok() {
audit?;
}
}
result
}
fn dispatch_tool(context: &Ctx, name: &str, arguments: Value) -> Result<Value> {
let database = &context.database;
match name {
"akurai_doctor" => Ok(doctor(context)),
"akurai_repo_list" => {
let arguments: RepositoryArguments = parse(arguments)?;
Ok(to_value(database.query_repositories(
&RepositoryQuery {
search: arguments.search,
visibility: arguments.visibility,
limit: arguments.limit,
offset: arguments.offset,
},
)?)?)
}
"akurai_repo_add" => {
let arguments: RepoAddArguments = parse(arguments)?;
config::validate_repository(&arguments.name, &arguments.url, &arguments.branch)?;
Ok(to_value(database.add_repository(
&arguments.name,
&arguments.url,
&arguments.branch,
)?)?)
}
"akurai_repo_host" => {
let arguments: RepoHostArguments = parse(arguments)?;
let path = hosted_git::host(&context.root, &arguments.name, &arguments.source)?;
match database.add_repository(
&arguments.name,
path.to_str()
.context("hosted repository path is not UTF-8")?,
&arguments.branch,
) {
Ok(repository) => Ok(json!({
"repository": repository,
"clone_url": format!("https://akurai-build.olibuijr.com/git/{}.git", arguments.name)
})),
Err(error) => {
if let Err(cleanup_error) = fs::remove_dir_all(&path) {
return Err(error.context(format!(
"also failed to clean up {}: {cleanup_error}",
path.display()
)));
}
Err(error)
}
}
}
"akurai_repo_sync" => {
let arguments: RepoSyncArguments = parse(arguments)?;
let path = hosted_git::sync(&context.root, &arguments.name, &arguments.source)?;
Ok(json!({"name": arguments.name, "path": path, "synchronized": true}))
}
"akurai_release" => {
let arguments: ReleaseArguments = parse(arguments)?;
// The repository must be registered before a release may target it.
database.repository(&arguments.repository)?;
let outcome = release::release(
&arguments.source,
&arguments.bump,
arguments.notes.as_deref(),
)?;
let mirror = hosted_git::sync(&context.root, &arguments.repository, &arguments.source)?;
let runner = context.runner()?;
let queued =
runner.queue(&arguments.repository, None, Some(&outcome.commit), "manual")?;
let run = if arguments.wait {
wait_for_run(database, queued.id, MAX_BLOCKING_WAIT, 2)?
} else {
json!({"id": queued.id, "status": "queued", "deduplicated": queued.deduplicated})
};
Ok(json!({
"release": to_value(outcome)?,
"mirror": mirror,
"run": run,
}))
}
"akurai_repo_rename" => {
let arguments: RepoRenameArguments = parse(arguments)?;
Ok(to_value(
database.rename_repository(&arguments.old, &arguments.new)?,
)?)
}
"akurai_repo_remove" => {
let arguments: RepoNameArguments = parse(arguments)?;
let repository = database.remove_repository(&arguments.name)?;
let mirror = hosted_git::remove(&context.root, &arguments.name)?;
Ok(json!({
"repository": repository,
"removed": true,
"mirror_deleted": mirror,
}))
}
"akurai_repo_visibility" => {
let arguments: RepoVisibilityArguments = parse(arguments)?;
Ok(to_value(database.set_repository_visibility(
&arguments.repository,
&arguments.visibility,
)?)?)
}
"akurai_repo_update" => {
let arguments: RepoUpdateArguments = parse(arguments)?;
if let Some(branch) = arguments.default_branch.as_deref() {
config::validate_ref(branch)?;
}
if let Some(url) = arguments.url.as_deref() {
config::validate_repository(
&arguments.repository,
url,
arguments.default_branch.as_deref().unwrap_or("main"),
)?;
}
Ok(to_value(database.update_repository(
&arguments.repository,
arguments.url.as_deref(),
arguments.default_branch.as_deref(),
)?)?)
}
"akurai_issue_list" => {
let arguments: IssueListArguments = parse(arguments)?;
let total = database.count_issues(&arguments.repository, arguments.state.as_deref())?;
let issues = database.list_issues(
&arguments.repository,
arguments.state.as_deref(),
arguments.limit,
arguments.offset,
)?;
Ok(
json!({"issues": issues, "total": total, "limit": arguments.limit, "offset": arguments.offset}),
)
}
"akurai_issue_create" => {
let arguments: IssueCreateArguments = parse(arguments)?;
Ok(to_value(database.create_issue(
&arguments.repository,
&arguments.title,
&arguments.body,
"machine",
"AkurAI Build",
)?)?)
}
"akurai_issue_update" => {
let arguments: IssueUpdateArguments = parse(arguments)?;
Ok(to_value(database.update_issue(
&arguments.repository,
arguments.number,
arguments.title.as_deref(),
arguments.body.as_deref(),
arguments.state.as_deref(),
)?)?)
}
"akurai_issue_comment" => {
let arguments: IssueCommentArguments = parse(arguments)?;
Ok(to_value(database.add_issue_comment(
&arguments.repository,
arguments.number,
&arguments.body,
"machine",
"AkurAI Build",
)?)?)
}
"akurai_repo_branches" => {
let arguments: RepoNameArguments = parse(arguments)?;
Ok(to_value(repo_branches(database, &arguments.name)?)?)
}
"akurai_repo_tree" => {
let arguments: RepoTreeArguments = parse(arguments)?;
let repository = database.repository(&arguments.repository)?;
let sub_path = arguments.path.unwrap_or_default();
Ok(to_value(context.tree.tree(
&repository.name,
&repository.url,
arguments.reference.as_deref(),
&sub_path,
)?)?)
}
"akurai_repo_blob" => {
let arguments: RepoBlobArguments = parse(arguments)?;
let repository = database.repository(&arguments.repository)?;
let reference = arguments
.reference
.as_deref()
.unwrap_or(&repository.default_branch);
Ok(to_value(context.tree.blob(
&repository.name,
&repository.url,
reference,
&arguments.path,
)?)?)
}
"akurai_init" => {
let arguments: InitArguments = parse(arguments)?;
ensure!(
arguments.force || !arguments.out.exists(),
"{} already exists; use force",
arguments.out.display()
);
fs::write(&arguments.out, include_str!("../akurai.example.yml"))?;
Ok(json!({"path": arguments.out, "created": true}))
}
"akurai_run_queue" => {
let arguments: RunQueueArguments = parse(arguments)?;
let runner = context.runner()?;
let queued = runner.queue(
&arguments.repository,
arguments.git_ref.as_deref(),
arguments.commit.as_deref(),
"manual",
)?;
if arguments.wait {
Ok(wait_for_run(database, queued.id, MAX_BLOCKING_WAIT, 2)?)
} else {
let run = database.run(queued.id)?;
Ok(
json!({"id": queued.id, "status": run.status, "deduplicated": queued.deduplicated}),
)
}
}
"akurai_runs" => {
let arguments: RunArguments = parse(arguments)?;
Ok(to_value(
database.query_runs(&RunQuery {
repository: arguments.repo.map(|value| value.values().join(",")),
statuses: arguments.status.map(StringList::values).unwrap_or_default(),
git_ref: arguments.git_ref,
triggers: arguments
.trigger
.map(StringList::values)
.unwrap_or_default(),
search: arguments.search,
before_id: None,
limit: arguments.limit,
offset: arguments.offset,
public_only: false,
})?,
)?)
}
"akurai_run_show" => {
let arguments: RunIdArguments = parse(arguments)?;
Ok(to_value(database.detail(arguments.id)?)?)
}
"akurai_run_cancel" => {
let arguments: RunIdArguments = parse(arguments)?;
database.cancel_run(arguments.id)?;
Ok(to_value(database.detail(arguments.id)?)?)
}
"akurai_run_wait" => {
let arguments: RunWaitArguments = parse(arguments)?;
ensure!(
(1..=14_400).contains(&arguments.timeout),
"timeout must be 1..=14400 seconds"
);
ensure!(
(1..=60).contains(&arguments.interval),
"interval must be 1..=60 seconds"
);
let budget = Duration::from_secs(arguments.timeout).min(MAX_BLOCKING_WAIT);
Ok(wait_for_run(
database,
arguments.id,
budget,
arguments.interval,
)?)
}
"akurai_run_logs" => {
let arguments: RunLogsArguments = parse(arguments)?;
let mut jobs = database.jobs(arguments.id)?;
if arguments.failed {
jobs.retain(|job| {
matches!(job.status.as_str(), "failed" | "interrupted" | "canceled")
});
}
Ok(to_value(
jobs.into_iter()
.map(|job| {
json!({"id": job.id, "name": job.name, "status": job.status, "logs": job.logs})
})
.collect::<Vec<_>>(),
)?)
}
"akurai_run_retry" => {
let arguments: RunRetryArguments = parse(arguments)?;
let id = database.retry(arguments.id)?;
if arguments.wait {
Ok(wait_for_run(database, id, MAX_BLOCKING_WAIT, 2)?)
} else {
Ok(json!({"id": id, "status": "queued"}))
}
}
"akurai_run_promote" => {
let arguments: RunPromoteArguments = parse(arguments)?;
let jobs = database.approve_environment_as(
arguments.id,
&arguments.environment,
&context.actor,
)?;
if arguments.wait {
Ok(wait_for_run(database, arguments.id, MAX_BLOCKING_WAIT, 2)?)
} else {
Ok(json!({
"id": arguments.id,
"environment": arguments.environment,
"approved_jobs": jobs,
"status": "queued"
}))
}
}
"akurai_artifact_get" => {
let arguments: ArtifactGetArguments = parse(arguments)?;
ensure!(
arguments.force || !arguments.output.exists(),
"{} exists; use force",
arguments.output.display()
);
let runner = context.runner()?;
let artifact = database.artifact(arguments.id)?;
let source = runner.artifact_path(&artifact)?;
if let Some(parent) = arguments
.output
.parent()
.filter(|path| !path.as_os_str().is_empty())
{
fs::create_dir_all(parent)?;
}
fs::copy(source, &arguments.output)?;
Ok(json!({
"id": arguments.id,
"output": arguments.output,
"sha256": artifact.sha256,
"bytes": artifact.bytes
}))
}
"akurai_delivery_metrics" => {
let arguments: DeliveryMetricsArguments = parse(arguments)?;
Ok(to_value(database.delivery_metrics(
arguments.repository.as_deref(),
arguments.environment.as_deref(),
arguments.window_seconds,
)?)?)
}
"akurai_pipeline_validate" => {
let arguments: PipelineValidateArguments = parse(arguments)?;
validate_pipeline(context, &arguments)
}
"akurai_cache_stats" => {
let arguments: CacheStatsArguments = parse(arguments)?;
cache_stats(database, &context.root, arguments.repository.as_deref())
}
"akurai_cache_prune" => {
let arguments: CachePruneArguments = parse(arguments)?;
cache_prune(
database,
&context.root,
arguments.repository.as_deref(),
arguments.max_age_days,
arguments.dry_run,
)
}
"akurai_worker_drain" => {
let arguments: WorkerDrainArguments = parse(arguments)?;
database.set_worker_draining(&arguments.worker, arguments.draining)?;
let worker = database
.workers()?
.into_iter()
.find(|worker| worker.id == arguments.worker)
.context("worker disappeared after drain update")?;
Ok(to_value(worker)?)
}
"akurai_audit_events" => {
let arguments: AuditEventsArguments = parse(arguments)?;
Ok(to_value(database.audit_events(
arguments.repository.as_deref(),
arguments.limit,
arguments.offset,
)?)?)
}
"akurai_installation_register" => {
let arguments: InstallationRegisterArguments = parse(arguments)?;
let revision = database.register_installation_revision(
&arguments.installation_id,
&arguments.organization_id,
&arguments.workspace_id,
&arguments.app_id,
&arguments.idempotency_key,
)?;
Ok(json!({
"installation_id": arguments.installation_id,
"revision": revision,
"idempotent": true
}))
}
"akurai_deployments_for_tenant" => {
let arguments: TenantDeploymentsArguments = parse(arguments)?;
Ok(to_value(database.deployments_for_tenant(
arguments.run_id,
&arguments.organization_id,
&arguments.workspace_id,
)?)?)
}
"akurai_workers" => Ok(to_value(database.workers()?)?),
_ => bail!("unknown tool: {name}"),
}
}
fn audit_descriptor(
context: &Ctx,
name: &str,
arguments: &Value,
) -> Option<(String, Option<String>, Option<String>, Option<String>)> {
let string = |key: &str| {
arguments
.get(key)
.and_then(Value::as_str)
.map(str::to_owned)
};
match name {
"akurai_repo_add" | "akurai_repo_host" | "akurai_repo_sync" => {
Some((name.to_owned(), string("name"), None, string("source")))
}
"akurai_repo_update" | "akurai_repo_visibility" => {
Some((name.to_owned(), string("repository"), None, None))
}
"akurai_repo_rename" => Some((name.to_owned(), string("old"), None, string("new"))),
"akurai_repo_remove" => Some((name.to_owned(), string("name"), None, None)),
"akurai_release" => Some((
name.to_owned(),
string("repository"),
None,
string("source"),
)),
"akurai_run_queue" => Some((
name.to_owned(),
string("repository"),
None,
arguments
.get("commit")
.or_else(|| arguments.get("git_ref"))
.and_then(Value::as_str)
.map(str::to_owned),
)),
"akurai_run_cancel" | "akurai_run_retry" => {
let id = arguments.get("id").and_then(Value::as_i64);
let run = id.and_then(|id| context.database.run(id).ok());
Some((
name.to_owned(),
run.as_ref().map(|run| run.repository.clone()),
None,
id.map(|id| id.to_string()),
))
}
"akurai_run_promote" => {
let id = arguments.get("id").and_then(Value::as_i64);
let run = id.and_then(|id| context.database.run(id).ok());
Some((
name.to_owned(),
run.as_ref().map(|run| run.repository.clone()),
string("environment"),
id.map(|id| id.to_string()),
))
}
"akurai_worker_drain" | "akurai_cache_prune" => Some((
name.to_owned(),
string("repository"),
None,
string("worker"),
)),
"akurai_issue_create" | "akurai_issue_update" | "akurai_issue_comment" => Some((
name.to_owned(),
string("repository"),
None,
string("number"),
)),
_ => None,
}
}
fn validate_pipeline(context: &Ctx, arguments: &PipelineValidateArguments) -> Result<Value> {
let repository = context.database.repository(&arguments.repository)?;
let reference = arguments
.reference
.as_deref()
.unwrap_or(&repository.default_branch);
config::validate_ref(reference)?;
let temporary = tempfile::tempdir().context("create pipeline validation workspace")?;
let workspace = temporary.path();
let init = git_process::command(workspace)
.args(["init", "--quiet"])
.output()
.context("initialize pipeline validation workspace")?;
ensure!(init.status.success(), "git init failed");
let remote = git_process::command(workspace)
.args(["remote", "add", "origin", &repository.url])
.output()
.context("configure pipeline validation remote")?;
ensure!(remote.status.success(), "git remote add failed");
let fetch = git_process::command(workspace)
.args(["fetch", "--depth=1", "origin", reference])
.output()
.context("fetch immutable pipeline ref")?;
ensure!(
fetch.status.success(),
"pipeline ref is not available from the repository"
);
let commit = git_process::output(workspace, ["rev-parse", "FETCH_HEAD"])?;
let commit = commit.trim().to_owned();
config::validate_commit(&commit)?;
let pipeline_name = [".akurai.yml", "akurai.yml"]
.into_iter()
.find(|name| {
git_process::command(workspace)
.args(["cat-file", "-e", &format!("{commit}:{name}")])
.output()
.is_ok_and(|output| output.status.success())
})
.context("repository has no .akurai.yml or akurai.yml at immutable ref")?;
let source = git_process::output(workspace, ["show", &format!("{commit}:{pipeline_name}")])?;
let jobs = config::parse(&source)?;
Ok(json!({
"repository": repository.name,
"ref": reference,
"commit": commit,
"pipeline": pipeline_name,
"valid": true,
"jobs": jobs.iter().map(|job| json!({"name": job.name, "executor": job.executor, "environment": job.environment})).collect::<Vec<_>>()
}))
}
#[derive(Serialize)]
struct CacheEntry {
repository: String,
repository_id: i64,
entry: String,
files: u64,
bytes: u64,
modified_at: Option<i64>,
}
fn cache_entry(
path: &Path,
repository: &str,
repository_id: i64,
entry: &str,
) -> Result<CacheEntry> {
let (files, bytes, modified_at) = cache_tree(path)?;
Ok(CacheEntry {
repository: repository.to_owned(),
repository_id,
entry: entry.to_owned(),
files,
bytes,
modified_at,
})
}
fn cache_tree(path: &Path) -> Result<(u64, u64, Option<i64>)> {
let metadata = fs::symlink_metadata(path)?;
if metadata.file_type().is_symlink() {
return Ok((0, 0, None));
}
let modified_at = metadata
.modified()
.ok()
.and_then(|time| time.duration_since(UNIX_EPOCH).ok())
.and_then(|duration| i64::try_from(duration.as_secs()).ok());
if metadata.is_file() {
return Ok((1, metadata.len(), modified_at));
}
if !metadata.is_dir() {
return Ok((0, 0, modified_at));
}
let mut files = 0;
let mut bytes = 0;
let mut newest = modified_at;
for child in fs::read_dir(path)? {
let child = child?;
let (child_files, child_bytes, child_modified) = cache_tree(&child.path())?;
files = files.saturating_add(child_files);
bytes = bytes.saturating_add(child_bytes);
if child_modified > newest {
newest = child_modified;
}
}
Ok((files, bytes, newest))
}
fn cache_entries(
database: &Database,
root: &Path,
repository: Option<&str>,
) -> Result<Vec<CacheEntry>> {
let cache_root = root.join("cache");
if !cache_root.is_dir() {
return Ok(Vec::new());
}
let repositories = if let Some(repository) = repository {
config::validate_repo_name(repository)?;
let repository = database.repository(repository)?;
let repository_id = repository.id;
let repository_name = repository.name;
vec![(
repository_id,
repository_name,
cache_root.join(repository_id.to_string()),
)]
} else {
database
.repositories()?
.into_iter()
.map(|repository| {
(
repository.id,
repository.name,
cache_root.join(repository.id.to_string()),
)
})
.collect()
};
let mut entries = Vec::new();
for (repository_id, repository, repository_root) in repositories {
if !repository_root.is_dir() {
continue;
}
for entry in fs::read_dir(&repository_root)? {
let entry = entry?;
let metadata = fs::symlink_metadata(entry.path())?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
continue;
}
entries.push(cache_entry(
&entry.path(),
&repository,
repository_id,
&entry.file_name().to_string_lossy(),
)?);
ensure!(entries.len() <= 10_000, "cache entry limit exceeded");
}
}
Ok(entries)
}
fn cache_stats(database: &Database, root: &Path, repository: Option<&str>) -> Result<Value> {
let entries = cache_entries(database, root, repository)?;
let bytes = entries.iter().map(|entry| entry.bytes).sum::<u64>();
let files = entries.iter().map(|entry| entry.files).sum::<u64>();
Ok(json!({
"repository": repository,
"entries": entries,
"entry_count": entries.len(),
"files": files,
"bytes": bytes
}))
}
fn cache_prune(
database: &Database,
root: &Path,
repository: Option<&str>,
max_age_days: u64,
dry_run: bool,
) -> Result<Value> {
ensure!(
(1..=3650).contains(&max_age_days),
"max_age_days must be 1..=3650"
);
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.context("system clock is before Unix epoch")?
.as_secs();
let cutoff = now.saturating_sub(max_age_days.saturating_mul(86_400));
let entries = cache_entries(database, root, repository)?;
let mut eligible = 0_u64;
let mut bytes = 0_u64;
let mut removed = Vec::new();
for entry in entries {
let Some(modified_at) = entry.modified_at else {
continue;
};
if modified_at < i64::try_from(cutoff)? {
eligible += 1;
bytes = bytes.saturating_add(entry.bytes);
if !dry_run {
let path = root
.join("cache")
.join(entry.repository_id.to_string())
.join(&entry.entry);
fs::remove_dir_all(path)?;
removed.push(format!("{}/{}", entry.repository, entry.entry));
}
}
}
Ok(json!({
"repository": repository,
"max_age_days": max_age_days,
"dry_run": dry_run,
"eligible": eligible,
"removed": removed.len(),
"bytes": bytes,
"entries": removed
}))
}
/// Poll a run until it reaches a persisted terminal or approval-required
/// state, or the blocking budget elapses. On budget exhaustion the latest
/// run detail is returned (typically still "queued"/"running") instead of an
/// error, so callers always get a response within the MCP client timeout and
/// can follow up with akurai_run_wait / akurai_run_show. The actual pipeline
/// work is performed by the serve daemon's worker loop, never inline here.
fn wait_for_run(
database: &Database,
run_id: i64,
budget: Duration,
interval: u64,
) -> Result<Value> {
let deadline = Instant::now() + budget;
loop {
let detail = database.detail(run_id)?;
let settled = matches!(
detail.run.status.as_str(),
"waiting" | "succeeded" | "failed" | "canceled" | "interrupted"
);
if settled || Instant::now() >= deadline {
return to_value(detail);
}
thread::sleep(Duration::from_secs(interval).min(deadline - Instant::now()));
}
}
fn doctor(context: &Ctx) -> Value {
let git = version("git", &["--version"]);
let docker = version("docker", &["info", "--format", "{{.ServerVersion}}"]);
let database = context
.database
.check_ready()
.map(|()| "ready".to_owned())
.unwrap_or_else(|error| format!("unavailable: {error}"));
json!({"database": database, "git": git, "docker": docker, "data": context.root})
}
fn version(program: &str, arguments: &[&str]) -> Value {
match Command::new(program).args(arguments).output() {
Ok(output) => json!({
"ok": output.status.success(),
"output": String::from_utf8_lossy(if output.status.success() { &output.stdout } else { &output.stderr }).trim()
}),
Err(error) => json!({"ok": false, "output": error.to_string()}),
}
}
fn repo_branches(database: &Database, name: &str) -> Result<Vec<git_process::RemoteBranch>> {
let repository = database.repository(name)?;
git_process::remote_branches(&repository.url, &repository.default_branch)
}
fn parse<T: for<'de> Deserialize<'de>>(arguments: Value) -> Result<T> {
Ok(serde_json::from_value(arguments)?)
}
fn to_value<T: Serialize>(data: T) -> Result<Value> {
Ok(serde_json::to_value(data)?)
}
#[derive(Deserialize, Default)]
#[serde(deny_unknown_fields)]
struct RepositoryArguments {
#[serde(alias = "q")]
search: Option<String>,
visibility: Option<String>,
#[serde(default = "default_repository_limit")]
limit: usize,
#[serde(default)]
offset: usize,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RepoAddArguments {
name: String,
url: String,
#[serde(default = "default_branch")]
branch: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RepoHostArguments {
name: String,
source: PathBuf,
#[serde(default = "default_branch")]
branch: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct IssueListArguments {
repository: String,
state: Option<String>,
#[serde(default = "default_issue_limit")]
limit: usize,
#[serde(default)]
offset: usize,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct IssueCreateArguments {
repository: String,
title: String,
#[serde(default)]
body: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct IssueUpdateArguments {
repository: String,
number: i64,
title: Option<String>,
body: Option<String>,
state: Option<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RepoUpdateArguments {
repository: String,
url: Option<String>,
default_branch: Option<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct PipelineValidateArguments {
repository: String,
#[serde(rename = "ref")]
reference: Option<String>,
}
#[derive(Deserialize, Default)]
#[serde(deny_unknown_fields)]
struct CacheStatsArguments {
repository: Option<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct CachePruneArguments {
repository: Option<String>,
max_age_days: u64,
#[serde(default)]
dry_run: bool,
}
#[derive(Deserialize, Default)]
#[serde(deny_unknown_fields)]
struct DeliveryMetricsArguments {
repository: Option<String>,
environment: Option<String>,
#[serde(default = "default_metrics_window")]
window_seconds: i64,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct WorkerDrainArguments {
worker: String,
draining: bool,
}
#[derive(Deserialize, Default)]
#[serde(deny_unknown_fields)]
struct AuditEventsArguments {
repository: Option<String>,
#[serde(default = "default_audit_limit")]
limit: usize,
#[serde(default)]
offset: usize,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct IssueCommentArguments {
repository: String,
number: i64,
body: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RepoSyncArguments {
name: String,
source: PathBuf,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ReleaseArguments {
repository: String,
source: PathBuf,
#[serde(default = "default_bump")]
bump: String,
#[serde(default)]
notes: Option<String>,
#[serde(default)]
wait: bool,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RepoRenameArguments {
old: String,
new: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RepoNameArguments {
name: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RepoVisibilityArguments {
repository: String,
visibility: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RepoTreeArguments {
repository: String,
#[serde(rename = "ref")]
reference: Option<String>,
path: Option<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RepoBlobArguments {
repository: String,
#[serde(rename = "ref")]
reference: Option<String>,
path: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct InitArguments {
out: PathBuf,
#[serde(default)]
force: bool,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RunQueueArguments {
repository: String,
git_ref: Option<String>,
commit: Option<String>,
#[serde(default)]
wait: bool,
}
#[derive(Deserialize, Default)]
#[serde(deny_unknown_fields)]
struct RunArguments {
#[serde(alias = "repository")]
repo: Option<StringList>,
status: Option<StringList>,
git_ref: Option<String>,
trigger: Option<StringList>,
#[serde(alias = "q")]
search: Option<String>,
#[serde(default = "default_run_limit")]
limit: usize,
#[serde(default)]
offset: usize,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RunIdArguments {
id: i64,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RunWaitArguments {
id: i64,
#[serde(default = "default_wait_timeout")]
timeout: u64,
#[serde(default = "default_wait_interval")]
interval: u64,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RunLogsArguments {
id: i64,
#[serde(default)]
failed: bool,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RunRetryArguments {
id: i64,
#[serde(default)]
wait: bool,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RunPromoteArguments {
id: i64,
environment: String,
#[serde(default)]
wait: bool,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ArtifactGetArguments {
id: i64,
output: PathBuf,
#[serde(default)]
force: bool,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct InstallationRegisterArguments {
installation_id: String,
organization_id: String,
workspace_id: String,
app_id: String,
idempotency_key: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct TenantDeploymentsArguments {
run_id: i64,
organization_id: String,
workspace_id: String,
}
#[derive(Deserialize)]
#[serde(untagged)]
enum StringList {
One(String),
Many(Vec<String>),
}
impl StringList {
fn values(self) -> Vec<String> {
let values = match self {
Self::One(value) => vec![value],
Self::Many(values) => values,
};
values
.into_iter()
.flat_map(|value| {
value
.split(',')
.map(str::trim)
.filter(|item| !item.is_empty())
.map(str::to_owned)
.collect::<Vec<_>>()
})
.collect()
}
}
const fn default_issue_limit() -> usize {
50
}
const fn default_audit_limit() -> usize {
100
}
const fn default_repository_limit() -> usize {
50
}
const fn default_metrics_window() -> i64 {
7 * 86_400
}
const fn default_run_limit() -> usize {
20
}
const fn default_wait_timeout() -> u64 {
3600
}
const fn default_wait_interval() -> u64 {
2
}
fn default_branch() -> String {
"main".to_owned()
}
fn default_bump() -> String {
"patch".to_owned()
}
fn tools() -> Vec<Value> {
vec![
json!({
"name": "akurai_doctor",
"description": "Run controller diagnostics (database readiness, git, docker) without changing state.",
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": false}
}),
json!({
"name": "akurai_repo_list",
"description": "Query registered Git repositories by text and visibility.",
"inputSchema": {
"type": "object",
"properties": {
"search": {"type": "string", "description": "LIKE search across name, URL, and default branch."},
"visibility": {"type": "string", "enum": ["private", "public"]},
"limit": {"type": "integer", "minimum": 1, "maximum": 500, "default": 50},
"offset": {"type": "integer", "minimum": 0, "maximum": 10000, "default": 0}
},
"additionalProperties": false
}
}),
json!({
"name": "akurai_repo_add",
"description": "Register an external Git repository by URL for CI.",
"inputSchema": {
"type": "object",
"properties": {
"name": {"type": "string"},
"url": {"type": "string"},
"branch": {"type": "string", "default": "main"}
},
"required": ["name", "url"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_repo_host",
"description": "Create a bare mirror owned by AkurAI Build from a trusted Titan checkout and register it for CI.",
"inputSchema": {
"type": "object",
"properties": {
"name": {"type": "string"},
"source": {"type": "string", "description": "Absolute path to the trusted Titan checkout."},
"branch": {"type": "string", "default": "main"}
},
"required": ["name", "source"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_repo_sync",
"description": "Synchronize an existing AkurAI Build mirror from its trusted Titan checkout.",
"inputSchema": {
"type": "object",
"properties": {
"name": {"type": "string"},
"source": {"type": "string", "description": "Absolute path to the trusted Titan checkout."}
},
"required": ["name", "source"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_release",
"description": "Release a product: bump the version, cut the changelog, commit + tag in the trusted checkout, sync the mirror, and queue a pipeline run for the release commit. The single maintained release path for every product.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string", "description": "Registered repository name."},
"source": {"type": "string", "description": "Absolute path to the trusted Titan checkout."},
"bump": {"type": "string", "enum": ["patch", "minor", "major"], "default": "patch"},
"notes": {"type": "string", "description": "Changelog body for the cut section; defaults to the changelog's Unreleased content."},
"wait": {"type": "boolean", "default": false, "description": "Block until the queued run reaches a terminal or approval-required state."}
},
"required": ["repository", "source"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_repo_rename",
"description": "Rename a registered repository.",
"inputSchema": {
"type": "object",
"properties": {"old": {"type": "string"}, "new": {"type": "string"}},
"required": ["old", "new"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_repo_remove",
"description": "Unregister a repository and delete its hosted mirror. Cascades runs and artifacts. Destructive.",
"inputSchema": {
"type": "object",
"properties": {"name": {"type": "string"}},
"required": ["name"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_repo_visibility",
"description": "Set a repository's dashboard visibility.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string"},
"visibility": {"type": "string", "enum": ["private", "public"]}
},
"required": ["repository", "visibility"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_repo_update",
"description": "Safely update a registered repository URL and/or default branch.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string", "minLength": 1, "maxLength": 64},
"url": {"type": "string", "minLength": 1, "maxLength": 2048},
"default_branch": {"type": "string", "minLength": 1, "maxLength": 200}
},
"required": ["repository"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_issue_list",
"description": "List numbered issues for a repository with bounded pagination.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string", "minLength": 1, "maxLength": 64},
"state": {"type": "string", "enum": ["open", "closed"]},
"limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 50},
"offset": {"type": "integer", "minimum": 0, "maximum": 10000, "default": 0}
},
"required": ["repository"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_issue_create",
"description": "Open a repository issue as the administrator integration.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string", "minLength": 1, "maxLength": 64},
"title": {"type": "string", "minLength": 1, "maxLength": 256},
"body": {"type": "string", "maxLength": 65536}
},
"required": ["repository", "title"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_issue_update",
"description": "Edit or open/close a repository issue as the administrator integration.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string", "minLength": 1, "maxLength": 64},
"number": {"type": "integer", "minimum": 1},
"title": {"type": "string", "minLength": 1, "maxLength": 256},
"body": {"type": "string", "maxLength": 65536},
"state": {"type": "string", "enum": ["open", "closed"]}
},
"required": ["repository", "number"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_issue_comment",
"description": "Add a comment to a repository issue as the administrator integration.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string", "minLength": 1, "maxLength": 64},
"number": {"type": "integer", "minimum": 1},
"body": {"type": "string", "minLength": 1, "maxLength": 65536}
},
"required": ["repository", "number", "body"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_repo_branches",
"description": "List remote branch heads for a registered repository via git ls-remote.",
"inputSchema": {
"type": "object",
"properties": {"name": {"type": "string"}},
"required": ["name"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_repo_tree",
"description": "Browse a repository's file tree at a Git ref and sub-path.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string"},
"ref": {"type": "string", "description": "Git ref; defaults to the repository's HEAD."},
"path": {"type": "string", "description": "Sub-path within the tree; defaults to the root."}
},
"required": ["repository"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_repo_blob",
"description": "Read a bounded UTF-8 file from a repository at a Git ref and path.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string"},
"ref": {"type": "string", "description": "Git ref; defaults to the repository's default branch."},
"path": {"type": "string", "description": "File path within the repository."}
},
"required": ["repository", "path"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_init",
"description": "Write a minimal .akurai.yml pipeline template to a path on the Titan filesystem.",
"inputSchema": {
"type": "object",
"properties": {
"out": {"type": "string"},
"force": {"type": "boolean", "default": false}
},
"required": ["out"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_run_queue",
"description": "Queue a pipeline run for a registered repository, optionally blocking until it reaches a terminal or approval-required state.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string"},
"git_ref": {"type": "string"},
"commit": {"type": "string", "description": "Full hexadecimal commit SHA to pin."},
"wait": {"type": "boolean", "default": false}
},
"required": ["repository"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_runs",
"description": "Query runs across one, many, or all repositories with composable filters.",
"inputSchema": {
"type": "object",
"properties": {
"repo": string_or_array("Repository name, comma-separated names, or an array."),
"repository": string_or_array("Repository name, comma-separated names, or an array."),
"status": string_or_array("Status or statuses: queued, running, waiting, succeeded, failed, canceled, interrupted."),
"git_ref": {"type": "string", "description": "Exact Git ref."},
"trigger": string_or_array("Trigger or triggers: manual, webhook, retry."),
"search": {"type": "string", "description": "LIKE search across repository, ref, commit SHA, and retained error."},
"limit": {"type": "integer", "minimum": 1, "maximum": 200, "default": 20},
"offset": {"type": "integer", "minimum": 0, "maximum": 10000, "default": 0}
},
"additionalProperties": false
}
}),
json!({
"name": "akurai_run_show",
"description": "Read one run with jobs, bounded logs, artifacts, and deployments.",
"inputSchema": {
"type": "object",
"properties": {"id": {"type": "integer", "minimum": 1}},
"required": ["id"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_run_cancel",
"description": "Cancel a queued or waiting run (and its unstarted jobs). Running and terminal runs are refused; the worker owns a running run.",
"inputSchema": {
"type": "object",
"properties": {"id": {"type": "integer", "minimum": 1}},
"required": ["id"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_run_wait",
"description": "Poll a run until it reaches a persisted terminal or approval-required state, or the 25s blocking budget elapses (then returns the latest run state; call again to keep waiting).",
"inputSchema": {
"type": "object",
"properties": {
"id": {"type": "integer", "minimum": 1},
"timeout": {"type": "integer", "minimum": 1, "maximum": 14400, "default": 3600},
"interval": {"type": "integer", "minimum": 1, "maximum": 60, "default": 2}
},
"required": ["id"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_run_logs",
"description": "Return job logs for a run, optionally filtered to failed/canceled/interrupted jobs.",
"inputSchema": {
"type": "object",
"properties": {
"id": {"type": "integer", "minimum": 1},
"failed": {"type": "boolean", "default": false}
},
"required": ["id"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_run_retry",
"description": "Queue the same immutable revision as a prior run, optionally waiting.",
"inputSchema": {
"type": "object",
"properties": {
"id": {"type": "integer", "minimum": 1},
"wait": {"type": "boolean", "default": false}
},
"required": ["id"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_run_promote",
"description": "Approve and resume a protected environment on a run; the worker daemon executes it. With wait, polls up to 25s and returns the latest run state. Production-impacting.",
"inputSchema": {
"type": "object",
"properties": {
"id": {"type": "integer", "minimum": 1},
"environment": {"type": "string"},
"wait": {"type": "boolean", "default": false}
},
"required": ["id", "environment"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_artifact_get",
"description": "Copy a build artifact to a path on the Titan filesystem.",
"inputSchema": {
"type": "object",
"properties": {
"id": {"type": "integer", "minimum": 1},
"output": {"type": "string"},
"force": {"type": "boolean", "default": false}
},
"required": ["id", "output"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_delivery_metrics",
"description": "Read source-derived DORA delivery metrics with freshness, sample size, and immutable run/deployment drill-down IDs.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string"},
"environment": {"type": "string"},
"window_seconds": {"type": "integer", "minimum": 1, "maximum": 2592000, "default": 604800}
},
"additionalProperties": false
}
}),
json!({
"name": "akurai_installation_register",
"description": "Register an installation revision from trusted provisioning, idempotently.",
"inputSchema": {
"type": "object",
"properties": {
"installation_id": {"type": "string"},
"organization_id": {"type": "string"},
"workspace_id": {"type": "string"},
"app_id": {"type": "string"},
"idempotency_key": {"type": "string"}
},
"required": ["installation_id", "organization_id", "workspace_id", "app_id", "idempotency_key"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_deployments_for_tenant",
"description": "Read managed deployments for a run within an authorized organization and workspace tenant.",
"inputSchema": {
"type": "object",
"properties": {
"run_id": {"type": "integer", "minimum": 1},
"organization_id": {"type": "string"},
"workspace_id": {"type": "string"}
},
"required": ["run_id", "organization_id", "workspace_id"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_pipeline_validate",
"description": "Fetch and validate a pipeline at an immutable resolved Git ref without queueing a run.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string", "minLength": 1, "maxLength": 64},
"ref": {"type": "string", "minLength": 1, "maxLength": 200, "description": "Branch, tag, or full immutable commit; branches are resolved to their current commit before validation."}
},
"required": ["repository"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_cache_stats",
"description": "Read bounded on-disk pipeline cache entries, file counts, bytes, and modification evidence.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string", "minLength": 1, "maxLength": 64}
},
"additionalProperties": false
}
}),
json!({
"name": "akurai_cache_prune",
"description": "Prune cache entries older than max_age_days; use dry_run to inspect eligible entries without deleting them.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string", "minLength": 1, "maxLength": 64},
"max_age_days": {"type": "integer", "minimum": 1, "maximum": 3650},
"dry_run": {"type": "boolean", "default": false}
},
"required": ["max_age_days"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_worker_drain",
"description": "Set or clear a worker drain flag. Draining workers finish their leased run but do not claim new work.",
"inputSchema": {
"type": "object",
"properties": {
"worker": {"type": "string", "minLength": 1, "maxLength": 64},
"draining": {"type": "boolean"}
},
"required": ["worker", "draining"],
"additionalProperties": false
}
}),
json!({
"name": "akurai_audit_events",
"description": "Read bounded consequential-operation audit events, optionally filtered by repository.",
"inputSchema": {
"type": "object",
"properties": {
"repository": {"type": "string", "minLength": 1, "maxLength": 64},
"limit": {"type": "integer", "minimum": 1, "maximum": 500, "default": 100},
"offset": {"type": "integer", "minimum": 0, "maximum": 10000, "default": 0}
},
"additionalProperties": false
}
}),
json!({
"name": "akurai_workers",
"description": "List Titan worker status, capacity, current run, heartbeat, and completed-run count.",
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": false}
}),
]
}
fn string_or_array(description: &str) -> Value {
json!({
"description": description,
"oneOf": [
{"type": "string"},
{"type": "array", "items": {"type": "string"}, "maxItems": 32}
]
})
}
fn tool_result(data: Value, is_error: bool) -> Value {
let text = if is_error {
data.as_str().unwrap_or("tool failed").to_owned()
} else {
serde_json::to_string_pretty(&data).unwrap_or_else(|_| "null".into())
};
json!({"content": [{"type": "text", "text": text}], "isError": is_error})
}
fn json_rpc_result(id: Value, result: Value) -> Value {
json!({"jsonrpc": "2.0", "id": id, "result": result})
}
fn json_rpc_error(id: Value, code: i64, message: impl Into<String>) -> Value {
json!({"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message.into()}})
}
#[cfg(test)]
mod tests {
use super::*;
const KEY: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
fn context() -> Ctx {
let database = Database::memory(KEY).expect("open in-memory database");
database.migrate().expect("migrate schema");
let root = std::env::temp_dir().join(format!("akurai-mcp-test-{}", std::process::id()));
Ctx {
database,
root: root.clone(),
allow_native: false,
tree: RepoTree::new(&root).expect("create tree root"),
actor: "test".to_owned(),
}
}
#[test]
fn lists_tools() {
let tools = tools();
let names: Vec<&str> = tools
.iter()
.map(|tool| tool["name"].as_str().expect("tool name"))
.collect();
assert!(names.contains(&"akurai_repo_add"));
assert!(names.contains(&"akurai_run_queue"));
assert!(names.contains(&"akurai_run_promote"));
assert!(names.contains(&"akurai_run_cancel"));
assert!(names.contains(&"akurai_repo_visibility"));
assert!(names.contains(&"akurai_issue_list"));
assert!(names.contains(&"akurai_issue_create"));
assert!(names.contains(&"akurai_issue_update"));
assert!(names.contains(&"akurai_issue_comment"));
assert!(names.contains(&"akurai_repo_branches"));
assert!(names.contains(&"akurai_repo_tree"));
assert!(names.contains(&"akurai_repo_blob"));
for name in [
"akurai_pipeline_validate",
"akurai_cache_stats",
"akurai_cache_prune",
"akurai_worker_drain",
"akurai_audit_events",
"akurai_repo_update",
] {
assert!(names.contains(&name), "missing {name}");
}
}
#[test]
fn operation_schemas_are_strict() {
for tool in tools() {
if matches!(
tool["name"].as_str(),
Some(
"akurai_pipeline_validate"
| "akurai_cache_stats"
| "akurai_cache_prune"
| "akurai_worker_drain"
| "akurai_audit_events"
| "akurai_repo_update"
)
) {
assert_eq!(tool["inputSchema"]["additionalProperties"], false);
}
}
}
#[test]
fn installation_registration_tool_is_idempotent() -> Result<()> {
let context = context();
let first = call_tool(
&context,
"akurai_installation_register",
json!({
"installation_id": "inst-1",
"organization_id": "org-1",
"workspace_id": "ws-1",
"app_id": "app-1",
"idempotency_key": "request-1"
}),
)?;
let second = call_tool(
&context,
"akurai_installation_register",
json!({
"installation_id": "inst-1",
"organization_id": "org-1",
"workspace_id": "ws-1",
"app_id": "app-1",
"idempotency_key": "request-1"
}),
)?;
assert_eq!(first["revision"], 1);
assert_eq!(second["revision"], 1);
Ok(())
}
#[test]
fn repo_lifecycle_through_tools() {
let context = context();
let added = call_tool(
&context,
"akurai_repo_add",
json!({"name": "demo", "url": "https://example.com/demo.git", "branch": "main"}),
)
.expect("add repository");
assert_eq!(added["name"], "demo");
let listed = call_tool(&context, "akurai_repo_list", json!({"search": "demo"}))
.expect("list repositories");
assert_eq!(listed.as_array().expect("array").len(), 1);
let renamed = call_tool(
&context,
"akurai_repo_rename",
json!({"old": "demo", "new": "demo2"}),
)
.expect("rename repository");
assert_eq!(renamed["name"], "demo2");
let visibility = call_tool(
&context,
"akurai_repo_visibility",
json!({"repository": "demo2", "visibility": "public"}),
)
.expect("set visibility");
assert_eq!(visibility["visibility"], "public");
let removed = call_tool(&context, "akurai_repo_remove", json!({"name": "demo2"}))
.expect("remove repository");
assert_eq!(removed["removed"], true);
}
#[test]
fn community_tools_share_numbering_and_state_invariants() -> Result<()> {
let context = context();
call_tool(
&context,
"akurai_repo_add",
json!({"name": "community", "url": "https://example.com/community.git", "branch": "main"}),
)?;
let first = call_tool(
&context,
"akurai_issue_create",
json!({"repository": "community", "title": "First", "body": "Details"}),
)?;
assert_eq!(first["number"], 1);
let comment = call_tool(
&context,
"akurai_issue_comment",
json!({"repository": "community", "number": 1, "body": "Discussed"}),
)?;
assert_eq!(comment["issue_number"], 1);
let closed = call_tool(
&context,
"akurai_issue_update",
json!({"repository": "community", "number": 1, "state": "closed"}),
)?;
assert_eq!(closed["state"], "closed");
let listed = call_tool(
&context,
"akurai_issue_list",
json!({"repository": "community", "state": "closed"}),
)?;
assert_eq!(listed["total"], 1);
assert_eq!(listed["issues"][0]["number"], 1);
let unknown = call_tool(
&context,
"akurai_issue_list",
json!({"repository": "community", "unknown": true}),
);
assert!(unknown.is_err());
Ok(())
}
#[test]
fn unknown_tool_errors() {
let context = context();
let error = call_tool(&context, "akurai_nope", json!({})).expect_err("should fail");
assert!(error.to_string().contains("unknown tool"));
}
#[test]
fn initialize_and_tools_list_round_trip() {
let context = context();
let init = handle(
&context,
&json!({"jsonrpc":"2.0","id":1,"method":"initialize"}),
)
.expect("initialize response");
assert_eq!(init["result"]["serverInfo"]["name"], "akurai-build");
let list = handle(
&context,
&json!({"jsonrpc":"2.0","id":2,"method":"tools/list"}),
)
.expect("tools/list response");
assert!(
!list["result"]["tools"]
.as_array()
.expect("array")
.is_empty()
);
}
#[test]
fn repository_blob_tool_reads_file_at_commit() -> Result<()> {
let temporary = tempfile::tempdir()?;
let source = temporary.path().join("source");
fs::create_dir(&source)?;
let git = |arguments: &[&str]| -> Result<()> {
let output = Command::new("git")
.current_dir(&source)
.args(arguments)
.output()?;
ensure!(
output.status.success(),
"git failed: {}",
String::from_utf8_lossy(&output.stderr)
);
Ok(())
};
git(&["init", "-b", "main"])?;
fs::write(source.join("fixture.txt"), "fixture")?;
git(&["add", "fixture.txt"])?;
git(&[
"-c",
"user.name=test",
"-c",
"user.email=test@test",
"commit",
"-m",
"init",
])?;
let output = Command::new("git")
.current_dir(&source)
.args(["rev-parse", "HEAD"])
.output()?;
ensure!(output.status.success(), "git rev-parse failed");
let sha = String::from_utf8(output.stdout)?.trim().to_owned();
let database = Database::memory(KEY)?;
database.migrate()?;
database.add_repository(
"demo",
source.to_str().context("UTF-8 source path")?,
"main",
)?;
let root = temporary.path().join("data");
let context = Ctx {
database,
root: root.clone(),
allow_native: false,
tree: RepoTree::new(&root)?,
actor: "test".to_owned(),
};
let blob = call_tool(
&context,
"akurai_repo_blob",
json!({"repository": "demo", "ref": sha, "path": "fixture.txt"}),
)?;
assert_eq!(blob["content"], "fixture");
assert_eq!(blob["path"], "fixture.txt");
Ok(())
}
}