Function drive

packages/infra/src/WorkflowEngineCosmos.ts:366–439

(
    executionId: string,
    payload: object,
    parent: string | undefined,
    entry: Registered
  )

  // --- Drive logic -------------------------------------------------------

  const drive = (
    executionId: string,
    payload: object,
    parent: string | undefined,
    entry: Registered
  ): Effect.Effect<void> =>
    Effect.gen(function*() {
      let local = locals.get(executionId)
      if (local?.fiber) {
        const polled = local.fiber.pollUnsafe()
        const stillRunning = !polled
        const completedNotResume = polled && polled._tag === "Success" && polled.value._tag === "Complete"
        if (stillRunning || completedNotResume) return
      }

      const stateOpt = yield* readExec(executionId)
      if (Option.isNone(stateOpt) || stateOpt.value.status === "complete") return

      // Best-effort claim: takes lease so recovery poller leaves us alone.
      // Failure is tolerated — local fiber still drives; OCC guards persisted
      // state so split-brain stays correct.
      const claimed = yield* tryClaim(stateOpt.value)
      const state = Option.isSome(claimed) ? claimed.value : stateOpt.value

      const instance = WorkflowInstance.initial(entry.workflow, executionId)
      instance.interrupted = state.interrupted
      if (!local) {
        local = { instance, fiber: undefined, parent }
        locals.set(executionId, local)
      } else {
        local.instance = instance
      }

      const onComplete = Effect.fnUntraced(function*(result: Workflow.Result<unknown, unknown>) {
        const current = yield* readExec(executionId)
        if (Option.isNone(current) || current.value.status === "complete") return
        const isComplete = result._tag === "Complete"
        const completedResult = isComplete ? yield* encodeResult(entry.workflow, result) : undefined
        yield* replaceExec({
          ...current.value,
          status: isComplete ? "complete" : current.value.status,
          suspended: result._tag === "Suspended",
          // Never downgrade a persisted interrupt: a concurrent `interrupt` may
          // have set the flag while this driver was suspending. Losing it would
          // leave a re-drive unable to collapse the suspension.
          interrupted: instance.interrupted || current.value.interrupted,
          completedResult,
          // Release lease on completion so the doc isn't seen as orphaned.
          worker: isComplete ? undefined : current.value.worker,
          leaseExpiresAt: isComplete ? undefined : current.value.leaseExpiresAt
        })
          .pipe(Effect.catchTag("OptimisticConcurrencyException", () => Effect.void))
        if (parent && isComplete) {
          yield* Effect.forkIn(driveById(parent), scope)
        }
      })

      local.fiber = yield* entry.execute(payload, executionId).pipe(