diff --git a/application/api/user/agents/portability.py b/application/api/user/agents/portability.py index 34ae3ec7..00b5888d 100644 --- a/application/api/user/agents/portability.py +++ b/application/api/user/agents/portability.py @@ -10,10 +10,18 @@ Re-import is idempotent at every level: the agent is matched by prompts by ``(name, content)``, custom models by ``(display_name, upstream_model_id, base_url)`` — so the same file applied twice updates rather than duplicating. + +Workflow agents additionally carry their graph under ``spec.workflow``. +The graph's agent nodes reference tools, sources and models by raw id in +their config; export rewrites those into the same portable identities the +top-level sections use (``tool-N`` keys into ``spec.tools``, source names, +custom-model display names), and import resolves them back and writes the +graph through the same validation gate as the workflow API. """ from __future__ import annotations +import copy import re from typing import Any, Optional @@ -39,6 +47,9 @@ from application.storage.db.repositories.user_custom_models import ( UserCustomModelsRepository, ) from application.storage.db.repositories.user_tools import UserToolsRepository +from application.storage.db.repositories.workflow_edges import WorkflowEdgesRepository +from application.storage.db.repositories.workflow_nodes import WorkflowNodesRepository +from application.storage.db.repositories.workflows import WorkflowsRepository from application.storage.db.session import db_readonly, db_session @@ -49,12 +60,18 @@ KIND = "Agent" # child resources a single import may reference / create. MAX_IMPORT_BYTES = 512 * 1024 MAX_LIST_ITEMS = 100 +MAX_WORKFLOW_NODES = 200 +MAX_WORKFLOW_EDGES = 400 class AgentImportError(Exception): """Raised when an agent YAML document is malformed or unsupported.""" +class AgentExportError(Exception): + """Raised when an agent cannot be exported as a document import would accept.""" + + class _SafeNoAliasLoader(yaml.SafeLoader): """SafeLoader that also rejects anchors/aliases (billion-laughs guard).""" @@ -145,12 +162,14 @@ def _safe_export_config(stored_config: dict, config_requirements: dict) -> dict: Export must never leak a credential. A key is emitted ONLY if ``config_requirements`` declares it and does not mark it secret. - Anything not described by requirements — e.g. ``api_tool``/MCP free-form - ``headers``, ``url`` query strings, ``server_url`` that routinely embed - tokens, or tools with empty ``config_requirements`` — is dropped. Reuse - on import matches the existing tool; create re-collects config from the - user. This blocklist-free approach can't be defeated by a stale or empty - ``config_requirements``. + Anything not described by requirements — e.g. MCP free-form ``headers``, + ``url`` query strings, ``server_url`` that routinely embed tokens, or + tools with empty ``config_requirements`` — is dropped. Reuse on import + matches the existing tool; create re-collects config from the user. This + blocklist-free approach can't be defeated by a stale or empty + ``config_requirements``. (``api_tool`` is the one exception: its + user-authored ``config["actions"]`` travel via + ``_export_api_tool_actions``, redacted under the same posture.) """ requirements = config_requirements or {} safe: dict[str, Any] = {} @@ -161,6 +180,224 @@ def _safe_export_config(stored_config: dict, config_requirements: dict) -> dict: return safe +# Param-type sections an action may carry. Fixed ``value``s are exported only +# for ``parameters`` (declared function args, treated as non-secret throughout: +# returned to the client, shown in the actions editor). The free-form HTTP +# types routinely hold credentials (an ``Authorization`` header fixed as a +# value), so for those only the ``filled_by_llm`` flag travels — same posture +# as ``_safe_export_config`` dropping free-form config. +# +# Scope note: ``actions``-column rows come from ``transform_actions`` over live +# tool metadata, which today only ever emits ``parameters`` sections — the +# HTTP types appear only in ``api_tool``'s ``config["actions"]``, which travel +# through ``_export_api_tool_actions`` / ``_sanitize_api_tool_actions`` below. +# The full tuple is kept here so a hand-edited file naming those sections is +# handled (flags applied, values refused) rather than silently ignored. +_ACTION_PARAM_TYPES = ("parameters", "query_params", "headers", "body") +_ACTION_VALUE_SAFE_TYPES = frozenset({"parameters"}) + +# Action keys copied verbatim by ``_export_api_tool_actions`` — everything +# except ``url`` (query-string-stripped) and the param sections (redacted). +_API_ACTION_SAFE_KEYS = ( + "name", + "description", + "method", + "active", + "require_approval", + "body_content_type", + "body_encoding_rules", +) + + +def _export_api_tool_actions(stored_config: dict) -> dict: + """Redacted copy of an ``api_tool``'s ``config["actions"]``. + + ``api_tool`` is the one tool type whose actions are user-authored data in + ``config["actions"]`` (a dict keyed by action name) rather than the + ``actions`` column (the executor reads exactly that — see ToolExecutor). + The action *shape* — endpoint, method, param schemas, ``filled_by_llm`` + flags, fixed ``parameters`` values — is what makes the tool work, so it + travels. Fixed values in the free-form HTTP sections and URL query + strings routinely hold credentials, so those are blanked — same posture + as ``_safe_export_config``. + """ + actions = stored_config.get("actions") + if not isinstance(actions, dict): + return {} + out: dict[str, Any] = {} + for name, action in actions.items(): + if not isinstance(action, dict): + continue + entry: dict[str, Any] = { + k: copy.deepcopy(action[k]) for k in _API_ACTION_SAFE_KEYS if k in action + } + entry["url"] = str(action.get("url") or "").split("?", 1)[0] + for param_type in _ACTION_PARAM_TYPES: + section = action.get(param_type) + if not isinstance(section, dict): + continue + section = copy.deepcopy(section) + props = section.get("properties") + if param_type not in _ACTION_VALUE_SAFE_TYPES and isinstance(props, dict): + for details in props.values(): + if isinstance(details, dict) and details.get("value"): + details["value"] = "" + entry[param_type] = section + out[str(name)] = entry + return out + + +def _sanitize_api_tool_actions(raw, tool_label: str, warnings: list) -> dict: + """Vet a hand-editable ``config["actions"]`` block before storing it. + + Unlike column actions — whose schema comes from live tool code, so the + YAML may contribute flags and scalar values only — ``api_tool`` actions + ARE user-authored schema, and the whole definition is accepted. Each + action's URL goes through the same SSRF gate as the tool-config routes + (``_validate_tool_urls``); an action with a missing or unsafe URL is + dropped with a warning rather than failing the import. + """ + if not isinstance(raw, dict): + return {} + if len(raw) > MAX_LIST_ITEMS: + raise AgentImportError(f"Too many api_tool actions (max {MAX_LIST_ITEMS})") + out: dict[str, Any] = {} + for name, action in raw.items(): + if not isinstance(action, dict): + continue + url = str(action.get("url") or "").strip() + if not url: + warnings.append( + f"Tool '{tool_label}' action '{name}' has no URL; dropped" + ) + continue + try: + validate_url(url) + except SSRFError: + warnings.append( + f"Tool '{tool_label}' action '{name}' has an unsafe URL; dropped" + ) + continue + # Export blanks fixed values in the free-form HTTP sections (they + # routinely hold credentials); a param left pinned with an empty + # value is silently omitted from the request at run time, so + # surface it now instead. + pinned_empty = [ + f"{param_type}.{pname}" + for param_type in _ACTION_PARAM_TYPES + if param_type not in _ACTION_VALUE_SAFE_TYPES + and isinstance(action.get(param_type), dict) + for pname, details in (action[param_type].get("properties") or {}).items() + if isinstance(details, dict) + and details.get("filled_by_llm") is False + and not details.get("value") + ] + if pinned_empty: + warnings.append( + f"Tool '{tool_label}' action '{name}': fixed values for " + f"{', '.join(pinned_empty)} don't travel in the file; set them " + "in the tool's settings or the request will omit them" + ) + out[str(name)] = action + return out + + +def _export_action_overrides(actions) -> list: + """Compact per-action customizations that deviate from factory defaults. + + ``transform_actions`` initializes every action ``active`` with every param + ``filled_by_llm: true, value: ""`` — anything else is a user customization + (a fixed ``chat_id``, a disabled action) that must survive export, or the + imported tool silently reverts to "let the LLM fill everything". + """ + out = [] + for action in actions or []: + if not isinstance(action, dict) or not action.get("name"): + continue + override: dict[str, Any] = {"name": action["name"]} + customized = False + if action.get("active", True) is False: + override["active"] = False + customized = True + for param_type in _ACTION_PARAM_TYPES: + section = action.get(param_type) + props = section.get("properties") if isinstance(section, dict) else None + if not isinstance(props, dict): + continue + for pname, details in props.items(): + if not isinstance(details, dict): + continue + filled = details.get("filled_by_llm", True) + value = details.get("value", "") + if filled and not value: + continue + entry: dict[str, Any] = {"filled_by_llm": bool(filled)} + if param_type in _ACTION_VALUE_SAFE_TYPES: + entry["value"] = value + override.setdefault(param_type, {})[pname] = entry + customized = True + if customized: + out.append(override) + return out + + +def _apply_action_overrides(base_actions: list, overrides, tool_type: str, warnings: list) -> list: + """Overlay exported customizations onto freshly transformed live metadata. + + Matched strictly by (action name, param-type, param name): the YAML + contributes flags and scalar values only, never schema — a hand-edited + file can't inject a doctored action definition — and the merge survives + tool-version drift. Unknown names are skipped with a warning. Values are + accepted only for the ``parameters`` type, mirroring export. + """ + if not isinstance(overrides, list) or not overrides: + return base_actions + by_name = {a.get("name"): a for a in base_actions if isinstance(a, dict)} + for override in overrides: + if not isinstance(override, dict): + continue + name = override.get("name") or "" + action = by_name.get(name) + if action is None: + warnings.append( + f"Tool '{tool_type}' action '{name}' no longer exists; customization skipped" + ) + continue + if override.get("active") is False: + action["active"] = False + for param_type in _ACTION_PARAM_TYPES: + override_props = override.get(param_type) + if not isinstance(override_props, dict): + continue + section = action.get(param_type) + props = section.get("properties") if isinstance(section, dict) else {} + for pname, entry in override_props.items(): + details = props.get(pname) if isinstance(props, dict) else None + if not isinstance(details, dict) or not isinstance(entry, dict): + warnings.append( + f"Tool '{tool_type}' action '{name}' has no parameter " + f"'{pname}'; customization skipped" + ) + continue + details["filled_by_llm"] = bool(entry.get("filled_by_llm", True)) + if param_type in _ACTION_VALUE_SAFE_TYPES: + value = entry.get("value", "") + if isinstance(value, (str, int, float, bool)) or value is None: + details["value"] = "" if value is None else value + elif not details["filled_by_llm"] and not details.get("value"): + # Pinned but valueless: the executor omits the param from + # the LLM schema AND skips the empty fixed value, so the + # request silently goes out without it. Values for the + # free-form HTTP types never travel in the file, so this + # can only be repaired in the tool's settings. + warnings.append( + f"Tool '{tool_type}' action '{name}': fixed value for " + f"{param_type} '{pname}' doesn't travel in the file; set " + "it in the tool's settings or the request will omit it" + ) + return base_actions + + # --------------------------------------------------------------------------- # Export # --------------------------------------------------------------------------- @@ -206,13 +443,26 @@ def _serialize_tools(conn, agent: dict, user: str) -> list: for tid in agent.get("tools") or []: tid_str = str(tid) if is_synthesized_tool_id(tid_str): - out.append({"type": synthesized_tool_name_for_id(tid_str), "builtin": True}) + out.append( + { + "type": synthesized_tool_name_for_id(tid_str), + "builtin": True, + "ref": tid_str, + } + ) continue row = repo.get_any(tid_str, user) if not row: continue requirements = row.get("config_requirements") or {} requires_secrets = _secret_field_names(requirements) + config = _safe_export_config(row.get("config") or {}, requirements) + if (row.get("name") or "") == "api_tool": + # api_tool's actions live in config, not the actions column — an + # export without them imports as a dead tool. + exported_actions = _export_api_tool_actions(row.get("config") or {}) + if exported_actions: + config["actions"] = exported_actions entry: dict[str, Any] = { "type": row.get("name") or "", # Raw custom_name (may be "") so import matching is symmetric and @@ -220,9 +470,12 @@ def _serialize_tools(conn, agent: dict, user: str) -> list: "name": row.get("custom_name") or "", "display_name": row.get("display_name") or "", "description": row.get("description") or "", - "config": _safe_export_config(row.get("config") or {}, requirements), + "config": config, "ref": tid_str, } + action_overrides = _export_action_overrides(row.get("actions")) + if action_overrides: + entry["actions"] = action_overrides if requires_secrets: entry["requires_secrets"] = requires_secrets out.append(entry) @@ -266,6 +519,160 @@ def _serialize_models(conn, agent: dict, user: str) -> dict: return {"default": default_value, "available": available} +def _effective_node_config(config: dict) -> dict: + """Return the dict a node's settings actually live in. + + The builder writes some node configs nested under ``config`` while other + payloads keep them flat; the engine reads + ``node.config.get("config", node.config)``. Mirror that here so reference + rewriting touches the dict the engine will read. + """ + nested = config.get("config") + return nested if isinstance(nested, dict) else config + + +def _strip_shadowed_ref_keys(config: dict) -> None: + """Drop flat ``tools``/``sources``/``model_id`` shadowed by a nested config. + + When a node stores its settings under ``config["config"]``, the engine + reads only the nested dict — flat leftovers of those keys are inert, but + on export they would leak the exporter's raw ids into the file, and on + import they would store a hand-edited file's raw ids verbatim. + """ + if isinstance(config.get("config"), dict): + for key in ("tools", "sources", "model_id"): + config.pop(key, None) + + +def _serialize_workflow(conn, agent: dict, user: str): + """Build the portable ``spec.workflow`` block plus its reference sections. + + Agent nodes hold raw tool/source/model ids; those are collected (deduped, + in first-appearance order) into the same top-level ``tools`` / ``sources`` + / ``model`` sections a classic export uses, and the node configs are + rewritten to portable keys: ``tool-N``, the source's name, a custom + model's display name. Returns ``(workflow_block, tools, sources, model)``; + the block is None when the agent has no resolvable workflow. + """ + empty_model = {"default": "", "available": []} + workflow_id = agent.get("workflow_id") + if not workflow_id: + return None, [], [], empty_model + row = WorkflowsRepository(conn).get(str(workflow_id), user) + if row is None: + return None, [], [], empty_model + # Lazy import, same as _apply_workflow — keeps route modules decoupled at load. + from application.api.user.workflows.routes import get_workflow_graph_version + + version = get_workflow_graph_version(row) + node_rows = WorkflowNodesRepository(conn).find_by_version(str(row["id"]), version) + edge_rows = WorkflowEdgesRepository(conn).find_by_version(str(row["id"]), version) + + # Pass 1: collect referenced ids across agent nodes, deduped and ordered. + tool_ids: list[str] = [] + source_ids: list[str] = [] + model_ids: list[str] = [] + for n in node_rows: + if n.get("node_type") != "agent": + continue + cfg = _effective_node_config(n.get("config") or {}) + for tid in cfg.get("tools") or []: + if str(tid) and str(tid) not in tool_ids: + tool_ids.append(str(tid)) + for sid in cfg.get("sources") or []: + if str(sid) and str(sid) not in source_ids: + source_ids.append(str(sid)) + mid = cfg.get("model_id") + if mid and str(mid) not in model_ids: + model_ids.append(str(mid)) + + # Serialize the references through the classic per-resource serializers so + # secret redaction and identity rules stay in one place. Unresolvable ids + # (dangling rows) are dropped from both the section and the node config; + # every emitted entry carries its ``ref``, which keys the maps below. + tools: list = [] + tool_key_by_id: dict[str, str] = {} + for entry in _serialize_tools(conn, {"tools": tool_ids}, user): + tool_key_by_id[str(entry["ref"])] = _tool_key(len(tools)) + tools.append(entry) + + sources: list = [] + source_name_by_id: dict[str, str] = {} + for entry in _serialize_sources(conn, {"extra_source_ids": source_ids}, user): + source_name_by_id[str(entry["ref"])] = entry["name"] + sources.append(entry) + + model = _serialize_models(conn, {"models": model_ids, "default_model_id": ""}, user) + model_name_by_id: dict[str, str] = {} + for entry in model["available"]: + if isinstance(entry, dict): + model_name_by_id[str(entry.get("ref") or "")] = entry.get("display_name") or "" + elif isinstance(entry, str): + model_name_by_id[entry] = entry + + # ``parse_agent_yaml`` bounds these sections at import (DoS cap), so a + # graph referencing more would export a file that import then rejects — + # refuse here with a clear message instead. + if ( + len(tools) > MAX_LIST_ITEMS + or len(sources) > MAX_LIST_ITEMS + or len(model["available"]) > MAX_LIST_ITEMS + ): + raise AgentExportError( + "This workflow references more than " + f"{MAX_LIST_ITEMS} tools, sources, or models and cannot be exported" + ) + + # Pass 2: emit nodes/edges with agent node configs rewritten to the + # portable keys. (Node settings sit under ``config`` here; the workflow + # API's wire shape calls that field ``data`` — import converts.) + nodes_out = [] + for n in node_rows: + config = copy.deepcopy(n.get("config") or {}) + if n.get("node_type") == "agent": + cfg = _effective_node_config(config) + cfg["tools"] = [ + tool_key_by_id[str(t)] + for t in cfg.get("tools") or [] + if str(t) in tool_key_by_id + ] + cfg["sources"] = [ + source_name_by_id[str(s)] + for s in cfg.get("sources") or [] + if str(s) in source_name_by_id + ] + if cfg.get("model_id"): + cfg["model_id"] = model_name_by_id.get(str(cfg["model_id"])) + _strip_shadowed_ref_keys(config) + nodes_out.append( + { + "id": n["node_id"], + "type": n["node_type"], + "title": n.get("title") or "", + "description": n.get("description") or "", + "position": n.get("position") or {"x": 0, "y": 0}, + "config": config, + } + ) + edges_out = [ + { + "id": e["edge_id"], + "source": e.get("source_id"), + "target": e.get("target_id"), + "sourceHandle": e.get("source_handle"), + "targetHandle": e.get("target_handle"), + } + for e in edge_rows + ] + block = { + "name": row.get("name") or "", + "description": row.get("description") or "", + "nodes": nodes_out, + "edges": edges_out, + } + return block, tools, sources, model + + def _import_config(spec: dict) -> dict: """Validate a spec's ``config`` through the same gate as the API. @@ -286,16 +693,28 @@ def _import_config(spec: dict) -> dict: def serialize_agent(conn, agent: dict, user: str) -> dict: """Build the portable export document for an agent row.""" + is_workflow = (agent.get("agent_type") or "") == "workflow" + if is_workflow: + # A workflow agent's behavior lives in its graph: the agent-level + # prompt/source/tool/model slots are unused, so the top-level sections + # instead carry the references collected from the graph's agent nodes. + workflow_block, tools, sources, model = _serialize_workflow(conn, agent, user) + prompt = "default" + else: + tools = _serialize_tools(conn, agent, user) + sources = _serialize_sources(conn, agent, user) + model = _serialize_models(conn, agent, user) + prompt = _serialize_prompt(conn, agent, user) spec = { "name": agent.get("name") or "", "description": agent.get("description") or "", "agent_type": agent.get("agent_type") or "classic", "retriever": agent.get("retriever") or "classic", "chunks": int(agent["chunks"]) if agent.get("chunks") is not None else None, - "prompt": _serialize_prompt(conn, agent, user), - "model": _serialize_models(conn, agent, user), - "sources": _serialize_sources(conn, agent, user), - "tools": _serialize_tools(conn, agent, user), + "prompt": prompt, + "model": model, + "sources": sources, + "tools": tools, "limits": { "limited_token_mode": bool(agent.get("limited_token_mode", False)), "token_limit": agent.get("token_limit"), @@ -306,6 +725,8 @@ def serialize_agent(conn, agent: dict, user: str) -> dict: "allow_system_prompt_override": bool(agent.get("allow_system_prompt_override", False)), "config": agent.get("config") or {}, } + if is_workflow: + spec["workflow"] = workflow_block return { "apiVersion": API_VERSION, "kind": KIND, @@ -342,6 +763,33 @@ def agent_to_yaml(export: dict) -> str: # --------------------------------------------------------------------------- +def _validate_workflow_shape(spec: dict) -> None: + """Shallow-validate ``spec.workflow`` (deep graph validation runs at apply). + + A missing key or explicit null is allowed — a draft workflow agent may have + no graph yet; only a present block must be well-formed and bounded. + """ + workflow = spec.get("workflow") + if workflow is None: + return + if not isinstance(workflow, dict): + raise AgentImportError("spec.workflow must be a mapping") + nodes = workflow.get("nodes") + edges = workflow.get("edges") + if not isinstance(nodes, list) or not isinstance(edges, list): + raise AgentImportError("spec.workflow.nodes and spec.workflow.edges must be lists") + if len(nodes) > MAX_WORKFLOW_NODES: + raise AgentImportError(f"Too many workflow nodes (max {MAX_WORKFLOW_NODES})") + if len(edges) > MAX_WORKFLOW_EDGES: + raise AgentImportError(f"Too many workflow edges (max {MAX_WORKFLOW_EDGES})") + for node in nodes: + if not isinstance(node, dict) or not node.get("id") or not node.get("type"): + raise AgentImportError("Every workflow node needs an 'id' and a 'type'") + for edge in edges: + if not isinstance(edge, dict) or not edge.get("id"): + raise AgentImportError("Every workflow edge needs an 'id'") + + def parse_agent_yaml(text: str) -> dict: """Parse and shallow-validate an agent YAML document.""" if text and len(text) > MAX_IMPORT_BYTES: @@ -362,7 +810,7 @@ def parse_agent_yaml(text: str) -> dict: if not spec.get("name"): raise AgentImportError("spec.name is required") if spec.get("agent_type") == "workflow": - raise AgentImportError("Workflow agents are not supported for import yet") + _validate_workflow_shape(spec) for key in ("sources", "tools"): value = spec.get(key) if isinstance(value, list) and len(value) > MAX_LIST_ITEMS: @@ -440,6 +888,38 @@ def _tool_key(index: int) -> str: # --------------------------------------------------------------------------- +def _plan_workflow_removal(conn, user: str, target: dict) -> Optional[dict]: + """Describe what an explicit ``workflow: null`` would destroy, if anything. + + Returns a ``delete`` block sized with the current graph, or None when the + workflow survives — the target is new, published (``_apply_workflow`` + keeps a published agent's graph), has no workflow, or another agent still + references it. + """ + if target.get("action") != "update" or not target.get("agent_id"): + return None + if (target.get("status") or "") == "published": + return None + agent_row = AgentsRepository(conn).get(str(target["agent_id"]), user) or {} + existing_id = agent_row.get("workflow_id") + if not existing_id: + return None + row = WorkflowsRepository(conn).get(str(existing_id), user) + if row is None: + return None + if AgentsRepository(conn).count_by_workflow(str(existing_id), user) > 1: + return None + # Lazy import, same as _apply_workflow — keeps route modules decoupled at load. + from application.api.user.workflows.routes import get_workflow_graph_version + + version = get_workflow_graph_version(row) + return { + "action": "delete", + "nodes": len(WorkflowNodesRepository(conn).find_by_version(str(row["id"]), version)), + "edges": len(WorkflowEdgesRepository(conn).find_by_version(str(row["id"]), version)), + } + + def plan_import(conn, user: str, doc: dict) -> dict: """Resolve every reference without writing; returns a resolution report.""" spec = doc["spec"] @@ -528,7 +1008,36 @@ def plan_import(conn, user: str, doc: dict) -> dict: } ) - return {"target": target, "sources": sources, "tools": tools, "prompt": prompt, "models": models} + workflow_plan = None + wf_spec = spec.get("workflow") + if spec.get("agent_type") == "workflow": + if isinstance(wf_spec, dict): + action = "create" + if target.get("action") == "update" and target.get("agent_id"): + agent_row = AgentsRepository(conn).get(str(target["agent_id"]), user) + existing_wf = (agent_row or {}).get("workflow_id") + if existing_wf and WorkflowsRepository(conn).get(str(existing_wf), user): + action = "update" + workflow_plan = { + "nodes": len(wf_spec.get("nodes") or []), + "edges": len(wf_spec.get("edges") or []), + "action": action, + } + elif "workflow" in spec: + # Explicit ``workflow: null`` (vs. an omitted key, which changes + # nothing). Apply clears the link and reaps the row when no other + # agent references it, destroying the graph, its run history and + # its artifacts — the dry-run has to say so before the user commits. + workflow_plan = _plan_workflow_removal(conn, user, target) + + return { + "target": target, + "sources": sources, + "tools": tools, + "prompt": prompt, + "models": models, + "workflow": workflow_plan, + } # --------------------------------------------------------------------------- @@ -546,26 +1055,48 @@ def _apply_prompt(conn, user: str, spec: dict) -> Optional[str]: return str(row["id"]) -def _apply_sources(conn, user: str, spec: dict, resolution: dict, warnings: list) -> list: +def _apply_sources(conn, user: str, spec: dict, resolution: dict, warnings: list): + """Resolve spec sources; returns ``(ids, id_by_name)``. + + ``id_by_name`` maps each spec source name to its resolved id (or None) so + a workflow graph's node configs — which reference sources by name — can + be rewritten from the same resolution. + """ repo = SourcesRepository(conn) mapping = resolution.get("sources") if not isinstance(mapping, dict): mapping = {} resolved: list[str] = [] - for src in spec.get("sources") or []: + id_by_name: dict[str, Optional[str]] = {} + entries = [src for src in spec.get("sources") or [] if isinstance(src, dict)] + # ``id_by_name`` (and a workflow node's source reference) keys on the name + # alone, but names aren't unique per user — ``find_by_name`` returns the + # oldest match, so duplicates collapse onto one source. Same hazard the + # custom-model display names carry below; surface it rather than silently + # rewiring a node to a different source. + names = [src.get("name") or "" for src in entries] + for dup in sorted({n for n in names if n and names.count(n) > 1}): + warnings.append( + f"Multiple sources share the name '{dup}'; references to it all " + "resolve to the oldest one" + ) + for src in entries: name = src.get("name") or "" + id_by_name.setdefault(name, None) mapped = mapping.get(name) if mapped: # Ownership-check the client-supplied id before linking (IDOR guard). owned = repo.get_any(str(mapped), user) if owned: resolved.append(str(owned["id"])) + id_by_name[name] = str(owned["id"]) else: warnings.append(f"Source mapping for '{name}' is not yours; ignored") continue match = repo.find_by_name(user, name) if match: resolved.append(str(match["id"])) + id_by_name[name] = str(match["id"]) else: warnings.append(f"Source '{name}' not found; left unattached") seen: set = set() @@ -574,7 +1105,7 @@ def _apply_sources(conn, user: str, spec: dict, resolution: dict, warnings: list if sid not in seen: seen.add(sid) out.append(sid) - return out + return out, id_by_name def _validate_tool_urls(tool_type: str, config: dict) -> Optional[str]: @@ -607,6 +1138,19 @@ def _create_tool_from_spec(conn, user: str, tool: dict, secrets: dict, warnings: config_requirements = inst.get_config_requirements() or {} config = dict(tool.get("config") or {}) config.update(secrets or {}) + if tool_type == "api_tool": + label = tool.get("display_name") or tool.get("name") or tool_type + config["actions"] = _sanitize_api_tool_actions( + config.get("actions"), label, warnings + ) + if not config["actions"]: + # Zero actions means a dead tool — the executor resolves api_tool + # calls exclusively through config["actions"]. Reachable via older + # exports (which couldn't carry actions) or a fully-dropped block. + warnings.append( + f"API tool '{label}' was created without any actions; " + "configure its actions in the tool's settings before use" + ) missing = [k for k in _secret_field_names(config_requirements, required_only=True) if not config.get(k)] if missing: warnings.append(f"Tool '{tool_type}' needs secret(s) {missing}; not created") @@ -621,6 +1165,7 @@ def _create_tool_from_spec(conn, user: str, tool: dict, secrets: dict, warnings: warnings.append(f"Tool '{tool_type}' secret encryption failed; not created") return None actions = transform_actions(inst.get_actions_metadata() or []) + actions = _apply_action_overrides(actions, tool.get("actions"), tool_type, warnings) created = UserToolsRepository(conn).create( user, tool_type, @@ -635,16 +1180,31 @@ def _create_tool_from_spec(conn, user: str, tool: dict, secrets: dict, warnings: return str(created["id"]) -def _apply_tools(conn, user: str, spec: dict, resolution: dict, warnings: list) -> list: +def _apply_tools(conn, user: str, spec: dict, resolution: dict, warnings: list): + """Resolve spec tools; returns ``(ids, id_by_key)``. + + ``id_by_key`` maps each spec entry's ``tool-N`` key to its resolved id + (or None when skipped/unavailable) so a workflow graph's node configs — + which reference tools by those keys — can be rewritten from the same + resolution. + """ repo = UserToolsRepository(conn) user_tools = repo.list_for_user(user) decisions = resolution.get("tools") if not isinstance(decisions, dict): decisions = {} out: list[str] = [] + id_by_key: dict[str, Optional[str]] = {} + + def _resolve(key: str, tid: Optional[str]) -> None: + id_by_key[key] = tid + if tid: + out.append(tid) + for index, tool in enumerate(spec.get("tools") or []): + key = _tool_key(index) tool_type = tool.get("type") or "" - decision = decisions.get(_tool_key(index)) + decision = decisions.get(key) if not isinstance(decision, dict): decision = {} choice = decision.get("decision") @@ -652,33 +1212,36 @@ def _apply_tools(conn, user: str, spec: dict, resolution: dict, warnings: list) if tool.get("builtin"): if synthesize_tool_by_name(tool_type) is None: warnings.append(f"Built-in tool '{tool_type}' unavailable; skipped") + _resolve(key, None) continue - out.append(default_tool_id(tool_type)) + _resolve(key, default_tool_id(tool_type)) continue if choice == "skip": + _resolve(key, None) continue if choice == "reuse" and decision.get("tool_id"): # Ownership-check the client-supplied id before linking (IDOR guard). tid = str(decision["tool_id"]) if is_synthesized_tool_id(tid) or repo.get_any(tid, user): - out.append(tid) + _resolve(key, tid) else: warnings.append(f"Tool reuse id '{tid}' is not yours; ignored") + _resolve(key, None) continue custom_name = tool.get("name") or "" match = _find_user_tool(user_tools, tool_type, custom_name) if match and choice != "create": - out.append(str(match["id"])) + _resolve(key, str(match["id"])) continue created_id = _create_tool_from_spec(conn, user, tool, decision.get("secrets") or {}, warnings) + _resolve(key, created_id) if created_id: - out.append(created_id) # Record so a later identical spec entry reuses rather than re-creates. user_tools.append({"id": created_id, "name": tool_type, "custom_name": custom_name}) - return out + return out, id_by_key def _apply_models(conn, user: str, spec: dict, resolution: dict, warnings: list): @@ -691,7 +1254,22 @@ def _apply_models(conn, user: str, spec: dict, resolution: dict, warnings: list) name_to_id: dict[str, str] = {} models: list[str] = [] - for entry in model_spec.get("available") or []: + available = model_spec.get("available") or [] + # ``name_to_id`` (and a workflow node's ``model_id`` reference) keys on + # display_name alone, while custom-model identity is the full + # (display_name, upstream_model_id, base_url) triple — duplicates + # collapse onto the last entry, so surface that instead of silently + # rewiring nodes. + custom_names = [ + e.get("display_name") or "" for e in available if isinstance(e, dict) + ] + for dup in sorted({n for n in custom_names if n and custom_names.count(n) > 1}): + warnings.append( + f"Multiple custom models share the display name '{dup}'; " + "references to it resolve to the last one in the file" + ) + + for entry in available: if isinstance(entry, str): if validate_model_id(entry, user): models.append(entry) @@ -749,13 +1327,189 @@ def _apply_models(conn, user: str, spec: dict, resolution: dict, warnings: list) default_id = str(owned["id"]) else: warnings.append(f"Default model '{raw_default}' not resolved") - return (models or None), (default_id or None) + return (models or None), (default_id or None), name_to_id def _limit(spec: dict, key: str): return (spec.get("limits") or {}).get(key) +# Sentinel: the file omitted ``spec.workflow`` entirely, so the agent's +# existing workflow link is left untouched (vs. explicit null = clear). +_WORKFLOW_UNCHANGED = object() + + +def _rewrite_node_refs( + conn, + user: str, + cfg: dict, + tool_ids_by_key: dict, + source_ids_by_name: dict, + model_ids_by_name: dict, + warnings: list, +) -> None: + """Resolve an agent node's portable references back to the importer's ids. + + Export writes ``tool-N`` keys / source names / model display names, and a + hand-edited file may also carry raw ids — those are accepted only when + they are a builtin synthetic id or ownership-check against the importing + user (IDOR guard). Anything unresolvable is dropped with a warning so the + graph imports degraded rather than pointing at foreign resources. + """ + tools_repo = UserToolsRepository(conn) + sources_repo = SourcesRepository(conn) + + resolved_tools = [] + for ref in cfg.get("tools") or []: + ref = str(ref) + if ref in tool_ids_by_key: + tid = tool_ids_by_key[ref] + if tid: + resolved_tools.append(tid) + else: + warnings.append(f"Workflow node tool '{ref}' not resolved; removed") + elif is_synthesized_tool_id(ref): + resolved_tools.append(ref) + elif looks_like_uuid(ref) and tools_repo.get_any(ref, user): + resolved_tools.append(ref) + else: + warnings.append(f"Workflow node tool '{ref}' not resolved; removed") + cfg["tools"] = resolved_tools + + resolved_sources = [] + for ref in cfg.get("sources") or []: + ref = str(ref) + if ref in source_ids_by_name: + sid = source_ids_by_name[ref] + if sid: + resolved_sources.append(sid) + else: + warnings.append(f"Workflow node source '{ref}' not resolved; removed") + elif looks_like_uuid(ref) and sources_repo.get_any(ref, user): + resolved_sources.append(ref) + else: + warnings.append(f"Workflow node source '{ref}' not resolved; removed") + # Distinct names can collapse onto one source (see ``_apply_sources``), so + # dedupe rather than storing the same id twice on the node. + cfg["sources"] = list(dict.fromkeys(resolved_sources)) + + raw_model = cfg.get("model_id") + if raw_model: + raw_model = str(raw_model) + if raw_model in model_ids_by_name: + cfg["model_id"] = model_ids_by_name[raw_model] + elif validate_model_id(raw_model, user): + cfg["model_id"] = raw_model + else: + warnings.append(f"Workflow node model '{raw_model}' not resolved; removed") + cfg["model_id"] = None + + +def _apply_workflow( + conn, + user: str, + spec: dict, + target: dict, + tool_ids_by_key: dict, + source_ids_by_name: dict, + model_ids_by_name: dict, + warnings: list, +): + """Write the imported graph; returns the workflow id, None, or the sentinel. + + The graph goes through the same validation gate as the workflow API + (``validate_workflow_structure``) so a hand-edited YAML can't install a + graph the write path would have rejected. When the matched agent already + owns a workflow the graph lands as a new version of it (mirroring + ``PUT /api/workflows/``); otherwise a new workflow is created. + """ + if "workflow" not in spec: + return _WORKFLOW_UNCHANGED + wf_spec = spec.get("workflow") + if not isinstance(wf_spec, dict): + return None + from application.api.user.workflows.routes import ( + _write_graph, + get_workflow_graph_version, + normalize_agent_node_json_schemas, + validate_workflow_structure, + ) + + nodes_data = [] + for node in wf_spec.get("nodes") or []: + config = copy.deepcopy(node.get("config") or {}) + if not isinstance(config, dict): + config = {} + if node.get("type") == "agent": + _rewrite_node_refs( + conn, + user, + _effective_node_config(config), + tool_ids_by_key, + source_ids_by_name, + model_ids_by_name, + warnings, + ) + _strip_shadowed_ref_keys(config) + position = node.get("position") + nodes_data.append( + { + "id": str(node.get("id")), + "type": str(node.get("type")), + "title": node.get("title") or "", + "description": node.get("description") or "", + "position": position if isinstance(position, dict) else {"x": 0, "y": 0}, + "data": config, + } + ) + edges_data = [ + { + "id": str(edge.get("id")), + "source": edge.get("source"), + "target": edge.get("target"), + "sourceHandle": edge.get("sourceHandle"), + "targetHandle": edge.get("targetHandle"), + } + for edge in wf_spec.get("edges") or [] + ] + + errors = validate_workflow_structure(nodes_data, edges_data, user_id=user) + if errors: + raise AgentImportError("Workflow validation failed: " + "; ".join(errors[:5])) + nodes_data = normalize_agent_node_json_schemas(nodes_data) + + wf_repo = WorkflowsRepository(conn) + name = wf_spec.get("name") or spec.get("name") or "Imported workflow" + description = wf_spec.get("description") or "" + + existing = None + if target.get("action") == "update" and target.get("agent_id"): + agent_row = AgentsRepository(conn).get(str(target["agent_id"]), user) + existing_id = (agent_row or {}).get("workflow_id") + if existing_id: + existing = wf_repo.get(str(existing_id), user) + if existing is not None: + pg_workflow_id = str(existing["id"]) + next_version = get_workflow_graph_version(existing) + 1 + _write_graph(conn, pg_workflow_id, next_version, nodes_data, edges_data) + wf_repo.update( + pg_workflow_id, + user, + { + "name": name, + "description": description, + "current_graph_version": next_version, + }, + ) + WorkflowNodesRepository(conn).delete_other_versions(pg_workflow_id, next_version) + WorkflowEdgesRepository(conn).delete_other_versions(pg_workflow_id, next_version) + return pg_workflow_id + + created = wf_repo.create(user, name, description=description) + _write_graph(conn, str(created["id"]), 1, nodes_data, edges_data) + return str(created["id"]) + + def apply_import(conn, user: str, doc: dict, resolution: Optional[dict] = None) -> dict: """Create or update an agent from a parsed YAML doc. @@ -778,9 +1532,13 @@ def apply_import(conn, user: str, doc: dict, resolution: Optional[dict] = None) target = _resolve_target(conn, user, metadata) prompt_id = _apply_prompt(conn, user, spec) - source_ids = _apply_sources(conn, user, spec, resolution, warnings) - tool_ids = _apply_tools(conn, user, spec, resolution, warnings) - models, default_model_id = _apply_models(conn, user, spec, resolution, warnings) + source_ids, source_ids_by_name = _apply_sources(conn, user, spec, resolution, warnings) + tool_ids, tool_ids_by_key = _apply_tools(conn, user, spec, resolution, warnings) + models, default_model_id, model_ids_by_name = _apply_models( + conn, user, spec, resolution, warnings + ) + + is_workflow = (spec.get("agent_type") or "classic") == "workflow" agents_repo = AgentsRepository(conn) is_update = bool(target.get("action") == "update" and target.get("agent_id")) @@ -817,6 +1575,62 @@ def apply_import(conn, user: str, doc: dict, resolution: Optional[dict] = None) "config": _import_config(spec), "slug": slug, } + orphaned_workflow_id: Optional[str] = None + if is_workflow: + # A workflow agent's behavior lives in its graph — the agent-level + # prompt/tool/model/source slots in the file exist only to carry the + # graph's references through resolution, so they never land on the row. + authoritative.update( + { + "prompt_id": None, + "tools": [], + "models": None, + "default_model_id": None, + "extra_source_ids": [], + } + ) + prior_workflow_id = None + if is_update: + prior_workflow_id = ( + agents_repo.get(str(target["agent_id"]), user) or {} + ).get("workflow_id") + published = (target.get("status") or "") == "published" + workflow_result = _apply_workflow( + conn, + user, + spec, + target, + tool_ids_by_key, + source_ids_by_name, + model_ids_by_name, + warnings, + ) + # A published workflow agent with no graph can't run — the create and + # update routes both reject that state, so import must not install it + # through the back door (e.g. a file that flips ``agent_type`` to + # workflow over a published classic agent). + if workflow_result is _WORKFLOW_UNCHANGED: + if published and not prior_workflow_id: + raise AgentImportError( + "A published workflow agent needs a workflow; this file has none" + ) + elif workflow_result is None and published: + if not prior_workflow_id: + raise AgentImportError( + "A published workflow agent needs a workflow; this file has none" + ) + warnings.append( + "File has no workflow; the published agent kept its existing one" + ) + else: + authoritative["workflow_id"] = workflow_result + if workflow_result is None and prior_workflow_id: + # Explicit ``workflow: null`` on a draft: remember the old + # link so the row can be reaped after the update — there is + # no workflow-list API independent of agents, so an + # unlinked workflow would be unreachable forever. + orphaned_workflow_id = str(prior_workflow_id) + # Optional fields — applied only when present so a partial file doesn't wipe them. optional = { "retriever": spec.get("retriever") or ("classic" if not source_ids else None), @@ -832,10 +1646,23 @@ def apply_import(conn, user: str, doc: dict, resolution: Optional[dict] = None) # Only a brand-new agent (the create path below) starts as a draft. fields = {**authoritative, **optional, "name": spec.get("name")} if agents_repo.update(str(target["agent_id"]), user, fields): + if orphaned_workflow_id and not agents_repo.count_by_workflow( + orphaned_workflow_id, user + ): + # Nothing references it anymore; delete (cascades nodes/edges + # and reaps run artifacts) rather than stranding the row. + WorkflowsRepository(conn).delete(orphaned_workflow_id, user) + # Destructive and irreversible — the graph, its run history and + # its artifacts (rows and stored bytes) all go. Say so. + warnings.append( + "File has no workflow; the agent's workflow was deleted " + "along with its run history and artifacts" + ) return { "agent_id": str(target["agent_id"]), "action": "updated", "status": target.get("status") or "draft", + "agent_type": authoritative["agent_type"], "slug": slug, "warnings": warnings, } @@ -846,6 +1673,7 @@ def apply_import(conn, user: str, doc: dict, resolution: Optional[dict] = None) "agent_id": str(row["id"]), "action": "created", "status": "draft", + "agent_type": authoritative["agent_type"], "slug": slug, "warnings": warnings, } @@ -906,15 +1734,13 @@ class ExportAgent(Resource): return make_response( jsonify({"success": False, "message": "Agent not found"}), 404 ) - if (agent.get("agent_type") or "") == "workflow": - return make_response( - jsonify( - {"success": False, "message": "Workflow agents can't be exported yet"} - ), - 400, - ) agent["slug"] = ensure_agent_slug(conn, agent, user) - export = serialize_agent(conn, agent, user) + try: + export = serialize_agent(conn, agent, user) + except AgentExportError as exc: + return make_response( + jsonify({"success": False, "message": str(exc)}), 400 + ) body = agent_to_yaml(export) filename = f"{export['metadata'].get('slug') or 'agent'}.agent.yaml" response = make_response(body, 200) @@ -961,6 +1787,11 @@ class ImportAgent(Resource): try: with db_session() as conn: result = apply_import(conn, user, doc, resolution) + except AgentImportError as exc: + # Apply-time rejection of the user's document (e.g. the workflow + # graph failing the API's validation gate) — a client error, with + # the detail surfaced, matching how PUT /workflows reports it. + return make_response(jsonify({"success": False, "message": str(exc)}), 400) except Exception: current_app.logger.error("Agent import failed", exc_info=True) return make_response( diff --git a/application/api/user/agents/routes.py b/application/api/user/agents/routes.py index fc8f3753..f71c7afb 100644 --- a/application/api/user/agents/routes.py +++ b/application/api/user/agents/routes.py @@ -29,8 +29,13 @@ from application.api.user.team_sharing import ( team_access_for, visible_with_access, ) +from application.agents.default_tools import is_synthesized_tool_id from application.storage.db.repositories.agent_folders import AgentFoldersRepository from application.storage.db.repositories.agents import AgentsRepository +from application.storage.db.repositories.user_custom_models import ( + UserCustomModelsRepository, +) +from application.storage.db.repositories.user_tools import UserToolsRepository from application.storage.db.repositories.conversations import ConversationsRepository from application.storage.db.repositories.shared_conversations import ( SharedConversationsRepository, @@ -38,8 +43,6 @@ from application.storage.db.repositories.shared_conversations import ( from application.storage.db.repositories.stack_logs import StackLogsRepository from application.storage.db.repositories.token_usage import TokenUsageRepository from application.storage.db.repositories.users import UsersRepository -from application.storage.db.repositories.workflow_edges import WorkflowEdgesRepository -from application.storage.db.repositories.workflow_nodes import WorkflowNodesRepository from application.storage.db.repositories.workflows import WorkflowsRepository from application.storage.db.session import db_readonly, db_session from application.utils import ( @@ -1473,15 +1476,13 @@ class DeleteAgent(Resource): workflow_id = agent.get("workflow_id") # For workflow-type agents, delete the owned workflow in the # same transaction. workflow_nodes/workflow_edges cascade - # via ON DELETE CASCADE so a single workflow delete suffices. + # via ON DELETE CASCADE so the single owner-scoped delete + # suffices — and it must stay the ONLY delete: an explicit + # (unscoped) node/edge cleanup here once let deleting an agent + # whose ``workflow_id`` pointed at another user's workflow gut + # that user's graph. if agent.get("agent_type") == "workflow" and workflow_id: try: - WorkflowNodesRepository(conn).delete_by_workflow( - str(workflow_id), - ) - WorkflowEdgesRepository(conn).delete_by_workflow( - str(workflow_id), - ) WorkflowsRepository(conn).delete(str(workflow_id), user) except Exception as wf_err: current_app.logger.warning( @@ -1619,6 +1620,71 @@ class GetTemplateAgents(Resource): return make_response(jsonify({"success": False}), 400) +def _filter_adoptable_tool_ids(conn, user: str, tool_ids) -> list: + """Keep only tool ids the adopter can actually run. + + Builtin synthetic ids resolve for everyone; a ``user_tools`` row must be + the adopter's own — the runtime resolves tools strictly owner-scoped, so + a template owner's tool id would just be dropped (with a log line) on + every run. Filtering at adopt time makes the gap visible in the builder + instead of at run time. + """ + tools_repo = UserToolsRepository(conn) + kept = [] + for tid in tool_ids or []: + tid = str(tid) + if is_synthesized_tool_id(tid) or ( + looks_like_uuid(tid) and tools_repo.get_any(tid, user) + ): + kept.append(tid) + return kept + + +def _strip_foreign_node_refs(conn, user: str): + """Build a ``clone_to_user`` node-config transform for adoption. + + Agent nodes carry raw tool/source/model ids from the template owner. The + engine resolves all three owner-scoped at run time (dropping misses with + only a log line), so unresolvable refs would survive the clone as ghost + references — shown in the builder, never used. Strip them instead, so + the adopter sees empty fields to fill in. + """ + + def transform(node_type: str, config: dict) -> dict: + if node_type != "agent": + return config + # Node settings may sit nested under ``config`` — mirror the engine, + # which reads ``node.config.get("config", node.config)``. + nested = config.get("config") + cfg = nested if isinstance(nested, dict) else config + if cfg.get("tools"): + cfg["tools"] = _filter_adoptable_tool_ids(conn, user, cfg["tools"]) + if cfg.get("sources"): + sources = cfg["sources"] + if not isinstance(sources, list): + sources = [sources] + cfg["sources"] = [ + str(s) + for s in sources + if s and can_access(conn, "source", str(s), user) + ] + model_id = cfg.get("model_id") + if ( + model_id + and looks_like_uuid(str(model_id)) + and not UserCustomModelsRepository(conn).get(str(model_id), user) + ): + cfg["model_id"] = None + if cfg is not config: + # Flat leftovers of these keys are ignored by the engine but would + # carry the template owner's raw ids along; drop them. + for key in ("tools", "sources", "model_id"): + config.pop(key, None) + return config + + return transform + + @agents_ns.route("/adopt_agent") class AdoptAgent(Resource): @api.doc(params={"id": "Agent ID"}, description="Adopt an agent by ID") @@ -1667,13 +1733,48 @@ class AdoptAgent(Resource): for col in ( "description", "agent_type", "retriever", "default_model_id", - "source_id", "prompt_id", "folder_id", "workflow_id", + "source_id", "prompt_id", "folder_id", "extra_source_ids", ): val = template.get(col) if val not in (None, ""): create_kwargs[col] = val + # ``workflow_id`` must NOT be copied by reference: the runtime + # resolves it strictly owner-scoped, so the adopter would get + # "Workflow not found or inaccessible" on every run. Deep-copy + # the template's graph into a workflow the adopter owns, + # stripping node refs (tools/sources/models) the adopter can't + # resolve — they are owner-scoped at run time too. + if template.get("workflow_id"): + cloned_workflow = WorkflowsRepository(conn).clone_to_user( + str(template["workflow_id"]), + user, + from_owner=template.get("user_id"), + node_config_transform=_strip_foreign_node_refs(conn, user), + ) + if cloned_workflow is None: + # A workflow agent without its graph can never run — + # fail the adopt instead of publishing a broken agent. + current_app.logger.error( + "Adopt: template %s references workflow %s not owned " + "by %s; adoption refused", + template.get("id"), + template.get("workflow_id"), + template.get("user_id"), + ) + return make_response( + jsonify( + { + "success": False, + "message": "This template's workflow is " + "unavailable; it cannot be adopted", + } + ), + 500, + ) + create_kwargs["workflow_id"] = str(cloned_workflow["id"]) + # The avatar must live under the adopter's own upload # directory: image paths are validated against their owner, # so a copied reference to the template owner's blob would @@ -1683,9 +1784,16 @@ class AdoptAgent(Resource): ) if adopted_image: create_kwargs["image"] = adopted_image - for col in ("tools", "json_schema", "models", "shared_metadata", "config"): + for col in ("json_schema", "models", "shared_metadata", "config"): if template.get(col) is not None: create_kwargs[col] = template[col] + # Tool ids resolve owner-scoped at run time, so the template + # owner's rows would silently drop on every run — keep only + # what the adopter can actually use (builtins, own tools). + if template.get("tools") is not None: + create_kwargs["tools"] = _filter_adoptable_tool_ids( + conn, user, template["tools"] + ) for col in ("chunks", "token_limit", "request_limit"): if template.get(col) is not None: create_kwargs[col] = template[col] diff --git a/application/storage/db/repositories/agents.py b/application/storage/db/repositories/agents.py index cdec8a2a..97234e4a 100644 --- a/application/storage/db/repositories/agents.py +++ b/application/storage/db/repositories/agents.py @@ -160,6 +160,19 @@ class AgentsRepository: ) return [row_to_dict(r) for r in result.fetchall()] + def count_by_workflow(self, workflow_id: str, user_id: str) -> int: + """Number of this user's agents linked to ``workflow_id``.""" + if not looks_like_uuid(workflow_id): + return 0 + result = self._conn.execute( + text( + "SELECT COUNT(*) FROM agents " + "WHERE workflow_id = CAST(:wf AS uuid) AND user_id = :user_id" + ), + {"wf": workflow_id, "user_id": user_id}, + ) + return int(result.scalar() or 0) + def get_by_id(self, agent_id: str) -> Optional[dict]: """Fetch an agent by id with NO ownership scoping. diff --git a/application/storage/db/repositories/workflow_edges.py b/application/storage/db/repositories/workflow_edges.py index 63610c2b..b41b9b48 100644 --- a/application/storage/db/repositories/workflow_edges.py +++ b/application/storage/db/repositories/workflow_edges.py @@ -149,16 +149,6 @@ class WorkflowEdgesRepository: row = result.fetchone() return str(row[0]) if row else None - def delete_by_workflow(self, workflow_id: str) -> int: - result = self._conn.execute( - text( - "DELETE FROM workflow_edges " - "WHERE workflow_id = CAST(:wf_id AS uuid)" - ), - {"wf_id": workflow_id}, - ) - return result.rowcount - def delete_by_version(self, workflow_id: str, graph_version: int) -> int: result = self._conn.execute( text( diff --git a/application/storage/db/repositories/workflow_nodes.py b/application/storage/db/repositories/workflow_nodes.py index e8ef1025..ca24a6f1 100644 --- a/application/storage/db/repositories/workflow_nodes.py +++ b/application/storage/db/repositories/workflow_nodes.py @@ -143,16 +143,6 @@ class WorkflowNodesRepository: row = result.fetchone() return row_to_dict(row) if row is not None else None - def delete_by_workflow(self, workflow_id: str) -> int: - result = self._conn.execute( - text( - "DELETE FROM workflow_nodes " - "WHERE workflow_id = CAST(:wf_id AS uuid)" - ), - {"wf_id": workflow_id}, - ) - return result.rowcount - def delete_by_version(self, workflow_id: str, graph_version: int) -> int: result = self._conn.execute( text( diff --git a/application/storage/db/repositories/workflows.py b/application/storage/db/repositories/workflows.py index bc9e4cc9..fb3f96c1 100644 --- a/application/storage/db/repositories/workflows.py +++ b/application/storage/db/repositories/workflows.py @@ -8,6 +8,7 @@ Covers CRUD on workflow metadata: from __future__ import annotations +import copy from typing import Optional from sqlalchemy import Connection, text @@ -115,6 +116,106 @@ class WorkflowsRepository: row = result.fetchone() return row[0] if row else None + def clone_to_user( + self, + workflow_id: str, + new_owner: str, + *, + from_owner: Optional[str] = None, + node_config_transform=None, + ) -> Optional[dict]: + """Deep-copy a workflow's current graph into a new workflow owned by ``new_owner``. + + The runtime resolves ``agents.workflow_id`` strictly owner-scoped, so a + workflow reference can never be shared across users by id — it must be + copied. Only the current graph version is copied; the clone starts at + version 1. The source is resolved scoped to ``from_owner`` when given + (callers pass the expected owner so a stray id can't clone another + tenant's graph). Omitting ``from_owner`` resolves the source with NO + ownership check — only do that when ``workflow_id`` comes from trusted + server-side state, never from client input. Returns the new workflow + row, or None if the source is missing. + + ``node_config_transform(node_type, config) -> dict`` lets the caller + rewrite each node's config as it is copied (it receives a deep copy) — + e.g. adoption strips refs the new owner can't resolve, since node + configs carry raw tool/source/model ids that are owner-scoped at run + time and would otherwise ride along as ghost references. + """ + from application.storage.db.repositories.workflow_edges import ( + WorkflowEdgesRepository, + ) + from application.storage.db.repositories.workflow_nodes import ( + WorkflowNodesRepository, + ) + + if not looks_like_uuid(workflow_id): + return None + source = ( + self.get(workflow_id, from_owner) + if from_owner is not None + else self.get_by_id(workflow_id) + ) + if source is None: + return None + try: + version = int(source.get("current_graph_version") or 1) + except (TypeError, ValueError): + version = 1 + if version <= 0: + version = 1 + + clone = self.create( + new_owner, source.get("name") or "", description=source.get("description") + ) + clone_id = str(clone["id"]) + + def _cloned_config(n: dict) -> dict: + config = n.get("config") or {} + if node_config_transform is not None: + config = node_config_transform(n["node_type"], copy.deepcopy(config)) + return config if isinstance(config, dict) else {} + + nodes_repo = WorkflowNodesRepository(self._conn) + node_rows = nodes_repo.find_by_version(workflow_id, version) + created_nodes = nodes_repo.bulk_create( + clone_id, + 1, + [ + { + "node_id": n["node_id"], + "node_type": n["node_type"], + "title": n.get("title"), + "description": n.get("description"), + "position": n.get("position") or {"x": 0, "y": 0}, + "config": _cloned_config(n), + } + for n in node_rows + ], + ) + uuid_by_node_id = {n["node_id"]: n["id"] for n in created_nodes} + + edges_repo = WorkflowEdgesRepository(self._conn) + edge_rows = edges_repo.find_by_version(workflow_id, version) + translated = [] + for e in edge_rows: + from_uuid = uuid_by_node_id.get(e.get("source_id")) + to_uuid = uuid_by_node_id.get(e.get("target_id")) + if not from_uuid or not to_uuid: + continue + translated.append( + { + "edge_id": e["edge_id"], + "from_node_id": from_uuid, + "to_node_id": to_uuid, + "source_handle": e.get("source_handle"), + "target_handle": e.get("target_handle"), + "config": e.get("config") or {}, + } + ) + edges_repo.bulk_create(clone_id, 1, translated) + return clone + def delete(self, workflow_id: str, user_id: str) -> bool: # Run artifacts parent to ``workflow_runs`` (which cascade-delete with the # workflow), but ``artifacts.workflow_run_id`` is a bare uuid with no FK, so diff --git a/frontend/src/agents/AgentCard.tsx b/frontend/src/agents/AgentCard.tsx index cc12413b..64ae3dc3 100644 --- a/frontend/src/agents/AgentCard.tsx +++ b/frontend/src/agents/AgentCard.tsx @@ -17,12 +17,14 @@ import Trash from '../assets/red-trash.svg'; import ThreeDots from '../assets/three-dots.svg'; import UnPin from '../assets/unpin.svg'; import { Avatar } from '../components/ui/avatar'; +import { Button } from '../components/ui/button'; import { DropdownMenu, DropdownMenuContent, DropdownMenuItem, DropdownMenuTrigger, } from '../components/ui/dropdown-menu'; +import { Modal } from '../components/ui/modal'; import ConfirmationModal from '../modals/ConfirmationModal'; import MoveToFolderModal from '../modals/MoveToFolderModal'; import { ActiveState } from '../models/misc'; @@ -67,6 +69,7 @@ export default function AgentCard({ useState('INACTIVE'); const [moveModalState, setMoveModalState] = useState('INACTIVE'); const [shareModalOpen, setShareModalOpen] = useState(false); + const [exportError, setExportError] = useState(null); const menuOptionsConfig: Record = { template: [ @@ -297,7 +300,18 @@ export default function AgentCard({ const handleExport = async () => { try { const response = await userService.exportAgent(agent.id ?? '', token); - if (!response.ok) throw new Error('Failed to export agent'); + if (!response.ok) { + const message = await response + .json() + .then((data) => data?.message) + .catch(() => null); + // Server-side refusals (e.g. a workflow referencing too many + // resources) carry an actionable message; flag it so the catch can + // tell it apart from a transport error like "Failed to fetch". + const error = new Error(message || t('agents.exportAgentFailed')); + error.name = 'ExportRefused'; + throw error; + } const yamlText = await response.text(); const blob = new Blob([yamlText], { type: 'application/x-yaml' }); const url = URL.createObjectURL(blob); @@ -310,6 +324,11 @@ export default function AgentCard({ URL.revokeObjectURL(url); } catch (error) { console.error('Error:', error); + setExportError( + error instanceof Error && error.name === 'ExportRefused' + ? error.message + : t('agents.exportAgentFailed'), + ); } }; @@ -461,6 +480,25 @@ export default function AgentCard({ cancelLabel="Cancel" variant="danger" /> + { + if (!open) setExportError(null); + }} + title={t('agents.exportAgentFailed')} + size="sm" + footer={ + + } + > +

