feat: implement rich error handling with retry policies and E2E testing framework

backend/src/database/schema/traces.ts

@@ -13,7 +13,7 @@ export const workflowTracesTable = pgTable(
     nodeRef: text('node_ref').notNull(),
     timestamp: timestamp('timestamp', { withTimezone: true }).notNull(),
     message: text('message'),
-    error: text('error'),
+    error: jsonb('error'),
     outputSummary: jsonb('output_summary'),
     level: text('level').notNull().default('info'),
     data: jsonb('data'),

backend/src/dsl/types.ts

@@ -14,6 +14,24 @@ export const WorkflowActionSchema = z.object({
       }),
     )
     .default({}),
+  retryPolicy: z
+    .object({
+      maxAttempts: z.number().int().optional(),
+      initialIntervalSeconds: z.number().optional(),
+      maximumIntervalSeconds: z.number().optional(),
+      backoffCoefficient: z.number().optional(),
+      nonRetryableErrorTypes: z.array(z.string()).optional(),
+      errorTypePolicies: z
+        .record(
+          z.string(),
+          z.object({
+            retryable: z.boolean().optional(),
+            retryDelayMs: z.number().optional(),
+          }),
+        )
+        .optional(),
+    })
+    .optional(),
 });
 
 export type WorkflowAction = z.infer<typeof WorkflowActionSchema>;

backend/src/events/event-ingest.service.ts

