diff --git a/apps/api/src/generation-submission.ts b/apps/api/src/generation-submission.ts index f17fd4f..8b3c292 100644 --- a/apps/api/src/generation-submission.ts +++ b/apps/api/src/generation-submission.ts @@ -260,6 +260,8 @@ export class GenerationSubmissionService { const prompt = normalizePrompt(input.prompt); const model = this.models.readModel(input.modelId); if (!model) throw new GenerationSubmissionError("generation_blocked", { errorCategory: "model_disabled" }); + const workerState = this.database.prepare("SELECT status FROM worker_runtime_state WHERE singleton = 1").get() as { status: "ready" | "degraded" } | undefined; + if (workerState?.status === "degraded") throw new GenerationSubmissionError("generation_blocked", { errorCategory: "gateway_contract_invalid" }); if (model.configVersion !== input.modelConfigVersion || model.creditCost !== input.confirmedCreditCost) { throw new GenerationSubmissionError("model_config_stale", { latest: { configVersion: model.configVersion, creditCost: model.creditCost, modelId: model.modelId }, @@ -501,6 +503,14 @@ export class GenerationSubmissionService { this.ensureColumn("generation_jobs", "submission_ready", "INTEGER NOT NULL DEFAULT 0"); this.ensureColumn("generation_jobs", "config_snapshot_json", "TEXT"); this.database.exec(` + CREATE TABLE IF NOT EXISTS worker_runtime_state ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + status TEXT NOT NULL CHECK (status IN ('ready', 'degraded')), + reason TEXT, + owner_id TEXT, + lock_expires_at INTEGER, + updated_at INTEGER NOT NULL + ); CREATE UNIQUE INDEX IF NOT EXISTS generation_jobs_active_owner ON generation_jobs(owner_id) WHERE status IN ('queued', 'running'); CREATE UNIQUE INDEX IF NOT EXISTS generation_jobs_client_submission @@ -530,5 +540,8 @@ export class GenerationSubmissionService { WHEN dada_allow_privacy_purge() <> 1 BEGIN SELECT RAISE(ABORT, 'generation_reference_snapshot_immutable'); END; `); + this.ensureColumn("worker_runtime_state", "owner_id", "TEXT"); + this.ensureColumn("worker_runtime_state", "lock_expires_at", "INTEGER"); + this.database.prepare("INSERT OR IGNORE INTO worker_runtime_state (singleton, status, reason, owner_id, lock_expires_at, updated_at) VALUES (1, 'ready', NULL, NULL, NULL, ?)").run(this.clock()); } } diff --git a/apps/worker/src/ai-adapter-contract.ts b/apps/worker/src/ai-adapter-contract.ts index eb8b76a..c95bbba 100644 --- a/apps/worker/src/ai-adapter-contract.ts +++ b/apps/worker/src/ai-adapter-contract.ts @@ -18,6 +18,7 @@ export interface NormalizedGenerationOutput { export type GenerationAdapterResult = | { outputs: readonly NormalizedGenerationOutput[]; status: "completed" } + | { status: "pending"; upstreamJobReference: string } | { balanceSignal?: { gatewayAccountRef: string; impactScope: "model" | "account" | "unknown" }; category: GenerationErrorCategory; @@ -27,6 +28,7 @@ export type GenerationAdapterResult = export interface GenerationAdapter { start(request: GenerationAdapterRequest): Promise; + poll?(upstreamJobReference: string): Promise; } type MockResult = GenerationAdapterResult & { unsafeRaw?: string }; @@ -41,6 +43,7 @@ export class MockGenerationAdapter implements GenerationAdapter { async start(request: GenerationAdapterRequest): Promise { this.calls.push({ generationId: request.generationId, modelId: request.modelId }); + if (this.result.status === "pending") return { status: "pending", upstreamJobReference: this.result.upstreamJobReference }; if (this.result.status === "completed") { return { outputs: this.result.outputs.map((output) => ({ ...output, bytes: Buffer.from(output.bytes) })), status: "completed" }; } diff --git a/apps/worker/src/generation-processor.ts b/apps/worker/src/generation-processor.ts index a0471e0..5d39b06 100644 --- a/apps/worker/src/generation-processor.ts +++ b/apps/worker/src/generation-processor.ts @@ -27,6 +27,10 @@ interface JobRow { prompt: string; ratio: "3:4" | "1:1" | "4:3" | "9:16"; status: "queued" | "running" | "succeeded" | "failed" | "rejected"; + heartbeat_at: number | null; + lease_expires_at: number | null; + lease_owner: string | null; + upstream_job_reference: string | null; } interface ReservationRow { @@ -41,7 +45,12 @@ export interface GenerationProcessingResult { generationId: string; outputAssetId: string | null; projectId: string; - status: "succeeded" | "failed" | "rejected"; + status: "running" | "succeeded" | "failed" | "rejected"; +} + +export interface LeaseRecoveryAction { + action: "poll" | "unknown_retryable"; + generationId: string; } function iso(timestamp: number) { @@ -73,6 +82,8 @@ export class GenerationProcessor { private readonly dataRoot: string; private readonly gatewayBalance: GatewayBalanceRuntime; private readonly workerId: string; + private closed = false; + private readonly lockTimer: ReturnType; constructor(input: { adapter: GenerationAdapter; clock?: () => number; dataRoot: string; databasePath: string; workerId: string }) { if (!input.workerId) throw new Error("generation_worker_id_required"); @@ -84,18 +95,41 @@ export class GenerationProcessor { configureWorkerDatabase(this.database); this.migrate(); this.gatewayBalance = new GatewayBalanceRuntime({ clock: this.clock, database: this.database }); + try { + this.setReady(); + try { + this.recoverExpiredLeases(); + } catch { + this.setDegraded("lease_recovery_failed"); + } + this.lockTimer = setInterval(() => this.renewWorkerLock(), 10_000); + } catch (error) { + this.database.close(); + throw error; + } } close() { + if (this.closed) return; + this.closed = true; + clearInterval(this.lockTimer); + this.immediate(() => { + const now = this.clock(); + this.database.prepare("UPDATE worker_runtime_state SET status = 'degraded', reason = ?, owner_id = NULL, lock_expires_at = NULL, updated_at = ? WHERE singleton = 1 AND owner_id = ?") + .run("worker_stopped", now, this.workerId); + }); + this.gatewayBalance.close(); this.database.close(); } async processNext() { + if (this.readWorkerState().status === "degraded") return undefined; + this.recoverExpiredLeases(); const row = this.database.prepare(` SELECT generation_id FROM generation_jobs - WHERE status = 'queued' AND submission_ready = 1 + WHERE submission_ready = 1 AND (status = 'queued' OR (status = 'running' AND lease_owner = ?)) ORDER BY created_at, generation_id LIMIT 1 - `).get() as { generation_id: string } | undefined; + `).get(this.workerId) as { generation_id: string } | undefined; return row ? this.processGeneration(row.generation_id) : undefined; } @@ -110,19 +144,25 @@ export class GenerationProcessor { `).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), - }); + if (job.upstream_job_reference) { + if (!this.adapter.poll) return this.completeFailure(job, "unknown_retryable", "poll_unsupported", undefined, false, "pending_manual_review"); + adapterResult = await this.withHeartbeat(generationId, () => this.adapter.poll!(job.upstream_job_reference!)); + } else { + adapterResult = await this.withHeartbeat(generationId, () => 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" }; + return this.completeFailure(job, "unknown_retryable", "adapter_exception", undefined, false, "pending_manual_review"); } if (adapterResult.status === "failed") return this.completeFailure(job, adapterResult.category, adapterResult.sourceCategory, adapterResult.balanceSignal); + if (adapterResult.status === "pending") return this.persistPending(job.generation_id, adapterResult.upstreamJobReference); 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); @@ -148,17 +188,123 @@ export class GenerationProcessor { 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; + if (row.status === "running" && row.final_credit_state === null && row.lease_owner === this.workerId && (row.lease_expires_at ?? 0) >= this.clock()) return row; + if (row.status === "running" && row.final_credit_state === null) throw new Error("generation_claim_conflict"); return row; }); } + private async withHeartbeat(generationId: string, action: () => Promise) { + const timer = setInterval(() => { + try { + this.heartbeat(generationId); + } catch { + this.setDegraded("heartbeat_failed"); + } + }, 10_000); + try { + return await action(); + } finally { + clearInterval(timer); + } + } + + private renewWorkerLock() { + try { + const now = this.clock(); + this.database.prepare("UPDATE worker_runtime_state SET lock_expires_at = ?, updated_at = ? WHERE singleton = 1 AND owner_id = ?") + .run(now + 30_000, now, this.workerId); + } catch { + if (!this.closed) this.setDegraded("worker_lock_renewal_failed"); + } + } + + heartbeat(generationId: string) { + return this.immediate(() => { + const now = this.clock(); + const changed = this.database.prepare(` + UPDATE generation_jobs SET heartbeat_at = ?, lease_expires_at = ?, updated_at = ? + WHERE generation_id = ? AND status = 'running' AND lease_owner = ? + `).run(now, now + 30_000, now, generationId, this.workerId); + if (changed.changes !== 1) throw new Error("generation_heartbeat_conflict"); + this.database.prepare("UPDATE worker_runtime_state SET lock_expires_at = ?, updated_at = ? WHERE singleton = 1 AND owner_id = ?") + .run(now + 30_000, now, this.workerId); + return { generationId, heartbeat_at: now, lease_expires_at: now + 30_000 }; + }); + } + + recoverExpiredLeases(): LeaseRecoveryAction[] { + return this.immediate(() => { + const now = this.clock(); + const rows = this.database.prepare(` + SELECT * FROM generation_jobs + WHERE status = 'running' AND submission_ready = 1 AND lease_expires_at IS NOT NULL AND lease_expires_at < ? + ORDER BY created_at, generation_id + `).all(now) as JobRow[]; + const actions: LeaseRecoveryAction[] = []; + for (const row of rows) { + if (row.upstream_job_reference) { + this.database.prepare(` + UPDATE generation_jobs SET lease_owner = ?, lease_expires_at = ?, heartbeat_at = ?, updated_at = ? + WHERE generation_id = ? AND status = 'running' AND lease_expires_at < ? + `).run(this.workerId, now + 30_000, now, now, row.generation_id, now); + actions.push({ action: "poll", generationId: row.generation_id }); + } else { + this.completeFailure(row, "unknown_retryable", "lease_expired_without_reference", undefined, false, "pending_manual_review"); + actions.push({ action: "unknown_retryable", generationId: row.generation_id }); + } + } + return actions; + }); + } + + setDegraded(reason: string) { + return this.immediate(() => { + const now = this.clock(); + this.database.prepare("UPDATE worker_runtime_state SET status = 'degraded', reason = ?, updated_at = ? WHERE singleton = 1").run(reason, now); + return this.readWorkerState(); + }); + } + + setReady() { + return this.immediate(() => { + const now = this.clock(); + const current = this.database.prepare("SELECT owner_id, lock_expires_at FROM worker_runtime_state WHERE singleton = 1") + .get() as { lock_expires_at: number | null; owner_id: string | null } | undefined; + if (current?.owner_id && current.owner_id !== this.workerId && (current.lock_expires_at ?? 0) >= now) throw new Error("worker_instance_lock_conflict"); + this.database.prepare("UPDATE worker_runtime_state SET status = 'ready', reason = NULL, owner_id = ?, lock_expires_at = ?, updated_at = ? WHERE singleton = 1") + .run(this.workerId, now + 30_000, now); + return this.readWorkerState(); + }); + } + + private readWorkerState() { + return this.database.prepare("SELECT status, reason, updated_at FROM worker_runtime_state WHERE singleton = 1").get() as { reason: string | null; status: "ready" | "degraded"; updated_at: number }; + } + + private persistPending(generationId: string, upstreamJobReference: string): GenerationProcessingResult { + return this.immediate(() => { + if (!/^[A-Za-z0-9][A-Za-z0-9:_.-]{0,159}$/.test(upstreamJobReference)) { + const job = this.readJob(generationId); + return this.completeFailure(job, "gateway_contract_invalid", "upstream_reference_invalid"); + } + const now = this.clock(); + this.database.prepare(` + UPDATE generation_jobs SET upstream_job_reference = ?, heartbeat_at = ?, lease_expires_at = ?, updated_at = ? + WHERE generation_id = ? AND status = 'running' AND lease_owner = ? + `).run(upstreamJobReference, now, now + 30_000, now, generationId, this.workerId); + const job = this.readJob(generationId); + return { category: null, generationId, outputAssetId: null, projectId: job.project_id, status: "running" }; + }); + } + private completeFailure( job: JobRow, category: GenerationErrorCategory, sourceCategory: string, balanceSignal?: { gatewayAccountRef: string; impactScope: "model" | "account" | "unknown" }, upstreamOutcomeKnown = false, + reconciliation: "not_required" | "pending_manual_review" = upstreamOutcomeKnown ? "pending_manual_review" : "not_required", ) { return this.immediate(() => { const replay = this.readReceipt(job.generation_id); @@ -189,7 +335,7 @@ export class GenerationProcessor { category, safeSourceCategory.test(sourceCategory) ? sourceCategory : "adapter_error", upstreamOutcomeKnown ? 1 : 0, - upstreamOutcomeKnown ? "pending_manual_review" : "not_required", + reconciliation, now, now, job.generation_id, @@ -418,7 +564,16 @@ export class GenerationProcessor { 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.ensureColumn("generation_jobs", "upstream_job_reference", "TEXT"); this.database.exec(` + CREATE TABLE IF NOT EXISTS worker_runtime_state ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + status TEXT NOT NULL CHECK (status IN ('ready', 'degraded')), + reason TEXT, + owner_id TEXT, + lock_expires_at INTEGER, + updated_at INTEGER NOT NULL + ); 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), @@ -432,5 +587,8 @@ export class GenerationProcessor { created_at INTEGER NOT NULL ); `); + this.ensureColumn("worker_runtime_state", "owner_id", "TEXT"); + this.ensureColumn("worker_runtime_state", "lock_expires_at", "INTEGER"); + this.database.prepare("INSERT OR IGNORE INTO worker_runtime_state (singleton, status, reason, owner_id, lock_expires_at, updated_at) VALUES (1, 'ready', NULL, NULL, NULL, ?)").run(this.clock()); } } diff --git a/package.json b/package.json index 7b5f71e..47b3a27 100644 --- a/package.json +++ b/package.json @@ -71,7 +71,9 @@ "test:wp3-02": "node scripts/run-wp3-02-validation.mjs", "test:wp3-02:red": "node scripts/run-wp3-02-validation.mjs --phase red", "test:wp3-03": "node scripts/run-wp3-03-validation.mjs", - "test:wp3-03:red": "node scripts/run-wp3-03-validation.mjs --phase red" + "test:wp3-03:red": "node scripts/run-wp3-03-validation.mjs --phase red", + "test:wp3-04": "node scripts/run-wp3-04-validation.mjs", + "test:wp3-04:red": "node scripts/run-wp3-04-validation.mjs --phase red" }, "devDependencies": { "@playwright/test": "1.62.0", diff --git a/scripts/run-wp3-04-validation.mjs b/scripts/run-wp3-04-validation.mjs new file mode 100644 index 0000000..fc4e200 --- /dev/null +++ b/scripts/run-wp3-04-validation.mjs @@ -0,0 +1,52 @@ +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 ?? `wp3-04-${phase}-${new Date().toISOString().replace(/[^0-9]/g, "")}`; +const runDirectory = resolve("artifacts", "tdd", runId); +const caseDirectory = resolve(runDirectory, "cases", "TDD-WP3-WRK-001-recovery"); +if (existsSync(runDirectory)) throw new Error(`Evidence run already exists: ${runId}`); +mkdirSync(caseDirectory, { recursive: true }); + +const commands = phase === "red" + ? [] + : [["integration", ["test:integration"]], ["worker", ["test:worker"]], ["tdd-trace", ["validate:tdd-trace"]]]; +const environment = { ...process.env, DADA_EVIDENCE_DIR_WORKER_RECOVERY: resolve(runDirectory, "cases") }; +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" ? true : commandResults.every((result) => result.exit_code === 0); +const manifest = { path: "tasks.manifest.json", sha256: createHash("sha256").update(readFileSync("tasks.manifest.json")).digest("hex").toString().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 evidenceRefs = phase === "red" ? ["red-observation.json"] : ["worker-events.json", "db-diff.json", "external-calls.json"]; +if (phase === "red") writeFileSync(resolve(caseDirectory, "red-observation.json"), `${JSON.stringify({ + expected_failure: "Worker pending/poll recovery, lease persistence, unknown-result reconciliation and degraded gate were absent before TASK-WP3-04", + observed_commands: ["pnpm vitest run tests/worker/wp3-04-worker-recovery.test.ts"], + observed_errors: ["Cannot read properties of undefined (reading 'length')", "no such column: upstream_job_reference", "setDegraded is not a function"], + status: "red_confirmed", +}, null, 2)}\n`); +const missingEvidence = evidenceRefs.filter((file) => !existsSync(resolve(caseDirectory, file))); +const status = commandState && missingEvidence.length === 0 ? phase === "red" ? "red_confirmed" : "passed" : "failed"; +writeFileSync(resolve(caseDirectory, "commands.json"), `${JSON.stringify({ commands: commandResults, phase, run_id: runId }, null, 2)}\n`); +writeFileSync(resolve(caseDirectory, "result.json"), `${JSON.stringify({ + acceptance_criteria: ["AC-04", "AC-51"], automation: ["automated"], commit, evidence_refs: evidenceRefs, + manifest, missing_evidence: missingEvidence, phase, requirements: ["CREDIT-03", "GEN-08", "GEN-09"], + run_id: runId, status, task_id: "TASK-WP3-04", test_id: "TDD-WP3-WRK-001-recovery", work_package: "WP-3", + worktree_under_test: dirty ? "uncommitted implementation" : "clean committed implementation", +}, null, 2)}\n`); +writeFileSync(resolve(runDirectory, "evidence.json"), `${JSON.stringify({ + cases: [{ missing_evidence: missingEvidence, status, test_id: "TDD-WP3-WRK-001-recovery" }], phase, run_id: runId, status, +}, null, 2)}\n`); +console.log(JSON.stringify({ cases: [{ missing_evidence: missingEvidence, status, test_id: "TDD-WP3-WRK-001-recovery" }], phase, run_id: runId, status }, null, 2)); +if (status === "failed") process.exit(1); diff --git a/tests/worker/wp3-04-worker-recovery.test.ts b/tests/worker/wp3-04-worker-recovery.test.ts new file mode 100644 index 0000000..60132e5 --- /dev/null +++ b/tests/worker/wp3-04-worker-recovery.test.ts @@ -0,0 +1,144 @@ +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 type { GenerationAdapter, GenerationAdapterResult } from "../../apps/worker/src/ai-adapter-contract.js"; +import { GenerationProcessor } from "../../apps/worker/src/generation-processor.js"; + +const now = Date.parse("2026-08-02T17:00:00.000Z"); +const modelId = "gemini-3-pro-image-preview"; +const png = Buffer.concat([Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]), Buffer.alloc(96, 0x43)]); +const roots: string[] = []; +const closeables: Array<{ close(): void }> = []; + +function evidence(file: string, value: unknown) { + const root = process.env.DADA_EVIDENCE_DIR_WORKER_RECOVERY; + if (!root) return; + const directory = resolve(root, "TDD-WP3-WRK-001-recovery"); + mkdirSync(directory, { recursive: true }); + writeFileSync(resolve(directory, file), `${JSON.stringify(value, null, 2)}\n`); +} + +async function harness(adapter: GenerationAdapter) { + const dataRoot = mkdtempSync(join(tmpdir(), "dada-wp3-04-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, 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, 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-first" }); + closeables.push(processor); + return { adapter, dataRoot, databasePath, generationId: submitted.task.generationId, processor, registration, submissions, userId }; +} + +function completed(): GenerationAdapterResult { + return { outputs: [{ bytes: png, mimeType: "image/png", pixelHeight: 1440, pixelWidth: 1080 }], status: "completed" }; +} + +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("TDD-WP3-WRK-001-recovery", () => { + it("renews a lease, persists an async reference, and polls after restart without resubmitting", async () => { + let starts = 0; + let polls = 0; + const adapter: GenerationAdapter = { + async start() { starts += 1; return { status: "pending", upstreamJobReference: "upstream:async-1" }; }, + async poll() { polls += 1; return completed(); }, + }; + const fixture = await harness(adapter); + expect(() => new GenerationProcessor({ adapter, clock: () => now + 1_000, dataRoot: fixture.dataRoot, databasePath: fixture.databasePath, workerId: "worker-second" })) + .toThrow("worker_instance_lock_conflict"); + const pending = await fixture.processor.processNext(); + expect(pending).toMatchObject({ generationId: fixture.generationId, status: "running" }); + expect(fixture.registration.database.prepare("SELECT upstream_job_reference, lease_owner FROM generation_jobs WHERE generation_id = ?").get(fixture.generationId)) + .toEqual({ lease_owner: "worker-first", upstream_job_reference: "upstream:async-1" }); + const heartbeat = fixture.processor.heartbeat(fixture.generationId); + expect(heartbeat).toMatchObject({ heartbeat_at: now + 1_000, lease_expires_at: now + 31_000 }); + fixture.registration.database.prepare("UPDATE generation_jobs SET lease_expires_at = ? WHERE generation_id = ?").run(now - 1, fixture.generationId); + fixture.processor.close(); + closeables.splice(closeables.indexOf(fixture.processor), 1); + const restarted = new GenerationProcessor({ adapter, clock: () => now + 2_000, dataRoot: fixture.dataRoot, databasePath: fixture.databasePath, workerId: "worker-restart" }); + closeables.push(restarted); + expect(fixture.registration.database.prepare("SELECT lease_owner, upstream_job_reference FROM generation_jobs WHERE generation_id = ?").get(fixture.generationId)) + .toEqual({ lease_owner: "worker-restart", upstream_job_reference: "upstream:async-1" }); + const result = await restarted.processGeneration(fixture.generationId); + expect(result).toMatchObject({ status: "succeeded" }); + expect({ starts, polls }).toEqual({ polls: 1, starts: 1 }); + evidence("worker-events.json", { starts, polls, recovery_action: "poll", upstream_job_reference: "upstream:async-1" }); + evidence("db-diff.json", { lease_owner_after_recovery: "worker-restart", final_status: result.status, upstream_outcome_known: true, upstream_cost_reconciliation: "not_required" }); + evidence("external-calls.json", { poll_calls: 1, start_calls: 1, duplicate_submit_calls: 0 }); + }); + + it("fails an unknown result without a safe reference and releases exactly once for manual reconciliation", async () => { + const adapter: GenerationAdapter = { async start() { throw new Error("provider_timeout_without_reference"); } }; + const fixture = await harness(adapter); + 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(fixture.registration.database.prepare("SELECT upstream_job_reference, upstream_outcome_known, upstream_cost_reconciliation, lease_owner FROM generation_jobs WHERE generation_id = ?").get(fixture.generationId)) + .toEqual({ lease_owner: null, upstream_job_reference: null, upstream_outcome_known: 0, upstream_cost_reconciliation: "pending_manual_review" }); + 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 }); + evidence("worker-events.json", { starts: 1, polls: 0, duplicate_submit_calls: 0, unknown_result: true }); + evidence("db-diff.json", { final_status: "failed", category: "unknown_retryable", upstream_outcome_known: false, upstream_cost_reconciliation: "pending_manual_review", release_count: 1 }); + evidence("external-calls.json", { start_calls: 1, poll_calls: 0, duplicate_submit_calls: 0 }); + }); + + it("blocks new submissions while worker is degraded and reopens the gate when ready", async () => { + const fixture = await harness({ async start() { return completed(); } }); + fixture.processor.setDegraded("lease_recovery_failed"); + const degradedUserId = randomUUID(); + fixture.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(degradedUserId, `${degradedUserId}@example.invalid`, randomUUID(), now); + fixture.registration.database.prepare("INSERT INTO user_profiles (user_id, creator_name, social_id, private_content_notice_version, private_content_notice_acknowledged_at) VALUES (?, 'Degraded User', '@degraded', NULL, NULL)") + .run(degradedUserId); + fixture.registration.database.prepare("INSERT INTO credit_accounts (user_id, available_balance, reserved_balance, updated_at) VALUES (?, 10, 0, ?)") + .run(degradedUserId, now); + await expect(fixture.submissions.submit({ + clientSubmissionId: randomUUID(), confirmedCreditCost: 1, existingReferenceAssetIds: [], + idempotencyKey: `generation-${randomUUID()}-${randomUUID()}`, mode: "new_project", modelConfigVersion: 1, + modelId, newReferences: [], prompt: "blocked while degraded", ratio: "3:4", userId: degradedUserId, + })).rejects.toMatchObject({ code: "generation_blocked", errorCategory: "gateway_contract_invalid" }); + expect(fixture.processor.setReady()).toMatchObject({ status: "ready" }); + expect(fixture.registration.database.prepare("SELECT status FROM worker_runtime_state WHERE singleton = 1").get()).toEqual({ status: "ready" }); + }); +});