Mike-Solar
e0db5ab2cf
- Core types: Rational time, EntityId, ActionBatch, PngBytes (no Pillow dep) - Config: ~/.quercus/config.toml with 0600 perms, env var override for CI - LLMProvider: Claude (anthropic SDK) and OpenAI-compatible endpoint (covers custom gateways and llama.cpp server mode); graceful degradation without API keys - Tool schemas: 26 curated tools with JSON Schema and read_only flags - AgentLoop: sync dialogue loop with confirmation gate (read-only batches bypass, rejected mutations are reported back to the LLM), PNG frame feedback for multimodal providers - Session log: JSONL recording with frame sha256 hashes, replayable - HostAdapter ABC (signatures only, aligned with OPP/1 method families) - Mock LLM provider (scripted/replay/record) for network-free CI Tests: 69 passed via uv run pytest (no network access)
75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
"""会话日志：JSONL 写读往返、帧哈希记录（不落图片本体）。"""
|
|
import hashlib
|
|
import json
|
|
|
|
from quercus_core.providers.base import AssistantTurn, ToolCall
|
|
from quercus_core.session.log import SessionLog, frame_sha256, replay
|
|
from quercus_core.types import Action, ActionBatch, PngBytes, ToolResult
|
|
|
|
PNG = PngBytes(b"\x89PNG\r\n\x1a\n" + bytes(range(64)))
|
|
|
|
|
|
def test_write_read_roundtrip(tmp_path):
|
|
path = tmp_path / "s1.jsonl"
|
|
log = SessionLog(path, session_id="sess-1")
|
|
log.log_user("你好")
|
|
log.log_assistant(
|
|
AssistantTurn(
|
|
text="",
|
|
tool_calls=(ToolCall(id="c1", name="get_frame", arguments={"time": {"num": 1, "den": 1}}),),
|
|
)
|
|
)
|
|
log.log_tool_result(ToolResult(tool="get_frame", ok=True, summary="帧", images=(PNG,)))
|
|
log.log_batch_decision(
|
|
ActionBatch(
|
|
label="AI 动作：get_frame",
|
|
actions=[Action(tool="get_frame", params={"time": {"num": 1, "den": 1}}, call_id="c1")],
|
|
id="b1",
|
|
),
|
|
approved=True,
|
|
)
|
|
|
|
events = replay(path)
|
|
assert len(events) == 4
|
|
assert events[0].kind == "user_message"
|
|
assert events[0].payload["text"] == "你好"
|
|
assert events[0].session_id == "sess-1"
|
|
assert events[1].kind == "assistant_turn"
|
|
assert events[1].payload["tool_calls"][0]["name"] == "get_frame"
|
|
assert events[2].kind == "tool_result"
|
|
assert events[3].kind == "batch_decision"
|
|
assert events[3].payload["approved"] is True
|
|
|
|
|
|
def test_frame_hash_recorded_not_bytes(tmp_path):
|
|
path = tmp_path / "s2.jsonl"
|
|
log = SessionLog(path, session_id="s")
|
|
log.log_tool_result(ToolResult(tool="get_frame", ok=True, summary="f", images=(PNG,)))
|
|
|
|
events = replay(path)
|
|
ev = events[0]
|
|
assert ev.kind == "tool_result"
|
|
assert ev.payload["image_hashes"] == [hashlib.sha256(PNG).hexdigest()]
|
|
# 日志文件里绝不出现图片本体字节
|
|
assert PNG not in path.read_bytes()
|
|
|
|
|
|
def test_jsonl_each_line_is_valid_json(tmp_path):
|
|
path = tmp_path / "s3.jsonl"
|
|
log = SessionLog(path)
|
|
log.log_user("x")
|
|
log.log_tool_result(ToolResult(tool="seek", ok=True, summary="ok"))
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
json.loads(line)
|
|
|
|
|
|
def test_frame_sha256_helper(fake_png):
|
|
assert frame_sha256(fake_png) == hashlib.sha256(fake_png).hexdigest()
|
|
|
|
|
|
def test_create_default_uses_home(tmp_home):
|
|
log = SessionLog.create_default()
|
|
assert log.path.parent == tmp_home / ".quercus" / "sessions"
|
|
assert log.path.suffix == ".jsonl"
|
|
assert log.path.name # <id>.jsonl
|