cancel_thread: new mcp tool — unify reminder + question cancel on both surfaces
This commit is contained in:
parent
fcd407da11
commit
b1d0a62cb9
11 changed files with 331 additions and 25 deletions
8
TODO.md
8
TODO.md
|
|
@ -40,13 +40,7 @@ how often the friction bites in normal use.
|
||||||
into the prompt builder in `hive-ag3nt::turn.rs`. Even better: add a
|
into the prompt builder in `hive-ag3nt::turn.rs`. Even better: add a
|
||||||
one-shot `recv_batch(max: u32)` MCP tool that returns up to `max`
|
one-shot `recv_batch(max: u32)` MCP tool that returns up to `max`
|
||||||
pending messages in a single round-trip.
|
pending messages in a single round-trip.
|
||||||
- **Self-management of own asks + reminders** — once I fire `ask` or
|
- ~~**Self-management of own asks + reminders**~~ ✓ landed — unified with `get_open_threads` rather than a separate listing surface. `OpenThread` gained a `Reminder { id, owner, message, due_at, age_seconds }` variant (sub-agent flavour filters by `owner == self`; manager unfiltered). New `mcp__hyperhive__cancel_thread(kind, id)` on both surfaces — `kind` is `"question"` (asker gets `[cancelled by <self>]` answer, unblocks) or `"reminder"` (hard-deleted before fire). Auth: sub-agent must own the row; manager bypasses for hive-wide cleanup. New helpers `OperatorQuestions::cancel` + `Broker::cancel_reminder_as` push the auth check down so both flavours stay aligned. Shared dispatch in `hive-c0re/src/questions.rs::handle_cancel_thread`.
|
||||||
`remind` I have no way to inspect or cancel them from the agent side.
|
|
||||||
Operator can cancel asks via dashboard; nothing for reminders at all
|
|
||||||
(TODO above). Want `list_my_asks() -> [{id, target, question, asked_at}]`
|
|
||||||
and `cancel_ask(id)` on the agent surface, plus `list_my_reminders()`
|
|
||||||
/ `cancel_reminder(id)`. Bounded by `asker == self` and `reminder.owner
|
|
||||||
== self` so no cross-agent meddling.
|
|
||||||
- **Optional `in_reply_to: <msg_id>` on send** — pure wire addition; no
|
- **Optional `in_reply_to: <msg_id>` on send** — pure wire addition; no
|
||||||
behavioural change. The dashboard could render conversation threads
|
behavioural change. The dashboard could render conversation threads
|
||||||
(already wants this for the agent-to-agent question UI in the
|
(already wants this for the agent-to-agent question UI in the
|
||||||
|
|
|
||||||
|
|
@ -7,7 +7,8 @@ Tools (hyperhive surface):
|
||||||
- (some agents only) **extra MCP tools** surfaced as `mcp__<server>__<tool>` — these are agent-specific (matrix client, scraper, db connector, etc.) declared in your `agent.nix` under `hyperhive.extraMcpServers`. Treat them as first-class tools alongside the hyperhive surface; the operator already auto-approved them at deploy time.
|
- (some agents only) **extra MCP tools** surfaced as `mcp__<server>__<tool>` — these are agent-specific (matrix client, scraper, db connector, etc.) declared in your `agent.nix` under `hyperhive.extraMcpServers`. Treat them as first-class tools alongside the hyperhive surface; the operator already auto-approved them at deploy time.
|
||||||
- `mcp__hyperhive__ask(question, options?, multi?, ttl_seconds?, to?)` — surface a structured question to the human operator (default, or `to: "operator"`) OR a peer agent (`to: "<agent-name>"`). Returns immediately with a question id — do NOT wait inline. When the recipient answers, a system message with event `question_answered { id, question, answer, answerer }` lands in your inbox; handle it on a future turn. Use this for clarifications, permission for risky actions, choice between options, or peer Q&A without burning regular inbox slots. `options` is advisory: a short fixed-choice list when applicable, otherwise leave empty for free text. `multi: true` lets the answerer pick multiple (checkboxes), answer comes back comma-joined. `ttl_seconds` auto-cancels with answer `[expired]` (and `answerer: "ttl-watchdog"`) when the decision becomes moot.
|
- `mcp__hyperhive__ask(question, options?, multi?, ttl_seconds?, to?)` — surface a structured question to the human operator (default, or `to: "operator"`) OR a peer agent (`to: "<agent-name>"`). Returns immediately with a question id — do NOT wait inline. When the recipient answers, a system message with event `question_answered { id, question, answer, answerer }` lands in your inbox; handle it on a future turn. Use this for clarifications, permission for risky actions, choice between options, or peer Q&A without burning regular inbox slots. `options` is advisory: a short fixed-choice list when applicable, otherwise leave empty for free text. `multi: true` lets the answerer pick multiple (checkboxes), answer comes back comma-joined. `ttl_seconds` auto-cancels with answer `[expired]` (and `answerer: "ttl-watchdog"`) when the decision becomes moot.
|
||||||
- `mcp__hyperhive__answer(id, answer)` — answer a question that was routed to YOU. You'll see one in your inbox as a `question_asked { id, asker, question, options, multi }` system event when a peer or the manager calls `ask(to: "<your-name>", ...)`. The answer surfaces in the asker's inbox as a `question_answered` event. Strict authorisation: you can only answer questions where you are the declared target.
|
- `mcp__hyperhive__answer(id, answer)` — answer a question that was routed to YOU. You'll see one in your inbox as a `question_asked { id, asker, question, options, multi }` system event when a peer or the manager calls `ask(to: "<your-name>", ...)`. The answer surfaces in the asker's inbox as a `question_answered` event. Strict authorisation: you can only answer questions where you are the declared target.
|
||||||
- `mcp__hyperhive__get_open_threads()` — list your loose ends: unanswered questions where you're asker (waiting on someone) or target (owing a reply). No args, cheap server-side sweep. Useful at turn start to remember what's outstanding without scanning inbox archaeology.
|
- `mcp__hyperhive__get_open_threads()` — list your loose ends: unanswered questions where you're asker (waiting on someone) or target (owing a reply), plus reminders you've scheduled that haven't fired. No args, cheap server-side sweep. Useful at turn start to remember what's outstanding without scanning inbox archaeology.
|
||||||
|
- `mcp__hyperhive__cancel_thread(kind, id)` — cancel one of your own open threads. `kind` is `"question"` (the asker — you, in this case — gets a `[cancelled by <you>]` answer so the waiter unblocks) or `"reminder"` (hard-deleted before it fires). `id` from the matching `get_open_threads` row or the original submission reply.
|
||||||
- `mcp__hyperhive__whoami()` — self-introspection: returns your canonical agent name (from socket identity, not the prompt-substituted label), role, and current hyperhive rev. No args. Use it when you want a trustworthy identity stamp for state files, commit messages, or cross-agent attribution that won't drift across renames.
|
- `mcp__hyperhive__whoami()` — self-introspection: returns your canonical agent name (from socket identity, not the prompt-substituted label), role, and current hyperhive rev. No args. Use it when you want a trustworthy identity stamp for state files, commit messages, or cross-agent attribution that won't drift across renames.
|
||||||
|
|
||||||
Need new packages, env vars, or other NixOS config for yourself? You can't edit your own config directly — message the manager (recipient `manager`) describing what you need + why. The manager evaluates the request (it doesn't rubber-stamp), edits `/agents/{label}/config/agent.nix` on your behalf, commits, and submits an approval that the operator can accept on the dashboard; on approve hive-c0re rebuilds your container with the new config.
|
Need new packages, env vars, or other NixOS config for yourself? You can't edit your own config directly — message the manager (recipient `manager`) describing what you need + why. The manager evaluates the request (it doesn't rubber-stamp), edits `/agents/{label}/config/agent.nix` on your behalf, commits, and submits an approval that the operator can accept on the dashboard; on approve hive-c0re rebuilds your container with the new config.
|
||||||
|
|
|
||||||
|
|
@ -12,7 +12,8 @@ Tools (hyperhive surface):
|
||||||
- `mcp__hyperhive__request_apply_commit(agent, commit_ref, description?)` — submit a config change for any agent (`hm1nd` for self) for operator approval. Pass an optional `description` and it appears on the dashboard approval card so the operator knows what changed without opening the diff. At submit time hive-c0re fetches your commit into the agent's applied repo and pins it as `proposal/<id>`; from that moment your proposed-side commit can be amended or force-pushed freely without changing what the operator will build.
|
- `mcp__hyperhive__request_apply_commit(agent, commit_ref, description?)` — submit a config change for any agent (`hm1nd` for self) for operator approval. Pass an optional `description` and it appears on the dashboard approval card so the operator knows what changed without opening the diff. At submit time hive-c0re fetches your commit into the agent's applied repo and pins it as `proposal/<id>`; from that moment your proposed-side commit can be amended or force-pushed freely without changing what the operator will build.
|
||||||
- `mcp__hyperhive__ask(question, options?, multi?, ttl_seconds?, to?)` — surface a structured question to the operator (default, or `to: "operator"`) OR a sub-agent (`to: "<agent-name>"`). Returns immediately with a question id; the answer arrives later as a system `question_answered { id, question, answer, answerer }` event in your inbox. Options are advisory: the dashboard always lets the operator type a free-text answer in addition. Set `multi: true` to render options as checkboxes (operator can pick multiple); the answer comes back as `, `-separated. Set `ttl_seconds` to auto-cancel after a deadline (capped at 6h server-side) — on expiry the answer is `[expired]` and `answerer` is `"ttl-watchdog"`. Do not poll inside the same turn — finish the current work and react when the event lands.
|
- `mcp__hyperhive__ask(question, options?, multi?, ttl_seconds?, to?)` — surface a structured question to the operator (default, or `to: "operator"`) OR a sub-agent (`to: "<agent-name>"`). Returns immediately with a question id; the answer arrives later as a system `question_answered { id, question, answer, answerer }` event in your inbox. Options are advisory: the dashboard always lets the operator type a free-text answer in addition. Set `multi: true` to render options as checkboxes (operator can pick multiple); the answer comes back as `, `-separated. Set `ttl_seconds` to auto-cancel after a deadline (capped at 6h server-side) — on expiry the answer is `[expired]` and `answerer` is `"ttl-watchdog"`. Do not poll inside the same turn — finish the current work and react when the event lands.
|
||||||
- `mcp__hyperhive__answer(id, answer)` — answer a question that was routed to YOU (a sub-agent did `ask(to: "manager", ...)`). The triggering event in your inbox is `question_asked { id, asker, question, options, multi }`. The answer surfaces in the asker's inbox as a `question_answered` event.
|
- `mcp__hyperhive__answer(id, answer)` — answer a question that was routed to YOU (a sub-agent did `ask(to: "manager", ...)`). The triggering event in your inbox is `question_asked { id, asker, question, options, multi }`. The answer surfaces in the asker's inbox as a `question_answered` event.
|
||||||
- `mcp__hyperhive__get_open_threads()` — hive-wide loose ends: every pending approval + every unanswered question across the swarm. Cheap server-side sweep, no args. Use to find stalled threads (sub-agent A asked B something three days ago and B never answered) before they rot.
|
- `mcp__hyperhive__get_open_threads()` — hive-wide loose ends: every pending approval + every unanswered question + every pending reminder across the swarm. Cheap server-side sweep, no args. Use to find stalled threads (sub-agent A asked B something three days ago and B never answered) before they rot.
|
||||||
|
- `mcp__hyperhive__cancel_thread(kind, id)` — cancel any question or reminder in the swarm (manager bypasses the owner check used on sub-agents). Use for hive-wide cleanup when a sub-agent is offline / can't withdraw its own ask / reminder.
|
||||||
- `mcp__hyperhive__whoami()` — self-introspection: canonical name (`manager`), role, current hyperhive rev. No args. Useful for boot announcements and cross-agent attribution that won't drift across config reloads.
|
- `mcp__hyperhive__whoami()` — self-introspection: canonical name (`manager`), role, current hyperhive rev. No args. Useful for boot announcements and cross-agent attribution that won't drift across config reloads.
|
||||||
|
|
||||||
Approval boundary: lifecycle ops on *existing* sub-agents (`kill`, `start`, `restart`) are at your discretion — no operator approval. *Creating* a new agent (`request_spawn`) and *changing* any agent's config (`request_apply_commit`) still go through the approval queue. The operator only signs off on changes; you run the day-to-day.
|
Approval boundary: lifecycle ops on *existing* sub-agents (`kill`, `start`, `restart`) are at your discretion — no operator approval. *Creating* a new agent (`request_spawn`) and *changing* any agent's config (`request_apply_commit`) still go through the approval queue. The operator only signs off on changes; you run the day-to-day.
|
||||||
|
|
|
||||||
|
|
@ -174,11 +174,37 @@ pub fn format_open_threads(resp: Result<SocketReply, anyhow::Error>) -> String {
|
||||||
"- question #{id} ({asker} → {to}, {age_seconds}s old): {question}"
|
"- question #{id} ({asker} → {to}, {age_seconds}s old): {question}"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
hive_sh4re::OpenThread::Reminder {
|
||||||
|
id,
|
||||||
|
owner,
|
||||||
|
message,
|
||||||
|
due_at,
|
||||||
|
age_seconds,
|
||||||
|
} => {
|
||||||
|
let _ = writeln!(
|
||||||
|
out,
|
||||||
|
"- reminder #{id} ({owner}, scheduled {age_seconds}s ago, due_at={due_at}): {message}"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
out
|
out
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Parse the user-facing `kind` string for `cancel_thread` into the
|
||||||
|
/// wire enum. Accepts a small alias set so claude doesn't have to
|
||||||
|
/// remember the exact spelling (`"q"` / `"r"` shorthand falls out
|
||||||
|
/// for free).
|
||||||
|
fn parse_cancel_kind(raw: &str) -> Result<hive_sh4re::CancelThreadKind, String> {
|
||||||
|
match raw.trim().to_ascii_lowercase().as_str() {
|
||||||
|
"question" | "q" => Ok(hive_sh4re::CancelThreadKind::Question),
|
||||||
|
"reminder" | "r" => Ok(hive_sh4re::CancelThreadKind::Reminder),
|
||||||
|
other => Err(format!(
|
||||||
|
"cancel_thread: unknown kind '{other}' (expected \"question\" or \"reminder\")"
|
||||||
|
)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Format helper for `whoami`: renders the identity block as a short
|
/// Format helper for `whoami`: renders the identity block as a short
|
||||||
/// human-readable string. Skips fields that are `None` so the output
|
/// human-readable string. Skips fields that are `None` so the output
|
||||||
/// doesn't carry dead placeholders.
|
/// doesn't carry dead placeholders.
|
||||||
|
|
@ -423,11 +449,13 @@ impl AgentServer {
|
||||||
#[tool(
|
#[tool(
|
||||||
description = "List loose ends pending against this agent: unanswered questions \
|
description = "List loose ends pending against this agent: unanswered questions \
|
||||||
where you are the asker (waiting on someone) or the target (someone's waiting on \
|
where you are the asker (waiting on someone) or the target (someone's waiting on \
|
||||||
you), plus — for the manager only — pending approvals you submitted that the \
|
you), pending reminders you scheduled, plus — for the manager only — pending \
|
||||||
operator hasn't acted on yet. Cheap server-side sweep, no args. Useful at turn \
|
approvals you submitted that the operator hasn't acted on yet. Cheap server-side \
|
||||||
start to remember what you owe / what's owed to you without scrolling inbox \
|
sweep, no args. Useful at turn start to remember what you owe / what's owed to \
|
||||||
history. Output is a short bulleted list with ids, ages in seconds, and the \
|
you without scrolling inbox history. Output is a short bulleted list with ids, \
|
||||||
relevant context. Empty result is reported clearly."
|
ages in seconds, and the relevant context. Each `question` or `reminder` row \
|
||||||
|
can be cancelled by passing its id + kind to `cancel_thread`. Empty result \
|
||||||
|
is reported clearly."
|
||||||
)]
|
)]
|
||||||
async fn get_open_threads(&self) -> String {
|
async fn get_open_threads(&self) -> String {
|
||||||
run_tool_envelope("get_open_threads", String::new(), async move {
|
run_tool_envelope("get_open_threads", String::new(), async move {
|
||||||
|
|
@ -453,6 +481,34 @@ impl AgentServer {
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tool(
|
||||||
|
description = "Cancel an open thread you own — a `question` you asked (the \
|
||||||
|
asker gets `[cancelled by <you>]` as the answer and unblocks) or a `reminder` \
|
||||||
|
you scheduled (hard-deleted before it fires). `kind` is `\"question\"` or \
|
||||||
|
`\"reminder\"`; `id` is the row id from the matching `get_open_threads` entry \
|
||||||
|
or the `question_queued` reply you got when you submitted. Auth: you can only \
|
||||||
|
cancel rows where you're the asker / owner. Returns `ok` or an error string."
|
||||||
|
)]
|
||||||
|
async fn cancel_thread(&self, Parameters(args): Parameters<CancelThreadArgs>) -> String {
|
||||||
|
let log = format!("{args:?}");
|
||||||
|
let kind_label = args.kind.clone();
|
||||||
|
let id = args.id;
|
||||||
|
run_tool_envelope("cancel_thread", log, async move {
|
||||||
|
let kind = match parse_cancel_kind(&args.kind) {
|
||||||
|
Ok(k) => k,
|
||||||
|
Err(e) => return e,
|
||||||
|
};
|
||||||
|
let (resp, retries) = self
|
||||||
|
.dispatch(hive_sh4re::AgentRequest::CancelThread { kind, id })
|
||||||
|
.await;
|
||||||
|
annotate_retries(
|
||||||
|
format_ack(resp, "cancel_thread", format!("cancelled {kind_label} {id}")),
|
||||||
|
retries,
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
|
||||||
#[tool(
|
#[tool(
|
||||||
description = "Schedule a reminder that lands in this agent's own inbox at a future \
|
description = "Schedule a reminder that lands in this agent's own inbox at a future \
|
||||||
time (sender will appear as `reminder`). Use for self-paced follow-ups: 'check task \
|
time (sender will appear as `reminder`). Use for self-paced follow-ups: 'check task \
|
||||||
|
|
@ -597,6 +653,18 @@ pub struct AnswerArgs {
|
||||||
pub answer: String,
|
pub answer: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct CancelThreadArgs {
|
||||||
|
/// Which kind of thread to cancel — `"question"` for an open
|
||||||
|
/// `ask` that's still waiting on an answer, `"reminder"` for a
|
||||||
|
/// scheduled `remind` that hasn't fired yet. Use the `kind`
|
||||||
|
/// field straight off the `get_open_threads` row.
|
||||||
|
pub kind: String,
|
||||||
|
/// Row id from the matching `get_open_threads` entry (or the
|
||||||
|
/// `question_queued` reply when you submitted it).
|
||||||
|
pub id: i64,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
pub struct RequestApplyCommitArgs {
|
pub struct RequestApplyCommitArgs {
|
||||||
/// Agent whose config repo the commit lives in (use `"hm1nd"` for the
|
/// Agent whose config repo the commit lives in (use `"hm1nd"` for the
|
||||||
|
|
@ -922,10 +990,12 @@ impl ManagerServer {
|
||||||
|
|
||||||
#[tool(
|
#[tool(
|
||||||
description = "Hive-wide loose ends: EVERY pending approval + EVERY unanswered \
|
description = "Hive-wide loose ends: EVERY pending approval + EVERY unanswered \
|
||||||
question across the swarm. Use to scan for stalled coordination — questions \
|
question + EVERY pending reminder across the swarm. Use to scan for stalled \
|
||||||
sub-agents asked each other that nobody's answering, approvals stuck waiting on \
|
coordination — questions sub-agents asked each other that nobody's answering, \
|
||||||
the operator, etc. No args. The sub-agent flavour of this tool only returns the \
|
approvals stuck waiting on the operator, reminders piling up on an offline \
|
||||||
agent's own threads; the manager flavour is unfiltered."
|
agent, etc. No args. The sub-agent flavour only returns the agent's own \
|
||||||
|
threads; the manager flavour is unfiltered. Cancel any question or reminder \
|
||||||
|
row via `cancel_thread` (manager bypasses the owner check)."
|
||||||
)]
|
)]
|
||||||
async fn get_open_threads(&self) -> String {
|
async fn get_open_threads(&self) -> String {
|
||||||
run_tool_envelope("get_open_threads", String::new(), async move {
|
run_tool_envelope("get_open_threads", String::new(), async move {
|
||||||
|
|
@ -951,6 +1021,34 @@ impl ManagerServer {
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tool(
|
||||||
|
description = "Cancel any open thread in the swarm — a `question` (cancels \
|
||||||
|
with the operator-override sentinel so the asker unblocks) or a `reminder` \
|
||||||
|
(hard-deleted before fire). `kind` is `\"question\"` or `\"reminder\"`; `id` \
|
||||||
|
is the row id from `get_open_threads` or the original submission reply. \
|
||||||
|
Manager surface bypasses the owner check on the sub-agent flavour — use for \
|
||||||
|
hive-wide cleanup of stuck or stale threads."
|
||||||
|
)]
|
||||||
|
async fn cancel_thread(&self, Parameters(args): Parameters<CancelThreadArgs>) -> String {
|
||||||
|
let log = format!("{args:?}");
|
||||||
|
let kind_label = args.kind.clone();
|
||||||
|
let id = args.id;
|
||||||
|
run_tool_envelope("cancel_thread", log, async move {
|
||||||
|
let kind = match parse_cancel_kind(&args.kind) {
|
||||||
|
Ok(k) => k,
|
||||||
|
Err(e) => return e,
|
||||||
|
};
|
||||||
|
let (resp, retries) = self
|
||||||
|
.dispatch(hive_sh4re::ManagerRequest::CancelThread { kind, id })
|
||||||
|
.await;
|
||||||
|
annotate_retries(
|
||||||
|
format_ack(resp, "cancel_thread", format!("cancelled {kind_label} {id}")),
|
||||||
|
retries,
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
|
||||||
#[tool(
|
#[tool(
|
||||||
description = "Fetch recent journal log lines for a sub-agent container. Useful \
|
description = "Fetch recent journal log lines for a sub-agent container. Useful \
|
||||||
for diagnosing MCP server registration failures, startup crashes, plugin install \
|
for diagnosing MCP server registration failures, startup crashes, plugin install \
|
||||||
|
|
@ -995,8 +1093,9 @@ impl ManagerServer {
|
||||||
sub-agent — non-blocking, answer arrives later as a `question_answered` event), \
|
sub-agent — non-blocking, answer arrives later as a `question_answered` event), \
|
||||||
`answer` (respond to a `question_asked` event directed at you), \
|
`answer` (respond to a `question_asked` event directed at you), \
|
||||||
`get_open_threads` (hive-wide loose ends — pending approvals + unanswered \
|
`get_open_threads` (hive-wide loose ends — pending approvals + unanswered \
|
||||||
questions across the swarm), `whoami` (self-introspection — canonical name, \
|
questions + pending reminders across the swarm), `cancel_thread` (cancel any \
|
||||||
role, current hyperhive rev). The manager's own config lives at \
|
question or reminder row by id), `whoami` (self-introspection — canonical \
|
||||||
|
name, role, current hyperhive rev). The manager's own config lives at \
|
||||||
`/agents/hm1nd/config/agent.nix`."
|
`/agents/hm1nd/config/agent.nix`."
|
||||||
)]
|
)]
|
||||||
impl ServerHandler for ManagerServer {}
|
impl ServerHandler for ManagerServer {}
|
||||||
|
|
@ -1039,6 +1138,7 @@ pub fn allowed_mcp_tools(flavor: Flavor) -> Vec<String> {
|
||||||
"remind",
|
"remind",
|
||||||
"get_open_threads",
|
"get_open_threads",
|
||||||
"whoami",
|
"whoami",
|
||||||
|
"cancel_thread",
|
||||||
],
|
],
|
||||||
Flavor::Manager => &[
|
Flavor::Manager => &[
|
||||||
"send",
|
"send",
|
||||||
|
|
@ -1055,6 +1155,7 @@ pub fn allowed_mcp_tools(flavor: Flavor) -> Vec<String> {
|
||||||
"get_open_threads",
|
"get_open_threads",
|
||||||
"remind",
|
"remind",
|
||||||
"whoami",
|
"whoami",
|
||||||
|
"cancel_thread",
|
||||||
],
|
],
|
||||||
};
|
};
|
||||||
let mut out: Vec<String> = names
|
let mut out: Vec<String> = names
|
||||||
|
|
|
||||||
|
|
@ -193,6 +193,13 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
|
||||||
role: "agent".to_owned(),
|
role: "agent".to_owned(),
|
||||||
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
|
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
|
||||||
},
|
},
|
||||||
|
AgentRequest::CancelThread { kind, id } => crate::questions::handle_cancel_thread(
|
||||||
|
coord, agent, *kind, *id,
|
||||||
|
)
|
||||||
|
.map_or_else(
|
||||||
|
|message| AgentResponse::Err { message },
|
||||||
|
|()| AgentResponse::Ok,
|
||||||
|
),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -405,6 +405,40 @@ impl Broker {
|
||||||
Ok(n)
|
Ok(n)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Cancel a pending reminder on behalf of `canceller`. Returns
|
||||||
|
/// the owner agent name on success (handy for logging). Auth
|
||||||
|
/// rules mirror `OperatorQuestions::cancel`: owner, operator, or
|
||||||
|
/// manager.
|
||||||
|
pub fn cancel_reminder_as(&self, id: i64, canceller: &str) -> Result<String> {
|
||||||
|
let conn = self.conn.lock().unwrap();
|
||||||
|
let owner: Option<String> = conn
|
||||||
|
.query_row(
|
||||||
|
"SELECT agent FROM reminders WHERE id = ?1 AND sent_at IS NULL",
|
||||||
|
params![id],
|
||||||
|
|row| row.get(0),
|
||||||
|
)
|
||||||
|
.optional()?;
|
||||||
|
let Some(owner) = owner else {
|
||||||
|
anyhow::bail!("reminder {id} not pending (already delivered or unknown)");
|
||||||
|
};
|
||||||
|
let authorised = canceller == owner
|
||||||
|
|| canceller == hive_sh4re::OPERATOR_RECIPIENT
|
||||||
|
|| canceller == hive_sh4re::MANAGER_AGENT;
|
||||||
|
if !authorised {
|
||||||
|
anyhow::bail!(
|
||||||
|
"reminder {id}: '{canceller}' not allowed to cancel (owner = '{owner}')"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
let n = conn.execute(
|
||||||
|
"DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL",
|
||||||
|
params![id],
|
||||||
|
)?;
|
||||||
|
if n == 0 {
|
||||||
|
anyhow::bail!("reminder {id} vanished between auth check and delete");
|
||||||
|
}
|
||||||
|
Ok(owner)
|
||||||
|
}
|
||||||
|
|
||||||
/// Get up to `limit` due reminders across all agents in a single query.
|
/// Get up to `limit` due reminders across all agents in a single query.
|
||||||
/// Returns `(agent, id, message, file_path)` tuples. Pass a small limit
|
/// Returns `(agent, id, message, file_path)` tuples. Pass a small limit
|
||||||
/// (e.g. 100) so a burst of overdue reminders doesn't flood the broker
|
/// (e.g. 100) so a burst of overdue reminders doesn't flood the broker
|
||||||
|
|
|
||||||
|
|
@ -348,6 +348,16 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
|
||||||
role: "manager".to_owned(),
|
role: "manager".to_owned(),
|
||||||
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
|
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
|
||||||
},
|
},
|
||||||
|
ManagerRequest::CancelThread { kind, id } => crate::questions::handle_cancel_thread(
|
||||||
|
coord,
|
||||||
|
MANAGER_AGENT,
|
||||||
|
*kind,
|
||||||
|
*id,
|
||||||
|
)
|
||||||
|
.map_or_else(
|
||||||
|
|message| ManagerResponse::Err { message },
|
||||||
|
|()| ManagerResponse::Ok,
|
||||||
|
),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -26,9 +26,12 @@ use crate::coordinator::Coordinator;
|
||||||
/// we keep the rule per-agent so the manager's MCP surface gets
|
/// we keep the rule per-agent so the manager's MCP surface gets
|
||||||
/// the same shape via a different code path);
|
/// the same shape via a different code path);
|
||||||
/// - unanswered questions where `agent` is the asker (waiting on
|
/// - unanswered questions where `agent` is the asker (waiting on
|
||||||
/// someone) OR the target (owes a reply).
|
/// someone) OR the target (owes a reply);
|
||||||
|
/// - pending reminders this agent scheduled (`owner == self`).
|
||||||
///
|
///
|
||||||
/// Newest-first within each kind, approvals before questions.
|
/// Ordered approvals → questions → reminders within the returned
|
||||||
|
/// vector. Within each kind, source-of-truth ordering (sqlite's
|
||||||
|
/// `pending()` queries return newest-first within their indexes).
|
||||||
pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<OpenThread>> {
|
pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<OpenThread>> {
|
||||||
let now = now_unix();
|
let now = now_unix();
|
||||||
let mut out = Vec::new();
|
let mut out = Vec::new();
|
||||||
|
|
@ -60,13 +63,25 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<OpenThread>> {
|
||||||
age_seconds: saturating_age(now, q.asked_at),
|
age_seconds: saturating_age(now, q.asked_at),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
for r in coord.broker.list_pending_reminders()? {
|
||||||
|
if r.agent != agent {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
out.push(OpenThread::Reminder {
|
||||||
|
id: r.id,
|
||||||
|
owner: r.agent,
|
||||||
|
message: r.message,
|
||||||
|
due_at: r.due_at,
|
||||||
|
age_seconds: saturating_age(now, r.created_at),
|
||||||
|
});
|
||||||
|
}
|
||||||
Ok(out)
|
Ok(out)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Hive-wide loose-ends view: EVERY pending approval + EVERY
|
/// Hive-wide loose-ends view: EVERY pending approval + EVERY
|
||||||
/// unanswered question. Manager surface only; sub-agents can't see
|
/// unanswered question + EVERY pending reminder. Manager surface
|
||||||
/// each other's threads via the agent surface (`for_agent` filters
|
/// only; sub-agents can't see each other's threads via the agent
|
||||||
/// by name).
|
/// surface (`for_agent` filters by name).
|
||||||
pub fn hive_wide(coord: &Coordinator) -> Result<Vec<OpenThread>> {
|
pub fn hive_wide(coord: &Coordinator) -> Result<Vec<OpenThread>> {
|
||||||
let now = now_unix();
|
let now = now_unix();
|
||||||
let mut out = Vec::new();
|
let mut out = Vec::new();
|
||||||
|
|
@ -88,6 +103,15 @@ pub fn hive_wide(coord: &Coordinator) -> Result<Vec<OpenThread>> {
|
||||||
age_seconds: saturating_age(now, q.asked_at),
|
age_seconds: saturating_age(now, q.asked_at),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
for r in coord.broker.list_pending_reminders()? {
|
||||||
|
out.push(OpenThread::Reminder {
|
||||||
|
id: r.id,
|
||||||
|
owner: r.agent,
|
||||||
|
message: r.message,
|
||||||
|
due_at: r.due_at,
|
||||||
|
age_seconds: saturating_age(now, r.created_at),
|
||||||
|
});
|
||||||
|
}
|
||||||
Ok(out)
|
Ok(out)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -196,6 +196,54 @@ impl OperatorQuestions {
|
||||||
Ok((question, asker, target))
|
Ok((question, asker, target))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Cancel a pending question on behalf of `canceller`. Returns
|
||||||
|
/// `(question, asker, target)` so the caller can fire the usual
|
||||||
|
/// `QuestionAnswered` event to the asker with a `[cancelled by
|
||||||
|
/// <canceller>]` sentinel.
|
||||||
|
///
|
||||||
|
/// Auth: the canceller must be one of:
|
||||||
|
/// - the original asker (an agent withdrawing their own ask),
|
||||||
|
/// - the operator (already covered by the existing `answer` path
|
||||||
|
/// but allowed here too for symmetry / dashboard cancel),
|
||||||
|
/// - the manager (privileged hive-wide cleanup).
|
||||||
|
///
|
||||||
|
/// Not the target — that's covered by `answer` (responding with
|
||||||
|
/// an actual reply, sentinel or otherwise).
|
||||||
|
pub fn cancel(
|
||||||
|
&self,
|
||||||
|
id: i64,
|
||||||
|
canceller: &str,
|
||||||
|
) -> Result<(String, String, Option<String>)> {
|
||||||
|
let conn = self.conn.lock().unwrap();
|
||||||
|
let row: Option<(String, String, Option<String>, Option<i64>)> = conn
|
||||||
|
.query_row(
|
||||||
|
"SELECT question, asker, target, answered_at FROM operator_questions WHERE id = ?1",
|
||||||
|
params![id],
|
||||||
|
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
|
||||||
|
)
|
||||||
|
.optional()?;
|
||||||
|
let Some((question, asker, target, answered_at)) = row else {
|
||||||
|
bail!("question {id} not found");
|
||||||
|
};
|
||||||
|
if answered_at.is_some() {
|
||||||
|
bail!("question {id} already answered/cancelled");
|
||||||
|
}
|
||||||
|
let authorised = canceller == asker
|
||||||
|
|| canceller == hive_sh4re::OPERATOR_RECIPIENT
|
||||||
|
|| canceller == hive_sh4re::MANAGER_AGENT;
|
||||||
|
if !authorised {
|
||||||
|
bail!(
|
||||||
|
"question {id}: '{canceller}' not allowed to cancel (asker = '{asker}')"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
let sentinel = format!("[cancelled by {canceller}]");
|
||||||
|
conn.execute(
|
||||||
|
"UPDATE operator_questions SET answer = ?1, answered_at = ?2 WHERE id = ?3",
|
||||||
|
params![sentinel, now_unix(), id],
|
||||||
|
)?;
|
||||||
|
Ok((question, asker, target))
|
||||||
|
}
|
||||||
|
|
||||||
#[allow(dead_code)]
|
#[allow(dead_code)]
|
||||||
pub fn get(&self, id: i64) -> Result<Option<OpQuestion>> {
|
pub fn get(&self, id: i64) -> Result<Option<OpQuestion>> {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
|
|
|
||||||
|
|
@ -129,6 +129,49 @@ pub fn handle_answer(
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Handle `CancelThread` from either surface. Dispatches by kind to
|
||||||
|
/// either `OperatorQuestions::cancel` or `Broker::cancel_reminder_as`,
|
||||||
|
/// both of which do their own auth check (canceller == owner /
|
||||||
|
/// asker, or `operator`, or `manager`). On question cancel, fires
|
||||||
|
/// the `QuestionAnswered` event back to the asker so the harness
|
||||||
|
/// loop can react (mirrors the operator-cancel dashboard path).
|
||||||
|
pub fn handle_cancel_thread(
|
||||||
|
coord: &Arc<Coordinator>,
|
||||||
|
canceller: &str,
|
||||||
|
kind: hive_sh4re::CancelThreadKind,
|
||||||
|
id: i64,
|
||||||
|
) -> Result<(), String> {
|
||||||
|
match kind {
|
||||||
|
hive_sh4re::CancelThreadKind::Question => {
|
||||||
|
let (question, asker, target) = coord
|
||||||
|
.questions
|
||||||
|
.cancel(id, canceller)
|
||||||
|
.map_err(|e| format!("{e:#}"))?;
|
||||||
|
let sentinel = format!("[cancelled by {canceller}]");
|
||||||
|
tracing::info!(%id, %canceller, %asker, "question cancelled");
|
||||||
|
coord.notify_agent(
|
||||||
|
&asker,
|
||||||
|
&hive_sh4re::HelperEvent::QuestionAnswered {
|
||||||
|
id,
|
||||||
|
question,
|
||||||
|
answer: sentinel.clone(),
|
||||||
|
answerer: canceller.to_owned(),
|
||||||
|
},
|
||||||
|
);
|
||||||
|
coord.emit_question_resolved(id, &sentinel, canceller, true, target.as_deref());
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
hive_sh4re::CancelThreadKind::Reminder => {
|
||||||
|
let owner = coord
|
||||||
|
.broker
|
||||||
|
.cancel_reminder_as(id, canceller)
|
||||||
|
.map_err(|e| format!("{e:#}"))?;
|
||||||
|
tracing::info!(%id, %canceller, %owner, "reminder cancelled");
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Real coverage needs a `Coordinator` fixture (broker + sqlite +
|
// Real coverage needs a `Coordinator` fixture (broker + sqlite +
|
||||||
// in-memory questions). Skipped for now — the normalisation branches
|
// in-memory questions). Skipped for now — the normalisation branches
|
||||||
// in `handle_ask` are short enough to read line-by-line; once we add
|
// in `handle_ask` are short enough to read line-by-line; once we add
|
||||||
|
|
|
||||||
|
|
@ -191,6 +191,11 @@ pub enum ReminderTiming {
|
||||||
/// caller (claude in the agent harness) is expected to render these
|
/// caller (claude in the agent harness) is expected to render these
|
||||||
/// as a short bulleted list — the per-row fields are all the context
|
/// as a short bulleted list — the per-row fields are all the context
|
||||||
/// needed without a follow-up fetch.
|
/// needed without a follow-up fetch.
|
||||||
|
///
|
||||||
|
/// `Question` and `Reminder` rows are cancellable via the
|
||||||
|
/// `CancelThread` request (and the `cancel_thread` MCP tool);
|
||||||
|
/// `Approval` rows are not (operator approves/denies via the
|
||||||
|
/// dashboard, manager has no withdraw path today).
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||||
pub enum OpenThread {
|
pub enum OpenThread {
|
||||||
|
|
@ -223,6 +228,33 @@ pub enum OpenThread {
|
||||||
/// Wall-clock seconds since `asked_at`. Saturates at zero.
|
/// Wall-clock seconds since `asked_at`. Saturates at zero.
|
||||||
age_seconds: u64,
|
age_seconds: u64,
|
||||||
},
|
},
|
||||||
|
/// A scheduled but un-delivered reminder. For agent-flavour calls:
|
||||||
|
/// only the agent's own reminders (where `owner == self`). For
|
||||||
|
/// manager-flavour calls: every pending reminder in the swarm.
|
||||||
|
/// `owner` is the agent who scheduled it; `due_at` is the absolute
|
||||||
|
/// unix timestamp the scheduler is targeting.
|
||||||
|
Reminder {
|
||||||
|
id: i64,
|
||||||
|
owner: String,
|
||||||
|
message: String,
|
||||||
|
due_at: i64,
|
||||||
|
/// Wall-clock seconds since the reminder was scheduled. Saturates
|
||||||
|
/// at zero on clock anomalies. (For time-until-fire, compute
|
||||||
|
/// `due_at - now` client-side from the wire timestamp.)
|
||||||
|
age_seconds: u64,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Kind discriminator for `CancelThread`. Maps to which underlying
|
||||||
|
/// store the dispatcher reaches into (`OperatorQuestions` vs
|
||||||
|
/// `Broker::reminders`). Approvals are deliberately not cancellable
|
||||||
|
/// — the operator approves/denies via the dashboard, manager has no
|
||||||
|
/// withdraw path today.
|
||||||
|
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
|
||||||
|
#[serde(rename_all = "snake_case")]
|
||||||
|
pub enum CancelThreadKind {
|
||||||
|
Question,
|
||||||
|
Reminder,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Requests on a per-agent socket. The agent's identity is the socket
|
/// Requests on a per-agent socket. The agent's identity is the socket
|
||||||
|
|
@ -319,6 +351,13 @@ pub enum AgentRequest {
|
||||||
/// identity after a rename or session-continue boundary where the
|
/// identity after a rename or session-continue boundary where the
|
||||||
/// system-prompt-substituted label is no longer reliable.
|
/// system-prompt-substituted label is no longer reliable.
|
||||||
Whoami,
|
Whoami,
|
||||||
|
/// Cancel an open thread the agent owns: a `Question` they asked
|
||||||
|
/// (returns `[cancelled by <self>]` as the answer to the asker)
|
||||||
|
/// or a `Reminder` they scheduled (hard-deletes the row).
|
||||||
|
/// Authorisation on the sub-agent surface: caller must own the
|
||||||
|
/// row. The manager surface uses the same wire variant but
|
||||||
|
/// accepts any id.
|
||||||
|
CancelThread { kind: CancelThreadKind, id: i64 },
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Responses on a per-agent socket.
|
/// Responses on a per-agent socket.
|
||||||
|
|
@ -625,6 +664,10 @@ pub enum ManagerRequest {
|
||||||
/// Manager-flavour self-introspection. Same wire shape as
|
/// Manager-flavour self-introspection. Same wire shape as
|
||||||
/// `AgentRequest::Whoami`, but `role` is always `"manager"`.
|
/// `AgentRequest::Whoami`, but `role` is always `"manager"`.
|
||||||
Whoami,
|
Whoami,
|
||||||
|
/// Cancel an open thread (question or reminder). Manager surface
|
||||||
|
/// can cancel any row (no owner check) — same dispatch as
|
||||||
|
/// `AgentRequest::CancelThread` but with privileged auth.
|
||||||
|
CancelThread { kind: CancelThreadKind, id: i64 },
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue