Journal SPEC

journal.qnt

module executionJournal {
    // =========================================================================
    // Identity Types
    // =========================================================================
    // Path-based identifier encoding position in the call tree (Dewey encoding).
    // Real system: { root: [u8; 32], path: Vec<u32> }
    // Spec: List[int] — e.g., [1] is root, [1, 0] is first child, [1, 0, 2] is grandchild.
    type PromiseId = List[int]

    // This should technically be the root of the promise ID however Quint has no way
    // of knowing this
    type ExecutionId = PromiseId

    // Concurrent region
    // Real System: [u8, 32]
    type JoinSetId = PromiseId

    // Hash of the WASM binary. Pins execution to exact code version.
    // Real system: [u8; 32]
    type ComponentDigest = str

    // Opaque data crossing the WASM boundary (bytes + codec).
    // We never inspect contents in the spec — only pass them through.
    type Payload = str

    // Wall-clock timestamp. Used for debugging/metadata, NOT for replay logic.
    // Modeled as int (logical tick) because the spec doesn't need real dates.
    type Timestamp = int

    // Duration for timers/retries. Modeled as int (logical units).
    type Duration = int

    // Delivery id for buffered signals (per signal name, monotonic).
    type SignalId = int

    // Retry policy metadata (opaque for now; not interpreted in the spec).
    type RetryPolicy = str

    // What kind of side effect invocation
    type InvokeKind =
        | Function    // function/task/workflow invocation
        | Http        // HTTP request to external service

    // What kind of await
    type AwaitKind =
        | Single              // waiting on one promise
        | Any                 // first of many (JoinSet js.next())
        | All                 // all must complete (JoinSet js.all())
        | AwaitSignal({ name: str, promise_id: PromiseId })  // waiting for named signal

    // =========================================================================
    // Event Types — 20 events, 5 categories
    // =========================================================================

    type EventType =
        // --- Category 1: Lifecycle (Soundness) ---
        // Formal basis: WF-net soundness — proper initiation and termination
        | ExecutionStarted({ component_digest: ComponentDigest,
                             input: Payload,
                             parent_id: PromiseId,
                             idempotency_key: str })
        | ExecutionCompleted({ result: Payload })
        | ExecutionFailed({ error: str })
        | CancelRequested({ reason: str })
        | ExecutionCancelled({ reason: str })

        // --- Category 2: Side Effects (Replay Correctness) ---
        // Formal basis: non-idempotent ops must cache results
        // All follow 3-phase: Scheduled -> Started -> Completed
        | InvokeScheduled({ promise_id: PromiseId,
                            kind: InvokeKind,
                            function_name: str,
                            input: Payload,
                            retry_policy: RetryPolicy })
        | InvokeStarted({ promise_id: PromiseId,
                          attempt: int })
        | InvokeCompleted({ promise_id: PromiseId,
                            result: Payload,
                            attempt: int })
        | InvokeRetrying({ promise_id: PromiseId,
                           failed_attempt: int,
                           error: str,
                           retry_at: Timestamp })

        // --- Category 3: Nondeterminism (Determinism Guarantee) ---
        // Formal basis: LTS determinism — entropy sources must be captured
        // Single-phase: just value capture
        | RandomGenerated({ promise_id: PromiseId,
                            value: str })
        | TimeRecorded({ promise_id: PromiseId,
                         time: Timestamp })

        // --- Category 4: Control Flow (State Reconstruction) ---
        // Formal basis: CSP trace semantics
        | TimerScheduled({ promise_id: PromiseId,
                           duration: Duration,
                           fire_at: Timestamp })
        | TimerFired({ promise_id: PromiseId })
        | SignalDelivered({ signal_name: str,
                            payload: Payload,
                            delivery_id: SignalId })
        // Signal consumed by workflow via await_signal() — has promise_id for replay cache
        | SignalReceived({ promise_id: PromiseId,
                            signal_name: str,
                            payload: Payload,
                            delivery_id: SignalId })
        | ExecutionAwaiting({ waiting_on: Set[PromiseId],
                              kind: AwaitKind })
        | ExecutionResumed

        // --- Category 5: Concurrency (Total Ordering) ---
        // Formal basis: Lamport — concurrent results need deterministic ordering
        | JoinSetCreated({ join_set_id: JoinSetId })
        | JoinSetSubmitted({ join_set_id: JoinSetId,
                             promise_id: PromiseId })
        | JoinSetAwaited({ join_set_id: JoinSetId,
                           promise_id: PromiseId,
                           result: Payload })

    // =========================================================================
    // Journal Entry — one row in the append-only log
    // =========================================================================
    // Maps to Rust's JournalEvent { sequence, timestamp, event }
    type JournalEntry = {
        sequence: int,
        timestamp: Timestamp,
        event: EventType,
    }

    // =========================================================================
    // Execution Status — journal state machine
    // =========================================================================
    // Running → Blocked ↔ Running → Cancelling → Terminal
    // Terminal states: Completed, Failed, Cancelled
    type ExecutionStatus =
        | Running                     // actively executing code
        | Blocked({ waiting_on: Set[PromiseId], kind: AwaitKind })    // waiting for promise(s) to resolve
        | Cancelling                  // cancellation requested, cleanup in progress
        | Completed                   // success (terminal)
        | Failed                      // error (terminal)
        | Cancelled                   // externally cancelled (terminal)

    // =========================================================================
    // Execution State — per-execution bundle
    // =========================================================================
    type ExecutionState = {
        journal: List[JournalEntry],
        status: ExecutionStatus,
        nextChildSeq: int,
        allocatedChildren: Set[PromiseId],
    }

    // All executions, keyed by their PromiseId
    var executions: PromiseId -> ExecutionState

    var clock: Timestamp

    // =========================================================================
    // Pure Helpers
    // =========================================================================

    // Construct a journal entry
    pure def mkEntry(seq: int, ts: Timestamp, evt: EventType): JournalEntry =
        { sequence: seq, timestamp: ts, event: evt }

    // Terminal states — execution is finished, no more events allowed
    pure val TERMINAL_STATES = Set(Completed, Failed, Cancelled)

    // Deterministic child PromiseId derivation (Dewey encoding)
    // parent.append(seq) — e.g., [1].append(0) = [1, 0]
    pure def childId(parent: PromiseId, seq: int): PromiseId =
        parent.append(seq)

    // Check if a promise_id has a resolution event in the journal
    pure def isResolved(journal: List[JournalEntry], pid: PromiseId): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | InvokeCompleted(e) => e.promise_id == pid
                | TimerFired(e) => e.promise_id == pid
                | SignalReceived(e) => e.promise_id == pid
                | _ => false
            }
        )

    // Check if a blocked execution's wait condition is satisfied
    pure def canResume(exec: ExecutionState): bool =
        match exec.status {
            | Blocked(b) => match b.kind {
                | Single => b.waiting_on.forall(pid => isResolved(exec.journal, pid))
                | All => b.waiting_on.forall(pid => isResolved(exec.journal, pid))
                | Any => b.waiting_on.exists(pid => isResolved(exec.journal, pid))
                | AwaitSignal(s) => isResolved(exec.journal, s.promise_id)
            }
            | _ => false
        }

    // Check if the event is a terminal event
    pure def isTerminalEvent(evt: EventType): bool =
      match evt {
          | ExecutionCompleted(_) => true
          | ExecutionFailed(_) => true
          | ExecutionCancelled(_) => true
          | _ => false
      }

    // Derive execution status by folding over journal
    // Only 7 event types change status; everything else passes through
    pure def deriveStatus(journal: List[JournalEntry]): ExecutionStatus =
        journal.foldl(Running, (status, entry) =>
            match entry.event {
                | ExecutionStarted(_) => Running
                | ExecutionAwaiting(e) => Blocked({ waiting_on: e.waiting_on, kind: e.kind })
                | ExecutionResumed => Running
                | CancelRequested(_) => Cancelling
                | ExecutionCompleted(_) => Completed
                | ExecutionFailed(_) => Failed
                | ExecutionCancelled(_) => Cancelled
                | _ => status
            }
        )

    // ----- Journal scanning helpers (for action guards) -----

    pure def hasCancelRequested(journal: List[JournalEntry]): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | CancelRequested(_) => true
                | _ => false
            }
        )

    pure def hasInvokeScheduled(journal: List[JournalEntry], pid: PromiseId): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | InvokeScheduled(e) => e.promise_id == pid
                | _ => false
            }
        )

    pure def hasInvokeStarted(journal: List[JournalEntry], pid: PromiseId): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | InvokeStarted(e) => e.promise_id == pid
                | _ => false
            }
        )

    pure def hasInvokeCompleted(journal: List[JournalEntry], pid: PromiseId): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | InvokeCompleted(e) => e.promise_id == pid
                | _ => false
            }
        )

    pure def hasTimerScheduled(journal: List[JournalEntry], pid: PromiseId): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | TimerScheduled(e) => e.promise_id == pid
                | _ => false
            }
        )

    pure def signalDeliveredCount(journal: List[JournalEntry], name: str): int =
        journal.indices().filter(i =>
            match journal.nth(i).event {
                | SignalDelivered(e) => e.signal_name == name
                | _ => false
            }
        ).size()

    pure def signalReceivedCount(journal: List[JournalEntry], name: str): int =
        journal.indices().filter(i =>
            match journal.nth(i).event {
                | SignalReceived(e) => e.signal_name == name
                | _ => false
            }
        ).size()

    pure def hasSignalDelivered(journal: List[JournalEntry], name: str, deliveryId: SignalId, payload: Payload): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | SignalDelivered(e) => e.signal_name == name and e.delivery_id == deliveryId and e.payload == payload
                | _ => false
            }
        )

    pure def hasJoinSetCreated(journal: List[JournalEntry], jsId: JoinSetId): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | JoinSetCreated(e) => e.join_set_id == jsId
                | _ => false
            }
        )

    pure def hasJoinSetAwaited(journal: List[JournalEntry], jsId: JoinSetId): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | JoinSetAwaited(e) => e.join_set_id == jsId
                | _ => false
            }
        )

    pure def hasJoinSetSubmittedFor(journal: List[JournalEntry], jsId: JoinSetId, pid: PromiseId): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | JoinSetSubmitted(e) => e.join_set_id == jsId and e.promise_id == pid
                | _ => false
            }
        )

    pure def hasJoinSetAwaitedFor(journal: List[JournalEntry], jsId: JoinSetId, pid: PromiseId): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | JoinSetAwaited(e) => e.join_set_id == jsId and e.promise_id == pid
                | _ => false
            }
        )

    pure def hasSubmittedToOtherJoinSet(journal: List[JournalEntry], pid: PromiseId, jsId: JoinSetId): bool =
        journal.indices().exists(i =>
            match journal.nth(i).event {
                | JoinSetSubmitted(e) => e.promise_id == pid and e.join_set_id != jsId
                | _ => false
            }
        )

    pure def retryCount(journal: List[JournalEntry], pid: PromiseId): int =
        journal.indices().filter(i =>
            match journal.nth(i).event {
                | InvokeRetrying(e) => e.promise_id == pid
                | _ => false
            }
        ).size()

    // =========================================================================
    // Initialization
    // =========================================================================
    action init = all {
        executions' = Map(),
        clock' = 0,
    }

    // =========================================================================
    // Lifecycle Actions
    // =========================================================================

    // Create a new execution — first event is always ExecutionStarted
    action startExecution(eid: PromiseId, ts: Timestamp, digest: ComponentDigest,
                          input: Payload, parent: PromiseId, key: str): bool = all {
        not(eid.in(executions.keys())),
        executions' = executions.put(eid, {
            journal: List(mkEntry(0, ts,
                ExecutionStarted({ component_digest: digest, input: input,
                                   parent_id: parent, idempotency_key: key }))),
            status: Running,
            nextChildSeq: 0,
            allocatedChildren: Set(),
        }),
    }

    // Complete execution successfully — terminal
    action completeExecution(eid: PromiseId, ts: Timestamp, result: Payload): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            not(exec.status.in(TERMINAL_STATES)),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        ExecutionCompleted({ result: result }))),
                status: Completed,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Fail execution — terminal
    action failExecution(eid: PromiseId, ts: Timestamp, error: str): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            not(exec.status.in(TERMINAL_STATES)),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        ExecutionFailed({ error: error }))),
                status: Failed,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Request cancellation — not terminal yet, allows cleanup
    action requestCancel(eid: PromiseId, ts: Timestamp, reason: str): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            not(exec.status.in(TERMINAL_STATES)),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        CancelRequested({ reason: reason }))),
                status: Cancelling,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Cancel execution — terminal
    action cancelExecution(eid: PromiseId, ts: Timestamp, reason: str): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            not(exec.status.in(TERMINAL_STATES)),
            hasCancelRequested(exec.journal),
            exec.status == Cancelling,
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        ExecutionCancelled({ reason: reason }))),
                status: Cancelled,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // =========================================================================
    // Side Effect + Nondeterminism Actions
    // =========================================================================

    // Schedule a side effect invocation — allocates child PromiseId
    action scheduleInvoke(eid: PromiseId, ts: Timestamp, childPid: PromiseId,
                          kind: InvokeKind, fname: str, input: Payload,
                          retryPolicy: RetryPolicy): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            exec.status == Running,
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        InvokeScheduled({ promise_id: childPid, kind: kind,
                                          function_name: fname, input: input,
                                          retry_policy: retryPolicy }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq + 1,
                allocatedChildren: exec.allocatedChildren.union(Set(childPid)),
            }),
        }
    }

    // Mark an invocation as started (records attempt number)
    action startInvoke(eid: PromiseId, ts: Timestamp, childPid: PromiseId,
                       attempt: int): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            not(exec.status.in(TERMINAL_STATES)),
            hasInvokeScheduled(exec.journal, childPid),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        InvokeStarted({ promise_id: childPid, attempt: attempt }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Mark an invocation as completed with its result
    action completeInvoke(eid: PromiseId, ts: Timestamp, childPid: PromiseId,
                          result: Payload, attempt: int): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            not(exec.status.in(TERMINAL_STATES)),
            hasInvokeStarted(exec.journal, childPid),
            not(hasInvokeCompleted(exec.journal, childPid)),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        InvokeCompleted({ promise_id: childPid, result: result,
                                          attempt: attempt }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Record a retry after a failed attempt
    action retryInvoke(eid: PromiseId, ts: Timestamp, childPid: PromiseId,
                       failedAttempt: int, error: str, retryAt: Timestamp): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            not(exec.status.in(TERMINAL_STATES)),
            hasInvokeStarted(exec.journal, childPid),
            not(hasInvokeCompleted(exec.journal, childPid)),
            retryCount(exec.journal, childPid) < MAX_RETRY_ATTEMPTS,
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        InvokeRetrying({ promise_id: childPid, failed_attempt: failedAttempt,
                                         error: error, retry_at: retryAt }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Capture a generated random value — allocates child PromiseId
    action generateRandom(eid: PromiseId, ts: Timestamp, childPid: PromiseId,
                          value: str): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            exec.status == Running,
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        RandomGenerated({ promise_id: childPid, value: value }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq + 1,
                allocatedChildren: exec.allocatedChildren.union(Set(childPid)),
            }),
        }
    }

    // Capture the current wall-clock time — allocates child PromiseId
    action recordTime(eid: PromiseId, ts: Timestamp, childPid: PromiseId,
                      time: Timestamp): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            exec.status == Running,
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        TimeRecorded({ promise_id: childPid, time: time }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq + 1,
                allocatedChildren: exec.allocatedChildren.union(Set(childPid)),
            }),
        }
    }

    // Schedule a timer — allocates child PromiseId
    action scheduleTimer(eid: PromiseId, ts: Timestamp, childPid: PromiseId,
                         duration: Duration, fireAt: Timestamp): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            exec.status == Running,
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        TimerScheduled({ promise_id: childPid, duration: duration,
                                         fire_at: fireAt }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq + 1,
                allocatedChildren: exec.allocatedChildren.union(Set(childPid)),
            }),
        }
    }

    // Timer fires — runtime event, execution may be Blocked
    action fireTimer(eid: PromiseId, ts: Timestamp, childPid: PromiseId): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            not(exec.status.in(TERMINAL_STATES)),
            hasTimerScheduled(exec.journal, childPid),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        TimerFired({ promise_id: childPid }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // External signal delivered to execution — no child allocation
    action deliverSignal(eid: PromiseId, ts: Timestamp, signalName: str, payload: Payload): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        val deliveryId = signalDeliveredCount(exec.journal, signalName) + 1
        all {
            not(exec.status.in(TERMINAL_STATES)),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        SignalDelivered({ signal_name: signalName,
                                          payload: payload,
                                          delivery_id: deliveryId }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Workflow consumes a signal — allocates child PromiseId
    action receiveSignal(eid: PromiseId, ts: Timestamp, childPid: PromiseId, signalName: str, payload: Payload): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        val deliveryId = signalReceivedCount(exec.journal, signalName) + 1
        all {
            exec.status == Running,
            signalDeliveredCount(exec.journal, signalName) >= deliveryId,
            hasSignalDelivered(exec.journal, signalName, deliveryId, payload),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        SignalReceived({ promise_id: childPid, signal_name: signalName,
                                         payload: payload,
                                         delivery_id: deliveryId }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq + 1,
                allocatedChildren: exec.allocatedChildren.union(Set(childPid)),
            }),
        }
    }

    // Workflow awaits promise(s) — transitions Running → Blocked
    action awaitExecution(eid: PromiseId, ts: Timestamp, waitingOn: Set[PromiseId], kind: AwaitKind): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            exec.status == Running,
            match kind {
                | AwaitSignal(s) => waitingOn.size() == 1 and waitingOn.contains(s.promise_id)
                | _ => true
            },
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        ExecutionAwaiting({ waiting_on: waitingOn, kind: kind }))),
                status: Blocked({ waiting_on: waitingOn, kind: kind }),
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Resume a blocked execution — writes ExecutionResumed to journal
    action resumeExecution(eid: PromiseId, ts: Timestamp): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            canResume(exec),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts, ExecutionResumed)),
                status: Running,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Create a join set for concurrent work
    action createJoinSet(eid: PromiseId, ts: Timestamp,
                        jsId: JoinSetId): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            exec.status == Running,
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        JoinSetCreated({ join_set_id: jsId }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Submit a promise to a join set
    action submitToJoinSet(eid: PromiseId, ts: Timestamp,
                           jsId: JoinSetId, childPid: PromiseId): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            exec.status == Running,
            hasJoinSetCreated(exec.journal, jsId),
            not(hasJoinSetAwaited(exec.journal, jsId)),
            not(hasSubmittedToOtherJoinSet(exec.journal, childPid, jsId)),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        JoinSetSubmitted({ join_set_id: jsId,
                                           promise_id: childPid }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // Consume a result from a join set
    action awaitJoinSet(eid: PromiseId, ts: Timestamp, jsId: JoinSetId,
                        childPid: PromiseId, result: Payload): bool = all {
        eid.in(executions.keys()),
        val exec = executions.get(eid)
        all {
            exec.status == Running,
            hasJoinSetSubmittedFor(exec.journal, jsId, childPid),
            hasInvokeCompleted(exec.journal, childPid),
            not(hasJoinSetAwaitedFor(exec.journal, jsId, childPid)),
            executions' = executions.set(eid, {
                journal: exec.journal.append(
                    mkEntry(exec.journal.length(), ts,
                        JoinSetAwaited({ join_set_id: jsId,
                                         promise_id: childPid,
                                         result: result }))),
                status: exec.status,
                nextChildSeq: exec.nextChildSeq,
                allocatedChildren: exec.allocatedChildren,
            }),
        }
    }

    // =========================================================================
    // Model checking bounds — small finite sets for state exploration
    // =========================================================================
    pure val EXEC_IDS = Set(List(1), List(2), List(3))
    pure val PAYLOADS = Set("p1", "p2")
    pure val DIGESTS = Set("d1")
    pure val FUNC_NAMES = Set("fn1", "fn2")
    pure val SIGNAL_NAMES = Set("sig1", "sig2")
    pure val JOINSET_IDS = Set(List(4), List(5))  // JoinSetId = List[int], distinct from EXEC_IDS
    pure val DURATIONS = Set(1, 2, 3)
    pure val ERRORS = Set("err1")
    pure val MAX_RETRY_ATTEMPTS = 3
    pure val IDEM_KEYS = Set("k1", "k2")
    pure val RETRY_POLICIES = Set("rp1", "rp2")

    // =========================================================================
    // Step action — nondeterministic dispatcher for model checking
    // =========================================================================
    action step = any {
        // --- Lifecycle ---
        nondet eid = oneOf(EXEC_IDS)
        nondet digest = oneOf(DIGESTS)
        nondet input = oneOf(PAYLOADS)
        nondet parent = oneOf(EXEC_IDS)
        nondet key = oneOf(IDEM_KEYS)
        all { startExecution(eid, clock, digest, input, parent, key), clock' = clock + 1 },

        nondet eid = oneOf(executions.keys())
        nondet result = oneOf(PAYLOADS)
        all { completeExecution(eid, clock, result), clock' = clock + 1 },

        nondet eid = oneOf(executions.keys())
        nondet error = oneOf(ERRORS)
        all { failExecution(eid, clock, error), clock' = clock + 1 },

        nondet eid = oneOf(executions.keys())
        nondet reason = oneOf(ERRORS)
        all { requestCancel(eid, clock, reason), clock' = clock + 1 },

        nondet eid = oneOf(executions.keys())
        nondet reason = oneOf(ERRORS)
        all { cancelExecution(eid, clock, reason), clock' = clock + 1 },

        // --- Side Effects (child-allocating: childPid derived deterministically) ---
        nondet eid = oneOf(executions.keys())
        nondet kind = oneOf(Set(Function, Http))
        nondet fname = oneOf(FUNC_NAMES)
        nondet input = oneOf(PAYLOADS)
        nondet retryPolicy = oneOf(RETRY_POLICIES)
        val childPid_si = childId(eid, executions.get(eid).nextChildSeq)
        all { scheduleInvoke(eid, clock, childPid_si, kind, fname, input, retryPolicy), clock' = clock + 1 },

        // --- Side Effects (referencing: childPid from already-allocated children) ---
        nondet eid = oneOf(executions.keys())
        nondet childPid_sti = oneOf(executions.get(eid).allocatedChildren)
        nondet attempt = oneOf(Set(1, 2, 3))
        all { startInvoke(eid, clock, childPid_sti, attempt), clock' = clock + 1 },

        nondet eid = oneOf(executions.keys())
        nondet childPid_ci = oneOf(executions.get(eid).allocatedChildren)
        nondet result = oneOf(PAYLOADS)
        nondet attempt = oneOf(Set(1, 2, 3))
        all { completeInvoke(eid, clock, childPid_ci, result, attempt), clock' = clock + 1 },

        nondet eid = oneOf(executions.keys())
        nondet childPid_ri = oneOf(executions.get(eid).allocatedChildren)
        nondet failedAttempt = oneOf(Set(1, 2, 3))
        nondet error = oneOf(ERRORS)
        all { retryInvoke(eid, clock, childPid_ri, failedAttempt, error, clock + 1), clock' = clock + 1 },

        // --- Nondeterminism (child-allocating) ---
        nondet eid = oneOf(executions.keys())
        nondet value = oneOf(Set("rand1", "rand2"))
        val childPid_gr = childId(eid, executions.get(eid).nextChildSeq)
        all { generateRandom(eid, clock, childPid_gr, value), clock' = clock + 1 },

        nondet eid = oneOf(executions.keys())
        val childPid_rt = childId(eid, executions.get(eid).nextChildSeq)
        all { recordTime(eid, clock, childPid_rt, clock), clock' = clock + 1 },

        // --- Control Flow (child-allocating) ---
        nondet eid = oneOf(executions.keys())
        nondet duration = oneOf(DURATIONS)
        val childPid_st = childId(eid, executions.get(eid).nextChildSeq)
        all { scheduleTimer(eid, clock, childPid_st, duration, clock + duration), clock' = clock + 1 },

        // --- Control Flow (referencing) ---
        nondet eid = oneOf(executions.keys())
        nondet childPid_ft = oneOf(executions.get(eid).allocatedChildren)
        all { fireTimer(eid, clock, childPid_ft), clock' = clock + 1 },

        nondet eid = oneOf(executions.keys())
        nondet signalName = oneOf(SIGNAL_NAMES)
        nondet payload = oneOf(PAYLOADS)
        all { deliverSignal(eid, clock, signalName, payload), clock' = clock + 1 },

        // --- Control Flow (child-allocating) ---
        nondet eid = oneOf(executions.keys())
        nondet signalName = oneOf(SIGNAL_NAMES)
        nondet payload = oneOf(PAYLOADS)
        val childPid_rs = childId(eid, executions.get(eid).nextChildSeq)
        all { receiveSignal(eid, clock, childPid_rs, signalName, payload), clock' = clock + 1 },

        // --- Control Flow (await — domain-meaningful patterns) ---
        nondet eid = oneOf(executions.keys())
        val children_aw = executions.get(eid).allocatedChildren
        all {
            children_aw.size() > 0,
            any {
                // Single: wait on one child
                nondet child = oneOf(children_aw)
                all { awaitExecution(eid, clock, Set(child), Single), clock' = clock + 1 },
                // Any: wait on all children, first to resolve unblocks
                all { awaitExecution(eid, clock, children_aw, Any), clock' = clock + 1 },
                // All: wait on all children, all must complete
                all { awaitExecution(eid, clock, children_aw, All), clock' = clock + 1 },
                // AwaitSignal: wait on one child for a named signal
                nondet child = oneOf(children_aw)
                nondet sigName = oneOf(SIGNAL_NAMES)
                all { awaitExecution(eid, clock, Set(child), AwaitSignal({ name: sigName, promise_id: child })), clock' = clock + 1 },
            },
        },

        // --- Resume ---
        nondet eid = oneOf(executions.keys())
        all { resumeExecution(eid, clock), clock' = clock + 1 },

        // --- Concurrency ---
        nondet eid = oneOf(executions.keys())
        nondet jsId = oneOf(JOINSET_IDS)
        all { createJoinSet(eid, clock, jsId), clock' = clock + 1 },

        nondet eid = oneOf(executions.keys())
        nondet jsId = oneOf(JOINSET_IDS)
        nondet childPid_stj = oneOf(executions.get(eid).allocatedChildren)
        all { submitToJoinSet(eid, clock, jsId, childPid_stj), clock' = clock + 1 },

        nondet eid = oneOf(executions.keys())
        nondet jsId = oneOf(JOINSET_IDS)
        nondet childPid_aj = oneOf(executions.get(eid).allocatedChildren)
        nondet result = oneOf(PAYLOADS)
        all { awaitJoinSet(eid, clock, jsId, childPid_aj, result), clock' = clock + 1 },
    }


    // =========================================================================
    // INVARIANTS
    // =========================================================================
    
    // INV-1:  The first event is always Execution Started
   val firstEventIsStarted = executions.keys().forall(eid =>
    val exec = executions.get(eid)
    exec.journal.length() == 0 or
      match exec.journal.nth(0).event {
        | ExecutionStarted(_) => true
        | _ => false
      }
  )

   /// INV-2: Journal sequence numbers are monotonically increasing (entry i has sequence == i)
  val journalMonotonicity = executions.keys().forall(eid =>
    val exec = executions.get(eid)
    exec.journal.indices().forall(i =>
      exec.journal.nth(i).sequence == i
    )
  )

    /// INV-3: Terminal events are always the last entry
    val terminalFinality = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            isTerminalEvent(exec.journal.nth(i).event) implies i == exec.journal.length() - 1
        )
    )

    /// INV-4: Stored status matches journal-derived status
    val statusJournalConsistency = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.status == deriveStatus(exec.journal)
    )

    /// S-3: At most one terminal event per journal
    val singleTerminal = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().filter(i =>
            isTerminalEvent(exec.journal.nth(i).event)
        ).size() <= 1
    )

    /// S-5: ExecutionCancelled requires preceding CancelRequested
    val cancelledRequiresRequested = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | ExecutionCancelled(_) => exec.journal.indices().exists(j =>
                    j < i and match exec.journal.nth(j).event {
                        | CancelRequested(_) => true
                        | _ => false
                    }
                )
                | _ => true
            }
        )
    )

    /// INV-5: InvokeStarted requires prior InvokeScheduled for same promise_id
    val phaseOrdering = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | InvokeStarted(e) => exec.journal.indices().exists(j =>
                    j < i and match exec.journal.nth(j).event {
                        | InvokeScheduled(s) => s.promise_id == e.promise_id
                        | _ => false
                    }
                )
                | _ => true
            }
        )
    )

    /// SE-2: InvokeCompleted requires preceding InvokeStarted for same promise_id
    val completedRequiresStarted = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | InvokeCompleted(e) => exec.journal.indices().exists(j =>
                    j < i and match exec.journal.nth(j).event {
                        | InvokeStarted(s) => s.promise_id == e.promise_id
                        | _ => false
                    }
                )
                | _ => true
            }
        )
    )

    /// SE-4: No InvokeStarted/Retrying after InvokeCompleted for same promise_id
    val noEventsAfterCompleted = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | InvokeCompleted(e) => exec.journal.indices().forall(j =>
                    j <= i or match exec.journal.nth(j).event {
                        | InvokeStarted(s) => s.promise_id != e.promise_id
                        | InvokeRetrying(s) => s.promise_id != e.promise_id
                        | _ => true
                    }
                )
                | _ => true
            }
        )
    )

    /// SE-3: InvokeRetrying requires preceding InvokeStarted for same promise_id
    val retryingRequiresStarted = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | InvokeRetrying(e) => exec.journal.indices().exists(j =>
                    j < i and match exec.journal.nth(j).event {
                        | InvokeStarted(s) => s.promise_id == e.promise_id
                        | _ => false
                    }
                )
                | _ => true
            }
        )
    )

    /// SE-5: Retry count per promise_id never exceeds MAX_RETRY_ATTEMPTS
    val retryBounded = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.allocatedChildren.forall(pid =>
            retryCount(exec.journal, pid) <= MAX_RETRY_ATTEMPTS
        )
    )

    /// CF-2: SignalReceived requires preceding SignalDelivered for same signal_name, delivery_id, payload
    val signalReceivedRequiresDelivered = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | SignalReceived(e) => exec.journal.indices().exists(j =>
                    j < i and match exec.journal.nth(j).event {
                        | SignalDelivered(s) => s.signal_name == e.signal_name and s.delivery_id == e.delivery_id and s.payload == e.payload
                        | _ => false
                    }
                )
                | _ => true
            }
        )
    )

    /// CF-3: Each delivery_id consumed by at most one SignalReceived
    val signalConsumedOnce = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            exec.journal.indices().forall(j =>
                (i != j) implies not(
                    match exec.journal.nth(i).event {
                        | SignalReceived(e1) => match exec.journal.nth(j).event {
                            | SignalReceived(e2) => e1.signal_name == e2.signal_name and e1.delivery_id == e2.delivery_id
                            | _ => false
                        }
                        | _ => false
                    }
                )
            )
        )
    )

    /// CF-4: AwaitSignal waiting_on matches promise_id
    val awaitSignalConsistent = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | ExecutionAwaiting(e) => match e.kind {
                    | AwaitSignal(s) => e.waiting_on.size() == 1 and e.waiting_on.contains(s.promise_id)
                    | _ => true
                }
                | _ => true
            }
        )
    )

    /// CF-1: TimerFired requires preceding TimerScheduled for same promise_id
    val timerFiredRequiresScheduled = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | TimerFired(e) => exec.journal.indices().exists(j =>
                    j < i and match exec.journal.nth(j).event {
                        | TimerScheduled(s) => s.promise_id == e.promise_id
                        | _ => false
                    }
                )
                | _ => true
            }
        )
    )

    /// JS-2: No JoinSetSubmitted after JoinSetAwaited for same join_set_id
    val noSubmitAfterAwait = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | JoinSetSubmitted(e) => exec.journal.indices().forall(j =>
                    j >= i or match exec.journal.nth(j).event {
                        | JoinSetAwaited(s) => s.join_set_id != e.join_set_id
                        | _ => true
                    }
                )
                | _ => true
            }
        )
    )

    /// JS-1: JoinSetSubmitted requires preceding JoinSetCreated for same join_set_id
    val submitRequiresCreated = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | JoinSetSubmitted(e) => exec.journal.indices().exists(j =>
                    j < i and match exec.journal.nth(j).event {
                        | JoinSetCreated(s) => s.join_set_id == e.join_set_id
                        | _ => false
                    }
                )
                | _ => true
            }
        )
    )

    /// JS-4: JoinSetAwaited requires preceding InvokeCompleted for same promise_id
    val awaitedRequiresCompleted = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | JoinSetAwaited(e) => exec.journal.indices().exists(j =>
                    j < i and match exec.journal.nth(j).event {
                        | InvokeCompleted(s) => s.promise_id == e.promise_id
                        | _ => false
                    }
                )
                | _ => true
            }
        )
    )

    /// JS-3: JoinSetAwaited requires preceding JoinSetSubmitted for same (join_set_id, promise_id)
    val awaitedRequiresMember = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            match exec.journal.nth(i).event {
                | JoinSetAwaited(e) => exec.journal.indices().exists(j =>
                    j < i and match exec.journal.nth(j).event {
                        | JoinSetSubmitted(s) => s.join_set_id == e.join_set_id and s.promise_id == e.promise_id
                        | _ => false
                    }
                )
                | _ => true
            }
        )
    )

    /// JS-6: Per join set, count of awaited <= count of submitted
    val consumeBounded = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        JOINSET_IDS.forall(jsId =>
            val awaited = exec.journal.indices().filter(i =>
                match exec.journal.nth(i).event {
                    | JoinSetAwaited(e) => e.join_set_id == jsId
                    | _ => false
                }
            ).size()
            val submitted = exec.journal.indices().filter(i =>
                match exec.journal.nth(i).event {
                    | JoinSetSubmitted(e) => e.join_set_id == jsId
                    | _ => false
                }
            ).size()
            awaited <= submitted
        )
    )

    /// JS-5: No two JoinSetAwaited for same (join_set_id, promise_id)
    val noDoubleConsume = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            exec.journal.indices().forall(j =>
                (i != j) implies not(
                    match exec.journal.nth(i).event {
                        | JoinSetAwaited(e1) => match exec.journal.nth(j).event {
                            | JoinSetAwaited(e2) => e1.join_set_id == e2.join_set_id and e1.promise_id == e2.promise_id
                            | _ => false
                        }
                        | _ => false
                    }
                )
            )
        )
    )

    /// JS-7: A promise_id appears in at most one join set
    val promiseSingleOwner = executions.keys().forall(eid =>
        val exec = executions.get(eid)
        exec.journal.indices().forall(i =>
            exec.journal.indices().forall(j =>
                match exec.journal.nth(i).event {
                    | JoinSetSubmitted(e1) => match exec.journal.nth(j).event {
                        | JoinSetSubmitted(e2) => e1.promise_id == e2.promise_id implies e1.join_set_id == e2.join_set_id
                        | _ => true
                    }
                    | _ => true
                }
            )
        )
    )

    /// INV-6: No duplicate promise_ids across executions
    val promiseIdUniqueness = executions.keys().forall(eid1 =>
        executions.keys().forall(eid2 =>
            eid1 == eid2 or
            executions.get(eid1).allocatedChildren.forall(c1 =>
                executions.get(eid2).allocatedChildren.forall(c2 =>
                    c1 != c2
                )
            )
        )
    )
}

