fix(pipeline): report progress only after article writes succeed

Co-Authored-By: Codex <noreply@openai.com>

Author: snowopsdev

pipeline/src/stages.ts

@@ -162,15 +162,7 @@ export async function runPipeline(
       // and a scheduled run's alerting sees stuck articles.
       try {
         const outcome = await stage.run(article, ctx)
-        processed.add(article.id)
-        finalStatusByArticle.set(article.id, outcome.status)
         const warnings = outcome.warnings ?? []
-        if (warnings.length > 0) {
-          warned += 1
-          for (const warning of warnings) {
-            console.warn(`[${stage.name}] article ${article.id} warning: ${warning}`)
-          }
-        }
         await ctx.payload.update({
           collection: 'articles',
           id: article.id,
@@ -192,6 +184,16 @@ export async function runPipeline(
             },
           },
         })
+        // A stage outcome is only progress once the article write succeeds.
+        // Keep the last persisted status when a later stage fails to save.
+        processed.add(article.id)
+        finalStatusByArticle.set(article.id, outcome.status)
+        if (warnings.length > 0) {
+          warned += 1
+          for (const warning of warnings) {
+            console.warn(`[${stage.name}] article ${article.id} warning: ${warning}`)
+          }
+        }
         console.log(
           `[${stage.name}] article ${article.id} "${article.keyword}" -> ${outcome.status}`,
         )

pipeline/test/runPipeline.test.ts

@@ -85,6 +85,39 @@ const quietly = async <T>(body: () => Promise<T>): Promise<T> => {
 }
 
 describe('runPipeline', () => {
+  it('reports only persisted progress when an article write fails', async () => {
+    const { ctx } = fakeCtx([articleAt(1, 'topic_selected')])
+    ctx.payload.update = async () => {
+      throw new Error('database write failed')
+    }
+    const summary = await quietly(() => runPipeline(ctx, {
+      stages: [stubStage('research', 'topic_selected', 'researched')],
+    }))
+    assert.equal(summary.failed, 1)
+    assert.deepEqual(summary.articleIds, [])
+    assert.deepEqual(summary.finalStatuses, {})
+    assert.equal(summary.failures[0]?.message, 'database write failed')
+  })
+
+  it('keeps the last persisted status when a later stage write fails', async () => {
+    const article = articleAt(1, 'topic_selected')
+    const { ctx } = fakeCtx([article])
+    ctx.payload.update = (async ({ data }: { data: { status: Article['status'] } }) => {
+      if (data.status === 'drafted') throw new Error('database write failed')
+      article.status = data.status
+      return article
+    }) as unknown as typeof ctx.payload.update
+    const summary = await quietly(() => runPipeline(ctx, {
+      stages: [
+        stubStage('research', 'topic_selected', 'researched'),
+        stubStage('generate', 'researched', 'drafted'),
+      ],
+    }))
+    assert.equal(summary.failed, 1)
+    assert.deepEqual(summary.articleIds, [1])
+    assert.deepEqual(summary.finalStatuses, { researched: 1 })
+  })
+
   it('reports no failures for a clean run and advances every article', async () => {
     const { ctx, updates } = fakeCtx([
       articleAt(1, 'topic_selected'),