Workflows

A workflow runs a business process in a known order. Reach for one when work needs multiple named steps, typed data flowing between them, an object action at the end, and a run history you can inspect.

A workflow chains nodes: steps (typed functions that produce data), action nodes (which run an object action), and interventions (which pause for a human decision — see Interventions).

When to use a workflow

Workflows coordinate the work; steps and actions do it.

Define steps

A step is a typed function: it declares the input it needs, the output it returns, and the code that runs. The .run(...) handler receives { input, sixb } — use sixb to read or write objects.

import { defineWorkflowStep, ref } from "@sixb/core"
import { Invoice } from "../ontology/invoice"

const loadInvoiceContext = defineWorkflowStep("load-invoice-context")
  .input({ invoice: ref(Invoice) })
  .output({
    invoice: ref(Invoice),
    invoiceNumber: "string",
    amountLabel: "string",
    status: "string",
  })
  .run(async ({ input, sixb }) => {
    const invoice = await sixb.objects(Invoice).get(input.invoice.primaryId)
    if (!invoice) {
      throw new Error(`[AcmeCorp] Invoice '${input.invoice.primaryId}' was not found.`)
    }

    const amount = Number(invoice.properties.amount ?? 0)
    const currency = String(invoice.properties.currency ?? "USD")

    return {
      invoice: input.invoice,
      invoiceNumber: String(invoice.properties.number ?? invoice.primaryId),
      amountLabel: `${currency} ${amount.toFixed(2)}`,
      status: String(invoice.properties.status ?? "unknown"),
    }
  })

A field typed ref(Invoice) carries an object reference of the shape { objectTypeId, primaryId }.

Compose the chain

defineWorkflow(id) declares the input, then .then(...) appends each node in order.

import { defineWorkflow, ref } from "@sixb/core"
import { sendReminder } from "../actions/send-reminder"
import { Invoice } from "../ontology/invoice"

export const invoiceReminder = defineWorkflow("invoice-reminder")
  .input({ invoice: ref(Invoice) })
  .then(loadInvoiceContext)
  .then(composeReminder)
  .then(reviewReminder)
  .then(sendReminder, ({ steps }) => ({
    subject: steps.composeReminder.invoice,
    params: {
      approved: steps.reviewReminder.approved,
      message: steps.reviewReminder.message,
      reviewerNote: steps.reviewReminder.reviewerNote,
    },
  }))

Each node id becomes a camelCase key on steps: compose-reminder is steps.composeReminder.

Pass data between nodes

Use a bare .then(node) when the previous output already matches the next node's input shape.

Use a mapper when the next node needs a different shape. A mapper receives { input, steps } — the original workflow input plus every earlier node's output — and returns the next node's input.

  .then(composeReminder, ({ input, steps }) => ({
    invoice: input.invoice,
    amountLabel: steps.loadInvoiceContext.amountLabel,
  }))

Action nodes

An action node runs an object action and waits for it to finish. Action nodes must use a mapper. For an object action, the mapper returns subject (the object the action runs on) and params:

  .then(sendReminder, ({ steps }) => ({
    subject: steps.composeReminder.invoice,
    params: {
      approved: steps.reviewReminder.approved,
      message: steps.reviewReminder.message,
      reviewerNote: steps.reviewReminder.reviewerNote,
    },
  }))

For a global action, omit subject and return only params.

Pause for a human with interventions

An intervention node suspends the run until someone submits a response, then resumes with that response on steps. Define it with defineIntervention(id, options?): .input(...) is the data the reviewer sees, .response(...) is the decision they submit, and .defaults(...) pre-fills the form.

import { defineIntervention, interventionField, ref } from "@sixb/core"

export const reviewReminder = defineIntervention("review-reminder", {
  description: "Approve or request changes before sending the invoice reminder.",
})
  .input({ invoice: ref(Invoice), message: "string" })
  .response({
    approved: interventionField("boolean", { required: true }),
    message: interventionField("string", { required: true }),
    reviewerNote: interventionField("string", { required: false }),
  })
  .defaults(({ input }) => ({
    approved: true,
    message: input.message,
  }))

