agent-dispatch-queue: public queued-job RPCs and messages - #1744
Open
erikhortsch wants to merge 15 commits into
Open
agent-dispatch-queue: public queued-job RPCs and messages#1744erikhortsch wants to merge 15 commits into
erikhortsch wants to merge 15 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: 547b4f1 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 |
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>
Two ways to say "no limit" in one API: JobGroup.concurrency_limit used plain
uint32 with zero meaning the dispatcher default, while DispatchLimit used
optional with unset meaning unlimited and zero meaning blocked. Both now use
zero-means-unset, and AddJobsRequest.group_concurrency_limit drops its optional
to match its own response field.
This gives up expressing "blocked" as a limit value, which was how a single
agent could be paused without deleting its row or touching its jobs. Pausing has
no other per-agent mechanism today - PauseJobGroup only covers groups - so if we
want it, it wants its own field rather than an overloaded zero. Same trade for
an exact-key row that exempts one agent from an all-agents default: with zero
meaning unset, the row falls through to the default instead of clearing it.
max_starts_per_minute becomes max_starts_per_second. Per-second could not
express the low-rate destination case ("3 calls per minute"), but that case left
with the label scope, and the scopes that remain - project and agent - are the
high-volume ones where per-second is the natural unit.
Drops the "not calendar minutes" note: the rolling window is an implementation
detail of how we count, not a promise the API needs to make.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
erikhortsch
commented
Aug 28, 2026
| // Dispatch starts allowed per second. The scheduler releases the budget in | ||
| // slices across admission rounds rather than at the top of each second. Zero | ||
| // leaves the scope unrated. | ||
| uint32 max_starts_per_second = 5; |
Contributor
Author
There was a problem hiding this comment.
Part of me wonders if this should be a float, so that rates like "1 every 3 seconds" could be set.
The comment sized min_free_capacity against arrival rate alone, which only holds if capacity comes back promptly. It does not. A dispatched job is never cancelled or paused, so capacity returns only as jobs finish, and the floor has to carry every non-queued arrival until then. That window is set by job length and by how bunched the starts are. Jobs that start together finish together: a batch of hour-long jobs admitted at once frees nothing for an hour, where thirty-second jobs free capacity continuously. So max_starts_per_second sizes this floor as much as arrival rate does, and spreading starts is usually cheaper than reserving capacity. Both fields now say so. Drops the "5 sessions/sec wants 10-20" example: it was only right for a window of a few seconds and misleads for anything longer-running. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Callbacks could only be set per job, so a caller adding a thousand jobs to one group repeated the same URL a thousand times and had no way to change it afterwards. AddJobsRequest.group_webhooks sets them once at group creation, ignored on later adds like group_concurrency_limit, and JobGroup echoes them. States the delivery rule the three levels now need: project, group and job callbacks all fire, and a URL listed at more than one level is delivered to once, most specific winning. Also clears signing_key on reads. QueuedJob embeds QueuedJobInput wholesale, so GetJob and ListJobs were handing every job's webhook signing keys back to any caller with read access; ListJobGroups would have done the same. Egress keeps webhooks on requests only and never echoes them on EgressInfo, which is the same problem solved by omission - we echo the URLs deliberately, so the keys have to come out explicitly. 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.