Compare commits

..
Author SHA1 Message Date
suyx 7de633d4c9 fix(POSTV1-05): 修复生成请求与 Worker 重入
Dada P0-A isolated Windows CI / validate-and-package (push) Waiting to run
2026-08-05 16:04:11 +08:00
suyx b995662388 fix(POSTV1-04): 修复生成服务运行时装配
Dada P0-A isolated Windows CI / validate-and-package (push) Waiting to run
2026-08-05 15:36:47 +08:00
8 changed files with 237 additions and 5 deletions
+16 -1
View File
@@ -16,7 +16,12 @@ import { MockResendAdapter } from "./resend-adapter.js";
import { readSecureConfigCandidate } from "./secure-config.js"; import { readSecureConfigCandidate } from "./secure-config.js";
import { StructuredJsonlLogger } from "./structured-log.js"; import { StructuredJsonlLogger } from "./structured-log.js";
import { attachApiSupervisorControl, initializeApiCredentialClients, receiveApiCredentials } from "./supervisor-channel.js"; import { attachApiSupervisorControl, initializeApiCredentialClients, receiveApiCredentials } from "./supervisor-channel.js";
import { ModelConfigurationService, portableRuntimeModelCandidates } from "./model-configuration.js"; import { GenerationSubmissionService } from "./generation-submission.js";
import {
GenerationModelConfigurationCatalog,
ModelConfigurationService,
portableRuntimeModelCandidates,
} from "./model-configuration.js";
import { MockAmapAdapter, type AmapAdapter } from "./amap-adapter.js"; import { MockAmapAdapter, type AmapAdapter } from "./amap-adapter.js";
import { StickerReleaseService } from "./sticker-releases.js"; import { StickerReleaseService } from "./sticker-releases.js";
import { createAdminDiagnosticsProvider, createAdminServicesStorageProvider } from "./admin-state.js"; import { createAdminDiagnosticsProvider, createAdminServicesStorageProvider } from "./admin-state.js";
@@ -28,6 +33,7 @@ let credits: CreditService | undefined;
let storage: ManagedStorage | undefined; let storage: ManagedStorage | undefined;
let latestExports: LatestExportService | undefined; let latestExports: LatestExportService | undefined;
let models: ModelConfigurationService | undefined; let models: ModelConfigurationService | undefined;
let generations: GenerationSubmissionService | undefined;
let recentAssets: RecentAssetService | undefined; let recentAssets: RecentAssetService | undefined;
let stickers: StickerReleaseService | undefined; let stickers: StickerReleaseService | undefined;
let amap: AmapAdapter = new MockAmapAdapter(); let amap: AmapAdapter = new MockAmapAdapter();
@@ -59,6 +65,11 @@ if (credentialChannelEnabled) {
stickers = new StickerReleaseService({ databasePath, storage }); stickers = new StickerReleaseService({ databasePath, storage });
latestExports = new LatestExportService({ databasePath, storage }); latestExports = new LatestExportService({ databasePath, storage });
models = new ModelConfigurationService({ database: registration.database, seedCandidates: portableRuntimeModelCandidates }); models = new ModelConfigurationService({ database: registration.database, seedCandidates: portableRuntimeModelCandidates });
generations = new GenerationSubmissionService({
credits,
models: new GenerationModelConfigurationCatalog(models),
storage,
});
recentAssets = new RecentAssetService({ database: registration.database }); recentAssets = new RecentAssetService({ database: registration.database });
registration.applySecureConfig(readSecureConfigCandidate(instanceConfigPath)); registration.applySecureConfig(readSecureConfigCandidate(instanceConfigPath));
} catch (error) { } catch (error) {
@@ -68,6 +79,8 @@ if (credentialChannelEnabled) {
stickers = undefined; stickers = undefined;
latestExports?.close(); latestExports?.close();
latestExports = undefined; latestExports = undefined;
generations?.close();
generations = undefined;
storage?.close(); storage?.close();
storage = undefined; storage = undefined;
credits?.close(); credits?.close();
@@ -102,6 +115,7 @@ const app = await createApp({
amap, amap,
...(browserSupportRelease ? { browserSupportRelease } : {}), ...(browserSupportRelease ? { browserSupportRelease } : {}),
...(credits ? { credits } : {}), ...(credits ? { credits } : {}),
...(generations ? { generations } : {}),
...(latestExports ? { latestExports } : {}), ...(latestExports ? { latestExports } : {}),
...(registration && localTestAuth ? { localTestAuth: true } : {}), ...(registration && localTestAuth ? { localTestAuth: true } : {}),
...(models ? { models } : {}), ...(models ? { models } : {}),
@@ -125,6 +139,7 @@ if (controlPipeIndex >= 0) {
await app.close(); await app.close();
amap.dispose?.(); amap.dispose?.();
latestExports?.close(); latestExports?.close();
generations?.close();
credits?.close(); credits?.close();
projects?.close(); projects?.close();
registration?.close(); registration?.close();
+41
View File
@@ -2,6 +2,8 @@ import { randomUUID, createHash } from "node:crypto";
import type BetterSqlite3 from "better-sqlite3"; import type BetterSqlite3 from "better-sqlite3";
import { serializeAuditSummary, auditRetentionMilliseconds } from "./audit-policy.js"; import { serializeAuditSummary, auditRetentionMilliseconds } from "./audit-policy.js";
import type { GenerationModelCatalog, GenerationModelSnapshot } from "./generation-submission.js";
import { projectRatios } from "./projects.js";
export const modelIds = [ export const modelIds = [
"gemini-3.1-flash-image-preview", "gemini-3.1-flash-image-preview",
@@ -59,6 +61,45 @@ export interface ModelConfigurationView {
models: ModelConfigView[]; models: ModelConfigView[];
} }
type ReadableModelConfiguration = Pick<ModelConfigurationService, "read">;
function generationRuntimeReason(reason: ModelRuntimeReason): GenerationModelSnapshot["runtimeAvailability"]["reason"] {
if (reason === "gateway_balance_insufficient") return reason;
if (reason === "contract_unverified" || reason === "contract_blocked") return "gateway_contract_invalid";
if (reason === "available") return null;
return "model_disabled";
}
export class GenerationModelConfigurationCatalog implements GenerationModelCatalog {
constructor(private readonly models: ReadableModelConfiguration) {}
readModel(modelId: string): GenerationModelSnapshot | undefined {
const configuration = this.models.read();
const model = configuration.models.find((entry) => entry.model_id === modelId);
if (!model) return undefined;
const supportedRatios = projectRatios.filter((ratio) => model.supported_ratios.includes(ratio));
return {
configSetVersion: configuration.config_set_version,
configVersion: model.config_version,
contractValidationStatus: model.contract_validation_status === "verified" ? "verified" : "unverified",
creditCost: model.credit_cost,
enabled: model.enabled,
modelId: model.model_id,
promptMaxLength: model.prompt_max_length,
referenceLimits: {
maxFileBytes: model.reference_limits.max_file_bytes,
maxFiles: model.reference_limits.max_files,
maxTotalBytes: model.reference_limits.max_total_bytes,
},
runtimeAvailability: {
availableForNewJobs: model.runtime_availability.available_for_new_jobs,
reason: generationRuntimeReason(model.runtime_availability.reason),
},
supportedRatios,
};
}
}
export class ModelConfigurationError extends Error { export class ModelConfigurationError extends Error {
constructor( constructor(
readonly code: readonly code:
@@ -0,0 +1,35 @@
export interface GenerationPollingProcessor {
processNext(): Promise<unknown>;
}
export class GenerationPollingLoop {
private closed = false;
private inFlight = false;
private readonly timer: ReturnType<typeof setInterval>;
constructor(
private readonly processor: GenerationPollingProcessor,
intervalMilliseconds = 250,
) {
if (!Number.isSafeInteger(intervalMilliseconds) || intervalMilliseconds <= 0) {
throw new Error("generation_polling_interval_invalid");
}
this.timer = setInterval(() => this.run(), intervalMilliseconds);
}
close() {
if (this.closed) return;
this.closed = true;
clearInterval(this.timer);
}
private run() {
if (this.closed || this.inFlight) return;
this.inFlight = true;
void this.processor.processNext()
.catch(() => undefined)
.finally(() => {
this.inFlight = false;
});
}
}
+5 -1
View File
@@ -16,6 +16,7 @@ const gptImageEndpoint = "https://oneapi.intelligrow.cn/v1/images/generations";
const gptImageReferenceEndpoint = "https://oneapi.intelligrow.cn/v1/images/edits"; const gptImageReferenceEndpoint = "https://oneapi.intelligrow.cn/v1/images/edits";
const maximumResponseBytes = 32 * 1024 * 1024; const maximumResponseBytes = 32 * 1024 * 1024;
const requestTimeoutMilliseconds = 180_000; const requestTimeoutMilliseconds = 180_000;
const geminiImageSystemInstruction = "Generate exactly one image from the user's description. Return the generated image and do not answer with text only.";
type FetchLike = typeof fetch; type FetchLike = typeof fetch;
@@ -141,7 +142,10 @@ function buildRequest(request: GenerationAdapterRequest) {
return { return {
body: JSON.stringify({ body: JSON.stringify({
extra_body: { google: { image_config: { aspect_ratio: request.ratio, image_size: "1K" } } }, extra_body: { google: { image_config: { aspect_ratio: request.ratio, image_size: "1K" } } },
messages: [{ content, role: "user" }], messages: [
{ content: geminiImageSystemInstruction, role: "system" },
{ content, role: "user" },
],
model: geminiProviderModelId, model: geminiProviderModelId,
stream: false, stream: false,
}), }),
+4 -3
View File
@@ -3,6 +3,7 @@ import { join } from "node:path";
import { WorkerAiCallGate } from "./ai-call-gate.js"; import { WorkerAiCallGate } from "./ai-call-gate.js";
import { runAiRuntimeProbe } from "./ai-runtime-probe.js"; import { runAiRuntimeProbe } from "./ai-runtime-probe.js";
import { GenerationPollingLoop } from "./generation-polling-loop.js";
import { GenerationProcessor } from "./generation-processor.js"; import { GenerationProcessor } from "./generation-processor.js";
import { OneApiGenerationAdapter } from "./oneapi-generation-adapter.js"; import { OneApiGenerationAdapter } from "./oneapi-generation-adapter.js";
import { readConfiguredLocalDataRoot } from "./runtime-config.js"; import { readConfiguredLocalDataRoot } from "./runtime-config.js";
@@ -52,13 +53,13 @@ if (!workerPort && process.argv.includes("--dada-ai-probe")) {
let projectCleanup: ProjectPurgeCleanup | undefined; let projectCleanup: ProjectPurgeCleanup | undefined;
let retentionTimer: ReturnType<typeof setInterval> | undefined; let retentionTimer: ReturnType<typeof setInterval> | undefined;
let processor: GenerationProcessor | undefined; let processor: GenerationProcessor | undefined;
let generationTimer: ReturnType<typeof setInterval> | undefined; let generationLoop: GenerationPollingLoop | undefined;
const control = attachWorkerSupervisorControl(controlPipe, () => { const control = attachWorkerSupervisorControl(controlPipe, () => {
clearInterval(keepAlive); clearInterval(keepAlive);
if (retentionTimer) clearInterval(retentionTimer); if (retentionTimer) clearInterval(retentionTimer);
retention?.close(); retention?.close();
projectCleanup?.close(); projectCleanup?.close();
if (generationTimer) clearInterval(generationTimer); generationLoop?.close();
processor?.close(); processor?.close();
storage?.close(); storage?.close();
}); });
@@ -102,7 +103,7 @@ if (!workerPort && process.argv.includes("--dada-ai-probe")) {
}); });
logger.write({ error_category: "none", status_category: "ready" }); logger.write({ error_category: "none", status_category: "ready" });
new WorkerAiCallGate({ getStorageStatus: () => storageStatus === "unavailable" ? storageStatus : (storage?.getStatus() ?? "unavailable"), logger }); new WorkerAiCallGate({ getStorageStatus: () => storageStatus === "unavailable" ? storageStatus : (storage?.getStatus() ?? "unavailable"), logger });
generationTimer = setInterval(() => { void processor?.processNext().catch(() => undefined); }, 250); generationLoop = new GenerationPollingLoop(processor);
} catch { } catch {
storageStatus = "unavailable"; storageStatus = "unavailable";
control.reportStatus("storage_unavailable"); control.reportStatus("storage_unavailable");
@@ -0,0 +1,100 @@
import { readFileSync } from "node:fs";
import { describe, expect, it } from "vitest";
import { GenerationModelConfigurationCatalog, type ModelConfigurationView } from "../../apps/api/src/model-configuration.js";
describe("POSTV1-04 generation runtime wiring", () => {
it("constructs, injects and closes the production generation submission service", () => {
const main = readFileSync("apps/api/src/main.ts", "utf8");
expect(main).toContain('import { GenerationSubmissionService } from "./generation-submission.js";');
expect(main).toContain("let generations: GenerationSubmissionService | undefined;");
expect(main).toContain("generations = new GenerationSubmissionService({");
expect(main).toContain("models: new GenerationModelConfigurationCatalog(models),");
expect(main).toContain("...(generations ? { generations } : {}),");
expect(main).toContain("generations?.close();");
});
it("maps the current model configuration into the generation submission contract", () => {
const configuration: ModelConfigurationView = {
config_set_version: 7,
configured_default_model_id: "gemini-3.1-flash-image-preview",
recommended_model_id: "gemini-3.1-flash-image-preview",
models: [{
config_version: 3,
contract_evidence_ref: "fixture-contract",
contract_validation_status: "verified",
credit_cost: 2,
display_name: "Fixture model",
enabled: true,
error_mapping_profile: {},
gateway_account_ref: "fixture-gateway",
is_default: true,
model_id: "gemini-3.1-flash-image-preview",
prompt_max_length: 1_000,
recommendation_priority: 1,
reference_limits: { max_file_bytes: 10, max_files: 2, max_total_bytes: 20 },
route_profile: {},
runtime_availability: { available_for_new_jobs: true, checked_at: "2026-08-05T00:00:00.000Z", reason: "available" },
safety_source: "provider",
supported_ratios: ["3:4", "invalid"],
}],
};
const catalog = new GenerationModelConfigurationCatalog({ read: () => configuration });
expect(catalog.readModel("gemini-3.1-flash-image-preview")).toEqual({
configSetVersion: 7,
configVersion: 3,
contractValidationStatus: "verified",
creditCost: 2,
enabled: true,
modelId: "gemini-3.1-flash-image-preview",
promptMaxLength: 1_000,
referenceLimits: { maxFileBytes: 10, maxFiles: 2, maxTotalBytes: 20 },
runtimeAvailability: { availableForNewJobs: true, reason: null },
supportedRatios: ["3:4"],
});
expect(catalog.readModel("missing")).toBeUndefined();
});
it("maps model and contract blocks into generation error categories", () => {
const configuration = {
config_set_version: 1,
configured_default_model_id: "gemini-3.1-flash-image-preview",
recommended_model_id: null,
models: [],
} satisfies ModelConfigurationView;
const catalog = new GenerationModelConfigurationCatalog({ read: () => configuration });
const base = {
config_version: 1,
contract_evidence_ref: null,
contract_validation_status: "verified" as const,
credit_cost: 1,
display_name: "Fixture model",
enabled: true,
error_mapping_profile: {},
gateway_account_ref: "fixture-gateway",
is_default: true,
model_id: "gemini-3.1-flash-image-preview" as const,
prompt_max_length: 1_000,
recommendation_priority: 1,
reference_limits: { max_file_bytes: 10, max_files: 2, max_total_bytes: 20 },
route_profile: {},
safety_source: "provider",
supported_ratios: ["3:4"],
};
configuration.models = [{
...base,
runtime_availability: { available_for_new_jobs: false, checked_at: "2026-08-05T00:00:00.000Z", reason: "contract_blocked" },
}];
expect(catalog.readModel(base.model_id)?.runtimeAvailability.reason).toBe("gateway_contract_invalid");
configuration.models = [{
...base,
runtime_availability: { available_for_new_jobs: false, checked_at: "2026-08-05T00:00:00.000Z", reason: "worker_degraded" },
}];
expect(catalog.readModel(base.model_id)?.runtimeAvailability.reason).toBe("model_disabled");
});
});
@@ -0,0 +1,28 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import { GenerationPollingLoop } from "../../apps/worker/src/generation-polling-loop.js";
describe("POSTV1-05 generation polling loop", () => {
afterEach(() => {
vi.useRealTimers();
});
it("does not start another processor call while the current call is unresolved", async () => {
vi.useFakeTimers();
let finishCurrentCall: (() => void) | undefined;
const processNext = vi.fn(() => new Promise<void>((resolve) => {
finishCurrentCall = resolve;
}));
const loop = new GenerationPollingLoop({ processNext }, 250);
await vi.advanceTimersByTimeAsync(1_000);
expect(processNext).toHaveBeenCalledTimes(1);
finishCurrentCall?.();
await Promise.resolve();
await vi.advanceTimersByTimeAsync(250);
expect(processNext).toHaveBeenCalledTimes(2);
loop.close();
});
});
@@ -26,6 +26,14 @@ describe("POSTV1-02 OneAPI runtime adapter", () => {
const headers = new Headers(init?.headers); const headers = new Headers(init?.headers);
expect(headers.get("authorization")).toBe("Bearer synthetic-runtime-token"); expect(headers.get("authorization")).toBe("Bearer synthetic-runtime-token");
expect(init?.redirect).toBe("error"); expect(init?.redirect).toBe("error");
const payload = JSON.parse(String(init?.body)) as { messages: Array<{ content: unknown; role: string }> };
expect(payload.messages).toEqual([
{
content: "Generate exactly one image from the user's description. Return the generated image and do not answer with text only.",
role: "system",
},
{ content: "一张用于本机验收的抽象色彩图", role: "user" },
]);
return new Response(JSON.stringify({ return new Response(JSON.stringify({
choices: [{ message: { content: `![result](data:image/png;base64,${source.toString("base64")})` } }], choices: [{ message: { content: `![result](data:image/png;base64,${source.toString("base64")})` } }],
}), { headers: { "content-type": "application/json" }, status: 200 }); }), { headers: { "content-type": "application/json" }, status: 200 });