AWL workloops

Declares perpetual work: a flow whose completion is an incident, armed by `every` or `on`, carrying invariants, retention, and a declared `retire`.

• form declaration
• positions top-level

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Declares a typed, defaulted value threaded iteration to iteration; `route start(name: expr)` rebinds it at the boundary.

• form header-clause
• positions workloop-header

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Declares a health object with a type, tolerance and confirming route. Health samples execute; AWL cannot write its typed current-state value.

• form header-clause
• positions workloop-header

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Declares when an unconfirmed invariant alarms — `N windows` of unhealthy samples or `unconfirmed for <duration>` — and has no default.

• form header-clause
• positions workloop-header

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Declares the required prior-generation window. The store applies it when current-state invariant values are written; AWL currently produces no such values.

• form header-clause
• positions workloop-header

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Starts a declared detached workflow that outlives the iteration and owes it nothing; every hatch carries a `key` dedupe identity.

• form statement
• positions step-body

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Declares the workloop's wind-down block: the recorded, non-failure way the loop stops, required on every workloop.

• form declaration
• positions top-level

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Declares the start contract for an independently running workflow this document may hatch. The contract supplies input only.

• form declaration
• positions top-level

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Completes a count-form tolerance: the number of consecutive unhealthy samples the invariant tolerates.

• form header-clause
• positions workloop-header

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Declares a typed answer re-registered during an iteration. It is queryable only while that generation is resident; a parked loop cannot answer it.

• form header-clause
• positions workloop-header

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Names the route whose firing lands as a green health sample for the invariant above it — required on every invariant, because one nothing can confirm alarms forever.

• form header-clause
• positions workloop-header

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

The iteration terminal after `route`: closes this iteration and opens the next generation, rebinding carries as the payload.

• form target-word
• positions route-target

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)

Opens a duration-form tolerance: the invariant alarms when unconfirmed for the declared window — the only form silence can trip.

• form header-clause
• positions workloop-header

//! A signal-armed loop must tolerate by duration: silence has no samples.
workloop mail_watch
  on nudge
  input box_dir: String
  signal nudge: Nudge
  report board: type Board
  invariant serving: type Board
    tolerance unconfirmed for 45m
    confirms start
  retention 7d

type Nudge { reason: String }
type Board { handled: Int }

worker handlers
  action handle(box_dir: String) -> Board

step handle_nudge
  handle(box_dir: box_dir) -> board
  answer board(handled: board.handled)
  route start

retire
  handle(box_dir: box_dir)

Connects `unconfirmed` to its window in a duration-form tolerance.

• form header-clause
• positions workloop-header

//! A signal-armed loop must tolerate by duration: silence has no samples.
workloop mail_watch
  on nudge
  input box_dir: String
  signal nudge: Nudge
  report board: type Board
  invariant serving: type Board
    tolerance unconfirmed for 45m
    confirms start
  retention 7d

type Nudge { reason: String }
type Board { handled: Int }

worker handlers
  action handle(box_dir: String) -> Board

step handle_nudge
  handle(box_dir: box_dir) -> board
  answer board(handled: board.handled)
  route start

retire
  handle(box_dir: box_dir)

Carries a hatch's mandatory dedupe identity: a retried or replayed iteration re-mints the same identity, so one subject hatches one workflow.

• form statement
• positions step-body

//! Poll a queue on cadence; hatch one detached run per task found.
workloop queue_watch
  every 25m
  input queue_dir: String
  carry seen: [String] = []
  report board: type Board
  invariant serving: type Board
    tolerance 3 windows
    confirms start
  retention 14d

type Board { swept: Int }
type Task  { id: String, kind: String, payload: String }
type Sweep { tasks: [Task], seen: [String], swept: Int }

worker sweepers
  action sweep(queue_dir: String, seen: [String]) -> Sweep

detached process_task(kind: String, payload: String)

step sweep_queue
  sweep(queue_dir: queue_dir, seen: seen) -> found
  answer board(swept: found.swept)
  outcome idle: if found.tasks is empty, route start
  outcome busy: else, route dispatch

step dispatch
  fork task in found.tasks
    hatch process_task(kind: task.kind, payload: task.payload) key task.id
  join
  route start(seen: found.seen)

retire
  sweep(queue_dir: queue_dir, seen: seen)