From 3c78540931b3c11f987c1f0205e025701f12bf51 Mon Sep 17 00:00:00 2001 From: Pavel Date: Tue, 18 Aug 2026 17:10:26 +0200 Subject: [PATCH 1/6] Workflow export --- application/api/user/agents/portability.py | 498 ++- application/api/user/agents/routes.py | 35 +- .../storage/db/repositories/workflows.py | 87 + frontend/src/agents/AgentCard.tsx | 11 +- frontend/src/locale/de.json | 3 + frontend/src/locale/en.json | 3 + frontend/src/locale/es.json | 3 + frontend/src/locale/jp.json | 3 + frontend/src/locale/ru.json | 3 + frontend/src/locale/zh-TW.json | 3 + frontend/src/locale/zh.json | 3 + frontend/src/modals/ImportAgentModal.tsx | 34 +- tests/api/test_agent_portability.py | 19 + tests/api/test_workflow_portability.py | 385 ++ tests/api/user/agents/test_workflow_agents.py | 387 ++ tests/api/user/test_agents_routes.py | 3645 ----------------- tests/api/user/test_workflows.py | 12 - 17 files changed, 1434 insertions(+), 3700 deletions(-) create mode 100644 tests/api/test_workflow_portability.py create mode 100644 tests/api/user/agents/test_workflow_agents.py delete mode 100644 tests/api/user/test_agents_routes.py delete mode 100644 tests/api/user/test_workflows.py diff --git a/application/api/user/agents/portability.py b/application/api/user/agents/portability.py index 34ae3ec7..e29ab510 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,6 +60,8 @@ 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): @@ -266,6 +279,137 @@ 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 _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 each reference 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. + tools: list = [] + tool_key_by_id: dict[str, str] = {} + for tid in tool_ids: + entries = _serialize_tools(conn, {"tools": [tid]}, user) + if not entries: + continue + tool_key_by_id[tid] = _tool_key(len(tools)) + tools.append(entries[0]) + + sources: list = [] + source_name_by_id: dict[str, str] = {} + for sid in source_ids: + entries = _serialize_sources(conn, {"source_id": sid}, user) + if not entries: + continue + source_name_by_id[sid] = entries[0]["name"] + sources.append(entries[0]) + + 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 + + # Pass 2: emit nodes/edges in the workflow API's wire shape, with agent + # node configs rewritten to the portable keys. + 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"])) + 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 +430,29 @@ 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: + workflow_block = None + 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 +463,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 +501,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 +548,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: @@ -528,7 +714,29 @@ 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" and 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, + } + + return { + "target": target, + "sources": sources, + "tools": tools, + "prompt": prompt, + "models": models, + "workflow": workflow_plan, + } # --------------------------------------------------------------------------- @@ -546,26 +754,36 @@ 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] = [] + id_by_name: dict[str, Optional[str]] = {} for src in spec.get("sources") or []: 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 +792,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]: @@ -635,16 +853,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 +885,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): @@ -749,13 +985,186 @@ 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") + cfg["sources"] = 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, + ) + 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 +1187,25 @@ 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" + workflow_result = _WORKFLOW_UNCHANGED + if is_workflow: + workflow_result = _apply_workflow( + conn, + user, + spec, + target, + tool_ids_by_key, + source_ids_by_name, + model_ids_by_name, + warnings, + ) agents_repo = AgentsRepository(conn) is_update = bool(target.get("action") == "update" and target.get("agent_id")) @@ -817,6 +1242,27 @@ def apply_import(conn, user: str, doc: dict, resolution: Optional[dict] = None) "config": _import_config(spec), "slug": slug, } + 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": [], + } + ) + if workflow_result is not _WORKFLOW_UNCHANGED: + if workflow_result is None and (target.get("status") or "") == "published": + warnings.append( + "File has no workflow; the published agent kept its existing one" + ) + else: + authoritative["workflow_id"] = workflow_result + # 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), @@ -836,6 +1282,7 @@ def apply_import(conn, user: str, doc: dict, resolution: Optional[dict] = None) "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 +1293,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,13 +1354,6 @@ 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) body = agent_to_yaml(export) @@ -961,6 +1402,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..83d2103a 100644 --- a/application/api/user/agents/routes.py +++ b/application/api/user/agents/routes.py @@ -38,8 +38,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 +1471,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( @@ -1667,13 +1663,32 @@ 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. + if template.get("workflow_id"): + cloned_workflow = WorkflowsRepository(conn).clone_to_user( + str(template["workflow_id"]), + user, + from_owner=template.get("user_id"), + ) + if cloned_workflow is not None: + create_kwargs["workflow_id"] = str(cloned_workflow["id"]) + else: + current_app.logger.warning( + "Adopt: template %s references missing workflow %s; " + "adopted agent created without one", + template.get("id"), template.get("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 diff --git a/application/storage/db/repositories/workflows.py b/application/storage/db/repositories/workflows.py index bc9e4cc9..42ba12f9 100644 --- a/application/storage/db/repositories/workflows.py +++ b/application/storage/db/repositories/workflows.py @@ -115,6 +115,93 @@ 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, + ) -> 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. + """ + 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"]) + + 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": n.get("config") or {}, + } + 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..1e5ae660 100644 --- a/frontend/src/agents/AgentCard.tsx +++ b/frontend/src/agents/AgentCard.tsx @@ -297,7 +297,13 @@ 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); + throw new Error(message || t('agents.exportAgentFailed')); + } const yamlText = await response.text(); const blob = new Blob([yamlText], { type: 'application/x-yaml' }); const url = URL.createObjectURL(blob); @@ -310,6 +316,9 @@ export default function AgentCard({ URL.revokeObjectURL(url); } catch (error) { console.error('Error:', error); + alert( + error instanceof Error ? error.message : t('agents.exportAgentFailed'), + ); } }; diff --git a/frontend/src/locale/de.json b/frontend/src/locale/de.json index 6f9701c7..ef1244ef 100644 --- a/frontend/src/locale/de.json +++ b/frontend/src/locale/de.json @@ -1030,6 +1030,8 @@ "importError": "Import failed", "willCreate": "Creates a new agent.", "willUpdate": "Updates an existing agent (matched by {{matchedBy}}).", + "workflowCreate": "Includes a workflow with {{nodes}} nodes — it will be created for you.", + "workflowUpdate": "Includes a workflow with {{nodes}} nodes — your existing workflow will be updated.", "sources": "Sources", "tools": "Tools", "models": "Models", @@ -1149,6 +1151,7 @@ "newAgent": "Neuer Agent", "importAgent": "Import", "exportAgent": "Export as YAML", + "exportAgentFailed": "Failed to export agent", "shareWithTeam": "Mit Team teilen", "backToAll": "Zurück zu allen Agenten", "pageHeader": { diff --git a/frontend/src/locale/en.json b/frontend/src/locale/en.json index 52adfe35..ff6e22f5 100644 --- a/frontend/src/locale/en.json +++ b/frontend/src/locale/en.json @@ -1035,6 +1035,8 @@ "importError": "Import failed", "willCreate": "Creates a new agent.", "willUpdate": "Updates an existing agent (matched by {{matchedBy}}).", + "workflowCreate": "Includes a workflow with {{nodes}} nodes — it will be created for you.", + "workflowUpdate": "Includes a workflow with {{nodes}} nodes — your existing workflow will be updated.", "sources": "Sources", "tools": "Tools", "models": "Models", @@ -1158,6 +1160,7 @@ "newAgent": "New Agent", "importAgent": "Import", "exportAgent": "Export as YAML", + "exportAgentFailed": "Failed to export agent", "shareWithTeam": "Share with team", "backToAll": "Back to all agents", "pageHeader": { diff --git a/frontend/src/locale/es.json b/frontend/src/locale/es.json index 62029b09..3065caa8 100644 --- a/frontend/src/locale/es.json +++ b/frontend/src/locale/es.json @@ -1030,6 +1030,8 @@ "importError": "Import failed", "willCreate": "Creates a new agent.", "willUpdate": "Updates an existing agent (matched by {{matchedBy}}).", + "workflowCreate": "Includes a workflow with {{nodes}} nodes — it will be created for you.", + "workflowUpdate": "Includes a workflow with {{nodes}} nodes — your existing workflow will be updated.", "sources": "Sources", "tools": "Tools", "models": "Models", @@ -1149,6 +1151,7 @@ "newAgent": "Nuevo Agente", "importAgent": "Import", "exportAgent": "Export as YAML", + "exportAgentFailed": "Failed to export agent", "shareWithTeam": "Compartir con el equipo", "backToAll": "Volver a todos los agentes", "pageHeader": { diff --git a/frontend/src/locale/jp.json b/frontend/src/locale/jp.json index a1d7d0a7..8bfd42ee 100644 --- a/frontend/src/locale/jp.json +++ b/frontend/src/locale/jp.json @@ -1030,6 +1030,8 @@ "importError": "Import failed", "willCreate": "Creates a new agent.", "willUpdate": "Updates an existing agent (matched by {{matchedBy}}).", + "workflowCreate": "Includes a workflow with {{nodes}} nodes — it will be created for you.", + "workflowUpdate": "Includes a workflow with {{nodes}} nodes — your existing workflow will be updated.", "sources": "Sources", "tools": "Tools", "models": "Models", @@ -1149,6 +1151,7 @@ "newAgent": "新しいエージェント", "importAgent": "Import", "exportAgent": "Export as YAML", + "exportAgentFailed": "Failed to export agent", "shareWithTeam": "チームと共有", "backToAll": "すべてのエージェントに戻る", "pageHeader": { diff --git a/frontend/src/locale/ru.json b/frontend/src/locale/ru.json index cb364aa1..a025eb32 100644 --- a/frontend/src/locale/ru.json +++ b/frontend/src/locale/ru.json @@ -1030,6 +1030,8 @@ "importError": "Import failed", "willCreate": "Creates a new agent.", "willUpdate": "Updates an existing agent (matched by {{matchedBy}}).", + "workflowCreate": "Includes a workflow with {{nodes}} nodes — it will be created for you.", + "workflowUpdate": "Includes a workflow with {{nodes}} nodes — your existing workflow will be updated.", "sources": "Sources", "tools": "Tools", "models": "Models", @@ -1149,6 +1151,7 @@ "newAgent": "Новый Агент", "importAgent": "Import", "exportAgent": "Export as YAML", + "exportAgentFailed": "Failed to export agent", "shareWithTeam": "Поделиться с командой", "backToAll": "Вернуться ко всем агентам", "pageHeader": { diff --git a/frontend/src/locale/zh-TW.json b/frontend/src/locale/zh-TW.json index f2739086..dd8bc6ee 100644 --- a/frontend/src/locale/zh-TW.json +++ b/frontend/src/locale/zh-TW.json @@ -1030,6 +1030,8 @@ "importError": "Import failed", "willCreate": "Creates a new agent.", "willUpdate": "Updates an existing agent (matched by {{matchedBy}}).", + "workflowCreate": "Includes a workflow with {{nodes}} nodes — it will be created for you.", + "workflowUpdate": "Includes a workflow with {{nodes}} nodes — your existing workflow will be updated.", "sources": "Sources", "tools": "Tools", "models": "Models", @@ -1149,6 +1151,7 @@ "newAgent": "新建代理", "importAgent": "Import", "exportAgent": "Export as YAML", + "exportAgentFailed": "Failed to export agent", "shareWithTeam": "與團隊分享", "backToAll": "返回所有代理", "pageHeader": { diff --git a/frontend/src/locale/zh.json b/frontend/src/locale/zh.json index bb0d7680..b7ca42a4 100644 --- a/frontend/src/locale/zh.json +++ b/frontend/src/locale/zh.json @@ -1030,6 +1030,8 @@ "importError": "Import failed", "willCreate": "Creates a new agent.", "willUpdate": "Updates an existing agent (matched by {{matchedBy}}).", + "workflowCreate": "Includes a workflow with {{nodes}} nodes — it will be created for you.", + "workflowUpdate": "Includes a workflow with {{nodes}} nodes — your existing workflow will be updated.", "sources": "Sources", "tools": "Tools", "models": "Models", @@ -1149,6 +1151,7 @@ "newAgent": "新建代理", "importAgent": "Import", "exportAgent": "Export as YAML", + "exportAgentFailed": "Failed to export agent", "shareWithTeam": "与团队共享", "backToAll": "返回所有代理", "pageHeader": { diff --git a/frontend/src/modals/ImportAgentModal.tsx b/frontend/src/modals/ImportAgentModal.tsx index 579c7f90..7a46a35a 100644 --- a/frontend/src/modals/ImportAgentModal.tsx +++ b/frontend/src/modals/ImportAgentModal.tsx @@ -54,6 +54,11 @@ type ImportPlan = { tools: PlanTool[]; prompt: { status: string; name?: string }; models: PlanModel[]; + workflow?: { + nodes: number; + edges: number; + action: 'create' | 'update'; + } | null; }; interface ImportAgentModalProps { @@ -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 = () => { @@ -187,16 +192,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 +226,9 @@ export default function ImportAgentModal({ + } + > +

