Control and coordination

Locks, waits, suspension, error signalling and source control. Every parameter with its type, default and description, plus worked examples.

Lock

Distributed mutex — serialize concurrent workflow runs on a named resource. Two backends: postgres (lease table) and dragonfly (Redis-compatible).

acquire returns an opaque token that must be supplied to refresh and release. There is no fencing token; the lease is enforced only at the lock store, not at any downstream resource.

Task: lock

Single task that dispatches on op: (acquire / refresh / release / peek).

Common parameters

Field	Type	Required	Default	Description
op	string	No	"acquire"	Operation: acquire / refresh / release / peek
backend	string	No	"postgres"	postgres or dragonfly
dsn	string	Yes	—	Postgres DSN or Redis URL
table	string	No	kis_locks	Postgres lease table name (ignored for dragonfly)
key	string	Yes	—	Lock key
setvar	string	No	—	Capture the result map into this workflow variable

Per-op parameters

Field	Type	Used by	Description
ttl	duration	acquire, refresh	Lease duration
wait	duration	acquire	Block-and-retry budget. Default 0 = fail fast
owner	string	acquire	Optional metadata stamped on the lease
token	string	refresh, release	Token from the matching acquire

Durations accept Go-style strings (5m) or a bare integer (seconds).

Returns

acquire: { success, acquired, attempts, key, token, expiresAt, error }. refresh: { success, refreshed, expiresAt, error }. release: { success, released, error }. peek: { success, held, token, owner, expiresAt, error }.

Examples

Serialize deployments (Postgres)

name: serialize-deployments
list: true
tasks:
  - name: take-deploy-lock
    lock:
      op: acquire
      backend: postgres
      dsn: "{{lock_dsn}}"
      key: deploy/prod
      ttl: 10m
      wait: 30s
      owner: "deploy-{{run_id}}"
      setvar: deploy_lock

  - name: deploy
    shell:
      script: ./deploy.sh prod

  - name: drop-deploy-lock
    lock:
      op: release
      backend: postgres
      dsn: "{{lock_dsn}}"
      key: deploy/prod
      token: "{{deploy_lock.token}}"

Cert rotation via Dragonfly

- name: rotation-lock
  lock:
    op: acquire
    backend: dragonfly
    dsn: "redis://dragonfly.internal:6379/0"
    key: certs/rotation
    ttl: 5m
    wait: 1m
    setvar: rot

Refresh during a long task

- name: refresh-mid
  lock:
    op: refresh
    dsn: "{{lock_dsn}}"
    key: migration/2026-05-19
    token: "{{mig.token}}"
    ttl: 2m

Peek before acting

- name: who-holds-it
  lock:
    op: peek
    backend: dragonfly
    dsn: "redis://dragonfly.internal:6379/0"
    key: nightly-report
    setvar: holder

- name: report
  when: "{{not holder.held}}"
  shell:
    script: ./nightly-report.sh

Namespace: lock

Same operations as separate Script functions: lock.acquire, lock.refresh, lock.release, lock.peek. See docs/namespaces/lock.md for the full namespace reference.

Wait

Polling primitive — blocks until a runtime condition becomes true. Active complement to the passive suspend task.

wait is a single task that dispatches on a for: discriminator. Today only for: http is implemented; tcp / dns / exec / file / db variants are planned.

Task: wait

Parameters

Field	Type	Required	Default	Description
for	string	No	"http"	Wait kind
setvar	string	No	—	Capture the result map into this workflow variable

for: http parameters

Field	Type	Required	Default	Description
url	string	Yes	—	Target URL
method	string	No	"GET"	HTTP method
headers	map	No	—	Request headers — sent on every poll
body	string	No	—	Request body — reused on every poll
status	int or [int]	No	200	Acceptable status code(s)
bodymatch	string	No	—	Regex the response body must match
bodycontains	string	No	—	Substring the response body must contain
headermatch	map	No	—	Per-header regex constraints
interval	duration	No	5s	Poll interval
timeout	duration	No	5m	Total timeout
initialdelay	duration	No	0	Delay before the first poll
insecuretls	bool	No	false	Skip TLS certificate verification
expectfailure	bool	No	false	Succeed when the URL stops responding

