jonandClaude Opus 4.8
f509034fdd
- Tool-using playbooks: read-only tool registry (search_memory, search_history, search_documents, list_models, get_time), per-playbook allowlist, agentic loop, and a "running tool" status indicator. - Document ingest / RAG: documents table + chunker + embed/cosine retrieval reusing the existing stack; Documents page (upload/paste, viewer, delete); top-k chunks injected into the system prompt. - Vision chat: attach images, base64 into /api/chat. - Voice I/O: Web Speech dictation + read-aloud (browser-native, no backend). - Chat controls: stop, regenerate, edit-and-resend; num_ctx knob in Settings. - Per-playbook model override. - History polish: per-conversation + bulk ShareGPT export. - Faster ncp start: UI up first, Ollama warms in the background, with a per-phase timing readout. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
"""Tool-using playbook loop — the read-only MVP.
|
|
|
|
Run from nexus-core/ with the Promethean venv active: pytest -q
|
|
|
|
Hermetic: a fake manager stands in for Ollama, so no network/model is needed.
|
|
Guards the two pieces that would silently break the feature: the allowlist
|
|
filter and the tool-call loop's terminate-on-content behaviour.
|
|
"""
|
|
import asyncio
|
|
|
|
from synapse import tools
|
|
from synapse.chat import _run_tool_loop
|
|
|
|
|
|
def test_schemas_for_drops_unknown_names():
|
|
schemas = tools.schemas_for(["search_memory", "not_a_tool"])
|
|
names = [s["function"]["name"] for s in schemas]
|
|
assert names == ["search_memory"]
|
|
assert tools.schemas_for([]) == []
|
|
|
|
|
|
async def _drain(gen):
|
|
return [s async for s in gen]
|
|
|
|
|
|
class _FakeManager:
|
|
"""Returns a tool_call on the first chat() call, plain content after."""
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
async def chat(self, **_):
|
|
self.calls += 1
|
|
if self.calls == 1:
|
|
return {
|
|
"role": "assistant",
|
|
"tool_calls": [
|
|
{"function": {"name": "search_memory", "arguments": {"query": "gpu"}}}
|
|
],
|
|
}
|
|
return {"role": "assistant", "content": "here is the answer"}
|
|
|
|
|
|
def test_tool_loop_runs_tool_then_stops(monkeypatch):
|
|
async def fake_dispatch(name, args):
|
|
assert name == "search_memory"
|
|
assert args == {"query": "gpu"}
|
|
return '[{"section": "GPU", "text": "Vega 20 4GB"}]'
|
|
|
|
monkeypatch.setattr(tools, "dispatch", fake_dispatch)
|
|
|
|
messages = [{"role": "user", "content": "what gpu do i have?"}]
|
|
schemas = tools.schemas_for(["search_memory"])
|
|
statuses = asyncio.run(_drain(
|
|
_run_tool_loop(_FakeManager(), messages, "m", schemas, None, None)
|
|
))
|
|
|
|
# one status sentinel per tool run
|
|
assert statuses == ["__status__search_memory"]
|
|
# messages mutated in place: user -> assistant(tool_calls) -> tool(result);
|
|
# the final content turn is NOT appended (the streaming turn regenerates it).
|
|
assert [m["role"] for m in messages] == ["user", "assistant", "tool"]
|
|
assert "Vega 20" in messages[-1]["content"]
|
|
|
|
|
|
def test_tool_loop_degrades_when_model_returns_no_dict():
|
|
class _NoToolManager:
|
|
async def chat(self, **_):
|
|
return None # model can't do tools / errored
|
|
|
|
messages = [{"role": "user", "content": "hi"}]
|
|
before = list(messages)
|
|
statuses = asyncio.run(_drain(_run_tool_loop(_NoToolManager(), messages, "m", [{}], None, None)))
|
|
assert statuses == [] # no tool ran
|
|
assert messages == before # untouched -> falls back to a plain stream
|
GitNexus · part of the Nexus stack