journal_dessign.md

Execution Journal Design

Authoritative reference for the Invariant execution journal. The journal is an append-only sequence of events that fully describes an execution's history. It is the source of truth — state is derived by replaying it.

Purpose: Durability (survive crashes), replay (reconstruct state), exactly-once semantics (cached results prevent re-execution of side effects).

Related docs: JOINSET_DESIGN.md (JoinSet deep-dive), execution_journal.qnt (Quint formal spec).

---

Journal Structure

pub struct ExecutionJournal {
    pub execution_id: ExecutionId,
    pub state: JournalState,        // Derived from events, cached
    pub events: Vec<JournalEvent>,  // Append-only
}

pub struct JournalEvent {
    pub sequence: u64,              // 0-indexed, monotonically increasing
    pub timestamp: DateTime<Utc>,   // Wall-clock (debugging only, NOT used in replay)
    pub event: EventType,
}

Version = events.len(). Flat structure, simple storage, natural time ordering.

---

Event Types (20 events, 5 categories)

Each category satisfies a distinct formal correctness property.

Category 1: Lifecycle (Soundness)

Formal basis: WF-net soundness — proper initiation and termination.

Event	When Recorded	Data
ExecutionStarted	Always first	component_digest, input, parent_id, idempotency_key
ExecutionCompleted	Function returns Ok	result
ExecutionFailed	Function returns Err or traps	error
CancelRequested	External cancel signal arrives	reason
ExecutionCancelled	Cancellation finalized	reason

