Files
multica/server/pkg/protocol/messages.go
Multica Eve c25a82eee0 perf(agents): fast model discovery on runtime switch (MUL-5444) (#6098)
* perf(agents): make runtime model discovery fast on runtime switch (MUL-5444)

Switching runtime in the agent creation form left the model picker
spinning for ~8-20s. Two costs stacked up:

- the list-models request sat in the store until the daemon's next
  scheduled heartbeat (0-15s, avg 7.5s of pure dead wait), and
- the daemon then enumerated the catalog locally (static for claude,
  but a CLI/ACP round trip up to ~15s for everyone else).

Both are addressed with the two standard techniques for a slow,
low-frequency, read-only operation: push instead of poll, and
stale-while-revalidate.

Push (removes the heartbeat wait):
- new additive `daemon:pending_work` hint, runtime-scoped, delivered
  through the existing daemon WS hub and the Redis relay so the API node
  holding the socket does the delivery.
- the daemon answers a hint with ONE immediate heartbeat and dispatches
  what it claimed. The hint deliberately carries no work, so nothing has
  to be un-claimed when delivery fails and a duplicate hint cannot
  duplicate work - PopPending stays the atomic claim.
- per-runtime coalescing plus a 1s floor keeps a caller-triggered hint
  from becoming a heartbeat amplifier.

Cache (removes the discovery wait on repeat opens):
- server-side per-runtime catalog cache (in-memory single-node, Redis
  multi-node) written on every successful report.
- a snapshot younger than 15min answers the POST immediately as an
  already-completed request; older than 60s it also enqueues a
  background refresh that only warms the cache.
- only supported, non-empty catalogs are cached; a completed-but-empty
  report invalidates instead, while a failed report keeps serving the
  last known good list.

Frontend: staleTime 60s -> 5min and gcTime 30min, so a runtime revisited
in the same session renders from cache and revalidates in the background
instead of showing the spinner again.

Compatibility: every wire change is additive. Old daemons ignore the
unknown hint type and keep using the scheduled heartbeat; new daemons
against an old server simply never receive one. The cached response is
shaped exactly like a completed live discovery apart from the optional
`cached` / `cached_at` markers.

Co-authored-by: multica-agent <github@multica.ai>

* fix(agents): address review on model discovery SWR (MUL-5444)

Sol-Boy's review on #6098 found the client cache could outlive the
server's own staleness promise, and that the two changed endpoints were
still cast rather than validated.

Must-fix 1 — client freshness now derives from the served answer.
`staleTime` was a flat 5min, so a 14-minute-old snapshot (which the
server returns while queueing its own refresh) was held as fresh for
another 5min: observable staleness became server window + client window,
and the refreshed catalog never reached the tab that triggered the
refresh. `staleTime` is now a function of the query data: a `cached`
answer is stale on arrival (bound stays the server's window alone, and
the next mount/focus picks up the refreshed snapshot), while a live
discovery — which just measured the truth — is trusted for the full 5min
so a cold runtime is never re-enumerated inside one form session.
`gcTime` stays 30min, so a revisited runtime still renders from cache and
revalidates in the background; the pickers gate their spinner on
`isLoading`, which stays false throughout.

Must-fix 2 — both model-discovery responses go through a zod schema.
`POST /api/runtimes/{id}/models` and its poll companion were casting
network JSON to `RuntimeModelListRequest`, which the root CLAUDE.md
API-compatibility rules forbid. Added a lenient schema (`status` stays
`z.string()`, `supported` defaults to true, `.loose()` keeps unknown
fields) plus a fallback record whose `status` is `failed`: a malformed
body now surfaces "discovery failed" with manual entry still usable
instead of a fabricated empty catalog or an endless spinner.
`resolveRuntimeModels` was tightened to match — only an explicit
`completed` is a catalog, so an unrecognised status is an error rather
than a silent empty list, and `supported` can no longer be `undefined`.

Nit — the in-memory catalog cache now deep-copies each entry's
`Thinking` (and its level slice) and `ServiceTiers`, so it delivers the
independent value its comment promises and matches the Redis backend's
JSON round-trip semantics.

Tests: staleTime policy for cached/live/no-data; a QueryObserver test
proving the refreshed catalog reaches the same client with no blank
loading state; unknown-status and omitted-`supported` handling; schema
tests for live, cached, old-backend and nine malformed shapes; client
tests that both endpoints degrade to an explicit failure; nested-field
mutation isolation for the cache.

Co-authored-by: multica-agent <github@multica.ai>

---------

Co-authored-by: Eve <eve@multica-ai.local>
Co-authored-by: multica-agent <github@multica.ai>
2026-07-29 16:03:26 +08:00

329 lines
15 KiB
Go

package protocol
import "encoding/json"
const (
DaemonCapabilitySkillBundlesV1 = "skill-bundles-v1"
DaemonCapabilityCoalescedCommentsV1 = "coalesced-comments-v1"
// DaemonCapabilityRPCV1 advertises that the daemon can carry
// request/response RPCs over the WebSocket control connection (MUL-4257).
// Gated so only daemons+servers that both support it route claim over WS;
// everyone else keeps using the HTTP claim endpoint.
DaemonCapabilityRPCV1 = "rpc-v1"
// AppCapabilityChatDraftRestoreV1 is advertised (X-Client-Capabilities) by
// app clients that understand the durable draft-restore recovery path:
// chat:cancel_finalized as an invalidation hint plus the draft-restores
// endpoint. Cancelling a started-but-empty chat task defers the
// empty/non-empty judgment (#5219), so its cancel response carries no
// synchronous restore — a client without this capability would silently
// drop the user's prompt, and keeps the legacy synchronous restore instead.
AppCapabilityChatDraftRestoreV1 = "chat-draft-restore-v1"
)
// RPCRequestPayload is the generic daemon→server request envelope carried in a
// protocol.Message of type EventDaemonRPCRequest. RequestID correlates the
// response; Method selects the server-side handler (e.g. "tasks.claim"); Body
// is the method-specific request JSON.
type RPCRequestPayload struct {
RequestID string `json:"request_id"`
Method string `json:"method"`
Body json.RawMessage `json:"body,omitempty"`
// TimeoutMs is the server-side execution budget in milliseconds. The server
// bounds the handler's context by it so a slow RPC is cancelled (its work
// rolled back) rather than committing after the daemon has already timed
// out waiting and fallen back to HTTP (MUL-4257). 0 means no server-side
// bound (connection-lifetime only).
TimeoutMs int64 `json:"timeout_ms,omitempty"`
}
// RPCResponsePayload is the server→daemon reply, carried in a
// protocol.Message of type EventDaemonRPCResponse. RequestID echoes the
// request. Status mirrors an HTTP status so the daemon can treat WS and HTTP
// outcomes uniformly. Exactly one of Body / Error is meaningful: Body on
// success (2xx), Error on failure.
type RPCResponsePayload struct {
RequestID string `json:"request_id"`
Status int `json:"status"`
Body json.RawMessage `json:"body,omitempty"`
Error string `json:"error,omitempty"`
}
// Message is the envelope for all WebSocket messages.
type Message struct {
Type string `json:"type"`
Payload json.RawMessage `json:"payload"`
}
// TaskDispatchPayload is sent from server to daemon when a task is assigned.
type TaskDispatchPayload struct {
TaskID string `json:"task_id"`
IssueID string `json:"issue_id"`
Title string `json:"title"`
Description string `json:"description"`
}
// TaskAvailablePayload is sent from server to daemon as a wakeup hint. The
// daemon still claims work through the existing HTTP claim endpoint.
type TaskAvailablePayload struct {
RuntimeID string `json:"runtime_id"`
TaskID string `json:"task_id,omitempty"`
}
// RuntimeProfilesChangedPayload is sent from server to daemon as a wakeup hint
// when a workspace custom runtime profile is created, edited, disabled, or
// deleted. The daemon still fetches profiles and registers runtimes through the
// existing HTTP endpoints.
type RuntimeProfilesChangedPayload struct {
WorkspaceID string `json:"workspace_id"`
RuntimeProfileID string `json:"runtime_profile_id,omitempty"`
}
// WorkspacesChangedPayload is an account-scoped hint that asks a daemon to
// reconcile its workspace membership set. The server remains authoritative;
// no workspace data is embedded in the event.
type WorkspacesChangedPayload struct{}
// PendingWorkKind values carried by PendingWorkPayload.Kind. The kind is
// advisory only — the daemon reacts identically to every kind (one immediate
// heartbeat, which claims whatever is queued) — so an unknown value from a
// newer server stays safe on an older daemon.
const (
PendingWorkKindModelList = "model_list"
)
// PendingWorkPayload is sent from server to daemon as a wakeup hint when a
// heartbeat-carried request is enqueued for a runtime. The daemon responds by
// sending one immediate heartbeat for RuntimeID instead of waiting for its next
// scheduled tick; the request itself is still claimed through the normal
// heartbeat path, so this event carries no work and is safe to lose, duplicate,
// or ignore (MUL-5444).
type PendingWorkPayload struct {
RuntimeID string `json:"runtime_id"`
Kind string `json:"kind,omitempty"`
}
// TaskProgressPayload is sent from daemon to server during task execution.
type TaskProgressPayload struct {
TaskID string `json:"task_id"`
Summary string `json:"summary"`
Step int `json:"step,omitempty"`
Total int `json:"total,omitempty"`
}
// TaskCompletedPayload is sent from daemon to server when a task finishes.
type TaskCompletedPayload struct {
TaskID string `json:"task_id"`
PRURL string `json:"pr_url,omitempty"`
Output string `json:"output,omitempty"`
}
// TaskMessagePayload represents a single agent execution message (tool call, text, etc.)
type TaskMessagePayload struct {
TaskID string `json:"task_id"`
IssueID string `json:"issue_id,omitempty"`
Seq int `json:"seq"`
Type string `json:"type"` // "text", "tool_use", "tool_result", "error"
Tool string `json:"tool,omitempty"` // tool name for tool_use/tool_result
Content string `json:"content,omitempty"` // text content
Input map[string]any `json:"input,omitempty"` // tool input (tool_use only)
Output string `json:"output,omitempty"` // tool output (tool_result only)
CreatedAt string `json:"created_at,omitempty"`
}
// DaemonRegisterPayload is sent from daemon to server on connection.
type DaemonRegisterPayload struct {
DaemonID string `json:"daemon_id"`
AgentID string `json:"agent_id"`
Runtimes []RuntimeInfo `json:"runtimes"`
}
// RuntimeInfo describes an available agent runtime on the daemon's machine.
type RuntimeInfo struct {
Type string `json:"type"`
Version string `json:"version"`
Status string `json:"status"`
}
// ChatMessagePayload is broadcast when a new chat message is created.
type ChatMessagePayload struct {
ChatSessionID string `json:"chat_session_id"`
MessageID string `json:"message_id"`
Role string `json:"role"`
Content string `json:"content"`
TaskID string `json:"task_id,omitempty"`
CreatedAt string `json:"created_at"`
}
// Chat message kinds (chat_message.message_kind). Additive: unknown values
// degrade to ChatMessageKindMessage on older readers.
const (
// ChatMessageKindMessage is an ordinary user/assistant message.
ChatMessageKindMessage = "message"
// ChatMessageKindNoResponse marks a direct-chat turn the agent completed
// without any text reply — a visible, deliberate terminal outcome rather
// than a silently-dropped turn (MUL-4351).
ChatMessageKindNoResponse = "no_response"
)
// ChatDonePayload is broadcast when an agent finishes responding to a chat
// message. Carries the freshly-persisted assistant ChatMessage so the client
// can write it into the messages cache inline — avoids a refetch round-trip
// during the live-timeline → AssistantMessage handoff that previously caused
// a visible flicker (#2123).
//
// MessageKind is additive (MUL-4351): older clients ignore it and fall back to
// the non-empty Content the server always sends, so a no_response turn still
// renders a real bubble instead of an empty one. Because direct-chat completion
// now always writes exactly one assistant row (message or no_response),
// MessageID/Content/CreatedAt/ElapsedMs are always populated for direct chat —
// the omitempty tags only elide fields for the legacy paths that broadcast
// without a row.
type ChatDonePayload struct {
ChatSessionID string `json:"chat_session_id"`
TaskID string `json:"task_id"`
MessageID string `json:"message_id,omitempty"`
Content string `json:"content,omitempty"`
ElapsedMs int64 `json:"elapsed_ms,omitempty"`
CreatedAt string `json:"created_at,omitempty"`
MessageKind string `json:"message_kind,omitempty"`
}
// Outcome values carried by ChatCancelFinalizedPayload.
const (
// ChatCancelOutcomeStopped: the transcript turned out non-empty, so a
// "Stopped." assistant message was persisted.
ChatCancelOutcomeStopped = "stopped"
// ChatCancelOutcomeRestored: the transcript stayed empty, so the
// triggering user message was deleted and its content should be
// restored into the composer as a draft.
ChatCancelOutcomeRestored = "restored"
)
// ChatCancelFinalizedPayload is broadcast when a cancelled chat task's
// deferred finalization settles (#5219). The cancel HTTP response cannot
// carry this outcome — it is only known after the daemon's transcript flush —
// so clients react to this event instead: outcome "stopped" inserts the
// assistant message (MessageID/Content/... describe the new row, shaped like
// ChatDonePayload), outcome "restored" removes the deleted user message from
// caches and prompts the initiator's client to fetch the durable draft
// restore from the creator-authorized endpoint. The restored prompt's content
// and attachments deliberately never ride this workspace-wide broadcast.
type ChatCancelFinalizedPayload struct {
Outcome string `json:"outcome"`
ChatSessionID string `json:"chat_session_id"`
TaskID string `json:"task_id"`
// InitiatorUserID is the human who triggered the cancelled task. Only
// this user's client needs to fetch the draft restore (the endpoint is
// creator-authorized regardless); clients treat a missing value as
// "not me".
InitiatorUserID string `json:"initiator_user_id,omitempty"`
MessageID string `json:"message_id,omitempty"`
// Content/MessageKind/CreatedAt/ElapsedMs describe the persisted
// "Stopped." assistant row and are set only for outcome "stopped" —
// the same exposure surface as chat:done.
Content string `json:"content,omitempty"`
MessageKind string `json:"message_kind,omitempty"`
CreatedAt string `json:"created_at,omitempty"`
ElapsedMs int64 `json:"elapsed_ms,omitempty"`
}
// ChatSessionReadPayload is broadcast when the creator marks a session as read.
// Fires to other devices so their unread counts stay in sync.
type ChatSessionReadPayload struct {
ChatSessionID string `json:"chat_session_id"`
}
// ChatSessionDeletedPayload is broadcast when a chat session is hard-deleted
// so other tabs/devices drop it from their session lists and reset the active
// pointer if it referenced the deleted session.
type ChatSessionDeletedPayload struct {
ChatSessionID string `json:"chat_session_id"`
}
// ChatSessionUpdatedPayload is broadcast when a user-editable field on a
// chat session changes (today: title via inline rename). Other tabs/devices
// patch the session row in their cached list so the dropdown stays in sync
// without a full refetch.
type ChatSessionUpdatedPayload struct {
ChatSessionID string `json:"chat_session_id"`
Title string `json:"title"`
// ProjectID is set only by the project-context update path. The double
// pointer distinguishes an omitted field from an explicit JSON null.
ProjectID **string `json:"project_id,omitempty"`
// Pinned is set only by the pin/unpin path; nil on a plain rename so a
// receiver leaves the existing pin state untouched.
Pinned *bool `json:"pinned,omitempty"`
// Status is set only by the archive/unarchive path ("active"/"archived");
// nil on rename/pin so a receiver leaves the existing status untouched.
Status *string `json:"status,omitempty"`
UpdatedAt string `json:"updated_at"`
}
// DaemonHeartbeatRequestPayload is sent from daemon to server over WebSocket
// to update last_seen_at and pull pending actions for a single runtime.
// Mirrors the body of POST /api/daemon/heartbeat so both transports share
// identical semantics.
type DaemonHeartbeatRequestPayload struct {
RuntimeID string `json:"runtime_id"`
SupportsBatchImport bool `json:"supports_batch_import,omitempty"`
}
// DaemonHeartbeatAckPayload is the server's reply to DaemonHeartbeatRequestPayload.
// JSON shape mirrors the HTTP heartbeat response so daemon code can decode either.
// ServerCapabilities is explicit server-to-daemon protocol negotiation. A
// daemon must not infer support from its own advertised client capabilities.
//
// RuntimeGone is the WebSocket replacement for the HTTP 404 "runtime not found"
// response. When the server discovers the runtime row was deleted (UI delete,
// 7-day offline GC), it sends back an ack with Status=HeartbeatStatusRuntimeGone
// and RuntimeGone=true rather than tearing down the connection with an error.
// The daemon reads this signal, prunes the stale runtime from its local state
// and re-registers; without it the dead UUID would keep heartbeating until the
// daemon process restarts.
type DaemonHeartbeatAckPayload struct {
RuntimeID string `json:"runtime_id"`
Status string `json:"status"`
ServerCapabilities []string `json:"server_capabilities,omitempty"`
RuntimeGone bool `json:"runtime_gone,omitempty"`
PendingUpdate *DaemonHeartbeatPendingUpdate `json:"pending_update,omitempty"`
PendingModelList *DaemonHeartbeatPendingModelList `json:"pending_model_list,omitempty"`
PendingLocalSkills *DaemonHeartbeatPendingLocalSkills `json:"pending_local_skills,omitempty"`
PendingLocalSkillImport *DaemonHeartbeatPendingLocalSkillImport `json:"pending_local_skill_import,omitempty"`
// PendingLocalSkillImports carries multiple import requests in a single
// heartbeat so the daemon can process them concurrently. Old daemons
// that don't know this field silently ignore it (standard JSON behavior)
// and fall back to the singular PendingLocalSkillImport above.
PendingLocalSkillImports []DaemonHeartbeatPendingLocalSkillImport `json:"pending_local_skill_imports,omitempty"`
}
// HeartbeatStatusRuntimeGone is the ack Status used when the runtime row no
// longer exists server-side. Companion to DaemonHeartbeatAckPayload.RuntimeGone.
const HeartbeatStatusRuntimeGone = "runtime_gone"
// DaemonHeartbeatPendingUpdate describes a CLI-update action the daemon
// should run for the runtime.
type DaemonHeartbeatPendingUpdate struct {
ID string `json:"id"`
TargetVersion string `json:"target_version"`
}
// DaemonHeartbeatPendingModelList describes a request for the daemon to
// enumerate the runtime's supported models.
type DaemonHeartbeatPendingModelList struct {
ID string `json:"id"`
}
// DaemonHeartbeatPendingLocalSkills describes a request for the runtime's
// local-skill inventory.
type DaemonHeartbeatPendingLocalSkills struct {
ID string `json:"id"`
}
// DaemonHeartbeatPendingLocalSkillImport describes a request to import a
// specific runtime local skill.
type DaemonHeartbeatPendingLocalSkillImport struct {
ID string `json:"id"`
SkillKey string `json:"skill_key"`
}