diff --git a/src/lib/event-feed.ts b/src/lib/event-feed.ts index 411ce08..36ddd59 100644 --- a/src/lib/event-feed.ts +++ b/src/lib/event-feed.ts @@ -9,6 +9,13 @@ export function decodeEvent(msg: SseMessage): FeedEvent | null { const inner = JSON.parse(outer.data) as Record; if (outer.type === "inflight") { + const operation = inner.operation as string; + if (operation === "remove") { + const id = typeof inner.id === "string" ? inner.id : undefined; + if (id) return { type: "inflight", operation: "remove", id }; + return null; + } + if (operation !== "snapshot" && operation !== "upsert" && operation !== "add") return null; const requests: InflightRequest[] = Array.isArray(inner.requests) ? (inner.requests as InflightRequest[]) : Array.isArray(inner.request) @@ -18,7 +25,7 @@ export function decodeEvent(msg: SseMessage): FeedEvent | null { : []; return { type: "inflight", - operation: inner.operation as "snapshot" | "add" | "remove", + operation: operation === "snapshot" ? "snapshot" : "add", requests, }; } diff --git a/src/lib/inflight-tracker.ts b/src/lib/inflight-tracker.ts index ad5dbba..63ed038 100644 --- a/src/lib/inflight-tracker.ts +++ b/src/lib/inflight-tracker.ts @@ -13,29 +13,25 @@ export interface ModelState { } export type FeedEvent = - | { type: "inflight"; operation: "snapshot" | "add" | "remove"; requests: InflightRequest[] } + | { type: "inflight"; operation: "snapshot" | "add"; requests: InflightRequest[] } + | { type: "inflight"; operation: "remove"; id: string } | { type: "modelStatus"; models: ModelState[] }; export type ModelRuntimeState = "stopped" | "loading" | "ready"; export class InflightTracker { - private counts = new Map(); + private requests = new Map(); private states = new Map(); apply(event: FeedEvent): void { if (event.type === "inflight") { if (event.operation === "snapshot") { - const counts = new Map(); - for (const r of event.requests) counts.set(r.model, (counts.get(r.model) ?? 0) + 1); - this.counts = counts; + this.requests = new Map(); + for (const r of event.requests) if (r.id) this.requests.set(r.id, r.model); } else if (event.operation === "add") { - for (const r of event.requests) this.counts.set(r.model, (this.counts.get(r.model) ?? 0) + 1); + for (const r of event.requests) if (r.id) this.requests.set(r.id, r.model); } else if (event.operation === "remove") { - for (const r of event.requests) { - const c = this.counts.get(r.model) ?? 0; - if (c <= 1) this.counts.delete(r.model); - else this.counts.set(r.model, c - 1); - } + this.requests.delete(event.id); } } else if (event.type === "modelStatus") { for (const m of event.models) this.states.set(m.id, normalizeState(m.state)); @@ -43,7 +39,9 @@ export class InflightTracker { } count(modelId: string): number { - return this.counts.get(modelId) ?? 0; + let n = 0; + for (const model of this.requests.values()) if (model === modelId) n++; + return n; } state(modelId: string): ModelRuntimeState | undefined { diff --git a/tests/event-feed.test.ts b/tests/event-feed.test.ts index 52167bd..2a91006 100644 --- a/tests/event-feed.test.ts +++ b/tests/event-feed.test.ts @@ -41,6 +41,32 @@ test("decodeEvent parses an inflight add with a singular request field", () => { } }); +test("decodeEvent parses an inflight upsert as an add", () => { + const msg: SseMessage = { + event: "message", + data: JSON.stringify({ + type: "inflight", + data: JSON.stringify({ operation: "upsert", request: { id: "820", model: "DeepSeek-V4-Flash-0731", elapsed_ms: 103 } }), + }), + }; + const ev = decodeEvent(msg); + assert.ok(ev && ev.type === "inflight"); + if (ev && ev.type === "inflight") { + assert.equal(ev.operation, "add"); + assert.equal(ev.requests[0].id, "820"); + assert.equal(ev.requests[0].elapsed_ms, 103); + } +}); + +test("decodeEvent parses an inflight remove with a bare id", () => { + const msg: SseMessage = { + event: "message", + data: JSON.stringify({ type: "inflight", data: JSON.stringify({ operation: "remove", id: "817" }) }), + }; + const ev = decodeEvent(msg); + assert.deepEqual(ev, { type: "inflight", operation: "remove", id: "817" }); +}); + test("decodeEvent parses modelStatus", () => { const msg: SseMessage = { event: "message", diff --git a/tests/inflight-tracker.test.ts b/tests/inflight-tracker.test.ts index 0396cbe..3f4b443 100644 --- a/tests/inflight-tracker.test.ts +++ b/tests/inflight-tracker.test.ts @@ -18,17 +18,32 @@ test("snapshot rebuilds counts for all in-flight requests", () => { assert.equal(tracker.count("C"), 0); }); -test("add and remove adjust per-model counts", () => { +test("add (upsert) is idempotent per request id", () => { const tracker = new InflightTracker(); tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }] }); + tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }] }); tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "2" }] }); assert.equal(tracker.count("A"), 2); - tracker.apply({ type: "inflight", operation: "remove", requests: [{ model: "A", id: "1" }] }); +}); + +test("remove is id-based and decrements the matching request", () => { + const tracker = new InflightTracker(); + tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }, { model: "B", id: "2" }] }); assert.equal(tracker.count("A"), 1); - tracker.apply({ type: "inflight", operation: "remove", requests: [{ model: "A", id: "2" }] }); - assert.equal(tracker.count("A"), 0); - tracker.apply({ type: "inflight", operation: "remove", requests: [{ model: "A", id: "9" }] }); + assert.equal(tracker.count("B"), 1); + tracker.apply({ type: "inflight", operation: "remove", id: "1" }); assert.equal(tracker.count("A"), 0); + assert.equal(tracker.count("B"), 1); + tracker.apply({ type: "inflight", operation: "remove", id: "999" }); + assert.equal(tracker.count("B"), 1); +}); + +test("snapshot replaces prior in-flight state", () => { + const tracker = new InflightTracker(); + tracker.apply({ type: "inflight", operation: "add", requests: [{ model: "A", id: "1" }] }); + tracker.apply({ type: "inflight", operation: "snapshot", requests: [{ model: "A", id: "2" }] }); + assert.equal(tracker.count("A"), 1); + assert.equal(tracker.count("B"), 0); }); test("modelStatus normalizes states", () => {