Files
LocalAI/core/services/workerctl/paths.go
T
Ettore Di Giacinto 7bba436135 fix(distributed): preserve master protocol extensions
Carry prefix-cache pressure and reported residency over PostgreSQL, keep truthful backend stop replies, and retain capacity-aware S3 staging release over the worker control tunnel after the rebase.

Adapt the overlapping master tests to the tunnel-native protocol and cover cross-replica cache events and departed-node cleanup.

Assisted-by: Codex:GPT-5 [apply_patch] [exec_command]
2026-09-20 03:05:35 +00:00

160 lines
6.4 KiB
Go

// Package workerctl names the HTTP control plane a worker serves to the
// frontends that manage it.
//
// It is a leaf on the standard library alone, and deliberately so: the worker
// registers these paths and the frontend calls them, so both sides must agree
// on the literals without either importing the other's package.
package workerctl
import "encoding/json"
// Prefix is the one path prefix the worker mounts its whole control plane
// under. Everything the frontend may command a worker to do lives below it,
// which is what lets the worker put the control plane behind a single
// authentication check instead of one per verb.
const Prefix = "/v1/control/"
// The control verbs. Each replaces one NATS subject; the request and reply
// bodies are the messaging DTOs those subjects already carried, unchanged, so
// a worker still reachable over NATS and one reachable over the tunnel answer
// with the same bytes.
const (
PathBackendInstall = "/v1/control/backend/install"
PathBackendUpgrade = "/v1/control/backend/upgrade"
PathBackendList = "/v1/control/backend/list"
PathBackendStop = "/v1/control/backend/stop"
PathBackendDelete = "/v1/control/backend/delete"
PathModelStop = "/v1/control/model/stop"
PathModelUnload = "/v1/control/model/unload"
PathModelDelete = "/v1/control/model/delete"
PathModelsRunning = "/v1/control/models/running"
PathNodeStop = "/v1/control/node/stop"
// The file-staging verbs. They are only ever served by a worker whose
// deployment configured an object store, because without one there is
// nothing for them to move a file to or from; a worker without one mounts
// them not at all, and the catch-all under Prefix answers for them. That is
// the same 404 an older build gives, which is what the frontend already
// reads as "this worker does not serve that verb" rather than as absence.
PathFilesRelease = "/v1/control/files/release"
PathFilesEnsure = "/v1/control/files/ensure"
PathFilesStage = "/v1/control/files/stage"
PathFilesTemp = "/v1/control/files/temp"
PathFilesListDir = "/v1/control/files/listdir"
)
// The verbs an AGENT worker serves.
//
// They live in the same package, under the same prefix and behind the same
// bearer check as the backend worker's, because a worker is a worker to the
// frontend: one control client, one path table, one set of failure meanings.
// What differs is which of them a given worker MOUNTS, which is why the two
// sets are named separately below.
const (
PathMCPToolExecute = "/v1/control/mcp/tools/execute"
PathMCPDiscovery = "/v1/control/mcp/discovery"
PathAgentExecute = "/v1/control/agent/execute"
PathAgentCancel = "/v1/control/agent/cancel"
PathMCPCIRun = "/v1/control/mcp/ci/run"
)
// BackendPaths returns every control verb a BACKEND worker serves.
//
// It exists so a spec can assert a property of the whole set rather than of a
// list it re-types, which would go stale the moment a verb is added.
func BackendPaths() []string {
return []string{
PathBackendInstall,
PathBackendUpgrade,
PathBackendList,
PathBackendStop,
PathBackendDelete,
PathModelStop,
PathModelUnload,
PathModelDelete,
PathModelsRunning,
PathNodeStop,
PathFilesRelease,
PathFilesEnsure,
PathFilesStage,
PathFilesTemp,
PathFilesListDir,
}
}
// AgentPaths returns every control verb an AGENT worker serves.
//
// PathBackendStop is in BOTH sets and that is the point rather than an
// oversight: one path, two implementations, one caller. A backend worker kills
// the process and recycles its port; an agent worker drops the MCP sessions it
// cached for that backend. The frontend issues the same RPC to either and does
// not branch on the node's type to pick a carrier.
//
// PathAgentCancel is served by an agent worker and by nothing else. It is what
// replaced the agent.<name>.cancel broadcast, and moving that family here is
// what removed the last reason an agent worker dialled a message bus.
func AgentPaths() []string {
return []string{
PathMCPToolExecute,
PathMCPDiscovery,
PathAgentExecute,
PathAgentCancel,
PathMCPCIRun,
PathBackendStop,
}
}
// AllPaths returns every control verb's path, each once.
//
// It is the union of the two sets above and is what package-wide properties
// (every path under the prefix, no two verbs sharing a path) are asserted
// over. A spec about what a PARTICULAR worker mounts must use BackendPaths or
// AgentPaths instead: asserting mounting over the union would require every
// worker to serve every verb, which is the opposite of what the split is for.
func AllPaths() []string {
seen := make(map[string]bool)
out := make([]string, 0, len(BackendPaths())+len(AgentPaths()))
for _, p := range append(BackendPaths(), AgentPaths()...) {
if seen[p] {
continue
}
seen[p] = true
out = append(out, p)
}
return out
}
// Envelope is one line of a streaming control response.
//
// Exactly one of Progress and Reply is set. Zero or more Progress lines are
// followed by exactly ONE Reply line, and the Reply line is the last thing on
// the body. That ordering is the contract: it is what lets the frontend stop
// reading, and it is what replaces the subscribe-before-request dance the NATS
// carrier needed, since progress and reply now share one response and nothing
// can arrive before the caller is listening.
//
// Progress carrying the reply's own bytes is also why the 8000-byte
// notification cap that bounded the NATS progress subject has no analogue here:
// a line is written into the response body the caller is already reading.
type Envelope struct {
Progress json.RawMessage `json:"progress,omitempty"`
Reply json.RawMessage `json:"reply,omitempty"`
// Subject names the broadcast this progress line asks the frontend reading
// the stream to make on its behalf. Empty means the line is for this caller
// alone, which is what every pre-existing progress line is.
//
// It is a REQUEST and not an instruction: the frontend checks it against an
// allow list derived from the node's type before publishing anything. A
// worker naming a subject it has no business on is refused and logged, and
// the stream continues.
//
// It qualifies a Progress line and never a Reply line. A reply is the
// worker's verdict about the work, and there is no version of "publish my
// verdict for me" that this control plane has to carry.
Subject string `json:"subject,omitempty"`
}
// ContentTypeStream is the media type of a streaming control response.
const ContentTypeStream = "application/x-ndjson"