makeDecisions(task: DecisionTask, cb: {(Error?)}): any {
   const input = task.getWorkflowInput()
   if (input.handler !== 'taskGraph') return cb(new Error('invalid handler for taskGrah'))
   const parameters = input.parameters
   try {
     this.decide(parameters, task)
   } catch (err) {
     task.failWorkflow('error in decider', JSON.stringify(err).slice(0, 250))
   }
   cb()
 }

 buildTaskMeta(task: DecisionTask, meta?: Object): Object {
   let wfMeta = task.getWorkflowInfo() as WorkflowWithParent
   let parentWf = task.getParentWorkflowInfo()
   if (parentWf) {
     wfMeta.parentWorkflowId = parentWf.workflowId
   }
   let taskMeta = {
     task: { type: 'taskGraph', id: task.id },
     workflow: wfMeta
   }
   return _.defaults(taskMeta || {}, meta || {})
 }

  decide(parameters: TaskGraphParameters, decisionTask: DecisionTask) {
    const graph = parameters.graph
    const groupedEvents = decisionTask.getGroupedEvents()
    let next = this.getNextNodes(graph, groupedEvents)
    let startCountByHandler = {}
    let startCountSubWorkflows = 0
    for (let node of next.nodes) {
      if (node.type === 'decision') {
        // TODO: somehow hand off to a child? need to make this more generic but just hard code for now...
        if (node.handler === 'taskGraph') {
          let tgNode = node as TaskGraphGraphNode
          const shouldThrottle = this.throttleWorkflows(tgNode, graph, groupedEvents, startCountSubWorkflows)
          if (!shouldThrottle) {
            startCountSubWorkflows++
            const maxRetry = tgNode.maxRetry || this.ftlConfig.getOpt('maxRetry')
            decisionTask.startChildWorkflow(tgNode.id, tgNode, {maxRetry: maxRetry})
          }
        }
        else if (node.handler === 'recordMarker') {
          decisionTask.addMarker(node.id, node.parameters.status)
        }
        else {
          this.logger.warn('couldn\'t find hander for child node', node)
        }
      }
      else {
        const shouldThrottle = this.throttle(node, graph, groupedEvents, startCountByHandler)
        if (!shouldThrottle) {
          startCountByHandler[node.handler] = startCountByHandler[node.handler] || 0
          startCountByHandler[node.handler]++
          const handlerActType = this.activities.getModule(node.handler)
          if (!handlerActType) throw new Error('missing activity type ' + node.handler)
          let opts = this.buildOpts(node)
          opts['maxRetry'] = node.maxRetry || handlerActType.getMaxRetry() || this.ftlConfig.getOpt('maxRetry')
          decisionTask.scheduleTask(node.id, node, handlerActType, opts)
        }
      }
    }
    const failedToReFail = decisionTask.rescheduleFailedEvents()
    const failedToReTimeOut = decisionTask.rescheduleTimedOutEvents()
    const failedToReStart = decisionTask.rescheduleFailedToSchedule()
    const failedToReschedule = failedToReFail
    .concat(failedToReTimeOut)
    .concat(failedToReStart)
    if (failedToReschedule.length > 0) {
      this.logger.warn('failed to reschedule all previously failed events')
      decisionTask.failWorkflow('failed to reschedule previously failed events', JSON.stringify(failedToReschedule).slice(0, 250))
    }
    else if (next.finished) {
      // TODO: better results
      decisionTask.completeWorkflow({status: 'success'})
    }

  }

 onDecisionMade(task: DecisionTask) {
   const finishTime = this.decisionTimers[task.id]
   delete this.decisionTimers[task.id]
   this.ftlConfig.metricReporter.decrement('decider.running')
   this.ftlConfig.metricReporter.increment('decider.completed')
   this.ftlConfig.metricReporter.timing('decider.timer', finishTime)
   this.logInfo('responded to decision task', this.buildTaskMeta(task, { results: task.getDecisionInfo() }))
   const failedWorkflows = task.decisions.filter((d) => d.decision.decisionType === 'FailWorkflowExecution')
   // there should only really be one failedWorkflow
   if (failedWorkflows.length) {
     const wf = failedWorkflows[0]
     this.ftlConfig.notifier.sendError('workflowFailed', {
       workflow: task.getWorkflowInfo(),
       control: task.getWorkflowTaskInput().control,
       parentWf: task.getParentWorkflowInfo(),
       originWorkflow: task.getOriginWorkflow(),
       details: wf.decision.failWorkflowExecutionDecisionAttributes!.details,
       reason: wf.decision.failWorkflowExecutionDecisionAttributes!.reason
     })
   }
   this.emit('decisionCompleted', task.decisions.map((d) => d.decision ))
 }