Menu
akurai-tasks
publicLatest change 350a37d39889e09d093509a69e7f06106958ffa2 - Harden task coordination and recovery by Ólafur Búi Ólafsson
#![forbid(unsafe_code)]
use akurai_http::{Request, Response, Server};
use akurai_json::Value;
use sha2::{Digest, Sha256};
use std::io::{self, BufRead};
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::time::{SystemTime, UNIX_EPOCH};
use tasks_api::{origin_allowed, Api, OidcConfig};
use tasks_core::{discover_sources, plan_file_board, CreateItem, Query, Store};
use tasks_mcp::Mcp;
const VERSION: &str = env!("CARGO_PKG_VERSION");
fn main() -> ExitCode {
let mut args: Vec<String> = std::env::args().skip(1).collect();
match run(&mut args) {
Ok(Some(value)) => {
println!("{}", value.to_json());
ExitCode::SUCCESS
}
Ok(None) => ExitCode::SUCCESS,
Err(error) => {
eprintln!("error: {error}");
ExitCode::FAILURE
}
}
}
fn run(args: &mut Vec<String>) -> Result<Option<Value>, String> {
if matches!(
args.first().map(String::as_str),
Some("help" | "--help" | "-h") | None
) {
print_help();
return Ok(None);
}
if matches!(
args.first().map(String::as_str),
Some("version" | "--version" | "-V")
) {
println!("akurai-tasks {VERSION}");
return Ok(None);
}
let db_path = take_option(args, "--db")?
.or_else(|| std::env::var("AKURAI_TASKS_DB").ok())
.unwrap_or_else(|| "data/tasks.db".to_string());
let mut actor = take_option(args, "--actor")?
.or_else(|| std::env::var("AKURAI_TASKS_ACTOR").ok())
.unwrap_or_else(|| "local-operator".to_string());
let idempotency = take_option(args, "--idempotency")?.unwrap_or_else(generated_key);
if args.first().map(String::as_str) == Some("restore") {
return Store::restore(
Path::new(&db_path),
Path::new(required_arg(args, 1, "restore needs BACKUP")?),
)
.map(Some)
.map_err(|error| error.to_string());
}
let _lock =
Store::acquire_exclusive_lock(Path::new(&db_path)).map_err(|error| error.to_string())?;
let store = Store::open(&db_path).map_err(|error| error.to_string())?;
if store
.bootstrap_complete()
.map_err(|error| error.to_string())?
&& !matches!(
args.first().map(String::as_str),
Some("serve" | "doctor" | "backup")
)
{
let credential = std::env::var("AKURAI_TASKS_CREDENTIAL")
.map_err(|_| "AKURAI_TASKS_CREDENTIAL is required after bootstrap".to_string())?;
actor = store
.authenticate_bearer(&credential)
.map_err(|error| error.to_string())?
.ok_or_else(|| "AKURAI_TASKS_CREDENTIAL is invalid or expired".to_string())?;
}
if args.first().map(String::as_str) == Some("mcp") {
run_mcp(Mcp::new(store, actor))?;
return Ok(None);
}
if args.first().map(String::as_str) == Some("serve") {
let host = take_option(args, "--host")?.unwrap_or_else(|| "127.0.0.1".into());
let port = take_option(args, "--port")?
.unwrap_or_else(|| "8090".into())
.parse::<u16>()
.map_err(|_| "invalid --port".to_string())?;
let frontend =
PathBuf::from(take_option(args, "--frontend")?.unwrap_or_else(|| "frontend".into()));
serve(store, actor, &host, port, frontend)?;
return Ok(None);
}
let value = match args.first().map(String::as_str) {
Some("project") => project(&store, &args[1..], &actor, &idempotency)?,
Some("item") | Some("task") => item(&store, &args[1..], &actor, &idempotency)?,
Some("board") => store
.board(required_arg(args, 1, "board needs PROJECT")?)
.map_err(|error| error.to_string())?,
Some("frontier") => store
.ready_frontier(required_arg(args, 1, "frontier needs PROJECT")?)
.map_err(|error| error.to_string())?,
Some("events") => store
.events(args.get(1).map(String::as_str))
.map_err(|error| error.to_string())?,
Some("query") | Some("search") => store
.query_items(&query_from_args(&args[1..])?)
.map_err(|error| error.to_string())?,
Some("doctor") => store.doctor().map_err(|error| error.to_string())?,
Some("backup") => store
.backup(Path::new(required_arg(args, 1, "backup needs OUTPUT")?))
.map_err(|error| error.to_string())?,
Some("import") => import_command(&store, &args[1..], &actor, &idempotency)?,
Some("auth") => auth_command(&store, &args[1..], &actor, &idempotency)?,
Some(other) => {
return Err(format!(
"unknown command '{other}' (try: akurai-tasks help)"
))
}
None => unreachable!(),
};
Ok(Some(value))
}
fn project(
store: &Store,
args: &[String],
actor: &str,
idempotency: &str,
) -> Result<Value, String> {
match args.first().map(String::as_str) {
Some("create") => {
let key = required_arg(args, 1, "project create needs KEY NAME")?;
let name = required_arg(args, 2, "project create needs KEY NAME")?;
let repos = args
.get(3)
.map(|value| {
value
.split(',')
.filter(|repo| !repo.is_empty())
.map(str::to_string)
.collect()
})
.unwrap_or_default();
store
.create_project(key, name, repos, actor, idempotency)
.map_err(|error| error.to_string())
}
Some("update") => {
let key = required_arg(args, 1, "project update needs KEY REPOS")?;
let repos = required_arg(args, 2, "project update needs KEY REPOS")?
.split(',')
.filter(|repo| !repo.is_empty())
.map(str::to_string)
.collect();
store
.update_project(key, repos, actor, idempotency)
.map_err(|error| error.to_string())
}
Some("list") => store.list_projects().map_err(|error| error.to_string()),
Some("show") => store
.get_project(required_arg(args, 1, "project show needs KEY")?)
.map_err(|error| error.to_string()),
Some("archive") => store
.archive_project(
required_arg(args, 1, "project archive needs KEY")?,
true,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("unarchive") => store
.archive_project(
required_arg(args, 1, "project unarchive needs KEY")?,
false,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("delete") => store
.delete_project(
required_arg(args, 1, "project delete needs KEY")?,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("purge") => store
.purge_project(
required_arg(args, 1, "project purge needs KEY")?,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some(other) => Err(format!("unknown project command '{other}'")),
None => Err(
"project needs create, update, list, show, archive, unarchive, delete, or purge".into(),
),
}
}
fn item(store: &Store, args: &[String], actor: &str, idempotency: &str) -> Result<Value, String> {
match args.first().map(String::as_str) {
Some("create") => {
let project = required_arg(args, 1, "item create needs PROJECT TITLE")?;
let title = required_arg(args, 2, "item create needs PROJECT TITLE")?;
let repo = option_value(args, "--repo")?;
let description = option_value(args, "--description")?.unwrap_or_default();
let priority = option_value(args, "--priority")?.unwrap_or_else(|| "normal".into());
store
.create_item(CreateItem {
project,
title,
description: &description,
repo: repo.as_deref(),
priority: &priority,
actor,
idempotency_key: idempotency,
})
.map_err(|error| error.to_string())
}
Some("list") => store
.list_items(required_arg(args, 1, "item list needs PROJECT")?)
.map_err(|error| error.to_string()),
Some("show") => store
.get_item(required_arg(args, 1, "item show needs ID")?)
.map_err(|error| error.to_string()),
Some("set-repo") => store
.set_item_repo(
required_arg(args, 1, "item set-repo needs ID REPO")?,
Some(required_arg(args, 2, "item set-repo needs ID REPO")?),
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("move") => store
.transition(
required_arg(args, 1, "item move needs ID STATE REVISION")?,
required_arg(args, 2, "item move needs ID STATE REVISION")?,
integer_arg(args, 3, "item move needs ID STATE REVISION")?,
None,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("claim") => store
.claim(
required_arg(args, 1, "item claim needs ID OWNER REVISION")?,
required_arg(args, 2, "item claim needs ID OWNER REVISION")?,
integer_arg(args, 3, "item claim needs ID OWNER REVISION")?,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("depend") => store
.add_dependency(
required_arg(args, 1, "item depend needs ID DEPENDENCY")?,
required_arg(args, 2, "item depend needs ID DEPENDENCY")?,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("comment") | Some("evidence") => store
.add_record(
args[0].as_str(),
required_arg(args, 1, "record needs ID BODY")?,
required_arg(args, 2, "record needs ID BODY")?,
None,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("handoff") => store
.handoff(
required_arg(args, 1, "handoff needs ID RECEIVER SUMMARY")?,
required_arg(args, 2, "handoff needs ID RECEIVER SUMMARY")?,
required_arg(args, 3, "handoff needs ID RECEIVER SUMMARY")?,
None,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("accept") => store
.accept_handoff(
required_arg(args, 1, "accept needs ID")?,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("block") => store
.set_blocked(
required_arg(args, 1, "block needs ID REASON REVISION")?,
Some(required_arg(args, 2, "block needs ID REASON REVISION")?),
integer_arg(args, 3, "block needs ID REASON REVISION")?,
None,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("unblock") => store
.set_blocked(
required_arg(args, 1, "unblock needs ID REVISION")?,
None,
integer_arg(args, 2, "unblock needs ID REVISION")?,
None,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some(other) => Err(format!("unknown item command '{other}'")),
None => Err("item needs a subcommand".into()),
}
}
fn query_from_args(args: &[String]) -> Result<Query, String> {
Ok(Query {
projects: option_values(args, "--project"),
repos: option_values(args, "--repo"),
states: option_values(args, "--state"),
priorities: option_values(args, "--priority"),
owners: option_values(args, "--owner"),
types: option_values(args, "--type"),
labels: option_values(args, "--label"),
parent: option_value(args, "--parent")?,
cycle: option_value(args, "--cycle")?,
text: option_value(args, "--text")?,
blocked: option_value(args, "--blocked")?
.map(|value| {
value
.parse::<bool>()
.map_err(|_| "--blocked must be true or false")
})
.transpose()?,
created_after: optional_integer(args, "--created-after")?,
created_before: optional_integer(args, "--created-before")?,
updated_after: optional_integer(args, "--updated-after")?,
updated_before: optional_integer(args, "--updated-before")?,
sort: option_value(args, "--sort")?.unwrap_or_else(|| "id".into()),
descending: args.iter().any(|arg| arg == "--desc"),
limit: optional_integer(args, "--limit")?
.unwrap_or(100)
.try_into()
.map_err(|_| "--limit must be positive")?,
byte_limit: optional_integer(args, "--byte-limit")?
.unwrap_or(1_048_576)
.try_into()
.map_err(|_| "--byte-limit must be positive")?,
cursor: option_value(args, "--cursor")?,
})
}
fn import_command(
store: &Store,
args: &[String],
actor: &str,
idempotency: &str,
) -> Result<Value, String> {
match args.first().map(String::as_str) {
Some("discover") => discover_sources(Path::new(required_arg(
args,
1,
"import discover needs ROOT",
)?))
.map_err(|error| error.to_string()),
Some("plan") => plan_file_board(
Path::new(required_arg(
args,
1,
"import plan needs BOARD PROJECT ROLE",
)?),
required_arg(args, 2, "import plan needs BOARD PROJECT ROLE")?,
required_arg(args, 3, "import plan needs BOARD PROJECT ROLE")?,
)
.map_err(|error| error.to_string()),
Some("apply") => {
let path = required_arg(args, 1, "import apply needs BUNDLE")?;
let text = std::fs::read_to_string(path).map_err(|error| error.to_string())?;
let bundle = akurai_json::parse(&text).map_err(|error| error.to_string())?;
store
.apply_import(&bundle, actor, idempotency)
.map_err(|error| error.to_string())
}
Some("verify") => store
.verify_import(
required_arg(args, 1, "import verify needs RUN")?,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
_ => Err("import needs discover, plan, apply, or verify".into()),
}
}
fn auth_command(
store: &Store,
args: &[String],
actor: &str,
idempotency: &str,
) -> Result<Value, String> {
match args.first().map(String::as_str) {
Some("bootstrap") => store
.bootstrap(
required_arg(args, 1, "auth bootstrap needs PRINCIPAL DISPLAY")?,
required_arg(args, 2, "auth bootstrap needs PRINCIPAL DISPLAY")?,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("principal") => store
.upsert_principal(
required_arg(args, 1, "auth principal needs ID KIND DISPLAY")?,
required_arg(args, 2, "auth principal needs ID KIND DISPLAY")?,
required_arg(args, 3, "auth principal needs ID KIND DISPLAY")?,
true,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
Some("role") => store
.grant_role(
required_arg(args, 1, "auth role needs PRINCIPAL SCOPE ROLE")?,
required_arg(args, 2, "auth role needs PRINCIPAL SCOPE ROLE")?,
required_arg(args, 3, "auth role needs PRINCIPAL SCOPE ROLE")?,
actor,
idempotency,
)
.map_err(|error| error.to_string()),
_ => Err("auth needs bootstrap, principal, or role".into()),
}
}
fn run_mcp(mcp: Mcp) -> Result<(), String> {
for line in io::stdin().lock().lines() {
let line = line.map_err(|error| error.to_string())?;
let response = mcp.handle_line(&line);
if !response.is_empty() {
println!("{response}");
}
}
Ok(())
}
fn serve(
store: Store,
_actor: String,
host: &str,
port: u16,
frontend: PathBuf,
) -> Result<(), String> {
let allowed_origin = std::env::var("AKURAI_TASKS_ALLOWED_ORIGIN").ok();
let fallback_token = std::env::var("AKURAI_TASKS_TOKEN").ok();
let mcp_token = std::env::var("AKURAI_TASKS_MCP_TOKEN")
.or_else(|_| std::env::var("AKURAI_TASKS_TOKEN"))
.map_err(|_| "AKURAI_TASKS_MCP_TOKEN is required".to_string())?;
let bootstrapped = store
.bootstrap_complete()
.map_err(|error| error.to_string())?;
if bootstrapped {
store
.provision_service_principal("omp-mcp", "OMP MCP service", &mcp_token)
.map_err(|error| error.to_string())?;
}
let api = if let (Ok(client_id), Ok(client_secret)) = (
std::env::var("AKURAI_TASKS_IDP_CLIENT_ID"),
std::env::var("AKURAI_TASKS_IDP_CLIENT_SECRET"),
) {
Api::production(
store.clone(),
allowed_origin.clone(),
OidcConfig {
issuer: std::env::var("AKURAI_TASKS_IDP_ISSUER")
.unwrap_or_else(|_| "https://auth.olibuijr.com".into()),
internal_url: std::env::var("AKURAI_TASKS_IDP_INTERNAL_URL")
.unwrap_or_else(|_| "http://127.0.0.1:3500".into()),
client_id,
client_secret,
redirect_uri: std::env::var("AKURAI_TASKS_IDP_REDIRECT")
.unwrap_or_else(|_| "https://akurai-tasks.olibuijr.com/auth/callback".into()),
allowed_emails: std::env::var("AKURAI_TASKS_ALLOW")
.unwrap_or_else(|_| "olibuijr@olibuijr.com".into())
.split(',')
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.collect(),
},
)?
} else {
Api::new(
store.clone(),
fallback_token.ok_or_else(|| "OIDC or AKURAI_TASKS_TOKEN is required".to_string())?,
allowed_origin.clone(),
)?
};
let mcp_store = store.clone();
let server = Server::bind(format!("{host}:{port}")).map_err(|error| error.to_string())?;
let address = server.local_addr().map_err(|error| error.to_string())?;
println!("AkurAI-Tasks");
println!(" database and interfaces ready");
println!(" → http://{address}");
server
.run(move |request: &Request| {
if request.path == "/mcp" {
if !origin_allowed(request, allowed_origin.as_deref()) {
return response(
403,
"application/json",
br#"{"error":"origin is not allowed"}"#.to_vec(),
);
}
if !matches!(request.method, akurai_http::Method::Post) {
return response(
405,
"application/json",
br#"{"error":"method not allowed"}"#.to_vec(),
);
}
let bearer = request
.header("Authorization")
.and_then(|value| value.strip_prefix("Bearer "));
let principal = bearer
.and_then(|token| mcp_store.authenticate_bearer(token).ok().flatten())
.or_else(|| {
(!bootstrapped
&& bearer.is_some_and(|token| secure_equal(token, &mcp_token)))
.then(|| "omp-mcp".to_string())
});
let Some(principal) = principal else {
return response(
401,
"application/json",
br#"{"error":"bearer authentication required"}"#.to_vec(),
);
};
let output =
Mcp::new(mcp_store.clone(), principal).handle_line(&request.body_str());
return response(
if output.is_empty() { 202 } else { 200 },
"application/json",
output.into_bytes(),
);
}
if request.path.starts_with("/api/")
|| request.path.starts_with("/auth/")
|| matches!(request.path.as_str(), "/api" | "/health" | "/ready")
{
return api.handle(request);
}
static_file(&frontend, request)
})
.map_err(|error| error.to_string())
}
fn static_file(_frontend: &Path, request: &Request) -> Response {
if !matches!(
request.method,
akurai_http::Method::Get | akurai_http::Method::Head
) {
return response(405, "text/plain", b"method not allowed".to_vec());
}
let relative = if request.path == "/" {
"index.html"
} else {
request.path.trim_start_matches('/')
};
let (content_type, bytes): (&str, &[u8]) = match relative {
"index.html" => (
"text/html; charset=utf-8",
include_bytes!("../../../frontend/index.html"),
),
"styles.css" => (
"text/css; charset=utf-8",
include_bytes!("../../../frontend/styles.css"),
),
"app.js" => (
"text/javascript; charset=utf-8",
include_bytes!("../../../frontend/app.js"),
),
_ => return Response::not_found(),
};
response(200, content_type, bytes.to_vec())
}
fn secure_equal(left: &str, right: &str) -> bool {
let left = Sha256::digest(left.as_bytes());
let right = Sha256::digest(right.as_bytes());
left.iter()
.zip(right.iter())
.fold(0u8, |diff, (a, b)| diff | (a ^ b))
== 0
}
fn response(status: u16, content_type: &str, body: Vec<u8>) -> Response {
let mut response = Response::ok()
.with_header("Content-Type", content_type)
.with_header("X-Content-Type-Options", "nosniff")
.with_header(
"Content-Security-Policy",
"default-src 'self'; style-src 'self'; script-src 'self'",
)
.with_body(content_type, body);
response.status = status;
response.reason = match status {
200 => "OK",
202 => "Accepted",
401 => "Unauthorized",
403 => "Forbidden",
405 => "Method Not Allowed",
_ => "Error",
}
.into();
response
}
fn take_option(args: &mut Vec<String>, name: &str) -> Result<Option<String>, String> {
let Some(index) = args.iter().position(|arg| arg == name) else {
return Ok(None);
};
if index + 1 >= args.len() {
return Err(format!("{name} needs a value"));
}
let value = args.remove(index + 1);
args.remove(index);
Ok(Some(value))
}
fn option_value(args: &[String], name: &str) -> Result<Option<String>, String> {
let Some(index) = args.iter().position(|arg| arg == name) else {
return Ok(None);
};
args.get(index + 1)
.cloned()
.ok_or_else(|| format!("{name} needs a value"))
.map(Some)
}
fn option_values(args: &[String], name: &str) -> Vec<String> {
let mut values = Vec::new();
let mut index = 0;
while index < args.len() {
if args[index] == name {
if let Some(value) = args.get(index + 1) {
values.extend(
value
.split(',')
.filter(|value| !value.is_empty())
.map(str::to_string),
);
}
index += 2;
} else {
index += 1;
}
}
values
}
fn optional_integer(args: &[String], name: &str) -> Result<Option<i64>, String> {
option_value(args, name)?
.map(|value| {
value
.parse::<i64>()
.map_err(|_| format!("{name} must be an integer"))
})
.transpose()
}
fn required_arg<'a>(args: &'a [String], index: usize, message: &str) -> Result<&'a str, String> {
args.get(index)
.map(String::as_str)
.ok_or_else(|| message.into())
}
fn integer_arg(args: &[String], index: usize, message: &str) -> Result<i64, String> {
required_arg(args, index, message)?
.parse()
.map_err(|_| format!("{} must be an integer", args[index]))
}
fn generated_key() -> String {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
format!("cli-{}-{nanos}", std::process::id())
}
fn print_help() {
println!("akurai-tasks {VERSION} — multi-project Kanban\n");
println!("USAGE: akurai-tasks [--db PATH] [--actor ID] [--idempotency KEY] COMMAND\n");
println!("COMMANDS");
println!(" serve [--host ADDR] [--port N] [--frontend DIR]");
println!(" mcp JSON-RPC over stdio; tools/list exposes 31 tools");
println!(
" project create KEY NAME [repo,repo] | update KEY repo,repo | list | show KEY | archive KEY | unarchive KEY | delete KEY | purge KEY"
);
println!(" item create PROJECT TITLE [--repo REPO] [--description TEXT] [--priority LEVEL]");
println!(" item list PROJECT | show ID | set-repo ID REPO | move ID STATE REVISION");
println!(" item claim ID OWNER REVISION | depend ID DEPENDENCY");
println!(" item comment ID BODY | evidence ID BODY");
println!(" item block ID REASON REVISION | unblock ID REVISION");
println!(" item handoff ID RECEIVER SUMMARY | accept ID");
println!(" query [--project KEY] [--repo REPO] [--state STATE] [--priority LEVEL]");
println!(" [--owner ID] [--type TYPE] [--label LABEL] [--parent ID] [--cycle ID]");
println!(" [--text TEXT] [--blocked BOOL] [--sort FIELD] [--desc] [--limit N]");
println!(" [--byte-limit N] [--cursor CURSOR]");
println!(" board PROJECT | frontier PROJECT | events [PROJECT] | doctor");
println!(" backup OUTPUT | restore BACKUP");
println!(" import discover ROOT | plan BOARD PROJECT ROLE | apply BUNDLE | verify RUN");
println!(" auth bootstrap PRINCIPAL DISPLAY | principal ID KIND DISPLAY");
println!(" auth role PRINCIPAL SCOPE ROLE");
println!("\nMCP PARITY");
println!(" Projects, repository binding, work list/get/create/query/transition, leases,");
println!(" dependencies, comments, evidence, block/unblock, handoffs, imports, events,");
println!(" board, Ready frontier, and system_doctor.");
}