{exportError}

+ { - if (acceptedFiles[0]) processFile(acceptedFiles[0]); - }, - multiple: false, - // Declared here (not via getInputProps) so drag-and-drop is filtered - // too, with drag-over rejection feedback; processFile's extension check - // stays as a backstop for odd MIME reports. - accept: { - 'application/x-yaml': ['.yaml', '.yml'], - 'text/yaml': ['.yaml', '.yml'], - }, - }); + 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; @@ -320,7 +329,11 @@ export default function ImportAgentModal({
@@ -346,18 +359,28 @@ export default function ImportAgentModal({ : t('modals.importAgent.willCreate')}
- {plan.workflow && ( -

- - {plan.workflow.action === 'update' - ? t('modals.importAgent.workflowUpdate', { - nodes: plan.workflow.nodes, - }) - : t('modals.importAgent.workflowCreate', { - nodes: plan.workflow.nodes, - })} -

- )} + {plan.workflow && + (plan.workflow.action === 'delete' ? ( + // Irreversible: the graph, its run history and its artifacts + // all go. Warn rather than confirming with a green check. +

+ + {t('modals.importAgent.workflowDelete', { + nodes: plan.workflow.nodes, + })} +

+ ) : ( +

+ + {plan.workflow.action === 'update' + ? t('modals.importAgent.workflowUpdate', { + nodes: plan.workflow.nodes, + }) + : t('modals.importAgent.workflowCreate', { + nodes: plan.workflow.nodes, + })} +

+ ))} {plan.sources.length > 0 && (
diff --git a/frontend/src/modals/ImportSpecModal.tsx b/frontend/src/modals/ImportSpecModal.tsx index 2f636087..a5bee57b 100644 --- a/frontend/src/modals/ImportSpecModal.tsx +++ b/frontend/src/modals/ImportSpecModal.tsx @@ -1,5 +1,5 @@ import { useState } from 'react'; -import { useDropzone } from 'react-dropzone'; +import { type FileRejection, useDropzone } from 'react-dropzone'; import { useTranslation } from 'react-i18next'; import { useSelector } from 'react-redux'; @@ -73,19 +73,28 @@ export default function ImportSpecModal({ setParsedResult(null); }; - const { getRootProps, getInputProps, isDragActive } = useDropzone({ - onDrop: (acceptedFiles: File[]) => { - 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. - accept: { - 'application/json': ['.json'], - 'application/x-yaml': ['.yaml', '.yml'], - 'text/yaml': ['.yaml', '.yml'], - }, - }); + 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) { + setFile(null); + setParsedResult(null); + setError(t('modals.importSpec.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. + accept: { + 'application/json': ['.json'], + 'application/x-yaml': ['.yaml', '.yml'], + 'text/yaml': ['.yaml', '.yml'], + }, + }); const handleParse = async () => { if (!file) return; @@ -209,7 +218,11 @@ export default function ImportSpecModal({
diff --git a/tests/api/test_workflow_portability.py b/tests/api/test_workflow_portability.py index 138d8858..68875846 100644 --- a/tests/api/test_workflow_portability.py +++ b/tests/api/test_workflow_portability.py @@ -408,3 +408,128 @@ def test_null_workflow_keeps_row_still_referenced_by_another_agent(pg_conn): assert updated["workflow_id"] is None # Still referenced — must not be deleted. assert WorkflowsRepository(pg_conn).get(str(wf["id"]), user) is not None + + +def test_publish_guard_rejects_workflow_agent_with_no_graph(pg_conn): + """A file that makes a published agent a graph-less workflow agent is refused. + + The create and update routes both reject that state; import must not + install it through the back door. + """ + user = "u_wf_publish_guard" + agent = AgentsRepository(pg_conn).create( + user, "Probe Bot", "published", agent_type="classic", slug="probe-bot" + ) + doc = { + "apiVersion": "docsgpt.arc53.com/v1", + "kind": "Agent", + "metadata": {"id": str(agent["id"]), "slug": "probe-bot"}, + "spec": {"name": "Probe Bot", "agent_type": "workflow"}, + } + + # Key omitted entirely. + with pytest.raises(AgentImportError, match="published workflow agent needs a workflow"): + apply_import(pg_conn, user, doc) + + # Explicit null, with no prior workflow to fall back on. + doc["spec"]["workflow"] = None + with pytest.raises(AgentImportError, match="published workflow agent needs a workflow"): + apply_import(pg_conn, user, doc) + + unchanged = AgentsRepository(pg_conn).get(str(agent["id"]), user) + assert unchanged["agent_type"] == "classic" + assert unchanged["status"] == "published" + + +def test_plan_distinguishes_absent_workflow_from_explicit_null(pg_conn): + """Omitting the key is a no-op; an explicit null destroys the graph. + + The two used to produce an identical plan, so the modal showed nothing + either way while the outcomes differed completely. + """ + user = "u_wf_plan_null" + agent, wf = _seed_workflow_agent(pg_conn, user) + AgentsRepository(pg_conn).update(str(agent["id"]), user, {"status": "draft"}) + doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user))) + + absent = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user))) + del absent["spec"]["workflow"] + assert plan_import(pg_conn, user, absent)["workflow"] is None + + doc["spec"]["workflow"] = None + assert plan_import(pg_conn, user, doc)["workflow"] == { + "action": "delete", + "nodes": 3, + "edges": 2, + } + + +def test_plan_reports_no_removal_when_workflow_survives(pg_conn): + """No delete block when the row lives on — published, or still referenced.""" + user = "u_wf_plan_keep" + agent, wf = _seed_workflow_agent(pg_conn, user) + doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user))) + doc["spec"]["workflow"] = None + + # Published: _apply_workflow keeps the existing graph. + assert plan_import(pg_conn, user, doc)["workflow"] is None + + # Draft, but a sibling agent still references the row. + AgentsRepository(pg_conn).update(str(agent["id"]), user, {"status": "draft"}) + AgentsRepository(pg_conn).create( + user, "Sibling", "draft", agent_type="workflow", workflow_id=str(wf["id"]) + ) + assert plan_import(pg_conn, user, doc)["workflow"] is None + + +def test_null_workflow_deletion_is_warned_about(pg_conn): + """The destructive reap must not be silent.""" + user = "u_wf_null_warn" + agent, wf = _seed_workflow_agent(pg_conn, user) + doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user))) + doc["spec"]["workflow"] = None + AgentsRepository(pg_conn).update(str(agent["id"]), user, {"status": "draft"}) + + result = apply_import(pg_conn, user, doc) + assert WorkflowsRepository(pg_conn).get(str(wf["id"]), user) is None + assert any("was deleted" in w and "run history" in w for w in result["warnings"]) + + +def test_duplicate_source_names_warn_and_dedupe(pg_conn): + """Two sources sharing a name collapse onto the oldest — say so.""" + user = "u_wf_dup_src" + repo = SourcesRepository(pg_conn) + older = repo.create("KB", user_id=user, type="file") + newer = repo.create("KB", user_id=user, type="file") + agent, _wf = _seed_workflow_agent(pg_conn, user, source_id=str(older["id"])) + + # Point the node at both same-named sources. + version = 1 + nodes_repo = WorkflowNodesRepository(pg_conn) + node = [n for n in nodes_repo.find_by_version(str(agent["workflow_id"]), version) + if n["node_type"] == "agent"][0] + config = dict(node["config"]) + config["sources"] = [str(older["id"]), str(newer["id"])] + nodes_repo.bulk_create( + str(agent["workflow_id"]), + version, + [{ + "node_id": node["node_id"], + "node_type": node["node_type"], + "title": node.get("title"), + "position": node.get("position"), + "config": config, + }], + ) + + doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user))) + assert [s["name"] for s in doc["spec"]["sources"]] == ["KB", "KB"] + + result = apply_import(pg_conn, user, doc) + assert any("share the name 'KB'" in w for w in result["warnings"]) + + updated = AgentsRepository(pg_conn).get(str(agent["id"]), user) + graph_nodes, _ = _graph(pg_conn, WorkflowsRepository(pg_conn).get(str(updated["workflow_id"]), user)) + agent_node = [n for n in graph_nodes if n["node_type"] == "agent"][0] + # Collapsed onto the oldest match, and not stored twice. + assert agent_node["config"]["sources"] == [str(older["id"])]