fix(worker): preserve outcomes across result reporting failures

.changeset/worker-result-boundaries.md

@@ -0,0 +1,8 @@
+---
+"@pgflow/core": patch
+"@pgflow/edge-worker": patch
+---
+
+Preserve handler outcomes when worker logging or task settlement fails, and classify interruption from the worker shutdown signal instead of the exception name. Successful handlers get one completion attempt even during shutdown; settlement errors reach the worker error boundary without replay or conversion into handler failures.
+
+Preserve falsy task outputs (`0`, `false`, and `""`) and store null or undefined outputs as JSON null instead of SQL NULL.

pkgs/core/schemas/0100_function_add_step.sql

@@ -1,7 +1,7 @@
 create or replace function pgflow.add_step(
   flow_slug text,
   step_slug text,
-  deps_slugs text [] default '{}',
+  deps_slugs text[] default '{}',
   max_attempts int default null,
   base_delay int default null,
   timeout int default null,

pkgs/core/schemas/0100_function_compare_flow_shapes.sql

@@ -4,7 +4,7 @@ create or replace function pgflow._compare_flow_shapes(
   p_local jsonb,
   p_db jsonb
 )
-returns text []
+returns text[]
 language plpgsql
 stable
 set search_path to ''

pkgs/core/schemas/0110_function_set_vt_batch.sql

@@ -12,8 +12,8 @@
 
 create or replace function pgflow.set_vt_batch(
   queue_name TEXT,
-  msg_ids BIGINT [],
-  vt_offsets INTEGER []
+  msg_ids BIGINT[],
+  vt_offsets INTEGER[]
 )
 returns table (
   msg_id BIGINT,

pkgs/core/schemas/0120_function_start_tasks.sql

@@ -25,7 +25,7 @@
 -- message is touched.
 create or replace function pgflow.start_tasks(
   flow_slug text,
-  msg_ids bigint [],
+  msg_ids bigint[],
   worker_id uuid,
   queue_name text,
   step_slug text default null

pkgs/core/src/PgflowSqlClient.ts

@@ -56,12 +56,13 @@ export class PgflowSqlClient<TFlow extends AnyFlow>
   }
 
   async completeTask(stepTask: StepTaskKey, output?: Json): Promise<void> {
+    // postgres.js binds sql.json(null) as SQL NULL; coalesce to JSON null.
     await this.sql`
       SELECT pgflow.complete_task(
         run_id => ${stepTask.run_id}::uuid,
         step_slug => ${stepTask.step_slug}::text,
         task_index => ${stepTask.task_index}::int,
-        output => ${this.sql.json(output || null)}::jsonb
+        output => coalesce(${this.sql.json(output ?? null)}::jsonb, 'null'::jsonb)
       );
     `;
   }

pkgs/edge-worker/src/flow/StepTaskExecutor.ts

@@ -1,17 +1,10 @@
 import type { AnyFlow } from '@pgflow/dsl';
-import type { StepTaskRecord } from '@pgflow/core';
+import type { Json, StepTaskRecord } from '@pgflow/core';
 import type { IPgflowClient } from '@pgflow/core';
 import type { IExecutor } from '../core/types.js';
 import type { Logger, TaskLogContext } from '../platform/types.js';
 import type { StepTaskHandlerContext } from '../core/context.js';
 
-class AbortError extends Error {
-  constructor() {
-    super('Operation aborted');
-    this.name = 'AbortError';
-  }
-}
-
 /**
  * Configuration for worker identity in logging
  */
@@ -81,31 +74,30 @@ export class StepTaskExecutor<TFlow extends AnyFlow, TContext extends StepTaskHa
   }
 
   async execute(): Promise<void> {
+    if (this.signal.aborted) {
+      this.logger.debug(`Aborted execution for step task ${this.msgId}`);
+      return;
+    }
+
     const logContext = this.buildLogContext();
+    // Telemetry before execution is not a handler failure.
+    this.logger.taskStarted(logContext);
+    const startTime = Date.now();
+    let result: Json;
 
     try {
-      if (this.signal.aborted) {
-        throw new AbortError();
-      }
-
-      // Check if already aborted before starting
+      // Logging may have triggered shutdown before the handler starts.
       this.signal.throwIfAborted();
 
       const stepSlug = this.stepTask.step_slug;
 
-      // Log task started at debug level
-      this.logger.taskStarted(logContext);
-
       // Get the step handler from the flow with proper typing
       const stepDef = this.flow.getStepDefinition(stepSlug);
 
       if (!stepDef) {
         throw new Error(`No step definition found for slug=${stepSlug}`);
       }
 
-      // Measure handler execution duration
-      const startTime = Date.now();
-
       // Route input based on step type and dependencies
       let handlerInput: unknown;
       if (stepDef.stepType === 'map') {
@@ -120,37 +112,64 @@ export class StepTaskExecutor<TFlow extends AnyFlow, TContext extends StepTaskHa
       }
 
       // !!! HANDLER EXECUTION !!!
-      const result = await stepDef.handler(handlerInput, this.context);
+      result = await stepDef.handler(handlerInput, this.context);
       // !!! HANDLER EXECUTION !!!
+    } catch (error) {
+      await this.handleExecutionError(error, logContext);
+      return;
+    }
 
-      const durationMs = Date.now() - startTime;
+    // A successful handler still gets one completion attempt during shutdown.
+    // v1 has no claim-replay contract: acknowledgement errors escape to the
+    // worker error boundary without retry (the RPC may already have committed).
+    // In-memory results cannot be recovered after process death.
+    const durationMs = Date.now() - startTime;
+    await this.reportResult(
+      () => this.logger.taskCompleted(logContext, durationMs),
+      () => this.adapter.completeTask(this.stepTask, result)
+    );
+  }
 
-      // Log task completed at verbose level
-      this.logger.taskCompleted(logContext, durationMs);
-      await this.adapter.completeTask(this.stepTask, result);
+  /** Report once even if telemetry throws, then surface both errors if needed. */
+  private async reportResult(log: () => void, report: () => Promise<void>) {
+    const errors: unknown[] = [];
+    try {
+      log();
     } catch (error) {
-      await this.handleExecutionError(error, logContext);
+      errors.push(error);
+    }
+    try {
+      await report();
+    } catch (error) {
+      errors.push(error);
+    }
+    if (errors.length === 1) {
+      throw errors[0];
+    }
+    if (errors.length > 1) {
+      throw new AggregateError(errors, 'Task result logging and reporting failed');
     }
   }
 
   /**
    * Handles the error that occurred during execution.
    *
-   * If the error is an AbortError, it means that the worker was aborted and stopping,
-   * the task will be picked up by another worker later.
-   *
-   * Otherwise, it marks the task as failed.
+   * Only the worker shutdown signal classifies an execution as interrupted.
+   * An unrelated AbortError from a handler is still a handler failure.
    */
   private async handleExecutionError(error: unknown, logContext: TaskLogContext) {
-    if (error instanceof Error && error.name === 'AbortError') {
+    if (this.signal.aborted) {
       this.logger.debug(`Aborted execution for step task ${this.msgId}`);
       // Do not mark as failed - the worker was aborted and stopping,
       // the task will be picked up by another worker later
     } else {
-      // Log task failed at verbose level using structured method
-      const errorObj = error instanceof Error ? error : new Error(String(error));
-      this.logger.taskFailed(logContext, errorObj);
-      await this.adapter.failTask(this.stepTask, error);
+      await this.reportResult(
+        () => {
+          const errorObj = error instanceof Error ? error : new Error(String(error));
+          this.logger.taskFailed(logContext, errorObj);
+        },
+        () => this.adapter.failTask(this.stepTask, error)
+      );
     }
   }
 }

pkgs/edge-worker/tests/integration/pgflowSqlClientOutput.test.ts

@@ -0,0 +1,48 @@
+import { assertEquals } from '@std/assert';
+import { PgflowSqlClient } from '@pgflow/core';
+import type { Json } from '@pgflow/core';
+import { withTransaction } from '../db.ts';
+
+const cases: Array<{ name: string; output: Json | undefined; json: string }> = [
+  { name: 'zero', output: 0, json: '0' },
+  { name: 'false', output: false, json: 'false' },
+  { name: 'empty string', output: '', json: '""' },
+  { name: 'JSON null', output: null, json: 'null' },
+  { name: 'undefined as JSON null', output: undefined, json: 'null' },
+];
+
+for (const { name, output, json } of cases) {
+  Deno.test(
+    `PgflowSqlClient.completeTask stores ${name} in PostgreSQL`,
+    withTransaction(async (sql) => {
+      const flowSlug = 'sql_client_output';
+      await sql`select pgflow.create_flow(${flowSlug})`;
+      await sql`select pgflow.add_step(${flowSlug}, 'work')`;
+      const [worker] = await sql<{ worker_id: string }[]>`
+        insert into pgflow.workers (worker_id, queue_name, function_name)
+        values (gen_random_uuid(), ${flowSlug}, 'output_test')
+        returning worker_id
+      `;
+      const client = new PgflowSqlClient(sql);
+      const run = await client.startFlow(flowSlug, {});
+      const messages = await client.readMessages(flowSlug, 10, 1, 1, 10);
+      assertEquals(messages.length, 1);
+      const tasks = await client.startTasks(
+        flowSlug,
+        messages.map((message) => message.msg_id),
+        worker.worker_id,
+        flowSlug
+      );
+      assertEquals(tasks.length, 1);
+
+      await client.completeTask(tasks[0], output);
+
+      const [stored] = await sql`
+        select status, output::text as json, output is null as sql_null
+        from pgflow.step_tasks
+        where run_id = ${run.run_id}::uuid and step_slug = 'work' and task_index = 0
+      `;
+      assertEquals(stored, { status: 'completed', json, sql_null: false });
+    })
+  );
+}