"""执行、存储和人机交互的失败路径回归；不访问模型网络。""" import asyncio import json from types import SimpleNamespace import httpx import pytest from langchain_core.messages import AIMessage, AIMessageChunk import config from runtime import concurrency from api.interrupts import pack from db import billing_store, chat_store, db_conn, run_store, schedule_store @pytest.fixture(autouse=True) def isolation(monkeypatch, tmp_path): monkeypatch.setenv('MAIN_FAN_ENABLED', '0') monkeypatch.setattr(config, "ROOT", tmp_path) (tmp_path / "skills").mkdir() concurrency.reset_for_tests() yield concurrency.reset_for_tests() class FakeAgent: def __init__(self, interrupt=None, wait=None): self.interrupt = interrupt self.wait = wait self.started = asyncio.Event() self.inputs = [] async def aget_state(self, config): return SimpleNamespace(tasks=[]) async def astream(self, graph_input, **kwargs): self.inputs.append((graph_input, kwargs)) self.started.set() yield {"type": "messages", "data": (AIMessageChunk(content="部分回答", usage_metadata={"input_tokens": 12, "output_tokens": 3, "total_tokens": 15}), {})} if self.wait: await self.wait.wait() if self.interrupt: yield {"type": "updates", "data": {"__interrupt__": [self.interrupt]}} def test_sqlite_rollback_and_nested_savepoint(): adapter = db_conn._SqliteAdapter(":memory:") adapter.execute("CREATE TABLE probe (id INTEGER)") with pytest.raises(RuntimeError): with adapter.transaction(): adapter.execute("INSERT INTO probe VALUES (1)") adapter.execute_many("INSERT INTO probe VALUES (?)", [(2,), (3,)]) raise RuntimeError() assert adapter.query("SELECT * FROM probe") == [] with adapter.transaction(): adapter.execute("INSERT INTO probe VALUES (4)") with pytest.raises(ValueError): with adapter.transaction(): adapter.execute("INSERT INTO probe VALUES (5)") raise ValueError() adapter.execute("INSERT INTO probe VALUES (6)") assert adapter.query("SELECT id FROM probe") == [{"id": 4}, {"id": 6}] def test_billing_transaction_rolls_back_quota_failure(monkeypatch): def fail(*args): raise RuntimeError("quota database failure") monkeypatch.setattr(billing_store, "_accumulate_quota", fail) with pytest.raises(RuntimeError): billing_store.record_usage("u", "s", "chat", config.MODEL, 10, 2, 0, db_conn.day_key()) assert billing_store.recent_usage("u") == [] def test_failed_settlement_survives_new_meter_and_deduplicates(monkeypatch, pricing_ready): from billing.meter import UsageMeter, retry_pending real = billing_store.record_usage meter = UsageMeter("u", "s", "chat", config.MODEL) meter.add(10, 2) def fail(**kwargs): raise RuntimeError("temporary outage") monkeypatch.setattr(billing_store, "record_usage", fail) assert not meter.settle()["settled"] assert not meter.settled assert db_conn.scalar("SELECT COUNT(*) FROM billing_pending") == 1 monkeypatch.setattr(billing_store, "record_usage", real) assert retry_pending() == 1 assert meter.settle()["settled"] # old request's retry cannot double-charge assert len(billing_store.recent_usage("u")) == 1 assert billing_store.get_quota("u", db_conn.day_key())["used_tokens"] == 12 assert db_conn.scalar("SELECT COUNT(*) FROM billing_pending") == 0 @pytest.mark.asyncio async def test_http_concurrency_returns_429(monkeypatch, pricing_ready): from main import app monkeypatch.setattr(config, "MAX_CONCURRENT_TASKS", 0) async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as c: response = await c.post("/api/chat", data={"message": "hello"}) assert response.status_code == 429 assert response.headers["retry-after"] assert not concurrency._sessions @pytest.mark.asyncio async def test_cross_user_message_and_schedule_access(pricing_ready): from main import app session = chat_store.create_session("alice") message_id = chat_store.save_message(session["id"], "user", "secret") schedule = schedule_store.create_schedule("alice", "private", "0 9 * * *", "private") async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as c: response = await c.delete(f"/api/messages/{message_id}", params={"user_name": "bob"}) assert response.status_code == 404 for method, suffix in [("GET", ""), ("DELETE", ""), ("POST", "/run"), ("GET", "/runs")]: response = await c.request(method, f"/api/schedules/{schedule['id']}{suffix}", params={"user_name": "bob"}) assert response.status_code == 404 assert chat_store.count_messages(session["id"]) == 1 assert schedule_store.get_schedule(schedule["id"]) @pytest.mark.asyncio async def test_same_session_concurrent_request_rejected(monkeypatch, pricing_ready): from main import app agent = FakeAgent(wait=asyncio.Event()) monkeypatch.setattr("agents.agent_registry.get_agent", lambda: agent) session = chat_store.create_session("default") async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as c: first = asyncio.create_task(c.post("/api/chat", data={"message": "one", "session_id": session["id"]})) await asyncio.wait_for(agent.started.wait(), 2) second = await c.post("/api/chat", data={"message": "two", "session_id": session["id"]}) assert second.status_code == 409 agent.wait.set() assert (await first).status_code == 200 assert concurrency.stats()["global_active"] == 0 assert len(run_store.list_runs(session["id"])) == 1 assert "two" not in [m["content"] for m in chat_store.get_messages(session["id"])] @pytest.mark.asyncio @pytest.mark.parametrize("cancel", [False, True]) async def test_stream_close_or_cancel_settles_and_releases(cancel, pricing_ready): from runtime.execution import admit, build_execution session = chat_store.create_session("default") agent = FakeAgent(wait=asyncio.Event()) admission = await admit("default", session["id"]) execution = await build_execution(admission, agent=agent, session_id=session["id"], thread_id=session["id"], user_name="default", message="test", config={"configurable": {"thread_id": session["id"]}}) generator = execution.events() while True: frame = await anext(generator) if frame["type"] == "token": break if cancel: pending = asyncio.create_task(anext(generator)) await asyncio.sleep(0) pending.cancel() with pytest.raises(asyncio.CancelledError): await pending await generator.aclose() assert not concurrency._sessions assert concurrency.stats()["global_active"] == 0 assert billing_store.get_quota("default", db_conn.day_key())["used_tokens"] == 15 assert run_store.list_runs(session["id"])[0]["status"] == "cancelled" assert "不完整" in chat_store.get_messages(session["id"])[-1]["content"] @pytest.mark.asyncio async def test_schedule_uses_shared_runtime_and_pauses(monkeypatch, pricing_ready): from scheduler.runner import run_schedule item = SimpleNamespace(id="approval-id", value={"action_requests": [{"name": "calculate", "args": {"expression": "1+1"}}]}) agent = FakeAgent(interrupt=item) monkeypatch.setattr("agents.agent_registry.get_agent", lambda: agent) schedule = schedule_store.create_schedule("u", "plan", "0 9 * * *", "question", "plan") result = await run_schedule(schedule["id"]) assert result["status"] == "waiting_input" assert schedule_store.latest_run(schedule["id"])["status"] == "waiting_input" assert run_store.list_runs(result["session_id"])[0]["status"] == "waiting_input" assert "request_plan_approval" in agent.inputs[0][0]["messages"][-1].content assert billing_store.get_quota("u", db_conn.day_key())["used_tokens"] == 15 assert concurrency.stats()["global_active"] == 0 @pytest.mark.asyncio async def test_resume_and_schedule_enforce_quota(pricing_ready): from main import app from scheduler.runner import run_schedule from billing.guard import QuotaExceededError session = chat_store.create_session("u") billing_store.set_quota_limit("u", db_conn.day_key(), 1) billing_store.record_usage("u", session["id"], "chat", config.MODEL, 1, 0, 0, db_conn.day_key()) schedule = schedule_store.create_schedule("u", "x", "0 9 * * *", "x") with pytest.raises(QuotaExceededError): await run_schedule(schedule["id"]) async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as c: response = await c.post("/api/chat/resume", json={"session_id": session["id"], "user_name": "u", "approved": True}) assert response.status_code == 429 assert not concurrency._sessions def test_workspace_isolation_and_skills(monkeypatch, tmp_path): from agents.workspace import backend_for_runtime, session_root (tmp_path / "skills" / "guide.md").write_text("shared guide") def backend(user, session): return backend_for_runtime(SimpleNamespace(config={"configurable": {"workspace_user": user, "workspace_session": session}})) first, second, other_user = backend("alice", "one"), backend("alice", "two"), backend("bob", "one") first.write("/workspace/private.txt", "secret") assert "secret" in str(first.read("/workspace/private.txt")) assert second.read("/workspace/private.txt").error assert other_user.read("/workspace/private.txt").error assert "shared guide" in str(first.read("/skills/guide.md")) assert session_root("../escape", "../../other").is_relative_to(tmp_path / "data" / "workspaces") with pytest.raises(ValueError): backend_for_runtime(SimpleNamespace(config={})) def test_batch_requires_current_interrupt_and_shows_all(): from api.interrupts import interrupt_frame, resume_value from api.models import ResumeRequest from api.sse_events import EventWriter from fastapi import HTTPException item = SimpleNamespace(id="a", value={"action_requests": [{"name": "one", "args": {}}, {"name": "two", "args": {}}]}) frame = json.loads(interrupt_frame(EventWriter("s"), pack(item))[6:]) assert len(frame["tool_approval"]["actions"]) == 2 state = SimpleNamespace(tasks=[SimpleNamespace(interrupts=[item])]) payload = ResumeRequest(session_id="s", approved=True, interrupt_token=frame["tool_approval"]["interrupt_token"]) assert len(resume_value(payload, state)["a"]["decisions"]) == 2 item.id = "newer" with pytest.raises(HTTPException) as error: resume_value(payload, state) assert error.value.status_code == 409 def test_declared_side_effect_tools_require_approval(): from tools.registry import register from agents.agent_factory import _build_interrupt_on @register(group="test", requires_approval=True) def external_write() -> str: """External change.""" return "done" assert _build_interrupt_on()["external_write"] is True def test_unknown_model_is_not_silently_priced(monkeypatch): from billing import pricing, guard monkeypatch.setattr(config, "MODEL", "unpriced-test-model") pricing.seed_default_pricing() assert billing_store.get_pricing(config.MODEL) is None with pytest.raises(guard.BillingConfigError): guard.precheck("u") def test_startup_marks_stale_runs_interrupted(): run_store.start("stale", "s", "u", "chat") run_store.recover_interrupted() assert run_store.list_runs("s")[0]["status"] == "interrupted" from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult from pydantic import Field class ScriptedModel(BaseChatModel): """真实图引擎，确定性模型；覆盖 middleware、工具和 checkpoint。""" responses: list[AIMessage] = Field(default_factory=list) calls: int = 0 @property def _llm_type(self): return "offline-scripted" def bind_tools(self, tools, **kwargs): return self def _next(self): response = self.responses[self.calls] self.calls += 1 return response def _generate(self, messages, stop=None, run_manager=None, **kwargs): return ChatResult(generations=[ChatGeneration(message=self._next())]) def _stream(self, messages, stop=None, run_manager=None, **kwargs): message = self._next() yield ChatGenerationChunk(message=AIMessageChunk(content=message.content, tool_calls=message.tool_calls, usage_metadata={"input_tokens": 20, "output_tokens": 5, "total_tokens": 25})) @pytest.mark.asyncio @pytest.mark.parametrize("tool_name,args", [ ("ask_user", {"questions": ["请选择区域"]}), ("request_plan_approval", {"plan": "读取文件后生成报告"}), ]) async def test_real_graph_interrupt_refresh_resume(monkeypatch, pricing_ready, tool_name, args): from agents import agent_factory, agent_memory from main import app model = ScriptedModel(responses=[ AIMessage(content="", tool_calls=[{"id": "call1", "name": tool_name, "args": args}]), AIMessage(content="已根据你的确认完成。"), ]) monkeypatch.setattr("agents.model_factory.get_model", lambda: model) graph = agent_factory.build_agent() monkeypatch.setattr("agents.agent_registry.get_agent", lambda: graph) async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as c: first = await c.post("/api/chat", data={"message": "test", "user_name": "u"}) assert first.status_code == 200, first.text events = [json.loads(line[6:]) for line in first.text.splitlines() if line.startswith("data: ")] assert not [e for e in events if e["type"] == "error"], events interrupted = next(e for e in events if e["type"] == "interrupt") sid = first.headers["x-session-id"] token = interrupted.get("interrupt_token") or interrupted["tool_approval"]["interrupt_token"] refreshed = (await c.get("/api/chat/interrupt", params={"session_id": sid, "user_name": "u"})).json()["interrupts"][0] assert (refreshed.get("interrupt_token") or refreshed["tool_approval"]["interrupt_token"]) == token assert events[-1]["waiting_input"] response = await c.post("/api/chat/resume", json={"session_id": sid, "user_name": "u", "interrupt_token": token, "approved": True, "answers": ["一区"]}) assert response.status_code == 200, response.text assert "已根据你的确认完成" in response.text assert run_store.list_runs(sid)[0]["status"] == "success" repeated = await c.post("/api/chat/resume", json={"session_id": sid, "user_name": "u", "interrupt_token": token, "approved": True}) assert repeated.status_code == 409 await agent_memory.delete_thread(sid) @pytest.mark.asyncio async def test_real_graph_workspace_write(monkeypatch, pricing_ready): from agents import agent_factory, agent_memory from agents.workspace import session_root from main import app model = ScriptedModel(responses=[ AIMessage(content="", tool_calls=[{"id": "w1", "name": "write_file", "args": {"file_path": "/workspace/result.txt", "content": "隔离的产物"}}]), AIMessage(content="已写入"), ]) monkeypatch.setattr("agents.model_factory.get_model", lambda: model) graph = agent_factory.build_agent() monkeypatch.setattr("agents.agent_registry.get_agent", lambda: graph) async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as c: result = await c.post("/api/chat", data={"message": "write", "user_name": "u"}) sid = result.headers["x-session-id"] assert (session_root("u", sid) / "workspace/result.txt").read_text() == "隔离的产物", result.text await agent_memory.delete_thread(sid) @pytest.mark.asyncio async def test_request_id_does_not_execute_twice(monkeypatch, pricing_ready): from main import app agent = FakeAgent() monkeypatch.setattr("agents.agent_registry.get_agent", lambda: agent) async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as c: body = {"message": "once", "request_id": "client-generated-key"} first = await c.post("/api/chat", data=body) second = await c.post("/api/chat", data=body) assert first.status_code == 200 assert second.status_code == 409 assert second.json()["detail"]["run_id"] == first.headers["x-run-id"] assert len(agent.inputs) == 1 assert len(chat_store.list_sessions("default")) == 1 @pytest.mark.asyncio async def test_real_subagent_shares_only_parent_workspace(monkeypatch, pricing_ready): from agents import agent_factory, agent_memory from agents.workspace import session_root from main import app model = ScriptedModel(responses=[ AIMessage(content="", tool_calls=[{"id": "t1", "name": "task", "args": {"subagent_type": "demo-searcher", "description": "write a report"}}]), AIMessage(content="", tool_calls=[{"id": "w2", "name": "write_file", "args": {"file_path": "/workspace/sub.txt", "content": "子任务产物"}}]), AIMessage(content="子任务完成"), AIMessage(content="汇总完成"), ]) monkeypatch.setattr("agents.model_factory.get_model", lambda: model) graph = agent_factory.build_agent() monkeypatch.setattr("agents.agent_registry.get_agent", lambda: graph) async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as c: result = await c.post("/api/chat", data={"message": "delegate", "user_name": "u"}) sid = result.headers["x-session-id"] assert (session_root("u", sid) / "workspace/sub.txt").read_text() == "子任务产物", result.text await agent_memory.delete_thread(sid) @pytest.mark.asyncio async def test_schedule_timeout_is_failed_and_billed(monkeypatch, pricing_ready): from scheduler.runner import run_schedule agent = FakeAgent(wait=asyncio.Event()) monkeypatch.setattr("agents.agent_registry.get_agent", lambda: agent) monkeypatch.setattr(config, "TASK_WALL_TIMEOUT", 0.05) schedule = schedule_store.create_schedule("u", "slow", "0 9 * * *", "wait") result = await run_schedule(schedule["id"]) assert result["status"] == "failed" assert schedule_store.latest_run(schedule["id"])["status"] == "failed" assert billing_store.get_quota("u", db_conn.day_key())["used_tokens"] == 15 assert concurrency.stats()["global_active"] == 0 @pytest.mark.asyncio async def test_upload_limit_removes_partial_file(monkeypatch, pricing_ready): from fastapi import UploadFile, HTTPException from io import BytesIO from api.chat_routes import _save_upload from agents.workspace import session_root monkeypatch.setattr("api.chat_routes.MAX_UPLOAD_SIZE", 5) with pytest.raises(HTTPException) as error: await _save_upload(UploadFile(filename="data.txt", file=BytesIO(b"too long")), "u", "s") assert error.value.status_code == 413 assert not list((session_root("u", "s") / "workspace/uploads").iterdir()) def test_child_and_summary_usage_count_without_leaking_text_or_context(): from api.sse_core import ChatStream, StreamRequest stream = ChatStream(StreamRequest(agent=None, session_id="s", thread_id="s", config={}, message="q")) main = AIMessageChunk(content="主回答", usage_metadata={"input_tokens": 100, "output_tokens": 10, "total_tokens": 110}) child = AIMessageChunk(content="子任务内部材料", usage_metadata={"input_tokens": 10, "output_tokens": 2, "total_tokens": 12}) summary = AIMessageChunk(content="内部摘要", usage_metadata={"input_tokens": 20, "output_tokens": 3, "total_tokens": 23}) assert stream._handle_messages((main, {}), ()) assert stream._handle_messages((child, {}), ("task:1",)) == [] assert stream._handle_messages((summary, {"lc_source": "summarization"}), ()) == [] assert stream.answer_parts == ["主回答"] assert stream.meter.total_tokens == 145 assert stream.meter.context_tokens == 100 assert not stream.meter.compacted @pytest.mark.asyncio async def test_scheduler_shutdown_waits_for_cleanup(monkeypatch): from scheduler import scheduler entered = asyncio.Event() cleaned = False calls = [] async def work(*args, **kwargs): nonlocal cleaned entered.set() try: await asyncio.Event().wait() finally: await asyncio.sleep(0) cleaned = True class Scheduler: def pause(self): calls.append("pause") def shutdown(self, **kwargs): assert cleaned calls.append("shutdown") monkeypatch.setattr(scheduler, "run_schedule", work) monkeypatch.setattr(scheduler, "_scheduler", Scheduler()) task = asyncio.create_task(scheduler._invoke("s")) await entered.wait() await scheduler.shutdown() assert task.done() and cleaned assert calls == ["pause", "shutdown"] assert not scheduler._active @pytest.mark.asyncio async def test_admission_revalidates_deleted_session(pricing_ready): from agents.execution import admit session = chat_store.create_session("u") chat_store.delete_session(session["id"]) with pytest.raises(concurrency.SessionMissingError): await admit("u", session["id"]) assert concurrency.stats()["global_active"] == 0 assert not concurrency._sessions @pytest.mark.asyncio async def test_cancellation_during_persistence_does_not_duplicate_answer(monkeypatch, pricing_ready): import threading from runtime.execution import admit, build_execution session = chat_store.create_session("u") entered, release = threading.Event(), threading.Event() original = chat_store.save_message def slow_save(session_id, role, content, **kwargs): if role == "assistant": entered.set() assert release.wait(3) return original(session_id, role, content, **kwargs) monkeypatch.setattr(chat_store, "save_message", slow_save) admission = await admit("u", session["id"]) execution = await build_execution(admission, agent=FakeAgent(), session_id=session["id"], thread_id=session["id"], user_name="u", message="question", config={"configurable": {"thread_id": session["id"]}}) async def consume(): async for _ in execution.events(): pass task = asyncio.create_task(consume()) assert await asyncio.to_thread(entered.wait, 3) task.cancel() await asyncio.sleep(0) release.set() with pytest.raises(asyncio.CancelledError): await task assert len([m for m in chat_store.get_messages(session["id"]) if m["msg_role"] == "assistant"]) == 1 assert len(billing_store.recent_usage("u")) == 1 assert concurrency.stats()["global_active"] == 0 @pytest.mark.asyncio @pytest.mark.parametrize("delegated", [False, True]) async def test_extension_approval_cannot_be_bypassed_by_main_or_subagent(monkeypatch, pricing_ready, delegated): import sys from extensions import registry as extensions from extensions import define_tool from tools import registry as tools from agents import agent_factory, agent_memory from main import app writes = [] @define_tool(name="adjust_fan", group="通风") def write_fan(speed: int) -> str: """调整风机速度。""" writes.append(speed) return "完成" def setup(catalog): catalog.register_agent(name="ventilation", description="通风控制", prompt="使用工具调整风机。", tools=[write_fan], approval_tools=[write_fan]) modules = extensions.discover_modules() monkeypatch.setitem(sys.modules, "extensions.ventilation.extension", SimpleNamespace(setup=setup)) monkeypatch.setattr(extensions, "discover_modules", lambda: [*modules, "extensions.ventilation.extension"]) tools.reset_for_tests() responses = [] if delegated: responses.append(AIMessage(content="", tool_calls=[{"id": "delegate", "name": "task", "args": {"subagent_type": "ventilation", "description": "调整风机"}}])) responses.extend([ AIMessage(content="", tool_calls=[{"id": "fan", "name": "adjust_fan", "args": {"speed": 10}}]), AIMessage(content="已完成"), AIMessage(content="已汇总"), ]) model = ScriptedModel(responses=responses) monkeypatch.setattr("agents.model_factory.get_model", lambda: model) graph = agent_factory.build_agent() assert tools.get_spec("adjust_fan").requires_approval monkeypatch.setattr("agents.agent_registry.get_agent", lambda: graph) async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client: first = await client.post("/api/chat", data={"message": "调整风机", "user_name": "u"}) events = [json.loads(line[6:]) for line in first.text.splitlines() if line.startswith("data: ")] assert not [e for e in events if e["type"] == "error"], events interrupted = next(e for e in events if e["type"] == "interrupt") assert writes == [] sid = first.headers["x-session-id"] token = interrupted["tool_approval"]["interrupt_token"] response = await client.post("/api/chat/resume", json={"session_id": sid, "user_name": "u", "interrupt_token": token, "approved": True}) assert response.status_code == 200, response.text assert writes == [10], response.text assert run_store.list_runs(sid)[0]["status"] == "success", response.text await agent_memory.delete_thread(sid)