feat(claw): the scoped agent token and the client-tool routes
The claw is the server half of the client tools. The middleware now knows two bearer classes: the human token answers every /v1 route as before; the agent token answers only the routes the contract allowlists, and only with the contract header at the matching version — consents, sessions, ingest, and the phones answer it 403, so a model holding the token cannot prompt-inject its way past the scoping to answer its own secret asks. Boot refuses the configuration where agent_token is the human token: the scoping is the safety, and one value playing both roles silently unscopes everything the shell holds. Three new routes write the tables the resident already reads: push to the notification queue, schedule and cancel tasks the resident's tick fires — the tools get native-shaped writes without the daemon growing a second brain. A JSON fallback gives the one error shape everywhere, unknown routes included. The daemon composes the tools' environment (RHEMA_URL, RHEMA_TOKEN, the tools' bin dir on PATH) and hands it to the resident's shell through the agent-tools env layer; with no agent_token it provisions nothing, exactly as before. The proof is the contract's conformance suite run against this router over real HTTP — same answers as the mock, checked at test time.
This commit is contained in:
1 parent
2fbd93fd79
commit
98bc0fd623
9 files changed
+418
-24
No files matched your search
@@ -46,6 +46,10 @@ rhema-provider-llama-cpp = { path = "../rhema-provider-llama-cpp" }
|
||||
rhema-provider-openai-compatible = { path = "../rhema-provider-openai-compatible" }
|
||||
rhema-provider-openrouter = { path = "../rhema-provider-openrouter" }
|
||||
rhema-rsd = { path = "../rhema-rsd" }
|
||||
# The tool contract: the agent-token allowlist and version the auth
|
||||
# middleware enforces (rhema-tools spec). The arrow runs claw ->
|
||||
# contract only — never the reverse.
|
||||
rhema-tools-contract = { path = "../../tools/contract" }
|
||||
# The resident tier's durable memory: its tools join the resolver chain
|
||||
rhema-memory = { path = "../rhema-memory" }
|
||||
# The shared ModelCard the backend registry's listings return
|
||||
|
||||
@@ -66,6 +66,8 @@ provider: anthropic
|
||||
model: claude-sonnet-4-5
|
||||
listen: 127.0.0.1:7799
|
||||
token: change-me # optional; when set, /v1/* requires it
|
||||
agent_token: change-me-too # the client tools' scoped bearer (rhema-tools)
|
||||
tools_bin: ~/.local/share/rhema-tools/bin # put the client tools on PATH
|
||||
poll_interval_secs: 120
|
||||
max_turn_steps: 2048 # tool rounds one chat turn may run
|
||||
watches: # a seed: imported while the table is empty
|
||||
@@ -101,6 +103,20 @@ its approval seam auto-approved: tool calls, the shell included,
|
||||
execute without a prompt. The daemon says so at startup, and the
|
||||
bearer-gated surface plus the watch list are the trust boundary.
|
||||
|
||||
`agent_token` provisions the client tools: a second bearer the daemon
|
||||
hands the agent's shell as `RHEMA_TOKEN` (with `RHEMA_URL`), valid
|
||||
only for the data routes the tool contract allowlists — watches,
|
||||
notifications, tasks — and only with the contract header at the
|
||||
matching version. It cannot reach consents, sessions, ingest, or the
|
||||
phones, so a model holding the token cannot prompt-inject its way past
|
||||
the scoping (see `specs/rhema-tools.md`; the daemon runs the
|
||||
contract's conformance suite against its own router at test time). It
|
||||
must differ from `token` — the daemon refuses to start when the two
|
||||
are the same. `tools_bin` layers a directory onto the shell PATH so
|
||||
the provisioned tools resolve. With no `agent_token`, the daemon
|
||||
provisions no tool environment: the agent's shell is exactly as it
|
||||
was.
|
||||
|
||||
The native surface:
|
||||
|
||||
```
|
||||
|
||||
@@ -56,6 +56,12 @@ pub enum ConfigError {
|
||||
/// path or workspace root cannot be located.
|
||||
#[error("cannot locate the state directory: neither XDG_STATE_HOME nor HOME is set")]
|
||||
NoStateHome,
|
||||
/// The agent token is the human token: boot refuses, because the
|
||||
/// agent token's safety is that it is scoped, and this copy is not.
|
||||
#[error(
|
||||
"agent_token is the same value as token; the agent token must be its own secret, scoped to the tool routes"
|
||||
)]
|
||||
SameAgentToken,
|
||||
/// One watch is unusable; the daemon refuses to start rather than
|
||||
/// poll a half-configured watch set.
|
||||
#[error("watch #{index} ({forge} {repo}): {reason}")]
|
||||
@@ -288,6 +294,16 @@ pub struct ClawConfig {
|
||||
/// open on its interface; set it when the port is reachable from
|
||||
/// anywhere but loopback. `/healthz` stays open either way.
|
||||
pub token: Option<String>,
|
||||
/// The scoped agent token the client tools hold: it answers only
|
||||
/// the data routes the tool contract allowlists — watches,
|
||||
/// notifications, tasks — never consents, sessions, or ingest, so
|
||||
/// handing it to the model's shell is safe (rhema-tools spec §5).
|
||||
/// Unset disables the client tools; the daemon then provisions no
|
||||
/// tool environment.
|
||||
pub agent_token: Option<String>,
|
||||
/// A directory of client-tool binaries (the rhema-tools project's)
|
||||
/// layered onto the agent's shell PATH. Unset leaves PATH alone.
|
||||
pub tools_bin: Option<PathBuf>,
|
||||
/// What to watch — the optional seed for the persistent watch table.
|
||||
/// Imported once, when the table is still empty; after that the
|
||||
/// table rules, and watches are managed in chat or over HTTP.
|
||||
@@ -319,6 +335,8 @@ impl Default for ClawConfig {
|
||||
model: None,
|
||||
listen: None,
|
||||
token: None,
|
||||
agent_token: None,
|
||||
tools_bin: None,
|
||||
transcript: None,
|
||||
watches: Vec::new(),
|
||||
calendars: Vec::new(),
|
||||
@@ -445,6 +463,14 @@ impl ClawConfig {
|
||||
/// watches from there; a daemon that starts with nothing to guard
|
||||
/// still serves chat.
|
||||
pub fn validate(&self) -> Result<(), ConfigError> {
|
||||
// The agent token's safety is that it is scoped; making it the
|
||||
// human token silently unscopes everything the model's shell
|
||||
// holds. Boot refuses the configuration instead.
|
||||
if let (Some(token), Some(agent_token)) = (&self.token, &self.agent_token) {
|
||||
if !token.is_empty() && token == agent_token {
|
||||
return Err(ConfigError::SameAgentToken);
|
||||
}
|
||||
}
|
||||
for (index, watch) in self.watches.iter().enumerate() {
|
||||
let fail = |reason: String| ConfigError::Watch {
|
||||
index,
|
||||
@@ -583,6 +609,31 @@ impl ClawConfig {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
#[test]
|
||||
fn the_agent_token_must_differ_from_the_human_token() {
|
||||
let config = ClawConfig {
|
||||
token: Some("same".to_owned()),
|
||||
agent_token: Some("same".to_owned()),
|
||||
..ClawConfig::default()
|
||||
};
|
||||
assert!(
|
||||
matches!(config.validate(), Err(ConfigError::SameAgentToken)),
|
||||
"one token playing both roles is refused"
|
||||
);
|
||||
let config = ClawConfig {
|
||||
agent_token: Some("other".to_owned()),
|
||||
..config
|
||||
};
|
||||
assert!(config.validate().is_ok(), "two tokens pass");
|
||||
// An agent token with no human token at all is the tools-only
|
||||
// surface: legal, and scoped by construction.
|
||||
let tools_only = ClawConfig {
|
||||
agent_token: Some("agent".to_owned()),
|
||||
..ClawConfig::default()
|
||||
};
|
||||
assert!(tools_only.validate().is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_transcript_is_on_unless_the_config_turns_it_off() {
|
||||
assert!(ClawConfig::default().transcript_enabled());
|
||||
|
||||
@@ -228,12 +228,34 @@ impl Daemon {
|
||||
// The forge handler resolves its watches through the table at
|
||||
// dispatch time, so a watch added mid-run is answerable without a
|
||||
// restart.
|
||||
// The client tools' connection, handed to the agent's shell as
|
||||
// environment: the URL the surface answers on, the scoped
|
||||
// agent token (the middleware keeps it to the contract's data
|
||||
// routes, so the model holding it cannot reach consents or
|
||||
// ingest), and the tools' bin dir on PATH. No token, no
|
||||
// environment — the tools are simply absent.
|
||||
let mut tool_env = std::collections::BTreeMap::new();
|
||||
if let Some(agent_token) = &config.agent_token {
|
||||
tool_env.insert(
|
||||
"RHEMA_URL".to_owned(),
|
||||
format!("http://{}", config.listen_addr()),
|
||||
);
|
||||
tool_env.insert("RHEMA_TOKEN".to_owned(), agent_token.clone());
|
||||
}
|
||||
if let Some(tools_bin) = &config.tools_bin {
|
||||
let inherited = std::env::var("PATH").unwrap_or_default();
|
||||
tool_env.insert(
|
||||
"PATH".to_owned(),
|
||||
format!("{}:{inherited}", tools_bin.display()),
|
||||
);
|
||||
}
|
||||
let forge_handler = ClawHandler::new(
|
||||
backends.clone(),
|
||||
workspace_root,
|
||||
watches.clone(),
|
||||
accounts.clone(),
|
||||
config.transcript_enabled(),
|
||||
tool_env.clone(),
|
||||
);
|
||||
// Three drivers over one store admit from the same table, so
|
||||
// each owns its slice of the key space: a forge pass must never
|
||||
@@ -364,6 +386,7 @@ impl Daemon {
|
||||
store,
|
||||
watches,
|
||||
notifications,
|
||||
tasks,
|
||||
bootstrap,
|
||||
hub,
|
||||
backends.clone(),
|
||||
@@ -375,6 +398,7 @@ impl Daemon {
|
||||
attachments,
|
||||
phone,
|
||||
config.token.clone(),
|
||||
config.agent_token.clone(),
|
||||
);
|
||||
|
||||
Ok(Self {
|
||||
|
||||
@@ -66,6 +66,10 @@ pub struct ClawHandler {
|
||||
/// reply client must authenticate as the identity the poll did, or
|
||||
/// the forge refuses the very post the poll was reading.
|
||||
accounts: Arc<dyn IdentityStore>,
|
||||
/// The client tools' environment, layered onto every dispatch's
|
||||
/// agent shell: the daemon URL, the scoped agent token, the tools'
|
||||
/// bin dir on PATH. Empty when the config issues no agent token.
|
||||
tool_env: std::collections::BTreeMap<String, String>,
|
||||
clients: Mutex<HashMap<String, Arc<dyn Forge>>>,
|
||||
}
|
||||
|
||||
@@ -78,6 +82,7 @@ impl ClawHandler {
|
||||
watches: Arc<ClawStore>,
|
||||
accounts: Arc<dyn IdentityStore>,
|
||||
transcript: bool,
|
||||
tool_env: std::collections::BTreeMap<String, String>,
|
||||
) -> Self {
|
||||
Self {
|
||||
backends,
|
||||
@@ -85,6 +90,7 @@ impl ClawHandler {
|
||||
workspace_root: workspace_root.into(),
|
||||
watches,
|
||||
accounts,
|
||||
tool_env,
|
||||
clients: Mutex::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
@@ -239,6 +245,7 @@ impl ClawHandler {
|
||||
config.transcript = self
|
||||
.transcript
|
||||
.then(rhema_transcript::TranscriptSettings::default);
|
||||
config.shell_env = self.tool_env.clone();
|
||||
config
|
||||
}
|
||||
|
||||
@@ -779,6 +786,7 @@ mod tests {
|
||||
watches,
|
||||
Arc::new(rhema_identity::VolatileIdentityStore::new()),
|
||||
true,
|
||||
std::collections::BTreeMap::new(),
|
||||
);
|
||||
|
||||
// A watch added to the table AFTER the handler was built resolves:
|
||||
@@ -820,8 +828,16 @@ mod tests {
|
||||
watches.clone(),
|
||||
accounts.clone(),
|
||||
true,
|
||||
std::collections::BTreeMap::new(),
|
||||
);
|
||||
let off = ClawHandler::new(
|
||||
registry,
|
||||
"/tmp/w",
|
||||
watches,
|
||||
accounts,
|
||||
false,
|
||||
std::collections::BTreeMap::new(),
|
||||
);
|
||||
let off = ClawHandler::new(registry, "/tmp/w", watches, accounts, false);
|
||||
assert!(
|
||||
on.agent_config("m", Path::new("/tmp/w"))
|
||||
.transcript
|
||||
|
||||
+243
-23
@@ -38,6 +38,9 @@ use rhema_autonomous_agent::Driver;
|
||||
use rhema_rsd::{
|
||||
DispatchError, Dispatcher, IngestResult, KeyRecord, KeyState, ResourceKey, SqliteStore, Store,
|
||||
};
|
||||
use rhema_tools_contract::Auth;
|
||||
use rhema_tools_contract::CONTRACT_HEADER;
|
||||
use rhema_tools_contract::CONTRACT_VERSION;
|
||||
use serde::Deserialize;
|
||||
use serde_json::{Value, json};
|
||||
use tokio::sync::broadcast::error::RecvError;
|
||||
@@ -47,6 +50,7 @@ use crate::backends::BackendRegistry;
|
||||
use crate::bootstrap::Bootstrap;
|
||||
use crate::hub::{EventHub, TurnEvent};
|
||||
use crate::notifications::NotificationStore;
|
||||
use crate::task_store::TaskStore;
|
||||
use crate::watch_store::{ClawStore, validate_watch};
|
||||
|
||||
/// How many `run_once` passes one POST may spend draining.
|
||||
@@ -64,6 +68,9 @@ pub struct ClawState {
|
||||
store: Arc<SqliteStore>,
|
||||
watches: Arc<ClawStore>,
|
||||
notifications: Arc<NotificationStore>,
|
||||
/// The scheduled-task table the `/v1/tasks` routes serve — the same
|
||||
/// rows the resident scheduler fires.
|
||||
tasks: Arc<TaskStore>,
|
||||
bootstrap: Arc<Bootstrap>,
|
||||
hub: Arc<EventHub>,
|
||||
backends: Arc<BackendRegistry>,
|
||||
@@ -84,6 +91,10 @@ pub struct ClawState {
|
||||
/// reported state the ingest door feeds.
|
||||
phone: Arc<crate::phone::PhoneStore>,
|
||||
token: Option<String>,
|
||||
/// The scoped agent token the client tools hold: a second bearer
|
||||
/// answering only the contract's data routes (see
|
||||
/// [`require_auth`]).
|
||||
agent_token: Option<String>,
|
||||
/// The generic sources `/v1/ingest` has registered on the
|
||||
/// dispatcher, so a source name registers exactly once however many
|
||||
/// events race in through it.
|
||||
@@ -92,17 +103,19 @@ pub struct ClawState {
|
||||
|
||||
impl ClawState {
|
||||
/// The state over the interactive dispatcher and its driver, the
|
||||
/// shared dispatch store, the watch table, the notification table,
|
||||
/// the bootstrap store, the event hub, the account store, the
|
||||
/// consent table, the attachment store, and the configured bearer
|
||||
/// `token`.
|
||||
#[allow(clippy::too_many_arguments)] // fifteen parts, each load-bearing
|
||||
/// shared dispatch store, the watch table, the notification and
|
||||
/// task tables, the bootstrap store, the event hub, the account
|
||||
/// store, the consent table, the attachment store, and the
|
||||
/// configured bearers — the human `token` and the scoped
|
||||
/// `agent_token`.
|
||||
#[allow(clippy::too_many_arguments)] // seventeen parts, each load-bearing
|
||||
pub fn new(
|
||||
dispatcher: Arc<Dispatcher>,
|
||||
driver: Arc<Driver>,
|
||||
store: Arc<SqliteStore>,
|
||||
watches: Arc<ClawStore>,
|
||||
notifications: Arc<NotificationStore>,
|
||||
tasks: Arc<TaskStore>,
|
||||
bootstrap: Arc<Bootstrap>,
|
||||
hub: Arc<EventHub>,
|
||||
backends: Arc<BackendRegistry>,
|
||||
@@ -114,6 +127,7 @@ impl ClawState {
|
||||
attachments: Arc<crate::attachments::AttachmentStore>,
|
||||
phone: Arc<crate::phone::PhoneStore>,
|
||||
token: Option<String>,
|
||||
agent_token: Option<String>,
|
||||
) -> Self {
|
||||
Self {
|
||||
dispatcher,
|
||||
@@ -121,6 +135,7 @@ impl ClawState {
|
||||
store,
|
||||
watches,
|
||||
notifications,
|
||||
tasks,
|
||||
bootstrap,
|
||||
hub,
|
||||
backends,
|
||||
@@ -132,6 +147,7 @@ impl ClawState {
|
||||
attachments,
|
||||
phone,
|
||||
token,
|
||||
agent_token,
|
||||
generic_sources: Arc::new(Mutex::new(BTreeSet::new())),
|
||||
}
|
||||
}
|
||||
@@ -224,7 +240,12 @@ pub fn router(state: ClawState) -> Router {
|
||||
.route("/phones", get(phones_list))
|
||||
.route("/phones/{device}", delete(phones_forget))
|
||||
.route("/ingest", post(ingest))
|
||||
.route("/notifications", get(notifications_list))
|
||||
.route(
|
||||
"/notifications",
|
||||
get(notifications_list).post(notifications_push),
|
||||
)
|
||||
.route("/tasks", get(tasks_list).post(tasks_add))
|
||||
.route("/tasks/{id}", delete(tasks_remove))
|
||||
.route("/notifications/{id}/ack", post(notifications_ack))
|
||||
.route("/agent", get(agent_get).put(agent_put))
|
||||
.route(
|
||||
@@ -244,6 +265,9 @@ pub fn router(state: ClawState) -> Router {
|
||||
.route("/chat/completions", post(crate::openai::chat_completions))
|
||||
.layer(middleware::from_fn_with_state(state.clone(), require_auth));
|
||||
Router::new()
|
||||
// The one error shape everywhere, unknown routes included: a
|
||||
// client's typo answers JSON it can read, not axum's bare text.
|
||||
.fallback(not_found)
|
||||
.route(
|
||||
"/",
|
||||
get(|| async {
|
||||
@@ -264,6 +288,15 @@ pub fn router(state: ClawState) -> Router {
|
||||
.with_state(state)
|
||||
}
|
||||
|
||||
/// The unknown-route answer: the contract's error shape.
|
||||
async fn not_found() -> Response {
|
||||
(
|
||||
StatusCode::NOT_FOUND,
|
||||
Json(json!({"error": "no such route"})),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
|
||||
/// Byte-length-safe constant-time equality over the two strings' UTF-8
|
||||
/// bytes: every compared byte is XOR-folded into one difference word
|
||||
/// along with the length difference, so the answer takes the same work
|
||||
@@ -279,33 +312,65 @@ fn tokens_match(presented: &str, expected: &str) -> bool {
|
||||
diff == 0
|
||||
}
|
||||
|
||||
/// The bearer check: with no configured token the surface is open; with
|
||||
/// one, `Authorization: Bearer <token>` is required on every `/v1`
|
||||
/// route. `/healthz` never reaches this.
|
||||
/// The bearer check, in two classes. With neither token configured the
|
||||
/// surface is open, as before. The human token answers every `/v1`
|
||||
/// route. The agent token — what the client tools hold — answers only
|
||||
/// the data routes the tool contract allowlists, and only with the
|
||||
/// contract header at the matching version: a tool cannot reach
|
||||
/// consents, sessions, or ingest, so a leaked agent token cannot
|
||||
/// answer the agent's own `ask_secret` (rhema-tools spec §5). Any
|
||||
/// other bearer is a 401. `/healthz` never reaches this.
|
||||
async fn require_auth(State(state): State<ClawState>, request: Request, next: Next) -> Response {
|
||||
let authorized = match &state.token {
|
||||
None => true,
|
||||
Some(expected) => request
|
||||
let presented = request
|
||||
.headers()
|
||||
.get(axum::http::header::AUTHORIZATION)
|
||||
.and_then(|value| value.to_str().ok())
|
||||
.is_some_and(|value| {
|
||||
value
|
||||
.strip_prefix("Bearer ")
|
||||
.is_some_and(|presented| tokens_match(presented, expected))
|
||||
}),
|
||||
.and_then(|value| value.strip_prefix("Bearer "));
|
||||
let class = match (&state.token, &state.agent_token, presented) {
|
||||
// Nothing configured: the open surface, bearers ignored.
|
||||
(None, None, _) => Auth::Human,
|
||||
(Some(expected), _, Some(presented)) if tokens_match(presented, expected) => Auth::Human,
|
||||
(_, Some(expected), Some(presented)) if tokens_match(presented, expected) => Auth::Agent,
|
||||
// A token configured and no bearer, or a bearer matching
|
||||
// neither.
|
||||
(_, _, _) => Auth::Invalid,
|
||||
};
|
||||
if authorized {
|
||||
next.run(request).await
|
||||
} else {
|
||||
// A body, however small: a silent 401 reads as a network fault
|
||||
// to every client debugging its token.
|
||||
(
|
||||
match class {
|
||||
Auth::Human => next.run(request).await,
|
||||
// Both mean "no credential": Auth::None never leaves the
|
||||
// classifier, but the match stays total.
|
||||
Auth::None | Auth::Invalid => (
|
||||
StatusCode::UNAUTHORIZED,
|
||||
Json(json!({"error": "missing or wrong bearer token"})),
|
||||
)
|
||||
.into_response(),
|
||||
Auth::Agent => {
|
||||
let version = request
|
||||
.headers()
|
||||
.get(CONTRACT_HEADER)
|
||||
.and_then(|value| value.to_str().ok())
|
||||
.is_some_and(|value| value == CONTRACT_VERSION.to_string());
|
||||
if !version {
|
||||
return (
|
||||
StatusCode::CONFLICT,
|
||||
Json(json!({"error": format!(
|
||||
"agent calls carry the {CONTRACT_HEADER} header at version {CONTRACT_VERSION}"
|
||||
)})),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
if rhema_tools_contract::agent_allowed(request.method().as_str(), request.uri().path())
|
||||
{
|
||||
next.run(request).await
|
||||
} else {
|
||||
(
|
||||
StatusCode::FORBIDDEN,
|
||||
Json(json!({"error": "this route is not the agent token's"})),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn sessions(
|
||||
@@ -1231,6 +1296,161 @@ async fn notifications_list(
|
||||
Ok(Json(json!({ "notifications": notifications })))
|
||||
}
|
||||
|
||||
/// The body `POST /v1/notifications` takes: the client tools' push.
|
||||
#[derive(Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct NotificationPushBody {
|
||||
title: String,
|
||||
body: String,
|
||||
device: Option<String>,
|
||||
}
|
||||
|
||||
/// The client tools' push: the same queue the notification poll
|
||||
/// serves, entered over the surface — one row, delivered at the
|
||||
/// device's next poll-and-ack (rhema-tools spec §6). The route the
|
||||
/// `POST /v1/notifications` client speaks.
|
||||
async fn notifications_push(
|
||||
State(state): State<ClawState>,
|
||||
Json(body): Json<NotificationPushBody>,
|
||||
) -> Response {
|
||||
if body.title.trim().is_empty() || body.body.trim().is_empty() {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({"error": "title and body must not be empty"})),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
let device = body
|
||||
.device
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|device| !device.is_empty())
|
||||
.unwrap_or(crate::notifications::DEFAULT_DEVICE);
|
||||
match state
|
||||
.notifications
|
||||
.push(device, body.title.trim(), body.body.trim())
|
||||
{
|
||||
Ok(record) => (StatusCode::OK, Json(record)).into_response(),
|
||||
Err(err) => (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(json!({"error": err.to_string()})),
|
||||
)
|
||||
.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// The body `POST /v1/tasks` takes: the client tools' schedule. `kind`
|
||||
/// picks the shape, exactly the two the resident scheduler fires.
|
||||
#[derive(Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct TaskAddBody {
|
||||
kind: String,
|
||||
prompt: String,
|
||||
at_millis: Option<i64>,
|
||||
interval_millis: Option<i64>,
|
||||
deliver_to: Option<String>,
|
||||
}
|
||||
|
||||
/// The scheduled-task table, oldest first — the same rows the
|
||||
/// resident's tick fires.
|
||||
async fn tasks_list(State(state): State<ClawState>) -> Result<Json<Value>, (StatusCode, String)> {
|
||||
let tasks = state
|
||||
.tasks
|
||||
.list()
|
||||
.map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
|
||||
Ok(Json(json!({ "tasks": tasks })))
|
||||
}
|
||||
|
||||
/// The client tools' schedule: one task, fired by the resident's tick
|
||||
/// when due (rhema-tools spec §6). The route the `POST /v1/tasks`
|
||||
/// client speaks.
|
||||
async fn tasks_add(State(state): State<ClawState>, Json(body): Json<TaskAddBody>) -> Response {
|
||||
if body.prompt.trim().is_empty() {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({"error": "prompt must not be empty"})),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
let now = now_millis();
|
||||
let scheduled = match body.kind.as_str() {
|
||||
"at" => {
|
||||
let Some(at_millis) = body.at_millis else {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({"error": "a one-shot task needs at_millis"})),
|
||||
)
|
||||
.into_response();
|
||||
};
|
||||
state.tasks.schedule_at(
|
||||
body.prompt.trim(),
|
||||
at_millis,
|
||||
now,
|
||||
body.deliver_to.as_deref(),
|
||||
)
|
||||
}
|
||||
"every" => {
|
||||
let Some(interval_millis) = body.interval_millis else {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({"error": "a recurring task needs interval_millis"})),
|
||||
)
|
||||
.into_response();
|
||||
};
|
||||
if interval_millis <= 0 {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({"error": "interval_millis must be positive"})),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
state.tasks.schedule_every(
|
||||
body.prompt.trim(),
|
||||
interval_millis,
|
||||
now,
|
||||
body.deliver_to.as_deref(),
|
||||
)
|
||||
}
|
||||
_ => {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
Json(json!({"error": "kind must be at or every"})),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
};
|
||||
match scheduled {
|
||||
Ok(record) => (StatusCode::OK, Json(record)).into_response(),
|
||||
Err(err) => (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(json!({"error": err.to_string()})),
|
||||
)
|
||||
.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Milliseconds since the Unix epoch — the task table's clock.
|
||||
fn now_millis() -> i64 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|duration| duration.as_millis() as i64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Cancels one task. Like the watch removal, the answer is
|
||||
/// `{"removed": false}` for an id already gone — a retried cancel
|
||||
/// converges instead of 404ing.
|
||||
async fn tasks_remove(State(state): State<ClawState>, Path(id): Path<i64>) -> Response {
|
||||
match state.tasks.remove(id) {
|
||||
Ok(removed) => (StatusCode::OK, Json(json!({"removed": removed}))).into_response(),
|
||||
Err(err) => (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(json!({"error": err.to_string()})),
|
||||
)
|
||||
.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Marks one notification delivered. Idempotent: `false` when the id is
|
||||
/// unknown or already acked.
|
||||
async fn notifications_ack(State(state): State<ClawState>, Path(id): Path<i64>) -> Response {
|
||||
|
||||
@@ -67,6 +67,8 @@ fn config(server: &MockServer, dir: &tempfile::TempDir) -> ClawConfig {
|
||||
model: Some("test-model".to_owned()),
|
||||
listen: None,
|
||||
token: None,
|
||||
agent_token: None,
|
||||
tools_bin: None,
|
||||
transcript: None,
|
||||
watches: vec![Watch {
|
||||
forge: ForgeKind::GitHub,
|
||||
|
||||
@@ -55,6 +55,8 @@ fn config(server: &MockServer, dir: &tempfile::TempDir) -> ClawConfig {
|
||||
model: Some("test-model".to_owned()),
|
||||
listen: None,
|
||||
token: None,
|
||||
agent_token: None,
|
||||
tools_bin: None,
|
||||
transcript: None,
|
||||
watches: vec![Watch {
|
||||
forge: ForgeKind::GitHub,
|
||||
|
||||
@@ -61,6 +61,8 @@ fn config(dir: &tempfile::TempDir, token: Option<&str>) -> ClawConfig {
|
||||
model: Some("test-model".to_owned()),
|
||||
listen: None,
|
||||
token: token.map(str::to_owned),
|
||||
agent_token: None,
|
||||
tools_bin: None,
|
||||
transcript: None,
|
||||
watches: Vec::new(),
|
||||
calendars: Vec::new(),
|
||||
@@ -1186,6 +1188,63 @@ async fn notifications_round_trip_over_http() {
|
||||
}
|
||||
|
||||
/// A registry whose only backend is the scripted mock.
|
||||
/// The tool contract's conformance suite against the daemon's real
|
||||
/// router, over real HTTP — the same suite the contract crate runs
|
||||
/// against its mock router. Same results is the definition of the
|
||||
/// daemon implementing the contract (rhema-tools spec §7): the scoped
|
||||
/// agent token reads and writes the data domains under the contract
|
||||
/// header, and consents, sessions, ingest, and the phones stay
|
||||
/// human-only.
|
||||
#[tokio::test]
|
||||
async fn the_surface_speaks_the_tool_contract() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let mut config = config(&dir, Some("human-token"));
|
||||
config.agent_token = Some("agent-token".to_owned());
|
||||
let daemon = Daemon::build(
|
||||
&config,
|
||||
mock_registry(Arc::new(MockProvider::new(Vec::new()))),
|
||||
rhema_rsd::DispatcherConfig::default(),
|
||||
)
|
||||
.expect("the daemon assembles");
|
||||
let base = serve(&daemon).await;
|
||||
let client = reqwest::Client::new();
|
||||
let result = rhema_tools_contract::conformance(|request: rhema_tools_contract::CallRequest| {
|
||||
let client = client.clone();
|
||||
let base = base.clone();
|
||||
async move {
|
||||
let method = reqwest::Method::from_bytes(request.method.as_bytes())
|
||||
.map_err(|e| e.to_string())?;
|
||||
let builder = client.request(method, format!("{base}{}", request.path));
|
||||
let builder = match request.auth {
|
||||
rhema_tools_contract::Auth::None => builder,
|
||||
rhema_tools_contract::Auth::Human => {
|
||||
builder.header(reqwest::header::AUTHORIZATION, "Bearer human-token")
|
||||
}
|
||||
rhema_tools_contract::Auth::Agent => {
|
||||
builder.header(reqwest::header::AUTHORIZATION, "Bearer agent-token")
|
||||
}
|
||||
rhema_tools_contract::Auth::Invalid => {
|
||||
builder.header(reqwest::header::AUTHORIZATION, "Bearer a-stranger")
|
||||
}
|
||||
};
|
||||
let builder = match request.contract {
|
||||
Some(version) => builder.header("x-rhema-contract", version.to_string()),
|
||||
None => builder,
|
||||
};
|
||||
let builder = match request.body {
|
||||
Some(body) => builder.json(&body),
|
||||
None => builder,
|
||||
};
|
||||
let response = builder.send().await.map_err(|e| e.to_string())?;
|
||||
let status = response.status().as_u16();
|
||||
let body = response.json().await.unwrap_or(serde_json::json!({}));
|
||||
Ok(rhema_tools_contract::CallResponse { status, body })
|
||||
}
|
||||
})
|
||||
.await;
|
||||
assert_eq!(result, Ok(()), "the daemon speaks the tool contract");
|
||||
}
|
||||
|
||||
fn mock_registry(provider: Arc<dyn rhema_core::ModelProvider>) -> Arc<rhema_claw::BackendRegistry> {
|
||||
Arc::new(
|
||||
rhema_claw::BackendRegistry::new(
|
||||
|
||||
Reference in new issue
Block a user