diff --git a/nuxt.config.ts b/nuxt.config.ts index 0f3ffb8..525502a 100644 --- a/nuxt.config.ts +++ b/nuxt.config.ts @@ -2,7 +2,6 @@ import type { Nitro } from 'nitropack' // https://nuxt.com/docs/api/configuration/nuxt-config export default defineNuxtConfig({ - modules: [ '@nuxt/content', '@nuxt/eslint', @@ -52,9 +51,7 @@ export default defineNuxtConfig({ }, }, - css: [ - '@/assets/css/main.css', - ], + css: ['@/assets/css/main.css'], ui: { colorMode: false, @@ -96,6 +93,10 @@ export default defineNuxtConfig({ tasks: true, }, scheduledTasks: { + // Re-submit nodes stuck in `pending` (review workflow never completed) to + // the moderation Workflow — safety net for swallowed/failed triggers. + // Hourly; only nodes pending > 30 min are considered stuck. + '0 * * * *': ['review:reconcile-pending'], // Recompute all trust scores daily at 3am UTC '0 3 * * *': ['compute:trust-scores'], // Reap expired OAuth codes/tokens + stale rate-limit windows at 3:15am UTC @@ -140,43 +141,47 @@ export default defineNuxtConfig({ d1_databases: [ { binding: 'DB', - database_name: process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' - ? 'communityfix_nuxt_content_staging' - : 'communityfix_nuxt_content', - database_id: process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' - ? 'bfd9a625-c402-4713-b887-7b545e5688ae' // communityfix_nuxt_content_staging - : '9c5ab51d-ec5e-4aed-9883-85d79bea6b85', // communityfix_nuxt_content (prod) + database_name: + process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' + ? 'communityfix_nuxt_content_staging' + : 'communityfix_nuxt_content', + database_id: + process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' + ? 'bfd9a625-c402-4713-b887-7b545e5688ae' // communityfix_nuxt_content_staging + : '9c5ab51d-ec5e-4aed-9883-85d79bea6b85', // communityfix_nuxt_content (prod) }, ], hyperdrive: [ { binding: 'HYPERDRIVE', - id: process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' - ? 'd9a5e6878b9342b4ac96844524a59d05' // communityfix-staging → Neon staging branch - : '9cfe49d0e805462086e8365bd604a062', // communityfix-prod → Neon production branch + id: + process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' + ? 'd9a5e6878b9342b4ac96844524a59d05' // communityfix-staging → Neon staging branch + : '9cfe49d0e805462086e8365bd604a062', // communityfix-prod → Neon production branch }, ], kv_namespaces: [ { binding: 'KV_AUTH', - id: process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' - ? 'b0d3e44f92ae4be0a0c57903404f62fb' // communityfix-auth-staging - : '10ae9211092f42d4a90cd17a938c360a', // communityfix-auth (prod) + id: + process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' + ? 'b0d3e44f92ae4be0a0c57903404f62fb' // communityfix-auth-staging + : '10ae9211092f42d4a90cd17a938c360a', // communityfix-auth (prod) }, ], - send_email: [ - { name: 'EMAIL' }, - ], + send_email: [{ name: 'EMAIL' }], workflows: [ { binding: 'MODERATION_WORKFLOW', class_name: 'ModerationWorkflow', - name: process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' - ? 'moderation-staging' - : 'moderation', - script_name: process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' - ? 'communityfix-moderation-staging' - : 'communityfix-moderation', + name: + process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' + ? 'moderation-staging' + : 'moderation', + script_name: + process.env.WORKERS_CI_BRANCH && process.env.WORKERS_CI_BRANCH !== 'master' + ? 'communityfix-moderation-staging' + : 'communityfix-moderation', }, ], }, diff --git a/server/tasks/review/reconcile-pending.ts b/server/tasks/review/reconcile-pending.ts new file mode 100644 index 0000000..45283fc --- /dev/null +++ b/server/tasks/review/reconcile-pending.ts @@ -0,0 +1,86 @@ +import { sql } from 'drizzle-orm' +import { triggerModeration, type ModerationKind } from '../../utils/moderation-trigger' + +// Safety net for moderation that never ran. A node flips to `pending` and the +// write path fires `triggerModeration`, but that call is best-effort: if the +// Workflow instance is never created (transient Cloudflare error, missing +// binding — both swallowed in moderation-trigger.ts) or the run dies before it +// records a verdict, the node is stranded `pending` with nothing to recover it. +// The Cloudflare Workflow gives durability *once a run exists*; this task is what +// guarantees a run eventually exists. It re-submits stuck nodes to the same +// durable Workflow, so a failed trigger heals itself within a cron cycle instead +// of waiting for an admin to click "Re-moderate". + +// Nodes still `pending` this long after their last edit are treated as stuck: +// any in-flight review (Workflow steps retry for a few minutes) has had time to +// land a verdict. Below this we leave them alone to avoid double-firing a review +// that is simply still running. The task itself runs hourly. +const STUCK_GRACE_MINUTES = 30 + +// Cap how many Workflow instances one run may start, so a backlog (or a bug that +// strands many nodes at once) can't blow past Cloudflare's instance-creation +// limits in a single tick — the next run picks up the remainder. +const BATCH_LIMIT = 100 + +// A pending node is "stuck" when NO terminal moderation outcome exists at or +// after its last edit. The decision-action guard keeps nodes that are +// *deliberately* pending out of the set: a node flagged uncertain or asked for +// more info carries a `flag_uncertain` / `request_info` log dated after +// `updated_at`, so it is left for the human. `updated_at` moves only on a real +// edit (updateIssue/updateCaseStudy set it; the Workflow's own approve/curate/ +// relocate writes do not), so a node edited after a prior verdict correctly +// re-qualifies as stuck if its re-trigger failed. +async function findStuckPending(): Promise> { + const db = useDB() + return (await db.execute(sql` + SELECT 'issue'::text AS kind, i.id AS id, i.updated_at AS updated_at + FROM issues i + WHERE i.status = 'pending' + AND i.updated_at < NOW() - (${STUCK_GRACE_MINUTES}::int * INTERVAL '1 minute') + AND NOT EXISTS ( + SELECT 1 FROM audit_logs a + WHERE a.issue_id = i.id + AND a.action IN ('approve', 'reject', 'flag_spam', 'flag_duplicate', 'flag_uncertain', 'request_info') + AND a.created_at >= i.updated_at + ) + UNION ALL + SELECT 'case-study'::text AS kind, cs.id AS id, cs.updated_at AS updated_at + FROM case_studies cs + WHERE cs.status = 'pending' + AND cs.updated_at < NOW() - (${STUCK_GRACE_MINUTES}::int * INTERVAL '1 minute') + AND NOT EXISTS ( + SELECT 1 FROM audit_logs a + WHERE a.details->>'caseStudyId' = cs.id::text + AND a.action IN ('approve', 'reject', 'flag_spam', 'flag_duplicate', 'flag_uncertain', 'request_info') + AND a.created_at >= cs.updated_at + ) + ORDER BY updated_at ASC + LIMIT ${BATCH_LIMIT} + `)) as unknown as Array<{ kind: ModerationKind; id: number }> +} + +export default defineTask({ + meta: { + name: 'review:reconcile-pending', + description: + 'Re-submit nodes stuck in pending (review never completed) to the moderation Workflow', + }, + async run() { + // Scheduled tasks run without an h3 event — scope one DB client across the + // nested useDB() calls (matches compute:trust-scores / oauth:purge). + return withScopedDB(async () => { + const stuck = await findStuckPending() + // Fire sequentially — cheap `Workflow.create()` calls, and serial submission + // is gentler on the instance-creation rate limit than a burst. + for (const node of stuck) await triggerModeration(node.kind, node.id) + if (stuck.length) { + console.log( + `[review:reconcile-pending] re-triggered ${stuck.length}: ${stuck + .map((n) => `${n.kind}#${n.id}`) + .join(', ')}`, + ) + } + return { result: `Re-triggered ${stuck.length} stuck node(s)`, nodes: stuck } + }) + }, +}) diff --git a/workers/moderation/src/index.ts b/workers/moderation/src/index.ts index 13cfe74..f522162 100644 --- a/workers/moderation/src/index.ts +++ b/workers/moderation/src/index.ts @@ -17,6 +17,7 @@ import { prepareCaseStudy, rejectCaseStudy, finalizeCaseStudy, + applyCaseStudyCurate, resolveLocation, applyLocationFix, prepareRevision, @@ -28,6 +29,7 @@ import { type TagResult, type SdgResult, type StructureVerdict, + type CaseStudyPrep, type CaseStudyModeration, type CurationResult, type IssueCurationResult, @@ -218,6 +220,33 @@ export class ModerationWorkflow extends WorkflowEntrypoint finalizeCaseStudy(ctx, prep, moderation)) + + await this.enrichCaseStudy(ctx, step, prep) + } + + private async enrichCaseStudy(ctx: Ctx, step: WorkflowStep, prep: CaseStudyPrep) { + const id = prep.cs.id + + const curateArm = step + .do('curate', STEP, () => + runStep( + ctx.anthropic, + 'case-study.curate', + { parentContext: prep.parentContext, originalJson: prep.originalJson }, + `case-study ${id}`, + ), + ) + .then((curated) => + step.do('apply-curate', STEP, () => applyCaseStudyCurate(ctx, prep, curated)), + ) + .catch((err) => + console.error(`[review-case-study] curation failed for case study ${id}:`, err), + ) + let locationArm: Promise = Promise.resolve() const loc = prep.cs.location as { x: number; y: number } | null if (loc) { @@ -239,20 +268,7 @@ export class ModerationWorkflow extends WorkflowEntrypoint - runStep( - ctx.anthropic, - 'case-study.curate', - { parentContext: prep.parentContext, originalJson: prep.originalJson }, - `case-study ${id}`, - ), - ), - locationArm, - ]) - await step.do('finalize', STEP, () => - finalizeCaseStudy(ctx, prep.cs, prep.solution, moderation, curated), - ) + await Promise.all([curateArm, locationArm]) } } diff --git a/workers/moderation/src/pipelines.ts b/workers/moderation/src/pipelines.ts index d2edacc..11cf18d 100644 --- a/workers/moderation/src/pipelines.ts +++ b/workers/moderation/src/pipelines.ts @@ -857,57 +857,113 @@ function diffStripped(before: CaseStudyRow, after: CurationResult): string[] { return stripped } +// Commit the approval. This must NOT depend on the curate step: a curation +// failure used to abort the whole run *before* this write (curate + finalize ran +// together under one Promise.all), stranding an already-approved case study back +// at `pending`. Finalize now runs first and curation is a separate best-effort +// enrichment pass (see `applyCaseStudyCurate`), mirroring the issue pipeline. export async function finalizeCaseStudy( ctx: Ctx, - cs: CaseStudyRow, - solution: SolutionRef, + prep: CaseStudyPrep, moderation: CaseStudyModeration, - curated: CurationResult, ): Promise { const { db } = ctx + const { cs } = prep const caseStudyId = cs.id - const cleanedText = [ - solution ? `Solution: ${solution.title}` : '', - solution?.summary ?? '', + // Embed from the approved original text. If curation runs it re-embeds from the + // tightened version; if it fails we still have a usable embedding here. + let newEmbedding: number[] | null = null + try { + newEmbedding = await embed(ctx.openai, prep.caseStudyText) + } catch (err) { + console.error( + `[review-case-study] Embedding regeneration failed for case study ${caseStudyId}:`, + err, + ) + } + + await db + .update(caseStudies) + .set({ status: 'approved', ...(newEmbedding ? { embedding: newEmbedding } : {}) }) + .where(eq(caseStudies.id, caseStudyId)) + + await createAuditLog(db, { + type: 'moderation', + action: 'approve', + userId: cs.authorId, + reason: moderation.reason, + details: { + caseStudyId, + solutionId: cs.solutionId, + promptVersion: STEPS['case-study.moderate'].version, + }, + }) + if (cs.authorId) await updateUserTrustScore(ctx, cs.authorId) +} + +// Drop null/undefined entries so optional case-study fields are omitted from the +// jsonb columns rather than persisted as explicit nulls (matches how the schema +// treats absent metric/source/link fields). `Metric`/`Source`/`LinkRow` carry +// only a required key plus optional ones, so stripping nulls === the old explicit +// per-field reconstruction. +function compact>( + obj: T, +): { [K in keyof T]: Exclude } { + return Object.fromEntries(Object.entries(obj).filter(([, v]) => v != null)) as { + [K in keyof T]: Exclude + } +} +function cleanRows>( + rows: T[] | null | undefined, +): Array<{ [K in keyof T]: Exclude }> | null { + return rows ? rows.map(compact) : null +} + +// The text an approved+curated case study is re-embedded from. +function curatedCaseStudyText( + cs: CaseStudyRow, + solution: SolutionRef, + curated: CurationResult, +): string { + return [ + ...(solution ? [`Solution: ${solution.title}`, solution.summary] : []), `Location: ${cs.locationName}`, - curated.implementer ? `Implementer: ${curated.implementer}` : '', + curated.implementer && `Implementer: ${curated.implementer}`, `Outcome: ${cs.outcome}`, - curated.description ?? '', - curated.lessonsLearned?.join('\n') ?? '', + curated.description, + curated.lessonsLearned?.join('\n'), ] .filter(Boolean) .join('\n') .trim() +} + +// Best-effort tightening of an already-approved case study (the issue-side +// `applyIssueCurate` analogue). Runs after `finalizeCaseStudy`, so a failure here +// only forfeits the cleanup — it can never revoke the approval. +export async function applyCaseStudyCurate( + ctx: Ctx, + prep: CaseStudyPrep, + curated: CurationResult, +): Promise { + const { db } = ctx + const { cs, solution } = prep + const caseStudyId = cs.id let newEmbedding: number[] | null = null try { - newEmbedding = await embed(ctx.openai, cleanedText) + newEmbedding = await embed(ctx.openai, curatedCaseStudyText(cs, solution, curated)) } catch (err) { console.error( - `[review-case-study] Embedding regeneration failed for case study ${caseStudyId}:`, + `[review-case-study] Re-embedding after curation failed for case study ${caseStudyId}:`, err, ) } - const cleanedMetrics = - curated.metrics?.map((m) => ({ - label: m.label, - ...(m.baseline != null ? { baseline: m.baseline } : {}), - ...(m.result != null ? { result: m.result } : {}), - ...(m.unit != null ? { unit: m.unit } : {}), - })) ?? null - const cleanedSources = - curated.sources?.map((s) => ({ url: s.url, ...(s.title != null ? { title: s.title } : {}) })) ?? - null - const cleanedLinks = - curated.links?.map((l) => ({ url: l.url, ...(l.title != null ? { title: l.title } : {}) })) ?? - null - await db .update(caseStudies) .set({ - status: 'approved', description: curated.description, lessonsLearned: curated.lessonsLearned, implementer: curated.implementer, @@ -917,26 +973,27 @@ export async function finalizeCaseStudy( scale: curated.scale, startDate: curated.startDate, endDate: curated.endDate, - metrics: cleanedMetrics, - sources: cleanedSources, - links: cleanedLinks, + metrics: cleanRows(curated.metrics), + sources: cleanRows(curated.sources), + links: cleanRows(curated.links), ...(newEmbedding ? { embedding: newEmbedding } : {}), }) .where(eq(caseStudies.id, caseStudyId)) await createAuditLog(db, { type: 'moderation', - action: 'approve', + action: 'curate', userId: cs.authorId, - reason: moderation.reason, + reason: curated.notes, details: { caseStudyId, solutionId: cs.solutionId, - curation: { notes: curated.notes, strippedFields: diffStripped(cs, curated) }, - promptVersion: STEPS['case-study.moderate'].version, + strippedFields: diffStripped(cs, curated), + notes: curated.notes, + promptVersion: STEPS['case-study.curate'].version, }, }) - if (cs.authorId) await updateUserTrustScore(ctx, cs.authorId) + console.log(`[review-case-study] Case study ${caseStudyId} curated`) } // ── Revision pre-screen ────────────────────────────────────────────────────