✨ feat: approval recovery, webhook triggers, fleet tests, agent safety

Workflow approval recovery:
- Add resumeFromApproval() to WorkflowService — resumes paused workflows
  when approval is approved, skips already-completed nodes, handles rejection
- Approve endpoint now fires workflow resumption automatically

Webhook trigger integration:
- Add POST /webhooks/workflow/:path endpoint for webhook-triggered workflows
- HMAC-SHA256 signature validation when webhookSecret is configured
- Webhook body injected as workflow input under __webhook namespace
- Add getByWebhookPath() query to WorkflowsRepository

Agent safety fixes:
- getCommunicationBus() now throws descriptive error instead of null deref
- spawnCounts Map cleaned up when conversations have no active sessions

Fleet Command tests (68 tests):
- Lifecycle, scheduling, task execution, error handling, budget enforcement
- Concurrency guards, dependency cascades, orphan recovery, shared context
- Boot/shutdown, task communication, event emissions, singleton management

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: ersinkoc

packages/gateway/src/db/repositories/workflows.ts

@@ -837,6 +837,28 @@ export class WorkflowsRepository extends BaseRepository {
     );
     return parseInt(row?.count ?? '0', 10);
   }
+
+  /**
+   * Find a workflow by its trigger node's webhookPath (global lookup, cross-user).
+   * Used by the webhook endpoint to match incoming requests to workflows.
+   */
+  async getByWebhookPath(webhookPath: string): Promise<Workflow | null> {
+    // Query all active workflows and find one with a matching trigger node webhookPath.
+    // Uses PostgreSQL JSONB operators for efficient filtering.
+    const row = await this.queryOne<WorkflowRow>(
+      `SELECT w.* FROM workflows w,
+       LATERAL jsonb_array_elements(
+         CASE WHEN jsonb_typeof(w.nodes) = 'array' THEN w.nodes ELSE '[]'::jsonb END
+       ) AS node
+       WHERE w.status = 'active'
+         AND node->>'type' = 'triggerNode'
+         AND node->'data'->>'triggerType' = 'webhook'
+         AND node->'data'->>'webhookPath' = $1
+       LIMIT 1`,
+      [webhookPath]
+    );
+    return row ? mapWorkflow(row) : null;
+  }
 }
 
 // ============================================================================

packages/gateway/src/routes/webhooks.ts

@@ -12,6 +12,10 @@ import { getServiceRegistry, Services } from '@ownpilot/core';
 import { getLog } from '../services/log.js';
 import { safeKeyCompare, apiError, apiResponse, ERROR_CODES, getErrorMessage } from './helpers.js';
 import { TriggersRepository, type WebhookConfig } from '../db/repositories/triggers.js';
+import {
+  WorkflowsRepository,
+  type TriggerNodeData,
+} from '../db/repositories/workflows.js';
 
 const log = getLog('Webhooks');
 
@@ -245,3 +249,94 @@ webhookRoutes.post('/trigger/:triggerId', async (c) => {
 
   return apiResponse(c, { message: 'Webhook received', triggerId });
 });
