worker

package
v0.25.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 9, 2026 License: AGPL-3.0 Imports: 61 Imported by: 0

Documentation

Overview

Package worker is Atlas's out-of-process job worker: the built-in Worker Types, served by a process that runs as a client of atlas's 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.

The protocol client itself — leasing, completing, failing, throwing, answering queries, asking before acting — lives in package workersdk, which imports nothing of the engine so that a Worker Type built outside this repository can depend on it (ADR-0535). This package holds what is specific to the Worker Types atlas ships (BuiltinConnectors and the handlers behind it) and keeps the client's names as aliases, so code written against worker.New, worker.Options or worker.BPMNError is unchanged.

Index

Constants

View Source 
const (
	DefaultLease      = workersdk.DefaultLease
	DefaultWait       = workersdk.DefaultWait
	DefaultMaxJobs    = workersdk.DefaultMaxJobs
	DefaultRetry      = workersdk.DefaultRetry
	DefaultMaxQueries = workersdk.DefaultMaxQueries
)

The protocol client's defaults.

View Source 
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.

View Source 
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.

View Source 
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 

func ProbeSQL(ctx context.Context, p sqldb.Product, dsn string) error

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 Refused added in v0.20.0 

func Refused() error

Refused is the error a handler returns for a refused action. See workersdk.Refused.

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 

func RunAiTask(ctx context.Context, j Job, models map[string]agent.Model) (map[string]any, error)

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 RunCardDAVJob added in v0.20.0 

func RunCardDAVJob(ctx context.Context, j Job, reg *carddav.Registry) (map[string]any, error)

RunCardDAVJob performs a resolved CardDAV task through a registry the caller owns. It is exported for the reason RunGoogleContactsJob is: a caller embedding this package can build a registry from its own secret store and get the identical call.

func RunClioJob added in v0.5.0 

func RunClioJob(ctx context.Context, j Job, reg *clio.Registry) (map[string]any, error)

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 

func RunDiscordJob(ctx context.Context, j Job, reg *discord.Registry) (map[string]any, error)

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 

func RunEntraJob(ctx context.Context, j Job, reg *entra.Registry) (map[string]any, error)

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 RunExchangeJob added in v0.15.0 

func RunExchangeJob(ctx context.Context, j Job, reg *exchange.Registry) (map[string]any, error)

RunExchangeJob performs a resolved Exchange task through a registry the caller owns. It is exported for the reason RunEntraJob is: 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 RunGoogleContactsJob added in v0.20.0 

func RunGoogleContactsJob(ctx context.Context, j Job, reg *googlecontacts.Registry) (map[string]any, error)

RunGoogleContactsJob performs a resolved Google Contacts task through a registry the caller owns. It is exported for the reason RunExchangeJob is: the environment is only the default place a worker's credentials come from, and a caller embedding this package can build a registry from its own secret store 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 

func RunJiraJob(ctx context.Context, j Job, reg *jira.Registry) (map[string]any, error)

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 

func RunMailJob(ctx context.Context, j Job, reg *mail.Registry) (map[string]any, error)

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 RunMailJobWith added in v0.20.0 

func RunMailJobWith(ctx context.Context, j Job, reg *mail.Registry, docs mail.DocumentStore) (map[string]any, error)

RunMailJobWith is RunMailJob with the object stores an export-attachment may put a document into; nil holds none.

func RunPaperlessJob added in v0.20.0 

func RunPaperlessJob(ctx context.Context, j Job, clients map[string]*paperless.Client) (map[string]any, error)

RunPaperlessJob runs one plain BPMN service-task job whose variable "paperless" is a metadata-only request (worker, operation and operation-specific attributes). The worker returns only the chosen result variable; absence of a result variable means that no API response is persisted into Atlas's variable events.

func RunRemedyJob added in v0.5.0 

func RunRemedyJob(ctx context.Context, j Job, reg *remedy.Registry) (map[string]any, error)

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 RunS3Job added in v0.9.0 

func RunS3Job(ctx context.Context, j Job, reg *s3.Registry) (map[string]any, error)

RunS3Job performs a resolved object-store 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 s3.Run with the in-process path, so no two of those can disagree about what a resolved object-store task means — only about which identities are in reach. On an offloaded installation this is also where a put's bytes are: they go from here to the store, and the engine only ever held the variable they were composed from.

