Compare commits

...
18 changed files with 437 additions and 118 deletions

View file

@ -53,11 +53,12 @@ hive-c0re/ host daemon + CLI (one binary, subcommand-dispatched)
transient/sockets) + tombstone enumeration + transient/sockets) + tombstone enumeration +
kick_agent + notify_agent (helper-event push) + kick_agent + notify_agent (helper-event push) +
last_containers cache + rescan_and_emit diff helper last_containers cache + rescan_and_emit diff helper
src/open_threads.rs loose-ends aggregator (pending approvals + src/loose_ends.rs loose-ends aggregator (pending approvals +
unanswered questions) — for_agent (filtered) and unanswered questions + pending reminders) —
hive_wide (manager surface). Backs for_agent (filtered) and hive_wide (manager
AgentRequest::GetOpenThreads + ManagerRequest:: surface). Backs AgentRequest::GetLooseEnds +
GetOpenThreads (the get_open_threads MCP tool). ManagerRequest::GetLooseEnds (the
get_loose_ends MCP tool).
src/actions.rs approve/deny/destroy (transient-aware) src/actions.rs approve/deny/destroy (transient-aware)
src/auto_update.rs startup rebuild scan + ensure_manager + src/auto_update.rs startup rebuild scan + ensure_manager +
meta::lock_update_hyperhive bump meta::lock_update_hyperhive bump

View file