+
+/**
+ * POST /webhooks/workflow/:path
+ *
+ * Receives external webhook calls that trigger workflows directly.
+ * Matches the incoming path against workflow trigger nodes with
+ * triggerType='webhook' and a matching webhookPath.
+ *
+ * Validates HMAC-SHA256 signature via X-Webhook-Signature header
+ * if webhookSecret is configured on the trigger node.
+ *
+ * The webhook payload is injected as workflow input variables
+ * under the __webhook namespace.
+ */
+webhookRoutes.post('/workflow/:path', async (c) => {
+  const webhookPath = c.req.param('path');
+
+  // Look up active workflow with matching webhookPath in its trigger node
+  const repo = new WorkflowsRepository();
+  const workflow = await repo.getByWebhookPath(`/hooks/${webhookPath}`);
+
+  if (!workflow) {
+    return apiError(
+      c,
+      { code: ERROR_CODES.NOT_FOUND, message: 'No workflow matches this webhook path' },
+      404
+    );
+  }
+
+  // Find the trigger node to check for webhook secret
+  const triggerNode = workflow.nodes.find((n) => n.type === 'triggerNode');
+  const triggerData = triggerNode?.data as TriggerNodeData | undefined;
+
+  // HMAC-SHA256 signature validation if webhookSecret is configured
+  if (triggerData?.webhookSecret) {
+    const signature = c.req.header('x-webhook-signature');
+    if (!signature) {
+      return apiError(
+        c,
+        { code: ERROR_CODES.ACCESS_DENIED, message: 'Missing X-Webhook-Signature header' },
+        403
+      );
+    }
+
+    const rawBody = await c.req.text();
+    const expected = createHmac('sha256', triggerData.webhookSecret).update(rawBody).digest('hex');
+
+    if (!safeKeyCompare(signature, expected)) {
+      return apiError(
+        c,
+        { code: ERROR_CODES.ACCESS_DENIED, message: 'Invalid webhook signature' },
+        403
+      );
+    }
+  }
+
+  // Parse the webhook body
+  let body: Record<string, unknown> = {};
+  try {
+    body = (await c.req.json()) as Record<string, unknown>;
+  } catch {
+    /* empty body is fine */
+  }
+
+  // Execute the workflow with webhook data as input
+  try {
+    const service = getServiceRegistry().get(Services.Workflow);
+    const userId = workflow.userId ?? 'default';
+
+    // Fire-and-forget: execute in background, don't block the webhook response
+    service
+      .executeWorkflow(workflow.id, userId, undefined, {
+        inputs: { __webhook: { path: webhookPath, body, receivedAt: new Date().toISOString() } },
+      })
+      .catch((err: Error) =>
+        log.error('Webhook workflow execution failed', { workflowId: workflow.id, error: err.message })
+      );
+
+    return apiResponse(c, {
+      message: 'Webhook received, workflow triggered',
+      workflowId: workflow.id,
+    });
+  } catch (error) {
+    log.error('Webhook workflow trigger failed', { error: getErrorMessage(error) });
+    return apiError(
+      c,
+      { code: ERROR_CODES.INTERNAL_ERROR, message: 'Failed to trigger workflow' },
+      500
+    );
+  }
+});

packages/gateway/src/routes/workflows.test.ts

@@ -57,8 +57,13 @@ vi.mock('../db/repositories/workflow-approvals.js', () => ({
   createWorkflowApprovalsRepository: () => mockApprovalsRepo,
 }));
 
+const mockWorkflowService = {
+  resumeFromApproval: vi.fn().mockResolvedValue(undefined),
+};
+
 vi.mock('../services/workflow-service.js', () => ({
   topologicalSort: vi.fn(), // default: no throw = valid DAG
+  getWorkflowService: () => mockWorkflowService,
 }));
 
 vi.mock('../services/workflow/dag-utils.js', () => ({

packages/gateway/src/routes/workflows.ts

@@ -21,7 +21,7 @@ import {
 import { ERROR_CODES } from './error-codes.js';
 import { createWorkflowsRepository } from '../db/repositories/workflows.js';
 import { createWorkflowApprovalsRepository } from '../db/repositories/workflow-approvals.js';
-import { topologicalSort } from '../services/workflow-service.js';
+import { topologicalSort, getWorkflowService } from '../services/workflow-service.js';
 import {
   detectCycle,
   type ValidationNode,
@@ -768,6 +768,29 @@ workflowRoutes.post('/approvals/:id/approve', async (c) => {
   }
 
   wsGateway.broadcast('approval:decided', { approvalId: id, status: 'approved' });
+
+  // Resume the workflow execution from the approval point
+  if (approval.workflowId && approval.workflowLogId) {
+    const service = getWorkflowService();
+    service
+      .resumeFromApproval(
+        approval.workflowId,
+        userId,
+        approval.nodeId,
+        'approved',
+        approval.workflowLogId
+      )
+      .catch((err: unknown) => {
+        // Log but don't fail the approval response — workflow resume is best-effort
+        const wfRepo = createWorkflowsRepository(userId);
+        wfRepo.updateLog(approval.workflowLogId, {
+          status: 'failed',
+          error: `Resume after approval failed: ${getErrorMessage(err)}`,
+          completedAt: new Date().toISOString(),
+        }).catch(() => { /* ignore log update failure */ });
+      });
+  }
+
   return apiResponse(c, approval);
 });