Durations accept Go-style strings (5s, 1m30s) or a bare integer (seconds).

Returns

Field	Type	Description
success	bool	Whether the wait condition was met
attempts	int	Number of polls performed
elapsedMs	int	Total elapsed milliseconds
lastStatus	int	HTTP status of the final poll
lastBody	string	Body of the final poll
lastHeaders	map	Headers of the final poll
error	string	Failure reason on timeout / cancellation

Examples

Wait for a service to come up

name: wait-for-health
list: true
tasks:
  - name: deploy
    shell:
      script: ./deploy.sh

  - name: ready
    wait:
      for: http
      url: http://localhost:8080/healthz
      status: 200
      interval: 2s
      timeout: 90s

Poll a job until it reports “done”

name: wait-for-job
tasks:
  - name: complete
    wait:
      for: http
      url: "https://api.example.com/jobs/{{job_id}}"
      headers:
        Authorization: "Bearer {{api_token}}"
      bodycontains: '"status":"done"'
      interval: 5s
      timeout: 30m

Wait for shutdown (invert)

- name: gone
  wait:
    for: http
    url: http://localhost:8080/healthz
    expectfailure: true
    interval: 1s
    timeout: 30s

Capture the result for downstream tasks

- name: ready
  wait:
    for: http
    url: https://api.example.com/state
    bodycontains: '"ready":true'
    setvar: state_check

- name: report
  print:
    message: "Ready after {{state_check.attempts}} polls in {{state_check.elapsedMs}}ms."

Namespace: wait

Execute the same polls from Script scripts.

Functions

Function	Description
wait.http()	Poll an HTTP endpoint until the response satisfies the criteria

Parameters

Parameter	Type	Default	Description
url	string	—	Target URL (required)
method	string	"GET"	HTTP method
headers	map	—	Request headers
body	string	—	Request body
status	int or [int]	200	Acceptable status code(s)
bodyMatch	string	—	Regex the body must match
bodyContains	string	—	Substring the body must contain
headerMatch	map	—	Per-header regex constraints
interval	int or string	5s	Poll interval ("5s" or integer seconds)
timeout	int or string	5m	Total timeout
initialDelay	int or string	0	Delay before the first poll
insecureTLS	bool	false	Skip TLS verification
expectFailure	bool	false	Succeed when the URL stops responding

Returns

Field	Type	Description
success	bool	Whether the condition was met
attempts	int	Polls performed
elapsedMs	int	Total elapsed milliseconds
lastStatus	int	Status of the final poll
lastBody	string	Body of the final poll
lastHeaders	map	Headers of the final poll
error	string	Failure reason

Examples

// Wait for an endpoint to return 200
let r = wait.http({
  url: "http://localhost:8080/healthz",
  interval: "2s",
  timeout: "90s"
})
if (!r.success) {
  throw "service never came up: " + r.error
}

// Block on a long-running job
let r = wait.http({
  url: "https://api.example.com/jobs/" + jobId,
  headers: { Authorization: "Bearer " + token },
  bodyContains: '"status":"done"',
  interval: "5s",
  timeout: "30m"
})

// Wait for shutdown
wait.http({
  url: "http://localhost:8080/healthz",
  expectFailure: true,
  interval: "1s",
  timeout: "30s"
})

// Match a status list and inspect the final response on timeout
let r = wait.http({
  url: "https://api.example.com/v",
  status: [200, 204],
  bodyMatch: "\"version\":\"2\\.\\d+\\.\\d+\"",
  timeout: "10m"
})
log("attempts=" + r.attempts + " lastStatus=" + r.lastStatus)

