vent_agent_2.0/tests/test_runtime_regressions.py

"""执行、存储和人机交互的失败路径回归；不访问模型网络。"""
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)