From f8325e5458b56c493cfbe49bef141410c2bba68b Mon Sep 17 00:00:00 2001 From: Victor Giers Date: Mon, 15 Jun 2026 15:01:12 +0200 Subject: [PATCH] Introduce core agent framework components: workflow persistence models, tool registry system, and reference binding logic. --- backend/agent/__init__.py | 5 + backend/agent/bindings.py | 67 +++++++++++ backend/agent/models.py | 116 ++++++++++++++++++ backend/agent/registry.py | 228 ++++++++++++++++++++++++++++++++++++ backend/agent/schemas.py | 126 ++++++++++++++++++++ backend/agent/validation.py | 156 ++++++++++++++++++++++++ 6 files changed, 698 insertions(+) create mode 100644 backend/agent/__init__.py create mode 100644 backend/agent/bindings.py create mode 100644 backend/agent/models.py create mode 100644 backend/agent/registry.py create mode 100644 backend/agent/schemas.py create mode 100644 backend/agent/validation.py diff --git a/backend/agent/__init__.py b/backend/agent/__init__.py new file mode 100644 index 0000000..a5d0d33 --- /dev/null +++ b/backend/agent/__init__.py @@ -0,0 +1,5 @@ +"""Native Heimgeist tools and validated workflow execution.""" + +from .registry import NativeToolProvider, ToolDefinition, ToolExecutionContext + +__all__ = ["NativeToolProvider", "ToolDefinition", "ToolExecutionContext"] diff --git a/backend/agent/bindings.py b/backend/agent/bindings.py new file mode 100644 index 0000000..405a8de --- /dev/null +++ b/backend/agent/bindings.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import re +from typing import Any, Dict, Iterable, List, Tuple + + +REFERENCE_RE = re.compile(r"\{\{\s*([A-Za-z0-9_.-]+)\s*\}\}") + + +class BindingError(ValueError): + pass + + +def lookup_reference(reference: str, values: Dict[str, Any]) -> Any: + parts = [part for part in str(reference or "").split(".") if part] + if not parts or parts[0] not in {"input", "run", "nodes"}: + raise BindingError(f"Invalid reference root: {reference}") + current: Any = values + traversed: List[str] = [] + for part in parts: + traversed.append(part) + if isinstance(current, dict) and part in current: + current = current[part] + continue + if isinstance(current, list) and part.isdigit() and int(part) < len(current): + current = current[int(part)] + continue + raise BindingError(f"Missing reference: {'.'.join(traversed)}") + return current + + +def resolve_bindings(value: Any, values: Dict[str, Any]) -> Any: + if isinstance(value, dict): + if set(value) == {"$ref"}: + return lookup_reference(str(value["$ref"]), values) + return {key: resolve_bindings(item, values) for key, item in value.items()} + if isinstance(value, list): + return [resolve_bindings(item, values) for item in value] + if not isinstance(value, str) or "{{" not in value: + return value + matches = list(REFERENCE_RE.finditer(value)) + if len(matches) == 1 and matches[0].span() == (0, len(value)): + return lookup_reference(matches[0].group(1), values) + output = value + for match in reversed(matches): + resolved = lookup_reference(match.group(1), values) + replacement = "" if resolved is None else str(resolved) + output = output[:match.start()] + replacement + output[match.end():] + if "{{" in output or "}}" in output: + raise BindingError("Malformed template expression") + return output + + +def iter_references(value: Any, path: str = "") -> Iterable[Tuple[str, str]]: + if isinstance(value, dict): + if set(value) == {"$ref"}: + yield path or "$", str(value["$ref"]) + return + for key, item in value.items(): + child = f"{path}.{key}" if path else key + yield from iter_references(item, child) + elif isinstance(value, list): + for index, item in enumerate(value): + yield from iter_references(item, f"{path}[{index}]") + elif isinstance(value, str): + for match in REFERENCE_RE.finditer(value): + yield path or "$", match.group(1) diff --git a/backend/agent/models.py b/backend/agent/models.py new file mode 100644 index 0000000..53ecc70 --- /dev/null +++ b/backend/agent/models.py @@ -0,0 +1,116 @@ +from datetime import datetime +import uuid + +from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint +from sqlalchemy.orm import relationship + +from ..database import Base + + +def _uuid() -> str: + return str(uuid.uuid4()) + + +class WorkflowDefinition(Base): + __tablename__ = "workflow_definitions" + + id = Column(String(36), primary_key=True, default=_uuid) + slug = Column(String(100), unique=True, nullable=False, index=True) + name = Column(String(160), nullable=False) + description = Column(Text, nullable=False, default="") + current_revision_id = Column(String(36), nullable=True) + built_in = Column(Boolean, nullable=False, default=False) + enabled = Column(Boolean, nullable=False, default=True) + routing_description = Column(Text, nullable=False, default="") + routing_examples_json = Column(Text, nullable=False, default="[]") + estimated_cost_class = Column(String(32), nullable=False, default="low") + required_capabilities_json = Column(Text, nullable=False, default="[]") + created_at = Column(DateTime, nullable=False, default=datetime.utcnow) + updated_at = Column(DateTime, nullable=False, default=datetime.utcnow, onupdate=datetime.utcnow) + + revisions = relationship( + "WorkflowRevision", + back_populates="workflow", + foreign_keys="WorkflowRevision.workflow_id", + cascade="all, delete-orphan", + ) + + +class WorkflowRevision(Base): + __tablename__ = "workflow_revisions" + __table_args__ = (UniqueConstraint("workflow_id", "version", name="uq_workflow_revision_version"),) + + id = Column(String(36), primary_key=True, default=_uuid) + workflow_id = Column(String(36), ForeignKey("workflow_definitions.id"), nullable=False, index=True) + version = Column(Integer, nullable=False) + graph_json = Column(Text, nullable=False) + created_at = Column(DateTime, nullable=False, default=datetime.utcnow) + created_by = Column(String(64), nullable=False, default="user") + trusted = Column(Boolean, nullable=False, default=False) + checksum = Column(String(64), nullable=False, index=True) + + workflow = relationship("WorkflowDefinition", back_populates="revisions", foreign_keys=[workflow_id]) + + +class WorkflowRun(Base): + __tablename__ = "workflow_runs" + + id = Column(String(36), primary_key=True, default=_uuid) + workflow_id = Column(String(36), ForeignKey("workflow_definitions.id"), nullable=False, index=True) + workflow_revision_id = Column(String(36), ForeignKey("workflow_revisions.id"), nullable=False) + session_id = Column(String(64), nullable=True, index=True) + user_message_id = Column(String(36), nullable=True) + status = Column(String(32), nullable=False, default="queued", index=True) + selection_mode = Column(String(24), nullable=False, default="direct") + inputs_json = Column(Text, nullable=False, default="{}") + outputs_json = Column(Text, nullable=True) + error_json = Column(Text, nullable=True) + metrics_json = Column(Text, nullable=False, default="{}") + started_at = Column(DateTime, nullable=True) + finished_at = Column(DateTime, nullable=True) + cancellation_requested = Column(Boolean, nullable=False, default=False) + + +class WorkflowNodeRun(Base): + __tablename__ = "workflow_node_runs" + __table_args__ = (UniqueConstraint("workflow_run_id", "node_id", "attempt", name="uq_node_run_attempt"),) + + id = Column(String(36), primary_key=True, default=_uuid) + workflow_run_id = Column(String(36), ForeignKey("workflow_runs.id"), nullable=False, index=True) + node_id = Column(String(120), nullable=False) + attempt = Column(Integer, nullable=False, default=1) + status = Column(String(32), nullable=False, default="queued") + resolved_inputs_json = Column(Text, nullable=True) + outputs_json = Column(Text, nullable=True) + error_json = Column(Text, nullable=True) + started_at = Column(DateTime, nullable=True) + finished_at = Column(DateTime, nullable=True) + metrics_json = Column(Text, nullable=False, default="{}") + + +class WorkflowEvent(Base): + __tablename__ = "workflow_events" + __table_args__ = (UniqueConstraint("workflow_run_id", "sequence", name="uq_workflow_event_sequence"),) + + id = Column(String(36), primary_key=True, default=_uuid) + workflow_run_id = Column(String(36), ForeignKey("workflow_runs.id"), nullable=False, index=True) + workflow_id = Column(String(36), nullable=False) + node_id = Column(String(120), nullable=True) + sequence = Column(Integer, nullable=False) + event_type = Column(String(64), nullable=False) + timestamp = Column(DateTime, nullable=False, default=datetime.utcnow) + payload_json = Column(Text, nullable=False, default="{}") + + +class WorkflowConfirmation(Base): + __tablename__ = "workflow_confirmations" + + id = Column(String(36), primary_key=True, default=_uuid) + workflow_run_id = Column(String(36), ForeignKey("workflow_runs.id"), nullable=False, index=True) + node_id = Column(String(120), nullable=False) + tool_name = Column(String(160), nullable=False) + status = Column(String(24), nullable=False, default="pending") + request_json = Column(Text, nullable=False, default="{}") + response_json = Column(Text, nullable=True) + created_at = Column(DateTime, nullable=False, default=datetime.utcnow) + resolved_at = Column(DateTime, nullable=True) diff --git a/backend/agent/registry.py b/backend/agent/registry.py new file mode 100644 index 0000000..89c410c --- /dev/null +++ b/backend/agent/registry.py @@ -0,0 +1,228 @@ +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +import json +from typing import Any, Awaitable, Callable, Dict, Iterable, Optional, Protocol + + +class SchemaValidationError(ValueError): + pass + + +def _type_matches(value: Any, expected: str) -> bool: + if expected == "null": + return value is None + if expected == "object": + return isinstance(value, dict) + if expected == "array": + return isinstance(value, list) + if expected == "string": + return isinstance(value, str) + if expected == "boolean": + return isinstance(value, bool) + if expected == "integer": + return isinstance(value, int) and not isinstance(value, bool) + if expected == "number": + return isinstance(value, (int, float)) and not isinstance(value, bool) + return True + + +def validate_json_schema(value: Any, schema: Dict[str, Any], path: str = "$", *, partial: bool = False) -> None: + if not isinstance(schema, dict): + raise SchemaValidationError(f"{path}: schema must be an object") + if "const" in schema and value != schema["const"]: + raise SchemaValidationError(f"{path}: must equal {schema['const']!r}") + if "enum" in schema and value not in schema["enum"]: + raise SchemaValidationError(f"{path}: must be one of {schema['enum']!r}") + for union_key in ("oneOf", "anyOf"): + if union_key in schema: + matches = 0 + errors = [] + for candidate in schema[union_key]: + try: + validate_json_schema(value, candidate, path, partial=partial) + matches += 1 + except SchemaValidationError as exc: + errors.append(str(exc)) + if (union_key == "oneOf" and matches != 1) or (union_key == "anyOf" and matches == 0): + raise SchemaValidationError(f"{path}: does not match {union_key}: {'; '.join(errors[:3])}") + return + + expected = schema.get("type") + if isinstance(expected, list): + if not any(_type_matches(value, item) for item in expected): + raise SchemaValidationError(f"{path}: expected one of {expected}, got {type(value).__name__}") + elif expected and not _type_matches(value, expected): + raise SchemaValidationError(f"{path}: expected {expected}, got {type(value).__name__}") + + if isinstance(value, dict): + properties = schema.get("properties") or {} + if not partial: + for required in schema.get("required") or []: + if required not in value: + raise SchemaValidationError(f"{path}.{required}: required value is missing") + if schema.get("additionalProperties") is False: + unknown = [key for key in value if key not in properties] + if unknown: + raise SchemaValidationError(f"{path}: unknown properties: {', '.join(unknown)}") + for key, item in value.items(): + if key in properties: + validate_json_schema(item, properties[key], f"{path}.{key}", partial=partial) + minimum = schema.get("minProperties") + if minimum is not None and len(value) < minimum: + raise SchemaValidationError(f"{path}: expected at least {minimum} properties") + elif isinstance(value, list): + if "minItems" in schema and len(value) < int(schema["minItems"]): + raise SchemaValidationError(f"{path}: expected at least {schema['minItems']} items") + if "maxItems" in schema and len(value) > int(schema["maxItems"]): + raise SchemaValidationError(f"{path}: expected at most {schema['maxItems']} items") + item_schema = schema.get("items") + if item_schema: + for index, item in enumerate(value): + validate_json_schema(item, item_schema, f"{path}[{index}]", partial=partial) + elif isinstance(value, str): + if "minLength" in schema and len(value) < int(schema["minLength"]): + raise SchemaValidationError(f"{path}: string is too short") + if "maxLength" in schema and len(value) > int(schema["maxLength"]): + raise SchemaValidationError(f"{path}: string is too long") + elif isinstance(value, (int, float)) and not isinstance(value, bool): + if "minimum" in schema and value < schema["minimum"]: + raise SchemaValidationError(f"{path}: must be at least {schema['minimum']}") + if "maximum" in schema and value > schema["maximum"]: + raise SchemaValidationError(f"{path}: must be at most {schema['maximum']}") + + +def validate_schema_definition(schema: Dict[str, Any], path: str = "$schema") -> None: + if not isinstance(schema, dict): + raise SchemaValidationError(f"{path}: schema must be an object") + expected = schema.get("type") + allowed = {"object", "array", "string", "boolean", "integer", "number", "null"} + if isinstance(expected, str) and expected not in allowed: + raise SchemaValidationError(f"{path}.type: unsupported type {expected!r}") + if isinstance(expected, list) and any(item not in allowed for item in expected): + raise SchemaValidationError(f"{path}.type: contains an unsupported type") + if expected == "object": + properties = schema.get("properties", {}) + if not isinstance(properties, dict): + raise SchemaValidationError(f"{path}.properties: must be an object") + for key, child in properties.items(): + validate_schema_definition(child, f"{path}.properties.{key}") + required = schema.get("required", []) + if not isinstance(required, list) or any(not isinstance(item, str) for item in required): + raise SchemaValidationError(f"{path}.required: must be a string array") + missing = [item for item in required if item not in properties] + if missing: + raise SchemaValidationError(f"{path}.required: unknown properties: {', '.join(missing)}") + if expected == "array" and "items" in schema: + validate_schema_definition(schema["items"], f"{path}.items") + for key in ("oneOf", "anyOf"): + if key in schema: + if not isinstance(schema[key], list) or not schema[key]: + raise SchemaValidationError(f"{path}.{key}: must be a non-empty array") + for index, child in enumerate(schema[key]): + validate_schema_definition(child, f"{path}.{key}[{index}]") + + +ToolHandler = Callable[[Dict[str, Any], "ToolExecutionContext"], Awaitable[Dict[str, Any]]] + + +@dataclass(frozen=True) +class ToolDefinition: + name: str + description: str + input_schema: Dict[str, Any] + output_schema: Dict[str, Any] + handler: ToolHandler + risk_level: str = "low" + timeout_seconds: float = 120.0 + result_size_limit: int = 120_000 + requires_confirmation: bool = False + llm_call: bool = False + + def manifest(self) -> Dict[str, Any]: + return { + "name": self.name, + "description": self.description, + "inputSchema": self.input_schema, + "outputSchema": self.output_schema, + "riskLevel": self.risk_level, + "timeoutSeconds": self.timeout_seconds, + "resultSizeLimit": self.result_size_limit, + "requiresConfirmation": self.requires_confirmation, + } + + def ollama_manifest(self) -> Dict[str, Any]: + return { + "type": "function", + "function": { + "name": self.name, + "description": self.description, + "parameters": self.input_schema, + }, + } + + +@dataclass +class ToolExecutionContext: + run_id: str + workflow_id: str + node_id: str + session_id: Optional[str] + selection_mode: str + explicit_user_action: bool + emit: Callable[[str, Dict[str, Any]], Awaitable[None]] + cancellation_event: asyncio.Event + db_factory: Callable[[], Any] + registry: Optional["NativeToolProvider"] = None + run_values: Dict[str, Any] = field(default_factory=dict) + request_confirmation: Optional[Callable[[ToolDefinition, Dict[str, Any]], Awaitable[bool]]] = None + consume_llm_call: Optional[Callable[[], None]] = None + show_thinking: bool = False + + +class ToolProvider(Protocol): + def list_tools(self) -> Iterable[Dict[str, Any]]: ... + def get_tool(self, name: str) -> ToolDefinition: ... + async def call_tool(self, name: str, arguments: Dict[str, Any], context: ToolExecutionContext) -> Dict[str, Any]: ... + + +class NativeToolProvider: + def __init__(self) -> None: + self._tools: Dict[str, ToolDefinition] = {} + + def register(self, definition: ToolDefinition) -> None: + if definition.name in self._tools: + raise ValueError(f"Duplicate tool name: {definition.name}") + validate_schema_definition(definition.input_schema, f"{definition.name}.inputSchema") + validate_schema_definition(definition.output_schema, f"{definition.name}.outputSchema") + self._tools[definition.name] = definition + + def list_tools(self) -> Iterable[Dict[str, Any]]: + return [tool.manifest() for tool in sorted(self._tools.values(), key=lambda item: item.name)] + + def get_tool(self, name: str) -> ToolDefinition: + try: + return self._tools[name] + except KeyError as exc: + raise KeyError(f"Unknown tool: {name}") from exc + + async def call_tool(self, name: str, arguments: Dict[str, Any], context: ToolExecutionContext) -> Dict[str, Any]: + definition = self.get_tool(name) + validate_json_schema(arguments, definition.input_schema) + if context.cancellation_event.is_set(): + raise asyncio.CancelledError() + if definition.llm_call and context.consume_llm_call: + context.consume_llm_call() + try: + result = await asyncio.wait_for( + definition.handler(arguments, context), + timeout=definition.timeout_seconds, + ) + except asyncio.TimeoutError as exc: + raise TimeoutError(f"Tool {name} exceeded {definition.timeout_seconds:g}s") from exc + validate_json_schema(result, definition.output_schema) + size = len(json.dumps(result, ensure_ascii=False, default=str)) + if size > definition.result_size_limit: + raise ValueError(f"Tool {name} result exceeded {definition.result_size_limit} characters") + return result diff --git a/backend/agent/schemas.py b/backend/agent/schemas.py new file mode 100644 index 0000000..fd3e461 --- /dev/null +++ b/backend/agent/schemas.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +from typing import Any, Dict, List, Literal, Optional + +from pydantic import BaseModel, ConfigDict, Field + + +class WorkflowPosition(BaseModel): + x: float = 0 + y: float = 0 + + +class WorkflowNode(BaseModel): + model_config = ConfigDict(extra="forbid") + + id: str + type: Literal["input", "output", "tool", "prompt", "agent", "condition", "merge", "template", "select", "limit"] + position: WorkflowPosition = Field(default_factory=WorkflowPosition) + config: Dict[str, Any] = Field(default_factory=dict) + + +class WorkflowEdge(BaseModel): + model_config = ConfigDict(extra="forbid") + + id: str + source: str + source_handle: str = "output" + target: str + target_handle: str = "input" + + +class WorkflowLimits(BaseModel): + maximum_nodes: int = 40 + maximum_tool_calls: int = 20 + maximum_llm_calls: int = 8 + maximum_runtime_seconds: int = 300 + maximum_result_chars: int = 120_000 + maximum_concurrency: int = 4 + + +class WorkflowGraph(BaseModel): + model_config = ConfigDict(extra="forbid") + + schema_version: int = 1 + inputs: Dict[str, Dict[str, Any]] = Field(default_factory=dict) + outputs: Dict[str, Dict[str, Any]] = Field(default_factory=dict) + nodes: List[WorkflowNode] + edges: List[WorkflowEdge] + limits: WorkflowLimits = Field(default_factory=WorkflowLimits) + + +class ValidationIssue(BaseModel): + code: str + message: str + node_id: Optional[str] = None + field_path: Optional[str] = None + + +class WorkflowValidationResult(BaseModel): + valid: bool + errors: List[ValidationIssue] = Field(default_factory=list) + warnings: List[ValidationIssue] = Field(default_factory=list) + + +class WorkflowCreateRequest(BaseModel): + name: str + slug: Optional[str] = None + description: str = "" + routing_description: str = "" + routing_examples: List[str] = Field(default_factory=list) + enabled: bool = True + graph: Dict[str, Any] + + +class WorkflowUpdateRequest(BaseModel): + name: Optional[str] = None + description: Optional[str] = None + routing_description: Optional[str] = None + routing_examples: Optional[List[str]] = None + enabled: Optional[bool] = None + + +class WorkflowRevisionRequest(BaseModel): + graph: Dict[str, Any] + created_by: str = "user" + + +class WorkflowValidateRequest(BaseModel): + graph: Dict[str, Any] + + +class WorkflowRunRequest(BaseModel): + session_id: Optional[str] = None + message: str = "" + model: str + selection_mode: Literal["direct", "automatic", "explicit"] = "direct" + workflow_id: Optional[str] = None + workflow_revision_id: Optional[str] = None + library_slug: Optional[str] = None + web_search_enabled: bool = False + searx_url: Optional[str] = None + searx_engines: List[str] = Field(default_factory=list) + attachments: List[Dict[str, Any]] = Field(default_factory=list) + vision_model: Optional[str] = None + transcription_model: Optional[str] = None + stream: bool = True + router_model: Optional[str] = None + show_thinking: bool = False + generation_options: Dict[str, Any] = Field(default_factory=dict) + sample_inputs: Dict[str, Any] = Field(default_factory=dict) + explicit_user_action: bool = False + + +class ConfirmationResponseRequest(BaseModel): + approved: bool + note: Optional[str] = None + + +class RouterSelectRequest(BaseModel): + message: str + recent_messages: List[Dict[str, Any]] = Field(default_factory=list) + attachments: List[Dict[str, Any]] = Field(default_factory=list) + library_slug: Optional[str] = None + model: Optional[str] = None + chat_model: Optional[str] = None + confidence_threshold: float = 0.55 diff --git a/backend/agent/validation.py b/backend/agent/validation.py new file mode 100644 index 0000000..6dcf242 --- /dev/null +++ b/backend/agent/validation.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +from collections import defaultdict, deque +from typing import Any, Dict, List, Optional + +from pydantic import ValidationError + +from .bindings import iter_references +from .registry import NativeToolProvider, SchemaValidationError, validate_json_schema, validate_schema_definition +from .schemas import ValidationIssue, WorkflowGraph, WorkflowValidationResult + + +HOST_LIMITS = { + "maximum_nodes": 80, + "maximum_tool_calls": 40, + "maximum_llm_calls": 16, + "maximum_runtime_seconds": 600, + "maximum_result_chars": 240_000, + "maximum_concurrency": 8, +} +RUN_REFERENCE_FIELDS = { + "session_id", "chat_model", "router_model", "vision_model", "transcription_model", + "library_slug", "searx_url", "searx_engines", "messages", "attachments", + "generation_options", "context_blocks", "target_message_id", "target_url", +} + + +def _issue(code: str, message: str, node_id: Optional[str] = None, field_path: Optional[str] = None) -> ValidationIssue: + return ValidationIssue(code=code, message=message, node_id=node_id, field_path=field_path) + + +def validate_workflow_graph(raw_graph: Dict[str, Any], registry: NativeToolProvider) -> WorkflowValidationResult: + errors: List[ValidationIssue] = [] + warnings: List[ValidationIssue] = [] + try: + graph = WorkflowGraph.model_validate(raw_graph) + except ValidationError as exc: + for item in exc.errors(): + errors.append(_issue("malformed_graph", item["msg"], field_path=".".join(str(p) for p in item["loc"]))) + return WorkflowValidationResult(valid=False, errors=errors) + + if graph.schema_version != 1: + errors.append(_issue("unsupported_schema_version", "Only workflow schema_version 1 is supported.", field_path="schema_version")) + + for name, schema in {**graph.inputs, **graph.outputs}.items(): + try: + validate_schema_definition(schema, f"schema.{name}") + except SchemaValidationError as exc: + errors.append(_issue("malformed_schema", str(exc), field_path=f"inputs.{name}")) + + node_ids = [node.id for node in graph.nodes] + edge_ids = [edge.id for edge in graph.edges] + for identifier, values, path in (("node", node_ids, "nodes"), ("edge", edge_ids, "edges")): + seen = set() + for value in values: + if not value: + errors.append(_issue(f"empty_{identifier}_id", f"{identifier.title()} IDs cannot be empty.", field_path=path)) + elif value in seen: + errors.append(_issue(f"duplicate_{identifier}_id", f"Duplicate {identifier} ID: {value}", field_path=path)) + seen.add(value) + + nodes = {node.id: node for node in graph.nodes} + input_nodes = [node for node in graph.nodes if node.type == "input"] + output_nodes = [node for node in graph.nodes if node.type == "output"] + if not input_nodes: + errors.append(_issue("missing_input_node", "A workflow must contain an input node.")) + if not output_nodes: + errors.append(_issue("missing_output_node", "A workflow must contain an output node.")) + + outgoing: Dict[str, List[Any]] = defaultdict(list) + incoming: Dict[str, List[Any]] = defaultdict(list) + for edge in graph.edges: + if edge.source not in nodes: + errors.append(_issue("missing_edge_source", f"Edge source does not exist: {edge.source}", field_path=f"edges.{edge.id}.source")) + if edge.target not in nodes: + errors.append(_issue("missing_edge_target", f"Edge target does not exist: {edge.target}", field_path=f"edges.{edge.id}.target")) + if edge.source in nodes and edge.target in nodes: + outgoing[edge.source].append(edge) + incoming[edge.target].append(edge) + + indegree = {node_id: len(incoming[node_id]) for node_id in nodes} + queue = deque(node_id for node_id, degree in indegree.items() if degree == 0) + visited: List[str] = [] + while queue: + node_id = queue.popleft() + visited.append(node_id) + for edge in outgoing[node_id]: + indegree[edge.target] -= 1 + if indegree[edge.target] == 0: + queue.append(edge.target) + if len(visited) != len(nodes): + errors.append(_issue("cycle", "Workflow schema version 1 must be a DAG; cycles are not allowed.")) + + reachable = set() + frontier = [node.id for node in input_nodes] + while frontier: + node_id = frontier.pop() + if node_id in reachable: + continue + reachable.add(node_id) + frontier.extend(edge.target for edge in outgoing[node_id]) + for node in graph.nodes: + if node.type not in {"input"} and node.id not in reachable: + errors.append(_issue("unreachable_node", f"Node {node.id} is not reachable from an input node.", node.id)) + + for node in graph.nodes: + if node.type == "tool": + tool_name = str(node.config.get("tool") or "") + try: + tool = registry.get_tool(tool_name) + except KeyError: + errors.append(_issue("unknown_tool", f"Unknown tool: {tool_name or '(missing)'}", node.id, "config.tool")) + else: + arguments = node.config.get("arguments", {}) + if not isinstance(arguments, dict): + errors.append(_issue("invalid_tool_arguments", "Tool arguments must be an object.", node.id, "config.arguments")) + else: + literal_arguments = { + key: value for key, value in arguments.items() + if not (isinstance(value, dict) and set(value) == {"$ref"}) + } + try: + validate_json_schema(literal_arguments, tool.input_schema, partial=True) + except SchemaValidationError as exc: + errors.append(_issue("invalid_tool_arguments", str(exc), node.id, "config.arguments")) + if node.type == "condition": + operation = node.config.get("operation") + allowed = {"equals", "not_equals", "exists", "greater_than", "less_than", "contains", "array_not_empty", "score_at_least"} + if operation not in allowed: + errors.append(_issue("invalid_condition", f"Unsupported condition operation: {operation}", node.id, "config.operation")) + handles = {edge.source_handle for edge in outgoing[node.id]} + if not {"true", "false"}.issubset(handles): + errors.append(_issue("invalid_branch", "Condition nodes require both true and false outgoing branches.", node.id)) + if node.type == "prompt" and not node.config.get("model_source"): + errors.append(_issue("missing_model_source", "Prompt nodes require model_source.", node.id, "config.model_source")) + + for field_path, reference in iter_references(node.config, f"nodes.{node.id}.config"): + parts = reference.split(".") + if parts[0] == "input" and (len(parts) < 2 or parts[1] not in graph.inputs): + errors.append(_issue("missing_reference", f"Unknown workflow input reference: {reference}", node.id, field_path)) + elif parts[0] == "run" and (len(parts) < 2 or parts[1] not in RUN_REFERENCE_FIELDS): + errors.append(_issue("missing_reference", f"Unknown run reference: {reference}", node.id, field_path)) + elif parts[0] == "nodes" and (len(parts) < 3 or parts[1] not in nodes): + errors.append(_issue("missing_reference", f"Unknown node reference: {reference}", node.id, field_path)) + elif parts[0] not in {"input", "run", "nodes"}: + errors.append(_issue("missing_reference", f"Invalid reference: {reference}", node.id, field_path)) + + limits = graph.limits.model_dump() + for key, maximum in HOST_LIMITS.items(): + value = limits[key] + if value < 1 or value > maximum: + errors.append(_issue("excessive_limit", f"{key} must be between 1 and {maximum}.", field_path=f"limits.{key}")) + if len(graph.nodes) > graph.limits.maximum_nodes: + errors.append(_issue("excessive_nodes", "The graph contains more nodes than its maximum_nodes limit.", field_path="nodes")) + + return WorkflowValidationResult(valid=not errors, errors=errors, warnings=warnings)