diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 627e5ba..c3f1ec1 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -270,12 +270,14 @@ function generationTaskResponse(task: GenerationTaskView) { return { confirmed_credit_cost: task.confirmedCreditCost, created_at: task.createdAt, + error_category: task.errorCategory, generation_id: task.generationId, model_config_version: task.modelConfigVersion, model_id: task.modelId, project_id: task.projectId, prompt: task.prompt, ratio: task.ratio, + reference_asset_ids: task.referenceAssetIds, reference_count: task.referenceCount, reserved_credits: task.reservedCredits, status: task.status, diff --git a/apps/api/src/generation-submission.ts b/apps/api/src/generation-submission.ts index f965d7f..f17fd4f 100644 --- a/apps/api/src/generation-submission.ts +++ b/apps/api/src/generation-submission.ts @@ -88,12 +88,16 @@ export type GenerationSubmissionFields = Omit).map((entry) => entry.managed_file_id); return { confirmedCreditCost: row.confirmed_credit_cost, createdAt: iso(row.created_at), + errorCategory: row.error_category, generationId: row.generation_id, modelConfigVersion: row.model_config_version, modelId: row.model_id, projectId: row.project_id, prompt: row.prompt, ratio: row.ratio, - referenceCount, + referenceAssetIds, + referenceCount: referenceAssetIds.length, reservedCredits: row.reserved_credits, status: row.status, updatedAt: iso(row.updated_at), diff --git a/apps/web/src/generated/api/types.gen.ts b/apps/web/src/generated/api/types.gen.ts index a6bd78e..ed91086 100644 --- a/apps/web/src/generated/api/types.gen.ts +++ b/apps/web/src/generated/api/types.gen.ts @@ -363,12 +363,14 @@ export type GenerationProjectItem = { export type GenerationTaskResponse = { "confirmed_credit_cost": number; "created_at": string; + "error_category": GenerationErrorCategory | null; "generation_id": string; "model_config_version": number; "model_id": string; "project_id": ProjectId; "prompt": string; "ratio": ProjectRatio; + "reference_asset_ids": Array; "reference_count": number; "reserved_credits": number; "status": GenerationTaskStatus; diff --git a/apps/web/src/project-pages.css b/apps/web/src/project-pages.css index bfba561..bf266f7 100644 --- a/apps/web/src/project-pages.css +++ b/apps/web/src/project-pages.css @@ -227,6 +227,17 @@ padding: 12px 14px; } +.model-status select { + min-width: 0; + width: 100%; + min-height: 38px; + padding: 6px 30px 6px 9px; + border: 1px solid #75756f; + color: #111111; + background: #ffffff; + font: inherit; +} + .model-status span, .ratio-control legend, .reference-input small { @@ -419,11 +430,20 @@ font-weight: 800; } -.current-task-active a { +.current-task-active a, +.current-task-active button { color: #111111; font-weight: 800; } +.current-task-active button { + justify-self: start; + min-height: 42px; + padding: 8px 12px; + border: 1px solid #111111; + background: #f2f500; +} + .capacity-critical, .generation-notice { display: flex; diff --git a/apps/web/src/project-pages.tsx b/apps/web/src/project-pages.tsx index 12b0c55..6c10840 100644 --- a/apps/web/src/project-pages.tsx +++ b/apps/web/src/project-pages.tsx @@ -51,18 +51,36 @@ interface ModelPayload { interface GenerationTaskPayload { confirmed_credit_cost: number; created_at: string; + error_category: GenerationErrorCategory | null; generation_id: string; model_config_version: number; model_id: string; project_id: string; prompt: string; ratio: Ratio; + reference_asset_ids?: string[]; reference_count: number; reserved_credits: number; status: "queued" | "running" | "succeeded" | "failed" | "rejected"; updated_at: string; } +type GenerationErrorCategory = "upstream_timeout" | "upstream_failed" | "safety_rejected" | "model_disabled" + | "gateway_balance_insufficient" | "gateway_contract_invalid" | "reference_invalid" + | "unknown_retryable" | "unknown_non_retryable"; + +const generationErrorActions: Record = { + gateway_balance_insufficient: "选择未受影响模型或联系管理员", + gateway_contract_invalid: "选择其他模型或联系管理员", + model_disabled: "选择其他模型或等待", + reference_invalid: "更换或移除参考图", + safety_rejected: "修改提示词或参考图", + unknown_non_retryable: "联系管理员", + unknown_retryable: "稍后重试", + upstream_failed: "稍后重试", + upstream_timeout: "使用原输入重试", +}; + interface ProjectSummary { current_image_id: string | null; deleted_at?: string | null; @@ -207,6 +225,7 @@ export function WorkspacePage() { const [session, setSession] = useState(); const [projects, setProjects] = useState(); const [models, setModels] = useState(); + const [selectedModelId, setSelectedModelId] = useState(); const [currentTask, setCurrentTask] = useState(); const [localData, setLocalData] = useState(); const [generationStateLoaded, setGenerationStateLoaded] = useState(false); @@ -217,6 +236,9 @@ export function WorkspacePage() { const [submitting, setSubmitting] = useState(false); const [generationNotice, setGenerationNotice] = useState(""); const [requiresReconfirmation, setRequiresReconfirmation] = useState(false); + const query = useMemo(() => new URLSearchParams(window.location.search), []); + const [targetProjectId, setTargetProjectId] = useState(() => query.get("retry") ?? query.get("continue") ?? undefined); + const [existingReferenceAssetIds, setExistingReferenceAssetIds] = useState([]); useEffect(() => { let active = true; @@ -245,12 +267,31 @@ export function WorkspacePage() { return () => { active = false; }; }, []); + useEffect(() => { + if (!targetProjectId || prompt) return; + readOptionalJson(`/api/v1/projects/${targetProjectId}`).then((project) => { + if (!project) return; + setPrompt(project.draft_prompt); + setRatio(project.ratio); + }).catch(() => setGenerationNotice("暂时无法读取原项目。")); + }, [prompt, targetProjectId]); + + useEffect(() => { + if (!currentTask || !["queued", "running"].includes(currentTask.status)) return; + const timer = window.setInterval(() => { + readOptionalJson(`/api/v1/generations/${currentTask.generation_id}`) + .then((task) => { if (task) setCurrentTask(task); }) + .catch(() => undefined); + }, 2_000); + return () => window.clearInterval(timer); + }, [currentTask?.generation_id, currentTask?.status]); + const selectedModel = useMemo(() => { if (!models) return undefined; - const modelId = models.recommended_model_id ?? models.configured_default_model_id; + const modelId = selectedModelId ?? models.recommended_model_id ?? models.configured_default_model_id; return models.models.find((model) => model.model_id === modelId && model.enabled && model.contract_validation_status === "verified" && model.runtime_availability.available_for_new_jobs); - }, [models]); + }, [models, selectedModelId]); const referenceBytes = references.reduce((sum, file) => sum + file.size, 0); const referencesValid = selectedModel !== undefined @@ -269,12 +310,13 @@ export function WorkspacePage() { const body = new FormData(); body.append("client_submission_id", crypto.randomUUID()); body.append("confirmed_credit_cost", String(selectedModel.credit_cost)); - body.append("creation_mode", "new_project"); - body.append("existing_reference_asset_ids", "[]"); + body.append("creation_mode", targetProjectId ? "existing_project" : "new_project"); + body.append("existing_reference_asset_ids", JSON.stringify(existingReferenceAssetIds)); body.append("model_config_version", String(selectedModel.config_version)); body.append("model_id", selectedModel.model_id); body.append("prompt", prompt.trim()); body.append("ratio", ratio); + if (targetProjectId) body.append("project_id", targetProjectId); body.append("reference_manifest", JSON.stringify(references.map((file) => ({ file_name: file.name, mime_type: file.type, @@ -323,6 +365,28 @@ export function WorkspacePage() { } } + function handleTerminalAction(task: GenerationTaskPayload) { + const category = task.error_category; + if (!category) return; + if (category === "unknown_non_retryable") { + setGenerationNotice("请联系超级管理员处理此任务。"); + return; + } + setPrompt(task.prompt); + setRatio(task.ratio); + setTargetProjectId(task.project_id); + setExistingReferenceAssetIds(task.reference_asset_ids ?? []); + setCurrentTask(undefined); + if (["model_disabled", "gateway_balance_insufficient", "gateway_contract_invalid"].includes(category)) { + setSelectedModelId(undefined); + setGenerationNotice("请选择当前可用模型后重新提交。"); + } else if (category === "safety_rejected" || category === "reference_invalid") { + setGenerationNotice("请修改输入后重新提交。"); + } else { + setGenerationNotice("已恢复原任务输入,可以重新提交。"); + } + } + async function confirmLatestModelConfiguration() { try { const next = await readJson("/api/v1/models"); @@ -378,7 +442,12 @@ export function WorkspacePage() {
模型 - {selectedModel ? selectedModel.model_id : "当前没有可用于新任务的模型"} + {models?.models.some((model) => model.enabled && model.contract_validation_status === "verified" && model.runtime_availability.available_for_new_jobs) ? ( + + ) : 当前没有可用于新任务的模型}
画面比例 @@ -432,14 +501,20 @@ export function WorkspacePage() {

当前任务

{currentTask ? (
- {currentTask.status === "queued" ? "排队中" : "处理中"} + {{ + failed: "生成失败", queued: "排队中", rejected: "请求未通过", running: "处理中", succeeded: "生成成功", + }[currentTask.status]} {currentTask.prompt}
画面比例
{currentTask.ratio}
参考图
{currentTask.reference_count} 张
-
点数
已冻结 {currentTask.reserved_credits} 点
+
点数
{currentTask.status === "succeeded" ? `已扣除 ${currentTask.confirmed_credit_cost} 点` : ["failed", "rejected"].includes(currentTask.status) ? "已释放冻结点" : `已冻结 ${currentTask.reserved_credits} 点`}
- 返回当前项目 + {currentTask.status === "succeeded" ? 进入编辑 : null} + {["queued", "running"].includes(currentTask.status) ? 返回当前项目 : null} + {["failed", "rejected"].includes(currentTask.status) && currentTask.error_category ? ( + + ) : null}
) : (
没有进行中的任务任务状态会持续显示在这里
diff --git a/apps/worker/src/ai-adapter-contract.ts b/apps/worker/src/ai-adapter-contract.ts new file mode 100644 index 0000000..eb8b76a --- /dev/null +++ b/apps/worker/src/ai-adapter-contract.ts @@ -0,0 +1,54 @@ +import type { GenerationErrorCategory } from "./generation-error-registry.js"; + +export interface GenerationAdapterRequest { + configSnapshot: Readonly>; + generationId: string; + modelId: string; + prompt: string; + ratio: "3:4" | "1:1" | "4:3" | "9:16"; + referenceAssetIds: readonly string[]; +} + +export interface NormalizedGenerationOutput { + bytes: Buffer; + mimeType: "image/jpeg" | "image/png" | "image/webp"; + pixelHeight: number; + pixelWidth: number; +} + +export type GenerationAdapterResult = + | { outputs: readonly NormalizedGenerationOutput[]; status: "completed" } + | { + balanceSignal?: { gatewayAccountRef: string; impactScope: "model" | "account" | "unknown" }; + category: GenerationErrorCategory; + sourceCategory: string; + status: "failed"; + }; + +export interface GenerationAdapter { + start(request: GenerationAdapterRequest): Promise; +} + +type MockResult = GenerationAdapterResult & { unsafeRaw?: string }; + +export class MockGenerationAdapter implements GenerationAdapter { + readonly calls: Array<{ generationId: string; modelId: string }> = []; + private readonly result: MockResult; + + constructor(result: MockResult) { + this.result = result; + } + + async start(request: GenerationAdapterRequest): Promise { + this.calls.push({ generationId: request.generationId, modelId: request.modelId }); + if (this.result.status === "completed") { + return { outputs: this.result.outputs.map((output) => ({ ...output, bytes: Buffer.from(output.bytes) })), status: "completed" }; + } + return { + ...(this.result.balanceSignal ? { balanceSignal: { ...this.result.balanceSignal } } : {}), + category: this.result.category, + sourceCategory: this.result.sourceCategory, + status: "failed", + }; + } +} diff --git a/apps/worker/src/gateway-balance-runtime.ts b/apps/worker/src/gateway-balance-runtime.ts new file mode 100644 index 0000000..b6ddf70 --- /dev/null +++ b/apps/worker/src/gateway-balance-runtime.ts @@ -0,0 +1,241 @@ +import { randomUUID } from "node:crypto"; + +import Database from "better-sqlite3"; +import type BetterSqlite3 from "better-sqlite3"; + +import { configureWorkerDatabase } from "./sqlite-connection.js"; + +const auditRetentionMilliseconds = 180 * 24 * 60 * 60 * 1_000; +const accountRefPattern = /^[a-z][a-z0-9_-]{2,119}$/; +const sourceCategoryPattern = /^[a-z][a-z0-9_]{0,79}$/; + +interface RuntimeRow { + available_for_new_jobs: number; + balance_blocked: number; + gateway_account_ref: string; + model_id: string; + runtime_reason: string; + runtime_version: number; + updated_at: number; +} + +interface BalanceRow { + affected_model_ids_json: string; + balance_status: "available" | "insufficient" | "unknown"; + detected_at: number; + gateway_account_ref: string; + impact_scope: "model" | "account" | "unknown"; + last_confirmed_at: number | null; + recovery_status: "not_required" | "awaiting_confirmation" | "confirmed"; + runtime_unavailable_model_ids_json: string; + source_category: string; +} + +export interface GatewayBalanceView { + affectedModelIds: string[]; + balanceStatus: "available" | "insufficient" | "unknown"; + detectedAt: string; + gatewayAccountRef: string; + impactScope: "model" | "account" | "unknown"; + lastConfirmedAt: string | null; + recoveryStatus: "not_required" | "awaiting_confirmation" | "confirmed"; + runtimeUnavailableModelIds: string[]; + sourceCategory: string; +} + +export class GatewayBalanceRuntime { + readonly database: BetterSqlite3.Database; + private readonly clock: () => number; + private readonly ownsDatabase: boolean; + + constructor(input: { clock?: () => number; database?: BetterSqlite3.Database; databasePath?: string }) { + if (!input.database && !input.databasePath) throw new Error("gateway_balance_database_required"); + this.database = input.database ?? new Database(input.databasePath!); + this.ownsDatabase = !input.database; + this.clock = input.clock ?? Date.now; + if (this.ownsDatabase) configureWorkerDatabase(this.database); + this.migrate(); + } + + close() { + if (this.ownsDatabase) this.database.close(); + } + + seedModels(models: Array<{ gatewayAccountRef: string; modelId: string }>) { + const now = this.clock(); + this.immediate(() => { + for (const model of models) { + if (!accountRefPattern.test(model.gatewayAccountRef) || !model.modelId) throw new Error("gateway_balance_model_invalid"); + this.database.prepare(` + INSERT INTO model_runtime_availability ( + model_id, gateway_account_ref, balance_blocked, available_for_new_jobs, runtime_reason, runtime_version, updated_at + ) VALUES (?, ?, 0, 1, 'available', 1, ?) + ON CONFLICT(model_id) DO UPDATE SET gateway_account_ref = excluded.gateway_account_ref + `).run(model.modelId, model.gatewayAccountRef, now); + } + }); + } + + recordInsufficient(input: { + eventId: string; + gatewayAccountRef: string; + impactScope: "model" | "account" | "unknown"; + modelId: string; + sourceCategory: string; + }) { + if (!input.eventId || !accountRefPattern.test(input.gatewayAccountRef) || !input.modelId + || !sourceCategoryPattern.test(input.sourceCategory)) throw new Error("gateway_balance_signal_invalid"); + return this.immediate(() => { + const receipt = this.database.prepare("SELECT response_json FROM gateway_balance_event_receipts WHERE event_id = ?") + .get(input.eventId) as { response_json: string } | undefined; + if (receipt) return JSON.parse(receipt.response_json) as GatewayBalanceView; + this.database.prepare(` + INSERT INTO model_runtime_availability ( + model_id, gateway_account_ref, balance_blocked, available_for_new_jobs, runtime_reason, runtime_version, updated_at + ) VALUES (?, ?, 0, 1, 'available', 1, ?) + ON CONFLICT(model_id) DO UPDATE SET gateway_account_ref = excluded.gateway_account_ref + `).run(input.modelId, input.gatewayAccountRef, this.clock()); + const rows = this.database.prepare("SELECT model_id FROM model_runtime_availability WHERE gateway_account_ref = ? ORDER BY model_id") + .all(input.gatewayAccountRef) as Array<{ model_id: string }>; + const newlyAffected = input.impactScope === "model" ? [input.modelId] : rows.map((row) => row.model_id); + const existing = this.database.prepare("SELECT runtime_unavailable_model_ids_json FROM gateway_balance_states WHERE gateway_account_ref = ?") + .get(input.gatewayAccountRef) as { runtime_unavailable_model_ids_json: string } | undefined; + const unavailable = [...new Set([...(existing ? JSON.parse(existing.runtime_unavailable_model_ids_json) as string[] : []), ...newlyAffected])].toSorted(); + const now = this.clock(); + for (const modelId of unavailable) { + this.database.prepare(` + UPDATE model_runtime_availability + SET balance_blocked = 1, available_for_new_jobs = 0, runtime_reason = 'gateway_balance_insufficient', + runtime_version = runtime_version + 1, updated_at = ? + WHERE model_id = ? AND gateway_account_ref = ? AND balance_blocked = 0 + `).run(now, modelId, input.gatewayAccountRef); + } + this.database.prepare(` + INSERT INTO gateway_balance_states ( + gateway_account_ref, balance_status, impact_scope, affected_model_ids_json, + runtime_unavailable_model_ids_json, detected_at, last_confirmed_at, recovery_status, source_category + ) VALUES (?, 'insufficient', ?, ?, ?, ?, NULL, 'awaiting_confirmation', ?) + ON CONFLICT(gateway_account_ref) DO UPDATE SET + balance_status = 'insufficient', impact_scope = excluded.impact_scope, + affected_model_ids_json = excluded.affected_model_ids_json, + runtime_unavailable_model_ids_json = excluded.runtime_unavailable_model_ids_json, + detected_at = excluded.detected_at, last_confirmed_at = NULL, + recovery_status = 'awaiting_confirmation', source_category = excluded.source_category + `).run(input.gatewayAccountRef, input.impactScope, JSON.stringify(newlyAffected.toSorted()), JSON.stringify(unavailable), now, input.sourceCategory); + this.database.prepare(` + INSERT INTO admin_operation_logs ( + log_id, actor_type, actor_ref, operation_type, target_type, target_ref, + result, before_summary, after_summary, occurred_at, expires_at + ) VALUES (?, 'system', 'generation_worker', 'gateway_balance_detected', 'gateway_balance_state', ?, 'succeeded', NULL, ?, ?, ?) + `).run(randomUUID(), input.gatewayAccountRef, JSON.stringify({ impact_scope: input.impactScope, runtime_unavailable_model_ids: unavailable }), now, now + auditRetentionMilliseconds); + const result = this.readState(input.gatewayAccountRef)!; + this.database.prepare("INSERT INTO gateway_balance_event_receipts (event_id, gateway_account_ref, response_json, created_at) VALUES (?, ?, ?, ?)") + .run(input.eventId, input.gatewayAccountRef, JSON.stringify(result), now); + return result; + }); + } + + readState(gatewayAccountRef: string) { + const row = this.database.prepare("SELECT * FROM gateway_balance_states WHERE gateway_account_ref = ?").get(gatewayAccountRef) as BalanceRow | undefined; + return row ? this.view(row) : undefined; + } + + listRuntime() { + const rows = this.database.prepare("SELECT * FROM model_runtime_availability ORDER BY model_id").all() as RuntimeRow[]; + return rows.map((row) => ({ + availableForNewJobs: row.available_for_new_jobs === 1, + balanceBlocked: row.balance_blocked === 1, + gatewayAccountRef: row.gateway_account_ref, + modelId: row.model_id, + runtimeReason: row.runtime_reason, + runtimeVersion: row.runtime_version, + updatedAt: new Date(row.updated_at).toISOString(), + })); + } + + restoreWithoutConfirmedRecovery(gatewayAccountRef: string, _actorId: string) { + const state = this.readState(gatewayAccountRef); + if (!state || state.recoveryStatus !== "confirmed") throw new Error("gateway_balance_recovery_unconfirmed"); + throw new Error("gateway_balance_recovery_owned_by_wp3"); + } + + private view(row: BalanceRow): GatewayBalanceView { + return { + affectedModelIds: JSON.parse(row.affected_model_ids_json) as string[], + balanceStatus: row.balance_status, + detectedAt: new Date(row.detected_at).toISOString(), + gatewayAccountRef: row.gateway_account_ref, + impactScope: row.impact_scope, + lastConfirmedAt: row.last_confirmed_at === null ? null : new Date(row.last_confirmed_at).toISOString(), + recoveryStatus: row.recovery_status, + runtimeUnavailableModelIds: JSON.parse(row.runtime_unavailable_model_ids_json) as string[], + sourceCategory: row.source_category, + }; + } + + private immediate(action: () => T) { + if (this.database.inTransaction) return action(); + this.database.exec("BEGIN IMMEDIATE"); + try { + const result = action(); + this.database.exec("COMMIT"); + return result; + } catch (error) { + if (this.database.inTransaction) this.database.exec("ROLLBACK"); + throw error; + } + } + + private migrate() { + this.database.exec(` + CREATE TABLE IF NOT EXISTS model_runtime_availability ( + model_id TEXT PRIMARY KEY, + gateway_account_ref TEXT NOT NULL, + balance_blocked INTEGER NOT NULL CHECK (balance_blocked IN (0, 1)), + available_for_new_jobs INTEGER NOT NULL CHECK (available_for_new_jobs IN (0, 1)), + runtime_reason TEXT NOT NULL CHECK (runtime_reason IN ( + 'available', 'configured_disabled', 'contract_unverified', 'contract_blocked', + 'gateway_balance_insufficient', 'gateway_paused', 'worker_degraded' + )), + runtime_version INTEGER NOT NULL CHECK (runtime_version > 0), + updated_at INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS gateway_balance_states ( + gateway_account_ref TEXT PRIMARY KEY, + balance_status TEXT NOT NULL CHECK (balance_status IN ('available', 'insufficient', 'unknown')), + impact_scope TEXT NOT NULL CHECK (impact_scope IN ('model', 'account', 'unknown')), + affected_model_ids_json TEXT NOT NULL CHECK (json_valid(affected_model_ids_json)), + runtime_unavailable_model_ids_json TEXT NOT NULL CHECK (json_valid(runtime_unavailable_model_ids_json)), + detected_at INTEGER NOT NULL, + last_confirmed_at INTEGER, + recovery_status TEXT NOT NULL CHECK (recovery_status IN ('not_required', 'awaiting_confirmation', 'confirmed')), + source_category TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS gateway_balance_event_receipts ( + event_id TEXT PRIMARY KEY, + gateway_account_ref TEXT NOT NULL, + response_json TEXT NOT NULL CHECK (json_valid(response_json)), + created_at INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS admin_operation_logs ( + log_id TEXT PRIMARY KEY, + actor_type TEXT NOT NULL CHECK (actor_type IN ('system', 'super_admin')), + actor_ref TEXT NOT NULL CHECK (dada_audit_ref_is_safe(actor_ref) = 1), + operation_type TEXT NOT NULL CHECK (dada_audit_ref_is_safe(operation_type) = 1), + target_type TEXT NOT NULL CHECK (dada_audit_ref_is_safe(target_type) = 1), + target_ref TEXT NOT NULL CHECK (dada_audit_ref_is_safe(target_ref) = 1), + result TEXT NOT NULL CHECK (result IN ('succeeded', 'failed')), + before_summary TEXT CHECK (before_summary IS NULL OR dada_audit_summary_is_safe(before_summary) = 1), + after_summary TEXT CHECK (after_summary IS NULL OR dada_audit_summary_is_safe(after_summary) = 1), + occurred_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL CHECK (expires_at = occurred_at + ${auditRetentionMilliseconds}) + ); + CREATE TRIGGER IF NOT EXISTS admin_operation_logs_no_update + BEFORE UPDATE ON admin_operation_logs BEGIN SELECT RAISE(ABORT, 'admin_operation_logs_immutable'); END; + CREATE TRIGGER IF NOT EXISTS admin_operation_logs_no_delete + BEFORE DELETE ON admin_operation_logs + WHEN dada_allow_retention_purge() <> 1 OR OLD.expires_at > dada_retention_purge_now() + BEGIN SELECT RAISE(ABORT, 'admin_operation_logs_immutable'); END; + `); + } +} diff --git a/apps/worker/src/generation-error-registry.ts b/apps/worker/src/generation-error-registry.ts new file mode 100644 index 0000000..25b1292 --- /dev/null +++ b/apps/worker/src/generation-error-registry.ts @@ -0,0 +1,38 @@ +export const generationErrorCategories = [ + "upstream_timeout", + "upstream_failed", + "safety_rejected", + "model_disabled", + "gateway_balance_insufficient", + "gateway_contract_invalid", + "reference_invalid", + "unknown_retryable", + "unknown_non_retryable", +] as const; + +export type GenerationErrorCategory = typeof generationErrorCategories[number]; +export type GenerationUserAction = "retry_same_input" | "wait_and_retry" | "edit_input" | "choose_model_or_contact_admin" | "contact_admin"; + +export interface GenerationErrorDefinition { + creditBehavior: "no_reserve" | "release_if_reserved"; + messageKey: string; + retryPolicy: "immediate" | "after_wait" | "after_edit" | "after_model_change" | "none"; + taskOutcome: "not_created" | "failed" | "rejected" | "failed_or_not_created"; + userAction: GenerationUserAction; +} + +export const generationErrorRegistry: Readonly>> = Object.freeze({ + upstream_timeout: Object.freeze({ creditBehavior: "release_if_reserved", messageKey: "generation.error.upstream_timeout", retryPolicy: "immediate", taskOutcome: "failed", userAction: "retry_same_input" }), + upstream_failed: Object.freeze({ creditBehavior: "release_if_reserved", messageKey: "generation.error.upstream_failed", retryPolicy: "after_wait", taskOutcome: "failed", userAction: "wait_and_retry" }), + safety_rejected: Object.freeze({ creditBehavior: "release_if_reserved", messageKey: "generation.error.safety_rejected", retryPolicy: "after_edit", taskOutcome: "rejected", userAction: "edit_input" }), + model_disabled: Object.freeze({ creditBehavior: "no_reserve", messageKey: "generation.error.model_disabled", retryPolicy: "after_model_change", taskOutcome: "not_created", userAction: "choose_model_or_contact_admin" }), + gateway_balance_insufficient: Object.freeze({ creditBehavior: "release_if_reserved", messageKey: "generation.error.gateway_balance_insufficient", retryPolicy: "after_model_change", taskOutcome: "failed_or_not_created", userAction: "choose_model_or_contact_admin" }), + gateway_contract_invalid: Object.freeze({ creditBehavior: "release_if_reserved", messageKey: "generation.error.gateway_contract_invalid", retryPolicy: "after_model_change", taskOutcome: "failed_or_not_created", userAction: "choose_model_or_contact_admin" }), + reference_invalid: Object.freeze({ creditBehavior: "release_if_reserved", messageKey: "generation.error.reference_invalid", retryPolicy: "after_edit", taskOutcome: "failed_or_not_created", userAction: "edit_input" }), + unknown_retryable: Object.freeze({ creditBehavior: "release_if_reserved", messageKey: "generation.error.unknown_retryable", retryPolicy: "after_wait", taskOutcome: "failed", userAction: "wait_and_retry" }), + unknown_non_retryable: Object.freeze({ creditBehavior: "release_if_reserved", messageKey: "generation.error.unknown_non_retryable", retryPolicy: "none", taskOutcome: "failed", userAction: "contact_admin" }), +}); + +export function isGenerationErrorCategory(value: unknown): value is GenerationErrorCategory { + return typeof value === "string" && generationErrorCategories.includes(value as GenerationErrorCategory); +} diff --git a/apps/worker/src/generation-processor.ts b/apps/worker/src/generation-processor.ts new file mode 100644 index 0000000..a0471e0 --- /dev/null +++ b/apps/worker/src/generation-processor.ts @@ -0,0 +1,436 @@ +import { createHash, randomUUID } from "node:crypto"; +import { existsSync, mkdirSync, renameSync, rmSync, writeFileSync } from "node:fs"; +import { dirname, join, resolve } from "node:path"; + +import Database from "better-sqlite3"; +import type BetterSqlite3 from "better-sqlite3"; + +import type { GenerationAdapter, GenerationAdapterResult, NormalizedGenerationOutput } from "./ai-adapter-contract.js"; +import { GatewayBalanceRuntime } from "./gateway-balance-runtime.js"; +import { generationErrorRegistry, type GenerationErrorCategory } from "./generation-error-registry.js"; +import { configureWorkerDatabase } from "./sqlite-connection.js"; + +const hardLimitBytes = 5_368_709_120; +const warningLimitBytes = 4_294_967_296; +const criticalLimitBytes = 4_831_838_208; +const maximumOutputBytes = 20 * 1_024 * 1_024; +const safeSourceCategory = /^[a-z][a-z0-9_]{0,79}$/; + +interface JobRow { + config_snapshot_json: string; + error_category: GenerationErrorCategory | null; + final_credit_state: "committed" | "released" | null; + generation_id: string; + model_id: string; + owner_id: string; + project_id: string; + prompt: string; + ratio: "3:4" | "1:1" | "4:3" | "9:16"; + status: "queued" | "running" | "succeeded" | "failed" | "rejected"; +} + +interface ReservationRow { + amount: number; + model_id: string; + status: "reserved" | "committed" | "released"; + user_id: string; +} + +export interface GenerationProcessingResult { + category: GenerationErrorCategory | null; + generationId: string; + outputAssetId: string | null; + projectId: string; + status: "succeeded" | "failed" | "rejected"; +} + +function iso(timestamp: number) { + return new Date(timestamp).toISOString(); +} + +function extensionFor(mimeType: NormalizedGenerationOutput["mimeType"]) { + return mimeType === "image/png" ? ".png" : mimeType === "image/jpeg" ? ".jpg" : ".webp"; +} + +function hasExpectedMagic(output: NormalizedGenerationOutput) { + const bytes = output.bytes; + if (output.mimeType === "image/png") return bytes.length >= 8 && bytes.subarray(0, 8).equals(Buffer.from([137, 80, 78, 71, 13, 10, 26, 10])); + if (output.mimeType === "image/jpeg") return bytes.length >= 3 && bytes[0] === 0xff && bytes[1] === 0xd8 && bytes[2] === 0xff; + return bytes.length >= 12 && bytes.subarray(0, 4).toString("ascii") === "RIFF" && bytes.subarray(8, 12).toString("ascii") === "WEBP"; +} + +function capacityClass(bytes: number) { + return { + capacity: bytes < warningLimitBytes ? "normal" : bytes < criticalLimitBytes ? "warning" : "critical", + status: bytes >= hardLimitBytes ? "full" : "active", + } as const; +} + +export class GenerationProcessor { + readonly database: BetterSqlite3.Database; + private readonly adapter: GenerationAdapter; + private readonly clock: () => number; + private readonly dataRoot: string; + private readonly gatewayBalance: GatewayBalanceRuntime; + private readonly workerId: string; + + constructor(input: { adapter: GenerationAdapter; clock?: () => number; dataRoot: string; databasePath: string; workerId: string }) { + if (!input.workerId) throw new Error("generation_worker_id_required"); + this.adapter = input.adapter; + this.clock = input.clock ?? Date.now; + this.dataRoot = resolve(input.dataRoot); + this.workerId = input.workerId; + this.database = new Database(input.databasePath); + configureWorkerDatabase(this.database); + this.migrate(); + this.gatewayBalance = new GatewayBalanceRuntime({ clock: this.clock, database: this.database }); + } + + close() { + this.database.close(); + } + + async processNext() { + const row = this.database.prepare(` + SELECT generation_id FROM generation_jobs + WHERE status = 'queued' AND submission_ready = 1 + ORDER BY created_at, generation_id LIMIT 1 + `).get() as { generation_id: string } | undefined; + return row ? this.processGeneration(row.generation_id) : undefined; + } + + async processGeneration(generationId: string): Promise { + const replay = this.readReceipt(generationId); + if (replay) return replay; + const job = this.claim(generationId); + if (["succeeded", "failed", "rejected"].includes(job.status)) return this.terminalResult(job); + + const references = this.database.prepare(` + SELECT managed_file_id FROM generation_reference_snapshots WHERE generation_id = ? ORDER BY position + `).all(generationId) as Array<{ managed_file_id: string }>; + let adapterResult: GenerationAdapterResult; + try { + adapterResult = await this.adapter.start({ + configSnapshot: JSON.parse(job.config_snapshot_json) as Record, + generationId, + modelId: job.model_id, + prompt: job.prompt, + ratio: job.ratio, + referenceAssetIds: references.map((row) => row.managed_file_id), + }); + } catch { + adapterResult = { category: "unknown_retryable", sourceCategory: "adapter_exception", status: "failed" }; + } + + if (adapterResult.status === "failed") return this.completeFailure(job, adapterResult.category, adapterResult.sourceCategory, adapterResult.balanceSignal); + const output = adapterResult.outputs.length === 1 ? adapterResult.outputs[0] : undefined; + if (!output || !this.isSafeOutput(job, output)) { + return this.completeFailure(job, adapterResult.outputs.length === 1 ? "unknown_retryable" : "gateway_contract_invalid", "output_validation_failed", undefined, true); + } + try { + return this.completeSuccess(job, output); + } catch { + return this.completeFailure(job, "unknown_retryable", "output_persist_failed", undefined, true); + } + } + + private claim(generationId: string) { + return this.immediate(() => { + const row = this.readJob(generationId); + if (row.status === "queued") { + const now = this.clock(); + const changed = this.database.prepare(` + UPDATE generation_jobs + SET status = 'running', lease_owner = ?, lease_expires_at = ?, heartbeat_at = ?, started_at = COALESCE(started_at, ?), + attempt_no = attempt_no + 1, updated_at = ? + WHERE generation_id = ? AND status = 'queued' AND submission_ready = 1 + `).run(this.workerId, now + 30_000, now, now, now, generationId); + if (changed.changes !== 1) throw new Error("generation_claim_conflict"); + return this.readJob(generationId); + } + if (row.status === "running" && row.final_credit_state === null) return row; + return row; + }); + } + + private completeFailure( + job: JobRow, + category: GenerationErrorCategory, + sourceCategory: string, + balanceSignal?: { gatewayAccountRef: string; impactScope: "model" | "account" | "unknown" }, + upstreamOutcomeKnown = false, + ) { + return this.immediate(() => { + const replay = this.readReceipt(job.generation_id); + if (replay) return replay; + const latest = this.readJob(job.generation_id); + if (["succeeded", "failed", "rejected"].includes(latest.status)) return this.terminalResult(latest); + const status = category === "safety_rejected" ? "rejected" as const : "failed" as const; + this.finalizeCredit(job.generation_id, status); + if (balanceSignal) { + this.gatewayBalance.seedModels([{ gatewayAccountRef: balanceSignal.gatewayAccountRef, modelId: job.model_id }]); + this.gatewayBalance.recordInsufficient({ + eventId: `generation-${job.generation_id}`, + gatewayAccountRef: balanceSignal.gatewayAccountRef, + impactScope: balanceSignal.impactScope, + modelId: job.model_id, + sourceCategory: "adapter_balance_signal", + }); + } + const now = this.clock(); + this.database.prepare(` + UPDATE generation_jobs + SET status = ?, error_category = ?, diagnostic_source_category = ?, final_credit_state = 'released', + upstream_outcome_known = ?, upstream_cost_reconciliation = ?, finished_at = ?, updated_at = ?, + lease_owner = NULL, lease_expires_at = NULL, heartbeat_at = NULL + WHERE generation_id = ? AND status = 'running' + `).run( + status, + category, + safeSourceCategory.test(sourceCategory) ? sourceCategory : "adapter_error", + upstreamOutcomeKnown ? 1 : 0, + upstreamOutcomeKnown ? "pending_manual_review" : "not_required", + now, + now, + job.generation_id, + ); + const result = this.terminalResult(this.readJob(job.generation_id)); + this.writeReceipt(result, now); + return result; + }); + } + + private completeSuccess(job: JobRow, output: NormalizedGenerationOutput) { + const fileId = randomUUID(); + const operationId = randomUUID(); + const extension = extensionFor(output.mimeType); + const relativePath = `content/generated/${job.owner_id}/${fileId}${extension}`; + const destinationPath = resolve(this.dataRoot, relativePath); + const stagingDirectory = resolve(this.dataRoot, "staging", operationId); + const stagingPath = join(stagingDirectory, "payload.tmp"); + if (!destinationPath.startsWith(`${this.dataRoot}\\`) && !destinationPath.startsWith(`${this.dataRoot}/`)) throw new Error("generation_output_path_invalid"); + + this.reserveStorage(operationId, output.bytes.byteLength); + let renamed = false; + try { + mkdirSync(stagingDirectory, { recursive: true }); + writeFileSync(stagingPath, output.bytes, { flag: "wx" }); + mkdirSync(dirname(destinationPath), { recursive: true }); + renameSync(stagingPath, destinationPath); + renamed = true; + rmSync(stagingDirectory, { force: true, recursive: true }); + return this.immediate(() => { + const replay = this.readReceipt(job.generation_id); + if (replay) return replay; + const now = this.clock(); + const project = this.database.prepare("SELECT name, state_version, current_image_id FROM projects WHERE project_id = ? AND status = 'active'") + .get(job.project_id) as { current_image_id: string | null; name: string; state_version: number } | undefined; + if (!project) throw new Error("generation_project_unavailable"); + const state = this.database.prepare("SELECT canvas_json FROM project_states WHERE project_id = ? ORDER BY state_version DESC LIMIT 1") + .get(job.project_id) as { canvas_json: string } | undefined; + if (!state) throw new Error("generation_project_state_unavailable"); + const canvas = JSON.parse(state.canvas_json) as { background: { asset_id: string | null }; [key: string]: unknown }; + if (project.current_image_id === null) canvas.background.asset_id = fileId; + this.database.prepare(` + INSERT INTO managed_files (file_id, file_kind, owner_ref, relative_path, byte_size, mime_type, sha256, status, created_at) + VALUES (?, 'generated', ?, ?, ?, ?, ?, 'committed', ?) + `).run(fileId, job.owner_id, relativePath, output.bytes.byteLength, output.mimeType, createHash("sha256").update(output.bytes).digest("hex"), iso(now)); + this.database.prepare("INSERT INTO project_resource_files (project_id, managed_file_id, resource_kind, created_at) VALUES (?, ?, 'generated', ?)") + .run(job.project_id, fileId, now); + this.database.prepare("INSERT INTO project_asset_refs (reference_id, managed_file_id, reference_type, created_at) VALUES (?, ?, 'project', ?)") + .run(`project:${job.project_id}:${fileId}`, fileId, iso(now)); + this.database.prepare(` + INSERT INTO generation_output_assets (generation_id, managed_file_id, pixel_width, pixel_height, created_at) + VALUES (?, ?, ?, ?, ?) + `).run(job.generation_id, fileId, output.pixelWidth, output.pixelHeight, now); + this.database.prepare("INSERT INTO project_images (image_id, project_id, generation_id, created_at) VALUES (?, ?, ?, ?)") + .run(fileId, job.project_id, job.generation_id, now); + this.database.prepare(` + UPDATE projects SET current_image_id = COALESCE(current_image_id, ?), updated_at = ?, state_version = state_version + 1 + WHERE project_id = ? + `).run(fileId, now, job.project_id); + this.database.prepare("INSERT INTO project_states (project_id, state_version, name, canvas_json, created_at) VALUES (?, ?, ?, ?, ?)") + .run(job.project_id, project.state_version + 1, project.name, JSON.stringify(canvas), now); + this.consumeStorage(operationId, output.bytes.byteLength, now); + this.finalizeCredit(job.generation_id, "succeeded"); + this.database.prepare(` + UPDATE generation_jobs + SET status = 'succeeded', error_category = NULL, final_credit_state = 'committed', output_asset_id = ?, + upstream_outcome_known = 1, upstream_cost_reconciliation = 'not_required', finished_at = ?, updated_at = ?, + lease_owner = NULL, lease_expires_at = NULL, heartbeat_at = NULL + WHERE generation_id = ? AND status = 'running' + `).run(fileId, now, now, job.generation_id); + const result = this.terminalResult(this.readJob(job.generation_id)); + this.writeReceipt(result, now); + return result; + }); + } catch (error) { + if (renamed && existsSync(destinationPath)) this.queueCompensation(relativePath, output.bytes.byteLength); + else rmSync(stagingDirectory, { force: true, recursive: true }); + this.releaseStorage(operationId); + throw error; + } + } + + private isSafeOutput(job: JobRow, output: NormalizedGenerationOutput) { + const pixels = { "1:1": [1080, 1080], "3:4": [1080, 1440], "4:3": [1440, 1080], "9:16": [1080, 1920] }[job.ratio]; + return output.bytes.byteLength > 0 && output.bytes.byteLength <= maximumOutputBytes && hasExpectedMagic(output) + && output.pixelWidth === pixels[0] && output.pixelHeight === pixels[1]; + } + + private finalizeCredit(generationId: string, outcome: "succeeded" | "failed" | "rejected") { + const reservation = this.database.prepare("SELECT * FROM credit_reservations WHERE generation_id = ?") + .get(generationId) as ReservationRow | undefined; + if (!reservation) throw new Error("generation_credit_reservation_missing"); + if (reservation.status !== "reserved") return; + const account = this.database.prepare("SELECT available_balance, reserved_balance FROM credit_accounts WHERE user_id = ?") + .get(reservation.user_id) as { available_balance: number; reserved_balance: number } | undefined; + if (!account || account.reserved_balance < reservation.amount) throw new Error("generation_credit_invariant_failed"); + const committed = outcome === "succeeded"; + const availableAfter = committed ? account.available_balance : account.available_balance + reservation.amount; + const reservedAfter = account.reserved_balance - reservation.amount; + const now = this.clock(); + const operationKey = `generation:${generationId}:${committed ? "commit" : "release"}`; + this.database.prepare("UPDATE credit_accounts SET available_balance = ?, reserved_balance = ?, updated_at = ? WHERE user_id = ?") + .run(availableAfter, reservedAfter, now, reservation.user_id); + this.database.prepare("UPDATE credit_reservations SET status = ?, finalized_at = ? WHERE generation_id = ? AND status = 'reserved'") + .run(committed ? "committed" : "released", now, generationId); + this.database.prepare(` + INSERT INTO credit_ledger ( + ledger_id, user_id, operation_key, entry_type, amount, available_before, available_after, + reserved_before, reserved_after, created_at, reference_type, reference_id, model_id, reason, entry_status + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'generation', ?, ?, NULL, ?) + `).run( + randomUUID(), reservation.user_id, operationKey, committed ? "generation_commit" : "generation_release", + committed ? -reservation.amount : reservation.amount, account.available_balance, availableAfter, + account.reserved_balance, reservedAfter, now, generationId, reservation.model_id, committed ? "committed" : "released", + ); + this.database.prepare(` + INSERT INTO outbox_events (event_id, operation_key, topic, aggregate_type, aggregate_id, payload_json, status, created_at, published_at) + VALUES (?, ?, ?, 'generation', ?, ?, 'pending', ?, NULL) + `).run(randomUUID(), operationKey, committed ? "generation_credit_committed" : "generation_credit_released", generationId, JSON.stringify({ outcome }), now); + } + + private reserveStorage(operationId: string, bytes: number) { + this.immediate(() => { + const state = this.database.prepare("SELECT managed_content_bytes, storage_status FROM local_backend_storage_state WHERE singleton = 1") + .get() as { managed_content_bytes: number; storage_status: string } | undefined; + const active = this.database.prepare("SELECT COALESCE(SUM(projected_bytes), 0) AS bytes FROM storage_reservations WHERE status = 'active'") + .get() as { bytes: number }; + if (!state || state.storage_status !== "active" || state.managed_content_bytes + active.bytes + bytes > hardLimitBytes) { + throw new Error("generation_storage_unavailable"); + } + this.database.prepare("INSERT INTO storage_reservations (reservation_id, operation_id, projected_bytes, status, created_at) VALUES (?, ?, ?, 'active', ?)") + .run(randomUUID(), operationId, bytes, iso(this.clock())); + }); + } + + private consumeStorage(operationId: string, bytes: number, now: number) { + this.database.prepare("UPDATE storage_reservations SET status = 'consumed', resolved_at = ? WHERE operation_id = ? AND status = 'active'") + .run(iso(now), operationId); + const state = this.database.prepare("SELECT managed_content_bytes FROM local_backend_storage_state WHERE singleton = 1") + .get() as { managed_content_bytes: number }; + const next = state.managed_content_bytes + bytes; + const classification = capacityClass(next); + this.database.prepare(` + UPDATE local_backend_storage_state + SET managed_content_bytes = ?, capacity_notice_level = ?, storage_status = ?, measured_at = ?, version = version + 1 + WHERE singleton = 1 + `).run(next, classification.capacity, classification.status, iso(now)); + } + + private releaseStorage(operationId: string) { + try { + this.database.prepare("UPDATE storage_reservations SET status = 'released', resolved_at = ? WHERE operation_id = ? AND status = 'active'") + .run(iso(this.clock()), operationId); + } catch { + // Startup storage reconciliation remains the fallback if SQLite is unavailable. + } + } + + private queueCompensation(relativePath: string, bytes: number) { + try { + this.database.prepare(` + INSERT OR IGNORE INTO file_cleanup_queue ( + cleanup_id, managed_file_id, relative_path, byte_size, counts_toward_managed, reason, status, created_at + ) VALUES (?, NULL, ?, ?, 0, 'compensation', 'pending', ?) + `).run(randomUUID(), relativePath, bytes, iso(this.clock())); + } catch { + // Startup reconciliation detects unindexed generated files as a final fallback. + } + } + + private readJob(generationId: string) { + const row = this.database.prepare("SELECT * FROM generation_jobs WHERE generation_id = ? AND submission_ready = 1").get(generationId) as JobRow | undefined; + if (!row) throw new Error("generation_not_found"); + return row; + } + + private terminalResult(row: JobRow): GenerationProcessingResult { + if (!["succeeded", "failed", "rejected"].includes(row.status)) throw new Error("generation_not_terminal"); + const output = this.database.prepare("SELECT managed_file_id FROM generation_output_assets WHERE generation_id = ?") + .get(row.generation_id) as { managed_file_id: string } | undefined; + return { + category: row.error_category, + generationId: row.generation_id, + outputAssetId: output?.managed_file_id ?? null, + projectId: row.project_id, + status: row.status as GenerationProcessingResult["status"], + }; + } + + private readReceipt(generationId: string) { + const row = this.database.prepare("SELECT response_json FROM generation_processor_receipts WHERE generation_id = ?") + .get(generationId) as { response_json: string } | undefined; + return row ? JSON.parse(row.response_json) as GenerationProcessingResult : undefined; + } + + private writeReceipt(result: GenerationProcessingResult, now: number) { + this.database.prepare("INSERT INTO generation_processor_receipts (generation_id, response_json, created_at) VALUES (?, ?, ?)") + .run(result.generationId, JSON.stringify(result), now); + } + + private immediate(action: () => T) { + if (this.database.inTransaction) return action(); + this.database.exec("BEGIN IMMEDIATE"); + try { + const result = action(); + this.database.exec("COMMIT"); + return result; + } catch (error) { + if (this.database.inTransaction) this.database.exec("ROLLBACK"); + throw error; + } + } + + private ensureColumn(table: string, column: string, definition: string) { + const columns = this.database.prepare(`PRAGMA table_info(${table})`).all() as Array<{ name: string }>; + if (!columns.some((entry) => entry.name === column)) this.database.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); + } + + private migrate() { + this.ensureColumn("generation_jobs", "lease_owner", "TEXT"); + this.ensureColumn("generation_jobs", "lease_expires_at", "INTEGER"); + this.ensureColumn("generation_jobs", "heartbeat_at", "INTEGER"); + this.ensureColumn("generation_jobs", "started_at", "INTEGER"); + this.ensureColumn("generation_jobs", "attempt_no", "INTEGER NOT NULL DEFAULT 0"); + this.ensureColumn("generation_jobs", "output_asset_id", "TEXT"); + this.ensureColumn("generation_jobs", "diagnostic_source_category", "TEXT"); + this.ensureColumn("generation_jobs", "upstream_outcome_known", "INTEGER NOT NULL DEFAULT 0"); + this.ensureColumn("generation_jobs", "upstream_cost_reconciliation", "TEXT NOT NULL DEFAULT 'not_applicable'"); + this.database.exec(` + CREATE TABLE IF NOT EXISTS generation_output_assets ( + generation_id TEXT PRIMARY KEY REFERENCES generation_jobs(generation_id), + managed_file_id TEXT NOT NULL UNIQUE REFERENCES managed_files(file_id), + pixel_width INTEGER NOT NULL CHECK (pixel_width > 0), + pixel_height INTEGER NOT NULL CHECK (pixel_height > 0), + created_at INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS generation_processor_receipts ( + generation_id TEXT PRIMARY KEY REFERENCES generation_jobs(generation_id), + response_json TEXT NOT NULL CHECK (json_valid(response_json)), + created_at INTEGER NOT NULL + ); + `); + } +} diff --git a/apps/worker/src/sqlite-connection.ts b/apps/worker/src/sqlite-connection.ts new file mode 100644 index 0000000..d51c7fd --- /dev/null +++ b/apps/worker/src/sqlite-connection.ts @@ -0,0 +1,47 @@ +import type BetterSqlite3 from "better-sqlite3"; + +const auditRefPattern = /^[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}$/; +const safeStringPattern = /^[A-Za-z0-9_.:@-]{1,160}$/; +const forbiddenKeys = new Set([ + "absolute_path", "api_key", "body", "code_hmac", "content", "credential", "email", "image", + "image_content", "password", "path", "prompt", "secret", "session_token", "verification_code", "whitelist", +]); +const forbiddenKeyFragments = ["content", "credential", "email", "image", "password", "path", "prompt", "secret", "token"]; + +function safeSummaryValue(value: unknown, depth: number): boolean { + if (depth > 3) return false; + if (value === null || typeof value === "boolean") return true; + if (typeof value === "number") return Number.isSafeInteger(value); + if (typeof value === "string") return safeStringPattern.test(value) && !value.includes("@"); + if (Array.isArray(value)) return value.length <= 20 && value.every((entry) => safeSummaryValue(entry, depth + 1)); + if (!value || typeof value !== "object") return false; + const entries = Object.entries(value); + return entries.length <= 32 && entries.every(([key, entry]) => ( + auditRefPattern.test(key) + && !forbiddenKeys.has(key.toLowerCase()) + && !forbiddenKeyFragments.some((fragment) => key.toLowerCase().includes(fragment)) + && safeSummaryValue(entry, depth + 1) + )); +} + +export function configureWorkerDatabase(database: BetterSqlite3.Database) { + database.pragma("journal_mode = WAL"); + database.pragma("foreign_keys = ON"); + database.pragma("synchronous = FULL"); + database.pragma("busy_timeout = 5000"); + database.function("dada_audit_ref_is_safe", { deterministic: true }, (value: unknown) => ( + typeof value === "string" && auditRefPattern.test(value) ? 1 : 0 + )); + database.function("dada_audit_summary_is_safe", { deterministic: true }, (value: unknown) => { + if (typeof value !== "string" || Buffer.byteLength(value, "utf8") > 2_048) return 0; + try { + return safeSummaryValue(JSON.parse(value), 0) ? 1 : 0; + } catch { + return 0; + } + }); + database.function("dada_allow_retention_purge", { deterministic: false }, () => 0); + database.function("dada_retention_purge_now", { deterministic: false }, () => 0); + database.function("dada_allow_privacy_purge", { deterministic: false }, () => 0); + database.function("dada_privacy_purge_subject", { deterministic: false }, () => ""); +} diff --git a/openapi/openapi.json b/openapi/openapi.json index ab98e4e..69088c6 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -2402,6 +2402,16 @@ "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}\\.[0-9]{3}Z$", "type": "string" }, + "error_category": { + "anyOf": [ + { + "$ref": "#/components/schemas/GenerationErrorCategory" + }, + { + "type": "null" + } + ] + }, "generation_id": { "pattern": "^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-8][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$", "type": "string" @@ -2426,6 +2436,14 @@ "ratio": { "$ref": "#/components/schemas/ProjectRatio" }, + "reference_asset_ids": { + "items": { + "pattern": "^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-8][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$", + "type": "string" + }, + "maxItems": 16, + "type": "array" + }, "reference_count": { "minimum": 0, "type": "integer" @@ -2445,12 +2463,14 @@ "required": [ "confirmed_credit_cost", "created_at", + "error_category", "generation_id", "model_config_version", "model_id", "project_id", "prompt", "ratio", + "reference_asset_ids", "reference_count", "reserved_credits", "status", diff --git a/package.json b/package.json index 5799118..bcea71f 100644 --- a/package.json +++ b/package.json @@ -14,7 +14,7 @@ "test:integration": "vitest run tests/integration", "test:api": "pnpm check:openapi && vitest run tests/api", "test:worker": "pnpm --filter @dada/worker build && node scripts/worker-smoke.mjs && vitest run tests/worker", - "test:e2e": "pnpm check:openapi && playwright test tests/e2e/event-sync.spec.ts tests/e2e/support-gate.spec.ts tests/e2e/local-data-boundary.spec.ts tests/e2e/storage-capacity.spec.ts tests/e2e/public-asset-cache.spec.ts tests/e2e/user-auth.spec.ts tests/e2e/admin-auth.spec.ts tests/e2e/entry-state-ui.spec.ts tests/e2e/session-invalid-ui.spec.ts tests/e2e/user-registration.spec.ts tests/e2e/account-settings.spec.ts tests/e2e/projects-workspace.spec.ts tests/e2e/project-autosave-conflict.spec.ts tests/e2e/project-trash.spec.ts tests/e2e/credits.spec.ts tests/e2e/generation-workspace.spec.ts --config playwright.config.ts", + "test:e2e": "pnpm check:openapi && playwright test tests/e2e/event-sync.spec.ts tests/e2e/support-gate.spec.ts tests/e2e/local-data-boundary.spec.ts tests/e2e/storage-capacity.spec.ts tests/e2e/public-asset-cache.spec.ts tests/e2e/user-auth.spec.ts tests/e2e/admin-auth.spec.ts tests/e2e/entry-state-ui.spec.ts tests/e2e/session-invalid-ui.spec.ts tests/e2e/user-registration.spec.ts tests/e2e/account-settings.spec.ts tests/e2e/projects-workspace.spec.ts tests/e2e/project-autosave-conflict.spec.ts tests/e2e/project-trash.spec.ts tests/e2e/credits.spec.ts tests/e2e/generation-workspace.spec.ts tests/e2e/generation-terminal-actions.spec.ts --config playwright.config.ts", "test:visual": "node scripts/validate-layer-scope.mjs VISUAL", "test:performance": "node scripts/validate-layer-scope.mjs PERFORMANCE", "test:security": "node scripts/verify-frozen-dependencies.mjs && node scripts/redaction-scan.mjs", @@ -61,7 +61,9 @@ "test:wp2-04": "node scripts/run-wp2-04-validation.mjs", "test:wp2-04:red": "node scripts/run-wp2-04-validation.mjs --phase red", "test:wp2-05": "node scripts/run-wp2-05-validation.mjs", - "test:wp2-05:red": "node scripts/run-wp2-05-validation.mjs --phase red" + "test:wp2-05:red": "node scripts/run-wp2-05-validation.mjs --phase red", + "test:wp2-06": "node scripts/run-wp2-06-validation.mjs", + "test:wp2-06:red": "node scripts/run-wp2-06-validation.mjs --phase red" }, "devDependencies": { "@playwright/test": "1.62.0", diff --git a/packages/shared-contracts/src/generations.ts b/packages/shared-contracts/src/generations.ts index b4adc53..bdbe92e 100644 --- a/packages/shared-contracts/src/generations.ts +++ b/packages/shared-contracts/src/generations.ts @@ -1,5 +1,6 @@ import { Type, type Static } from "@sinclair/typebox"; +import { GenerationErrorCategorySchema } from "./api.js"; import { ProjectIdSchema, ProjectRatioSchema } from "./projects.js"; const uuidPattern = "^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-8][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$"; @@ -16,12 +17,14 @@ export const GenerationTaskStatusSchema = Type.Union([ export const GenerationTaskResponseSchema = Type.Object({ confirmed_credit_cost: Type.Integer({ minimum: 1 }), created_at: Type.String({ pattern: isoTimestampPattern }), + error_category: Type.Union([Type.Ref(GenerationErrorCategorySchema), Type.Null()]), generation_id: Type.String({ pattern: uuidPattern }), model_config_version: Type.Integer({ minimum: 1 }), model_id: Type.String({ maxLength: 160, minLength: 1 }), project_id: Type.Ref(ProjectIdSchema), prompt: Type.String({ maxLength: 4_000, minLength: 1 }), ratio: Type.Ref(ProjectRatioSchema), + reference_asset_ids: Type.Array(Type.String({ pattern: uuidPattern }), { maxItems: 16 }), reference_count: Type.Integer({ minimum: 0 }), reserved_credits: Type.Integer({ minimum: 0 }), status: Type.Ref(GenerationTaskStatusSchema), diff --git a/scripts/run-wp2-06-validation.mjs b/scripts/run-wp2-06-validation.mjs new file mode 100644 index 0000000..7494c1c --- /dev/null +++ b/scripts/run-wp2-06-validation.mjs @@ -0,0 +1,79 @@ +import { createHash } from "node:crypto"; +import { spawnSync } from "node:child_process"; +import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { resolve } from "node:path"; + +const phaseIndex = process.argv.indexOf("--phase"); +const phase = phaseIndex >= 0 ? process.argv[phaseIndex + 1] : "green"; +if (!["red", "green"].includes(phase)) throw new Error(`Unsupported phase: ${phase}`); +const runId = process.env.DADA_TDD_RUN_ID ?? `wp2-06-${phase}-${new Date().toISOString().replace(/[^0-9]/g, "")}`; +const runDirectory = resolve("artifacts", "tdd", runId); +const casesDirectory = resolve(runDirectory, "cases"); +if (existsSync(runDirectory)) throw new Error(`Evidence run already exists: ${runId}`); +mkdirSync(casesDirectory, { recursive: true }); + +const errorCases = [ + "upstream-timeout", "upstream-failed", "safety-rejected", "model-disabled", "gateway-balance", + "gateway-contract", "reference-invalid", "unknown-retryable", "unknown-non-retryable", +].map((suffix) => ({ + acceptance: ["AC-04", "AC-53"], + evidence: [ + "response.json", + "db-diff.json", + ...(suffix === "model-disabled" || suffix.startsWith("gateway-") ? ["external-calls.json"] : ["worker-events.json"]), + `screenshots/${suffix === "gateway-balance" ? "gateway_balance_insufficient" : suffix === "gateway-contract" ? "gateway_contract_invalid" : suffix.replaceAll("-", "_")}.png`, + ], + id: `TDD-WP2-ERR-001-${suffix}`, + requirements: ["GEN-16", "17.18"], +})); +const cases = [ + { acceptance: ["AC-04", "AC-51"], evidence: ["response.json", "db-diff.json", "worker-events.json", "external-calls.json"], id: "TDD-WP2-BAL-001-balance-impact", requirements: ["GEN-14", "GEN-15", "17.17"] }, + ...errorCases, + { acceptance: ["AC-03"], evidence: ["worker-events.json", "db-diff.json", "fs-after.json", "screenshots/succeeded.png"], id: "TDD-WP2-GEN-002-success", requirements: ["CREDIT-03", "GEN-06", "GEN-08", "GEN-11"] }, +]; +for (const item of cases) mkdirSync(resolve(casesDirectory, item.id), { recursive: true }); + +const commands = phase === "red" + ? [ + ["unit", ["exec", "vitest", "run", "tests/unit/wp2-06-generation-error-registry.test.ts"]], + ["integration", ["exec", "vitest", "run", "tests/integration/wp2-06-gateway-balance.test.ts"]], + ["worker", ["exec", "vitest", "run", "tests/worker/wp2-06-generation-processor.test.ts"]], + ["e2e", ["exec", "playwright", "test", "tests/e2e/generation-terminal-actions.spec.ts", "--config", "playwright.config.ts"]], + ] + : [["unit", ["test:unit"]], ["integration", ["test:integration"]], ["api", ["test:api"]], ["worker", ["test:worker"]], ["e2e", ["test:e2e"]], ["tdd-trace", ["validate:tdd-trace"]]]; +const environment = { ...process.env, DADA_EVIDENCE_DIR_GENERATION_RUNTIME: casesDirectory }; +const commandResults = []; +for (const [name, args] of commands) { + const command = `pnpm ${args.join(" ")}`; + const started_at = new Date().toISOString(); + const result = spawnSync(process.env.ComSpec ?? "cmd.exe", ["/d", "/s", "/c", command], { encoding: "utf8", env: environment }); + if (result.stdout) process.stdout.write(result.stdout); + if (result.stderr) process.stderr.write(result.stderr); + commandResults.push({ command, exit_code: result.status ?? 1, finished_at: new Date().toISOString(), name, started_at }); +} +const commandState = phase === "red" ? commandResults.every((result) => result.exit_code !== 0) : commandResults.every((result) => result.exit_code === 0); +const manifest = { path: "tasks.manifest.json", sha256: createHash("sha256").update(readFileSync("tasks.manifest.json")).digest("hex").toUpperCase() }; +const commit = spawnSync("git", ["rev-parse", "--short", "HEAD"], { encoding: "utf8" }).stdout.trim(); +const dirty = spawnSync("git", ["status", "--porcelain"], { encoding: "utf8" }).stdout.trim().length > 0; +const summaries = []; +for (const item of cases) { + const directory = resolve(casesDirectory, item.id); + writeFileSync(resolve(directory, "commands.json"), `${JSON.stringify({ commands: commandResults, phase, run_id: runId }, null, 2)}\n`); + if (phase === "red") writeFileSync(resolve(directory, "red-observation.json"), `${JSON.stringify({ expected_failure: "Generation terminal processor, fixed error registry, gateway balance scope and terminal UI actions are absent", status: commandState ? "red_confirmed" : "failed" }, null, 2)}\n`); + const evidenceRefs = phase === "red" ? ["red-observation.json"] : item.evidence; + const missingEvidence = evidenceRefs.filter((file) => !existsSync(resolve(directory, file))); + const targetStatus = phase === "red" ? "red_confirmed" : "passed"; + const status = commandState && missingEvidence.length === 0 ? targetStatus : "failed"; + writeFileSync(resolve(directory, "result.json"), `${JSON.stringify({ + acceptance_criteria: item.acceptance, automation: ["automated"], commit, evidence_refs: evidenceRefs, + layer: ["DB", "API", "WRK", "UNIT", "E2E"], manifest, missing_evidence: missingEvidence, phase, + requirements: item.requirements, run_id: runId, status, task_id: "TASK-WP2-06", test_id: item.id, + work_package: "WP-2", worktree_under_test: dirty ? "uncommitted implementation" : "clean committed implementation", + }, null, 2)}\n`); + summaries.push({ missing_evidence: missingEvidence, status, test_id: item.id }); +} +const targetStatus = phase === "red" ? "red_confirmed" : "passed"; +const status = summaries.every((item) => item.status === targetStatus) ? targetStatus : "failed"; +writeFileSync(resolve(runDirectory, "evidence.json"), `${JSON.stringify({ cases: summaries, phase, run_id: runId, status }, null, 2)}\n`); +console.log(JSON.stringify({ cases: summaries, phase, run_id: runId, status }, null, 2)); +if (status !== targetStatus) process.exit(1); diff --git a/tests/e2e/generation-terminal-actions.spec.ts b/tests/e2e/generation-terminal-actions.spec.ts new file mode 100644 index 0000000..747dea8 --- /dev/null +++ b/tests/e2e/generation-terminal-actions.spec.ts @@ -0,0 +1,85 @@ +import { mkdirSync } from "node:fs"; +import { resolve } from "node:path"; + +import { expect, test } from "@playwright/test"; + +const webUrl = "http://127.0.0.1:4173"; +const categories = [ + ["upstream_timeout", "使用原输入重试"], + ["upstream_failed", "稍后重试"], + ["safety_rejected", "修改提示词或参考图"], + ["model_disabled", "选择其他模型或等待"], + ["gateway_balance_insufficient", "选择未受影响模型或联系管理员"], + ["gateway_contract_invalid", "选择其他模型或联系管理员"], + ["reference_invalid", "更换或移除参考图"], + ["unknown_retryable", "稍后重试"], + ["unknown_non_retryable", "联系管理员"], +] as const; + +test.use({ trace: "off" }); + +test("TASK-WP2-06 renders the single frozen action for every generation category", async ({ page }) => { + let activeCategory: typeof categories[number][0] = "upstream_timeout"; + await page.route("**/api/v1/auth/session", (route) => route.fulfill({ contentType: "application/json", json: { + audience: "user", authenticated: true, credits: { available_balance: 10, reserved_balance: 0 }, + csrf_token: "csrf-terminal-fixture-00000000000000000000000000000000", + local_data: { capacity_status: "normal", hard_limit_bytes: 5_368_709_120, managed_content_bytes: 0 }, + user: { creator_name: "Terminal User", role: "user", social_id: "@terminal", status: "active", user_id: "00000000-0000-4000-8000-000000000801" }, + } })); + await page.route("**/api/v1/projects?status=active", (route) => route.fulfill({ contentType: "application/json", json: { active_count: 1, active_limit: 20, projects: [] } })); + await page.route("**/api/v1/models", (route) => route.fulfill({ contentType: "application/json", json: { + config_set_version: 1, configured_default_model_id: "gemini-3.1-flash-image-preview", models: [], recommended_model_id: null, + } })); + await page.route("**/api/v1/generations/current", (route) => route.fulfill({ contentType: "application/json", json: { + confirmed_credit_cost: 1, created_at: "2026-08-02T14:00:00.000Z", error_category: activeCategory, + generation_id: "00000000-0000-4000-8000-000000000802", model_config_version: 1, + model_id: "gemini-3.1-flash-image-preview", project_id: "00000000-0000-4000-8000-000000000803", + prompt: "失败任务", ratio: "3:4", reference_count: 0, reserved_credits: 0, + status: activeCategory === "safety_rejected" ? "rejected" : "failed", updated_at: "2026-08-02T14:01:00.000Z", + } })); + const evidenceRoot = process.env.DADA_EVIDENCE_DIR_GENERATION_RUNTIME; + for (const [category, action] of categories) { + activeCategory = category; + await page.goto(`${webUrl}/app`); + await expect(page.getByRole("button", { name: action })).toBeVisible(); + await expect(page.locator(".current-task").getByRole("button")).toHaveCount(1); + await expect(page.getByText(/UPSTREAM_SECRET|internal\.invalid|stack trace/i)).toHaveCount(0); + if (evidenceRoot) { + const suffix = category === "gateway_balance_insufficient" + ? "gateway-balance" + : category === "gateway_contract_invalid" + ? "gateway-contract" + : category.replaceAll("_", "-"); + const directory = resolve(evidenceRoot, `TDD-WP2-ERR-001-${suffix}`, "screenshots"); + mkdirSync(directory, { recursive: true }); + await page.screenshot({ fullPage: true, path: resolve(directory, `${category}.png`) }); + } + } +}); + +test("TASK-WP2-06 succeeded task exposes its result without an error action", async ({ page }) => { + await page.route("**/api/v1/auth/session", (route) => route.fulfill({ contentType: "application/json", json: { + audience: "user", authenticated: true, credits: { available_balance: 9, reserved_balance: 0 }, + csrf_token: "csrf-success-fixture-00000000000000000000000000000000", + local_data: { capacity_status: "normal", hard_limit_bytes: 5_368_709_120, managed_content_bytes: 104 }, + user: { creator_name: "Success User", role: "user", social_id: "@success", status: "active", user_id: "00000000-0000-4000-8000-000000000811" }, + } })); + await page.route("**/api/v1/projects?status=active", (route) => route.fulfill({ contentType: "application/json", json: { active_count: 1, active_limit: 20, projects: [] } })); + await page.route("**/api/v1/models", (route) => route.fulfill({ contentType: "application/json", json: { config_set_version: 1, configured_default_model_id: "gemini-3.1-flash-image-preview", models: [], recommended_model_id: null } })); + await page.route("**/api/v1/generations/current", (route) => route.fulfill({ contentType: "application/json", json: { + confirmed_credit_cost: 1, created_at: "2026-08-02T14:00:00.000Z", error_category: null, + generation_id: "00000000-0000-4000-8000-000000000812", model_config_version: 1, + model_id: "gemini-3.1-flash-image-preview", project_id: "00000000-0000-4000-8000-000000000813", + prompt: "成功任务", ratio: "3:4", reference_count: 0, reserved_credits: 0, + status: "succeeded", updated_at: "2026-08-02T14:01:00.000Z", + } })); + await page.goto(`${webUrl}/app`); + await expect(page.getByText("生成成功", { exact: true })).toBeVisible(); + await expect(page.getByRole("link", { name: "进入编辑" })).toBeVisible(); + const evidenceRoot = process.env.DADA_EVIDENCE_DIR_GENERATION_RUNTIME; + if (evidenceRoot) { + const directory = resolve(evidenceRoot, "TDD-WP2-GEN-002-success", "screenshots"); + mkdirSync(directory, { recursive: true }); + await page.screenshot({ fullPage: true, path: resolve(directory, "succeeded.png") }); + } +}); diff --git a/tests/integration/wp2-06-gateway-balance.test.ts b/tests/integration/wp2-06-gateway-balance.test.ts new file mode 100644 index 0000000..ce4a121 --- /dev/null +++ b/tests/integration/wp2-06-gateway-balance.test.ts @@ -0,0 +1,128 @@ +import { randomUUID } from "node:crypto"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; + +import { afterEach, describe, expect, it } from "vitest"; + +import { CreditService } from "../../apps/api/src/credits.js"; +import { GenerationSubmissionService, StaticGenerationModelCatalog } from "../../apps/api/src/generation-submission.js"; +import { ManagedStorage } from "../../apps/api/src/managed-storage.js"; +import { ProjectService } from "../../apps/api/src/projects.js"; +import { RegistrationService } from "../../apps/api/src/registration.js"; +import { MockResendAdapter } from "../../apps/api/src/resend-adapter.js"; +import { GatewayBalanceRuntime } from "../../apps/worker/src/gateway-balance-runtime.js"; + +const roots: string[] = []; +const now = Date.parse("2026-08-02T14:00:00.000Z"); +const models = ["gemini-3.1-flash-image-preview", "gemini-3-pro-image-preview", "gpt-image-2"]; +const closeables: Array<{ close(): void }> = []; +const balanceEvidence: unknown[] = []; + +function evidence(file: string, value: unknown) { + const root = process.env.DADA_EVIDENCE_DIR_GENERATION_RUNTIME; + if (!root) return; + const directory = resolve(root, "TDD-WP2-BAL-001-balance-impact"); + mkdirSync(directory, { recursive: true }); + writeFileSync(resolve(directory, file), `${JSON.stringify(value, null, 2)}\n`); +} + +afterEach(() => { + for (const value of closeables.splice(0).reverse()) value.close(); + for (const root of roots.splice(0)) rmSync(root, { force: true, recursive: true }); +}); + +describe("TASK-WP2-06 gateway balance scope", () => { + it.each([ + ["model", [models[0]]], + ["account", models], + ["unknown", models], + ] as const)("applies %s impact without rewriting model configuration", (impactScope, expected) => { + const root = mkdtempSync(join(tmpdir(), `dada-wp2-06-balance-${impactScope}-`)); + roots.push(root); + const databasePath = join(root, "dada.sqlite3"); + const registration = new RegistrationService({ + challengePepper: Buffer.alloc(32, 0x31), + clock: () => now, + currentPrivacyNoticeVersion: "p0a-registration-notice-v1", + databasePath, + invitePepper: Buffer.alloc(32, 0x32), + resend: new MockResendAdapter(), + sessionPepper: Buffer.alloc(32, 0x33), + }); + const database = registration.database; + database.exec("CREATE TABLE model_config_integrity (model_id TEXT PRIMARY KEY, config_hash TEXT NOT NULL, enabled INTEGER NOT NULL, is_default INTEGER NOT NULL, recommendation_priority INTEGER NOT NULL)"); + const insertConfig = database.prepare("INSERT INTO model_config_integrity VALUES (?, ?, 1, ?, ?)"); + models.forEach((modelId, index) => insertConfig.run(modelId, `hash-${index + 1}`, index === 0 ? 1 : 0, index + 1)); + const beforeConfig = database.prepare("SELECT * FROM model_config_integrity ORDER BY recommendation_priority").all(); + registration.close(); + + const runtime = new GatewayBalanceRuntime({ clock: () => now, databasePath }); + runtime.seedModels(models.map((modelId) => ({ gatewayAccountRef: "gateway-account-primary", modelId }))); + const eventId = randomUUID(); + const first = runtime.recordInsufficient({ + eventId, gatewayAccountRef: "gateway-account-primary", impactScope, modelId: models[0], sourceCategory: "adapter_balance_signal", + }); + const replay = runtime.recordInsufficient({ + eventId, gatewayAccountRef: "gateway-account-primary", impactScope, modelId: models[0], sourceCategory: "adapter_balance_signal", + }); + expect(replay).toEqual(first); + expect(first.runtimeUnavailableModelIds.toSorted()).toEqual([...expected].toSorted()); + expect(runtime.listRuntime().filter((item) => !item.availableForNewJobs).map((item) => item.modelId).toSorted()).toEqual([...expected].toSorted()); + expect(() => runtime.restoreWithoutConfirmedRecovery("gateway-account-primary", randomUUID())).toThrow("gateway_balance_recovery_unconfirmed"); + const afterConfig = runtime.database.prepare("SELECT * FROM model_config_integrity ORDER BY recommendation_priority").all(); + expect(afterConfig).toEqual(beforeConfig); + expect(runtime.database.prepare("SELECT COUNT(*) AS count FROM admin_operation_logs WHERE operation_type = 'gateway_balance_detected'").get()).toEqual({ count: 1 }); + expect(() => runtime.database.prepare("UPDATE admin_operation_logs SET result = 'failed'").run()).toThrow("admin_operation_logs_immutable"); + expect(() => runtime.database.prepare("DELETE FROM admin_operation_logs").run()).toThrow("admin_operation_logs_immutable"); + balanceEvidence.push({ impact_scope: impactScope, response: first, runtime: runtime.listRuntime() }); + evidence("response.json", { scopes: balanceEvidence }); + evidence("db-diff.json", { config_unchanged: afterConfig, scopes: balanceEvidence }); + evidence("worker-events.json", { replay_deduplicated: true, scopes: balanceEvidence }); + evidence("external-calls.json", { automatic_recharge_calls: 0, recovery_calls: 0 }); + runtime.close(); + }); + + it("blocks a disabled model before job creation or credit reservation", async () => { + const root = mkdtempSync(join(tmpdir(), "dada-wp2-06-model-disabled-")); + roots.push(root); + const databasePath = join(root, "dada.sqlite3"); + const registration = new RegistrationService({ + challengePepper: Buffer.alloc(32, 0x41), clock: () => now, + currentPrivacyNoticeVersion: "p0a-registration-notice-v1", databasePath, + invitePepper: Buffer.alloc(32, 0x42), resend: new MockResendAdapter(), sessionPepper: Buffer.alloc(32, 0x43), + }); + const projects = new ProjectService({ clock: () => now, databasePath }); + const credits = new CreditService({ clock: () => now, databasePath }); + const storage = new ManagedStorage({ dataRoot: root, databasePath }); + const submissions = new GenerationSubmissionService({ + clock: () => now, credits, storage, + models: new StaticGenerationModelCatalog([{ + configSetVersion: 1, configVersion: 1, contractValidationStatus: "verified", creditCost: 1, + enabled: false, modelId: models[0], promptMaxLength: 1_000, + referenceLimits: { maxFileBytes: 1_024, maxFiles: 2, maxTotalBytes: 2_048 }, + runtimeAvailability: { availableForNewJobs: false, reason: "model_disabled" }, supportedRatios: ["3:4"], + }]), + }); + closeables.push(submissions, storage, credits, projects, registration); + const userId = randomUUID(); + registration.database.prepare(` + INSERT INTO users (user_id, normalized_email, role, status, counts_toward_stage_limit, registration_id, created_at) + VALUES (?, ?, 'user', 'active', 1, ?, ?) + `).run(userId, `${userId}@example.invalid`, randomUUID(), now); + registration.database.prepare("INSERT INTO user_profiles (user_id, creator_name, social_id) VALUES (?, 'Disabled User', '@disabled')").run(userId); + registration.database.prepare("INSERT INTO credit_accounts (user_id, available_balance, reserved_balance, updated_at) VALUES (?, 10, 0, ?)").run(userId, now); + await expect(submissions.submit({ + clientSubmissionId: randomUUID(), confirmedCreditCost: 1, existingReferenceAssetIds: [], + idempotencyKey: `generation-${randomUUID()}-${randomUUID()}`, mode: "new_project", modelConfigVersion: 1, + modelId: models[0], newReferences: [], prompt: "禁用模型", ratio: "3:4", userId, + })).rejects.toMatchObject({ code: "generation_blocked", errorCategory: "model_disabled" }); + const counts = Object.fromEntries(["projects", "generation_jobs", "credit_reservations", "outbox_events"].map((table) => [ + table, (registration.database.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get() as { count: number }).count, + ])); + expect(counts).toEqual({ credit_reservations: 0, generation_jobs: 0, outbox_events: 0, projects: 0 }); + evidence("../TDD-WP2-ERR-001-model-disabled/response.json", { error_category: "model_disabled", status: "not_created", user_action: "choose_model_or_contact_admin" }); + evidence("../TDD-WP2-ERR-001-model-disabled/db-diff.json", counts); + evidence("../TDD-WP2-ERR-001-model-disabled/external-calls.json", { adapter_calls: 0 }); + }); +}); diff --git a/tests/unit/wp2-06-generation-error-registry.test.ts b/tests/unit/wp2-06-generation-error-registry.test.ts new file mode 100644 index 0000000..aa14db7 --- /dev/null +++ b/tests/unit/wp2-06-generation-error-registry.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from "vitest"; + +import { generationErrorRegistry } from "../../apps/worker/src/generation-error-registry.js"; + +describe("TASK-WP2-06 fixed generation error registry", () => { + it("contains exactly the nine frozen categories and no supplier-facing text", () => { + expect(Object.keys(generationErrorRegistry).toSorted()).toEqual([ + "gateway_balance_insufficient", "gateway_contract_invalid", "model_disabled", "reference_invalid", + "safety_rejected", "unknown_non_retryable", "unknown_retryable", "upstream_failed", "upstream_timeout", + ]); + expect(generationErrorRegistry).toMatchObject({ + gateway_balance_insufficient: { creditBehavior: "release_if_reserved", taskOutcome: "failed_or_not_created", userAction: "choose_model_or_contact_admin" }, + gateway_contract_invalid: { creditBehavior: "release_if_reserved", taskOutcome: "failed_or_not_created", userAction: "choose_model_or_contact_admin" }, + model_disabled: { creditBehavior: "no_reserve", taskOutcome: "not_created", userAction: "choose_model_or_contact_admin" }, + reference_invalid: { creditBehavior: "release_if_reserved", taskOutcome: "failed_or_not_created", userAction: "edit_input" }, + safety_rejected: { creditBehavior: "release_if_reserved", taskOutcome: "rejected", userAction: "edit_input" }, + unknown_non_retryable: { creditBehavior: "release_if_reserved", retryPolicy: "none", taskOutcome: "failed", userAction: "contact_admin" }, + unknown_retryable: { creditBehavior: "release_if_reserved", retryPolicy: "after_wait", taskOutcome: "failed", userAction: "wait_and_retry" }, + upstream_failed: { creditBehavior: "release_if_reserved", retryPolicy: "after_wait", taskOutcome: "failed", userAction: "wait_and_retry" }, + upstream_timeout: { creditBehavior: "release_if_reserved", retryPolicy: "immediate", taskOutcome: "failed", userAction: "retry_same_input" }, + }); + expect(JSON.stringify(generationErrorRegistry)).not.toMatch(/supplier|stack|https?:\/\//i); + expect(Object.isFrozen(generationErrorRegistry)).toBe(true); + }); +}); diff --git a/tests/worker/wp2-06-generation-processor.test.ts b/tests/worker/wp2-06-generation-processor.test.ts new file mode 100644 index 0000000..e0d46f7 --- /dev/null +++ b/tests/worker/wp2-06-generation-processor.test.ts @@ -0,0 +1,140 @@ +import { randomUUID } from "node:crypto"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; + +import { afterEach, describe, expect, it } from "vitest"; + +import { CreditService } from "../../apps/api/src/credits.js"; +import { GenerationSubmissionService, StaticGenerationModelCatalog } from "../../apps/api/src/generation-submission.js"; +import { ManagedStorage } from "../../apps/api/src/managed-storage.js"; +import { ProjectService } from "../../apps/api/src/projects.js"; +import { RegistrationService } from "../../apps/api/src/registration.js"; +import { MockResendAdapter } from "../../apps/api/src/resend-adapter.js"; +import { MockGenerationAdapter } from "../../apps/worker/src/ai-adapter-contract.js"; +import { GenerationProcessor } from "../../apps/worker/src/generation-processor.js"; + +const now = Date.parse("2026-08-02T14:00:00.000Z"); +const modelId = "gemini-3.1-flash-image-preview"; +const png = Buffer.concat([Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]), Buffer.alloc(96, 0x42)]); +const roots: string[] = []; +const closeables: Array<{ close(): void }> = []; + +function evidence(caseId: string, file: string, value: unknown) { + const root = process.env.DADA_EVIDENCE_DIR_GENERATION_RUNTIME; + if (!root) return; + const directory = resolve(root, caseId); + mkdirSync(directory, { recursive: true }); + writeFileSync(resolve(directory, file), `${JSON.stringify(value, null, 2)}\n`); +} + +async function harness(adapter: MockGenerationAdapter) { + const dataRoot = mkdtempSync(join(tmpdir(), "dada-wp2-06-worker-")); + roots.push(dataRoot); + mkdirSync(join(dataRoot, "db"), { recursive: true }); + const databasePath = join(dataRoot, "db", "dada.sqlite3"); + const registration = new RegistrationService({ + challengePepper: Buffer.alloc(32, 0x31), clock: () => now, currentPrivacyNoticeVersion: "p0a-registration-notice-v1", + databasePath, invitePepper: Buffer.alloc(32, 0x32), resend: new MockResendAdapter(), sessionPepper: Buffer.alloc(32, 0x33), + }); + const projects = new ProjectService({ clock: () => now, databasePath }); + const credits = new CreditService({ clock: () => now, databasePath }); + const storage = new ManagedStorage({ dataRoot, databasePath }); + const submissions = new GenerationSubmissionService({ + clock: () => now, credits, + models: new StaticGenerationModelCatalog([{ + configSetVersion: 1, configVersion: 1, contractValidationStatus: "verified", creditCost: 1, enabled: true, modelId, + promptMaxLength: 1_000, referenceLimits: { maxFileBytes: 1_024, maxFiles: 2, maxTotalBytes: 2_048 }, + runtimeAvailability: { availableForNewJobs: true, reason: null }, supportedRatios: ["3:4"], + }]), storage, + }); + closeables.push(submissions, storage, credits, projects, registration); + const userId = randomUUID(); + registration.database.prepare("INSERT INTO users (user_id, normalized_email, role, status, counts_toward_stage_limit, registration_id, created_at) VALUES (?, ?, 'user', 'active', 1, ?, ?)") + .run(userId, `${userId}@example.invalid`, randomUUID(), now); + registration.database.prepare(` + INSERT INTO user_profiles (user_id, creator_name, social_id, private_content_notice_version, private_content_notice_acknowledged_at) + VALUES (?, 'Worker User', '@worker', NULL, NULL) + `).run(userId); + registration.database.prepare("INSERT INTO credit_accounts (user_id, available_balance, reserved_balance, updated_at) VALUES (?, 10, 0, ?)") + .run(userId, now); + const submitted = await submissions.submit({ + clientSubmissionId: randomUUID(), confirmedCreditCost: 1, existingReferenceAssetIds: [], + idempotencyKey: `generation-${randomUUID()}-${randomUUID()}`, mode: "new_project", modelConfigVersion: 1, + modelId, newReferences: [], prompt: "Worker 终态", ratio: "3:4", userId, + }); + const processor = new GenerationProcessor({ adapter, clock: () => now + 1_000, dataRoot, databasePath, workerId: "worker-fixture" }); + closeables.push(processor); + return { adapter, credits, dataRoot, generationId: submitted.task.generationId, processor, projectId: submitted.task.projectId, registration, userId }; +} + +afterEach(() => { + for (const value of closeables.splice(0).reverse()) value.close(); + for (const root of roots.splice(0)) rmSync(root, { force: true, recursive: true }); +}); + +describe("TASK-WP2-06 generation terminal processing", () => { + it("persists one safe output before succeeding and commits credits once", async () => { + const fixture = await harness(new MockGenerationAdapter({ outputs: [{ bytes: png, mimeType: "image/png", pixelHeight: 1440, pixelWidth: 1080 }], status: "completed" })); + const first = await fixture.processor.processNext(); + const replay = await fixture.processor.processGeneration(fixture.generationId); + expect(first).toMatchObject({ generationId: fixture.generationId, status: "succeeded" }); + expect(replay).toEqual(first); + expect(fixture.registration.database.prepare("SELECT status, error_category, final_credit_state FROM generation_jobs WHERE generation_id = ?").get(fixture.generationId)) + .toEqual({ error_category: null, final_credit_state: "committed", status: "succeeded" }); + expect(fixture.registration.database.prepare("SELECT attempt_no FROM generation_jobs WHERE generation_id = ?").get(fixture.generationId)).toEqual({ attempt_no: 1 }); + expect(fixture.registration.database.prepare("SELECT COUNT(*) AS count FROM project_images WHERE project_id = ?").get(fixture.projectId)).toEqual({ count: 1 }); + expect(fixture.registration.database.prepare("SELECT COUNT(*) AS count FROM managed_files WHERE file_kind = 'generated'").get()).toEqual({ count: 1 }); + evidence("TDD-WP2-GEN-002-success", "worker-events.json", { adapter_calls: fixture.adapter.calls, first, replay }); + evidence("TDD-WP2-GEN-002-success", "db-diff.json", { credits: fixture.credits.readAccount(fixture.userId), generation: first, history_count: 1 }); + evidence("TDD-WP2-GEN-002-success", "fs-after.json", { generated_files: 1, staging_files: 0 }); + }); + + it("releases credits and requires reconciliation when persistence fails after upstream success", async () => { + const adapter = new MockGenerationAdapter({ outputs: [{ bytes: png, mimeType: "image/png", pixelHeight: 1440, pixelWidth: 1080 }], status: "completed" }); + const fixture = await harness(adapter); + fixture.registration.database.prepare("UPDATE local_backend_storage_state SET storage_status = 'unavailable' WHERE singleton = 1").run(); + const first = await fixture.processor.processNext(); + const replay = await fixture.processor.processGeneration(fixture.generationId); + expect(first).toMatchObject({ category: "unknown_retryable", status: "failed" }); + expect(replay).toEqual(first); + expect(adapter.calls).toHaveLength(1); + expect(fixture.credits.readAccount(fixture.userId)).toMatchObject({ availableBalance: 10, reservedBalance: 0 }); + expect(fixture.registration.database.prepare(` + SELECT final_credit_state, upstream_cost_reconciliation FROM generation_jobs WHERE generation_id = ? + `).get(fixture.generationId)).toEqual({ final_credit_state: "released", upstream_cost_reconciliation: "pending_manual_review" }); + }); + + it.each([ + "upstream_timeout", "upstream_failed", "safety_rejected", "gateway_balance_insufficient", + "gateway_contract_invalid", "reference_invalid", "unknown_retryable", "unknown_non_retryable", + ] as const)("settles %s once without output or raw supplier details", async (category) => { + const adapter = new MockGenerationAdapter({ + category, + ...(category === "gateway_balance_insufficient" ? { balanceSignal: { gatewayAccountRef: "gateway-account-primary", impactScope: "model" as const } } : {}), + sourceCategory: "sanitized_fixture", status: "failed", unsafeRaw: "UPSTREAM_SECRET https://internal.invalid stack trace", + }); + const fixture = await harness(adapter); + const first = await fixture.processor.processNext(); + const replay = await fixture.processor.processGeneration(fixture.generationId); + const expectedStatus = category === "safety_rejected" ? "rejected" : "failed"; + expect(first).toMatchObject({ category, generationId: fixture.generationId, status: expectedStatus }); + expect(replay).toEqual(first); + expect(fixture.credits.readAccount(fixture.userId)).toMatchObject({ availableBalance: 10, reservedBalance: 0 }); + expect(fixture.registration.database.prepare("SELECT COUNT(*) AS count FROM managed_files WHERE file_kind = 'generated'").get()).toEqual({ count: 0 }); + expect(fixture.registration.database.prepare("SELECT COUNT(*) AS count FROM credit_ledger WHERE reference_id = ? AND entry_type = 'generation_release'").get(fixture.generationId)).toEqual({ count: 1 }); + expect(JSON.stringify(first)).not.toMatch(/UPSTREAM_SECRET|internal\.invalid|stack trace/i); + const suffix = category === "gateway_balance_insufficient" + ? "gateway-balance" + : category === "gateway_contract_invalid" + ? "gateway-contract" + : category.replaceAll("_", "-"); + const caseId = `TDD-WP2-ERR-001-${suffix}`; + evidence(caseId, "response.json", first); + evidence(caseId, "worker-events.json", { adapter_calls: adapter.calls, replay_deduplicated: true }); + evidence(caseId, "db-diff.json", { credits: fixture.credits.readAccount(fixture.userId), generated_files: 0 }); + if (["gateway_balance_insufficient", "gateway_contract_invalid"].includes(category)) { + evidence(caseId, "external-calls.json", { leaked_supplier_fields: 0, total_calls: 1 }); + } + }); +});