@@ -13,7 +13,7 @@ interface KafkaTraceEventPayload {
   timestamp: string;
   level: string;
   message?: string;
-  error?: string;
+  error?: unknown;
   outputSummary?: unknown;
   data?: Record<string, unknown> | null;
   sequence: number;

backend/src/trace/trace.repository.ts

@@ -21,7 +21,7 @@ export interface PersistedTraceEvent {
   sequence: number;
   level: string;
   message?: string;
-  error?: string;
+  error?: unknown;
   outputSummary?: unknown;
   data?: Record<string, unknown> | null;
 }

backend/src/trace/trace.service.ts

@@ -56,7 +56,7 @@ export class TraceService {
     timestamp: Date;
     type: PersistedTraceEventType;
     message: string | null;
-    error: string | null;
+    error: unknown;
     outputSummary: unknown | null;
     level: string;
     data: unknown | null;
@@ -77,7 +77,7 @@ export class TraceService {
       level,
       timestamp: record.timestamp.toISOString(),
       message: record.message ?? undefined,
-      error: record.error ? { message: record.error } : undefined,
+      error: this.toTraceError(record.error),
       outputSummary,
     };
 
@@ -154,6 +154,42 @@ export class TraceService {
     return { payload, metadata };
   }
 
+  private toTraceError(error: unknown): TraceEventPayload['error'] {
+    if (!error) {
+      return undefined;
+    }
+
+    if (typeof error === 'string') {
+      return { message: error };
+    }
+
+    if (typeof error === 'object' && error !== null) {
+      const errObj = error as Record<string, unknown>;
+
+      // Extract fieldErrors if present and valid
+      let fieldErrors: Record<string, string[]> | undefined;
+      if ('fieldErrors' in errObj && errObj.fieldErrors !== null && typeof errObj.fieldErrors === 'object') {
+        const fieldErrorsObj = errObj.fieldErrors as Record<string, unknown>;
+        const isValidFieldErrors = Object.values(fieldErrorsObj).every(
+          (value) => Array.isArray(value) && value.every((item) => typeof item === 'string')
+        );
+        if (isValidFieldErrors) {
+          fieldErrors = fieldErrorsObj as Record<string, string[]>;
+        }
+      }
+
+      return {
+        message: typeof errObj.message === 'string' ? errObj.message : String(error),
+        type: typeof errObj.type === 'string' ? errObj.type : undefined,
+        stack: typeof errObj.stack === 'string' ? errObj.stack : undefined,
+        details: this.toRecord(errObj.details),
+        fieldErrors,
+      };
+    }
+
+    return { message: String(error) };
+  }
+
   private parseMetadata(metadataRaw: unknown): TraceEventMetadata | undefined {
     if (!metadataRaw || typeof metadataRaw !== 'object' || Array.isArray(metadataRaw)) {
       return undefined;

backend/src/workflows/workflows.controller.ts

@@ -129,6 +129,18 @@ const traceErrorSchema = {
     message: { type: 'string' },
     stack: { type: 'string' },
     code: { type: 'string' },
+    type: { type: 'string' },
+    details: {
+      type: 'object',
+      additionalProperties: true,
+    },
+    fieldErrors: {
+      type: 'object',
+      additionalProperties: {
+        type: 'array',
+        items: { type: 'string' },
+      },
+    },
   },
   additionalProperties: false,
 };

backend/src/workflows/workflows.service.ts

@@ -729,7 +729,22 @@ export class WorkflowsService {
     );
 
     const nodeOverrides = options.nodeOverrides ?? {};
-    const definitionWithOverrides = this.applyNodeOverrides(compiledDefinition, nodeOverrides);
+    let definitionWithOverrides = this.applyNodeOverrides(compiledDefinition, nodeOverrides);
+
+    // Inject retry policies from component registry
+    definitionWithOverrides = {
+      ...definitionWithOverrides,
+      actions: definitionWithOverrides.actions.map((action) => {
+        const component = componentRegistry.get(action.componentId);
+        if (component?.retryPolicy) {
+          return {
+            ...action,
+            retryPolicy: component.retryPolicy,
+          };
+        }
+        return action;
+      }),
+    };
     const normalizedKey = this.normalizeIdempotencyKey(options.idempotencyKey);
     const runId = options.runId ?? (normalizedKey ? this.runIdFromIdempotencyKey(normalizedKey) : `shipsec-run-${randomUUID()}`);
     const triggerMetadata = options.trigger ?? this.buildEntryPointTriggerMetadata(auth);

bun.lock

@@ -12,6 +12,7 @@
         "long": "^5.3.2",
       },
       "devDependencies": {
+        "@types/bun": "^1.3.5",
         "@types/node": "^24.10.1",
         "bun-types": "^1.3.4",
         "openapi-typescript": "^7.10.1",
@@ -950,6 +951,8 @@
 
     "@types/body-parser": ["@types/body-parser@1.19.6", "", { "dependencies": { "@types/connect": "*", "@types/node": "*" } }, "sha512-HLFeCYgz89uk22N5Qg3dvGvsv46B8GLvKKo1zKG4NybA8U2DiEO3w9lqGg29t/tfLRJpJ6iQxnVw4OnB7MoM9g=="],
 
+    "@types/bun": ["@types/bun@1.3.5", "", { "dependencies": { "bun-types": "1.3.5" } }, "sha512-RnygCqNrd3srIPEWBd5LFeUYG7plCoH2Yw9WaZGyNmdTEei+gWaHqydbaIRkIkcbXwhBT94q78QljxN0Sk838w=="],
+
     "@types/connect": ["@types/connect@3.4.38", "", { "dependencies": { "@types/node": "*" } }, "sha512-K6uROf1LD88uDQqJCktA4yzL1YYAK6NgfsI0v/mTgyPKWsX1CnJ0XPSDhViejru1GcRkLWb8RlzFYJRqGUbaug=="],
 
     "@types/cookiejar": ["@types/cookiejar@2.1.5", "", {}, "sha512-he+DHOWReW0nghN24E1WUqM0efK4kI9oTqDm6XmK8ZPe2djZ90BSNdGnIyCLzCPw7/pogPlGbzI2wHGGmi4O/Q=="],
@@ -2968,6 +2971,8 @@
 
     "@tokenizer/inflate/fflate": ["fflate@0.8.2", "", {}, "sha512-cPJU47OaAoCbg0pBvzsgpTPhmhqI5eJjh/JIu8tPj5q+T7iLvW/JAYUqmE7KOB4R1ZyEhzBaIQpQpardBF5z8A=="],
 
+    "@types/bun/bun-types": ["bun-types@1.3.5", "", { "dependencies": { "@types/node": "*" } }, "sha512-inmAYe2PFLs0SUbFOWSVD24sg1jFlMPxOjOSSCYqUgn4Hsc3rDc7dFvfVYjFPNHtov6kgUeulV4SxbuIV/stPw=="],
+
     "@types/express/@types/express-serve-static-core": ["@types/express-serve-static-core@5.1.0", "", { "dependencies": { "@types/node": "*", "@types/qs": "*", "@types/range-parser": "*", "@types/send": "*" } }, "sha512-jnHMsrd0Mwa9Cf4IdOzbz543y4XJepXrbia2T4b6+spXC2We3t1y6K44D3mR8XMFSXMCf3/l7rCgddfx7UNVBA=="],
 
     "@typescript-eslint/typescript-estree/minimatch": ["minimatch@9.0.5", "", { "dependencies": { "brace-expansion": "^2.0.1" } }, "sha512-G6T0ZX48xgozx7587koeX9Ys2NYy6Gmv//P89sEte9V9whIapMNF4idKxnW2QtCcLiTWlb/wfCabAtAFWhhBow=="],

bunfig.toml

@@ -3,3 +3,4 @@ members = ["frontend", "backend", "worker", "packages/**"]
 
 [test]
 preload = ["./test/setup.ts"]
+coveragePathIgnorePatterns = ["e2e-tests/**"]

e2e-tests/README.md

@@ -0,0 +1,19 @@
+# E2E Tests
+
+End-to-end tests for workflow execution with real backend, worker, and infrastructure.
+
+## Prerequisites
+
+Local development environment must be running:
+```bash
+docker compose -p shipsec up -d
+pm2 start pm2.config.cjs
+```
+
+## Running Tests
+
+```bash
+bun test:e2e
+```
+
+Tests are skipped if services aren't available. Set `RUN_E2E=true` to enable.

e2e-tests/cleanup.ts

@@ -0,0 +1,100 @@
+/**
+ * Cleanup Script for E2E Tests
+ *
+ * Removes test workflows and runs created during E2E testing.
+ * This keeps the workspace clean and prevents test artifact accumulation.
+ */
+
+const API_BASE = 'http://localhost:3211/api/v1';
+const HEADERS = {
+  'Content-Type': 'application/json',
+  'x-internal-token': 'local-internal-token',
+};
+
+const TEST_WORKFLOW_PREFIX = 'Test:';
+const TEST_CLEANUP_DELAY_MS = 1000; // Delay before cleanup to allow final reads
+
+/**
+ * Fetch all workflows
+ */
+async function getWorkflows() {
+  const res = await fetch(`${API_BASE}/workflows`, { headers: HEADERS });
+  if (!res.ok) {
+    throw new Error(`Failed to fetch workflows: ${res.status}`);
+  }
+  return res.json();
+}
+
+/**
+ * Fetch runs for a specific workflow
+ */
+async function getWorkflowRuns(workflowId: string) {
+  const res = await fetch(`${API_BASE}/workflows/${workflowId}/runs`, { headers: HEADERS });
+  if (!res.ok) {
+    console.warn(`Failed to fetch runs for workflow ${workflowId}: ${res.status}`);
+    return [];
+  }
+  return res.json();
+}
+
+/**
+ * Delete a workflow (and all its runs)
+ */
+async function deleteWorkflow(workflowId: string, workflowName: string) {
+  const res = await fetch(`${API_BASE}/workflows/${workflowId}`, {
+    method: 'DELETE',
+    headers: HEADERS,
+  });
+
+  if (res.ok) {
+    console.log(`  ✓ Deleted: ${workflowName}`);
+  } else if (res.status === 404) {
+    console.warn(`  ⊘ Not found: ${workflowName} (already deleted)`);
+  } else {
+    console.warn(`  ✗ Failed to delete: ${workflowName} (${res.status})`);
+  }
+}
+
+/**
+ * Main cleanup function
+ */
+async function cleanup() {
+  console.log('🧹 E2E Test Cleanup');
+  console.log('');
+
+  try {
+    // Fetch all workflows
+    const workflows = await getWorkflows();
+
+    // Filter test workflows (those starting with "Error Test:")
+    const testWorkflows = workflows.filter((wf: any) =>
+      wf.name?.startsWith(TEST_WORKFLOW_PREFIX)
+    );
+
+    if (testWorkflows.length === 0) {
+      console.log('✨ No test workflows found - workspace is clean');
+      return;
+    }
+
+    console.log(`Found ${testWorkflows.length} test workflow(s):`);
+
+    // Delete each test workflow
+    for (const workflow of testWorkflows) {
+      await deleteWorkflow(workflow.id, workflow.name);
+    }
+
+    console.log('');
+    console.log(`✨ Cleanup complete - deleted ${testWorkflows.length} test workflow(s)`);
+
+  } catch (error) {
+    console.error('');
+    console.error('❌ Cleanup failed:', error);
+    process.exit(1);
+  }
+}
+
+// Run cleanup
+cleanup().catch(error => {
+  console.error('Fatal error during cleanup:', error);
+  process.exit(1);
+});