diff --git a/CHANGELOG.md b/CHANGELOG.md index 28e7e71e..c3fcba1d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,18 @@ SPDX-License-Identifier: Apache-2.0 ## Unreleased +- Run state, history, tasks, deadlines, and incidents name each run of a step by + its instance key, which equals the step ID outside loops, so existing workflows + keep byte-identical transitions and history. Deadline scans, incident + resolution, replay, exported history, and debug sessions find the step behind + any instance key. +- Human task views add `instance_key` and `iteration` for tasks created inside a + loop iteration; both are omitted for every other task. +- The capacity check of one event's work runs at its end and after every 64 + steps or 1 MiB of new step output, instead of after every step. Work over + capacity is still refused, but the refusal can happen at a later step of the + same event, and when a later step of that event fails, the run opens that + step's incident instead. - Keep typing in the selected form field when deferred row autofocus completes. Rapid row additions focus the newest item without overriding a later choice. - Update Studio's YAML parser to 2.8.3 so deeply nested documents produce a diff --git a/docs/guides/human-tasks.md b/docs/guides/human-tasks.md index aa28983c..280a3a07 100644 --- a/docs/guides/human-tasks.md +++ b/docs/guides/human-tasks.md @@ -332,6 +332,11 @@ The list shows tasks you are eligible for; a task manager also sees the other tasks in their authorized workspace. The [business key](execution-management.md#1-choose-identifiers-with-different-jobs) groups runs for one case; it is not unique. +A task's `node_id` is the step that created it. A task created inside a loop +iteration also carries `instance_key`, the key of that run of the step (for +example `review[3]`), and `iteration`, its loop indexes (`[3]`). Both fields are +absent for every other task. + **In Studio**, open **My tasks**: the task shows the status "Ready to claim". The status filter next to the search box, **All tasks** by default, also offers **Ready to claim**, **Claimed**, **Completed**, **Expired**, and **Canceled**. diff --git a/docs/reference/compiler.md b/docs/reference/compiler.md index a21dbfff..861990e7 100644 --- a/docs/reference/compiler.md +++ b/docs/reference/compiler.md @@ -362,6 +362,15 @@ JavaScript (with the `u` flag) and JSON Schema `pattern` use the anchored form `^(?:...)$`. API models declare key fields as `InstanceKeyText` (or `InstanceKeyTextOrEmpty` where `""` is allowed). +At run time the run state, history events, step results, tasks, deadlines, and +incidents name each run of a step by its instance key. Outside loops and repeated +activations that key is the step ID, so existing workflows store exactly what they +stored before. A connector's idempotency key is `{run_id}:{instance key}`: retries +of one run of a step share it, and two runs of a step never do. Expressions still +read a step's output as `/steps/`. A branch frame is keyed by its owner +step, its block name, and the loop indexes of the steps it holds: `checks/credit` +outside loops, `checks/credit[3]` in iteration 3. + ## Import an artifact `import_artifact(bytes_or_text_or_object, *, limits=ArtifactLimits(), features=None)` diff --git a/docs/reference/history-and-replay.md b/docs/reference/history-and-replay.md index cb271ada..2692b1de 100644 --- a/docs/reference/history-and-replay.md +++ b/docs/reference/history-and-replay.md @@ -202,6 +202,10 @@ sidecar, committed in the same transaction as the fact: action, release, and deadline facts. Receipts bind the generation and the selected wait, and bounded signal metadata records how signals were arbitrated. +- Facts and issued commands name each run of a step by its + [instance key](compiler.md#instance-keys), which is the step ID outside loops. + Replay finds the step from the step ID part of a canonical key; any other + spelling makes the history inconsistent. - The expected transition is the SHA-256 of a canonical `classified-v1` safe projection of the kernel state, identified as `sha256-rfc8785-v1`; no duplicate snapshot is stored. diff --git a/src/firefly_weave/contracts/human_tasks.py b/src/firefly_weave/contracts/human_tasks.py index 8a8c7ce8..6472da4a 100644 --- a/src/firefly_weave/contracts/human_tasks.py +++ b/src/firefly_weave/contracts/human_tasks.py @@ -19,9 +19,10 @@ from typing import Literal from uuid import UUID -from pydantic import AwareDatetime, Field +from pydantic import AwareDatetime, Field, model_validator from firefly_weave.contracts.definitions import ContractModel, ResourceName +from firefly_weave.contracts.instance_keys import InstanceKeyTextOrEmpty, node_of, split_instance from firefly_weave.contracts.values import JsonObjectData @@ -76,6 +77,8 @@ class HumanTask(ContractModel): id: UUID run_id: UUID node_id: str + instance_key: InstanceKeyTextOrEmpty = Field(default="", exclude_if=lambda value: not value) + iteration: list[int] = Field(default_factory=list, exclude_if=lambda value: not value) revision: int = Field(ge=1) status: Literal["ready", "claimed", "completed", "expired", "cancelled"] title: str @@ -91,6 +94,14 @@ class HumanTask(ContractModel): decision_actor_id: UUID | None = None output: JsonObjectData | None = None + @model_validator(mode="after") + def _one_step(self) -> "HumanTask": + if self.instance_key and (self.instance_key == self.node_id or node_of(self.instance_key) != self.node_id): + raise ValueError("instance_key must be empty or a longer key of the same step") + if self.iteration != (list(split_instance(self.instance_key).indexes) if self.instance_key else []): + raise ValueError("iteration must list the bracketed indexes of instance_key") + return self + class HumanTaskCommand(ContractModel): expected_revision: int = Field(ge=1) diff --git a/src/firefly_weave/human_tasks/service.py b/src/firefly_weave/human_tasks/service.py index df7615f6..b7742ea0 100644 --- a/src/firefly_weave/human_tasks/service.py +++ b/src/firefly_weave/human_tasks/service.py @@ -44,6 +44,7 @@ TaskGroup, TaskGroupRequest, ) +from firefly_weave.contracts.instance_keys import instance_view from firefly_weave.contracts.values import JsonObject from firefly_weave.definitions.models import CatalogError from firefly_weave.persistence.idempotency import Idempotency, lock @@ -56,9 +57,11 @@ def task_view(row: dict[str, Any]) -> HumanTask: - return HumanTask.model_validate_json( - json.dumps({key: row[key] for key in HumanTask.model_fields if key in row}, default=str) - ) + """The API view of a stored task: its static step ID, plus its instance key and iteration inside loops.""" + values = {key: row[key] for key in HumanTask.model_fields if key in row} + identity = instance_view(row["node_id"]) + values.update(node_id=identity.node_id, instance_key=identity.instance_key, iteration=identity.iteration) + return HumanTask.model_validate_json(json.dumps(values, default=str)) async def human(tx: Transaction, identifier: UUID) -> Principal: diff --git a/src/firefly_weave/operations/debug/simulator.py b/src/firefly_weave/operations/debug/simulator.py index 0bc214a3..ce692e44 100644 --- a/src/firefly_weave/operations/debug/simulator.py +++ b/src/firefly_weave/operations/debug/simulator.py @@ -27,6 +27,7 @@ from firefly_weave.compiler.ir import ActionNode, HumanTaskNode, SignalNode, WaitNode from firefly_weave.contracts.definitions import ActionDefinition, load_definition from firefly_weave.contracts.diagnostics import Diagnostic +from firefly_weave.contracts.instance_keys import InvalidInstanceKey, node_of from firefly_weave.contracts.values import JsonObject, JsonValue from firefly_weave.operations.debug.models import ( Boundary, @@ -169,7 +170,7 @@ def set_breakpoints(self, node_ids: set[str]) -> None: def change() -> None: self.data.breakpoints = sorted(node_ids) - if self.data.paused not in node_ids: + if self.data.paused is not None and node_of(self.data.paused) not in node_ids: self.data.paused = None self._mutate(change) @@ -187,7 +188,7 @@ def run() -> None: self.data.paused = None for _ in range(self.limits.continue_boundaries): selected = self._selected() - if selected in self.data.breakpoints and selected != skip: + if selected is not None and node_of(selected) in self.data.breakpoints and selected != skip: self.data.paused = selected return skip = None @@ -212,8 +213,8 @@ def _step(self) -> bool: if cursor.admitted: if d.node_count >= self.limits.node_reductions: raise DebugError("WV-DEBUG-LIMIT") - node = self._prepared.nodes[cursor.ready[0]] - boundary = Boundary(kind="node", node_id=node.id, node_kind=node.kind) + node = self._prepared.nodes[node_of(cursor.ready[0])] + boundary = Boundary(kind="node", node_id=cursor.ready[0], node_kind=node.kind) d.node_count += 1 else: boundary = Boundary(kind="event", event_type=cursor.event.type) @@ -228,7 +229,10 @@ def _step(self) -> bool: if isinstance(command, Deadline): d.deadlines.append(command) elif isinstance(command, TaskIntent): + # A step mock applies to every instance of the step. key = "node:" + command.node_id + if key not in d.mocks: + key = "node:" + node_of(command.node_id) if key not in d.mocks: key = "action:" + command.action_reference if key not in d.mocks: @@ -261,7 +265,7 @@ def _eligible(self) -> RuntimeEvent | None: continue d.deadlines.remove(deadline) return self._event( - "wait_elapsed" if deadline.node_id in durations else "timed_out", + "wait_elapsed" if node_of(deadline.node_id) in durations else "timed_out", {"node_id": deadline.node_id, "deadline": deadline.deadline.isoformat()}, ) for node in self.ir.graph.nodes: @@ -288,7 +292,11 @@ def _eligible(self) -> RuntimeEvent | None: def human_decision(self, node_id: str, payload: JsonValue) -> DebugView: def receive() -> None: - node = next((n for n in self.ir.graph.nodes if isinstance(n, HumanTaskNode) and n.id == node_id), None) + try: + step = node_of(node_id) + except InvalidInstanceKey: + raise DebugError("WV-DEBUG-HUMAN-TASK") from None + node = next((n for n in self.ir.graph.nodes if isinstance(n, HumanTaskNode) and n.id == step), None) if node is None or node_id not in self.data.state.active or self.data.state.status in TERMINAL: raise DebugError("WV-DEBUG-HUMAN-TASK") validate( diff --git a/src/firefly_weave/operations/exports.py b/src/firefly_weave/operations/exports.py index 362ce1a3..5850810e 100644 --- a/src/firefly_weave/operations/exports.py +++ b/src/firefly_weave/operations/exports.py @@ -17,12 +17,14 @@ """Pure safe evidence projection and stable comparison; no storage or executors.""" import json +from collections.abc import Mapping from typing import cast from firefly_weave.compiler.api import CompiledArtifact from firefly_weave.compiler.canonical import canonical_digest -from firefly_weave.compiler.ir import ActionNode, HumanTaskNode, SignalNode +from firefly_weave.compiler.ir import ActionNode, HumanTaskNode, IRNode, SignalNode from firefly_weave.contracts.definitions import ActionDefinition, load_definition +from firefly_weave.contracts.instance_keys import InvalidInstanceKey, split_instance from firefly_weave.contracts.values import JsonObject, JsonValue from firefly_weave.contracts.workers import TaskError from firefly_weave.operations.redaction import Omission, project @@ -30,6 +32,16 @@ from firefly_weave.runtime.models import RuntimeEvent, Transition +def event_node(nodes: Mapping[str, IRNode], key: object) -> IRNode | None: + """The step named by an event's instance key; None unless the key is canonical and names a step.""" + if not isinstance(key, str): + return None + try: + return nodes.get(split_instance(key).node_id) + except InvalidInstanceKey: + return None + + def lock_digest(artifact: CompiledArtifact) -> str: return canonical_digest({"dependencies": artifact.executable["dependencies"]}) @@ -101,7 +113,7 @@ def safe_event(event: RuntimeEvent, artifact: CompiledArtifact | None, *, unavai ): assert artifact is not None ir = workflow(artifact) - node = next((n for n in ir.graph.nodes if n.id == data.get("node_id")), None) + node = event_node({n.id: n for n in ir.graph.nodes}, data.get("node_id")) schemas: list[JsonObject] = [] if isinstance(node, ActionNode) and event.type in {"task_completed", "incident_resolved"}: definition = load_definition(next(d.document for d in ir.dependencies if d.digest == node.dependency)) diff --git a/src/firefly_weave/operations/incidents.py b/src/firefly_weave/operations/incidents.py index d7065863..d0235886 100644 --- a/src/firefly_weave/operations/incidents.py +++ b/src/firefly_weave/operations/incidents.py @@ -32,6 +32,7 @@ from firefly_weave.compiler.ir import ActionNode from firefly_weave.contracts.access import Scope from firefly_weave.contracts.definitions import ActionDefinition, load_definition +from firefly_weave.contracts.instance_keys import node_of from firefly_weave.contracts.operations import IncidentResolution, IncidentView from firefly_weave.definitions.models import CatalogError from firefly_weave.persistence.uow import Transaction @@ -134,7 +135,9 @@ async def resolve( raise CatalogError(409, "WV-RUNTIME-STATE", "Incident is no longer active") artifact = import_artifact(row["artifact"]) ir = workflow(artifact) - node = next((n for n in ir.graph.nodes if n.id == incident["node_id"]), None) + # A run-wide incident has no step; a step incident names its instance key. + step = None if incident["node_id"] is None else node_of(incident["node_id"]) + node = next((n for n in ir.graph.nodes if n.id == step), None) task = None if request.kind != "terminate": if not isinstance(node, ActionNode) or incident["incident_key"] not in previous.state.incidents: @@ -146,7 +149,7 @@ async def resolve( tasks = await repository.rows( f"SELECT * FROM task_intents WHERE {SCOPE} AND run_id=:run AND node_id=:node FOR UPDATE", run=previous.id, - node=node.id, + node=incident["node_id"], ) if not tasks or tasks[0]["status"] not in {"failed", "incident"}: raise CatalogError(409, "WV-RUNTIME-STATE", "No failed task matches this incident") diff --git a/src/firefly_weave/operations/replay.py b/src/firefly_weave/operations/replay.py index c6a761bd..4560b1c1 100644 --- a/src/firefly_weave/operations/replay.py +++ b/src/firefly_weave/operations/replay.py @@ -25,6 +25,7 @@ from firefly_weave.compiler.api import CompiledArtifact, import_artifact from firefly_weave.compiler.ir import ActionNode, HumanTaskNode, IRNode, SignalNode, WaitNode from firefly_weave.contracts.definitions import ActionDefinition, load_definition +from firefly_weave.contracts.instance_keys import node_of from firefly_weave.contracts.operations import ( DeadlineFact, HumanTaskFact, @@ -33,7 +34,7 @@ ReplayReport, TaskFact, ) -from firefly_weave.operations.exports import bounded_size, lock_digest, safe_event, transition_digest +from firefly_weave.operations.exports import bounded_size, event_node, lock_digest, safe_event, transition_digest from firefly_weave.operations.redaction import project from firefly_weave.runtime.kernel import deadline_order, transition, validate, validate_action, workflow from firefly_weave.runtime.models import Deadline, HumanTaskIntent, RunState, RuntimeEvent, TaskIntent @@ -120,7 +121,7 @@ def report(status: str, code: str, sequence: int) -> ReplayReport: missing=True, ) if event.type not in {"started", "cancelled", "timed_out", "incident_resolved", "paused", "resumed"}: - require(event.data.get("node_id") in nodes, "WV-REPLAY-EVENT") + require(event_node(nodes, event.data.get("node_id")) is not None, "WV-REPLAY-EVENT") projected = safe_event(event, artifact) changes = {key for key in event.data if key not in projected.data} require(changes <= {"reason", "evidence_reference"}, "WV-REPLAY-REDACTED", missing=True) @@ -148,7 +149,7 @@ def report(status: str, code: str, sequence: int) -> ReplayReport: assert state is not None require(state.status not in TERMINAL, "WV-REPLAY-TERMINAL") node_id = str(event.data.get("node_id", "")) - node = nodes.get(node_id) + node = event_node(nodes, node_id) receipt_id = UUID(str(event.data["receipt_id"])) if event.type == "incident_resolved" else event.id require(receipt_id not in accepted_receipts, "WV-REPLAY-DUPLICATE-RECEIPT") if event.type in {"task_completed", "task_failed", "incident_opened", "recovery_scheduled"}: @@ -386,4 +387,4 @@ def _verify_wait( require( event.data.get("wait_id") == str(issued.wait_id) and bool(due) and due[0] == issued, "WV-REPLAY-DEADLINE" ) - require((event.type == "wait_elapsed") == (node_id in duration_nodes), "WV-REPLAY-DEADLINE") + require((event.type == "wait_elapsed") == (node_of(node_id) in duration_nodes), "WV-REPLAY-DEADLINE") diff --git a/src/firefly_weave/runtime/capacity.py b/src/firefly_weave/runtime/capacity.py index 52a71a02..ecbe5cba 100644 --- a/src/firefly_weave/runtime/capacity.py +++ b/src/firefly_weave/runtime/capacity.py @@ -31,6 +31,10 @@ TERMINAL_STATE_BYTES = 128 * 1024 * 1024 MAX_REDUCTIONS = 10_000 MAX_INTENTS = 1_000 +# One event's transition is admitted at its final checkpoint, and earlier once this many reductions or this many +# bytes of new step output have accumulated since the last admission. +ADMISSION_REDUCTIONS = 64 +ADMISSION_OUTPUT_BYTES = 1024 * 1024 capacity_failures: ContextVar[list[bool] | None] = ContextVar("weave_capacity_failures", default=None) diff --git a/src/firefly_weave/runtime/control.py b/src/firefly_weave/runtime/control.py index fb80760a..4798e384 100644 --- a/src/firefly_weave/runtime/control.py +++ b/src/firefly_weave/runtime/control.py @@ -14,26 +14,42 @@ # Author: Firefly Software Foundation # SPDX-License-Identifier: Apache-2.0 -"""Pure lexical frames and deterministic direct-branch capacity accounting.""" +"""Pure lexical frames keyed by instance keys, and deterministic direct-branch capacity accounting.""" from firefly_weave.compiler.ir import Node +from firefly_weave.contracts.instance_keys import node_of, split_instance from firefly_weave.contracts.values import JsonObject, JsonValue from firefly_weave.runtime.models import BranchState, JoinState, RunState -def branch_key(owner: str, name: str) -> str: +def loop_suffix(key: str) -> str: + """The loop iterations of an instance key as key text, outer loop first; "" outside loops.""" + if node_of(key) == key: + return "" + return "".join(f"[{index}]" for index in split_instance(key).indexes) + + +def branch_key(owner: str, name: str, suffix: str = "") -> str: + """A frame key: the owner step ID, its block name and the loop suffix of the steps inside it.""" # IR author IDs and branch names are resource names; separators cannot collide. - return f"{owner}/{name}" + return f"{owner}/{name}{suffix}" -def frame(state: RunState, node: Node) -> BranchState | None: +def frame_key(node: Node, key: str) -> str | None: + """The frame that holds the instance ``key`` of ``node``; None at the top level.""" if not node.scope: return None - return state.branches[branch_key(*node.scope[-2:])] + owner, name = node.scope[-2:] + return branch_key(owner, name, loop_suffix(key)) + + +def frame(state: RunState, node: Node, key: str) -> BranchState | None: + name = frame_key(node, key) + return None if name is None else state.branches[name] -def bindings(state: RunState, node: Node) -> JsonObject: - branch = frame(state, node) +def bindings(state: RunState, node: Node, key: str) -> JsonObject: + branch = frame(state, node, key) return state.steps if branch is None else {**branch.visible, **branch.steps} @@ -44,7 +60,8 @@ def joined_output(state: RunState, join: JoinState) -> JsonValue: return {branch.name: branch.output for branch in sorted(branches, key=lambda branch: branch.name)} -def available(state: RunState, join: JoinState) -> list[BranchState]: +def available(state: RunState, join: JoinState) -> list[str]: + """Keys of the pending branches of ``join`` that may start now, in declaration order.""" running = sum(state.branches[key].status == "running" for key in join.branches) - pending = [state.branches[key] for key in join.branches if state.branches[key].status == "pending"] + pending = [key for key in join.branches if state.branches[key].status == "pending"] return pending[: max(0, join.concurrency - running)] diff --git a/src/firefly_weave/runtime/kernel.py b/src/firefly_weave/runtime/kernel.py index dcb29a5f..e9e3d8a5 100644 --- a/src/firefly_weave/runtime/kernel.py +++ b/src/firefly_weave/runtime/kernel.py @@ -48,17 +48,29 @@ ) from firefly_weave.compiler.schemas import validate_payload from firefly_weave.contracts.definitions import ActionDefinition, ContractModel, WorkerImplementation, load_definition +from firefly_weave.contracts.instance_keys import node_of from firefly_weave.contracts.values import JsonObject, JsonValue from firefly_weave.runtime.capacity import ( + ADMISSION_OUTPUT_BYTES, + ADMISSION_REDUCTIONS, MAX_REDUCTIONS, STATE_BYTES, TERMINAL_STATE_BYTES, + TRANSITION_BYTES, RuntimeCapacityError, admit_transition, capacity_failures, logical_size, ) -from firefly_weave.runtime.control import available, bindings, branch_key, frame, joined_output +from firefly_weave.runtime.control import ( + available, + bindings, + branch_key, + frame, + frame_key, + joined_output, + loop_suffix, +) from firefly_weave.runtime.models import ( BranchState, ControlCommand, @@ -86,8 +98,11 @@ def checked_deadline(timestamp: datetime, duration_seconds: int) -> datetime: def deadline_order(node_id: str, due: datetime, duration_nodes: set[str]) -> tuple[int, datetime, str]: - """Shared observed-due arbitration: run timeout, signal timeout, then duration.""" - return (0 if node_id == "@run" else 2 if node_id in duration_nodes else 1, due, node_id) + """Shared observed-due arbitration: run timeout, signal timeout, then duration. + + ``node_id`` is a deadline's instance key and ``duration_nodes`` holds static step IDs. + """ + return (0 if node_id == "@run" else 2 if node_of(node_id) in duration_nodes else 1, due, node_id) def workflow(artifact: CompiledArtifact) -> WorkflowIR: @@ -204,13 +219,23 @@ def transition(state: RunState, event: RuntimeEvent, artifact: CompiledArtifact capacity_failures.reset(token) logical_size(state, STATE_BYTES) cursor = begin(state, event, artifact) - reductions = 0 + reductions = unadmitted = output_bytes = measured = 0 prepared = prepare(artifact) if artifact is not None and not state.unavailable else None while not cursor.done: reductions += 1 if reductions > MAX_REDUCTIONS: raise RuntimeCapacityError() cursor = _advance_owned(cursor, prepared, operational=True) + # Admission walks the whole transition, so it is amortized: at the final checkpoint, and earlier once + # enough reductions or new step output have accumulated. Only where a capacity error surfaces can move. + unadmitted += 1 + changes = cursor.result.steps + for change in changes[measured:]: + output_bytes += logical_size(change.output, TRANSITION_BYTES) + measured = len(changes) + if cursor.done or unadmitted >= ADMISSION_REDUCTIONS or output_bytes >= ADMISSION_OUTPUT_BYTES: + admit_transition(cursor.result) + unadmitted = output_bytes = 0 return cursor.result @@ -280,8 +305,6 @@ def _advance_owned( current = cursor.current or str(event.data.get("node_id", ir.graph.entry)) def checkpoint(value: Transition, *, done: bool = True) -> KernelCursor: - if operational: - admit_transition(value) return cursor.model_copy( update={ "origin": state, @@ -299,14 +322,21 @@ def action(node: ActionNode) -> tuple[str, ActionDefinition]: assert isinstance(definition, ActionDefinition) return dependency.reference, definition - def complete(node: Node, value: JsonValue, decision: JsonObject | None = None) -> None: - branch = frame(current_state, node) + def successor(key: str) -> str: + # Repeated activations ("#") and loop yields ("~") continue through their owner node. + if "#" in key or "~" in key: + raise KernelError("WV-RUNTIME-EVENT") + return successors[node_of(key)] + loop_suffix(key) + + def complete(node: Node, key: str, value: JsonValue, decision: JsonObject | None = None) -> None: + # Expressions read steps by their static IDs, so bindings keep the step ID; history keeps the instance key. + branch = frame(current_state, node, key) target = current_state.steps if branch is None else branch.steps target[node.id] = {"output": value} - result.steps.append(StepChange(node_id=node.id, status="completed", output=value, decision=decision)) + result.steps.append(StepChange(node_id=key, status="completed", output=value, decision=decision)) - def mapping(node: Node, expression: JsonObject) -> JsonValue: - return evaluate(expression, {"input": current_state.input, "steps": bindings(current_state, node)}) + def mapping(node: Node, key: str, expression: JsonObject) -> JsonValue: + return evaluate(expression, {"input": current_state.input, "steps": bindings(current_state, node, key)}) def guard(path: str, purpose: str, value: JsonValue) -> None: for item in ir.guards: @@ -323,11 +353,12 @@ def incident(node_id: str, code: str, generation: int = 0) -> None: ) def spawn(join: JoinState) -> None: - for branch in available(current_state, join): - branch = branch.model_copy(update={"status": "running"}) - current_state.branches[branch_key(branch.owner, branch.name)] = branch + suffix = loop_suffix(join.owner) + for key in available(current_state, join): + branch = current_state.branches[key].model_copy(update={"status": "running"}) + current_state.branches[key] = branch result.commands.append(ControlCommand(kind="spawn_branch", node_id=join.owner, branch=branch.name)) - queue.append(branch.target) + queue.append(branch.target + suffix) def terminate(status: str, code: str | None = None) -> None: nonlocal current_state @@ -355,15 +386,15 @@ def accept(completion: RuntimeEvent) -> None: identifier = completion.data.get("node_id") if not isinstance(identifier, str) or identifier not in current_state.active: raise KernelError("WV-RUNTIME-EVENT") - node = nodes[identifier] + node = nodes[node_of(identifier)] if isinstance(node, WaitNode) and completion.type == "wait_elapsed": due = current_state.waits.get(identifier) if due is None or completion.data.get("deadline") != due.isoformat() or completion.timestamp < due: raise KernelError("WV-RUNTIME-EARLY-WAKEUP") - complete(node, None) + complete(node, identifier, None) current_state.active.remove(identifier) current_state.waits.pop(identifier) - queue.append(successors[identifier]) + queue.append(successor(identifier)) return if "output" not in completion.data: raise KernelError("WV-RUNTIME-EVENT") @@ -388,10 +419,10 @@ def accept(completion: RuntimeEvent) -> None: raise KernelError("WV-RUNTIME-EVENT") value = completion.data["output"] validate(ir, schema, value) - complete(node, value) + complete(node, identifier, value) current_state.active.remove(identifier) current_state.waits.pop(identifier, None) - queue.append(successors[identifier]) + queue.append(successor(identifier)) try: if not cursor.admitted: @@ -499,7 +530,7 @@ def accept(completion: RuntimeEvent) -> None: code = "WV-TASK-FAILED" if event.type == "task_failed" else event.data.get("code") if not isinstance(code, str) or not code: raise KernelError("WV-RUNTIME-EVENT") - if event.type == "task_failed" and not isinstance(nodes[current], ActionNode): + if event.type == "task_failed" and not isinstance(nodes[node_of(current)], ActionNode): raise KernelError("WV-RUNTIME-EVENT") incident(current, code, generation) return checkpoint(result.model_copy(update={"state": current_state})) @@ -529,24 +560,25 @@ def accept(completion: RuntimeEvent) -> None: return checkpoint(result.model_copy(update={"state": current_state}), done=not queue) while queue: current = queue.pop(0) - node = nodes[current] + node = nodes[node_of(current)] + suffix = loop_suffix(current) if isinstance(node, StartNode): - queue.append(successors[current]) + queue.append(successor(current)) elif isinstance(node, EndNode): - output = mapping(node, cast(JsonObject, node.output.model_dump(by_alias=True))) + output = mapping(node, current, cast(JsonObject, node.output.model_dump(by_alias=True))) validate(ir, ir.schemas[ir.output_schema], output) current_state = current_state.model_copy(update={"output": output, "status": "succeeded"}) return checkpoint(result.model_copy(update={"state": current_state})) elif isinstance(node, TransformNode): - value = mapping(node, cast(JsonObject, node.value.model_dump(by_alias=True))) + value = mapping(node, current, cast(JsonObject, node.value.model_dump(by_alias=True))) guard(node.path + "/value", "transform_output", value) - complete(node, value) - queue.append(successors[current]) + complete(node, current, value) + queue.append(successor(current)) elif isinstance(node, DecisionTableNode): dependency = next( d for d in ir.dependencies if d.digest == node.dependency and d.kind == "DecisionTable" ) - value = mapping(node, cast(JsonObject, node.input.model_dump(by_alias=True))) + value = mapping(node, current, cast(JsonObject, node.input.model_dump(by_alias=True))) guard(node.path + "/with", "decision_input", value) decision = evaluate_decision_table( cast(JsonObject, dependency.document["spec"]), @@ -556,10 +588,11 @@ def accept(completion: RuntimeEvent) -> None: guard(node.path, "decision_output", decision.output) complete( node, + current, decision.output, {"matched_rule_ids": list(decision.matched_rule_ids), "used_default": decision.used_default}, ) - queue.append(successors[current]) + queue.append(successor(current)) elif isinstance(node, FailNode): terminate("failed", node.code) current_state = current_state.model_copy( @@ -567,7 +600,7 @@ def accept(completion: RuntimeEvent) -> None: ) return checkpoint(result.model_copy(update={"state": current_state})) elif isinstance(node, (SwitchNode, ParallelNode)): - if node.id in current_state.joins: + if current in current_state.joins: raise KernelError("WV-RUNTIME-ACTIVATION") if isinstance(node, ParallelNode): targets = [(b.name, b.target) for b in node.branches] @@ -576,85 +609,87 @@ def accept(completion: RuntimeEvent) -> None: targets = [("default", node.default)] concurrency = 1 for case in node.cases: - choice = mapping(node, cast(JsonObject, case.when.model_dump(by_alias=True))) + choice = mapping(node, current, cast(JsonObject, case.when.model_dump(by_alias=True))) if not isinstance(choice, bool): raise KernelError("WV-RUNTIME-PREDICATE") if choice: targets = [(case.branch, case.target)] break - parent = frame(current_state, node) if operational: current_size = logical_size(current_state, STATE_BYTES) - visible_size = logical_size(bindings(current_state, node), STATE_BYTES) + visible_size = logical_size(bindings(current_state, node, current), STATE_BYTES) if current_size + len(targets) * (visible_size + 1024) > STATE_BYTES: raise RuntimeCapacityError() keys = [] for name, target in targets: - key = branch_key(node.id, name) + key = branch_key(node.id, name, suffix) keys.append(key) current_state.branches[key] = BranchState( owner=node.id, name=name, target=target, - parent=branch_key(parent.owner, parent.name) if parent else None, - visible=deepcopy(bindings(current_state, node)), + parent=frame_key(node, current), + visible=deepcopy(bindings(current_state, node, current)), ) join = JoinState( - owner=node.id, - node_id=node.join, + owner=current, + node_id=node.join + suffix, mode="all" if isinstance(node, ParallelNode) else "selected", branches=keys, concurrency=concurrency, required_count=len(keys), ) - current_state.joins[node.id] = join + current_state.joins[current] = join spawn(join) elif isinstance(node, BranchOutputNode): - branch = frame(current_state, node) + branch = frame(current_state, node, current) assert branch is not None if branch.status != "running": raise KernelError("WV-RUNTIME-ACTIVATION") - value = mapping(node, cast(JsonObject, node.output.model_dump(by_alias=True))) + value = mapping(node, current, cast(JsonObject, node.output.model_dump(by_alias=True))) guard(node.path, "branch_output", value) measure_value(value) branch = branch.model_copy(update={"output": value, "status": "completed"}) - current_state.branches[branch_key(node.owner, node.branch)] = branch - result.steps.append(StepChange(node_id=node.id, status="completed", output=value)) - result.commands.append(ControlCommand(kind="accept_output", node_id=node.id, branch=node.branch)) - join = current_state.joins[node.owner] - join.completed.append(branch_key(node.owner, node.branch)) + key = branch_key(node.owner, node.branch, suffix) + current_state.branches[key] = branch + result.steps.append(StepChange(node_id=current, status="completed", output=value)) + result.commands.append(ControlCommand(kind="accept_output", node_id=current, branch=node.branch)) + owner = node.owner + suffix + join = current_state.joins[owner] + join.completed.append(key) join.completed.sort() join = join.model_copy(update={"completed_count": len(join.completed)}) - current_state.joins[node.owner] = join + current_state.joins[owner] = join if join.completed_count == join.required_count: queue.append(join.node_id) else: spawn(join) elif isinstance(node, JoinNode): - join = current_state.joins[node.owner] + owner = node.owner + suffix + join = current_state.joins[owner] if join.status != "waiting" or join.completed_count != join.required_count: raise KernelError("WV-RUNTIME-JOIN") value = joined_output(current_state, join) measure_value(value) - complete(nodes[node.owner], value) - result.steps.append(StepChange(node_id=node.id, status="completed", output=value)) - result.commands.append(ControlCommand(kind="join", node_id=node.id)) - current_state.joins[node.owner] = join.model_copy(update={"status": "joined"}) - queue.append(successors[node.id]) + complete(nodes[node.owner], owner, value) + result.steps.append(StepChange(node_id=current, status="completed", output=value)) + result.commands.append(ControlCommand(kind="join", node_id=current)) + current_state.joins[owner] = join.model_copy(update={"status": "joined"}) + queue.append(successor(current)) elif isinstance(node, (ActionNode, SignalNode, WaitNode, HumanTaskNode)): - if node.id in current_state.active: + if current in current_state.active: raise KernelError("WV-RUNTIME-ACTIVATION") - current_state.active.append(node.id) - result.steps.append(StepChange(node_id=node.id, status="waiting")) + current_state.active.append(current) + result.steps.append(StepChange(node_id=current, status="waiting")) if isinstance(node, HumanTaskNode): - title = mapping(node, cast(JsonObject, node.title.model_dump(by_alias=True))) - context = mapping(node, cast(JsonObject, node.context.model_dump(by_alias=True))) + title = mapping(node, current, cast(JsonObject, node.title.model_dump(by_alias=True))) + context = mapping(node, current, cast(JsonObject, node.context.model_dump(by_alias=True))) validate(ir, {"type": "string", "minLength": 1, "maxLength": 512}, title) validate(ir, {"type": "object"}, context) expiry = checked_deadline(event.timestamp, node.expiry_seconds) if node.expiry_seconds else None result.commands.append( HumanTaskIntent( - node_id=node.id, + node_id=current, assignment=node.assignment, title=cast(str, title), context=cast(JsonObject, context), @@ -665,16 +700,16 @@ def accept(completion: RuntimeEvent) -> None: ) ) if expiry: - current_state.waits[node.id] = expiry - result.commands.append(Deadline(node_id=node.id, deadline=expiry)) + current_state.waits[current] = expiry + result.commands.append(Deadline(node_id=current, deadline=expiry)) elif isinstance(node, ActionNode): reference, definition = action(node) - value = mapping(node, cast(JsonObject, node.input.model_dump(by_alias=True))) + value = mapping(node, current, cast(JsonObject, node.input.model_dump(by_alias=True))) validate_action(ir, definition, "input", value) implementation = definition.spec.implementation result.commands.append( TaskIntent( - node_id=node.id, + node_id=current, action_digest=node.dependency, action_reference=reference, input=value, @@ -692,10 +727,10 @@ def accept(completion: RuntimeEvent) -> None: wait_deadline = checked_deadline( event.timestamp, node.duration_seconds if isinstance(node, WaitNode) else node.timeout_seconds ) - current_state.waits[node.id] = wait_deadline + current_state.waits[current] = wait_deadline result.commands.append( Deadline( - node_id=node.id, + node_id=current, deadline=wait_deadline, name=node.name if isinstance(node, SignalNode) else None, ) diff --git a/src/firefly_weave/runtime/repository.py b/src/firefly_weave/runtime/repository.py index 7693f872..689661b5 100644 --- a/src/firefly_weave/runtime/repository.py +++ b/src/firefly_weave/runtime/repository.py @@ -419,7 +419,8 @@ async def scanner_runs(self, limit: int) -> list[dict[str, Any]]: "AND EXISTS (SELECT 1 FROM jsonb_array_elements(CASE WHEN " "jsonb_typeof(runs.artifact->'executable'->'graph'->'nodes')='array' " "THEN runs.artifact->'executable'->'graph'->'nodes' ELSE '[]'::jsonb END) n " - "WHERE n->>'kind' IN ('signal','humanTask') AND n->>'id'=d.node_id AND NOT EXISTS " + "WHERE n->>'kind' IN ('signal','humanTask') AND n->>'id'=substring(d.node_id from '^[^[#~]+') " + "AND NOT EXISTS " "(SELECT 1 FROM signal_receipts s WHERE s.run_id=runs.id AND s.name=n->>'name' " "AND s.accepted_at action = next( d for d in ir.dependencies if d.kind == "Action" and d.digest == verified.task["payload"]["action_digest"] ) - node = next(n for n in ir.graph.nodes if n.id == verified.task["node_id"]) + step = node_of(verified.task["node_id"]) + node = next(n for n in ir.graph.nodes if n.id == step) assert isinstance(node, ActionNode) bundle = {d.reference: d.document for d in ir.dependencies if d.kind == "Schema"} try: diff --git a/src/firefly_weave/runtime/terminal_capacity.py b/src/firefly_weave/runtime/terminal_capacity.py index 3ea10b11..80d38251 100644 --- a/src/firefly_weave/runtime/terminal_capacity.py +++ b/src/firefly_weave/runtime/terminal_capacity.py @@ -22,6 +22,7 @@ from uuid import NAMESPACE_URL, UUID, uuid4, uuid5 from firefly_weave.compiler.api import import_artifact +from firefly_weave.contracts.instance_keys import node_of from firefly_weave.contracts.integration_events import EventMetadata, IntegrationEvent from firefly_weave.contracts.operations import RecordedEvidence, SignalFact, WaitReceipt from firefly_weave.contracts.runtime import CapacityRunAcknowledgment @@ -53,13 +54,13 @@ async def due_terminal(repository: RuntimeRepository, row: dict[str, Any]) -> Ru due = await repository.rows( f"SELECT id,node_id,deadline FROM run_deadlines WHERE {SCOPE} AND run_id=:run " "AND NOT consumed AND deadline<=clock_timestamp() " - "AND (node_id='@run' OR node_id=ANY(cast(:signals AS text[]))) " + "AND (node_id='@run' OR substring(node_id from '^[^[#~]+')=ANY(cast(:signals AS text[]))) " "ORDER BY (node_id='@run') DESC,deadline,node_id LIMIT 1001", run=row["id"], signals=list(signals or {}), ) for deadline in due: - name = (signals or {}).get(deadline["node_id"]) + name = (signals or {}).get(node_of(deadline["node_id"])) if deadline["node_id"] != "@run" and name is None: continue if name is not None and await repository.rows( diff --git a/src/firefly_weave/runtime/waits.py b/src/firefly_weave/runtime/waits.py index 4ebbca3c..d1085d98 100644 --- a/src/firefly_weave/runtime/waits.py +++ b/src/firefly_weave/runtime/waits.py @@ -22,6 +22,7 @@ from firefly_weave.compiler.api import import_artifact from firefly_weave.compiler.canonical import canonical_digest from firefly_weave.compiler.ir import SignalNode, WaitNode +from firefly_weave.contracts.instance_keys import node_of from firefly_weave.runtime.kernel import deadline_order, workflow from firefly_weave.runtime.kernel import transition_async as transition from firefly_weave.runtime.models import RuntimeEvent @@ -59,7 +60,8 @@ async def settle(repository: RuntimeRepository, row: dict[str, Any], *, terminal if deadline["deadline"] > now: continue # A buffered receipt strictly before this wait deadline wins. Overall timeout always stops the run. - signal = next((n for n in ir.graph.nodes if isinstance(n, SignalNode) and n.id == deadline["node_id"]), None) + step = node_of(deadline["node_id"]) + signal = next((n for n in ir.graph.nodes if isinstance(n, SignalNode) and n.id == step), None) earlier = ( [] if signal is None @@ -73,7 +75,7 @@ async def settle(repository: RuntimeRepository, row: dict[str, Any], *, terminal ) if earlier: continue - elapsed = deadline["node_id"] in duration_nodes + elapsed = step in duration_nodes if terminal_only and elapsed: continue if not elapsed: @@ -181,7 +183,7 @@ async def _unavailable_timeout(repository: RuntimeRepository, row: dict[str, Any if candidate["node_id"] == "@run": selected = candidate break - name = signals.get(candidate["node_id"]) + name = signals.get(node_of(candidate["node_id"])) if name is not None: earlier = await repository.rows( f"SELECT 1 FROM signal_receipts WHERE {SCOPE} AND run_id=:run AND name=:name " diff --git a/tests/acceptance/journeys.toml b/tests/acceptance/journeys.toml index da851f35..3294013c 100644 --- a/tests/acceptance/journeys.toml +++ b/tests/acceptance/journeys.toml @@ -41,7 +41,7 @@ "editor-hardening" = false "language-contracts" = false "text-operators" = true -"instance-keys" = false +"instance-keys" = true "loop-compiler" = false "loop-runtime" = false "loop-simulation" = false diff --git a/tests/contracts/test_golden_instance_keys.py b/tests/contracts/test_golden_instance_keys.py new file mode 100644 index 00000000..f301e92d --- /dev/null +++ b/tests/contracts/test_golden_instance_keys.py @@ -0,0 +1,141 @@ +# 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 + +"""Steps run inside a loop iteration exactly as they run outside one, under their instance keys. + +Every recorded golden transition is replayed a second time with each step instance key carrying the loop +suffix ``[3][0]``: the transition must equal the recorded one with the same suffix added to every instance +key it holds. Expression bindings keep static step IDs, so they are left as recorded. +""" + +import json +from pathlib import Path +from typing import Any + +import pytest + +from firefly_weave.compiler.api import import_artifact +from firefly_weave.runtime.kernel import transition +from firefly_weave.runtime.models import RunState, RuntimeEvent + +CORPUS = json.loads(Path("tests/fixtures/golden/corpus.json").read_text(encoding="utf-8")) +SUFFIX = "[3][0]" + + +def lift_key(key: str) -> str: + return key if key == "@run" else key + SUFFIX + + +def lift_incident(key: str) -> str: + node, _, generation = key.rpartition(":") + return f"{lift_key(node)}:{generation}" + + +def lift_event(event: dict[str, Any]) -> dict[str, Any]: + data = dict(event["data"]) + if "node_id" in data: + data["node_id"] = lift_key(data["node_id"]) + if data.get("incident_key", "@legacy") != "@legacy": + data["incident_key"] = lift_incident(data["incident_key"]) + return {**event, "data": data} + + +def lift_state(state: dict[str, Any]) -> dict[str, Any]: + branches = { + key + SUFFIX: {**branch, "parent": None if branch["parent"] is None else branch["parent"] + SUFFIX} + for key, branch in state["branches"].items() + } + joins = { + lift_key(key): { + **join, + "owner": lift_key(join["owner"]), + "node_id": lift_key(join["node_id"]), + "branches": [branch + SUFFIX for branch in join["branches"]], + "completed": [branch + SUFFIX for branch in join["completed"]], + } + for key, join in state["joins"].items() + } + return { + **state, + "active": [lift_key(key) for key in state["active"]], + "waits": {lift_key(key): due for key, due in state["waits"].items()}, + "branches": branches, + "joins": joins, + "incidents": { + lift_incident(key): {**incident, "node_id": lift_key(incident["node_id"])} + for key, incident in state["incidents"].items() + }, + "deferred_results": [lift_event(event) for event in state["deferred_results"]], + } + + +def lift_transition(recorded: dict[str, Any]) -> dict[str, Any]: + return { + "state": lift_state(recorded["state"]), + "commands": [{**command, "node_id": lift_key(command["node_id"])} for command in recorded["commands"]], + "steps": [{**step, "node_id": lift_key(step["node_id"])} for step in recorded["steps"]], + } + + +def liftable(event: dict[str, Any]) -> bool: + """Events that continue steps; a start, a cancellation, a run timeout or a terminate resolution names none.""" + data = event["data"] + if event["type"] in {"started", "cancelled"} or data.get("node_id") == "@run": + return False + return not (event["type"] == "incident_resolved" and data.get("kind") == "terminate") + + +CASES = [ + pytest.param(source["name"], scenario, id=f"{source['name']}-{scenario['name']}") + for source in CORPUS["sources"] + for scenario in source["scenarios"] +] + + +def test_the_lifted_replay_covers_every_kind_of_continuation(): + kinds = { + step["event"]["type"] + for source in CORPUS["sources"] + for scenario in source["scenarios"] + for step in scenario["stream"] + if liftable(step["event"]) + } + assert kinds == { + "human_completed", + "incident_resolved", + "paused", + "resumed", + "signal_received", + "task_completed", + "task_failed", + "timed_out", + "wait_elapsed", + } + + +@pytest.mark.parametrize(("name", "scenario"), CASES) +def test_recorded_transitions_keep_their_shape_inside_a_loop_iteration(name, scenario): + artifact = import_artifact(next(source for source in CORPUS["sources"] if source["name"] == name)["artifact"]) + state: dict[str, Any] = RunState.model_validate_json(json.dumps({"input": scenario["input"]})).model_dump( + mode="json" + ) + for step in scenario["stream"]: + if liftable(step["event"]): + lifted = RunState.model_validate_json(json.dumps(lift_state(state))) + event = RuntimeEvent.model_validate_json(json.dumps(lift_event(step["event"]))) + result = transition(lifted, event, artifact) + assert result.model_dump(mode="json") == lift_transition(step["transition"]), event.sequence + state = step["transition"]["state"] diff --git a/tests/integration/test_incidents.py b/tests/integration/test_incidents.py index 18eff56c..06d44e94 100644 --- a/tests/integration/test_incidents.py +++ b/tests/integration/test_incidents.py @@ -808,3 +808,45 @@ def record(value): else: state = (await client.get(f"{env_url}/runs/{failed_run.id}", headers=operation_headers)).json()["state"] assert state["status"] == "succeeded" and state["output"] == output + + +@pytest.mark.parametrize("worker_runtime_fixture", ["safe"], indirect=True) +async def test_a_failed_step_instance_is_retried_under_its_instance_key( + client, operation_headers, env_url, failed_run, task_service, transaction_factory, worker_ids, access_db +): + # Move the failed step into iteration 2 of a loop, as a loop body would have stored it. + async with access_db[1].begin() as session: + await session.execute( + text( + """UPDATE runs SET state=jsonb_set(jsonb_set(state,'{active}','["work[2]"]'),'{incidents}',""" + """(SELECT jsonb_object_agg(replace(key,'work:','work[2]:'),jsonb_set(value,'{node_id}','"work[2]"'))""" + """ FROM jsonb_each(state->'incidents'))) WHERE id=:id""" + ), + {"id": failed_run.id}, + ) + await session.execute( + text( + """UPDATE task_intents SET node_id='work[2]',payload=jsonb_set(payload,'{node_id}','"work[2]"'),""" + """operation_key=operation_key||'[2]' WHERE node_id='work'""" + ) + ) + await session.execute( + text("UPDATE incidents SET node_id='work[2]',incident_key=replace(incident_key,'work:','work[2]:')") + ) + await session.execute(text("UPDATE step_instances SET node_id='work[2]' WHERE node_id='work'")) + incident = await incident_for(client, operation_headers, env_url, failed_run) + assert incident["incident_key"].startswith("work[2]:") + response = await client.post( + f"{env_url}/incidents/{incident['id']}/resolve", + headers={**operation_headers, "If-Match": str(incident["revision"])}, + json={"kind": "retry_safe", "reason": "retry read", "receipt_id": str(uuid4())}, + ) + assert response.status_code == 200, response.text + async with transaction_factory() as tx: + lease = (await task_service.claim(tx, worker_ids[0], 1))[0] + assert lease.proof.generation == 2 and lease.operation_key == f"{failed_run.id}:work[2]" + await task_service.complete(tx, lease.proof, uuid4(), 19) + async with access_db[1]() as session: + state = await session.scalar(text("SELECT state FROM runs WHERE id=:id"), {"id": failed_run.id}) + step = await session.scalar(text("SELECT status FROM step_instances WHERE node_id='work[2]'")) + assert (state["status"], state["output"], step) == ("succeeded", 19, "completed") diff --git a/tests/integration/test_instance_keys.py b/tests/integration/test_instance_keys.py new file mode 100644 index 00000000..ca228beb --- /dev/null +++ b/tests/integration/test_instance_keys.py @@ -0,0 +1,69 @@ +# 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 + +"""Stored instance keys drive waits through the real deadline scanner and PostgreSQL.""" + +import asyncio +import json +from uuid import uuid4 + +import pytest +from sqlalchemy import text + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def worker_runtime_fixture(worker_runtime_fixture): + document = json.loads(worker_runtime_fixture["source"]) + document["spec"]["steps"].append({"id": "pause", "kind": "wait", "durationSeconds": 1}) + return {**worker_runtime_fixture, "source": json.dumps(document)} + + +async def test_a_wait_of_a_step_instance_elapses_and_continues_the_run( + queued_task, task_service, transaction_factory, worker_ids, services, access_db, worker_setup +): + from firefly_weave.access.audit import AuditContext + from firefly_weave.runtime.deadlines import DeadlineService + from firefly_weave.runtime.service import RuntimeService + + async with transaction_factory() as tx: + lease = (await task_service.claim(tx, worker_ids[0], 1))[0] + await task_service.complete(tx, lease.proof, uuid4(), 5) + # Move the waiting step into iteration 2 of a loop, as a loop body would have stored it. + async with access_db[1].begin() as session: + await session.execute( + text( + """UPDATE runs SET state=jsonb_set(jsonb_set(state,'{active}','["pause[2]"]'),'{waits}',""" + """jsonb_build_object('pause[2]',state->'waits'->'pause')) WHERE id=:id""" + ), + {"id": queued_task.id}, + ) + await session.execute(text("UPDATE run_deadlines SET node_id='pause[2]' WHERE node_id='pause'")) + await session.execute(text("UPDATE step_instances SET node_id='pause[2]' WHERE node_id='pause'")) + await asyncio.sleep(1.05) + graph = services(access_db[0]) + async with transaction_factory() as tx: + report = await graph.resolve(DeadlineService).scan(tx, 10) + assert (report.elapsed, report.timed_out) == (1, 0) + actor, scope = worker_setup[3:5] + state = (await graph.resolve(RuntimeService).read(actor, scope, queued_task.id, context=AuditContext())).state + assert state.status == "succeeded" and state.output == 5 and state.waits == {} + async with access_db[1]() as session: + event = (await session.execute(text("SELECT data FROM run_events WHERE type='wait_elapsed'"))).scalar_one() + consumed = await session.scalar(text("SELECT consumed FROM run_deadlines WHERE node_id='pause[2]'")) + step = await session.scalar(text("SELECT status FROM step_instances WHERE node_id='pause[2]'")) + assert event["node_id"] == "pause[2]" and consumed and step == "completed" diff --git a/tests/integration/test_secret_admission.py b/tests/integration/test_secret_admission.py index b3e6844a..4c027a9d 100644 --- a/tests/integration/test_secret_admission.py +++ b/tests/integration/test_secret_admission.py @@ -912,3 +912,32 @@ async def small_page(self, release, references, *, limit=100): for row in (await session.execute(text("SELECT state,request,activation FROM runs"))).all(): assert canary not in json.dumps(list(row)) await assert_absent(access_db[1], canary, canonical_digest(canary), retained={"runs": retained_runs}) + + +@pytest.mark.parametrize("worker_runtime_fixture", ["signal"], indirect=True) +async def test_legacy_signal_deadline_of_a_step_instance_still_times_out( + task_service, transaction_factory, queued_task, worker_ids, access_db, services +): + from firefly_weave.runtime.deadlines import DeadlineService + + async with access_db[1].begin() as session: + await session.execute(text("UPDATE runs SET state=state-'admission_policy'")) + async with transaction_factory() as tx: + assert await task_service.claim(tx, worker_ids[0], 1) == [] + async with access_db[1].begin() as session: + assert await session.scalar(text("SELECT terminal_signals_verified FROM run_policy_blocks")) + await session.execute( + text( + "UPDATE run_deadlines SET node_id='approval[2]',deadline=clock_timestamp()-interval '1 second' " + "WHERE node_id='approval'" + ) + ) + service = services(access_db[0]).resolve(DeadlineService) + async with transaction_factory() as tx: + assert await service.tick(tx, 1) == 1 + async with transaction_factory() as tx: + assert await service.tick(tx, 1) == 0 + async with access_db[1]() as session: + assert await session.scalar(text("SELECT state->>'status' FROM runs")) == "timed_out" + event = (await session.execute(text("SELECT data FROM run_events WHERE type='timed_out'"))).scalar_one() + assert event["node_id"] == "approval[2]" and event["code"] == "WV-LEGACY-UNAVAILABLE" diff --git a/tests/integration/test_terminal_capacity.py b/tests/integration/test_terminal_capacity.py index 6e2b8a7c..93013bef 100644 --- a/tests/integration/test_terminal_capacity.py +++ b/tests/integration/test_terminal_capacity.py @@ -272,3 +272,31 @@ async def test_capacity_rejection_still_allows_issued_signal_timeout( if mode != "timeout": assert not await owner.scalar(text("SELECT count(*) FROM wait_wakeups")) assert not await owner.scalar(text("SELECT count(*) FROM signal_receipts WHERE consumed")) + + +async def test_sql_signal_timeout_of_a_step_instance_matches_the_kernel( + waiting_run, access_db, author, services, monkeypatch +): + import firefly_weave.runtime.repository as repository + from firefly_weave.persistence.uow import UnitOfWork + from firefly_weave.runtime.deadlines import DeadlineService + + identifier = UUID(waiting_run) + async with access_db[1].begin() as owner: + await owner.execute( + text( + "UPDATE run_deadlines SET node_id='approval[2]',deadline=clock_timestamp()-interval '1 second' " + "WHERE node_id='approval'" + ) + ) + before = (await owner.execute(text("SELECT * FROM runs WHERE id=:id"), {"id": identifier})).mappings().one() + await exhaust_pools(owner, author[2].project_id) + monkeypatch.setattr(repository, "RUN_FETCH_BYTES", 1) + async with UnitOfWork(access_db[0]).open(author[2]) as tx: + assert await services(access_db[0]).resolve(DeadlineService).tick(tx, 100) == 1 + async with access_db[1]() as owner: + await assert_equivalent(owner, before, "timed_out") + data = await owner.scalar( + text("SELECT data FROM run_events WHERE run_id=:id AND type='timed_out'"), {"id": identifier} + ) + assert data["node_id"] == "approval[2]" diff --git a/tests/unit/operations/test_instance_key_lookups.py b/tests/unit/operations/test_instance_key_lookups.py new file mode 100644 index 00000000..45b63875 --- /dev/null +++ b/tests/unit/operations/test_instance_key_lookups.py @@ -0,0 +1,140 @@ +# 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 + +"""History, debugging and task checks find a step from any of its instance keys.""" + +import json +from datetime import UTC, datetime +from types import SimpleNamespace +from uuid import uuid4 + +import pytest + +from firefly_weave.compiler.api import compile_source +from firefly_weave.compiler.catalog import CatalogSnapshot +from firefly_weave.operations.debug.simulator import Simulator +from firefly_weave.operations.exports import event_node, safe_event +from firefly_weave.runtime.kernel import workflow +from firefly_weave.runtime.models import RuntimeEvent + +NOW = datetime(2026, 1, 1, tzinfo=UTC) + + +def pause_flow(): + result = compile_source( + json.dumps( + { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": {"name": "pause", "version": "1.0.0"}, + "spec": { + "inputSchema": {}, + "outputSchema": {}, + "steps": [{"id": "pause", "kind": "wait", "durationSeconds": 5}], + "output": {"literal": 1}, + }, + } + ), + format="json", + catalog=CatalogSnapshot.empty(), + ) + assert result.ok, result.to_bytes() + return result.artifact + + +@pytest.mark.parametrize(("key", "step"), [("work", "work"), ("work[2]", "work"), ("work[2][0]#1.3~4", "work")]) +def test_an_event_key_names_its_step(worker_runtime_fixture, key, step): + nodes = {node.id: node for node in workflow(worker_runtime_fixture["artifact"]).graph.nodes} + assert event_node(nodes, key) is nodes[step] + + +@pytest.mark.parametrize("key", ["work[02]", "work[2", "", "missing[1]", None, 3]) +def test_a_malformed_or_unknown_key_names_no_step(worker_runtime_fixture, key): + nodes = {node.id: node for node in workflow(worker_runtime_fixture["artifact"]).graph.nodes} + assert event_node(nodes, key) is None + + +@pytest.mark.parametrize(("key", "kept"), [("work[2]", True), ("work[02]", False)]) +def test_history_keeps_the_output_of_a_step_instance(worker_runtime_fixture, key, kept): + event = RuntimeEvent( + id=uuid4(), + type="task_completed", + timestamp=NOW, + sequence=2, + data={"node_id": key, "generation": 1, "output": 9}, + ) + projected = safe_event(event, worker_runtime_fixture["artifact"]) + assert ("output" in projected.data) is kept + assert projected.data["node_id"] == key + + +def test_task_output_checks_find_the_step_of_an_instance(worker_runtime_fixture): + from firefly_weave.runtime.service import RuntimeService + + artifact = worker_runtime_fixture["artifact"] + action = next(d for d in workflow(artifact).dependencies if d.kind == "Action") + verified = SimpleNamespace( + run={"artifact": json.loads(artifact.to_bytes())}, + task={"node_id": "work[2]", "payload": {"action_digest": action.digest}}, + ) + assert RuntimeService.classify_task_output(RuntimeService.__new__(RuntimeService), verified, 9) is None + + +def test_a_debug_session_elapses_a_wait_of_a_step_instance(): + simulator = Simulator(pause_flow(), mocks={}, input={}, now=NOW) + waiting = simulator.continue_until_breakpoint() + assert waiting.status == "waiting" and waiting.active_nodes == ["pause"] + data = json.loads(simulator.serialize()) + data["state"]["active"] = ["pause[2]"] + data["state"]["waits"] = {"pause[2]": data["state"]["waits"]["pause"]} + data["deadlines"] = [{**deadline, "node_id": "pause[2]"} for deadline in data["deadlines"]] + lifted = Simulator.restore(json.dumps(data)) + lifted.advance_time(5) + done = lifted.continue_until_breakpoint() + assert done.status == "succeeded" + assert (done.events[-1].type, done.events[-1].data["node_id"]) == ("wait_elapsed", "pause[2]") + + +def test_a_step_mock_and_a_static_breakpoint_apply_to_every_instance(worker_runtime_fixture): + simulator = Simulator(worker_runtime_fixture["artifact"], mocks={"node:work": 9}, input=3, now=NOW) + started = simulator.next() + assert started.boundary.kind == "event" + data = json.loads(simulator.serialize()) + assert data["cursor"]["ready"] == ["@start"] + data["cursor"]["ready"] = ["@start[2]"] + lifted = Simulator.restore(json.dumps(data)) + lifted.set_breakpoints({"work"}) + paused = lifted.continue_until_breakpoint() + assert paused.status == "paused" and paused.selected_node == "work[2]" + assert paused.boundary.node_id == "@start[2]" and paused.boundary.node_kind == "start" + done = lifted.continue_until_breakpoint() + assert done.status == "succeeded" and done.variables["output"] == 9 + assert done.events[-1].data == {"node_id": "work[2]", "output": 9} + + +def test_re_setting_a_step_breakpoint_keeps_an_instance_paused(worker_runtime_fixture): + simulator = Simulator(worker_runtime_fixture["artifact"], mocks={"node:work": 9}, input=3, now=NOW) + simulator.next() + data = json.loads(simulator.serialize()) + data["cursor"]["ready"] = ["@start[2]"] + lifted = Simulator.restore(json.dumps(data)) + lifted.set_breakpoints({"work"}) + assert lifted.continue_until_breakpoint().selected_node == "work[2]" + lifted.set_breakpoints({"work"}) + kept = lifted.inspect() + assert kept.status == "paused" and kept.selected_node == "work[2]" + lifted.set_breakpoints(set()) + assert lifted.inspect().status != "paused" diff --git a/tests/unit/runtime/test_amortized_admission.py b/tests/unit/runtime/test_amortized_admission.py new file mode 100644 index 00000000..3a80a2da --- /dev/null +++ b/tests/unit/runtime/test_amortized_admission.py @@ -0,0 +1,95 @@ +# 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 + +"""One event's transition is admitted at its end and after every 64 reductions or mebibyte of new output.""" + +import pytest +from test_kernel import artifact, event + +from firefly_weave.runtime import kernel +from firefly_weave.runtime.kernel import transition +from firefly_weave.runtime.models import RunState + + +def chain(count: int, *, value: dict | None = None): + steps = [{"id": f"copy{i}", "kind": "transform", "value": value or {"literal": i}} for i in range(count)] + return artifact(steps, {"literal": True}) + + +def admissions(monkeypatch) -> list[int]: + calls: list[int] = [] + admit = kernel.admit_transition + + def counted(result): + calls.append(len(result.steps)) + admit(result) + + monkeypatch.setattr(kernel, "admit_transition", counted) + return calls + + +@pytest.mark.parametrize( + ("steps", "expected"), + [(1, [1]), (61, [61]), (62, [62, 62]), (125, [62, 125]), (126, [62, 126, 126])], +) +def test_admission_runs_at_the_end_and_after_every_64_reductions(monkeypatch, steps, expected): + calls = admissions(monkeypatch) + result = transition(RunState(), event(), chain(steps)) + assert result.state.status == "succeeded" and len(result.steps) == steps + assert calls == expected + + +def test_admission_runs_once_a_mebibyte_of_new_step_output_accumulates(monkeypatch): + calls = admissions(monkeypatch) + result = transition(RunState(input="x" * 300_000), event(), chain(8, value={"ref": "/input"})) + assert result.state.status == "succeeded" + assert calls == [4, 8, 8] + + +def test_an_amortized_admission_still_refuses_an_oversized_transition(): + flow = chain(40, value={"ref": "/input"}) + initial = RunState(input="x" * 900_000) + with pytest.raises(ValueError, match="WV-RUNTIME-LIMIT"): + transition(initial, event(), flow) + assert initial.steps == {} and initial.status == "queued" + + +@pytest.mark.parametrize(("length", "expected"), [(262_142, [4, 8, 8]), (262_141, [5, 8])]) +def test_the_output_threshold_is_exactly_one_mebibyte(monkeypatch, length, expected): + # Each output is the input string: its length plus two quote bytes. Four outputs of 262,144 bytes reach + # exactly 1,048,576 and admit; one byte less each and the fifth output is the one that crosses. + assert 4 * (length + 2) - kernel.ADMISSION_OUTPUT_BYTES in {0, -4} + calls = admissions(monkeypatch) + result = transition(RunState(input="x" * length), event(), chain(8, value={"ref": "/input"})) + assert result.state.status == "succeeded" + assert calls == expected + + +def test_an_expression_failure_after_unadmitted_growth_ends_in_an_incident(monkeypatch): + # Admission is amortized, so the 33rd copy of a 1 MB input is not yet admitted when the next step fails. + # The failure rolls the event's work back behind an incident, so nothing oversized is returned and the + # outcome is deterministic. Admitting every reduction would have refused the event first. + steps = [{"id": f"copy{i}", "kind": "transform", "value": {"ref": "/input"}} for i in range(33)] + steps.append({"id": "broken", "kind": "transform", "value": {"ref": "/input/missing"}}) + flow = artifact(steps, {"literal": True}) + initial = RunState(input="x" * 1_000_000) + result = transition(initial, event(), flow) + assert (result.state.status, result.state.incident) == ("suspended", "WV-EXPR-MISSING") + assert not result.steps and result.state.steps == {} + kernel.admit_transition(result) + monkeypatch.setattr(kernel, "ADMISSION_REDUCTIONS", 1) + with pytest.raises(ValueError, match="WV-RUNTIME-LIMIT"): + transition(initial, event(), flow) diff --git a/tests/unit/runtime/test_human_task.py b/tests/unit/runtime/test_human_task.py index 5968e20b..c0154a67 100644 --- a/tests/unit/runtime/test_human_task.py +++ b/tests/unit/runtime/test_human_task.py @@ -18,6 +18,7 @@ from datetime import timedelta +import pytest from test_kernel import artifact, event from firefly_weave.runtime.kernel import transition @@ -124,3 +125,42 @@ def test_simulator_requires_explicit_native_human_decision(): done = simulator.continue_until_breakpoint() assert done.status == "succeeded" assert simulator.data.events[-1].type == "human_completed" + + +def waiting_review_instance(): + import json + + from firefly_weave.operations.debug.simulator import Simulator + + simulator = Simulator( + artifact([human()], {"ref": "/steps/review/output"}), mocks={}, input={}, now=event().timestamp + ) + assert simulator.continue_until_breakpoint().active_nodes == ["review"] + data = json.loads(simulator.serialize()) + data["state"]["active"] = ["review[2]"] + return Simulator.restore(json.dumps(data)) + + +def test_simulator_decides_a_human_task_instance(): + from firefly_weave.operations.debug.models import DebugCommand + + simulator = waiting_review_instance() + simulator.command( + DebugCommand(kind="human_decision", name="review[2]", payload={"decision": "approve", "data": {"note": "ok"}}) + ) + done = simulator.continue_until_breakpoint() + assert done.status == "succeeded" + completed = simulator.data.events[-1] + assert (completed.type, completed.data["node_id"]) == ("human_completed", "review[2]") + + +@pytest.mark.parametrize("name", ["[2]", "#a", "~1", "review[02]", "review"]) +def test_simulator_rejects_a_malformed_or_inactive_human_task_key(name): + from firefly_weave.operations.debug.models import DebugCommand, DebugError + + simulator = waiting_review_instance() + with pytest.raises(DebugError) as raised: + simulator.command( + DebugCommand(kind="human_decision", name=name, payload={"decision": "approve", "data": {"note": "ok"}}) + ) + assert raised.value.code == "WV-DEBUG-HUMAN-TASK" diff --git a/tests/unit/runtime/test_instance_identity.py b/tests/unit/runtime/test_instance_identity.py new file mode 100644 index 00000000..f9851faf --- /dev/null +++ b/tests/unit/runtime/test_instance_identity.py @@ -0,0 +1,272 @@ +# 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 kernel keys runtime state by instance keys and looks up steps by their static IDs.""" + +import json +from datetime import UTC, datetime +from uuid import uuid4 + +import pytest +from hypothesis import given +from hypothesis import strategies as st +from test_kernel import event + +from firefly_weave.contracts.instance_keys import InstanceKey, InvalidInstanceKey, format_instance +from firefly_weave.contracts.values import MAX_SAFE_INTEGER +from firefly_weave.runtime.control import frame_key, loop_suffix +from firefly_weave.runtime.kernel import deadline_order, prepare, transition +from firefly_weave.runtime.models import RunState, RuntimeEvent + +NOW = datetime(2026, 9, 30, tzinfo=UTC) + + +def lifted(state: RunState, suffix: str) -> dict: + """The JSON of a parallel run state moved into the loop iteration ``suffix``.""" + data = state.model_dump(mode="json") + data["active"] = [key + suffix for key in data["active"]] + data["branches"] = {key + suffix: branch for key, branch in data["branches"].items()} + data["joins"] = { + key + suffix: { + **join, + "owner": join["owner"] + suffix, + "node_id": join["node_id"] + suffix, + "branches": [key + suffix for key in join["branches"]], + } + for key, join in data["joins"].items() + } + return data + + +def completed(state: RunState, node_id: str, output: int, flow) -> RunState: + completion = RuntimeEvent( + id=uuid4(), + type="task_completed", + timestamp=NOW, + sequence=state.accepted_sequence + 1, + data={"node_id": node_id, "output": output}, + ) + return transition(state, completion, flow).state + + +@pytest.mark.parametrize( + ("key", "suffix"), + [ + ("send", ""), + ("@join:fork", ""), + ("send[3]", "[3]"), + ("send[3][0]", "[3][0]"), + ("@join:fork[2]", "[2]"), + ("@branch:fork:0[12]", "[12]"), + ("support[3]#2.1.review", "[3]"), + ("support#2", ""), + ("notify[1]~2", "[1]"), + ], +) +def test_the_loop_suffix_is_the_bracketed_part_of_a_key(key, suffix): + assert loop_suffix(key) == suffix + + +@given( + st.builds( + InstanceKey, + st.from_regex(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,12}|@join:[a-z]{1,5}|@branch:[a-z]{1,5}:[0-9]", fullmatch=True), + st.lists(st.integers(0, MAX_SAFE_INTEGER), max_size=4).map(tuple), + st.lists(st.from_regex(r"[A-Za-z0-9]{1,6}", fullmatch=True), max_size=3).map(tuple), + st.none() | st.integers(1, MAX_SAFE_INTEGER), + ) +) +def test_every_key_starts_with_its_step_and_loop_suffix(parsed): + key = format_instance(parsed) + suffix = loop_suffix(key) + assert suffix == "".join(f"[{index}]" for index in parsed.indexes) + assert key.startswith(parsed.node_id + suffix) + + +@pytest.mark.parametrize("key", ["send[03]", "send[", "send[1]x", "[1]"]) +def test_a_non_canonical_key_has_no_loop_suffix(key): + with pytest.raises(InvalidInstanceKey): + loop_suffix(key) + + +def test_frames_add_the_loop_suffix_of_the_steps_they_hold(parallel_artifact): + nodes = prepare(parallel_artifact).nodes + assert frame_key(nodes["a"], "a") == "fork/a" + assert frame_key(nodes["a"], "a[2]") == "fork/a[2]" + assert frame_key(nodes["a"], "a[2][7]") == "fork/a[2][7]" + assert frame_key(nodes["fork"], "fork[2]") is None + + +def test_a_parallel_inside_an_iteration_runs_in_frames_of_that_iteration(parallel_artifact, parallel_state): + state = RunState.model_validate_json(json.dumps(lifted(parallel_state, "[2]"))) + assert set(state.branches) == {"fork/a[2]", "fork/b[2]"} + first = completed(state, "a[2]", 1, parallel_artifact) + assert first.branches["fork/a[2]"].status == "completed" and first.active == ["b[2]"] + assert first.joins["fork[2]"].completed == ["fork/a[2]"] + done = completed(first, "b[2]", 2, parallel_artifact) + assert done.status == "succeeded" and done.output == {"a": 1, "b": 2} + assert done.joins["fork[2]"].status == "joined" + assert done.steps["fork"] == {"output": {"a": 1, "b": 2}} + + +def test_iterations_of_one_parallel_never_share_frames_or_joins(parallel_artifact, parallel_state): + second, third = lifted(parallel_state, "[2]"), lifted(parallel_state, "[3]") + both = { + **second, + "active": second["active"] + third["active"], + "branches": {**second["branches"], **third["branches"]}, + "joins": {**second["joins"], **third["joins"]}, + } + state = completed(RunState.model_validate_json(json.dumps(both)), "a[2]", 1, parallel_artifact) + assert state.branches["fork/a[2]"].status == "completed" + assert state.branches["fork/a[3]"].status == "running" and state.branches["fork/a[3]"].output is None + assert state.joins["fork[2]"].completed == ["fork/a[2]"] and state.joins["fork[3]"].completed == [] + assert state.active == ["b[2]", "a[3]", "b[3]"] + + +def test_a_repeated_activation_has_no_successor_of_its_own(worker_runtime_fixture): + flow = worker_runtime_fixture["artifact"] + waiting = transition(RunState(input=3), event(), flow).state + assert waiting.active == ["work"] + state = RunState.model_validate_json(json.dumps({**waiting.model_dump(mode="json"), "active": ["work#1"]})) + result = transition(state, event("task_completed", 2, {"node_id": "work#1", "output": 9}), flow) + assert result.state.status == "suspended" and result.state.incident == "WV-RUNTIME-EVENT" + assert not result.steps and result.state.active == ["work#1"] + + +def test_deadline_order_reads_the_step_of_an_instance_key(): + due = datetime(2026, 9, 30, tzinfo=UTC) + assert deadline_order("pause[2]", due, {"pause"}) == (2, due, "pause[2]") + assert deadline_order("approval[2]", due, {"pause"}) == (1, due, "approval[2]") + assert deadline_order("@run", due, {"pause"}) == (0, due, "@run") + + +def compiled(catalog, steps, output): + from firefly_weave.compiler.api import compile_source + + document = { + "apiVersion": "weave/v1alpha1", + "kind": "Workflow", + "metadata": {"name": "nested", "version": "1.0.0"}, + "spec": {"inputSchema": {}, "outputSchema": {}, "steps": steps, "output": output}, + } + result = compile_source(json.dumps(document), format="json", catalog=catalog) + assert result.ok, result.to_bytes() + return result.artifact + + +def echo(identifier: str) -> dict: + return {"id": identifier, "kind": "action", "uses": "echo-action@2.0.0", "with": {"ref": "/input"}} + + +def nested_switch(catalog): + """A switch inside a parallel branch; its predicate and its case output read the branch's step ``seed``.""" + pick = { + "id": "pick", + "kind": "switch", + "cases": [ + { + "when": {"op": {"name": "eq", "args": [{"ref": "/steps/seed/output"}, {"literal": 1}]}}, + "steps": [], + "output": {"ref": "/steps/seed/output"}, + } + ], + "default": {"steps": [], "output": {"literal": 0}}, + } + outer = { + "id": "outer", + "kind": "parallel", + "concurrency": 2, + "branches": { + "a": {"steps": [echo("seed"), pick], "output": {"ref": "/steps/pick/output"}}, + "b": {"steps": [echo("other")], "output": {"ref": "/steps/other/output"}}, + }, + } + return compiled(catalog, [outer], {"ref": "/steps/outer/output"}) + + +def test_a_switch_nested_in_an_iteration_reads_and_opens_frames_of_that_iteration(worker_runtime_fixture): + flow = nested_switch(worker_runtime_fixture["catalog"]) + started = transition(RunState(input=1), event(), flow).state + assert started.active == ["seed", "other"] + second, third = lifted(started, "[2]"), lifted(started, "[3]") + both = { + **second, + "active": second["active"] + third["active"], + "branches": {**second["branches"], **third["branches"]}, + "joins": {**second["joins"], **third["joins"]}, + } + state = completed(RunState.model_validate_json(json.dumps(both)), "seed[2]", 1, flow) + state = completed(state, "seed[3]", 5, flow) + inner = {key: branch for key, branch in state.branches.items() if branch.owner == "pick"} + # Iteration 2 took the case (seed is 1 there), iteration 3 the default (seed is 5 there). + assert {key: branch.name for key, branch in inner.items()} == { + "pick/case:0[2]": "case:0", + "pick/default[3]": "default", + } + assert inner["pick/case:0[2]"].parent == "outer/a[2]" and inner["pick/default[3]"].parent == "outer/a[3]" + assert inner["pick/case:0[2]"].visible == {"seed": {"output": 1}} + assert inner["pick/default[3]"].visible == {"seed": {"output": 5}} + assert state.branches["outer/a[2]"].output == 1 and state.branches["outer/a[3]"].output == 0 + assert state.joins["pick[2]"].owner == "pick[2]" and state.joins["pick[3]"].owner == "pick[3]" + + +@pytest.mark.parametrize(("iteration", "refused"), [("[3]", False), ("[2]", True)]) +def test_a_block_activates_once_per_iteration(worker_runtime_fixture, iteration, refused): + # Branches with literal outputs finish inside one event, so only the join check can refuse a repeat. + fork = { + "id": "fork", + "kind": "parallel", + "concurrency": 2, + "branches": {n: {"steps": [], "output": {"literal": n}} for n in ("a", "b")}, + } + flow = compiled(worker_runtime_fixture["catalog"], [echo("first"), fork], {"ref": "/steps/fork/output"}) + waiting = transition(RunState(input=1), event(), flow).state + joined = completed(waiting, "first", 7, flow) + assert joined.status == "succeeded" and joined.joins["fork"].status == "joined" + # Iteration 2 has already run its parallel; the step before it is delivered again in one iteration. + data = waiting.model_dump(mode="json") + data.update({key: lifted(joined, "[2]")[key] for key in ("branches", "joins")}, active=["first" + iteration]) + result = transition( + RunState.model_validate_json(json.dumps(data)), + event("task_completed", waiting.accepted_sequence + 1, {"node_id": "first" + iteration, "output": 8}), + flow, + ) + if refused: + assert result.state.status == "suspended" and result.state.incident == "WV-RUNTIME-ACTIVATION" + assert not result.steps and set(result.state.joins) == {"fork[2]"} + else: + assert result.state.status == "succeeded" and result.state.output == {"a": "a", "b": "b"} + assert set(result.state.joins) == {"fork[2]", "fork[3]"} + assert result.state.branches["fork/a[3]"].status == "completed" + + +def test_a_loop_yield_key_has_no_successor_of_its_own(): + from test_kernel import artifact + + flow = artifact([{"id": "pause", "kind": "wait", "durationSeconds": 5}], {"literal": 1}) + waiting = transition(RunState(), event(), flow).state + assert waiting.active == ["pause"] + due = waiting.waits["pause"] + data = waiting.model_dump(mode="json") + data["active"] = ["pause~1"] + data["waits"] = {"pause~1": data["waits"]["pause"]} + elapsed = event("wait_elapsed", 2, {"node_id": "pause~1", "deadline": due.isoformat()}) + result = transition( + RunState.model_validate_json(json.dumps(data)), elapsed.model_copy(update={"timestamp": due}), flow + ) + assert result.state.status == "suspended" and result.state.incident == "WV-RUNTIME-EVENT" + assert not result.steps and result.state.active == ["pause~1"] and "pause~1" in result.state.waits diff --git a/tests/unit/test_acceptance_journeys.py b/tests/unit/test_acceptance_journeys.py index 5780c377..a8ca77fe 100644 --- a/tests/unit/test_acceptance_journeys.py +++ b/tests/unit/test_acceptance_journeys.py @@ -129,7 +129,7 @@ def test_the_committed_map_is_valid(): enablement = journeys.Enablement.load(ROOT / "tests/acceptance/journeys.toml") assert enablement.capabilities["acceptance-foundations"] is True landed = [name for name, value in enablement.capabilities.items() if value] - assert landed == ["text-operators", "ai-gateway", "acceptance-foundations"] + assert landed == ["text-operators", "instance-keys", "ai-gateway", "acceptance-foundations"] assert enablement.journey_steps("J0") == tuple(f"J0.{n}" for n in range(1, 15)) assert enablement.profiles["pr"] == ("J0", "J1", "J2", "J3", "J5", "J8", "J9") assert enablement.criteria["J3"] == ("SC1", "SC3", "SC7") diff --git a/tests/unit/test_human_task_view.py b/tests/unit/test_human_task_view.py new file mode 100644 index 00000000..df438ab4 --- /dev/null +++ b/tests/unit/test_human_task_view.py @@ -0,0 +1,72 @@ +# 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 human task names its step, and inside loops also its instance key and iteration.""" + +from datetime import UTC, datetime +from uuid import uuid4 + +import pytest +from pydantic import ValidationError + +from firefly_weave.contracts.human_tasks import HumanTask +from firefly_weave.human_tasks.service import task_view + + +def row(node_id: str) -> dict: + return { + "tenant_id": uuid4(), + "id": uuid4(), + "run_id": uuid4(), + "node_id": node_id, + "revision": 1, + "status": "ready", + "title": "Review the invoice", + "context": {}, + "form_schema": {}, + "decisions": ["approve"], + "assignment": {"binding_id": str(uuid4()), "revision": 1}, + "created_at": datetime(2026, 10, 13, tzinfo=UTC), + } + + +def test_a_task_outside_loops_keeps_its_exact_view(): + view = task_view(row("review")).model_dump(mode="json") + assert view["node_id"] == "review" + assert "instance_key" not in view and "iteration" not in view + + +@pytest.mark.parametrize(("key", "iteration"), [("review[3]", [3]), ("review[3][0]", [3, 0])]) +def test_a_task_inside_loops_names_its_step_key_and_iteration(key, iteration): + view = task_view(row(key)) + assert (view.node_id, view.instance_key, view.iteration) == ("review", key, iteration) + assert view.model_dump(mode="json")["iteration"] == iteration + + +@pytest.mark.parametrize( + ("node_id", "instance_key", "iteration"), + [("review", "review", []), ("review", "approve[1]", [1]), ("review", "review[1]", []), ("review", "", [1])], +) +def test_a_task_view_refuses_an_identity_of_another_step(node_id, instance_key, iteration): + with pytest.raises(ValidationError): + HumanTask.model_validate( + { + **task_view(row("review")).model_dump(), + "node_id": node_id, + "instance_key": instance_key, + "iteration": iteration, + } + )