diff --git a/CHANGELOG.md b/CHANGELOG.md index 625e952e..330c5f0d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,25 @@ SPDX-License-Identifier: Apache-2.0 ## Unreleased +- The compiler checks and compiles **Call a workflow** (`callWorkflow`) against + the exact published version it calls: that version must opt in with + `spec.callable` (any workflow, or only `allowedCallers`), `with` is checked + against its input schema, a `businessKey` must be text of 1 to 200 characters, + and the step returns the called workflow's output, a `{status, runId, output, + error}` envelope with `onFailure: continue`, or `null` with `mode: detach`. New + diagnostics `WV-COMP-UNKNOWN_WORKFLOW`, `CALL_NOT_CALLABLE`, `CALL_RECURSION`, + `CALL_DEPTH` (more than 8 calls deep, the new compiler limit + `max_call_depth`), and `CALL_MODE` point at the field to fix. The called + version is locked as an interface dependency: the caller's artifact carries + that version's definition, so its digest matches the published one, but + compiling, importing, and activating the caller read only its input and output + schemas. Activation takes `workflow_activation_ids`, one live + activation of each called version in the same environment, and refuses wrong + or missing pins with `422 WV-READINESS-CALL`. This platform does not start child + runs yet, so it refuses to publish, import, or activate a workflow that calls + another with `WV-IR-UNSUPPORTED`; workflows without calls compile to the same + bytes as before. `weave workflow explain` and `weave workflow graph` show calls, + and the SDK's `WorkflowBuilder.with_callable()` writes `spec.callable`. - Loops run. A **Loop over items** (`forEach`) step runs its body once per element, `concurrency` at a time, in one run, and its output lists one result per element in element order, whatever order the iterations finish in. Each diff --git a/docs/concepts.md b/docs/concepts.md index cad01085..57963dd1 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -305,7 +305,7 @@ Use this table to translate what you know: | Error end event, terminate end event | **Fail** (`fail`) | Ends the whole run as failed with your code and message, including any other parallel branches | | Error boundary event | Not in the language | A failed Action retries when that is safe; otherwise the run is suspended with an incident for an operator. To branch on a business outcome, return it in the Action's output and test it with a Decision | | Message throw or end event | **Call an action**, or an integration event subscription | A subscription notifies another system when a run's status changes | -| Call activity, subprocess | **Call a workflow** (`callWorkflow`) | Defined in the [workflow language](contracts.md#steps), but this version does not compile or run it: the compiler reports `WV-COMP-UNSUPPORTED_FEATURE`. Until then, keep the part inside the same workflow, or make it a separate workflow that your product starts | +| Call activity, subprocess | **Call a workflow** (`callWorkflow`) | Defined in the [workflow language](contracts.md#steps). This version compiles and checks it against the exact version it calls, but the platform does not run calls yet and refuses to publish a workflow that uses one (`WV-IR-UNSUPPORTED`). Until then, keep the part inside the same workflow, or make it a separate workflow that your product starts | | Loop, multi-instance activity | **Loop over items** (`forEach`) | Runs its body once per element of a list, `concurrency` at a time, inside the same run, and collects one result per element in element order; see [loops at run time](reference/compiler.md#loops-at-run-time). A failure in one element stops the whole run, as in a Parallel | | Compensation | Not in the language | Model compensating Actions as ordinary steps; Weave never undoes an external effect | | Instance migration | Not available | A running run keeps its version; new runs use the new activation | diff --git a/docs/contracts.md b/docs/contracts.md index 762be564..c246e94b 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -220,10 +220,9 @@ each need a language feature: `flow.forEach`, `flow.callWorkflow`, `text.concat` and `text.join`. This version compiles and runs `concat`, `join`, and `forEach`; see [loops at run time](reference/compiler.md#loops-at-run-time). A platform that does not list `flow.forEach` refuses to publish, import, or activate a -workflow that uses a loop (`WV-IR-UNSUPPORTED`). The shape of -`callWorkflow` is final, so `load_definition` accepts it, but this version of the -compiler reports `WV-COMP-UNSUPPORTED_FEATURE` at each use instead of compiling -it. A platform runs a construct only when its +workflow that uses a loop (`WV-IR-UNSUPPORTED`). It also compiles +`callWorkflow` (see **Workflow calls** below) and refuses to publish a workflow +that calls another until it runs calls. A platform runs a construct only when its [language manifest](#language-manifest) lists the feature. Step IDs can never contain `[`, `#`, or `~`, which keeps [instance keys](reference/compiler.md#instance-keys) unambiguous. @@ -246,6 +245,37 @@ iterations; the compiler suggests a lower `maxItems` for the inner loop when the need more. The [compiler reference](reference/compiler.md#troubleshooting) lists the diagnostics. +**Workflow calls.** A `callWorkflow` step names an exact published workflow +version in `uses`, and that version must allow the call with `spec.callable`: +`{}` allows any workflow in the project, and `allowedCallers` limits it to the +named workflows. `with` is checked against the called version's `inputSchema`, +and `businessKey`, when set, must be text of 1 to 200 characters. The step's +output depends on how it runs: + +| `mode` and `onFailure` | Output | +| --- | --- | +| `wait`, without `onFailure` or with `stop` | The called version's output, checked against its `outputSchema` | +| `wait` with `continue` | `{status, runId, output, error}`: `status` (`succeeded`, `failed`, `cancelled`, or `timed_out`) and `runId` are always present; read `output` and `error` with `coalesce`, since either may be missing | +| `detach` | `null`; the step starts the call and continues | + +A call cannot lead back to the calling version, and the longest chain of calls a +workflow starts may be at most 8 deep (`max_call_depth`). The compiler locks the +called version as a `Workflow` dependency of the caller and reads only its input +and output schemas: the called workflow's own steps, actions, and connections +stay with its own activation. The caller's artifact still includes the called +version's definition, so the dependency digest matches the published one. Each +called version is included once, without its own dependencies, and counts toward +the 32 MiB artifact limit. Anyone who can read the caller's version or export +its artifact can see the called version's steps; secret values never appear, +because literals bound to secret inputs are refused. When you activate a +workflow that calls others, `workflow_activation_ids` pins one activation of +each called version in the same environment, by reference (for example +`{"notify-customer@1.0.0": "ACTIVATION_UUID"}`). Activation refuses a missing or +extra pin, a pin to another workflow or version, a retired activation or +version, a version that does not allow this workflow, and a chain of calls +deeper than 8 with `422 WV-READINESS-CALL`; the activation records each pin in +`workflow_call_pins` and the longest chain in `call_depth`. + **Human tasks** need an environment assignment binding at activation, and completing one requires the current claim and task permission. A workflow with a human task compiles to IR version `weave/ir-v1alpha2` or higher, because the @@ -383,6 +413,7 @@ The operator's `Limits()` defaults bound compilation: | `max_loop_items` | 10,000 | The largest `maxItems` a loop may declare | | `max_loop_depth` | 3 | Loops nested inside one another | | `max_run_iterations` | 100,000 | Iterations all loops of one workflow may need, each loop multiplied by the `maxItems` of the loops around it | +| `max_call_depth` | 8 | The longest chain of workflow calls one workflow may start | Definitions cannot change these budgets. Building a model alone does not apply them; the compiler does. @@ -458,7 +489,9 @@ backend is absent. The `integration` and `e2e` pytest markers are registered. | A field is rejected although it looks right | Field names are camelCase and case-sensitive; snake_case keys are rejected | Use the published name, such as `timeoutSeconds` | | `null` is rejected for an optional field | Optional fields may be omitted but not set to null | Remove the field | | A step's `retry` or `timeoutSeconds` is rejected | Retry and per-attempt timeout belong to the action definition | Move them to the action; use `spec.timeoutSeconds` for a workflow-wide timeout | -| `WV-COMP-UNSUPPORTED_FEATURE` | The document uses `callWorkflow`, which this version of the compiler does not compile yet | Keep the document for a later version, or replace the construct with the steps the compiler supports | +| `WV-COMP-UNSUPPORTED_FEATURE` | The document uses a construct whose language feature this compiler does not compile; the message names the feature | Upgrade the CLI or SDK that compiles it, or replace the construct with steps this compiler supports | +| `WV-IR-UNSUPPORTED` when publishing a workflow with `callWorkflow` | The compiler accepts the call, but this platform does not run workflow calls yet, so it refuses to publish it; `result.missing_features` is `["flow.callWorkflow"]` | Keep the workflow as a draft until the platform lists `flow.callWorkflow` in its language manifest; the called workflow itself publishes normally | +| `WV-READINESS-CALL` on activation | A called workflow version has no pinned activation; a pin names something the workflow does not call; the pinned activation or its version is retired, runs another version, or does not allow this workflow; the pinned version differs from the one the workflow was compiled against; or the chain of calls is deeper than 8 | Pin one live activation of each called version in `workflow_activation_ids` | | `WV-IR-UNSUPPORTED` when publishing a workflow with `forEach` | The platform you publish to is older and does not list `flow.forEach` in its language manifest; `result.missing_features` is `["flow.forEach"]` | Upgrade that platform, or keep the workflow as a draft until it lists the feature | | `WV-DECISION-OPERATOR` | A decision table rule uses `concat` or `join` | Build the text in a Transform step before the table and pass it in as input | | `WV-IR-UNSUPPORTED` on activation, a catalog read, or a run read | The platform does not list an IR version or a language feature the workflow uses; `result.missing_features` names the missing features and is empty when the IR version itself is unknown. A catalog list shows such a version as an unavailable item with `reason: ir_unsupported` and the same `missing_features`. A run that uses such a feature waits for an upgrade; signals and task results for it answer the same way, and run lists show it as unavailable | Upgrade the platform, or activate a version that does not use the feature | diff --git a/docs/reference/compiler.md b/docs/reference/compiler.md index bd4c2e4e..9c55580d 100644 --- a/docs/reference/compiler.md +++ b/docs/reference/compiler.md @@ -230,8 +230,9 @@ Executables also define a sorted `features` list of the language features they need (`text.concat`, `text.join`, `flow.forEach`, `flow.callWorkflow`, `ai.agent`, `ai.memory`). It is omitted when empty, so existing digests do not change. A workflow that uses `concat` or `join` in any expression compiles to -`weave/ir-v1alpha4` with `text.concat`, `text.join`, or both, and a workflow with -a `forEach` step lists `flow.forEach`; v1alpha4 always has a non-empty list, and +`weave/ir-v1alpha4` with `text.concat`, `text.join`, or both, a workflow with +a `forEach` step lists `flow.forEach`, and a workflow with a `callWorkflow` step +lists `flow.callWorkflow`; v1alpha4 always has a non-empty list, and the validator recomputes the version and the list from the graph. The version is the highest level any construct needs, so a workflow with a human task and `join` is v1alpha4, not v1alpha2. A platform imports and activates such an artifact only when its capabilities list every feature in @@ -252,6 +253,14 @@ locks: code. - Schema guards and signal nodes reference interned schemas instead of copying them into every step. +- A `callWorkflow` node refers to its called workflow's digest, locked as a + `Workflow` dependency whose document is that version's definition. The caller + uses only its `inputSchema`, `outputSchema`, and the named schemas they + reference: import never rechecks the called workflow's steps, and its actions + and connections are not dependencies of the caller. The node carries `with`, + `mode` (`wait` or `detach`), and, when set, `onFailure` and `businessKey`; the + guards `call_input` (at `…/with`) and `call_output` (at the step) hold the + called version's schemas, and a called version must allow the caller by name. Action and connector executables keep their strict manifest spec and dependencies; they have no workflow graph. @@ -611,7 +620,13 @@ is not classified. Safe ordinary artifacts keep their canonical digests. | Incident `WV-RUNTIME-LOOP-LIMIT` on a loop | `items` had more elements than the loop's `maxItems` | Terminate the run; raise `maxItems` (at most 10,000) or split the list before the loop | | Incident `WV-RUNTIME-LOOP-OUTPUT` on a loop | The collected results would exceed the 1 MiB payload limit | Terminate the run; return only what later steps need from the body, or use `collect: nonNull` to drop empty results | | `WV-COMP-UNKNOWN_ACTION` (or `_CONNECTOR`, `_TASK`, `_ADAPTER`) | The catalog does not contain that exact reference | Add the exact version to the catalog, or fix the reference | -| `WV-COMP-UNSUPPORTED_FEATURE` | The workflow uses a construct whose language feature this compiler does not compile yet (`callWorkflow`); the message names the feature | Keep the document for a later version, or use the steps and operators this version compiles | +| `WV-COMP-UNSUPPORTED_FEATURE` | The workflow uses a construct whose language feature this compiler does not compile; the message names the feature | Upgrade the CLI or SDK that compiles it, or use the steps and operators this version compiles | +| `WV-COMP-UNKNOWN_WORKFLOW` at `…/uses` | The catalog does not contain the exact workflow version a call names | Publish that version first, or fix the reference | +| `WV-COMP-CALL_NOT_CALLABLE` at `…/uses` | The called version has no `spec.callable`, or its `allowedCallers` does not name this workflow | Publish a new version of the called workflow whose `callable` allows this workflow | +| `WV-COMP-CALL_RECURSION` at `…/uses` | Following the calls of the called version leads back to this exact version | Call a workflow that does not lead back to this one | +| `WV-COMP-CALL_DEPTH` at `…/uses` | The chain of calls this call starts is deeper than 8 (`max_call_depth`), or never ends | Call a workflow that starts fewer calls of its own | +| `WV-COMP-CALL_MODE` at `…/onFailure` | `onFailure` is set on a call with `mode: detach` | Remove `onFailure`, or set `mode: wait` | +| `WV-COMP-REFERENCE_PRESENCE` reading `/steps//output/output` | A call with `onFailure: continue` has a child output only when the child succeeded | Read it with `coalesce`, or branch on `/steps//output/status` first | | `WV-COMP-LOOP_ITEMS` at `…/items` | A loop's `items` can never be a list | Point `items` at a list, such as an array field of the input or of a step's output | | `WV-COMP-UNKNOWN_COMPATIBILITY` at `…/items` | The compiler cannot prove that `items` is a list; the run checks it | Declare the field as an array where it comes from | | `WV-COMP-LOOP_SCOPE` at `…/ref` | `/item`, `/index`, or `/loops/` is read outside a loop body, in the loop's own `items`, or names a loop that does not enclose the expression | Read loop data only inside the loop's body; use `/loops//item` for an enclosing loop | diff --git a/examples/language/order-intake.workflow.yaml b/examples/language/order-intake.workflow.yaml index 5de2aa7e..f405da61 100644 --- a/examples/language/order-intake.workflow.yaml +++ b/examples/language/order-intake.workflow.yaml @@ -14,8 +14,8 @@ # Author: Firefly Software Foundation # SPDX-License-Identifier: Apache-2.0 -# Call a workflow and wait for its result, then start another call without waiting. This version compiles -# concat and join but reports "not yet supported" for callWorkflow; the document shape is final. +# Call a workflow and wait for its result, then start another call without waiting. Both calls compile +# against notify-customer@1.0.0; a platform publishes this workflow once it runs workflow calls. apiVersion: weave/v1alpha1 kind: Workflow metadata: @@ -26,7 +26,7 @@ spec: type: object properties: customerId: {type: string} - orderId: {type: string} + orderId: {type: string, minLength: 1, maxLength: 200} tags: type: array items: {type: string} diff --git a/scripts/language_fixtures.py b/scripts/language_fixtures.py index ee146ceb..afbb3829 100644 --- a/scripts/language_fixtures.py +++ b/scripts/language_fixtures.py @@ -18,7 +18,8 @@ The fixtures pin the stable language contracts: instance-key parse and format cases with the ``INSTANCE_KEY_PATTERN`` expression, the text conversion table for ``concat`` and ``join``, template segments -with their saved form, ``scope_at`` suggestions inside and around loops, and the language manifest snapshot. +with their saved form, ``scope_at`` suggestions inside and around loops and after workflow calls, and the language +manifest snapshot. Expected values are written here by hand; the generator refuses to write a fixture that the implementation disagrees with. Scope cases pin each suggestion's reference, source, loop and type label by hand and record the rest of each entry as ``scope_at`` returns it. @@ -281,6 +282,18 @@ def _loop(identifier: str, items: str, *steps: dict[str, Any], **options: Any) - _LINES = _loop( "line", "/item", {"id": "pair", "kind": "transform", "value": {"ref": "/loops/batch/index"}}, maxItems=50 ) + + +def _call(identifier: str, **options: Any) -> dict[str, Any]: + return { + "id": identifier, + "kind": "callWorkflow", + "uses": "notify-customer@1.0.0", + "with": {"ref": "/input"}, + **options, + } + + _INPUT_ROWS = [ ("/input", "input", None, "Group"), ("/input/invoices", "input", None, "List of groups"), @@ -346,6 +359,39 @@ def _loop(identifier: str, items: str, *steps: dict[str, Any], **options: Any) - [*_INPUT_ROWS, ("/item", "item", "batch", "List of groups"), ("/index", "index", "batch", "Whole number")], ), ("a field that is not an expression has no scope", _workflow(_NOTIFY), "/spec/steps/0/maxItems", None), + ( + "a call's input sees what is visible before the call", + _workflow(_call("notify")), + "/spec/steps/0/with", + _INPUT_ROWS, + ), + ( + "after a call that waits, its result is the called workflow's output, unknown without a catalog", + _workflow(_call("notify"), output={"ref": "/steps/notify/output"}), + "/spec/output", + [*_INPUT_ROWS, ("/steps/notify/output", "step", None, "Any value")], + ), + ( + "a call that continues on failure returns its status, run ID, the child's output and the error", + _workflow(_call("notify", onFailure="continue"), output={"ref": "/steps/notify/output"}), + "/spec/output", + [ + *_INPUT_ROWS, + ("/steps/notify/output", "step", None, "Group"), + ("/steps/notify/output/status", "step", None, "Choice"), + ("/steps/notify/output/runId", "step", None, "Text"), + ("/steps/notify/output/output", "step", None, "Any value"), + ("/steps/notify/output/error", "step", None, "Group"), + ("/steps/notify/output/error/code", "step", None, "Text"), + ("/steps/notify/output/error/message", "step", None, "Text"), + ], + ), + ( + "a call that does not wait returns nothing", + _workflow(_call("notify", mode="detach"), output={"ref": "/steps/notify/output"}), + "/spec/output", + [*_INPUT_ROWS, ("/steps/notify/output", "step", None, "Empty")], + ), ) diff --git a/src/firefly_weave/cli/workflow.py b/src/firefly_weave/cli/workflow.py index 6d4a96fe..0da1e502 100644 --- a/src/firefly_weave/cli/workflow.py +++ b/src/firefly_weave/cli/workflow.py @@ -40,6 +40,7 @@ ActionNode, ArtifactEnvelope, BranchOutputNode, + CallWorkflowNode, EndNode, ForEachNode, SwitchNode, @@ -88,6 +89,9 @@ def _explain(artifact: CompiledArtifact) -> None: detail = f" uses={dependencies[node.dependency].reference}" if isinstance(node, ActionNode) else "" if isinstance(node, ForEachNode): detail = f" concurrency={node.concurrency} maxItems={node.max_items} collect={node.collect}" + if isinstance(node, CallWorkflowNode): + failure = f" onFailure={node.on_failure}" if node.on_failure is not None else "" + detail = f" uses={dependencies[node.dependency].reference} mode={node.mode}{failure}" click.echo(f" {node.id} ({node.kind}) {location}{detail}") expressions: list[Expression] = [] if isinstance(node, ActionNode): @@ -100,6 +104,10 @@ def _explain(artifact: CompiledArtifact) -> None: expressions.extend(case.when for case in node.cases) elif isinstance(node, ForEachNode): expressions.append(node.items) + elif isinstance(node, CallWorkflowNode): + expressions.append(node.input) + if node.business_key is not None: + expressions.append(node.business_key) for expression in expressions: for reference in _references(expression): click.echo(f" reads {reference}") diff --git a/src/firefly_weave/compiler/admission.py b/src/firefly_weave/compiler/admission.py index dd399ea6..49a29a55 100644 --- a/src/firefly_weave/compiler/admission.py +++ b/src/firefly_weave/compiler/admission.py @@ -164,6 +164,10 @@ def walk(value: JsonValue) -> None: resource = catalog.resolve("DecisionTable", str(value.get("uses", ""))) spec = cast(JsonObject, resource.definition.value.get("spec", {})) if resource else {} binding(spec.get("inputSchema"), value["with"]) + if value.get("kind") == "callWorkflow" and "with" in value: + resource = catalog.resolve("Workflow", str(value.get("uses", ""))) + spec = cast(JsonObject, resource.definition.value.get("spec", {})) if resource else {} + binding(spec.get("inputSchema"), value["with"]) if value.get("kind") == "DecisionTable" and isinstance(value.get("spec"), dict): spec = cast(JsonObject, value["spec"]) output = spec.get("outputSchema") @@ -217,6 +221,11 @@ def admit_artifact(executable: JsonObject) -> None: for resource in catalog.resources.values(): if resource.kind == "Schema": Classification(resource.definition.value, bundle, SchemaLimits()).literals() + elif resource.kind == "Workflow": + # A called workflow is an interface: its steps were admitted when it was published. + spec = cast(JsonObject, resource.definition.value["spec"]) + interface = {key: spec[key] for key in ("inputSchema", "outputSchema")} + admit_authoring({"kind": "Workflow", "spec": interface}, catalog) else: admit_authoring(resource.definition.value, catalog) if "spec" in executable: @@ -233,6 +242,9 @@ def admit_artifact(executable: JsonObject) -> None: r for r in resources.values() if r.kind == "DecisionTable" and r.digest == node["dependency"] ) admit_authoring({"kind": "decisionTable", "uses": table.reference, "with": node["with"]}, catalog) + elif node["kind"] == "callWorkflow": + called = next(r for r in resources.values() if r.kind == "Workflow" and r.digest == node["dependency"]) + admit_authoring({"kind": "callWorkflow", "uses": called.reference, "with": node["with"]}, catalog) elif node["kind"] == "end": admit_binding( cast(JsonObject, schemas[cast(str, executable["outputSchema"])]), diff --git a/src/firefly_weave/compiler/analyzer.py b/src/firefly_weave/compiler/analyzer.py index 6bd8ef51..dd980fc4 100644 --- a/src/firefly_weave/compiler/analyzer.py +++ b/src/firefly_weave/compiler/analyzer.py @@ -30,6 +30,15 @@ ActionConfigIssue, ActionConfigValidator, ) +from firefly_weave.compiler.calls import ( + BUSINESS_KEY_SCHEMA, + CALL_MESSAGES, + CallGraph, + call_output_schema, + check_call_target, + detached_failure_paths, + without_detached_failures, +) from firefly_weave.compiler.catalog import CatalogResource, CatalogSnapshot, FrozenDocument, ResourceKind from firefly_weave.compiler.decision_tables import ( DECISION_MESSAGES, @@ -106,6 +115,8 @@ "LOOP_BUDGET": ( "Nested loops could run more than {limit:,} iterations in one run; lower maxItems on an inner loop." ), + "UNKNOWN_WORKFLOW": CALL_MESSAGES["UNKNOWN_WORKFLOW"], + "CALL_MODE": CALL_MESSAGES["CALL_MODE"], } @@ -263,6 +274,9 @@ def __init__( self.loops: tuple[LoopScope, ...] = () # Every loop as (path, product of the enclosing loops' maxItems, its maxItems, whether it holds no loop). self.loop_chains: list[tuple[str, int, int, bool]] = [] + # The analyzed workflow's exact reference, and the calls each catalog workflow makes. + self.reference = "" + self.calls = CallGraph(catalog) def location(self, path: str) -> SourceRange | None: while path not in self.source.locations and path: @@ -384,6 +398,7 @@ def resolve(self, kind: ResourceKind, reference: str, path: str) -> CatalogResou "Connector": "UNKNOWN_CONNECTOR", "TaskCapability": "UNKNOWN_TASK", "Adapter": "UNKNOWN_ADAPTER", + "Workflow": "UNKNOWN_WORKFLOW", }.get(kind, "UNKNOWN_RESOURCE") self.issue(code, path, stage="resolution") return None @@ -399,6 +414,11 @@ def resolve(self, kind: ResourceKind, reference: str, path: str) -> CatalogResou elif kind == "TaskCapability": self.schema(cast(JsonObject, value["inputSchema"]), path) self.schema(cast(JsonObject, value["outputSchema"]), path) + elif kind == "Workflow": + # A called workflow is an interface: its steps were checked when it was published. + spec = cast(JsonObject, value["spec"]) + self.schema(cast(JsonObject, spec["inputSchema"]), path) + self.schema(cast(JsonObject, spec["outputSchema"]), path) self.checked[key] = before == self.error_count return resource if self.checked[key] else None @@ -568,6 +588,10 @@ def preflight(self, steps: list[JsonObject], path: str, depth: int = 1) -> bool: self.count(cast(JsonObject, step[key]), step_path + "/" + key) if kind == "forEach": self.count(cast(JsonObject, step["items"]), step_path + "/items") + if kind == "callWorkflow": + self.count(cast(JsonObject, step["with"]), step_path + "/with") + if "businessKey" in step: + self.count(cast(JsonObject, step["businessKey"]), step_path + "/businessKey") for _, child, child_path in self.branch_values(step, step_path): if "when" in child: self.count(cast(JsonObject, child["when"]), child_path + "/when") @@ -910,6 +934,8 @@ def block( elif kind == "forEach": output, body = self.loop(step, location, visible, scope) branches.append(body) + elif kind == "callWorkflow": + output = self.call(step, location, visible) elif kind in {"switch", "parallel"}: if kind == "parallel" and cast(int, step["concurrency"]) > self.concurrency: self.issue("CONCURRENCY_LIMIT", location + "/concurrency") @@ -1037,6 +1063,37 @@ def loop( ) return output_schema(body_schema, max_items, cast(str, step["collect"])), branch + def call(self, step: JsonObject, location: str, visible: dict[str, JsonObject]) -> JsonObject: + """Check one callWorkflow step against the workflow it calls; returns the step's output schema.""" + identifier, reference = cast(str, step["id"]), cast(str, step["uses"]) + mode, on_failure = cast(str, step["mode"]), cast(str | None, step.get("onFailure")) + resource = self.resolve("Workflow", reference, location + "/uses") + callee = cast(JsonObject, resource.definition.value["spec"]) if resource else None + if resource is not None: + target = check_call_target( + self.reference, reference, self.catalog, max_call_depth=self.limits.max_call_depth, graph=self.calls + ) + for finding in target.issues: + self.issue(finding.code, location + "/uses", stage=finding.stage, message=finding.message) + input_schema = cast(JsonObject, callee["inputSchema"]) if callee else None + self.expression(cast(JsonObject, step["with"]), location + "/with", visible, input_schema) + if input_schema is not None: + self.guard(location + "/with", "call_input", input_schema) + if "businessKey" in step: + self.expression( + cast(JsonObject, step["businessKey"]), location + "/businessKey", visible, BUSINESS_KEY_SCHEMA + ) + if mode == "detach": + return call_output_schema(mode, on_failure, {}) + if callee is None: + # Offline, the called workflow's output stays unconstrained; the continue envelope around it is known. + if self.catalog_absent and on_failure != "continue": + self.pending_outputs.add(identifier) + return call_output_schema(mode, on_failure, {}) + output = cast(JsonObject, callee["outputSchema"]) + self.guard(location, "call_output", output) + return call_output_schema(mode, on_failure, output) + def loop_items(self, schema: JsonObject, items: JsonObject, path: str) -> None: """Items must be a list: provably not is LOOP_ITEMS; unproved warns and keeps a loop_items guard.""" try: @@ -1131,15 +1188,21 @@ def run(self, *, partial: bool = False) -> AnalysisResult: if isinstance(kind, str) and kind in {"Workflow", "Action", "Connector", "DecisionTable"} else "definition" ) - issues = validate_contract_payload(contract, self.source.value, limits=self.contract_limits) + # onFailure on a call that does not wait is reported at its field, not as an anonymous contract violation. + detached = detached_failure_paths(self.source.value) if contract == "workflow" else [] + for path in detached: + self.issue("CALL_MODE", path) + document = without_detached_failures(self.source.value) if detached else self.source.value + issues = validate_contract_payload(contract, document, limits=self.contract_limits) for issue in issues: self.add(issue) - if issues: + if issues or detached: return self.finish(None, None) model = load_definition(self.source.value) value = model.model_dump(by_alias=True) definition = FrozenDocument.from_value(value) self.kind, self.definition = model.kind, definition + self.reference = f"{model.metadata.name}@{model.metadata.version}" existing = self.catalog.resolve(model.kind, f"{model.metadata.name}@{model.metadata.version}") if existing is not None and existing.digest != definition.digest: self.issue("IMMUTABLE_VERSION", "/metadata/version", stage="resolution") diff --git a/src/firefly_weave/compiler/calls.py b/src/firefly_weave/compiler/calls.py new file mode 100644 index 00000000..a17e299b --- /dev/null +++ b/src/firefly_weave/compiler/calls.py @@ -0,0 +1,241 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + + +"""Workflow calls: which workflow versions a workflow may call, how deep calls go, and what a call step returns. + +A ``callWorkflow`` step names an exact published workflow version. The called version must opt in with +``spec.callable`` (optionally limited to named callers), must not lead back to the calling workflow, and the +longest chain of calls it starts must stay within ``Limits.max_call_depth``. ``check_call_target`` applies these +checks to one call; anything else that starts workflows (such as an AI agent's workflow tool) applies the same +function. +""" + +from __future__ import annotations + +from collections import deque +from collections.abc import Iterator +from copy import deepcopy +from dataclasses import dataclass, field +from typing import Final, Literal, cast + +from firefly_weave.compiler.catalog import CatalogResource, CatalogSnapshot +from firefly_weave.compiler.ir import allows_caller +from firefly_weave.compiler.source_map import pointer_child +from firefly_weave.contracts.language_features import MAX_CALL_DEPTH +from firefly_weave.contracts.values import JsonObject, JsonValue + +# How a waiting call that continues on failure reports the child run's outcome. +CALL_STATUSES: Final = ("succeeded", "failed", "cancelled", "timed_out") +BUSINESS_KEY_SCHEMA: Final[JsonObject] = {"type": "string", "minLength": 1, "maxLength": 200} +CALL_ERROR_SCHEMA: Final[JsonObject] = { + "type": "object", + "properties": {"code": {"type": "string"}, "message": {"type": "string"}}, + "required": ["code", "message"], + "additionalProperties": False, +} +CALL_MESSAGES: Final[dict[str, str]] = { + "UNKNOWN_WORKFLOW": "The catalog does not contain this exact workflow version.", + "CALL_NOT_CALLABLE": ( + "This workflow version does not allow calls from {caller}; " + "add {caller} to its callable.allowedCallers in a new version." + ), + "CALL_RECURSION": "This call would reach the calling workflow again.", + "CALL_DEPTH": "The chain of workflow calls is deeper than {limit:,}.", + "CALL_MODE": "onFailure applies only when the call waits for its result.", +} + + +@dataclass(frozen=True) +class CallIssue: + """One finding about a call target; ``code`` has no ``WV-COMP-`` prefix.""" + + code: str + message: str + stage: Literal["resolution", "semantic"] + + +@dataclass(frozen=True) +class CallTarget: + """The resolved called workflow (``None`` when the catalog lacks it) and every finding about the call.""" + + resource: CatalogResource | None + issues: tuple[CallIssue, ...] + + +@dataclass +class CallGraph: + """The calls each catalog workflow makes, read once per analysis.""" + + catalog: CatalogSnapshot + _calls: dict[str, tuple[str, ...]] = field(default_factory=dict) + + def calls(self, reference: str) -> tuple[str, ...]: + """The exact references ``reference`` calls; none when the catalog lacks it.""" + if reference not in self._calls: + resource = self.catalog.resolve("Workflow", reference) + self._calls[reference] = tuple(called_references(resource.definition.value)) if resource else () + return self._calls[reference] + + def reaches(self, start: str, target: str) -> bool: + """Whether ``start`` is ``target`` or calls it through any chain of calls.""" + seen, pending = {start}, deque([start]) + while pending: + reference = pending.popleft() + if reference == target: + return True + for callee in self.calls(reference): + if callee not in seen: + seen.add(callee) + pending.append(callee) + return False + + def depth(self, start: str) -> int | None: + """The longest chain of calls ``start`` makes (0 when it calls nothing); ``None`` when calls form a cycle.""" + state: dict[str, bool] = {start: False} + order: list[str] = [] + stack: list[tuple[str, Iterator[str]]] = [(start, iter(self.calls(start)))] + while stack: + reference, callees = stack[-1] + callee = next(callees, None) + if callee is None: + stack.pop() + state[reference] = True + order.append(reference) + elif callee not in state: + state[callee] = False + stack.append((callee, iter(self.calls(callee)))) + elif not state[callee]: + return None + depths: dict[str, int] = {} + for reference in order: + depths[reference] = max((depths[callee] + 1 for callee in self.calls(reference)), default=0) + return depths[start] + + +def workflow_steps(document: JsonValue) -> Iterator[tuple[JsonObject, str]]: + """Every step of a workflow definition with its path, blocks included; never recursive, never validating.""" + spec = document.get("spec") if isinstance(document, dict) else None + if not isinstance(spec, dict): + return + blocks: deque[tuple[JsonValue, str]] = deque([(spec.get("steps"), "/spec/steps")]) + while blocks: + steps, path = blocks.popleft() + for index, step in enumerate(steps if isinstance(steps, list) else []): + if not isinstance(step, dict): + continue + location = f"{path}/{index}" + yield step, location + kind = step.get("kind") + if kind == "switch": + cases = step.get("cases") + for number, case in enumerate(cases if isinstance(cases, list) else []): + if isinstance(case, dict): + blocks.append((case.get("steps"), f"{location}/cases/{number}/steps")) + if isinstance(step.get("default"), dict): + blocks.append((cast(JsonObject, step["default"]).get("steps"), location + "/default/steps")) + elif kind == "parallel" and isinstance(step.get("branches"), dict): + for name, branch in sorted(cast(JsonObject, step["branches"]).items()): + if isinstance(branch, dict): + blocks.append((branch.get("steps"), pointer_child(location + "/branches", name) + "/steps")) + elif kind == "forEach" and isinstance(step.get("body"), dict): + blocks.append((cast(JsonObject, step["body"]).get("steps"), location + "/body/steps")) + + +def called_references(document: JsonValue) -> list[str]: + """The exact workflow references a workflow definition calls, in document order.""" + return [ + cast(str, step["uses"]) + for step, _ in workflow_steps(document) + if step.get("kind") == "callWorkflow" and isinstance(step.get("uses"), str) + ] + + +def detached_failure_paths(document: JsonValue) -> list[str]: + """Paths of ``onFailure`` on calls that do not wait for their result, which the definition model rejects.""" + return [ + path + "/onFailure" + for step, path in workflow_steps(document) + if step.get("kind") == "callWorkflow" and step.get("mode") == "detach" and "onFailure" in step + ] + + +def without_detached_failures(document: JsonObject) -> JsonObject: + """A copy of a workflow definition without ``onFailure`` on calls that do not wait for their result.""" + copy = deepcopy(document) + for step, _ in workflow_steps(copy): + if step.get("kind") == "callWorkflow" and step.get("mode") == "detach": + step.pop("onFailure", None) + return copy + + +def check_call_target( + caller: str, + reference: str, + catalog: CatalogSnapshot, + *, + max_call_depth: int = MAX_CALL_DEPTH, + graph: CallGraph | None = None, +) -> CallTarget: + """Resolve the workflow ``caller`` (``name@version``) calls at ``reference`` and check that the call is allowed. + + Findings, in this order: the catalog lacks the exact version (``UNKNOWN_WORKFLOW``, nothing else is checked); + the version does not allow calls from the caller's name (``CALL_NOT_CALLABLE``); the call leads back to the + caller's exact version (``CALL_RECURSION``); or the chain of calls this call starts is longer than + ``max_call_depth`` or never ends (``CALL_DEPTH``). + """ + resource = catalog.resolve("Workflow", reference) + if resource is None: + return CallTarget(None, (CallIssue("UNKNOWN_WORKFLOW", CALL_MESSAGES["UNKNOWN_WORKFLOW"], "resolution"),)) + graph = graph or CallGraph(catalog) + name = caller.partition("@")[0] + issues: list[CallIssue] = [] + spec = resource.definition.value["spec"] + if not allows_caller(cast(JsonObject, spec).get("callable"), name): + message = CALL_MESSAGES["CALL_NOT_CALLABLE"].format(caller=name) + issues.append(CallIssue("CALL_NOT_CALLABLE", message, "resolution")) + if graph.reaches(reference, caller): + issues.append(CallIssue("CALL_RECURSION", CALL_MESSAGES["CALL_RECURSION"], "semantic")) + else: + depth = graph.depth(reference) + if depth is None or depth + 1 > max_call_depth: + message = CALL_MESSAGES["CALL_DEPTH"].format(limit=max_call_depth) + issues.append(CallIssue("CALL_DEPTH", message, "semantic")) + return CallTarget(resource, tuple(issues)) + + +def continue_envelope(output: JsonObject) -> JsonObject: + """The output of a waiting call that continues when the child run does not succeed.""" + return { + "type": "object", + "properties": { + "status": {"type": "string", "enum": list(CALL_STATUSES)}, + "runId": {"type": "string", "format": "uuid"}, + "output": output, + "error": CALL_ERROR_SCHEMA, + }, + "required": ["status", "runId"], + "additionalProperties": False, + } + + +def call_output_schema(mode: str, on_failure: str | None, callee_output: JsonObject) -> JsonObject: + """A call step's output: the called workflow's output, the continue envelope, or null for a detached call.""" + if mode == "detach": + return {"type": "null"} + if on_failure == "continue": + return continue_envelope(callee_output) + return callee_output diff --git a/src/firefly_weave/compiler/ir.py b/src/firefly_weave/compiler/ir.py index 7098b935..ce2949b4 100644 --- a/src/firefly_weave/compiler/ir.py +++ b/src/firefly_weave/compiler/ir.py @@ -33,27 +33,19 @@ from firefly_weave.contracts.definitions import ( ActionDefinition, ActionSpec, - ActionStep, ArrayExpression, ConnectionRequirement, ConnectorDefinition, ConnectorSpec, ContractModel, DecisionTableSpec, - DecisionTableStep, Expression, - ForEachStep, - HumanTaskStep, JsonPointer, - LLMStep, Metadata, ObjectExpression, OpExpression, - ParallelStep, PositiveInt, ResourceName, - SignalStep, - SwitchStep, WorkerImplementation, WorkflowSpec, load_definition, @@ -109,6 +101,8 @@ class Guard(ContractModel): "human_output", "branch_output", "loop_items", + "call_input", + "call_output", "workflow_input", "workflow_output", ] @@ -178,6 +172,25 @@ class FailNode(Node): message: Annotated[UnicodeString, Field(min_length=1)] +class CallWorkflowNode(Node): + """Starts the exact workflow version ``dependency`` names as a child run with the input ``with``.""" + + kind: Literal["callWorkflow"] + dependency: Digest + input: Expression = Field(alias="with") + mode: Literal["wait", "detach"] + on_failure: Literal["stop", "continue"] | None = Field( + default=None, alias="onFailure", exclude_if=lambda value: value is None + ) + business_key: Expression | None = Field(default=None, alias="businessKey", exclude_if=lambda value: value is None) + + @model_validator(mode="after") + def failure_needs_wait(self) -> CallWorkflowNode: + if self.mode == "detach" and self.on_failure is not None: + raise ValueError("onFailure applies only when the call waits for its result") + return self + + class SwitchCase(ContractModel): branch: UnicodeString when: Expression @@ -241,6 +254,7 @@ class JoinNode(Node): | SignalNode | HumanTaskNode | FailNode + | CallWorkflowNode | SwitchNode | ParallelNode | ForEachNode @@ -369,7 +383,17 @@ def topology(self) -> IRGraph: if not node.completes and actual: raise ValueError("Noncompleting join has successor") if isinstance( - node, (StartNode, ActionNode, DecisionTableNode, TransformNode, WaitNode, SignalNode, JoinNode) + node, + ( + StartNode, + ActionNode, + DecisionTableNode, + TransformNode, + WaitNode, + SignalNode, + CallWorkflowNode, + JoinNode, + ), ): expected_count = 0 if isinstance(node, JoinNode) and not node.completes else 1 if len(actual) != expected_count or any( @@ -469,6 +493,11 @@ def workflow_requirements(graph: IRGraph) -> tuple[str, list[LanguageFeature]]: elif isinstance(node, ForEachNode): features.add("flow.forEach") expressions.append(node.items) + elif isinstance(node, CallWorkflowNode): + features.add("flow.callWorkflow") + expressions.append(node.input) + if node.business_key is not None: + expressions.append(node.business_key) while expressions: expression = expressions.pop() if isinstance(expression, OpExpression): @@ -485,6 +514,14 @@ def workflow_requirements(graph: IRGraph) -> tuple[str, list[LanguageFeature]]: return IR_LEVELS[3 if features else level], sorted(features) +def allows_caller(callable_field: JsonValue, caller: str) -> bool: + """Whether a workflow's ``callable`` field (definition JSON; ``None`` when absent) lets ``caller`` call it.""" + if not isinstance(callable_field, dict): + return False + allowed = callable_field.get("allowedCallers") + return allowed is None or (isinstance(allowed, list) and caller in allowed) + + def accepted_ir_versions(features: Collection[str]) -> list[str]: """The IR versions a platform running ``features`` accepts: v1alpha4 joins with the first feature.""" return list(IR_LEVELS if features else IR_LEVELS[:3]) @@ -584,11 +621,18 @@ def workflow_references(self) -> WorkflowIR: raise ValueError("Unknown workflow schema") actions = {d.digest for d in self.dependencies if d.kind == "Action"} tables = {d.digest for d in self.dependencies if d.kind == "DecisionTable"} + workflows = {d.digest: d.document for d in self.dependencies if d.kind == "Workflow"} for node in self.graph.nodes: if isinstance(node, ActionNode) and node.dependency not in actions: raise ValueError("Unknown action dependency") if isinstance(node, DecisionTableNode) and node.dependency not in tables: raise ValueError("Unknown decision table dependency") + if isinstance(node, CallWorkflowNode): + if node.dependency not in workflows: + raise ValueError("Unknown called workflow dependency") + callee = cast(JsonObject, workflows[node.dependency]["spec"]) + if not allows_caller(callee.get("callable"), self.metadata.name): + raise ValueError("The called workflow does not allow calls from this workflow") if isinstance(node, SignalNode) and node.schema_ref not in self.schemas: raise ValueError("Unknown signal schema") return self @@ -692,31 +736,9 @@ def binding( for descriptor in spec.actions.values(): schema_roots.extend((descriptor.input_schema, descriptor.output_schema)) elif isinstance(spec, WorkflowSpec): + # A workflow dependency is a called workflow's interface: its steps were checked when it was published + # and run in its own child run, so only its input and output schemas belong to this executable. schema_roots.extend((spec.input_schema, spec.output_schema)) - connections(spec.connections) - pending = list(spec.steps) - while pending: - step = pending.pop() - if isinstance(step, (ActionStep, LLMStep)): - require("Action", step.uses) - action = definitions[("Action", step.uses)] - assert isinstance(action, ActionDefinition) - binding(step.connection, action.spec.connection, spec.connections) - elif isinstance(step, DecisionTableStep): - require("DecisionTable", step.uses) - elif isinstance(step, HumanTaskStep): - schema_roots.append(step.form_schema) - elif isinstance(step, SignalStep): - schema_roots.append(step.payload_schema) - elif isinstance(step, SwitchStep): - pending.extend(step.default.steps) - for case in step.cases: - pending.extend(case.steps) - elif isinstance(step, ParallelStep): - for branch in step.branches.values(): - pending.extend(branch.steps) - elif isinstance(step, ForEachStep): - pending.extend(step.body.steps) if isinstance(executable, WorkflowIR): connections(executable.connections) actions = { diff --git a/src/firefly_weave/compiler/language_support.py b/src/firefly_weave/compiler/language_support.py index 2ae38d87..98413caa 100644 --- a/src/firefly_weave/compiler/language_support.py +++ b/src/firefly_weave/compiler/language_support.py @@ -29,7 +29,9 @@ from firefly_weave.contracts.language_features import KIND_FEATURES, OPERATOR_FEATURES, LanguageFeature from firefly_weave.contracts.values import JsonObject, JsonValue -COMPILED_FEATURES: Final[frozenset[LanguageFeature]] = frozenset({"flow.forEach", "text.concat", "text.join"}) +COMPILED_FEATURES: Final[frozenset[LanguageFeature]] = frozenset( + {"flow.callWorkflow", "flow.forEach", "text.concat", "text.join"} +) # Expression positions per step kind; blocks are walked separately. _EXPRESSIONS: Final[dict[str, tuple[str, ...]]] = { diff --git a/src/firefly_weave/compiler/lowering.py b/src/firefly_weave/compiler/lowering.py index 51b3b817..23df1c95 100644 --- a/src/firefly_weave/compiler/lowering.py +++ b/src/firefly_weave/compiler/lowering.py @@ -89,6 +89,7 @@ def __init__(self, result: AnalysisResult, limits: ArtifactLimits) -> None: self.edges: list[JsonValue] = [] self.actions = {r.reference: r.digest for r in result.resolved_resources if r.kind == "Action"} self.tables = {r.reference: r.digest for r in result.resolved_resources if r.kind == "DecisionTable"} + self.workflows = {r.reference: r.digest for r in result.resolved_resources if r.kind == "Workflow"} def schema(self, document: FrozenDocument) -> str: digest = document.digest @@ -198,6 +199,15 @@ def block(self, steps: tuple[AnalyzedStep, ...], successor: str) -> str: ) elif step.kind == "decisionTable": base.update({"dependency": self.tables[cast(str, value["uses"])], "with": value["with"]}) + elif step.kind == "callWorkflow": + base.update( + { + "dependency": self.workflows[cast(str, value["uses"])], + "with": value["with"], + "mode": value["mode"], + } + ) + base.update({key: value[key] for key in ("onFailure", "businessKey") if key in value}) elif step.kind == "signal": base.update( { diff --git a/src/firefly_weave/contracts/catalog.py b/src/firefly_weave/contracts/catalog.py index 6c0c3030..df8096e9 100644 --- a/src/firefly_weave/contracts/catalog.py +++ b/src/firefly_weave/contracts/catalog.py @@ -22,7 +22,7 @@ from pydantic import Field, model_validator from firefly_weave.contracts.access import Scope -from firefly_weave.contracts.definitions import ContractModel, ResourceName, SemVer +from firefly_weave.contracts.definitions import ContractModel, ResourceName, SemVer, VersionedReference from firefly_weave.contracts.human_tasks import AssignmentPin from firefly_weave.contracts.values import JsonObjectData from firefly_weave.contracts.workers import ConnectorExecutionPin @@ -67,6 +67,8 @@ class ActivationRequest(ContractModel): connection_revision_ids: dict[ResourceName, UUID] = Field(default_factory=dict) connector_release_ids: dict[UUID, UUID] = Field(default_factory=dict, exclude_if=lambda v: not v) worker_release_ids: dict[ResourceName, UUID] = Field(default_factory=dict) + # One activation of each workflow version this workflow calls, in this environment. + workflow_activation_ids: dict[VersionedReference, UUID] = Field(default_factory=dict, exclude_if=lambda v: not v) @model_validator(mode="after") def environment_required(self) -> "ActivationRequest": @@ -75,6 +77,17 @@ def environment_required(self) -> "ActivationRequest": return self +class WorkflowCallPin(ContractModel): + """The activation a call to ``reference`` starts its child runs with, fixed when the caller is activated.""" + + reference: VersionedReference + activation_id: UUID + activation_revision: int = Field(ge=1) + version_id: UUID + artifact_digest: Digest + definition_digest: Digest + + class Activation(ContractModel): assignment_pins: dict[ResourceName, AssignmentPin] = Field(default_factory=dict, exclude_if=lambda v: not v) connector_execution_pins: list[ConnectorExecutionPin] = Field(default_factory=list, exclude_if=lambda v: not v) @@ -83,3 +96,6 @@ class Activation(ContractModel): name: ResourceName request: ActivationRequest retired: bool = False + # The called workflows' activations, by reference, and the longest chain of calls this activation can start. + workflow_call_pins: list[WorkflowCallPin] = Field(default_factory=list, exclude_if=lambda v: not v) + call_depth: int = Field(default=0, ge=0, exclude_if=lambda v: v == 0) diff --git a/src/firefly_weave/contracts/language_features.py b/src/firefly_weave/contracts/language_features.py index a2f76ac6..7d4275f1 100644 --- a/src/firefly_weave/contracts/language_features.py +++ b/src/firefly_weave/contracts/language_features.py @@ -40,8 +40,8 @@ ) WORKFLOW_FIELD_FEATURES: Final[Mapping[str, LanguageFeature]] = MappingProxyType({"callable": "flow.callWorkflow"}) -# Language limits the manifest publishes. Limits uses the loop limits as defaults and the compiler enforces them; -# workflow calls enforce the call depth in a later release. +# Language limits the manifest publishes. Limits uses them as defaults; the compiler enforces the loop limits and the +# call depth, and activation enforces the call depth again through the pinned workflows. DEFAULT_LOOP_MAX_ITEMS: Final = 1000 MAX_LOOP_ITEMS: Final = 10_000 MAX_LOOP_DEPTH: Final = 3 diff --git a/src/firefly_weave/contracts/limits.py b/src/firefly_weave/contracts/limits.py index 0551ca61..eb988038 100644 --- a/src/firefly_weave/contracts/limits.py +++ b/src/firefly_weave/contracts/limits.py @@ -18,7 +18,12 @@ from pydantic import BaseModel, ConfigDict, Field -from firefly_weave.contracts.language_features import MAX_LOOP_DEPTH, MAX_LOOP_ITEMS, MAX_RUN_ITERATIONS +from firefly_weave.contracts.language_features import ( + MAX_CALL_DEPTH, + MAX_LOOP_DEPTH, + MAX_LOOP_ITEMS, + MAX_RUN_ITERATIONS, +) class Limits(BaseModel): @@ -36,3 +41,5 @@ class Limits(BaseModel): max_loop_items: int = Field(default=MAX_LOOP_ITEMS, gt=0) max_loop_depth: int = Field(default=MAX_LOOP_DEPTH, gt=0) max_run_iterations: int = Field(default=MAX_RUN_ITERATIONS, gt=0) + # Workflow calls: the longest chain of calls one workflow may start, counting each call once. + max_call_depth: int = Field(default=MAX_CALL_DEPTH, gt=0) diff --git a/src/firefly_weave/definitions/calls.py b/src/firefly_weave/definitions/calls.py new file mode 100644 index 00000000..554089c2 --- /dev/null +++ b/src/firefly_weave/definitions/calls.py @@ -0,0 +1,183 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + + +"""Workflow call pins: the exact activation each call of an activated workflow starts its child runs with. + +An activation of a workflow that calls others pins one activation of every called version in the same environment, +checked by the rules ``call_pins`` applies. Test activations of drafts use the same rules. +``call_tree_side_effect`` summarizes the strongest side effect a run can cause through the pinned calls. +""" + +import json +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass +from typing import Any, cast, get_args +from uuid import UUID + +from firefly_weave.compiler.ir import allows_caller +from firefly_weave.contracts.access import Scope +from firefly_weave.contracts.catalog import Activation, WorkflowCallPin +from firefly_weave.contracts.definitions import SideEffect +from firefly_weave.contracts.language_features import MAX_CALL_DEPTH +from firefly_weave.contracts.values import JsonObject +from firefly_weave.definitions.models import CatalogError +from firefly_weave.definitions.repository import DefinitionRepository + +# Side effects from the weakest to the strongest. +SIDE_EFFECTS: tuple[SideEffect, ...] = get_args(SideEffect.__value__) + +type CallTreeResolver = Callable[[UUID], Awaitable[tuple[JsonObject, Mapping[str, UUID]]]] +type CalleeLoader = Callable[[UUID], Awaitable["PinnedCallee | None"]] + + +@dataclass(frozen=True) +class PinnedCallee: + """A pinned activation as stored, the retirement of its name's latest revision, and its version row.""" + + activation: Activation + latest_retired: bool + version: Mapping[str, Any] + + +def readiness(message: str) -> CatalogError: + return CatalogError(422, "WV-READINESS-CALL", message) + + +def called_dependencies(executable: JsonObject) -> dict[str, str]: + """Each workflow version an executable calls, by reference, with the definition digest it was compiled against.""" + return { + cast(str, dependency["reference"]): cast(str, dependency["digest"]) + for dependency in cast(list[JsonObject], executable.get("dependencies", [])) + if dependency["kind"] == "Workflow" + } + + +async def call_pins( + caller: str, + scope: Scope, + executable: JsonObject, + supplied: Mapping[str, UUID], + load: CalleeLoader, + *, + max_call_depth: int = MAX_CALL_DEPTH, +) -> tuple[list[WorkflowCallPin], int]: + """Check the activations pinned for ``caller``'s calls; returns the pins and the activation's call depth. + + ``supplied`` maps each called reference to an activation ID; ``load`` reads one pinned activation of + ``scope``'s environment (``None`` when there is none), and runs only once the pinned references match the calls. + Every failure is ``WV-READINESS-CALL``. + """ + called = called_dependencies(executable) + if set(supplied) != set(called): + raise readiness("Pin exactly one activation of each workflow version this workflow calls") + pins: list[WorkflowCallPin] = [] + depth = 0 + for reference in sorted(called): + callee = await load(supplied[reference]) + if callee is None or callee.activation.id != supplied[reference] or callee.activation.request.scope != scope: + raise readiness("A pinned workflow activation was not found in this environment") + activation, version = callee.activation, callee.version + name, number = reference.split("@") + if activation.name != name or version["kind"] != "Workflow" or version["version"] != number: + raise readiness("A pinned activation does not run the called workflow version") + if activation.retired or callee.latest_retired or version["retired"]: + raise readiness("A pinned workflow activation or its version is retired") + if version["digest"] != activation.request.artifact_digest: + raise readiness("A pinned activation does not match its published version") + if version["definition_digest"] != called[reference]: + raise readiness("A pinned version differs from the version this workflow was compiled against") + if not allows_caller(cast(JsonObject, version["document"]["spec"]).get("callable"), caller): + raise readiness("A called workflow version does not allow calls from this workflow") + if activation.call_depth + 1 > max_call_depth: + raise readiness(f"The chain of workflow calls is deeper than {max_call_depth:,}") + depth = max(depth, activation.call_depth + 1) + pins.append( + WorkflowCallPin( + reference=reference, + activation_id=activation.id, + activation_revision=activation.revision, + version_id=activation.request.version_id, + artifact_digest=activation.request.artifact_digest, + definition_digest=version["definition_digest"], + ) + ) + return pins, depth + + +async def pinned_callee(repository: DefinitionRepository, scope: Scope, identifier: UUID) -> PinnedCallee | None: + """Load a pinned activation of ``scope``'s environment, the latest revision of its name, and its version.""" + rows = await repository.rows( + "SELECT payload FROM activation_revisions WHERE tenant_id=:tenant AND project_id=:project " + "AND environment_id=:environment AND id=:id", + environment=scope.environment_id, + id=identifier, + ) + if not rows: + return None + activation = Activation.model_validate_json(json.dumps(rows[0]["payload"])) + latest = await repository.rows( + "SELECT payload FROM activation_revisions WHERE tenant_id=:tenant AND project_id=:project " + "AND environment_id=:environment AND name=:name ORDER BY revision DESC LIMIT 1", + environment=scope.environment_id, + name=activation.name, + ) + version = await repository.version(activation.request.version_id) + return PinnedCallee(activation, bool(latest[0]["payload"]["retired"]), version) + + +def side_effect(executable: JsonObject) -> SideEffect: + """The strongest side effect among an executable's actions; ``read_only`` when it has none.""" + strongest: SideEffect = SIDE_EFFECTS[0] + for dependency in cast(list[JsonObject], executable.get("dependencies", [])): + if dependency["kind"] == "Action": + effect = cast(SideEffect, cast(JsonObject, cast(JsonObject, dependency["document"])["spec"])["sideEffect"]) + strongest = max(strongest, effect, key=SIDE_EFFECTS.index) + return strongest + + +async def call_tree_side_effect( + executable: JsonObject, pins: Mapping[str, UUID], resolve: CallTreeResolver +) -> SideEffect: + """The strongest side effect of a workflow and of every workflow its pinned calls start, transitively. + + ``resolve`` returns a pinned activation's executable and its own pins, by reference. + """ + strongest = side_effect(executable) + seen: set[UUID] = set() + pending = list(pins.values()) + while pending: + identifier = pending.pop() + if identifier in seen: + continue + seen.add(identifier) + callee, callee_pins = await resolve(identifier) + strongest = max(strongest, side_effect(callee), key=SIDE_EFFECTS.index) + pending.extend(callee_pins.values()) + return strongest + + +def pinned_call_tree(repository: DefinitionRepository, scope: Scope) -> CallTreeResolver: + """A resolver for ``call_tree_side_effect`` over the activations of ``scope``'s environment.""" + + async def resolve(identifier: UUID) -> tuple[JsonObject, Mapping[str, UUID]]: + callee = await pinned_callee(repository, scope, identifier) + if callee is None: + raise readiness("A pinned workflow activation was not found in this environment") + executable = cast(JsonObject, cast(JsonObject, callee.version["artifact"])["executable"]) + return executable, {pin.reference: pin.activation_id for pin in callee.activation.workflow_call_pins} + + return resolve diff --git a/src/firefly_weave/definitions/service.py b/src/firefly_weave/definitions/service.py index 0161add1..82e93a7c 100644 --- a/src/firefly_weave/definitions/service.py +++ b/src/firefly_weave/definitions/service.py @@ -515,6 +515,15 @@ async def activate( connector_pins = await self.execution_readiness( actor, scope, request, artifact, capability="release.activate", context=context, tx=tx ) + from firefly_weave.definitions.calls import call_pins, pinned_callee + + workflow_call_pins, call_depth = await call_pins( + row["name"], + scope, + artifact.executable, + request.workflow_activation_ids, + lambda identifier: pinned_callee(repository, scope, identifier), + ) from firefly_weave.compiler.ir import ActionNode, HumanTaskNode from firefly_weave.human_tasks.service import pin_assignments from firefly_weave.runtime.kernel import workflow @@ -556,6 +565,8 @@ async def activate( request=request, connector_execution_pins=connector_pins, assignment_pins=assignment_pins, + workflow_call_pins=workflow_call_pins, + call_depth=call_depth, ) await self._insert_activation(repository, activation) for slot, identifier in request.connection_revision_ids.items(): @@ -899,6 +910,8 @@ async def retire_activation( retired=True, connector_execution_pins=selected.connector_execution_pins, assignment_pins=selected.assignment_pins, + workflow_call_pins=selected.workflow_call_pins, + call_depth=selected.call_depth, ) await self._insert_activation(repository, activation) await audit( diff --git a/src/firefly_weave/runtime/kernel.py b/src/firefly_weave/runtime/kernel.py index dfa85282..20dc8f3b 100644 --- a/src/firefly_weave/runtime/kernel.py +++ b/src/firefly_weave/runtime/kernel.py @@ -31,6 +31,7 @@ from firefly_weave.compiler.ir import ( ActionNode, BranchOutputNode, + CallWorkflowNode, DecisionTableNode, EndNode, FailNode, @@ -813,6 +814,10 @@ def accept(completion: RuntimeEvent) -> None: update={"output": {"code": node.code, "message": node.message}} ) return checkpoint(result.model_copy(update={"state": current_state})) + elif isinstance(node, CallWorkflowNode): + # This runtime does not start child runs: import refuses workflows that call others, and a call + # that still arrives here stops behind an incident instead of being skipped. + raise KernelError("WV-RUNTIME-ARTIFACT") elif isinstance(node, (SwitchNode, ParallelNode)): if current in current_state.joins: raise KernelError("WV-RUNTIME-ACTIVATION") diff --git a/src/firefly_weave/sdk/builder.py b/src/firefly_weave/sdk/builder.py index a308c217..ac05eef3 100644 --- a/src/firefly_weave/sdk/builder.py +++ b/src/firefly_weave/sdk/builder.py @@ -97,6 +97,10 @@ def with_timeout(self, seconds: PositiveInt) -> Self: """Set a positive timeout; omission is the constructor's default.""" return self._with_spec(timeoutSeconds=seconds) + def with_callable(self, allowed_callers: list[ResourceName] | None = None) -> Self: + """Let other workflows call this version: any workflow in the project, or only ``allowed_callers``.""" + return self._with_spec(callable={} if allowed_callers is None else {"allowedCallers": list(allowed_callers)}) + def to_definition(self) -> WorkflowDefinition: """Return a validated, detached canonical model.""" return _snapshot(self._definition) diff --git a/src/firefly_weave/sdk/visualization.py b/src/firefly_weave/sdk/visualization.py index 0314f572..593496c2 100644 --- a/src/firefly_weave/sdk/visualization.py +++ b/src/firefly_weave/sdk/visualization.py @@ -37,6 +37,7 @@ "switch": "Choose a branch", "parallel": "Run branches", "forEach": "Loop over items", + "callWorkflow": "Call a workflow", "branch-output": "Collect branch result", "join": "Continue after branches", "fail": "Stop with a failure", diff --git a/studio/src/app/designer/diagnostic-copy.ts b/studio/src/app/designer/diagnostic-copy.ts index 2b47c864..fc69aaa7 100644 --- a/studio/src/app/designer/diagnostic-copy.ts +++ b/studio/src/app/designer/diagnostic-copy.ts @@ -96,6 +96,26 @@ const compiler: Record = { UNKNOWN_RESOURCE: { text: "Something this definition depends on isn't in the project catalog.", }, + UNKNOWN_WORKFLOW: { + text: "This workflow version isn't published in the project.", + hint: "Check its name and version, or publish that workflow first.", + }, + CALL_NOT_CALLABLE: { + text: "That workflow version doesn't accept calls from this workflow.", + hint: "Publish a new version of it with the Called by a workflow trigger that allows this workflow.", + }, + CALL_RECURSION: { + text: "This call would end up calling this workflow again.", + hint: "Call a workflow that doesn't lead back to this one.", + }, + CALL_DEPTH: { + text: "Workflows would call each other more levels deep than the platform allows.", + hint: "Call a workflow that starts fewer workflows of its own.", + }, + CALL_MODE: { + text: "This step can continue after the called run fails only when it waits for the called workflow's result.", + hint: "Set When it runs to Wait for its result, or clear If the called run fails.", + }, CATALOG_PENDING: { text: "This is checked against the project catalog when you validate while connected to a platform.", }, diff --git a/studio/tests/diagnostic-copy.test.ts b/studio/tests/diagnostic-copy.test.ts index 58a36976..a100555d 100644 --- a/studio/tests/diagnostic-copy.test.ts +++ b/studio/tests/diagnostic-copy.test.ts @@ -63,6 +63,11 @@ const compilerCodes = [ "LOOP_MAX_ITEMS", "LOOP_BUDGET", "LOOP_FANOUT", + "UNKNOWN_WORKFLOW", + "CALL_NOT_CALLABLE", + "CALL_RECURSION", + "CALL_DEPTH", + "CALL_MODE", ].map((code) => `WV-COMP-${code}`); describe("plain-language diagnostic copy", () => { @@ -80,6 +85,19 @@ describe("plain-language diagnostic copy", () => { expect(view.technical).toBe("Definition violates the contract."); } }); + it("names the call settings by their Studio labels for CALL_MODE", () => { + const view = describeDiagnostic({ + code: "WV-COMP-CALL_MODE", + severity: "error", + message: "onFailure applies only to mode wait.", + }); + expect(view.text).toBe( + "This step can continue after the called run fails only when it waits for the called workflow's result.", + ); + expect(view.hint).toBe( + "Set When it runs to Wait for its result, or clear If the called run fails.", + ); + }); it("explains the dominance error in the author's terms", () => { const view = describeDiagnostic({ code: "WV-COMP-UNAVAILABLE_REFERENCE", @@ -210,12 +228,13 @@ describe.skipIf(!pythonAvailable())( "-c", ` import inspect, json, re -from firefly_weave.compiler import analyzer, api, action_config -text = inspect.getsource(analyzer) + inspect.getsource(api) +from firefly_weave.compiler import analyzer, api, action_config, calls +text = inspect.getsource(analyzer) + inspect.getsource(api) + inspect.getsource(calls) codes = set(re.findall(r'(?:self\\.issue|issue)\\(\\s*"([A-Z_]+)"', text)) codes |= set(re.findall(r'"(UNKNOWN_[A-Z_]+)"', text)) codes |= set(re.findall(r'WV-COMP-([A-Z_]+)', text)) codes |= set(action_config.ACTION_CONFIG_CODES) +codes |= set(calls.CALL_MESSAGES) print(json.dumps(sorted("WV-COMP-" + c for c in codes))) `, ], diff --git a/studio/tests/fixtures/language/scope.json b/studio/tests/fixtures/language/scope.json index b50c7f2f..cdd0fd9f 100644 --- a/studio/tests/fixtures/language/scope.json +++ b/studio/tests/fixtures/language/scope.json @@ -2096,6 +2096,1096 @@ "truncated": false, "entries": [] } + }, + { + "name": "a call's input sees what is visible before the call", + "document": { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": { + "name": "scope-cases", + "version": "1.0.0" + }, + "spec": { + "inputSchema": { + "type": "object", + "properties": { + "invoices": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "batches": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + } + }, + "required": [ + "invoices", + "batches" + ], + "additionalProperties": false + }, + "outputSchema": {}, + "steps": [ + { + "id": "notify", + "kind": "callWorkflow", + "uses": "notify-customer@1.0.0", + "with": { + "ref": "/input" + } + } + ], + "output": { + "literal": null + } + } + }, + "path": "/spec/steps/0/with", + "result": { + "found": true, + "evaluated": true, + "truncated": false, + "entries": [ + { + "ref": "/input", + "source": "input", + "path": [], + "label": "Workflow input", + "breadcrumb": "Workflow input", + "schema": { + "type": "object", + "properties": { + "invoices": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "batches": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + } + }, + "required": [ + "invoices", + "batches" + ], + "additionalProperties": false + }, + "types": [ + "object" + ], + "typeLabel": "Group", + "optional": false + }, + { + "ref": "/input/invoices", + "source": "input", + "path": [ + "invoices" + ], + "label": "invoices", + "breadcrumb": "Workflow input › invoices", + "schema": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "types": [ + "array" + ], + "typeLabel": "List of groups", + "optional": false + }, + { + "ref": "/input/batches", + "source": "input", + "path": [ + "batches" + ], + "label": "batches", + "breadcrumb": "Workflow input › batches", + "schema": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + }, + "types": [ + "array" + ], + "typeLabel": "List of lists", + "optional": false + } + ] + } + }, + { + "name": "after a call that waits, its result is the called workflow's output, unknown without a catalog", + "document": { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": { + "name": "scope-cases", + "version": "1.0.0" + }, + "spec": { + "inputSchema": { + "type": "object", + "properties": { + "invoices": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "batches": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + } + }, + "required": [ + "invoices", + "batches" + ], + "additionalProperties": false + }, + "outputSchema": {}, + "steps": [ + { + "id": "notify", + "kind": "callWorkflow", + "uses": "notify-customer@1.0.0", + "with": { + "ref": "/input" + } + } + ], + "output": { + "ref": "/steps/notify/output" + } + } + }, + "path": "/spec/output", + "result": { + "found": true, + "evaluated": true, + "truncated": false, + "entries": [ + { + "ref": "/input", + "source": "input", + "path": [], + "label": "Workflow input", + "breadcrumb": "Workflow input", + "schema": { + "type": "object", + "properties": { + "invoices": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "batches": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + } + }, + "required": [ + "invoices", + "batches" + ], + "additionalProperties": false + }, + "types": [ + "object" + ], + "typeLabel": "Group", + "optional": false + }, + { + "ref": "/input/invoices", + "source": "input", + "path": [ + "invoices" + ], + "label": "invoices", + "breadcrumb": "Workflow input › invoices", + "schema": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "types": [ + "array" + ], + "typeLabel": "List of groups", + "optional": false + }, + { + "ref": "/input/batches", + "source": "input", + "path": [ + "batches" + ], + "label": "batches", + "breadcrumb": "Workflow input › batches", + "schema": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + }, + "types": [ + "array" + ], + "typeLabel": "List of lists", + "optional": false + }, + { + "ref": "/steps/notify/output", + "source": "step", + "stepId": "notify", + "stepKind": "callWorkflow", + "path": [], + "label": "notify", + "breadcrumb": "notify", + "schema": {}, + "types": [], + "typeLabel": "Any value", + "optional": false + } + ] + } + }, + { + "name": "a call that continues on failure returns its status, run ID, the child's output and the error", + "document": { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": { + "name": "scope-cases", + "version": "1.0.0" + }, + "spec": { + "inputSchema": { + "type": "object", + "properties": { + "invoices": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "batches": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + } + }, + "required": [ + "invoices", + "batches" + ], + "additionalProperties": false + }, + "outputSchema": {}, + "steps": [ + { + "id": "notify", + "kind": "callWorkflow", + "uses": "notify-customer@1.0.0", + "with": { + "ref": "/input" + }, + "onFailure": "continue" + } + ], + "output": { + "ref": "/steps/notify/output" + } + } + }, + "path": "/spec/output", + "result": { + "found": true, + "evaluated": true, + "truncated": false, + "entries": [ + { + "ref": "/input", + "source": "input", + "path": [], + "label": "Workflow input", + "breadcrumb": "Workflow input", + "schema": { + "type": "object", + "properties": { + "invoices": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "batches": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + } + }, + "required": [ + "invoices", + "batches" + ], + "additionalProperties": false + }, + "types": [ + "object" + ], + "typeLabel": "Group", + "optional": false + }, + { + "ref": "/input/invoices", + "source": "input", + "path": [ + "invoices" + ], + "label": "invoices", + "breadcrumb": "Workflow input › invoices", + "schema": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "types": [ + "array" + ], + "typeLabel": "List of groups", + "optional": false + }, + { + "ref": "/input/batches", + "source": "input", + "path": [ + "batches" + ], + "label": "batches", + "breadcrumb": "Workflow input › batches", + "schema": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + }, + "types": [ + "array" + ], + "typeLabel": "List of lists", + "optional": false + }, + { + "ref": "/steps/notify/output", + "source": "step", + "stepId": "notify", + "stepKind": "callWorkflow", + "path": [], + "label": "notify", + "breadcrumb": "notify", + "schema": { + "type": "object", + "properties": { + "status": { + "type": "string", + "enum": [ + "succeeded", + "failed", + "cancelled", + "timed_out" + ] + }, + "runId": { + "type": "string", + "format": "uuid" + }, + "output": {}, + "error": { + "type": "object", + "properties": { + "code": { + "type": "string" + }, + "message": { + "type": "string" + } + }, + "required": [ + "code", + "message" + ], + "additionalProperties": false + } + }, + "required": [ + "status", + "runId" + ], + "additionalProperties": false + }, + "types": [ + "object" + ], + "typeLabel": "Group", + "optional": false + }, + { + "ref": "/steps/notify/output/status", + "source": "step", + "stepId": "notify", + "stepKind": "callWorkflow", + "path": [ + "status" + ], + "label": "status", + "breadcrumb": "notify › status", + "schema": { + "type": "string", + "enum": [ + "succeeded", + "failed", + "cancelled", + "timed_out" + ] + }, + "types": [ + "string" + ], + "typeLabel": "Choice", + "optional": false + }, + { + "ref": "/steps/notify/output/runId", + "source": "step", + "stepId": "notify", + "stepKind": "callWorkflow", + "path": [ + "runId" + ], + "label": "runId", + "breadcrumb": "notify › runId", + "schema": { + "type": "string", + "format": "uuid" + }, + "types": [ + "string" + ], + "typeLabel": "Text", + "optional": false + }, + { + "ref": "/steps/notify/output/output", + "source": "step", + "stepId": "notify", + "stepKind": "callWorkflow", + "path": [ + "output" + ], + "label": "output", + "breadcrumb": "notify › output", + "schema": {}, + "types": [], + "typeLabel": "Any value", + "optional": true + }, + { + "ref": "/steps/notify/output/error", + "source": "step", + "stepId": "notify", + "stepKind": "callWorkflow", + "path": [ + "error" + ], + "label": "error", + "breadcrumb": "notify › error", + "schema": { + "type": "object", + "properties": { + "code": { + "type": "string" + }, + "message": { + "type": "string" + } + }, + "required": [ + "code", + "message" + ], + "additionalProperties": false + }, + "types": [ + "object" + ], + "typeLabel": "Group", + "optional": true + }, + { + "ref": "/steps/notify/output/error/code", + "source": "step", + "stepId": "notify", + "stepKind": "callWorkflow", + "path": [ + "error", + "code" + ], + "label": "code", + "breadcrumb": "notify › error › code", + "schema": { + "type": "string" + }, + "types": [ + "string" + ], + "typeLabel": "Text", + "optional": true + }, + { + "ref": "/steps/notify/output/error/message", + "source": "step", + "stepId": "notify", + "stepKind": "callWorkflow", + "path": [ + "error", + "message" + ], + "label": "message", + "breadcrumb": "notify › error › message", + "schema": { + "type": "string" + }, + "types": [ + "string" + ], + "typeLabel": "Text", + "optional": true + } + ] + } + }, + { + "name": "a call that does not wait returns nothing", + "document": { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": { + "name": "scope-cases", + "version": "1.0.0" + }, + "spec": { + "inputSchema": { + "type": "object", + "properties": { + "invoices": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "batches": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + } + }, + "required": [ + "invoices", + "batches" + ], + "additionalProperties": false + }, + "outputSchema": {}, + "steps": [ + { + "id": "notify", + "kind": "callWorkflow", + "uses": "notify-customer@1.0.0", + "with": { + "ref": "/input" + }, + "mode": "detach" + } + ], + "output": { + "ref": "/steps/notify/output" + } + } + }, + "path": "/spec/output", + "result": { + "found": true, + "evaluated": true, + "truncated": false, + "entries": [ + { + "ref": "/input", + "source": "input", + "path": [], + "label": "Workflow input", + "breadcrumb": "Workflow input", + "schema": { + "type": "object", + "properties": { + "invoices": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "batches": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + } + }, + "required": [ + "invoices", + "batches" + ], + "additionalProperties": false + }, + "types": [ + "object" + ], + "typeLabel": "Group", + "optional": false + }, + { + "ref": "/input/invoices", + "source": "input", + "path": [ + "invoices" + ], + "label": "invoices", + "breadcrumb": "Workflow input › invoices", + "schema": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + }, + "types": [ + "array" + ], + "typeLabel": "List of groups", + "optional": false + }, + { + "ref": "/input/batches", + "source": "input", + "path": [ + "batches" + ], + "label": "batches", + "breadcrumb": "Workflow input › batches", + "schema": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": { + "number": { + "type": "string" + }, + "customerEmail": { + "type": "string", + "title": "Customer email" + } + }, + "required": [ + "number" + ], + "additionalProperties": false + } + } + }, + "types": [ + "array" + ], + "typeLabel": "List of lists", + "optional": false + }, + { + "ref": "/steps/notify/output", + "source": "step", + "stepId": "notify", + "stepKind": "callWorkflow", + "path": [], + "label": "notify", + "breadcrumb": "notify", + "schema": { + "type": "null" + }, + "types": [ + "null" + ], + "typeLabel": "Empty", + "optional": false + } + ] + } } ] } diff --git a/tests/acceptance/journeys.toml b/tests/acceptance/journeys.toml index f13d8273..a49d652f 100644 --- a/tests/acceptance/journeys.toml +++ b/tests/acceptance/journeys.toml @@ -45,7 +45,7 @@ "loop-compiler" = true "loop-runtime" = true "loop-simulation" = false -"workflow-call-compiler" = false +"workflow-call-compiler" = true "workflow-call-runtime" = false "workflow-call-simulation" = false "workflow-call-fast-path" = false diff --git a/tests/integration/test_workflow_call_activation.py b/tests/integration/test_workflow_call_activation.py new file mode 100644 index 00000000..629d8fe9 --- /dev/null +++ b/tests/integration/test_workflow_call_activation.py @@ -0,0 +1,106 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + + +"""Activating a workflow that calls others pins one live activation of each called workflow version.""" + +import json +from pathlib import Path + +import pytest +from sqlalchemy import text + +from firefly_weave.access.audit import AuditContext +from firefly_weave.compiler.parser import parse_source +from firefly_weave.contracts import language_features +from firefly_weave.contracts.catalog import Activation, ActivationRequest + +pytestmark = pytest.mark.integration + +EXAMPLES = Path("examples/language") +CALLS = ("flow.callWorkflow", "text.concat", "text.join") + + +def source(name): + return json.dumps(parse_source((EXAMPLES / name).read_text(), format="yaml").value) + + +async def stored(access_db, identifier): + async with access_db[1]() as session: + payload = await session.scalar( + text("SELECT payload FROM activation_revisions WHERE id=:id"), {"id": identifier} + ) + return Activation.model_validate_json(json.dumps(payload)) + + +async def publish_and_activate(definitions, actor, scope, name, key, **pins): + published = await definitions.publish(actor, scope, "Workflow", source(name), "json", key, context=AuditContext()) + request = ActivationRequest(scope=scope, version_id=published.id, artifact_digest=published.digest, **pins) + return published, request + + +@pytest.fixture +def platform(worker_setup, services, access_db): + from firefly_weave.definitions.service import DefinitionService + + actor, scope = worker_setup[3:5] + return services(access_db[0]).resolve(DefinitionService), actor, scope + + +async def test_activation_pins_one_live_activation_of_each_called_workflow(platform, access_db, monkeypatch): + from firefly_weave.definitions.calls import call_tree_side_effect, pinned_call_tree + from firefly_weave.definitions.models import CatalogError + from firefly_weave.definitions.repository import DefinitionRepository + + monkeypatch.setattr(language_features, "ADVERTISED_FEATURES", CALLS) + definitions, actor, scope = platform + _, callee_request = await publish_and_activate(definitions, actor, scope, "notify-customer.workflow.yaml", "callee") + callee = await definitions.activate(actor, scope, callee_request, "callee", context=AuditContext()) + assert (callee.workflow_call_pins, callee.call_depth) == ([], 0) + published, request = await publish_and_activate(definitions, actor, scope, "order-intake.workflow.yaml", "caller") + + with pytest.raises(CatalogError) as unpinned: + await definitions.activate(actor, scope, request, "unpinned", context=AuditContext()) + assert (unpinned.value.status, unpinned.value.code) == (422, "WV-READINESS-CALL") + + pinned = request.model_copy(update={"workflow_activation_ids": {"notify-customer@1.0.0": callee.id}}) + activation = await definitions.activate(actor, scope, pinned, "pinned", context=AuditContext()) + [pin] = activation.workflow_call_pins + assert (pin.reference, pin.activation_id, pin.activation_revision) == ("notify-customer@1.0.0", callee.id, 1) + assert (pin.version_id, pin.artifact_digest) == (callee.request.version_id, callee.request.artifact_digest) + assert activation.call_depth == 1 + assert await stored(access_db, activation.id) == activation + + async with definitions.uow.open(scope, mutation=False) as tx: + resolve = pinned_call_tree(DefinitionRepository(tx), scope) + row = await DefinitionRepository(tx).version(published.id) + pins = {pin.reference: pin.activation_id} + assert await call_tree_side_effect(row["artifact"]["executable"], pins, resolve) == "read_only" + + await definitions.retire_activation(actor, scope, callee.id, "retire-callee", 1, context=AuditContext()) + with pytest.raises(CatalogError) as retired: + await definitions.activate(actor, scope, pinned, "after-retire", 1, context=AuditContext()) + assert (retired.value.code, retired.value.message) == ( + "WV-READINESS-CALL", + "A pinned workflow activation or its version is retired", + ) + + retired_caller = await definitions.retire_activation( + actor, scope, activation.id, "retire-caller", 1, context=AuditContext() + ) + assert retired_caller.retired + assert (retired_caller.workflow_call_pins, retired_caller.call_depth) == (activation.workflow_call_pins, 1) + assert await stored(access_db, retired_caller.id) == retired_caller diff --git a/tests/integration/test_workflow_call_publication.py b/tests/integration/test_workflow_call_publication.py new file mode 100644 index 00000000..736c53cc --- /dev/null +++ b/tests/integration/test_workflow_call_publication.py @@ -0,0 +1,58 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + + +"""A platform publishes a workflow that calls another only once it runs workflow calls.""" + +import json +from pathlib import Path + +import pytest +from sqlalchemy import text + +from firefly_weave.access.audit import AuditContext +from firefly_weave.compiler.parser import parse_source + +pytestmark = pytest.mark.integration + +EXAMPLES = Path("examples/language") + + +def source(name): + return json.dumps(parse_source((EXAMPLES / name).read_text(), format="yaml").value) + + +async def test_a_platform_that_does_not_run_calls_refuses_to_publish_a_workflow_that_calls( + worker_setup, services, access_db +): + from firefly_weave.definitions.models import CatalogError + from firefly_weave.definitions.service import DefinitionService + + actor, scope = worker_setup[3:5] + definitions = services(access_db[0]).resolve(DefinitionService) + callee = await definitions.publish( + actor, scope, "Workflow", source("notify-customer.workflow.yaml"), "json", "callee", context=AuditContext() + ) + assert callee.name == "notify-customer" + with pytest.raises(CatalogError) as refused: + await definitions.publish( + actor, scope, "Workflow", source("order-intake.workflow.yaml"), "json", "caller", context=AuditContext() + ) + assert (refused.value.status, refused.value.code) == (422, "WV-IR-UNSUPPORTED") + assert refused.value.result == {"reason": "ir_unsupported", "missing_features": ["flow.callWorkflow"]} + async with access_db[1]() as session: + stored = await session.scalar(text("SELECT count(*) FROM definition_versions WHERE name='order-intake'")) + assert stored == 0 diff --git a/tests/unit/cli/test_graph.py b/tests/unit/cli/test_graph.py index 9cdb5ff5..b467bb67 100644 --- a/tests/unit/cli/test_graph.py +++ b/tests/unit/cli/test_graph.py @@ -271,3 +271,28 @@ def test_graph_draws_a_compiled_loop_without_running_it(tmp_path): assert "Loop over items" in content text = CliRunner().invoke(cli, ["workflow", "graph", str(path)]).output assert "[label] Loop over items\n +-- each: body --> [@branch:label:0]" in text + + +def test_graph_draws_a_compiled_call_without_running_it(tmp_path): + from firefly_weave.compiler.catalog import CatalogSnapshot + from firefly_weave.compiler.parser import parse_source + from firefly_weave.contracts.definitions import load_definition + + examples = Path("examples/language") + callee = load_definition( + parse_source((examples / "notify-customer.workflow.yaml").read_text(), format="yaml").value + ) + result = compile_source( + (examples / "order-intake.workflow.yaml").read_text(), + format="yaml", + catalog=CatalogSnapshot.from_definitions([callee]), + ) + assert result.ok and result.artifact is not None + path = tmp_path / "compiled.json" + path.write_bytes(result.artifact.to_bytes()) + for format in ("text", "mermaid", "svg"): + drawn = CliRunner().invoke(cli, ["workflow", "graph", str(path), "--format", format, "--output", "json"]) + assert drawn.exit_code == 0, drawn.output + assert "Call a workflow" in json.loads(drawn.output)["content"] + text = CliRunner().invoke(cli, ["workflow", "graph", str(path)]).output + assert "[notify] Call a workflow\n" in text and "[audit] Call a workflow\n" in text diff --git a/tests/unit/cli/test_offline.py b/tests/unit/cli/test_offline.py index 99106a07..925cb0e7 100644 --- a/tests/unit/cli/test_offline.py +++ b/tests/unit/cli/test_offline.py @@ -357,3 +357,51 @@ def test_explain_prints_a_loop_with_its_settings_and_its_each_edge(tmp_path): "@branch:label:0 -> @join:label (join branch=body)", ): assert expected in result.output + + +def test_explain_prints_a_call_with_its_called_workflow_and_mode(tmp_path): + from firefly_weave.compiler.catalog import FrozenDocument + from firefly_weave.contracts.definitions import load_definition + + examples = Path("examples/language") + callee = load_definition( + parse_source((examples / "notify-customer.workflow.yaml").read_text(), format="yaml").value + ) + document = callee.model_dump(by_alias=True) + lock = tmp_path / "calls.lock.json" + lock.write_text( + json.dumps( + { + "definitions": [{"document": document, "digest": FrozenDocument.from_value(document).digest}], + "tasks": [], + "adapters": [], + "schemas": {}, + } + ) + ) + result = invoke("workflow", "explain", examples / "order-intake.workflow.yaml", "--catalog", lock) + assert result.exit_code == 0, result.output + for expected in ( + " Workflow notify-customer@1.0.0 sha256=", + "notify (callWorkflow)", + " uses=notify-customer@1.0.0 mode=wait onFailure=continue\n", + "audit (callWorkflow)", + " uses=notify-customer@1.0.0 mode=detach\n", + " reads /input/customerId\n", + " reads /input/orderId\n", + " reads /input/tags\n", + "notify -> audit (next)", + " /spec/steps/0 call_output schema=", + " /spec/steps/1/with call_input schema=", + ): + assert expected in result.output + # The business key is read even when nothing in `with` reads the same input. + keyed = tmp_path / "order-intake.workflow.yaml" + source = (examples / "order-intake.workflow.yaml").read_text() + keyed.write_text( + source.replace('{literal: "Order "}, {ref: /input/orderId}', '{literal: "Order "}, {literal: "new"}') + ) + result = invoke("workflow", "explain", keyed, "--catalog", lock) + assert result.exit_code == 0, result.output + assert "{ref: /input/orderId}" not in keyed.read_text().split("businessKey")[0] + assert " reads /input/orderId\n" in result.output diff --git a/tests/unit/compiler/test_call_compiler.py b/tests/unit/compiler/test_call_compiler.py new file mode 100644 index 00000000..2e0d20d1 --- /dev/null +++ b/tests/unit/compiler/test_call_compiler.py @@ -0,0 +1,422 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + + +"""The compiler types, checks and lowers callWorkflow steps against the exact workflow version they call.""" + +import json +from pathlib import Path + +import pytest + +from firefly_weave.compiler.api import compile_source, import_artifact, validate_authoring +from firefly_weave.compiler.canonical import canonical_digest +from firefly_weave.compiler.catalog import CatalogResource, CatalogSnapshot, FrozenDocument +from firefly_weave.compiler.ir import IR_VERSION, IR_VERSION_EXTENSIONS, WorkflowIR +from firefly_weave.compiler.parser import parse_source +from firefly_weave.contracts.definitions import load_definition +from firefly_weave.contracts.limits import Limits + +EXAMPLES = Path("examples/language") +OUTPUT = { + "type": "object", + "properties": {"customerId": {"type": "string"}, "delivered": {"type": "boolean"}}, + "required": ["customerId", "delivered"], + "additionalProperties": False, +} +INPUT = { + "type": "object", + "properties": {"customerId": {"type": "string"}, "message": {"type": "string"}}, + "required": ["customerId", "message"], + "additionalProperties": False, +} +CALLER_INPUT = { + "type": "object", + "properties": {"customerId": {"type": "string"}, "orderId": {"type": "string", "minLength": 1, "maxLength": 200}}, + "required": ["customerId", "orderId"], + "additionalProperties": False, +} +WITH = {"object": {"customerId": {"ref": "/input/customerId"}, "message": {"literal": "Order received"}}} + + +def callee(reference="notify-customer@1.0.0", *, callable_field=None, steps=None, input_schema=None): + name, version = reference.split("@") + spec = { + "inputSchema": input_schema or INPUT, + "outputSchema": OUTPUT, + "steps": steps or [], + "output": {"object": {"customerId": {"ref": "/input/customerId"}, "delivered": {"literal": True}}}, + } + spec["callable"] = {} if callable_field is None else callable_field + return { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": {"name": name, "version": version}, + "spec": spec, + } + + +def call(identifier="notify", uses="notify-customer@1.0.0", **fields): + return {"id": identifier, "kind": "callWorkflow", "uses": uses, "with": WITH, **fields} + + +def caller(*steps, output=None, output_schema=None, name="order-intake"): + return { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": {"name": name, "version": "1.0.0"}, + "spec": { + "inputSchema": CALLER_INPUT, + "outputSchema": output_schema or {}, + "steps": list(steps), + "output": output or {"literal": None}, + }, + } + + +def catalog(*documents, **extra): + return CatalogSnapshot.from_definitions([load_definition(d) for d in documents], **extra) + + +def compiled(document, *documents, limits=None, **extra): + options = {"limits": limits} if limits else {} + return compile_source(document, format="object", catalog=catalog(*documents, **extra), **options) + + +def findings(result, severity=None): + return [ + (d.code, d.path, d.severity, d.stage) for d in result.diagnostics if severity is None or d.severity == severity + ] + + +def test_the_call_example_compiles_to_call_nodes_with_the_called_workflow_as_an_interface_dependency(): + load = lambda name: parse_source((EXAMPLES / name).read_text(), format="yaml").value # noqa: E731 + result = compile_source( + (EXAMPLES / "order-intake.workflow.yaml").read_text(), + format="yaml", + catalog=catalog(load("notify-customer.workflow.yaml")), + ) + assert result.ok, findings(result) + assert findings(result) == [] + executable = json.loads(result.artifact.to_bytes())["executable"] + assert executable["irVersion"] == IR_VERSION_EXTENSIONS + assert executable["features"] == ["flow.callWorkflow", "text.concat", "text.join"] + [dependency] = [d for d in executable["dependencies"] if d["kind"] == "Workflow"] + assert dependency["reference"] == "notify-customer@1.0.0" + assert dependency["document"] == load_definition(load("notify-customer.workflow.yaml")).model_dump(by_alias=True) + calls = {node["id"]: node for node in executable["graph"]["nodes"] if node["kind"] == "callWorkflow"} + assert {key: calls["notify"][key] for key in ("dependency", "mode", "onFailure", "businessKey")} == { + "dependency": dependency["digest"], + "mode": "wait", + "onFailure": "continue", + "businessKey": {"ref": "/input/orderId"}, + } + assert "onFailure" not in calls["audit"] and "businessKey" not in calls["audit"] + assert (calls["audit"]["mode"], calls["audit"]["dependency"]) == ("detach", dependency["digest"]) + edges = {(e["source"], e["target"], e["kind"]) for e in executable["graph"]["edges"]} + assert {("@start", "notify", "next"), ("notify", "audit", "next"), ("audit", "@end", "next")} <= edges + assert [(g["path"], g["purpose"]) for g in executable["guards"]] == [ + ("/spec/inputSchema", "workflow_input"), + ("/spec/steps/0/with", "call_input"), + ("/spec/steps/0", "call_output"), + ("/spec/steps/1/with", "call_input"), + ("/spec/output", "workflow_output"), + ] + assert WorkflowIR.model_validate(executable).features == executable["features"] + + +def test_being_callable_does_not_change_what_the_called_workflow_compiles_to(): + without = callee() + del without["spec"]["callable"] + plain, opted_in = ( + json.loads(compiled(document).artifact.to_bytes())["executable"] for document in (without, callee()) + ) + assert plain == opted_in + assert (plain["irVersion"], "features" in plain) == (IR_VERSION, False) + + +def test_a_waiting_call_that_stops_on_failure_returns_the_called_workflow_output(): + good = caller(call(), output={"ref": "/steps/notify/output/delivered"}, output_schema={"type": "boolean"}) + assert compiled(good, callee()).ok + wrong = caller(call(), output={"ref": "/steps/notify/output/delivered"}, output_schema={"type": "string"}) + assert findings(compiled(wrong, callee()), "error") == [ + ("WV-COMP-TYPE_MISMATCH", "/spec/output", "error", "semantic") + ] + + +def test_a_waiting_call_that_continues_returns_a_status_envelope_whose_output_may_be_missing(): + continued = call(onFailure="continue") + status = caller( + continued, + output={"ref": "/steps/notify/output/status"}, + output_schema={"type": "string", "enum": ["succeeded", "failed", "cancelled", "timed_out"]}, + ) + assert findings(compiled(status, callee())) == [] + child = caller(continued, output={"ref": "/steps/notify/output/output/delivered"}) + assert findings(compiled(child, callee())) == [ + ("WV-COMP-REFERENCE_PRESENCE", "/spec/output", "warning", "semantic") + ] + fallback = caller( + continued, + output={ + "op": {"name": "coalesce", "args": [{"ref": "/steps/notify/output/output/delivered"}, {"literal": False}]} + }, + output_schema={"type": "boolean"}, + ) + assert findings(compiled(fallback, callee())) == [] + + +def test_a_detached_call_returns_nothing(): + detached = call(mode="detach") + assert compiled( + caller(detached, output={"ref": "/steps/notify/output"}, output_schema={"type": "null"}), callee() + ).ok + assert findings( + compiled(caller(detached, output={"ref": "/steps/notify/output"}, output_schema={"type": "string"}), callee()), + "error", + ) == [("WV-COMP-TYPE_MISMATCH", "/spec/output", "error", "semantic")] + + +def test_the_call_input_is_checked_against_the_called_workflow_input(): + wrong = call() + wrong["with"] = {"object": {"customerId": {"literal": 7}, "message": {"literal": "x"}}} + assert findings(compiled(caller(wrong), callee()), "error") == [ + ("WV-COMP-TYPE_MISMATCH", "/spec/steps/0/with", "error", "semantic") + ] + + +@pytest.mark.parametrize( + ("business_key", "expected"), + [ + pytest.param({"ref": "/input/orderId"}, [], id="bounded-text"), + pytest.param( + {"ref": "/input/customerId"}, + [("WV-COMP-UNKNOWN_COMPATIBILITY", "/spec/steps/0/businessKey", "warning", "semantic")], + id="unbounded-text", + ), + pytest.param( + {"literal": 7}, [("WV-COMP-TYPE_MISMATCH", "/spec/steps/0/businessKey", "error", "semantic")], id="number" + ), + pytest.param( + {"literal": "x" * 201}, + [("WV-COMP-TYPE_MISMATCH", "/spec/steps/0/businessKey", "error", "semantic")], + id="too-long", + ), + ], +) +def test_a_business_key_is_text_of_one_to_200_characters(business_key, expected): + assert findings(compiled(caller(call(businessKey=business_key)), callee())) == expected + + +def test_an_unknown_version_and_a_version_that_does_not_allow_the_caller_are_resolution_errors(): + unknown = compiled(caller(call(uses="notify-customer@9.9.9")), callee()) + assert findings(unknown) == [("WV-COMP-UNKNOWN_WORKFLOW", "/spec/steps/0/uses", "error", "resolution")] + assert unknown.diagnostics[0].message == "The catalog does not contain this exact workflow version." + closed = compiled(caller(call()), callee(callable_field={"allowedCallers": ["invoice-run"]})) + assert findings(closed) == [("WV-COMP-CALL_NOT_CALLABLE", "/spec/steps/0/uses", "error", "resolution")] + assert "allow calls from order-intake" in closed.diagnostics[0].message + + +def test_a_call_back_to_the_calling_version_is_recursion(): + itself = caller(call(uses="order-intake@1.0.0")) + itself["spec"]["callable"] = {} + itself["spec"]["steps"][0]["with"] = {"ref": "/input"} + result = compiled(itself, itself) + assert findings(result) == [("WV-COMP-CALL_RECURSION", "/spec/steps/0/uses", "error", "semantic")] + + +def test_the_chain_of_calls_is_limited_by_the_call_depth(): + inner = callee("inner@1.0.0") + outer = callee(steps=[{**call("deeper", "inner@1.0.0"), "with": {"ref": "/input"}}]) + assert compiled(caller(call()), outer, inner, limits=Limits(max_call_depth=2)).ok + too_deep = compiled(caller(call()), outer, inner, limits=Limits(max_call_depth=1)) + assert findings(too_deep) == [("WV-COMP-CALL_DEPTH", "/spec/steps/0/uses", "error", "semantic")] + assert too_deep.diagnostics[0].message == "The chain of workflow calls is deeper than 1." + + +def test_on_failure_on_a_call_that_does_not_wait_is_reported_at_its_field(): + detached = call(mode="detach", onFailure="continue") + nested = caller( + { + "id": "pick", + "kind": "switch", + "cases": [{"when": {"literal": True}, "steps": [detached], "output": {"literal": None}}], + "default": {"steps": [], "output": {"literal": None}}, + } + ) + result = compiled(nested, callee()) + assert findings(result) == [("WV-COMP-CALL_MODE", "/spec/steps/0/cases/0/steps/0/onFailure", "error", "semantic")] + assert result.diagnostics[0].message == "onFailure applies only when the call waits for its result." + broken = caller(detached, {"id": "late", "kind": "wait"}) + assert [code for code, *_ in findings(compiled(broken, callee()))] == [ + "WV-COMP-CALL_MODE", + "WV-SCHEMA-INVALID_INSTANCE", + ] + + +def test_without_a_catalog_calls_are_pending_and_their_envelope_is_still_typed(): + pending = validate_authoring(caller(call(), output={"ref": "/steps/notify/output/anything"}), format="object") + assert findings(pending) == [("WV-COMP-CATALOG_PENDING", "/spec/steps/0/uses", "info", "resolution")] + envelope = validate_authoring( + caller( + call(onFailure="continue"), output={"ref": "/steps/notify/output/status"}, output_schema={"type": "integer"} + ), + format="object", + ) + assert findings(envelope, "error") == [("WV-COMP-TYPE_MISMATCH", "/spec/output", "error", "semantic")] + + +def test_a_secret_input_of_the_called_workflow_never_takes_a_literal(): + secret = callee( + input_schema={**INPUT, "properties": {**INPUT["properties"], "message": {"type": "string", "x-secret": True}}} + ) + result = compiled(caller(call()), secret) + assert not result.ok and result.artifact is None + assert [d.code for d in result.diagnostics] == ["WV-SCHEMA-SECRET_VALUE"] + + +def forged_with_secret_callee_input(artifact): + """The artifact with its called workflow's ``message`` input marked secret, every digest recomputed.""" + value = json.loads(artifact.to_bytes()) + executable = value["executable"] + (dependency,) = executable["dependencies"] + message = dependency["document"]["spec"]["inputSchema"]["properties"]["message"] + message["x-secret"] = True + document = FrozenDocument.from_value(dependency["document"]) + digest = CatalogResource("Workflow", dependency["reference"], document).digest + for node in executable["graph"]["nodes"]: + if node.get("dependency") == dependency["digest"]: + node["dependency"] = digest + dependency["digest"] = digest + value["digest"] = canonical_digest(executable) + return value + + +def test_import_refuses_a_forged_caller_that_binds_a_literal_to_a_secret_input_of_the_called_workflow(): + artifact = compiled(caller(call()), callee()).artifact + imported = import_artifact(artifact.to_bytes(), features=["flow.callWorkflow"]) + assert imported.digest == artifact.digest + with pytest.raises(ValueError, match="classified or unclassifiable"): + import_artifact(forged_with_secret_callee_input(artifact), features=["flow.callWorkflow"]) + + +def test_an_imported_caller_never_needs_what_the_called_workflow_uses_inside(worker_runtime_fixture): + action = worker_runtime_fixture["action"].model_dump(by_alias=True) + worker = callee( + steps=[{"id": "work", "kind": "action", "uses": "echo-action@2.0.0", "with": {"literal": {"n": 1}}}] + ) + task = next( + r.definition.value for r in worker_runtime_fixture["catalog"].resources.values() if r.kind == "TaskCapability" + ) + result = compiled(caller(call()), worker, action, tasks=[task]) + assert result.ok, findings(result) + executable = json.loads(result.artifact.to_bytes())["executable"] + assert sorted(d["kind"] for d in executable["dependencies"]) == ["Workflow"] + imported = import_artifact(result.artifact.to_bytes(), features=["flow.callWorkflow"]) + assert imported.digest == result.artifact.digest + + +LOOP_INPUT = { + "type": "object", + "properties": {"customers": {"type": "array", "items": {"type": "string"}, "maxItems": 10}}, + "required": ["customers"], + "additionalProperties": False, +} +PER_CUSTOMER = {"object": {"customerId": {"ref": "/item"}, "message": {"literal": "Order received"}}} + + +def per_customer(*body): + document = caller( + { + "id": "each", + "kind": "forEach", + "items": {"ref": "/input/customers"}, + "maxItems": 10, + "body": {"steps": list(body), "output": {"literal": None}}, + } + ) + document["spec"]["inputSchema"] = LOOP_INPUT + return document + + +def test_calls_inside_a_loop_body_compile_import_and_stop_like_any_other_call(): + from datetime import UTC, datetime + from uuid import uuid4 + + from firefly_weave.compiler.ir import UnsupportedIR + from firefly_weave.runtime.kernel import transition + from firefly_weave.runtime.models import RunState, RuntimeEvent + + loop = per_customer( + call(**{"with": PER_CUSTOMER}), call("status", **{"with": PER_CUSTOMER, "onFailure": "continue"}) + ) + result = compiled(loop, callee()) + assert result.ok, findings(result) + executable = json.loads(result.artifact.to_bytes())["executable"] + assert (executable["irVersion"], executable["features"]) == ( + IR_VERSION_EXTENSIONS, + ["flow.callWorkflow", "flow.forEach"], + ) + assert [(g["path"], g["purpose"]) for g in executable["guards"] if g["purpose"].startswith("call_")] == [ + ("/spec/steps/0/body/steps/0/with", "call_input"), + ("/spec/steps/0/body/steps/0", "call_output"), + ("/spec/steps/0/body/steps/1/with", "call_input"), + ("/spec/steps/0/body/steps/1", "call_output"), + ] + with pytest.raises(UnsupportedIR) as refused: + import_artifact(result.artifact.to_bytes()) + assert refused.value.missing == ("flow.callWorkflow",) + artifact = import_artifact(result.artifact.to_bytes(), features=["flow.callWorkflow", "flow.forEach"]) + assert artifact.digest == result.artifact.digest + + started = RuntimeEvent(id=uuid4(), type="started", timestamp=datetime(2026, 10, 13, tzinfo=UTC), sequence=1) + stopped = transition(RunState(input={"customers": ["c-1", "c-2"]}), started, result.artifact) + assert stopped.state.status == "suspended" + assert [(item.node_id, item.code) for item in stopped.state.incidents.values()] == [ + ("notify[0]", "WV-RUNTIME-ARTIFACT") + ] + assert stopped.steps == [] + + +def test_call_checks_reach_into_loop_bodies(): + detached = call(**{"with": PER_CUSTOMER, "mode": "detach", "onFailure": "continue"}) + assert findings(compiled(per_customer(detached), callee())) == [ + ("WV-COMP-CALL_MODE", "/spec/steps/0/body/steps/0/onFailure", "error", "semantic") + ] + pending = validate_authoring(per_customer(call(**{"with": PER_CUSTOMER})), format="object") + assert findings(pending) == [("WV-COMP-CATALOG_PENDING", "/spec/steps/0/body/steps/0/uses", "info", "resolution")] + secret = callee( + input_schema={**INPUT, "properties": {**INPUT["properties"], "message": {"type": "string", "x-secret": True}}} + ) + refused = compiled(per_customer(call(**{"with": PER_CUSTOMER})), secret) + assert not refused.ok and [d.code for d in refused.diagnostics] == ["WV-SCHEMA-SECRET_VALUE"] + calls_back = {**call("back", "order-intake@1.0.0"), "with": {"ref": "/input"}} + looping = callee( + steps=[ + { + "id": "each", + "kind": "forEach", + "items": {"literal": [1]}, + "maxItems": 1, + "body": {"steps": [calls_back], "output": {"literal": None}}, + } + ] + ) + itself = caller(call()) + itself["spec"]["callable"] = {} + assert findings(compiled(itself, looping, itself)) == [ + ("WV-COMP-CALL_RECURSION", "/spec/steps/0/uses", "error", "semantic") + ] diff --git a/tests/unit/compiler/test_call_ir.py b/tests/unit/compiler/test_call_ir.py new file mode 100644 index 00000000..d6efdad4 --- /dev/null +++ b/tests/unit/compiler/test_call_ir.py @@ -0,0 +1,213 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + +"""The call node, the called workflow as an interface dependency, and the call depth budget.""" + +import copy +import json +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from firefly_weave.compiler.catalog import CatalogResource, FrozenDocument +from firefly_weave.compiler.ir import ( + IR_VERSION_EXTENSIONS, + CallWorkflowNode, + Guard, + IRGraph, + WorkflowIR, + allows_caller, + workflow_requirements, +) +from firefly_weave.contracts.language import language_manifest +from firefly_weave.contracts.limits import Limits + +EXECUTABLE = json.loads(Path("tests/fixtures/canonical/empty-workflow.executable.json").read_text()) +CONCAT = {"op": {"name": "concat", "args": [{"literal": "order "}, {"ref": "/input/id"}]}} + + +def callee(callable_field=None, *, steps=None, input_schema=None, connections=None): + """notify-customer@1.0.0; ``callable_field`` is its ``spec.callable`` (omitted when None).""" + spec = { + "inputSchema": input_schema or {"type": "object"}, + "outputSchema": {"type": "object"}, + "steps": steps or [], + "output": {"ref": "/input"}, + } + if connections is not None: + spec["connections"] = connections + if callable_field is not None: + spec["callable"] = callable_field + document = { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": {"name": "notify-customer", "version": "1.0.0"}, + "spec": spec, + } + resource = CatalogResource("Workflow", "notify-customer@1.0.0", FrozenDocument.from_value(document)) + return { + "kind": "Workflow", + "reference": "notify-customer@1.0.0", + "digest": resource.digest, + "document": resource.definition.value, + } + + +def edge(source, target, kind="next", branch=None): + return {"source": source, "target": target, "kind": kind, "branch": branch} + + +def call_graph(dependency, **fields): + """@start -> notify (callWorkflow) -> @end.""" + node = { + "id": "notify", + "kind": "callWorkflow", + "scope": [], + "path": "/spec/steps/0", + "dependency": dependency, + "with": {"ref": "/input"}, + "mode": "wait", + **fields, + } + return { + "entry": "@start", + "exit": "@end", + "nodes": [ + {"id": "@start", "kind": "start", "scope": [], "path": "/spec"}, + {"id": "@end", "kind": "end", "scope": [], "path": "/spec/output", "output": {"literal": None}}, + node, + ], + "edges": [edge("@start", "notify"), edge("notify", "@end")], + } + + +def executable(dependency=None, **fields): + dependency = dependency or callee({}) + return { + **copy.deepcopy(EXECUTABLE), + "irVersion": IR_VERSION_EXTENSIONS, + "features": ["flow.callWorkflow"], + "dependencies": [dependency], + "graph": call_graph(dependency["digest"], **fields), + } + + +def test_a_call_graph_is_valid_in_both_modes(): + for fields in ({}, {"onFailure": "continue", "businessKey": {"ref": "/input/id"}}, {"mode": "detach"}): + graph = IRGraph.model_validate(call_graph("a" * 64, **fields)) + call = next(item for item in graph.nodes if isinstance(item, CallWorkflowNode)) + assert (call.mode, call.on_failure) == (fields.get("mode", "wait"), fields.get("onFailure")) + + +def test_the_call_node_serializes_its_canonical_field_names_and_omits_unset_options(): + for fields in ({}, {"onFailure": "stop", "businessKey": {"ref": "/input/id"}}): + graph = call_graph("a" * 64, **fields) + dumped = IRGraph.model_validate(graph).model_dump(by_alias=True) + assert dumped["nodes"][2] == graph["nodes"][2] + assert set(IRGraph.model_validate(call_graph("a" * 64)).model_dump(by_alias=True)["nodes"][2]) == { + "id", + "kind", + "scope", + "path", + "dependency", + "with", + "mode", + } + + +def test_a_detached_call_cannot_say_what_happens_when_it_fails(): + with pytest.raises(ValidationError, match="onFailure applies only when the call waits"): + IRGraph.model_validate(call_graph("a" * 64, mode="detach", onFailure="continue")) + + +@pytest.mark.parametrize( + "edges", + [ + pytest.param([edge("@start", "notify")], id="no-successor"), + pytest.param([edge("@start", "notify"), edge("notify", "@end", "case", "x")], id="control-edge"), + pytest.param([edge("@start", "notify"), edge("notify", "@end", "next", "body")], id="branch-on-next"), + ], +) +def test_a_call_continues_through_exactly_one_next_edge(edges): + graph = call_graph("a" * 64) + graph["edges"] = edges + with pytest.raises(ValidationError, match="Invalid sequential edges"): + IRGraph.model_validate(graph) + + +def test_a_call_selects_the_extension_ir_and_reads_its_input_and_business_key(): + plain = IRGraph.model_validate(call_graph("a" * 64)) + assert workflow_requirements(plain) == (IR_VERSION_EXTENSIONS, ["flow.callWorkflow"]) + for fields in ({"with": CONCAT}, {"businessKey": CONCAT}): + graph = IRGraph.model_validate(call_graph("a" * 64, **fields)) + assert workflow_requirements(graph) == (IR_VERSION_EXTENSIONS, ["flow.callWorkflow", "text.concat"]) + + +def test_a_workflow_with_a_call_validates_its_features_and_its_called_workflow(): + workflow = WorkflowIR.model_validate(executable()) + assert workflow.features == ["flow.callWorkflow"] + with pytest.raises(ValidationError, match="Language features must be exactly"): + WorkflowIR.model_validate({**executable(), "features": ["flow.forEach"]}) + with pytest.raises(ValidationError, match="Unknown called workflow dependency"): + WorkflowIR.model_validate({**executable(), "graph": call_graph("b" * 64)}) + + +@pytest.mark.parametrize( + ("callable_field", "allowed"), + [ + pytest.param(None, False, id="not-callable"), + pytest.param({}, True, id="any-caller"), + pytest.param({"allowedCallers": ["sample"]}, True, id="named-caller"), + pytest.param({"allowedCallers": ["order-intake"]}, False, id="other-caller"), + ], +) +def test_a_called_workflow_must_allow_the_calling_workflow_by_name(callable_field, allowed): + assert allows_caller(callable_field, "sample") is allowed + document = executable(callee(callable_field)) + if allowed: + assert WorkflowIR.model_validate(document).metadata.name == "sample" + else: + with pytest.raises(ValidationError, match="does not allow calls from this workflow"): + WorkflowIR.model_validate(document) + + +def test_a_called_workflow_is_an_interface_whose_steps_this_executable_never_needs(): + steps = [{"id": "send", "kind": "action", "uses": "missing@1.0.0", "with": {"ref": "/input"}}] + assert WorkflowIR.model_validate(executable(callee({}, steps=steps))).dependencies[0].kind == "Workflow" + named = callee({}, input_schema={"$ref": "customer#/$defs/request"}) + with pytest.raises(ValidationError, match="Unresolved executable dependency"): + WorkflowIR.model_validate(executable(named)) + + +def test_the_called_workflow_connections_stay_with_its_own_activation(): + connected = callee({}, connections={"crm": {"connector": "crm-connector@1.0.0"}}) + validated = WorkflowIR.model_validate(executable(connected)) + assert [dependency.kind for dependency in validated.dependencies] == ["Workflow"] + assert validated.connections == {} + + +def test_call_guards_have_their_own_purposes(): + for purpose in ("call_input", "call_output"): + assert Guard.model_validate({"path": "/spec/steps/0", "purpose": purpose, "schemaRef": "a" * 64}).purpose == ( + purpose + ) + + +def test_call_depth_is_an_operator_budget_with_the_published_default(): + assert Limits().max_call_depth == language_manifest().limits.max_call_depth == 8 + with pytest.raises(ValidationError): + Limits(max_call_depth=0) diff --git a/tests/unit/compiler/test_call_targets.py b/tests/unit/compiler/test_call_targets.py new file mode 100644 index 00000000..cda5fb6e --- /dev/null +++ b/tests/unit/compiler/test_call_targets.py @@ -0,0 +1,200 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + + +"""Which workflow versions a workflow may call, how deep calls go, and what a call step returns.""" + +import pytest + +from firefly_weave.compiler.calls import ( + BUSINESS_KEY_SCHEMA, + CallGraph, + call_output_schema, + called_references, + check_call_target, + continue_envelope, + detached_failure_paths, + without_detached_failures, +) +from firefly_weave.compiler.catalog import CatalogSnapshot +from firefly_weave.contracts.definitions import load_definition + + +def call(identifier, uses, **fields): + return {"id": identifier, "kind": "callWorkflow", "uses": uses, "with": {"ref": "/input"}, **fields} + + +def workflow(reference, *steps, callable_field=None): + name, version = reference.split("@") + spec = {"inputSchema": {}, "outputSchema": {}, "steps": list(steps), "output": {"literal": None}} + if callable_field is not None: + spec["callable"] = callable_field + return { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": {"name": name, "version": version}, + "spec": spec, + } + + +def catalog(*documents): + return CatalogSnapshot.from_definitions([load_definition(document) for document in documents]) + + +def codes(target): + return [(issue.code, issue.stage) for issue in target.issues] + + +def test_an_unknown_workflow_version_is_the_only_finding(): + target = check_call_target( + "order-intake@1.0.0", "notify@1.0.0", catalog(workflow("notify@2.0.0", callable_field={})) + ) + assert target.resource is None + assert codes(target) == [("UNKNOWN_WORKFLOW", "resolution")] + assert target.issues[0].message == "The catalog does not contain this exact workflow version." + + +@pytest.mark.parametrize( + ("callable_field", "allowed"), + [ + pytest.param(None, False, id="not-callable"), + pytest.param({}, True, id="any-workflow"), + pytest.param({"allowedCallers": ["order-intake", "invoice-run"]}, True, id="named"), + pytest.param({"allowedCallers": ["invoice-run"]}, False, id="other-names"), + ], +) +def test_a_version_allows_calls_from_any_workflow_or_only_the_named_ones(callable_field, allowed): + target = check_call_target( + "order-intake@1.0.0", "notify@1.0.0", catalog(workflow("notify@1.0.0", callable_field=callable_field)) + ) + assert target.resource is not None and target.resource.reference == "notify@1.0.0" + assert codes(target) == ([] if allowed else [("CALL_NOT_CALLABLE", "resolution")]) + if not allowed: + assert target.issues[0].message == ( + "This workflow version does not allow calls from order-intake; " + "add order-intake to its callable.allowedCallers in a new version." + ) + + +def test_a_call_that_leads_back_to_the_calling_version_is_recursion(): + itself = catalog(workflow("loop@1.0.0", call("again", "loop@1.0.0"), callable_field={})) + assert codes(check_call_target("loop@1.0.0", "loop@1.0.0", itself)) == [("CALL_RECURSION", "semantic")] + indirect = catalog( + workflow("a@1.0.0", call("b", "b@1.0.0"), callable_field={}), + workflow("b@1.0.0", call("a", "a@1.0.0"), callable_field={}), + ) + assert codes(check_call_target("a@1.0.0", "b@1.0.0", indirect)) == [("CALL_RECURSION", "semantic")] + + +def test_a_called_workflow_that_calls_back_a_caller_not_yet_published_is_recursion(): + callee_only = catalog(workflow("b@1.0.0", call("a", "a@1.0.0"), callable_field={})) + assert codes(check_call_target("a@1.0.0", "b@1.0.0", callee_only)) == [("CALL_RECURSION", "semantic")] + + +def test_calling_an_older_version_of_the_same_workflow_is_not_recursion(): + older = catalog(workflow("report@1.0.0", callable_field={})) + assert codes(check_call_target("report@2.0.0", "report@1.0.0", older)) == [] + + +def chain(length): + """c1 calls c2 ... and c calls nothing; every version allows any caller.""" + return [ + workflow( + f"c{index}@1.0.0", *([call("next", f"c{index + 1}@1.0.0")] if index < length else []), callable_field={} + ) + for index in range(1, length + 1) + ] + + +def test_the_longest_chain_of_calls_stays_within_the_call_depth(): + calls = catalog(*chain(3)) + graph = CallGraph(calls) + assert [graph.depth(f"c{index}@1.0.0") for index in (1, 2, 3)] == [2, 1, 0] + assert codes(check_call_target("top@1.0.0", "c1@1.0.0", calls, max_call_depth=3)) == [] + too_deep = check_call_target("top@1.0.0", "c1@1.0.0", calls, max_call_depth=2) + assert codes(too_deep) == [("CALL_DEPTH", "semantic")] + assert too_deep.issues[0].message == "The chain of workflow calls is deeper than 2." + assert codes(check_call_target("top@1.0.0", "c1@1.0.0", catalog(*chain(8)))) == [] + assert codes(check_call_target("top@1.0.0", "c1@1.0.0", catalog(*chain(9)))) == [("CALL_DEPTH", "semantic")] + + +def test_calls_that_never_end_without_reaching_the_caller_are_too_deep(): + cycle = catalog( + workflow("b@1.0.0", call("c", "c@1.0.0"), callable_field={}), + workflow("c@1.0.0", call("b", "b@1.0.0"), callable_field={}), + ) + assert CallGraph(cycle).depth("b@1.0.0") is None + assert codes(check_call_target("a@1.0.0", "b@1.0.0", cycle)) == [("CALL_DEPTH", "semantic")] + + +def test_a_missing_callee_of_a_callee_counts_as_calling_nothing(): + partial = catalog(workflow("b@1.0.0", call("gone", "gone@1.0.0"), callable_field={})) + assert CallGraph(partial).depth("b@1.0.0") == 1 + assert codes(check_call_target("a@1.0.0", "b@1.0.0", partial)) == [] + + +def test_calls_are_found_in_every_block_and_detached_failure_settings_by_path(): + document = workflow( + "w@1.0.0", + call("first", "a@1.0.0", mode="detach", onFailure="continue"), + { + "id": "pick", + "kind": "switch", + "cases": [{"when": {"literal": True}, "steps": [call("inCase", "b@1.0.0")], "output": {"literal": None}}], + "default": {"steps": [call("inDefault", "c@1.0.0", mode="detach")], "output": {"literal": None}}, + }, + { + "id": "both", + "kind": "parallel", + "concurrency": 2, + "branches": { + "x/y": {"steps": [call("inBranch", "d@1.0.0", onFailure="stop")], "output": {"literal": None}} + }, + }, + { + "id": "each", + "kind": "forEach", + "items": {"ref": "/input"}, + "body": { + "steps": [call("inBody", "e@1.0.0", mode="detach", onFailure="stop")], + "output": {"literal": None}, + }, + }, + ) + assert called_references(document) == ["a@1.0.0", "b@1.0.0", "c@1.0.0", "d@1.0.0", "e@1.0.0"] + assert detached_failure_paths(document) == ["/spec/steps/0/onFailure", "/spec/steps/3/body/steps/0/onFailure"] + cleaned = without_detached_failures(document) + assert detached_failure_paths(cleaned) == [] and detached_failure_paths(document) != [] + assert cleaned["spec"]["steps"][2]["branches"]["x/y"]["steps"][0]["onFailure"] == "stop" + assert called_references({"kind": "Workflow", "spec": {"steps": [7, {"kind": "callWorkflow"}]}}) == [] + assert called_references({"kind": "Action"}) == [] + + +def test_a_call_step_returns_the_callee_output_the_continue_envelope_or_nothing(): + output = {"type": "object", "properties": {"delivered": {"type": "boolean"}}} + assert call_output_schema("wait", None, output) == output + assert call_output_schema("wait", "stop", output) == output + assert call_output_schema("detach", None, output) == {"type": "null"} + envelope = call_output_schema("wait", "continue", output) + assert envelope == continue_envelope(output) + assert envelope["required"] == ["status", "runId"] + assert envelope["properties"]["status"] == { + "type": "string", + "enum": ["succeeded", "failed", "cancelled", "timed_out"], + } + assert envelope["properties"]["output"] == output + assert set(envelope["properties"]["error"]["required"]) == {"code", "message"} + assert BUSINESS_KEY_SCHEMA == {"type": "string", "minLength": 1, "maxLength": 200} diff --git a/tests/unit/compiler/test_contracts.py b/tests/unit/compiler/test_contracts.py index 04220358..8d7986f4 100644 --- a/tests/unit/compiler/test_contracts.py +++ b/tests/unit/compiler/test_contracts.py @@ -63,6 +63,7 @@ def test_definition_budget_cannot_coerce_strings(): "max_loop_items": 10_000, "max_loop_depth": 3, "max_run_iterations": 100_000, + "max_call_depth": 8, } diff --git a/tests/unit/compiler/test_language_examples.py b/tests/unit/compiler/test_language_examples.py index c9a699ca..c0883421 100644 --- a/tests/unit/compiler/test_language_examples.py +++ b/tests/unit/compiler/test_language_examples.py @@ -14,7 +14,7 @@ # Author: Firefly Software Foundation # SPDX-License-Identifier: Apache-2.0 -"""The language examples have their final shape and report exactly the features not compiled yet.""" +"""The language examples have their final shape, and every feature they use compiles.""" from pathlib import Path @@ -29,12 +29,8 @@ EXPECTED = { "notify-customer.workflow.yaml": [], "order-summary.workflow.yaml": [], - # Text operators and loops compile; workflow calls are still reported. "notify-overdue.workflow.yaml": [], - "order-intake.workflow.yaml": [ - ("/spec/steps/0/kind", "flow.callWorkflow"), - ("/spec/steps/1/kind", "flow.callWorkflow"), - ], + "order-intake.workflow.yaml": [], } diff --git a/tests/unit/compiler/test_loop_ir.py b/tests/unit/compiler/test_loop_ir.py index eae17703..3a3b83ad 100644 --- a/tests/unit/compiler/test_loop_ir.py +++ b/tests/unit/compiler/test_loop_ir.py @@ -232,12 +232,13 @@ def test_the_workflow_validator_requires_the_loop_feature(): WorkflowIR.model_validate({**executable, "features": ["text.concat"]}) -def test_the_dependency_closure_follows_steps_inside_loop_bodies(): +def test_a_workflow_dependency_is_an_interface_even_when_its_loop_bodies_use_actions(): workflow = { "apiVersion": "weave/v1alpha1", "kind": "Workflow", "metadata": {"name": "outer", "version": "1.0.0"}, "spec": { + "callable": {}, "inputSchema": {}, "outputSchema": {}, "steps": [ @@ -261,8 +262,8 @@ def test_the_dependency_closure_follows_steps_inside_loop_bodies(): "digest": resource.digest, "document": resource.definition.value, } - with pytest.raises(ValidationError, match="Unresolved executable dependency"): - WorkflowIR.model_validate({**EXECUTABLE, "dependencies": [dependency]}) + # A called workflow's steps run in its own child run, so the caller never needs the actions they use. + assert WorkflowIR.model_validate({**EXECUTABLE, "dependencies": [dependency]}).dependencies[0].kind == "Workflow" def test_loop_limits_are_operator_budgets_with_the_published_defaults(): diff --git a/tests/unit/compiler/test_unsupported_features.py b/tests/unit/compiler/test_unsupported_features.py index e182d281..48780c93 100644 --- a/tests/unit/compiler/test_unsupported_features.py +++ b/tests/unit/compiler/test_unsupported_features.py @@ -231,13 +231,13 @@ def _switch(cases: list[dict], default: dict | None = None) -> dict: ] -def test_text_and_loops_compile_and_nothing_is_advertised_that_the_compiler_cannot_compile(): - assert set(COMPILED_FEATURES) == {"flow.forEach", "text.concat", "text.join"} +def test_text_loops_and_calls_compile_and_nothing_is_advertised_that_the_compiler_cannot_compile(): + assert set(COMPILED_FEATURES) == {"flow.callWorkflow", "flow.forEach", "text.concat", "text.join"} assert set(ADVERTISED_FEATURES) <= COMPILED_FEATURES -def test_with_text_and_loops_compiled_only_calls_are_reported(): - assert _uses(EVERYTHING) == [("/spec/steps/0/body/steps/1/kind", "flow.callWorkflow")] +def test_every_construct_a_definition_can_hold_today_compiles(): + assert _uses(EVERYTHING) == [] def test_scan_finds_every_kind_and_operator_at_its_field(nothing_compiled): diff --git a/tests/unit/runtime/test_call_refusal.py b/tests/unit/runtime/test_call_refusal.py new file mode 100644 index 00000000..8b5b80f5 --- /dev/null +++ b/tests/unit/runtime/test_call_refusal.py @@ -0,0 +1,65 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + + +"""A runtime that does not start child runs refuses a compiled call; it never skips the step.""" + +from datetime import UTC, datetime +from pathlib import Path +from uuid import uuid4 + +import pytest + +from firefly_weave.compiler.api import compile_source, import_artifact +from firefly_weave.compiler.catalog import CatalogSnapshot +from firefly_weave.compiler.ir import UnsupportedIR +from firefly_weave.compiler.parser import parse_source +from firefly_weave.contracts.definitions import load_definition +from firefly_weave.runtime.kernel import transition +from firefly_weave.runtime.models import RunState, RuntimeEvent + +EXAMPLES = Path("examples/language") + + +@pytest.fixture +def artifact(): + callee = parse_source((EXAMPLES / "notify-customer.workflow.yaml").read_text(), format="yaml").value + result = compile_source( + (EXAMPLES / "order-intake.workflow.yaml").read_text(), + format="yaml", + catalog=CatalogSnapshot.from_definitions([load_definition(callee)]), + ) + assert result.ok, result.to_bytes() + return result.artifact + + +def test_import_refuses_a_call_while_the_platform_does_not_list_the_call_feature(artifact): + with pytest.raises(UnsupportedIR) as refused: + import_artifact(artifact.to_bytes()) + assert refused.value.missing == ("flow.callWorkflow",) + features = ["flow.callWorkflow", "text.concat", "text.join"] + assert import_artifact(artifact.to_bytes(), features=features).digest == artifact.digest + + +def test_the_kernel_opens_an_incident_at_a_call_it_cannot_start(artifact): + started = RuntimeEvent(id=uuid4(), type="started", timestamp=datetime(2026, 10, 13, tzinfo=UTC), sequence=1) + run_input = {"customerId": "c-1", "orderId": "o-1", "tags": ["new"]} + result = transition(RunState(input=run_input), started, artifact) + assert result.state.status == "suspended" + assert [(item.node_id, item.code) for item in result.state.incidents.values()] == [ + ("notify", "WV-RUNTIME-ARTIFACT") + ] + assert result.steps == [] diff --git a/tests/unit/sdk/test_builder.py b/tests/unit/sdk/test_builder.py index abd25602..20e240b7 100644 --- a/tests/unit/sdk/test_builder.py +++ b/tests/unit/sdk/test_builder.py @@ -430,3 +430,16 @@ def test_human_task_builder_uses_native_contract_and_new_ir(): result = compile_source(builder.to_document(), format="object", catalog=CatalogSnapshot.empty()) assert result.ok and result.artifact.executable["irVersion"] == "weave/ir-v1alpha2" assert builder.to_definition().spec.steps[0].kind == "humanTask" + + +def test_with_callable_lets_any_workflow_or_only_named_workflows_call_this_version(): + builder = new_builder() + assert "callable" not in builder.to_document()["spec"] + assert builder.with_callable().to_document()["spec"]["callable"] == {} + named = builder.with_callable(["order-intake", "invoice-run"]) + assert named.to_document()["spec"]["callable"] == {"allowedCallers": ["order-intake", "invoice-run"]} + assert named.to_definition().spec.callable is not None + with pytest.raises(ValidationError): + builder.with_callable(["order-intake", "order-intake"]) + result = compile_source(named.to_document(), format="object", catalog=CatalogSnapshot.empty()) + assert result.ok and "features" not in result.artifact.executable diff --git a/tests/unit/test_acceptance_journeys.py b/tests/unit/test_acceptance_journeys.py index e464a77d..6a95038b 100644 --- a/tests/unit/test_acceptance_journeys.py +++ b/tests/unit/test_acceptance_journeys.py @@ -134,6 +134,7 @@ def test_the_committed_map_is_valid(): "instance-keys", "loop-compiler", "loop-runtime", + "workflow-call-compiler", "ai-gateway", "ai-models-screen", "acceptance-foundations", diff --git a/tests/unit/test_workflow_call_pins.py b/tests/unit/test_workflow_call_pins.py new file mode 100644 index 00000000..247e2959 --- /dev/null +++ b/tests/unit/test_workflow_call_pins.py @@ -0,0 +1,207 @@ +# Copyright 2026 Firefly Software Foundation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# Author: Firefly Software Foundation +# SPDX-License-Identifier: Apache-2.0 + + +"""Activation pins for workflow calls, and the side effect a run can cause through them.""" + +from uuid import UUID, uuid4 + +import pytest + +from firefly_weave.contracts.access import Scope +from firefly_weave.contracts.catalog import Activation, ActivationRequest, WorkflowCallPin +from firefly_weave.definitions.calls import PinnedCallee, call_pins, call_tree_side_effect, side_effect +from firefly_weave.definitions.models import CatalogError + +TENANT, PROJECT, ENVIRONMENT = (UUID(int=index) for index in (1, 2, 3)) +SCOPE = Scope(tenant_id=TENANT, project_id=PROJECT, environment_id=ENVIRONMENT) +DEFINITION, ARTIFACT = "d" * 64, "a" * 64 +REFERENCE = "notify-customer@1.0.0" + + +def caller_executable(*references): + return {"dependencies": [{"kind": "Workflow", "reference": r, "digest": DEFINITION} for r in references]} + + +def callee(*, name="notify-customer", version="1.0.0", callable_field=None, call_depth=0, scope=SCOPE, **row): + activation = Activation( + id=uuid4(), + revision=2, + name=name, + request=ActivationRequest(version_id=uuid4(), artifact_digest=ARTIFACT, scope=scope), + retired=row.pop("activation_retired", False), + call_depth=call_depth, + ) + spec = {} if callable_field is None else {"callable": callable_field} + version_row = { + "kind": "Workflow", + "name": name, + "version": version, + "retired": False, + "digest": ARTIFACT, + "definition_digest": DEFINITION, + "document": {"spec": spec}, + **row, + } + return PinnedCallee(activation, row.pop("latest_retired", False), version_row) + + +async def pins_for(pinned, *references, supplied=None, max_call_depth=8): + """Pin ``pinned`` for every called reference unless ``supplied`` says otherwise.""" + references = references or (REFERENCE,) + supplied = {r: pinned.activation.id for r in references} if supplied is None else supplied + + async def load(identifier): + return pinned if identifier == pinned.activation.id else None + + return await call_pins( + "order-intake", SCOPE, caller_executable(*references), supplied, load, max_call_depth=max_call_depth + ) + + +async def refused(pinned, **options): + with pytest.raises(CatalogError) as error: + await pins_for(pinned, **options) + assert (error.value.status, error.value.code) == (422, "WV-READINESS-CALL") + return error.value.message + + +async def test_a_valid_pin_records_the_exact_callee_activation_version_and_digests(): + pinned = callee(callable_field={}, call_depth=2) + pins, depth = await pins_for(pinned) + assert pins == [ + WorkflowCallPin( + reference=REFERENCE, + activation_id=pinned.activation.id, + activation_revision=2, + version_id=pinned.activation.request.version_id, + artifact_digest=ARTIFACT, + definition_digest=DEFINITION, + ) + ] + assert depth == 3 + + +async def test_a_workflow_without_calls_needs_no_pins_and_takes_no_extra_ones(): + async def load(identifier): + raise AssertionError("nothing to load") + + assert await call_pins("order-intake", SCOPE, caller_executable(), {}, load) == ([], 0) + with pytest.raises(CatalogError, match="Pin exactly one activation"): + await call_pins("order-intake", SCOPE, caller_executable(), {REFERENCE: uuid4()}, load) + + +async def test_missing_or_extra_pins_are_refused_before_anything_is_loaded(): + pinned = callee(callable_field={}) + assert "Pin exactly one activation" in await refused(pinned, supplied={}) + extra = {REFERENCE: pinned.activation.id, "other@1.0.0": pinned.activation.id} + assert "Pin exactly one activation" in await refused(pinned, supplied=extra) + + +@pytest.mark.parametrize( + ("pinned", "message"), + [ + pytest.param( + callee(callable_field={}, scope=Scope(tenant_id=TENANT, project_id=PROJECT, environment_id=uuid4())), + "not found in this environment", + id="other-environment", + ), + pytest.param(callee(name="invoice", callable_field={}), "does not run the called workflow", id="other-name"), + pytest.param( + callee(version="2.0.0", callable_field={}), "does not run the called workflow", id="other-version" + ), + pytest.param(callee(callable_field={}, kind="Action"), "does not run the called workflow", id="not-a-workflow"), + pytest.param(callee(callable_field={}, activation_retired=True), "retired", id="retired-activation"), + pytest.param(callee(callable_field={}, latest_retired=True), "retired", id="retired-name"), + pytest.param(callee(callable_field={}, retired=True), "retired", id="retired-version"), + pytest.param(callee(callable_field={}, digest="b" * 64), "does not match its published", id="artifact"), + pytest.param(callee(callable_field={}, definition_digest="c" * 64), "compiled against", id="other-definition"), + pytest.param(callee(), "does not allow calls from this workflow", id="not-callable"), + pytest.param( + callee(callable_field={"allowedCallers": ["invoice-run"]}), "does not allow calls", id="other-callers" + ), + pytest.param(callee(callable_field={}, call_depth=8), "deeper than 8", id="too-deep"), + ], +) +async def test_a_pin_is_refused_unless_it_is_a_live_callable_activation_of_the_exact_version(pinned, message): + assert message in await refused(pinned) + + +async def test_a_pin_to_an_unknown_activation_is_refused(): + pinned = callee(callable_field={}) + assert "not found in this environment" in await refused(pinned, supplied={REFERENCE: uuid4()}) + + +async def test_a_loader_that_answers_another_activation_does_not_pin_it(): + pinned = callee(callable_field={}) + + async def load(identifier): + return pinned + + with pytest.raises(CatalogError, match="not found in this environment"): + await call_pins("order-intake", SCOPE, caller_executable(REFERENCE), {REFERENCE: uuid4()}, load) + + +async def test_the_call_depth_limit_counts_the_pinned_chain(): + assert (await pins_for(callee(callable_field={}, call_depth=1), max_call_depth=2))[1] == 2 + assert "deeper than 2" in await refused(callee(callable_field={}, call_depth=2), max_call_depth=2) + + +def executable_with(*effects): + return { + "dependencies": [ + {"kind": "Action", "reference": f"a{i}@1.0.0", "document": {"spec": {"sideEffect": effect}}} + for i, effect in enumerate(effects) + ] + } + + +def test_the_side_effect_of_a_workflow_is_the_strongest_of_its_actions(): + assert side_effect({"dependencies": []}) == "read_only" + assert side_effect(executable_with("read_only", "idempotency_key", "idempotent")) == "idempotency_key" + assert side_effect(executable_with("idempotent", "non_idempotent")) == "non_idempotent" + + +async def test_the_side_effect_of_a_call_tree_follows_every_pin_once(): + child, grandchild = uuid4(), uuid4() + tree = { + child: (executable_with("idempotent"), {"grand@1.0.0": grandchild}), + grandchild: (executable_with("non_idempotent"), {"child@1.0.0": child}), + } + visits = [] + + async def resolve(identifier): + visits.append(identifier) + return tree[identifier] + + assert await call_tree_side_effect(executable_with("read_only"), {"child@1.0.0": child}, resolve) == ( + "non_idempotent" + ) + assert visits == [child, grandchild] + assert await call_tree_side_effect(executable_with("idempotent"), {}, resolve) == "idempotent" + + +def test_activation_fields_for_calls_are_omitted_when_unused(): + request = ActivationRequest(version_id=uuid4(), artifact_digest=ARTIFACT, scope=SCOPE) + activation = Activation(id=uuid4(), revision=1, name="order-intake", request=request) + assert "workflow_activation_ids" not in request.model_dump() + assert {"workflow_call_pins", "call_depth"}.isdisjoint(activation.model_dump()) + pinned = request.model_copy(update={"workflow_activation_ids": {REFERENCE: uuid4()}}) + assert ActivationRequest.model_validate_json(pinned.model_dump_json()) == pinned + with pytest.raises(ValueError): + ActivationRequest.model_validate( + {**request.model_dump(mode="json"), "workflow_activation_ids": {"notify": str(uuid4())}} + )