Category 2: Side Effects (Replay Correctness)

Formal basis: Non-idempotent operations must cache results. All follow 3-phase pattern: Scheduled -> Started -> Completed.

Why 3-phase: Scheduled = intent (for replay matching, enables exactly-once via intent logging). Started = in-flight (timeout detection). Completed = result (cache for replay). Retrying = transient failure, will retry.

Event	Phase	Data
InvokeScheduled	Scheduled	promise_id, kind, function_name, input, retry_policy
InvokeStarted	Started	promise_id, attempt
InvokeCompleted	Completed	promise_id, result, attempt
InvokeRetrying	Retry	promise_id, failed_attempt, error, retry_at

InvokeKind categorizes the invocation type. The kind field lives on InvokeScheduled only — later phases inherit kind via promise_id lookup.

pub enum InvokeKind {
    Function,   // function/task/workflow invocation
    Http,       // HTTP request to external service
    // Future: Database, Grpc, Message, etc.
}

This is extensible: new side effect types (DB queries, gRPC calls) are added as InvokeKind variants, not new event types. All share the same 3-phase structure and replay semantics.

Category 3: Nondeterminism (Determinism Guarantee)

Formal basis: LTS determinism — entropy sources must be captured. Single-phase (pure value capture, no execution to track).

Event	When Recorded	Data
RandomGenerated	random() called	promise_id, value
TimeRecorded	now() called	promise_id, time