See docs/namespaces/wait.md for the dedicated namespace reference.

Suspend

Pause workflow execution until triggered with data. Can suspend at different phases of task execution.

Parameters

Parameter	Type	Required	Default	Description
when	string	No	execute	When to suspend: execute, before, after
message	string	No	-	Message to log when suspending

When Options

Value	Description
execute	Suspend during Execute phase, resumes when triggered with data
before	Suspend before Execute runs
after	Suspend after Execute completes

Examples

Basic Suspend for Manual Approval

name: suspend-approval
list: true
tasks:
  - name: deploy-staging
    shell: ./deploy.sh staging

  - name: wait-for-approval
    suspend:
      message: "Waiting for production deployment approval"

  - name: deploy-production
    shell: ./deploy.sh production

Suspend Before Execute

name: suspend-before
tasks:
  - name: wait-before-start
    suspend:
      when: before
      message: "Workflow paused before starting task"

Manual Gate in Pipeline

name: suspend-gate
list: true
tasks:
  - name: run-tests
    shell: npm test

  - name: manual-qa-gate
    suspend:
      message: "Waiting for QA team approval"

  - name: publish
    shell: npm publish

From a script

There is no suspend namespace, and there cannot be one. Suspending means stopping a durable run and resuming it later from its record; a script has no record to resume from.

When work needs to pause for an approval or an external event, that is the signal to write a flow. See Wait.

Throw

Test utility task that throws errors at configurable phases. Useful for testing error handling in workflows.

Parameters

Parameter	Type	Required	Default	Description
phase	string	No	execute	When to throw: execute, before, after, all
message	string	No	-	Custom error message

Phase Options

Value	Description
execute	Throw error during Execute phase (default)
before	Throw error during BeforeExecute phase
after	Throw error during AfterExecute phase
all	Throw error in all phases

Examples

Simple Error

name: throw-simple
tasks:
  - name: fail-task
    throw:
      message: "This task intentionally fails"

Error at Specific Phase

name: throw-before
tasks:
  - name: fail-before-execute
    throw:
      phase: before
      message: "Error during initialization"

Test Error Handling

name: throw-test-handling
tasks:
  - name: trigger-error
    throw:
      phase: execute
      message: "Simulated failure for testing"
    error:
      go: handle-error

  - name: handle-error
    print:
      message: "Error was caught and handled"

Error was caught and handled
execution engine instance Status: errored  Errored Nodes: [trigger-error]

Note the status. Handling a failure does not unmake it — the run is still reported errored, with the node that failed named. That is what makes throw: useful for rehearsing an error path: you can see both that the handler ran and that the run was honest about why.

Error routing needs graph mode. In a list: true flow the error: key is dropped, so an error-handling example has to declare its edges.

From a script

There is no throw namespace. Use the language:

if (!r.success) throw new Error(`deploy failed: ${r.error}`);

The task exists because a flow’s YAML has no way to raise an error; a script does.

Git

Git operations using go-git (pure Go).

Operations

Operation	Description
clone	Clone a repository
commit	Create a commit
push	Push to remote
pull	Pull from remote
status	Get repository status
add-remote	Add a remote
remove-remote	Remove a remote

Parameters

Parameter	Type	Required	Description
command	string	Yes	Operation to perform
directory	string	Yes*	Repository directory
url	string	Yes*	Repository URL (for clone)
file	string	No	File to add and commit
content	string	No	Content to write to file
message	string	No	Commit message
author	string	No	Author name
email	string	No	Author email
remote	string	No	Remote name (default: origin)
username	string	No	HTTP basic auth username
password	string	No	HTTP basic auth password/token
depth	int	No	Clone depth (0 = full)

Examples

Clone Repository

name: git-clone
tasks:
  - name: clone
    git:
      command: clone
      url: https://github.com/user/repo.git
      directory: /path/to/dest
      username: "{{git_user}}"
      password: "{{git_token}}"

