Skip to content
3 changes: 3 additions & 0 deletions packages/server/src/routes/connections.routes.ts
Original file line number Diff line number Diff line change
Expand Up @@ -460,6 +460,7 @@ export async function connectionsRoutes(app: FastifyInstance) {
conn.claudeFastMode === "true",
conn.treatAsLocalEndpoint === "true",
conn.defaultParameters,
conn.id,
);
let responseText = "";
for await (const chunk of provider.chat([{ role: "user", content: "Reply with OK." }], {
Expand Down Expand Up @@ -502,6 +503,7 @@ export async function connectionsRoutes(app: FastifyInstance) {
conn.claudeFastMode === "true",
conn.treatAsLocalEndpoint === "true",
conn.defaultParameters,
conn.id,
);
let responseText = "";
for await (const chunk of provider.chat([{ role: "user", content: "Reply with OK." }], {
Expand Down Expand Up @@ -1369,6 +1371,7 @@ export async function connectionsRoutes(app: FastifyInstance) {
conn.claudeFastMode === "true",
conn.treatAsLocalEndpoint === "true",
conn.defaultParameters,
conn.id,
);

let fullResponse = "";
Expand Down
3 changes: 3 additions & 0 deletions packages/server/src/routes/encounter.routes.ts
Original file line number Diff line number Diff line change
Expand Up @@ -589,6 +589,7 @@ export async function encounterRoutes(app: FastifyInstance) {
conn.claudeFastMode === "true",
conn.treatAsLocalEndpoint === "true",
conn.defaultParameters,
conn.id,
);

const characterIds: string[] = JSON.parse(chat.characterIds as string);
Expand Down Expand Up @@ -699,6 +700,7 @@ export async function encounterRoutes(app: FastifyInstance) {
conn.claudeFastMode === "true",
conn.treatAsLocalEndpoint === "true",
conn.defaultParameters,
conn.id,
);

const characterIds: string[] = JSON.parse(chat.characterIds as string);
Expand Down Expand Up @@ -800,6 +802,7 @@ export async function encounterRoutes(app: FastifyInstance) {
conn.claudeFastMode === "true",
conn.treatAsLocalEndpoint === "true",
conn.defaultParameters,
conn.id,
);

const characterIds: string[] = JSON.parse(chat.characterIds as string);
Expand Down
1 change: 1 addition & 0 deletions packages/server/src/routes/generate/dry-run-route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1615,6 +1615,7 @@ export async function registerDryRunRoute(app: FastifyInstance) {
conn.claudeFastMode === "true",
conn.treatAsLocalEndpoint === "true",
conn.defaultParameters,
connId ?? undefined,
);

// ── Mirror /api/generate: normalize + fit prompt to context ──
Expand Down
1 change: 1 addition & 0 deletions packages/server/src/routes/lorebooks.routes.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1193,6 +1193,7 @@ export async function lorebooksRoutes(app: FastifyInstance) {
resolvedConn.claudeFastMode === "true",
resolvedConn.treatAsLocalEndpoint === "true",
resolvedConn.defaultParameters,
resolvedConn.id,
);
})();
const embeddingModel = useLocalSidecar ? LOCAL_SIDECAR_MODEL : body.model;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import type { DB } from "../../db/connection.js";
import { fitMessagesToContext, type BaseLLMProvider, type ChatMessage, type ChatOptions } from "../llm/base-provider.js";
import { getLocalSidecarProvider, LOCAL_SIDECAR_MODEL } from "../llm/local-sidecar.js";
import { createLLMProvider } from "../llm/provider-registry.js";
import { unwrapConnectionAdmissionProvider } from "../generation/connection-admission.js";
import { createConnectionsStorage } from "../storage/connections.storage.js";

export function createCapabilityLanguageModelHost(db: DB): CapabilityLanguageModelHost {
Expand All @@ -21,7 +22,7 @@ export function createCapabilityLanguageModelHost(db: DB): CapabilityLanguageMod
};
const resolvedModel = (provider: BaseLLMProvider, connectionId: string, model: string) =>
Object.freeze({
name: provider.constructor.name,
name: unwrapConnectionAdmissionProvider(provider).constructor.name,
connectionId,
model,
maxContext: provider.maxContextValue,
Expand Down Expand Up @@ -77,6 +78,8 @@ export function createCapabilityLanguageModelHost(db: DB): CapabilityLanguageMod
connection.maxTokensOverride,
connection.claudeFastMode === "true",
connection.treatAsLocalEndpoint === "true",
undefined,
connection.id,
),
connection.id,
requireModel(model ?? connection.model),
Expand Down
196 changes: 196 additions & 0 deletions packages/server/src/services/generation/connection-admission.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,196 @@
import type { ChatCompletionResult, ChatMessage, ChatOptions, LLMUsage } from "../llm/base-provider.js";
import { BaseLLMProvider } from "../llm/base-provider.js";
import { logger } from "../../lib/logger.js";

// Admission keys identify one physical provider endpoint. Text work keys on the configured
// connection id; image work keys on the resolved base URL plus the endpoint id where the
// backend needs one, because most image callers have no connection row in scope and
// unregistered foreground work would defeat the priority rule. The keyspaces do not overlap.
type ConnectionState = {
foregroundActive: number;
backgroundActive: boolean;
lastForegroundFinishedAt: number;
};

const states = new Map<string, ConnectionState>();
export const BACKGROUND_CONNECTION_IDLE_MS = 30_000;
Comment on lines +9 to +16

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🔵 Trivial | 💤 Low value

states grows without bound and is never pruned.

states keeps one entry per connection id and per image admission key forever. Entries stay after the connection is deleted or the endpoint changes. The entry is small, so this is a slow leak rather than an outage, but nothing removes idle entries.

Consider deleting the state when foregroundActive === 0, backgroundActive === false, and the idle window has elapsed. Do this inside the release paths.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/server/src/services/generation/connection-admission.ts` around lines
9 - 16, Prune idle entries from the states map within the foreground and
background release paths. When a state has foregroundActive === 0,
backgroundActive === false, and lastForegroundFinishedAt is older than
BACKGROUND_CONNECTION_IDLE_MS, delete its corresponding key; preserve state
entries while either activity remains active or the idle window has not elapsed.


export type ConnectionAttemptOutcome = "completed" | "failed";
export type ConnectionAttemptFinalizer = (outcome: ConnectionAttemptOutcome) => void | Promise<void>;
export type ConnectionAdmissionMode =
| { kind: "foreground" }
| {
kind: "background";
beforeAttempt?: () => void | ConnectionAttemptFinalizer | Promise<void | ConnectionAttemptFinalizer>;
};

export class BackgroundConnectionBusyError extends Error {
constructor(readonly connectionId: string) {
super(`Connection ${connectionId} is not available for background generation.`);
this.name = "BackgroundConnectionBusyError";
}
}

export class ConnectionAttemptRejectedError extends Error {
constructor(readonly cause: unknown) {
super("Connection attempt was rejected before provider work started.", { cause });
this.name = "ConnectionAttemptRejectedError";
}
}

export class ConnectionAttemptFinalizationError extends Error {
constructor(readonly cause: unknown) {
super("Connection attempt accounting failed after provider work finished.", { cause });
this.name = "ConnectionAttemptFinalizationError";
}
}

export function isConnectionAdmissionFailure(error: unknown): boolean {
return (
error instanceof BackgroundConnectionBusyError ||
error instanceof ConnectionAttemptRejectedError ||
error instanceof ConnectionAttemptFinalizationError
);
}

function stateFor(connectionId: string): ConnectionState {
const existing = states.get(connectionId);
if (existing) return existing;
const state = { foregroundActive: 0, backgroundActive: false, lastForegroundFinishedAt: 0 };
states.set(connectionId, state);
return state;
}

export function beginForegroundConnection(connectionId: string): () => void {
const state = stateFor(connectionId);
state.foregroundActive += 1;
let released = false;
return () => {
if (released) return;
released = true;
state.foregroundActive -= 1;
state.lastForegroundFinishedAt = Date.now();
};
}

export function tryBackgroundConnection(
connectionId: string,
at: Date,
): { acquired: false } | { acquired: true; release: () => void } {
const state = stateFor(connectionId);
if (
state.backgroundActive ||
state.foregroundActive > 0 ||
at.getTime() - state.lastForegroundFinishedAt < BACKGROUND_CONNECTION_IDLE_MS
) {
return { acquired: false };
}
state.backgroundActive = true;
return {
acquired: true,
release: () => {
state.backgroundActive = false;
},
};
}

export function resetConnectionAdmissionForTests(): void {
states.clear();
}

async function beginConnectionAttempt(
connectionId: string,
mode: ConnectionAdmissionMode,
): Promise<{ release: () => void; finalize?: ConnectionAttemptFinalizer }> {
if (mode.kind === "foreground") return { release: beginForegroundConnection(connectionId) };

const admission = tryBackgroundConnection(connectionId, new Date());
if (!admission.acquired) throw new BackgroundConnectionBusyError(connectionId);
try {
return { release: admission.release, finalize: (await mode.beforeAttempt?.()) || undefined };
} catch (error) {
admission.release();
throw new ConnectionAttemptRejectedError(error);
}
}

/**
* Record the attempt outcome without letting accounting failures overwrite a provider
* error: the operation's own failure is what the caller (and the fallback chain) needs to
* see. A finalization failure only surfaces when the operation itself succeeded.
*/
async function finalizeConnectionAttempt(
attempt: { release: () => void; finalize?: ConnectionAttemptFinalizer },
outcome: ConnectionAttemptOutcome,
): Promise<void> {
try {
await attempt.finalize?.(outcome);
} catch (error) {
if (outcome === "completed") throw new ConnectionAttemptFinalizationError(error);
logger.error(error, "[connection-admission] Attempt accounting failed after a failed provider call");
} finally {
attempt.release();
}
}
Comment on lines +122 to +134

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Description: Inspect background finalizers passed as beforeAttempt to confirm what a failed finalize means for streamed output.
set -euo pipefail

rg -n -C 12 'beforeAttempt' --glob '*.ts' packages/server/src scripts

Repository: Pasta-Devs/Marinara-Engine

Length of output: 9339


🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf '--- connection-admission relevant section ---\n'
sed -n '70,190p' packages/server/src/services/generation/connection-admission.ts | cat -n

printf '\n--- ConnectionFallbackProvider relevant references ---\n'
rg -n -C 8 'finalizeConnectionAttempt|ChatCompletion|chat\(|chatComplete|ConnectionAttemptFinalizationError|chatComplete' packages/server/src/services/llm/connection-fallback-provider.ts packages/server/src/services/generation/connection-admission.ts

printf '\n--- all finalizeConnectionAttempt usages ---\n'
rg -n -C 8 'finalizeConnectionAttempt' packages/server/src

Repository: Pasta-Devs/Marinara-Engine

Length of output: 255


🏁 Script executed:

#!/bin/bash
set -euo pipefail

printf '%s\n' '--- connection-admission relevant section ---'
sed -n '70,190p' packages/server/src/services/generation/connection-admission.ts | cat -n

printf '%s\n' ''
printf '%s\n' '--- ConnectionFallbackProvider relevant references ---'
rg -n -C 8 'finalizeConnectionAttempt|ChatCompletion|chat\(|chatComplete|ConnectionAttemptFinalizationError|chatComplete' packages/server/src/services/llm/connection-fallback-provider.ts packages/server/src/services/generation/connection-admission.ts

printf '%s\n$' ''
printf '%s\n' '--- all finalizeConnectionAttempt usages ---'
rg -n -C 8 'finalizeConnectionAttempt' packages/server/src

Repository: Pasta-Devs/Marinara-Engine

Length of output: 29615


Avoid promoting late finalizer failures to stream failure after output is emitted.

ConnectionAdmissionProvider.chat marks the streaming call completed after yield* this.provider.chat(...) returns, then finalizeConnectionAttempt can later throw ConnectionAttemptFinalizationError. ConnectionFallbackProvider.chat includes that admission failure in its catch and rethrows it when emittedUsableOutput is true, so the consumer reports a failed generation for a token stream that already delivered content. Keep the throw for non-streaming paths only, or log and suppress it when tokens were already delivered.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/server/src/services/generation/connection-admission.ts` around lines
122 - 134, The completed-outcome path in finalizeConnectionAttempt must not
promote late finalizer errors into a stream failure after usable tokens have
been emitted. Adjust the coordination between ConnectionAdmissionProvider.chat,
finalizeConnectionAttempt, and ConnectionFallbackProvider.chat so completed
finalization errors are retained for non-streaming calls but logged and
suppressed for already-emitting streams, preserving normal propagation before
output is delivered.


export async function withConnectionAdmission<T>(
connectionId: string,
mode: ConnectionAdmissionMode,
operation: () => Promise<T>,
): Promise<T> {
const attempt = await beginConnectionAttempt(connectionId, mode);
let outcome: ConnectionAttemptOutcome = "failed";
try {
const result = await operation();
outcome = "completed";
return result;
} finally {
await finalizeConnectionAttempt(attempt, outcome);
}
}

export class ConnectionAdmissionProvider extends BaseLLMProvider {
constructor(
readonly provider: BaseLLMProvider,
private readonly connectionId: string,
private readonly mode: ConnectionAdmissionMode = { kind: "foreground" },
) {
super("", "", provider.maxContextValue ?? undefined, null, provider.maxTokensOverrideValue);
}

async *chat(messages: ChatMessage[], options: ChatOptions): AsyncGenerator<string, LLMUsage | void, unknown> {
const attempt = await beginConnectionAttempt(this.connectionId, this.mode);
let outcome: ConnectionAttemptOutcome = "failed";
try {
const result = yield* this.provider.chat(messages, options);
outcome = "completed";
return result;
} finally {
await finalizeConnectionAttempt(attempt, outcome);
}
}
Comment on lines +161 to +171

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Description: Find manual iteration of provider chat generators that may skip generator close.
set -euo pipefail

echo "--- manual .next() iteration over chat generators ---"
rg -n -C 10 --glob '*.ts' '\.chat\s*\(' packages/server/src | rg -n -C 6 '\.next\s*\(' || true

echo "--- explicit generator close calls ---"
rg -n -C 4 --glob '*.ts' '\.return\s*\(\s*undefined\s*\)' packages/server/src || true

echo "--- for await consumers of chat ---"
rg -n -C 4 --glob '*.ts' 'for await' packages/server/src | rg -n 'chat' || true

Repository: Pasta-Devs/Marinara-Engine

Length of output: 5412


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "--- locate relevant files ---"
fd -a 'connection-admission|connection-fallback-provider|base-provider|game\.routes|connections\.routes|providers' packages/server/src | sed 's#^\./##'

echo "--- connection-admission outline and relevant section ---"
ast-grep outline packages/server/src/services/generation/connection-admission.ts --view compact || true
sed -n '1,230p' packages/server/src/services/generation/connection-admission.ts

echo "--- base-provider chat generation outline/search around generateContentWithStream ---"
ast-grep outline packages/server/src/services/llm/base-provider.ts --view compact || true
sed -n '620,760p' packages/server/src/services/llm/base-provider.ts

Repository: Pasta-Devs/Marinara-Engine

Length of output: 13692


Close provider.chat(...) generators when calling chatComplete.

BaseLLMProvider.chatComplete() calls gen.next() in a loop but does not call return() on the AsyncGenerator, so the ConnectionAdmissionProvider.chat() finally block never runs. Manual callers must close an await gen.next() loop with await gen.return(undefined) in finally; for await closes the generator automatically.

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/server/src/services/generation/connection-admission.ts` around lines
161 - 171, Update BaseLLMProvider.chatComplete() to close the AsyncGenerator
returned by provider.chat() by calling await gen.return(undefined) in a finally
block around its manual gen.next() loop, ensuring
ConnectionAdmissionProvider.chat() executes finalizeConnectionAttempt. Preserve
the existing iteration and result handling.


async chatComplete(messages: ChatMessage[], options: ChatOptions): Promise<ChatCompletionResult> {
return withConnectionAdmission(this.connectionId, this.mode, () => this.provider.chatComplete(messages, options));
}

async embed(texts: string[], model: string, signal?: AbortSignal): Promise<number[][]> {
return withConnectionAdmission(this.connectionId, this.mode, () => this.provider.embed(texts, model, signal));
}
}

/**
* Look through the admission decorator to the provider that actually talks to the backend.
* Diagnostics that report a provider class need the concrete one, not the wrapper.
*/
export function unwrapConnectionAdmissionProvider(provider: BaseLLMProvider): BaseLLMProvider {
return provider instanceof ConnectionAdmissionProvider ? provider.provider : provider;
}

export function withConnectionAdmissionProvider(
provider: BaseLLMProvider,
connectionId: string,
mode: ConnectionAdmissionMode = { kind: "foreground" },
): BaseLLMProvider {
return new ConnectionAdmissionProvider(provider, connectionId, mode);
}
39 changes: 37 additions & 2 deletions packages/server/src/services/image/image-generation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,11 @@ import { generateRunPodComfyUI } from "./runpod-comfyui.service.js";
import { logger, logDebugOverride } from "../../lib/logger.js";
import { assertInsideDir, normalizeLoopbackUrl, safeFetch, validateOutboundUrl } from "../../utils/security.js";
import { notifyGenerationFallback, type GenerationFallbackNotifier } from "../generation/fallback-notification.js";
import {
isConnectionAdmissionFailure,
withConnectionAdmission,
type ConnectionAdmissionMode,
} from "../generation/connection-admission.js";
import {
COMFYUI_MAX_REFERENCE_IMAGES,
findMissingComfyReferenceSlots,
Expand Down Expand Up @@ -115,6 +120,8 @@ export interface ImageGenRequest {
signal?: AbortSignal;
/** Emit the final provider request even when the global log level is above debug. */
debugMode?: boolean;
/** Defaults to foreground: the caller is servicing a user-visible request. */
admissionMode?: ConnectionAdmissionMode;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
/** Called immediately before a configured fallback connection is attempted. */
onFallback?: GenerationFallbackNotifier;
/** Optional one-shot backup connection used only when the primary image request fails. */
Expand Down Expand Up @@ -194,6 +201,23 @@ function resolveImageBackend(source: string, baseUrl: string, serviceHint: strin
const IMAGE_GEN_TIMEOUT = Number(process.env.IMAGE_GEN_TIMEOUT_MS ?? 1_800_000);
const COMFYUI_GEN_TIMEOUT_SECONDS = Number(process.env.COMFYUI_GEN_TIMEOUT ?? 2400);

/**
* Identify the physical image endpoint an admission slot belongs to. RunPod connections share
* one API base URL and select the actual endpoint with a separate id, so the base URL alone
* would make two independent endpoints contend for a single slot. Every other backend's base
* URL already is the physical target, and a stale `imageEndpointId` left on an imported or
* copied connection must not split one ComfyUI/A1111 endpoint into separate slots.
*/
export function imageAdmissionKey(
normalizedBaseUrl: string,
resolvedSource: string,
imageEndpointId?: string,
): string {
if (resolvedSource !== "runpod_comfyui") return normalizedBaseUrl;
const endpointId = imageEndpointId?.trim();
return endpointId ? `${normalizedBaseUrl}#${endpointId}` : normalizedBaseUrl;
}

/**
* Generate an image using the configured image generation connection.
* Returns the base64 data and metadata needed to save it.
Expand All @@ -213,7 +237,7 @@ export async function generateImage(
: IMAGE_GEN_TIMEOUT;

try {
return await withImageGenerationDeadline(request, generationTimeoutMs, async (signal) => {
const physicalRequest = () => withImageGenerationDeadline(request, generationTimeoutMs, async (signal) => {
const allowLocalUrls =
request.allowLocalUrls ?? (await shouldAllowLocalUrlsForImageConnection(normalizedBaseUrl, resolvedSource));
const scopedRequest = {
Expand Down Expand Up @@ -269,9 +293,20 @@ export async function generateImage(
return generateOpenAI(normalizedBaseUrl, apiKey, scopedRequest);
}
});
// Admit on the resolved endpoint rather than a connection id: every caller reaches this
// function, but only some have a connection row in scope, and foreground work that
// registers nothing would let background preparation start on top of it.
// ponytail: image work keys on the endpoint URL while text work keys on the connection
// id, so the two do not hold each other off on a connection used for both. Unify the
// key if that overlap ever shows up in practice.
return await withConnectionAdmission(
imageAdmissionKey(normalizedBaseUrl, resolvedSource, request.imageEndpointId),
request.admissionMode ?? { kind: "foreground" },
Comment thread
coderabbitai[bot] marked this conversation as resolved.
physicalRequest,
);
} catch (error) {
const fallback = request.fallback;
if (!fallback || request.signal?.aborted) throw error;
if (!fallback || request.signal?.aborted || isConnectionAdmissionFailure(error)) throw error;
logger.warn(
error,
"[illustrator-fallback] Primary image generation failed; retrying with connection %s (%s)",
Expand Down
Loading
Loading