Category 4: Control Flow (State Reconstruction)

Formal basis: CSP trace semantics — branching decisions define execution paths.

Event	When Recorded	Data
TimerScheduled	sleep(duration) called	promise_id, duration, fire_at
TimerFired	Duration elapsed	promise_id
SignalDelivered	External signal arrives at execution	signal_name, payload, delivery_id
SignalReceived	Workflow consumes signal via await_signal()	promise_id, signal_name, payload, delivery_id
ExecutionAwaiting	Workflow blocks	waiting_on: Vec<PromiseId>, kind: AwaitKind
ExecutionResumed	Blocked → Running (wait satisfied)	—

AwaitKind: Single | Any (first of many) | All (all must complete) | Signal(name). Signal await assigns a PromiseId via the local sequence counter, same as invoke/timer/random.

Signal two-event model: SignalDelivered is the durable buffer — recorded when an external signal arrives, no promise_id (external to the call tree). Each delivery assigns a per-signal-name monotonic delivery_id. SignalReceived is the consumption event — recorded when await_signal() matches a delivered signal, carries a promise_id for the replay cache, and records the consumed delivery_id. Consumption is FIFO per signal name (oldest unconsumed delivery). If the signal is already buffered when await_signal() is called, SignalReceived is recorded immediately (no blocking). If not, the workflow blocks until the signal arrives.