@ -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_loose_ends` (renamed from `get_open_threads` per the naming pass). `LooseEnd` enum (renamed from `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_loose_end(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_loose_end`. Per-agent web UI's `/api/open-threads``/api/loose-ends` too, with reminder-row rendering added.
`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

View file

@ -393,43 +393,43 @@
} }
renderStateBadge(); renderStateBadge();
} }
// Open-threads section: same data the get_open_threads MCP tool // Loose-ends section: same data the get_loose_ends MCP tool
// returns. Best-effort fetch on cold load + after every turn_end // returns. Best-effort fetch on cold load + after every turn_end
// (a turn likely answered or asked something). Silent failure // (a turn likely answered or asked something). Silent failure
// keeps the section hidden rather than surfacing an empty banner. // keeps the section hidden rather than surfacing an empty banner.
let lastOpenThreadsCount = 0; let lastLooseEndsCount = 0;
async function refreshOpenThreads() { async function refreshLooseEnds() {
try { try {
const resp = await fetch('/api/open-threads'); const resp = await fetch('/api/loose-ends');
if (!resp.ok) { if (!resp.ok) {
renderOpenThreads([]); renderLooseEnds([]);
return; return;
} }
const data = await resp.json(); const data = await resp.json();
renderOpenThreads(data.threads || []); renderLooseEnds(data.loose_ends || []);
} catch (err) { } catch (err) {
console.warn('open-threads fetch failed', err); console.warn('loose-ends fetch failed', err);
renderOpenThreads([]); renderLooseEnds([]);
} }
} }
function renderOpenThreads(threads) { function renderLooseEnds(threads) {
const root = $('open-threads-section'); const root = $('loose-ends-section');
const list = $('open-threads-list'); const list = $('loose-ends-list');
const summary = $('open-threads-summary'); const summary = $('loose-ends-summary');
if (!root || !list || !summary) return; if (!root || !list || !summary) return;
if (!threads.length) { if (!threads.length) {
root.hidden = true; root.hidden = true;
lastOpenThreadsCount = 0; lastLooseEndsCount = 0;
return; return;
} }
root.hidden = false; root.hidden = false;
summary.textContent = 'open threads · ' + threads.length; summary.textContent = 'loose ends · ' + threads.length;
list.innerHTML = ''; list.innerHTML = '';
// Auto-expand on first appearance of any open thread so the // Auto-expand on first appearance of any open thread so the
// operator notices new loose ends; collapse only on operator // operator notices new loose ends; collapse only on operator
// click (sticky after that). // click (sticky after that).
if (lastOpenThreadsCount === 0) root.open = true; if (lastLooseEndsCount === 0) root.open = true;
lastOpenThreadsCount = threads.length; lastLooseEndsCount = threads.length;
const fmtAge = (s) => { const fmtAge = (s) => {
if (s < 60) return s + 's'; if (s < 60) return s + 's';
if (s < 3600) return Math.floor(s / 60) + 'm'; if (s < 3600) return Math.floor(s / 60) + 'm';
@ -455,6 +455,18 @@
el('span', { class: 'inbox-ts' }, fmtAge(t.age_seconds || 0) + ' ago'), el('span', { class: 'inbox-ts' }, fmtAge(t.age_seconds || 0) + ' ago'),
el('div', { class: 'inbox-body' }, t.question || ''), el('div', { class: 'inbox-body' }, t.question || ''),
); );
} else if (t.kind === 'reminder') {
// due_at is an absolute unix-seconds value; show time-until-fire
// (negative when overdue, fmtAge handles 0/positive case here).
const now = Math.floor(Date.now() / 1000);
const dueIn = (t.due_at || 0) - now;
const dueLabel = dueIn >= 0 ? 'in ' + fmtAge(dueIn) : fmtAge(-dueIn) + ' overdue';
li.append(
el('span', { class: 'inbox-from' }, '⏰ reminder #' + t.id), ' ',
el('span', { class: 'inbox-sep' }, t.owner + ' · due ' + dueLabel), ' ',
el('span', { class: 'inbox-ts' }, 'scheduled ' + fmtAge(t.age_seconds || 0) + ' ago'),
el('div', { class: 'inbox-body' }, t.message || ''),
);
} else { } else {
li.append(el('span', { class: 'inbox-body' }, JSON.stringify(t))); li.append(el('span', { class: 'inbox-body' }, JSON.stringify(t)));
} }
@ -614,7 +626,7 @@
// Open-threads aren't part of /api/state (kept on the broker // Open-threads aren't part of /api/state (kept on the broker
// db, fetched via the per-agent socket). Cold-load fetches // db, fetched via the per-agent socket). Cold-load fetches
// it here; turn_end refreshes it via the renderer below. // it here; turn_end refreshes it via the renderer below.
refreshOpenThreads(); refreshLooseEnds();
// Skip the re-render if nothing structurally changed. The most // Skip the re-render if nothing structurally changed. The most
// common case is `online` polling itself — without this guard, the // common case is `online` polling itself — without this guard, the
// operator's <input value> gets clobbered every cycle. // operator's <input value> gets clobbered every cycle.
@ -956,7 +968,7 @@
} else { } else {
setBannerActive(false); setState('idle'); setBannerActive(false); setState('idle');
// Likely answered/asked/scheduled something — refresh. // Likely answered/asked/scheduled something — refresh.
refreshOpenThreads(); refreshLooseEnds();
} }
const cls = ev.ok ? 'turn-end-ok' : 'turn-end-fail'; const cls = ev.ok ? 'turn-end-ok' : 'turn-end-fail';
api.row(cls, api.row(cls,

View file

@ -29,9 +29,9 @@
<ul id="inbox-list"></ul> <ul id="inbox-list"></ul>
</details> </details>
<details id="open-threads-section" class="agent-inbox" hidden> <details id="loose-ends-section" class="agent-inbox" hidden>
<summary><span id="open-threads-summary">open threads</span></summary> <summary><span id="loose-ends-summary">loose ends</span></summary>
<ul id="open-threads-list"></ul> <ul id="loose-ends-list"></ul>
</details> </details>
<div class="terminal-wrap"> <div class="terminal-wrap">

View file

@ -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_loose_ends()` — 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_loose_end(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_loose_ends` 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.

View file

@ -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_loose_ends()` — 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_loose_end(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.

View file

@ -240,7 +240,7 @@ async fn serve(
| AgentResponse::Status { .. } | AgentResponse::Status { .. }
| AgentResponse::Recent { .. } | AgentResponse::Recent { .. }
| AgentResponse::QuestionQueued { .. } | AgentResponse::QuestionQueued { .. }
| AgentResponse::OpenThreads { .. } | AgentResponse::LooseEnds { .. }
| AgentResponse::PendingRemindersCount { .. } | AgentResponse::PendingRemindersCount { .. }
| AgentResponse::Whoami { .. }, | AgentResponse::Whoami { .. },
) => { ) => {
@ -320,11 +320,11 @@ fn now_unix() -> i64 {
async fn fetch_agent_post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) { async fn fetch_agent_post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) {
let threads = match client::request::<_, AgentResponse>( let threads = match client::request::<_, AgentResponse>(
socket, socket,
&AgentRequest::GetOpenThreads, &AgentRequest::GetLooseEnds,
) )
.await .await
{ {
Ok(AgentResponse::OpenThreads { threads }) => u64::try_from(threads.len()).ok(), Ok(AgentResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(),
_ => None, _ => None,
}; };
let reminders = match client::request::<_, AgentResponse>( let reminders = match client::request::<_, AgentResponse>(

View file

@ -208,7 +208,7 @@ async fn serve(
| ManagerResponse::QuestionQueued { .. } | ManagerResponse::QuestionQueued { .. }
| ManagerResponse::Recent { .. } | ManagerResponse::Recent { .. }
| ManagerResponse::Logs { .. } | ManagerResponse::Logs { .. }
| ManagerResponse::OpenThreads { .. } | ManagerResponse::LooseEnds { .. }
| ManagerResponse::PendingRemindersCount { .. } | ManagerResponse::PendingRemindersCount { .. }
| ManagerResponse::Whoami { .. }, | ManagerResponse::Whoami { .. },
) => { ) => {
@ -259,11 +259,11 @@ fn now_unix() -> i64 {
async fn fetch_manager_post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) { async fn fetch_manager_post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) {
let threads = match client::request::<_, ManagerResponse>( let threads = match client::request::<_, ManagerResponse>(
socket, socket,
&ManagerRequest::GetOpenThreads, &ManagerRequest::GetLooseEnds,
) )
.await .await
{ {
Ok(ManagerResponse::OpenThreads { threads }) => u64::try_from(threads.len()).ok(), Ok(ManagerResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(),
_ => None, _ => None,
}; };
let reminders = match client::request::<_, ManagerResponse>( let reminders = match client::request::<_, ManagerResponse>(

View file

@ -40,7 +40,7 @@ pub enum SocketReply {
QuestionQueued(i64), QuestionQueued(i64),
Recent(Vec<hive_sh4re::InboxRow>), Recent(Vec<hive_sh4re::InboxRow>),
Logs(String), Logs(String),
OpenThreads(Vec<hive_sh4re::OpenThread>), LooseEnds(Vec<hive_sh4re::LooseEnd>),
PendingRemindersCount(u64), PendingRemindersCount(u64),
Whoami { Whoami {
name: String, name: String,
@ -59,7 +59,7 @@ impl From<hive_sh4re::AgentResponse> for SocketReply {
hive_sh4re::AgentResponse::Status { unread } => Self::Status(unread), hive_sh4re::AgentResponse::Status { unread } => Self::Status(unread),
hive_sh4re::AgentResponse::Recent { rows } => Self::Recent(rows), hive_sh4re::AgentResponse::Recent { rows } => Self::Recent(rows),
hive_sh4re::AgentResponse::QuestionQueued { id } => Self::QuestionQueued(id), hive_sh4re::AgentResponse::QuestionQueued { id } => Self::QuestionQueued(id),
hive_sh4re::AgentResponse::OpenThreads { threads } => Self::OpenThreads(threads), hive_sh4re::AgentResponse::LooseEnds { loose_ends } => Self::LooseEnds(loose_ends),
hive_sh4re::AgentResponse::PendingRemindersCount { count } => { hive_sh4re::AgentResponse::PendingRemindersCount { count } => {
Self::PendingRemindersCount(count) Self::PendingRemindersCount(count)
} }
@ -87,7 +87,7 @@ impl From<hive_sh4re::ManagerResponse> for SocketReply {
hive_sh4re::ManagerResponse::QuestionQueued { id } => Self::QuestionQueued(id), hive_sh4re::ManagerResponse::QuestionQueued { id } => Self::QuestionQueued(id),
hive_sh4re::ManagerResponse::Recent { rows } => Self::Recent(rows), hive_sh4re::ManagerResponse::Recent { rows } => Self::Recent(rows),
hive_sh4re::ManagerResponse::Logs { content } => Self::Logs(content), hive_sh4re::ManagerResponse::Logs { content } => Self::Logs(content),
hive_sh4re::ManagerResponse::OpenThreads { threads } => Self::OpenThreads(threads), hive_sh4re::ManagerResponse::LooseEnds { loose_ends } => Self::LooseEnds(loose_ends),
hive_sh4re::ManagerResponse::PendingRemindersCount { count } => { hive_sh4re::ManagerResponse::PendingRemindersCount { count } => {
Self::PendingRemindersCount(count) Self::PendingRemindersCount(count)
} }
@ -128,16 +128,16 @@ pub fn format_recv(resp: Result<SocketReply, anyhow::Error>) -> String {
} }
} }
/// Format helper for `get_open_threads`: renders a short bulleted list /// Format helper for `get_loose_ends`: renders a short bulleted list
/// of pending approvals + questions. Empty list collapses to a clear /// of pending approvals + questions. Empty list collapses to a clear
/// marker so claude doesn't go hunting for a payload that isn't there. /// marker so claude doesn't go hunting for a payload that isn't there.
pub fn format_open_threads(resp: Result<SocketReply, anyhow::Error>) -> String { pub fn format_loose_ends(resp: Result<SocketReply, anyhow::Error>) -> String {
use std::fmt::Write as _; use std::fmt::Write as _;
let threads = match resp { let threads = match resp {
Ok(SocketReply::OpenThreads(t)) => t, Ok(SocketReply::LooseEnds(t)) => t,
Ok(SocketReply::Err(m)) => return format!("get_open_threads failed: {m}"), Ok(SocketReply::Err(m)) => return format!("get_loose_ends failed: {m}"),
Ok(other) => return format!("get_open_threads unexpected response: {other:?}"), Ok(other) => return format!("get_loose_ends unexpected response: {other:?}"),
Err(e) => return format!("get_open_threads transport error: {e:#}"), Err(e) => return format!("get_loose_ends transport error: {e:#}"),
}; };
if threads.is_empty() { if threads.is_empty() {
return "(no open threads)".to_owned(); return "(no open threads)".to_owned();
@ -145,7 +145,7 @@ pub fn format_open_threads(resp: Result<SocketReply, anyhow::Error>) -> String {
let mut out = format!("{} open thread(s):\n", threads.len()); let mut out = format!("{} open thread(s):\n", threads.len());
for t in &threads { for t in &threads {
match t { match t {
hive_sh4re::OpenThread::Approval { hive_sh4re::LooseEnd::Approval {
id, id,
agent, agent,
commit_ref, commit_ref,
@ -161,7 +161,7 @@ pub fn format_open_threads(resp: Result<SocketReply, anyhow::Error>) -> String {
"- approval #{id} ({agent} @ {commit_ref}, {age_seconds}s old){desc}" "- approval #{id} ({agent} @ {commit_ref}, {age_seconds}s old){desc}"
); );
} }
hive_sh4re::OpenThread::Question { hive_sh4re::LooseEnd::Question {
id, id,
asker, asker,
target, target,
@ -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::LooseEnd::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_loose_end` 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_loose_end_kind(raw: &str) -> Result<hive_sh4re::CancelLooseEndKind, String> {
match raw.trim().to_ascii_lowercase().as_str() {
"question" | "q" => Ok(hive_sh4re::CancelLooseEndKind::Question),
"reminder" | "r" => Ok(hive_sh4re::CancelLooseEndKind::Reminder),
other => Err(format!(
"cancel_loose_end: 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,16 +449,18 @@ 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_loose_end`. Empty result \
is reported clearly."
)] )]
async fn get_open_threads(&self) -> String { async fn get_loose_ends(&self) -> String {
run_tool_envelope("get_open_threads", String::new(), async move { run_tool_envelope("get_loose_ends", String::new(), async move {
let (resp, retries) = self.dispatch(hive_sh4re::AgentRequest::GetOpenThreads).await; let (resp, retries) = self.dispatch(hive_sh4re::AgentRequest::GetLooseEnds).await;
annotate_retries(format_open_threads(resp), retries) annotate_retries(format_loose_ends(resp), retries)
}) })
.await .await
} }
@ -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_loose_ends` 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_loose_end(&self, Parameters(args): Parameters<CancelLooseEndArgs>) -> String {
let log = format!("{args:?}");
let kind_label = args.kind.clone();
let id = args.id;
run_tool_envelope("cancel_loose_end", log, async move {
let kind = match parse_loose_end_kind(&args.kind) {
Ok(k) => k,
Err(e) => return e,
};
let (resp, retries) = self
.dispatch(hive_sh4re::AgentRequest::CancelLooseEnd { kind, id })
.await;
annotate_retries(
format_ack(resp, "cancel_loose_end", 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 CancelLooseEndArgs {
/// 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_loose_ends` row.
pub kind: String,
/// Row id from the matching `get_loose_ends` 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,17 +990,19 @@ 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_loose_end` (manager bypasses the owner check)."
)] )]
async fn get_open_threads(&self) -> String { async fn get_loose_ends(&self) -> String {
run_tool_envelope("get_open_threads", String::new(), async move { run_tool_envelope("get_loose_ends", String::new(), async move {
let (resp, retries) = self let (resp, retries) = self
.dispatch(hive_sh4re::ManagerRequest::GetOpenThreads) .dispatch(hive_sh4re::ManagerRequest::GetLooseEnds)
.await; .await;
annotate_retries(format_open_threads(resp), retries) annotate_retries(format_loose_ends(resp), retries)
}) })
.await .await
} }
@ -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_loose_ends` 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_loose_end(&self, Parameters(args): Parameters<CancelLooseEndArgs>) -> String {
let log = format!("{args:?}");
let kind_label = args.kind.clone();
let id = args.id;
run_tool_envelope("cancel_loose_end", log, async move {
let kind = match parse_loose_end_kind(&args.kind) {
Ok(k) => k,
Err(e) => return e,
};
let (resp, retries) = self
.dispatch(hive_sh4re::ManagerRequest::CancelLooseEnd { kind, id })
.await;
annotate_retries(
format_ack(resp, "cancel_loose_end", 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 \
@ -994,9 +1092,10 @@ impl ManagerServer {
any agent including yourself), `ask` (structured question to the operator or a \ any agent including yourself), `ask` (structured question to the operator or a \
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_loose_ends` (hive-wide loose ends pending approvals + unanswered \
questions across the swarm), `whoami` (self-introspection canonical name, \ questions + pending reminders across the swarm), `cancel_loose_end` (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 {}
@ -1037,8 +1136,9 @@ pub fn allowed_mcp_tools(flavor: Flavor) -> Vec<String> {
"ask", "ask",
"answer", "answer",
"remind", "remind",
"get_open_threads", "get_loose_ends",
"whoami", "whoami",
"cancel_loose_end",
], ],
Flavor::Manager => &[ Flavor::Manager => &[
"send", "send",
@ -1052,9 +1152,10 @@ pub fn allowed_mcp_tools(flavor: Flavor) -> Vec<String> {
"ask", "ask",
"answer", "answer",
"get_logs", "get_logs",
"get_open_threads", "get_loose_ends",
"remind", "remind",
"whoami", "whoami",
"cancel_loose_end",
], ],
}; };
let mut out: Vec<String> = names let mut out: Vec<String> = names

