diff --git a/app/src/fileflash/agents/runtime/llm.py b/app/src/fileflash/agents/runtime/llm.py index 1596728..12a3ddc 100644 --- a/app/src/fileflash/agents/runtime/llm.py +++ b/app/src/fileflash/agents/runtime/llm.py @@ -263,13 +263,14 @@ async def _run_tool_loop( tool_results: list[dict[str, Any]] = [] for call in tool_calls: tool_output = await tool_executor(call["tool"], call["input"]) - tool_results.append( - { - "type": "tool_result", - "tool_use_id": call["id"], - "content": _tool_result_content(tool_output), - } - ) + tool_result: dict[str, Any] = { + "type": "tool_result", + "tool_use_id": call["id"], + "content": _tool_result_content(tool_output), + } + if _is_tool_error_payload(tool_output): + tool_result["is_error"] = True + tool_results.append(tool_result) messages.append({"role": "user", "content": tool_results}) loop_kwargs = dict(request_kwargs) loop_kwargs["messages"] = messages @@ -472,6 +473,12 @@ def _tool_result_content(payload: dict[str, Any]) -> list[dict[str, str]]: return [{"type": "text", "text": text}] +def _is_tool_error_payload(payload: Any) -> bool: + if not isinstance(payload, dict): + return False + return bool(payload.get("_toolError") is True or payload.get("isError") is True) + + def _reasoning_params(reasoning_effort: str) -> dict[str, Any]: effort = (reasoning_effort or "adaptive").strip().lower() if effort == "adaptive": diff --git a/app/src/fileflash/agents/runtime/plan_runner.py b/app/src/fileflash/agents/runtime/plan_runner.py index 6caaa75..33e8483 100644 --- a/app/src/fileflash/agents/runtime/plan_runner.py +++ b/app/src/fileflash/agents/runtime/plan_runner.py @@ -78,6 +78,10 @@ async def _run( metadata = await _collect_context_metadata(db, user_id=user_id, request=request) allowed_tools = _skill_tool_whitelist(skill) allowed_tool_set = set(allowed_tools) + exploration_tools = tuple( + tool_name for tool_name in allowed_tools if REGISTRY.get(tool_name).side_effect == "read" + ) + exploration_tool_set = set(exploration_tools) planner_router = ToolRouter(db=db, user_id=user_id) tool_call_budget = min(self.settings.agent_job_max_tool_calls, 32) planned_tool_calls = 0 @@ -85,19 +89,6 @@ async def _run( async def _planning_tool_executor(tool_name: str, args: dict[str, Any]) -> dict[str, Any]: nonlocal planned_tool_calls - if tool_name not in allowed_tool_set: - raise ApiError( - status_code=400, - code=400, - message=f"Planner attempted disallowed tool: {tool_name}", - ) - spec = REGISTRY.get(tool_name) - if spec.side_effect != "read": - raise ApiError( - status_code=400, - code=400, - message=f"Planner exploratory tool call must be read-only: {tool_name}", - ) planned_tool_calls += 1 if planned_tool_calls > tool_call_budget: raise ApiError( @@ -105,6 +96,39 @@ async def _planning_tool_executor(tool_name: str, args: dict[str, Any]) -> dict[ code=400, message="Planner exceeded exploratory tool-call budget", ) + if tool_name not in allowed_tool_set: + blocked = _blocked_planning_tool_result( + tool_name=tool_name, + reason="Planner attempted a tool that is not allowed by the selected skill.", + ) + if len(planning_evidence) < 12: + planning_evidence.append( + AgentPlanningEvidence( + step=planned_tool_calls, + tool=tool_name, + input=_evidence_mapping(args), + output_preview=_evidence_preview(blocked), + ) + ) + return blocked + if tool_name not in exploration_tool_set: + blocked = _blocked_planning_tool_result( + tool_name=tool_name, + reason=( + "Planner exploratory tool call must be read-only. " + "Move write operations into final proposedActions." + ), + ) + if len(planning_evidence) < 12: + planning_evidence.append( + AgentPlanningEvidence( + step=planned_tool_calls, + tool=tool_name, + input=_evidence_mapping(args), + output_preview=_evidence_preview(blocked), + ) + ) + return blocked output = await planner_router.dispatch(ToolCall(tool_name=tool_name, arguments=args)) if len(planning_evidence) < 12: planning_evidence.append( @@ -123,19 +147,29 @@ async def _planning_tool_executor(tool_name: str, args: dict[str, Any]) -> dict[ request=request, skill=skill, allowed_tools=allowed_tools, + exploration_tools=exploration_tools, metadata=metadata, ), max_tokens=request.hints.budget_tokens, reasoning_effort=request.hints.reasoning_effort, - tools=REGISTRY.anthropic_tools_for(allowed_tools), + tools=REGISTRY.anthropic_tools_for(exploration_tools), tool_executor=_planning_tool_executor, max_tool_roundtrips=6, ) + effective_max_steps: int | None + if self.settings.is_development_env: + effective_max_steps = None + else: + effective_max_steps = min( + request.hints.max_steps, + self.settings.agent_job_max_tool_calls, + ) + actions = _normalize_actions( llm_payload=llm_payload, allowed_tools=allowed_tools, - max_steps=min(request.hints.max_steps, self.settings.agent_job_max_tool_calls), + max_steps=effective_max_steps, ) chosen_skill = _chosen_skill(skill) llm_summary = str( @@ -458,6 +492,7 @@ def _user_prompt( request: PlanAgentRequest, skill: AgentSkill | AgentSkillCatalogEntry | None, allowed_tools: tuple[str, ...], + exploration_tools: tuple[str, ...], metadata: dict[str, Any], ) -> str: payload = { @@ -467,10 +502,12 @@ def _user_prompt( "dataPolicy": request.data_policy.model_dump(by_alias=True), "skill": _skill_payload(skill), "allowedTools": list(allowed_tools), + "explorationTools": list(exploration_tools), "toolSchemas": _tool_schemas(allowed_tools), "toolUseMode": ( - "Use read-only tools first when facts are missing. " - "Write tools must appear only in final proposedActions, not exploratory tool_use steps." + "Use exploratory tool_use only with read-only tools listed in explorationTools. " + "Never call write tools with tool_use; put write operations only in final proposedActions " + "and use '$stepN.field' references for cross-step ids." ), "plannerDefaults": { "organizeRequest": { @@ -539,18 +576,32 @@ def _tool_schemas(allowed_tools: tuple[str, ...]) -> list[dict[str, Any]]: return REGISTRY.schemas_for(allowed_tools) +def _blocked_planning_tool_result(*, tool_name: str, reason: str) -> dict[str, Any]: + return { + "_plannerBlocked": True, + "_toolError": True, + "errorCode": "planner.exploration_tool_blocked", + "tool": tool_name, + "message": reason, + "guidance": ( + "Do not execute write tools during exploration. " + "Place write actions in final proposedActions and reference prior outputs via '$stepN.field'." + ), + } + + def _normalize_actions( *, llm_payload: dict[str, Any], allowed_tools: tuple[str, ...], - max_steps: int, + max_steps: int | None, ) -> list[AgentProposedAction]: raw_actions = llm_payload.get("proposedActions", llm_payload.get("proposed_actions")) if raw_actions is None: raw_actions = llm_payload.get("actions") if not isinstance(raw_actions, list): raise ApiError(status_code=502, code=502, message="Agent plan JSON missing proposedActions") - if len(raw_actions) > max_steps: + if max_steps is not None and len(raw_actions) > max_steps: raise ApiError(status_code=400, code=400, message="Agent plan exceeds maxSteps") allowed = set(allowed_tools) diff --git a/app/src/fileflash/schemas/agent.py b/app/src/fileflash/schemas/agent.py index 97a25dd..47c273d 100644 --- a/app/src/fileflash/schemas/agent.py +++ b/app/src/fileflash/schemas/agent.py @@ -47,7 +47,7 @@ class AgentDataPolicy(CamelModel): class AgentHints(CamelModel): prefer_skill_id: str | None = None - max_steps: int = Field(default=12, ge=1, le=100) + max_steps: int = Field(default=12, ge=1) budget_tokens: int = Field(default=8_000, ge=1) reasoning_effort: AgentReasoningEffort = "adaptive" diff --git a/app/src/fileflash/services/agent/plan_service.py b/app/src/fileflash/services/agent/plan_service.py index 4ab7994..f5fc3a5 100644 --- a/app/src/fileflash/services/agent/plan_service.py +++ b/app/src/fileflash/services/agent/plan_service.py @@ -40,7 +40,10 @@ async def enqueue_plan(self, *, user_id: int, payload: PlanAgentRequest) -> Plan code=400, message="Agent token budget exceeds server limit", ) - if payload.hints.max_steps > self.settings.agent_job_max_tool_calls: + if ( + not self.settings.is_development_env + and payload.hints.max_steps > self.settings.agent_job_max_tool_calls + ): raise ApiError(status_code=400, code=400, message="Agent maxSteps exceeds server limit") await self._enforce_limits(user_id=user_id) diff --git a/app/src/fileflash/services/rate_limiter.py b/app/src/fileflash/services/rate_limiter.py index 00450a9..c872fe2 100644 --- a/app/src/fileflash/services/rate_limiter.py +++ b/app/src/fileflash/services/rate_limiter.py @@ -6,6 +6,7 @@ from redis.exceptions import RedisError logger = logging.getLogger(__name__) +_LOOP_RUNTIME_ERROR_MARKERS = ("different loop", "attached to a different loop", "event loop is closed") class RedisRateLimiter: @@ -37,9 +38,41 @@ async def allow_weighted(self, key: str, limit: int, window_seconds: int, weight except RedisError: logger.exception("Redis unavailable, rate limiter degraded for key=%s", key) return True + except RuntimeError as exc: + if not _is_loop_runtime_error(exc): + raise + logger.exception("Redis loop error, rate limiter degraded for key=%s", key) + await self._drop_cached_client() + return True async def close(self) -> None: - if self._redis is not None: - await self._redis.aclose() - self._redis = None + client = self._redis + self._redis = None + if client is None: + return + try: + await client.aclose() + except RuntimeError as exc: + if _is_loop_runtime_error(exc): + logger.warning("Ignore Redis close loop error: %s", exc) + return + raise + + async def _drop_cached_client(self) -> None: + client = self._redis + self._redis = None + if client is None: + return + try: + await client.aclose() + except RuntimeError as exc: + if _is_loop_runtime_error(exc): + logger.warning("Ignore Redis reset loop error: %s", exc) + return + raise + + +def _is_loop_runtime_error(exc: RuntimeError) -> bool: + message = str(exc).lower() + return any(marker in message for marker in _LOOP_RUNTIME_ERROR_MARKERS) diff --git a/app/tests/test_agent_plan_execute_runtime.py b/app/tests/test_agent_plan_execute_runtime.py index 256bdeb..245bf4b 100644 --- a/app/tests/test_agent_plan_execute_runtime.py +++ b/app/tests/test_agent_plan_execute_runtime.py @@ -69,6 +69,7 @@ def settings(**overrides): "agent_enabled": True, "agent_job_max_tokens": 50_000, "agent_job_max_tool_calls": 100, + "is_development_env": False, "agent_user_concurrent_limit": 2, "agent_user_daily_limit": 50, "agent_llm_base_url": None, @@ -124,6 +125,82 @@ async def test_plan_enqueue_returns_frontend_shape_and_sets_phase(): assert jobs.kwargs["payload"]["hints"]["reasoningEffort"] == "adaptive" +@pytest.mark.asyncio +async def test_plan_enqueue_rejects_max_steps_above_server_limit_in_non_dev(): + db = DummyDb() + jobs = FakeJobs() + service = PlanService( + db=db, + settings=settings(), + jobs=jobs, # type: ignore[arg-type] + plans=AgentPlanRepository(db), # type: ignore[arg-type] + settings_repo=AgentSettingsRepository(db), # type: ignore[arg-type] + work_sessions=AgentWorkSessionRepository(db), # type: ignore[arg-type] + ) + payload = PlanAgentRequest.model_validate( + { + "input": "整理当前文件夹", + "context": { + "rootFolderId": "root", + "selectedFileIds": [], + "selectedFolderIds": [], + "currentPath": "/My Files", + }, + "executionPolicy": "confirm", + "hints": { + "preferSkillId": None, + "maxSteps": 101, + "budgetTokens": 8000, + "reasoningEffort": "adaptive", + }, + } + ) + + with pytest.raises(ApiError) as exc: + await service.enqueue_plan(user_id=7, payload=payload) + + assert exc.value.status_code == 400 + assert exc.value.message == "Agent maxSteps exceeds server limit" + + +@pytest.mark.asyncio +async def test_plan_enqueue_allows_max_steps_above_server_limit_in_dev(): + db = DummyDb() + db.scalar = AsyncMock(side_effect=[0, 0]) + jobs = FakeJobs() + service = PlanService( + db=db, + settings=settings(is_development_env=True), + jobs=jobs, # type: ignore[arg-type] + plans=AgentPlanRepository(db), # type: ignore[arg-type] + settings_repo=AgentSettingsRepository(db), # type: ignore[arg-type] + work_sessions=AgentWorkSessionRepository(db), # type: ignore[arg-type] + ) + payload = PlanAgentRequest.model_validate( + { + "input": "整理当前文件夹", + "context": { + "rootFolderId": "root", + "selectedFileIds": [], + "selectedFolderIds": [], + "currentPath": "/My Files", + }, + "executionPolicy": "confirm", + "hints": { + "preferSkillId": None, + "maxSteps": 5000, + "budgetTokens": 8000, + "reasoningEffort": "adaptive", + }, + } + ) + + result = await service.enqueue_plan(user_id=7, payload=payload) + + assert result.job_id == "123" + assert jobs.kwargs["payload"]["hints"]["maxSteps"] == 5000 + + @pytest.mark.asyncio async def test_anthropic_planner_client_uses_sdk_and_parses_text_blocks(): class FakeMessages: @@ -750,6 +827,72 @@ async def test_plan_runner_generates_stable_hash(monkeypatch: pytest.MonkeyPatch assert "reasoningEffort" in planner.await_args.kwargs["user_prompt"] +@pytest.mark.asyncio +async def test_plan_runner_ignores_requested_max_steps_in_dev(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(plan_module, "_choose_skill", AsyncMock(return_value=None)) + monkeypatch.setattr( + plan_module, + "_collect_context_metadata", + AsyncMock(return_value={"scope": "currentFolder", "files": [], "folders": []}), + ) + monkeypatch.setattr(plan_module, "_upsert_agent_plan", AsyncMock(return_value=None)) + planner = AsyncMock( + return_value={ + "summary": "Move files into folders.", + "proposedActions": [ + { + "step": 1, + "tool": "drive.createFolder", + "input": {"parentFolderId": "root", "name": "Docs"}, + }, + { + "step": 2, + "tool": "drive.moveFile", + "input": {"fileId": "1", "targetFolderId": "2"}, + }, + ], + } + ) + request = PlanAgentRequest.model_validate( + { + "input": "organize", + "context": { + "rootFolderId": "root", + "selectedFileIds": [], + "selectedFolderIds": [], + "currentPath": "/My Files", + }, + "executionPolicy": "autopilot", + "hints": { + "preferSkillId": None, + "maxSteps": 1, + "budgetTokens": 8000, + "reasoningEffort": "adaptive", + }, + } + ) + job = BackgroundJob( + job_id=3221, + task_type="agent.plan", + status="running", + payload=request.model_dump(by_alias=True), + result={}, + requested_by=7, + scheduled_at=datetime.now(UTC), + created_at=datetime.now(UTC), + updated_at=datetime.now(UTC), + ) + runner = PlanRunner( + settings=settings(is_development_env=True), + planner_client=SimpleNamespace(create_plan=planner), # type: ignore[arg-type] + ) + + result = await runner.run(db=DummyDb(), job=job) # type: ignore[arg-type] + + assert len(result.proposed_actions) == 2 + assert [action.step for action in result.proposed_actions] == [1, 2] + + @pytest.mark.asyncio async def test_plan_runner_commits_after_upsert(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(plan_module, "_choose_skill", AsyncMock(return_value=None)) @@ -1085,7 +1228,7 @@ async def test_plan_runner_delegates_count_question_with_search_term_to_planner( @pytest.mark.asyncio -async def test_plan_runner_rejects_write_tool_in_exploratory_loop( +async def test_plan_runner_blocks_write_tool_in_exploratory_loop_and_continues( monkeypatch: pytest.MonkeyPatch, ): monkeypatch.setattr(plan_module, "_choose_skill", AsyncMock(return_value=None)) @@ -1095,11 +1238,34 @@ async def test_plan_runner_rejects_write_tool_in_exploratory_loop( AsyncMock(return_value={"scope": "currentFolder", "rootFolderId": "root", "files": [], "folders": []}), ) monkeypatch.setattr(plan_module, "_upsert_agent_plan", AsyncMock(return_value=None)) + dispatch_mock = AsyncMock(return_value={"ok": True}) + monkeypatch.setattr( + plan_module, + "ToolRouter", + lambda **kwargs: SimpleNamespace(dispatch=dispatch_mock), + ) async def fake_create_plan(**kwargs): # noqa: ANN003 tool_executor = kwargs["tool_executor"] - await tool_executor("drive.moveFile", {"fileId": "1", "targetFolderId": "2"}) - return {"summary": "should not reach", "proposedActions": []} + blocked = await tool_executor("drive.createFolder", {"parentFolderId": "root", "name": "Movies"}) + assert blocked["_plannerBlocked"] is True + assert blocked["_toolError"] is True + assert blocked["tool"] == "drive.createFolder" + return { + "summary": "create then move", + "proposedActions": [ + { + "step": 1, + "tool": "drive.createFolder", + "input": {"parentFolderId": "root", "name": "Movies"}, + }, + { + "step": 2, + "tool": "drive.moveFile", + "input": {"fileId": "1", "targetFolderId": "$step1.folderId"}, + }, + ], + } runner = PlanRunner( settings=settings(), @@ -1129,10 +1295,69 @@ async def fake_create_plan(**kwargs): # noqa: ANN003 updated_at=datetime.now(UTC), ) - with pytest.raises(ApiError) as exc: - await runner.run(db=DummyDb(), job=job) # type: ignore[arg-type] - assert exc.value.status_code == 400 - assert "read-only" in exc.value.message + result = await runner.run(db=DummyDb(), job=job) # type: ignore[arg-type] + assert len(result.proposed_actions) == 2 + assert result.proposed_actions[0].tool == "drive.createFolder" + assert result.proposed_actions[1].input["targetFolderId"] == "$step1.folderId" + dispatch_mock.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_plan_runner_passes_only_read_tools_to_planner_tool_use( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setattr(plan_module, "_choose_skill", AsyncMock(return_value=None)) + monkeypatch.setattr( + plan_module, + "_collect_context_metadata", + AsyncMock(return_value={"scope": "currentFolder", "rootFolderId": "root", "files": [], "folders": []}), + ) + monkeypatch.setattr(plan_module, "_upsert_agent_plan", AsyncMock(return_value=None)) + captured_tools: list[dict[str, Any]] = [] + + async def fake_create_plan(**kwargs): # noqa: ANN003 + nonlocal captured_tools + tools = kwargs.get("tools") + if isinstance(tools, list): + captured_tools = list(tools) + return {"summary": "ok", "proposedActions": []} + + runner = PlanRunner( + settings=settings(), + planner_client=SimpleNamespace(create_plan=fake_create_plan), # type: ignore[arg-type] + ) + request = PlanAgentRequest.model_validate( + { + "input": "整理当前文件夹", + "context": { + "rootFolderId": "root", + "selectedFileIds": [], + "selectedFolderIds": [], + "currentPath": "/My Files", + }, + "executionPolicy": "confirm", + } + ) + job = BackgroundJob( + job_id=341, + task_type="agent.plan", + status="running", + payload=request.model_dump(by_alias=True), + result={}, + requested_by=7, + scheduled_at=datetime.now(UTC), + created_at=datetime.now(UTC), + updated_at=datetime.now(UTC), + ) + + await runner.run(db=DummyDb(), job=job) # type: ignore[arg-type] + + assert captured_tools + internal_names = {str(item.get("internalName") or "") for item in captured_tools} + assert "drive.createFolder" not in internal_names + assert "drive.moveFile" not in internal_names + for internal_name in internal_names: + assert plan_module.REGISTRY.get(internal_name).side_effect == "read" @pytest.mark.asyncio @@ -1519,6 +1744,30 @@ def test_normalize_actions_rejects_symbolic_placeholder_target_folder(): assert "newFolderId" in exc.value.message +def test_normalize_actions_allows_unbounded_steps_when_max_steps_none(): + actions = plan_module._normalize_actions( + llm_payload={ + "summary": "organize movies", + "proposedActions": [ + { + "step": 1, + "tool": "drive.createFolder", + "input": {"parentFolderId": "root", "name": "Movies"}, + }, + { + "step": 2, + "tool": "drive.moveFile", + "input": {"fileId": "13", "targetFolderId": "$step1.folderId"}, + }, + ], + }, + allowed_tools=("drive.createFolder", "drive.moveFile"), + max_steps=None, + ) + + assert len(actions) == 2 + + def test_normalize_actions_accepts_previous_step_reference(): actions = plan_module._normalize_actions( llm_payload={ diff --git a/app/tests/test_rate_limiter.py b/app/tests/test_rate_limiter.py index 386c160..6720f6f 100644 --- a/app/tests/test_rate_limiter.py +++ b/app/tests/test_rate_limiter.py @@ -13,12 +13,22 @@ class FakeRedis: - def __init__(self, *, fail: bool = False) -> None: + def __init__( + self, + *, + fail: bool = False, + runtime_error: RuntimeError | None = None, + close_runtime_error: RuntimeError | None = None, + ) -> None: self.fail = fail + self.runtime_error = runtime_error + self.close_runtime_error = close_runtime_error self.values: dict[str, int] = {} self.expired: list[tuple[str, int]] = [] async def incrby(self, key: str, amount: int) -> int: + if self.runtime_error is not None: + raise self.runtime_error if self.fail: raise RedisError("down") self.values[key] = self.values.get(key, 0) + amount @@ -27,6 +37,10 @@ async def incrby(self, key: str, amount: int) -> int: async def expire(self, key: str, window_seconds: int) -> None: self.expired.append((key, window_seconds)) + async def aclose(self) -> None: + if self.close_runtime_error is not None: + raise self.close_runtime_error + @pytest.mark.asyncio async def test_allow_weighted_uses_incrby_and_sets_ttl() -> None: @@ -59,6 +73,36 @@ async def test_allow_weighted_degrades_open_when_redis_fails() -> None: assert await limiter.allow_weighted("k", limit=1, window_seconds=60, weight=10) is True +@pytest.mark.asyncio +async def test_allow_weighted_degrades_open_and_resets_client_on_loop_runtime_error() -> None: + limiter = RedisRateLimiter("redis://example") + limiter._redis = FakeRedis( + runtime_error=RuntimeError("Future pending attached to a different loop") + ) # type: ignore[assignment] + + assert await limiter.allow_weighted("k", limit=1, window_seconds=60, weight=10) is True + assert limiter._redis is None + + +@pytest.mark.asyncio +async def test_allow_weighted_reraises_non_loop_runtime_error() -> None: + limiter = RedisRateLimiter("redis://example") + limiter._redis = FakeRedis(runtime_error=RuntimeError("unexpected runtime failure")) # type: ignore[assignment] + + with pytest.raises(RuntimeError, match="unexpected runtime failure"): + await limiter.allow_weighted("k", limit=1, window_seconds=60, weight=10) + + +@pytest.mark.asyncio +async def test_close_ignores_loop_runtime_error() -> None: + limiter = RedisRateLimiter("redis://example") + limiter._redis = FakeRedis(close_runtime_error=RuntimeError("Event loop is closed")) # type: ignore[assignment] + + await limiter.close() + + assert limiter._redis is None + + class FakeRateLimiter: def __init__(self) -> None: self.allow = AsyncMock(return_value=False)