Workflow export

This commit is contained in:
Pavel committed 2026-08-18 17:10:26 +02:00
1 parent a51131d3fc
commit 3c78540931
17 files changed
+1434 -3700

No files matched your search

+472 -26
View File
@@ -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/<id>``); 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(
+25 -10
View File
@@ -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
@@ -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
+10 -1
View File
@@ -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'),
);
}
};
+3
View File
@@ -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": {
+3
View File
@@ -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": {
+3
View File
@@ -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": {
+3
View File
@@ -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": {
+3
View File
@@ -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": {
+3
View File
@@ -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": {
+3
View File
@@ -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": {
+28 -6
View File
@@ -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<Record<string, string>>({});
const [warnings, setWarnings] = useState<string[] | null>(null);
const [importedStatus, setImportedStatus] = useState<string | null>(null);
const [goToAgentId, setGoToAgentId] = useState<string | null>(null);
const [goToEditPath, setGoToEditPath] = useState<string | null>(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({
<Button
type="button"
onClick={() => {
const id = goToAgentId;
const path = goToEditPath;
handleClose();
if (id) navigate(`/agents/edit/${id}`);
if (path) navigate(path);
}}
className="rounded-3xl px-5"
>
@@ -328,6 +337,19 @@ export default function ImportAgentModal({
: t('modals.importAgent.willCreate')}
</div>
{plan.workflow && (
<p className="flex items-center gap-2 text-sm">
<CheckCircle2 className="h-4 w-4 text-green-600 dark:text-green-400" />
{plan.workflow.action === 'update'
? t('modals.importAgent.workflowUpdate', {
nodes: plan.workflow.nodes,
})
: t('modals.importAgent.workflowCreate', {
nodes: plan.workflow.nodes,
})}
</p>
)}
{plan.sources.length > 0 && (
<Section title={t('modals.importAgent.sources')}>
{plan.sources.map((s) =>
+19
View File
@@ -286,10 +286,29 @@ def test_parse_rejects_bad_documents():
parse_agent_yaml("kind: Agent\napiVersion: other/v1\nspec:\n name: x\n")
with pytest.raises(AgentImportError):
parse_agent_yaml("kind: Agent\napiVersion: docsgpt.arc53.com/v1\nspec: {}\n")
# A workflow agent without a graph parses (drafts may have none)…
doc = parse_agent_yaml(
"kind: Agent\napiVersion: docsgpt.arc53.com/v1\n"
"spec:\n name: x\n agent_type: workflow\n"
)
assert doc["spec"]["agent_type"] == "workflow"
# …but a present workflow block must be well-formed.
with pytest.raises(AgentImportError):
parse_agent_yaml(
"kind: Agent\napiVersion: docsgpt.arc53.com/v1\n"
"spec:\n name: x\n agent_type: workflow\n workflow: nonsense\n"
)
with pytest.raises(AgentImportError):
parse_agent_yaml(
"kind: Agent\napiVersion: docsgpt.arc53.com/v1\n"
"spec:\n name: x\n agent_type: workflow\n"
" workflow:\n nodes: {}\n edges: []\n"
)
with pytest.raises(AgentImportError):
parse_agent_yaml(
"kind: Agent\napiVersion: docsgpt.arc53.com/v1\n"
"spec:\n name: x\n agent_type: workflow\n"
" workflow:\n nodes:\n - id: n1\n edges: []\n"
)
+385
View File
@@ -0,0 +1,385 @@
"""Workflow-agent YAML export/import (application.api.user.agents.portability).
Real SQL against the ephemeral ``pg_conn`` fixture, like
``test_agent_portability.py``. Covers the graph block: reference rewriting
(tools → ``tool-N`` keys, sources → names, custom models → display names),
secret redaction inside node configs, graph validation at apply, and
idempotent re-import as a new graph version of the same workflow.
"""
from __future__ import annotations
import pytest
from application.agents.default_tools import default_tool_id
from application.api.user.agents.portability import (
AgentImportError,
agent_to_yaml,
apply_import,
parse_agent_yaml,
plan_import,
serialize_agent,
)
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.sources import SourcesRepository
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
pytestmark = pytest.mark.integration
def _graph(conn, workflow_row):
version = int(workflow_row.get("current_graph_version") or 1)
nodes = WorkflowNodesRepository(conn).find_by_version(str(workflow_row["id"]), version)
edges = WorkflowEdgesRepository(conn).find_by_version(str(workflow_row["id"]), version)
return nodes, edges
def _seed_workflow_agent(conn, user, *, tool_id=None, source_id=None, model_id=None):
"""Workflow agent with a start → agent → end graph referencing real resources."""
agent_cfg = {
"agent_type": "classic",
"system_prompt": "Answer briefly.",
"tools": [t for t in [tool_id, default_tool_id("read_document")] if t],
"sources": [s for s in [source_id] if s],
"model_id": model_id,
}
wf = WorkflowsRepository(conn).create(user, "Pipeline", description="wd")
wf_id = str(wf["id"])
nodes = WorkflowNodesRepository(conn).bulk_create(
wf_id,
1,
[
{"node_id": "start-1", "node_type": "start", "title": "Start"},
{"node_id": "agent-1", "node_type": "agent", "title": "Answer", "config": agent_cfg},
{"node_id": "end-1", "node_type": "end", "title": "End"},
],
)
by_str = {n["node_id"]: n["id"] for n in nodes}
WorkflowEdgesRepository(conn).bulk_create(
wf_id,
1,
[
{"edge_id": "e1", "from_node_id": by_str["start-1"], "to_node_id": by_str["agent-1"]},
{"edge_id": "e2", "from_node_id": by_str["agent-1"], "to_node_id": by_str["end-1"]},
],
)
agent = AgentsRepository(conn).create(
user,
"Flow Bot",
"published",
description="wf agent",
agent_type="workflow",
workflow_id=wf_id,
)
return agent, wf
def _seed_tool(conn, user):
return UserToolsRepository(conn).create(
user,
"brave",
config={"encrypted_credentials": "SECRETBLOB", "region": "US"},
custom_name="My Brave",
display_name="Brave Search",
description="search",
config_requirements={"token": {"secret": True, "required": True}, "region": {}},
actions=[],
)
def test_export_rewrites_node_refs_and_redacts_secrets(pg_conn):
user = "u_wf_export"
tool = _seed_tool(pg_conn, user)
src = SourcesRepository(pg_conn).create("KB", user_id=user, type="file")
model = UserCustomModelsRepository(pg_conn).create(
user, "llama-3", "My Llama", "https://models.example.com/v1", "sk-plain"
)
agent, wf = _seed_workflow_agent(
pg_conn,
user,
tool_id=str(tool["id"]),
source_id=str(src["id"]),
model_id=str(model["id"]),
)
export = serialize_agent(pg_conn, agent, user)
text = agent_to_yaml(export)
spec = export["spec"]
assert "SECRETBLOB" not in text and "sk-plain" not in text
# Raw ids never appear inside the graph block (the sections may carry
# ``ref`` hints, same as classic export).
workflow_text = agent_to_yaml(spec["workflow"])
assert str(tool["id"]) not in workflow_text
assert str(src["id"]) not in workflow_text
assert str(model["id"]) not in workflow_text
assert spec["workflow"]["name"] == "Pipeline"
assert [n["id"] for n in spec["workflow"]["nodes"]] == ["agent-1", "end-1", "start-1"]
agent_node = next(n for n in spec["workflow"]["nodes"] if n["id"] == "agent-1")
assert agent_node["config"]["tools"] == ["tool-0", "tool-1"]
assert agent_node["config"]["sources"] == ["KB"]
assert agent_node["config"]["model_id"] == "My Llama"
# The collected sections use the classic serializers' portable identities.
assert spec["tools"][0]["type"] == "brave"
assert spec["tools"][0]["requires_secrets"] == ["token"]
assert spec["tools"][1] == {"type": "read_document", "builtin": True}
assert spec["sources"][0]["name"] == "KB"
assert spec["model"]["available"][0]["display_name"] == "My Llama"
assert spec["prompt"] == "default"
edges = spec["workflow"]["edges"]
assert {(e["source"], e["target"]) for e in edges} == {
("start-1", "agent-1"),
("agent-1", "end-1"),
}
def test_export_without_workflow_emits_null_block(pg_conn):
user = "u_wf_export2"
agent = AgentsRepository(pg_conn).create(
user, "Draft Flow", "draft", agent_type="workflow"
)
export = serialize_agent(pg_conn, agent, user)
assert export["spec"]["workflow"] is None
assert export["spec"]["tools"] == []
assert export["spec"]["sources"] == []
def test_round_trip_same_user_reuses_everything(pg_conn):
user = "u_wf_round"
tool = _seed_tool(pg_conn, user)
src = SourcesRepository(pg_conn).create("KB", user_id=user, type="file")
model = UserCustomModelsRepository(pg_conn).create(
user, "llama-3", "My Llama", "https://models.example.com/v1", "sk-plain"
)
agent, wf = _seed_workflow_agent(
pg_conn,
user,
tool_id=str(tool["id"]),
source_id=str(src["id"]),
model_id=str(model["id"]),
)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
result = apply_import(pg_conn, user, doc)
assert result["action"] == "updated"
assert result["agent_id"] == str(agent["id"])
updated = AgentsRepository(pg_conn).get(str(agent["id"]), user)
# Same workflow row, bumped to a new graph version.
assert str(updated["workflow_id"]) == str(wf["id"])
wf_row = WorkflowsRepository(pg_conn).get(str(wf["id"]), user)
assert int(wf_row["current_graph_version"]) == 2
nodes, edges = _graph(pg_conn, wf_row)
assert len(nodes) == 3 and len(edges) == 2
agent_node = next(n for n in nodes if n["node_id"] == "agent-1")
# References resolved back to the same owned resources — no duplicates.
assert set(agent_node["config"]["tools"]) == {
str(tool["id"]), default_tool_id("read_document"),
}
assert agent_node["config"]["sources"] == [str(src["id"])]
assert agent_node["config"]["model_id"] == str(model["id"])
assert len(UserToolsRepository(pg_conn).list_for_user(user)) == 1
# Old graph versions are pruned.
assert WorkflowNodesRepository(pg_conn).find_by_version(str(wf["id"]), 1) == []
def test_import_into_fresh_user_creates_workflow(pg_conn):
owner = "u_wf_owner"
tool = _seed_tool(pg_conn, owner)
src = SourcesRepository(pg_conn).create("KB", user_id=owner, type="file")
agent, _ = _seed_workflow_agent(
pg_conn, owner, tool_id=str(tool["id"]), source_id=str(src["id"])
)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, owner)))
importer = "u_wf_importer"
result = apply_import(
pg_conn,
importer,
doc,
{"tools": {"tool-0": {"decision": "create", "secrets": {"token": "tok"}}}},
)
assert result["action"] == "created"
created = AgentsRepository(pg_conn).get(result["agent_id"], importer)
assert created["status"] == "draft"
assert created["workflow_id"] is not None
wf_row = WorkflowsRepository(pg_conn).get(str(created["workflow_id"]), importer)
assert wf_row is not None and wf_row["name"] == "Pipeline"
nodes, edges = _graph(pg_conn, wf_row)
assert len(nodes) == 3 and len(edges) == 2
agent_node = next(n for n in nodes if n["node_id"] == "agent-1")
# The tool was created for the importer; the missing source was dropped.
my_tool = UserToolsRepository(pg_conn).list_for_user(importer)[0]
assert set(agent_node["config"]["tools"]) == {
str(my_tool["id"]), default_tool_id("read_document"),
}
assert agent_node["config"]["sources"] == []
assert any("Source 'KB' not found" in w for w in result["warnings"])
assert any("not resolved" in w for w in result["warnings"])
def test_plan_reports_workflow_block(pg_conn):
user = "u_wf_plan"
agent, wf = _seed_workflow_agent(pg_conn, user)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
plan = plan_import(pg_conn, user, doc)
assert plan["workflow"] == {"nodes": 3, "edges": 2, "action": "update"}
other_plan = plan_import(pg_conn, "someone_else", doc)
assert other_plan["workflow"]["action"] == "create"
def test_import_rejects_invalid_graph(pg_conn):
user = "u_wf_invalid"
doc = {
"apiVersion": "docsgpt.arc53.com/v1",
"kind": "Agent",
"metadata": {"slug": "broken-flow"},
"spec": {
"name": "Broken Flow",
"agent_type": "workflow",
"workflow": {
"name": "Broken",
# No start/end nodes — the API's validation gate must reject it.
"nodes": [{"id": "a", "type": "agent", "config": {}}],
"edges": [],
},
},
}
with pytest.raises(AgentImportError, match="Workflow validation failed"):
apply_import(pg_conn, user, doc)
# Nothing half-imported.
assert AgentsRepository(pg_conn).list_for_user(user) == []
assert WorkflowsRepository(pg_conn).list_for_user(user) == []
def test_import_drops_foreign_raw_ids_from_node_config(pg_conn):
"""A hand-edited file carrying someone else's raw ids must not link them."""
victim = "u_wf_victim"
victim_tool = _seed_tool(pg_conn, victim)
victim_src = SourcesRepository(pg_conn).create("Secret KB", user_id=victim, type="file")
doc = {
"apiVersion": "docsgpt.arc53.com/v1",
"kind": "Agent",
"metadata": {"slug": "sneaky-flow"},
"spec": {
"name": "Sneaky Flow",
"agent_type": "workflow",
"workflow": {
"name": "Sneaky",
"nodes": [
{"id": "s", "type": "start"},
{
"id": "a",
"type": "agent",
"config": {
"tools": [str(victim_tool["id"])],
"sources": [str(victim_src["id"])],
},
},
{"id": "e", "type": "end"},
],
"edges": [
{"id": "e1", "source": "s", "target": "a"},
{"id": "e2", "source": "a", "target": "e"},
],
},
},
}
result = apply_import(pg_conn, "u_wf_attacker", doc)
created = AgentsRepository(pg_conn).get(result["agent_id"], "u_wf_attacker")
wf_row = WorkflowsRepository(pg_conn).get(str(created["workflow_id"]), "u_wf_attacker")
nodes, _ = _graph(pg_conn, wf_row)
agent_node = next(n for n in nodes if n["node_id"] == "a")
assert agent_node["config"]["tools"] == []
assert agent_node["config"]["sources"] == []
assert len(result["warnings"]) == 2
def test_import_route_rejects_invalid_graph_as_client_error(pg_conn):
"""Apply-time graph rejection is a 400 carrying the validation detail.
Regression: an ``AgentImportError`` raised while applying (the file parses,
but the graph fails the workflow API's validation gate) used to fall into
the route's generic handler and surface as a 500 with no explanation.
"""
from contextlib import contextmanager
from unittest.mock import patch
from flask import Flask, request
from application.api.user.agents.portability import ImportAgent
@contextmanager
def _conn():
yield pg_conn
yaml_text = (
"kind: Agent\napiVersion: docsgpt.arc53.com/v1\n"
"spec:\n"
" name: Broken Flow\n"
" agent_type: workflow\n"
" workflow:\n"
" name: Broken\n"
" nodes:\n"
" - id: a\n"
" type: agent\n"
" edges: []\n"
)
app = Flask(__name__)
with patch(
"application.api.user.agents.portability.db_session", _conn
), app.test_request_context(
"/api/import_agent", method="POST", json={"yaml": yaml_text}
):
request.decoded_token = {"sub": "u_wf_route"}
response = ImportAgent().post()
assert response.status_code == 400
body = response.get_json()
assert body["success"] is False
assert "Workflow validation failed" in body["message"]
assert AgentsRepository(pg_conn).list_for_user("u_wf_route") == []
def test_import_without_workflow_keeps_published_agents_graph(pg_conn):
user = "u_wf_keep"
agent, wf = _seed_workflow_agent(pg_conn, user)
doc = parse_agent_yaml(agent_to_yaml(serialize_agent(pg_conn, agent, user)))
del doc["spec"]["workflow"]
result = apply_import(pg_conn, user, doc)
assert result["action"] == "updated"
updated = AgentsRepository(pg_conn).get(str(agent["id"]), user)
assert str(updated["workflow_id"]) == str(wf["id"])
def test_import_with_null_workflow_keeps_published_but_clears_draft(pg_conn):
user = "u_wf_null"
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 agent: explicit null must not break it — kept with a warning.
result = apply_import(pg_conn, user, doc)
updated = AgentsRepository(pg_conn).get(str(agent["id"]), user)
assert str(updated["workflow_id"]) == str(wf["id"])
assert any("kept its existing" in w for w in result["warnings"])
# Draft agent: explicit null clears the link.
AgentsRepository(pg_conn).update(str(agent["id"]), user, {"status": "draft"})
result = apply_import(pg_conn, user, doc)
updated = AgentsRepository(pg_conn).get(str(agent["id"]), user)
assert updated["workflow_id"] is None
@@ -0,0 +1,387 @@
"""Workflow-agent behavior of the agents routes, against real Postgres.
Rebuilds the workflow-related coverage of the retired Mongo-era
``tests/api/user/test_agents_routes.py`` on the ``pg_conn`` fixture:
adopting a workflow template (the graph must be deep-copied, never
shared by id), deleting a workflow agent (only the owner's graph may
go), and creating/updating agents with a workflow reference.
"""
from contextlib import contextmanager
from unittest.mock import patch
import pytest
from flask import Flask
@pytest.fixture
def app():
return Flask(__name__)
@contextmanager
def _patch_db(conn):
@contextmanager
def _yield():
yield conn
with patch(
"application.api.user.agents.routes.db_session", _yield
), patch(
"application.api.user.agents.routes.db_readonly", _yield
):
yield
def _seed_workflow(pg_conn, user, *, name="WF"):
"""Create a start → agent → end workflow; returns (workflow, nodes, edges)."""
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
wf = WorkflowsRepository(pg_conn).create(user, name, description="d")
wf_id = str(wf["id"])
nodes = WorkflowNodesRepository(pg_conn).bulk_create(
wf_id,
1,
[
{"node_id": "start-1", "node_type": "start", "title": "Start"},
{
"node_id": "agent-1",
"node_type": "agent",
"title": "Agent",
"config": {"system_prompt": "hi", "tools": [], "sources": []},
},
{"node_id": "end-1", "node_type": "end", "title": "End"},
],
)
by_str = {n["node_id"]: n["id"] for n in nodes}
edges = WorkflowEdgesRepository(pg_conn).bulk_create(
wf_id,
1,
[
{
"edge_id": "e1",
"from_node_id": by_str["start-1"],
"to_node_id": by_str["agent-1"],
},
{
"edge_id": "e2",
"from_node_id": by_str["agent-1"],
"to_node_id": by_str["end-1"],
},
],
)
return wf, nodes, edges
def _graph(pg_conn, workflow_id, version=1):
from application.storage.db.repositories.workflow_edges import (
WorkflowEdgesRepository,
)
from application.storage.db.repositories.workflow_nodes import (
WorkflowNodesRepository,
)
nodes = WorkflowNodesRepository(pg_conn).find_by_version(str(workflow_id), version)
edges = WorkflowEdgesRepository(pg_conn).find_by_version(str(workflow_id), version)
return nodes, edges
class TestCloneToUser:
def test_clones_graph_for_new_owner(self, pg_conn):
from application.storage.db.repositories.workflows import WorkflowsRepository
wf, _, _ = _seed_workflow(pg_conn, "owner-a", name="Source WF")
clone = WorkflowsRepository(pg_conn).clone_to_user(
str(wf["id"]), "owner-b", from_owner="owner-a"
)
assert clone is not None
assert str(clone["id"]) != str(wf["id"])
assert clone["user_id"] == "owner-b"
assert clone["name"] == "Source WF"
nodes, edges = _graph(pg_conn, clone["id"])
assert {n["node_id"] for n in nodes} == {"start-1", "agent-1", "end-1"}
agent_node = next(n for n in nodes if n["node_id"] == "agent-1")
assert agent_node["config"] == {
"system_prompt": "hi", "tools": [], "sources": [],
}
# Edges are re-keyed onto the clone's node rows, not the source's.
clone_node_uuids = {n["id"] for n in nodes}
assert len(edges) == 2
for e in edges:
assert e["from_node_id"] in clone_node_uuids
assert e["to_node_id"] in clone_node_uuids
# The source graph is untouched.
src_nodes, src_edges = _graph(pg_conn, wf["id"])
assert len(src_nodes) == 3 and len(src_edges) == 2
def test_wrong_owner_returns_none(self, pg_conn):
from application.storage.db.repositories.workflows import WorkflowsRepository
wf, _, _ = _seed_workflow(pg_conn, "owner-a")
assert (
WorkflowsRepository(pg_conn).clone_to_user(
str(wf["id"]), "thief", from_owner="not-the-owner"
)
is None
)
def test_missing_workflow_returns_none(self, pg_conn):
from application.storage.db.repositories.workflows import WorkflowsRepository
assert (
WorkflowsRepository(pg_conn).clone_to_user(
"00000000-0000-0000-0000-000000000000", "u"
)
is None
)
class TestAdoptWorkflowAgent:
def test_adopt_clones_the_graph(self, app, pg_conn):
from application.api.user.agents.routes import AdoptAgent
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.workflows import WorkflowsRepository
wf, _, _ = _seed_workflow(pg_conn, "__system__", name="Template WF")
repo = AgentsRepository(pg_conn)
template = repo.create(
"__system__",
"WF Template",
"template",
agent_type="workflow",
workflow_id=str(wf["id"]),
)
with _patch_db(pg_conn), app.test_request_context(
f"/api/adopt_agent?id={template['id']}", method="POST"
):
from flask import request
request.decoded_token = {"sub": "u-wf-adopter"}
response = AdoptAgent().post()
assert response.status_code == 200
adopted = next(
a for a in repo.list_for_user("u-wf-adopter") if a["name"] == "WF Template"
)
assert adopted["workflow_id"] is not None
assert str(adopted["workflow_id"]) != str(wf["id"])
cloned = WorkflowsRepository(pg_conn).get(
str(adopted["workflow_id"]), "u-wf-adopter"
)
assert cloned is not None
nodes, edges = _graph(pg_conn, cloned["id"])
assert {n["node_id"] for n in nodes} == {"start-1", "agent-1", "end-1"}
assert len(edges) == 2
def test_adopt_survives_unresolvable_workflow(self, app, pg_conn):
"""A template pointing at a graph its owner doesn't own adopts without one."""
from application.api.user.agents.routes import AdoptAgent
from application.storage.db.repositories.agents import AgentsRepository
wf, _, _ = _seed_workflow(pg_conn, "someone-else")
repo = AgentsRepository(pg_conn)
template = repo.create(
"__system__",
"Broken WF Template",
"template",
agent_type="workflow",
workflow_id=str(wf["id"]),
)
with _patch_db(pg_conn), app.test_request_context(
f"/api/adopt_agent?id={template['id']}", method="POST"
):
from flask import request
request.decoded_token = {"sub": "u-wf-adopter2"}
response = AdoptAgent().post()
assert response.status_code == 200
adopted = next(
a
for a in repo.list_for_user("u-wf-adopter2")
if a["name"] == "Broken WF Template"
)
assert adopted["workflow_id"] is None
# The unowned graph was not cloned or mutated.
nodes, edges = _graph(pg_conn, wf["id"])
assert len(nodes) == 3 and len(edges) == 2
class TestDeleteWorkflowAgent:
def test_delete_removes_owned_workflow(self, app, pg_conn):
from application.api.user.agents.routes import DeleteAgent
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.workflows import WorkflowsRepository
user = "u-wf-del"
wf, _, _ = _seed_workflow(pg_conn, user)
repo = AgentsRepository(pg_conn)
agent = repo.create(
user, "WF Agent", "draft", agent_type="workflow", workflow_id=str(wf["id"])
)
with _patch_db(pg_conn), app.test_request_context(
f"/api/delete_agent?id={agent['id']}", method="DELETE"
):
from flask import request
request.decoded_token = {"sub": user}
response = DeleteAgent().delete()
assert response.status_code == 200
assert repo.get_any(str(agent["id"]), user) is None
assert WorkflowsRepository(pg_conn).get_by_id(str(wf["id"])) is None
nodes, edges = _graph(pg_conn, wf["id"])
assert nodes == [] and edges == []
def test_delete_leaves_unowned_workflow_intact(self, app, pg_conn):
"""Regression: deleting an agent whose ``workflow_id`` points at another
user's workflow (the pre-clone adopted shape) must not touch that graph.
The old explicit node/edge cleanup was not owner-scoped and gutted it."""
from application.api.user.agents.routes import DeleteAgent
from application.storage.db.repositories.agents import AgentsRepository
from application.storage.db.repositories.workflows import WorkflowsRepository
wf, _, _ = _seed_workflow(pg_conn, "__system__", name="Shared Template WF")
repo = AgentsRepository(pg_conn)
agent = repo.create(
"u-wf-del2",
"Adopted WF Agent",
"draft",
agent_type="workflow",
workflow_id=str(wf["id"]),
)
with _patch_db(pg_conn), app.test_request_context(
f"/api/delete_agent?id={agent['id']}", method="DELETE"
):
from flask import request
request.decoded_token = {"sub": "u-wf-del2"}
response = DeleteAgent().delete()
assert response.status_code == 200
assert repo.get_any(str(agent["id"]), "u-wf-del2") is None
# The system's workflow and its whole graph survive.
assert WorkflowsRepository(pg_conn).get_by_id(str(wf["id"])) is not None
nodes, edges = _graph(pg_conn, wf["id"])
assert len(nodes) == 3 and len(edges) == 2
class TestCreateWorkflowAgent:
def test_create_published_with_owned_workflow(self, app, pg_conn):
from application.api.user.agents.routes import CreateAgent
from application.storage.db.repositories.agents import AgentsRepository
user = "u-wf-create"
wf, _, _ = _seed_workflow(pg_conn, user)
with _patch_db(pg_conn), app.test_request_context(
"/api/create_agent",
method="POST",
json={
"name": "My WF Agent",
"description": "d",
"agent_type": "workflow",
"status": "published",
"workflow": str(wf["id"]),
},
):
from flask import request
request.decoded_token = {"sub": user}
response = CreateAgent().post()
assert response.status_code == 201
agent = next(
a
for a in AgentsRepository(pg_conn).list_for_user(user)
if a["name"] == "My WF Agent"
)
assert str(agent["workflow_id"]) == str(wf["id"])
assert agent["status"] == "published"
def test_create_published_with_unowned_workflow_returns_404(self, app, pg_conn):
from application.api.user.agents.routes import CreateAgent
wf, _, _ = _seed_workflow(pg_conn, "someone-else")
with _patch_db(pg_conn), app.test_request_context(
"/api/create_agent",
method="POST",
json={
"name": "Stolen WF Agent",
"description": "d",
"agent_type": "workflow",
"status": "published",
"workflow": str(wf["id"]),
},
):
from flask import request
request.decoded_token = {"sub": "u-wf-thief"}
response = CreateAgent().post()
assert response.status_code == 404
class TestUpdateWorkflowAgent:
def test_update_sets_and_clears_workflow(self, app, pg_conn):
from application.api.user.agents.routes import UpdateAgent
from application.storage.db.repositories.agents import AgentsRepository
user = "u-wf-update"
wf, _, _ = _seed_workflow(pg_conn, user)
repo = AgentsRepository(pg_conn)
agent = repo.create(user, "WF Agent", "draft", agent_type="workflow")
agent_id = str(agent["id"])
with _patch_db(pg_conn), app.test_request_context(
f"/api/update_agent/{agent_id}",
method="PUT",
json={"workflow": str(wf["id"])},
):
from flask import request
request.decoded_token = {"sub": user}
response = UpdateAgent().put(agent_id)
assert response.status_code == 200
assert str(repo.get(agent_id, user)["workflow_id"]) == str(wf["id"])
# Clearing is allowed while the agent stays a draft.
with _patch_db(pg_conn), app.test_request_context(
f"/api/update_agent/{agent_id}",
method="PUT",
json={"workflow": None},
):
from flask import request
request.decoded_token = {"sub": user}
response = UpdateAgent().put(agent_id)
assert response.status_code == 200
assert repo.get(agent_id, user)["workflow_id"] is None
def test_publish_without_workflow_is_rejected(self, app, pg_conn):
from application.api.user.agents.routes import UpdateAgent
from application.storage.db.repositories.agents import AgentsRepository
user = "u-wf-update2"
agent = AgentsRepository(pg_conn).create(
user, "WF Agent", "draft", agent_type="workflow"
)
agent_id = str(agent["id"])
with _patch_db(pg_conn), app.test_request_context(
f"/api/update_agent/{agent_id}",
method="PUT",
json={"status": "published"},
):
from flask import request
request.decoded_token = {"sub": user}
response = UpdateAgent().put(agent_id)
assert response.status_code == 400
File diff suppressed because it is too large. Load diff
-12
View File
@@ -1,12 +0,0 @@
"""Tests for application/api/user/workflows/routes.py.
Previously asserted on bson.ObjectId serialization. Workflow persistence is
now Postgres-backed; coverage will be rebuilt via pg_conn + WorkflowsRepository.
"""
import pytest
@pytest.mark.skip(reason="needs PG fixture rewrite - tracked separately")
def test_user_workflows_routes_pending_pg_rewrite():
pass