mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-08 01:16:45 +08:00
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 <[email protected]> * 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 <[email protected]> --------- Co-authored-by: Eve <[email protected]> Co-authored-by: multica-agent <[email protected]>
This commit is contained in:
Multica Eve
co-authored by
multica-agent
Eve
multica-agent
Eve
parent
280fa28e0d
commit
c25a82eee0
@@ -1509,3 +1509,87 @@ describe("ApiClient", () => {
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("ApiClient model discovery response schema", () => {
|
||||
const completed = {
|
||||
id: "req-1",
|
||||
runtime_id: "rt-1",
|
||||
status: "completed",
|
||||
supported: true,
|
||||
created_at: "2026-07-29T00:00:00Z",
|
||||
updated_at: "2026-07-29T00:00:01Z",
|
||||
models: [{ id: "claude-sonnet-4-6", label: "Claude Sonnet 4.6" }],
|
||||
};
|
||||
|
||||
function stubJSON(body: unknown) {
|
||||
vi.stubGlobal(
|
||||
"fetch",
|
||||
vi.fn().mockResolvedValue(
|
||||
new Response(JSON.stringify(body), {
|
||||
status: 200,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
it("parses a live completed discovery", async () => {
|
||||
stubJSON(completed);
|
||||
|
||||
const result = await new ApiClient("https://api.example.test")
|
||||
.initiateListModels("rt-1");
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: "completed",
|
||||
supported: true,
|
||||
models: [{ id: "claude-sonnet-4-6" }],
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps the cache markers on a server-cached snapshot", async () => {
|
||||
stubJSON({ ...completed, cached: true, cached_at: "2026-07-29T00:00:00Z" });
|
||||
|
||||
const result = await new ApiClient("https://api.example.test")
|
||||
.initiateListModels("rt-1");
|
||||
|
||||
expect(result.cached).toBe(true);
|
||||
expect(result.cached_at).toBe("2026-07-29T00:00:00Z");
|
||||
});
|
||||
|
||||
// The picker drives a state machine off `status`, so a malformed body must
|
||||
// become an explicit failure — not a fabricated empty catalog, and not an
|
||||
// endless "discovering models" spinner.
|
||||
it("degrades a malformed initiate response to an explicit failure", async () => {
|
||||
stubJSON({ status: 7, models: "nope" });
|
||||
|
||||
const result = await new ApiClient("https://api.example.test")
|
||||
.initiateListModels("rt-1");
|
||||
|
||||
expect(result.status).toBe("failed");
|
||||
expect(result.supported).toBe(true);
|
||||
expect(result.error).toBe("invalid model discovery response");
|
||||
expect(result.runtime_id).toBe("rt-1");
|
||||
});
|
||||
|
||||
it("degrades a malformed poll response to an explicit failure", async () => {
|
||||
stubJSON("not-an-object");
|
||||
|
||||
const result = await new ApiClient("https://api.example.test")
|
||||
.getListModelsResult("rt-1", "req-9");
|
||||
|
||||
expect(result.status).toBe("failed");
|
||||
expect(result.id).toBe("req-9");
|
||||
expect(result.runtime_id).toBe("rt-1");
|
||||
});
|
||||
|
||||
it("stays usable against a backend that omits supported", async () => {
|
||||
const { supported: _omitted, ...withoutSupported } = completed;
|
||||
stubJSON(withoutSupported);
|
||||
|
||||
const result = await new ApiClient("https://api.example.test")
|
||||
.getListModelsResult("rt-1", "req-1");
|
||||
|
||||
expect(result.supported).toBe(true);
|
||||
expect(result.status).toBe("completed");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -295,6 +295,8 @@ import {
|
||||
EMPTY_GITHUB_CONNECT_RESPONSE,
|
||||
EMPTY_LIST_GITHUB_INSTALLATIONS_RESPONSE,
|
||||
EMPTY_LIST_GITHUB_REPOSITORIES_RESPONSE,
|
||||
RuntimeModelListRequestSchema,
|
||||
MALFORMED_RUNTIME_MODEL_LIST_REQUEST,
|
||||
} from "./schemas";
|
||||
|
||||
/** Identifies the calling client to the server.
|
||||
@@ -1680,15 +1682,40 @@ export class ApiClient {
|
||||
return this.fetch(`/api/runtimes/${runtimeId}/update/${updateId}`);
|
||||
}
|
||||
|
||||
// Both discovery endpoints feed a UI state machine (poll while
|
||||
// pending/running, then render or fail), so the response is validated rather
|
||||
// than cast: an unparseable body degrades to an explicit "failed" record that
|
||||
// shows the discovery error and keeps manual model entry usable, instead of a
|
||||
// fabricated empty catalog or an endless spinner (MUL-5444).
|
||||
async initiateListModels(runtimeId: string): Promise<RuntimeModelListRequest> {
|
||||
return this.fetch(`/api/runtimes/${runtimeId}/models`, { method: "POST" });
|
||||
const raw = await this.fetch<unknown>(`/api/runtimes/${runtimeId}/models`, {
|
||||
method: "POST",
|
||||
});
|
||||
return parseWithFallback<RuntimeModelListRequest>(
|
||||
raw,
|
||||
RuntimeModelListRequestSchema,
|
||||
{ ...MALFORMED_RUNTIME_MODEL_LIST_REQUEST, runtime_id: runtimeId },
|
||||
{ endpoint: "POST /api/runtimes/{id}/models" },
|
||||
);
|
||||
}
|
||||
|
||||
async getListModelsResult(
|
||||
runtimeId: string,
|
||||
requestId: string,
|
||||
): Promise<RuntimeModelListRequest> {
|
||||
return this.fetch(`/api/runtimes/${runtimeId}/models/${requestId}`);
|
||||
const raw = await this.fetch<unknown>(
|
||||
`/api/runtimes/${runtimeId}/models/${requestId}`,
|
||||
);
|
||||
return parseWithFallback<RuntimeModelListRequest>(
|
||||
raw,
|
||||
RuntimeModelListRequestSchema,
|
||||
{
|
||||
...MALFORMED_RUNTIME_MODEL_LIST_REQUEST,
|
||||
id: requestId,
|
||||
runtime_id: runtimeId,
|
||||
},
|
||||
{ endpoint: "GET /api/runtimes/{id}/models/{requestId}" },
|
||||
);
|
||||
}
|
||||
|
||||
async initiateListLocalSkills(
|
||||
|
||||
@@ -24,6 +24,8 @@ import {
|
||||
IssueTriggerPreviewSchema,
|
||||
ListIssuesResponseSchema,
|
||||
ListPropertiesResponseSchema,
|
||||
MALFORMED_RUNTIME_MODEL_LIST_REQUEST,
|
||||
RuntimeModelListRequestSchema,
|
||||
SearchProjectsResponseSchema,
|
||||
RuntimeHourlyActivityListSchema,
|
||||
RuntimeUsageByAgentListSchema,
|
||||
@@ -989,3 +991,119 @@ describe("CommentTriggerPreviewSchema.blocked", () => {
|
||||
expect(parsed.blocked.map((b) => b.target_id)).toEqual(["s1", "a1"]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("RuntimeModelListRequestSchema", () => {
|
||||
const completed = {
|
||||
id: "req-1",
|
||||
runtime_id: "rt-1",
|
||||
status: "completed",
|
||||
supported: true,
|
||||
created_at: "2026-07-29T00:00:00Z",
|
||||
updated_at: "2026-07-29T00:00:01Z",
|
||||
models: [
|
||||
{
|
||||
id: "gpt-5.6-sol",
|
||||
label: "GPT-5.6-Sol",
|
||||
provider: "openai",
|
||||
default: true,
|
||||
thinking: {
|
||||
supported_levels: [{ value: "high", label: "High" }],
|
||||
default_level: "low",
|
||||
},
|
||||
service_tiers: [{ id: "fast", name: "Fast" }],
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
it("parses a live completed discovery, keeping the fields the UI branches on", () => {
|
||||
const parsed = parseWithFallback(
|
||||
completed,
|
||||
RuntimeModelListRequestSchema,
|
||||
MALFORMED_RUNTIME_MODEL_LIST_REQUEST,
|
||||
{ endpoint: "test" },
|
||||
);
|
||||
expect(parsed.status).toBe("completed");
|
||||
expect(parsed.supported).toBe(true);
|
||||
expect(parsed.models?.[0]?.default).toBe(true);
|
||||
expect(parsed.models?.[0]?.thinking?.supported_levels).toEqual([
|
||||
{ value: "high", label: "High" },
|
||||
]);
|
||||
expect(parsed.models?.[0]?.service_tiers).toEqual([{ id: "fast", name: "Fast" }]);
|
||||
expect(parsed.cached).toBeUndefined();
|
||||
});
|
||||
|
||||
it("keeps the additive cache markers when the server serves a snapshot", () => {
|
||||
const parsed = parseWithFallback(
|
||||
{ ...completed, cached: true, cached_at: "2026-07-29T00:00:00Z" },
|
||||
RuntimeModelListRequestSchema,
|
||||
MALFORMED_RUNTIME_MODEL_LIST_REQUEST,
|
||||
{ endpoint: "test" },
|
||||
);
|
||||
expect(parsed.cached).toBe(true);
|
||||
expect(parsed.cached_at).toBe("2026-07-29T00:00:00Z");
|
||||
});
|
||||
|
||||
// A backend that predates MUL-5444 sends neither marker; an even older one
|
||||
// may omit `supported`. Both must stay usable rather than reading as
|
||||
// "runtime manages the model itself" off an undefined.
|
||||
it("defaults supported to true on an older backend that omits it", () => {
|
||||
const { supported: _omitted, ...withoutSupported } = completed;
|
||||
const parsed = parseWithFallback(
|
||||
withoutSupported,
|
||||
RuntimeModelListRequestSchema,
|
||||
MALFORMED_RUNTIME_MODEL_LIST_REQUEST,
|
||||
{ endpoint: "test" },
|
||||
);
|
||||
expect(parsed.supported).toBe(true);
|
||||
expect(parsed.cached).toBeUndefined();
|
||||
});
|
||||
|
||||
it("passes an unknown status through instead of failing the whole response", () => {
|
||||
const parsed = parseWithFallback(
|
||||
{ ...completed, status: "superseded" },
|
||||
RuntimeModelListRequestSchema,
|
||||
MALFORMED_RUNTIME_MODEL_LIST_REQUEST,
|
||||
{ endpoint: "test" },
|
||||
);
|
||||
expect(parsed.status).toBe("superseded");
|
||||
});
|
||||
|
||||
// Malformed bodies must land on the "failed" fallback: `completed` would
|
||||
// fabricate an empty catalog and `pending` would spin the picker until the
|
||||
// client-side poll timeout.
|
||||
it("falls back to an explicit failure on a malformed body", () => {
|
||||
for (const malformed of [
|
||||
null,
|
||||
"nope",
|
||||
42,
|
||||
{},
|
||||
{ status: 7 },
|
||||
{ ...completed, status: undefined },
|
||||
{ ...completed, supported: "yes" },
|
||||
{ ...completed, models: "nope" },
|
||||
{ ...completed, models: [{ label: "no id" }] },
|
||||
]) {
|
||||
const parsed = parseWithFallback(
|
||||
malformed,
|
||||
RuntimeModelListRequestSchema,
|
||||
MALFORMED_RUNTIME_MODEL_LIST_REQUEST,
|
||||
{ endpoint: "test" },
|
||||
);
|
||||
expect(parsed.status).toBe("failed");
|
||||
expect(parsed.supported).toBe(true);
|
||||
expect(parsed.error).toBe("invalid model discovery response");
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps unknown server fields instead of stripping them", () => {
|
||||
const parsed = parseWithFallback(
|
||||
{ ...completed, future_field: "keep me" },
|
||||
RuntimeModelListRequestSchema,
|
||||
MALFORMED_RUNTIME_MODEL_LIST_REQUEST,
|
||||
{ endpoint: "test" },
|
||||
);
|
||||
expect((parsed as unknown as { future_field?: string }).future_field).toBe(
|
||||
"keep me",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -39,6 +39,7 @@ import type {
|
||||
ListWebhookDeliveriesResponse,
|
||||
NotificationPreferenceResponse,
|
||||
ResourceLabelsResponse,
|
||||
RuntimeModelListRequest,
|
||||
SearchIssuesResponse,
|
||||
SearchProjectsResponse,
|
||||
Squad,
|
||||
@@ -1820,3 +1821,80 @@ export const CreateBillingPortalSessionResponseSchema = z.object({
|
||||
export const EMPTY_CREATE_BILLING_PORTAL_SESSION_RESPONSE: CreateBillingPortalSessionResponse = {
|
||||
url: "",
|
||||
};
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Runtime model discovery (`POST /api/runtimes/:id/models`,
|
||||
// `GET /api/runtimes/:id/models/:requestId`). Both endpoints return the same
|
||||
// request record, and the UI drives a state machine off `status`, so the two
|
||||
// fields that decide behaviour are pinned: `status` gates the polling loop and
|
||||
// `supported` gates whether the picker is usable at all. Everything else stays
|
||||
// lenient per the rules at the top of this file.
|
||||
//
|
||||
// `status` deliberately stays `z.string()` (a newer server may add a state);
|
||||
// `resolveRuntimeModels` treats anything it does not recognise as an explicit
|
||||
// failure rather than a completed-but-empty catalog. `supported` defaults to
|
||||
// true so a server old enough to omit it keeps the picker enabled instead of
|
||||
// rendering "managed by runtime" off an `undefined`.
|
||||
//
|
||||
// `cached` / `cached_at` are additive markers for a snapshot served from the
|
||||
// server-side catalog cache (MUL-5444); an older backend omits them.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const RuntimeModelThinkingLevelSchema = z.object({
|
||||
value: z.string(),
|
||||
label: z.string().default(""),
|
||||
description: z.string().optional(),
|
||||
}).loose();
|
||||
|
||||
const RuntimeModelThinkingSchema = z.object({
|
||||
supported_levels: z.array(RuntimeModelThinkingLevelSchema).default([]),
|
||||
default_level: z.string().optional(),
|
||||
}).loose();
|
||||
|
||||
const RuntimeModelServiceTierSchema = z.object({
|
||||
id: z.string(),
|
||||
name: z.string().default(""),
|
||||
description: z.string().optional(),
|
||||
}).loose();
|
||||
|
||||
// A model entry with no `id` is unselectable — `onChange(m.id)` would persist
|
||||
// an empty model — so `id` is required and a malformed entry drops the whole
|
||||
// response to the fallback rather than rendering a dead row.
|
||||
const RuntimeModelSchema = z.object({
|
||||
id: z.string(),
|
||||
label: z.string().default(""),
|
||||
provider: z.string().optional(),
|
||||
default: z.boolean().optional(),
|
||||
thinking: RuntimeModelThinkingSchema.nullable().optional()
|
||||
.transform((v) => v ?? undefined),
|
||||
service_tiers: z.array(RuntimeModelServiceTierSchema).optional(),
|
||||
}).loose();
|
||||
|
||||
export const RuntimeModelListRequestSchema = z.object({
|
||||
id: z.string().default(""),
|
||||
runtime_id: z.string().default(""),
|
||||
status: z.string(),
|
||||
models: z.array(RuntimeModelSchema).optional(),
|
||||
supported: z.boolean().default(true),
|
||||
error: z.string().optional(),
|
||||
created_at: z.string().default(""),
|
||||
updated_at: z.string().default(""),
|
||||
cached: z.boolean().optional(),
|
||||
cached_at: z.string().optional(),
|
||||
}).loose();
|
||||
|
||||
// Fallback for an unparseable model-discovery response. `failed` is the only
|
||||
// honest choice: `completed` would fabricate an empty catalog (and silently
|
||||
// clear a saved model when `supported` is read as false), while `pending`
|
||||
// would spin the picker until the client-side poll timeout. `failed` surfaces
|
||||
// "discovery failed" immediately and leaves the creatable manual-entry field
|
||||
// working, which is the same degradation as a real discovery failure.
|
||||
export const MALFORMED_RUNTIME_MODEL_LIST_REQUEST: RuntimeModelListRequest = {
|
||||
id: "",
|
||||
runtime_id: "",
|
||||
status: "failed",
|
||||
supported: true,
|
||||
error: "invalid model discovery response",
|
||||
created_at: "",
|
||||
updated_at: "",
|
||||
};
|
||||
|
||||
@@ -0,0 +1,251 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { QueryClient, QueryObserver } from "@tanstack/react-query";
|
||||
import {
|
||||
LIVE_MODELS_STALE_TIME_MS,
|
||||
resolveRuntimeModels,
|
||||
runtimeModelsKeys,
|
||||
runtimeModelsOptions,
|
||||
staleTimeFor,
|
||||
} from "./models";
|
||||
import type { RuntimeModelListRequest, RuntimeModelsResult } from "../types/agent";
|
||||
|
||||
const initiateListModels = vi.fn();
|
||||
const getListModelsResult = vi.fn();
|
||||
|
||||
vi.mock("../api", () => ({
|
||||
api: {
|
||||
initiateListModels: (runtimeId: string) => initiateListModels(runtimeId),
|
||||
getListModelsResult: (runtimeId: string, requestId: string) =>
|
||||
getListModelsResult(runtimeId, requestId),
|
||||
},
|
||||
}));
|
||||
|
||||
const catalog = [{ id: "claude-sonnet-4-6", label: "Claude Sonnet 4.6" }];
|
||||
const refreshedCatalog = [{ id: "claude-opus-5", label: "Claude Opus 5" }];
|
||||
|
||||
function request(
|
||||
overrides: Partial<RuntimeModelListRequest>,
|
||||
): RuntimeModelListRequest {
|
||||
return {
|
||||
id: "req-1",
|
||||
runtime_id: "rt-1",
|
||||
status: "pending",
|
||||
supported: true,
|
||||
created_at: "2026-07-29T00:00:00Z",
|
||||
updated_at: "2026-07-29T00:00:00Z",
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function cachedResponse(
|
||||
models: RuntimeModelListRequest["models"],
|
||||
cachedAt: string,
|
||||
): RuntimeModelListRequest {
|
||||
return request({
|
||||
status: "completed",
|
||||
models,
|
||||
cached: true,
|
||||
cached_at: cachedAt,
|
||||
});
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
initiateListModels.mockReset();
|
||||
getListModelsResult.mockReset();
|
||||
});
|
||||
|
||||
describe("resolveRuntimeModels", () => {
|
||||
// The server answers a warm runtime straight from its catalog cache
|
||||
// (MUL-5444). That response is already terminal, so discovery must resolve on
|
||||
// the POST alone — one round trip, no polling, no spinner.
|
||||
it("resolves from a cached completed response without polling", async () => {
|
||||
initiateListModels.mockResolvedValue(
|
||||
cachedResponse(catalog, "2026-07-29T00:00:00Z"),
|
||||
);
|
||||
|
||||
const result = await resolveRuntimeModels("rt-1");
|
||||
|
||||
expect(result).toEqual({
|
||||
models: catalog,
|
||||
supported: true,
|
||||
cached: true,
|
||||
cachedAt: "2026-07-29T00:00:00Z",
|
||||
});
|
||||
expect(getListModelsResult).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("marks a live discovery as not cached", async () => {
|
||||
initiateListModels.mockResolvedValue(
|
||||
request({ status: "completed", models: catalog }),
|
||||
);
|
||||
|
||||
const result = await resolveRuntimeModels("rt-1");
|
||||
|
||||
expect(result.cached).toBe(false);
|
||||
expect(result.cachedAt).toBeUndefined();
|
||||
});
|
||||
|
||||
it("still polls a pending response until the daemon reports back", async () => {
|
||||
initiateListModels.mockResolvedValue(request({ status: "pending" }));
|
||||
getListModelsResult
|
||||
.mockResolvedValueOnce(request({ status: "running" }))
|
||||
.mockResolvedValueOnce(request({ status: "completed", models: catalog }));
|
||||
|
||||
const result = await resolveRuntimeModels("rt-1");
|
||||
|
||||
expect(result.models).toEqual(catalog);
|
||||
expect(result.supported).toBe(true);
|
||||
expect(getListModelsResult).toHaveBeenCalledTimes(2);
|
||||
expect(getListModelsResult).toHaveBeenLastCalledWith("rt-1", "req-1");
|
||||
});
|
||||
|
||||
it("surfaces a failed discovery as an error", async () => {
|
||||
initiateListModels.mockResolvedValue(
|
||||
request({ status: "failed", error: "claude not installed" }),
|
||||
);
|
||||
|
||||
await expect(resolveRuntimeModels("rt-1")).rejects.toThrow(
|
||||
"claude not installed",
|
||||
);
|
||||
});
|
||||
|
||||
// A status this client does not know (newer server, or the malformed-response
|
||||
// fallback record) must NOT read as "completed with no models" — that renders
|
||||
// an authoritative-looking empty dropdown. The schema keeps `status` lenient,
|
||||
// so such a value really can reach this code at runtime even though the TS
|
||||
// union does not admit it.
|
||||
it("treats an unrecognised status as an explicit failure", async () => {
|
||||
initiateListModels.mockResolvedValue({
|
||||
...request({}),
|
||||
status: "superseded" as RuntimeModelListRequest["status"],
|
||||
});
|
||||
|
||||
await expect(resolveRuntimeModels("rt-1")).rejects.toThrow(
|
||||
/status: superseded/,
|
||||
);
|
||||
});
|
||||
|
||||
it("defaults supported to true when the server omits it", async () => {
|
||||
initiateListModels.mockResolvedValue({
|
||||
...request({ status: "completed", models: catalog }),
|
||||
supported: undefined as unknown as boolean,
|
||||
});
|
||||
|
||||
const result = await resolveRuntimeModels("rt-1");
|
||||
|
||||
expect(result.supported).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("staleTimeFor", () => {
|
||||
// The server may serve a snapshot up to its own serve window old and refresh
|
||||
// it in the background. If the client also held that response as fresh, the
|
||||
// observable staleness would be server window + client window. Zero keeps the
|
||||
// bound at the server's window alone.
|
||||
it("treats a cached answer as immediately revalidatable", () => {
|
||||
expect(staleTimeFor({ models: catalog, supported: true, cached: true })).toBe(0);
|
||||
});
|
||||
|
||||
it("trusts a live answer for the full window", () => {
|
||||
expect(staleTimeFor({ models: catalog, supported: true, cached: false })).toBe(
|
||||
LIVE_MODELS_STALE_TIME_MS,
|
||||
);
|
||||
expect(staleTimeFor({ models: catalog, supported: true })).toBe(
|
||||
LIVE_MODELS_STALE_TIME_MS,
|
||||
);
|
||||
});
|
||||
|
||||
it("has nothing to trust without data", () => {
|
||||
expect(staleTimeFor(undefined)).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe("runtimeModelsOptions", () => {
|
||||
it("keeps unused entries long enough to render instantly on return", () => {
|
||||
const options = runtimeModelsOptions("rt-1");
|
||||
expect(options.gcTime).toBeGreaterThanOrEqual(LIVE_MODELS_STALE_TIME_MS);
|
||||
expect(options.queryKey).toEqual(["runtimes", "models", "rt-1"]);
|
||||
expect(options.enabled).toBe(true);
|
||||
});
|
||||
|
||||
it("resolves freshness from the served answer, not a fixed window", () => {
|
||||
const options = runtimeModelsOptions("rt-1");
|
||||
expect(typeof options.staleTime).toBe("function");
|
||||
const staleTime = options.staleTime as (query: {
|
||||
state: { data?: { models: []; supported: boolean; cached?: boolean } };
|
||||
}) => number;
|
||||
expect(
|
||||
staleTime({ state: { data: { models: [], supported: true, cached: true } } }),
|
||||
).toBe(0);
|
||||
expect(
|
||||
staleTime({ state: { data: { models: [], supported: true, cached: false } } }),
|
||||
).toBe(LIVE_MODELS_STALE_TIME_MS);
|
||||
});
|
||||
|
||||
it("stays disabled without a runtime", () => {
|
||||
expect(runtimeModelsOptions(null).enabled).toBe(false);
|
||||
});
|
||||
|
||||
// The regression Sol-Boy flagged on PR #6098: a stale-but-served snapshot must
|
||||
// reach the SAME client once the server-side refresh lands, and the picker
|
||||
// must not blink an empty loading state while that happens.
|
||||
it("picks up the refreshed catalog on revisit without a blank loading state", async () => {
|
||||
initiateListModels
|
||||
// First open: server hands back a 14-minute-old snapshot and queues its
|
||||
// own background refresh.
|
||||
.mockResolvedValueOnce(cachedResponse(catalog, "2026-07-29T00:00:00Z"))
|
||||
// Revisit: the refresh has landed, so the same endpoint now serves the
|
||||
// new catalog.
|
||||
.mockResolvedValueOnce(
|
||||
cachedResponse(refreshedCatalog, "2026-07-29T00:14:00Z"),
|
||||
);
|
||||
|
||||
const client = new QueryClient({
|
||||
defaultOptions: { queries: { retry: false } },
|
||||
});
|
||||
const options = runtimeModelsOptions("rt-1");
|
||||
|
||||
await client.fetchQuery(options);
|
||||
expect(
|
||||
client.getQueryData(runtimeModelsKeys.forRuntime("rt-1")),
|
||||
).toMatchObject({ models: catalog, cached: true });
|
||||
|
||||
// A cached answer is stale on arrival, so remounting the picker revalidates.
|
||||
const query = client
|
||||
.getQueryCache()
|
||||
.find<RuntimeModelsResult>({
|
||||
queryKey: runtimeModelsKeys.forRuntime("rt-1"),
|
||||
})!;
|
||||
expect(query.isStaleByTime(staleTimeFor(query.state.data))).toBe(true);
|
||||
|
||||
const observer = new QueryObserver(client, options);
|
||||
const emissions: { isLoading: boolean; ids: string[] }[] = [];
|
||||
const unsubscribe = observer.subscribe((result) => {
|
||||
emissions.push({
|
||||
isLoading: result.isLoading,
|
||||
ids: (result.data?.models ?? []).map((m) => m.id),
|
||||
});
|
||||
});
|
||||
|
||||
try {
|
||||
await vi.waitFor(() => {
|
||||
const last = emissions.at(-1);
|
||||
expect(last?.ids).toEqual(refreshedCatalog.map((m) => m.id));
|
||||
});
|
||||
} finally {
|
||||
unsubscribe();
|
||||
client.clear();
|
||||
}
|
||||
|
||||
// Every emission during the background revalidation kept a rendered
|
||||
// catalog: the pickers gate their spinner on `isLoading`, so this is what
|
||||
// "no blank loading state" means concretely.
|
||||
expect(emissions.every((e) => e.isLoading === false)).toBe(true);
|
||||
expect(emissions.every((e) => e.ids.length > 0)).toBe(true);
|
||||
// The first emission is the stale snapshot still on screen while the
|
||||
// revalidation runs — proof the new catalog arrived by replacing rendered
|
||||
// data, not after a gap.
|
||||
expect(emissions[0]?.ids).toEqual(catalog.map((m) => m.id));
|
||||
expect(initiateListModels).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
@@ -11,6 +11,17 @@ export const runtimeModelsKeys = {
|
||||
const POLL_INTERVAL_MS = 500;
|
||||
const POLL_TIMEOUT_MS = 30_000;
|
||||
|
||||
// How long a LIVE discovery result (one the daemon just produced) is trusted
|
||||
// without re-asking. Discovery is a round trip to the user's machine, and a
|
||||
// catalog only changes when they upgrade a CLI, switch accounts, or edit a
|
||||
// provider config, so re-running it inside one form session is pure waste.
|
||||
export const LIVE_MODELS_STALE_TIME_MS = 5 * 60_000;
|
||||
// Kept long so a runtime revisited later in the same session renders from the
|
||||
// client cache instead of the "discovering models" spinner. Freshness is
|
||||
// governed by staleTime; gcTime only decides how long the entry survives while
|
||||
// unused.
|
||||
export const MODELS_GC_TIME_MS = 30 * 60_000;
|
||||
|
||||
// resolveRuntimeModels initiates a list-models request against the daemon
|
||||
// (via heartbeat piggyback) and polls until the daemon reports back or
|
||||
// the request times out. Returns both the models list and a
|
||||
@@ -18,6 +29,11 @@ const POLL_TIMEOUT_MS = 30_000;
|
||||
// per-agent model selection entirely (hermes today) — the UI uses
|
||||
// this to disable its dropdown instead of accepting a value that
|
||||
// wouldn't be honoured at runtime.
|
||||
//
|
||||
// `cached` reports that the server answered from its catalog cache rather than
|
||||
// a live daemon round trip (MUL-5444). It is not cosmetic: it feeds the
|
||||
// staleTime policy below so the client never extends the server's staleness
|
||||
// window past what the server itself promises.
|
||||
export async function resolveRuntimeModels(
|
||||
runtimeId: string,
|
||||
): Promise<RuntimeModelsResult> {
|
||||
@@ -31,10 +47,40 @@ export async function resolveRuntimeModels(
|
||||
await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
|
||||
current = await api.getListModelsResult(runtimeId, initial.id);
|
||||
}
|
||||
if (current.status === "failed" || current.status === "timeout") {
|
||||
throw new Error(current.error || "model discovery failed");
|
||||
// Only an explicit `completed` is a catalog. Anything else — failed, timeout,
|
||||
// or a status this client does not know (newer server, or a response that fell
|
||||
// back to the malformed-record shape) — is surfaced as an error so the picker
|
||||
// shows "discovery failed" and keeps manual entry available. Treating an
|
||||
// unrecognised status as success would render an empty dropdown that looks
|
||||
// authoritative.
|
||||
if (current.status !== "completed") {
|
||||
throw new Error(
|
||||
current.error || `model discovery failed (status: ${current.status})`,
|
||||
);
|
||||
}
|
||||
return { models: current.models ?? [], supported: current.supported };
|
||||
return {
|
||||
models: current.models ?? [],
|
||||
supported: current.supported !== false,
|
||||
cached: current.cached === true,
|
||||
cachedAt: current.cached_at,
|
||||
};
|
||||
}
|
||||
|
||||
// staleTimeFor is the freshness policy, split out so it can be unit-tested
|
||||
// without a QueryClient.
|
||||
//
|
||||
// A cached answer is treated as stale immediately. The server serves a snapshot
|
||||
// up to `modelCatalogServeWindow` old and queues its own background refresh;
|
||||
// if the client ALSO held that response as fresh for minutes, the observable
|
||||
// staleness would be server window + client window (and the refreshed catalog
|
||||
// would never reach the tab that triggered the refresh). Returning 0 keeps the
|
||||
// bound at the server's window alone: the next mount / focus revalidates, the
|
||||
// refreshed snapshot is picked up, and because the query already holds data
|
||||
// React Query refetches in the background — `isLoading` stays false, so no
|
||||
// picker flashes an empty loading state (MUL-5444).
|
||||
export function staleTimeFor(data: RuntimeModelsResult | undefined): number {
|
||||
if (!data) return 0;
|
||||
return data.cached ? 0 : LIVE_MODELS_STALE_TIME_MS;
|
||||
}
|
||||
|
||||
export function runtimeModelsOptions(runtimeId: string | null | undefined) {
|
||||
@@ -44,9 +90,8 @@ export function runtimeModelsOptions(runtimeId: string | null | undefined) {
|
||||
: runtimeModelsKeys.all(),
|
||||
queryFn: () => resolveRuntimeModels(runtimeId as string),
|
||||
enabled: Boolean(runtimeId),
|
||||
// Models rarely change; cache for 60s to match the server-side
|
||||
// cache in agent.ListModels.
|
||||
staleTime: 60_000,
|
||||
staleTime: (query) => staleTimeFor(query.state.data),
|
||||
gcTime: MODELS_GC_TIME_MS,
|
||||
retry: false,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1042,6 +1042,15 @@ export interface RuntimeModelListRequest {
|
||||
error?: string;
|
||||
created_at: string;
|
||||
updated_at: string;
|
||||
/**
|
||||
* True when the server answered from its own catalog cache instead of a live
|
||||
* daemon round trip (MUL-5444). Informational only: such a response already
|
||||
* arrives with `status: "completed"` and a populated `models`, so callers
|
||||
* that ignore this field behave exactly as before. `cached_at` is the
|
||||
* snapshot's capture time.
|
||||
*/
|
||||
cached?: boolean;
|
||||
cached_at?: string;
|
||||
}
|
||||
|
||||
// Result shape returned by resolveRuntimeModels — includes the
|
||||
@@ -1050,6 +1059,15 @@ export interface RuntimeModelListRequest {
|
||||
export interface RuntimeModelsResult {
|
||||
models: RuntimeModel[];
|
||||
supported: boolean;
|
||||
/**
|
||||
* True when the server answered from its catalog cache rather than a live
|
||||
* daemon round trip (MUL-5444). Drives the query's freshness policy: a
|
||||
* cached answer is immediately revalidatable so the client never extends the
|
||||
* server's staleness window.
|
||||
*/
|
||||
cached?: boolean;
|
||||
/** Capture time of the served snapshot, when the answer was cached. */
|
||||
cachedAt?: string;
|
||||
}
|
||||
|
||||
export type RuntimeLocalSkillStatus =
|
||||
|
||||
@@ -239,10 +239,14 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus
|
||||
if notifier, ok := opts.DaemonWakeup.(handler.WorkspaceSetRefreshNotifier); ok {
|
||||
h.DaemonWorkspaceRefresh = notifier
|
||||
}
|
||||
if notifier, ok := opts.DaemonWakeup.(handler.DaemonPendingWorkNotifier); ok {
|
||||
h.DaemonPendingWork = notifier
|
||||
}
|
||||
}
|
||||
if rdb != nil {
|
||||
h.UpdateStore = handler.NewRedisUpdateStore(rdb)
|
||||
h.ModelListStore = handler.NewRedisModelListStore(rdb)
|
||||
h.ModelCatalogCache = handler.NewRedisModelCatalogCache(rdb)
|
||||
h.LocalSkillListStore = handler.NewRedisLocalSkillListStore(rdb)
|
||||
h.LocalSkillImportStore = handler.NewRedisLocalSkillImportStore(rdb)
|
||||
h.LivenessStore = handler.NewRedisLivenessStore(rdb)
|
||||
|
||||
@@ -74,8 +74,22 @@ const (
|
||||
// intentionally independent from AgentTimeout, which only governs the
|
||||
// provider process after the task reaches running.
|
||||
defaultTaskPrepareTimeout = 5 * time.Minute
|
||||
// pendingWorkHeartbeatTimeout bounds the out-of-band heartbeat a
|
||||
// server-pushed daemon:pending_work hint triggers (MUL-5444). Short on
|
||||
// purpose: the hint is only a latency optimisation, and the scheduled
|
||||
// heartbeat still picks the request up if this attempt fails.
|
||||
pendingWorkHeartbeatTimeout = 15 * time.Second
|
||||
// pendingWorkHintBookkeepingTTL is how long a runtime's last-hint timestamp
|
||||
// is retained before it is swept — purely to keep the map bounded.
|
||||
pendingWorkHintBookkeepingTTL = 10 * time.Minute
|
||||
)
|
||||
|
||||
// pendingWorkHintMinInterval is the floor between two hint-driven heartbeats
|
||||
// for the same runtime. Keeps an interactive first open instant while stopping a
|
||||
// caller-triggered hint from becoming a heartbeat amplifier. A var so tests can
|
||||
// shrink it, same as the other timing knobs in this package.
|
||||
var pendingWorkHintMinInterval = time.Second
|
||||
|
||||
func repoCheckoutModeFor(provider, goos string) string {
|
||||
if provider == "codex" && goos == "linux" {
|
||||
return repoCheckoutModeIsolated
|
||||
@@ -332,6 +346,16 @@ type Daemon struct {
|
||||
reregisterNextAttempt map[string]time.Time // workspace_id -> earliest time the next re-register attempt may run
|
||||
reregisterLastCompletedAt map[string]time.Time // workspace_id -> wall-clock at which the last SUCCESSFUL re-register call returned (failures intentionally not stamped — see recordRegisterCompletion)
|
||||
|
||||
// pendingWorkMu guards pendingWorkInflight and pendingWorkLastRun, which
|
||||
// coalesce and rate-limit server-pushed "heartbeat now" hints (MUL-5444).
|
||||
// Several UI surfaces can request the same runtime's model list within
|
||||
// milliseconds; without the guard each hint would fire its own out-of-band
|
||||
// heartbeat, and an authenticated caller looping the list-models endpoint
|
||||
// could turn that into a heartbeat amplifier.
|
||||
pendingWorkMu sync.Mutex
|
||||
pendingWorkInflight map[string]struct{} // runtime_id -> hint-driven heartbeat in flight
|
||||
pendingWorkLastRun map[string]time.Time // runtime_id -> when the last hint-driven heartbeat started
|
||||
|
||||
cancelFunc context.CancelFunc // set by Run(); called by triggerRestart
|
||||
rootCtx context.Context // set by Run(); used by long-running recoveries that must survive per-runtime ctx cancellation
|
||||
restartBinary string // non-empty after a successful update; path to the new binary
|
||||
@@ -428,6 +452,8 @@ func New(cfg Config, logger *slog.Logger) *Daemon {
|
||||
deletingCodexStores: make(map[string]bool),
|
||||
localPathLocks: NewLocalPathLocker(),
|
||||
runtimeGoneInflight: make(map[string]struct{}),
|
||||
pendingWorkInflight: make(map[string]struct{}),
|
||||
pendingWorkLastRun: make(map[string]time.Time),
|
||||
reregisterNextAttempt: make(map[string]time.Time),
|
||||
reregisterLastCompletedAt: make(map[string]time.Time),
|
||||
cancelPollInterval: 5 * time.Second,
|
||||
@@ -2784,6 +2810,98 @@ func (d *Daemon) handleHeartbeatActions(ctx context.Context, runtimeID string, r
|
||||
}
|
||||
}
|
||||
|
||||
// handlePendingWorkHint reacts to a server-pushed daemon:pending_work frame by
|
||||
// sending ONE immediate heartbeat for the runtime, then dispatching whatever
|
||||
// that heartbeat claimed (MUL-5444).
|
||||
//
|
||||
// Why a heartbeat and not the work itself: the hint deliberately carries no
|
||||
// request payload, so the server never has to un-claim anything when delivery
|
||||
// fails, and a duplicated or replayed hint cannot duplicate work — the claim
|
||||
// stays atomic inside the store's PopPending. A hint that never arrives (daemon
|
||||
// offline, WS gap, relay drop) costs nothing but the pre-existing wait for the
|
||||
// next scheduled heartbeat.
|
||||
//
|
||||
// The HTTP heartbeat is used on purpose rather than queueing a WS frame: the
|
||||
// hint arrives on the read pump, the WS write path may be backed up or tearing
|
||||
// down, and this is a human-interactive, low-frequency path (a user opened a
|
||||
// model picker) where one extra request is cheaper than a missed wakeup. Note it
|
||||
// intentionally bypasses the wsHeartbeatRecentlyAcked suppression that the
|
||||
// scheduled HTTP tick honours — that suppression exists to avoid duplicate
|
||||
// periodic writes, not to block an explicitly requested pull.
|
||||
func (d *Daemon) handlePendingWorkHint(runtimeID, kind string) {
|
||||
if runtimeID == "" {
|
||||
return
|
||||
}
|
||||
if d.findRuntime(runtimeID) == nil {
|
||||
// Not one of ours (stale relay fanout, or the runtime was just pruned).
|
||||
return
|
||||
}
|
||||
|
||||
d.pendingWorkMu.Lock()
|
||||
if d.pendingWorkInflight == nil {
|
||||
d.pendingWorkInflight = make(map[string]struct{})
|
||||
}
|
||||
if d.pendingWorkLastRun == nil {
|
||||
d.pendingWorkLastRun = make(map[string]time.Time)
|
||||
}
|
||||
// Drop long-idle bookkeeping so the map can't grow with every runtime this
|
||||
// process has ever seen.
|
||||
for id, at := range d.pendingWorkLastRun {
|
||||
if time.Since(at) > pendingWorkHintBookkeepingTTL {
|
||||
delete(d.pendingWorkLastRun, id)
|
||||
}
|
||||
}
|
||||
if _, inflight := d.pendingWorkInflight[runtimeID]; inflight {
|
||||
d.pendingWorkMu.Unlock()
|
||||
return
|
||||
}
|
||||
// Rate limit per runtime. The hint is caller-triggered (any workspace member
|
||||
// hitting the list-models endpoint), so without a floor a request loop would
|
||||
// become a heartbeat amplifier. A suppressed hint costs nothing but the
|
||||
// pre-existing wait for the scheduled heartbeat.
|
||||
if last, ok := d.pendingWorkLastRun[runtimeID]; ok && time.Since(last) < pendingWorkHintMinInterval {
|
||||
d.pendingWorkMu.Unlock()
|
||||
d.logger.Debug("pending work hint throttled", "runtime_id", runtimeID, "kind", kind)
|
||||
return
|
||||
}
|
||||
d.pendingWorkInflight[runtimeID] = struct{}{}
|
||||
d.pendingWorkLastRun[runtimeID] = time.Now()
|
||||
d.pendingWorkMu.Unlock()
|
||||
defer func() {
|
||||
d.pendingWorkMu.Lock()
|
||||
delete(d.pendingWorkInflight, runtimeID)
|
||||
d.pendingWorkMu.Unlock()
|
||||
}()
|
||||
|
||||
// Root context: handleHeartbeatActions hands this ctx to the actual work
|
||||
// (model discovery shells out to a CLI for up to ~15s), so it must outlive
|
||||
// this function. Only the heartbeat request itself is time-bounded.
|
||||
ctx := d.recoveryContext()
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
hbCtx, cancel := context.WithTimeout(ctx, pendingWorkHeartbeatTimeout)
|
||||
resp, err := d.client.SendHeartbeat(hbCtx, runtimeID)
|
||||
cancel()
|
||||
if err != nil {
|
||||
if isRuntimeNotFoundError(err) {
|
||||
go d.handleRuntimeGone(runtimeID)
|
||||
return
|
||||
}
|
||||
d.logger.Debug("pending work hint heartbeat failed", "runtime_id", runtimeID, "kind", kind, "error", err)
|
||||
return
|
||||
}
|
||||
if resp == nil {
|
||||
return
|
||||
}
|
||||
if resp.RuntimeGone {
|
||||
go d.handleRuntimeGone(runtimeID)
|
||||
return
|
||||
}
|
||||
d.logger.Debug("pending work hint served", "runtime_id", runtimeID, "kind", kind)
|
||||
d.handleHeartbeatActions(ctx, runtimeID, resp)
|
||||
}
|
||||
|
||||
// handleModelList resolves the provider's supported models (via static
|
||||
// catalog or by shelling out to the agent CLI) and reports the result
|
||||
// back to the server. Model discovery failures are reported as empty
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/multica-ai/multica/server/pkg/protocol"
|
||||
)
|
||||
|
||||
// pendingWorkHintDaemon builds a Daemon that knows exactly one runtime and
|
||||
// talks to an httptest server, so the hint path is exercised against the real
|
||||
// daemon.Client instead of a mock. The per-runtime hint floor is disabled by
|
||||
// default; TestHandlePendingWorkHint_ThrottlesRepeatHints restores it.
|
||||
func pendingWorkHintDaemon(t *testing.T, handler http.HandlerFunc) (*Daemon, *int32) {
|
||||
t.Helper()
|
||||
withPendingWorkHintMinInterval(t, 0)
|
||||
var calls int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
atomic.AddInt32(&calls, 1)
|
||||
handler(w, r)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
d := &Daemon{
|
||||
cfg: Config{HeartbeatInterval: time.Hour},
|
||||
client: NewClient(srv.URL),
|
||||
logger: slog.Default(),
|
||||
runtimeIndex: map[string]Runtime{"rt-1": {ID: "rt-1", Provider: "claude"}},
|
||||
pendingWorkInflight: make(map[string]struct{}),
|
||||
pendingWorkLastRun: make(map[string]time.Time),
|
||||
}
|
||||
return d, &calls
|
||||
}
|
||||
|
||||
func withPendingWorkHintMinInterval(t *testing.T, d time.Duration) {
|
||||
t.Helper()
|
||||
prev := pendingWorkHintMinInterval
|
||||
pendingWorkHintMinInterval = d
|
||||
t.Cleanup(func() { pendingWorkHintMinInterval = prev })
|
||||
}
|
||||
|
||||
// TestHandlePendingWorkHint_SendsImmediateHeartbeat is the core of MUL-5444:
|
||||
// a server-pushed hint must produce a heartbeat right now instead of leaving the
|
||||
// queued model-list request to wait for the next scheduled tick (up to a full
|
||||
// HeartbeatInterval, 15s by default).
|
||||
func TestHandlePendingWorkHint_SendsImmediateHeartbeat(t *testing.T) {
|
||||
var gotPath, gotRuntime string
|
||||
d, calls := pendingWorkHintDaemon(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
gotPath = r.URL.Path
|
||||
var body struct {
|
||||
RuntimeID string `json:"runtime_id"`
|
||||
}
|
||||
_ = json.NewDecoder(r.Body).Decode(&body)
|
||||
gotRuntime = body.RuntimeID
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte(`{"runtime_id":"rt-1","status":"ok"}`))
|
||||
})
|
||||
|
||||
d.handlePendingWorkHint("rt-1", protocol.PendingWorkKindModelList)
|
||||
|
||||
if got := atomic.LoadInt32(calls); got != 1 {
|
||||
t.Fatalf("expected exactly 1 heartbeat, got %d", got)
|
||||
}
|
||||
if gotPath != "/api/daemon/heartbeat" {
|
||||
t.Fatalf("heartbeat path = %q", gotPath)
|
||||
}
|
||||
if gotRuntime != "rt-1" {
|
||||
t.Fatalf("heartbeat runtime_id = %q, want rt-1", gotRuntime)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHandlePendingWorkHint_IgnoresUnknownRuntime keeps a stale or
|
||||
// cross-machine relay fanout from making this daemon heartbeat for a runtime it
|
||||
// does not own.
|
||||
func TestHandlePendingWorkHint_IgnoresUnknownRuntime(t *testing.T) {
|
||||
d, calls := pendingWorkHintDaemon(t, func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte(`{"runtime_id":"other","status":"ok"}`))
|
||||
})
|
||||
|
||||
d.handlePendingWorkHint("not-mine", protocol.PendingWorkKindModelList)
|
||||
d.handlePendingWorkHint("", protocol.PendingWorkKindModelList)
|
||||
|
||||
if got := atomic.LoadInt32(calls); got != 0 {
|
||||
t.Fatalf("expected no heartbeat for an unknown runtime, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHandlePendingWorkHint_CoalescesConcurrentHints pins the stampede guard:
|
||||
// several UI surfaces (model picker, thinking level, service tier) each request
|
||||
// the catalog for the same runtime within milliseconds, and one heartbeat
|
||||
// already claims whatever is queued.
|
||||
func TestHandlePendingWorkHint_CoalescesConcurrentHints(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
arrived := make(chan struct{}, 1)
|
||||
d, calls := pendingWorkHintDaemon(t, func(w http.ResponseWriter, _ *http.Request) {
|
||||
select {
|
||||
case arrived <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
<-release
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte(`{"runtime_id":"rt-1","status":"ok"}`))
|
||||
})
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
d.handlePendingWorkHint("rt-1", protocol.PendingWorkKindModelList)
|
||||
}()
|
||||
|
||||
// Wait until the first hint is actually inside the HTTP call, then fire the
|
||||
// siblings that must be dropped.
|
||||
select {
|
||||
case <-arrived:
|
||||
case <-time.After(2 * time.Second):
|
||||
close(release)
|
||||
wg.Wait()
|
||||
t.Fatal("first hint never reached the server")
|
||||
}
|
||||
for i := 0; i < 3; i++ {
|
||||
d.handlePendingWorkHint("rt-1", protocol.PendingWorkKindModelList)
|
||||
}
|
||||
close(release)
|
||||
wg.Wait()
|
||||
|
||||
if got := atomic.LoadInt32(calls); got != 1 {
|
||||
t.Fatalf("expected concurrent hints to coalesce into 1 heartbeat, got %d", got)
|
||||
}
|
||||
|
||||
// The guard must release afterwards, otherwise the runtime would ignore
|
||||
// every later hint for the lifetime of the process.
|
||||
d.handlePendingWorkHint("rt-1", protocol.PendingWorkKindModelList)
|
||||
if got := atomic.LoadInt32(calls); got != 2 {
|
||||
t.Fatalf("expected a later hint to heartbeat again, got %d total", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHandlePendingWorkHint_ThrottlesRepeatHints pins the amplification guard:
|
||||
// the hint is triggered by any workspace member hitting the list-models
|
||||
// endpoint, so back-to-back requests must not turn into back-to-back heartbeats.
|
||||
func TestHandlePendingWorkHint_ThrottlesRepeatHints(t *testing.T) {
|
||||
d, calls := pendingWorkHintDaemon(t, func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write([]byte(`{"runtime_id":"rt-1","status":"ok"}`))
|
||||
})
|
||||
withPendingWorkHintMinInterval(t, time.Minute)
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
d.handlePendingWorkHint("rt-1", protocol.PendingWorkKindModelList)
|
||||
}
|
||||
|
||||
if got := atomic.LoadInt32(calls); got != 1 {
|
||||
t.Fatalf("expected repeat hints inside the floor to collapse into 1 heartbeat, got %d", got)
|
||||
}
|
||||
|
||||
// Past the floor, a hint is served again — the throttle is a floor, not a
|
||||
// one-shot latch.
|
||||
withPendingWorkHintMinInterval(t, time.Nanosecond)
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
d.handlePendingWorkHint("rt-1", protocol.PendingWorkKindModelList)
|
||||
if got := atomic.LoadInt32(calls); got != 2 {
|
||||
t.Fatalf("expected a hint past the floor to heartbeat, got %d total", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHandlePendingWorkHint_SurvivesHeartbeatFailure documents that a failed
|
||||
// hint is a no-op: the scheduled heartbeat remains the correctness path, so the
|
||||
// daemon must not panic or retry-storm here.
|
||||
func TestHandlePendingWorkHint_SurvivesHeartbeatFailure(t *testing.T) {
|
||||
d, calls := pendingWorkHintDaemon(t, func(w http.ResponseWriter, _ *http.Request) {
|
||||
http.Error(w, `{"error":"boom"}`, http.StatusInternalServerError)
|
||||
})
|
||||
|
||||
d.handlePendingWorkHint("rt-1", protocol.PendingWorkKindModelList)
|
||||
|
||||
if got := atomic.LoadInt32(calls); got != 1 {
|
||||
t.Fatalf("expected exactly 1 attempt, got %d", got)
|
||||
}
|
||||
}
|
||||
@@ -402,6 +402,19 @@ func (d *Daemon) readTaskWakeupMessagesForConnection(conn *websocket.Conn, taskW
|
||||
if d.workspaceChanges != nil {
|
||||
d.workspaceChanges.broadcast()
|
||||
}
|
||||
case protocol.EventDaemonPendingWork:
|
||||
var payload protocol.PendingWorkPayload
|
||||
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
|
||||
d.logger.Debug("pending work websocket invalid payload", "error", err)
|
||||
continue
|
||||
}
|
||||
if payload.RuntimeID == "" {
|
||||
d.logger.Debug("pending work websocket missing runtime_id")
|
||||
continue
|
||||
}
|
||||
// Own goroutine: the hint triggers an HTTP heartbeat plus the work it
|
||||
// claims, and the read pump must stay free for the next frame.
|
||||
go d.handlePendingWorkHint(payload.RuntimeID, payload.Kind)
|
||||
case protocol.EventDaemonHeartbeatAck:
|
||||
var ack HeartbeatResponse
|
||||
if err := json.Unmarshal(msg.Payload, &ack); err != nil {
|
||||
|
||||
@@ -328,6 +328,14 @@ func (h *Hub) NotifyWorkspacesChanged(userID string) {
|
||||
h.notifyWorkspacesChanged(userID, "")
|
||||
}
|
||||
|
||||
// NotifyPendingWork tells daemons watching runtimeID that a heartbeat-carried
|
||||
// request is queued, so they can heartbeat now instead of waiting for the next
|
||||
// scheduled tick (MUL-5444). Best-effort like every other hub notification: the
|
||||
// daemon's own heartbeat schedule remains the correctness path.
|
||||
func (h *Hub) NotifyPendingWork(runtimeID, kind string) {
|
||||
h.notifyPendingWork(runtimeID, kind, "")
|
||||
}
|
||||
|
||||
func (h *Hub) notifyTaskAvailable(runtimeID, taskID, eventID string) {
|
||||
if h == nil || runtimeID == "" {
|
||||
return
|
||||
@@ -366,6 +374,22 @@ func (h *Hub) notifyWorkspacesChanged(userID, eventID string) {
|
||||
h.notifyUserFrame(userID, data, eventID)
|
||||
}
|
||||
|
||||
func (h *Hub) notifyPendingWork(runtimeID, kind, eventID string) {
|
||||
if h == nil || runtimeID == "" {
|
||||
return
|
||||
}
|
||||
data, err := pendingWorkFrame(runtimeID, kind)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
delivered, deduped := h.notifyFrame(runtimeID, data, eventID)
|
||||
if delivered {
|
||||
M.WakeupDeliveredHit.Add(1)
|
||||
} else if !deduped {
|
||||
M.WakeupDeliveredMiss.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Hub) DeliverDaemonRuntime(scopeID string, frame []byte, eventID string) {
|
||||
if h == nil {
|
||||
return
|
||||
@@ -411,6 +435,19 @@ func (h *Hub) DeliverDaemonRuntime(scopeID string, frame []byte, eventID string)
|
||||
} else if !deduped {
|
||||
M.WakeupDeliveredMiss.Add(1)
|
||||
}
|
||||
case protocol.EventDaemonPendingWork:
|
||||
var payload protocol.PendingWorkPayload
|
||||
if err := json.Unmarshal(msg.Payload, &payload); err != nil || payload.RuntimeID == "" {
|
||||
slog.Debug("daemon websocket relay: invalid pending_work payload", "error", err, "scope_id", scopeID, "event_id", eventID)
|
||||
M.WakeupDeliveredMiss.Add(1)
|
||||
return
|
||||
}
|
||||
delivered, deduped := h.notifyFrame(payload.RuntimeID, frame, eventID)
|
||||
if delivered {
|
||||
M.WakeupDeliveredHit.Add(1)
|
||||
} else if !deduped {
|
||||
M.WakeupDeliveredMiss.Add(1)
|
||||
}
|
||||
default:
|
||||
M.WakeupDeliveredMiss.Add(1)
|
||||
return
|
||||
@@ -528,6 +565,16 @@ func workspacesChangedFrame() ([]byte, error) {
|
||||
})
|
||||
}
|
||||
|
||||
func pendingWorkFrame(runtimeID, kind string) ([]byte, error) {
|
||||
return json.Marshal(protocol.Message{
|
||||
Type: protocol.EventDaemonPendingWork,
|
||||
Payload: mustMarshalRaw(protocol.PendingWorkPayload{
|
||||
RuntimeID: runtimeID,
|
||||
Kind: kind,
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
func mustMarshalRaw(v any) json.RawMessage {
|
||||
data, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
|
||||
@@ -99,3 +99,32 @@ func (n *RelayNotifier) NotifyWorkspacesChanged(userID string) {
|
||||
}
|
||||
M.WakeupPublishedTotal.Add(1)
|
||||
}
|
||||
|
||||
// NotifyPendingWork fans a runtime-scoped "heartbeat now" hint out to the local
|
||||
// hub and, when Redis is configured, through the relay so the API node that
|
||||
// actually holds the daemon's WebSocket delivers it (MUL-5444). Shard key is the
|
||||
// runtime ID: hints for one runtime stay ordered relative to each other, and a
|
||||
// dropped hint only costs the daemon its normal heartbeat delay.
|
||||
func (n *RelayNotifier) NotifyPendingWork(runtimeID, kind string) {
|
||||
if runtimeID == "" {
|
||||
return
|
||||
}
|
||||
eventID := ulid.Make().String()
|
||||
if n.local != nil {
|
||||
n.local.notifyPendingWork(runtimeID, kind, eventID)
|
||||
}
|
||||
if n.relay == nil {
|
||||
return
|
||||
}
|
||||
frame, err := pendingWorkFrame(runtimeID, kind)
|
||||
if err != nil {
|
||||
M.WakeupPublishErrors.Add(1)
|
||||
return
|
||||
}
|
||||
if err := n.relay.PublishWithID(realtime.ScopeDaemonRuntime, runtimeID, "", frame, eventID); err != nil {
|
||||
M.WakeupPublishErrors.Add(1)
|
||||
slog.Warn("daemon websocket pending work publish failed", "error", err, "runtime_id", runtimeID, "kind", kind)
|
||||
return
|
||||
}
|
||||
M.WakeupPublishedTotal.Add(1)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
package daemonws
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
"github.com/multica-ai/multica/server/pkg/protocol"
|
||||
)
|
||||
|
||||
// TestNotifyPendingWork pins the hint that removes the up-to-one-heartbeat
|
||||
// wait before a queued model-list request is picked up (MUL-5444): daemons
|
||||
// watching the runtime must receive a runtime-scoped daemon:pending_work frame.
|
||||
func TestNotifyPendingWork(t *testing.T) {
|
||||
M.Reset()
|
||||
defer M.Reset()
|
||||
|
||||
hub := NewHub()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hub.HandleWebSocket(w, r, ClientIdentity{RuntimeIDs: []string{"runtime-1"}})
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
conn := dialPendingWorkClient(t, hub, server.URL, "runtime-1")
|
||||
defer conn.Close()
|
||||
|
||||
hub.NotifyPendingWork("runtime-1", protocol.PendingWorkKindModelList)
|
||||
|
||||
payload := readPendingWorkFrame(t, conn)
|
||||
if payload.RuntimeID != "runtime-1" {
|
||||
t.Fatalf("payload.RuntimeID = %q, want runtime-1", payload.RuntimeID)
|
||||
}
|
||||
if payload.Kind != protocol.PendingWorkKindModelList {
|
||||
t.Fatalf("payload.Kind = %q, want %q", payload.Kind, protocol.PendingWorkKindModelList)
|
||||
}
|
||||
}
|
||||
|
||||
// TestDeliverDaemonRuntime_PendingWork covers the multi-node path: the node
|
||||
// that holds the daemon's socket receives the frame off the Redis relay and has
|
||||
// to route it by the payload's runtime_id, not by the relay shard key.
|
||||
func TestDeliverDaemonRuntime_PendingWork(t *testing.T) {
|
||||
M.Reset()
|
||||
defer M.Reset()
|
||||
|
||||
hub := NewHub()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hub.HandleWebSocket(w, r, ClientIdentity{RuntimeIDs: []string{"runtime-1"}})
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
conn := dialPendingWorkClient(t, hub, server.URL, "runtime-1")
|
||||
defer conn.Close()
|
||||
|
||||
frame, err := pendingWorkFrame("runtime-1", protocol.PendingWorkKindModelList)
|
||||
if err != nil {
|
||||
t.Fatalf("pendingWorkFrame: %v", err)
|
||||
}
|
||||
hub.DeliverDaemonRuntime("runtime-1", frame, "event-1")
|
||||
|
||||
payload := readPendingWorkFrame(t, conn)
|
||||
if payload.RuntimeID != "runtime-1" {
|
||||
t.Fatalf("payload.RuntimeID = %q, want runtime-1", payload.RuntimeID)
|
||||
}
|
||||
|
||||
// Same event id must not be delivered twice — a mirrored relay can publish
|
||||
// the same hint through two paths.
|
||||
hub.DeliverDaemonRuntime("runtime-1", frame, "event-1")
|
||||
if err := conn.SetReadDeadline(time.Now().Add(150 * time.Millisecond)); err != nil {
|
||||
t.Fatalf("SetReadDeadline: %v", err)
|
||||
}
|
||||
if _, _, err := conn.ReadMessage(); err == nil {
|
||||
t.Fatal("expected no second delivery for a duplicate event id")
|
||||
}
|
||||
}
|
||||
|
||||
// TestNotifyPendingWork_IgnoresEmptyRuntime guards against a hint with no
|
||||
// runtime scope fanning out to every connected daemon.
|
||||
func TestNotifyPendingWork_IgnoresEmptyRuntime(t *testing.T) {
|
||||
M.Reset()
|
||||
defer M.Reset()
|
||||
|
||||
hub := NewHub()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hub.HandleWebSocket(w, r, ClientIdentity{RuntimeIDs: []string{"runtime-1"}})
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
conn := dialPendingWorkClient(t, hub, server.URL, "runtime-1")
|
||||
defer conn.Close()
|
||||
|
||||
hub.NotifyPendingWork("", protocol.PendingWorkKindModelList)
|
||||
|
||||
if err := conn.SetReadDeadline(time.Now().Add(150 * time.Millisecond)); err != nil {
|
||||
t.Fatalf("SetReadDeadline: %v", err)
|
||||
}
|
||||
if _, _, err := conn.ReadMessage(); err == nil {
|
||||
t.Fatal("expected no frame for an empty runtime id")
|
||||
}
|
||||
}
|
||||
|
||||
func dialPendingWorkClient(t *testing.T, hub *Hub, serverURL, runtimeID string) *websocket.Conn {
|
||||
t.Helper()
|
||||
wsURL := "ws" + strings.TrimPrefix(serverURL, "http")
|
||||
conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("Dial: %v", err)
|
||||
}
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for hub.RuntimeConnectionCount(runtimeID) == 0 {
|
||||
if time.Now().After(deadline) {
|
||||
conn.Close()
|
||||
t.Fatalf("runtime connection %s was not registered", runtimeID)
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
return conn
|
||||
}
|
||||
|
||||
func readPendingWorkFrame(t *testing.T, conn *websocket.Conn) protocol.PendingWorkPayload {
|
||||
t.Helper()
|
||||
if err := conn.SetReadDeadline(time.Now().Add(time.Second)); err != nil {
|
||||
t.Fatalf("SetReadDeadline: %v", err)
|
||||
}
|
||||
_, raw, err := conn.ReadMessage()
|
||||
if err != nil {
|
||||
t.Fatalf("ReadMessage: %v", err)
|
||||
}
|
||||
var msg protocol.Message
|
||||
if err := json.Unmarshal(raw, &msg); err != nil {
|
||||
t.Fatalf("unmarshal message: %v", err)
|
||||
}
|
||||
if msg.Type != protocol.EventDaemonPendingWork {
|
||||
t.Fatalf("message type = %q, want %q", msg.Type, protocol.EventDaemonPendingWork)
|
||||
}
|
||||
var payload protocol.PendingWorkPayload
|
||||
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
|
||||
t.Fatalf("unmarshal payload: %v", err)
|
||||
}
|
||||
return payload
|
||||
}
|
||||
@@ -139,6 +139,15 @@ type WorkspaceSetRefreshNotifier interface {
|
||||
NotifyWorkspacesChanged(userID string)
|
||||
}
|
||||
|
||||
// DaemonPendingWorkNotifier pushes a runtime-scoped "heartbeat now" hint to the
|
||||
// daemon so a queued heartbeat-carried request (model discovery) is picked up
|
||||
// immediately instead of on the daemon's next scheduled tick (MUL-5444).
|
||||
// Satisfied by both *daemonws.Hub (single-node) and *daemonws.RelayNotifier
|
||||
// (multi-node, fans out through Redis).
|
||||
type DaemonPendingWorkNotifier interface {
|
||||
NotifyPendingWork(runtimeID, kind string)
|
||||
}
|
||||
|
||||
type Handler struct {
|
||||
Queries *db.Queries
|
||||
DB dbExecutor
|
||||
@@ -162,6 +171,16 @@ type Handler struct {
|
||||
Storage storage.Storage
|
||||
CFSigner *auth.CloudFrontSigner
|
||||
Analytics analytics.Client
|
||||
// DaemonPendingWork pushes "heartbeat now" hints for queued
|
||||
// heartbeat-carried requests (MUL-5444). Optional: when nil,
|
||||
// requestDaemonPendingWork falls back to the local DaemonHub, which is the
|
||||
// correct delivery scope for a single-node deployment.
|
||||
DaemonPendingWork DaemonPendingWorkNotifier
|
||||
// ModelCatalogCache serves the last known good model list for a runtime so
|
||||
// the picker can render without waiting for a daemon round trip
|
||||
// (stale-while-revalidate, MUL-5444). Nil-safe: every call site treats a nil
|
||||
// cache as a permanent miss and falls back to the full discovery flow.
|
||||
ModelCatalogCache ModelCatalogCache
|
||||
// Metrics is the shared business-metrics collector built by main.go.
|
||||
// May be nil in tests / self-hosted with the metrics listener disabled;
|
||||
// every Record* method is nil-safe and obsmetrics.RecordEvent treats a
|
||||
@@ -309,6 +328,7 @@ func New(queries *db.Queries, txStarter txStarter, hub *realtime.Hub, bus *event
|
||||
EmailService: emailService,
|
||||
UpdateStore: NewInMemoryUpdateStore(),
|
||||
ModelListStore: NewInMemoryModelListStore(),
|
||||
ModelCatalogCache: NewInMemoryModelCatalogCache(),
|
||||
LocalSkillListStore: NewInMemoryLocalSkillListStore(),
|
||||
LocalSkillImportStore: NewInMemoryLocalSkillImportStore(),
|
||||
LivenessStore: NewNoopLivenessStore(),
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Runtime model catalog cache (stale-while-revalidate)
|
||||
// ---------------------------------------------------------------------------
|
||||
//
|
||||
// Listing a runtime's models is a round trip to the user's machine: the request
|
||||
// waits for the daemon's next heartbeat, the daemon shells out to the provider
|
||||
// CLI (or drives an ACP handshake) and only then reports back. Even with the
|
||||
// pending-work push hint (MUL-5444) that is seconds of latency on a UI surface
|
||||
// people open repeatedly while filling in one form — switch runtime, look at the
|
||||
// models, switch back.
|
||||
//
|
||||
// The catalog itself changes only when the user upgrades a CLI, logs into a
|
||||
// different account, or edits a provider config, so it is a textbook
|
||||
// stale-while-revalidate candidate: answer from the last known good snapshot
|
||||
// immediately, and refresh in the background so the NEXT open is also warm.
|
||||
//
|
||||
// Windows are deliberately conservative:
|
||||
// - modelCatalogServeWindow bounds how stale an answer the API will hand a
|
||||
// client without waiting for the daemon.
|
||||
// - modelCatalogRevalidateAfter bounds how long a snapshot can be served
|
||||
// without triggering a background refresh, so a CLI upgrade converges
|
||||
// within one open instead of lingering for the whole serve window.
|
||||
//
|
||||
// Only successful, non-empty, `supported` catalogs are cached. An empty list is
|
||||
// almost always a transient discovery failure (CLI not logged in, timeout) —
|
||||
// caching it would pin the picker empty (same reasoning as agent.cachedDiscovery
|
||||
// in the daemon).
|
||||
|
||||
const (
|
||||
// modelCatalogServeWindow is how long a cached catalog may answer a
|
||||
// list-models request without waiting for the daemon.
|
||||
modelCatalogServeWindow = 15 * time.Minute
|
||||
// modelCatalogRevalidateAfter is the age past which serving from cache also
|
||||
// enqueues a background refresh.
|
||||
modelCatalogRevalidateAfter = 60 * time.Second
|
||||
)
|
||||
|
||||
// ModelCatalogSnapshot is the last known good model list for one runtime.
|
||||
type ModelCatalogSnapshot struct {
|
||||
RuntimeID string `json:"runtime_id"`
|
||||
Models []ModelEntry `json:"models"`
|
||||
Supported bool `json:"supported"`
|
||||
StoredAt time.Time `json:"stored_at"`
|
||||
}
|
||||
|
||||
// Age reports how long ago the snapshot was captured.
|
||||
func (s *ModelCatalogSnapshot) Age(now time.Time) time.Duration {
|
||||
if s == nil {
|
||||
return 0
|
||||
}
|
||||
return now.Sub(s.StoredAt)
|
||||
}
|
||||
|
||||
// ModelCatalogCache stores the last successful model catalog per runtime. Both
|
||||
// methods are best-effort from the caller's perspective: a Get error means
|
||||
// "answer the slow way" and a Put error means "the next open is cold". Neither
|
||||
// may fail a request.
|
||||
//
|
||||
// Implementations must be safe for concurrent use.
|
||||
type ModelCatalogCache interface {
|
||||
Get(ctx context.Context, runtimeID string) (*ModelCatalogSnapshot, error)
|
||||
Put(ctx context.Context, runtimeID string, models []ModelEntry, supported bool) error
|
||||
// Invalidate drops any snapshot for the runtime. Used when the cached
|
||||
// catalog can no longer be trusted (e.g. the runtime row was deleted).
|
||||
Invalidate(ctx context.Context, runtimeID string) error
|
||||
}
|
||||
|
||||
// cacheableModelCatalog reports whether a completed discovery result is worth
|
||||
// remembering. `supported=false` runtimes have no picker at all, and an empty
|
||||
// catalog is treated as a transient failure rather than an authoritative
|
||||
// "this runtime has no models".
|
||||
func cacheableModelCatalog(models []ModelEntry, supported bool) bool {
|
||||
return supported && len(models) > 0
|
||||
}
|
||||
|
||||
// cloneModelEntries deep-copies a catalog so the in-memory backend hands out
|
||||
// values a caller cannot mutate into the shared cache. A shallow slice copy is
|
||||
// not enough: ModelEntry carries a *ModelThinking (with its own level slice) and
|
||||
// a ServiceTiers slice, all of which would still alias the cached objects. The
|
||||
// Redis backend gets this for free by round-tripping through JSON, and the two
|
||||
// implementations must not differ in whether the returned value is independent.
|
||||
func cloneModelEntries(models []ModelEntry) []ModelEntry {
|
||||
if models == nil {
|
||||
return nil
|
||||
}
|
||||
out := make([]ModelEntry, len(models))
|
||||
for i, m := range models {
|
||||
clone := m
|
||||
if m.Thinking != nil {
|
||||
thinking := *m.Thinking
|
||||
if m.Thinking.SupportedLevels != nil {
|
||||
thinking.SupportedLevels = append([]ThinkingLevel(nil), m.Thinking.SupportedLevels...)
|
||||
}
|
||||
clone.Thinking = &thinking
|
||||
}
|
||||
if m.ServiceTiers != nil {
|
||||
clone.ServiceTiers = append([]ModelServiceTier(nil), m.ServiceTiers...)
|
||||
}
|
||||
out[i] = clone
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// InMemoryModelCatalogCache is the single-node implementation. Adequate for
|
||||
// self-hosted and tests; multi-node deploys should use the Redis backend so
|
||||
// every API replica shares one warm catalog.
|
||||
type InMemoryModelCatalogCache struct {
|
||||
mu sync.Mutex
|
||||
entries map[string]ModelCatalogSnapshot
|
||||
retainFor time.Duration
|
||||
}
|
||||
|
||||
func NewInMemoryModelCatalogCache() *InMemoryModelCatalogCache {
|
||||
return &InMemoryModelCatalogCache{
|
||||
entries: make(map[string]ModelCatalogSnapshot),
|
||||
retainFor: modelCatalogServeWindow,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *InMemoryModelCatalogCache) Get(_ context.Context, runtimeID string) (*ModelCatalogSnapshot, error) {
|
||||
if runtimeID == "" {
|
||||
return nil, nil
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
entry, ok := c.entries[runtimeID]
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
if time.Since(entry.StoredAt) > c.retainFor {
|
||||
delete(c.entries, runtimeID)
|
||||
return nil, nil
|
||||
}
|
||||
// Copy so a caller mutating the response cannot corrupt the cache.
|
||||
snapshot := entry
|
||||
snapshot.Models = cloneModelEntries(entry.Models)
|
||||
return &snapshot, nil
|
||||
}
|
||||
|
||||
func (c *InMemoryModelCatalogCache) Put(_ context.Context, runtimeID string, models []ModelEntry, supported bool) error {
|
||||
if runtimeID == "" || !cacheableModelCatalog(models, supported) {
|
||||
return nil
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
// Garbage-collect expired entries so the map can't grow unbounded as
|
||||
// runtimes come and go.
|
||||
now := time.Now()
|
||||
for id, entry := range c.entries {
|
||||
if now.Sub(entry.StoredAt) > c.retainFor {
|
||||
delete(c.entries, id)
|
||||
}
|
||||
}
|
||||
|
||||
c.entries[runtimeID] = ModelCatalogSnapshot{
|
||||
RuntimeID: runtimeID,
|
||||
Models: cloneModelEntries(models),
|
||||
Supported: supported,
|
||||
StoredAt: now,
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *InMemoryModelCatalogCache) Invalidate(_ context.Context, runtimeID string) error {
|
||||
if runtimeID == "" {
|
||||
return nil
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
delete(c.entries, runtimeID)
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// Redis-backed ModelCatalogCache so every API replica shares one warm model
|
||||
// catalog per runtime. Without it each node would keep its own map and a user
|
||||
// bouncing between replicas behind a load balancer would still pay the full
|
||||
// daemon round trip on most opens (same multi-node reasoning as
|
||||
// RedisModelListStore).
|
||||
//
|
||||
// Key layout — one plain string key per runtime, TTL = serve window, so
|
||||
// expiry is Redis's job and there is nothing to sweep:
|
||||
//
|
||||
// mul:runtime_model_catalog:<runtime_id> → JSON-encoded ModelCatalogSnapshot
|
||||
//
|
||||
// No hash tag: every operation here is single-key, so cluster slot co-location
|
||||
// buys nothing (unlike the Lua-scripted pending queues).
|
||||
const modelCatalogKeyPrefix = "mul:runtime_model_catalog:"
|
||||
|
||||
func modelCatalogKey(runtimeID string) string { return modelCatalogKeyPrefix + runtimeID }
|
||||
|
||||
// RedisModelCatalogCache is the multi-node implementation of ModelCatalogCache.
|
||||
type RedisModelCatalogCache struct {
|
||||
rdb *redis.Client
|
||||
ttl time.Duration
|
||||
}
|
||||
|
||||
func NewRedisModelCatalogCache(rdb *redis.Client) *RedisModelCatalogCache {
|
||||
return &RedisModelCatalogCache{rdb: rdb, ttl: modelCatalogServeWindow}
|
||||
}
|
||||
|
||||
func (c *RedisModelCatalogCache) Get(ctx context.Context, runtimeID string) (*ModelCatalogSnapshot, error) {
|
||||
if runtimeID == "" {
|
||||
return nil, nil
|
||||
}
|
||||
raw, err := c.rdb.Get(ctx, modelCatalogKey(runtimeID)).Bytes()
|
||||
if errors.Is(err, redis.Nil) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get model catalog: %w", err)
|
||||
}
|
||||
var snapshot ModelCatalogSnapshot
|
||||
if err := json.Unmarshal(raw, &snapshot); err != nil {
|
||||
// A snapshot we cannot decode is indistinguishable from a cold cache;
|
||||
// drop it so the next Put replaces the bad value.
|
||||
c.rdb.Del(ctx, modelCatalogKey(runtimeID))
|
||||
return nil, fmt.Errorf("decode model catalog: %w", err)
|
||||
}
|
||||
// Guard against a snapshot whose TTL outlived the serve window (TTL is set
|
||||
// from the same constant, but a config change or a manual write could
|
||||
// diverge) — the age check is the contract, not the TTL.
|
||||
if snapshot.StoredAt.IsZero() || time.Since(snapshot.StoredAt) > c.ttl {
|
||||
return nil, nil
|
||||
}
|
||||
return &snapshot, nil
|
||||
}
|
||||
|
||||
func (c *RedisModelCatalogCache) Put(ctx context.Context, runtimeID string, models []ModelEntry, supported bool) error {
|
||||
if runtimeID == "" || !cacheableModelCatalog(models, supported) {
|
||||
return nil
|
||||
}
|
||||
snapshot := ModelCatalogSnapshot{
|
||||
RuntimeID: runtimeID,
|
||||
Models: models,
|
||||
Supported: supported,
|
||||
StoredAt: time.Now(),
|
||||
}
|
||||
data, err := json.Marshal(snapshot)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal model catalog: %w", err)
|
||||
}
|
||||
if err := c.rdb.Set(ctx, modelCatalogKey(runtimeID), data, c.ttl).Err(); err != nil {
|
||||
return fmt.Errorf("persist model catalog: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *RedisModelCatalogCache) Invalidate(ctx context.Context, runtimeID string) error {
|
||||
if runtimeID == "" {
|
||||
return nil
|
||||
}
|
||||
if err := c.rdb.Del(ctx, modelCatalogKey(runtimeID)).Err(); err != nil {
|
||||
return fmt.Errorf("invalidate model catalog: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Reuses the newRedisTestClient helper from
|
||||
// runtime_local_skills_redis_store_test.go: same Redis instance, same gating on
|
||||
// REDIS_TEST_URL, same FlushDB-per-test isolation. The in-memory twin of these
|
||||
// assertions lives in runtime_model_catalog_test.go — both backends must agree,
|
||||
// because a deployment silently swaps between them based on REDIS_URL.
|
||||
|
||||
func TestRedisModelCatalogCache_RoundTrip(t *testing.T) {
|
||||
rdb := newRedisTestClient(t)
|
||||
ctx := context.Background()
|
||||
cache := NewRedisModelCatalogCache(rdb)
|
||||
|
||||
if got, err := cache.Get(ctx, "runtime-1"); err != nil || got != nil {
|
||||
t.Fatalf("cold cache should miss: got=%+v err=%v", got, err)
|
||||
}
|
||||
|
||||
if err := cache.Put(ctx, "runtime-1", sampleCatalog(), true); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
|
||||
got, err := cache.Get(ctx, "runtime-1")
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if got == nil {
|
||||
t.Fatal("expected a cached snapshot")
|
||||
}
|
||||
if got.RuntimeID != "runtime-1" || len(got.Models) != 2 {
|
||||
t.Fatalf("unexpected snapshot: %+v", got)
|
||||
}
|
||||
if !got.Models[0].Default {
|
||||
t.Error("the default badge must survive the Redis round trip")
|
||||
}
|
||||
if got.StoredAt.IsZero() {
|
||||
t.Error("StoredAt must be persisted — the serve/revalidate windows depend on it")
|
||||
}
|
||||
if age := got.Age(time.Now()); age < 0 || age > time.Minute {
|
||||
t.Errorf("implausible snapshot age %s", age)
|
||||
}
|
||||
|
||||
// A sibling API node reading the same key must see the same snapshot; that
|
||||
// is the whole reason this backend exists.
|
||||
sibling := NewRedisModelCatalogCache(rdb)
|
||||
if shared, err := sibling.Get(ctx, "runtime-1"); err != nil || shared == nil {
|
||||
t.Fatalf("snapshot not visible to a sibling node: got=%+v err=%v", shared, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRedisModelCatalogCache_SkipsUncacheableResults(t *testing.T) {
|
||||
rdb := newRedisTestClient(t)
|
||||
ctx := context.Background()
|
||||
cache := NewRedisModelCatalogCache(rdb)
|
||||
|
||||
if err := cache.Put(ctx, "runtime-empty", nil, true); err != nil {
|
||||
t.Fatalf("put empty: %v", err)
|
||||
}
|
||||
if got, _ := cache.Get(ctx, "runtime-empty"); got != nil {
|
||||
t.Fatalf("empty catalog must not be cached: %+v", got)
|
||||
}
|
||||
|
||||
if err := cache.Put(ctx, "runtime-unsupported", sampleCatalog(), false); err != nil {
|
||||
t.Fatalf("put unsupported: %v", err)
|
||||
}
|
||||
if got, _ := cache.Get(ctx, "runtime-unsupported"); got != nil {
|
||||
t.Fatalf("unsupported runtime must not be cached: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRedisModelCatalogCache_InvalidateAndExpiry(t *testing.T) {
|
||||
rdb := newRedisTestClient(t)
|
||||
ctx := context.Background()
|
||||
cache := NewRedisModelCatalogCache(rdb)
|
||||
|
||||
if err := cache.Put(ctx, "runtime-1", sampleCatalog(), true); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
if err := cache.Invalidate(ctx, "runtime-1"); err != nil {
|
||||
t.Fatalf("invalidate: %v", err)
|
||||
}
|
||||
if got, _ := cache.Get(ctx, "runtime-1"); got != nil {
|
||||
t.Fatalf("expected a miss after Invalidate: %+v", got)
|
||||
}
|
||||
|
||||
// The key must carry a TTL so a runtime that never comes back cannot pin a
|
||||
// snapshot in Redis forever.
|
||||
if err := cache.Put(ctx, "runtime-1", sampleCatalog(), true); err != nil {
|
||||
t.Fatalf("re-put: %v", err)
|
||||
}
|
||||
ttl, err := rdb.TTL(ctx, modelCatalogKey("runtime-1")).Result()
|
||||
if err != nil {
|
||||
t.Fatalf("ttl: %v", err)
|
||||
}
|
||||
if ttl <= 0 || ttl > modelCatalogServeWindow {
|
||||
t.Fatalf("ttl = %s, want (0, %s]", ttl, modelCatalogServeWindow)
|
||||
}
|
||||
|
||||
// A snapshot older than the serve window is a miss even if the key is still
|
||||
// present (the age check is the contract, not the TTL).
|
||||
stale := NewRedisModelCatalogCache(rdb)
|
||||
stale.ttl = time.Nanosecond
|
||||
if got, err := stale.Get(ctx, "runtime-1"); err != nil || got != nil {
|
||||
t.Fatalf("expected a miss past the serve window: got=%+v err=%v", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRedisModelCatalogCache_DropsUndecodableSnapshot(t *testing.T) {
|
||||
rdb := newRedisTestClient(t)
|
||||
ctx := context.Background()
|
||||
cache := NewRedisModelCatalogCache(rdb)
|
||||
|
||||
if err := rdb.Set(ctx, modelCatalogKey("runtime-1"), "not-json", time.Minute).Err(); err != nil {
|
||||
t.Fatalf("seed: %v", err)
|
||||
}
|
||||
if got, err := cache.Get(ctx, "runtime-1"); err == nil || got != nil {
|
||||
t.Fatalf("expected a decode error and no snapshot: got=%+v err=%v", got, err)
|
||||
}
|
||||
// The poisoned value must be gone so it cannot fail every later read.
|
||||
if n, err := rdb.Exists(ctx, modelCatalogKey("runtime-1")).Result(); err != nil || n != 0 {
|
||||
t.Fatalf("undecodable snapshot was not dropped: exists=%d err=%v", n, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,380 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func sampleCatalog() []ModelEntry {
|
||||
return []ModelEntry{
|
||||
{ID: "claude-sonnet-4-6", Label: "Claude Sonnet 4.6", Provider: "anthropic", Default: true},
|
||||
{ID: "claude-opus-5", Label: "Claude Opus 5", Provider: "anthropic"},
|
||||
}
|
||||
}
|
||||
|
||||
// TestInMemoryModelCatalogCache_RoundTrip is the happy path behind the
|
||||
// stale-while-revalidate fast path (MUL-5444): a completed discovery is
|
||||
// remembered so the next picker open answers without a daemon round trip.
|
||||
func TestInMemoryModelCatalogCache_RoundTrip(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cache := NewInMemoryModelCatalogCache()
|
||||
|
||||
if got, err := cache.Get(ctx, "rt-1"); err != nil || got != nil {
|
||||
t.Fatalf("cold cache should miss: got=%+v err=%v", got, err)
|
||||
}
|
||||
if err := cache.Put(ctx, "rt-1", sampleCatalog(), true); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
|
||||
got, err := cache.Get(ctx, "rt-1")
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if got == nil {
|
||||
t.Fatal("expected a cached snapshot")
|
||||
}
|
||||
if len(got.Models) != 2 || got.Models[0].ID != "claude-sonnet-4-6" {
|
||||
t.Fatalf("unexpected models: %+v", got.Models)
|
||||
}
|
||||
if !got.Models[0].Default {
|
||||
t.Error("the default badge must survive the cache round trip")
|
||||
}
|
||||
if !got.Supported {
|
||||
t.Error("supported flag lost")
|
||||
}
|
||||
if got.Age(time.Now()) < 0 {
|
||||
t.Error("snapshot age must not be negative")
|
||||
}
|
||||
// Other runtimes keep their own entry — a catalog is per machine + per CLI
|
||||
// install, never shareable across runtimes.
|
||||
if other, err := cache.Get(ctx, "rt-2"); err != nil || other != nil {
|
||||
t.Fatalf("unrelated runtime should miss: got=%+v err=%v", other, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestInMemoryModelCatalogCache_ReturnsIndependentCopies stops a caller that
|
||||
// mutates the response (e.g. filtering the list) from corrupting the shared
|
||||
// cache for every later reader.
|
||||
func TestInMemoryModelCatalogCache_ReturnsIndependentCopies(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cache := NewInMemoryModelCatalogCache()
|
||||
if err := cache.Put(ctx, "rt-1", sampleCatalog(), true); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
|
||||
first, err := cache.Get(ctx, "rt-1")
|
||||
if err != nil || first == nil {
|
||||
t.Fatalf("get: %+v %v", first, err)
|
||||
}
|
||||
first.Models[0].ID = "mutated"
|
||||
|
||||
second, err := cache.Get(ctx, "rt-1")
|
||||
if err != nil || second == nil {
|
||||
t.Fatalf("get: %+v %v", second, err)
|
||||
}
|
||||
if second.Models[0].ID != "claude-sonnet-4-6" {
|
||||
t.Fatalf("cache was corrupted by a caller mutation: %+v", second.Models[0])
|
||||
}
|
||||
}
|
||||
|
||||
// TestInMemoryModelCatalogCache_IsolatesNestedFields extends the copy guarantee
|
||||
// to everything a ModelEntry points at. A shallow slice copy would leave the
|
||||
// *ModelThinking, its level slice, and ServiceTiers aliasing the cached objects,
|
||||
// which also made this backend behave differently from the Redis one (JSON
|
||||
// round-trip always yields an independent value).
|
||||
func TestInMemoryModelCatalogCache_IsolatesNestedFields(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cache := NewInMemoryModelCatalogCache()
|
||||
|
||||
source := []ModelEntry{{
|
||||
ID: "gpt-5.6-sol",
|
||||
Label: "GPT-5.6-Sol",
|
||||
Thinking: &ModelThinking{
|
||||
DefaultLevel: "low",
|
||||
SupportedLevels: []ThinkingLevel{{Value: "low", Label: "Low"}},
|
||||
},
|
||||
ServiceTiers: []ModelServiceTier{{ID: "fast", Name: "Fast"}},
|
||||
}}
|
||||
if err := cache.Put(ctx, "rt-1", source, true); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
|
||||
// Mutating the caller's own slice after Put must not reach the cache.
|
||||
source[0].Thinking.DefaultLevel = "mutated-by-writer"
|
||||
source[0].Thinking.SupportedLevels[0].Label = "mutated-by-writer"
|
||||
source[0].ServiceTiers[0].Name = "mutated-by-writer"
|
||||
|
||||
first, err := cache.Get(ctx, "rt-1")
|
||||
if err != nil || first == nil {
|
||||
t.Fatalf("get: %+v %v", first, err)
|
||||
}
|
||||
if first.Models[0].Thinking.DefaultLevel != "low" ||
|
||||
first.Models[0].Thinking.SupportedLevels[0].Label != "Low" ||
|
||||
first.Models[0].ServiceTiers[0].Name != "Fast" {
|
||||
t.Fatalf("writer mutation leaked into the cache: %+v", first.Models[0])
|
||||
}
|
||||
|
||||
// Mutating a returned snapshot must not reach the cache either.
|
||||
first.Models[0].Thinking.DefaultLevel = "mutated-by-reader"
|
||||
first.Models[0].Thinking.SupportedLevels[0].Value = "mutated-by-reader"
|
||||
first.Models[0].ServiceTiers[0].ID = "mutated-by-reader"
|
||||
|
||||
second, err := cache.Get(ctx, "rt-1")
|
||||
if err != nil || second == nil {
|
||||
t.Fatalf("get: %+v %v", second, err)
|
||||
}
|
||||
if second.Models[0].Thinking == first.Models[0].Thinking {
|
||||
t.Error("thinking pointer is shared between snapshots")
|
||||
}
|
||||
if second.Models[0].Thinking.DefaultLevel != "low" ||
|
||||
second.Models[0].Thinking.SupportedLevels[0].Value != "low" ||
|
||||
second.Models[0].ServiceTiers[0].ID != "fast" {
|
||||
t.Fatalf("reader mutation leaked into the cache: %+v", second.Models[0])
|
||||
}
|
||||
}
|
||||
|
||||
// TestInMemoryModelCatalogCache_SkipsUncacheableResults pins the same rule the
|
||||
// daemon's own discovery cache uses: an empty catalog is a transient failure
|
||||
// (CLI not logged in, timeout), and caching it would pin the picker empty for
|
||||
// the whole serve window.
|
||||
func TestInMemoryModelCatalogCache_SkipsUncacheableResults(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cache := NewInMemoryModelCatalogCache()
|
||||
|
||||
if err := cache.Put(ctx, "rt-empty", nil, true); err != nil {
|
||||
t.Fatalf("put empty: %v", err)
|
||||
}
|
||||
if got, _ := cache.Get(ctx, "rt-empty"); got != nil {
|
||||
t.Fatalf("empty catalog must not be cached: %+v", got)
|
||||
}
|
||||
|
||||
if err := cache.Put(ctx, "rt-unsupported", sampleCatalog(), false); err != nil {
|
||||
t.Fatalf("put unsupported: %v", err)
|
||||
}
|
||||
if got, _ := cache.Get(ctx, "rt-unsupported"); got != nil {
|
||||
t.Fatalf("unsupported runtime must not be cached: %+v", got)
|
||||
}
|
||||
|
||||
if err := cache.Put(ctx, "", sampleCatalog(), true); err != nil {
|
||||
t.Fatalf("put empty runtime id: %v", err)
|
||||
}
|
||||
if got, _ := cache.Get(ctx, ""); got != nil {
|
||||
t.Fatalf("empty runtime id must not be cached: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestInMemoryModelCatalogCache_ExpiresAndInvalidates bounds how stale an
|
||||
// answer the fast path can serve, and proves an explicit drop works.
|
||||
func TestInMemoryModelCatalogCache_ExpiresAndInvalidates(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
cache := NewInMemoryModelCatalogCache()
|
||||
cache.retainFor = 20 * time.Millisecond
|
||||
|
||||
if err := cache.Put(ctx, "rt-1", sampleCatalog(), true); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
time.Sleep(40 * time.Millisecond)
|
||||
if got, err := cache.Get(ctx, "rt-1"); err != nil || got != nil {
|
||||
t.Fatalf("expected expiry past the serve window: got=%+v err=%v", got, err)
|
||||
}
|
||||
|
||||
cache.retainFor = modelCatalogServeWindow
|
||||
if err := cache.Put(ctx, "rt-1", sampleCatalog(), true); err != nil {
|
||||
t.Fatalf("re-put: %v", err)
|
||||
}
|
||||
if err := cache.Invalidate(ctx, "rt-1"); err != nil {
|
||||
t.Fatalf("invalidate: %v", err)
|
||||
}
|
||||
if got, _ := cache.Get(ctx, "rt-1"); got != nil {
|
||||
t.Fatalf("expected a miss after Invalidate: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// failingModelCatalogCache reports a backend error on every read.
|
||||
type failingModelCatalogCache struct{}
|
||||
|
||||
func (failingModelCatalogCache) Get(context.Context, string) (*ModelCatalogSnapshot, error) {
|
||||
return nil, errors.New("redis down")
|
||||
}
|
||||
func (failingModelCatalogCache) Put(context.Context, string, []ModelEntry, bool) error { return nil }
|
||||
func (failingModelCatalogCache) Invalidate(context.Context, string) error { return nil }
|
||||
|
||||
// TestCachedModelCatalog_DegradesToMiss proves the cache can never fail a
|
||||
// request: a nil cache, a backend error, or a snapshot that is no longer
|
||||
// serveable all fall back to the normal daemon round trip.
|
||||
func TestCachedModelCatalog_DegradesToMiss(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
if got := (&Handler{}).cachedModelCatalog(ctx, "rt-1"); got != nil {
|
||||
t.Fatalf("nil cache should miss, got %+v", got)
|
||||
}
|
||||
|
||||
h := &Handler{ModelCatalogCache: failingModelCatalogCache{}}
|
||||
if got := h.cachedModelCatalog(ctx, "rt-1"); got != nil {
|
||||
t.Fatalf("failing cache should miss, got %+v", got)
|
||||
}
|
||||
|
||||
// A snapshot that somehow holds an unusable catalog is treated as a miss
|
||||
// rather than answering the picker with an empty list.
|
||||
inmem := NewInMemoryModelCatalogCache()
|
||||
inmem.entries["rt-1"] = ModelCatalogSnapshot{RuntimeID: "rt-1", Supported: true, StoredAt: time.Now()}
|
||||
h = &Handler{ModelCatalogCache: inmem}
|
||||
if got := h.cachedModelCatalog(ctx, "rt-1"); got != nil {
|
||||
t.Fatalf("empty cached catalog should miss, got %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// pendingWorkRecorder records the runtime-scoped hints a handler pushes.
|
||||
type pendingWorkRecorder struct {
|
||||
mu sync.Mutex
|
||||
hints []string
|
||||
}
|
||||
|
||||
func (r *pendingWorkRecorder) NotifyPendingWork(runtimeID, kind string) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.hints = append(r.hints, runtimeID+":"+kind)
|
||||
}
|
||||
|
||||
func (r *pendingWorkRecorder) count() int {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return len(r.hints)
|
||||
}
|
||||
|
||||
// TestRevalidateModelCatalog_EnqueuesAndHints covers the "revalidate" half of
|
||||
// stale-while-revalidate: serving a stale snapshot must queue a refresh AND
|
||||
// push the wakeup hint, so the next open is both fast and fresh.
|
||||
func TestRevalidateModelCatalog_EnqueuesAndHints(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := NewInMemoryModelListStore()
|
||||
rec := &pendingWorkRecorder{}
|
||||
h := &Handler{ModelListStore: store, DaemonPendingWork: rec}
|
||||
|
||||
h.revalidateModelCatalog(ctx, "rt-1")
|
||||
|
||||
pending, err := store.HasPending(ctx, "rt-1")
|
||||
if err != nil {
|
||||
t.Fatalf("has pending: %v", err)
|
||||
}
|
||||
if !pending {
|
||||
t.Fatal("expected a background refresh request to be enqueued")
|
||||
}
|
||||
if rec.count() != 1 {
|
||||
t.Fatalf("expected exactly 1 pending-work hint, got %d", rec.count())
|
||||
}
|
||||
|
||||
// Stampede control: a second open while the first refresh is still queued
|
||||
// must not pile on another request.
|
||||
h.revalidateModelCatalog(ctx, "rt-1")
|
||||
if rec.count() != 1 {
|
||||
t.Fatalf("expected the queued refresh to suppress a second hint, got %d", rec.count())
|
||||
}
|
||||
}
|
||||
|
||||
// TestRequestDaemonPendingWork_PrefersNotifier keeps the notifier optional: a
|
||||
// handler without one must not panic (single-node deployments fall back to the
|
||||
// local hub, which is nil in unit tests).
|
||||
func TestRequestDaemonPendingWork_PrefersNotifier(t *testing.T) {
|
||||
rec := &pendingWorkRecorder{}
|
||||
h := &Handler{DaemonPendingWork: rec}
|
||||
h.requestDaemonPendingWork("rt-1", "model_list")
|
||||
if rec.count() != 1 {
|
||||
t.Fatalf("expected the notifier to receive the hint, got %d", rec.count())
|
||||
}
|
||||
|
||||
h.requestDaemonPendingWork("", "model_list")
|
||||
if rec.count() != 1 {
|
||||
t.Fatalf("an empty runtime id must not produce a hint, got %d", rec.count())
|
||||
}
|
||||
|
||||
// No notifier and no hub: still a no-op, never a panic.
|
||||
(&Handler{}).requestDaemonPendingWork("rt-1", "model_list")
|
||||
}
|
||||
|
||||
// TestCacheableModelCatalog pins which completed discovery results are worth
|
||||
// remembering. It is also the branch ReportModelListResult uses to decide
|
||||
// between warming the cache and dropping a snapshot the runtime no longer
|
||||
// advertises.
|
||||
func TestCacheableModelCatalog(t *testing.T) {
|
||||
if !cacheableModelCatalog(sampleCatalog(), true) {
|
||||
t.Error("a supported, non-empty catalog must be cacheable")
|
||||
}
|
||||
if cacheableModelCatalog(sampleCatalog(), false) {
|
||||
t.Error("a runtime that ignores model selection must not be cached")
|
||||
}
|
||||
if cacheableModelCatalog(nil, true) {
|
||||
t.Error("an empty catalog is a transient failure, not a cacheable answer")
|
||||
}
|
||||
if cacheableModelCatalog([]ModelEntry{}, true) {
|
||||
t.Error("an empty (non-nil) catalog is not cacheable either")
|
||||
}
|
||||
}
|
||||
|
||||
// TestCachedModelListResponse_WireShape pins what a cache hit looks like on the
|
||||
// wire. Existing clients only branch on status/models, so the response must be
|
||||
// indistinguishable from a completed live discovery apart from the additive
|
||||
// `cached` marker.
|
||||
func TestCachedModelListResponse_WireShape(t *testing.T) {
|
||||
storedAt := time.Now().Add(-2 * time.Minute)
|
||||
resp := &ModelListRequest{
|
||||
ID: randomID(),
|
||||
RuntimeID: "rt-1",
|
||||
Status: ModelListCompleted,
|
||||
Models: sampleCatalog(),
|
||||
Supported: true,
|
||||
CreatedAt: storedAt,
|
||||
UpdatedAt: storedAt,
|
||||
Cached: true,
|
||||
CachedAt: &storedAt,
|
||||
}
|
||||
raw, err := json.Marshal(resp)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
|
||||
var decoded struct {
|
||||
ID string `json:"id"`
|
||||
Status ModelListStatus `json:"status"`
|
||||
Models []ModelEntry `json:"models"`
|
||||
Supported bool `json:"supported"`
|
||||
Cached bool `json:"cached"`
|
||||
CachedAt *time.Time `json:"cached_at"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &decoded); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if decoded.Status != ModelListCompleted {
|
||||
t.Fatalf("status = %q, want completed so clients stop polling", decoded.Status)
|
||||
}
|
||||
if decoded.ID == "" {
|
||||
t.Error("a cache hit still needs a request id for client-side bookkeeping")
|
||||
}
|
||||
if len(decoded.Models) != 2 || !decoded.Supported {
|
||||
t.Fatalf("cache hit must carry the catalog: %+v supported=%v", decoded.Models, decoded.Supported)
|
||||
}
|
||||
if !decoded.Cached || decoded.CachedAt == nil {
|
||||
t.Fatalf("cache hit must be marked: cached=%v cached_at=%v", decoded.Cached, decoded.CachedAt)
|
||||
}
|
||||
|
||||
// A live (non-cached) response must not carry the markers at all, so older
|
||||
// clients see exactly the previous payload.
|
||||
live, err := json.Marshal(&ModelListRequest{ID: "x", RuntimeID: "rt-1", Status: ModelListPending, Supported: true})
|
||||
if err != nil {
|
||||
t.Fatalf("marshal live: %v", err)
|
||||
}
|
||||
var liveFields map[string]any
|
||||
if err := json.Unmarshal(live, &liveFields); err != nil {
|
||||
t.Fatalf("unmarshal live: %v", err)
|
||||
}
|
||||
if _, ok := liveFields["cached"]; ok {
|
||||
t.Error("live responses must omit the cached marker")
|
||||
}
|
||||
if _, ok := liveFields["cached_at"]; ok {
|
||||
t.Error("live responses must omit cached_at")
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/multica-ai/multica/server/pkg/protocol"
|
||||
)
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -60,6 +61,15 @@ type ModelListRequest struct {
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
RunStartedAt *time.Time `json:"-"`
|
||||
// Cached marks a response answered from the server-side catalog cache
|
||||
// instead of a live daemon round trip (MUL-5444). Purely informational —
|
||||
// Status is already "completed" and Models is already populated, so a client
|
||||
// that ignores this field behaves exactly as before. CachedAt carries the
|
||||
// snapshot's capture time for clients that want to surface freshness.
|
||||
// Neither field is ever persisted in the request store; they only exist on
|
||||
// the synthetic cache-hit response.
|
||||
Cached bool `json:"cached,omitempty"`
|
||||
CachedAt *time.Time `json:"cached_at,omitempty"`
|
||||
}
|
||||
|
||||
// ModelEntry mirrors agent.Model for the wire. `Default` tags the
|
||||
@@ -290,8 +300,17 @@ func modelListRequestTerminal(status ModelListStatus) bool {
|
||||
// Handlers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// InitiateListModels creates a pending model list request for a runtime.
|
||||
// Called by the frontend; the daemon picks it up on its next heartbeat.
|
||||
// InitiateListModels answers a "list this runtime's models" request.
|
||||
//
|
||||
// Fast path: a cached catalog younger than modelCatalogServeWindow is returned
|
||||
// as an already-completed request, so the picker renders immediately instead of
|
||||
// waiting for the daemon (stale-while-revalidate). Serving a snapshot older than
|
||||
// modelCatalogRevalidateAfter also enqueues a background refresh, which nobody
|
||||
// polls — its only job is to warm the cache for the next open.
|
||||
//
|
||||
// Slow path: enqueue a pending request the daemon claims on its next heartbeat,
|
||||
// and push a wakeup hint so "next heartbeat" is now rather than up to one
|
||||
// HeartbeatInterval away.
|
||||
func (h *Handler) InitiateListModels(w http.ResponseWriter, r *http.Request) {
|
||||
runtimeID := chi.URLParam(r, "runtimeId")
|
||||
runtimeUUID, ok := parseUUIDOrBadRequest(w, runtimeID, "runtime_id")
|
||||
@@ -311,15 +330,104 @@ func (h *Handler) InitiateListModels(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, http.StatusServiceUnavailable, "runtime is offline")
|
||||
return
|
||||
}
|
||||
resolvedRuntimeID := uuidToString(rt.ID)
|
||||
|
||||
req, err := h.ModelListStore.Create(r.Context(), uuidToString(rt.ID))
|
||||
if cached := h.cachedModelCatalog(r.Context(), resolvedRuntimeID); cached != nil {
|
||||
age := cached.Age(time.Now())
|
||||
if age >= modelCatalogRevalidateAfter {
|
||||
h.revalidateModelCatalog(r.Context(), resolvedRuntimeID)
|
||||
}
|
||||
storedAt := cached.StoredAt
|
||||
writeJSON(w, http.StatusOK, &ModelListRequest{
|
||||
// Synthetic ID: no store record backs a cache hit. Clients only poll
|
||||
// GET /models/{id} while status is pending/running, which this
|
||||
// response never is.
|
||||
ID: randomID(),
|
||||
RuntimeID: resolvedRuntimeID,
|
||||
Status: ModelListCompleted,
|
||||
Models: cached.Models,
|
||||
Supported: cached.Supported,
|
||||
CreatedAt: storedAt,
|
||||
UpdatedAt: storedAt,
|
||||
Cached: true,
|
||||
CachedAt: &storedAt,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
req, err := h.ModelListStore.Create(r.Context(), resolvedRuntimeID)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, "failed to enqueue model list request: "+err.Error())
|
||||
return
|
||||
}
|
||||
h.requestDaemonPendingWork(resolvedRuntimeID, protocol.PendingWorkKindModelList)
|
||||
writeJSON(w, http.StatusOK, req)
|
||||
}
|
||||
|
||||
// cachedModelCatalog returns a usable cached catalog, or nil when the cache is
|
||||
// absent, cold, or unreadable. Cache problems must never fail the request — the
|
||||
// caller just falls back to the daemon round trip.
|
||||
func (h *Handler) cachedModelCatalog(ctx context.Context, runtimeID string) *ModelCatalogSnapshot {
|
||||
if h.ModelCatalogCache == nil {
|
||||
return nil
|
||||
}
|
||||
snapshot, err := h.ModelCatalogCache.Get(ctx, runtimeID)
|
||||
if err != nil {
|
||||
slog.Warn("model catalog cache read failed", "error", err, "runtime_id", runtimeID)
|
||||
return nil
|
||||
}
|
||||
if snapshot == nil || !cacheableModelCatalog(snapshot.Models, snapshot.Supported) {
|
||||
return nil
|
||||
}
|
||||
return snapshot
|
||||
}
|
||||
|
||||
// revalidateModelCatalog enqueues a background discovery request whose result
|
||||
// only updates the cache. No client polls it; the store's own timeout sweeps the
|
||||
// record if the daemon never answers.
|
||||
//
|
||||
// Stampede control: skip when a request for this runtime is already queued. A
|
||||
// running request that has already been claimed is not visible to HasPending, so
|
||||
// two opens in the same second can enqueue two refreshes — bounded, cheap
|
||||
// (the daemon memoizes discovery for 60s), and strictly better than the
|
||||
// alternative of never refreshing.
|
||||
func (h *Handler) revalidateModelCatalog(ctx context.Context, runtimeID string) {
|
||||
if h.ModelListStore == nil {
|
||||
return
|
||||
}
|
||||
pending, err := h.ModelListStore.HasPending(ctx, runtimeID)
|
||||
if err != nil {
|
||||
slog.Debug("model catalog revalidate probe failed", "error", err, "runtime_id", runtimeID)
|
||||
return
|
||||
}
|
||||
if pending {
|
||||
return
|
||||
}
|
||||
if _, err := h.ModelListStore.Create(ctx, runtimeID); err != nil {
|
||||
slog.Debug("model catalog revalidate enqueue failed", "error", err, "runtime_id", runtimeID)
|
||||
return
|
||||
}
|
||||
h.requestDaemonPendingWork(runtimeID, protocol.PendingWorkKindModelList)
|
||||
}
|
||||
|
||||
// requestDaemonPendingWork nudges the daemon to heartbeat now. Best-effort by
|
||||
// design: the daemon's scheduled heartbeat remains the correctness path, so a
|
||||
// missing notifier or an offline daemon only costs latency. Prefers the relay
|
||||
// notifier (reaches the API node holding the daemon's socket) and falls back to
|
||||
// the local hub, which is the whole cluster in single-node deployments.
|
||||
func (h *Handler) requestDaemonPendingWork(runtimeID, kind string) {
|
||||
if runtimeID == "" {
|
||||
return
|
||||
}
|
||||
if h.DaemonPendingWork != nil {
|
||||
h.DaemonPendingWork.NotifyPendingWork(runtimeID, kind)
|
||||
return
|
||||
}
|
||||
if h.DaemonHub != nil {
|
||||
h.DaemonHub.NotifyPendingWork(runtimeID, kind)
|
||||
}
|
||||
}
|
||||
|
||||
// GetModelListRequest returns the status of a model list request.
|
||||
func (h *Handler) GetModelListRequest(w http.ResponseWriter, r *http.Request) {
|
||||
requestID := chi.URLParam(r, "requestId")
|
||||
@@ -391,6 +499,25 @@ func (h *Handler) ReportModelListResult(w http.ResponseWriter, r *http.Request)
|
||||
writeError(w, http.StatusInternalServerError, "failed to persist completion")
|
||||
return
|
||||
}
|
||||
// Warm the catalog cache so the next picker open renders instantly
|
||||
// (MUL-5444). A cache write failure is not the daemon's problem — the
|
||||
// report itself succeeded — so it is logged, not surfaced.
|
||||
//
|
||||
// A completed-but-uncacheable result (empty catalog, or a runtime that
|
||||
// does not honour per-agent model selection) is the freshest truth we
|
||||
// have: drop any older snapshot instead of letting the fast path keep
|
||||
// serving a catalog the runtime no longer advertises. Failed reports
|
||||
// deliberately leave the cache alone — serving the last known good list
|
||||
// through a transient discovery failure is the point of the cache.
|
||||
if h.ModelCatalogCache != nil {
|
||||
if cacheableModelCatalog(body.Models, supported) {
|
||||
if err := h.ModelCatalogCache.Put(r.Context(), runtimeID, body.Models, supported); err != nil {
|
||||
slog.Warn("model catalog cache write failed", "error", err, "runtime_id", runtimeID)
|
||||
}
|
||||
} else if err := h.ModelCatalogCache.Invalidate(r.Context(), runtimeID); err != nil {
|
||||
slog.Warn("model catalog cache invalidate failed", "error", err, "runtime_id", runtimeID)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if err := h.ModelListStore.Fail(r.Context(), requestID, body.Error); err != nil {
|
||||
slog.Error("ModelListStore Fail failed", "error", err, "request_id", requestID)
|
||||
|
||||
@@ -132,6 +132,14 @@ const (
|
||||
EventDaemonTaskAvailable = "daemon:task_available"
|
||||
EventDaemonRuntimeProfilesChanged = "daemon:runtime_profiles_changed"
|
||||
EventDaemonWorkspacesChanged = "daemon:workspaces_changed"
|
||||
// EventDaemonPendingWork is a runtime-scoped hint that a heartbeat-carried
|
||||
// request (today: model-list discovery) is queued for that runtime. Without
|
||||
// it the daemon only learns about the request on its next scheduled
|
||||
// heartbeat, which adds up to one HeartbeatInterval (15s by default) of
|
||||
// dead wait to an interactive UI flow (MUL-5444). The hint carries no work
|
||||
// itself: the daemon still pulls the request through the normal heartbeat
|
||||
// claim, so a lost or duplicated hint is harmless.
|
||||
EventDaemonPendingWork = "daemon:pending_work"
|
||||
// Generic daemon→server request/response over the WebSocket control
|
||||
// connection (MUL-4257). The daemon sends EventDaemonRPCRequest with a
|
||||
// correlation id + method + body; the server replies EventDaemonRPCResponse
|
||||
|
||||
@@ -84,6 +84,25 @@ type RuntimeProfilesChangedPayload struct {
|
||||
// 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"`
|
||||
|
||||
Reference in New Issue
Block a user