diff --git a/app/src/components/orchestration/AgentChatPanel.tsx b/app/src/components/orchestration/AgentChatPanel.tsx index 2cad3f2538..10ba4100b5 100644 --- a/app/src/components/orchestration/AgentChatPanel.tsx +++ b/app/src/components/orchestration/AgentChatPanel.tsx @@ -19,12 +19,14 @@ import debugFactory from 'debug'; import { type KeyboardEvent, type ReactNode, + type Ref, useCallback, useEffect, useRef, useState, } from 'react'; +import { useStickToBottom } from '../../hooks/useStickToBottom'; import { useT } from '../../lib/i18n/I18nContext'; import { orchestrationClient, @@ -47,6 +49,10 @@ import SessionTranscript from './SessionTranscript'; const debug = debugFactory('orchestration:agent-chat'); +// Stable identity for an empty transcript so `useStickToBottom`'s layout effect +// doesn't re-run every render when the selected chat has no messages yet. +const EMPTY_MESSAGES: readonly unknown[] = []; + function sessionLabel(session: SessionSummary): string { return session.label?.trim() || session.sessionId; } @@ -60,10 +66,12 @@ function sessionLabel(session: SessionSummary): string { function ChatPageScaffold({ header, footer, + scrollRef, children, }: { header?: ReactNode; footer?: ReactNode; + scrollRef?: Ref; children: ReactNode; }) { const footerRef = useRef(null); @@ -91,6 +99,8 @@ function ChatPageScaffold({
{header}
{children} @@ -184,6 +194,11 @@ function AgentComposer({ function SessionChatView({ session }: { session: SessionSummary }) { const { t } = useT(); const { state, messages, refresh } = useSessionTranscript(session.sessionId); + const { containerRef: scrollRef } = useStickToBottom( + messages, + session.sessionId, + session.sessionId + ); const [body, setBody] = useState(''); const [sending, setSending] = useState(false); const [sendError, setSendError] = useState(null); @@ -232,6 +247,7 @@ function SessionChatView({ session }: { session: SessionSummary }) { return ( @@ -343,6 +359,14 @@ export default function AgentChatPanel({ sendMessage, } = useOrchestrationChats(t); const contactSessions = useContactSessions(); + // Keep the transcript pinned to the newest message (and disengage when the + // user scrolls up). Called before the `openSession` early return so hook order + // stays stable; `selectedId` as thread + reset key snaps fresh on tab switch. + const { containerRef: masterScrollRef } = useStickToBottom( + selected?.messages ?? EMPTY_MESSAGES, + selectedId, + selectedId + ); const [composerBody, setComposerBody] = useState(''); const [sending, setSending] = useState(false); @@ -456,6 +480,7 @@ export default function AgentChatPanel({ // switching chip in the footer. When subconscious is active, the steering // directive + "Run review" ride alongside the chip (no header bar). {showComposer ? ( diff --git a/app/src/components/orchestration/__tests__/AgentChatPanel.test.tsx b/app/src/components/orchestration/__tests__/AgentChatPanel.test.tsx index c8a7bf0820..540f16167d 100644 --- a/app/src/components/orchestration/__tests__/AgentChatPanel.test.tsx +++ b/app/src/components/orchestration/__tests__/AgentChatPanel.test.tsx @@ -163,4 +163,76 @@ describe('AgentChatPanel', () => { render(); expect(screen.getByText(/load failed/)).toBeInTheDocument(); }); + + // ── Autoscroll (regression: new master message snapped to the TOP) ───────── + const msg = (id: string) => ({ + id, + from: 'you', + body: id, + timestamp: '2026-07-08T00:00:00Z', + encrypted: false, + }); + + // jsdom does no layout, so `scrollTop`/`scrollHeight`/`clientHeight` are inert. + // Back them with a stored value so the stick-to-bottom snap is observable. + const fakeMetrics = (el: HTMLElement, scrollHeight: number, clientHeight: number) => { + let top = 0; + Object.defineProperty(el, 'scrollHeight', { value: scrollHeight, configurable: true }); + Object.defineProperty(el, 'clientHeight', { value: clientHeight, configurable: true }); + Object.defineProperty(el, 'scrollTop', { + get: () => top, + set: v => { + top = v; + }, + configurable: true, + }); + }; + + it('pins the newest master message to the bottom on a new message (not the top)', () => { + chatsApi.current = { + ...chatsApi.current, + selectedId: 'master', + selected: { id: 'master', title: 'Master', messages: [msg('m1')] }, + }; + const { rerender } = render(); + const scroll = screen.getByTestId('orch-chat-scroll') as HTMLDivElement; + fakeMetrics(scroll, 1000, 400); + scroll.scrollTop = 0; // as if reset to the top by the loading-spinner swap + + chatsApi.current = { + ...chatsApi.current, + selected: { id: 'master', title: 'Master', messages: [msg('m1'), msg('m2')] }, + }; + rerender(); + + expect(scroll.scrollTop).toBe(1000); // snapped to the bottom, not left at 0 + }); + + it('does not yank the master chat down when the user has scrolled up', () => { + chatsApi.current = { + ...chatsApi.current, + selectedId: 'master', + selected: { id: 'master', title: 'Master', messages: [msg('m1')] }, + }; + const { rerender } = render(); + const scroll = screen.getByTestId('orch-chat-scroll') as HTMLDivElement; + fakeMetrics(scroll, 1000, 400); + scroll.scrollTop = 0; // 600px from the bottom, past the 80px threshold + fireEvent.scroll(scroll); // disengages stickiness + + chatsApi.current = { + ...chatsApi.current, + selected: { id: 'master', title: 'Master', messages: [msg('m1'), msg('m2')] }, + }; + rerender(); + + expect(scroll.scrollTop).toBe(0); // left where the user parked it + }); + + it('renders with no selected chat (covers the empty-messages fallback)', () => { + chatsApi.current = { ...chatsApi.current, selected: undefined as never }; + render(); + // Exercises `selected?.messages ?? EMPTY_MESSAGES`; the panel still renders. + expect(screen.getByTestId('orch-agent-tab-master')).toBeInTheDocument(); + }); }); diff --git a/src/openhuman/agent_orchestration/background_completions.rs b/src/openhuman/agent_orchestration/background_completions.rs index 1ff7512fa7..8e7be4ca7c 100644 --- a/src/openhuman/agent_orchestration/background_completions.rs +++ b/src/openhuman/agent_orchestration/background_completions.rs @@ -33,6 +33,12 @@ pub(crate) struct CompletedBackgroundAgent { /// batch of threads is deleted. const CANCELLED_TOMBSTONE_CAP: usize = 512; +/// Upper bound on the collected-task tombstone set. A completion records within +/// seconds of the parent collecting it inline, so only recently collected task +/// ids can still be racing a late record; older tombstones are evicted in +/// insertion order. +const COLLECTED_TOMBSTONE_CAP: usize = 512; + /// Shared state behind a single mutex so the cancellation check in /// [`record_completion`] is atomic against the tombstone+sweep in /// [`discard_for_thread`] — otherwise the cooperative-abort race could enqueue a @@ -49,6 +55,15 @@ struct QueueState { cancelled_threads: HashSet, /// Insertion order for `cancelled_threads`, used to bound the set. cancelled_order: VecDeque, + /// Task ids the parent already collected inline via `wait_subagent` and will + /// present in its own turn. A completion for a collected task is dropped by + /// [`record_completion`] (closing the wait/record ordering race) and any + /// already-queued entry is swept by [`mark_collected`], so background + /// delivery never re-answers a result the master already surfaced (the + /// duplicate-response bug). + collected_tasks: HashSet, + /// Insertion order for `collected_tasks`, used to bound the set. + collected_order: VecDeque, } impl QueueState { @@ -63,6 +78,19 @@ impl QueueState { } } } + + /// Tombstone `task_id` so a completion that records after the parent + /// collected it inline is dropped rather than delivered again. + fn tombstone_collected(&mut self, task_id: &str) { + if self.collected_tasks.insert(task_id.to_string()) { + self.collected_order.push_back(task_id.to_string()); + while self.collected_order.len() > COLLECTED_TOMBSTONE_CAP { + if let Some(evicted) = self.collected_order.pop_front() { + self.collected_tasks.remove(&evicted); + } + } + } + } } static QUEUE: OnceLock> = OnceLock::new(); @@ -104,6 +132,18 @@ pub(crate) fn record_completion( return; } } + // The parent already collected this result inline (`wait_subagent`) and + // presents it in its own turn, so a background-delivery turn for it would + // just re-answer the same thing. Drop it (closes the wait-before-record + // race; the record-before-wait order is handled by the sweep in + // `mark_collected`). + if state.collected_tasks.contains(&entry.task_id) { + log::debug!( + "[background_completions] dropping completion task_id={} already collected inline", + entry.task_id + ); + return; + } let pending = state.pending.entry(parent_session).or_default(); if pending.iter().any(|c| c.task_id == entry.task_id) { return; @@ -172,6 +212,33 @@ pub(crate) fn discard_for_thread(thread_id: &str) -> usize { removed } +/// Mark `task_id` as collected inline by the parent (via `wait_subagent`) so its +/// background completion is not independently delivered as a second, duplicate +/// answer. Tombstones the id — bounded — so a completion that records *after* +/// this call (the wait-before-record ordering) is dropped by +/// [`record_completion`], and sweeps any entry already queued for it across all +/// sessions (the record-before-wait ordering). Both orderings resolve +/// atomically under the single queue mutex. Returns whether a queued entry was +/// removed. +pub(crate) fn mark_collected(task_id: &str) -> bool { + let mut state = queue() + .lock() + .expect("background_completions queue poisoned"); + state.tombstone_collected(task_id); + let mut removed = false; + for pending in state.pending.values_mut() { + let before = pending.len(); + pending.retain(|c| c.task_id != task_id); + removed |= pending.len() != before; + } + // Drop now-empty session buckets so the map doesn't accumulate keys. + state.pending.retain(|_, v| !v.is_empty()); + log::debug!( + "[background_completions] mark_collected task_id={task_id} removed_queued={removed}" + ); + removed +} + /// Wipe every queued completion across all sessions. Called on a full thread /// purge. Tombstones are left intact (the per-thread protection set by /// [`discard_for_thread`]); the purge path tombstones each in-flight sub-agent's @@ -401,4 +468,67 @@ mod tests { assert!(!has_pending("sess-c2")); assert_eq!(clear_all(), 0); } + + #[test] + fn mark_collected_sweeps_the_queued_entry() { + let _guard = test_guard(); + let s = "sess-mc-sweep"; + record_completion(s, "mc-sub-1", "researcher", "collected", None); + record_completion(s, "mc-sub-2", "researcher", "keep", None); + + // The parent collected sub-1 inline, so it must not be delivered again; + // sub-2 (never waited on) survives for normal idle delivery. + assert!(mark_collected("mc-sub-1"), "swept the queued entry"); + assert_eq!(pending_count(s), 1); + let drained = take_pending(s); + assert_eq!(drained[0].task_id, "mc-sub-2"); + } + + #[test] + fn record_after_mark_collected_is_dropped_by_tombstone() { + let _guard = test_guard(); + // Collecting inline tombstones the task id... + assert!( + !mark_collected("mc-late"), + "nothing queued yet, so nothing swept" + ); + // ...so a completion that records *after* (the wait-before-record order) + // is dropped rather than queued for a duplicate delivery turn. + record_completion("sess-mc-race", "mc-late", "researcher", "stale", None); + assert_eq!( + pending_count("sess-mc-race"), + 0, + "a completion collected inline must not be re-delivered" + ); + } + + #[test] + fn mark_collected_is_task_scoped() { + let _guard = test_guard(); + let s = "sess-mc-scope"; + // Only the collected task is suppressed; an un-waited sibling still + // surfaces (the genuinely-later fire-and-forget feature is preserved). + mark_collected("mc-scope-1"); + record_completion(s, "mc-scope-2", "researcher", "later", None); + assert_eq!(pending_count(s), 1); + assert!(has_pending(s)); + take_pending(s); + } + + #[test] + fn collected_tombstone_is_bounded() { + let _guard = test_guard(); + for i in 0..(COLLECTED_TOMBSTONE_CAP + 50) { + mark_collected(&format!("mc-bound-{i}")); + } + let len = queue() + .lock() + .expect("queue poisoned") + .collected_tasks + .len(); + assert!( + len <= COLLECTED_TOMBSTONE_CAP, + "collected tombstone must stay bounded, got {len}" + ); + } } diff --git a/src/openhuman/agent_orchestration/tools/wait_subagent.rs b/src/openhuman/agent_orchestration/tools/wait_subagent.rs index 3e9d6dbe97..6dc4ae9b75 100644 --- a/src/openhuman/agent_orchestration/tools/wait_subagent.rs +++ b/src/openhuman/agent_orchestration/tools/wait_subagent.rs @@ -161,6 +161,16 @@ impl Tool for WaitSubagentTool { resolved_task_id, iterations ); + // The parent is collecting this result inline and will present + // it in this turn, so suppress the detached completion's separate + // background-delivery turn — otherwise the same result is + // re-answered as a duplicate. Only the Completed arm marks: + // AwaitingUser/Failed never record a completion, and a + // still-Running/TimedOut sub-agent has no terminal result yet, so + // a genuinely-later completion must still surface. + crate::openhuman::agent_orchestration::background_completions::mark_collected( + &resolved_task_id, + ); let status = wait_status_payload( resume_ref.as_ref(), &resolved_task_id, diff --git a/src/openhuman/agent_registry/agents/orchestrator/prompt.md b/src/openhuman/agent_registry/agents/orchestrator/prompt.md index e0c080a33e..e95122d0ab 100644 --- a/src/openhuman/agent_registry/agents/orchestrator/prompt.md +++ b/src/openhuman/agent_registry/agents/orchestrator/prompt.md @@ -22,6 +22,7 @@ Follow this sequence for every user message: - Find the matching toolkit in the **Connected Integrations** section and call `delegate_to_integrations_agent` with that `toolkit`. - **Do this even if remembered context could plausibly answer.** The user wants the live source of truth, not a stale summary. - If the relevant toolkit is **not** in **Connected Integrations**, call `composio_connect { toolkit: "" }` **directly** to raise an **inline connect card** so the user can authorize in one click, then continue the task once it returns `connected: true`. Do **not** refuse based on the Connected Integrations list (that is only what is *already* connected, not what is *connectable*), do **not** make "go to Settings → Connections" your first move, and do **not** silently fall back to memory retrieval (see "Connecting external services" below). + - **Scope gate (required):** treat this as an external-service request ONLY if the ask actually operates on that service's own data or actions — its inbox/messages, files, calendar events, docs, tickets, etc. A service merely being *connected* is **not** a reason to touch it. General-knowledge answers, web/news lookups, headlines, date/time, and math must **not** spawn `delegate_to_integrations_agent` (or any email/inbox/calendar fetch) even when Gmail/Notion/etc. are connected — route those to a direct tool (Step 3) or the matching non-integration specialist (Step 4). When the request neither names nor clearly implies a specific service's own data or actions, do not reach into one — a clear implication ("check my inbox", "send an email") still counts as naming it and should be delivered; only a request that references no service at all (e.g. "today's date") stays off delegation. 3. **Can I solve this with direct tools?** - Yes: use direct tools (`memory_recall`, `read_workspace_state`, `composio_list_connections`, task tools, etc.). - **Memory is direct work.** Recalling a fact (`memory_recall`), storing a fact (`memory_store`), or saving a preference (`save_preference`) needs **no sub-agent** — call the direct tool and confirm the result. After a `memory_store` write, follow the memory protocol and call `update_memory_md` (targeting `MEMORY.md`) to keep the index in sync with the store; `save_preference` writes the preference store and needs no such reconcile. Reserve the `retrieve_memory` and `manage_profile_memory` **delegates** for genuinely deep work: multi-hop memory-tree walks, ingest/index into the long-term tree, reconciling overlapping notes, people-graph/alias management, or persona edits. Do **not** spawn a sub-agent for a single recall or a single "remember this". diff --git a/src/openhuman/agent_registry/agents/orchestrator/prompt.rs b/src/openhuman/agent_registry/agents/orchestrator/prompt.rs index 9e324f9a47..83248b1059 100644 --- a/src/openhuman/agent_registry/agents/orchestrator/prompt.rs +++ b/src/openhuman/agent_registry/agents/orchestrator/prompt.rs @@ -248,7 +248,10 @@ fn render_delegation_guide(integrations: &[ConnectedIntegration]) -> String { "## Connected Integrations\n\n\ IMPORTANT: You MUST use the `delegate_to_integrations_agent` tool for any request \ involving connected services. You do NOT have direct access to these services — all \ - interaction must go through delegation. Never claim you cannot access a connected \ + interaction must go through delegation. Delegate here ONLY when the request actually \ + operates on a connected service's data or actions; a connected service is not a reason \ + to touch it for general-knowledge, web/news, headline, date/time, or math questions. \ + Never claim you cannot access a connected \ service without first attempting delegation.\n\n\ The following services have an active connection. Their tool implementations \ live inside the `integrations_agent` sub-agent — NOT in your own tool list. \ @@ -614,6 +617,46 @@ mod tests { ); } + #[test] + fn build_scope_gates_integrations_delegation() { + // Regression: a connected service (e.g. Gmail) is not, by itself, a + // reason to operate on it — a general-knowledge / web / date ask that + // names no service must NOT spawn `delegate_to_integrations_agent`. + // Guards both the static Step-2 scope gate and the rendered + // delegation-guide clause. + let no_integrations = build(&ctx_with(&[])).unwrap(); + assert!( + no_integrations + .contains("General-knowledge answers, web/news lookups, headlines, date/time"), + "Step-2 scope gate must keep general/web/date asks off integrations delegation" + ); + assert!( + no_integrations + .contains("neither names nor clearly implies a specific service's own data"), + "Step-2 scope gate must forbid reaching into an unreferenced service" + ); + + let gmail = vec![ConnectedIntegration { + toolkit: "gmail".into(), + description: "Email access.".into(), + tools: Vec::new(), + gated_tools: Vec::new(), + connected: true, + connections: Vec::new(), + non_active_status: None, + }]; + let with_gmail = build(&ctx_with(&gmail)).unwrap(); + assert!( + with_gmail + .contains("a connected service is not a reason to touch it for general-knowledge"), + "delegation guide must carry the scoping clause when integrations are connected" + ); + // The existing always-delegate contract for real service asks is preserved. + assert!(with_gmail.contains( + "Never claim you cannot access a connected service without first attempting delegation" + )); + } + #[test] fn build_does_not_route_scope_errors_as_disconnected() { let body = build(&ctx_with(&[])).unwrap(); diff --git a/src/openhuman/orchestration/effect_executor.rs b/src/openhuman/orchestration/effect_executor.rs index 6e48647073..0cfb96867b 100644 --- a/src/openhuman/orchestration/effect_executor.rs +++ b/src/openhuman/orchestration/effect_executor.rs @@ -322,6 +322,16 @@ async fn run_local_agent_and_forward( body: body.clone(), timestamp: now.clone(), seq, + // Bookkeeping row: this raw `[local sub-agent … completed]` dump is + // forwarded to the brain (below) purely so it can synthesize a + // reply — the user reads that synthesized `send_dm` reply, not this. + // Tagging it with an excluded event_kind keeps the row (so `seq` + // stays allocated and the envelope forwards) but hides it from the + // master transcript, previews, and unread counts (store.rs filters + // out 'status'/'lifecycle'/'unknown'/'session_info'). Without this, + // a brain that spawns many sub-agents floods the chat with raw + // dumps. See list_messages_by_session / count_unread in store.rs. + event_kind: Some("lifecycle".to_string()), ..Default::default() }, )?; @@ -340,7 +350,27 @@ async fn run_local_agent_and_forward( ts, "tool_completion", ); - super::cloud::push_event(&config, &envelope).await?; + if let Err(e) = super::cloud::push_event(&config, &envelope).await { + // Forward failed (offline / signed out / retries exhausted): the brain + // never got this result and the completion row was persisted hidden, so + // it would vanish entirely. Un-hide it — it is now the only copy — and + // nudge the renderer so the user sees the result rather than losing it. + let completion_id = format!("tool-completion:{task_id}:{seq}"); + if let Err(store_err) = super::store::with_connection(&config.workspace_dir, |conn| { + super::store::clear_message_event_kind(conn, &completion_id) + }) { + log::warn!( + target: LOG, + "[orchestration] run_local_agent.unhide_failed task={task_id}: {store_err}" + ); + } + super::bus::notify_orchestration_message( + counterpart, + session_id, + super::types::ChatKind::Master.as_str(), + ); + return Err(e); + } log::debug!( target: LOG, "[orchestration] run_local_agent.forwarded task={task_id} session={session_id} seq={seq} ok={ok}" @@ -359,11 +389,54 @@ pub async fn handle_tool_call(data: &Value) -> Option<(String, Value)> { return None; } }; + // Dedup redelivered side-effecting local-execution tools (run_local_agent): + // `orch:tool_call` is at-least-once, so the same call can arrive twice, and + // without this each redelivery re-spawns the sub-agent AND forwards another + // `tool_completion` — which wakes another brain cycle and can surface a + // duplicate reply. Read-only tools (device_status) are idempotent and left + // un-guarded. Mirrors the guard in `handle_send_dm`. A successful async ack + // stays latched (a redelivery re-acks without re-spawning); a claim whose + // dispatch FAILS is released below so the redelivery re-runs and returns the + // real error instead of a fabricated accept. + if super::exec_gate::is_local_execution_tool(&frame.name) && is_duplicate_call(&frame.call_id) { + log::debug!( + target: LOG, + "[orchestration] tool_call.duplicate call_id={} name={} (re-acking, no re-dispatch)", + frame.call_id, + frame.name + ); + return Some(( + frame.call_id.clone(), + tool_result_frame( + &frame.call_id, + true, + json!({ "accepted": true, "status": "running", "duplicate": true }), + None, + ), + )); + } let (ok, result, error) = match dispatch_device_tool(&frame.name, &frame.args, &frame.cycle_id).await { Ok(value) => (true, value, None), Err(e) => (false, Value::Null, Some(e)), }; + // A claimed local-execution call whose dispatch FAILED — an A2A-gate denial + // (dispatch_device_tool restricts run_local_agent to Master cycles), an + // unknown cycle origin, or invalid args — must release its claim so an + // at-least-once redelivery re-runs it and returns the same real error. Left + // latched, the dedup fast-path above would fabricate an `accepted/running` ok + // for a call that never ran, masking the denial and stranding the brain on a + // `tool_completion` that never comes. + if !ok && super::exec_gate::is_local_execution_tool(&frame.name) { + // Diagnostic for the denied/invalid path; no raw args or error body. + log::warn!( + target: LOG, + "[orchestration] tool_call.dispatch_failed call_id={} name={} released_claim=true", + frame.call_id, + frame.name + ); + release_call(&frame.call_id); + } Some(( frame.call_id.clone(), tool_result_frame(&frame.call_id, ok, result, error.as_deref()), @@ -777,4 +850,95 @@ mod tests { "expected toolkit error, got: {msg}" ); } + + #[tokio::test] + async fn duplicate_run_local_agent_tool_call_is_reacked_without_redispatch() { + // A redelivered run_local_agent call_id must NOT re-spawn the sub-agent: + // the guard re-acks accepted/running without dispatching (dispatch would + // run_local_agent → spawn → forward a duplicate tool_completion). + let call_id = "call-dup-run-local-agent-test"; + assert!( + !is_duplicate_call(call_id), + "first claim is not a duplicate" + ); + let frame = serde_json::json!({ + "callId": call_id, + "name": "run_local_agent", + "cycleId": "cyc:openhuman:local:master:1", + "args": { "agent_id": "researcher", "prompt": "x" }, + }); + let (cid, result) = handle_tool_call(&frame).await.expect("frame parses"); + assert_eq!(cid, call_id); + assert_eq!(result["ok"].as_bool(), Some(true)); + assert_eq!( + result["result"]["duplicate"].as_bool(), + Some(true), + "redelivery re-acked as duplicate without re-dispatch" + ); + release_call(call_id); + } + + #[tokio::test] + async fn read_only_device_status_is_not_dedup_guarded() { + // Read-only tools are idempotent: even a previously-seen call_id still + // dispatches and returns real data — the guard is scoped to + // side-effecting local-execution tools, never device_status. + let call_id = "call-device-status-test"; + is_duplicate_call(call_id); // claim it as if already seen + let frame = serde_json::json!({ + "callId": call_id, + "name": "device_status", + "cycleId": "cyc:openhuman:local:master:1", + "args": {}, + }); + let (_, result) = handle_tool_call(&frame).await.expect("frame parses"); + assert_eq!(result["ok"].as_bool(), Some(true)); + assert!( + result["result"]["platform"].is_string(), + "real status returned, not the duplicate placeholder" + ); + assert!(result["result"].get("duplicate").is_none()); + release_call(call_id); + } + + #[tokio::test] + async fn failed_run_local_agent_dispatch_releases_claim_for_redelivery() { + // A run_local_agent on a non-Master (e.g. A2A) cycle is denied by the + // gate. Its claim must be released so an at-least-once redelivery re-runs + // and is denied AGAIN — never fabricated as accepted/duplicate (which + // would strand the brain waiting on a tool_completion that never comes). + let call_id = "call-a2a-denied-release-test"; + let frame = serde_json::json!({ + "callId": call_id, + "name": "run_local_agent", + "cycleId": "cyc:openhuman:a2a:@peer:5", // unregistered → not Master → denied + "args": { "agent_id": "researcher", "prompt": "x" }, + }); + let (_, first) = handle_tool_call(&frame).await.expect("frame parses"); + assert_eq!( + first["ok"].as_bool(), + Some(false), + "non-Master run_local_agent is denied" + ); + // Redelivery: the failed claim was released, so it re-dispatches and is + // denied again — not the duplicate re-ack. + let (_, second) = handle_tool_call(&frame).await.expect("frame parses"); + assert_eq!( + second["ok"].as_bool(), + Some(false), + "redelivery re-denied, not fabricated-accepted" + ); + assert!(second["result"].get("duplicate").is_none()); + // Same real denial both times — not a fabricated/different error. + assert_eq!(first["error"], second["error"]); + assert!( + second["error"] + .as_str() + .unwrap_or_default() + .contains("restricted to the Master chat"), + "the real non-Master denial: {}", + second["error"] + ); + release_call(call_id); + } } diff --git a/src/openhuman/orchestration/store.rs b/src/openhuman/orchestration/store.rs index 4b1d552c35..df06f96cbd 100644 --- a/src/openhuman/orchestration/store.rs +++ b/src/openhuman/orchestration/store.rs @@ -354,6 +354,18 @@ pub fn insert_message(conn: &Connection, m: &OrchestrationMessage) -> Result 0) } +/// Clear a message's `event_kind` (set it NULL), making a previously hidden +/// bookkeeping row visible again in the transcript / unread counts. Used to +/// un-hide a `tool_completion` row whose forward to the hosted brain failed, so +/// its result is not silently lost from the UI. Returns whether a row changed. +pub fn clear_message_event_kind(conn: &Connection, id: &str) -> Result { + let changed = conn.execute( + "UPDATE messages SET event_kind = NULL WHERE id = ?1", + params![id], + )?; + Ok(changed > 0) +} + /// Count persisted messages for a session (test/observability helper). pub fn count_messages(conn: &Connection, agent_id: &str, session_id: &str) -> Result { Ok(conn.query_row( @@ -949,6 +961,30 @@ mod tests { .unwrap(); } + #[test] + fn clear_message_event_kind_unhides_a_hidden_row() { + let tmp = tempfile::tempdir().unwrap(); + with_connection(tmp.path(), |conn| { + upsert_session(conn, &session("@a", "master", 1))?; + let hidden = OrchestrationMessage { + event_kind: Some("lifecycle".into()), + ..msg("tool-completion:cyc:1", "@a", "master", 1) + }; + insert_message(conn, &hidden)?; + // Hidden from the transcript while event_kind is an excluded kind. + assert!(list_messages_by_session(conn, "master", 50, None)?.is_empty()); + // Un-hiding it (clear event_kind) makes it visible again. + assert!(clear_message_event_kind(conn, "tool-completion:cyc:1")?); + let visible = list_messages_by_session(conn, "master", 50, None)?; + assert_eq!(visible.len(), 1); + assert_eq!(visible[0].id, "tool-completion:cyc:1"); + // A non-existent id changes nothing. + assert!(!clear_message_event_kind(conn, "nope")?); + Ok(()) + }) + .unwrap(); + } + #[test] fn persists_and_reads_back_tool_result_outcome() { let tmp = tempfile::tempdir().unwrap();