fix(core,web): retry transport and quota LLM errors, disable dead sessions on auth failure (#82)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
1 parent
c369e089a7
commit
3d3495528c
54 files changed
+2278
-161
No files matched your search
@@ -234,6 +234,13 @@ export interface ModelsResponse {
|
||||
defaultModel?: ModelRefDto;
|
||||
/** Vision model used as a proxy reader for read_image (describes images when the session model has vision=false). */
|
||||
visionModel?: ModelRefDto;
|
||||
/**
|
||||
* When the Project's model/credential config last changed (ISO; the config file's mtime,
|
||||
* so it survives restarts). The web's auth-dead gate compares it against the last auth
|
||||
* abort: an abort OLDER than the last credential update no longer disables the composer
|
||||
* (the key was fixed since). Absent when the Project has no config file yet.
|
||||
*/
|
||||
updatedAt?: string;
|
||||
models: ModelInfo[];
|
||||
}
|
||||
|
||||
@@ -609,6 +616,16 @@ export interface ApprovalDecisionRequest {
|
||||
decision: "allow" | "deny";
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /api/sessions/:sessionId/retry-now — skip the in-progress reconnect backoff and
|
||||
* fire the next retry immediately (the "retry now" button on the reconnect countdown).
|
||||
* `skipped: false` is the benign "no reconnect wait in progress" case (idle session, or
|
||||
* the wait elapsed in a timing race), not an error.
|
||||
*/
|
||||
export interface RetryNowResponse {
|
||||
skipped: boolean;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// SSE server events (OmniMessage uses the default event, only server_event here)
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -626,6 +643,14 @@ export type ServerEvent =
|
||||
| { type: "session_title"; sessionId: string; title: string }
|
||||
/** Last-Event-ID has been evicted from the buffer: the frontend should re-fetch the history endpoint before continuing to consume this connection. */
|
||||
| { type: "resync_required" }
|
||||
/**
|
||||
* The Project's model credentials changed (PUT /models): cached runtimes have been
|
||||
* invalidated server-side, so an auth-dead Session can continue — the frontend clears
|
||||
* its auth-dead composer state immediately. Published to every existing Session channel
|
||||
* of the Project; tabs without a live channel learn the same fact from the models
|
||||
* response's `updatedAt` on their next load.
|
||||
*/
|
||||
| { type: "credentials_updated" }
|
||||
/** Placeholder handshake on the user channel (reserved for automated task notifications). */
|
||||
| { type: "hello" }
|
||||
/** New session registered (pushed over the parent session's channel for subagent sessions): frontend refreshes the list in place. */
|
||||
|
||||
@@ -10,11 +10,25 @@ import type {
|
||||
ModelsUpdateRequest,
|
||||
ModelTestRequest,
|
||||
ModelUpdateEntry,
|
||||
ServerEvent,
|
||||
} from "../../api/types.js";
|
||||
import type { AppEnv } from "../../auth/middleware.js";
|
||||
import { badRequest, readJson, requireString, requireValidId } from "../validate.js";
|
||||
import type { AppDeps } from "../../app.js";
|
||||
|
||||
/**
|
||||
* Live unlock for auth-dead composers: after a models/credential update, publish
|
||||
* `credentials_updated` to every EXISTING channel of this Project's Sessions (`peek` —
|
||||
* never creates channels: a tab that isn't subscribed learns the same fact from the models
|
||||
* response's `updatedAt` when it next loads).
|
||||
*/
|
||||
function publishCredentialsUpdated(deps: AppDeps, projectId: string): void {
|
||||
const event: ServerEvent = { type: "credentials_updated" };
|
||||
for (const row of deps.sessionsRepo.listByProject(projectId)) {
|
||||
deps.channels.peek(row.sessionId)?.publish(event, "server_event");
|
||||
}
|
||||
}
|
||||
|
||||
/** Validate a paired reference object ({ provider, modelId }); shape mismatch throws 400. */
|
||||
function parseRef(value: unknown, label: string): ModelRefDto {
|
||||
if (value === null || typeof value !== "object" || Array.isArray(value)) {
|
||||
@@ -143,7 +157,15 @@ export function modelsRoutes(deps: AppDeps): Hono<AppEnv> {
|
||||
const projectId = requireValidId(c, "projectId");
|
||||
deps.projectService.requireProjectOwner(c.var.user.userId, projectId);
|
||||
const req = parseModelsUpdate(await readJson(c));
|
||||
return c.json(await deps.projectConfigService.updateModels(projectId, req));
|
||||
const res = await deps.projectConfigService.updateModels(projectId, req);
|
||||
// Effective-value semantics (mirrors the vault route): no hot swap into a Task already
|
||||
// in flight, but every cached runtime in this Project is invalidated — the next Task on
|
||||
// any of its Sessions re-resumes and reads the new api_key / base_url. Without this, a
|
||||
// key edit would not reach a loaded Session until the 30-minute idle sweep or a restart.
|
||||
deps.manager.invalidateProjectRuntimes(projectId);
|
||||
// Live unlock: open tabs clear their auth-dead composer immediately (no reload needed).
|
||||
publishCredentialsUpdated(deps, projectId);
|
||||
return c.json(res);
|
||||
});
|
||||
|
||||
// Connectivity test (owner): the model reference `(provider, modelId)` is sent as a pair
|
||||
|
||||
@@ -21,6 +21,7 @@ import type {
|
||||
SessionCreateResponse,
|
||||
SessionResponse,
|
||||
SessionsResponse,
|
||||
RetryNowResponse,
|
||||
TaskCreateResponse,
|
||||
} from "../../api/types.js";
|
||||
import { PREVIEW_TOKEN_TTL_MS, resolvePreviewTarget } from "../../services/preview-token.js";
|
||||
@@ -420,6 +421,16 @@ export function sessionsRoutes(deps: AppDeps): Hono<AppEnv> {
|
||||
return c.body(null, aborted ? 202 : 204);
|
||||
});
|
||||
|
||||
// "Retry now" on the reconnect countdown: skip the remaining backoff wait and fire the
|
||||
// next retry immediately (attempt counter unchanged). Benign either way — 200 with
|
||||
// skipped:false when no reconnect wait is in progress, so a timing race (the wait
|
||||
// elapsed just before the click) never surfaces as an error.
|
||||
app.post("/:sessionId/retry-now", (c) => {
|
||||
const row = resolveSession(c);
|
||||
const skipped = deps.manager.retryNow(row.sessionId);
|
||||
return c.json({ skipped } satisfies RetryNowResponse);
|
||||
});
|
||||
|
||||
app.post("/:sessionId/compact", async (c) => {
|
||||
const row = resolveSession(c);
|
||||
const { sessionId } = await deps.manager.startCompact(row.sessionId);
|
||||
|
||||
@@ -78,6 +78,8 @@ export interface RuntimeSession {
|
||||
compactability(): CompactAvailability;
|
||||
/** Queues a mid-run steering message (core `Session.steer`); false when no Task is running. */
|
||||
steer(text: string): boolean;
|
||||
/** Skips the in-progress reconnect backoff, firing the next retry immediately (core `Session.skipReconnectWait`); false when no wait is in progress. */
|
||||
skipReconnectWait(): boolean;
|
||||
toolPermission(name: string): "r" | "rw" | undefined;
|
||||
/**
|
||||
* Out-of-band one-shot request for title generation (core `Session.generateTitle`,
|
||||
@@ -396,6 +398,22 @@ export class SessionManager {
|
||||
this.agentGenerations.set(key, this.generationOf(projectId, agentId) + 1);
|
||||
}
|
||||
|
||||
/**
|
||||
* After a Project's models/credentials change: invalidate every cached runtime in this
|
||||
* Project, so the next Task re-resumes with the new api_key / base_url. Same
|
||||
* effective-value semantics as invalidateAgentRuntimes — no hot swap into a Task already
|
||||
* in flight. Iterating the active table is complete here, not a shortcut: the generation
|
||||
* map only matters for runtimes that are already cached, and an Agent with no cached
|
||||
* entry builds fresh through the loader anyway.
|
||||
*/
|
||||
invalidateProjectRuntimes(projectId: string): void {
|
||||
const agentIds = new Set<string>();
|
||||
for (const e of this.entries.values()) {
|
||||
if (e.projectId === projectId) agentIds.add(e.agentId);
|
||||
}
|
||||
for (const agentId of agentIds) this.invalidateAgentRuntimes(projectId, agentId);
|
||||
}
|
||||
|
||||
// —— Task / compaction drive ——
|
||||
|
||||
/**
|
||||
@@ -558,6 +576,23 @@ export class SessionManager {
|
||||
entry.lastActivityMs = Date.now();
|
||||
}
|
||||
|
||||
/**
|
||||
* "Retry now" on the reconnect countdown: skip the in-progress backoff wait and fire
|
||||
* the next retry immediately (the attempt counter is unchanged — the skipped wait does
|
||||
* not consume an extra attempt). Returns false as a benign no-op when the session has
|
||||
* no active runtime or no reconnect wait is in progress — timing races (the wait
|
||||
* elapsed right before the click) must not surface as errors. Mirrors the steer seam:
|
||||
* manager → RuntimeSession → core Session → engine.
|
||||
*/
|
||||
retryNow(sessionId: string): boolean {
|
||||
const entry = this.entries.get(sessionId);
|
||||
// Both running Tasks and compactions can be parked in a reconnect backoff.
|
||||
if (!entry || entry.status === "idle") return false;
|
||||
const skipped = entry.session.skipReconnectWait();
|
||||
if (skipped) entry.lastActivityMs = Date.now();
|
||||
return skipped;
|
||||
}
|
||||
|
||||
/**
|
||||
* Interrupt the current Task/compaction: pending approvals converge to deny first,
|
||||
* then the AbortSignal fires. Returns false if nothing is in progress (the route
|
||||
|
||||
@@ -9,21 +9,23 @@
|
||||
* (its state wraps up accordingly, see close).
|
||||
*
|
||||
* LLM (source = `llm`): reads the status of `request_end` —
|
||||
* - `failed` → unexpected (not retryable: auth failure, invalid params, etc., needs a human);
|
||||
* - `failed` / `auth` → unexpected (not retryable: credentials rejected, invalid params,
|
||||
* etc., needs a human; both share the llm_failed code — no separate taxonomy);
|
||||
* - `timeout` / `malformed` → expected (the engine already reconnects and retries, part
|
||||
* of normal operation);
|
||||
* - `aborted` / `completed` are not recorded (the former is a user-initiated interrupt,
|
||||
* not an error).
|
||||
*
|
||||
* The message uses the real reason: `request_end` only carries status, and **the only
|
||||
* place core carries the actual failure-reason text is the `abort` event's reason**
|
||||
* (e.g. `llm request error: 401 …` / `malformed response failed after N retries`). So a
|
||||
* `request_end` failure is first held pending, not persisted immediately, and is
|
||||
* resolved at the next request boundary:
|
||||
* The message uses the real reason: `request_end` carries the failure detail on
|
||||
* non-completed statuses (`message`, from LLMOutcome), and the `abort` event's reason is
|
||||
* core's failure-reason prose (e.g. `llm request error: 401 …` / `malformed response
|
||||
* failed after N retries`). A `request_end` failure is first held pending (status + its
|
||||
* own detail), not persisted immediately, and is resolved at the next request boundary:
|
||||
* - Immediately followed by `abort` → use its reason as the message (the real reason);
|
||||
* - Immediately followed by `request_begin` (the engine is retrying) → no reason text
|
||||
* left to wait for, use the status text;
|
||||
* - Still unresolved when the run ends → close persists it as a fallback.
|
||||
* - Immediately followed by `request_begin` (the engine is retrying — no abort will ever
|
||||
* arrive) → use the staged request_end's own `message` (the real detail, e.g. a quota
|
||||
* code), falling back to the generic status text when it carried none;
|
||||
* - Still unresolved when the run ends → close persists it the same way.
|
||||
* Exception: when reason is a user-interrupt message (`aborted …`), it's not trusted —
|
||||
* "the user clicked stop during backoff" isn't the reason for this timeout, so the
|
||||
* status text is used instead (that timeout is a genuine failure and is still recorded).
|
||||
@@ -71,11 +73,14 @@ export const TOOL_NAMES_MAX = 1000;
|
||||
export const ORIGIN_CTX_MAX = 200;
|
||||
|
||||
/** Recorded LLM failure states (`aborted` / `completed` are not errors and aren't included here). */
|
||||
type LlmFailure = "failed" | "timeout" | "malformed";
|
||||
type LlmFailure = "failed" | "timeout" | "malformed" | "auth";
|
||||
|
||||
/** LLM failure state → error code, classification, and fallback message (used when the abort reason isn't available). */
|
||||
const LLM_FAILURES: Record<LlmFailure, { code: string; kind: ErrorKind; text: string }> = {
|
||||
failed: { code: "llm_failed", kind: "unexpected", text: "LLM request failed (not retryable)." },
|
||||
// Credentials rejection shares the failed bucket (no new taxonomy): same code/kind, the
|
||||
// real reason arrives via the abort reason or the event's own message as usual.
|
||||
auth: { code: "llm_failed", kind: "unexpected", text: "LLM request failed (not retryable)." },
|
||||
timeout: {
|
||||
code: "llm_timeout",
|
||||
kind: "expected",
|
||||
@@ -92,7 +97,7 @@ const LLM_FAILURES: Record<LlmFailure, { code: string; kind: ErrorKind; text: st
|
||||
type ToolFailure = "failed" | "timeout";
|
||||
|
||||
function isLlmFailure(s: unknown): s is LlmFailure {
|
||||
return s === "failed" || s === "timeout" || s === "malformed";
|
||||
return s === "failed" || s === "timeout" || s === "malformed" || s === "auth";
|
||||
}
|
||||
|
||||
function isToolFailure(s: unknown): s is ToolFailure {
|
||||
@@ -123,8 +128,8 @@ function toolFailureText(output: string): string {
|
||||
}
|
||||
|
||||
export class StreamErrorWatcher {
|
||||
/** LLM failures awaiting a real reason: origin → failure state (see file header; each session has at most one in-flight Request). */
|
||||
private readonly pending = new Map<string, LlmFailure>();
|
||||
/** LLM failures awaiting a real reason: origin → failure state + the request_end's own detail (see file header; each session has at most one in-flight Request). */
|
||||
private readonly pending = new Map<string, { status: LlmFailure; message?: string }>();
|
||||
/** Names of in-flight tool calls: `origin \0 tool_call_id` → tool name (dequeued once output arrives). */
|
||||
private readonly toolNames = new Map<string, string>();
|
||||
/** Subagent identity: origin → that subagent's `{agentId, sessionId}` (see the file header's attribution section). */
|
||||
@@ -196,11 +201,23 @@ export class StreamErrorWatcher {
|
||||
// —— LLM ——
|
||||
|
||||
private observeLlm(msg: OmniMessage): void {
|
||||
const p = msg.payload as { type?: string; status?: StopReason; reason?: string | null };
|
||||
const p = msg.payload as {
|
||||
type?: string;
|
||||
status?: StopReason;
|
||||
reason?: string | null;
|
||||
message?: string;
|
||||
};
|
||||
const key = originKey(msg);
|
||||
if (p.type === "request_end") {
|
||||
this.flush(key); // Defensive: if a previous failure is still pending (normally resolved by request_begin), persist it first
|
||||
if (isLlmFailure(p.status)) this.pending.set(key, p.status);
|
||||
if (isLlmFailure(p.status)) {
|
||||
this.pending.set(key, {
|
||||
status: p.status,
|
||||
// The event's own failure detail (LLMOutcome.message): the message of record when
|
||||
// no abort follows (the retry path).
|
||||
...(typeof p.message === "string" && p.message.trim() ? { message: p.message } : {}),
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
// A new attempt begins (the engine is retrying): no reason text left to wait for the previous failure, persist using the status text.
|
||||
@@ -221,13 +238,15 @@ export class StreamErrorWatcher {
|
||||
* up from it (see file header).
|
||||
*/
|
||||
private flush(key: string, reason?: string | null): void {
|
||||
const status = this.pending.get(key);
|
||||
if (status === undefined) return;
|
||||
const entry = this.pending.get(key);
|
||||
if (entry === undefined) return;
|
||||
this.pending.delete(key);
|
||||
const spec = LLM_FAILURES[status];
|
||||
const spec = LLM_FAILURES[entry.status];
|
||||
const trimmed = reason?.trim();
|
||||
// A user-interrupt message isn't a failure reason (see file header); fall back to the status text.
|
||||
const message = trimmed && !isUserAbortReason(trimmed) ? trimmed : spec.text;
|
||||
// Message priority: the abort reason (core's failure prose) → the staged request_end's
|
||||
// own detail (the retry path: no abort ever arrives) → the generic status text. A
|
||||
// user-interrupt message isn't a failure reason (see file header).
|
||||
const message = trimmed && !isUserAbortReason(trimmed) ? trimmed : (entry.message ?? spec.text);
|
||||
this.errors.record({
|
||||
source: "llm",
|
||||
err: message,
|
||||
|
||||
@@ -121,6 +121,20 @@ export class ProjectConfigService {
|
||||
return projectConfigPath(this.root, projectId);
|
||||
}
|
||||
|
||||
/**
|
||||
* When the Project config (models/credentials) last changed: the config file's mtime as
|
||||
* ISO — an honest persistent source that survives server restarts (every models update
|
||||
* rewrites the file). undefined when the file doesn't exist yet.
|
||||
*/
|
||||
private async configUpdatedAt(projectId: string): Promise<string | undefined> {
|
||||
try {
|
||||
const st = await fs.stat(this.filePath(projectId));
|
||||
return st.mtime.toISOString();
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/** Reads the raw TOML object; returns an empty object if the file doesn't exist (does not write to disk). */
|
||||
async readRaw(projectId: string): Promise<RawTable> {
|
||||
let raw: string;
|
||||
@@ -390,9 +404,11 @@ export class ProjectConfigService {
|
||||
provider: ref.provider,
|
||||
modelId: ref.model_id,
|
||||
});
|
||||
const updatedAt = await this.configUpdatedAt(projectId);
|
||||
return {
|
||||
...(defaultRef !== undefined ? { defaultModel: toDto(defaultRef) } : {}),
|
||||
...(visionRef !== undefined ? { visionModel: toDto(visionRef) } : {}),
|
||||
...(updatedAt !== undefined ? { updatedAt } : {}),
|
||||
models,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -406,6 +406,66 @@ describe("stream-error-watcher (LLM / Environment errors)", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("request_end(auth) shares the llm_failed/unexpected bucket; message takes the abort reason", () => {
|
||||
// Credentials rejection is its own stop reason but no new error taxonomy: same code
|
||||
// and kind as failed, resolved by the abort that follows like any failed exit.
|
||||
const got = feed([
|
||||
requestBegin(),
|
||||
requestEnd("auth", "401 invalid x-api-key (invalid_api_key)"),
|
||||
abortEvent("llm request error: 401 invalid x-api-key (invalid_api_key)"),
|
||||
]);
|
||||
expect(got).toHaveLength(1);
|
||||
expect(got[0]).toMatchObject({
|
||||
source: "llm",
|
||||
kind: "unexpected",
|
||||
code: "llm_failed",
|
||||
message: "llm request error: 401 invalid x-api-key (invalid_api_key)",
|
||||
});
|
||||
});
|
||||
|
||||
it("a retried failure keeps its real detail: request_end(timeout).message lands in the record", () => {
|
||||
// The retry path: the engine reconnects (request_begin) and eventually succeeds, so no
|
||||
// abort ever arrives for the staged failure — the request_end's own failure detail
|
||||
// (LLMOutcome.message, e.g. a quota code) is the message of record, not the generic
|
||||
// "timed out" status text. This is what the Cost center shows for a retried quota 403.
|
||||
const got = feed([
|
||||
requestBegin(),
|
||||
requestEnd("timeout", "403 no active subscription (insufficient_user_quota)"),
|
||||
requestBegin(),
|
||||
requestEnd("completed"),
|
||||
]);
|
||||
expect(got).toHaveLength(1);
|
||||
expect(got[0]).toMatchObject({
|
||||
source: "llm",
|
||||
kind: "expected",
|
||||
code: "llm_timeout",
|
||||
message: "403 no active subscription (insufficient_user_quota)",
|
||||
});
|
||||
});
|
||||
|
||||
it("an abort reason still outranks the staged request_end detail (failed exit path)", () => {
|
||||
const got = feed([
|
||||
requestBegin(),
|
||||
requestEnd("failed", "401 invalid x-api-key (invalid_api_key)"),
|
||||
abortEvent("llm request error: 401 invalid x-api-key (invalid_api_key)"),
|
||||
]);
|
||||
expect(got).toHaveLength(1);
|
||||
// The abort's prose (with core's "llm request error" framing) wins over the raw detail.
|
||||
expect(got[0]!.message).toBe("llm request error: 401 invalid x-api-key (invalid_api_key)");
|
||||
});
|
||||
|
||||
it("interrupt during backoff with a staged detail: the detail wins over the status text", () => {
|
||||
// The interrupt message is distrusted (not the failure's reason), but the staged
|
||||
// request_end detail IS the failure's reason — prefer it over the generic status text.
|
||||
const got = feed([
|
||||
requestBegin(),
|
||||
requestEnd("timeout", "403 quota exceeded (insufficient_user_quota)"),
|
||||
abortEvent("aborted during reconnect backoff"),
|
||||
]);
|
||||
expect(got).toHaveLength(1);
|
||||
expect(got[0]!.message).toBe("403 quota exceeded (insufficient_user_quota)");
|
||||
});
|
||||
|
||||
it("aborted (user clicked Stop) is not an error: not recorded; neither is completed", () => {
|
||||
expect(
|
||||
feed([
|
||||
|
||||
@@ -25,6 +25,7 @@ function parkingFakeSession(sessionId: string, runs: string[][]): RuntimeSession
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run(input: OmniMessage[], opts: { approve: ApproveFn; signal: AbortSignal }) {
|
||||
runs.push(input.map((m) => (m.payload as { text?: string }).text ?? ""));
|
||||
const tc = toolCall({ name: "exec_command", arguments: "{}", toolCallId: "tc-fu" });
|
||||
|
||||
@@ -13,7 +13,7 @@ import type { AddressInfo } from "node:net";
|
||||
import { readFile, stat } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import { MODEL_CATALOG } from "@prismshadow/penguin-core";
|
||||
import { MODEL_CATALOG, userText } from "@prismshadow/penguin-core";
|
||||
import type {
|
||||
ModelsResponse,
|
||||
ModelTestResponse,
|
||||
@@ -21,7 +21,9 @@ import type {
|
||||
SessionCreateResponse,
|
||||
} from "../src/api/types.js";
|
||||
import { ProjectConfigService } from "../src/services/project-config-service.js";
|
||||
import { apiClient, createTestApp, loginAdmin, provisionUser } from "./helpers.js";
|
||||
import type { RuntimeSession } from "../src/runtime/session-manager.js";
|
||||
import type { ChannelEvent } from "../src/runtime/channel.js";
|
||||
import { apiClient, createTestApp, loginAdmin, provisionUser, waitFor } from "./helpers.js";
|
||||
import type { TestApp } from "./helpers.js";
|
||||
|
||||
/** Catalog paired refs (config primary key = (provider, model_id)). */
|
||||
@@ -586,3 +588,134 @@ describe("模型引用改键与连通性测试", () => {
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("models update reaches loaded sessions (invalidation + live unlock)", () => {
|
||||
let t: TestApp;
|
||||
let api: ReturnType<typeof apiClient>;
|
||||
let projectId: string;
|
||||
/** Loader call count: how many times the manager (re)built a runtime from the index. */
|
||||
let loads: number;
|
||||
|
||||
/** Minimal fake runtime Session (no LLM calls; mirrors vault.test.ts). */
|
||||
const fakeRuntimeSession = (sessionId: string): RuntimeSession => ({
|
||||
sessionId,
|
||||
toolPermission: () => "rw",
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run() {},
|
||||
async *compact() {},
|
||||
});
|
||||
|
||||
const insertSession = (sessionId: string, project: string): void => {
|
||||
t.deps.sessionsRepo.insert({
|
||||
sessionId,
|
||||
projectId: project,
|
||||
agentId: "default_agent",
|
||||
modelId: "m1",
|
||||
provider: "custom",
|
||||
workspace: t.root,
|
||||
approvalMode: "allow-all",
|
||||
title: null,
|
||||
createdAt: new Date().toISOString(),
|
||||
});
|
||||
};
|
||||
|
||||
const putModels = (apiKey: string) =>
|
||||
api.put(`/api/projects/${projectId}/models`, {
|
||||
defaultModel: { provider: "custom", modelId: "m-inv" },
|
||||
models: [{ provider: "custom", modelId: "m-inv", apiKey }],
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
loads = 0;
|
||||
t = await createTestApp({
|
||||
loader: {
|
||||
load: async (row) => {
|
||||
loads++;
|
||||
return fakeRuntimeSession(row.sessionId);
|
||||
},
|
||||
},
|
||||
});
|
||||
const { cookie } = await provisionUser(t.app, "inv_owner");
|
||||
api = apiClient(t.app, cookie);
|
||||
const created = (await (
|
||||
await api.post("/api/projects", { projectId: "inv_owner-models", name: "invalidation" })
|
||||
).json()) as ProjectCreateResponse;
|
||||
projectId = created.project.projectId;
|
||||
});
|
||||
afterEach(async () => {
|
||||
await t.cleanup();
|
||||
});
|
||||
|
||||
it("PUT invalidates the Project's cached Session runtimes: the next Task re-resumes with the new key", async () => {
|
||||
insertSession("models-sess-1", projectId);
|
||||
const idle = () => t.deps.manager.statusOf("models-sess-1") === "idle";
|
||||
|
||||
// First Task builds the runtime (load #1); the second reuses the active-table entry.
|
||||
await t.deps.manager.startTask("models-sess-1", [userText("a")]);
|
||||
await waitFor(idle);
|
||||
await t.deps.manager.startTask("models-sess-1", [userText("b")]);
|
||||
await waitFor(idle);
|
||||
expect(loads).toBe(1);
|
||||
|
||||
// Key update via the API: the cached runtime is stale, so the next Task re-resumes
|
||||
// (the loader re-reads the Project config — the new api_key reaches the next Task).
|
||||
expect((await putModels("sk-fresh-key-000111")).status).toBe(200);
|
||||
await t.deps.manager.startTask("models-sess-1", [userText("c")]);
|
||||
await waitFor(idle);
|
||||
expect(loads).toBe(2);
|
||||
|
||||
// Reads don't invalidate: the rebuilt runtime is reused.
|
||||
expect((await api.get(`/api/projects/${projectId}/models`)).status).toBe(200);
|
||||
await t.deps.manager.startTask("models-sess-1", [userText("d")]);
|
||||
await waitFor(idle);
|
||||
expect(loads).toBe(2);
|
||||
});
|
||||
|
||||
it("PUT publishes credentials_updated to the Project's existing session channels only", async () => {
|
||||
insertSession("models-sess-live", projectId);
|
||||
insertSession("models-sess-cold", projectId);
|
||||
// A session of ANOTHER project must not receive the event.
|
||||
const other = (await (
|
||||
await api.post("/api/projects", { projectId: "inv_owner-other", name: "other" })
|
||||
).json()) as ProjectCreateResponse;
|
||||
insertSession("other-sess", other.project.projectId);
|
||||
|
||||
// Only the "live" session has an open channel (a subscribed tab); "cold" has none.
|
||||
const events: ChannelEvent[] = [];
|
||||
t.deps.channels.get("models-sess-live").subscribe((e) => events.push(e));
|
||||
const otherEvents: ChannelEvent[] = [];
|
||||
t.deps.channels.get("other-sess").subscribe((e) => otherEvents.push(e));
|
||||
|
||||
expect((await putModels("sk-live-key-000222")).status).toBe(200);
|
||||
|
||||
const types = events
|
||||
.filter((e) => e.event === "server_event")
|
||||
.map((e) => (JSON.parse(e.data) as { type: string }).type);
|
||||
expect(types).toContain("credentials_updated");
|
||||
// Cross-project channels stay silent, and no channel is created for unsubscribed sessions.
|
||||
expect(
|
||||
otherEvents
|
||||
.filter((e) => e.event === "server_event")
|
||||
.map((e) => (JSON.parse(e.data) as { type: string }).type),
|
||||
).not.toContain("credentials_updated");
|
||||
expect(t.deps.channels.peek("models-sess-cold")).toBeUndefined();
|
||||
});
|
||||
|
||||
it("GET/PUT responses expose updatedAt (config file mtime) for the web's auth-dead gate", async () => {
|
||||
const before = Date.now() - 60_000;
|
||||
const put = await putModels("sk-ts-key-000333");
|
||||
expect(put.status).toBe(200);
|
||||
const putBody = (await put.json()) as ModelsResponse;
|
||||
expect(typeof putBody.updatedAt).toBe("string");
|
||||
expect(Date.parse(putBody.updatedAt!)).toBeGreaterThan(before);
|
||||
|
||||
const get = await api.get(`/api/projects/${projectId}/models`);
|
||||
const getBody = (await get.json()) as ModelsResponse;
|
||||
expect(typeof getBody.updatedAt).toBe("string");
|
||||
// Reads don't bump it: still the PUT's write time (same file mtime).
|
||||
expect(Date.parse(getBody.updatedAt!)).toBeGreaterThanOrEqual(Date.parse(putBody.updatedAt!));
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,99 @@
|
||||
/**
|
||||
* Integration tests for POST /api/sessions/:id/retry-now ("retry now" on the reconnect
|
||||
* countdown — skips the in-progress backoff wait):
|
||||
* - 200 {skipped:true} while the runtime reports a wait was skipped;
|
||||
* - 200 {skipped:false} as a benign no-op when the session is idle or no wait is in
|
||||
* progress (timing races must not surface as errors);
|
||||
* - 404 for foreign/unknown sessions (via the shared resolveSession lookup).
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import { approvalDecision, assistantText, toolCall, userText } from "@prismshadow/penguin-core";
|
||||
import type { ApproveFn, OmniMessage } from "@prismshadow/penguin-core";
|
||||
import type { RetryNowResponse } from "../src/api/types.js";
|
||||
import type { SessionRow } from "../src/db/repos/sessions.js";
|
||||
import type { RuntimeSession } from "../src/runtime/session-manager.js";
|
||||
import { apiClient, createTestApp, provisionUser, waitFor } from "./helpers.js";
|
||||
import type { TestApp } from "./helpers.js";
|
||||
|
||||
const SID = "session-2026-07-27-10-00-00-ccdd0002";
|
||||
|
||||
/** Fake Session that parks on one approval (keeps the Task running) and records skip calls. */
|
||||
function backoffFakeSession(sessionId: string, skips: number[]): RuntimeSession {
|
||||
return {
|
||||
sessionId,
|
||||
toolPermission: () => "rw",
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => {
|
||||
skips.push(Date.now());
|
||||
return true; // pretend a reconnect wait was in progress and got skipped
|
||||
},
|
||||
async *run(_input: OmniMessage[], opts: { approve: ApproveFn; signal: AbortSignal }) {
|
||||
const tc = toolCall({ name: "exec_command", arguments: "{}", toolCallId: "tc-retry" });
|
||||
yield tc;
|
||||
const decision = await opts.approve(tc);
|
||||
yield approvalDecision(decision, "tc-retry");
|
||||
yield assistantText("done");
|
||||
},
|
||||
async *compact() {},
|
||||
};
|
||||
}
|
||||
|
||||
describe("retry-now route", () => {
|
||||
let t: TestApp;
|
||||
let api: ReturnType<typeof apiClient>;
|
||||
let outsider: ReturnType<typeof apiClient>;
|
||||
let skips: number[];
|
||||
|
||||
beforeEach(async () => {
|
||||
t = await createTestApp();
|
||||
const { cookie } = await provisionUser(t.app, "retrier");
|
||||
const other = await provisionUser(t.app, "outsider_r");
|
||||
api = apiClient(t.app, cookie);
|
||||
outsider = apiClient(t.app, other.cookie);
|
||||
const row: SessionRow = {
|
||||
sessionId: SID,
|
||||
projectId: "retrier-default_project",
|
||||
agentId: "default_agent",
|
||||
modelId: "m1",
|
||||
provider: "custom",
|
||||
workspace: "/tmp/w",
|
||||
approvalMode: "always-ask",
|
||||
title: null,
|
||||
createdAt: new Date().toISOString(),
|
||||
};
|
||||
t.deps.sessionsRepo.insert(row);
|
||||
skips = [];
|
||||
t.deps.manager.adopt(row, backoffFakeSession(SID, skips));
|
||||
});
|
||||
afterEach(async () => {
|
||||
await t.cleanup();
|
||||
});
|
||||
|
||||
it("idle → benign 200 {skipped:false}; the runtime is never poked", async () => {
|
||||
const res = await api.post(`/api/sessions/${SID}/retry-now`, {});
|
||||
expect(res.status).toBe(200);
|
||||
expect((await res.json()) as RetryNowResponse).toEqual({ skipped: false });
|
||||
expect(skips).toEqual([]);
|
||||
});
|
||||
|
||||
it("running (parked in a wait) → 200 {skipped:true}, the skip reaches the runtime", async () => {
|
||||
await t.deps.manager.startTask(SID, [userText("go")]);
|
||||
await waitFor(() => t.deps.manager.pendingApprovalCount(SID) === 1);
|
||||
|
||||
const res = await api.post(`/api/sessions/${SID}/retry-now`, {});
|
||||
expect(res.status).toBe(200);
|
||||
expect((await res.json()) as RetryNowResponse).toEqual({ skipped: true });
|
||||
expect(skips).toHaveLength(1);
|
||||
|
||||
t.deps.manager.decideApproval(SID, "tc-retry", "allow");
|
||||
await waitFor(() => t.deps.manager.statusOf(SID) === "idle");
|
||||
});
|
||||
|
||||
it("foreign and unknown sessions → 404 (same auth semantics as the other session routes)", async () => {
|
||||
expect((await outsider.post(`/api/sessions/${SID}/retry-now`, {})).status).toBe(404);
|
||||
expect((await api.post(`/api/sessions/session-ghost/retry-now`, {})).status).toBe(404);
|
||||
expect(skips).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -58,6 +58,7 @@ function approvalFakeSession(sessionId: string, toolName = "exec_command"): Runt
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run(_input: OmniMessage[], opts: { approve: ApproveFn; signal: AbortSignal }) {
|
||||
const tc = toolCall({ name: toolName, arguments: "{}", toolCallId: "tc-1" });
|
||||
yield tc;
|
||||
@@ -165,6 +166,7 @@ describe("session-manager", () => {
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run(_input: OmniMessage[], opts: { thinkingLevel?: string }) {
|
||||
seen.push(opts.thinkingLevel);
|
||||
yield assistantText("ok");
|
||||
@@ -192,6 +194,7 @@ describe("session-manager", () => {
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run(): AsyncGenerator<OmniMessage> {
|
||||
yield requestBegin();
|
||||
yield toolCall({ name: "exec_command", arguments: "{}", toolCallId: "tc-1" });
|
||||
@@ -335,6 +338,7 @@ describe("session-manager", () => {
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
// eslint-disable-next-line require-yield
|
||||
async *run(input: OmniMessage[]): AsyncGenerator<OmniMessage> {
|
||||
runInputs.push(input.map((m) => (m.payload as { text?: string }).text ?? ""));
|
||||
@@ -417,6 +421,7 @@ describe("session-manager", () => {
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run() {
|
||||
// The parent-level run_subagent call (no origin): its prompt becomes the sub-session title.
|
||||
yield toolCall({
|
||||
@@ -493,6 +498,7 @@ describe("session-manager", () => {
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run(_input, opts) {
|
||||
const tc1 = toolCall({ name: "t1", arguments: "{}", toolCallId: "tc-1" });
|
||||
yield tc1;
|
||||
@@ -699,6 +705,41 @@ describe("session-manager", () => {
|
||||
expect(loads).toBe(2);
|
||||
});
|
||||
|
||||
it("invalidateProjectRuntimes: every cached entry in the Project is rebuilt; other Projects unaffected", async () => {
|
||||
// Two Agents with cached runtimes in p1, one in p2 — a models/credential update to p1
|
||||
// must reach BOTH of p1's Agents (collected from the active table) and leave p2 alone.
|
||||
const rowA2: SessionRow = { ...ROW, sessionId: "session-2", agentId: "a2" };
|
||||
const rowP2: SessionRow = { ...ROW, sessionId: "session-3", projectId: "p2" };
|
||||
sessions.insert(rowA2);
|
||||
sessions.insert(rowP2);
|
||||
let loads = 0;
|
||||
const loader: SessionLoader = {
|
||||
load: async (row) => {
|
||||
loads++;
|
||||
return approvalFakeSession(row.sessionId);
|
||||
},
|
||||
};
|
||||
const manager = makeManager(loader);
|
||||
for (const sid of ["session-1", "session-2", "session-3"]) {
|
||||
sessions.updateApprovalMode(sid, "allow-all");
|
||||
}
|
||||
manager.adopt(ROW, approvalFakeSession("session-1"));
|
||||
manager.adopt(rowA2, approvalFakeSession("session-2"));
|
||||
manager.adopt(rowP2, approvalFakeSession("session-3"));
|
||||
|
||||
manager.invalidateProjectRuntimes("p1");
|
||||
// Both p1 entries are stale and rebuild through the loader on their next Task.
|
||||
await manager.startTask("session-1", [userText("a")]);
|
||||
await waitFor(() => manager.statusOf("session-1") === "idle");
|
||||
await manager.startTask("session-2", [userText("b")]);
|
||||
await waitFor(() => manager.statusOf("session-2") === "idle");
|
||||
expect(loads).toBe(2);
|
||||
// The p2 entry keeps its cached runtime.
|
||||
await manager.startTask("session-3", [userText("c")]);
|
||||
await waitFor(() => manager.statusOf("session-3") === "idle");
|
||||
expect(loads).toBe(2);
|
||||
});
|
||||
|
||||
it("sweepIdle: entries that are running / have pending approvals are not evicted", async () => {
|
||||
const manager = makeManager(loaderOf(approvalFakeSession("session-1")));
|
||||
await manager.startTask("session-1", [userText("go")]);
|
||||
@@ -731,6 +772,7 @@ describe("session-manager", () => {
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run() {
|
||||
yield thinkingMessage("thinking");
|
||||
yield assistantText("answer A");
|
||||
@@ -785,6 +827,7 @@ describe("session-manager", () => {
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run() {
|
||||
yield assistantText("x".repeat(600));
|
||||
// Sub-session text doesn't count toward the main-session body threshold
|
||||
@@ -836,6 +879,7 @@ describe("session-manager", () => {
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run() {
|
||||
yield toolCall({
|
||||
name: "run_subagent",
|
||||
|
||||
@@ -64,6 +64,7 @@ function approvalFakeSession(sessionId: string): RuntimeSession {
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run(_input: OmniMessage[], opts: { approve: ApproveFn; signal: AbortSignal }) {
|
||||
const tc = toolCall({ name: "exec_command", arguments: "{}", toolCallId: "tc-sse" });
|
||||
yield tc;
|
||||
|
||||
@@ -27,6 +27,7 @@ function steeringFakeSession(sessionId: string, steered: string[]): RuntimeSessi
|
||||
steered.push(text);
|
||||
return true;
|
||||
},
|
||||
skipReconnectWait: () => false,
|
||||
async *run(_input: OmniMessage[], opts: { approve: ApproveFn; signal: AbortSignal }) {
|
||||
const tc = toolCall({ name: "exec_command", arguments: "{}", toolCallId: "tc-steer" });
|
||||
yield tc;
|
||||
|
||||
@@ -22,6 +22,7 @@ function fakeRuntimeSession(sessionId: string): RuntimeSession {
|
||||
generateTitle: async () => ({ title: null, usage: null }),
|
||||
compactability: () => "ok" as const,
|
||||
steer: () => false,
|
||||
skipReconnectWait: () => false,
|
||||
async *run() {},
|
||||
async *compact() {},
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user