func RunSQLJob 

func RunSQLJob(ctx context.Context, j Job, reg *sqldb.Registry) (map[string]any, error)

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 

func RunSharePointJob(ctx context.Context, j Job, reg *sharepoint.Registry) (map[string]any, error)

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.

func RunSharePointServerJob added in v0.25.0 

func RunSharePointServerJob(ctx context.Context, j Job, reg *sharepointserver.Registry) (map[string]any, error)

RunSharePointServerJob performs a resolved task through a registry the caller owns.

func RunTeamsPhoneJob added in v0.15.0 

func RunTeamsPhoneJob(ctx context.Context, j Job, reg *teamsphone.Registry) (map[string]any, error)

RunTeamsPhoneJob performs a resolved Teams Phone job through a registry the caller owns. It is exported for RunEntraJob's reason: the environment is only the default place a worker's tenants come from.

func RunWikiJob added in v0.21.0 

func RunWikiJob(ctx context.Context, j Job, reg *wikiwikibangbang.Registry) (map[string]any, error)

RunWikiJob performs a resolved wikiwikibangbang task through a registry the caller owns. It is exported for the reason RunCardDAVJob is: a caller embedding this package can build a registry from its own secret store and get the identical call.

Types

type Authorization added in v0.20.0 

type Authorization = workersdk.Authorization

Authorization is the server's answer to a handler asking before it acts (ADR-0475).

func Authorize added in v0.20.0 

func Authorize(ctx context.Context) (*Authorization, error)

Authorize asks the server whether the job in ctx may act. See workersdk.Authorize.

type BPMNError added in v0.10.0 

type BPMNError = workersdk.BPMNError

BPMNError ends a job in a BPMN error instead of a result (ADR-0089).

type CmdExec 

type CmdExec = workersdk.CmdExec

CmdExec runs an operating-system command per job.

type CompletingExec added in v0.5.0 

type CompletingExec = workersdk.CompletingExec

CompletingExec is work that completes with more than variables.

type CompletingExecFunc added in v0.5.0 

type CompletingExecFunc = workersdk.CompletingExecFunc

CompletingExecFunc adapts a function to CompletingExec.

type ConnectorPayload 

type ConnectorPayload = workersdk.ConnectorPayload

ConnectorPayload is a resolved task: what to do, with no engine concepts in it.

type Connectors 

type Connectors struct {
	Handlers map[string]Exec
	// Queries are the form lookups the configured kinds answer, keyed by job type
	// (ADR-0473). A kind that answers them serves jobs too, and
	// answers them for the same Workers.
	Queries map[string]QueryFunc
	Names   []string
	// Endpoints are the endpoints the held names' configurations name, reduced to
	// scheme, host and port, keyed by worker name. A worker reports them beside the
	// names on every poll (ADR-0471).
	Endpoints map[string]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 = workersdk.DecisionReport

DecisionReport is a worker's account of a decision it evaluated.

type Exec 

type Exec = workersdk.Exec

Exec does one job's work.

type ExecFunc 

type ExecFunc = workersdk.ExecFunc

ExecFunc adapts a function to Exec.

type Job 

type Job = workersdk.Job

Job is one leased job as the engine hands it over.

type Options 

type Options = workersdk.Options

Options configures a Worker.

type Outcome added in v0.5.0 

type Outcome = workersdk.Outcome

Outcome is everything a job's work produced.

func RunAgentRound added in v0.5.0 

func RunAgentRound(ctx context.Context, j Job, models map[string]agent.Model) (Outcome, error)

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 

func RunTemisJob(ctx context.Context, j Job, reg *temis.Registry) (Outcome, error)

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 QueryFunc added in v0.20.0 

type QueryFunc = workersdk.QueryFunc

QueryFunc answers one form lookup (ADR-0473).

type ToolCallReport added in v0.5.0 

type ToolCallReport = workersdk.ToolCallReport

ToolCallReport is a worker's account of one tool an agent chose.

type Worker 

type Worker = workersdk.Worker

Worker leases jobs of the types it handles and reports what happened.

func New 

func New(opts Options) *Worker

New builds a worker, filling in defaults. See workersdk.New.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.