feat(tools): per-call approval for action tools

3-way action_tool_policy (off/ask/allow). In "ask", the chat stream stays
open and the tool loop awaits approval: emits event:tool_request, the UI
shows Approve/Deny, POST /chat/approve resumes the same stream. Declined
actions return a denied result; a timeout denies.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jon
2026-07-23 14:39:00 -05:00
co-authored by Claude Opus 4.8
parent 492020b547
commit 52b3c5c3f0
6 changed files with 190 additions and 30 deletions
+50
View File
@@ -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)]