Commit and Push

name: git-commit-push
list: true
tasks:
  - name: commit
    git:
      command: commit
      directory: /path/to/repo
      file: config.yaml
      content: "key: value"
      message: "Update config"
      author: John Doe
      email: [email protected]

  - name: push
    git:
      command: push
      directory: /path/to/repo
      remote: origin
      username: "{{git_user}}"
      password: "{{git_token}}"

Namespace: git

Git version control functions.

Functions

Function	Description
git.clone()	Clone a repository
git.commit()	Create a commit
git.push()	Push to remote
git.pull()	Pull from remote
git.status()	Get repository status
git.addRemote()	Add a remote
git.removeRemote()	Remove a remote

Clone Parameters

Parameter	Type	Default	Description
url	string	-	Repository URL (required)
directory	string	-	Local directory path (required)
username	string	-	Auth username
password	string	-	Auth password/token
depth	int	0	Clone depth (0 = full)

Commit Parameters

Parameter	Type	Default	Description
directory	string	-	Repository directory (required)
message	string	-	Commit message
author	string	-	Author name (required)
email	string	-	Author email (required)
file	string	-	File to add/modify
content	string	-	Content to write to file

Push/Pull Parameters

Parameter	Type	Default	Description
directory	string	-	Repository directory (required)
remote	string	"origin"	Remote name
username	string	-	Auth username
password	string	-	Auth password/token

Examples

// Clone a repository
let result = git.clone({
    url: "https://github.com/user/repo.git",
    directory: "/path/to/dest",
    username: input.git_user,
    password: input.git_token
})

// Get repository status
let status = git.status({
    directory: "/path/to/repo"
})
if (status.success) {
    log("Is clean:", status.clean)
    log("Changed files:", status.files)
}

// Commit changes
git.commit({
    directory: "/path/to/repo",
    file: "config.yaml",
    content: "key: value\n",
    message: "Update configuration",
    author: "John Doe",
    email: "[email protected]"
})

// Push to remote
git.push({
    directory: "/path/to/repo",
    remote: "origin",
    username: input.git_user,
    password: input.git_token
})

// Pull from remote
git.pull({
    directory: "/path/to/repo",
    remote: "origin",
    username: input.git_user,
    password: input.git_token
})

// Add remote
git.addRemote({
    directory: "/path/to/repo",
    name: "upstream",
    url: "https://github.com/upstream/repo.git"
})

// Remove remote
git.removeRemote({
    directory: "/path/to/repo",
    name: "upstream"
})

DNS

DNS record CRUD (via libdns) and system-resolver lookups. Single task that dispatches on op:. Six providers in v1: cloudflare, route53, gcloud, azure, digitalocean, godaddy.

The same providers are reused by the letsencrypt task for DNS-01 challenges, so credentials map between them.

Task: dns

Common parameters (CRUD)

Field	Type	Required	Default	Description
op	string	No	list	list / set / append / delete / lookup
provider	string	Yes (CRUD)	—	Provider name (see spec)
credentials	map	Yes (CRUD)	—	Provider-specific auth (see spec for keys)
zone	string	Yes (CRUD)	—	Zone name (example.com)
timeout	duration	No	—	Wall-clock cap
setvar	string	No	—	Capture result map into a workflow variable

Record shape

Field	Type	Description
type	string	A, AAAA, CNAME, TXT, MX, SRV, NS, etc.
name	string	Relative to zone (www); @ for apex
value	string	For MX: hostname. For SRV: "port target". Otherwise the literal value
ttl	duration	TTL
priority	int	MX or SRV priority
weight	int	SRV weight

Returns

Op	Adds to base {success, error}
list / set / append / lookup	records
delete	deleted

Examples

Add a record

- name: add-www
  dns:
    op: append
    provider: cloudflare
    credentials:
      api_token: "{{ cf_token }}"
    zone: example.com
    records:
      - type: A
        name: www
        value: 203.0.113.10
        ttl: 300