View file

@ -106,7 +106,7 @@ pub async fn serve(
.route("/api/compact", post(post_compact)) .route("/api/compact", post(post_compact))
.route("/api/model", post(post_set_model)) .route("/api/model", post(post_set_model))
.route("/api/new-session", post(post_new_session)) .route("/api/new-session", post(post_new_session))
.route("/api/open-threads", get(api_open_threads)) .route("/api/loose-ends", get(api_loose_ends))
.with_state(state); .with_state(state);
let addr = SocketAddr::from(([0, 0, 0, 0], port)); let addr = SocketAddr::from(([0, 0, 0, 0], port));
let listener = bind_with_retry(addr, "web UI").await?; let listener = bind_with_retry(addr, "web UI").await?;
@ -240,25 +240,25 @@ struct SessionView {
exit_note: Option<String>, exit_note: Option<String>,
} }
/// Proxy this agent's open-threads via the per-agent socket. The /// Proxy this agent's loose-ends list via the per-agent socket. The
/// web UI surfaces the result as a collapsible section in the page /// web UI surfaces the result as a collapsible section in the page
/// so the operator can see at a glance what's pending against the /// so the operator can see at a glance what's pending against the
/// agent (questions asked by it, peer questions targeting it, /// agent (questions asked by it, peer questions targeting it,
/// approvals for the manager). Same data the /// reminders it scheduled, approvals for the manager). Same data
/// `mcp__hyperhive__get_open_threads` tool sees from inside the /// the `mcp__hyperhive__get_loose_ends` tool sees from inside the
/// container. /// container.
async fn api_open_threads(State(state): State<AppState>) -> Response { async fn api_loose_ends(State(state): State<AppState>) -> Response {
let threads: Vec<hive_sh4re::OpenThread> = match state.flavor() { let loose_ends: Vec<hive_sh4re::LooseEnd> = match state.flavor() {
Flavor::Agent => { Flavor::Agent => {
match client::request::<_, hive_sh4re::AgentResponse>( match client::request::<_, hive_sh4re::AgentResponse>(
&state.socket, &state.socket,
&hive_sh4re::AgentRequest::GetOpenThreads, &hive_sh4re::AgentRequest::GetLooseEnds,
) )
.await .await
{ {
Ok(hive_sh4re::AgentResponse::OpenThreads { threads }) => threads, Ok(hive_sh4re::AgentResponse::LooseEnds { loose_ends }) => loose_ends,
Ok(hive_sh4re::AgentResponse::Err { message }) => { Ok(hive_sh4re::AgentResponse::Err { message }) => {
return error_response(&format!("get_open_threads: {message}")); return error_response(&format!("get_loose_ends: {message}"));
} }
Ok(other) => return error_response(&format!("unexpected response: {other:?}")), Ok(other) => return error_response(&format!("unexpected response: {other:?}")),
Err(e) => return error_response(&format!("transport: {e:#}")), Err(e) => return error_response(&format!("transport: {e:#}")),
@ -267,20 +267,20 @@ async fn api_open_threads(State(state): State<AppState>) -> Response {
Flavor::Manager => { Flavor::Manager => {
match client::request::<_, hive_sh4re::ManagerResponse>( match client::request::<_, hive_sh4re::ManagerResponse>(
&state.socket, &state.socket,
&hive_sh4re::ManagerRequest::GetOpenThreads, &hive_sh4re::ManagerRequest::GetLooseEnds,
) )
.await .await
{ {
Ok(hive_sh4re::ManagerResponse::OpenThreads { threads }) => threads, Ok(hive_sh4re::ManagerResponse::LooseEnds { loose_ends }) => loose_ends,
Ok(hive_sh4re::ManagerResponse::Err { message }) => { Ok(hive_sh4re::ManagerResponse::Err { message }) => {
return error_response(&format!("get_open_threads: {message}")); return error_response(&format!("get_loose_ends: {message}"));
} }
Ok(other) => return error_response(&format!("unexpected response: {other:?}")), Ok(other) => return error_response(&format!("unexpected response: {other:?}")),
Err(e) => return error_response(&format!("transport: {e:#}")), Err(e) => return error_response(&format!("transport: {e:#}")),
} }
} }
}; };
axum::Json(serde_json::json!({ "threads": threads })).into_response() axum::Json(serde_json::json!({ "loose_ends": loose_ends })).into_response()
} }
async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> { async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> {

View file

@ -174,8 +174,8 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
timing, timing,
file_path, file_path,
} => handle_remind(coord, agent, message, timing, file_path.as_deref()), } => handle_remind(coord, agent, message, timing, file_path.as_deref()),
AgentRequest::GetOpenThreads => match crate::open_threads::for_agent(coord, agent) { AgentRequest::GetLooseEnds => match crate::loose_ends::for_agent(coord, agent) {
Ok(threads) => AgentResponse::OpenThreads { threads }, Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
Err(e) => AgentResponse::Err { Err(e) => AgentResponse::Err {
message: format!("{e:#}"), message: format!("{e:#}"),
}, },
@ -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::CancelLooseEnd { kind, id } => crate::questions::handle_cancel_loose_end(
coord, agent, *kind, *id,
)
.map_or_else(
|message| AgentResponse::Err { message },
|()| AgentResponse::Ok,
),
} }
} }

View file

@ -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

View file

@ -1,7 +1,7 @@
//! Loose-ends aggregator. Walks the `approvals` + `operator_questions` //! Loose-ends aggregator. Walks the `approvals` + `operator_questions`
//! tables once per call and assembles a `Vec<OpenThread>` for either //! tables once per call and assembles a `Vec<LooseEnd>` for either
//! a single agent (`for_agent`) or the whole hive (`hive_wide`). Both //! a single agent (`for_agent`) or the whole hive (`hive_wide`). Both
//! `AgentRequest::GetOpenThreads` and `ManagerRequest::GetOpenThreads` //! `AgentRequest::GetLooseEnds` and `ManagerRequest::GetLooseEnds`
//! land here so the routing logic + age-seconds derivation stay in //! land here so the routing logic + age-seconds derivation stay in
//! one place. //! one place.
//! //!
@ -16,7 +16,7 @@
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
use anyhow::Result; use anyhow::Result;
use hive_sh4re::{MANAGER_AGENT, OpenThread}; use hive_sh4re::{MANAGER_AGENT, LooseEnd};
use crate::coordinator::Coordinator; use crate::coordinator::Coordinator;
@ -26,10 +26,13 @@ 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
pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<OpenThread>> { /// 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<LooseEnd>> {
let now = now_unix(); let now = now_unix();
let mut out = Vec::new(); let mut out = Vec::new();
// Approvals are only submitted by the manager today. When that // Approvals are only submitted by the manager today. When that
@ -38,7 +41,7 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<OpenThread>> {
// here on that column — for now MANAGER_AGENT == sole submitter. // here on that column — for now MANAGER_AGENT == sole submitter.
if agent == MANAGER_AGENT { if agent == MANAGER_AGENT {
for a in coord.approvals.pending()? { for a in coord.approvals.pending()? {
out.push(OpenThread::Approval { out.push(LooseEnd::Approval {
id: a.id, id: a.id,
agent: a.agent, agent: a.agent,
commit_ref: a.commit_ref, commit_ref: a.commit_ref,
@ -52,7 +55,7 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<OpenThread>> {
if !role_match { if !role_match {
continue; continue;
} }
out.push(OpenThread::Question { out.push(LooseEnd::Question {
id: q.id, id: q.id,
asker: q.asker, asker: q.asker,
target: q.target, target: q.target,
@ -60,18 +63,30 @@ 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(LooseEnd::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<LooseEnd>> {
let now = now_unix(); let now = now_unix();
let mut out = Vec::new(); let mut out = Vec::new();
for a in coord.approvals.pending()? { for a in coord.approvals.pending()? {
out.push(OpenThread::Approval { out.push(LooseEnd::Approval {
id: a.id, id: a.id,
agent: a.agent, agent: a.agent,
commit_ref: a.commit_ref, commit_ref: a.commit_ref,
@ -80,7 +95,7 @@ pub fn hive_wide(coord: &Coordinator) -> Result<Vec<OpenThread>> {
}); });
} }
for q in coord.questions.pending_all()? { for q in coord.questions.pending_all()? {
out.push(OpenThread::Question { out.push(LooseEnd::Question {
id: q.id, id: q.id,
asker: q.asker, asker: q.asker,
target: q.target, target: q.target,
@ -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(LooseEnd::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)
} }

View file

@ -20,10 +20,10 @@ mod events_vacuum;
mod forge; mod forge;
mod lifecycle; mod lifecycle;
mod limits; mod limits;
mod loose_ends;
mod manager_server; mod manager_server;
mod meta; mod meta;
mod migrate; mod migrate;
mod open_threads;
mod operator_questions; mod operator_questions;
mod questions; mod questions;
mod reminder_scheduler; mod reminder_scheduler;

View file

@ -329,8 +329,8 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
}, },
} }
} }
ManagerRequest::GetOpenThreads => match crate::open_threads::hive_wide(coord) { ManagerRequest::GetLooseEnds => match crate::loose_ends::hive_wide(coord) {
Ok(threads) => ManagerResponse::OpenThreads { threads }, Ok(loose_ends) => ManagerResponse::LooseEnds { loose_ends },
Err(e) => ManagerResponse::Err { Err(e) => ManagerResponse::Err {
message: format!("{e:#}"), message: format!("{e:#}"),
}, },
@ -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::CancelLooseEnd { kind, id } => crate::questions::handle_cancel_loose_end(
coord,
MANAGER_AGENT,
*kind,
*id,
)
.map_or_else(
|message| ManagerResponse::Err { message },
|()| ManagerResponse::Ok,
),
} }
} }

View file

@ -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();

View file

@ -129,6 +129,49 @@ pub fn handle_answer(
Ok(()) Ok(())
} }
/// Handle `CancelLooseEnd` 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_loose_end(
coord: &Arc<Coordinator>,
canceller: &str,
kind: hive_sh4re::CancelLooseEndKind,
id: i64,
) -> Result<(), String> {
match kind {
hive_sh4re::CancelLooseEndKind::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::CancelLooseEndKind::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

View file

@ -185,16 +185,21 @@ pub enum ReminderTiming {
At { unix_timestamp: i64 }, At { unix_timestamp: i64 },
} }
/// One row in the response to `GetOpenThreads`. Tagged enum so new /// One row in the response to `GetLooseEnds`. Tagged enum so new
/// thread kinds (forge PRs, long-running approvals from a privileged /// thread kinds (forge PRs, long-running approvals from a privileged
/// bot, etc) can land later without breaking existing handlers. The /// bot, etc) can land later without breaking existing handlers. The
/// 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
/// `CancelLooseEnd` request (and the `cancel_loose_end` 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 LooseEnd {
/// A pending approval. For agent-flavour `GetOpenThreads` calls /// A pending approval. For agent-flavour `GetLooseEnds` calls
/// this only surfaces when the agent itself is the manager /// this only surfaces when the agent itself is the manager
/// (sub-agents don't submit approvals). For manager-flavour calls /// (sub-agents don't submit approvals). For manager-flavour calls
/// it lists every pending approval in the swarm. `agent` is the /// it lists every pending approval in the swarm. `agent` is the
@ -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 `CancelLooseEnd`. 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 CancelLooseEndKind {
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
@ -308,7 +340,7 @@ pub enum AgentRequest {
/// agent submitted them (which only ever happens for the /// agent submitted them (which only ever happens for the
/// manager); questions surface where the agent is `asker` or /// manager); questions surface where the agent is `asker` or
/// `target`. Cheap O(n) sweep server-side — no caching. /// `target`. Cheap O(n) sweep server-side — no caching.
GetOpenThreads, GetLooseEnds,
/// Count of this agent's pending (un-delivered) reminders. Used /// Count of this agent's pending (un-delivered) reminders. Used
/// by the harness's per-turn stats sink to snapshot "what was /// by the harness's per-turn stats sink to snapshot "what was
/// queued at turn-end time" without paying for a full list. /// queued at turn-end time" without paying for a full list.
@ -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.
CancelLooseEnd { kind: CancelLooseEndKind, id: i64 },
} }
/// Responses on a per-agent socket. /// Responses on a per-agent socket.
@ -340,9 +379,9 @@ pub enum AgentResponse {
/// `Ask` result: the queued question id. The answer lands later /// `Ask` result: the queued question id. The answer lands later
/// as `HelperEvent::QuestionAnswered` in this agent's inbox. /// as `HelperEvent::QuestionAnswered` in this agent's inbox.
QuestionQueued { id: i64 }, QuestionQueued { id: i64 },
/// `GetOpenThreads` result: list of loose ends pending against /// `GetLooseEnds` result: list of loose ends pending against
/// this agent. Ordered newest-first within each kind. /// this agent. Ordered newest-first within each kind.
OpenThreads { threads: Vec<OpenThread> }, LooseEnds { loose_ends: Vec<LooseEnd> },
/// `CountPendingReminders` result. /// `CountPendingReminders` result.
PendingRemindersCount { count: u64 }, PendingRemindersCount { count: u64 },
/// `Whoami` result: identity + role + the current hyperhive rev /// `Whoami` result: identity + role + the current hyperhive rev
@ -616,15 +655,19 @@ pub enum ManagerRequest {
/// Hive-wide loose-ends view: EVERY pending approval + EVERY /// Hive-wide loose-ends view: EVERY pending approval + EVERY
/// unanswered question in the swarm. Used by the manager to scan /// unanswered question in the swarm. Used by the manager to scan
/// for stalled coordination — the per-agent equivalent on the /// for stalled coordination — the per-agent equivalent on the
/// sub-agent surface is `AgentRequest::GetOpenThreads` which /// sub-agent surface is `AgentRequest::GetLooseEnds` which
/// only returns rows where the agent itself is asker / target. /// only returns rows where the agent itself is asker / target.
GetOpenThreads, GetLooseEnds,
/// Count of the manager's own pending reminders. Mirror of /// Count of the manager's own pending reminders. Mirror of
/// `AgentRequest::CountPendingReminders` on the manager surface. /// `AgentRequest::CountPendingReminders` on the manager surface.
CountPendingReminders, CountPendingReminders,
/// 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::CancelLooseEnd` but with privileged auth.
CancelLooseEnd { kind: CancelLooseEndKind, id: i64 },
} }
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
@ -657,11 +700,11 @@ pub enum ManagerResponse {
Logs { Logs {
content: String, content: String,
}, },
/// `GetOpenThreads` result: hive-wide loose ends (approvals + /// `GetLooseEnds` result: hive-wide loose ends (approvals +
/// unanswered questions). Same `OpenThread` variants as the /// unanswered questions). Same `LooseEnd` variants as the
/// agent surface; the manager's view is unfiltered. /// agent surface; the manager's view is unfiltered.
OpenThreads { LooseEnds {
threads: Vec<OpenThread>, loose_ends: Vec<LooseEnd>,
}, },
/// `CountPendingReminders` result. /// `CountPendingReminders` result.
PendingRemindersCount { PendingRemindersCount {