Give each run of a step its own instance key

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

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**.

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/<step ID>`. 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)`

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.

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)

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:

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(

src/firefly_weave/operations/exports.py

@@ -17,19 +17,31 @@
 """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
 from firefly_weave.runtime.kernel import action_schemas, workflow
 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))

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")

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")

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)
 
 

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)]