agent-dispatch-queue: public queued-job RPCs and messages - #1744
Open
erikhortsch wants to merge 10 commits into
Open
agent-dispatch-queue: public queued-job RPCs and messages#1744erikhortsch wants to merge 10 commits into
erikhortsch wants to merge 10 commits into
Conversation
Add the AgentDispatchQueue twirp service: AddJobs/GetJob/ListJobs/RemoveJob/ RetryJobs, the five group RPCs (Unimplemented until AP-976), and queue config Get/Update. QueuedJobInput mirrors CreateAgentDispatchRequest so a queued job and an immediate dispatch describe the same work. A standalone service rather than new methods on AgentDispatchService: twirp generates no Unimplemented embed, so adding methods there is a compile break for every implementer, and EnsureAdminPermission is room-scoped while these RPCs are project-scoped. Follows the CloudAgent precedent. Requests carry no project_id; the project comes from the caller's credentials. Authorized by the new AgentGrant.DispatchAdmin. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
🦋 Changeset detectedLatest commit: 194c3a2 The changes in this PR will be included in the next version bump. This PR includes changesets to release 2 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
3 tasks
Make the queue config extensible and name the reserve for what it controls. UpdateDispatchQueueConfigRequest carried a bare uint32, so every future limit would have been a breaking change or a duplicated field list. It now carries DispatchQueueConfig, replaced wholesale. reserved_inbound_concurrency becomes outbound_headroom. The value is a margin below the project's capacity where outbound admission stops, not an allocation sized to inbound volume: admission already reads capacity remaining, so inbound lowers outbound's budget on its own. The old name led every reader to size it to peak inbound concurrency instead of to arrivals during one admission round. Add DispatchLimit/DispatchLimitScope for per-agent and per-label limits, with max_concurrent and a rolling-60s max_starts_per_minute. An empty key means every distinct value gets its own budget, so one rule covers all agents or all label values without enumerating them. QueuedJobInput gains labels to scope them; labels name classes, never identities, to bound cardinality and keep PII out of a config surface. Group concurrency was already enforced by the scheduler but unreachable from the API. Expose it inline as AddJobsRequest.group_concurrency_limit and JobGroup.concurrency_limit rather than as a DispatchLimit scope, since groups are per-campaign and would churn the project config. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
QueuedJobInput could only express a deadline. A caller wanting a job to run later had to hold it and submit at the right moment, which is the queue's job. not_before rather than run_after: capacity still gates dispatch, so this is a floor on eligibility, not a trigger that fires at the given time. It pairs with expires_at as a half-open window, matching nbf/notBefore elsewhere. Eligibility filters the FIFO scan; ordering stays created_at, so a held job keeps its place in line rather than going to the back when it becomes eligible. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
outbound_headroom becomes min_free_capacity. The old name read backwards: "outbound headroom" sounds like the headroom outbound gets, when it is the capacity the queue must leave alone, so a reader who took it the intuitive way would size it to their queue volume and starve everything else. It also named a beneficiary that does not exist. What the field protects is whatever is not this queue - sessions a user started, direct CreateDispatch calls, which may themselves be outbound - so inbound/outbound was the wrong axis. Naming the invariant instead leaves it correct for workloads that are not calls at all. not_before becomes start_after, and RetryJobs gains it, so a retry can be held back rather than re-queued immediately. Drop "campaign" throughout; it promises scheduling and reporting this does not do. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The comment explained min_free_capacity as a margin covering arrivals between admission rounds, as if the problem were the queue recomputing on a delay. It is not. Requests that do not come through the queue have nowhere to wait: if the queue holds the project's last capacity, they fail with quota exceeded. That asymmetry - queued work waits, everything else fails - is the whole reason the floor exists, and it changes how to size it: to the demand you cannot afford to reject, not to a recompute interval. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
erikhortsch
marked this pull request as ready for review
August 27, 2026 23:13
Per-destination limits ("call this number at most X times") are not in v1, so
the label scope and the job labels that fed it come out. v1 limits are project
and agent scope, plus the per-group limit set at AddJobsRequest.
Removes DISPATCH_LIMIT_SCOPE_LABEL, DispatchLimit.label and
QueuedJobInput.labels. Field numbers 3 and 10 are left as gaps rather than
renumbered: nothing has been tagged, but a silent renumber is a worse failure
for anyone already generating from this branch than a cosmetic hole.
Labels were the only field carrying customer-supplied per-job classification,
so dropping them also drops the PII and cardinality validation they needed at
the edge. Both come back with the scope.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Adds a
livekit.AgentDispatchQueuetwirp service for queued agent dispatch: dispatches thatshould run later, as project capacity allows, rather than failing at the concurrency cap the
way
AgentDispatchService.CreateDispatchdoes.AddJobs(batch, optionalgroup_id),GetJob,ListJobs,RemoveJob,RetryJobsListJobGroups,PauseJobGroup,ResumeJobGroup,CancelJobGroup,DeleteJobGroupGetDispatchQueueConfig/UpdateDispatchQueueConfigQueuedJob,QueuedJobInput,QueuedJobStatus,JobGroup,DispatchQueueConfigAlso adds
AgentGrant.DispatchAdminto authorize the service.Contract only — no server implementation lands here. As with
CloudAgent, this is served bycloud; OSS livekit-server does not mount the path.
Why a separate service
The obvious alternative is adding these RPCs to the existing
AgentDispatchService, so callerskeep one client for immediate and queued dispatch. Two things rule it out:
Unimplemented*Serverembed. The generated server struct embeds theinterface itself, so adding a method to
AgentDispatchServiceis a hard compile break forevery implementer, with nothing to fall back on.
AgentDispatchServiceauthorizes per-room. Its methods require an admin grant scoped toreq.Room. Queue operations are project-scoped with no room, so that check cannot express them.CloudAgentis the existing precedent for a project-scoped, cloud-served surface in this repo,and this follows it.
Notes on the shape
QueuedJobInputmirrorsCreateAgentDispatchRequest—agent_name,room_name,metadata,attributes,deployment,restart_policy— so a queued job and an immediate dispatchdescribe the same work, rather than hiding it behind an opaque payload string.
project_idon any request; the project comes from the caller's credentials.group_id; unset applies at project scope.QUEUED_JOB_STATUS_UNSPECIFIEDis the zero value, so an unset status cannot read as a live job.Verification
go build ./...clean;go test -race ./auth/... ./livekit/...passes.AgentDispatchQueueexposes all 12 RPCs; both the protobuf and JSON clients satisfy it, andAddJobsRequestround-trips preservinggroup_id,room_name, andWebhookConfig.signing_key.AgentDispatchService: the diff againstmainforlivekit_agent_dispatch.protoand its bindings is zero lines, so existing implementers are unaffected.