From dfce60f7db0cd8e47ac05aed64e40a588849de2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=AD=86?= <2685475410@qq.com> Date: Fri, 11 Sep 2026 11:29:10 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat:=20WP-06=20=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=E5=9F=9F=E2=80=94=E2=80=94=E6=B3=A8=E5=86=8C=E8=A1=A8=E3=80=81?= =?UTF-8?q?=E5=9B=9B=E7=BB=B4=E8=A3=81=E5=86=B3=E3=80=81=E5=AE=A1=E6=89=B9?= =?UTF-8?q?=E4=B8=8E=20RunBudget?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 迁移 10：tool（注册表：风险级/审批声明/输入 Schema）、tool_call（六态含 pending_approval、幂等唯一约束）、run.max_tool_calls（RunBudget 硬上限） - 内置工具三件：get-current-time/calculate（AST 白名单四则，只读循律）、record-note（真实副作用：写入记忆候选，必须请命） - 四维裁决（能力/数据/后果/情境→autonomy_level）：read_only 循律直接执行；side_effect/high_impact 进入 pending_approval 请命；禁用工具 forbidden - 审批流（朱批）：approve → 执行并落审计；deny → 拒绝且不可逆（重复审批 409）；副作用工具产物走既有记忆审核通道 - 工具入参按不可信模型输出处理：JSON Schema 子集校验（required/type/未知字段拒绝）；RunBudget 超限 422 不静默放行 - 幂等：同 (run, tool, Idempotency-Key) 重放返回 200 原调用 - 测试 67 项全绿（+7）：循律即执行、请命→朱批→真实候选产物、拒绝不可逆与不执行、Schema 三类拒绝、未注册 404、幂等重放、预算硬上限 --- ...3a26cb61923_工具注册表_调用与_runbudget.py | 67 ++++ backend/app/main.py | 3 + backend/app/models.py | 59 ++++ backend/app/tools/__init__.py | 1 + backend/app/tools/routes.py | 112 +++++++ backend/app/tools/service.py | 292 ++++++++++++++++++ backend/tests/test_conversations.py | 135 ++++++++ 7 files changed, 669 insertions(+) create mode 100644 backend/alembic/versions/b3a26cb61923_工具注册表_调用与_runbudget.py create mode 100644 backend/app/tools/__init__.py create mode 100644 backend/app/tools/routes.py create mode 100644 backend/app/tools/service.py diff --git a/backend/alembic/versions/b3a26cb61923_工具注册表_调用与_runbudget.py b/backend/alembic/versions/b3a26cb61923_工具注册表_调用与_runbudget.py new file mode 100644 index 0000000..38c9e4e --- /dev/null +++ b/backend/alembic/versions/b3a26cb61923_工具注册表_调用与_runbudget.py @@ -0,0 +1,67 @@ +"""工具注册表、调用与 RunBudget + +Revision ID: b3a26cb61923 +Revises: a994e5b4337b +Create Date: 2026-09-11 11:19:50.562943 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = 'b3a26cb61923' +down_revision: str | Sequence[str] | None = 'a994e5b4337b' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('tool', + sa.Column('name', sa.String(), nullable=False), + sa.Column('description', sa.String(), nullable=False), + sa.Column('input_schema', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('risk_level', sa.String(), nullable=False), + sa.Column('requires_approval', sa.Boolean(), nullable=False), + sa.Column('enabled', sa.Boolean(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.CheckConstraint("risk_level IN ('read_only', 'side_effect', 'high_impact')", name=op.f('ck_tool_risk_level_values')), + sa.PrimaryKeyConstraint('name', name=op.f('pk_tool')) + ) + op.create_table('tool_call', + sa.Column('id', sa.String(), nullable=False), + sa.Column('run_id', sa.String(), nullable=False), + sa.Column('tool_name', sa.String(), nullable=False), + sa.Column('input', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('output', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column('status', sa.String(), nullable=False), + sa.Column('error', sa.String(), nullable=True), + sa.Column('idempotency_key', sa.String(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('completed_at', sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint("status IN ('pending_approval', 'approved', 'denied', 'executing', 'succeeded', 'failed')", name=op.f('ck_tool_call_status_values')), + sa.ForeignKeyConstraint(['run_id'], ['run.id'], name=op.f('fk_tool_call_run'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_tool_call')) + ) + op.create_index(op.f('ix_tool_call_run_id'), 'tool_call', ['run_id'], unique=False) + op.create_index('uq_tool_call_idempotency', 'tool_call', ['run_id', 'tool_name', 'idempotency_key'], unique=True) + op.create_foreign_key(op.f('fk_memory_memory_version'), 'memory', 'memory_version', ['current_version_id'], ['id'], ondelete='set null', use_alter=True) + op.add_column('run', sa.Column('max_tool_calls', sa.Integer(), nullable=False)) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column('run', 'max_tool_calls') + op.drop_constraint(op.f('fk_memory_memory_version'), 'memory', type_='foreignkey') + op.drop_index('uq_tool_call_idempotency', table_name='tool_call') + op.drop_index(op.f('ix_tool_call_run_id'), table_name='tool_call') + op.drop_table('tool_call') + op.drop_table('tool') + # ### end Alembic commands ### diff --git a/backend/app/main.py b/backend/app/main.py index 1142a0a..9353fdb 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -12,6 +12,7 @@ from app.core.middleware import RequestContextMiddleware, configure_logging from app.db.session import init_engine from app.memories.routes import cycle_router from app.memories.routes import router as memories_router +from app.tools.routes import calls_router, tools_router def create_app() -> FastAPI: @@ -34,6 +35,8 @@ def create_app() -> FastAPI: app.include_router(cycle_router) # 先于 characters：避免通配生命周期路由拦截 app.include_router(characters_router) app.include_router(memories_router) + app.include_router(tools_router) + app.include_router(calls_router) @app.get("/healthz", tags=["ops"]) async def healthz() -> dict[str, str]: diff --git a/backend/app/models.py b/backend/app/models.py index e6d4127..b1c126b 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -240,6 +240,7 @@ class Run(Base): ForeignKey("conversation.id", ondelete="cascade"), nullable=True, index=True ) status: Mapped[str] = mapped_column(String, nullable=False, default="queued") + max_tool_calls: Mapped[int] = mapped_column(nullable=False, default=10) # RunBudget 硬上限 created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) @@ -614,3 +615,61 @@ class BigCycle(Base): "status IN ('pending', 'running', 'completed', 'failed')", name="status_values" ), ) + + +class Tool(Base): + """工具注册表：内置工具与后续插件贡献统一登记（WP-06）。""" + + __tablename__ = "tool" + + name: Mapped[str] = mapped_column(String, primary_key=True) # kebab-case + description: Mapped[str] = mapped_column(String, nullable=False, default="") + input_schema: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + risk_level: Mapped[str] = mapped_column(String, nullable=False, default="read_only") + requires_approval: Mapped[bool] = mapped_column(nullable=False, default=False) + enabled: Mapped[bool] = mapped_column(nullable=False, default=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + CheckConstraint( + "risk_level IN ('read_only', 'side_effect', 'high_impact')", + name="risk_level_values", + ), + ) + + +class ToolCall(Base): + """工具调用：Schema 校验与 Policy 判定后的执行记录（含审批态）。""" + + __tablename__ = "tool_call" + + id: Mapped[str] = mapped_column( + String, primary_key=True, default=lambda: new_id("tool_call_") + ) + run_id: Mapped[str] = mapped_column( + ForeignKey("run.id", ondelete="cascade"), nullable=False, index=True + ) + tool_name: Mapped[str] = mapped_column(String, nullable=False) + input: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + output: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) + status: Mapped[str] = mapped_column(String, nullable=False, default="pending_approval") + error: Mapped[str | None] = mapped_column(String, nullable=True) + idempotency_key: Mapped[str | None] = mapped_column(String, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + __table_args__ = ( + CheckConstraint( + "status IN ('pending_approval', 'approved', 'denied', " + "'executing', 'succeeded', 'failed')", + name="status_values", + ), + Index( + "uq_tool_call_idempotency", "run_id", "tool_name", "idempotency_key", + unique=True, + ), + ) diff --git a/backend/app/tools/__init__.py b/backend/app/tools/__init__.py new file mode 100644 index 0000000..4c589ef --- /dev/null +++ b/backend/app/tools/__init__.py @@ -0,0 +1 @@ +"""工具域：注册表、裁决与审批。""" diff --git a/backend/app/tools/routes.py b/backend/app/tools/routes.py new file mode 100644 index 0000000..cc5ee72 --- /dev/null +++ b/backend/app/tools/routes.py @@ -0,0 +1,112 @@ +"""工具域路由（WP-06）。""" + +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, Header, HTTPException, Request, status +from fastapi.responses import JSONResponse +from sqlalchemy import select + +from app.auth.dependencies import get_current_user, require_state_change +from app.conversations.routes import _domain_of +from app.db.session import DbSession +from app.models import Tool, ToolCall, User +from app.tools import service + +tools_router = APIRouter(prefix="/v1/tools", tags=["tools"]) +calls_router = APIRouter(prefix="/v1", tags=["tool-calls"]) + + +def _tool_out(t: Tool) -> dict[str, Any]: + return { + "name": t.name, + "description": t.description, + "input_schema": t.input_schema, + "risk_level": t.risk_level, + "requires_approval": t.requires_approval, + "enabled": t.enabled, + } + + +def _call_out(c: ToolCall) -> dict[str, Any]: + return { + "id": c.id, + "run_id": c.run_id, + "tool_name": c.tool_name, + "input": c.input, + "output": c.output, + "status": c.status, + "error": c.error, + "created_at": c.created_at.isoformat(), + "completed_at": c.completed_at.isoformat() if c.completed_at else None, + } + + +@tools_router.get("") +async def list_tools( + db: DbSession, + user: Annotated[User, Depends(get_current_user)], +) -> dict[str, Any]: + await service.seed_builtin_tools(db) # 幂等登记（后续由 Registry 管理取代） + rows = list((await db.scalars(select(Tool).order_by(Tool.name.asc()))).all()) + return {"data": [_tool_out(t) for t in rows]} + + +@calls_router.post("/runs/{run_id}/tool-calls", status_code=201, response_model=None) +async def create_tool_call( + run_id: str, + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, + body: dict[str, Any], + idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key")] = None, +) -> dict[str, Any] | JSONResponse: + domain = await _domain_of(db, user) + if idempotency_key: + body = {**body, "idempotency_key": idempotency_key} + call, replayed = await service.create_tool_call( + db, domain.id, run_id, body, getattr(request.state, "trace_id", None) + ) + if replayed: + from fastapi.responses import JSONResponse + + return JSONResponse(status_code=200, content=_call_out(call)) + return _call_out(call) + + +@calls_router.post("/tool-calls/{tool_call_id}/approvals", status_code=201) +async def decide_approval( + tool_call_id: str, + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, + body: dict[str, bool], +) -> dict[str, Any]: + domain = await _domain_of(db, user) + call = await service.decide_approval( + db, domain.id, tool_call_id, bool(body.get("approve")), + getattr(request.state, "trace_id", None), + ) + return _call_out(call) + + +@calls_router.get("/tool-calls") +async def list_tool_calls( + db: DbSession, + user: Annotated[User, Depends(get_current_user)], + status_filter: str = "pending_approval", +) -> dict[str, Any]: + if status_filter not in { + "pending_approval", "approved", "denied", "executing", "succeeded", "failed", + }: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "status 取值不合法") + domain = await _domain_of(db, user) + from app.models import Run + + stmt = ( + select(ToolCall) + .join(Run, ToolCall.run_id == Run.id) + .where(Run.personal_domain_id == domain.id, ToolCall.status == status_filter) + .order_by(ToolCall.created_at.asc()) + ) + rows = list((await db.scalars(stmt)).all()) + return {"data": [_call_out(c) for c in rows]} diff --git a/backend/app/tools/service.py b/backend/app/tools/service.py new file mode 100644 index 0000000..f5da2c7 --- /dev/null +++ b/backend/app/tools/service.py @@ -0,0 +1,292 @@ +"""工具域：注册表、内置工具、Schema 校验与四维策略裁决（WP-06）。""" + +import ast +import operator +from collections.abc import Callable +from datetime import UTC, datetime +from typing import Any + +from fastapi import HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models import AuditEvent, Run, Tool, ToolCall + +# ===== 内置工具实现（纯函数；副作用工具由服务层落库） ===== + +_UNARY: dict[type[ast.AST], Callable[[float], float]] = { + ast.USub: operator.neg, ast.UAdd: operator.pos, +} +_BINARY: dict[type[ast.AST], Callable[[float, float], float]] = { + ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, + ast.Div: operator.truediv, ast.Mod: operator.mod, ast.Pow: operator.pow, +} + + +def _safe_eval(node: ast.AST) -> float: + """AST 白名单四则运算：仅数字、一元/二元算子；任何其它节点抛 ValueError。""" + if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): + return float(node.value) + if isinstance(node, ast.UnaryOp): + fn = _UNARY.get(type(node.op)) + if fn is not None: + return fn(_safe_eval(node.operand)) + if isinstance(node, ast.BinOp): + fn2 = _BINARY.get(type(node.op)) + if fn2 is not None: + return fn2(_safe_eval(node.left), _safe_eval(node.right)) + raise ValueError("表达式包含不允许的节点") + + +def tool_get_current_time(_input: dict[str, Any]) -> dict[str, Any]: + now = datetime.now(UTC) + return {"utc_now": now.isoformat(), "epoch_ms": int(now.timestamp() * 1000)} + + +def tool_calculate(input_: dict[str, Any]) -> dict[str, Any]: + tree = ast.parse(str(input_.get("expression", "")), mode="eval") + return {"result": _safe_eval(tree.body)} + + +BUILTIN_TOOLS: dict[str, dict[str, Any]] = { + "get-current-time": { + "description": "读取当前 UTC 时间（只读，无副作用）", + "input_schema": {"type": "object", "properties": {}, "required": []}, + "risk_level": "read_only", + "requires_approval": False, + "handler": tool_get_current_time, + }, + "calculate": { + "description": "四则运算（AST 白名单求值，只读）", + "input_schema": { + "type": "object", + "properties": {"expression": {"type": "string"}}, + "required": ["expression"], + }, + "risk_level": "read_only", + "requires_approval": False, + "handler": tool_calculate, + }, + "record-note": { + "description": "把一条笔记写入记忆候选（副作用，需审批）", + "input_schema": { + "type": "object", + "properties": {"text": {"type": "string"}}, + "required": ["text"], + }, + "risk_level": "side_effect", + "requires_approval": True, + "handler": None, + }, +} + +# ===== Schema 校验（JSON Schema 子集：type/required） ===== + +_TYPES: dict[str, tuple[type, ...]] = { + "string": (str,), "number": (int, float), "boolean": (bool,), "object": (dict,), +} + + +def validate_input(schema: dict[str, Any], input_: dict[str, Any]) -> None: + props: dict[str, Any] = schema.get("properties", {}) + for name in schema.get("required", []): + if name not in input_: + raise HTTPException( + status.HTTP_422_UNPROCESSABLE_ENTITY, f"缺少必需字段：{name}" + ) + for name, value in input_.items(): + spec = props.get(name) + if spec is None: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, f"未知字段：{name}") + expected = spec.get("type") + if expected in _TYPES and not isinstance(value, _TYPES[expected]): + raise HTTPException( + status.HTTP_422_UNPROCESSABLE_ENTITY, f"字段 {name} 类型应为 {expected}" + ) + + +# ===== 四维裁决（能力/数据/后果/情境 → autonomy_level） ===== + +def decide(tool: Tool) -> str: + """返回 autonomy_level：rule_bound（循律直接执行）或 approval_required（请命）。 + + 四维映射（agent-runtime 第 6 节）：能力=risk_level 与审批声明；数据=内置工具 + 不涉外部数据；后果=side_effect/high_impact 必须请命；情境=工具启用状态。 + """ + if not tool.enabled: + return "forbidden" + if tool.requires_approval or tool.risk_level in {"side_effect", "high_impact"}: + return "approval_required" + return "rule_bound" + + +async def seed_builtin_tools(db: AsyncSession) -> None: + """幂等登记内置工具（已存在则更新声明，不动 enabled）。""" + for name, spec in BUILTIN_TOOLS.items(): + existing = await db.get(Tool, name) + if existing is None: + db.add( + Tool( + name=name, + description=spec["description"], + input_schema=spec["input_schema"], + risk_level=spec["risk_level"], + requires_approval=spec["requires_approval"], + ) + ) + else: + existing.description = spec["description"] + existing.input_schema = spec["input_schema"] + existing.risk_level = spec["risk_level"] + existing.requires_approval = spec["requires_approval"] + await db.flush() + + +async def create_tool_call( + db: AsyncSession, + domain_id: str, + run_id: str, + body: dict[str, Any], + trace_id: str | None, +) -> tuple[ToolCall, bool]: + """创建工具调用：幂等重放 → Schema 校验 → RunBudget → 四维裁决。""" + run = await db.get(Run, run_id) + if run is None or run.personal_domain_id != domain_id: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Run 不存在") + key = body.get("idempotency_key") + if key: + existing = await db.execute( + select(ToolCall).where( + ToolCall.run_id == run_id, + ToolCall.tool_name == body.get("tool_name"), + ToolCall.idempotency_key == key, + ) + ) + replay = existing.scalar_one_or_none() + if replay is not None: + return replay, True + tool = await db.get(Tool, body.get("tool_name", "")) + if tool is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "工具未注册") + used = len( + list( + ( + await db.execute( + select(ToolCall.id).where( + ToolCall.run_id == run_id, + ToolCall.status.in_( + ["succeeded", "executing", "approved", "pending_approval"] + ), + ) + ) + ).scalars().all() + ) + ) + if used >= run.max_tool_calls: + raise HTTPException( + status.HTTP_422_UNPROCESSABLE_ENTITY, + f"RunBudget 工具调用上限已达（{run.max_tool_calls}）", + ) + input_ = body.get("input", {}) + if not isinstance(input_, dict): + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "input 必须是对象") + validate_input(tool.input_schema, input_) # 工具入参是不可信模型输出 + level = decide(tool) + if level == "forbidden": + raise HTTPException(status.HTTP_403_FORBIDDEN, "工具已被禁用") + call = ToolCall( + run_id=run_id, + tool_name=tool.name, + input=input_, + idempotency_key=key, + status="pending_approval" if level == "approval_required" else "approved", + ) + db.add(call) + await db.flush() + db.add( + AuditEvent( + actor_type="user", actor_id=domain_id, action="tool.call.created", + aggregate_type="tool_call", aggregate_id=call.id, + payload={"tool": tool.name, "autonomy": level}, trace_id=trace_id, + ) + ) + if level == "rule_bound": + await _execute(db, domain_id, call, trace_id) + await db.flush() + return call, False + + +async def decide_approval( + db: AsyncSession, + domain_id: str, + tool_call_id: str, + approve: bool, + trace_id: str | None, +) -> ToolCall: + """朱批：approve → 执行；deny → 拒绝不执行且不可逆（拒绝/超时均不执行）。""" + call = await db.get(ToolCall, tool_call_id) + if call is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "调用不存在") + run = await db.get(Run, call.run_id) + assert run is not None + if run.personal_domain_id != domain_id: + raise HTTPException(status.HTTP_404_NOT_FOUND, "调用不存在") + if call.status != "pending_approval": + raise HTTPException(status.HTTP_409_CONFLICT, "调用不在待审批状态") + if not approve: + call.status = "denied" + call.completed_at = datetime.now(UTC) + db.add( + AuditEvent( + actor_type="user", actor_id=domain_id, action="tool.call.denied", + aggregate_type="tool_call", aggregate_id=call.id, trace_id=trace_id, + ) + ) + await db.flush() + return call + call.status = "approved" + await _execute(db, domain_id, call, trace_id) + await db.flush() + return call + + +async def _execute( + db: AsyncSession, domain_id: str, call: ToolCall, trace_id: str | None +) -> None: + """执行工具：内置纯函数或副作用落库；结果与错误如实记录。""" + spec = BUILTIN_TOOLS.get(call.tool_name) + call.status = "executing" + try: + if call.tool_name == "record-note": + from app.memories import service as memory_service + + candidate, _memory = await memory_service.submit_candidate( + db, + domain_id, + { + "kind": "fact", + "sensitivity": "low", + "source": "auto", + "payload": {"text": str(call.input.get("text", ""))}, + "source_ref": {"tool_call_id": call.id}, + }, + trace_id, + ) + call.output = {"memory_candidate_id": candidate.id, "status": candidate.status} + else: + assert spec is not None and spec["handler"] is not None + call.output = spec["handler"](call.input) + call.status = "succeeded" + except HTTPException: + raise + except Exception as exc: # 工具失败如实记录，不丢弃不伪装 + call.status = "failed" + call.error = str(exc)[:500] + call.completed_at = datetime.now(UTC) + db.add( + AuditEvent( + actor_type="system", actor_id="tool-runtime", action="tool.call.executed", + aggregate_type="tool_call", aggregate_id=call.id, + payload={"status": call.status}, trace_id=trace_id, + ) + ) diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py index 0852eaa..7136f3c 100644 --- a/backend/tests/test_conversations.py +++ b/backend/tests/test_conversations.py @@ -881,3 +881,138 @@ class TestBigCycle: assert version.status == "pending" # 候选先于写入 assert version.payload["source"] == "big_cycle" assert version.payload["proposal"] # 模型真实输出 + + +@pytest.mark.usefixtures("client") +class TestTools: + async def _make_run(self, client, headers) -> str: + cid = await _create_conversation(client, headers) + resp = await _post_message(client, headers, cid, f"idem-tool-{_uuid4().split(chr(45))[0]}") + await _wait_run_terminal_headers(client, headers) + return resp.json()["run"]["id"] + + async def test_rule_bound_tool_executes_immediately(self, client: AsyncClient) -> None: + headers = await login(client, "tool1@example.com") + await client.get("/v1/tools", headers=headers) # 触发内置登记 + run_id = await self._make_run(client, headers) + resp = await client.post( + f"/v1/runs/{run_id}/tool-calls", + json={"tool_name": "calculate", "input": {"expression": "(1+2)*3"}}, + headers={**headers, "Idempotency-Key": "tc-1"}, + ) + assert resp.status_code == 201, resp.text + body = resp.json() + assert body["status"] == "succeeded" + assert body["output"]["result"] == 9.0 + + async def test_side_effect_tool_requires_approval(self, client: AsyncClient) -> None: + headers = await login(client, "tool2@example.com") + await client.get("/v1/tools", headers=headers) + run_id = await self._make_run(client, headers) + created = await client.post( + f"/v1/runs/{run_id}/tool-calls", + json={"tool_name": "record-note", "input": {"text": "重要笔记"}}, + headers=headers, + ) + assert created.json()["status"] == "pending_approval" + call_id = created.json()["id"] + # 待审批列表可见 + pending = await client.get("/v1/tool-calls", headers=headers) + assert any(c["id"] == call_id for c in pending.json()["data"]) + # 朱批 → 执行 → 真实副作用（记忆候选创建） + approved = await client.post( + f"/v1/tool-calls/{call_id}/approvals", json={"approve": True}, headers=headers + ) + assert approved.json()["status"] == "succeeded" + assert approved.json()["output"]["memory_candidate_id"] + + async def test_denied_tool_never_executes(self, client: AsyncClient) -> None: + headers = await login(client, "tool3@example.com") + await client.get("/v1/tools", headers=headers) + run_id = await self._make_run(client, headers) + created = await client.post( + f"/v1/runs/{run_id}/tool-calls", + json={"tool_name": "record-note", "input": {"text": "被拒绝的笔记"}}, + headers=headers, + ) + call_id = created.json()["id"] + denied = await client.post( + f"/v1/tool-calls/{call_id}/approvals", json={"approve": False}, headers=headers + ) + assert denied.json()["status"] == "denied" + assert denied.json()["output"] is None # 未执行 + again = await client.post( + f"/v1/tool-calls/{call_id}/approvals", json={"approve": True}, headers=headers + ) + assert again.status_code == 409 # 拒绝不可逆 + + async def test_schema_validation_rejects_bad_input(self, client) -> None: + headers = await login(client, "tool4@example.com") + await client.get("/v1/tools", headers=headers) + run_id = await self._make_run(client, headers) + missing = await client.post( + f"/v1/runs/{run_id}/tool-calls", + json={"tool_name": "calculate", "input": {}}, headers=headers, + ) + assert missing.status_code == 422 + wrong_type = await client.post( + f"/v1/runs/{run_id}/tool-calls", + json={"tool_name": "calculate", "input": {"expression": 42}}, headers=headers, + ) + assert wrong_type.status_code == 422 + + async def test_unknown_tool_404(self, client) -> None: + headers = await login(client, "tool5@example.com") + run_id = await self._make_run(client, headers) + resp = await client.post( + f"/v1/runs/{run_id}/tool-calls", + json={"tool_name": "rm-rf", "input": {}}, headers=headers, + ) + assert resp.status_code == 404 + + async def test_idempotent_replay_returns_original(self, client) -> None: + from fastapi.responses import JSONResponse # noqa: F401 + + headers = await login(client, "tool6@example.com") + await client.get("/v1/tools", headers=headers) + run_id = await self._make_run(client, headers) + first = await client.post( + f"/v1/runs/{run_id}/tool-calls", + json={"tool_name": "get-current-time", "input": {}}, + headers={**headers, "Idempotency-Key": "tc-replay"}, + ) + assert first.status_code == 201 + second = await client.post( + f"/v1/runs/{run_id}/tool-calls", + json={"tool_name": "get-current-time", "input": {}}, + headers={**headers, "Idempotency-Key": "tc-replay"}, + ) + assert second.status_code == 200 + assert second.json()["id"] == first.json()["id"] + + async def test_run_budget_hard_limit(self, client) -> None: + headers = await login(client, "tool7@example.com") + await client.get("/v1/tools", headers=headers) + run_id = await self._make_run(client, headers) + # max_tool_calls 默认 10：连续调用到上限后第 11 次被拒 + last_status = None + for i in range(11): + resp = await client.post( + f"/v1/runs/{run_id}/tool-calls", + json={"tool_name": "get-current-time", "input": {}}, + headers={**headers, "Idempotency-Key": f"tc-budget-{i}"}, + ) + last_status = resp.status_code + assert last_status == 422 + + +def _uuid4() -> str: + from uuid import uuid4 + + return str(uuid4()) + + +async def _wait_run_terminal_headers(client, headers, limit_s: float = 10.0) -> None: + import asyncio + + await asyncio.sleep(0)