queue

Documentation

Overview

Package queue is the internal work queue over Postgres (FOR UPDATE SKIP LOCKED, per the plan's component 4). Five kinds share the work_items table — model_turn, tool_exec, web_exec, outputs_harvest and mcp_exec — and there are two ways to take one, which is the distinction to hold on to.

Claim is the in-process lane, for a caller holding a database handle: the brain claims model_turn, and the platform executor claims the rest. Poll is the wire lane behind the work API, and it serves exactly one thing — a tool_exec item on a self_hosted environment. A BYOC worker therefore never sees a model_turn row, nor any other kind, nor another environment's work.

That single split is what makes the platform executor and the BYOC worker the same pull protocol at two deployment points: one tool_exec item goes in-process when its environment is cloud and out to the customer's worker when it is self_hosted, and Claim's own predicate says so (kind <> 'tool_exec' OR e.kind = 'cloud'). Each Kind below records why it falls where it does.

Enqueue is idempotent per (session, kind) while a live item exists, so event-append triggers can fire without double-scheduling; a claim leases the item and an expired lease makes it claimable again.

Index

• Constants
• Variables
• type DB
• type HeartbeatResult
• type Item
• type Kind
• type LeaseKeeper
  • func (k *LeaseKeeper) Close() error
  • func (k *LeaseKeeper) Progress()
• type Queue
  • func New(pool *pgxpool.Pool) *Queue
  • func (q *Queue) Ack(ctx context.Context, envID, workID domain.ID) (*Work, error)
  • func (q *Queue) Assert(ctx context.Context, db DB, item *Item) error
  • func (q *Queue) CancelSession(ctx context.Context, db DB, sessionID domain.ID) error
  • func (q *Queue) Claim(ctx context.Context, kind Kind, ttl time.Duration) (*Item, error)
  • func (q *Queue) Complete(ctx context.Context, db DB, item *Item) error
  • func (q *Queue) Enqueue(ctx context.Context, db DB, envID, sessionID domain.ID, kind Kind) (bool, error)
  • func (q *Queue) Extend(ctx context.Context, item *Item, ttl time.Duration) error
  • func (q *Queue) GetWork(ctx context.Context, envID, workID domain.ID) (*Work, error)
  • func (q *Queue) Heartbeat(ctx context.Context, envID, workID domain.ID, expected string, ...) (*HeartbeatResult, error)
  • func (q *Queue) KeepLease(ctx context.Context, item *Item, ttl, stall time.Duration) (context.Context, *LeaseKeeper)
  • func (q *Queue) ListWork(ctx context.Context, envID domain.ID, after bool, afterT time.Time, ...) ([]*Work, error)
  • func (q *Queue) Poll(ctx context.Context, envID domain.ID, reclaim time.Duration) (*Work, error)
  • func (q *Queue) RecordPoll(ctx context.Context, envID domain.ID, workerID string) error
  • func (q *Queue) RegisterMetrics() (metric.Registration, error)
  • func (q *Queue) Requeue(ctx context.Context, db DB, item *Item) error
  • func (q *Queue) Stats(ctx context.Context, envID domain.ID) (*WorkStats, error)
  • func (q *Queue) Stop(ctx context.Context, envID, workID domain.ID, force bool) (*Work, error)
  • func (q *Queue) UpdateMetadata(ctx context.Context, envID, workID domain.ID, upserts map[string]string, ...) (*Work, error)
• type Work
• type WorkStats

Constants

const (
	MetricQueueDepth          = "queue.depth"
	MetricQueuePending        = "queue.pending"
	MetricQueueWorkersPolling = "queue.workers_polling"
)

The work-queue depth/pending/workers_polling gauges — the OTLP metric mirror of the /work/stats endpoint's BetaSelfHostedWorkQueueStats shape, reported per self_hosted environment. They are asynchronous: an operator reads a queue's backlog at collection time, not a running total, so the SDK samples them through the callback RegisterMetrics installs. The names are exported so the telemetry contract test can assert they reach an OTLP collector.

const NoHeartbeat = "NO_HEARTBEAT"

NoHeartbeat is the sentinel a worker's first heartbeat sends as expected_last_heartbeat to claim an unclaimed lease (the wire's optimistic concurrency: subsequent heartbeats echo the server's prior value).

Variables

var (
	ErrWorkNotFound      = errors.New("queue: work item not found")
	ErrWorkConflict      = errors.New("queue: work item is in a conflicting state")
	ErrHeartbeatMismatch = errors.New("queue: heartbeat precondition failed")
)

The wire work API's state-machine outcomes, mapped by the API layer onto HTTP statuses: not-found → 404, conflict → 409, heartbeat mismatch → 412.

var ErrLeaseLost = errors.New("queue: work item lease lost")

ErrLeaseLost reports that the item is no longer this claimant's: its lease expired and another claim took over, or it already finished.

var ErrWorkStalled = errors.New("queue: work item stalled")

ErrWorkStalled reports that the holder stopped making progress for longer than its keeper's stall budget, so the keeper cancelled the work and stopped renewing. It is a distinct outcome from ErrLeaseLost on purpose: the lease was never taken from this claimant, it was given up (#383).

Types

type DB

type DB interface {
	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}

DB is the slice of pgx shared by pools and transactions, so Enqueue can join the caller's transaction (event append + status flip + enqueue must commit atomically).

type HeartbeatResult

type HeartbeatResult struct {
	LastHeartbeat time.Time
	State         string
	LeaseExtended bool
	TTLSeconds    int64
}

HeartbeatResult is the wire heartbeat response projection.

type Item

type Item struct {
	ID            domain.ID
	EnvironmentID domain.ID
	SessionID     domain.ID
	Kind          Kind
	// Lease is the claim's expiry as recorded by the database. It is the
	// claimant's proof of ownership: Extend and Complete match it against
	// the row (the same optimistic-concurrency shape as the reference work
	// API's expected_last_heartbeat), so a claimant that lost its lease to
	// a reclaim gets ErrLeaseLost instead of silently finishing someone
	// else's item.
	Lease time.Time
	// Reclaimed marks an item whose previous claimant let the lease expire —
	// the session was mid-turn when its brain died, so the new claimant
	// should surface recovery (session.status_rescheduled) before replaying.
	Reclaimed bool
	// TraceContext is the W3C trace context captured at enqueue, so the
	// claimant can parent its work on the turn that produced it — the same
	// context a BYOC worker gets from its poll response (see Work). Nil when
	// the item was enqueued with no active span.
	TraceContext map[string]string
}

Item is one claimed unit of work.

type Kind

type Kind string

Kind discriminates the work streams.

const (
	// ModelTurn drives the brain: one turn of the orchestration loop, claimed
	// in-process by whichever brain replica gets there first. It is never
	// offered on the wire — a turn calls the model with the deployment's
	// credentials and appends to the event log, neither of which a customer's
	// worker has any way to do.
	ModelTurn Kind = "model_turn"
	// ToolExec runs a session's suspended tool calls in a sandbox, and is the
	// one kind that reaches both deployment points: Claim serves it in-process
	// when the environment is cloud, Poll serves it over the wire when the
	// environment is self_hosted. Nothing else crosses that line.
	ToolExec Kind = "tool_exec"
	// WebExec is the web_fetch/web_search work (docs/plan/15_web-tools.md):
	// run by the platform executor's web driver in its own process — no
	// sandbox — for cloud AND self_hosted sessions alike. Poll never serves
	// it; the official worker implements only the six sandbox tools.
	WebExec Kind = "web_exec"
	// OutputsHarvest is the deliverables harvest (docs/plan/21_outcomes.md,
	// Decision 8): the cloud executor snapshots /mnt/session/outputs/ into
	// the files registry before a grading pass. Internal-only — the brain
	// enqueues it for cloud environments alone, and Poll never serves it (a
	// self_hosted sandbox is unreachable from the platform; the reference
	// worker has no file lane).
	OutputsHarvest Kind = "outputs_harvest"
	// MCPExec is the MCP work (docs/plan/29_mcp-toolset.md): the executor's
	// MCP driver discovers a session's MCP servers' tools into mcp_catalogs,
	// and answers the session's MCP tool calls. One item does one of the two —
	// a pass with a call outstanding answers calls and does not discover, since
	// the turn is stopped on that call while a listing is only ever wanted by a
	// turn that has not started. Like WebExec it runs in
	// the platform executor's process for cloud AND self_hosted sessions
	// alike — the SDK states three times that MCP tools are server-side, the
	// work API has no MCP surface, and the BYOC worker's contract is
	// agent.tool_use + agent.custom_tool_use only — so Poll never serves it.
	MCPExec Kind = "mcp_exec"
)

type LeaseKeeper

type LeaseKeeper struct {
	// contains filtered or unexported fields
}

LeaseKeeper renews a claimed item's lease on a timer while its holder works, so a long tool run or model stream cannot let the lease lapse and hand the item to a second claimant mid-work. Each renewal is bounded by the lease it is racing, so a stalled database cannot hang the holder behind an unreturnable Extend, and losing the lease cancels the work context the holder runs under.

One keeper serves both consumers — the brain's turn loop (a model can think far longer than any inter-chunk gap, e.g. a long time-to-first-token on a big replayed context) and the executor's item processing (a slow image pull or a long-running tool). The timing semantics below are subtle enough that they must exist once, not once per consumer.

func (*LeaseKeeper) Close

func (k *LeaseKeeper) Close() error

Close stops the keeper and reports the first extension failure. The goroutine has exited when Close returns, so the item's lease value is stable again for the settling append to use as its ownership proof.

func (*LeaseKeeper) Progress

func (k *LeaseKeeper) Progress()

Progress reports that the work has moved: another tool answered, another mount landed. It is what keeps a long but healthy item alive, so it belongs at the steps a wedge would stop — never on a timer, which would report progress a wedged holder is not making. Safe from any goroutine. A keeper started with no stall budget ignores what it records, but the store still happens, so a caller that reports on a hot path pays for it whether or not anything reads it.

type Queue

type Queue struct {
	// contains filtered or unexported fields
}

Queue hands out work over one Postgres pool.

func New

func New(pool *pgxpool.Pool) *Queue

func (*Queue) Ack

func (q *Queue) Ack(ctx context.Context, envID, workID domain.ID) (*Work, error)

Ack acknowledges a polled work item, transitioning queued → starting. It is idempotent: only the queued→starting edge stamps acknowledged_at and installs the startup lease, so a re-ack of an already-advanced item returns it unchanged. The startup lease (ackStartupLeaseSeconds) governs a starting item until its first heartbeat replaces it, so Poll reclaims a dead worker's starting item on a real lease, not the short un-acked poll reservation. An item not visible to the work API (missing, wrong environment, or not a self_hosted tool_exec item) is ErrWorkNotFound.

func (*Queue) Assert

func (q *Queue) Assert(ctx context.Context, db DB, item *Item) error

Assert verifies the claimant still owns the item, inside the caller's transaction. Session state written mid-turn (the reclaim recovery announcement) must carry this proof like every other state write: a claimant that stalled past its lease could otherwise flip a session another brain has since settled.

It locks the work row rather than merely reading it, because an unlocked read is only a proof at the instant it runs: a reclaim committing between the read and the caller's commit would leave two holders writing the same session. The lock lasts as long as the caller's transaction, which is the only place this belongs — handed a pool it is a bare read again, the row lock ending with the implicit transaction the statement ran in. The lock closes that window from both sides — Claim takes its rows FOR UPDATE SKIP LOCKED, so it skips a row being asserted rather than blocking on it, and a reclaim that got there first has already changed lease_expires_at, which this read then fails on. The window is widest where nothing renews the lease while the caller settles, which is exactly the stalled executor's partial commit (#383).

func (*Queue) CancelSession

func (q *Queue) CancelSession(ctx context.Context, db DB, sessionID domain.ID) error

CancelSession stops every work item this session still has in flight, inside the caller's transaction — what a user.interrupt does to the turn it ends. Unlike Complete it carries no lease proof, because it is not a claimant finishing its own work: it takes the work away from whoever holds it. That is the point. A claimant only ever writes under a lease it re-asserts in the same transaction (Complete, Requeue, Assert), so a brain or executor still running the interrupted turn finds its item gone and rolls its whole settlement back, which is what keeps a stale half-turn — duplicate tool results, tool intents nothing may answer — off the append-only log. Cancelling under the session row lock the interrupt already holds is what makes that a decision and not a race.

It also frees the (session, kind) slot Enqueue is idempotent on, so a user.message in the same batch can schedule the redirect turn immediately.

func (*Queue) Claim

func (q *Queue) Claim(ctx context.Context, kind Kind, ttl time.Duration) (*Item, error)

Claim leases the oldest available item of the kind: queued items first-come first-served, plus active items whose lease expired (their claimant died). It returns nil with no error when there is nothing to do.

tool_exec claims are scoped to cloud environments — the platform-managed executor is the cloud hands. A self_hosted environment's tool_exec work is served only by Poll (a BYOC worker), never Claim, so an item a worker has polled can never also be run by the executor. Every other kind is claimed for every environment: the brain (model calls) and the executor's web driver (web_fetch/web_search) run on the platform regardless of where a session's sandbox lives.

func (*Queue) Complete

func (q *Queue) Complete(ctx context.Context, db DB, item *Item) error

Complete marks the item finished, in the caller's transaction when one is passed (a turn's settlement completes its item atomically with the state it writes, so a concurrent trigger serialized behind the same session lock always sees either a live item or a completed one — never a gap). Losing the lease first (another claimant took over after expiry) is an error: the caller's work may have raced the replacement's and must not be treated as cleanly finished.

