use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::path::{Path, PathBuf};
use crate::agent::Taint;
use crate::session::Session;
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum OutboxKind {
#[default]
Message,
Publish,
}
impl OutboxKind {
pub fn as_str(&self) -> &'static str {
match self {
OutboxKind::Message => "message",
OutboxKind::Publish => "publish",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OutboxItem {
pub id: String,
pub status: String,
pub tool: String,
#[serde(default)]
pub kind: OutboxKind,
pub args_before: Value,
pub args: Value,
pub summary: String,
#[serde(default)]
pub session_id: Option<String>,
#[serde(default)]
pub workspace: Option<PathBuf>,
#[serde(default)]
pub taint: Taint,
pub created_at: String,
#[serde(default)]
pub resolved_at: Option<String>,
#[serde(default)]
pub reason: Option<String>,
#[serde(default)]
pub error: Option<String>,
}
impl OutboxItem {
pub fn edited(&self) -> bool {
self.args != self.args_before
}
pub fn mineable_as_writing(&self) -> bool {
self.kind == OutboxKind::Message && self.status == "sent" && self.edited()
}
}
pub struct OutboxRoute {
pub store: OutboxStore,
routed: std::collections::BTreeSet<String>,
publishes: std::collections::BTreeSet<String>,
session_id: std::sync::Mutex<Option<String>>,
}
impl OutboxRoute {
pub fn new(
store: OutboxStore,
routed: impl IntoIterator<Item = String>,
publishes: impl IntoIterator<Item = String>,
) -> Self {
OutboxRoute {
store,
routed: routed.into_iter().collect(),
publishes: publishes.into_iter().collect(),
session_id: std::sync::Mutex::new(None),
}
}
pub fn routes(&self, tool: &str) -> bool {
self.routed.contains(tool)
}
pub fn routed(&self) -> impl Iterator<Item = &str> {
self.routed.iter().map(String::as_str)
}
pub fn kind_of(&self, tool: &str) -> OutboxKind {
if self.publishes.contains(tool) {
OutboxKind::Publish
} else {
OutboxKind::Message
}
}
pub fn publishes(&self) -> impl Iterator<Item = &str> {
self.publishes.iter().map(String::as_str)
}
pub fn set_session_id(&self, id: &str) {
if let Ok(mut slot) = self.session_id.lock() {
*slot = Some(id.to_string());
}
}
pub fn session_id(&self) -> Option<String> {
self.session_id.lock().ok().and_then(|s| s.clone())
}
}
pub struct OutboxStore {
root: PathBuf,
}
pub struct OutboxLock {
_file: std::fs::File,
}
impl OutboxStore {
pub fn default_root() -> Result<PathBuf> {
if let Ok(dir) = std::env::var("MECHA_OUTBOX_DIR") {
return Ok(PathBuf::from(dir));
}
Ok(crate::work::mecha_home()?.join("outbox"))
}
pub fn open(root: impl Into<PathBuf>) -> Result<Self> {
let root = root.into();
crate::create_private_dir(&root).with_context(|| format!("creating {}", root.display()))?;
Ok(OutboxStore { root })
}
pub fn open_existing_default() -> Option<Self> {
let root = Self::default_root().ok()?;
root.is_dir().then_some(OutboxStore { root })
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn stage(
&self,
tool: &str,
kind: OutboxKind,
args: Value,
taint: Taint,
session_id: Option<String>,
workspace: Option<PathBuf>,
) -> Result<OutboxItem> {
let item = OutboxItem {
id: Session::new_id(),
status: "pending".into(),
tool: tool.to_string(),
kind,
summary: summarize(tool, &args),
args_before: args.clone(),
args,
session_id,
workspace,
taint,
created_at: chrono::Utc::now().to_rfc3339(),
resolved_at: None,
reason: None,
error: None,
};
self.write_item(&item)?;
Ok(item)
}
pub fn items(&self) -> Result<Vec<OutboxItem>> {
let mut out = Vec::new();
for entry in std::fs::read_dir(&self.root)? {
let path = entry?.path();
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
match serde_json::from_str(&std::fs::read_to_string(&path)?) {
Ok(item) => out.push(item),
Err(e) => {
tracing::warn!("skipping unreadable outbox item {}: {e}", path.display())
}
}
}
out.sort_by(|a: &OutboxItem, b: &OutboxItem| a.id.cmp(&b.id));
Ok(out)
}
pub fn item(&self, id: &str) -> Result<OutboxItem> {
let all = self.items()?;
let matches: Vec<&OutboxItem> = all.iter().filter(|i| i.id.starts_with(id)).collect();
match matches.len() {
0 => anyhow::bail!("no outbox item matching `{id}`"),
1 => Ok(matches[0].clone()),
n => anyhow::bail!(
"`{id}` matches {n} outbox items: {}",
matches
.iter()
.map(|i| i.id.as_str())
.collect::<Vec<_>>()
.join(", ")
),
}
}
pub fn update_args(&self, id: &str, args: Value) -> Result<OutboxItem> {
let mut item = self.item(id)?;
anyhow::ensure!(
item.status == "pending",
"outbox item {} is {}, not pending",
item.id,
item.status
);
item.args = args;
item.summary = summarize(&item.tool, &item.args);
self.write_item(&item)?;
Ok(item)
}
pub fn resolve(&self, id: &str, status: &str, reason: Option<String>) -> Result<OutboxItem> {
let mut item = self.item(id)?;
anyhow::ensure!(
item.status == "pending",
"outbox item {} is {}, not pending",
item.id,
item.status
);
item.status = status.to_string();
item.resolved_at = Some(chrono::Utc::now().to_rfc3339());
item.reason = reason;
item.error = None;
self.write_item(&item)?;
Ok(item)
}
pub fn record_error(&self, id: &str, error: &str) -> Result<()> {
let mut item = self.item(id)?;
item.error = Some(error.to_string());
self.write_item(&item)
}
fn write_item(&self, item: &OutboxItem) -> Result<()> {
let path = self.root.join(format!("{}.json", item.id));
let tmp = path.with_extension("json.tmp");
std::fs::write(&tmp, serde_json::to_string_pretty(item)?)?;
std::fs::rename(&tmp, &path)?;
Ok(())
}
pub fn lock(&self) -> Result<OutboxLock> {
use std::os::unix::io::AsRawFd;
let file = std::fs::OpenOptions::new()
.create(true)
.truncate(false)
.write(true)
.open(self.root.join(".lock"))?;
if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) } != 0 {
return Err(std::io::Error::last_os_error()).context("locking the outbox");
}
Ok(OutboxLock { _file: file })
}
}
pub fn diff_args(before: &Value, after: &Value) -> String {
let pretty = |v: &Value| serde_json::to_string_pretty(v).unwrap_or_else(|_| v.to_string());
let b = pretty(before);
let a = pretty(after);
let b_lines: Vec<&str> = b.lines().collect();
let a_lines: Vec<&str> = a.lines().collect();
let mut out = String::new();
for line in &b_lines {
if !a_lines.contains(line) {
out.push_str(&format!(" - {line}\n"));
}
}
for line in &a_lines {
if !b_lines.contains(line) {
out.push_str(&format!(" + {line}\n"));
}
}
if out.is_empty() {
out.push_str(" (no textual change)\n");
}
out
}
fn summarize(tool: &str, args: &Value) -> String {
let text = headline(args).unwrap_or_else(|| serde_json::to_string(args).unwrap_or_default());
format!("{tool} {}", clip(text, 80))
}
fn headline(args: &Value) -> Option<String> {
let map = args.as_object()?;
let field = |key: &str| {
map.get(key)
.and_then(|v| match v {
Value::String(s) => Some(s.clone()),
Value::Array(a) => Some(
a.iter()
.filter_map(|x| x.as_str())
.collect::<Vec<_>>()
.join(", "),
),
_ => None,
})
.filter(|s| !s.trim().is_empty())
};
let to = field("to");
let subject = field("subject").or_else(|| field("title"));
match (to, subject) {
(Some(to), Some(subject)) => Some(format!("to {to} — \"{subject}\"")),
(Some(to), None) => Some(format!("to {to}")),
(None, Some(subject)) => Some(format!("\"{subject}\"")),
(None, None) => None,
}
}
fn clip(mut text: String, max: usize) -> String {
if text.len() > max {
let cut = (0..=max)
.rev()
.find(|&i| text.is_char_boundary(i))
.unwrap_or(0);
text.truncate(cut);
text.push('…');
}
text
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn scratch(name: &str) -> PathBuf {
let dir =
std::env::temp_dir().join(format!("mecha-outbox-test-{name}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
dir
}
#[test]
fn a_summary_leads_with_who_and_what_when_the_arguments_say() {
assert_eq!(
summarize(
"mail__send",
&json!({"to": "a@x.org", "subject": "Tuesday?", "body_markdown": "long…"})
),
"mail__send to a@x.org — \"Tuesday?\""
);
assert_eq!(
summarize(
"mail__send",
&json!({"to": ["a@x.org", "b@x.org"], "body": "hi"})
),
"mail__send to a@x.org, b@x.org"
);
assert_eq!(
summarize(
"cal__event_create",
&json!({"title": "Standup", "start": "…"})
),
"cal__event_create \"Standup\""
);
let plain = summarize("factory__bundle_publish", &json!({"bundle": "/tmp/x"}));
assert!(plain.contains("bundle"), "{plain}");
let long = summarize("t", &json!({"to": "x".repeat(200)}));
assert!(long.len() < 120, "{}", long.len());
assert!(long.ends_with('…'), "{long}");
assert_eq!(
summarize("t", &json!({"to": "", "body": "x"})),
r#"t {"body":"x","to":""}"#
);
}
#[test]
fn an_item_round_trips_and_lists_in_id_order() {
let root = scratch("roundtrip");
let store = OutboxStore::open(&root).unwrap();
let a = store
.stage(
"web__fetch",
OutboxKind::Message,
json!({"url": "https://a"}),
Taint::default(),
None,
None,
)
.unwrap();
let b = store
.stage(
"email__send",
OutboxKind::Message,
json!({"to": "x@y"}),
Taint {
private: true,
untrusted: true,
},
Some("sess-1".into()),
None,
)
.unwrap();
let items = store.items().unwrap();
assert_eq!(items.len(), 2);
assert_eq!(items[0].id, a.id.min(b.id.clone()));
let loaded = store.item(&b.id).unwrap();
assert_eq!(loaded.tool, "email__send");
assert!(loaded.taint.trifecta_armed());
assert_eq!(loaded.session_id.as_deref(), Some("sess-1"));
assert_eq!(loaded.args, loaded.args_before);
assert!(!loaded.edited());
let _ = std::fs::remove_dir_all(&root);
}
#[test]
fn a_prefix_that_matches_two_items_is_an_error_not_a_guess() {
let root = scratch("prefix");
let store = OutboxStore::open(&root).unwrap();
store
.stage(
"t",
OutboxKind::Message,
json!({}),
Taint::default(),
None,
None,
)
.unwrap();
store
.stage(
"t",
OutboxKind::Message,
json!({}),
Taint::default(),
None,
None,
)
.unwrap();
let err = store.item("2").unwrap_err();
assert!(err.to_string().contains("matches 2"), "{err}");
let _ = std::fs::remove_dir_all(&root);
}
#[test]
fn editing_replaces_args_and_never_touches_the_baseline() {
let root = scratch("edit");
let store = OutboxStore::open(&root).unwrap();
let item = store
.stage(
"web__fetch",
OutboxKind::Message,
json!({"url": "https://a"}),
Taint::default(),
None,
None,
)
.unwrap();
let edited = store
.update_args(&item.id, json!({"url": "https://b"}))
.unwrap();
assert!(edited.edited());
assert_eq!(edited.args_before, json!({"url": "https://a"}));
assert_eq!(edited.args, json!({"url": "https://b"}));
let _ = std::fs::remove_dir_all(&root);
}
#[test]
fn the_writing_miner_takes_edited_messages_and_never_publishes() {
let root = scratch("mineable");
let store = OutboxStore::open(&root).unwrap();
let cases = [
(OutboxKind::Message, "sent", true, true),
(OutboxKind::Publish, "sent", true, false),
(OutboxKind::Message, "sent", false, false),
(OutboxKind::Message, "rejected", true, false),
(OutboxKind::Message, "pending", true, false),
];
for (kind, status, edited, expected) in cases {
let mut item = store
.stage(
"x__send",
kind,
json!({"path": "/tmp/a"}),
Taint::default(),
None,
None,
)
.unwrap();
item.status = status.into();
if edited {
item.args = json!({"path": "/tmp/b"});
}
assert_eq!(
item.mineable_as_writing(),
expected,
"{kind:?} / {status} / edited={edited}"
);
}
let _ = std::fs::remove_dir_all(&root);
}
#[test]
fn a_routes_kind_comes_from_config_and_defaults_to_message() {
let root = scratch("kindof");
let store = OutboxStore::open(&root).unwrap();
let route = OutboxRoute::new(
store,
[
"mail__send".to_string(),
"factory__bundle_publish".to_string(),
],
["factory__bundle_publish".to_string()],
);
assert_eq!(
route.kind_of("factory__bundle_publish"),
OutboxKind::Publish
);
assert_eq!(route.kind_of("mail__send"), OutboxKind::Message);
assert_eq!(route.kind_of("never__heard_of_it"), OutboxKind::Message);
let _ = std::fs::remove_dir_all(&root);
}
#[test]
fn an_item_recorded_before_kinds_existed_loads_as_a_message() {
let item: OutboxItem = serde_json::from_value(json!({
"id": "20260101-000000-abc",
"status": "sent",
"tool": "mail__send",
"args_before": {"body": "a"},
"args": {"body": "b"},
"summary": "mail__send",
"created_at": "2026-01-01T00:00:00Z",
}))
.unwrap();
assert_eq!(item.kind, OutboxKind::Message);
assert!(item.mineable_as_writing());
assert_eq!(item.workspace, None);
}
#[test]
fn a_staged_call_records_the_jail_it_was_drafted_under() {
let root = scratch("workspace");
let store = OutboxStore::open(&root).unwrap();
let jail = PathBuf::from("/home/someone/.mecha/work/morning");
let item = store
.stage(
"factory__bundle_publish",
OutboxKind::Publish,
json!({"bundle": "site", "id": "brief"}),
Taint::default(),
None,
Some(jail.clone()),
)
.unwrap();
assert_eq!(item.workspace.as_ref(), Some(&jail));
let loaded = store.item(&item.id).unwrap();
assert_eq!(loaded.workspace.as_ref(), Some(&jail));
let _ = std::fs::remove_dir_all(&root);
}
#[test]
fn resolution_rewrites_in_place_and_only_pending_resolves() {
let root = scratch("resolve");
let store = OutboxStore::open(&root).unwrap();
let item = store
.stage(
"t",
OutboxKind::Message,
json!({}),
Taint::default(),
None,
None,
)
.unwrap();
let sent = store.resolve(&item.id, "sent", None).unwrap();
assert_eq!(sent.status, "sent");
assert!(sent.resolved_at.is_some());
assert_eq!(
store.items().unwrap().len(),
1,
"resolved in place, not archived"
);
let err = store.resolve(&item.id, "rejected", None).unwrap_err();
assert!(err.to_string().contains("not pending"), "{err}");
let err = store.update_args(&item.id, json!({"x": 1})).unwrap_err();
assert!(err.to_string().contains("not pending"), "{err}");
let _ = std::fs::remove_dir_all(&root);
}
#[test]
fn a_failed_release_records_the_error_and_stays_pending() {
let root = scratch("error");
let store = OutboxStore::open(&root).unwrap();
let item = store
.stage(
"t",
OutboxKind::Message,
json!({}),
Taint::default(),
None,
None,
)
.unwrap();
store.record_error(&item.id, "server unreachable").unwrap();
let loaded = store.item(&item.id).unwrap();
assert_eq!(loaded.status, "pending");
assert_eq!(loaded.error.as_deref(), Some("server unreachable"));
let sent = store.resolve(&item.id, "sent", None).unwrap();
assert_eq!(sent.error, None);
let _ = std::fs::remove_dir_all(&root);
}
}