ExecutionAwaiting is the explicit suspend (IEEE 1849). ExecutionResumed is the explicit resume — recorded when the blocked wait condition is satisfied.

Category 5: Concurrency (Total Ordering)

Formal basis: Lamport timestamps — concurrent results need deterministic ordering for replay.

Event	When Recorded	Data
JoinSetCreated	join_set() called	join_set_id (child-allocating)
JoinSetSubmitted	js.submit(...) called	join_set_id, promise_id
JoinSetAwaited	js.next() returns	join_set_id, promise_id, result

JoinSetId identity: JoinSetId is a newtype wrapper around PromiseId. The join_set() call allocates a child position via nextChildSeq++, same as invoke/random/time/timer/signal. This keeps JoinSets consistent with the identity model — every SDK call occupies a deterministic position in the call tree. The newtype prevents misuse: a JoinSetId cannot be awaited, completed, or submitted to another JoinSet.

pub struct JoinSetId(pub PromiseId);

Why PromiseId, not string or counter: (1) Unique by construction — same Dewey encoding guarantee as all other IDs. (2) Deterministic on replay — position-based, not dependent on runtime values. (3) Zero new mechanism — reuses nextChildSeq and allocatedChildren. (4) A PromiseId newtype captures the structural relationship between JoinSets and the identity hierarchy with typed safety.

