From 5861e98c2d44335e33464bdb64215b43c5923175 Mon Sep 17 00:00:00 2001 From: Raffael Meyer <14891507+barredterra@users.noreply.github.com> Date: Mon, 13 Jul 2026 03:48:24 +0200 Subject: [PATCH 1/5] feat: move from mozilla any-agent to langchain deepagents (#44) (cherry picked from commit e7a2dbdc3dccb6c12132d83f148f00ebd9fa9df0) --- README.md | 32 +- ask_alyf/ask_alyf/agent.py | 2010 ++--------------- ask_alyf/ask_alyf/api.py | 2 +- ask_alyf/ask_alyf/deep_agent_backend.py | 355 +++ .../test_ask_alyf_conversation.py | 26 +- ask_alyf/ask_alyf/field_agent.py | 35 +- ask_alyf/ask_alyf/history.py | 155 ++ ask_alyf/ask_alyf/subagents.py | 83 + ask_alyf/ask_alyf/test_code_tools.py | 319 ++- ask_alyf/ask_alyf/test_deep_agent_backend.py | 145 ++ ask_alyf/ask_alyf/test_field_agent.py | 40 +- ask_alyf/ask_alyf/toolset.py | 1156 ++++++++++ pyproject.toml | 3 +- 13 files changed, 2369 insertions(+), 1992 deletions(-) create mode 100644 ask_alyf/ask_alyf/deep_agent_backend.py create mode 100644 ask_alyf/ask_alyf/history.py create mode 100644 ask_alyf/ask_alyf/subagents.py create mode 100644 ask_alyf/ask_alyf/test_deep_agent_backend.py create mode 100644 ask_alyf/ask_alyf/toolset.py diff --git a/README.md b/README.md index 99c7ac5..b16f5f3 100644 --- a/README.md +++ b/README.md @@ -77,7 +77,9 @@ Common settings include: ## Conversation History -Conversations are stored in the **Ask ALYF Conversation** DocType. This keeps conversations available across page reloads and provides an audit trail for Agent mode proposals and actions. +Conversations are stored in the **Ask ALYF Conversation** DocType, which keeps messages available across page reloads and provides an audit trail for Agent mode proposals and actions. + +The coordinator rebuilds the full conversation history as native LangChain messages on each turn from the stored `messages_json` field. This keeps the source of truth in a single, user-visible DocType that respects owner-only permissions. Old conversations are deleted after 90 days by default. The retention period can be configured per site through **Log Settings**. @@ -117,15 +119,33 @@ Files, printing, charts, and navigation: Optional tools: -- `source_code_analyzer` searches installed app files when _Allow Code Search_ is enabled - `run_read_only_sql` runs read-only SQL for Administrator and System Manager users - `get_app_version`, `read_github_releases`, and `read_documentation_page` help answer app and documentation questions +### Subagents + +The coordinator delegates specialized work to restricted Deep Agents subagents via the built-in `task` tool. Each subagent has an explicit, minimal tool list so the parent's proposal and mutation tools are never inherited. + +- `source-code-analyzer` is registered only when _Allow Code Search_ is enabled. It searches and reads installed app code through the read-only `/source/` virtual filesystem mount and returns a structured `SourceCodeAnalysisResult`. +- `document-planner` is registered only in Agent mode. It inspects metadata and documents with read-only tools and returns a `DocumentPlannerResult` that the coordinator turns into a confirmed proposal. +- A built-in `general-purpose` subagent is provided by Deep Agents for open-ended delegation. + +### Todos and Virtual Filesystem + +The coordinator uses Deep Agents' built-in `write_todos` planning state to track its own progress. Todos are internal scratch state and are not mapped to Frappe **ToDo** records. + +The agent operates against a restricted composite virtual filesystem: + +- `/workspace/` is an in-memory scratch mount backed by `StateBackend`. It is _not_ directly writable by the model: filesystem writes are denied by the harness profile (which excludes `write_file`, `edit_file`, and `execute`) and by a blanket `FilesystemPermission` deny rule on `/**`. Planning and progress tracking happen through the built-in `write_todos` tool, not through filesystem writes. +- `/source/` exposes installed app source code read-only, confined to installed app roots and gated by _Allow Code Search_. +- `/attachments/` exposes Frappe **File** documents read-only, with per-read permission checks. + +Host filesystem writes, shell execution, and the built-in `write_file`/`edit_file`/`execute` tools are excluded from the model-visible tool surface. + ### Agent Mode Tools Agent mode includes the Ask mode tools plus tools that prepare confirmed actions: -- `document_planner` prepares safe `insert`, `save`, or `set_value` payloads before a proposal is shown - `insert` creates a document - `batch_insert` creates multiple documents of the same DocType - `save` updates a document @@ -150,8 +170,10 @@ Permission errors, validation errors, missing records, and LLM provider failures ## Dependencies -- [any-agent](https://mozilla-ai.github.io/any-agent/) for agent orchestration -- [any-llm-sdk](https://github.com/mozilla-ai/any-llm) for LLM provider access +- [Deep Agents](https://github.com/langchain-ai/deepagents) for agent orchestration, native messages, subagents, todos, and the virtual filesystem +- [LangChain](https://github.com/langchain-ai/langchain) and [langgraph](https://github.com/langchain-ai/langgraph) for the agent runtime +- [langchain-openai](https://github.com/langchain-ai/langchain) for the OpenAI-compatible chat model adapter +- [any-llm-sdk](https://github.com/mozilla-ai/any-llm) for LLM provider discovery and vision extraction - [PyMuPDF](https://pymupdf.readthedocs.io/) for PDF handling ## Installation diff --git a/ask_alyf/ask_alyf/agent.py b/ask_alyf/ask_alyf/agent.py index 2cb8223..de3c683 100644 --- a/ask_alyf/ask_alyf/agent.py +++ b/ask_alyf/ask_alyf/agent.py @@ -1,60 +1,57 @@ -import functools -import inspect -import json -import re from collections.abc import Callable -from dataclasses import dataclass, field from typing import Any -from urllib.parse import quote -from uuid import uuid4 import frappe -from any_agent import AgentConfig, AgentFramework, AnyAgent -from any_llm import AnyLLM +from deepagents import ( + FilesystemPermission, + HarnessProfile, + SubAgent, + create_deep_agent, + register_harness_profile, +) from frappe import _ +from langchain.agents import create_agent +from langchain_core.messages import AnyMessage, HumanMessage +from langchain_openai import ChatOpenAI from ask_alyf.ask_alyf import tools -from ask_alyf.ask_alyf.skill_utils import ( - build_available_skills_instruction, - get_accessible_skill_doc, +from ask_alyf.ask_alyf.deep_agent_backend import build_ask_alyf_backend +from ask_alyf.ask_alyf.history import history_item_to_native_message +from ask_alyf.ask_alyf.skill_utils import build_available_skills_instruction +from ask_alyf.ask_alyf.subagents import ( + DOCUMENT_PLANNER_INSTRUCTIONS, + SOURCE_CODE_ANALYZER_INSTRUCTIONS, + DocumentPlannerResult, + SourceCodeAnalysisResult, ) -from ask_alyf.ask_alyf.utils import parse_newline_list - -SPECIALIST_JSON_OUTPUT_INSTRUCTION = ( - "Return only a valid JSON object. " - "Do not wrap the JSON in markdown fences. " - "Do not add explanatory prose before or after the JSON." +from ask_alyf.ask_alyf.toolset import ( + ask_alyfRuntime, + ask_alyfToolset, + clear_messages_on_tool_error, ) -ALLOWED_DOCUMENT_PLANNER_TOOLS = frozenset({"insert", "save", "set_value"}) - -@dataclass -class ask_alyfRuntime: - conversation_name: str - mode: str - request_context: dict[str, Any] - conversation_history: list[dict[str, Any]] = field(default_factory=list) - pending_operations: list[dict[str, Any]] = field(default_factory=list) - document_extractions: list[dict[str, Any]] = field(default_factory=list) - attached_files: list[dict[str, Any]] = field(default_factory=list) +# Deep Agents exposes built-in filesystem write tools (``write_file``, +# ``edit_file``) and a shell ``execute`` tool by default. Ask ALYF must never +# let the model mutate the host filesystem or run shell commands directly, so +# we register a provider-wide OpenAI harness profile that strips those tools +# from every Deep Agents graph built from a ``ChatOpenAI`` model. Read-only +# filesystem tools (``ls``, ``read_file``, ``glob``, ``grep``) remain available +# and operate on the restricted composite VFS (workspace, source, attachments). +ASK_ALYF_EXCLUDED_TOOLS = frozenset({"write_file", "edit_file", "execute"}) +_ASK_ALYF_PROFILE_REGISTERED = False - def emit_status(self, text: str): - """Send a short status update to the current user.""" - frappe.publish_realtime( - "ask_alyf_status", - {"conversation": self.conversation_name, "text": text}, - user=frappe.session.user, - ) - def remember_document_extraction(self, extraction: dict[str, Any], *, extraction_prompt: str = ""): - """Store a normalized extraction result for reuse in later turns.""" - self.document_extractions.append( - _build_document_extraction_history_entry(extraction, extraction_prompt=extraction_prompt) - ) +def _ensure_ask_alyf_harness_profile() -> None: + """Register the Ask ALYF harness profile once per process. - def remember_attached_file(self, file_entry: dict[str, Any]): - """Store a file attachment to be shown in the conversation history.""" - self.attached_files.append(file_entry) + Registration is additive and idempotent, so concurrent calls are safe. + """ + global _ASK_ALYF_PROFILE_REGISTERED + if _ASK_ALYF_PROFILE_REGISTERED: + return + profile = HarnessProfile(excluded_tools=ASK_ALYF_EXCLUDED_TOOLS) + register_harness_profile("openai", profile) + _ASK_ALYF_PROFILE_REGISTERED = True def _get_api_key_from_settings(settings) -> str: @@ -64,1632 +61,28 @@ def _get_api_key_from_settings(settings) -> str: return api_key -def _get_model_id_from_settings(settings) -> str: - model_id = (settings.model or "").strip() - if not model_id: +def _get_model_name_from_settings(settings) -> str: + model_name = (settings.model or "").strip() + if not model_name: frappe.throw(_("Configure a model in Ask ALYF Settings before sending messages.")) + return model_name - try: - AnyLLM.split_model_provider(model_id) - return model_id - except ValueError: - pass - - if settings.llm_provider in {"OpenAI", "OpenAI Compatible"}: - return f"openai:{model_id}" - return model_id +def build_chat_model(settings, *, temperature: float = 0.2) -> ChatOpenAI: + """Build a LangChain `ChatOpenAI` from Ask ALYF Settings values. - -def _build_internal_agent_config( - *, - settings, - name: str, - instructions: str, - tool_defs: list[Callable[..., Any]], -): - return AgentConfig( - name=name, - model_id=_get_model_id_from_settings(settings), - api_base=(settings.base_url or "").strip() or None, + Supports OpenAI and OpenAI-compatible `base_url` configurations by + preserving the existing settings-driven model/api_key/base_url resolution. + """ + return ChatOpenAI( + model=_get_model_name_from_settings(settings), api_key=_get_api_key_from_settings(settings), - instructions=instructions, - tools=tool_defs, - model_args={"temperature": 0.1}, - ) - - -async def _create_internal_agent_async( - *, - settings, - name: str, - instructions: str, - tool_defs: list[Callable[..., Any]], -): - return await AnyAgent.create_async( - AgentFramework.TINYAGENT, - _build_internal_agent_config( - settings=settings, - name=name, - instructions=instructions, - tool_defs=tool_defs, - ), + base_url=(settings.base_url or "").strip() or None, + temperature=temperature, ) -def _build_specialist_history_context(runtime: ask_alyfRuntime, limit: int = 6) -> str: - history = getattr(runtime, "conversation_history", None) or [] - if not history: - return "" - - lines = [_("Recent conversation context:")] - for item in history[-limit:]: - lines.extend(_build_history_item_lines(item)) - return "\n".join(lines) - - -def _parse_json_object_output(raw_output: Any) -> dict[str, Any] | None: - if isinstance(raw_output, dict): - return raw_output - if not isinstance(raw_output, str): - return None - - text = raw_output.strip() - if not text: - return None - - candidates = [text] - fenced_match = re.search(r"```(?:json)?\s*(\{.*\})\s*```", text, flags=re.DOTALL) - if fenced_match: - candidates.insert(0, fenced_match.group(1).strip()) - - start = text.find("{") - end = text.rfind("}") - if start != -1 and end != -1 and start < end: - candidates.append(text[start : end + 1]) - - for candidate in candidates: - try: - parsed = json.loads(candidate) - except Exception: - continue - if isinstance(parsed, dict): - return parsed - - return None - - -def _coerce_string_list(value: Any, *, limit: int = 12) -> list[str]: - if not isinstance(value, list): - return [] - - items: list[str] = [] - for entry in value: - if not isinstance(entry, str): - continue - text = entry.strip() - if not text: - continue - items.append(text[:500]) - if len(items) >= limit: - break - - return items - - -def _coerce_evidence_entries(value: Any) -> list[dict[str, Any]]: - if not isinstance(value, list): - return [] - - evidence: list[dict[str, Any]] = [] - for entry in value[:12]: - if not isinstance(entry, dict): - continue - - path = entry.get("path") - if not isinstance(path, str) or not path.strip(): - continue - - start_line = entry.get("start_line") - end_line = entry.get("end_line") - note = entry.get("note") or entry.get("reason") or "" - item: dict[str, Any] = {"path": path.strip()} - if isinstance(start_line, int) and start_line > 0: - item["start_line"] = start_line - if isinstance(end_line, int) and end_line > 0: - item["end_line"] = end_line - if isinstance(note, str) and note.strip(): - item["note"] = note.strip()[:500] - evidence.append(item) - - return evidence - - -def _normalize_source_code_analysis(result: dict[str, Any], *, raw_output: str = "") -> dict[str, Any]: - answer = result.get("answer") - summary = result.get("summary") - uncertainty = result.get("uncertainty") - - answer_text = answer.strip() if isinstance(answer, str) else "" - summary_text = summary.strip() if isinstance(summary, str) else "" - uncertainty_text = uncertainty.strip() if isinstance(uncertainty, str) else "" - evidence = _coerce_evidence_entries(result.get("evidence")) - searched_paths = _coerce_string_list(result.get("searched_paths")) - - if not answer_text: - answer_text = summary_text or raw_output.strip() - if not summary_text: - summary_text = answer_text - - return { - "answer": answer_text, - "summary": summary_text, - "evidence": evidence, - "uncertainty": uncertainty_text, - "searched_paths": searched_paths, - } - - -def _build_document_planner_failure( - message: str, - *, - recommended_tool: str = "", - payload: dict[str, Any] | None = None, -) -> dict[str, Any]: - return { - "ready": False, - "recommended_tool": recommended_tool if recommended_tool in ALLOWED_DOCUMENT_PLANNER_TOOLS else "", - "payload": payload or {}, - "reason": "", - "missing_information": [message], - "checks": [], - "warnings": [], - } - - -def _normalize_document_plan( - result: dict[str, Any], - *, - default_doctype: str = "", - default_operation: str = "", - default_name: str = "", - raw_output: str = "", -) -> dict[str, Any]: - requested_operation = default_operation.strip().lower() - recommended_tool = result.get("recommended_tool") - if isinstance(recommended_tool, str): - recommended_tool = recommended_tool.strip().lower() - else: - recommended_tool = "" - - if recommended_tool not in ALLOWED_DOCUMENT_PLANNER_TOOLS: - recommended_tool = ( - requested_operation if requested_operation in ALLOWED_DOCUMENT_PLANNER_TOOLS else "" - ) - - payload = result.get("payload") - payload = payload if isinstance(payload, dict) else {} - if default_doctype and "doctype" not in payload: - payload["doctype"] = default_doctype - if default_name and recommended_tool in {"save", "set_value"} and "name" not in payload: - payload["name"] = default_name - - reason = result.get("reason") - reason_text = reason.strip()[:1000] if isinstance(reason, str) else "" - missing_information = _coerce_string_list(result.get("missing_information")) - checks = _coerce_string_list(result.get("checks")) - warnings = _coerce_string_list(result.get("warnings")) - ready = bool(result.get("ready")) and not missing_information and bool(recommended_tool) - - if recommended_tool in {"insert", "save"}: - values = payload.get("values") - if not isinstance(values, dict): - payload["values"] = {} - ready = False - warnings.append("DocumentPlanner did not return an object in payload.values.") - - if recommended_tool == "save": - name = payload.get("name") - if not isinstance(name, str) or not name.strip(): - ready = False - missing_information.append("Target document name is missing.") - - if recommended_tool == "set_value": - fieldname = payload.get("fieldname") - name = payload.get("name") - if not isinstance(fieldname, str) or not fieldname.strip(): - ready = False - missing_information.append("Target fieldname is missing.") - if "value" not in payload: - ready = False - missing_information.append("Target field value is missing.") - if not isinstance(name, str) or not name.strip(): - ready = False - missing_information.append("Target document name is missing.") - - doctype = payload.get("doctype") - if recommended_tool and (not isinstance(doctype, str) or not doctype.strip()): - ready = False - missing_information.append("Target DocType is missing.") - - if raw_output and not result: - warnings.append("DocumentPlanner returned invalid JSON.") - - return { - "ready": ready, - "recommended_tool": recommended_tool, - "payload": payload, - "reason": reason_text, - "missing_information": list(dict.fromkeys(missing_information)), - "checks": checks, - "warnings": list(dict.fromkeys(warnings)), - } - - -class ask_alyfToolset: - def __init__(self, runtime: ask_alyfRuntime, settings=None): - self.runtime = runtime - self.settings = settings - self._source_code_analyzer = None - self._document_planner = None - - def _get_settings(self): - if self.settings is None: - self.settings = tools.get_settings() - return self.settings - - def _get_source_code_analyzer(self): - if self._source_code_analyzer is None: - self._source_code_analyzer = SourceCodeAnalyzer( - runtime=self.runtime, - settings=self._get_settings(), - toolset=self, - ) - return self._source_code_analyzer - - def _get_document_planner(self): - if self._document_planner is None: - self._document_planner = DocumentPlanner( - runtime=self.runtime, - settings=self._get_settings(), - toolset=self, - ) - return self._document_planner - - def _proposal( - self, - kind: str, - tool: str, - summary: str, - reason: str = "", - *, - validation_error_status: str, - prepared_status: str, - requires_confirmation: bool = True, - **payload: Any, - ) -> dict[str, Any]: - """Create a pending operation proposal and append it to the list.""" - validation_error = tools.validate_pending_operation_payload(kind, tool, payload) - if validation_error: - self.runtime.emit_status(validation_error_status) - return { - "success": False, - "requires_confirmation": False, - "error": validation_error, - } - - proposal = { - "kind": kind, - "tool": tool, - "summary": summary, - "reason": reason, - "requires_confirmation": bool(requires_confirmation), - "payload": payload, - "call_id": uuid4().hex, - } - self.runtime.pending_operations.append(proposal) - self.runtime.emit_status(prepared_status) - return { - "success": True, - "requires_confirmation": bool(requires_confirmation), - "proposal": proposal, - } - - def _backend_proposal( - self, - tool: str, - summary: str, - reason: str = "", - *, - validation_error_status: str, - prepared_status: str, - **payload: Any, - ) -> dict[str, Any]: - return self._proposal( - tools.OPERATION_KIND_BACKEND, - tool, - summary, - reason, - validation_error_status=validation_error_status, - prepared_status=prepared_status, - requires_confirmation=True, - **payload, - ) - - def _frontend_proposal( - self, - tool: str, - summary: str, - reason: str = "", - *, - validation_error_status: str, - prepared_status: str, - requires_confirmation: bool | None = None, - **payload: Any, - ) -> dict[str, Any]: - if requires_confirmation is None: - requires_confirmation = tools.get_frontend_action_requires_confirmation(tool) - return self._proposal( - tools.OPERATION_KIND_FRONTEND, - tool, - summary, - reason, - validation_error_status=validation_error_status, - prepared_status=prepared_status, - requires_confirmation=requires_confirmation, - **payload, - ) - - def get_list( - self, - doctype: str, - fields: str | list[str] | None = None, - filters: dict[str, Any] | list[Any] | None = None, - order_by: str | None = None, - limit: int = 20, - group_by: str | None = None, - ) -> list[dict[str, Any]]: - """List documents with filters, fields, ordering, and optional grouping. - - Args: - doctype: The DocType to query. - fields: Optional field name or field list to return. - filters: Optional Frappe filters. - order_by: Optional ordering expression. - limit: Maximum number of rows to return. - group_by: Optional SQL group by expression. - - Returns: - A list of matching documents. - """ - self.runtime.emit_status(_("Fetching list...")) - return tools.get_list( - doctype=doctype, - fields=fields, - filters=filters, - order_by=order_by, - limit=limit, - group_by=group_by, - ) - - def get_count( - self, - doctype: str, - filters: dict[str, Any] | list[Any] | None = None, - ) -> int: - """Count documents that match the given filters. - - Args: - doctype: The DocType to query. - filters: Optional Frappe filters. - - Returns: - The number of matching documents. - """ - self.runtime.emit_status(_("Counting documents...")) - return tools.get_count(doctype=doctype, filters=filters) - - def get( - self, - doctype: str, - name: str | None = None, - filters: dict[str, Any] | None = None, - ) -> dict[str, Any]: - """Read a single document by name or filters. - - Args: - doctype: The DocType to query. - name: Optional document name. - filters: Optional filters when name is not known. - - Returns: - The matching document. - """ - self.runtime.emit_status(_("Fetching document...")) - return tools.get_document(doctype=doctype, name=name, filters=filters) - - def get_value( - self, - doctype: str, - fieldname: str | list[str], - filters: dict[str, Any] | list[Any] | str | None = None, - ) -> Any: - """Read one or more field values from a document. - - Args: - doctype: The DocType to query. - fieldname: One field name or a list of field names. - filters: Name or filters to identify the record. - - Returns: - The requested value or values. - """ - self.runtime.emit_status(_("Fetching value...")) - return tools.get_value(doctype=doctype, fieldname=fieldname, filters=filters) - - def get_single_value(self, doctype: str, field: str) -> Any: - """Read a field value from a Single DocType. - - Args: - doctype: The Single DocType to query. - field: The field name to read. - - Returns: - The field value. - """ - self.runtime.emit_status(_("Fetching single value...")) - return tools.get_single_value(doctype=doctype, field=field) - - def get_meta(self, doctype: str) -> dict[str, Any]: - """Inspect metadata, fields, and permissions for a DocType. - - Args: - doctype: The DocType to inspect. - - Returns: - A metadata dictionary for the DocType. - """ - self.runtime.emit_status(_("Loading metadata...")) - return tools.get_meta(doctype=doctype) - - def has_permission(self, doctype: str, docname: str, perm_type: str = "read") -> dict[str, bool]: - """Check whether the current user has a specific permission on a document. - - Args: - doctype: The DocType to check. - docname: The document name. - perm_type: The permission type to evaluate. - - Returns: - A dictionary containing the boolean permission result. - """ - self.runtime.emit_status(_("Checking permissions...")) - return tools.has_permission(doctype=doctype, docname=docname, perm_type=perm_type) - - def get_doc_permissions(self, doctype: str, docname: str) -> dict[str, Any]: - """Get the evaluated permission map for a document. - - Args: - doctype: The DocType to check. - docname: The document name. - - Returns: - The evaluated permission dictionary. - """ - self.runtime.emit_status(_("Evaluating permissions...")) - return tools.get_doc_permissions(doctype=doctype, docname=docname) - - def list_accessible_doctypes(self, permission_type: str = "read") -> list[str]: - """List DocTypes that the current user can access. - - Args: - permission_type: The permission type to test. - - Returns: - A list of DocType names. - """ - self.runtime.emit_status(_("Listing accessible DocTypes...")) - return tools.list_accessible_doctypes(permission_type=permission_type) - - def list_accessible_reports(self) -> list[dict[str, Any]]: - """List reports that the current user can access. - - Returns: - A list of report metadata dictionaries. - """ - self.runtime.emit_status(_("Listing accessible reports...")) - return tools.list_accessible_reports() - - def translate_ui_labels( - self, - labels: list[str], - language: str | None = None, - ) -> dict[str, Any]: - """Translate UI labels so responses match what the user sees on screen. - - Use this whenever request context language is not English before mentioning - button names, tab names, DocType labels, field labels, menu items, or status labels. - - Args: - labels: English UI labels to translate. - language: Optional target language code (defaults to request context language). - - Returns: - A dictionary with the resolved language and translated labels. - """ - self.runtime.emit_status(_("Translating UI labels...")) - request_language = self.runtime.request_context.get("lang") or self.runtime.request_context.get( - "locale" - ) - return tools.translate_ui_labels(labels=labels, language=language or request_language) - - def search_code(self, query: str, relative_path: str = "", limit: int = 20) -> list[dict[str, Any]]: - """Search installed app code for matching text. - - Args: - query: The text to search for. - relative_path: Optional installed-app-relative path. Providing at least the app name will make it much faster. - limit: Maximum number of matches to return. - - Returns: - A list of code search matches. - """ - self.runtime.emit_status(_("Searching code...")) - return tools.search_code(query=query, relative_path=relative_path, limit=limit) - - def read_code_file(self, path: str, start_line: int = 1, end_line: int = 200) -> dict[str, Any]: - """Read a file from installed app code. - - Args: - path: Bench-relative path inside an installed app, such as apps/my_app/my_app/module.py. - start_line: First line to include. - end_line: Last line to include. - - Returns: - The selected file content and line range. - """ - self.runtime.emit_status(_("Reading code file...")) - return tools.read_code_file(path=path, start_line=start_line, end_line=end_line) - - def ls( - self, - app_name: str, - relative_path: str = "", - recursive: bool = False, - include_hidden: bool = False, - limit: int = 200, - ) -> dict[str, Any]: - """List files or directories in an installed app, similar to Debian ls. - - Args: - app_name: The installed app name. - relative_path: Optional path inside the app. - recursive: Whether to include nested descendants. - include_hidden: Whether to include hidden files and folders. - limit: Maximum number of entries to return. - - Returns: - A directory listing payload. - """ - self.runtime.emit_status(_("Listing code files...")) - return tools.ls( - app_name=app_name, - relative_path=relative_path, - recursive=recursive, - include_hidden=include_hidden, - limit=limit, - ) - - def find( - self, - app_name: str, - name_pattern: str = "*", - relative_path: str = "", - entry_type: str = "any", - include_hidden: bool = False, - limit: int = 200, - ) -> dict[str, Any]: - """Find files or directories in an installed app, similar to Debian find. - - Args: - app_name: The installed app name. - name_pattern: Shell-style filename pattern, such as *.py. - relative_path: Optional path inside the app to search from. - entry_type: One of any, file, or directory. - include_hidden: Whether to include hidden files and folders. - limit: Maximum number of matches to return. - - Returns: - A find-style search payload. - """ - self.runtime.emit_status(_("Finding code paths...")) - return tools.find( - app_name=app_name, - name_pattern=name_pattern, - relative_path=relative_path, - entry_type=entry_type, - include_hidden=include_hidden, - limit=limit, - ) - - def grep( - self, - app_name: str, - query: str, - relative_path: str = "", - file_pattern: str = "*", - case_sensitive: bool = False, - include_hidden: bool = False, - limit: int = 50, - ) -> dict[str, Any]: - """Search file contents in an installed app, similar to Debian grep. - - Args: - app_name: The installed app name. - query: The text to search for. - relative_path: Optional path inside the app to search from. - file_pattern: Optional shell-style filename filter, such as *.py. - case_sensitive: Whether matching should be case-sensitive. - include_hidden: Whether to include hidden files and folders. - limit: Maximum number of matches to return. - - Returns: - A grep-style search payload. - """ - self.runtime.emit_status(_("Searching file contents...")) - return tools.grep( - app_name=app_name, - query=query, - relative_path=relative_path, - file_pattern=file_pattern, - case_sensitive=case_sensitive, - include_hidden=include_hidden, - limit=limit, - ) - - async def source_code_analyzer(self, question: str, relative_path: str = "") -> dict[str, Any]: - """Analyze installed app source code via a restricted specialist agent. - - Use this for code questions that require searching, reading, and interpreting - multiple files. The specialist only has access to read-only code tools. - - Args: - question: The code question to investigate. - relative_path: Optional installed-app-relative path to narrow the search. - - Returns: - A structured summary with an answer and supporting evidence. - """ - self.runtime.emit_status(_("Delegating code analysis...")) - return await self._get_source_code_analyzer().analyze( - question=question, - relative_path=relative_path, - ) - - def get_file_id( - self, - reference_doctype: str, - reference_name: str, - reference_field: str = "", - file_url: str = "", - file_name: str = "", - ) -> str: - """Resolve a unique File ID from attachment reference filters. - - Args: - reference_doctype: The DocType the file is attached to. - reference_name: The document name the file is attached to. - reference_field: Optional attachment field name. - file_url: Optional file URL to disambiguate matches. - file_name: Optional file name to disambiguate matches. - - Returns: - The matching File ID. - """ - self.runtime.emit_status(_("Resolving file ID...")) - return tools.get_file_id( - reference_doctype=reference_doctype, - reference_name=reference_name, - reference_field=reference_field, - file_url=file_url, - file_name=file_name, - ) - - def read_file_record(self, file_id: str) -> dict[str, Any]: - """Read the content of a File record the user can access. - - Args: - file_id: The File ID (`name`) to read. - - Returns: - The file metadata and content. - """ - self.runtime.emit_status(_("Reading file...")) - return tools.read_file_record(file_id=file_id) - - async def extract_document_data(self, file_id: str, extraction_prompt: str = "") -> dict[str, Any]: - """Extract structured data from a PDF or image file using vision AI. - - Call this tool when a user uploads or references a document (invoice, receipt, - contract, etc.) and wants to extract information from it. The file is rendered - as images and sent to a vision-capable model that reads text, tables, and layouts. - - Supports PDF files (up to 10 pages) and images (PNG, JPG, GIF, WebP). - - Args: - file_id: The File ID (`name`) to process. - extraction_prompt: Optional instructions for what data to extract. - If omitted, a general-purpose extraction prompt is used. - - Returns: - A dictionary with the file ID, file name, number of pages processed, - and the extracted data as a JSON object. - """ - self.runtime.emit_status(_("Extracting document data...")) - result = await tools.extract_document_data(file_id=file_id, extraction_prompt=extraction_prompt) - self.runtime.remember_document_extraction(result, extraction_prompt=extraction_prompt) - return result - - def get_print( - self, - doctype: str, - name: str, - print_format: str = "", - letterhead: str = "", - language: str = "", - ) -> dict[str, Any]: - """Generate a PDF print of a document and attach it to the conversation. - - Uses the default print format and default letter head unless overridden. - Permissions are enforced automatically by the tool. - - The generated PDF is automatically shown to the user as a clickable file - attachment in the conversation UI. Do not repeat the file name, URL, or - link in your text — just confirm briefly what was printed. - - Args: - doctype: The DocType of the document to print. - name: The document name. - print_format: Optional print format name. Defaults to the DocType's default. - letterhead: Optional letter head name. Defaults to the site's default. - language: Optional language code for the print. Defaults to the document's - language field if it exists, otherwise the current session language. - - Returns: - A dictionary with the generated file metadata (name, file_name, file_url). - """ - self.runtime.emit_status(_("Generating print...")) - file_entry = tools.get_print( - doctype=doctype, - name=name, - conversation_name=self.runtime.conversation_name, - print_format=print_format, - letterhead=letterhead, - language=language, - ) - self.runtime.remember_attached_file(file_entry) - return file_entry - - def run_read_only_sql(self, query: str) -> list[dict[str, Any]]: - """Run a read-only SQL query when the current user is allowed to do so. - - Args: - query: A single read-only SQL query. - - Returns: - The SQL result rows. - """ - self.runtime.emit_status(_("Running SQL query...")) - return tools.run_read_only_sql(query=query) - - def get_app_version(self, app_name: str) -> str: - """Read the installed version for an app. - - Args: - app_name: The installed app name. - - Returns: - The app version string. - """ - self.runtime.emit_status(_("Reading app version...")) - return tools.get_app_version(app_name=app_name) - - def read_github_releases(self, app_name: str, limit: int = 5) -> list[dict[str, Any]]: - """Read recent GitHub releases for an installed app. - - Args: - app_name: The installed app name. - limit: Maximum number of releases to return. - - Returns: - A list of release dictionaries. - """ - self.runtime.emit_status(_("Reading GitHub releases...")) - return tools.read_github_releases(app_name=app_name, limit=limit) - - def read_documentation_page(self, app_name: str, relative_path: str = "") -> dict[str, Any]: - """Fetch a documentation page for an installed app. - - Args: - app_name: The installed app name. - relative_path: Optional path relative to the configured docs URL. - - Returns: - A documentation payload containing the page content. - """ - self.runtime.emit_status(_("Reading documentation...")) - return tools.read_documentation_page(app_name=app_name, relative_path=relative_path) - - def read_skill(self, name: str) -> dict[str, str]: - """Read the full content of an Ask ALYF skill available to the current user. - - Use this when the instructions list a skill that seems relevant. Pass the - exact skill `name` shown in that list. - - Args: - name: The exact Ask ALYF Skill name. - - Returns: - A dictionary containing the skill name, title, and markdown description. - """ - self.runtime.emit_status(_("Reading skill...")) - skill_doc = get_accessible_skill_doc(name) - return { - "name": skill_doc.name, - "title": skill_doc.title, - "description": skill_doc.description or "", - } - - def write_skill( - self, - title: str, - description: str, - roles: list[str], - reason: str = "", - ) -> dict[str, Any]: - """Propose creating a reusable Ask ALYF skill. - - Use this when the user wants to save durable instructions that can later be - loaded with `read_skill`. - - Args: - title: The skill title shown to users. - description: The markdown skill content. - roles: Roles that should be allowed to use the skill. - Accepts a list or a comma/newline-separated string. - reason: Optional explanation of why the skill should be created. - - Returns: - A pending action proposal that requires confirmation. - """ - clean_title = (title or "").strip() - clean_description = (description or "").strip() - clean_roles = parse_newline_list(roles) - - if not clean_title: - frappe.throw(_("Skill title is required.")) - if not clean_description: - frappe.throw(_("Skill description is required.")) - if not clean_roles: - frappe.throw(_("At least one role is required.")) - - return self._backend_proposal( - "insert", - _("Create skill '{0}'").format(clean_title), - reason, - validation_error_status=_("Skill proposal needs correction."), - prepared_status=_("Prepared skill proposal."), - doctype="Ask ALYF Skill", - values={ - "title": clean_title, - "description": clean_description, - "roles": [{"role": role} for role in clean_roles], - }, - ) - - async def document_planner( - self, - user_request: str, - doctype: str = "", - operation: str = "insert", - name: str = "", - values_hint: dict[str, Any] | None = None, - ) -> dict[str, Any]: - """Plan a document create or update flow via a read-only specialist agent. - - Use this before non-trivial insert, save, or set_value operations that need - schema inspection or Link resolution. The specialist only has read tools. - - Args: - user_request: The user's requested outcome in natural language. - doctype: The target DocType to plan for. - operation: One of insert, save, or set_value. - name: Optional existing document name for save or set_value. - values_hint: Optional partial field values already known. - - Returns: - A structured plan with readiness, payload, missing information, and checks. - """ - self.runtime.emit_status(_("Planning document change...")) - return await self._get_document_planner().plan( - user_request=user_request, - doctype=doctype, - operation=operation, - name=name, - values_hint=values_hint, - ) - - def insert(self, doctype: str, values: dict[str, Any], reason: str = "") -> dict[str, Any]: - """Propose creating a new document. Use get_meta first to know the schema. - For child tables, values must be a list of row objects. - - Args: - doctype: The DocType to create. - values: The field values for the new document. - reason: Optional explanation of why this change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - return self._backend_proposal( - "insert", - _("Create {0}").format(_(doctype)), - reason, - validation_error_status=_("Create proposal needs correction."), - prepared_status=_("Prepared create proposal."), - doctype=doctype, - values=values, - ) - - def batch_insert(self, doctype: str, records: list[dict[str, Any]], reason: str = "") -> dict[str, Any]: - """Propose creating multiple documents of the same DocType. - Use get_meta first to know the schema for each record. - For child tables, each record must use a list of row objects. - - Args: - doctype: The DocType to create. - records: A list of field-value dictionaries, one per new document. - reason: Optional explanation of why this batch change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - record_count = len(records) if isinstance(records, list) else 0 - return self._backend_proposal( - "batch_insert", - _("Create {0} {1} records").format(record_count, _(doctype)), - reason, - validation_error_status=_("Batch create proposal needs correction."), - prepared_status=_("Prepared batch create proposal."), - doctype=doctype, - records=records, - ) - - def save( - self, - doctype: str, - name: str, - values: dict[str, Any], - reason: str = "", - ) -> dict[str, Any]: - """Propose updating an existing document. - For child tables, values must be a list of row objects. - - Args: - doctype: The DocType to update. - name: The document name. - values: The fields to update. - reason: Optional explanation of why this change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - return self._backend_proposal( - "save", - _("Update {0} {1}").format(_(doctype), name), - reason, - validation_error_status=_("Update proposal needs correction."), - prepared_status=_("Prepared update proposal."), - doctype=doctype, - name=name, - values=values, - ) - - def set_value( - self, - doctype: str, - name: str, - fieldname: str, - value: Any, - reason: str = "", - ) -> dict[str, Any]: - """Propose setting a single field on a document. - - Args: - doctype: The DocType to update. - name: The document name. - fieldname: The field to set. - value: The new value. - reason: Optional explanation of why this change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - return self._backend_proposal( - "set_value", - _("Set {0} on {1} {2}").format(fieldname, _(doctype), name), - reason, - validation_error_status=_("Set value proposal needs correction."), - prepared_status=_("Prepared set value proposal."), - doctype=doctype, - name=name, - fieldname=fieldname, - value=value, - ) - - def submit(self, doctype: str, name: str, reason: str = "") -> dict[str, Any]: - """Propose submitting a document. - - Args: - doctype: The DocType to submit. - name: The document name. - reason: Optional explanation of why this change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - return self._backend_proposal( - "submit", - _("Submit {0} {1}").format(_(doctype), name), - reason, - validation_error_status=_("Submit proposal needs correction."), - prepared_status=_("Prepared submit proposal."), - doctype=doctype, - name=name, - ) - - def cancel(self, doctype: str, name: str, reason: str = "") -> dict[str, Any]: - """Propose cancelling a document. - - Args: - doctype: The DocType to cancel. - name: The document name. - reason: Optional explanation of why this change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - return self._backend_proposal( - "cancel", - _("Cancel {0} {1}").format(_(doctype), name), - reason, - validation_error_status=_("Cancel proposal needs correction."), - prepared_status=_("Prepared cancel proposal."), - doctype=doctype, - name=name, - ) - - def amend(self, doctype: str, name: str, reason: str = "") -> dict[str, Any]: - """Propose amending a cancelled document. - - Args: - doctype: The DocType to amend. - name: The document name. - reason: Optional explanation of why this change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - return self._backend_proposal( - "amend", - _("Amend {0} {1}").format(_(doctype), name), - reason, - validation_error_status=_("Amend proposal needs correction."), - prepared_status=_("Prepared amend proposal."), - doctype=doctype, - name=name, - ) - - def delete(self, doctype: str, name: str, reason: str = "") -> dict[str, Any]: - """Propose deleting a document. - - Args: - doctype: The DocType to delete. - name: The document name. - reason: Optional explanation of why this change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - return self._backend_proposal( - "delete", - _("Delete {0} {1}").format(_(doctype), name), - reason, - validation_error_status=_("Delete proposal needs correction."), - prepared_status=_("Prepared delete proposal."), - doctype=doctype, - name=name, - ) - - def rename_doc( - self, - doctype: str, - name: str, - new_name: str, - merge: bool = False, - reason: str = "", - ) -> dict[str, Any]: - """Propose renaming a document. - - Args: - doctype: The DocType to rename. - name: The current document name. - new_name: The target document name. - merge: Whether to merge into an existing target. - reason: Optional explanation of why this change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - return self._backend_proposal( - "rename_doc", - _("Rename {0} {1} to {2}").format(_(doctype), name, new_name), - reason, - validation_error_status=_("Rename proposal needs correction."), - prepared_status=_("Prepared rename proposal."), - doctype=doctype, - name=name, - new_name=new_name, - merge=merge, - ) - - def attach_file( - self, - doctype: str, - name: str, - file_id: str, - reason: str = "", - ) -> dict[str, Any]: - """Propose attaching an existing file to a document. - - Args: - doctype: The DocType to update. - name: The document name. - file_id: The File ID (`name`) to attach. - reason: Optional explanation of why this change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - file_doc = tools._get_accessible_file_doc(file_id=file_id) - return self._backend_proposal( - "attach_file", - _("Attach file {0} to {1} {2}").format( - _build_file_markdown_link(file_doc), - _(doctype), - name, - ), - reason, - validation_error_status=_("Attach file proposal needs correction."), - prepared_status=_("Prepared attach file proposal."), - doctype=doctype, - name=name, - file_id=file_id, - ) - - def run_whitelisted_method( - self, - method: str, - args: dict[str, Any] | None = None, - reason: str = "", - ) -> dict[str, Any]: - """Propose calling a whitelisted method. - - Args: - method: The dotted Python path of the whitelisted method. - args: Optional keyword arguments for the method. - reason: Optional explanation of why this change is needed. - - Returns: - A pending action proposal that requires confirmation. - """ - return self._backend_proposal( - "run_method", - _("Call {0}").format(method), - reason, - validation_error_status=_("Method call proposal needs correction."), - prepared_status=_("Prepared method call proposal."), - method=method, - args=args or {}, - ) - - def set_route(self, route: list[str], reason: str = "") -> dict[str, Any]: - """Propose navigating to a Desk route on the frontend. - - Args: - route: Route parts used by frappe.set_route. - reason: Optional explanation of why this navigation helps the user. - - Returns: - A pending frontend operation proposal. - """ - route_label = "/".join(part for part in (route or []) if isinstance(part, str)) - return self._frontend_proposal( - "set_route", - _("Navigate to {0}").format(route_label or _("target route")), - reason, - validation_error_status=_("Route action needs correction."), - prepared_status=_("Prepared route action."), - route=route, - ) - - def new_doc( - self, - doctype: str, - route_options: dict[str, Any] | None = None, - reason: str = "", - ) -> dict[str, Any]: - """Propose opening a new document form in the frontend. - - Args: - doctype: The target DocType for frappe.new_doc. - route_options: Optional route options to prefill form values. - reason: Optional explanation of why this action helps the user. - - Returns: - A pending frontend operation proposal. - """ - return self._frontend_proposal( - "new_doc", - _("Open new {0}").format(_(doctype)), - reason, - validation_error_status=_("New document action needs correction."), - prepared_status=_("Prepared new document action."), - doctype=doctype, - route_options=route_options or {}, - ) - - def scroll_to_field(self, fieldname: str, reason: str = "") -> dict[str, Any]: - """Propose scrolling to a field on the active form. - - Args: - fieldname: The fieldname to scroll to. - reason: Optional explanation of why this action helps the user. - - Returns: - A pending frontend operation proposal. - """ - return self._frontend_proposal( - "scroll_to_field", - _("Scroll to field {0}").format(fieldname), - reason, - validation_error_status=_("Scroll action needs correction."), - prepared_status=_("Prepared scroll action."), - fieldname=fieldname, - ) - - def frm_set_value( - self, - fieldname: str, - value: Any, - doctype: str | None = None, - docname: str | None = None, - reason: str = "", - ) -> dict[str, Any]: - """Propose setting a field value on the active frontend form. - - Args: - fieldname: The target fieldname. - value: The value to apply. - doctype: Optional expected active form DocType. - docname: Optional expected active form document name. - reason: Optional explanation of why this action helps the user. - - Returns: - A pending frontend operation proposal. - """ - payload: dict[str, Any] = {"fieldname": fieldname, "value": value} - if doctype: - payload["doctype"] = doctype - if docname: - payload["docname"] = docname - - return self._frontend_proposal( - "frm_set_value", - _("Set field {0} on current form").format(fieldname), - reason, - validation_error_status=_("Set field action needs correction."), - prepared_status=_("Prepared set field action."), - **payload, - ) - - def frm_add_child( - self, - fieldname: str, - values: dict[str, Any] | None = None, - doctype: str | None = None, - docname: str | None = None, - reason: str = "", - ) -> dict[str, Any]: - """Propose adding a child table row on the active frontend form. - - Args: - fieldname: The child table fieldname. - values: Optional row values for the new child row. - doctype: Optional expected active form DocType. - docname: Optional expected active form document name. - reason: Optional explanation of why this action helps the user. - - Returns: - A pending frontend operation proposal. - """ - payload: dict[str, Any] = {"fieldname": fieldname, "values": values or {}} - if doctype: - payload["doctype"] = doctype - if docname: - payload["docname"] = docname - - return self._frontend_proposal( - "frm_add_child", - _("Add a row to {0} on current form").format(fieldname), - reason, - validation_error_status=_("Add child row action needs correction."), - prepared_status=_("Prepared add child row action."), - **payload, - ) - - def show_chart( - self, - frappe_charts: list[dict[str, Any]], - reason: str = "", - ) -> dict[str, Any]: - """Show one or more Frappe Charts under this assistant message. - - The desk creates the DOM element; each item in `frappe_charts` is the `options` - argument to `new frappe.Chart(container, options)` (Frappe Charts on the client). - - Shape (one object per chart): - - `type`: bar, line, scatter, pie, percentage, donut, or axis-mixed - - `data`: `{ "labels": [...], "datasets": [ { "values": [...], "name"?: str, "type"?: "bar"|"line" } ] }` - — every `values` list must be the same length as `labels` - - Optional: `title`, `height` (0 for default; if set, at least 240 — Frappe Charts reserves ~130px chrome), - `colors` (array; empty for defaults), - `barOptions`, `lineOptions`, `axisOptions` (see Frappe Charts docs) - - Args: - frappe_charts: One or more chart option objects. - reason: Optional short note for the user. - - Returns: - A pending frontend operation proposal (auto-executed in the browser). - """ - count = len(frappe_charts) if isinstance(frappe_charts, list) else 0 - summary = _("Show chart") if count == 1 else _("Show {0} charts").format(count) - return self._frontend_proposal( - "show_chart", - summary, - reason, - validation_error_status=_("Chart action needs correction."), - prepared_status=_("Prepared chart display."), - requires_confirmation=False, - frappe_charts=frappe_charts, - ) - - -class SourceCodeAnalyzer: - def __init__(self, runtime: ask_alyfRuntime, settings, toolset: ask_alyfToolset): - self.runtime = runtime - self.settings = settings - self.tool_defs = [ - toolset.search_code, - toolset.read_code_file, - toolset.ls, - toolset.find, - toolset.grep, - ] - self.instructions = """ -You are SourceCodeAnalyzer, an internal Ask ALYF specialist for installed app code. - -You can only use the provided source-code tools. - -Rules: -- Search or list first, then read the smallest relevant file ranges. -- Prefer the narrowest path scope available. -- Do not answer from memory when the tools can verify it. -- If evidence is incomplete or ambiguous, say so clearly. -- Include bench-relative paths and line ranges in evidence whenever possible. -- Return a compact JSON object with keys: - - `answer` (string) - - `summary` (string) - - `evidence` (list of objects with `path`, optional `start_line`, optional `end_line`, and optional `note`) - - `uncertainty` (string) - - `searched_paths` (list of strings) - -Return only a valid JSON object. -Do not wrap the JSON in markdown fences. -Do not add explanatory prose before or after the JSON. -""".strip() - self.agent = None - - async def _get_agent(self): - if self.agent is None: - self.agent = await _create_internal_agent_async( - settings=self.settings, - name="SourceCodeAnalyzer", - instructions=self.instructions, - tool_defs=self.tool_defs, - ) - return self.agent - - async def analyze(self, question: str, relative_path: str = "") -> dict[str, Any]: - clean_question = (question or "").strip() - if not clean_question: - return _normalize_source_code_analysis({}) - - history_context = _build_specialist_history_context(self.runtime) - request_context = frappe.as_json(getattr(self.runtime, "request_context", {}), indent=2) - path_hint = (relative_path or "").strip() or _("all installed app code roots available to you") - prompt = "\n".join( - [ - "Investigate this source-code question for the parent Ask ALYF agent.", - f"Question: {clean_question}", - f"Preferred scope: {path_hint}", - "Use the preferred scope when it is specific enough.", - "", - "Request context JSON:", - request_context, - "", - history_context or _("No recent conversation context."), - ] - ) - agent = await self._get_agent() - trace = await agent.run_async(prompt) - raw_output = str(trace.final_output or "").strip() - parsed = _parse_json_object_output(raw_output) or {} - return _normalize_source_code_analysis(parsed, raw_output=raw_output) - - -class DocumentPlanner: - def __init__(self, runtime: ask_alyfRuntime, settings, toolset: ask_alyfToolset): - self.runtime = runtime - self.settings = settings - self.tool_defs = [ - toolset.get_list, - toolset.get, - toolset.get_value, - toolset.get_single_value, - toolset.get_meta, - toolset.has_permission, - toolset.get_doc_permissions, - toolset.list_accessible_doctypes, - ] - self.instructions = f""" -You are DocumentPlanner, an internal Ask ALYF specialist for planning Frappe document changes. - -You only have read-only access to metadata and documents. You never execute writes. - -You may only plan these operations: -- `insert` -- `save` -- `set_value` - -Rules: -- Always inspect `get_meta` before planning `insert`, `save`, or `set_value`. -- Use the read tools to resolve Link targets or confirm existing values when possible. -- Never invent document names, Link targets, or required values. -- Treat `values_hint` as tentative until it is confirmed by the user or by a read tool. -- If information is missing, set `ready` to `false` and list each missing item in `missing_information`. -- The `payload` must match the parent tool signature for the recommended operation. -- Return a JSON object with keys: - - `ready` (boolean) - - `recommended_tool` (`insert`, `save`, or `set_value`) - - `payload` (object) - - `reason` (string) - - `missing_information` (list of strings) - - `checks` (list of strings) - - `warnings` (list of strings) - -{SPECIALIST_JSON_OUTPUT_INSTRUCTION} -""".strip() - self.agent = None - - async def _get_agent(self): - if self.agent is None: - self.agent = await _create_internal_agent_async( - settings=self.settings, - name="DocumentPlanner", - instructions=self.instructions, - tool_defs=self.tool_defs, - ) - return self.agent - - async def plan( - self, - user_request: str, - doctype: str = "", - operation: str = "insert", - name: str = "", - values_hint: dict[str, Any] | None = None, - ) -> dict[str, Any]: - clean_doctype = (doctype or "").strip() - clean_operation = (operation or "").strip().lower() or "insert" - clean_name = (name or "").strip() - clean_request = (user_request or "").strip() - clean_values_hint = values_hint if isinstance(values_hint, dict) else {} - - if not clean_doctype: - return _build_document_planner_failure("Target DocType is required.") - if clean_operation not in ALLOWED_DOCUMENT_PLANNER_TOOLS: - return _build_document_planner_failure( - "Operation must be one of insert, save, or set_value.", - recommended_tool="insert", - payload={"doctype": clean_doctype}, - ) - if not clean_request: - return _build_document_planner_failure( - "User request is required.", - recommended_tool=clean_operation, - payload={"doctype": clean_doctype}, - ) - - history_context = _build_specialist_history_context(self.runtime) - request_context = frappe.as_json(getattr(self.runtime, "request_context", {}), indent=2) - prompt = "\n".join( - [ - "Plan the next document action for the parent Ask ALYF agent.", - f"User request: {clean_request}", - f"Requested operation: {clean_operation}", - f"Target DocType: {clean_doctype}", - f"Target document name: {clean_name or '(not provided)'}", - "", - "values_hint JSON:", - frappe.as_json(clean_values_hint, indent=2), - "", - "Request context JSON:", - request_context, - "", - history_context or _("No recent conversation context."), - ] - ) - agent = await self._get_agent() - trace = await agent.run_async(prompt) - raw_output = str(trace.final_output or "").strip() - parsed = _parse_json_object_output(raw_output) or {} - return _normalize_document_plan( - parsed, - default_doctype=clean_doctype, - default_operation=clean_operation, - default_name=clean_name, - raw_output=raw_output, - ) - - -def _clear_messages_on_tool_error(func): - """Wrap a tool callable so that queued Frappe messages are discarded on - exception. The error still propagates to the agent framework (so the LLM - sees it), but the user won't receive a popup.""" - - if inspect.iscoroutinefunction(func): - - @functools.wraps(func) - async def async_wrapper(*args, **kwargs): - try: - return await func(*args, **kwargs) - except Exception: - frappe.clear_messages() - raise - - return async_wrapper - - @functools.wraps(func) - def wrapper(*args, **kwargs): - try: - return func(*args, **kwargs) - except Exception: - frappe.clear_messages() - raise - - return wrapper +# --- Deep Agents coordinator -------------------------------------------------- class ask_alyfAgentRunner: @@ -1697,28 +90,28 @@ def __init__(self, runtime: ask_alyfRuntime): self.runtime = runtime self.settings = tools.get_settings() self.toolset = ask_alyfToolset(runtime, settings=self.settings) - self.agent = AnyAgent.create( - AgentFramework.TINYAGENT, - AgentConfig( - name="Ask ALYF", - model_id=self._get_model_id(), - api_base=(self.settings.base_url or "").strip() or None, - api_key=self._get_api_key(), - instructions=self._build_instructions(), - tools=self._build_tools(), - model_args={"temperature": 0.2}, - ), + _ensure_ask_alyf_harness_profile() + self.model = build_chat_model(self.settings, temperature=0.2) + app_roots = tools.get_installed_app_roots() if self.settings.is_code_search_enabled() else {} + self.backend = build_ask_alyf_backend(app_roots) + self.agent = create_deep_agent( + model=self.model, + tools=self._build_tools(), + system_prompt=self._build_instructions(), + backend=self.backend, + subagents=self._build_subagents(), + permissions=self._build_permissions(), + name="ask_alyf", ) - def _get_api_key(self) -> str: - return _get_api_key_from_settings(self.settings) - - def _get_model_id(self) -> str: - return _get_model_id_from_settings(self.settings) - def _can_write_skill(self) -> bool: return bool(frappe.has_permission("Ask ALYF Skill", ptype="create")) + def _build_permissions(self) -> list[FilesystemPermission]: + # Defense-in-depth on top of the harness profile tool exclusions: even + # if a write tool somehow remained visible, the VFS denies every write. + return [FilesystemPermission(operations=["write"], paths=["/**"], mode="deny")] + def _build_instructions(self) -> str: context = frappe.as_json(self.runtime.request_context, indent=2) excluded_doctypes = ", ".join(sorted(tools.get_excluded_doctypes())) or "None" @@ -1727,8 +120,9 @@ def _build_instructions(self) -> str: code_search_usage_instruction = "" if self.settings.is_code_search_enabled(): code_search_usage_instruction = ( - "\n- When code search is enabled, use `source_code_analyzer` for code questions " - "instead of reasoning from memory." + "\n- When code search is enabled, delegate code questions to the " + "`source-code-analyzer` subagent via the `task` tool instead of " + "reasoning from memory." ) base_instructions = f""" @@ -1756,7 +150,7 @@ def _build_instructions(self) -> str: - `Agent` mode supports mutation workflows with write tools while still handling read-only questions with read tools. Every write tool call creates a pending proposal that requires user confirmation before execution. Multiple proposals can be created in a single turn — if the request needs several writes, propose them all now. The user will confirm or reject each one individually. - Frontend action tools can navigate or adjust the current form in the browser, or display Frappe Charts under the assistant message via `show_chart` (pass `frappe_charts` as a list of chart option objects; validated server-side). See the `show_chart` tool docstring for the options shape. - Frontend actions with `requires_confirmation` must be confirmed before the browser executes them. -- In `Agent` mode, prefer `document_planner` before non-trivial `insert`, `save`, or `set_value` operations. If it returns `ready=false`, ask the user for the missing information instead of guessing. If it returns `ready=true`, use the matching write tool with the returned payload. +- In `Agent` mode, prefer the `document-planner` subagent (via the `task` tool) before non-trivial `insert`, `save`, or `set_value` operations. If it returns `ready=false`, ask the user for the missing information instead of guessing. If it returns `ready=true`, use the matching write tool with the returned payload. - When the user wants to create multiple documents of the same DocType, prefer `batch_insert` instead of preparing many separate `insert` proposals. - Before insert or save, call get_meta for the target DocType and follow field types exactly. - Child table fields (fieldtype Table) must be arrays of row objects, never plain strings. @@ -1799,15 +193,11 @@ def _build_tools(self) -> list[Callable[..., Any]]: self.toolset.read_documentation_page, ] - if self.settings.is_code_search_enabled(): - tool_defs.append(self.toolset.source_code_analyzer) - if self.runtime.mode == "Agent": if self._can_write_skill(): tool_defs.append(self.toolset.write_skill) tool_defs.extend( [ - self.toolset.document_planner, self.toolset.insert, self.toolset.batch_insert, self.toolset.save, @@ -1824,169 +214,94 @@ def _build_tools(self) -> list[Callable[..., Any]]: ] ) - return [_clear_messages_on_tool_error(fn) for fn in tool_defs] + return [clear_messages_on_tool_error(fn) for fn in tool_defs] - def run(self, message: str, conversation_history: list[dict[str, Any]]) -> dict[str, Any]: - trace = self.agent.run(build_prompt(message, conversation_history)) - return { - "response": str(trace.final_output or "").strip(), - "pending_operations": self.runtime.pending_operations, - "document_extractions": self.runtime.document_extractions, - "attached_files": self.runtime.attached_files, - } - - -def build_prompt(message: str, conversation_history: list[dict[str, Any]]) -> str: - if not conversation_history: - return message - - lines = [ - "Use the prior conversation as context when answering the final user message.", - "", - "Conversation history:", - ] - for item in conversation_history: - lines.extend(_build_history_item_lines(item)) - - lines.extend(["", f"User: {message}"]) - return "\n".join(lines) - - -def _build_document_extraction_history_entry( - extraction: dict[str, Any], - *, - extraction_prompt: str = "", -) -> dict[str, Any]: - """Normalize extraction metadata stored on assistant messages.""" - return { - "file_id": extraction.get("file_id") or extraction.get("name"), - "file_name": extraction.get("file_name"), - "pages_processed": extraction.get("pages_processed"), - "total_pages": extraction.get("total_pages"), - "truncated": bool(extraction.get("truncated")), - "warning": extraction.get("warning"), - "extraction_prompt": extraction_prompt, - "extracted_data": extraction.get("extracted_data"), - } - - -def _build_history_item_lines(item: dict[str, Any]) -> list[str]: - """Render one stored message and its metadata into prompt lines.""" - role = (item.get("role") or "user").capitalize() - content = item.get("content") or "" - lines = [f"{role}: {content}"] - - metadata = item.get("metadata") - if not isinstance(metadata, dict): - return lines - - lines.extend(_build_attachment_metadata_lines(metadata.get("files"))) - lines.extend(_build_document_extraction_lines(metadata.get("document_extractions"))) - return lines - - -def _build_attachment_metadata_lines(files: Any) -> list[str]: - """Render attachment metadata into compact prompt lines.""" - if not isinstance(files, list): - return [] - - lines: list[str] = [] - for file_entry in files: - if not isinstance(file_entry, dict): - continue - - parts = [] - file_id = (file_entry.get("name") or file_entry.get("file_id") or "").strip() - file_name = (file_entry.get("file_name") or "").strip() - if file_id: - parts.append(f"id={file_id}") - if file_name: - parts.append(f"name={file_name}") - if parts: - lines.append(f"Attachment metadata: {', '.join(parts)}") - - return lines - - -def _build_document_extraction_lines(document_extractions: Any) -> list[str]: - """Render persisted document extraction metadata and JSON into prompt lines.""" - if not isinstance(document_extractions, list): - return [] - - lines: list[str] = [] - for extraction in document_extractions: - if not isinstance(extraction, dict): - continue - - summary = _build_document_extraction_summary(extraction) - if summary: - lines.append(summary) - - extraction_prompt = (extraction.get("extraction_prompt") or "").strip() - if extraction_prompt: - lines.append(f"Extraction request: {extraction_prompt}") - - warning = (extraction.get("warning") or "").strip() - if warning: - lines.append(f"Extraction warning: {warning}") - - extracted_data_text = _stringify_extracted_data(extraction.get("extracted_data")) - if extracted_data_text: - lines.append("Extracted document data (JSON):") - lines.append(extracted_data_text) - - return lines - - -def _build_document_extraction_summary(extraction: dict[str, Any]) -> str: - """Summarize a stored extraction record for prompt reuse.""" - parts = [] - file_id = (extraction.get("file_id") or extraction.get("name") or "").strip() - file_name = (extraction.get("file_name") or "").strip() - pages_processed = extraction.get("pages_processed") - total_pages = extraction.get("total_pages") - if file_id: - parts.append(f"id={file_id}") - if file_name: - parts.append(f"name={file_name}") - if isinstance(pages_processed, int): - parts.append(f"pages_processed={pages_processed}") - if isinstance(total_pages, int): - parts.append(f"total_pages={total_pages}") - return f"Stored document extraction: {', '.join(parts)}" if parts else "" + def _build_subagents(self) -> list[SubAgent]: + # ``general-purpose`` is auto-added by Deep Agents; we only declare the + # two restricted specialists here. Each supplies an explicit minimal + # tool list so the parent's proposal/mutation tools are never inherited. + subagents: list[SubAgent] = [] + if self.settings.is_code_search_enabled(): + subagents.append( + { + "name": "source-code-analyzer", + "description": ( + "Analyze installed app source code. Delegate code questions that require " + "searching, reading, and interpreting multiple files to this specialist. " + "It only has read-only filesystem tools scoped to the /source/ virtual mount." + ), + "system_prompt": SOURCE_CODE_ANALYZER_INSTRUCTIONS, + "tools": [], + "response_format": SourceCodeAnalysisResult, + } + ) -def _build_file_markdown_link(file_doc) -> str: - """Build a Markdown link for a File document label.""" - file_label = _escape_markdown_link_label( - (file_doc.file_name or file_doc.name or "").strip() or file_doc.name - ) - file_url = _quote_markdown_link_destination(file_doc.file_url) - if not file_url: - return file_label - return f"[{file_label}]({file_url})" + if self.runtime.mode == "Agent": + subagents.append( + { + "name": "document-planner", + "description": ( + "Plan document create or update flows. Delegate non-trivial insert, save, " + "or set_value operations to this read-only specialist so it can inspect " + "metadata, resolve Link targets, and return a ready-to-execute payload." + ), + "system_prompt": DOCUMENT_PLANNER_INSTRUCTIONS, + "tools": [ + self.toolset.get_list, + self.toolset.get, + self.toolset.get_value, + self.toolset.get_single_value, + self.toolset.get_meta, + self.toolset.has_permission, + self.toolset.get_doc_permissions, + self.toolset.list_accessible_doctypes, + ], + "response_format": DocumentPlannerResult, + } + ) + return subagents -def _escape_markdown_link_label(label: str) -> str: - """Escape characters that would break a Markdown link label.""" - return label.replace("\\", "\\\\").replace("[", "\\[").replace("]", "\\]") + def _build_input_messages(self, message: str) -> list[AnyMessage]: + """Build the native message list for the next invocation. + Without a checkpointer, the full stored history is rebuilt as native + messages each time and the new user message is appended. + """ + history = self.runtime.conversation_history or [] + messages: list[AnyMessage] = [] + for item in history: + native = history_item_to_native_message(item) + if native is not None: + messages.append(native) -def _quote_markdown_link_destination(url: str | None) -> str: - """Quote a URL so it is safe inside Markdown link syntax.""" - clean_url = (url or "").strip() - if not clean_url: - return "" - return quote(clean_url, safe="/#:?&=%") + messages.append(HumanMessage(content=message)) + return messages + def run(self, message: str, conversation_history: list[dict[str, Any]]) -> dict[str, Any]: + self.runtime.conversation_history = conversation_history or [] + input_messages = self._build_input_messages(message) + result = self.agent.invoke({"messages": input_messages}) + response_text = "" + result_messages = result.get("messages") if isinstance(result, dict) else None + if result_messages: + last = result_messages[-1] + content = getattr(last, "content", None) + if isinstance(content, str): + response_text = content.strip() + elif content is None: + response_text = "" + else: + # Some providers return content as a list of blocks; coerce to text. + response_text = str(content).strip() -def _stringify_extracted_data(extracted_data: Any) -> str: - """Render stored extraction data as JSON text for the prompt.""" - if isinstance(extracted_data, str): - return extracted_data.strip() - if extracted_data is None: - return "" - return frappe.as_json(extracted_data, indent=2) + return { + "response": response_text, + "pending_operations": self.runtime.pending_operations, + "document_extractions": self.runtime.document_extractions, + "attached_files": self.runtime.attached_files, + } def run_message( @@ -2004,3 +319,18 @@ def run_message( ) runner = ask_alyfAgentRunner(runtime) return runner.run(message, conversation_history) + + +# ``create_agent`` is re-exported so the field agent (and any future stateless +# caller) can build a plain LangChain agent without the Deep Agents middleware +# stack. +def build_stateless_agent(model, tools, *, system_prompt: str): + """Build a plain LangChain agent with no checkpointer, no subagents, no VFS. + + Used by the field agent which must remain stateless and tool-restricted. + """ + return create_agent( + model=model, + tools=tools, + system_prompt=system_prompt, + ) diff --git a/ask_alyf/ask_alyf/api.py b/ask_alyf/ask_alyf/api.py index f3c29d4..c12bba3 100644 --- a/ask_alyf/ask_alyf/api.py +++ b/ask_alyf/ask_alyf/api.py @@ -37,7 +37,7 @@ def _truncate_doc_for_size( Args: doc: The document dict to (potentially) truncate in-place. table_fieldnames: Set of fieldnames whose fieldtype is "Table". When - ``None`` or empty, any list-of-dicts value is treated as a child table. + `None` or empty, any list-of-dicts value is treated as a child table. threshold_bytes: Payload size limit in bytes (default 16 KB). max_rows: Maximum rows to retain per child table (default 20). diff --git a/ask_alyf/ask_alyf/deep_agent_backend.py b/ask_alyf/ask_alyf/deep_agent_backend.py new file mode 100644 index 0000000..ca21c6f --- /dev/null +++ b/ask_alyf/ask_alyf/deep_agent_backend.py @@ -0,0 +1,355 @@ +from __future__ import annotations + +import asyncio +import base64 +import re +from pathlib import Path + +import frappe +import wcmatch.glob as wcglob +from deepagents.backends import CompositeBackend, StateBackend +from deepagents.backends.protocol import ( + BackendProtocol, + EditResult, + FileData, + FileInfo, + GlobResult, + GrepMatch, + GrepResult, + LsResult, + ReadResult, + WriteResult, +) +from frappe import _ + +from ask_alyf.ask_alyf.tools import ( + _get_accessible_file_doc, + build_code_path_entry, + get_path_parts, + is_hidden_path, + is_path_within, + iter_scoped_entries, + iter_scoped_files, + normalize_app_relative_path, + to_bench_relative_path, +) + + +def _source_read_only_error() -> str: + return _("Source code is read-only") + + +def _attachment_read_only_error() -> str: + return _("Attachments are read-only") + + +def _bench_relative_to_source_path(bench_relative: str) -> str: + """Convert a bench-relative path (`apps/frappe/...`) to a source virtual path (`/frappe/...`).""" + parts = bench_relative.split("/") + if parts and parts[0] == "apps": + parts = parts[1:] + return "/" + "/".join(parts) + + +def _decode_file_bytes(raw: bytes) -> tuple[str, str]: + """Decode bytes into `(content, encoding)` for `FileData`.""" + try: + return raw.decode("utf-8"), "utf-8" + except UnicodeDecodeError: + return base64.b64encode(raw).decode("ascii"), "base64" + + +def _slice_text(content: str, offset: int, limit: int) -> str | None: + """Return the requested line window or `None` if offset exceeds file length.""" + if not content or content.strip() == "": + return content + lines = content.splitlines(keepends=True) + if offset >= len(lines): + return None + end = min(offset + limit, len(lines)) + return "".join(lines[offset:end]).replace("\r\n", "\n").replace("\r", "\n") + + +class ReadOnlySourceBackend(BackendProtocol): + """Read-only backend exposing installed app source code under `/source/`. + + The composite router strips the `/source/` prefix before dispatching, so + methods here receive paths like `/frappe/frappe/__init__.py`. The first + segment is the installed app name; the remainder is resolved within that + app root, confined to the app root, and traversal is rejected. + + App roots are pre-computed in the main thread and passed to the constructor + so that no Frappe database calls happen from Deep Agents worker threads. + """ + + def __init__(self, app_roots: dict[str, Path]) -> None: + self._app_roots = app_roots + + def _split_app(self, path: str) -> tuple[str, str]: + """Split a backend path into `(app_name, full_relative)`.""" + relative = path.lstrip("/") + parts = get_path_parts(relative) + if not parts: + raise ValueError(_("Path must start with an installed app name.")) + return parts[0], relative + + def _resolve(self, path: str) -> tuple[Path, Path]: + app_name, full_relative = self._split_app(path) + app_root = self._app_roots.get(app_name) + if not app_root: + raise ValueError(_("App '{0}' is not installed.").format(app_name)) + relative_target = normalize_app_relative_path(app_name, full_relative) + target = (app_root / relative_target).resolve() + if not is_path_within(app_root.resolve(), target): + raise ValueError(_("Code access is restricted to installed app directories.")) + return app_root, target + + def _collect_files(self, path: str | None) -> list[tuple[Path, Path]]: + relative = (path or "").lstrip("/") + if not relative: + files: list[tuple[Path, Path]] = [] + for _app_name, app_root in self._app_roots.items(): + for entry in iter_scoped_files(app_root, app_root, include_hidden=False): + files.append((app_root, entry)) + return files + app_root, target = self._resolve(relative) + return [(app_root, fp) for fp in iter_scoped_files(app_root, target, include_hidden=False)] + + def ls(self, path: str) -> LsResult: + try: + relative = path.lstrip("/") + if not relative: + entries = [ + FileInfo(path=f"/{name}", is_dir=True, size=0, modified_at="") + for name in sorted(self._app_roots) + ] + return LsResult(entries=entries) + + app_root, target = self._resolve(relative) + entries: list[FileInfo] = [] + if target.is_file(): + entry = build_code_path_entry(app_root, target) + entries.append( + FileInfo( + path=_bench_relative_to_source_path(entry["path"]), + is_dir=False, + size=int(entry.get("size") or 0), + modified_at="", + ) + ) + else: + for child in iter_scoped_entries(app_root, target, recursive=False, include_hidden=False): + entry = build_code_path_entry(app_root, child) + entries.append( + FileInfo( + path=_bench_relative_to_source_path(entry["path"]), + is_dir=entry["type"] == "directory", + size=int(entry.get("size") or 0), + modified_at="", + ) + ) + entries.sort(key=lambda info: info.get("path", "")) + return LsResult(entries=entries) + except Exception as exc: + return LsResult(error=str(exc)) + + def read(self, file_path: str, offset: int = 0, limit: int = 2000) -> ReadResult: + try: + relative = file_path.lstrip("/") + if not relative: + return ReadResult(error=_("Source file path is required")) + app_root, target = self._resolve(relative) + if is_hidden_path(app_root, target): + return ReadResult(error=_("Access to hidden paths is not allowed")) + if not target.is_file(): + return ReadResult(error=_("Path is not a file: {0}").format(file_path)) + raw = target.read_bytes() + content, encoding = _decode_file_bytes(raw) + if encoding == "base64": + return ReadResult(file_data=FileData(content=content, encoding=encoding)) + sliced = _slice_text(content, offset, limit) + if sliced is None: + return ReadResult(error=_("Line offset {0} exceeds file length").format(offset)) + return ReadResult(file_data=FileData(content=sliced, encoding=encoding)) + except Exception as exc: + return ReadResult(error=str(exc)) + + def grep(self, pattern: str, path: str | None = None, glob: str | None = None) -> GrepResult: + try: + try: + regex = re.compile(pattern) + except re.error as exc: + return GrepResult(error=_("Invalid regex pattern: {0}").format(str(exc))) + files = self._collect_files(path) + matches: list[GrepMatch] = [] + for _app_root, file_path in files: + source_path = _bench_relative_to_source_path(to_bench_relative_path(file_path)) + if glob: + relative = source_path.lstrip("/") + if not wcglob.globmatch(relative, glob, flags=wcglob.BRACE | wcglob.GLOBSTAR): + continue + try: + content = file_path.read_text(encoding="utf-8") + except Exception: + continue + for line_num, line in enumerate(content.split("\n"), 1): + if regex.search(line): + matches.append(GrepMatch(path=source_path, line=int(line_num), text=line)) + return GrepResult(matches=matches) + except Exception as exc: + return GrepResult(error=str(exc)) + + def glob(self, pattern: str, path: str | None = None) -> GlobResult: + try: + files = self._collect_files(path) + bare_pattern = pattern.lstrip("/") + matches: list[FileInfo] = [] + for _app_root, file_path in files: + source_path = _bench_relative_to_source_path(to_bench_relative_path(file_path)) + relative = source_path.lstrip("/") + if wcglob.globmatch(relative, bare_pattern, flags=wcglob.BRACE | wcglob.GLOBSTAR): + try: + size = int(file_path.stat().st_size) + except OSError: + size = 0 + matches.append(FileInfo(path=source_path, is_dir=False, size=size, modified_at="")) + matches.sort(key=lambda info: info.get("path", "")) + return GlobResult(matches=matches) + except Exception as exc: + return GlobResult(error=str(exc)) + + def write(self, file_path: str, content: str) -> WriteResult: + return WriteResult(error=_source_read_only_error()) + + def edit( + self, + file_path: str, + old_string: str, + new_string: str, + replace_all: bool = False, + ) -> EditResult: + return EditResult(error=_source_read_only_error()) + + async def als(self, path: str) -> LsResult: + return await asyncio.to_thread(self.ls, path) + + async def aread(self, file_path: str, offset: int = 0, limit: int = 2000) -> ReadResult: + return await asyncio.to_thread(self.read, file_path, offset, limit) + + async def agrep(self, pattern: str, path: str | None = None, glob: str | None = None) -> GrepResult: + return await asyncio.to_thread(self.grep, pattern, path, glob) + + async def aglob(self, pattern: str, path: str | None = None) -> GlobResult: + return await asyncio.to_thread(self.glob, pattern, path) + + async def awrite(self, file_path: str, content: str) -> WriteResult: + return WriteResult(error=_source_read_only_error()) + + async def aedit( + self, + file_path: str, + old_string: str, + new_string: str, + replace_all: bool = False, + ) -> EditResult: + return EditResult(error=_source_read_only_error()) + + +class ReadOnlyAttachmentBackend(BackendProtocol): + """Read-only backend exposing Frappe **File** attachments under `/attachments/`. + + The composite router strips the `/attachments/` prefix before dispatching, + so methods here receive paths like `/FILE00123` that resolve to a Frappe + **File** document by name. Every read enforces existence and read permission + via the shared `_get_accessible_file_doc` helper. + """ + + def ls(self, path: str) -> LsResult: + return LsResult(entries=[]) + + def read(self, file_path: str, offset: int = 0, limit: int = 2000) -> ReadResult: + try: + name = file_path.lstrip("/") + if not name: + return ReadResult(error=_("Attachment name is required")) + file_doc = _get_accessible_file_doc(file_id=name) + content = file_doc.get_content() + if isinstance(content, bytes): + text, encoding = _decode_file_bytes(content) + else: + text, encoding = str(content), "utf-8" + if encoding == "base64": + return ReadResult(file_data=FileData(content=text, encoding=encoding)) + sliced = _slice_text(text, offset, limit) + if sliced is None: + return ReadResult(error=_("Line offset {0} exceeds file length").format(offset)) + return ReadResult(file_data=FileData(content=sliced, encoding=encoding)) + except Exception as exc: + return ReadResult(error=str(exc)) + + def grep(self, pattern: str, path: str | None = None, glob: str | None = None) -> GrepResult: + return GrepResult(matches=[]) + + def glob(self, pattern: str, path: str | None = None) -> GlobResult: + return GlobResult(matches=[]) + + def write(self, file_path: str, content: str) -> WriteResult: + return WriteResult(error=_attachment_read_only_error()) + + def edit( + self, + file_path: str, + old_string: str, + new_string: str, + replace_all: bool = False, + ) -> EditResult: + return EditResult(error=_attachment_read_only_error()) + + async def als(self, path: str) -> LsResult: + return await asyncio.to_thread(self.ls, path) + + async def aread(self, file_path: str, offset: int = 0, limit: int = 2000) -> ReadResult: + return await asyncio.to_thread(self.read, file_path, offset, limit) + + async def agrep(self, pattern: str, path: str | None = None, glob: str | None = None) -> GrepResult: + return await asyncio.to_thread(self.grep, pattern, path, glob) + + async def aglob(self, pattern: str, path: str | None = None) -> GlobResult: + return await asyncio.to_thread(self.glob, pattern, path) + + async def awrite(self, file_path: str, content: str) -> WriteResult: + return WriteResult(error=_attachment_read_only_error()) + + async def aedit( + self, + file_path: str, + old_string: str, + new_string: str, + replace_all: bool = False, + ) -> EditResult: + return EditResult(error=_attachment_read_only_error()) + + +def build_ask_alyf_backend(app_roots: dict[str, Path]) -> CompositeBackend: + """Build the restricted composite virtual filesystem for Ask ALYF agents. + + Args: + app_roots: Pre-computed installed app roots from the main thread. + Passed to `ReadOnlySourceBackend` so it never calls Frappe + database APIs from Deep Agents worker threads. + + Mounts: + - `/workspace/` (default): in-memory writable scratch via + `StateBackend` — the only mount that accepts writes. + - `/source/`: read-only access to installed app source code, confined + to installed app roots. + - `/attachments/`: read-only access to Frappe **File** documents with + per-read permission checks. + """ + return CompositeBackend( + default=StateBackend(), + routes={ + "/source/": ReadOnlySourceBackend(app_roots), + "/attachments/": ReadOnlyAttachmentBackend(), + }, + ) diff --git a/ask_alyf/ask_alyf/doctype/ask_alyf_conversation/test_ask_alyf_conversation.py b/ask_alyf/ask_alyf/doctype/ask_alyf_conversation/test_ask_alyf_conversation.py index c374226..7a87db2 100644 --- a/ask_alyf/ask_alyf/doctype/ask_alyf_conversation/test_ask_alyf_conversation.py +++ b/ask_alyf/ask_alyf/doctype/ask_alyf_conversation/test_ask_alyf_conversation.py @@ -7,7 +7,8 @@ import frappe from frappe.tests import UnitTestCase -from ask_alyf.ask_alyf import agent, api, tools +from ask_alyf.ask_alyf import api, tools +from ask_alyf.ask_alyf.history import history_item_to_native_message from ask_alyf.ask_alyf.utils import dumps, loads @@ -104,16 +105,21 @@ def test_process_message_job_persists_document_extractions(self): self.assertEqual(messages[-1]["content"], "I found the supplier and total.") self.assertEqual(messages[-1]["metadata"].get("document_extractions"), document_extractions) - prompt = agent.build_prompt( - "Create the Purchase Invoice from that document.", - messages, + prompt = history_item_to_native_message( + { + "role": "user", + "content": "Create the Purchase Invoice from that document.", + "metadata": {"document_extractions": document_extractions}, + } ) - self.assertIn("Stored document extraction:", prompt) - self.assertIn("Extraction request: Extract line items and totals.", prompt) - self.assertIn("id=FILE-0001", prompt) - self.assertNotIn("/private/files/invoice.pdf", prompt) - self.assertIn('"supplier": "ACME"', prompt) - self.assertIn('"total": "123.45"', prompt) + self.assertIsNotNone(prompt) + self.assertIn("Create the Purchase Invoice from that document.", prompt.content) + self.assertIn("Stored document extraction:", prompt.content) + self.assertIn("Extraction request: Extract line items and totals.", prompt.content) + self.assertIn("id=FILE-0001", prompt.content) + self.assertNotIn("/private/files/invoice.pdf", prompt.content) + self.assertIn('"supplier": "ACME"', prompt.content) + self.assertIn('"total": "123.45"', prompt.content) def test_attach_file_persists_file_url_in_message_metadata(self): conversation = self.make_conversation(messages=[]) diff --git a/ask_alyf/ask_alyf/field_agent.py b/ask_alyf/ask_alyf/field_agent.py index b10ed0b..f83812f 100644 --- a/ask_alyf/ask_alyf/field_agent.py +++ b/ask_alyf/ask_alyf/field_agent.py @@ -5,14 +5,10 @@ from typing import Any import frappe -from any_agent import AgentConfig, AgentFramework, AnyAgent from ask_alyf.ask_alyf import field_contexts, tools -from ask_alyf.ask_alyf.agent import ( - _build_internal_agent_config, - _clear_messages_on_tool_error, - ask_alyfToolset, -) +from ask_alyf.ask_alyf.agent import build_chat_model, build_stateless_agent +from ask_alyf.ask_alyf.toolset import ask_alyfToolset, clear_messages_on_tool_error # Read-only tool method names exposed to the field agent. # Excludes: source_code_analyzer (unbounded latency in request thread), @@ -141,23 +137,26 @@ def run_field_agent( toolset = ask_alyfToolset(runtime=runtime, settings=settings) tool_defs = [ - _clear_messages_on_tool_error(getattr(toolset, method_name)) for method_name in FIELD_AGENT_TOOLS + clear_messages_on_tool_error(getattr(toolset, method_name)) for method_name in FIELD_AGENT_TOOLS ] - agent = AnyAgent.create( - AgentFramework.TINYAGENT, - _build_internal_agent_config( - settings=settings, - name="Ask ALYF Field Agent", - instructions=instructions, - tool_defs=tool_defs, - ), - ) + model = build_chat_model(settings, temperature=0.1) + agent = build_stateless_agent(model, tool_defs, system_prompt=instructions) try: - trace = agent.run(prompt) + result = agent.invoke({"messages": [{"role": "user", "content": prompt}]}) except Exception: frappe.log_error("Ask ALYF Field Agent Error") raise - return str(trace.final_output or "").strip() + result_messages = result.get("messages") if isinstance(result, dict) else None + if result_messages: + last = result_messages[-1] + content = getattr(last, "content", None) + if isinstance(content, str): + return content.strip() + elif content is not None: + # Some providers return content as a list of blocks; coerce to text. + return str(content).strip() + + return "" diff --git a/ask_alyf/ask_alyf/history.py b/ask_alyf/ask_alyf/history.py new file mode 100644 index 0000000..cb81e3b --- /dev/null +++ b/ask_alyf/ask_alyf/history.py @@ -0,0 +1,155 @@ +from typing import Any +from urllib.parse import quote + +import frappe +from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, SystemMessage + + +def _build_document_extraction_history_entry( + extraction: dict[str, Any], + *, + extraction_prompt: str = "", +) -> dict[str, Any]: + """Normalize extraction metadata stored on assistant messages.""" + return { + "file_id": extraction.get("file_id") or extraction.get("name"), + "file_name": extraction.get("file_name"), + "pages_processed": extraction.get("pages_processed"), + "total_pages": extraction.get("total_pages"), + "truncated": bool(extraction.get("truncated")), + "warning": extraction.get("warning"), + "extraction_prompt": extraction_prompt, + "extracted_data": extraction.get("extracted_data"), + } + + +def _build_attachment_metadata_lines(files: Any) -> list[str]: + """Render attachment metadata into compact prompt lines.""" + if not isinstance(files, list): + return [] + + lines: list[str] = [] + for file_entry in files: + if not isinstance(file_entry, dict): + continue + + parts = [] + file_id = (file_entry.get("name") or file_entry.get("file_id") or "").strip() + file_name = (file_entry.get("file_name") or "").strip() + if file_id: + parts.append(f"id={file_id}") + if file_name: + parts.append(f"name={file_name}") + if parts: + lines.append(f"Attachment metadata: {', '.join(parts)}") + + return lines + + +def _build_document_extraction_lines(document_extractions: Any) -> list[str]: + """Render persisted document extraction metadata and JSON into prompt lines.""" + if not isinstance(document_extractions, list): + return [] + + lines: list[str] = [] + for extraction in document_extractions: + if not isinstance(extraction, dict): + continue + + summary = _build_document_extraction_summary(extraction) + if summary: + lines.append(summary) + + extraction_prompt = (extraction.get("extraction_prompt") or "").strip() + if extraction_prompt: + lines.append(f"Extraction request: {extraction_prompt}") + + warning = (extraction.get("warning") or "").strip() + if warning: + lines.append(f"Extraction warning: {warning}") + + extracted_data_text = _stringify_extracted_data(extraction.get("extracted_data")) + if extracted_data_text: + lines.append("Extracted document data (JSON):") + lines.append(extracted_data_text) + + return lines + + +def _build_document_extraction_summary(extraction: dict[str, Any]) -> str: + """Summarize a stored extraction record for prompt reuse.""" + parts = [] + file_id = (extraction.get("file_id") or extraction.get("name") or "").strip() + file_name = (extraction.get("file_name") or "").strip() + pages_processed = extraction.get("pages_processed") + total_pages = extraction.get("total_pages") + if file_id: + parts.append(f"id={file_id}") + if file_name: + parts.append(f"name={file_name}") + if isinstance(pages_processed, int): + parts.append(f"pages_processed={pages_processed}") + if isinstance(total_pages, int): + parts.append(f"total_pages={total_pages}") + return f"Stored document extraction: {', '.join(parts)}" if parts else "" + + +def _build_file_markdown_link(file_doc) -> str: + """Build a Markdown link for a File document label.""" + file_label = _escape_markdown_link_label( + (file_doc.file_name or file_doc.name or "").strip() or file_doc.name + ) + file_url = _quote_markdown_link_destination(file_doc.file_url) + if not file_url: + return file_label + return f"[{file_label}]({file_url})" + + +def _escape_markdown_link_label(label: str) -> str: + """Escape characters that would break a Markdown link label.""" + return label.replace("\\", "\\\\").replace("[", "\\[").replace("]", "\\]") + + +def _quote_markdown_link_destination(url: str | None) -> str: + """Quote a URL so it is safe inside Markdown link syntax.""" + clean_url = (url or "").strip() + if not clean_url: + return "" + return quote(clean_url, safe="/#:?&=%") + + +def _stringify_extracted_data(extracted_data: Any) -> str: + """Render stored extraction data as JSON text for the prompt.""" + if isinstance(extracted_data, str): + return extracted_data.strip() + if extracted_data is None: + return "" + return frappe.as_json(extracted_data, indent=2) + + +def history_item_to_native_message(item: dict[str, Any]) -> AnyMessage | None: + """Convert one stored conversation history item into a native LangChain message. + + Attachment and extraction metadata are appended to the message content so the + model sees the same context the old ``build_prompt`` flattening provided. + """ + role = (item.get("role") or "user").lower() + content = item.get("content") or "" + metadata = item.get("metadata") + if not isinstance(metadata, dict): + metadata = None + + parts = [content] if content else [] + if metadata: + parts.extend(_build_attachment_metadata_lines(metadata.get("files"))) + parts.extend(_build_document_extraction_lines(metadata.get("document_extractions"))) + + text = "\n".join(part for part in parts if part).strip() + if not text: + return None + + if role == "assistant": + return AIMessage(content=text) + if role == "system": + return SystemMessage(content=text) + return HumanMessage(content=text) diff --git a/ask_alyf/ask_alyf/subagents.py b/ask_alyf/ask_alyf/subagents.py new file mode 100644 index 0000000..ceb59d2 --- /dev/null +++ b/ask_alyf/ask_alyf/subagents.py @@ -0,0 +1,83 @@ +from typing import Any + +from pydantic import BaseModel, Field + + +class SourceCodeAnalysisEvidence(BaseModel): + path: str + start_line: int | None = None + end_line: int | None = None + note: str | None = None + + +class SourceCodeAnalysisResult(BaseModel): + answer: str = "" + summary: str = "" + evidence: list[SourceCodeAnalysisEvidence] = Field(default_factory=list) + uncertainty: str = "" + searched_paths: list[str] = Field(default_factory=list) + + +class DocumentPlannerResult(BaseModel): + ready: bool = False + recommended_tool: str = "" + payload: dict[str, Any] = Field(default_factory=dict) + reason: str = "" + missing_information: list[str] = Field(default_factory=list) + checks: list[str] = Field(default_factory=list) + warnings: list[str] = Field(default_factory=list) + + +SOURCE_CODE_ANALYZER_INSTRUCTIONS = """ +You are SourceCodeAnalyzer, an internal Ask ALYF specialist for installed app code. + +You can only use the provided source-code tools (ls, read_file, glob, grep) against the `/source/` virtual mount. + +Rules: +- Search or list first, then read the smallest relevant file ranges. +- Prefer the narrowest path scope available. +- Do not answer from memory when the tools can verify it. +- If evidence is incomplete or ambiguous, say so clearly. +- Include `/source/`-relative paths and line ranges in evidence whenever possible. +- Return a compact JSON object with keys: + - `answer` (string) + - `summary` (string) + - `evidence` (list of objects with `path`, optional `start_line`, optional `end_line`, and optional `note`) + - `uncertainty` (string) + - `searched_paths` (list of strings) + +Return only a valid JSON object. +Do not wrap the JSON in markdown fences. +Do not add explanatory prose before or after the JSON. +""".strip() + +DOCUMENT_PLANNER_INSTRUCTIONS = """ +You are DocumentPlanner, an internal Ask ALYF specialist for planning Frappe document changes. + +You only have read-only access to metadata and documents. You never execute writes. + +You may only plan these operations: +- `insert` +- `save` +- `set_value` + +Rules: +- Always inspect `get_meta` before planning `insert`, `save`, or `set_value`. +- Use the read tools to resolve Link targets or confirm existing values when possible. +- Never invent document names, Link targets, or required values. +- Treat `values_hint` as tentative until it is confirmed by the user or by a read tool. +- If information is missing, set `ready` to `false` and list each missing item in `missing_information`. +- The `payload` must match the parent tool signature for the recommended operation. +- Return a JSON object with keys: + - `ready` (boolean) + - `recommended_tool` (`insert`, `save`, or `set_value`) + - `payload` (object) + - `reason` (string) + - `missing_information` (list of strings) + - `checks` (list of strings) + - `warnings` (list of strings) + +Return only a valid JSON object. +Do not wrap the JSON in markdown fences. +Do not add explanatory prose before or after the JSON. +""".strip() diff --git a/ask_alyf/ask_alyf/test_code_tools.py b/ask_alyf/ask_alyf/test_code_tools.py index 0165b1e..41036cf 100644 --- a/ask_alyf/ask_alyf/test_code_tools.py +++ b/ask_alyf/ask_alyf/test_code_tools.py @@ -1,16 +1,15 @@ import asyncio from types import SimpleNamespace -from unittest.mock import AsyncMock, patch +from unittest.mock import patch import frappe from frappe.tests import UnitTestCase +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage from ask_alyf.ask_alyf import tools -from ask_alyf.ask_alyf.agent import ( - _clear_messages_on_tool_error, - ask_alyfAgentRunner, - ask_alyfToolset, -) +from ask_alyf.ask_alyf.agent import ASK_ALYF_EXCLUDED_TOOLS, ask_alyfAgentRunner +from ask_alyf.ask_alyf.history import history_item_to_native_message +from ask_alyf.ask_alyf.toolset import ask_alyfToolset, clear_messages_on_tool_error class FakeSettings(SimpleNamespace): @@ -51,28 +50,38 @@ def make_runner(self, *, allow_code_search: bool, mode: str = "Ask"): runner.toolset = ask_alyfToolset(runtime, settings=runner.settings) return runner - def test_build_tools_only_adds_source_code_analyzer_when_enabled(self): + def test_subagents_register_source_code_analyzer_only_when_code_search_enabled(self): raw_code_tool_names = {"search_code", "read_code_file", "ls", "find", "grep"} disabled_runner = self.make_runner(allow_code_search=False) + disabled_subagents = disabled_runner._build_subagents() + disabled_names = {sub["name"] for sub in disabled_subagents} + self.assertNotIn("source-code-analyzer", disabled_names) + # Raw code tools are never exposed as parent tools regardless of setting. disabled_tool_names = {tool.__name__ for tool in disabled_runner._build_tools()} - self.assertNotIn("source_code_analyzer", disabled_tool_names) self.assertFalse(raw_code_tool_names.intersection(disabled_tool_names)) enabled_runner = self.make_runner(allow_code_search=True) + enabled_subagents = enabled_runner._build_subagents() + enabled_names = {sub["name"] for sub in enabled_subagents} + self.assertIn("source-code-analyzer", enabled_names) + # Source code tools are not parent-visible; they live inside the subagent. enabled_tool_names = {tool.__name__ for tool in enabled_runner._build_tools()} - self.assertIn("source_code_analyzer", enabled_tool_names) self.assertFalse(raw_code_tool_names.intersection(enabled_tool_names)) - def test_build_tools_only_adds_document_planner_in_agent_mode(self): + def test_subagents_register_document_planner_only_in_agent_mode(self): ask_runner = self.make_runner(allow_code_search=False, mode="Ask") + ask_subagents = ask_runner._build_subagents() + ask_names = {sub["name"] for sub in ask_subagents} + self.assertNotIn("document-planner", ask_names) ask_tool_names = {tool.__name__ for tool in ask_runner._build_tools()} - self.assertNotIn("document_planner", ask_tool_names) self.assertNotIn("batch_insert", ask_tool_names) agent_runner = self.make_runner(allow_code_search=False, mode="Agent") + agent_subagents = agent_runner._build_subagents() + agent_names = {sub["name"] for sub in agent_subagents} + self.assertIn("document-planner", agent_names) agent_tool_names = {tool.__name__ for tool in agent_runner._build_tools()} - self.assertIn("document_planner", agent_tool_names) self.assertIn("batch_insert", agent_tool_names) def test_build_tools_only_adds_write_skill_when_user_can_create_skills(self): @@ -353,115 +362,205 @@ def test_execute_action_batch_insert_collects_successes_and_failures(self): self.assertIn("Created 2 of 3 ToDo records.", result["message"]) self.assertIn("row 2: Missing description", result["message"]) - def test_source_code_analyzer_tool_delegates_to_specialist(self): - runtime = self.make_runtime(mode="Ask") - toolset = ask_alyfToolset(runtime) - expected = { - "answer": "The main agent runner is in agent.py.", - "summary": "Found the main runner.", - "evidence": [{"path": "apps/ask_alyf/ask_alyf/ask_alyf/agent.py", "start_line": 989}], - "uncertainty": "", - "searched_paths": ["apps/ask_alyf/ask_alyf/ask_alyf"], - } - specialist = SimpleNamespace(analyze=AsyncMock(return_value=expected)) - - with patch.object(toolset, "_get_source_code_analyzer", return_value=specialist): - result = asyncio.run( - toolset.source_code_analyzer( - "Where is the main agent runner defined?", - relative_path="ask_alyf/ask_alyf", - ) - ) + def test_source_code_analyzer_subagent_descriptor_has_empty_tools(self): + runner = self.make_runner(allow_code_search=True, mode="Ask") + subagents = runner._build_subagents() + analyzer = next(sub for sub in subagents if sub["name"] == "source-code-analyzer") + + self.assertEqual(analyzer["tools"], []) + # The descriptor must not inherit any parent proposal/mutation tools. + self.assertNotIn("insert", {getattr(t, "__name__", "") for t in analyzer["tools"]}) + self.assertNotIn("save", {getattr(t, "__name__", "") for t in analyzer["tools"]}) + + def test_source_code_analyzer_subagent_has_response_format(self): + from ask_alyf.ask_alyf.subagents import SourceCodeAnalysisResult + + runner = self.make_runner(allow_code_search=True, mode="Ask") + analyzer = next(sub for sub in runner._build_subagents() if sub["name"] == "source-code-analyzer") + self.assertIs(analyzer["response_format"], SourceCodeAnalysisResult) + + def test_document_planner_subagent_descriptor_has_read_only_tools(self): + runner = self.make_runner(allow_code_search=False, mode="Agent") + subagents = runner._build_subagents() + planner = next(sub for sub in subagents if sub["name"] == "document-planner") + + tool_names = {getattr(t, "__name__", "") for t in planner["tools"]} + # Only read-only metadata/tools are allowed. + self.assertIn("get_meta", tool_names) + self.assertIn("get", tool_names) + # No proposal or mutation tools leak into the planner. + for forbidden in ("insert", "save", "set_value", "submit", "cancel", "delete", "batch_insert"): + self.assertNotIn(forbidden, tool_names) + # No frontend action tools either. + for forbidden in ("set_route", "new_doc", "show_chart", "frm_set_value"): + self.assertNotIn(forbidden, tool_names) + + def test_document_planner_subagent_has_response_format(self): + from ask_alyf.ask_alyf.subagents import DocumentPlannerResult + + runner = self.make_runner(allow_code_search=False, mode="Agent") + planner = next(sub for sub in runner._build_subagents() if sub["name"] == "document-planner") + self.assertIs(planner["response_format"], DocumentPlannerResult) + + def test_harness_profile_excludes_write_edit_execute(self): + from ask_alyf.ask_alyf.agent import _ensure_ask_alyf_harness_profile - specialist.analyze.assert_awaited_once_with( - question="Where is the main agent runner defined?", - relative_path="ask_alyf/ask_alyf", + self.assertEqual( + ASK_ALYF_EXCLUDED_TOOLS, + frozenset({"write_file", "edit_file", "execute"}), ) - self.assertEqual(result, expected) - - def test_source_code_analyzer_initializes_internal_agent_async(self): - runtime = self.make_runtime(mode="Ask") - toolset = ask_alyfToolset(runtime, settings=FakeSettings(allow_code_search=True)) - fake_trace = SimpleNamespace(final_output='{"answer":"Verified","summary":"Verified"}') - fake_agent = SimpleNamespace(run_async=AsyncMock(return_value=fake_trace)) - analyzer = toolset._get_source_code_analyzer() + # Registration is idempotent and safe to call repeatedly. + _ensure_ask_alyf_harness_profile() + _ensure_ask_alyf_harness_profile() + + def test_build_permissions_denies_write_everywhere(self): + from deepagents import FilesystemPermission + + runner = self.make_runner(allow_code_search=False, mode="Ask") + permissions = runner._build_permissions() + self.assertEqual(len(permissions), 1) + perm = permissions[0] + self.assertIsInstance(perm, FilesystemPermission) + self.assertEqual(perm.operations, ["write"]) + self.assertEqual(perm.paths, ["/**"]) + self.assertEqual(perm.mode, "deny") + + def test_coordinator_exposes_built_in_write_todos_tool(self): + """The Deep Agents coordinator must include the built-in `write_todos` + planning tool in its model-visible tool surface. `write_todos` is added + by `TodoListMiddleware` during agent assembly and is the planning + mechanism (filesystem writes are denied by the harness profile and + permissions). + """ + from deepagents import create_deep_agent + from langchain_openai import ChatOpenAI + + from ask_alyf.ask_alyf.deep_agent_backend import build_ask_alyf_backend + + runner = self.make_runner(allow_code_search=False, mode="Ask") + agent = create_deep_agent( + model=ChatOpenAI(model="dummy", api_key="dummy"), + tools=runner._build_tools(), + system_prompt="dummy", + backend=build_ask_alyf_backend({}), + subagents=runner._build_subagents(), + permissions=runner._build_permissions(), + name="ask_alyf", + ) + tool_names = set(agent.nodes["tools"].bound.tools_by_name.keys()) + self.assertIn("write_todos", tool_names) - with patch( - "ask_alyf.ask_alyf.agent._create_internal_agent_async", - AsyncMock(return_value=fake_agent), - ) as create_agent: - first = asyncio.run(analyzer.analyze("Where is tax logic implemented?", relative_path="erpnext")) - second = asyncio.run(analyzer.analyze("Where is tax logic implemented?", relative_path="erpnext")) - - create_agent.assert_awaited_once() - self.assertEqual(fake_agent.run_async.await_count, 2) - self.assertEqual(first["answer"], "Verified") - self.assertEqual(second["summary"], "Verified") - - def test_document_planner_tool_delegates_to_specialist(self): - runtime = self.make_runtime(mode="Agent") - toolset = ask_alyfToolset(runtime) - expected = { - "ready": True, - "recommended_tool": "insert", - "payload": {"doctype": "ToDo", "values": {"description": "Call customer"}}, - "reason": "The user asked to create a new ToDo.", - "missing_information": [], - "checks": ["Loaded ToDo metadata."], - "warnings": [], + def test_history_item_to_native_message_maps_roles(self): + self.assertIsInstance( + history_item_to_native_message({"role": "user", "content": "hi"}), + HumanMessage, + ) + self.assertIsInstance( + history_item_to_native_message({"role": "assistant", "content": "ok"}), + AIMessage, + ) + self.assertIsInstance( + history_item_to_native_message({"role": "system", "content": "note"}), + SystemMessage, + ) + self.assertIsNone(history_item_to_native_message({"role": "user", "content": ""})) + + def test_history_item_to_native_message_appends_extraction_metadata(self): + item = { + "role": "user", + "content": "What is on this invoice?", + "metadata": { + "files": [{"name": "FILE-0001", "file_name": "invoice.pdf"}], + "document_extractions": [ + { + "file_id": "FILE-0001", + "file_name": "invoice.pdf", + "pages_processed": 2, + "total_pages": 2, + "extraction_prompt": "Extract line items and totals.", + "extracted_data": {"supplier": "ACME", "total": "123.45"}, + } + ], + }, } - specialist = SimpleNamespace(plan=AsyncMock(return_value=expected)) - - with patch.object(toolset, "_get_document_planner", return_value=specialist): - result = asyncio.run( - toolset.document_planner( - user_request="Create a ToDo to call the customer.", - doctype="ToDo", - operation="insert", - values_hint={"description": "Call customer"}, - ) - ) - - specialist.plan.assert_awaited_once_with( - user_request="Create a ToDo to call the customer.", - doctype="ToDo", - operation="insert", - name="", - values_hint={"description": "Call customer"}, + message = history_item_to_native_message(item) + self.assertIsInstance(message, HumanMessage) + text = message.content + self.assertIn("What is on this invoice?", text) + self.assertIn("Stored document extraction: id=FILE-0001, name=invoice.pdf", text) + self.assertIn("Extraction request: Extract line items and totals.", text) + self.assertIn('"supplier": "ACME"', text) + self.assertIn('"total": "123.45"', text) + # File URLs are never leaked into the native message context. + self.assertNotIn("/private/files/", text) + self.assertNotIn("/files/", text) + + def test_build_input_messages_rebuilds_full_history(self): + runner = self.make_runner(allow_code_search=False, mode="Ask") + runner.runtime.conversation_history = [ + {"role": "user", "content": "one"}, + {"role": "assistant", "content": "one-ans"}, + {"role": "system", "content": "result"}, + ] + messages = runner._build_input_messages("two") + # Full history is rebuilt as native messages + the new user turn. + self.assertEqual(len(messages), 4) + self.assertIsInstance(messages[0], HumanMessage) + self.assertIsInstance(messages[1], AIMessage) + self.assertIsInstance(messages[2], SystemMessage) + self.assertIsInstance(messages[-1], HumanMessage) + self.assertEqual(messages[-1].content, "two") + + def test_run_preserves_result_envelope_and_proposal_shapes(self): + runner = self.make_runner(allow_code_search=False, mode="Agent") + runner.agent = SimpleNamespace( + invoke=lambda _input, config=None: {"messages": [SimpleNamespace(content="Done.")]} ) - self.assertEqual(result, expected) + result = runner.run("do something", conversation_history=[]) + self.assertEqual(result["response"], "Done.") + self.assertEqual(result["pending_operations"], []) + self.assertEqual(result["document_extractions"], []) + self.assertEqual(result["attached_files"], []) + + def test_agent_mode_tools_include_proposals_but_no_host_mutation_or_shell(self): + """Model-visible tools must contain proposal operations but no direct + Frappe mutation, host filesystem, delete-of-host, or shell execution.""" + host_mutation_tools = {"write_file", "edit_file", "execute", "shell", "run_shell"} + proposal_tools = { + "insert", + "batch_insert", + "save", + "set_value", + "submit", + "cancel", + "amend", + "delete", + "rename_doc", + "attach_file", + "run_whitelisted_method", + } - def test_document_planner_initializes_internal_agent_async(self): - runtime = self.make_runtime(mode="Agent") - toolset = ask_alyfToolset(runtime, settings=FakeSettings(allow_code_search=False)) - fake_trace = SimpleNamespace( - final_output='{"ready":true,"recommended_tool":"insert","payload":{"doctype":"ToDo","values":{}},"reason":"ok","missing_information":[],"checks":[],"warnings":[]}' - ) - fake_agent = SimpleNamespace(run_async=AsyncMock(return_value=fake_trace)) - planner = toolset._get_document_planner() + agent_runner = self.make_runner(allow_code_search=False, mode="Agent") + agent_tool_names = {tool.__name__ for tool in agent_runner._build_tools()} + # Proposal ops are present (they only create pending proposals, never mutate directly). + self.assertTrue(proposal_tools.issubset(agent_tool_names)) + # No host filesystem, shell, or direct execution capability leaks in. + self.assertFalse(host_mutation_tools.intersection(agent_tool_names)) - with patch( - "ask_alyf.ask_alyf.agent._create_internal_agent_async", - AsyncMock(return_value=fake_agent), - ) as create_agent: - result = asyncio.run( - planner.plan( - user_request="Create a ToDo.", - doctype="ToDo", - operation="insert", - ) - ) + def test_ask_mode_tools_exclude_all_write_proposals_and_host_mutation(self): + host_mutation_tools = {"write_file", "edit_file", "execute", "shell", "run_shell"} + proposal_tools = {"insert", "save", "set_value", "submit", "cancel", "delete", "batch_insert"} - create_agent.assert_awaited_once() - fake_agent.run_async.assert_awaited_once() - self.assertTrue(result["ready"]) - self.assertEqual(result["recommended_tool"], "insert") + ask_runner = self.make_runner(allow_code_search=False, mode="Ask") + ask_tool_names = {tool.__name__ for tool in ask_runner._build_tools()} + self.assertFalse(proposal_tools.intersection(ask_tool_names)) + self.assertFalse(host_mutation_tools.intersection(ask_tool_names)) def test_clear_messages_wrapper_preserves_async_tools(self): async def fake_tool(file_id): return {"file_id": file_id} - wrapped = _clear_messages_on_tool_error(fake_tool) + wrapped = clear_messages_on_tool_error(fake_tool) result = asyncio.run(wrapped("FILE-0001")) self.assertTrue(asyncio.iscoroutinefunction(wrapped)) @@ -471,9 +570,9 @@ def test_clear_messages_wrapper_clears_messages_for_async_tool_errors(self): async def fake_tool(): raise RuntimeError("boom") - wrapped = _clear_messages_on_tool_error(fake_tool) + wrapped = clear_messages_on_tool_error(fake_tool) - with patch("ask_alyf.ask_alyf.agent.frappe.clear_messages") as clear_messages: + with patch("ask_alyf.ask_alyf.toolset.frappe.clear_messages") as clear_messages: with self.assertRaisesRegex(RuntimeError, "boom"): asyncio.run(wrapped()) diff --git a/ask_alyf/ask_alyf/test_deep_agent_backend.py b/ask_alyf/ask_alyf/test_deep_agent_backend.py new file mode 100644 index 0000000..a545b04 --- /dev/null +++ b/ask_alyf/ask_alyf/test_deep_agent_backend.py @@ -0,0 +1,145 @@ +# Copyright (c) 2026, ALYF GmbH and Contributors +# See license.txt + +from __future__ import annotations + +from unittest.mock import patch + +import frappe +from deepagents.backends import StateBackend +from frappe.tests import UnitTestCase + +from ask_alyf.ask_alyf import tools +from ask_alyf.ask_alyf.deep_agent_backend import ( + ReadOnlyAttachmentBackend, + ReadOnlySourceBackend, + _attachment_read_only_error, + _source_read_only_error, + build_ask_alyf_backend, +) + + +def _app_roots(): + return tools.get_installed_app_roots() + + +class UnitTestAskALYFVirtualFilesystem(UnitTestCase): + def test_build_ask_alyf_backend_mounts_workspace_source_and_attachments(self): + backend = build_ask_alyf_backend(_app_roots()) + route_prefixes = {prefix for prefix, _ in backend.sorted_routes} + self.assertIn("/source/", route_prefixes) + self.assertIn("/attachments/", route_prefixes) + # The default backend is the writable StateBackend. + self.assertIsInstance(backend.default, StateBackend) + + def test_workspace_routes_to_writable_default_backend(self): + backend = build_ask_alyf_backend(_app_roots()) + target, stripped = backend._get_backend_and_key("/workspace/scratch.txt") + self.assertIs(target, backend.default) + # /workspace/ is not a declared route, so the path is passed through unchanged. + self.assertEqual(stripped, "/workspace/scratch.txt") + + def test_source_routes_to_read_only_source_backend(self): + backend = build_ask_alyf_backend(_app_roots()) + target, stripped = backend._get_backend_and_key("/source/ask_alyf/agent.py") + self.assertIsInstance(target, ReadOnlySourceBackend) + self.assertEqual(stripped, "/ask_alyf/agent.py") + + def test_attachments_routes_to_read_only_attachment_backend(self): + backend = build_ask_alyf_backend(_app_roots()) + target, stripped = backend._get_backend_and_key("/attachments/FILE-0001") + self.assertIsInstance(target, ReadOnlyAttachmentBackend) + self.assertEqual(stripped, "/FILE-0001") + + def test_source_backend_rejects_writes_and_edits(self): + backend = build_ask_alyf_backend(_app_roots()) + write_result = backend.write("/source/ask_alyf/new_file.py", "print('hi')") + self.assertEqual(write_result.error, _source_read_only_error()) + + edit_result = backend.edit( + "/source/ask_alyf/agent.py", + "old", + "new", + ) + self.assertEqual(edit_result.error, _source_read_only_error()) + + def test_attachments_backend_rejects_writes_and_edits(self): + backend = build_ask_alyf_backend(_app_roots()) + write_result = backend.write("/attachments/FILE-0001", "overwrite") + self.assertEqual(write_result.error, _attachment_read_only_error()) + + edit_result = backend.edit( + "/attachments/FILE-0001", + "old", + "new", + ) + self.assertEqual(edit_result.error, _attachment_read_only_error()) + + def test_source_backend_blocks_path_traversal(self): + backend = build_ask_alyf_backend(_app_roots()) + result = backend.read("/source/ask_alyf/../../../../etc/passwd") + # Traversal escapes the app root; the read returns an error result + # rather than host filesystem contents. + self.assertTrue(result.error) + self.assertIsNone(result.file_data) + # The error must be the confinement error, not a code-search-disabled error. + self.assertIn("installed app", result.error) + + def test_source_backend_excludes_hidden_paths(self): + backend = build_ask_alyf_backend(_app_roots()) + result = backend.read("/source/ask_alyf/.gitignore") + self.assertEqual(result.error, "Access to hidden paths is not allowed") + self.assertIsNone(result.file_data) + + def test_source_backend_ls_root_handles_empty_app_roots_gracefully(self): + backend = build_ask_alyf_backend({}) + result = backend.ls("/source/") + self.assertFalse(result.error) + self.assertEqual(result.entries, []) + + def test_source_backend_read_returns_content_for_real_app_file(self): + backend = build_ask_alyf_backend(_app_roots()) + result = backend.read("/source/ask_alyf/ask_alyf/ask_alyf/agent.py") + # The file exists in the installed ask_alyf app, so we get content back. + self.assertIsNone(result.error) + self.assertIsNotNone(result.file_data) + self.assertIn("ask_alyfAgentRunner", result.file_data["content"]) + + def test_attachment_backend_read_enforces_permission_checks(self): + backend = build_ask_alyf_backend(_app_roots()) + with patch( + "ask_alyf.ask_alyf.deep_agent_backend._get_accessible_file_doc", + side_effect=frappe.PermissionError("no read access"), + ): + result = backend.read("/attachments/FILE-PRIVATE") + self.assertTrue(result.error) + self.assertIsNone(result.file_data) + + def test_attachment_backend_read_rejects_empty_name(self): + backend = build_ask_alyf_backend(_app_roots()) + result = backend.read("/attachments/") + self.assertTrue(result.error) + self.assertIsNone(result.file_data) + + def test_workspace_scratch_does_not_survive_across_backend_instances(self): + """Each `build_ask_alyf_backend(_app_roots())` call produces a fresh, independent + `StateBackend`. There is no checkpointer, so scratch written through one + backend instance cannot surface in another. + """ + backend1 = build_ask_alyf_backend(_app_roots()) + self.assertIsInstance(backend1.default, StateBackend) + + backend2 = build_ask_alyf_backend(_app_roots()) + self.assertIsInstance(backend2.default, StateBackend) + # Distinct instances: scratch state is per-backend, never shared. + self.assertIsNot(backend1.default, backend2.default) + + # StateBackend reads/writes through the live LangGraph config and keeps no + # standalone store, so reading outside a graph execution has no scratch to + # return. This is what makes /workspace/ an in-memory scratch mount whose + # contents do not survive across separate backend (and therefore separate + # agent) instances. + with self.assertRaises(RuntimeError): + backend1.default.read("/workspace/scratch.txt") + with self.assertRaises(RuntimeError): + backend2.default.read("/workspace/scratch.txt") diff --git a/ask_alyf/ask_alyf/test_field_agent.py b/ask_alyf/ask_alyf/test_field_agent.py index 0dc5cff..8312f65 100644 --- a/ask_alyf/ask_alyf/test_field_agent.py +++ b/ask_alyf/ask_alyf/test_field_agent.py @@ -64,13 +64,10 @@ def test_get_field_context_unmapped_via_build_generic_fallback_directly(self): class UnitTestFieldAgent(UnitTestCase): - def _make_fake_trace(self, output: str = "OK"): - return SimpleNamespace(final_output=output) - def _make_fake_agent(self, output: str = "OK"): - trace = self._make_fake_trace(output) + """Build a fake stateless agent whose `invoke` returns a native message list.""" agent = MagicMock() - agent.run.return_value = trace + agent.invoke.return_value = {"messages": [SimpleNamespace(content=output)]} return agent def test_run_field_agent_creates_no_conversation(self): @@ -78,7 +75,7 @@ def test_run_field_agent_creates_no_conversation(self): fake_agent = self._make_fake_agent("{{ doc.name }}") with ( - patch("ask_alyf.ask_alyf.field_agent.AnyAgent.create", return_value=fake_agent), + patch("ask_alyf.ask_alyf.field_agent.build_stateless_agent", return_value=fake_agent), patch("ask_alyf.ask_alyf.field_agent.tools.get_settings") as mock_settings, ): mock_settings.return_value = SimpleNamespace( @@ -110,7 +107,7 @@ def test_run_field_agent_no_realtime_published(self): fake_agent = self._make_fake_agent("generated output") with ( - patch("ask_alyf.ask_alyf.field_agent.AnyAgent.create", return_value=fake_agent), + patch("ask_alyf.ask_alyf.field_agent.build_stateless_agent", return_value=fake_agent), patch("ask_alyf.ask_alyf.field_agent.tools.get_settings") as mock_settings, patch.object(frappe, "publish_realtime") as mock_realtime, ): @@ -134,6 +131,35 @@ def test_run_field_agent_no_realtime_published(self): mock_realtime.assert_not_called() + def test_run_field_agent_uses_stateless_agent_without_checkpointer(self): + """run_field_agent must build a stateless agent with no checkpointer or subagents.""" + from ask_alyf.ask_alyf import field_agent + + fake_agent = self._make_fake_agent("ok") + with ( + patch("ask_alyf.ask_alyf.field_agent.build_stateless_agent", return_value=fake_agent) as build, + patch("ask_alyf.ask_alyf.field_agent.tools.get_settings") as mock_settings, + ): + mock_settings.return_value = SimpleNamespace( + model="gpt-test", + llm_provider="OpenAI", + base_url="", + get_password=lambda _f, raise_exception=False: "test-key", + ) + field_agent.run_field_agent( + doctype="Print Format", + fieldname="html", + fieldtype="Code", + current_value="", + doc={"name": "TEST-PF"}, + prompt="generate header", + ) + + build.assert_called_once() + # build_stateless_agent(model, tools, system_prompt=...) — no checkpointer/subagents/VFS kwargs. + kwargs = build.call_args.kwargs + self.assertEqual(set(kwargs.keys()), {"system_prompt"}) + class UnitTestFieldAgentEndpoint(UnitTestCase): def test_field_agent_run_endpoint_perm_gate_role(self): diff --git a/ask_alyf/ask_alyf/toolset.py b/ask_alyf/ask_alyf/toolset.py new file mode 100644 index 0000000..a5974c5 --- /dev/null +++ b/ask_alyf/ask_alyf/toolset.py @@ -0,0 +1,1156 @@ +import functools +import inspect +from dataclasses import dataclass, field +from typing import Any +from uuid import uuid4 + +import frappe +from frappe import _ + +from ask_alyf.ask_alyf import tools +from ask_alyf.ask_alyf.history import ( + _build_document_extraction_history_entry, + _build_file_markdown_link, +) +from ask_alyf.ask_alyf.skill_utils import get_accessible_skill_doc +from ask_alyf.ask_alyf.utils import parse_newline_list + + +@dataclass +class ask_alyfRuntime: + conversation_name: str + mode: str + request_context: dict[str, Any] + conversation_history: list[dict[str, Any]] = field(default_factory=list) + pending_operations: list[dict[str, Any]] = field(default_factory=list) + document_extractions: list[dict[str, Any]] = field(default_factory=list) + attached_files: list[dict[str, Any]] = field(default_factory=list) + + def emit_status(self, text: str): + """Send a short status update to the current user.""" + frappe.publish_realtime( + "ask_alyf_status", + {"conversation": self.conversation_name, "text": text}, + user=frappe.session.user, + ) + + def remember_document_extraction(self, extraction: dict[str, Any], *, extraction_prompt: str = ""): + """Store a normalized extraction result for reuse in later turns.""" + self.document_extractions.append( + _build_document_extraction_history_entry(extraction, extraction_prompt=extraction_prompt) + ) + + def remember_attached_file(self, file_entry: dict[str, Any]): + """Store a file attachment to be shown in the conversation history.""" + self.attached_files.append(file_entry) + + +class ask_alyfToolset: + def __init__(self, runtime: ask_alyfRuntime, settings=None): + self.runtime = runtime + self.settings = settings + + def _get_settings(self): + if self.settings is None: + self.settings = tools.get_settings() + return self.settings + + def _proposal( + self, + kind: str, + tool: str, + summary: str, + reason: str = "", + *, + validation_error_status: str, + prepared_status: str, + requires_confirmation: bool = True, + **payload: Any, + ) -> dict[str, Any]: + """Create a pending operation proposal and append it to the list.""" + validation_error = tools.validate_pending_operation_payload(kind, tool, payload) + if validation_error: + self.runtime.emit_status(validation_error_status) + return { + "success": False, + "requires_confirmation": False, + "error": validation_error, + } + + proposal = { + "kind": kind, + "tool": tool, + "summary": summary, + "reason": reason, + "requires_confirmation": bool(requires_confirmation), + "payload": payload, + "call_id": uuid4().hex, + } + self.runtime.pending_operations.append(proposal) + self.runtime.emit_status(prepared_status) + return { + "success": True, + "requires_confirmation": bool(requires_confirmation), + "proposal": proposal, + } + + def _backend_proposal( + self, + tool: str, + summary: str, + reason: str = "", + *, + validation_error_status: str, + prepared_status: str, + **payload: Any, + ) -> dict[str, Any]: + return self._proposal( + tools.OPERATION_KIND_BACKEND, + tool, + summary, + reason, + validation_error_status=validation_error_status, + prepared_status=prepared_status, + requires_confirmation=True, + **payload, + ) + + def _frontend_proposal( + self, + tool: str, + summary: str, + reason: str = "", + *, + validation_error_status: str, + prepared_status: str, + requires_confirmation: bool | None = None, + **payload: Any, + ) -> dict[str, Any]: + if requires_confirmation is None: + requires_confirmation = tools.get_frontend_action_requires_confirmation(tool) + return self._proposal( + tools.OPERATION_KIND_FRONTEND, + tool, + summary, + reason, + validation_error_status=validation_error_status, + prepared_status=prepared_status, + requires_confirmation=requires_confirmation, + **payload, + ) + + def get_list( + self, + doctype: str, + fields: str | list[str] | None = None, + filters: dict[str, Any] | list[Any] | None = None, + order_by: str | None = None, + limit: int = 20, + group_by: str | None = None, + ) -> list[dict[str, Any]]: + """List documents with filters, fields, ordering, and optional grouping. + + Args: + doctype: The DocType to query. + fields: Optional field name or field list to return. + filters: Optional Frappe filters. + order_by: Optional ordering expression. + limit: Maximum number of rows to return. + group_by: Optional SQL group by expression. + + Returns: + A list of matching documents. + """ + self.runtime.emit_status(_("Fetching list...")) + return tools.get_list( + doctype=doctype, + fields=fields, + filters=filters, + order_by=order_by, + limit=limit, + group_by=group_by, + ) + + def get_count( + self, + doctype: str, + filters: dict[str, Any] | list[Any] | None = None, + ) -> int: + """Count documents that match the given filters. + + Args: + doctype: The DocType to query. + filters: Optional Frappe filters. + + Returns: + The number of matching documents. + """ + self.runtime.emit_status(_("Counting documents...")) + return tools.get_count(doctype=doctype, filters=filters) + + def get( + self, + doctype: str, + name: str | None = None, + filters: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Read a single document by name or filters. + + Args: + doctype: The DocType to query. + name: Optional document name. + filters: Optional filters when name is not known. + + Returns: + The matching document. + """ + self.runtime.emit_status(_("Fetching document...")) + return tools.get_document(doctype=doctype, name=name, filters=filters) + + def get_value( + self, + doctype: str, + fieldname: str | list[str], + filters: dict[str, Any] | list[Any] | str | None = None, + ) -> Any: + """Read one or more field values from a document. + + Args: + doctype: The DocType to query. + fieldname: One field name or a list of field names. + filters: Name or filters to identify the record. + + Returns: + The requested value or values. + """ + self.runtime.emit_status(_("Fetching value...")) + return tools.get_value(doctype=doctype, fieldname=fieldname, filters=filters) + + def get_single_value(self, doctype: str, field: str) -> Any: + """Read a field value from a Single DocType. + + Args: + doctype: The Single DocType to query. + field: The field name to read. + + Returns: + The field value. + """ + self.runtime.emit_status(_("Fetching single value...")) + return tools.get_single_value(doctype=doctype, field=field) + + def get_meta(self, doctype: str) -> dict[str, Any]: + """Inspect metadata, fields, and permissions for a DocType. + + Args: + doctype: The DocType to inspect. + + Returns: + A metadata dictionary for the DocType. + """ + self.runtime.emit_status(_("Loading metadata...")) + return tools.get_meta(doctype=doctype) + + def has_permission(self, doctype: str, docname: str, perm_type: str = "read") -> dict[str, bool]: + """Check whether the current user has a specific permission on a document. + + Args: + doctype: The DocType to check. + docname: The document name. + perm_type: The permission type to evaluate. + + Returns: + A dictionary containing the boolean permission result. + """ + self.runtime.emit_status(_("Checking permissions...")) + return tools.has_permission(doctype=doctype, docname=docname, perm_type=perm_type) + + def get_doc_permissions(self, doctype: str, docname: str) -> dict[str, Any]: + """Get the evaluated permission map for a document. + + Args: + doctype: The DocType to check. + docname: The document name. + + Returns: + The evaluated permission dictionary. + """ + self.runtime.emit_status(_("Evaluating permissions...")) + return tools.get_doc_permissions(doctype=doctype, docname=docname) + + def list_accessible_doctypes(self, permission_type: str = "read") -> list[str]: + """List DocTypes that the current user can access. + + Args: + permission_type: The permission type to test. + + Returns: + A list of DocType names. + """ + self.runtime.emit_status(_("Listing accessible DocTypes...")) + return tools.list_accessible_doctypes(permission_type=permission_type) + + def list_accessible_reports(self) -> list[dict[str, Any]]: + """List reports that the current user can access. + + Returns: + A list of report metadata dictionaries. + """ + self.runtime.emit_status(_("Listing accessible reports...")) + return tools.list_accessible_reports() + + def translate_ui_labels( + self, + labels: list[str], + language: str | None = None, + ) -> dict[str, Any]: + """Translate UI labels so responses match what the user sees on screen. + + Use this whenever request context language is not English before mentioning + button names, tab names, DocType labels, field labels, menu items, or status labels. + + Args: + labels: English UI labels to translate. + language: Optional target language code (defaults to request context language). + + Returns: + A dictionary with the resolved language and translated labels. + """ + self.runtime.emit_status(_("Translating UI labels...")) + request_language = self.runtime.request_context.get("lang") or self.runtime.request_context.get( + "locale" + ) + return tools.translate_ui_labels(labels=labels, language=language or request_language) + + def search_code(self, query: str, relative_path: str = "", limit: int = 20) -> list[dict[str, Any]]: + """Search installed app code for matching text. + + Args: + query: The text to search for. + relative_path: Optional installed-app-relative path. Providing at least the app name will make it much faster. + limit: Maximum number of matches to return. + + Returns: + A list of code search matches. + """ + self.runtime.emit_status(_("Searching code...")) + return tools.search_code(query=query, relative_path=relative_path, limit=limit) + + def read_code_file(self, path: str, start_line: int = 1, end_line: int = 200) -> dict[str, Any]: + """Read a file from installed app code. + + Args: + path: Bench-relative path inside an installed app, such as apps/my_app/my_app/module.py. + start_line: First line to include. + end_line: Last line to include. + + Returns: + The selected file content and line range. + """ + self.runtime.emit_status(_("Reading code file...")) + return tools.read_code_file(path=path, start_line=start_line, end_line=end_line) + + def ls( + self, + app_name: str, + relative_path: str = "", + recursive: bool = False, + include_hidden: bool = False, + limit: int = 200, + ) -> dict[str, Any]: + """List files or directories in an installed app, similar to Debian ls. + + Args: + app_name: The installed app name. + relative_path: Optional path inside the app. + recursive: Whether to include nested descendants. + include_hidden: Whether to include hidden files and folders. + limit: Maximum number of entries to return. + + Returns: + A directory listing payload. + """ + self.runtime.emit_status(_("Listing code files...")) + return tools.ls( + app_name=app_name, + relative_path=relative_path, + recursive=recursive, + include_hidden=include_hidden, + limit=limit, + ) + + def find( + self, + app_name: str, + name_pattern: str = "*", + relative_path: str = "", + entry_type: str = "any", + include_hidden: bool = False, + limit: int = 200, + ) -> dict[str, Any]: + """Find files or directories in an installed app, similar to Debian find. + + Args: + app_name: The installed app name. + name_pattern: Shell-style filename pattern, such as *.py. + relative_path: Optional path inside the app to search from. + entry_type: One of any, file, or directory. + include_hidden: Whether to include hidden files and folders. + limit: Maximum number of matches to return. + + Returns: + A find-style search payload. + """ + self.runtime.emit_status(_("Finding code paths...")) + return tools.find( + app_name=app_name, + name_pattern=name_pattern, + relative_path=relative_path, + entry_type=entry_type, + include_hidden=include_hidden, + limit=limit, + ) + + def grep( + self, + app_name: str, + query: str, + relative_path: str = "", + file_pattern: str = "*", + case_sensitive: bool = False, + include_hidden: bool = False, + limit: int = 50, + ) -> dict[str, Any]: + """Search file contents in an installed app, similar to Debian grep. + + Args: + app_name: The installed app name. + query: The text to search for. + relative_path: Optional path inside the app to search from. + file_pattern: Optional shell-style filename filter, such as *.py. + case_sensitive: Whether matching should be case-sensitive. + include_hidden: Whether to include hidden files and folders. + limit: Maximum number of matches to return. + + Returns: + A grep-style search payload. + """ + self.runtime.emit_status(_("Searching file contents...")) + return tools.grep( + app_name=app_name, + query=query, + relative_path=relative_path, + file_pattern=file_pattern, + case_sensitive=case_sensitive, + include_hidden=include_hidden, + limit=limit, + ) + + def get_file_id( + self, + reference_doctype: str, + reference_name: str, + reference_field: str = "", + file_url: str = "", + file_name: str = "", + ) -> str: + """Resolve a unique File ID from attachment reference filters. + + Args: + reference_doctype: The DocType the file is attached to. + reference_name: The document name the file is attached to. + reference_field: Optional attachment field name. + file_url: Optional file URL to disambiguate matches. + file_name: Optional file name to disambiguate matches. + + Returns: + The matching File ID. + """ + self.runtime.emit_status(_("Resolving file ID...")) + return tools.get_file_id( + reference_doctype=reference_doctype, + reference_name=reference_name, + reference_field=reference_field, + file_url=file_url, + file_name=file_name, + ) + + def read_file_record(self, file_id: str) -> dict[str, Any]: + """Read the content of a File record the user can access. + + Args: + file_id: The File ID (`name`) to read. + + Returns: + The file metadata and content. + """ + self.runtime.emit_status(_("Reading file...")) + return tools.read_file_record(file_id=file_id) + + def extract_document_data(self, file_id: str, extraction_prompt: str = "") -> dict[str, Any]: + """Extract structured data from a PDF or image file using vision AI. + + Call this tool when a user uploads or references a document (invoice, receipt, + contract, etc.) and wants to extract information from it. The file is rendered + as images and sent to a vision-capable model that reads text, tables, and layouts. + + Supports PDF files (up to 10 pages) and images (PNG, JPG, GIF, WebP). + + Args: + file_id: The File ID (`name`) to process. + extraction_prompt: Optional instructions for what data to extract. + If omitted, a general-purpose extraction prompt is used. + + Returns: + A dictionary with the file ID, file name, number of pages processed, + and the extracted data as a JSON object. + """ + self.runtime.emit_status(_("Extracting document data...")) + import asyncio + + result = asyncio.run( + tools.extract_document_data(file_id=file_id, extraction_prompt=extraction_prompt) + ) + self.runtime.remember_document_extraction(result, extraction_prompt=extraction_prompt) + return result + + def get_print( + self, + doctype: str, + name: str, + print_format: str = "", + letterhead: str = "", + language: str = "", + ) -> dict[str, Any]: + """Generate a PDF print of a document and attach it to the conversation. + + Uses the default print format and default letter head unless overridden. + Permissions are enforced automatically by the tool. + + The generated PDF is automatically shown to the user as a clickable file + attachment in the conversation UI. Do not repeat the file name, URL, or + link in your text — just confirm briefly what was printed. + + Args: + doctype: The DocType of the document to print. + name: The document name. + print_format: Optional print format name. Defaults to the DocType's default. + letterhead: Optional letter head name. Defaults to the site's default. + language: Optional language code for the print. Defaults to the document's + language field if it exists, otherwise the current session language. + + Returns: + A dictionary with the generated file metadata (name, file_name, file_url). + """ + self.runtime.emit_status(_("Generating print...")) + file_entry = tools.get_print( + doctype=doctype, + name=name, + conversation_name=self.runtime.conversation_name, + print_format=print_format, + letterhead=letterhead, + language=language, + ) + self.runtime.remember_attached_file(file_entry) + return file_entry + + def run_read_only_sql(self, query: str) -> list[dict[str, Any]]: + """Run a read-only SQL query when the current user is allowed to do so. + + Args: + query: A single read-only SQL query. + + Returns: + The SQL result rows. + """ + self.runtime.emit_status(_("Running SQL query...")) + return tools.run_read_only_sql(query=query) + + def get_app_version(self, app_name: str) -> str: + """Read the installed version for an app. + + Args: + app_name: The installed app name. + + Returns: + The app version string. + """ + self.runtime.emit_status(_("Reading app version...")) + return tools.get_app_version(app_name=app_name) + + def read_github_releases(self, app_name: str, limit: int = 5) -> list[dict[str, Any]]: + """Read recent GitHub releases for an installed app. + + Args: + app_name: The installed app name. + limit: Maximum number of releases to return. + + Returns: + A list of release dictionaries. + """ + self.runtime.emit_status(_("Reading GitHub releases...")) + return tools.read_github_releases(app_name=app_name, limit=limit) + + def read_documentation_page(self, app_name: str, relative_path: str = "") -> dict[str, Any]: + """Fetch a documentation page for an installed app. + + Args: + app_name: The installed app name. + relative_path: Optional path relative to the configured docs URL. + + Returns: + A documentation payload containing the page content. + """ + self.runtime.emit_status(_("Reading documentation...")) + return tools.read_documentation_page(app_name=app_name, relative_path=relative_path) + + def read_skill(self, name: str) -> dict[str, str]: + """Read the full content of an Ask ALYF skill available to the current user. + + Use this when the instructions list a skill that seems relevant. Pass the + exact skill `name` shown in that list. + + Args: + name: The exact Ask ALYF Skill name. + + Returns: + A dictionary containing the skill name, title, and markdown description. + """ + self.runtime.emit_status(_("Reading skill...")) + skill_doc = get_accessible_skill_doc(name) + return { + "name": skill_doc.name, + "title": skill_doc.title, + "description": skill_doc.description or "", + } + + def write_skill( + self, + title: str, + description: str, + roles: list[str], + reason: str = "", + ) -> dict[str, Any]: + """Propose creating a reusable Ask ALYF skill. + + Use this when the user wants to save durable instructions that can later be + loaded with `read_skill`. + + Args: + title: The skill title shown to users. + description: The markdown skill content. + roles: Roles that should be allowed to use the skill. + Accepts a list or a comma/newline-separated string. + reason: Optional explanation of why the skill should be created. + + Returns: + A pending action proposal that requires confirmation. + """ + clean_title = (title or "").strip() + clean_description = (description or "").strip() + clean_roles = parse_newline_list(roles) + + if not clean_title: + frappe.throw(_("Skill title is required.")) + if not clean_description: + frappe.throw(_("Skill description is required.")) + if not clean_roles: + frappe.throw(_("At least one role is required.")) + + return self._backend_proposal( + "insert", + _("Create skill '{0}'").format(clean_title), + reason, + validation_error_status=_("Skill proposal needs correction."), + prepared_status=_("Prepared skill proposal."), + doctype="Ask ALYF Skill", + values={ + "title": clean_title, + "description": clean_description, + "roles": [{"role": role} for role in clean_roles], + }, + ) + + def insert(self, doctype: str, values: dict[str, Any], reason: str = "") -> dict[str, Any]: + """Propose creating a new document. Use get_meta first to know the schema. + For child tables, values must be a list of row objects. + + Args: + doctype: The DocType to create. + values: The field values for the new document. + reason: Optional explanation of why this change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + return self._backend_proposal( + "insert", + _("Create {0}").format(_(doctype)), + reason, + validation_error_status=_("Create proposal needs correction."), + prepared_status=_("Prepared create proposal."), + doctype=doctype, + values=values, + ) + + def batch_insert(self, doctype: str, records: list[dict[str, Any]], reason: str = "") -> dict[str, Any]: + """Propose creating multiple documents of the same DocType. + Use get_meta first to know the schema for each record. + For child tables, each record must use a list of row objects. + + Args: + doctype: The DocType to create. + records: A list of field-value dictionaries, one per new document. + reason: Optional explanation of why this batch change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + record_count = len(records) if isinstance(records, list) else 0 + return self._backend_proposal( + "batch_insert", + _("Create {0} {1} records").format(record_count, _(doctype)), + reason, + validation_error_status=_("Batch create proposal needs correction."), + prepared_status=_("Prepared batch create proposal."), + doctype=doctype, + records=records, + ) + + def save( + self, + doctype: str, + name: str, + values: dict[str, Any], + reason: str = "", + ) -> dict[str, Any]: + """Propose updating an existing document. + For child tables, values must be a list of row objects. + + Args: + doctype: The DocType to update. + name: The document name. + values: The fields to update. + reason: Optional explanation of why this change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + return self._backend_proposal( + "save", + _("Update {0} {1}").format(_(doctype), name), + reason, + validation_error_status=_("Update proposal needs correction."), + prepared_status=_("Prepared update proposal."), + doctype=doctype, + name=name, + values=values, + ) + + def set_value( + self, + doctype: str, + name: str, + fieldname: str, + value: Any, + reason: str = "", + ) -> dict[str, Any]: + """Propose setting a single field on a document. + + Args: + doctype: The DocType to update. + name: The document name. + fieldname: The field to set. + value: The new value. + reason: Optional explanation of why this change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + return self._backend_proposal( + "set_value", + _("Set {0} on {1} {2}").format(fieldname, _(doctype), name), + reason, + validation_error_status=_("Set value proposal needs correction."), + prepared_status=_("Prepared set value proposal."), + doctype=doctype, + name=name, + fieldname=fieldname, + value=value, + ) + + def submit(self, doctype: str, name: str, reason: str = "") -> dict[str, Any]: + """Propose submitting a document. + + Args: + doctype: The DocType to submit. + name: The document name. + reason: Optional explanation of why this change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + return self._backend_proposal( + "submit", + _("Submit {0} {1}").format(_(doctype), name), + reason, + validation_error_status=_("Submit proposal needs correction."), + prepared_status=_("Prepared submit proposal."), + doctype=doctype, + name=name, + ) + + def cancel(self, doctype: str, name: str, reason: str = "") -> dict[str, Any]: + """Propose cancelling a document. + + Args: + doctype: The DocType to cancel. + name: The document name. + reason: Optional explanation of why this change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + return self._backend_proposal( + "cancel", + _("Cancel {0} {1}").format(_(doctype), name), + reason, + validation_error_status=_("Cancel proposal needs correction."), + prepared_status=_("Prepared cancel proposal."), + doctype=doctype, + name=name, + ) + + def amend(self, doctype: str, name: str, reason: str = "") -> dict[str, Any]: + """Propose amending a cancelled document. + + Args: + doctype: The DocType to amend. + name: The document name. + reason: Optional explanation of why this change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + return self._backend_proposal( + "amend", + _("Amend {0} {1}").format(_(doctype), name), + reason, + validation_error_status=_("Amend proposal needs correction."), + prepared_status=_("Prepared amend proposal."), + doctype=doctype, + name=name, + ) + + def delete(self, doctype: str, name: str, reason: str = "") -> dict[str, Any]: + """Propose deleting a document. + + Args: + doctype: The DocType to delete. + name: The document name. + reason: Optional explanation of why this change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + return self._backend_proposal( + "delete", + _("Delete {0} {1}").format(_(doctype), name), + reason, + validation_error_status=_("Delete proposal needs correction."), + prepared_status=_("Prepared delete proposal."), + doctype=doctype, + name=name, + ) + + def rename_doc( + self, + doctype: str, + name: str, + new_name: str, + merge: bool = False, + reason: str = "", + ) -> dict[str, Any]: + """Propose renaming a document. + + Args: + doctype: The DocType to rename. + name: The current document name. + new_name: The target document name. + merge: Whether to merge into an existing target. + reason: Optional explanation of why this change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + return self._backend_proposal( + "rename_doc", + _("Rename {0} {1} to {2}").format(_(doctype), name, new_name), + reason, + validation_error_status=_("Rename proposal needs correction."), + prepared_status=_("Prepared rename proposal."), + doctype=doctype, + name=name, + new_name=new_name, + merge=merge, + ) + + def attach_file( + self, + doctype: str, + name: str, + file_id: str, + reason: str = "", + ) -> dict[str, Any]: + """Propose attaching an existing file to a document. + + Args: + doctype: The DocType to update. + name: The document name. + file_id: The File ID (`name`) to attach. + reason: Optional explanation of why this change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + file_doc = tools._get_accessible_file_doc(file_id=file_id) + return self._backend_proposal( + "attach_file", + _("Attach file {0} to {1} {2}").format( + _build_file_markdown_link(file_doc), + _(doctype), + name, + ), + reason, + validation_error_status=_("Attach file proposal needs correction."), + prepared_status=_("Prepared attach file proposal."), + doctype=doctype, + name=name, + file_id=file_id, + ) + + def run_whitelisted_method( + self, + method: str, + args: dict[str, Any] | None = None, + reason: str = "", + ) -> dict[str, Any]: + """Propose calling a whitelisted method. + + Args: + method: The dotted Python path of the whitelisted method. + args: Optional keyword arguments for the method. + reason: Optional explanation of why this change is needed. + + Returns: + A pending action proposal that requires confirmation. + """ + return self._backend_proposal( + "run_method", + _("Call {0}").format(method), + reason, + validation_error_status=_("Method call proposal needs correction."), + prepared_status=_("Prepared method call proposal."), + method=method, + args=args or {}, + ) + + def set_route(self, route: list[str], reason: str = "") -> dict[str, Any]: + """Propose navigating to a Desk route on the frontend. + + Args: + route: Route parts used by frappe.set_route. + reason: Optional explanation of why this navigation helps the user. + + Returns: + A pending frontend operation proposal. + """ + route_label = "/".join(part for part in (route or []) if isinstance(part, str)) + return self._frontend_proposal( + "set_route", + _("Navigate to {0}").format(route_label or _("target route")), + reason, + validation_error_status=_("Route action needs correction."), + prepared_status=_("Prepared route action."), + route=route, + ) + + def new_doc( + self, + doctype: str, + route_options: dict[str, Any] | None = None, + reason: str = "", + ) -> dict[str, Any]: + """Propose opening a new document form in the frontend. + + Args: + doctype: The target DocType for frappe.new_doc. + route_options: Optional route options to prefill form values. + reason: Optional explanation of why this action helps the user. + + Returns: + A pending frontend operation proposal. + """ + return self._frontend_proposal( + "new_doc", + _("Open new {0}").format(_(doctype)), + reason, + validation_error_status=_("New document action needs correction."), + prepared_status=_("Prepared new document action."), + doctype=doctype, + route_options=route_options or {}, + ) + + def scroll_to_field(self, fieldname: str, reason: str = "") -> dict[str, Any]: + """Propose scrolling to a field on the active form. + + Args: + fieldname: The fieldname to scroll to. + reason: Optional explanation of why this action helps the user. + + Returns: + A pending frontend operation proposal. + """ + return self._frontend_proposal( + "scroll_to_field", + _("Scroll to field {0}").format(fieldname), + reason, + validation_error_status=_("Scroll action needs correction."), + prepared_status=_("Prepared scroll action."), + fieldname=fieldname, + ) + + def frm_set_value( + self, + fieldname: str, + value: Any, + doctype: str | None = None, + docname: str | None = None, + reason: str = "", + ) -> dict[str, Any]: + """Propose setting a field value on the active frontend form. + + Args: + fieldname: The target fieldname. + value: The value to apply. + doctype: Optional expected active form DocType. + docname: Optional expected active form document name. + reason: Optional explanation of why this action helps the user. + + Returns: + A pending frontend operation proposal. + """ + payload: dict[str, Any] = {"fieldname": fieldname, "value": value} + if doctype: + payload["doctype"] = doctype + if docname: + payload["docname"] = docname + + return self._frontend_proposal( + "frm_set_value", + _("Set field {0} on current form").format(fieldname), + reason, + validation_error_status=_("Set field action needs correction."), + prepared_status=_("Prepared set field action."), + **payload, + ) + + def frm_add_child( + self, + fieldname: str, + values: dict[str, Any] | None = None, + doctype: str | None = None, + docname: str | None = None, + reason: str = "", + ) -> dict[str, Any]: + """Propose adding a child table row on the active frontend form. + + Args: + fieldname: The child table fieldname. + values: Optional row values for the new child row. + doctype: Optional expected active form DocType. + docname: Optional expected active form document name. + reason: Optional explanation of why this action helps the user. + + Returns: + A pending frontend operation proposal. + """ + payload: dict[str, Any] = {"fieldname": fieldname, "values": values or {}} + if doctype: + payload["doctype"] = doctype + if docname: + payload["docname"] = docname + + return self._frontend_proposal( + "frm_add_child", + _("Add a row to {0} on current form").format(fieldname), + reason, + validation_error_status=_("Add child row action needs correction."), + prepared_status=_("Prepared add child row action."), + **payload, + ) + + def show_chart( + self, + frappe_charts: list[dict[str, Any]], + reason: str = "", + ) -> dict[str, Any]: + """Show one or more Frappe Charts under this assistant message. + + The desk creates the DOM element; each item in `frappe_charts` is the `options` + argument to `new frappe.Chart(container, options)` (Frappe Charts on the client). + + Shape (one object per chart): + - `type`: bar, line, scatter, pie, percentage, donut, or axis-mixed + - `data`: `{ "labels": [...], "datasets": [ { "values": [...], "name"?: str, "type"?: "bar"|"line" } ] }` + — every `values` list must be the same length as `labels` + - Optional: `title`, `height` (0 for default; if set, at least 240 — Frappe Charts reserves ~130px chrome), + `colors` (array; empty for defaults), + `barOptions`, `lineOptions`, `axisOptions` (see Frappe Charts docs) + + Args: + frappe_charts: One or more chart option objects. + reason: Optional short note for the user. + + Returns: + A pending frontend operation proposal (auto-executed in the browser). + """ + count = len(frappe_charts) if isinstance(frappe_charts, list) else 0 + summary = _("Show chart") if count == 1 else _("Show {0} charts").format(count) + return self._frontend_proposal( + "show_chart", + summary, + reason, + validation_error_status=_("Chart action needs correction."), + prepared_status=_("Prepared chart display."), + requires_confirmation=False, + frappe_charts=frappe_charts, + ) + + +def clear_messages_on_tool_error(func): + """Wrap a tool callable so that queued Frappe messages are discarded on + exception. The error still propagates to the agent framework (so the LLM + sees it), but the user won't receive a popup.""" + + if inspect.iscoroutinefunction(func): + + @functools.wraps(func) + async def async_wrapper(*args, **kwargs): + try: + return await func(*args, **kwargs) + except Exception: + frappe.clear_messages() + raise + + return async_wrapper + + @functools.wraps(func) + def wrapper(*args, **kwargs): + try: + return func(*args, **kwargs) + except Exception: + frappe.clear_messages() + raise + + return wrapper diff --git a/pyproject.toml b/pyproject.toml index 070f955..1d7300c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,8 +10,9 @@ readme = "README.md" dynamic = ["version"] dependencies = [ # "frappe~=16.0.0" # Installed and managed by bench. - "any-agent", "any-llm-sdk", + "deepagents==0.6.12", + "langchain-openai==1.3.5", "pymupdf", ] From 26a062c5cf584a7d1bca0356231a154d71f6aa14 Mon Sep 17 00:00:00 2001 From: barredterra <14891507+barredterra@users.noreply.github.com> Date: Mon, 13 Jul 2026 04:21:05 +0200 Subject: [PATCH 2/5] fix(agent): guard harness profile registration with a lock Concurrent ask_alyfAgentRunner constructions in one process could both observe the registered flag as false and call register_harness_profile twice. Registration merges on duplicate keys, so an identical double register was functionally safe but logged a warning and did a redundant merge. Guard the check-then-set with a threading.Lock. --- ask_alyf/ask_alyf/agent.py | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/ask_alyf/ask_alyf/agent.py b/ask_alyf/ask_alyf/agent.py index de3c683..9084246 100644 --- a/ask_alyf/ask_alyf/agent.py +++ b/ask_alyf/ask_alyf/agent.py @@ -1,3 +1,4 @@ +import threading from collections.abc import Callable from typing import Any @@ -39,19 +40,23 @@ # and operate on the restricted composite VFS (workspace, source, attachments). ASK_ALYF_EXCLUDED_TOOLS = frozenset({"write_file", "edit_file", "execute"}) _ASK_ALYF_PROFILE_REGISTERED = False +_ASK_ALYF_PROFILE_LOCK = threading.Lock() def _ensure_ask_alyf_harness_profile() -> None: """Register the Ask ALYF harness profile once per process. - Registration is additive and idempotent, so concurrent calls are safe. + Registration is additive and idempotent, but the check-then-set is guarded + by a lock so concurrent ``ask_alyfAgentRunner`` constructions (e.g. two + gunicorn threads or RQ jobs in one process) cannot both register. """ global _ASK_ALYF_PROFILE_REGISTERED - if _ASK_ALYF_PROFILE_REGISTERED: - return - profile = HarnessProfile(excluded_tools=ASK_ALYF_EXCLUDED_TOOLS) - register_harness_profile("openai", profile) - _ASK_ALYF_PROFILE_REGISTERED = True + with _ASK_ALYF_PROFILE_LOCK: + if _ASK_ALYF_PROFILE_REGISTERED: + return + profile = HarnessProfile(excluded_tools=ASK_ALYF_EXCLUDED_TOOLS) + register_harness_profile("openai", profile) + _ASK_ALYF_PROFILE_REGISTERED = True def _get_api_key_from_settings(settings) -> str: From a0f6d6a6fe0875c7923c83926844d8d5165dce6d Mon Sep 17 00:00:00 2001 From: barredterra <14891507+barredterra@users.noreply.github.com> Date: Mon, 13 Jul 2026 04:21:28 +0200 Subject: [PATCH 3/5] docs(agent): clarify source-code-analyzer subagent VFS tooling The source-code-analyzer subagent is declared with tools: [], which looks like it receives no tools at all. Deep Agents always injects FilesystemMiddleware (and thus the read-only VFS tools ls, read_file, glob, grep) into every subagent stack regardless of the tools field; the tools list only controls extra tools inherited from the parent. Document this so the empty list is not mistaken for a bug. --- ask_alyf/ask_alyf/agent.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/ask_alyf/ask_alyf/agent.py b/ask_alyf/ask_alyf/agent.py index 9084246..39869d5 100644 --- a/ask_alyf/ask_alyf/agent.py +++ b/ask_alyf/ask_alyf/agent.py @@ -225,6 +225,16 @@ def _build_subagents(self) -> list[SubAgent]: # ``general-purpose`` is auto-added by Deep Agents; we only declare the # two restricted specialists here. Each supplies an explicit minimal # tool list so the parent's proposal/mutation tools are never inherited. + # + # Note on ``tools: []`` for source-code-analyzer: Deep Agents always + # injects ``FilesystemMiddleware`` (and thus the read-only VFS tools + # ``ls``, ``read_file``, ``glob``, ``grep``) into every subagent stack + # regardless of the ``tools`` field — the ``tools`` list only controls + # *extra* tools inherited from the parent. An empty list therefore means + # "no parent tools", not "no tools at all"; the analyzer still gets the + # ``/source/``-scoped read tools it needs while staying free of mutation + # tools. Write tools are stripped by the harness profile + deny + # permission, so the VFS remains strictly read-only here. subagents: list[SubAgent] = [] if self.settings.is_code_search_enabled(): From d8d7894ebf190e6790e09518f7d16b14088f304c Mon Sep 17 00:00:00 2001 From: barredterra <14891507+barredterra@users.noreply.github.com> Date: Mon, 13 Jul 2026 04:21:34 +0200 Subject: [PATCH 4/5] chore(deps): declare langchain, langchain-core and langgraph directly agent.py and history.py import from langchain and langchain_core, and langgraph underpins the deepagents runtime, but all three arrived only as transitive dependencies of deepagents and langchain-openai. A downstream resolver could pull a compatible-but-different version and break the imports with no install-time signal. Pin them to the ranges required by the pinned deepagents/langchain-openai versions. --- pyproject.toml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 1d7300c..c53fa9a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,12 @@ dependencies = [ # "frappe~=16.0.0" # Installed and managed by bench. "any-llm-sdk", "deepagents==0.6.12", + # Imported directly (langchain.agents, langchain_core.messages); pinned to + # the ranges required by deepagents/langchain-openai so a downstream + # resolver cannot silently pull an incompatible version. + "langchain>=1.3.11,<2.0.0", + "langchain-core>=1.4.9,<2.0.0", + "langgraph>=1.2.5,<1.3.0", "langchain-openai==1.3.5", "pymupdf", ] From c500373f8f5c8fcce15ceba8152c4ba0ce7ebd16 Mon Sep 17 00:00:00 2001 From: barredterra <14891507+barredterra@users.noreply.github.com> Date: Mon, 13 Jul 2026 04:33:31 +0200 Subject: [PATCH 5/5] refactor(toolset): remove orphaned code-search toolset methods Drop the five ask_alyfToolset wrappers search_code, read_code_file, ls, find, and grep. Code search now runs through the source-code-analyzer subagent against the /source/ virtual mount, so these wrappers were no longer registered as agent tools and had no callers. The underlying tools.* functions remain, still used by tests and the VFS backend. --- ask_alyf/ask_alyf/toolset.py | 124 ----------------------------------- 1 file changed, 124 deletions(-) diff --git a/ask_alyf/ask_alyf/toolset.py b/ask_alyf/ask_alyf/toolset.py index a5974c5..66af924 100644 --- a/ask_alyf/ask_alyf/toolset.py +++ b/ask_alyf/ask_alyf/toolset.py @@ -322,130 +322,6 @@ def translate_ui_labels( ) return tools.translate_ui_labels(labels=labels, language=language or request_language) - def search_code(self, query: str, relative_path: str = "", limit: int = 20) -> list[dict[str, Any]]: - """Search installed app code for matching text. - - Args: - query: The text to search for. - relative_path: Optional installed-app-relative path. Providing at least the app name will make it much faster. - limit: Maximum number of matches to return. - - Returns: - A list of code search matches. - """ - self.runtime.emit_status(_("Searching code...")) - return tools.search_code(query=query, relative_path=relative_path, limit=limit) - - def read_code_file(self, path: str, start_line: int = 1, end_line: int = 200) -> dict[str, Any]: - """Read a file from installed app code. - - Args: - path: Bench-relative path inside an installed app, such as apps/my_app/my_app/module.py. - start_line: First line to include. - end_line: Last line to include. - - Returns: - The selected file content and line range. - """ - self.runtime.emit_status(_("Reading code file...")) - return tools.read_code_file(path=path, start_line=start_line, end_line=end_line) - - def ls( - self, - app_name: str, - relative_path: str = "", - recursive: bool = False, - include_hidden: bool = False, - limit: int = 200, - ) -> dict[str, Any]: - """List files or directories in an installed app, similar to Debian ls. - - Args: - app_name: The installed app name. - relative_path: Optional path inside the app. - recursive: Whether to include nested descendants. - include_hidden: Whether to include hidden files and folders. - limit: Maximum number of entries to return. - - Returns: - A directory listing payload. - """ - self.runtime.emit_status(_("Listing code files...")) - return tools.ls( - app_name=app_name, - relative_path=relative_path, - recursive=recursive, - include_hidden=include_hidden, - limit=limit, - ) - - def find( - self, - app_name: str, - name_pattern: str = "*", - relative_path: str = "", - entry_type: str = "any", - include_hidden: bool = False, - limit: int = 200, - ) -> dict[str, Any]: - """Find files or directories in an installed app, similar to Debian find. - - Args: - app_name: The installed app name. - name_pattern: Shell-style filename pattern, such as *.py. - relative_path: Optional path inside the app to search from. - entry_type: One of any, file, or directory. - include_hidden: Whether to include hidden files and folders. - limit: Maximum number of matches to return. - - Returns: - A find-style search payload. - """ - self.runtime.emit_status(_("Finding code paths...")) - return tools.find( - app_name=app_name, - name_pattern=name_pattern, - relative_path=relative_path, - entry_type=entry_type, - include_hidden=include_hidden, - limit=limit, - ) - - def grep( - self, - app_name: str, - query: str, - relative_path: str = "", - file_pattern: str = "*", - case_sensitive: bool = False, - include_hidden: bool = False, - limit: int = 50, - ) -> dict[str, Any]: - """Search file contents in an installed app, similar to Debian grep. - - Args: - app_name: The installed app name. - query: The text to search for. - relative_path: Optional path inside the app to search from. - file_pattern: Optional shell-style filename filter, such as *.py. - case_sensitive: Whether matching should be case-sensitive. - include_hidden: Whether to include hidden files and folders. - limit: Maximum number of matches to return. - - Returns: - A grep-style search payload. - """ - self.runtime.emit_status(_("Searching file contents...")) - return tools.grep( - app_name=app_name, - query=query, - relative_path=relative_path, - file_pattern=file_pattern, - case_sensitive=case_sensitive, - include_hidden=include_hidden, - limit=limit, - ) - def get_file_id( self, reference_doctype: str,