Documentation
¶
Overview ¶
Package queue is the internal work queue over Postgres (FOR UPDATE SKIP LOCKED, per the plan's component 4). Two kinds share the work_items table: model_turn drives the brain, tool_exec drives executors (consumed from slice 6). 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 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) 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) 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) 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 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.
Functions ¶
This section is empty.
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
}
Item is one claimed unit of work.
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue hands out work over one Postgres pool.
func (*Queue) Ack ¶
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 ¶
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.
func (*Queue) Claim ¶
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. model_turn work is claimed for every environment: the brain (model calls) runs on the platform regardless of where a session's sandbox lives.
func (*Queue) Complete ¶
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 ¶
Extend renews the claimant's lease mid-work (long provider streams) and returns the new lease proof.
func (*Queue) GetWork ¶
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) 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 ¶
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. A revived stale worker learns it lost the item on its next heartbeat (the echoed last_heartbeat no longer matches → 412). 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 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 ¶
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) Requeue ¶
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 ¶
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 ¶
Stop stops a work item and returns the updated item (the wire Stop responds with the BetaSelfHostedWork, like ack/heartbeat — not an empty 204). force stops any not-yet-stopped item immediately (→ stopped); a graceful stop moves a live (queued/starting/active) item to stopping so the worker can wind down. 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 ¶
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.
Source Files