Upsert (set replaces existing RRset)

- name: spf
  dns:
    op: set
    provider: cloudflare
    credentials: { api_token: "{{ cf_token }}" }
    zone: example.com
    records:
      - type: TXT
        name: "@"
        value: "v=spf1 include:_spf.google.com -all"

MX records (priority surfaced as its own field)

- name: mx
  dns:
    op: set
    provider: cloudflare
    credentials: { api_token: "{{ cf_token }}" }
    zone: example.com
    records:
      - { type: MX, name: "@", value: "primary-mx.example.com.", priority: 10 }
      - { type: MX, name: "@", value: "secondary-mx.example.com.", priority: 20 }

Verify propagation via system resolver

- name: lookup
  dns:
    op: lookup
    name: www.example.com
    type: A
    server: 1.1.1.1:53
    timeout: 10s
    setvar: result

- name: report
  print:
    message: "Resolved to {{ result.records[0].value }}"

Namespace: dns

Same five operations as Script functions: dns.list / dns.set / dns.append / dns.delete / dns.lookup. See docs/namespaces/dns.md for the full namespace reference.

// Add a record
dns.append({
  provider: "cloudflare",
  credentials: { api_token: cfToken },
  zone: "example.com",
  records: [{ type: "A", name: "www", value: "203.0.113.10", ttl: 300 }]
})

// Verify propagation
let r = dns.lookup({ name: "www.example.com", type: "A", server: "1.1.1.1:53" })
if (r.records.length === 0) {
  throw "DNS not propagated yet"
}

Scraper

Web scraping using an external scraper service. Extracts content from web pages.

Parameters

Parameter	Type	Required	Default	Description
scraperurl	string	Yes	-	Scraper service URL
urls	list	Yes	-	List of URLs to scrape
setvar	string	No	-	Variable to store results
continueonerror	bool	No	false	Continue on scraping errors
timeout	string	No	5m	Scraping timeout
contextlimit	int	No	15000	Max characters per chunk

URL Configuration

Property	Type	Default	Description
url	string	-	URL to scrape
depth	int	1	Link crawl depth
limit	int	2	Max pages to scrape
secure	bool	false	Use secure mode
structured	bool	false	Return structured data
authenticate	map	-	Authentication config

Examples

Basic Scraping

name: scraper-basic
vars:
  scraper_url: "https://scraper.example.com"
tasks:
  - name: scrape-docs
    scraper:
      scraperurl: "{{scraper_url}}"
      urls:
        - url: "https://docs.example.com/guide"
          depth: 2
          limit: 10
      setvar: doc_content

Multiple URLs

name: scraper-multi
tasks:
  - name: scrape-competitors
    scraper:
      scraperurl: "{{scraper_url}}"
      continueonerror: true
      timeout: 10m
      urls:
        - url: "https://competitor1.com/pricing"
        - url: "https://competitor2.com/pricing"
        - url: "https://competitor3.com/pricing"
      setvar: pricing_data

With Authentication

name: scraper-auth
tasks:
  - name: scrape-protected
    scraper:
      scraperurl: "{{scraper_url}}"
      urls:
        - url: "https://internal.example.com/docs"
          depth: 3
          limit: 50
          authenticate:
            type: bearer
            token: "{{api_token}}"
      setvar: internal_docs

From a script

A scrape namespace is bound — scrape.web, .pdf, .excel, .docx, .pptx — but every one of them needs the pipeline environment that AI Flow builds around a running flow, for service routing. Called from a plain kis script run they report that and name the task to use:

scrape.web requires a pipeline environment with ProxySelector for service routing.
Use the aiflow YAML interface (scrapedata task) instead.

So the flow task is the surface here. The namespace is usable from a script: task inside an AI Flow, where the environment exists. See the AI library for which of its functions carry the same requirement.