{exportError}

+
(null); const [yamlText, setYamlText] = useState(''); const [fileName, setFileName] = useState(''); @@ -85,7 +90,7 @@ export default function ImportAgentModal({ const [modelKeys, setModelKeys] = useState>({}); const [warnings, setWarnings] = useState(null); const [importedStatus, setImportedStatus] = useState(null); - const [goToAgentId, setGoToAgentId] = useState(null); + const [goToEditPath, setGoToEditPath] = useState(null); const reset = () => { setYamlText(''); @@ -99,7 +104,7 @@ export default function ImportAgentModal({ setModelKeys({}); setWarnings(null); setImportedStatus(null); - setGoToAgentId(null); + setGoToEditPath(null); }; const handleClose = () => { @@ -107,9 +112,7 @@ export default function ImportAgentModal({ reset(); }; - const handleFileChange = async (e: React.ChangeEvent) => { - const selectedFile = e.target.files?.[0]; - if (!selectedFile) return; + const processFile = async (selectedFile: File) => { const name = selectedFile.name.toLowerCase(); if (!name.endsWith('.yaml') && !name.endsWith('.yml')) { setError(t('modals.importAgent.invalidFileType')); @@ -121,6 +124,29 @@ export default function ImportAgentModal({ setYamlText(await selectedFile.text()); }; + const { getRootProps, getInputProps, isDragActive, isDragReject } = + useDropzone({ + onDrop: (acceptedFiles: File[], fileRejections: FileRejection[]) => { + // A rejected file never reaches acceptedFiles, so without this the + // drop is a silent no-op and any previously picked file stays staged. + if (fileRejections.length > 0) { + setFileName(''); + setYamlText(''); + setError(t('modals.importAgent.invalidFileType')); + return; + } + if (acceptedFiles[0]) processFile(acceptedFiles[0]); + }, + multiple: false, + // Declared here (not via getInputProps) so drag-and-drop is filtered + // too; processFile's extension check stays as a backstop for odd MIME + // reports. + accept: { + 'application/x-yaml': ['.yaml', '.yml'], + 'text/yaml': ['.yaml', '.yml'], + }, + }); + const handleAnalyze = async () => { if (!yamlText) return; setLoading(true); @@ -187,16 +213,20 @@ export default function ImportAgentModal({ return; } const agentId = data.agent_id as string; + const editPath = + data.agent_type === 'workflow' + ? `/agents/workflow/edit/${agentId}` + : `/agents/edit/${agentId}`; if (data.warnings && data.warnings.length > 0) { // Keep the modal open so the user sees what was skipped. - setGoToAgentId(agentId); + setGoToEditPath(editPath); setWarnings(data.warnings as string[]); setImportedStatus((data.status as string) || null); setPlan(null); return; } handleClose(); - navigate(`/agents/edit/${agentId}`); + navigate(editPath); } catch { setError(t('modals.importAgent.importError')); } finally { @@ -217,9 +247,9 @@ export default function ImportAgentModal({