Documentation
¶
Overview ¶
Package worker is Atlas's out-of-process job worker: the binary, run as a client of its own HTTP API, leasing jobs of named types and reporting what happened (ADR-0157 step 5, over the protocol of ADR-0007).
It exists so that a service task's work does not have to run on the engine's single writer. An in-process worker's outbound call is charged to the whole engine for its duration; a job this worker takes is charged to the worker.
**What it does not do, and why.** ADR-0157 says the standard worker is "the same binary running the *existing* handler code". That turned out not to be reachable for the Worker Types: every one of them resolves its configuration from the compiled process through a ProcessLookup over the local state store (`mail.Handler(store, lookup, registry)` and its siblings), which a process outside the engine has neither of. Relocating those needs the worker detail to travel with the job — its own slice. What this worker serves instead is the kind that has always been meant for an external worker: a **model-authored** job type, a `<zeebe:taskDefinition type>` whose work is the customer's own code.
The contract for that code is the one the script worker already established: the job arrives as JSON on stdin, and whatever JSON object the command writes to stdout becomes the variables the job completes with. A non-zero exit fails the job, carrying the command's stderr as the message an operator reads on the incident.
Index ¶
- Constants
- func KnownConnectorKinds() []string
- func ProbeSQL(ctx context.Context, p sqldb.Product, dsn string) error
- func RunADJob(ctx context.Context, j Job, dialer ad.Dialer, secret ad.SecretResolver, ...) (map[string]any, error)
- func RunAiTask(ctx context.Context, j Job, models map[string]agent.Model) (map[string]any, error)
- func RunClioJob(ctx context.Context, j Job, reg *clio.Registry) (map[string]any, error)
- func RunDiscordJob(ctx context.Context, j Job, reg *discord.Registry) (map[string]any, error)
- func RunEntraJob(ctx context.Context, j Job, reg *entra.Registry) (map[string]any, error)
- func RunGoogleSheetsJob(ctx context.Context, j Job, reg *googlesheets.Registry) (map[string]any, error)
- func RunJiraJob(ctx context.Context, j Job, reg *jira.Registry) (map[string]any, error)
- func RunLdapJob(ctx context.Context, j Job, dialer ldap.Dialer, secret ldap.SecretResolver) (map[string]any, error)
- func RunMailJob(ctx context.Context, j Job, reg *mail.Registry) (map[string]any, error)
- func RunRemedyJob(ctx context.Context, j Job, reg *remedy.Registry) (map[string]any, error)
- func RunSQLJob(ctx context.Context, j Job, reg *sqldb.Registry) (map[string]any, error)
- func RunSharePointJob(ctx context.Context, j Job, reg *sharepoint.Registry) (map[string]any, error)
- type CmdExec
- type CompletingExec
- type CompletingExecFunc
- type ConnectorPayload
- type Connectors
- type DecisionReport
- type Exec
- type ExecFunc
- type Job
- type Options
- type Outcome
- type ToolCallReport
- type Worker
Constants ¶
const ( DefaultLease = 5 * time.Minute DefaultWait = 30 * time.Second DefaultMaxJobs = 1 DefaultRetry = 5 * time.Second )
Defaults, chosen so an unconfigured worker is safe rather than surprising: it takes one job at a time, holds it for five minutes, and parks for thirty seconds per poll.
const ADMockViewURLEnv = "ATLAS_AD_MOCK_VIEW_URL"
ADMockViewURLEnv is the Atlas endpoint this worker posts its mock forest to. A supervised worker is handed it at spawn; an external one is given it by hand, or not at all — in which case it keeps its forest to itself, as before. Exported because the engine renders the same name, and a test there asserts the two agree.
const SQLMockViewURLEnv = "ATLAS_SQL_MOCK_VIEW_URL"
SQLMockViewURLEnv is the Atlas endpoint this worker posts its mock journal to. A supervised worker is handed it at spawn and only while the mockup is on; an external one is given it by hand, or not at all — in which case it keeps its journal to itself. Exported because the engine renders the same name, and a test there asserts the two agree.
const WorkerIDEnv = "ATLAS_WORKER_ID"
WorkerIDEnv is this worker's own id, the one the Workers view shows. `atlas worker` takes it from --id and puts it here: a report has to say whose directory it is, and the workers a worker builds are built from the environment alone.
Variables ¶
This section is empty.
Functions ¶
func KnownConnectorKinds ¶
func KnownConnectorKinds() []string
KnownConnectorKinds are the kinds BuiltinConnectors implements. A kind may serve several job types — script serves one per language — which is why a caller must not check this by counting handlers.
It is not only the error a misspelling produces. `atlas --supervise-connector` refuses a kind that is not here *at startup*, so a kind implemented below but missing from this list cannot be given a supervised worker at all: the server exits rather than starts. That is how the jira kind shipped able to run on a worker an operator launched by hand and unable to be supervised by the server — the list is hand-written and the case below was added without it. TestKnownConnectorKindsMatchesWhatIsImplemented holds the two together now, in both directions.
func ProbeSQL ¶ added in v0.5.0
ProbeSQL opens one database and checks that it answers. It is what the Console's worker check calls, handed to the server as api.WithSQLProbe by whoever assembles the binary.
It lives here and not in the api package for the reason the blank imports above do: this is where the drivers are, and the engine deliberately links none of them (ADR-0173). The pool it opens is closed again immediately — a check is one connection, not a registration.
func RunADJob ¶
func RunADJob(ctx context.Context, j Job, dialer ad.Dialer, secret ad.SecretResolver, dirs *ad.Registry) (map[string]any, error)
RunADJob performs a resolved AD job with the caller's own dialer, secret store and directories. It is exported for the same reason RunMailJob and the rest are: the environment is only the default place a worker's credentials come from, and a caller embedding this package can resolve references from a vault of its own and get the identical operation.
dirs holds the Console-configured directories a task addresses by name; it may be nil for a worker that serves only tasks carrying their own url.
func RunAiTask ¶ added in v0.5.0
RunAiTask works one ai task: one call to a language model, the answer into the variable the model named (ADR-0256).
It shares everything below the question with RunAgentRound — the same provider map, the same adapters, the same wire — because a one-shot call is a round offered no tools, and an adapter with nothing to call answers in words.
The one thing it does not share is where the answer goes. A round answers into whatever variable this worker is configured for; a task answers into the variable the *model* named, which is the point of the task. So the single output is renamed here, and a decision that came back as a tool call is refused: an ai task offered no tools, and a model that called one anyway has answered a question nobody asked.
func RunClioJob ¶ added in v0.5.0
RunClioJob performs a resolved clio task through a registry the caller owns. It is exported for the same reason RunRemedyJob and RunMailJob are: the environment is only the default place a worker's instances come from, and a caller embedding this package can build a registry from a vault or an instance profile and get the identical call.
It shares clio.Run with the in-process path, so no two of those can disagree about what a resolved clio task means — only about which endpoints are in reach.
func RunDiscordJob ¶ added in v0.5.0
RunDiscordJob performs a resolved Discord task through a registry the caller owns. It is exported for the same reason RunJiraJob and RunGoogleSheetsJob are: the environment is only the default place a worker's identities come from, and a caller embedding this package can build a registry from a vault or an instance profile and get the identical call.
It shares discord.Run with the in-process path, so no two of those can disagree about what a resolved Discord task means — only about which identities are in reach.
func RunEntraJob ¶
RunEntraJob performs a resolved Entra job through a registry the caller owns. It is exported for the same reason RunMailJob and RunSQLJob are: the environment is only the default place a worker's credentials come from, and a caller embedding this package can build a registry from a vault or a managed identity and get the identical call.
func RunGoogleSheetsJob ¶ added in v0.5.0
func RunGoogleSheetsJob(ctx context.Context, j Job, reg *googlesheets.Registry) (map[string]any, error)
RunGoogleSheetsJob performs a resolved Google Sheets task through a registry the caller owns. It is exported for the same reason RunJiraJob and RunSharePointJob are: the environment is only the default place a worker's identities come from, and a caller embedding this package can build a registry from a vault or an instance profile and get the identical call.
It shares googlesheets.Run with the in-process path, so no two of those can disagree about what a resolved spreadsheet task means — only about which identities are in reach.
func RunJiraJob ¶ added in v0.5.0
RunJiraJob performs a resolved Jira job through a registry the caller owns. It is exported for the same reason RunRemedyJob and RunMailJob are: the environment is only the default place a worker's credentials come from, and a caller embedding this package can build a registry from a vault or an instance profile and get the identical call.
It shares jira.Run with the in-process path, so no two of those can disagree about what a resolved Jira task means — only about which credentials are in reach.
func RunLdapJob ¶ added in v0.5.0
func RunLdapJob(ctx context.Context, j Job, dialer ldap.Dialer, secret ldap.SecretResolver) (map[string]any, error)
RunLdapJob performs a resolved LDAP job with the caller's own dialer and secret store. It is exported for the same reason RunADJob and the rest are: the environment is only the default place a worker's credentials come from, and a caller embedding this package can resolve references from a vault of its own and get the identical operation.
dialer is a ldap.Pool in a worker that leases many jobs — an LDAP bind is expensive enough that ADR-0154 pooled it, and a worker that dials per job would give that back the moment the work moved out of the engine.
func RunMailJob ¶
RunMailJob sends a resolved mail job through a registry the caller owns. It is exported because the environment is only the *default* place a worker's mail credentials come from: a caller embedding this package can build a registry from a vault, a file, or an instance profile and still get the identical send. It shares mail.Run with the in-process path, so no two of those can disagree about what a resolved mail task means — only about which credentials are in reach.
func RunRemedyJob ¶ added in v0.5.0
RunRemedyJob creates a resolved Remedy job's entry through a registry the caller owns. It is exported for the same reason RunMailJob and RunEntraJob are: the environment is only the default place a worker's credentials come from, and a caller embedding this package can build a registry from a vault or an instance profile and get the identical call.
It shares remedy.Run with the in-process path, so no two of those can disagree about what a resolved Remedy task means — only about which credentials are in reach.
func RunSQLJob ¶
RunSQLJob executes a resolved SQL job through a registry the caller owns. It is exported for the same reason RunMailJob is: the environment is only the default place a worker's connection strings come from, and a caller embedding this package can build a registry from a vault or an instance profile and get the identical execution.
func RunSharePointJob ¶ added in v0.5.0
RunSharePointJob performs a resolved SharePoint task through a registry the caller owns. It is exported for the same reason RunJiraJob and RunMailJob are: the environment is only the default place a worker's instances come from, and a caller embedding this package can build a registry from a vault or an instance profile and get the identical call.
It shares sharepoint.Run with the in-process path, so no two of those can disagree about what a resolved SharePoint task means — only about which instances are in reach.
Types ¶
type CmdExec ¶
type CmdExec struct {
Name string
Args []string
// Env is added to the process's own environment.
Env []string
}
CmdExec runs an operating-system command per job: the job arrives as JSON on stdin, and the JSON object the command writes to stdout becomes the variables the job completes with (no output completes with none). A non-zero exit fails the job and carries the command's stderr, which is what an operator reads on the incident.
type CompletingExec ¶ added in v0.5.0
CompletingExec is the widest exec shape: work that completes with more than variables. Exec is the decision-less special case, and the ladder is deliberately the one the job package already has (Handler / OutputHandler / CompletingHandler) so the two sides of the protocol are described the same way.
A handler implementing both is used as a CompletingExec.
type CompletingExecFunc ¶ added in v0.5.0
CompletingExecFunc adapts a function to CompletingExec.
func (CompletingExecFunc) Run ¶ added in v0.5.0
Run implements Exec by discarding the decision half, so a CompletingExecFunc can stand wherever an Exec is required.
func (CompletingExecFunc) RunCompleting ¶ added in v0.5.0
RunCompleting implements CompletingExec.
type ConnectorPayload ¶
ConnectorPayload is a resolved task: what to do, with no engine concepts in it.
type Connectors ¶
type Connectors struct {
Handlers map[string]Exec
Names []string
// Unconfigured are kinds this worker was asked to serve and holds no
// configuration for, so it does not subscribe to them. It is reported at startup
// rather than swallowed: "mail is not being served here" is the answer to why a
// mail task is waiting, and the worker's log is in the Workers console.
Unconfigured []string
}
Connectors is what a worker was configured to serve: the handlers, keyed by the job type each answers, and the worker *names* this worker holds credentials for.
The names are reported to the engine on every poll, because they are the half of "can this worker be served" that only the worker knows — once a kind is offloaded the engine holds no credential for it and cannot read another process's environment. The Workers view subtracts them from what deployed models reference to show the names configured nowhere (ADR-0168).
func BuiltinConnectors ¶
func BuiltinConnectors(env func(string) string, kinds ...string) (Connectors, error)
BuiltinConnectors returns handlers for the named Worker Types, keyed by the job type each serves. env looks up this worker's configuration; pass os.Getenv outside of tests.
An unknown kind name yields nothing, and the caller compares counts to catch it: a worker is configured from its own command line, so a name it does not implement is a mistake to report at startup rather than a queue to lease work from and then fail. A kind that *is* implemented but misconfigured returns an error here for the same reason — the operator is still watching at startup, and a per-job discovery would spend a retry budget learning what was knowable before the first poll.
type DecisionReport ¶ added in v0.5.0
type DecisionReport struct {
DecisionID string `json:"decisionId"`
Inputs map[string]any `json:"inputs,omitempty"`
Outputs map[string]any `json:"outputs,omitempty"`
Trace string `json:"trace,omitempty"`
}
DecisionReport is a worker's account of a decision it evaluated: which decision, the inputs it was handed, the outputs it got and the service's trace.
It is an account, not a record. The engine stamps which element instance it belongs to from the job's own lease, so a report cannot attach itself to a task other than the one it holds.
type Exec ¶
Exec does one job's work and returns the variables it completed with, or an error that fails the job.
type Job ¶
type Job struct {
JobKey uint64 `json:"jobKey"`
Type string `json:"type"`
ProcessInstanceKey uint64 `json:"processInstanceKey"`
ElementInstanceKey uint64 `json:"elementInstanceKey"`
ProcessDefKey uint64 `json:"processDefKey"`
ElementID string `json:"elementId"`
Retries int32 `json:"retries"`
LeaseToken uint64 `json:"leaseToken"`
Variables map[string]any `json:"variables"`
// Worker is a task the engine resolved into plain values, present
// only for a kind the server no longer serves itself (ADR-0168). Nil for a plain
// job-worker task, whose work is the customer's own command.
Connector *ConnectorPayload `json:"connector,omitempty"`
}
Job is one leased job as the engine hands it over.
type Options ¶
type Options struct {
// Server is the base URL of the Atlas to work for, e.g. http://localhost:8080.
Server string
// ID names this worker. It rides on every lease and every report, and it is what
// the Workers view shows — so give each deployment its own.
ID string
// Token, when set, is sent as a bearer credential on every request.
Token string
// Handlers is what this worker serves, keyed by job type. A worker with none is a
// misconfiguration rather than a process that polls forever over nothing.
Handlers map[string]Exec
// Lease is how long the engine holds a job for this worker. It must comfortably
// exceed how long the work takes: an elapsed lease puts the job back on offer, and
// the report that finally arrives is fenced out.
Lease time.Duration
// Wait is how long a poll may block waiting for work before coming back empty.
Wait time.Duration
// Retry is how long to wait after a failed poll before trying again. A worker
// must not exit because the server was restarting, or because the process using
// its job type is not deployed yet — both are ordinary and both resolve.
Retry time.Duration
// MaxJobs is how many jobs one poll may lease. Keep it to what this worker can
// actually run: leased work nobody is running is work nobody else can take either.
MaxJobs int
// Workers are the worker names this worker holds credentials for, reported
// to the engine on every poll. Only the worker knows them — once a kind is
// offloaded the engine holds no credential for it — and the Workers view
// subtracts them from what deployed models reference to show which names are
// configured nowhere (ADR-0168). Empty for a worker that serves only plain job
// types, which need no credential of their own.
Connectors []string
// HTTP is the client to use; nil means a default one bounded by Lease + Wait.
HTTP *http.Client
}
Options configures a Worker.
type Outcome ¶ added in v0.5.0
type Outcome struct {
Variables map[string]any
Decision *DecisionReport
// ToolCalls is a worker's account of what an agent chose to run next
// (ADR-0253/ADR-0254). It is the second thing in the sentence above: a round's
// answer is usually not variables at all, it is a choice of activities, and a
// worker deciding the round is the only one who knows what the model said.
//
// Empty is an answer, not an omission — the agent saying the run is finished —
// so a round that ends completes with variables like any other job.
ToolCalls []ToolCallReport
}
Outcome is everything a job's work produced: the variables it completes with, and — for a central business rule task — the decision evaluation to retain.
It mirrors [job.Completion] on the in-process side, and for the same reason: a decision's record is durable (ADR-0066), so once the evaluation happens on a worker the worker has to be able to report it. Every other kind completes with variables alone, which is why Exec stays the ordinary shape.
func RunAgentRound ¶ added in v0.5.0
RunAgentRound decides one round of an agent-driven ad-hoc subprocess and reports the choice back as an Outcome (ADR-0253/ADR-0254).
It is the second runner that completes with more than variables, and for a reason of the same shape as the first: a round's answer is usually a *choice of activities*, which variables cannot express. What it reports is an account, not a decision the engine takes on trust — the container is read from the lease, and every tool name is resolved against that container's compiled index before anything is activated. A worker cannot widen an agent's reach.
It is exported for the same reason RunMailJob is: the environment is only the default place a worker's models come from, and a caller embedding this package may build them its own way.
func RunTemisJob ¶ added in v0.5.0
RunTemisJob evaluates a resolved central decision through a registry the caller owns, and reports it as an Outcome — the variables the task writes back *and* the evaluation to retain.
It is the only runner that returns a decision, because a business rule task is the only job whose completion carries one (ADR-0066). The inputs are echoed back in the report rather than recomputed: they are what this evaluation was actually asked, and the engine cannot rebuild them at completion time because the instance has moved on since the lease.
type ToolCallReport ¶ added in v0.5.0
type ToolCallReport struct {
Tool string `json:"tool"`
CallId string `json:"callId,omitempty"`
Arguments map[string]any `json:"arguments,omitempty"`
}
ToolCallReport is a worker's account of one tool an agent chose for the next round: which of the container's tools to run, the model's own id for the call, and the arguments it supplied for that tool's declared parameters.
Like a DecisionReport it is an account, not a record — and the difference matters more here, because a tool call does not describe work that happened, it *causes* work to happen. So the engine reads which container this is from the leased job, and resolves Tool against that container's compiled tool index before anything is activated. A worker cannot name an activity the model does not carry and have it run; naming one it does not carry raises the incident ADR-0253 already defines.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker leases jobs of the types it handles and reports what happened. Build one with New.
Source Files