diff --git a/interface/web/src/Chatbot.jsx b/interface/web/src/Chatbot.jsx index 6fec25b..f331f1b 100644 --- a/interface/web/src/Chatbot.jsx +++ b/interface/web/src/Chatbot.jsx @@ -16,6 +16,7 @@ export function Chatbot({ conversationId, setConversationId, onConversationChang const [memoryToast, setMemoryToast] = useState(null); const [images, setImages] = useState([]); // {name, b64} for vision models const [activeTool, setActiveTool] = useState(null); // playbook tool currently running + const [pendingApproval, setPendingApproval] = useState(null); // [{name, arguments}] awaiting yes/no const [editingIdx, setEditingIdx] = useState(null); // user message being edited const [editText, setEditText] = useState(""); const [listening, setListening] = useState(false); // mic dictation active @@ -267,6 +268,11 @@ export function Chatbot({ conversationId, setConversationId, onConversationChang pendingEventType = null; continue; } + if (pendingEventType === "tool_request") { + try { setPendingApproval(JSON.parse(payload)); } catch { /* ignore */ } + pendingEventType = null; + continue; + } if (pendingEventType === "sources") { try { const src = JSON.parse(payload).sources; @@ -284,6 +290,7 @@ export function Chatbot({ conversationId, setConversationId, onConversationChang // slow post-processing (title, memory) on the still-open stream. setLoading(false); setActiveTool(null); + setPendingApproval(null); pendingEventType = null; continue; } @@ -309,6 +316,7 @@ export function Chatbot({ conversationId, setConversationId, onConversationChang try { token = JSON.parse(payload); } catch { /* plain text fallback */ } if (activeTool) setActiveTool(null); // tokens started -> tools done + if (pendingApproval) setPendingApproval(null); setMessages(prev => { const updated = [...prev]; updated[assistantIndex] = { @@ -373,10 +381,25 @@ export function Chatbot({ conversationId, setConversationId, onConversationChang await streamAssistant({ message: newText.trim(), history, images: [], assistantIndex: base.length }); }; + // Approve or deny the pending action tool(s); the open chat stream resumes. + const resolveApproval = async (approve) => { + const req = pendingApproval || []; + setPendingApproval(null); + const decisions = {}; + req.forEach(a => { decisions[a.name] = approve; }); + try { + await fetch(`${API_BASE}/chat/approve`, { + method: "POST", headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ conversation_id: conversationId, decisions }), + }); + } catch { /* ignore */ } + }; + const stopGeneration = () => { if (abortRef.current) abortRef.current.abort(); setLoading(false); setActiveTool(null); + if (pendingApproval) resolveApproval(false); // stopping = deny pending actions }; const updateAssistant = (index, text) => { @@ -647,6 +670,29 @@ export function Chatbot({ conversationId, setConversationId, onConversationChang
+ {pendingApproval && ( +
+
+ ⚠️ The assistant wants to run: + {" "} + {pendingApproval.map((a, i) => ( + + {a.name}({a.arguments ? Object.values(a.arguments).join(", ") : ""}) + + ))} +
+
+ + +
+
+ )} {activeTool && (
🔧 running tool: {activeTool}… diff --git a/interface/web/src/Settings.jsx b/interface/web/src/Settings.jsx index 0e84a9d..935cc34 100644 --- a/interface/web/src/Settings.jsx +++ b/interface/web/src/Settings.jsx @@ -8,7 +8,7 @@ const DEFAULTS = { num_ctx: 0, // context window in tokens; 0 = model default rag_top_k: 3, // document chunks injected into chat rag_min_score: 0.6, // min cosine similarity for a chunk to count - allow_action_tools: false, // consent gate for web/fetch/memory-write tools + action_tool_policy: "off", // off | ask (per-call approval) | allow system_prompt: "", timeout: 120, gpu_offload: -1, // -1 = Auto; 0–100 = percent of layers forced onto the GPU @@ -243,18 +243,20 @@ export function Settings() {
- -
- Lets playbooks run tools that act: web search, fetching a URL, and writing - to memory. Off by default — a playbook can list them, but they only fire - when this is on. + + +
+ Controls tools that act (vs. read-only). A playbook must also list the tool. + {form.action_tool_policy === "ask" && " You'll get an Approve/Deny prompt in chat."} + {form.action_tool_policy === "allow" && " Actions run without confirmation."}
diff --git a/synapse/chat.py b/synapse/chat.py index f14a241..8e43c29 100644 --- a/synapse/chat.py +++ b/synapse/chat.py @@ -131,18 +131,28 @@ async def _normalize_to_async_generator(maybe_iterable) -> AsyncGenerator[str, N yield str(item) -async def _run_tool_loop(manager, messages, model, tool_schemas, temperature, num_gpu): - """Let the model call read-only tools before the final streamed answer. +# Per-call approval waiters, keyed by conversation_id. The chat stream stays open +# and the loop awaits the Event; POST /chat/approve fills decisions and sets it. +# ponytail: in-memory, single-process — fine for a local single-user app; needs a +# shared store only if this ever runs multi-worker. +pending_approvals: Dict[str, Dict[str, Any]] = {} +_APPROVAL_TIMEOUT = 300 # seconds; a timeout is treated as "deny all" + + +async def _run_tool_loop(manager, messages, model, tool_schemas, temperature, num_gpu, + conversation_id="", policy="allow"): + """Let the model call tools before the final streamed answer. Mutates `messages` IN PLACE, appending the assistant tool-call turns and - their `role:"tool"` results, and yields a `__status__` sentinel before - each tool runs (surfaced to the UI as a "running tool" indicator). - Non-streamed — tool calls arrive as whole messages. Degrades to an untouched - `messages` if the model can't do tool calling. + their `role:"tool"` results, and yields `__status__` sentinels. + When policy == "ask" and a turn contains action tools, yields an + `__approve__` sentinel and awaits the user's decision (via + `pending_approvals`) before running them; declined actions get a "denied" + result the model can react to. Degrades to untouched `messages` if the model + can't do tool calling. ponytail: the turn that finally returns content is thrown away and the answer - is re-generated by the streaming turn (one wasted call). Simpler than - streaming a maybe-already-complete message; revisit if latency matters. + is re-generated by the streaming turn (one wasted call). """ for _ in range(MAX_TOOL_STEPS): msg = await manager.chat( @@ -155,9 +165,32 @@ async def _run_tool_loop(manager, messages, model, tool_schemas, temperature, nu if not calls: break messages.append(msg) + + # If any action tool needs per-call approval, pause and wait for the user. + decisions = None + action_calls = [c for c in calls if _tools.is_action(c.get("function", {}).get("name", ""))] + if policy == "ask" and action_calls: + event = asyncio.Event() + pending_approvals[conversation_id] = {"event": event, "decisions": {}} + yield "__approve__" + _json.dumps([ + {"name": c.get("function", {}).get("name", ""), + "arguments": c.get("function", {}).get("arguments")} + for c in action_calls + ]) + try: + await asyncio.wait_for(event.wait(), timeout=_APPROVAL_TIMEOUT) + decisions = pending_approvals[conversation_id]["decisions"] + except asyncio.TimeoutError: + decisions = {} # no answer in time -> deny all actions + finally: + pending_approvals.pop(conversation_id, None) + for c in calls: fn = c.get("function", {}) name = fn.get("name", "") + if decisions is not None and _tools.is_action(name) and not decisions.get(name, False): + messages.append({"role": "tool", "content": _json.dumps({"denied": f"user declined {name}"})}) + continue yield f"__status__{name}" result = await _tools.dispatch(name, fn.get("arguments")) messages.append({"role": "tool", "content": result}) @@ -195,12 +228,16 @@ async def stream_chat_response( user_msg["images"] = images messages.append(user_msg) - # Tool-using playbooks: run read-only tool calls, then stream the final answer - # with their results already in the messages array. + # Tool-using playbooks: run tool calls, then stream the final answer with + # their results already in the messages array. tool_schemas = metadata.get("tools") if tool_schemas: try: - async for status in _run_tool_loop(manager, messages, model, tool_schemas, temperature, num_gpu): + async for status in _run_tool_loop( + manager, messages, model, tool_schemas, temperature, num_gpu, + conversation_id=metadata.get("conversation_id", ""), + policy=metadata.get("action_tool_policy", "allow"), + ): yield status except Exception: _logger.exception("tool loop failed; streaming without tools") diff --git a/synapse/main.py b/synapse/main.py index e314cc2..5f6bb22 100644 --- a/synapse/main.py +++ b/synapse/main.py @@ -17,6 +17,7 @@ from fastapi.responses import StreamingResponse, FileResponse from .nexus_config import settings, VERSION, DEFAULT_CHAT_MODEL from .chat import generate_chat_response, stream_chat_response, _synapse_trace +from . import chat as _chat from .ollama_manager import initialize_ollama, initialize_ollama_async, get_ollama_manager from .playbook_manager import PlaybookManager from . import tools as _tools @@ -370,15 +371,19 @@ async def chat_stream_endpoint(payload: Dict[str, Any]): metadata["images"] = images # Tool-using playbook: advertise the active playbook's allowlisted tools. - # Action tools (web/fetch/write) are withheld unless the consent gate is on. + # Action tools follow action_tool_policy: off (withheld) / ask (per-call + # approval, handled in the tool loop) / allow (run freely). + _policy = app_settings.get("action_tool_policy", "off") if _main_pb and getattr(_main_pb, "tools", None): - allow_actions = bool(app_settings.get("allow_action_tools", False)) + allow_actions = _policy != "off" schemas = _tools.schemas_for(_main_pb.tools, allow_actions) if schemas: metadata["tools"] = schemas + metadata["action_tool_policy"] = _policy + metadata["conversation_id"] = conversation_id _granted = [t for t in _main_pb.tools if not _tools.is_action(t) or allow_actions] _withheld = [t for t in _main_pb.tools if _tools.is_action(t) and not allow_actions] - _synapse_trace(f" TOOLS : {', '.join(_granted)}\n") + _synapse_trace(f" TOOLS : {', '.join(_granted)} [actions: {_policy}]\n") if _withheld: _synapse_trace(f" WITHHELD: {', '.join(_withheld)} (action tools off)\n") @@ -412,6 +417,10 @@ async def chat_stream_endpoint(payload: Dict[str, Any]): if chunk.startswith("__status__"): yield f"event: status\ndata: {_json.dumps({'tool': chunk[10:]})}\n\n" continue + if chunk.startswith("__approve__"): + # Loop is paused awaiting the user; forward the pending actions. + yield f"event: tool_request\ndata: {chunk[11:]}\n\n" + continue response_chunks.append(chunk) yield f"data: {_json.dumps(chunk)}\n\n" except _asyncio.TimeoutError: @@ -480,6 +489,20 @@ async def chat_stream_endpoint(payload: Dict[str, Any]): except Exception as e: raise HTTPException(status_code=500, detail=str(e)) + +@app.post("/chat/approve") +async def chat_approve(payload: Dict[str, Any] = Body(...)): + """Resolve a pending per-call tool approval. `decisions` maps tool name -> + bool; the awaiting chat stream resumes and runs the approved actions.""" + conversation_id = payload.get("conversation_id") or "" + decisions = payload.get("decisions") or {} + waiter = _chat.pending_approvals.get(conversation_id) + if not waiter: + raise HTTPException(status_code=404, detail="no pending approval for this conversation") + waiter["decisions"] = {k: bool(v) for k, v in decisions.items()} + waiter["event"].set() + return {"status": "resumed"} + # ------------------------- # Settings # ------------------------- diff --git a/synapse/memory/store.py b/synapse/memory/store.py index b0b831d..668da32 100644 --- a/synapse/memory/store.py +++ b/synapse/memory/store.py @@ -998,9 +998,11 @@ class PersistentMemoryStore: "rag_min_score": 0.6, # Active project/workspace; "" = all documents (unscoped). "active_project": "", - # Consent gate for tools that act (web_search/fetch_url/remember). Off by - # default: a playbook can list them, but they only run when this is on. - "allow_action_tools": False, + # Consent policy for tools that act (web_search/fetch_url/remember): + # "off" — withheld from the model entirely (default) + # "ask" — offered, but each call waits for per-call user approval + # "allow" — offered and run freely + "action_tool_policy": "off", "system_prompt": "", "timeout": 120, # How long Ollama keeps the model resident in VRAM between messages. diff --git a/tests/test_tools.py b/tests/test_tools.py index 968e690..932ede3 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -41,6 +41,56 @@ def test_action_tools_registered(): assert names == ["web_search", "remember"] +class _ActionManager: + """Returns a `remember` (action) tool_call once, then plain content.""" + def __init__(self): + self.n = 0 + + async def chat(self, **_): + self.n += 1 + if self.n == 1: + return {"role": "assistant", + "tool_calls": [{"function": {"name": "remember", "arguments": {"text": "x"}}}]} + return {"role": "assistant", "content": "done"} + + +def _drive_with_decision(decision, monkeypatch): + from synapse import chat as chatmod + + async def fake_dispatch(name, args): + return "saved-ok" + monkeypatch.setattr(tools, "dispatch", fake_dispatch) + + async def run(): + messages = [{"role": "user", "content": "remember x"}] + gen = chatmod._run_tool_loop(_ActionManager(), messages, "m", [{}], None, None, + conversation_id="conv", policy="ask") + statuses = [] + async for s in gen: + statuses.append(s) + if s.startswith("__approve__"): + w = chatmod.pending_approvals["conv"] + w["decisions"] = {"remember": decision} + w["event"].set() + return statuses, messages + + return asyncio.run(run()) + + +def test_ask_policy_pauses_then_runs_on_approve(monkeypatch): + statuses, messages = _drive_with_decision(True, monkeypatch) + assert any(s.startswith("__approve__") for s in statuses) # paused for approval + assert "__status__remember" in statuses # approved -> ran + assert any(m["role"] == "tool" and "saved-ok" in m["content"] for m in messages) + + +def test_ask_policy_skips_on_deny(monkeypatch): + statuses, messages = _drive_with_decision(False, monkeypatch) + assert any(s.startswith("__approve__") for s in statuses) + assert "__status__remember" not in statuses # denied -> never ran + assert any(m["role"] == "tool" and "declined" in m["content"] for m in messages) + + def test_action_tools_gated_by_consent(): allow = ["search_memory", "web_search", "remember", "fetch_url"] on = [s["function"]["name"] for s in tools.schemas_for(allow, allow_actions=True)]