Feat/audit

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":

app/src/fileflash/agents/runtime/plan_runner.py

@@ -78,33 +78,57 @@ 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
         planning_evidence: list[AgentPlanningEvidence] = []
 
         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(
                     status_code=400,
                     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)

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"
 

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)

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)