JoinSetAwaited is a replay marker, not a state transition. It records which result was consumed at this point — whether or not the workflow blocked. ExecutionAwaiting handles blocking; JoinSetAwaited handles replay ordering. See JOINSET_DESIGN.md for full design.

Category Summary

Category	Formal Property	Guarantee	Events
Lifecycle	Soundness	Proper start/end	5
Side Effects	Replay Correctness	External ops not re-executed	4
Nondeterminism	Determinism	Same random/time on replay	2
Control Flow	State Reconstruction	Same execution path on replay	6
Concurrency	Total Ordering	Same result order on replay	3

Together: forall execution E, replay(journal(E)) = E

---

State Machine

                    ExecutionStarted
                          |
                          v
                      ┌────────┐
              ┌──────>│Running │<──────┐
              │       └───┬────┘       │
              │           │            │
         result arrives   │     ExecutionAwaiting
              │           │            │
              │     ┌─────┴─────┐      │
              └─────│  Blocked  │──────┘
                    └───────────┘
                          │
              ┌───────────┼───────────┐
              v           v           v
         Completed     Failed    Cancelling
                                      │
                                      v
                                  Cancelled

CancelRequested transitions Running or Blocked to Cancelling. ExecutionCancelled transitions Cancelling to Cancelled (terminal).

pub enum JournalState {
    Running,
    Blocked { waiting_on: Vec<PromiseId>, kind: AwaitKind },
    Cancelling,   // cancel requested, cleanup in progress
    Completed,    // terminal
    Failed,       // terminal
    Cancelled,    // terminal
}

State derivation (from events, not stored independently):

Precondition: journal is non-empty (invariant S-2 guarantees first event is ExecutionStarted).

Fold over journal, carrying status forward. Only 7 event types change status:

Event	Status
ExecutionStarted	Running
CancelRequested	Cancelling
ExecutionAwaiting	Blocked(waiting_on, kind)
ExecutionResumed	Running
ExecutionCompleted	Completed
ExecutionFailed	Failed
ExecutionCancelled	Cancelled
Everything else	unchanged

Wait satisfaction (guard for ExecutionResumed) depends on AwaitKind:

• Single / All: all waiting_on promises in completedPromises
• Any: at least one waiting_on promise in completedPromises
• Signal(name): the signal's promise_id in completedPromises (a SignalReceived event exists for this pid)

---

Identity: Path-Based PromiseId

pub struct PromiseId {
    root: [u8; 32],      // hash(component_digest, parent, idempotency_key)
    path: Vec<u32>,       // sequence numbers at each depth
}
pub type ExecutionId = PromiseId;  // path.is_empty() == true
pub struct JoinSetId(pub PromiseId);  // child-allocating, distinct type

Encodes position in call tree. Properties: unique (position is unique), deterministic (same code path -> same sequence -> same IDs), recursive-safe (each depth has own counter).

workflow_main()        -> "root"        (depth 0)
  invoke!("task_a")    -> "root.0"      (depth 1, seq 0)
  invoke!("task_a")    -> "root.1"      (depth 1, seq 1 — same fn, different position)
task_a(x)              -> "root.0"
  time.now()           -> "root.0.0"    (depth 2, seq 0)
  invoke!("task_c")    -> "root.0.1"    (depth 2, seq 1)
task_c(z)              -> "root.0.1"
  random.u64()         -> "root.0.1.0"  (depth 3, seq 0)

What PromiseId encodes:

• promise_id.execution_root() -> which journal to load
• promise_id.parent() -> who to notify on completion
• promise_id.child(seq) -> create child for nth operation

---

Replay Protocol

Non-determinism eliminated by design. ComponentDigest pins execution to exact WASM binary. WASM sandbox eliminates ambient nondeterminism. Same code + same journal = same replay, always. No command matching, no trace comparison, no alignment checks.

1. Load journal from storage
2. Build HashMap<PromiseId, CachedResult> from completed events
3. Execute function from beginning
4. On each operation:
   - Generate child_id = current_promise_id.child(local_sequence++)
   - Lookup child_id in HashMap
   - HIT  -> return cached result, continue
   - MISS -> record InvokeScheduled, queue child work item, TRAP
5. On completion: record result, queue parent (respecting JoinSet rules)

Yield mechanism: WASM trap. No state serialization. Execution replays from beginning on resume.

Work queue: Unified. Starting workflows, executing tasks, resuming parents after child completion — all the same: pick WorkItem, load journal, execute/replay.

---

Retry Handling

PromiseId = logical operation (stable across retries). Attempt = physical execution.

invoke!("flaky_api", x) -> promise_id = "root.0" for ALL attempts

Journal:
| promise_id | event            | attempt |
|------------|------------------|---------|
| root.0     | InvokeScheduled  | -       |
| root.0     | InvokeStarted    | 1       |
| root.0     | InvokeRetrying   | 1       |  <- failed
| root.0     | InvokeStarted    | 2       |
| root.0     | InvokeCompleted  | 2       |  <- success

On replay: generate "root.0", lookup in cache, return cached result immediately. Retry history preserved but not needed for replay.

---

Design Decisions

Decision	Choice	Formal Basis
Event categories	5 categories by correctness property	Process calculi, LTS, WF-net
Side effect phases	3-phase (Scheduled/Started/Completed)	Intent logging, XES lifecycle
Nondeterminism phases	1-phase (value capture)	LTS determinism
Journal structure	Append-only, flat	Event sourcing, WAL
Ordering	Sequence numbers (Lamport clock)	Lamport 1978
Identity	Path-based PromiseId	Deterministic, unique, O(1) lookup
ExecutionId	Type alias for root PromiseId	Same concept at depth 0
Non-determinism detection	Eliminated by design	ComponentDigest + WASM sandbox
Cancel	2-phase (CancelRequested → ExecutionCancelled)	XES pi_abort
Unified invoke	No activity/child workflow split	Burckhardt formal model
JoinSet closure	No submit after first await	Structured concurrency, WCP2/3/9
JoinSet ownership	Promise in at most one set	Linearizability, scope ownership
JoinSetAwaited	Replay marker, not state transition	Burckhardt, Schneider
JoinSetId	PromiseId newtype, child-allocating	Identity model consistency, Dewey encoding
Yield	WASM trap	No state serialization needed
Resume	Explicit ExecutionResumed event	Journal is source of truth; status derived from last event
Work scheduling	Unified queue	All execution types same mechanism
Signals	Two-event (SignalDelivered + SignalReceived)	Durable buffer + promise-keyed consumption

