diff --git a/docs/workflow-runtime-b0.md b/docs/workflow-runtime-b0.md new file mode 100644 index 0000000..6669401 --- /dev/null +++ b/docs/workflow-runtime-b0.md @@ -0,0 +1,44 @@ +# B0：可审计 Workflow DSL 契约 + +本文件记录通用 AI 应用平台阶段 B 已落地的基础运行时。它先固定安全、可版本化的工作流语言，再将 +SQLite 存储、受限执行、审批暂停/恢复和发布评测门禁连接到同一契约；不能以未受约束的 JSON、回调 +或可执行脚本替代该契约。 + +## 当前范围 + +`WorkflowSpec` 是 schema version 为 `1` 的不可变 DAG。它具备： + +- 显式输入、节点、资源引用和公共输出； +- 严格 JSON 解码、精确字段集合、拒绝重复键和非有限数字； +- 规范化 JSON 与 SHA-256 digest，供 Revision、Run、评测和审计共同引用； +- 静态检查输入引用、节点输出、直接依赖、环和不连通节点； +- 仅可绑定 `knowledge_base` 与 `model_profile` 的不含密钥资源 ID。 + +首个原生节点集合是 `knowledge.retrieve`、`prompt.render`、`model.generate`、`grounding.validate`、 +`condition` 和 `human.approval`。每一种节点都有固定输入、输出、资源类型和参数白名单。DSL 不包含代码、 +shell、Provider URL、Provider 密钥、动态工具名或自由表达式。 + +## 已实现的运行时保证 + +- `Workflow / Draft / Revision / Deployment / Run / StepRun / Approval / Evaluation` 均为租户隔离的耐久记录； +- Draft 采用乐观并发版本，Revision 和 Evaluation 不可变，发布会原子替换活跃 Revision； +- 每次执行都记录输入摘要、规范摘要、节点输入/输出摘要、节点状态和安全错误码； +- 原生节点只能由平台注册的执行器实现。运行时拒绝任意代码、动态导入、shell 和定义内凭据； +- 步数、生成调用次数和总时长均受 Revision 预算限制；任一限制触发时运行失败且保留审计记录； +- 人工审批会耐久地暂停 Run。获批后从受限、租户保护的状态恢复；拒绝会终止该 Run； +- 发布前必须存在一份与目标 Revision 摘要完全一致、全部用例通过的评测记录； +- 进程中断时，正在执行的 Run/Step 会标为 `interrupted`，不会被自动重放。 + +## 设计边界 + +- 此切片不改变既有 `knowledge_chat` Application 的兼容契约。 +- 当前原生执行器以明确注册的端口接入；检索和生成的生产适配器必须在组合根中显式提供，不能由 DSL + 决定实现或连接位置。 +- 循环、通用重试和第三方工具尚未进入 DSL。它们必须先具备对应的预算、幂等、审批和审计语义。 +- `WorkflowSpec.digest` 是配置可追溯性，而不是模型输出逐字可复现性的承诺。 + +## 后续切片 + +1. 在 HTTP 与 SDK 边界公开 Workflow 的管理、执行、审批和评测资源，同时维持现有应用接口兼容。 +2. 为检索和生成提供生产组合根适配器，并对资源级权限、证据覆盖和拒答语义做端到端评测。 +3. 引入经过幂等设计的异步队列、超时取消与失败补偿；在此之前不开放通用重试或循环。 diff --git a/rag_system/workflow_contracts.py b/rag_system/workflow_contracts.py new file mode 100644 index 0000000..1ac0fed --- /dev/null +++ b/rag_system/workflow_contracts.py @@ -0,0 +1,632 @@ +"""Strict, storage-neutral Workflow DSL contracts. + +This module is the first B0 platform contract. It deliberately describes only +the bounded native nodes that the platform can audit; a workflow cannot contain +arbitrary code, dynamic imports, shell commands, or provider credentials. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from enum import StrEnum +from types import MappingProxyType +from typing import Any, TypeAlias, TypeVar, cast + +from rag_system.application_contracts import validate_model_profile_id +from rag_system.json_contract import JsonContractError, decode_json_object +from rag_system.knowledge_base_contracts import validate_resource_id + + +WORKFLOW_DSL_SCHEMA_VERSION = 1 +MAX_WORKFLOW_INPUTS = 32 +MAX_WORKFLOW_NODES = 64 +MAX_WORKFLOW_OUTPUTS = 32 +MAX_NODE_PARAMETERS = 16 + +_NAME_PATTERN = re.compile(r"[a-z][a-z0-9_]{0,63}") +_INPUT_REFERENCE_PATTERN = re.compile(r"input\.([a-z][a-z0-9_]{0,63})") +_NODE_REFERENCE_PATTERN = re.compile( + r"node\.([a-z][a-z0-9_]{0,63})\.([a-z][a-z0-9_]{0,63})" +) + +ParameterValue: TypeAlias = str | int | bool + + +class WorkflowContractError(Exception): + """Base class for invalid Workflow DSL values.""" + + +class WorkflowValidationError(WorkflowContractError, ValueError): + """A workflow is unsafe, ambiguous, or outside the supported node set.""" + + +class WorkflowNodeKind(StrEnum): + """Native operations that will receive audited runtime implementations.""" + + KNOWLEDGE_RETRIEVE = "knowledge.retrieve" + PROMPT_RENDER = "prompt.render" + MODEL_GENERATE = "model.generate" + GROUNDING_VALIDATE = "grounding.validate" + CONDITION = "condition" + HUMAN_APPROVAL = "human.approval" + + +class WorkflowResourceKind(StrEnum): + KNOWLEDGE_BASE = "knowledge_base" + MODEL_PROFILE = "model_profile" + + +@dataclass(frozen=True, slots=True) +class WorkflowInput: + """One named, JSON-compatible workflow input. + + Value schemas are intentionally deferred until the runtime has a typed + expression and structured-output contract. Names are still fixed here so + every reference is statically verifiable. + """ + + name: str + required: bool = True + + def __post_init__(self) -> None: + object.__setattr__(self, "name", _validate_name(self.name, "workflow input name")) + if not isinstance(self.required, bool): + raise WorkflowValidationError("workflow input required must be a boolean") + + +@dataclass(frozen=True, slots=True) +class WorkflowResourceRef: + """A secret-free reference to a platform-managed resource.""" + + resource_kind: WorkflowResourceKind + resource_id: str + + def __post_init__(self) -> None: + if not isinstance(self.resource_kind, WorkflowResourceKind): + raise WorkflowValidationError("workflow resource kind is invalid") + try: + if self.resource_kind is WorkflowResourceKind.KNOWLEDGE_BASE: + normalized = validate_resource_id(self.resource_id) + else: + normalized = validate_model_profile_id(self.resource_id) + except ValueError as error: + raise WorkflowValidationError("workflow resource ID is invalid") from error + object.__setattr__(self, "resource_id", normalized) + + +@dataclass(frozen=True, slots=True) +class WorkflowInputBinding: + """Route one declared input or upstream output into a native node input.""" + + target: str + source: str + + def __post_init__(self) -> None: + object.__setattr__(self, "target", _validate_name(self.target, "binding target")) + object.__setattr__(self, "source", _validate_reference(self.source)) + + +@dataclass(frozen=True, slots=True) +class WorkflowNode: + """One bounded native node and its static data dependencies.""" + + node_id: str + node_kind: WorkflowNodeKind + depends_on: tuple[str, ...] = () + input_bindings: tuple[WorkflowInputBinding, ...] = () + output_names: tuple[str, ...] = () + resources: tuple[WorkflowResourceRef, ...] = () + parameters: Mapping[str, ParameterValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + object.__setattr__(self, "node_id", _validate_name(self.node_id, "workflow node ID")) + if not isinstance(self.node_kind, WorkflowNodeKind): + raise WorkflowValidationError("workflow node kind is invalid") + + dependencies = _normalize_names(self.depends_on, "workflow dependencies") + if self.node_id in dependencies: + raise WorkflowValidationError("a workflow node cannot depend on itself") + bindings = _normalize_items( + self.input_bindings, WorkflowInputBinding, "workflow input bindings" + ) + outputs = _normalize_names(self.output_names, "workflow output names") + resources = _normalize_items(self.resources, WorkflowResourceRef, "workflow resources") + parameters = _normalize_parameters(self.parameters) + + if len({binding.target for binding in bindings}) != len(bindings): + raise WorkflowValidationError("workflow node input bindings cannot share a target") + if len(set(outputs)) != len(outputs): + raise WorkflowValidationError("workflow node output names cannot contain duplicates") + if len({(item.resource_kind, item.resource_id) for item in resources}) != len(resources): + raise WorkflowValidationError("workflow node resources cannot contain duplicates") + + _validate_node_shape(self.node_kind, bindings, outputs, resources, parameters) + object.__setattr__(self, "depends_on", dependencies) + object.__setattr__( + self, "input_bindings", tuple(sorted(bindings, key=lambda item: item.target)) + ) + object.__setattr__(self, "output_names", tuple(sorted(outputs))) + object.__setattr__( + self, + "resources", + tuple(sorted(resources, key=lambda item: (item.resource_kind.value, item.resource_id))), + ) + object.__setattr__(self, "parameters", MappingProxyType(parameters)) + + +@dataclass(frozen=True, slots=True) +class WorkflowOutput: + """A named public result selected from an upstream node output.""" + + name: str + source: str + + def __post_init__(self) -> None: + object.__setattr__(self, "name", _validate_name(self.name, "workflow output name")) + object.__setattr__(self, "source", _validate_reference(self.source)) + if _NODE_REFERENCE_PATTERN.fullmatch(self.source) is None: + raise WorkflowValidationError("workflow outputs must reference a node output") + + +@dataclass(frozen=True, slots=True) +class WorkflowSpec: + """An immutable, canonicalizable DAG for an auditable workflow revision.""" + + schema_version: int + inputs: tuple[WorkflowInput, ...] + nodes: tuple[WorkflowNode, ...] + outputs: tuple[WorkflowOutput, ...] + + def __post_init__(self) -> None: + if ( + isinstance(self.schema_version, bool) + or not isinstance(self.schema_version, int) + or self.schema_version != WORKFLOW_DSL_SCHEMA_VERSION + ): + raise WorkflowValidationError("workflow DSL schema version is unsupported") + inputs = _normalize_items(self.inputs, WorkflowInput, "workflow inputs") + nodes = _normalize_items(self.nodes, WorkflowNode, "workflow nodes") + outputs = _normalize_items(self.outputs, WorkflowOutput, "workflow outputs") + if not 1 <= len(inputs) <= MAX_WORKFLOW_INPUTS: + raise WorkflowValidationError("workflow has an invalid input count") + if not 1 <= len(nodes) <= MAX_WORKFLOW_NODES: + raise WorkflowValidationError("workflow has an invalid node count") + if not 1 <= len(outputs) <= MAX_WORKFLOW_OUTPUTS: + raise WorkflowValidationError("workflow has an invalid output count") + if len({item.name for item in inputs}) != len(inputs): + raise WorkflowValidationError("workflow input names cannot contain duplicates") + if len({item.node_id for item in nodes}) != len(nodes): + raise WorkflowValidationError("workflow node IDs cannot contain duplicates") + if len({item.name for item in outputs}) != len(outputs): + raise WorkflowValidationError("workflow output names cannot contain duplicates") + + _validate_workflow_graph(inputs, nodes, outputs) + object.__setattr__(self, "inputs", tuple(sorted(inputs, key=lambda item: item.name))) + object.__setattr__(self, "nodes", tuple(sorted(nodes, key=lambda item: item.node_id))) + object.__setattr__(self, "outputs", tuple(sorted(outputs, key=lambda item: item.name))) + + @property + def digest(self) -> str: + """Stable SHA-256 over the canonical JSON representation.""" + + return hashlib.sha256(self.to_json().encode("utf-8")).hexdigest() + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": self.schema_version, + "inputs": [ + {"name": item.name, "required": item.required} + for item in self.inputs + ], + "nodes": [ + { + "id": item.node_id, + "kind": item.node_kind.value, + "depends_on": list(item.depends_on), + "input_bindings": [ + {"target": binding.target, "source": binding.source} + for binding in item.input_bindings + ], + "output_names": list(item.output_names), + "resources": [ + {"kind": resource.resource_kind.value, "id": resource.resource_id} + for resource in item.resources + ], + "parameters": dict(item.parameters), + } + for item in self.nodes + ], + "outputs": [ + {"name": item.name, "source": item.source} + for item in self.outputs + ], + } + + def to_json(self) -> str: + return json.dumps(self.to_dict(), ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + @classmethod + def from_json(cls, content: str) -> WorkflowSpec: + try: + return cls.from_dict(decode_json_object(content)) + except JsonContractError as error: + raise WorkflowValidationError("workflow DSL must be one strict JSON object") from error + + @classmethod + def from_dict(cls, value: object) -> WorkflowSpec: + payload = _require_mapping(value, "workflow DSL") + _require_exact_keys( + payload, {"schema_version", "inputs", "nodes", "outputs"}, "workflow DSL" + ) + return cls( + schema_version=_require_int(payload["schema_version"], "workflow schema version"), + inputs=tuple( + _input_from_dict(item) for item in _require_list(payload["inputs"], "inputs") + ), + nodes=tuple(_node_from_dict(item) for item in _require_list(payload["nodes"], "nodes")), + outputs=tuple( + _output_from_dict(item) + for item in _require_list(payload["outputs"], "outputs") + ), + ) + + +def _input_from_dict(value: object) -> WorkflowInput: + payload = _require_mapping(value, "workflow input") + _require_exact_keys(payload, {"name", "required"}, "workflow input") + return WorkflowInput( + name=_require_text(payload["name"], "workflow input name"), + required=_require_bool(payload["required"], "workflow input required"), + ) + + +def _node_from_dict(value: object) -> WorkflowNode: + payload = _require_mapping(value, "workflow node") + _require_exact_keys( + payload, + {"id", "kind", "depends_on", "input_bindings", "output_names", "resources", "parameters"}, + "workflow node", + ) + bindings = tuple( + _binding_from_dict(item) + for item in _require_list(payload["input_bindings"], "input bindings") + ) + resources = tuple( + _resource_from_dict(item) for item in _require_list(payload["resources"], "resources") + ) + parameters = _require_mapping(payload["parameters"], "workflow parameters") + try: + kind = WorkflowNodeKind(_require_text(payload["kind"], "workflow node kind")) + except (TypeError, ValueError) as error: + raise WorkflowValidationError("workflow node kind is invalid") from error + return WorkflowNode( + node_id=_require_text(payload["id"], "workflow node ID"), + node_kind=kind, + depends_on=tuple( + _require_text(item, "workflow dependency") + for item in _require_list(payload["depends_on"], "dependencies") + ), + input_bindings=bindings, + output_names=tuple( + _require_text(item, "workflow output name") + for item in _require_list(payload["output_names"], "output names") + ), + resources=resources, + parameters=_parse_parameters(parameters), + ) + + +def _binding_from_dict(value: object) -> WorkflowInputBinding: + payload = _require_mapping(value, "workflow input binding") + _require_exact_keys(payload, {"target", "source"}, "workflow input binding") + return WorkflowInputBinding( + target=_require_text(payload["target"], "binding target"), + source=_require_text(payload["source"], "binding source"), + ) + + +def _resource_from_dict(value: object) -> WorkflowResourceRef: + payload = _require_mapping(value, "workflow resource") + _require_exact_keys(payload, {"kind", "id"}, "workflow resource") + try: + kind = WorkflowResourceKind(_require_text(payload["kind"], "workflow resource kind")) + except (TypeError, ValueError) as error: + raise WorkflowValidationError("workflow resource kind is invalid") from error + return WorkflowResourceRef( + resource_kind=kind, + resource_id=_require_text(payload["id"], "workflow resource ID"), + ) + + +def _output_from_dict(value: object) -> WorkflowOutput: + payload = _require_mapping(value, "workflow output") + _require_exact_keys(payload, {"name", "source"}, "workflow output") + return WorkflowOutput( + name=_require_text(payload["name"], "workflow output name"), + source=_require_text(payload["source"], "workflow output source"), + ) + + +def _validate_workflow_graph( + inputs: tuple[WorkflowInput, ...], + nodes: tuple[WorkflowNode, ...], + outputs: tuple[WorkflowOutput, ...], +) -> None: + input_names = {item.name for item in inputs} + node_by_id = {item.node_id: item for item in nodes} + for node in nodes: + dependencies = set(node.depends_on) + if not dependencies <= set(node_by_id): + raise WorkflowValidationError("workflow node depends on an unknown node") + referenced_dependencies: set[str] = set() + for binding in node.input_bindings: + input_match = _INPUT_REFERENCE_PATTERN.fullmatch(binding.source) + if input_match is not None: + if input_match.group(1) not in input_names: + raise WorkflowValidationError("workflow binding references an unknown input") + continue + node_match = _NODE_REFERENCE_PATTERN.fullmatch(binding.source) + if node_match is None: + raise WorkflowValidationError("workflow binding source is invalid") + source_node_id, source_output = node_match.groups() + source_node = node_by_id.get(source_node_id) + if source_node is None or source_output not in source_node.output_names: + raise WorkflowValidationError("workflow binding references an unknown node output") + referenced_dependencies.add(source_node_id) + if dependencies != referenced_dependencies: + raise WorkflowValidationError( + "workflow dependencies must exactly match node output bindings" + ) + + _reject_cycles(node_by_id) + output_nodes: set[str] = set() + for output in outputs: + match = _NODE_REFERENCE_PATTERN.fullmatch(output.source) + if match is None: + raise WorkflowValidationError("workflow output source is invalid") + node_id, output_name = match.groups() + referenced_node = node_by_id.get(node_id) + if referenced_node is None or output_name not in referenced_node.output_names: + raise WorkflowValidationError("workflow output references an unknown node output") + output_nodes.add(node_id) + + reachable: set[str] = set() + + def mark_reachable(node_id: str) -> None: + if node_id in reachable: + return + reachable.add(node_id) + for dependency in node_by_id[node_id].depends_on: + mark_reachable(dependency) + + for node_id in output_nodes: + mark_reachable(node_id) + if reachable != set(node_by_id): + raise WorkflowValidationError("workflow cannot contain disconnected nodes") + + +def _reject_cycles(nodes: Mapping[str, WorkflowNode]) -> None: + visiting: set[str] = set() + visited: set[str] = set() + + def visit(node_id: str) -> None: + if node_id in visiting: + raise WorkflowValidationError("workflow graph cannot contain a cycle") + if node_id in visited: + return + visiting.add(node_id) + for dependency in nodes[node_id].depends_on: + visit(dependency) + visiting.remove(node_id) + visited.add(node_id) + + for node_id in nodes: + visit(node_id) + + +def _validate_node_shape( + kind: WorkflowNodeKind, + bindings: tuple[WorkflowInputBinding, ...], + outputs: tuple[str, ...], + resources: tuple[WorkflowResourceRef, ...], + parameters: Mapping[str, ParameterValue], +) -> None: + expected_inputs = { + WorkflowNodeKind.KNOWLEDGE_RETRIEVE: {"query"}, + WorkflowNodeKind.PROMPT_RENDER: {"question", "evidence"}, + WorkflowNodeKind.MODEL_GENERATE: {"prompt"}, + WorkflowNodeKind.GROUNDING_VALIDATE: {"answer", "evidence"}, + WorkflowNodeKind.CONDITION: {"validation"}, + WorkflowNodeKind.HUMAN_APPROVAL: {"message"}, + }[kind] + expected_outputs = { + WorkflowNodeKind.KNOWLEDGE_RETRIEVE: {"evidence"}, + WorkflowNodeKind.PROMPT_RENDER: {"prompt"}, + WorkflowNodeKind.MODEL_GENERATE: {"answer"}, + WorkflowNodeKind.GROUNDING_VALIDATE: {"validation"}, + WorkflowNodeKind.CONDITION: {"decision"}, + WorkflowNodeKind.HUMAN_APPROVAL: {"decision"}, + }[kind] + if {item.target for item in bindings} != expected_inputs: + raise WorkflowValidationError("workflow node has unsupported input bindings") + if set(outputs) != expected_outputs: + raise WorkflowValidationError("workflow node has unsupported output names") + + if kind is WorkflowNodeKind.KNOWLEDGE_RETRIEVE: + _require_single_resource(resources, WorkflowResourceKind.KNOWLEDGE_BASE) + elif kind is WorkflowNodeKind.MODEL_GENERATE: + _require_single_resource(resources, WorkflowResourceKind.MODEL_PROFILE) + elif resources: + raise WorkflowValidationError("workflow node kind cannot bind platform resources") + _validate_parameters(kind, parameters) + + +def _require_single_resource( + resources: tuple[WorkflowResourceRef, ...], expected_kind: WorkflowResourceKind +) -> None: + if len(resources) != 1 or resources[0].resource_kind is not expected_kind: + raise WorkflowValidationError("workflow node has an invalid resource binding") + + +def _validate_parameters(kind: WorkflowNodeKind, parameters: Mapping[str, ParameterValue]) -> None: + allowed = { + WorkflowNodeKind.KNOWLEDGE_RETRIEVE: {"max_results"}, + WorkflowNodeKind.PROMPT_RENDER: {"template"}, + WorkflowNodeKind.MODEL_GENERATE: {"max_output_tokens"}, + WorkflowNodeKind.GROUNDING_VALIDATE: {"require_citations"}, + WorkflowNodeKind.CONDITION: {"rule"}, + WorkflowNodeKind.HUMAN_APPROVAL: {"timeout_seconds"}, + }[kind] + if not set(parameters) <= allowed: + raise WorkflowValidationError("workflow node has unsupported parameters") + if kind is WorkflowNodeKind.PROMPT_RENDER: + template = parameters.get("template") + if not isinstance(template, str) or not template.strip(): + raise WorkflowValidationError("prompt.render requires a non-empty template") + if "max_results" in parameters and not _is_int_between(parameters["max_results"], 1, 20): + raise WorkflowValidationError("knowledge.retrieve max_results must be between 1 and 20") + if "max_output_tokens" in parameters and not _is_int_between( + parameters["max_output_tokens"], 1, 8_192 + ): + raise WorkflowValidationError("model.generate max_output_tokens must be between 1 and 8192") + if "require_citations" in parameters and not isinstance(parameters["require_citations"], bool): + raise WorkflowValidationError("grounding.validate require_citations must be a boolean") + if "rule" in parameters and parameters["rule"] != "evidence_sufficient": + raise WorkflowValidationError("condition rule is unsupported") + if "timeout_seconds" in parameters and not _is_int_between( + parameters["timeout_seconds"], 60, 604_800 + ): + raise WorkflowValidationError( + "human.approval timeout_seconds must be between 60 and 604800" + ) + + +def _normalize_names(value: object, description: str) -> tuple[str, ...]: + items = _require_sequence(value, description) + names = tuple(_validate_name(item, description) for item in items) + if len(set(names)) != len(names): + raise WorkflowValidationError(f"{description} cannot contain duplicates") + return tuple(sorted(names)) + + +T = TypeVar("T") + + +def _normalize_items(value: object, expected_type: type[T], description: str) -> tuple[T, ...]: + items = _require_sequence(value, description) + if any(not isinstance(item, expected_type) for item in items): + raise WorkflowValidationError(f"{description} contains an invalid item") + return tuple(cast(T, item) for item in items) + + +def _normalize_parameters(value: object) -> dict[str, ParameterValue]: + parameters = _require_mapping(value, "workflow parameters") + if len(parameters) > MAX_NODE_PARAMETERS: + raise WorkflowValidationError("workflow node has too many parameters") + normalized: dict[str, ParameterValue] = {} + for key, item in parameters.items(): + normalized_key = _validate_name(key, "workflow parameter name") + if isinstance(item, bool): + normalized[normalized_key] = item + elif isinstance(item, int): + normalized[normalized_key] = item + elif isinstance(item, str) and len(item) <= 8_000 and _safe_text(item): + normalized[normalized_key] = item + else: + raise WorkflowValidationError("workflow parameter value is invalid") + return dict(sorted(normalized.items())) + + +def _parse_parameters(value: Mapping[str, object]) -> dict[str, ParameterValue]: + """Narrow untrusted JSON values before constructing a typed node.""" + + return _normalize_parameters(value) + + +def _validate_name(value: object, description: str) -> str: + if not isinstance(value, str) or _NAME_PATTERN.fullmatch(value) is None: + raise WorkflowValidationError(f"{description} has an invalid format") + return value + + +def _validate_reference(value: object) -> str: + if not isinstance(value, str) or ( + _INPUT_REFERENCE_PATTERN.fullmatch(value) is None + and _NODE_REFERENCE_PATTERN.fullmatch(value) is None + ): + raise WorkflowValidationError("workflow reference has an invalid format") + return value + + +def _require_sequence(value: object, description: str) -> tuple[object, ...]: + if isinstance(value, (str, bytes)) or not isinstance(value, Sequence): + raise WorkflowValidationError(f"{description} must be a sequence") + return tuple(value) + + +def _require_mapping(value: object, description: str) -> Mapping[str, object]: + if not isinstance(value, Mapping) or any(not isinstance(key, str) for key in value): + raise WorkflowValidationError(f"{description} must be an object") + return value + + +def _require_text(value: object, description: str) -> str: + if not isinstance(value, str): + raise WorkflowValidationError(f"{description} must be text") + return value + + +def _require_bool(value: object, description: str) -> bool: + if not isinstance(value, bool): + raise WorkflowValidationError(f"{description} must be a boolean") + return value + + +def _require_int(value: object, description: str) -> int: + if isinstance(value, bool) or not isinstance(value, int): + raise WorkflowValidationError(f"{description} must be an integer") + return value + + +def _require_list(value: object, description: str) -> list[object]: + if not isinstance(value, list): + raise WorkflowValidationError(f"{description} must be an array") + return value + + +def _require_exact_keys( + payload: Mapping[str, object], expected: set[str], description: str +) -> None: + if set(payload) != expected: + raise WorkflowValidationError(f"{description} has an invalid shape") + + +def _is_int_between(value: ParameterValue, minimum: int, maximum: int) -> bool: + return isinstance(value, int) and not isinstance(value, bool) and minimum <= value <= maximum + + +def _safe_text(value: str) -> bool: + return all(ord(character) >= 32 or character in {"\n", "\r", "\t"} for character in value) + + +__all__ = [ + "MAX_NODE_PARAMETERS", + "MAX_WORKFLOW_INPUTS", + "MAX_WORKFLOW_NODES", + "MAX_WORKFLOW_OUTPUTS", + "WORKFLOW_DSL_SCHEMA_VERSION", + "WorkflowContractError", + "WorkflowInput", + "WorkflowInputBinding", + "WorkflowNode", + "WorkflowNodeKind", + "WorkflowOutput", + "WorkflowResourceKind", + "WorkflowResourceRef", + "WorkflowSpec", + "WorkflowValidationError", +] diff --git a/rag_system/workflow_models.py b/rag_system/workflow_models.py new file mode 100644 index 0000000..bd1e6fa --- /dev/null +++ b/rag_system/workflow_models.py @@ -0,0 +1,472 @@ +"""Storage-neutral identities and durable state contracts for workflows.""" + +from __future__ import annotations + +import json +import re +from collections.abc import Mapping +from dataclasses import dataclass +from enum import StrEnum +from types import MappingProxyType +from typing import Any, cast + +from rag_system.application_contracts import ( + is_valid_timestamp, + validate_change_summary, + validate_display_name, + validate_project_id, + validate_subject, +) +from rag_system.tenancy import TenantId +from rag_system.workflow_contracts import WorkflowSpec + + +_WORKFLOW_ID = re.compile(r"wf_[A-Za-z0-9_-]{32}") +_WORKFLOW_REVISION_ID = re.compile(r"wfr_[A-Za-z0-9_-]{32}") +_WORKFLOW_DEPLOYMENT_ID = re.compile(r"wfd_[A-Za-z0-9_-]{32}") +_WORKFLOW_RUN_ID = re.compile(r"wrun_[A-Za-z0-9_-]{32}") +_WORKFLOW_STEP_RUN_ID = re.compile(r"wstep_[A-Za-z0-9_-]{32}") +_WORKFLOW_APPROVAL_ID = re.compile(r"wappr_[A-Za-z0-9_-]{32}") +_WORKFLOW_EVALUATION_ID = re.compile(r"weval_[A-Za-z0-9_-]{32}") +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class WorkflowModelError(ValueError): + """A durable workflow value violates the platform contract.""" + + +class WorkflowStatus(StrEnum): + ACTIVE = "active" + ARCHIVED = "archived" + + +class WorkflowDeploymentStatus(StrEnum): + ACTIVE = "active" + SUPERSEDED = "superseded" + + +class WorkflowRunStatus(StrEnum): + CREATED = "created" + QUEUED = "queued" + RUNNING = "running" + WAITING_APPROVAL = "waiting_approval" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + INTERRUPTED = "interrupted" + + +class WorkflowStepStatus(StrEnum): + PENDING = "pending" + RUNNING = "running" + WAITING_APPROVAL = "waiting_approval" + SUCCEEDED = "succeeded" + FAILED = "failed" + INTERRUPTED = "interrupted" + + +class ApprovalDecision(StrEnum): + APPROVED = "approved" + REJECTED = "rejected" + + +@dataclass(frozen=True, slots=True) +class ExecutionBudget: + """Revision-scoped bounds that every run must obey before executing nodes.""" + + max_steps: int = 32 + max_model_calls: int = 4 + max_wall_seconds: int = 120 + + def __post_init__(self) -> None: + _validate_int(self.max_steps, "max_steps", minimum=1, maximum=64) + _validate_int(self.max_model_calls, "max_model_calls", minimum=0, maximum=32) + _validate_int(self.max_wall_seconds, "max_wall_seconds", minimum=1, maximum=3_600) + + +@dataclass(frozen=True, slots=True) +class Workflow: + workflow_id: str + tenant_id: TenantId + project_id: str + display_name: str + active_revision_id: str | None + status: WorkflowStatus + created_at: float + updated_at: float + + def __post_init__(self) -> None: + validate_workflow_id(self.workflow_id) + if not isinstance(self.tenant_id, TenantId): + raise WorkflowModelError("workflow tenant is invalid") + validate_project_id(self.project_id) + object.__setattr__(self, "display_name", validate_display_name(self.display_name)) + if self.active_revision_id is not None: + validate_workflow_revision_id(self.active_revision_id) + if not isinstance(self.status, WorkflowStatus): + raise WorkflowModelError("workflow status is invalid") + _validate_time_range(self.created_at, self.updated_at) + + +@dataclass(frozen=True, slots=True) +class WorkflowDraft: + workflow_id: str + version: int + specification: WorkflowSpec | None + budget: ExecutionBudget | None + updated_at: float + updated_by: str + change_summary: str = "" + + def __post_init__(self) -> None: + validate_workflow_id(self.workflow_id) + _validate_int(self.version, "draft version", minimum=0, maximum=2_147_483_647) + if self.specification is not None and not isinstance(self.specification, WorkflowSpec): + raise WorkflowModelError("workflow draft specification is invalid") + if self.budget is not None and not isinstance(self.budget, ExecutionBudget): + raise WorkflowModelError("workflow draft budget is invalid") + if (self.specification is None) != (self.budget is None): + raise WorkflowModelError("workflow draft specification and budget must be set together") + if not is_valid_timestamp(self.updated_at): + raise WorkflowModelError("workflow draft timestamp is invalid") + object.__setattr__(self, "updated_by", validate_subject(self.updated_by)) + if self.specification is None: + if self.change_summary: + raise WorkflowModelError("an empty workflow draft cannot have a change summary") + else: + object.__setattr__(self, "change_summary", validate_change_summary(self.change_summary)) + + +@dataclass(frozen=True, slots=True) +class WorkflowRevision: + revision_id: str + workflow_id: str + revision_number: int + specification: WorkflowSpec + budget: ExecutionBudget + created_at: float + created_by: str + change_summary: str + + def __post_init__(self) -> None: + validate_workflow_revision_id(self.revision_id) + validate_workflow_id(self.workflow_id) + _validate_int(self.revision_number, "workflow revision number", minimum=1, maximum=2_147_483_647) + if not isinstance(self.specification, WorkflowSpec): + raise WorkflowModelError("workflow revision specification is invalid") + if not isinstance(self.budget, ExecutionBudget): + raise WorkflowModelError("workflow revision budget is invalid") + if not is_valid_timestamp(self.created_at): + raise WorkflowModelError("workflow revision timestamp is invalid") + object.__setattr__(self, "created_by", validate_subject(self.created_by)) + object.__setattr__(self, "change_summary", validate_change_summary(self.change_summary)) + + @property + def specification_digest(self) -> str: + return self.specification.digest + + +@dataclass(frozen=True, slots=True) +class WorkflowDeployment: + deployment_id: str + workflow_id: str + revision_id: str + deployed_at: float + deployed_by: str + status: WorkflowDeploymentStatus = WorkflowDeploymentStatus.ACTIVE + + def __post_init__(self) -> None: + validate_workflow_deployment_id(self.deployment_id) + validate_workflow_id(self.workflow_id) + validate_workflow_revision_id(self.revision_id) + if not is_valid_timestamp(self.deployed_at): + raise WorkflowModelError("workflow deployment timestamp is invalid") + object.__setattr__(self, "deployed_by", validate_subject(self.deployed_by)) + if not isinstance(self.status, WorkflowDeploymentStatus): + raise WorkflowModelError("workflow deployment status is invalid") + + +@dataclass(frozen=True, slots=True) +class WorkflowRun: + run_id: str + workflow_id: str + revision_id: str + specification_digest: str + status: WorkflowRunStatus + created_at: float + updated_at: float + created_by: str + input_digest: str + error_code: str | None = None + + def __post_init__(self) -> None: + validate_workflow_run_id(self.run_id) + validate_workflow_id(self.workflow_id) + validate_workflow_revision_id(self.revision_id) + _validate_digest(self.specification_digest, "workflow specification digest") + _validate_digest(self.input_digest, "workflow input digest") + if not isinstance(self.status, WorkflowRunStatus): + raise WorkflowModelError("workflow run status is invalid") + _validate_time_range(self.created_at, self.updated_at) + object.__setattr__(self, "created_by", validate_subject(self.created_by)) + if self.error_code is not None: + object.__setattr__(self, "error_code", _validate_error_code(self.error_code)) + + +@dataclass(frozen=True, slots=True) +class WorkflowStepRun: + step_run_id: str + run_id: str + node_id: str + status: WorkflowStepStatus + started_at: float | None + finished_at: float | None + input_digest: str | None + output_digest: str | None + error_code: str | None = None + + def __post_init__(self) -> None: + validate_workflow_step_run_id(self.step_run_id) + validate_workflow_run_id(self.run_id) + if not isinstance(self.node_id, str) or not self.node_id: + raise WorkflowModelError("workflow step node ID is invalid") + if not isinstance(self.status, WorkflowStepStatus): + raise WorkflowModelError("workflow step status is invalid") + _validate_optional_timestamp(self.started_at, "workflow step start timestamp") + _validate_optional_timestamp(self.finished_at, "workflow step finish timestamp") + if self.started_at is None and self.finished_at is not None: + raise WorkflowModelError("workflow step cannot finish before it starts") + if self.started_at is not None and self.finished_at is not None and self.finished_at < self.started_at: + raise WorkflowModelError("workflow step finish timestamp is invalid") + _validate_optional_digest(self.input_digest, "workflow step input digest") + _validate_optional_digest(self.output_digest, "workflow step output digest") + if self.error_code is not None: + object.__setattr__(self, "error_code", _validate_error_code(self.error_code)) + + +@dataclass(frozen=True, slots=True) +class WorkflowApproval: + approval_id: str + run_id: str + node_id: str + requested_at: float + requested_by: str + decision: ApprovalDecision | None = None + decided_at: float | None = None + decided_by: str | None = None + + def __post_init__(self) -> None: + validate_workflow_approval_id(self.approval_id) + validate_workflow_run_id(self.run_id) + if not isinstance(self.node_id, str) or not self.node_id: + raise WorkflowModelError("workflow approval node ID is invalid") + if not is_valid_timestamp(self.requested_at): + raise WorkflowModelError("workflow approval timestamp is invalid") + object.__setattr__(self, "requested_by", validate_subject(self.requested_by)) + if self.decision is None: + if self.decided_at is not None or self.decided_by is not None: + raise WorkflowModelError("pending workflow approval cannot be decided") + return + if not isinstance(self.decision, ApprovalDecision): + raise WorkflowModelError("workflow approval decision is invalid") + if not is_valid_timestamp(self.decided_at) or _timestamp(self.decided_at) < _timestamp( + self.requested_at + ): + raise WorkflowModelError("workflow approval decision timestamp is invalid") + object.__setattr__(self, "decided_by", validate_subject(self.decided_by)) + + +@dataclass(frozen=True, slots=True) +class WorkflowRunState: + """Tenant-protected resumable state for a paused workflow run. + + The workflow definition never contains secrets. Runtime values are stored + only while a run is active or awaiting approval, and are deliberately + bounded to keep a single run from exhausting the local durable profile. + """ + + run_id: str + input_values: Mapping[str, Any] + node_outputs: Mapping[str, Mapping[str, Any]] + updated_at: float + + def __post_init__(self) -> None: + validate_workflow_run_id(self.run_id) + if not is_valid_timestamp(self.updated_at): + raise WorkflowModelError("workflow run state timestamp is invalid") + encoded_inputs = _canonical_json_object(self.input_values, "workflow run inputs") + normalized_inputs = json.loads(encoded_inputs) + normalized_outputs: dict[str, Mapping[str, Any]] = {} + if not isinstance(self.node_outputs, Mapping): + raise WorkflowModelError("workflow run outputs are invalid") + for node_id, output in self.node_outputs.items(): + if not isinstance(node_id, str) or not re.fullmatch(r"[a-z][a-z0-9_]{0,63}", node_id): + raise WorkflowModelError("workflow run output node ID is invalid") + encoded_output = _canonical_json_object(output, "workflow node output") + normalized_outputs[node_id] = MappingProxyType(json.loads(encoded_output)) + encoded_outputs = json.dumps( + {key: dict(value) for key, value in sorted(normalized_outputs.items())}, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + if len((encoded_inputs + encoded_outputs).encode("utf-8")) > 512 * 1024: + raise WorkflowModelError("workflow run state is too large") + object.__setattr__(self, "input_values", MappingProxyType(normalized_inputs)) + object.__setattr__(self, "node_outputs", MappingProxyType(normalized_outputs)) + + @property + def input_json(self) -> str: + return json.dumps(dict(self.input_values), ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + @property + def outputs_json(self) -> str: + return json.dumps( + {key: dict(value) for key, value in self.node_outputs.items()}, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + + +@dataclass(frozen=True, slots=True) +class WorkflowEvaluation: + """Immutable release evidence tied to the exact workflow specification.""" + + evaluation_id: str + workflow_id: str + revision_id: str + specification_digest: str + generated_at: float + case_count: int + passed_case_count: int + + def __post_init__(self) -> None: + validate_workflow_evaluation_id(self.evaluation_id) + validate_workflow_id(self.workflow_id) + validate_workflow_revision_id(self.revision_id) + _validate_digest(self.specification_digest, "workflow evaluation specification digest") + if not is_valid_timestamp(self.generated_at): + raise WorkflowModelError("workflow evaluation timestamp is invalid") + _validate_int(self.case_count, "workflow evaluation case count", minimum=1, maximum=100_000) + _validate_int( + self.passed_case_count, "workflow evaluation passing case count", + minimum=0, maximum=self.case_count, + ) + + @property + def passed(self) -> bool: + """Production gating is intentionally strict for the first runtime profile.""" + + return self.passed_case_count == self.case_count + + +def validate_workflow_id(value: object) -> str: + return _validate_id(value, _WORKFLOW_ID, "workflow ID") + + +def validate_workflow_revision_id(value: object) -> str: + return _validate_id(value, _WORKFLOW_REVISION_ID, "workflow revision ID") + + +def validate_workflow_deployment_id(value: object) -> str: + return _validate_id(value, _WORKFLOW_DEPLOYMENT_ID, "workflow deployment ID") + + +def validate_workflow_run_id(value: object) -> str: + return _validate_id(value, _WORKFLOW_RUN_ID, "workflow run ID") + + +def validate_workflow_step_run_id(value: object) -> str: + return _validate_id(value, _WORKFLOW_STEP_RUN_ID, "workflow step run ID") + + +def validate_workflow_approval_id(value: object) -> str: + return _validate_id(value, _WORKFLOW_APPROVAL_ID, "workflow approval ID") + + +def validate_workflow_evaluation_id(value: object) -> str: + return _validate_id(value, _WORKFLOW_EVALUATION_ID, "workflow evaluation ID") + + +def _validate_id(value: object, pattern: re.Pattern[str], description: str) -> str: + if not isinstance(value, str) or pattern.fullmatch(value) is None: + raise WorkflowModelError(f"{description} has an invalid format") + return value + + +def _validate_int(value: object, description: str, *, minimum: int, maximum: int) -> None: + if isinstance(value, bool) or not isinstance(value, int) or not minimum <= value <= maximum: + raise WorkflowModelError(f"{description} is invalid") + + +def _validate_time_range(created_at: object, updated_at: object) -> None: + if not is_valid_timestamp(created_at) or not is_valid_timestamp(updated_at): + raise WorkflowModelError("workflow timestamps are invalid") + if _timestamp(updated_at) < _timestamp(created_at): + raise WorkflowModelError("workflow timestamps are invalid") + + +def _validate_optional_timestamp(value: object, description: str) -> None: + if value is not None and not is_valid_timestamp(value): + raise WorkflowModelError(f"{description} is invalid") + + +def _timestamp(value: object) -> float: + return float(cast(int | float, value)) + + +def _validate_digest(value: object, description: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise WorkflowModelError(f"{description} is invalid") + + +def _validate_optional_digest(value: object, description: str) -> None: + if value is not None: + _validate_digest(value, description) + + +def _validate_error_code(value: object) -> str: + if not isinstance(value, str) or not re.fullmatch(r"[a-z][a-z0-9_]{0,63}", value): + raise WorkflowModelError("workflow error code is invalid") + return value + + +def _canonical_json_object(value: object, description: str) -> str: + if not isinstance(value, Mapping): + raise WorkflowModelError(f"{description} are invalid") + try: + encoded = json.dumps(dict(value), ensure_ascii=False, separators=(",", ":"), sort_keys=True) + decoded = json.loads(encoded) + except (TypeError, ValueError) as error: + raise WorkflowModelError(f"{description} must be JSON-compatible") from error + if not isinstance(decoded, dict): + raise WorkflowModelError(f"{description} are invalid") + return encoded + + +__all__ = [ + "ApprovalDecision", + "ExecutionBudget", + "Workflow", + "WorkflowApproval", + "WorkflowDeployment", + "WorkflowDeploymentStatus", + "WorkflowDraft", + "WorkflowEvaluation", + "WorkflowModelError", + "WorkflowRevision", + "WorkflowRun", + "WorkflowRunState", + "WorkflowRunStatus", + "WorkflowStatus", + "WorkflowStepRun", + "WorkflowStepStatus", + "validate_workflow_approval_id", + "validate_workflow_deployment_id", + "validate_workflow_evaluation_id", + "validate_workflow_id", + "validate_workflow_revision_id", + "validate_workflow_run_id", + "validate_workflow_step_run_id", +] diff --git a/rag_system/workflow_native_nodes.py b/rag_system/workflow_native_nodes.py new file mode 100644 index 0000000..73269a3 --- /dev/null +++ b/rag_system/workflow_native_nodes.py @@ -0,0 +1,115 @@ +"""Safe built-in implementations for deterministic workflow node kinds. + +Resource-backed retrieval and generation remain explicit composition-root +ports. These helpers only implement operations that can be performed without +provider selection, credentials, or arbitrary expression evaluation. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any, cast + +from rag_system.tenancy import Principal +from rag_system.workflow_contracts import WorkflowNode, WorkflowNodeKind +from rag_system.workflow_runtime import NativeWorkflowNodeExecutor, WorkflowNodeExecutionError + + +def built_in_node_executors( + *, + retrieve: NativeWorkflowNodeExecutor, + generate: NativeWorkflowNodeExecutor, +) -> dict[WorkflowNodeKind, NativeWorkflowNodeExecutor]: + """Return the closed native node set for one configured runtime profile.""" + + if not callable(retrieve) or not callable(generate): + raise TypeError("retrieve and generate executors must be callable") + return { + WorkflowNodeKind.KNOWLEDGE_RETRIEVE: retrieve, + WorkflowNodeKind.PROMPT_RENDER: cast(NativeWorkflowNodeExecutor, render_prompt), + WorkflowNodeKind.MODEL_GENERATE: generate, + WorkflowNodeKind.GROUNDING_VALIDATE: cast(NativeWorkflowNodeExecutor, validate_grounding), + WorkflowNodeKind.CONDITION: cast(NativeWorkflowNodeExecutor, evaluate_condition), + } + + +def render_prompt( + _principal: Principal, node: WorkflowNode, values: Mapping[str, Any] +) -> Mapping[str, Any]: + """Perform literal placeholder substitution, never template evaluation.""" + + _require_kind(node, WorkflowNodeKind.PROMPT_RENDER) + template = node.parameters.get("template") + if not isinstance(template, str): + raise WorkflowNodeExecutionError("prompt_template_invalid") + question = _text(values.get("question")) + evidence = _render_evidence(values.get("evidence")) + return {"prompt": template.replace("{{ question }}", question).replace("{{ evidence }}", evidence)} + + +def validate_grounding( + _principal: Principal, node: WorkflowNode, values: Mapping[str, Any] +) -> Mapping[str, Any]: + """Produce a simple, explainable gate used by the fixed condition node.""" + + _require_kind(node, WorkflowNodeKind.GROUNDING_VALIDATE) + answer = _text(values.get("answer")).strip() + evidence = values.get("evidence") + has_evidence = _evidence_present(evidence) + require_citations = bool(node.parameters.get("require_citations", False)) + return { + "validation": { + "answer_present": bool(answer), + "evidence_present": has_evidence, + "evidence_sufficient": bool(answer) and (has_evidence or not require_citations), + } + } + + +def evaluate_condition( + _principal: Principal, node: WorkflowNode, values: Mapping[str, Any] +) -> Mapping[str, Any]: + """Evaluate only the statically declared ``evidence_sufficient`` rule.""" + + _require_kind(node, WorkflowNodeKind.CONDITION) + validation = values.get("validation") + if not isinstance(validation, Mapping) or not isinstance(validation.get("evidence_sufficient"), bool): + raise WorkflowNodeExecutionError("condition_input_invalid") + return {"decision": "allow" if validation["evidence_sufficient"] else "refuse"} + + +def _require_kind(node: WorkflowNode, expected: WorkflowNodeKind) -> None: + if not isinstance(node, WorkflowNode) or node.node_kind is not expected: + raise WorkflowNodeExecutionError("native_node_contract_invalid") + + +def _text(value: object) -> str: + if not isinstance(value, str): + raise WorkflowNodeExecutionError("native_node_input_invalid") + return value + + +def _render_evidence(value: object) -> str: + if isinstance(value, str): + return value + if isinstance(value, (list, tuple)): + return "\n".join(_text(item) for item in value) + if isinstance(value, Mapping): + return "\n".join(f"{key}: {item}" for key, item in sorted(value.items()) if isinstance(key, str)) + raise WorkflowNodeExecutionError("native_node_input_invalid") + + +def _evidence_present(value: object) -> bool: + if isinstance(value, str): + return bool(value.strip()) + if isinstance(value, (list, tuple, Mapping)): + return bool(value) + return False + + +__all__ = [ + "built_in_node_executors", + "evaluate_condition", + "render_prompt", + "validate_grounding", +] diff --git a/rag_system/workflow_runtime.py b/rag_system/workflow_runtime.py new file mode 100644 index 0000000..89877f1 --- /dev/null +++ b/rag_system/workflow_runtime.py @@ -0,0 +1,371 @@ +"""Bounded, resumable execution for immutable workflow revisions. + +The runtime deliberately accepts only registered native node executors. It +does not evaluate user code, invoke shells, or resolve credentials from a +workflow definition. Every state transition and node result is persisted +through :mod:`workflow_store` before execution advances. +""" + +from __future__ import annotations + +import hashlib +import json +import secrets +import time +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from typing import Any, Protocol + +from rag_system.tenancy import Principal +from rag_system.workflow_contracts import WorkflowNode, WorkflowNodeKind, WorkflowSpec +from rag_system.workflow_models import ( + ApprovalDecision, + WorkflowApproval, + WorkflowRevision, + WorkflowRun, + WorkflowRunState, + WorkflowRunStatus, + WorkflowStepRun, + WorkflowStepStatus, +) +from rag_system.workflow_store import WorkflowStore + + +class WorkflowRuntimeError(Exception): + """A safe, structured workflow execution refusal or failure.""" + + +class WorkflowNotPublishedError(WorkflowRuntimeError): + def __init__(self) -> None: + super().__init__("Workflow is not published.") + + +class WorkflowExecutionBudgetError(WorkflowRuntimeError): + def __init__(self, code: str) -> None: + super().__init__("Workflow execution budget was exceeded.") + self.code = code + + +class WorkflowNodeExecutionError(WorkflowRuntimeError): + def __init__(self, code: str = "node_execution_failed") -> None: + super().__init__("Workflow node execution failed.") + self.code = code + + +class NativeWorkflowNodeExecutor(Protocol): + """Implementation port for one pre-approved native node kind.""" + + def __call__( + self, + principal: Principal, + node: WorkflowNode, + values: Mapping[str, Any], + ) -> Mapping[str, Any]: ... + + +@dataclass(frozen=True, slots=True) +class WorkflowExecution: + run: WorkflowRun + outputs: Mapping[str, Any] | None + pending_approval_id: str | None = None + + def __post_init__(self) -> None: + if self.pending_approval_id is not None and self.run.status is not WorkflowRunStatus.WAITING_APPROVAL: + raise ValueError("only a waiting run can have a pending approval") + if self.run.status is WorkflowRunStatus.SUCCEEDED and self.outputs is None: + raise ValueError("a successful run must have outputs") + if self.run.status is not WorkflowRunStatus.SUCCEEDED and self.outputs is not None: + raise ValueError("only a successful run can have outputs") + + +class WorkflowRuntime: + """Execute one active revision with hard bounds and durable pause/resume.""" + + def __init__( + self, + store: WorkflowStore, + executors: Mapping[WorkflowNodeKind, NativeWorkflowNodeExecutor], + *, + clock: Callable[[], float] = time.time, + ) -> None: + if not callable(clock): + raise TypeError("clock must be callable") + if not isinstance(executors, Mapping): + raise TypeError("executors must be a mapping") + normalized: dict[WorkflowNodeKind, NativeWorkflowNodeExecutor] = {} + for kind, executor in executors.items(): + if not isinstance(kind, WorkflowNodeKind) or not callable(executor): + raise TypeError("workflow executors must have native node kinds") + normalized[kind] = executor + self._store = store + self._executors = normalized + self._clock = clock + + def start( + self, principal: Principal, workflow_id: str, input_values: Mapping[str, Any] + ) -> WorkflowExecution: + workflow = self._store.get_workflow(principal, workflow_id) + if workflow.active_revision_id is None: + raise WorkflowNotPublishedError() + revision = self._store.get_revision(principal, workflow.workflow_id, workflow.active_revision_id) + inputs = _validate_inputs(revision.specification, input_values) + now = self._now() + run = WorkflowRun( + run_id=_new_id("wrun"), + workflow_id=workflow.workflow_id, + revision_id=revision.revision_id, + specification_digest=revision.specification_digest, + status=WorkflowRunStatus.CREATED, + created_at=now, + updated_at=now, + created_by=principal.subject, + input_digest=_digest(inputs), + ) + created = self._store.create_run(principal, run) + self._store.save_run_state( + principal, + WorkflowRunState( + run_id=created.run_id, input_values=inputs, node_outputs={}, updated_at=now + ), + ) + queued = self._store.transition_run( + principal, created.run_id, status=WorkflowRunStatus.QUEUED, updated_at=now + ) + return self._run(principal, revision, queued) + + def resume(self, principal: Principal, run_id: str) -> WorkflowExecution: + run = self._store.get_run(principal, run_id) + if run.status is not WorkflowRunStatus.WAITING_APPROVAL: + raise WorkflowRuntimeError("Workflow run is not awaiting approval.") + approvals = self._store.list_approvals(principal, run.run_id) + pending = next((item for item in approvals if item.decision is None), None) + if pending is not None: + return WorkflowExecution(run=run, outputs=None, pending_approval_id=pending.approval_id) + revision = self._store.get_revision(principal, run.workflow_id, run.revision_id) + now = self._now() + queued = self._store.transition_run( + principal, run.run_id, status=WorkflowRunStatus.QUEUED, updated_at=now + ) + return self._run(principal, revision, queued) + + def _run( + self, principal: Principal, revision: WorkflowRevision, run: WorkflowRun + ) -> WorkflowExecution: + now = self._now() + running = self._store.transition_run( + principal, run.run_id, status=WorkflowRunStatus.RUNNING, updated_at=now + ) + state = self._store.get_run_state(principal, running.run_id) + node_outputs = {key: dict(value) for key, value in state.node_outputs.items()} + try: + for node in _execution_order(revision.specification): + if node.node_id in node_outputs: + continue + self._enforce_budget(revision, running, node_outputs) + node_inputs = _resolve_node_inputs(node, state.input_values, node_outputs) + if node.node_kind is WorkflowNodeKind.HUMAN_APPROVAL: + approval = self._approval_for_node(principal, running.run_id, node.node_id) + if approval is None: + now = self._now() + created = self._store.create_approval( + principal, + WorkflowApproval( + approval_id=_new_id("wappr"), run_id=running.run_id, + node_id=node.node_id, requested_at=now, requested_by=principal.subject, + ), + ) + self._save_step(principal, running.run_id, node, node_inputs, + WorkflowStepStatus.WAITING_APPROVAL, now, None, None) + waiting = self._store.transition_run( + principal, running.run_id, status=WorkflowRunStatus.WAITING_APPROVAL, + updated_at=now, + ) + self._save_state(principal, waiting.run_id, state.input_values, node_outputs, now) + return WorkflowExecution(waiting, None, created.approval_id) + if approval.decision is None: + waiting = self._store.transition_run( + principal, running.run_id, status=WorkflowRunStatus.WAITING_APPROVAL, + updated_at=self._now(), + ) + return WorkflowExecution(waiting, None, approval.approval_id) + if approval.decision is ApprovalDecision.REJECTED: + self._save_step(principal, running.run_id, node, node_inputs, + WorkflowStepStatus.FAILED, self._now(), self._now(), None, + error_code="approval_rejected") + failed = self._store.transition_run( + principal, running.run_id, status=WorkflowRunStatus.FAILED, + updated_at=self._now(), error_code="approval_rejected", + ) + return WorkflowExecution(failed, None) + output = {"decision": ApprovalDecision.APPROVED.value} + else: + output = self._execute_node(principal, node, node_inputs) + node_outputs[node.node_id] = output + now = self._now() + self._save_step(principal, running.run_id, node, node_inputs, + WorkflowStepStatus.SUCCEEDED, now, now, output) + self._save_state(principal, running.run_id, state.input_values, node_outputs, now) + outputs = _public_outputs(revision.specification, node_outputs) + succeeded = self._store.transition_run( + principal, running.run_id, status=WorkflowRunStatus.SUCCEEDED, updated_at=self._now() + ) + return WorkflowExecution(succeeded, outputs) + except WorkflowExecutionBudgetError as error: + failed = self._store.transition_run( + principal, running.run_id, status=WorkflowRunStatus.FAILED, + updated_at=self._now(), error_code=error.code, + ) + return WorkflowExecution(failed, None) + except WorkflowRuntimeError as error: + failed = self._store.transition_run( + principal, running.run_id, status=WorkflowRunStatus.FAILED, + updated_at=self._now(), error_code=getattr(error, "code", "node_execution_failed"), + ) + return WorkflowExecution(failed, None) + + def _execute_node( + self, principal: Principal, node: WorkflowNode, values: Mapping[str, Any] + ) -> dict[str, Any]: + executor = self._executors.get(node.node_kind) + if executor is None: + raise WorkflowNodeExecutionError("node_executor_unavailable") + try: + output = executor(principal, node, values) + except WorkflowRuntimeError: + raise + except Exception as error: + raise WorkflowNodeExecutionError() from error + if not isinstance(output, Mapping) or set(output) != set(node.output_names): + raise WorkflowNodeExecutionError("node_output_invalid") + try: + encoded = json.dumps(output, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + decoded = json.loads(encoded) + except (TypeError, ValueError) as error: + raise WorkflowNodeExecutionError("node_output_invalid") from error + if not isinstance(decoded, dict) or len(encoded.encode("utf-8")) > 256 * 1024: + raise WorkflowNodeExecutionError("node_output_invalid") + return decoded + + def _enforce_budget( + self, revision: WorkflowRevision, run: WorkflowRun, outputs: Mapping[str, Mapping[str, Any]] + ) -> None: + if len(outputs) >= revision.budget.max_steps: + raise WorkflowExecutionBudgetError("step_budget_exceeded") + if self._now() - run.created_at > revision.budget.max_wall_seconds: + raise WorkflowExecutionBudgetError("wall_time_budget_exceeded") + model_nodes = {item.node_id for item in revision.specification.nodes if item.node_kind is WorkflowNodeKind.MODEL_GENERATE} + if len(model_nodes & set(outputs)) >= revision.budget.max_model_calls: + raise WorkflowExecutionBudgetError("model_call_budget_exceeded") + + def _approval_for_node( + self, principal: Principal, run_id: str, node_id: str + ) -> WorkflowApproval | None: + items = [item for item in self._store.list_approvals(principal, run_id) if item.node_id == node_id] + if len(items) > 1: + raise WorkflowNodeExecutionError("approval_state_invalid") + return items[0] if items else None + + def _save_step( + self, principal: Principal, run_id: str, node: WorkflowNode, values: Mapping[str, Any], + status: WorkflowStepStatus, started_at: float, finished_at: float | None, + output: Mapping[str, Any] | None, *, error_code: str | None = None, + ) -> None: + self._store.save_step_run( + principal, + WorkflowStepRun( + step_run_id=_step_id(run_id, node.node_id), run_id=run_id, node_id=node.node_id, + status=status, started_at=started_at, finished_at=finished_at, + input_digest=_digest(values), output_digest=_digest(output) if output is not None else None, + error_code=error_code, + ), + ) + + def _save_state(self, principal: Principal, run_id: str, inputs: Mapping[str, Any], + outputs: Mapping[str, Mapping[str, Any]], updated_at: float) -> None: + self._store.save_run_state( + principal, + WorkflowRunState(run_id=run_id, input_values=inputs, node_outputs=outputs, updated_at=updated_at), + ) + + def _now(self) -> float: + value = self._clock() + if isinstance(value, bool) or not isinstance(value, (int, float)) or value < 0: + raise RuntimeError("clock returned an invalid timestamp") + return float(value) + + +def _execution_order(specification: WorkflowSpec) -> tuple[WorkflowNode, ...]: + nodes = {node.node_id: node for node in specification.nodes} + remaining = {node_id: set(node.depends_on) for node_id, node in nodes.items()} + ordered: list[WorkflowNode] = [] + while remaining: + ready = sorted(node_id for node_id, dependencies in remaining.items() if not dependencies) + if not ready: + raise WorkflowRuntimeError("Workflow graph is invalid.") + for node_id in ready: + ordered.append(nodes[node_id]) + del remaining[node_id] + completed = set(ready) + for dependencies in remaining.values(): + dependencies.difference_update(completed) + return tuple(ordered) + + +def _validate_inputs(specification: WorkflowSpec, values: Mapping[str, Any]) -> dict[str, Any]: + if not isinstance(values, Mapping): + raise WorkflowRuntimeError("Workflow inputs must be an object.") + expected = {item.name: item for item in specification.inputs} + if not set(values) <= set(expected): + raise WorkflowRuntimeError("Workflow inputs contain an unknown field.") + if any(item.required and item.name not in values for item in expected.values()): + raise WorkflowRuntimeError("Workflow inputs are incomplete.") + try: + encoded = json.dumps(values, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + decoded = json.loads(encoded) + except (TypeError, ValueError) as error: + raise WorkflowRuntimeError("Workflow inputs must be JSON-compatible.") from error + if not isinstance(decoded, dict) or len(encoded.encode("utf-8")) > 256 * 1024: + raise WorkflowRuntimeError("Workflow inputs are too large.") + return decoded + + +def _resolve_node_inputs(node: WorkflowNode, inputs: Mapping[str, Any], + outputs: Mapping[str, Mapping[str, Any]]) -> dict[str, Any]: + resolved: dict[str, Any] = {} + for binding in node.input_bindings: + parts = binding.source.split(".") + if parts[0] == "input": + resolved[binding.target] = inputs[parts[1]] + else: + resolved[binding.target] = outputs[parts[1]][parts[2]] + return resolved + + +def _public_outputs(specification: WorkflowSpec, outputs: Mapping[str, Mapping[str, Any]]) -> dict[str, Any]: + return {output.name: outputs[output.source.split(".")[1]][output.source.split(".")[2]] for output in specification.outputs} + + +def _digest(value: Mapping[str, Any]) -> str: + encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def _new_id(prefix: str) -> str: + return f"{prefix}_{secrets.token_hex(16)}" + + +def _step_id(run_id: str, node_id: str) -> str: + """Stable per-run ID lets resume update the same node audit record.""" + + return "wstep_" + hashlib.sha256(f"{run_id}:{node_id}".encode()).hexdigest()[:32] + + +__all__ = [ + "NativeWorkflowNodeExecutor", + "WorkflowExecution", + "WorkflowExecutionBudgetError", + "WorkflowNodeExecutionError", + "WorkflowNotPublishedError", + "WorkflowRuntime", + "WorkflowRuntimeError", +] diff --git a/rag_system/workflow_service.py b/rag_system/workflow_service.py new file mode 100644 index 0000000..a9204f3 --- /dev/null +++ b/rag_system/workflow_service.py @@ -0,0 +1,228 @@ +"""Authorised lifecycle commands for versioned, deployable workflows.""" + +from __future__ import annotations + +import secrets +import time +from collections.abc import Callable, Sequence +from dataclasses import dataclass + +from rag_system.application_contracts import is_valid_timestamp +from rag_system.application_ports import ApplicationRepository, KnowledgeBaseRepository +from rag_system.knowledge_base_contracts import KnowledgeBaseStatus +from rag_system.tenancy import Principal +from rag_system.workflow_contracts import WorkflowResourceKind, WorkflowSpec +from rag_system.workflow_models import ( + ApprovalDecision, + ExecutionBudget, + Workflow, + WorkflowDeployment, + WorkflowDraft, + WorkflowEvaluation, + WorkflowRevision, + WorkflowStatus, +) +from rag_system.workflow_store import WorkflowStore + + +class WorkflowServiceError(Exception): + """Base class for public workflow-management failures.""" + + +class WorkflowAuthorizationError(WorkflowServiceError): + def __init__(self) -> None: + super().__init__("The operation is not permitted.") + + +class WorkflowServiceValidationError(WorkflowServiceError, ValueError): + """A requested workflow change violates its lifecycle contract.""" + + +class WorkflowResourceUnavailableError(WorkflowServiceError): + def __init__(self) -> None: + super().__init__("A required workflow resource is unavailable.") + + +@dataclass(frozen=True, slots=True) +class PublishedWorkflow: + workflow: Workflow + deployment: WorkflowDeployment + + +class WorkflowService: + """Keep mutable drafts, immutable revisions, checks, and publishing separate.""" + + def __init__( + self, + store: WorkflowStore, + projects: ApplicationRepository, + knowledge_bases: KnowledgeBaseRepository, + *, + clock: Callable[[], float] = time.time, + trusted_model_profile_ids: Sequence[str] = ("default",), + ) -> None: + if not callable(clock): + raise TypeError("clock must be callable") + profiles = frozenset(trusted_model_profile_ids) + if not profiles or any(not isinstance(value, str) or not value for value in profiles): + raise ValueError("trusted_model_profile_ids are invalid") + self._store = store + self._projects = projects + self._knowledge_bases = knowledge_bases + self._clock = clock + self._trusted_model_profile_ids = profiles + + def create_workflow(self, principal: Principal, project_id: str, display_name: str) -> Workflow: + _require_writer(principal) + # A workflow cannot become an orphaned tenant resource. + self._projects.get_project(principal, project_id) + now = self._now() + return self._store.create_workflow( + principal, + Workflow( + workflow_id=_new_id("wf"), tenant_id=principal.tenant_id, project_id=project_id, + display_name=display_name, active_revision_id=None, status=WorkflowStatus.ACTIVE, + created_at=now, updated_at=now, + ), + ) + + def get_draft(self, principal: Principal, workflow_id: str) -> WorkflowDraft: + _require_writer(principal) + return self._store.get_draft(principal, workflow_id) + + def update_draft( + self, principal: Principal, workflow_id: str, specification: WorkflowSpec, budget: ExecutionBudget, + *, expected_version: int, change_summary: str, + ) -> WorkflowDraft: + _require_writer(principal) + if isinstance(expected_version, bool) or not isinstance(expected_version, int) or expected_version < 0: + raise WorkflowServiceValidationError("expected draft version is invalid") + workflow = self._store.get_workflow(principal, workflow_id) + if workflow.status is WorkflowStatus.ARCHIVED: + raise WorkflowServiceValidationError("Archived workflows cannot update drafts.") + self._verify_resources(principal, specification) + return self._store.update_draft( + principal, + WorkflowDraft( + workflow_id=workflow.workflow_id, version=expected_version + 1, + specification=specification, budget=budget, updated_at=self._now(), + updated_by=principal.subject, change_summary=change_summary, + ), + expected_version=expected_version, + ) + + def create_revision_from_draft( + self, principal: Principal, workflow_id: str, *, expected_version: int + ) -> WorkflowRevision: + _require_writer(principal) + draft = self._store.get_draft(principal, workflow_id) + if draft.version != expected_version: + raise WorkflowServiceValidationError("Workflow draft has changed.") + if draft.specification is None or draft.budget is None: + raise WorkflowServiceValidationError("Workflow draft has not been configured.") + return self.create_revision( + principal, workflow_id, draft.specification, draft.budget, + change_summary=draft.change_summary, + ) + + def create_revision( + self, principal: Principal, workflow_id: str, specification: WorkflowSpec, + budget: ExecutionBudget, *, change_summary: str) -> WorkflowRevision: + _require_writer(principal) + workflow = self._store.get_workflow(principal, workflow_id) + if workflow.status is WorkflowStatus.ARCHIVED: + raise WorkflowServiceValidationError("Archived workflows cannot accept revisions.") + self._verify_resources(principal, specification) + revisions = self._store.list_revisions(principal, workflow.workflow_id, limit=100) + return self._store.create_revision( + principal, + WorkflowRevision( + revision_id=_new_id("wfr"), workflow_id=workflow.workflow_id, + revision_number=max((item.revision_number for item in revisions), default=0) + 1, + specification=specification, budget=budget, created_at=self._now(), + created_by=principal.subject, change_summary=change_summary, + ), + ) + + def record_evaluation(self, principal: Principal, evaluation: WorkflowEvaluation) -> WorkflowEvaluation: + _require_writer(principal) + revision = self._store.get_revision(principal, evaluation.workflow_id, evaluation.revision_id) + if revision.specification_digest != evaluation.specification_digest: + raise WorkflowServiceValidationError("Evaluation does not match the immutable revision.") + return self._store.save_evaluation(principal, evaluation) + + def publish( + self, principal: Principal, workflow_id: str, revision_id: str, *, + expected_active_revision_id: str | None, + ) -> PublishedWorkflow: + _require_operator(principal) + workflow = self._store.get_workflow(principal, workflow_id) + if workflow.status is WorkflowStatus.ARCHIVED: + raise WorkflowServiceValidationError("Archived workflows cannot be published.") + revision = self._store.get_revision(principal, workflow.workflow_id, revision_id) + self._verify_resources(principal, revision.specification) + evaluations = self._store.list_evaluations(principal, workflow.workflow_id, revision.revision_id) + if not any(item.passed for item in evaluations): + raise WorkflowServiceValidationError("A passing evaluation is required before publication.") + now = self._now() + deployment = WorkflowDeployment( + deployment_id=_new_id("wfd"), workflow_id=workflow.workflow_id, + revision_id=revision.revision_id, deployed_at=now, deployed_by=principal.subject, + ) + return PublishedWorkflow( + workflow=self._store.publish( + principal, deployment, updated_at=now, + expected_active_revision_id=expected_active_revision_id, + ), + deployment=deployment, + ) + + def decide_approval( + self, principal: Principal, approval_id: str, decision: ApprovalDecision + ) -> None: + _require_operator(principal) + self._store.decide_approval(principal, approval_id, decision=decision, decided_at=self._now()) + + def _verify_resources(self, principal: Principal, specification: WorkflowSpec) -> None: + for node in specification.nodes: + for resource in node.resources: + if resource.resource_kind is WorkflowResourceKind.MODEL_PROFILE: + if resource.resource_id not in self._trusted_model_profile_ids: + raise WorkflowServiceValidationError("The model profile is unavailable.") + continue + try: + record = self._knowledge_bases.get(principal, resource.resource_id) + except Exception as error: + raise WorkflowResourceUnavailableError() from error + if record.status is not KnowledgeBaseStatus.READY: + raise WorkflowResourceUnavailableError() + + def _now(self) -> float: + value = float(self._clock()) + if not is_valid_timestamp(value): + raise WorkflowServiceValidationError("clock returned an invalid timestamp") + return value + + +def _new_id(prefix: str) -> str: + return f"{prefix}_{secrets.token_hex(16)}" + + +def _require_writer(principal: Principal) -> None: + if not isinstance(principal, Principal) or not principal.has_role("writer"): + raise WorkflowAuthorizationError() + + +def _require_operator(principal: Principal) -> None: + if not isinstance(principal, Principal) or not principal.has_role("operator"): + raise WorkflowAuthorizationError() + + +__all__ = [ + "PublishedWorkflow", + "WorkflowAuthorizationError", + "WorkflowResourceUnavailableError", + "WorkflowService", + "WorkflowServiceError", + "WorkflowServiceValidationError", +] diff --git a/rag_system/workflow_store.py b/rag_system/workflow_store.py new file mode 100644 index 0000000..ea550f1 --- /dev/null +++ b/rag_system/workflow_store.py @@ -0,0 +1,1113 @@ +"""Durable, tenant-scoped SQLite storage for versioned workflow resources.""" + +from __future__ import annotations + +import json +import sqlite3 +from contextlib import AbstractContextManager +from pathlib import Path +from threading import RLock +from typing import Any, cast + +from rag_system.sqlite_support import SqliteDatabase +from rag_system.tenancy import Principal, TenantId +from rag_system.workflow_models import ( + ApprovalDecision, + ExecutionBudget, + Workflow, + WorkflowApproval, + WorkflowDeployment, + WorkflowDeploymentStatus, + WorkflowDraft, + WorkflowEvaluation, + WorkflowModelError, + WorkflowRevision, + WorkflowRun, + WorkflowRunState, + WorkflowRunStatus, + WorkflowStatus, + WorkflowStepRun, + WorkflowStepStatus, + validate_workflow_id, + validate_workflow_revision_id, + validate_workflow_run_id, +) +from rag_system.workflow_contracts import WorkflowSpec, WorkflowValidationError + + +_SCHEMA_VERSION = 3 +_MAX_LIST_LIMIT = 100 + + +class WorkflowStoreError(WorkflowModelError): + """Base class for workflow-store failures.""" + + +class WorkflowStoreSchemaError(WorkflowStoreError): + def __init__(self) -> None: + super().__init__("Workflow store schema or stored data is invalid.") + + +class WorkflowStoreStorageError(WorkflowStoreError): + def __init__(self) -> None: + super().__init__("Workflow store operation failed.") + + +class WorkflowUnavailableError(WorkflowStoreError): + def __init__(self) -> None: + super().__init__("Workflow is unavailable.") + + +class WorkflowRevisionUnavailableError(WorkflowStoreError): + def __init__(self) -> None: + super().__init__("Workflow revision is unavailable.") + + +class WorkflowRunUnavailableError(WorkflowStoreError): + def __init__(self) -> None: + super().__init__("Workflow run is unavailable.") + + +class WorkflowDraftConflictError(WorkflowStoreError): + def __init__(self) -> None: + super().__init__("Workflow draft has changed.") + + +class WorkflowPublishConflictError(WorkflowStoreError): + def __init__(self) -> None: + super().__init__("Workflow publication has changed.") + + +class WorkflowApprovalUnavailableError(WorkflowStoreError): + def __init__(self) -> None: + super().__init__("Workflow approval is unavailable.") + + +class WorkflowStore: + """Use short transactions and tenant filters for all workflow state.""" + + def __init__(self, database_path: str | Path, *, timeout_seconds: float = 5.0) -> None: + if isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, (int, float)): + raise WorkflowModelError("workflow store timeout is invalid") + if not 0 < timeout_seconds <= 60: + raise WorkflowModelError("workflow store timeout is invalid") + path = Path(database_path) + if path.exists() and not path.is_file(): + raise WorkflowModelError("workflow database path must reference a file") + path.parent.mkdir(parents=True, exist_ok=True) + self._database_path = path.resolve() + self._database = SqliteDatabase(self._database_path, timeout_seconds=float(timeout_seconds)) + self._write_lock = RLock() + self._initialize() + + @property + def database_path(self) -> Path: + return self._database_path + + def create_workflow(self, principal: Principal, workflow: Workflow) -> Workflow: + tenant = _tenant(principal) + _same_tenant(tenant, workflow.tenant_id) + with self._write_lock, self._write() as connection: + try: + connection.execute( + """INSERT INTO workflows ( + workflow_id, tenant_id, project_id, display_name, active_revision_id, + status, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + ( + workflow.workflow_id, + tenant.value, + workflow.project_id, + workflow.display_name, + workflow.active_revision_id, + workflow.status.value, + workflow.created_at, + workflow.updated_at, + ), + ) + connection.execute( + """INSERT INTO workflow_drafts ( + workflow_id, version, specification_json, budget_json, updated_at, updated_by, change_summary + ) VALUES (?, 0, NULL, NULL, ?, ?, '')""", + (workflow.workflow_id, workflow.updated_at, principal.subject), + ) + except sqlite3.IntegrityError as error: + raise WorkflowStoreStorageError() from error + return workflow + + def get_workflow(self, principal: Principal, workflow_id: str) -> Workflow: + tenant = _tenant(principal) + clean_id = _safe_workflow_id(workflow_id) + with self._read() as connection: + row = connection.execute( + "SELECT * FROM workflows WHERE workflow_id = ? AND tenant_id = ?", (clean_id, tenant.value) + ).fetchone() + if row is None: + raise WorkflowUnavailableError() + return _workflow_from_row(row) + + def list_workflows( + self, principal: Principal, project_id: str, *, limit: int = 50 + ) -> tuple[Workflow, ...]: + tenant = _tenant(principal) + clean_limit = _limit(limit) + with self._read() as connection: + rows = connection.execute( + """SELECT * FROM workflows WHERE tenant_id = ? AND project_id = ? + ORDER BY updated_at DESC, workflow_id DESC LIMIT ?""", + (tenant.value, project_id, clean_limit), + ).fetchall() + return tuple(_workflow_from_row(row) for row in rows) + + def archive_workflow(self, principal: Principal, workflow_id: str, *, updated_at: float) -> Workflow: + tenant = _tenant(principal) + clean_id = _safe_workflow_id(workflow_id) + with self._write_lock, self._write() as connection: + row = _require_workflow(connection, tenant, clean_id) + workflow = _workflow_from_row(row) + if workflow.status is WorkflowStatus.ARCHIVED: + return workflow + connection.execute( + "UPDATE workflows SET status = ?, updated_at = ? WHERE workflow_id = ? AND tenant_id = ?", + (WorkflowStatus.ARCHIVED.value, updated_at, clean_id, tenant.value), + ) + return Workflow( + workflow_id=workflow.workflow_id, + tenant_id=workflow.tenant_id, + project_id=workflow.project_id, + display_name=workflow.display_name, + active_revision_id=workflow.active_revision_id, + status=WorkflowStatus.ARCHIVED, + created_at=workflow.created_at, + updated_at=updated_at, + ) + + def get_draft(self, principal: Principal, workflow_id: str) -> WorkflowDraft: + tenant = _tenant(principal) + clean_id = _safe_workflow_id(workflow_id) + with self._read() as connection: + _require_workflow(connection, tenant, clean_id) + row = connection.execute( + "SELECT * FROM workflow_drafts WHERE workflow_id = ?", (clean_id,) + ).fetchone() + if row is None: + raise WorkflowStoreSchemaError() + return _draft_from_row(row) + + def update_draft( + self, + principal: Principal, + draft: WorkflowDraft, + *, + expected_version: int, + ) -> WorkflowDraft: + tenant = _tenant(principal) + if not isinstance(expected_version, int) or isinstance(expected_version, bool) or expected_version < 0: + raise WorkflowModelError("workflow draft expected version is invalid") + with self._write_lock, self._write() as connection: + _require_workflow(connection, tenant, draft.workflow_id) + current = connection.execute( + "SELECT version FROM workflow_drafts WHERE workflow_id = ?", (draft.workflow_id,) + ).fetchone() + if current is None: + raise WorkflowStoreSchemaError() + if int(current["version"]) != expected_version or draft.version != expected_version + 1: + raise WorkflowDraftConflictError() + connection.execute( + """UPDATE workflow_drafts SET version = ?, specification_json = ?, budget_json = ?, + updated_at = ?, updated_by = ?, change_summary = ? WHERE workflow_id = ?""", + ( + draft.version, + _specification_json(draft.specification), + _budget_json(draft.budget), + draft.updated_at, + draft.updated_by, + draft.change_summary, + draft.workflow_id, + ), + ) + return draft + + def create_revision(self, principal: Principal, revision: WorkflowRevision) -> WorkflowRevision: + tenant = _tenant(principal) + with self._write_lock, self._write() as connection: + workflow = _workflow_from_row(_require_workflow(connection, tenant, revision.workflow_id)) + if workflow.status is WorkflowStatus.ARCHIVED: + raise WorkflowUnavailableError() + row = connection.execute( + "SELECT COALESCE(MAX(revision_number), 0) AS current_number FROM workflow_revisions WHERE workflow_id = ?", + (revision.workflow_id,), + ).fetchone() + if row is None or revision.revision_number != int(row["current_number"]) + 1: + raise WorkflowStoreStorageError() + try: + connection.execute( + """INSERT INTO workflow_revisions ( + revision_id, workflow_id, revision_number, specification_json, specification_digest, + budget_json, created_at, created_by, change_summary + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + revision.revision_id, + revision.workflow_id, + revision.revision_number, + revision.specification.to_json(), + revision.specification_digest, + _budget_json(revision.budget), + revision.created_at, + revision.created_by, + revision.change_summary, + ), + ) + except sqlite3.IntegrityError as error: + raise WorkflowStoreStorageError() from error + return revision + + def get_revision( + self, principal: Principal, workflow_id: str, revision_id: str + ) -> WorkflowRevision: + tenant = _tenant(principal) + clean_workflow_id = _safe_workflow_id(workflow_id) + clean_revision_id = _safe_revision_id(revision_id) + with self._read() as connection: + _require_workflow(connection, tenant, clean_workflow_id) + row = connection.execute( + "SELECT * FROM workflow_revisions WHERE workflow_id = ? AND revision_id = ?", + (clean_workflow_id, clean_revision_id), + ).fetchone() + if row is None: + raise WorkflowRevisionUnavailableError() + return _revision_from_row(row) + + def list_revisions( + self, principal: Principal, workflow_id: str, *, limit: int = 50 + ) -> tuple[WorkflowRevision, ...]: + tenant = _tenant(principal) + clean_id = _safe_workflow_id(workflow_id) + with self._read() as connection: + _require_workflow(connection, tenant, clean_id) + rows = connection.execute( + """SELECT * FROM workflow_revisions WHERE workflow_id = ? + ORDER BY revision_number DESC LIMIT ?""", + (clean_id, _limit(limit)), + ).fetchall() + return tuple(_revision_from_row(row) for row in rows) + + def publish( + self, + principal: Principal, + deployment: WorkflowDeployment, + *, + updated_at: float, + expected_active_revision_id: str | None, + ) -> Workflow: + tenant = _tenant(principal) + with self._write_lock, self._write() as connection: + row = _require_workflow(connection, tenant, deployment.workflow_id) + workflow = _workflow_from_row(row) + if workflow.status is WorkflowStatus.ARCHIVED: + raise WorkflowUnavailableError() + if workflow.active_revision_id != expected_active_revision_id: + raise WorkflowPublishConflictError() + _require_revision(connection, deployment.workflow_id, deployment.revision_id) + connection.execute( + """UPDATE workflow_deployments SET status = ? + WHERE workflow_id = ? AND status = ?""", + ( + WorkflowDeploymentStatus.SUPERSEDED.value, + deployment.workflow_id, + WorkflowDeploymentStatus.ACTIVE.value, + ), + ) + try: + connection.execute( + """INSERT INTO workflow_deployments ( + deployment_id, workflow_id, revision_id, deployed_at, deployed_by, status + ) VALUES (?, ?, ?, ?, ?, ?)""", + ( + deployment.deployment_id, + deployment.workflow_id, + deployment.revision_id, + deployment.deployed_at, + deployment.deployed_by, + WorkflowDeploymentStatus.ACTIVE.value, + ), + ) + except sqlite3.IntegrityError as error: + raise WorkflowStoreStorageError() from error + connection.execute( + """UPDATE workflows SET active_revision_id = ?, updated_at = ? + WHERE workflow_id = ? AND tenant_id = ?""", + (deployment.revision_id, updated_at, deployment.workflow_id, tenant.value), + ) + return Workflow( + workflow_id=workflow.workflow_id, + tenant_id=workflow.tenant_id, + project_id=workflow.project_id, + display_name=workflow.display_name, + active_revision_id=deployment.revision_id, + status=workflow.status, + created_at=workflow.created_at, + updated_at=updated_at, + ) + + def create_run(self, principal: Principal, run: WorkflowRun) -> WorkflowRun: + tenant = _tenant(principal) + with self._write_lock, self._write() as connection: + _require_workflow(connection, tenant, run.workflow_id) + revision = _revision_from_row(_require_revision(connection, run.workflow_id, run.revision_id)) + if revision.specification_digest != run.specification_digest: + raise WorkflowStoreStorageError() + try: + connection.execute( + """INSERT INTO workflow_runs ( + run_id, workflow_id, revision_id, specification_digest, status, created_at, + updated_at, created_by, input_digest, error_code + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + run.run_id, + run.workflow_id, + run.revision_id, + run.specification_digest, + run.status.value, + run.created_at, + run.updated_at, + run.created_by, + run.input_digest, + run.error_code, + ), + ) + except sqlite3.IntegrityError as error: + raise WorkflowStoreStorageError() from error + return run + + def get_run(self, principal: Principal, run_id: str) -> WorkflowRun: + tenant = _tenant(principal) + clean_id = _safe_run_id(run_id) + with self._read() as connection: + row = connection.execute( + """SELECT runs.* FROM workflow_runs AS runs + JOIN workflows AS workflows ON workflows.workflow_id = runs.workflow_id + WHERE runs.run_id = ? AND workflows.tenant_id = ?""", + (clean_id, tenant.value), + ).fetchone() + if row is None: + raise WorkflowRunUnavailableError() + return _run_from_row(row) + + def transition_run( + self, + principal: Principal, + run_id: str, + *, + status: WorkflowRunStatus, + updated_at: float, + error_code: str | None = None, + ) -> WorkflowRun: + if not isinstance(status, WorkflowRunStatus): + raise WorkflowModelError("workflow run status is invalid") + current = self.get_run(principal, run_id) + _validate_run_transition(current.status, status) + tenant = _tenant(principal) + with self._write_lock, self._write() as connection: + changed = connection.execute( + """UPDATE workflow_runs SET status = ?, updated_at = ?, error_code = ? + WHERE run_id = ? AND workflow_id IN ( + SELECT workflow_id FROM workflows WHERE tenant_id = ? + )""", + (status.value, updated_at, error_code, current.run_id, tenant.value), + ).rowcount + if changed != 1: + raise WorkflowRunUnavailableError() + return WorkflowRun( + run_id=current.run_id, + workflow_id=current.workflow_id, + revision_id=current.revision_id, + specification_digest=current.specification_digest, + status=status, + created_at=current.created_at, + updated_at=updated_at, + created_by=current.created_by, + input_digest=current.input_digest, + error_code=error_code, + ) + + def save_step_run(self, principal: Principal, step: WorkflowStepRun) -> WorkflowStepRun: + self.get_run(principal, step.run_id) + with self._write_lock, self._write() as connection: + try: + connection.execute( + """INSERT INTO workflow_step_runs ( + step_run_id, run_id, node_id, status, started_at, finished_at, + input_digest, output_digest, error_code + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(step_run_id) DO UPDATE SET + status = excluded.status, started_at = excluded.started_at, + finished_at = excluded.finished_at, input_digest = excluded.input_digest, + output_digest = excluded.output_digest, error_code = excluded.error_code""", + ( + step.step_run_id, + step.run_id, + step.node_id, + step.status.value, + step.started_at, + step.finished_at, + step.input_digest, + step.output_digest, + step.error_code, + ), + ) + except sqlite3.IntegrityError as error: + raise WorkflowStoreStorageError() from error + return step + + def list_step_runs(self, principal: Principal, run_id: str) -> tuple[WorkflowStepRun, ...]: + self.get_run(principal, run_id) + with self._read() as connection: + rows = connection.execute( + "SELECT * FROM workflow_step_runs WHERE run_id = ? ORDER BY step_run_id", (run_id,) + ).fetchall() + return tuple(_step_from_row(row) for row in rows) + + def save_run_state(self, principal: Principal, state: WorkflowRunState) -> WorkflowRunState: + """Persist bounded runtime values needed to resume an approved run.""" + + self.get_run(principal, state.run_id) + with self._write_lock, self._write() as connection: + try: + connection.execute( + """INSERT INTO workflow_run_state (run_id, input_json, outputs_json, updated_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(run_id) DO UPDATE SET input_json = excluded.input_json, + outputs_json = excluded.outputs_json, updated_at = excluded.updated_at""", + (state.run_id, state.input_json, state.outputs_json, state.updated_at), + ) + except sqlite3.IntegrityError as error: + raise WorkflowStoreStorageError() from error + return state + + def get_run_state(self, principal: Principal, run_id: str) -> WorkflowRunState: + run = self.get_run(principal, run_id) + with self._read() as connection: + row = connection.execute( + "SELECT * FROM workflow_run_state WHERE run_id = ?", (run.run_id,) + ).fetchone() + if row is None: + raise WorkflowRunUnavailableError() + return _run_state_from_row(row) + + def save_evaluation(self, principal: Principal, evaluation: WorkflowEvaluation) -> WorkflowEvaluation: + tenant = _tenant(principal) + with self._write_lock, self._write() as connection: + _require_workflow(connection, tenant, evaluation.workflow_id) + revision = _revision_from_row( + _require_revision(connection, evaluation.workflow_id, evaluation.revision_id) + ) + if revision.specification_digest != evaluation.specification_digest: + raise WorkflowStoreStorageError() + try: + connection.execute( + """INSERT INTO workflow_evaluations ( + evaluation_id, workflow_id, revision_id, specification_digest, generated_at, + case_count, passed_case_count + ) VALUES (?, ?, ?, ?, ?, ?, ?)""", + ( + evaluation.evaluation_id, evaluation.workflow_id, evaluation.revision_id, + evaluation.specification_digest, evaluation.generated_at, evaluation.case_count, + evaluation.passed_case_count, + ), + ) + except sqlite3.IntegrityError as error: + raise WorkflowStoreStorageError() from error + return evaluation + + def list_evaluations( + self, principal: Principal, workflow_id: str, revision_id: str, *, limit: int = 50 + ) -> tuple[WorkflowEvaluation, ...]: + tenant = _tenant(principal) + clean_workflow_id = _safe_workflow_id(workflow_id) + clean_revision_id = _safe_revision_id(revision_id) + with self._read() as connection: + _require_workflow(connection, tenant, clean_workflow_id) + _require_revision(connection, clean_workflow_id, clean_revision_id) + rows = connection.execute( + """SELECT * FROM workflow_evaluations WHERE workflow_id = ? AND revision_id = ? + ORDER BY generated_at DESC, evaluation_id DESC LIMIT ?""", + (clean_workflow_id, clean_revision_id, _limit(limit)), + ).fetchall() + return tuple(_evaluation_from_row(row) for row in rows) + + def create_approval(self, principal: Principal, approval: WorkflowApproval) -> WorkflowApproval: + self.get_run(principal, approval.run_id) + with self._write_lock, self._write() as connection: + try: + connection.execute( + """INSERT INTO workflow_approvals ( + approval_id, run_id, node_id, requested_at, requested_by, decision, decided_at, decided_by + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + ( + approval.approval_id, + approval.run_id, + approval.node_id, + approval.requested_at, + approval.requested_by, + None, + None, + None, + ), + ) + except sqlite3.IntegrityError as error: + raise WorkflowStoreStorageError() from error + return approval + + def get_approval(self, principal: Principal, approval_id: str) -> WorkflowApproval: + tenant = _tenant(principal) + with self._read() as connection: + row = connection.execute( + """SELECT approvals.* FROM workflow_approvals AS approvals + JOIN workflow_runs AS runs ON runs.run_id = approvals.run_id + JOIN workflows AS workflows ON workflows.workflow_id = runs.workflow_id + WHERE approvals.approval_id = ? AND workflows.tenant_id = ?""", + (approval_id, tenant.value), + ).fetchone() + if row is None: + raise WorkflowApprovalUnavailableError() + return _approval_from_row(row) + + def list_approvals(self, principal: Principal, run_id: str) -> tuple[WorkflowApproval, ...]: + run = self.get_run(principal, run_id) + with self._read() as connection: + rows = connection.execute( + "SELECT * FROM workflow_approvals WHERE run_id = ? ORDER BY requested_at, approval_id", + (run.run_id,), + ).fetchall() + return tuple(_approval_from_row(row) for row in rows) + + def decide_approval( + self, + principal: Principal, + approval_id: str, + *, + decision: ApprovalDecision, + decided_at: float, + ) -> WorkflowApproval: + if not isinstance(decision, ApprovalDecision): + raise WorkflowModelError("workflow approval decision is invalid") + tenant = _tenant(principal) + with self._write_lock, self._write() as connection: + row = connection.execute( + """SELECT approvals.* FROM workflow_approvals AS approvals + JOIN workflow_runs AS runs ON runs.run_id = approvals.run_id + JOIN workflows AS workflows ON workflows.workflow_id = runs.workflow_id + WHERE approvals.approval_id = ? AND workflows.tenant_id = ?""", + (approval_id, tenant.value), + ).fetchone() + if row is None or row["decision"] is not None: + raise WorkflowApprovalUnavailableError() + changed = connection.execute( + """UPDATE workflow_approvals SET decision = ?, decided_at = ?, decided_by = ? + WHERE approval_id = ? AND decision IS NULL""", + (decision.value, decided_at, principal.subject, approval_id), + ).rowcount + if changed != 1: + raise WorkflowApprovalUnavailableError() + return WorkflowApproval( + approval_id=str(row["approval_id"]), + run_id=str(row["run_id"]), + node_id=str(row["node_id"]), + requested_at=float(row["requested_at"]), + requested_by=str(row["requested_by"]), + decision=decision, + decided_at=decided_at, + decided_by=principal.subject, + ) + + def recover_interrupted_runs(self, principal: Principal, *, updated_at: float) -> int: + tenant = _tenant(principal) + with self._write_lock, self._write() as connection: + rows = connection.execute( + """SELECT runs.run_id FROM workflow_runs AS runs + JOIN workflows AS workflows ON workflows.workflow_id = runs.workflow_id + WHERE workflows.tenant_id = ? AND runs.status = ?""", + (tenant.value, WorkflowRunStatus.RUNNING.value), + ).fetchall() + run_ids = tuple(str(row["run_id"]) for row in rows) + if not run_ids: + return 0 + placeholders = ",".join("?" for _ in run_ids) + connection.execute( + f"UPDATE workflow_runs SET status = ?, updated_at = ?, error_code = ? WHERE run_id IN ({placeholders})", + (WorkflowRunStatus.INTERRUPTED.value, updated_at, "runtime_interrupted", *run_ids), + ) + connection.execute( + f"""UPDATE workflow_step_runs SET status = ?, finished_at = ?, error_code = ? + WHERE run_id IN ({placeholders}) AND status = ?""", + ( + WorkflowStepStatus.INTERRUPTED.value, + updated_at, + "runtime_interrupted", + *run_ids, + WorkflowStepStatus.RUNNING.value, + ), + ) + return len(run_ids) + + def _initialize(self) -> None: + with self._write_lock, self._write() as connection: + version = int(connection.execute("PRAGMA user_version").fetchone()[0]) + if version == 0: + _create_schema(connection) + connection.execute(f"PRAGMA user_version = {_SCHEMA_VERSION}") + elif version == 1: + _migrate_v1_to_v2(connection) + _migrate_v2_to_v3(connection) + connection.execute(f"PRAGMA user_version = {_SCHEMA_VERSION}") + elif version == 2: + _migrate_v2_to_v3(connection) + connection.execute(f"PRAGMA user_version = {_SCHEMA_VERSION}") + elif version != _SCHEMA_VERSION: + raise WorkflowStoreSchemaError() + _validate_schema(connection) + + def _read(self) -> AbstractContextManager[sqlite3.Connection]: + return cast( + AbstractContextManager[sqlite3.Connection], self._database.read(WorkflowStoreStorageError) + ) + + def _write(self) -> AbstractContextManager[sqlite3.Connection]: + return cast( + AbstractContextManager[sqlite3.Connection], + self._database.immediate_transaction( + WorkflowStoreStorageError, + pass_through=( + WorkflowStoreError, + WorkflowModelError, + WorkflowValidationError, + ), + ), + ) + + +def _create_schema(connection: sqlite3.Connection) -> None: + connection.executescript( + """ + CREATE TABLE workflows ( + workflow_id TEXT PRIMARY KEY, + tenant_id TEXT NOT NULL, + project_id TEXT NOT NULL, + display_name TEXT NOT NULL, + active_revision_id TEXT, + status TEXT NOT NULL, + created_at REAL NOT NULL, + updated_at REAL NOT NULL + ); + CREATE INDEX workflows_tenant_project_updated + ON workflows (tenant_id, project_id, updated_at DESC, workflow_id DESC); + CREATE TABLE workflow_drafts ( + workflow_id TEXT PRIMARY KEY REFERENCES workflows(workflow_id) ON DELETE RESTRICT, + version INTEGER NOT NULL, + specification_json TEXT, + budget_json TEXT, + updated_at REAL NOT NULL, + updated_by TEXT NOT NULL, + change_summary TEXT NOT NULL + ); + CREATE TABLE workflow_revisions ( + revision_id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL REFERENCES workflows(workflow_id) ON DELETE RESTRICT, + revision_number INTEGER NOT NULL, + specification_json TEXT NOT NULL, + specification_digest TEXT NOT NULL, + budget_json TEXT NOT NULL, + created_at REAL NOT NULL, + created_by TEXT NOT NULL, + change_summary TEXT NOT NULL, + UNIQUE (workflow_id, revision_number) + ); + CREATE TABLE workflow_deployments ( + deployment_id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL REFERENCES workflows(workflow_id) ON DELETE RESTRICT, + revision_id TEXT NOT NULL REFERENCES workflow_revisions(revision_id) ON DELETE RESTRICT, + deployed_at REAL NOT NULL, + deployed_by TEXT NOT NULL, + status TEXT NOT NULL + ); + CREATE UNIQUE INDEX workflow_deployments_one_active + ON workflow_deployments (workflow_id) WHERE status = 'active'; + CREATE TABLE workflow_runs ( + run_id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL REFERENCES workflows(workflow_id) ON DELETE RESTRICT, + revision_id TEXT NOT NULL REFERENCES workflow_revisions(revision_id) ON DELETE RESTRICT, + specification_digest TEXT NOT NULL, + status TEXT NOT NULL, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + created_by TEXT NOT NULL, + input_digest TEXT NOT NULL, + error_code TEXT + ); + CREATE INDEX workflow_runs_workflow_updated + ON workflow_runs (workflow_id, updated_at DESC, run_id DESC); + CREATE TABLE workflow_step_runs ( + step_run_id TEXT PRIMARY KEY, + run_id TEXT NOT NULL REFERENCES workflow_runs(run_id) ON DELETE RESTRICT, + node_id TEXT NOT NULL, + status TEXT NOT NULL, + started_at REAL, + finished_at REAL, + input_digest TEXT, + output_digest TEXT, + error_code TEXT + ); + CREATE INDEX workflow_step_runs_run ON workflow_step_runs (run_id, step_run_id); + CREATE TABLE workflow_approvals ( + approval_id TEXT PRIMARY KEY, + run_id TEXT NOT NULL REFERENCES workflow_runs(run_id) ON DELETE RESTRICT, + node_id TEXT NOT NULL, + requested_at REAL NOT NULL, + requested_by TEXT NOT NULL, + decision TEXT, + decided_at REAL, + decided_by TEXT + ); + CREATE UNIQUE INDEX workflow_approvals_pending_node + ON workflow_approvals (run_id, node_id) WHERE decision IS NULL; + CREATE TABLE workflow_run_state ( + run_id TEXT PRIMARY KEY REFERENCES workflow_runs(run_id) ON DELETE RESTRICT, + input_json TEXT NOT NULL, + outputs_json TEXT NOT NULL, + updated_at REAL NOT NULL + ); + CREATE TABLE workflow_evaluations ( + evaluation_id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL REFERENCES workflows(workflow_id) ON DELETE RESTRICT, + revision_id TEXT NOT NULL REFERENCES workflow_revisions(revision_id) ON DELETE RESTRICT, + specification_digest TEXT NOT NULL, + generated_at REAL NOT NULL, + case_count INTEGER NOT NULL, + passed_case_count INTEGER NOT NULL + ); + CREATE INDEX workflow_evaluations_revision_generated + ON workflow_evaluations (workflow_id, revision_id, generated_at DESC, evaluation_id DESC); + """ + ) + + +def _migrate_v1_to_v2(connection: sqlite3.Connection) -> None: + connection.execute( + """CREATE TABLE workflow_run_state ( + run_id TEXT PRIMARY KEY REFERENCES workflow_runs(run_id) ON DELETE RESTRICT, + input_json TEXT NOT NULL, + outputs_json TEXT NOT NULL, + updated_at REAL NOT NULL + )""" + ) + + +def _migrate_v2_to_v3(connection: sqlite3.Connection) -> None: + connection.executescript( + """CREATE TABLE workflow_evaluations ( + evaluation_id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL REFERENCES workflows(workflow_id) ON DELETE RESTRICT, + revision_id TEXT NOT NULL REFERENCES workflow_revisions(revision_id) ON DELETE RESTRICT, + specification_digest TEXT NOT NULL, + generated_at REAL NOT NULL, + case_count INTEGER NOT NULL, + passed_case_count INTEGER NOT NULL + ); + CREATE INDEX workflow_evaluations_revision_generated + ON workflow_evaluations (workflow_id, revision_id, generated_at DESC, evaluation_id DESC);""" + ) + + +def _validate_schema(connection: sqlite3.Connection) -> None: + required = { + "workflows", + "workflow_drafts", + "workflow_revisions", + "workflow_deployments", + "workflow_runs", + "workflow_step_runs", + "workflow_approvals", + "workflow_run_state", + "workflow_evaluations", + } + actual = { + str(row["name"]) + for row in connection.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall() + } + if not required <= actual: + raise WorkflowStoreSchemaError() + + +def _workflow_from_row(row: sqlite3.Row) -> Workflow: + try: + return Workflow( + workflow_id=str(row["workflow_id"]), + tenant_id=TenantId(str(row["tenant_id"])), + project_id=str(row["project_id"]), + display_name=str(row["display_name"]), + active_revision_id=row["active_revision_id"], + status=WorkflowStatus(str(row["status"])), + created_at=float(row["created_at"]), + updated_at=float(row["updated_at"]), + ) + except (KeyError, TypeError, ValueError) as error: + raise WorkflowStoreSchemaError() from error + + +def _draft_from_row(row: sqlite3.Row) -> WorkflowDraft: + try: + specification = _specification_from_json(row["specification_json"]) + budget = _budget_from_json(row["budget_json"]) + return WorkflowDraft( + workflow_id=str(row["workflow_id"]), + version=int(row["version"]), + specification=specification, + budget=budget, + updated_at=float(row["updated_at"]), + updated_by=str(row["updated_by"]), + change_summary=str(row["change_summary"]), + ) + except (TypeError, ValueError) as error: + raise WorkflowStoreSchemaError() from error + + +def _revision_from_row(row: sqlite3.Row) -> WorkflowRevision: + try: + specification = _specification_from_json(row["specification_json"]) + budget = _budget_from_json(row["budget_json"]) + if specification is None or budget is None or specification.digest != str(row["specification_digest"]): + raise ValueError + return WorkflowRevision( + revision_id=str(row["revision_id"]), + workflow_id=str(row["workflow_id"]), + revision_number=int(row["revision_number"]), + specification=specification, + budget=budget, + created_at=float(row["created_at"]), + created_by=str(row["created_by"]), + change_summary=str(row["change_summary"]), + ) + except (TypeError, ValueError) as error: + raise WorkflowStoreSchemaError() from error + + +def _run_from_row(row: sqlite3.Row) -> WorkflowRun: + try: + return WorkflowRun( + run_id=str(row["run_id"]), + workflow_id=str(row["workflow_id"]), + revision_id=str(row["revision_id"]), + specification_digest=str(row["specification_digest"]), + status=WorkflowRunStatus(str(row["status"])), + created_at=float(row["created_at"]), + updated_at=float(row["updated_at"]), + created_by=str(row["created_by"]), + input_digest=str(row["input_digest"]), + error_code=row["error_code"], + ) + except (TypeError, ValueError) as error: + raise WorkflowStoreSchemaError() from error + + +def _step_from_row(row: sqlite3.Row) -> WorkflowStepRun: + try: + return WorkflowStepRun( + step_run_id=str(row["step_run_id"]), + run_id=str(row["run_id"]), + node_id=str(row["node_id"]), + status=WorkflowStepStatus(str(row["status"])), + started_at=row["started_at"], + finished_at=row["finished_at"], + input_digest=row["input_digest"], + output_digest=row["output_digest"], + error_code=row["error_code"], + ) + except (TypeError, ValueError) as error: + raise WorkflowStoreSchemaError() from error + + +def _run_state_from_row(row: sqlite3.Row) -> WorkflowRunState: + try: + inputs = json.loads(str(row["input_json"])) + outputs = json.loads(str(row["outputs_json"])) + return WorkflowRunState( + run_id=str(row["run_id"]), + input_values=inputs, + node_outputs=outputs, + updated_at=float(row["updated_at"]), + ) + except (TypeError, ValueError, json.JSONDecodeError) as error: + raise WorkflowStoreSchemaError() from error + + +def _approval_from_row(row: sqlite3.Row) -> WorkflowApproval: + try: + raw_decision = row["decision"] + return WorkflowApproval( + approval_id=str(row["approval_id"]), + run_id=str(row["run_id"]), + node_id=str(row["node_id"]), + requested_at=float(row["requested_at"]), + requested_by=str(row["requested_by"]), + decision=ApprovalDecision(str(raw_decision)) if raw_decision is not None else None, + decided_at=float(row["decided_at"]) if row["decided_at"] is not None else None, + decided_by=str(row["decided_by"]) if row["decided_by"] is not None else None, + ) + except (TypeError, ValueError) as error: + raise WorkflowStoreSchemaError() from error + + +def _evaluation_from_row(row: sqlite3.Row) -> WorkflowEvaluation: + try: + return WorkflowEvaluation( + evaluation_id=str(row["evaluation_id"]), workflow_id=str(row["workflow_id"]), + revision_id=str(row["revision_id"]), specification_digest=str(row["specification_digest"]), + generated_at=float(row["generated_at"]), case_count=int(row["case_count"]), + passed_case_count=int(row["passed_case_count"]), + ) + except (TypeError, ValueError) as error: + raise WorkflowStoreSchemaError() from error + + +def _specification_json(specification: WorkflowSpec | None) -> str | None: + return None if specification is None else specification.to_json() + + +def _specification_from_json(value: object) -> WorkflowSpec | None: + if value is None: + return None + if not isinstance(value, str) or len(value.encode("utf-8")) > 256 * 1024: + raise ValueError + return WorkflowSpec.from_json(value) + + +def _budget_json(budget: ExecutionBudget | None) -> str | None: + if budget is None: + return None + return json.dumps( + { + "max_steps": budget.max_steps, + "max_model_calls": budget.max_model_calls, + "max_wall_seconds": budget.max_wall_seconds, + }, + separators=(",", ":"), + sort_keys=True, + ) + + +def _budget_from_json(value: object) -> ExecutionBudget | None: + if value is None: + return None + if not isinstance(value, str): + raise ValueError + payload: Any = json.loads(value) + if not isinstance(payload, dict) or set(payload) != { + "max_steps", + "max_model_calls", + "max_wall_seconds", + }: + raise ValueError + return ExecutionBudget(**payload) + + +def _tenant(principal: Principal) -> TenantId: + if not isinstance(principal, Principal): + raise WorkflowModelError("workflow principal is invalid") + return principal.tenant_id + + +def _same_tenant(expected: TenantId, actual: TenantId) -> None: + if expected != actual: + raise WorkflowUnavailableError() + + +def _safe_workflow_id(value: object) -> str: + try: + return validate_workflow_id(value) + except ValueError as error: + raise WorkflowUnavailableError() from error + + +def _safe_revision_id(value: object) -> str: + try: + return validate_workflow_revision_id(value) + except ValueError as error: + raise WorkflowRevisionUnavailableError() from error + + +def _safe_run_id(value: object) -> str: + try: + return validate_workflow_run_id(value) + except ValueError as error: + raise WorkflowRunUnavailableError() from error + + +def _limit(value: object) -> int: + if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= _MAX_LIST_LIMIT: + raise WorkflowModelError("workflow list limit is invalid") + return value + + +def _require_workflow(connection: sqlite3.Connection, tenant: TenantId, workflow_id: str) -> sqlite3.Row: + row = connection.execute( + "SELECT * FROM workflows WHERE workflow_id = ? AND tenant_id = ?", (workflow_id, tenant.value) + ).fetchone() + if row is None: + raise WorkflowUnavailableError() + return cast(sqlite3.Row, row) + + +def _require_revision( + connection: sqlite3.Connection, workflow_id: str, revision_id: str +) -> sqlite3.Row: + row = connection.execute( + "SELECT * FROM workflow_revisions WHERE workflow_id = ? AND revision_id = ?", + (workflow_id, revision_id), + ).fetchone() + if row is None: + raise WorkflowRevisionUnavailableError() + return cast(sqlite3.Row, row) + + +def _validate_run_transition(current: WorkflowRunStatus, target: WorkflowRunStatus) -> None: + allowed = { + WorkflowRunStatus.CREATED: {WorkflowRunStatus.QUEUED, WorkflowRunStatus.CANCELLED}, + WorkflowRunStatus.QUEUED: { + WorkflowRunStatus.RUNNING, + WorkflowRunStatus.CANCELLED, + WorkflowRunStatus.INTERRUPTED, + }, + WorkflowRunStatus.RUNNING: { + WorkflowRunStatus.WAITING_APPROVAL, + WorkflowRunStatus.SUCCEEDED, + WorkflowRunStatus.FAILED, + WorkflowRunStatus.CANCELLED, + WorkflowRunStatus.INTERRUPTED, + }, + WorkflowRunStatus.WAITING_APPROVAL: { + WorkflowRunStatus.QUEUED, + WorkflowRunStatus.FAILED, + WorkflowRunStatus.CANCELLED, + }, + WorkflowRunStatus.SUCCEEDED: set(), + WorkflowRunStatus.FAILED: set(), + WorkflowRunStatus.CANCELLED: set(), + WorkflowRunStatus.INTERRUPTED: set(), + } + if target not in allowed[current]: + raise WorkflowStoreStorageError() + + +__all__ = [ + "WorkflowApprovalUnavailableError", + "WorkflowDraftConflictError", + "WorkflowPublishConflictError", + "WorkflowRevisionUnavailableError", + "WorkflowRunUnavailableError", + "WorkflowStore", + "WorkflowStoreError", + "WorkflowStoreSchemaError", + "WorkflowStoreStorageError", + "WorkflowUnavailableError", +] diff --git a/tests/test_architecture.py b/tests/test_architecture.py index 2caed43..552816f 100644 --- a/tests/test_architecture.py +++ b/tests/test_architecture.py @@ -32,6 +32,12 @@ "rag_system/loader_contracts.py", "rag_system/retrieval_experiments.py", "rag_system/runtime_profile.py", + "rag_system/workflow_contracts.py", + "rag_system/workflow_models.py", + "rag_system/workflow_store.py", + "rag_system/workflow_runtime.py", + "rag_system/workflow_service.py", + "rag_system/workflow_native_nodes.py", ) FORBIDDEN_FRAMEWORK_PREFIXES = ( "fastapi", diff --git a/tests/test_workflow_contracts.py b/tests/test_workflow_contracts.py new file mode 100644 index 0000000..6974837 --- /dev/null +++ b/tests/test_workflow_contracts.py @@ -0,0 +1,210 @@ +from __future__ import annotations + +import json +import unittest +from dataclasses import FrozenInstanceError + +from rag_system.workflow_contracts import ( + WORKFLOW_DSL_SCHEMA_VERSION, + WorkflowInput, + WorkflowInputBinding, + WorkflowNode, + WorkflowNodeKind, + WorkflowOutput, + WorkflowResourceKind, + WorkflowResourceRef, + WorkflowSpec, + WorkflowValidationError, +) + + +KNOWLEDGE_BASE_ID = "kb_12345678901234567890123456789012" + + +def _binding(target: str, source: str) -> WorkflowInputBinding: + return WorkflowInputBinding(target=target, source=source) + + +def _workflow(*, reverse_order: bool = False) -> WorkflowSpec: + retrieve = WorkflowNode( + node_id="retrieve", + node_kind=WorkflowNodeKind.KNOWLEDGE_RETRIEVE, + input_bindings=(_binding("query", "input.question"),), + output_names=("evidence",), + resources=( + WorkflowResourceRef(WorkflowResourceKind.KNOWLEDGE_BASE, KNOWLEDGE_BASE_ID), + ), + parameters={"max_results": 5}, + ) + prompt = WorkflowNode( + node_id="prompt", + node_kind=WorkflowNodeKind.PROMPT_RENDER, + depends_on=("retrieve",), + input_bindings=( + _binding("evidence", "node.retrieve.evidence"), + _binding("question", "input.question"), + ), + output_names=("prompt",), + parameters={"template": "Question: {{ question }}\nEvidence: {{ evidence }}"}, + ) + generate = WorkflowNode( + node_id="generate", + node_kind=WorkflowNodeKind.MODEL_GENERATE, + depends_on=("prompt",), + input_bindings=(_binding("prompt", "node.prompt.prompt"),), + output_names=("answer",), + resources=(WorkflowResourceRef(WorkflowResourceKind.MODEL_PROFILE, "default"),), + parameters={"max_output_tokens": 1_024}, + ) + validate = WorkflowNode( + node_id="validate", + node_kind=WorkflowNodeKind.GROUNDING_VALIDATE, + depends_on=("generate", "retrieve"), + input_bindings=( + _binding("answer", "node.generate.answer"), + _binding("evidence", "node.retrieve.evidence"), + ), + output_names=("validation",), + parameters={"require_citations": True}, + ) + condition = WorkflowNode( + node_id="condition", + node_kind=WorkflowNodeKind.CONDITION, + depends_on=("validate",), + input_bindings=(_binding("validation", "node.validate.validation"),), + output_names=("decision",), + parameters={"rule": "evidence_sufficient"}, + ) + nodes = (retrieve, prompt, generate, validate, condition) + if reverse_order: + nodes = tuple(reversed(nodes)) + return WorkflowSpec( + schema_version=WORKFLOW_DSL_SCHEMA_VERSION, + inputs=(WorkflowInput("question"),), + nodes=nodes, + outputs=( + WorkflowOutput("answer", "node.generate.answer"), + WorkflowOutput("decision", "node.condition.decision"), + ), + ) + + +class WorkflowContractTests(unittest.TestCase): + def test_valid_workflow_is_immutable_and_has_a_canonical_digest(self) -> None: + workflow = _workflow() + + self.assertEqual(workflow.nodes[0].node_id, "condition") + self.assertEqual(workflow.digest, _workflow(reverse_order=True).digest) + self.assertEqual(WorkflowSpec.from_json(workflow.to_json()), workflow) + with self.assertRaises(FrozenInstanceError): + workflow.nodes = () # type: ignore[misc] + with self.assertRaises(TypeError): + workflow.nodes[0].parameters["max_results"] = 1 # type: ignore[index] + + def test_json_decoder_rejects_unknown_and_duplicate_fields(self) -> None: + payload = _workflow().to_dict() + payload["unexpected"] = True + with self.assertRaises(WorkflowValidationError): + WorkflowSpec.from_json(json.dumps(payload)) + duplicate = _workflow().to_json().replace( + '"schema_version":1', '"schema_version":1,"schema_version":1', 1 + ) + with self.assertRaises(WorkflowValidationError): + WorkflowSpec.from_json(duplicate) + + def test_node_shape_rejects_unbounded_or_incorrect_capabilities(self) -> None: + with self.assertRaises(WorkflowValidationError): + WorkflowNode( + node_id="generate", + node_kind=WorkflowNodeKind.MODEL_GENERATE, + input_bindings=(_binding("prompt", "input.question"),), + output_names=("answer",), + resources=(), + ) + with self.assertRaises(WorkflowValidationError): + WorkflowNode( + node_id="prompt", + node_kind=WorkflowNodeKind.PROMPT_RENDER, + input_bindings=( + _binding("question", "input.question"), + _binding("evidence", "input.question"), + ), + output_names=("prompt",), + parameters={"template": "ok", "shell": "powershell"}, + ) + with self.assertRaises(WorkflowValidationError): + WorkflowNode( + node_id="condition", + node_kind=WorkflowNodeKind.CONDITION, + input_bindings=(_binding("validation", "input.question"),), + output_names=("decision",), + parameters={"rule": "eval(user_code)"}, + ) + + def test_graph_requires_declared_inputs_direct_dependencies_and_reachable_nodes(self) -> None: + retrieve = WorkflowNode( + node_id="retrieve", + node_kind=WorkflowNodeKind.KNOWLEDGE_RETRIEVE, + input_bindings=(_binding("query", "input.question"),), + output_names=("evidence",), + resources=( + WorkflowResourceRef(WorkflowResourceKind.KNOWLEDGE_BASE, KNOWLEDGE_BASE_ID), + ), + ) + prompt = WorkflowNode( + node_id="prompt", + node_kind=WorkflowNodeKind.PROMPT_RENDER, + depends_on=(), + input_bindings=( + _binding("question", "input.question"), + _binding("evidence", "node.retrieve.evidence"), + ), + output_names=("prompt",), + parameters={"template": "{{ question }}"}, + ) + with self.assertRaisesRegex(WorkflowValidationError, "dependencies"): + WorkflowSpec( + schema_version=WORKFLOW_DSL_SCHEMA_VERSION, + inputs=(WorkflowInput("question"),), + nodes=(retrieve, prompt), + outputs=(WorkflowOutput("prompt", "node.prompt.prompt"),), + ) + + orphan = WorkflowNode( + node_id="orphan", + node_kind=WorkflowNodeKind.KNOWLEDGE_RETRIEVE, + input_bindings=(_binding("query", "input.question"),), + output_names=("evidence",), + resources=( + WorkflowResourceRef(WorkflowResourceKind.KNOWLEDGE_BASE, KNOWLEDGE_BASE_ID), + ), + ) + workflow = _workflow() + with self.assertRaisesRegex(WorkflowValidationError, "disconnected"): + WorkflowSpec( + schema_version=workflow.schema_version, + inputs=workflow.inputs, + nodes=(*workflow.nodes, orphan), + outputs=workflow.outputs, + ) + + def test_workflow_rejects_invalid_resources_parameters_and_output_references(self) -> None: + with self.assertRaises(WorkflowValidationError): + WorkflowResourceRef(WorkflowResourceKind.MODEL_PROFILE, "https://model.example") + with self.assertRaises(WorkflowValidationError): + WorkflowNode( + node_id="retrieve", + node_kind=WorkflowNodeKind.KNOWLEDGE_RETRIEVE, + input_bindings=(_binding("query", "input.question"),), + output_names=("evidence",), + resources=( + WorkflowResourceRef(WorkflowResourceKind.KNOWLEDGE_BASE, KNOWLEDGE_BASE_ID), + ), + parameters={"max_results": True}, + ) + with self.assertRaises(WorkflowValidationError): + WorkflowOutput("answer", "input.question") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_workflow_native_nodes.py b/tests/test_workflow_native_nodes.py new file mode 100644 index 0000000..30198d1 --- /dev/null +++ b/tests/test_workflow_native_nodes.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import unittest + +from rag_system.tenancy import Principal, TenantId +from rag_system.workflow_contracts import WorkflowInputBinding, WorkflowNode, WorkflowNodeKind +from rag_system.workflow_native_nodes import evaluate_condition, render_prompt, validate_grounding + + +PRINCIPAL = Principal("writer", TenantId("tenant-a"), frozenset({"writer"})) + + +class NativeWorkflowNodeTests(unittest.TestCase): + def test_prompt_grounding_and_condition_are_deterministic(self) -> None: + prompt = WorkflowNode( + "prompt", WorkflowNodeKind.PROMPT_RENDER, + input_bindings=( + WorkflowInputBinding("question", "input.question"), + WorkflowInputBinding("evidence", "input.evidence"), + ), output_names=("prompt",), parameters={"template": "Q={{ question }} E={{ evidence }}"}, + ) + validation = WorkflowNode( + "validate", WorkflowNodeKind.GROUNDING_VALIDATE, + input_bindings=( + WorkflowInputBinding("answer", "input.answer"), + WorkflowInputBinding("evidence", "input.evidence"), + ), output_names=("validation",), parameters={"require_citations": True}, + ) + condition = WorkflowNode( + "condition", WorkflowNodeKind.CONDITION, + input_bindings=(WorkflowInputBinding("validation", "input.validation"),), + output_names=("decision",), parameters={"rule": "evidence_sufficient"}, + ) + + rendered = render_prompt(PRINCIPAL, prompt, {"question": "q", "evidence": ["e1", "e2"]}) + checked = validate_grounding(PRINCIPAL, validation, {"answer": "a", "evidence": ["e1"]}) + + self.assertEqual(rendered, {"prompt": "Q=q E=e1\ne2"}) + self.assertEqual(evaluate_condition(PRINCIPAL, condition, checked), {"decision": "allow"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_workflow_runtime.py b/tests/test_workflow_runtime.py new file mode 100644 index 0000000..7270443 --- /dev/null +++ b/tests/test_workflow_runtime.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +import secrets +import tempfile +import unittest +from pathlib import Path + +from rag_system.tenancy import Principal, TenantId +from rag_system.workflow_contracts import ( + WORKFLOW_DSL_SCHEMA_VERSION, + WorkflowInput, + WorkflowInputBinding, + WorkflowNode, + WorkflowNodeKind, + WorkflowOutput, + WorkflowResourceKind, + WorkflowResourceRef, + WorkflowSpec, +) +from rag_system.workflow_models import ( + ApprovalDecision, + ExecutionBudget, + Workflow, + WorkflowDeployment, + WorkflowDraft, + WorkflowRevision, + WorkflowRunStatus, + WorkflowStatus, +) +from rag_system.workflow_runtime import WorkflowRuntime +from rag_system.workflow_store import WorkflowStore + + +PROJECT_ID = "prj_12345678901234567890123456789012" +KNOWLEDGE_BASE_ID = "kb_12345678901234567890123456789012" + + +def _id(prefix: str) -> str: + return f"{prefix}_{secrets.token_hex(16)}" + + +def _specification(*, approval: bool = False) -> WorkflowSpec: + retrieve = WorkflowNode( + node_id="retrieve", node_kind=WorkflowNodeKind.KNOWLEDGE_RETRIEVE, + input_bindings=(WorkflowInputBinding("query", "input.question"),), + output_names=("evidence",), + resources=(WorkflowResourceRef(WorkflowResourceKind.KNOWLEDGE_BASE, KNOWLEDGE_BASE_ID),), + ) + prompt = WorkflowNode( + node_id="prompt", node_kind=WorkflowNodeKind.PROMPT_RENDER, depends_on=("retrieve",), + input_bindings=( + WorkflowInputBinding("question", "input.question"), + WorkflowInputBinding("evidence", "node.retrieve.evidence"), + ), output_names=("prompt",), parameters={"template": "{{ question }} {{ evidence }}"}, + ) + generate = WorkflowNode( + node_id="generate", node_kind=WorkflowNodeKind.MODEL_GENERATE, depends_on=("prompt",), + input_bindings=(WorkflowInputBinding("prompt", "node.prompt.prompt"),), + output_names=("answer",), + resources=(WorkflowResourceRef(WorkflowResourceKind.MODEL_PROFILE, "default"),), + ) + nodes = [retrieve, prompt, generate] + outputs = [WorkflowOutput("answer", "node.generate.answer")] + if approval: + review = WorkflowNode( + node_id="review", node_kind=WorkflowNodeKind.HUMAN_APPROVAL, depends_on=("generate",), + input_bindings=(WorkflowInputBinding("message", "node.generate.answer"),), + output_names=("decision",), parameters={"timeout_seconds": 60}, + ) + nodes.append(review) + outputs.append(WorkflowOutput("decision", "node.review.decision")) + return WorkflowSpec( + schema_version=WORKFLOW_DSL_SCHEMA_VERSION, inputs=(WorkflowInput("question"),), + nodes=tuple(nodes), outputs=tuple(outputs), + ) + + +def _executors(): + return { + WorkflowNodeKind.KNOWLEDGE_RETRIEVE: lambda _p, _n, values: {"evidence": f"source:{values['query']}"}, + WorkflowNodeKind.PROMPT_RENDER: lambda _p, _n, values: {"prompt": f"{values['question']}|{values['evidence']}"}, + WorkflowNodeKind.MODEL_GENERATE: lambda _p, _n, values: {"answer": f"answer:{values['prompt']}"}, + } + + +class WorkflowRuntimeTests(unittest.TestCase): + def setUp(self) -> None: + self.tempdir = tempfile.TemporaryDirectory() + self.store = WorkflowStore(Path(self.tempdir.name) / "workflows.sqlite3") + self.principal = Principal("operator", TenantId("tenant-a"), frozenset({"reader", "writer", "operator"})) + self.now = 100.0 + self.runtime = WorkflowRuntime(self.store, _executors(), clock=lambda: self.now) + + def tearDown(self) -> None: + self.tempdir.cleanup() + + def _publish(self, specification: WorkflowSpec, budget: ExecutionBudget | None = None) -> Workflow: + budget = budget or ExecutionBudget() + workflow = self.store.create_workflow( + self.principal, + Workflow(_id("wf"), self.principal.tenant_id, PROJECT_ID, "Workflow", None, + WorkflowStatus.ACTIVE, self.now, self.now), + ) + draft = WorkflowDraft(workflow.workflow_id, 1, specification, budget, self.now, self.principal.subject, "Initial") + self.store.update_draft(self.principal, draft, expected_version=0) + revision = self.store.create_revision( + self.principal, + WorkflowRevision(_id("wfr"), workflow.workflow_id, 1, specification, budget, + self.now, self.principal.subject, "Initial"), + ) + return self.store.publish( + self.principal, + WorkflowDeployment(_id("wfd"), workflow.workflow_id, revision.revision_id, self.now, self.principal.subject), + updated_at=self.now, expected_active_revision_id=None, + ) + + def test_executes_native_nodes_and_persists_auditable_steps(self) -> None: + workflow = self._publish(_specification()) + + execution = self.runtime.start(self.principal, workflow.workflow_id, {"question": "hello"}) + + self.assertEqual(execution.run.status, WorkflowRunStatus.SUCCEEDED) + self.assertEqual(execution.outputs, {"answer": "answer:hello|source:hello"}) + self.assertEqual( + [step.status.value for step in self.store.list_step_runs(self.principal, execution.run.run_id)], + ["succeeded", "succeeded", "succeeded"], + ) + state = self.store.get_run_state(self.principal, execution.run.run_id) + self.assertIn("generate", state.node_outputs) + + def test_approval_pauses_then_resumes_from_durable_state(self) -> None: + workflow = self._publish(_specification(approval=True)) + waiting = self.runtime.start(self.principal, workflow.workflow_id, {"question": "hello"}) + + self.assertEqual(waiting.run.status, WorkflowRunStatus.WAITING_APPROVAL) + self.assertIsNotNone(waiting.pending_approval_id) + self.store.decide_approval( + self.principal, waiting.pending_approval_id or "", decision=ApprovalDecision.APPROVED, + decided_at=self.now + 1, + ) + self.now += 1 + execution = self.runtime.resume(self.principal, waiting.run.run_id) + + self.assertEqual(execution.run.status, WorkflowRunStatus.SUCCEEDED) + self.assertEqual(execution.outputs["decision"] if execution.outputs else None, "approved") + self.assertEqual(len(self.store.list_step_runs(self.principal, waiting.run.run_id)), 4) + + def test_budget_failure_stops_before_disallowed_model_call(self) -> None: + workflow = self._publish(_specification(), ExecutionBudget(max_steps=10, max_model_calls=0, max_wall_seconds=60)) + + execution = self.runtime.start(self.principal, workflow.workflow_id, {"question": "hello"}) + + self.assertEqual(execution.run.status, WorkflowRunStatus.FAILED) + self.assertEqual(execution.run.error_code, "model_call_budget_exceeded") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_workflow_service.py b/tests/test_workflow_service.py new file mode 100644 index 0000000..f2817a9 --- /dev/null +++ b/tests/test_workflow_service.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import secrets +import tempfile +import unittest +from pathlib import Path + +from rag_system.knowledge_base_contracts import KnowledgeBaseStatus +from rag_system.tenancy import Principal, TenantId +from rag_system.workflow_contracts import ( + WORKFLOW_DSL_SCHEMA_VERSION, WorkflowInput, WorkflowInputBinding, WorkflowNode, + WorkflowNodeKind, WorkflowOutput, WorkflowResourceKind, WorkflowResourceRef, WorkflowSpec, +) +from rag_system.workflow_models import ExecutionBudget, WorkflowEvaluation +from rag_system.workflow_service import WorkflowService, WorkflowServiceValidationError +from rag_system.workflow_store import WorkflowStore + + +PROJECT_ID = "prj_12345678901234567890123456789012" +KNOWLEDGE_BASE_ID = "kb_12345678901234567890123456789012" + + +class _Projects: + def get_project(self, _principal, project_id): + if project_id != PROJECT_ID: + raise ValueError + return object() + + +class _KnowledgeBases: + def get(self, _principal, resource_id): + if resource_id != KNOWLEDGE_BASE_ID: + raise ValueError + return type("Record", (), {"status": KnowledgeBaseStatus.READY})() + + +def _specification() -> WorkflowSpec: + retrieve = WorkflowNode( + node_id="retrieve", node_kind=WorkflowNodeKind.KNOWLEDGE_RETRIEVE, + input_bindings=(WorkflowInputBinding("query", "input.question"),), output_names=("evidence",), + resources=(WorkflowResourceRef(WorkflowResourceKind.KNOWLEDGE_BASE, KNOWLEDGE_BASE_ID),), + ) + prompt = WorkflowNode( + node_id="prompt", node_kind=WorkflowNodeKind.PROMPT_RENDER, depends_on=("retrieve",), + input_bindings=(WorkflowInputBinding("question", "input.question"), WorkflowInputBinding("evidence", "node.retrieve.evidence")), + output_names=("prompt",), parameters={"template": "{{ question }}"}, + ) + generate = WorkflowNode( + node_id="generate", node_kind=WorkflowNodeKind.MODEL_GENERATE, depends_on=("prompt",), + input_bindings=(WorkflowInputBinding("prompt", "node.prompt.prompt"),), output_names=("answer",), + resources=(WorkflowResourceRef(WorkflowResourceKind.MODEL_PROFILE, "default"),), + ) + return WorkflowSpec(WORKFLOW_DSL_SCHEMA_VERSION, (WorkflowInput("question"),), (retrieve, prompt, generate), (WorkflowOutput("answer", "node.generate.answer"),)) + + +class WorkflowServiceTests(unittest.TestCase): + def setUp(self) -> None: + self.tempdir = tempfile.TemporaryDirectory() + self.store = WorkflowStore(Path(self.tempdir.name) / "workflows.sqlite3") + self.principal = Principal("operator", TenantId("tenant-a"), frozenset({"reader", "writer", "operator"})) + self.service = WorkflowService(self.store, _Projects(), _KnowledgeBases(), clock=lambda: 1.0) + + def tearDown(self) -> None: + self.tempdir.cleanup() + + def test_publishing_requires_passing_evaluation_bound_to_revision(self) -> None: + workflow = self.service.create_workflow(self.principal, PROJECT_ID, "Workflow") + draft = self.service.update_draft( + self.principal, workflow.workflow_id, _specification(), ExecutionBudget(), + expected_version=0, change_summary="Initial", + ) + revision = self.service.create_revision_from_draft( + self.principal, workflow.workflow_id, expected_version=draft.version + ) + with self.assertRaisesRegex(WorkflowServiceValidationError, "evaluation"): + self.service.publish(self.principal, workflow.workflow_id, revision.revision_id, expected_active_revision_id=None) + + evaluation = WorkflowEvaluation( + evaluation_id=f"weval_{secrets.token_hex(16)}", workflow_id=workflow.workflow_id, + revision_id=revision.revision_id, specification_digest=revision.specification_digest, + generated_at=1.0, case_count=2, passed_case_count=2, + ) + self.service.record_evaluation(self.principal, evaluation) + published = self.service.publish( + self.principal, workflow.workflow_id, revision.revision_id, expected_active_revision_id=None + ) + self.assertEqual(published.workflow.active_revision_id, revision.revision_id) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_workflow_store.py b/tests/test_workflow_store.py new file mode 100644 index 0000000..373601d --- /dev/null +++ b/tests/test_workflow_store.py @@ -0,0 +1,238 @@ +from __future__ import annotations + +import hashlib +import secrets +import tempfile +import unittest +from pathlib import Path + +from rag_system.tenancy import Principal, TenantId +from rag_system.workflow_contracts import ( + WORKFLOW_DSL_SCHEMA_VERSION, + WorkflowInput, + WorkflowInputBinding, + WorkflowNode, + WorkflowNodeKind, + WorkflowOutput, + WorkflowResourceKind, + WorkflowResourceRef, + WorkflowSpec, +) +from rag_system.workflow_models import ( + ApprovalDecision, + ExecutionBudget, + Workflow, + WorkflowApproval, + WorkflowDeployment, + WorkflowRevision, + WorkflowRun, + WorkflowRunStatus, + WorkflowStatus, + WorkflowStepRun, + WorkflowStepStatus, +) +from rag_system.workflow_store import ( + WorkflowDraftConflictError, + WorkflowStore, + WorkflowUnavailableError, +) + + +PROJECT_ID = "prj_12345678901234567890123456789012" +KNOWLEDGE_BASE_ID = "kb_12345678901234567890123456789012" + + +def _id(prefix: str) -> str: + return f"{prefix}_{secrets.token_hex(16)}" + + +def _digest(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _specification() -> WorkflowSpec: + retrieve = WorkflowNode( + node_id="retrieve", + node_kind=WorkflowNodeKind.KNOWLEDGE_RETRIEVE, + input_bindings=(WorkflowInputBinding("query", "input.question"),), + output_names=("evidence",), + resources=(WorkflowResourceRef(WorkflowResourceKind.KNOWLEDGE_BASE, KNOWLEDGE_BASE_ID),), + ) + prompt = WorkflowNode( + node_id="prompt", + node_kind=WorkflowNodeKind.PROMPT_RENDER, + depends_on=("retrieve",), + input_bindings=( + WorkflowInputBinding("question", "input.question"), + WorkflowInputBinding("evidence", "node.retrieve.evidence"), + ), + output_names=("prompt",), + parameters={"template": "{{ question }}\n{{ evidence }}"}, + ) + generate = WorkflowNode( + node_id="generate", + node_kind=WorkflowNodeKind.MODEL_GENERATE, + depends_on=("prompt",), + input_bindings=(WorkflowInputBinding("prompt", "node.prompt.prompt"),), + output_names=("answer",), + resources=(WorkflowResourceRef(WorkflowResourceKind.MODEL_PROFILE, "default"),), + ) + return WorkflowSpec( + schema_version=WORKFLOW_DSL_SCHEMA_VERSION, + inputs=(WorkflowInput("question"),), + nodes=(retrieve, prompt, generate), + outputs=(WorkflowOutput("answer", "node.generate.answer"),), + ) + + +class WorkflowStoreTests(unittest.TestCase): + def setUp(self) -> None: + self.tempdir = tempfile.TemporaryDirectory() + self.store = WorkflowStore(Path(self.tempdir.name) / "workflows.sqlite3") + self.principal = Principal("writer", TenantId("tenant-a"), frozenset({"reader", "writer"})) + self.other = Principal("reader", TenantId("tenant-b"), frozenset({"reader"})) + self.workflow = Workflow( + workflow_id=_id("wf"), + tenant_id=self.principal.tenant_id, + project_id=PROJECT_ID, + display_name="Trusted answer", + active_revision_id=None, + status=WorkflowStatus.ACTIVE, + created_at=1.0, + updated_at=1.0, + ) + + def tearDown(self) -> None: + self.tempdir.cleanup() + + def test_draft_revision_and_publish_are_durable_and_tenant_scoped(self) -> None: + self.store.create_workflow(self.principal, self.workflow) + draft = self.store.get_draft(self.principal, self.workflow.workflow_id) + configured = type(draft)( + workflow_id=draft.workflow_id, + version=1, + specification=_specification(), + budget=ExecutionBudget(max_steps=10, max_model_calls=2, max_wall_seconds=60), + updated_at=2.0, + updated_by=self.principal.subject, + change_summary="Initial trusted workflow", + ) + self.store.update_draft(self.principal, configured, expected_version=0) + revision = self.store.create_revision( + self.principal, + WorkflowRevision( + revision_id=_id("wfr"), + workflow_id=self.workflow.workflow_id, + revision_number=1, + specification=configured.specification, + budget=configured.budget, + created_at=3.0, + created_by=self.principal.subject, + change_summary="Initial trusted workflow", + ), + ) + active = self.store.publish( + self.principal, + WorkflowDeployment( + deployment_id=_id("wfd"), + workflow_id=self.workflow.workflow_id, + revision_id=revision.revision_id, + deployed_at=4.0, + deployed_by=self.principal.subject, + ), + updated_at=4.0, + expected_active_revision_id=None, + ) + + self.assertEqual(active.active_revision_id, revision.revision_id) + self.assertEqual(self.store.get_revision(self.principal, active.workflow_id, revision.revision_id), revision) + with self.assertRaises(WorkflowUnavailableError): + self.store.get_workflow(self.other, active.workflow_id) + with self.assertRaises(WorkflowDraftConflictError): + self.store.update_draft(self.principal, configured, expected_version=0) + + def test_run_steps_approval_and_interruption_recovery(self) -> None: + self.store.create_workflow(self.principal, self.workflow) + specification = _specification() + draft = self.store.get_draft(self.principal, self.workflow.workflow_id) + configured = type(draft)( + workflow_id=draft.workflow_id, + version=1, + specification=specification, + budget=ExecutionBudget(), + updated_at=2.0, + updated_by=self.principal.subject, + change_summary="Initial workflow", + ) + self.store.update_draft(self.principal, configured, expected_version=0) + revision = self.store.create_revision( + self.principal, + WorkflowRevision( + revision_id=_id("wfr"), + workflow_id=self.workflow.workflow_id, + revision_number=1, + specification=specification, + budget=ExecutionBudget(), + created_at=3.0, + created_by=self.principal.subject, + change_summary="Initial workflow", + ), + ) + run = self.store.create_run( + self.principal, + WorkflowRun( + run_id=_id("wrun"), + workflow_id=self.workflow.workflow_id, + revision_id=revision.revision_id, + specification_digest=revision.specification_digest, + status=WorkflowRunStatus.CREATED, + created_at=4.0, + updated_at=4.0, + created_by=self.principal.subject, + input_digest=_digest("question"), + ), + ) + self.store.transition_run(self.principal, run.run_id, status=WorkflowRunStatus.QUEUED, updated_at=5.0) + running = self.store.transition_run( + self.principal, run.run_id, status=WorkflowRunStatus.RUNNING, updated_at=6.0 + ) + self.store.save_step_run( + self.principal, + WorkflowStepRun( + step_run_id=_id("wstep"), + run_id=running.run_id, + node_id="retrieve", + status=WorkflowStepStatus.RUNNING, + started_at=6.0, + finished_at=None, + input_digest=_digest("question"), + output_digest=None, + ), + ) + self.assertEqual(self.store.recover_interrupted_runs(self.principal, updated_at=7.0), 1) + self.assertEqual(self.store.get_run(self.principal, run.run_id).status, WorkflowRunStatus.INTERRUPTED) + self.assertEqual(self.store.list_step_runs(self.principal, run.run_id)[0].status, WorkflowStepStatus.INTERRUPTED) + + waiting = self.store.create_run( + self.principal, + WorkflowRun( + run_id=_id("wrun"), workflow_id=self.workflow.workflow_id, revision_id=revision.revision_id, + specification_digest=revision.specification_digest, status=WorkflowRunStatus.WAITING_APPROVAL, + created_at=8.0, updated_at=8.0, created_by=self.principal.subject, input_digest=_digest("second"), + ), + ) + approval = self.store.create_approval( + self.principal, + WorkflowApproval( + approval_id=_id("wappr"), run_id=waiting.run_id, node_id="approval", + requested_at=8.0, requested_by=self.principal.subject, + ), + ) + decided = self.store.decide_approval( + self.principal, approval.approval_id, decision=ApprovalDecision.APPROVED, decided_at=9.0 + ) + self.assertEqual(decided.decision, ApprovalDecision.APPROVED) + + +if __name__ == "__main__": + unittest.main()