Add it to the chain, then read the response in a later node (see the sendReminder mapper above, which reads steps.reviewReminder.approved). See Interventions for the full reference and how to submit a response.

Start a run

Registered workflows are reachable through sixb.workflows.

const result = await sixb.workflows.request(invoiceReminder, {
  input: {
    invoice: { objectTypeId: "Invoice", primaryId: "inv-001" },
  },
})

console.log(result.runId, result.created)

request(...) returns { workflowId, runId, queuedAt, jobId?, created }. created is false when a run with the same runId already existed, so no duplicate job was enqueued.

Use sixb.workflows.requestById(...) from server routes and dynamic cases where you only have the id. The other primitives expose the same grouped shape through actions.request, syncs.request, pipelines.request, and agents.request: everything is queued, never run inline.

await sixb.workflows.requestById({
  workflowId: "invoice-reminder",
  input: {
    invoice: { objectTypeId: "Invoice", primaryId: "inv-001" },
  },
})

The name is uniform; the guarantee is not, and the return type says so. A workflow run row exists as soon as workflows.requestById resolves, which is what lets it report created. Sync and pipeline runs are created by the worker that claims the job, so those two verbs report only that the request was accepted — passing the same runId twice still produces one run, but the second call cannot tell you it was a duplicate.

Start from a schedule

A workflow can auto-start from a schedule with .when(...). A schedule says "run now" — it provides no business object, so scheduled workflows must take empty input. If the run needs an invoice or customer, start it from your app or the API instead.

import { defineSchedule, defineWorkflow } from "@sixb/core"

export const daily = defineSchedule("daily-invoice-reminders").cron("0 9 * * *")

export const dailyInvoiceReminders = defineWorkflow("daily-invoice-reminders")
  .input({})
  .when(daily)
  .then(findOverdueInvoices)

Bind to an event schedule

A workflow can reference an event schedule with .when(schedule, mapper?). Use a mapper when the selected event must be converted into the workflow input.

export const reviewHighValuePayment = defineWorkflow("review-high-value-payment")
  .input({
    invoice: ref(Invoice),
    payment: ref(Payment),
    amount: "double",
  })
  .when(highValuePaymentLinked, ({ event }) => ({
    invoice: event.source,
    payment: event.target,
    amount: event.link.p.amount,
  }))
  .then(reviewPayment)

Run lifecycle

A run moves through these statuses, recorded per run and per node so you can inspect history:

Nodes run sequentially. Step input and output are validated against their schemas at runtime, so a malformed shape fails the run rather than passing bad data downstream.

GET /api/workflows carries a latestRun per workflow. It is null when the workflow has never run, and also when the storage provider keeps no run history — every published provider does, so only a partial custom provider hits that second case.

Register and run

Put workflow definitions under workflows/ and export them. createSixb() discovers them automatically, along with the actions and ontology they reference.

your-project/
  actions/
    send-reminder.ts
  ontology/
    invoice.ts
  workflows/
    invoice-reminder.ts
  sixb.config.ts

You can also register them explicitly. createSixb() is async:

import { createSixb } from "@sixb/core"
import { sendReminder } from "./actions/send-reminder"
import { Invoice } from "./ontology/invoice"
import { invoiceReminder } from "./workflows/invoice-reminder"

export const sixb = await createSixb({
  ontologies: [Invoice],
  actions: [sendReminder],
  workflows: [invoiceReminder],
})

sixb dev runs workflow workers automatically when workflows are registered. For a dedicated worker process in production, run:

sixb worker workflow

Related

• Interventions — human-in-the-loop pauses and submitting responses
• Actions — commands run by action nodes
• Rules — watch objects and react
• Automation — schedules that auto-start workflows