Documentation
¶
Overview ¶
Package worker is the BYOC (bring-your-own-compute) consumer of the work queue: the customer-hosted twin of internal/executor. Where the executor runs inside the platform with direct database access, a worker runs in the customer's own network and reaches the control plane only over the wire — authenticating with its environment key, reading a session's suspended tool calls through the session events API, running the built-in toolset in a local sandbox, and posting the results back as user.tool_result events. Platform executor and BYOC worker are the same pull protocol at two deployment points; this is the self_hosted one.
The whole loop lives here. lease.go polls the work queue, claims an item and heartbeats it (NewWorker and Run, which cmd/worker wires to configuration); toolexec.go is the driver that runs a session's outstanding tools once; files.go and skills.go materialize a session's files and skills into the sandbox before the tools that read them run. client.go builds the SDK client, and only that client ever talks to the platform — a worker holds no database handle, which is the property that lets it run on compute the platform cannot reach.
Two boundaries are worth knowing before changing anything here. The wire is the whole contract, and it is narrow: Poll serves this worker exactly one thing — a tool_exec item on its own self_hosted environment. Never a model_turn row, never web_exec, outputs_harvest or mcp_exec, and never another environment's work. Those kinds are claimed in-process by the brain and the platform executor, which hold the database handle this package deliberately does not (internal/queue's Kind constants record why each falls where it does). And a run that stops reporting progress for StallTimeout is cancelled, after which the worker deliberately stops beating rather than releasing the lease — the lease lapses server-side and the control plane re-offers the item, which is the one path that survives a worker wedged inside a call that ignores cancellation.
Index ¶
- Constants
- func NewClient(baseURL, envKey string) sdk.Client
- func RunSessionTools(ctx context.Context, client sdk.Client, provider sandbox.Provider, ...) error
- func SetupFiles(ctx context.Context, client sdk.Client, sessionID string, sb sandbox.Sandbox, ...) error
- func SetupSkills(ctx context.Context, client sdk.Client, sessionID string, sb sandbox.Sandbox, ...) error
- type Config
- type ToolExecConfig
- type Worker
Constants ¶
const ( // MetricSkillsMaterialized counts per-skill materialization outcomes. MetricSkillsMaterialized = "skills.materialized" // MetricSkillsMaterializeDuration is one whole materialization pass. MetricSkillsMaterializeDuration = "skills.materialize.duration" // MetricFilesMaterialized counts per-file mount materialization outcomes — // the executor twin's name on the worker meter. MetricFilesMaterialized = "files.materialized" // MetricFilesMaterializeDuration is one whole file-materialization pass. MetricFilesMaterializeDuration = "files.materialize.duration" )
Variables ¶
This section is empty.
Functions ¶
func NewClient ¶
NewClient builds the SDK client a worker uses to reach the control plane's session API. The worker authenticates with its environment key as a Bearer token — the wire's worker credential, scoped to one environment's work queue and distinct from the management x-api-key. The control plane routes a session-events request to its environment-key lane only when a Bearer is present and no x-api-key is; WithoutEnvironmentDefaults guarantees the latter by keeping the SDK from autoloading an ambient ANTHROPIC_API_KEY (which it would otherwise send as x-api-key) underneath the explicit options.
baseURL points at the control plane (e.g. an on-prem deployment's URL), never a hard-coded api.anthropic.com — a worker talks to the platform it belongs to.
func RunSessionTools ¶
func RunSessionTools(ctx context.Context, client sdk.Client, provider sandbox.Provider, sessionID string, cfg ToolExecConfig) error
RunSessionTools is the BYOC worker's tool-exec driver: given a session whose turn has suspended for built-in tool calls, it runs every unanswered tool in the session's sandbox and posts a user.tool_result for each back through the session events API. It is the self_hosted twin of the platform executor's per-item processing, with two deployment differences: the transport is HTTP (the worker has no database), and the result event is user.tool_result, not agent.tool_result — the control plane resumes the brain when a result completes the outstanding set, so the worker never enqueues a turn itself.
Results are posted per tool as each completes, so a backend fault partway through leaves the tools that did run answered on the log; a reclaiming pass re-derives only the still-unanswered ones. This matches the executor's partial-commit-on-fault: a tool-level failure (missing file, nonzero exit) still yields a result the model must see, and only a backend fault (sandbox gone) stops the set with the rest left for the reclaim.
The sandbox is provisioned only when there is unanswered work, so a call against an already-answered session (a redundant reclaim) is one bounded read with nothing to run.
Session liveness is the caller's gate, not this driver's. The platform executor refuses to run a stale session's tools by loading its status under the session row lock (executor.sessionForRun) before provisioning — but it does so in its per-item orchestration, not in its runTools core, which this driver is the analog of. The BYOC caller (the lease loop, PR C2b) owns the same session load: it must read the session (for the same reason it must load the egress policy this cfg.Networking carries) and skip a session that is not running or is archived, mirroring sessionForRun. The control plane is only a partial backstop here — a post to an archived session is refused (400), but a post to a merely not-running one appends without resuming — so the complete gate belongs in the caller, not in a reliance on the append being rejected.
func SetupFiles ¶ added in v0.2.0
func SetupFiles(ctx context.Context, client sdk.Client, sessionID string, sb sandbox.Sandbox, workdir string, progress func()) error
SetupFiles is the BYOC-worker twin of the executor's materializeFiles: it reads the session's file mounts over the wire and streams each file's bytes from GET /v1/files/{id}/content (the environment-key content lane) into the sandbox at its mount_path, recording the same sentinel and metrics. Wire-only — no database, no object store: the control plane's environment-scoped lane is the authority on which files this environment may read, so a file no session in the environment mounts answers 404 and is tolerated as a not_found miss. Only the session read is fatal; a per-file failure is logged, counted, and skipped, never failing the run.
func SetupSkills ¶ added in v0.2.0
func SetupSkills(ctx context.Context, client sdk.Client, sessionID string, sb sandbox.Sandbox, workdir string, progress func()) error
SetupSkills materializes the session agent's skills into the sandbox — the BYOC twin of the executor's materialization and a re-expression of the reference worker's SetupSkills (anthropic-sdk-go tools/agenttoolset): session GET with the environment key, per skill an alias resolution over the versions list (newest numeric wins), a version GET for the name, the /content download, and extraction under the reference guards — all wire, no database, writing through the sandbox file API instead of the host filesystem. Per-skill failure is logged and skipped, never fatal; only the session read fails the call, mirroring the reference. A sentinel under {workdir}/skills/ records the resolved set so a reclaiming pass over a live sandbox skips rewriting unchanged skills (the reference re-extracts every time, but its workdir is host-shared across sessions and cleaned per item; this sandbox is per-session, so skipping is safe and cheaper).
Types ¶
type Config ¶
type Config struct {
EnvironmentID string
// WorkerID identifies this worker for the control plane's poll metrics
// (Anthropic-Worker-ID). Auto-generated as "<hostname>-<random>" when empty,
// as the reference does.
WorkerID string
Image string
Workdir string
Networking domain.Networking
// Hardening caps every sandbox this worker provisions (#65) — the BYOC twin
// of the platform executor's. cmd/worker reads the same SANDBOX_* variables
// the executor does; the zero value hardens nothing, so a test builds a
// Config without acquiring a limit it did not ask for.
Hardening sandbox.Hardening
// EmptyPollSleep is the wait between empty polls (default 1s), on top of
// the server-side block_ms hold — kept because the reference client
// sleeps between empty polls the same way with block_ms set, so an idle
// worker's cadence stays wire-identical: one poll per block + sleep.
EmptyPollSleep time.Duration
// HeartbeatInterval, when > 0, fixes the heartbeat cadence; otherwise it is
// derived from each heartbeat response's ttl (ttl/2, clamped to
// [heartbeatFloor, heartbeatCap]) as the reference does. Tests set a small
// value; production leaves it 0.
HeartbeatInterval time.Duration
// StallTimeout bounds how long a run may report no progress before the
// heartbeat gives up on it — cancelling the run and, deliberately, beating
// no more, so the lease lapses server-side and the control plane re-offers
// the item (#383). It is the BYOC twin of the platform executor's knob, for
// the same wedge: a sandbox call that never returns leaves this worker
// heartbeating a run that will never finish, so the item is neither
// progressing nor reclaimable. Progress is a step finishing, not a byte
// moving, so the budget must clear the longest single step a healthy run
// takes: toolset.MaxTimeout for one `bash`, a cold image pull, a large
// mount. No two steps share one interval: the liveness read is reported at
// the top of RunSessionTools, ahead of the paging unanswered-use scan, so
// two bounded control-plane calls do not add up inside one silence.
// It bounds *silence*, never duration — a run that keeps finishing
// steps runs as long as it likes. Not an off switch: 0 takes the default
// (WORKER_STALL_TIMEOUT).
StallTimeout time.Duration
}
Config configures the BYOC worker lease loop. The worker owns its sandbox shape (Image/Workdir/Networking) rather than loading a per-session egress policy — a self_hosted worker runs on the customer's own compute and the wire exposes no per-session networking to it, so this mirrors the platform executor's Config, whose sandbox settings are likewise a deployment choice.
type ToolExecConfig ¶
type ToolExecConfig struct {
Image string
Workdir string
Networking domain.Networking
// Hardening is the containment the worker's sandboxes are created with, the
// BYOC twin of the platform executor's (#65). cmd/worker resolves it from
// the same environment variables the executor reads, so a customer-hosted
// sandbox is capped the same way a platform-managed one is.
Hardening sandbox.Hardening
// Progress, when set, is called each time the run finishes a step — the
// sandbox provisioned, the skills or files materialized, a tool answered.
// The lease loop watches it to tell a long run from a wedged one (#383);
// a caller that runs the driver directly leaves it nil.
Progress func()
}
ToolExecConfig is the sandbox shape a worker provisions for a session's tools. A self_hosted environment's wire config carries no image (the sandbox image is a deployment choice, not part of the domain model), so Image and Workdir come from the worker's own configuration — mirroring the platform executor's Config. Networking is the session's egress policy, read from the session's environment and threaded in by the caller.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker is the BYOC lease loop, the self_hosted twin of the platform executor. It polls the control plane's self_hosted work queue over HTTP, acknowledges an item, keeps its lease alive with heartbeats while the C2a tool-exec driver runs the session's tools in a local sandbox, and force-stops the item when the run ends. One session at a time, mirroring the reference `ant beta:worker`.
func NewWorker ¶
NewWorker builds a worker over an SDK client (see NewClient) and a local sandbox provider (the customer's Docker/K8s).
func (*Worker) Run ¶
Run polls until ctx is cancelled, handling one work item at a time. A poll that fails with an auth error (a bad environment key) is fatal and returns the error; any other poll error backs off and retries, so a transient network blip does not kill the worker. Cancellation (SIGINT/SIGTERM via the caller's signal context) ends the loop with a nil error.
Source Files