func (*Queue) Enqueue

func (q *Queue) Enqueue(ctx context.Context, db DB, envID, sessionID domain.ID, kind Kind) (bool, error)

Enqueue inserts a queued item unless a live (queued/starting/active) item for the same session and kind exists. It reports whether a new item was created; false means an existing live item already covers the work.

func (*Queue) Extend

func (q *Queue) Extend(ctx context.Context, item *Item, ttl time.Duration) error

Extend renews the claimant's lease mid-work (long provider streams) and returns the new lease proof.

func (*Queue) GetWork

func (q *Queue) GetWork(ctx context.Context, envID, workID domain.ID) (*Work, error)

GetWork returns one work item visible to the work API (see workAPIScope), or ErrWorkNotFound.

func (*Queue) Heartbeat

func (q *Queue) Heartbeat(ctx context.Context, envID, workID domain.ID, expected string, ttlSeconds int64) (*HeartbeatResult, error)

Heartbeat applies the wire's optimistic-concurrency heartbeat. The first heartbeat (expected == NoHeartbeat) claims the lease of a just-acked (starting) item and moves it to active; subsequent heartbeats echo the server's prior last_heartbeat and extend the lease while the item is active. A heartbeat on an active item the control plane has since moved to stopping/stopped succeeds without extending the lease, so the worker learns to wind down. An item not visible to the work API is ErrWorkNotFound; a visible item whose precondition does not hold (the expected value is not the row's current last_heartbeat, or the first-heartbeat preconditions fail) is ErrHeartbeatMismatch (412).

func (*Queue) KeepLease

func (q *Queue) KeepLease(ctx context.Context, item *Item, ttl, stall time.Duration) (context.Context, *LeaseKeeper)

KeepLease starts a keeper that extends item's lease to ttl at every ttl/3 until Close, and returns a child context cancelled when a renewal fails (the lease is lost). Run the work under the returned context and call Close when it finishes; Close reports the first renewal failure, if any.

stall bounds how long the holder may report no progress before the keeper gives up on it: the work is cancelled and, deliberately, the lease is left to lapse, so the item becomes reclaimable exactly as it would if the process had died. That is the containment #383 was missing — a wedged sandbox call leaves the row untouched, so renewal succeeds forever and the documented crash recovery never fires, because nothing crashed. A stall of zero or less keeps a keeper that renews for as long as its holder lives, which is what the brain's turn loop wants: its silence is bounded a layer down, by provider.StallGuard on the model stream itself.

Progress is reported by the holder (a tool finished, a mount landed), never inferred here, because only the holder can tell a long step from a stuck one. Detection costs up to one tick on top of the budget, since both live on the same ticker — and the tick is the shorter of ttl/3 and the budget itself, so a lease much longer than the budget cannot stretch that cost with it. One case costs more, deliberately: the check and the renewal share this goroutine, so a renewal blocked on a stalled database delays the next check by however long it blocks. That is bounded by what the lease has left (the renewal's own budget) and self-limiting — the blocked attempt either times out, which cancels the holder anyway, or returns to a tick that is already buffered and checks at once. A second goroutine would close the gap and is not worth its synchronisation: the outcome in that window is a cancelled holder either way, and only the error naming it differs.

func (*Queue) ListWork

func (q *Queue) ListWork(ctx context.Context, envID domain.ID, after bool, afterT time.Time, afterID string, fetch int) ([]*Work, error)

ListWork returns a page of work items visible to the work API (see workAPIScope) for the environment, newest first by (created_at, id). It fetches up to `fetch` rows so the caller can pass limit+1 and detect a further page. When after is true, the (afterT, afterID) keyset position excludes rows at or newer than it, continuing a previous page.

func (*Queue) Poll

func (q *Queue) Poll(ctx context.Context, envID domain.ID, reclaim time.Duration) (*Work, error)

Poll reserves the oldest queued tool_exec item for one environment and hands it back to a BYOC worker. This is the wire work API's poll: unlike Claim (the executor's queued→active lease), poll is a soft reservation — the item stays queued, and the separate ack transitions it to starting. The reservation is recorded as a lease pushed out by reclaim, so a concurrent poll won't re-hand-out the same item until the window lapses (the wire's reclaim_older_than_ms — the reference's "reclaim un-ack'd work" knob). It returns nil with no error when the environment's tool_exec queue is empty. model_turn work drives the platform's own brain and is never offered to a worker.

Poll reclaims two kinds of stranded item. A still-queued (un-acked) reservation whose window lapsed is re-offered — the wire's reclaim_older_than_ms knob, carried in the reclaim argument. AND a dead worker's already-acked (starting) or heartbeating (active) item whose lease has lapsed (lease_expires_at < now(), i.e. the worker stopped heartbeating) is reclaimed: it is reset to a fresh queued reservation (state → queued; last_heartbeat, acknowledged_at, started_at cleared, so it is indistinguishable on the wire from a never-run queued item) so the next worker can re-poll, re-ack, and re-claim it with a fresh NO_HEARTBEAT — the mirror of Claim's expired-active reclaim for cloud. Note the lease a starting/active item is reclaimed on is a real lease (Ack installs a startup lease, heartbeats extend it), not the un-acked poll reservation. The active-item reclaim keys on the lapsed lease, NOT on reclaim_older_than_ms (which stays the un-acked-reservation window, per the wire). The C2a driver re-derives work from the still-unanswered tool uses, so a reclaimed run re-executes only unanswered tools.

Poll also FINALIZES the one stranded item it must never re-offer. A graceful stop is finished by the worker that was asked for it, but a worker that dies mid-wind-down never gets there, and re-offering a stopping item would resurrect work a caller asked to stop. Its lease lapses like any other holder's, and that is the signal: a stopping item whose lease has lapsed has nobody left to finish it, so the poll settles it terminally (→ stopped, stopped_at stamped, lease cleared) instead of leaving it non-terminal forever (#25). Such an item always carries a lease to lapse, because a graceful stop only enters stopping from active (see Stop) — the null-lease arm is not for it, but for the one row the new state machine does not write: during a rolling upgrade a not-yet-upgraded replica can still park a never-polled queued item, which has no lease at all, in stopping. Migration 0014 finalizes the ones written before the upgrade; this catches the ones written during it. The finalize rides along as a data-modifying CTE, which Postgres runs to completion whether or not the main query selects anything, and takes its rows with SKIP LOCKED so concurrent polls of one environment never block on each other. It is bounded so a poll costs a bounded write even in the pathological case (an environment whose whole fleet died mid-wind-down across many sessions); the remainder drains on the polls that follow, and no row can starve because a finalized one leaves the set. No ORDER BY: every row in the set is equally abandoned, so which 50 go first does not matter, and ordering would buy a sort the bound exists to avoid. It needs no environment-kind guard of its own: only the work API's Stop produces a stopping row, and that is already scoped to a self_hosted tool_exec item.

Every RE-hand-out mints a fresh work id (#62). A work item's identity is stable only while one worker holds it, because the wire's lifecycle calls carry no ownership proof — stop's body is {force} only, and the work object has no generation/version field — so re-offering the same id let a hung-then-revived worker's force-stop land on whatever worker held the item next, re-stranding the session. Under a rotated id that worker's stop, ack, and heartbeat all address an id that no longer exists (ErrWorkNotFound → 404) while the replacement's item is untouched. The three 404s are safe by different routes, not one: our worker's heartbeat reads a 4xx as a lost lease and cancels the run, a 404 stop is logged and dropped, and a 404 ack is dropped as an empty poll (routine hand-off, not a fault to back off from). The row, and with it the item's metadata and trace context, carries over unchanged.

The FIRST hand-out keeps the id enqueue minted — no worker has ever held it, so there is nothing to invalidate, and a client that listed the queued item can still address it. That is exactly the lease_expires_at IS NULL case: every other branch matched on a lease the item was previously handed out under. Ids are opaque to the client, so the wire is unchanged. Claim needs no equivalent: the cloud executor proves ownership with the lease itself (see Item.Lease).

Poll serves only self_hosted environments — the mirror of Claim scoping tool_exec to cloud. The two are therefore mutually exclusive by environment kind, so an item a worker has polled is never also run by the executor even if an environment key were misconfigured against a cloud environment.

func (*Queue) RecordPoll

func (q *Queue) RecordPoll(ctx context.Context, envID domain.ID, workerID string) error

RecordPoll upserts a BYOC worker's most recent poll time for the environment, feeding the workers_polling stat. It is best-effort telemetry off the poll path: a worker identifies itself with the Anthropic-Worker-ID header, and a poll without an id is simply not recorded (the wire documents workers_polling as requiring worker_id).

The same statement reaps the environment's rows that have aged past the workers_polling window (excluding this worker, which it is refreshing) so the table stays bounded by recently-active workers — without the reap, default worker ids being minted fresh per process (worker.defaultWorkerID) would leak one permanent row per process start. Reaping only rows already outside the window can never drop one that workers_polling would count.

func (*Queue) RegisterMetrics

func (q *Queue) RegisterMetrics() (metric.Registration, error)

RegisterMetrics installs the observable queue gauges on the global meter provider. It is called once per process (the control plane, which already owns the /work/stats view), after telemetry.Init has installed that provider, and the returned error is only a registration failure. The callback enumerates the self_hosted environments and reports each one's Stats, so depth/pending/ workers_polling carry an environment.id and mean exactly what /work/stats does. A Stats failure aborts that one collection rather than a turn — an observable callback has nothing to fail but the sample.

With more than one control-plane replica each would report the same per- environment values (the stats are global to the database), so a dashboard must read the gauge as last/max across instances, not a sum — the standard caveat for a gauge sampled from shared state. v1 is single-replica.

The returned Registration must be Unregister-ed before the pool it reads is closed: the meter provider's shutdown does a final collection, and a callback that outlived the pool would query a closed pool (a benign but noisy error and a dropped final sample). The control plane unregisters ahead of pool.Close.

func (*Queue) Requeue

func (q *Queue) Requeue(ctx context.Context, db DB, item *Item) error

Requeue hands a claimed item back to the queue inside the caller's transaction: the claimant discovered follow-on work for the same session (input that arrived mid-turn) and chains it under the item's existing live slot — an Enqueue would be suppressed by it. Requires the lease.

func (*Queue) Stats

func (q *Queue) Stats(ctx context.Context, envID domain.ID) (*WorkStats, error)

Stats computes the work-queue statistics for a self_hosted environment, scoped like the rest of the work API (see workAPIScope). Both depth and pending count only queued items — the wire's "acknowledged" is our Ack (queued→starting), so an acked item has left the queue and counts toward neither:

• depth — queued items available to be picked up: no reservation, or a poll reservation whose lease has lapsed (the same lease_expires_at < now() boundary Poll uses to re-offer an item).
• pending — queued items polled but not yet acked: a live poll reservation.
• oldest_queued_at — the oldest queued item's created_at (depth + pending), null when no item is queued.
• workers_polling — distinct workers whose last poll landed within the window (recorded by RecordPoll off the poll path). Its subquery carries the same self_hosted gate as workAPIScope, so all four fields report on the same queue — a non-self_hosted environment is zero across the board.

It is one snapshot: an aggregate-only SELECT always returns exactly one row, so an empty queue reports zeros with a null oldest_queued_at.

func (*Queue) Stop

func (q *Queue) Stop(ctx context.Context, envID, workID domain.ID, force bool) (*Work, error)

Stop stops a work item and returns the updated item, which the wire never carries: Stop answers a bodiless 204, unlike ack/heartbeat, so the API handler discards it and only this package's state-machine tests read it. force stops any not-yet-stopped item immediately (→ stopped); a graceful stop asks the item's worker to wind down, so it moves the item to stopping only when a worker has claimed the lease with a heartbeat (active). Whether that worker is still alive is not knowable here, and does not need to be: an abandoned wind-down is finalized by the next Poll of the environment (see Poll).

An item no worker has claimed is stopped outright instead, because it has no way back out of stopping. That state is left by the holder learning of it from its next heartbeat, winding its tools down and stopping the item (internal/worker/lease.go). A still-queued item has no holder to learn; an acked-but-never-heartbeated (starting) one does have a worker, but no channel left to be told on — its only remaining beat is the claim, which Heartbeat refuses once the row is no longer starting. Parking either in stopping would strand it there forever with a null stopped_at: Poll never re-offers a stopping item, and nothing else would ever finish the transition (#25). Neither has anything in flight to wind down, so the graceful stop simply completes.

Completing it can outrun a worker that was handed the item by one round trip: the worker starts its run concurrently with its claim beat (the reference's own ordering), so one that polled and acked may already be provisioning when the beat comes back 412 and cancels it. That window costs nothing — a stopped item is never re-offered by Poll, so no second worker can be handed it, and it is the same window force: true has always had ("immediately stop work without graceful shutdown"). Parking the item in stopping instead would not shorten it by a microsecond, since the worker cancels when its beat returns whatever the answer is, and would reintroduce #25 in miniature: an item whose worker then died would wait on a poll an emptied environment may never see.

Stopping an item that is already past the requested transition (e.g. graceful-stopping a stopping item, or stopping a stopped one) is ErrWorkConflict; an item not visible to the work API is ErrWorkNotFound.

func (*Queue) UpdateMetadata

func (q *Queue) UpdateMetadata(ctx context.Context, envID, workID domain.ID, upserts map[string]string, deletes []string) (*Work, error)

UpdateMetadata applies a metadata patch to a work item and returns the updated item (the wire Update responds with the BetaSelfHostedWork). upserts sets or overwrites keys; deletes removes keys; both are applied in one atomic UPDATE (metadata || upserts, then minus deletes), so a concurrent worker state transition on the same row cannot be lost to a read-modify-write and two overlapping patches cannot drop each other's writes — work items carry no optimistic version to guard a read-modify-write with, unlike the versioned resources. The patch is orthogonal to lifecycle: any item visible to the work API (see workAPIScope) is patchable in any state. An item not visible is ErrWorkNotFound.

type Work

type Work struct {
	ID              domain.ID
	EnvironmentID   domain.ID
	SessionID       domain.ID
	State           string
	Metadata        map[string]string
	CreatedAt       time.Time
	AcknowledgedAt  *time.Time // set by ack (queued → starting)
	StartedAt       *time.Time // set by the first heartbeat (→ active)
	StopRequestedAt *time.Time // set by stop
	StoppedAt       *time.Time // set when the item reaches stopped
	// LastHeartbeat is the wire's latest_heartbeat_at — null until the worker
	// heartbeats, which a freshly polled (still-queued) item has not.
	LastHeartbeat *time.Time
	// TraceContext is the W3C trace context (traceparent/tracestate) captured at
	// enqueue from the active span, so the executor or worker that runs the item
	// can parent its tool-execution spans on the turn that produced the work. It
	// is control-plane-internal (nil when enqueued with no active span) and never
	// rendered into the wire work object's metadata — a poll carries it in a
	// response header instead (see the API layer).
	TraceContext map[string]string
}

Work is a work_items row projected for the wire work API (poll/get/list and the state-transition endpoints). Unlike Item (a claimant's lease proof for the internal executor), it carries the fields a BetaSelfHostedWork response renders, including the lifecycle timestamps the state machine populates. Each nullable timestamp is null until its transition is reached (a queued item has none of them).

type WorkStats

type WorkStats struct {
	Depth          int64
	Pending        int64
	OldestQueuedAt *time.Time
	WorkersPolling int64
}

WorkStats is the work-queue statistics projection for the wire stats endpoint (BetaSelfHostedWorkQueueStats). depth and pending partition the queued state by whether a poll reservation is still live; OldestQueuedAt is the oldest queued item's timestamp (nil when the queue holds none); WorkersPolling counts the distinct recently-polling workers.