---

Invariants

All invariants must hold at every journal state.

Structural Invariants

ID	Name	Property
S-1	monotonic_sequence	Sequence numbers strictly increasing
S-2	starts_with_started	First event is always ExecutionStarted
S-3	single_terminal	At most one terminal event
S-4	terminal_is_last	Terminal event is the final event
S-5	cancelled_requires_requested	ExecutionCancelled requires preceding CancelRequested

Side Effect Invariants

ID	Name	Property
SE-1	started_requires_scheduled	InvokeStarted(pid) requires preceding InvokeScheduled(pid)
SE-2	completed_requires_started	InvokeCompleted(pid) requires preceding InvokeStarted(pid)
SE-3	retrying_requires_started	InvokeRetrying(pid, attempt) requires preceding InvokeStarted(pid, attempt)
SE-4	no_events_after_completed	No InvokeStarted/Retrying after InvokeCompleted for same pid

Control Flow Invariants

ID	Name	Property
CF-1	timer_fired_requires_scheduled	TimerFired(pid) requires preceding TimerScheduled(pid)
CF-2	signal_received_requires_delivered	SignalReceived(name, delivery_id, payload) requires preceding SignalDelivered(name, delivery_id, payload)
CF-3	signal_consumed_once	Each delivery_id is consumed by at most one SignalReceived
CF-4	await_signal_consistent	AwaitSignal.promise_id must match the single waiting_on promise_id

JoinSet Invariants

ID	Name	Property
JS-1	submit_requires_created	JoinSetSubmitted(js) requires preceding JoinSetCreated(js)
JS-2	no_submit_after_await	No JoinSetSubmitted(js) after any JoinSetAwaited(js)
JS-3	awaited_requires_member	JoinSetAwaited(js, pid) requires preceding JoinSetSubmitted(js, pid)
JS-4	awaited_requires_completed	JoinSetAwaited(_, pid) requires preceding InvokeCompleted(pid)
JS-5	no_double_consume	No two JoinSetAwaited for same (js_id, pid) pair
JS-6	consume_bounded	Per set: count(JoinSetAwaited) <= count(JoinSetSubmitted)
JS-7	promise_single_owner	A promise_id appears in at most one join set

---

Journal Sequence: Full Example

A workflow that invokes a task, generates randomness, uses a JoinSet, and completes.

seq  event                                                 state
───  ─────                                                 ─────
 0   ExecutionStarted(component_digest, input)             Running
 1   RandomGenerated("root.0", 0x1a2b)                    Running
 2   InvokeScheduled("root.1", "fetch_user", {id:42})     Running
 3   ExecutionAwaiting(["root.1"], Single)                 Blocked
     ... child executes ...
 4   InvokeStarted("root.1", attempt=1)                   Blocked
 5   InvokeCompleted("root.1", Ok(User{...}), attempt=1)  Blocked
 6   ExecutionResumed                                      Running
 7   JoinSetCreated("root.2")                              Running
 8   InvokeScheduled("root.3", "send_email", {...})        Running
 9   JoinSetSubmitted("root.2", "root.3")                  Running
10   InvokeScheduled("root.4", "send_sms", {...})          Running
11   JoinSetSubmitted("root.2", "root.4")                  Running
12   ExecutionAwaiting(["root.3","root.4"], Any)            Blocked
     ... root.4 completes first ...
13   InvokeStarted("root.4", attempt=1)                   Blocked
14   InvokeCompleted("root.4", Ok(...), attempt=1)         Blocked
15   ExecutionResumed                                      Running
16   JoinSetAwaited("root.2", "root.4", Ok(...))           Running
     ... workflow calls js.next() again ...
17   ExecutionAwaiting(["root.3"], Any)                     Blocked
18   InvokeStarted("root.3", attempt=1)                   Blocked
19   InvokeRetrying("root.3", attempt=1, err, retry_at)    Blocked
20   InvokeStarted("root.3", attempt=2)                   Blocked
21   InvokeCompleted("root.3", Ok(...), attempt=2)         Blocked
22   ExecutionResumed                                      Running
23   JoinSetAwaited("root.2", "root.3", Ok(...))           Running
24   ExecutionCompleted(Ok(final_result))                   Completed

Journal Sequence: Signal Examples

Buffered signal (arrives before await_signal())

seq  event                                                          state
───  ─────                                                          ─────
 0   ExecutionStarted(component_digest, input)                      Running
 1   InvokeScheduled("root.0", "create_order", {...})               Running
 2   ExecutionAwaiting(["root.0"], Single)                          Blocked
     ... child executes ...
 3   InvokeStarted("root.0", attempt=1)                            Blocked
 4   InvokeCompleted("root.0", Ok(order), attempt=1)                Blocked
 5   ExecutionResumed                                               Running
 6   SignalDelivered("user_approval", {approved: true}, delivery_id=1)              Running  ← arrives while running
      ... workflow calls await_signal("user_approval") ...
 7   SignalReceived("root.1", "user_approval", {approved: true}, delivery_id=1)    Running  ← consumed instantly, no block
 8   ExecutionCompleted(Ok(result))                                  Completed

Blocking signal (arrives after await_signal())

seq  event                                                          state
───  ─────                                                          ─────
 0   ExecutionStarted(component_digest, input)                      Running
 1   InvokeScheduled("root.0", "create_order", {...})               Running
 2   ExecutionAwaiting(["root.0"], Single)                          Blocked
     ... child executes ...
 3   InvokeStarted("root.0", attempt=1)                            Blocked
 4   InvokeCompleted("root.0", Ok(order), attempt=1)                Blocked
 5   ExecutionResumed                                               Running
     ... workflow calls await_signal("user_approval") ...
 6   ExecutionAwaiting(["root.1"], Signal("user_approval"))         Blocked  ← no signal yet, trap
     ... hours pass, user approves ...
 7   SignalDelivered("user_approval", {approved: true}, delivery_id=1)              Blocked
 8   SignalReceived("root.1", "user_approval", {approved: true}, delivery_id=1)    Blocked  ← resolved but not resumed yet
 9   ExecutionResumed                                               Running
10   ExecutionCompleted(Ok(result))                                  Completed

In both cases, the replayer indexes SignalReceived by promise_id into CachedResult::Signal(payload). On replay, await_signal("user_approval") generates "root.1" → cache HIT → returns payload instantly. SignalDelivered events are ignored by the replayer (they exist only for durable buffering).

---