Merge pull request #191 from ShipSecAI/feature/error-retry-policies

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

Author: betterclever

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

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

bunfig.toml

@@ -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.