feat: inline node dispatch — 13 built-in nodes (math, json, string, flow) run in-process

- Add 'runtime' column (http|inline) to platform_function_definitions
- Split doWork() into doWorkInline() and doWorkHttp() paths
- InlineNodeRegistry with 13 implementations: math/add, math/multiply,
  math/subtract, const/number, const/string, const/boolean, json/select,
  json/object, json/merge, json/split, string/template, flow/guard, coerce
- 23 unit tests for inline node implementations
- register-functions.ts includes inline nodes in seed SQL
- FlowsPanel MOCK_FUNCTIONS show inline nodes with Zap badge
- pgpm runtime column with deploy/revert/verify scripts

Author: pyramation

job/compute-worker/__tests__/inline.test.ts

@@ -0,0 +1,175 @@
+/**
+ * Unit tests for the inline node registry and implementations.
+ */
+
+import { executeInline, getInlineImpl, listInlineNodes } from '../src/inline';
+
+describe('inline node registry', () => {
+  test('lists all registered nodes', () => {
+    const nodes = listInlineNodes();
+    expect(nodes.length).toBeGreaterThanOrEqual(13);
+    const names = nodes.map(n => n.name);
+    expect(names).toContain('math/add');
+    expect(names).toContain('math/multiply');
+    expect(names).toContain('math/subtract');
+    expect(names).toContain('json/select');
+    expect(names).toContain('json/merge');
+    expect(names).toContain('json/split');
+    expect(names).toContain('string/template');
+    expect(names).toContain('flow/guard');
+    expect(names).toContain('coerce');
+  });
+
+  test('getInlineImpl returns function for registered nodes', () => {
+    expect(getInlineImpl('math/add')).toBeTruthy();
+    expect(getInlineImpl('unknown/node')).toBeNull();
+  });
+
+  test('throws for unregistered node', async () => {
+    await expect(executeInline('nonexistent', {})).rejects.toThrow(
+      'No inline implementation for "nonexistent"'
+    );
+  });
+});
+
+describe('math nodes', () => {
+  test('add', async () => {
+    const result = await executeInline('math/add', { a: 5, b: 3 });
+    expect(result).toEqual({ sum: 8 });
+  });
+
+  test('add with missing inputs defaults to 0', async () => {
+    const result = await executeInline('math/add', {});
+    expect(result).toEqual({ sum: 0 });
+  });
+
+  test('multiply', async () => {
+    const result = await executeInline('math/multiply', { a: 4, b: 7 });
+    expect(result).toEqual({ product: 28 });
+  });
+
+  test('subtract', async () => {
+    const result = await executeInline('math/subtract', { a: 10, b: 3 });
+    expect(result).toEqual({ difference: 7 });
+  });
+});
+
+describe('const nodes', () => {
+  test('const/number', async () => {
+    const result = await executeInline('const/number', {}, { value: 42 });
+    expect(result).toEqual({ value: 42 });
+  });
+
+  test('const/string', async () => {
+    const result = await executeInline('const/string', {}, { value: 'hello' });
+    expect(result).toEqual({ value: 'hello' });
+  });
+
+  test('const/boolean', async () => {
+    const result = await executeInline('const/boolean', {}, { value: true });
+    expect(result).toEqual({ value: true });
+  });
+});
+
+describe('json nodes', () => {
+  test('json/select extracts nested value', async () => {
+    const result = await executeInline(
+      'json/select',
+      { obj: { user: { name: 'Dan', age: 30 } } },
+      { path: 'user.name' }
+    );
+    expect(result).toEqual({ value: 'Dan' });
+  });
+
+  test('json/select returns undefined for missing path', async () => {
+    const result = await executeInline(
+      'json/select',
+      { obj: { a: 1 } },
+      { path: 'b.c' }
+    );
+    expect(result).toEqual({ value: undefined });
+  });
+
+  test('json/select handles array index', async () => {
+    const result = await executeInline(
+      'json/select',
+      { obj: { items: ['a', 'b', 'c'] } },
+      { path: 'items.1' }
+    );
+    expect(result).toEqual({ value: 'b' });
+  });
+
+  test('json/object builds from inputs', async () => {
+    const result = await executeInline('json/object', { x: 1, y: 'hello' });
+    expect(result).toEqual({ value: { x: 1, y: 'hello' } });
+  });
+
+  test('json/merge combines two objects', async () => {
+    const result = await executeInline('json/merge', {
+      a: { x: 1, y: 2 },
+      b: { y: 3, z: 4 },
+    });
+    expect(result).toEqual({ value: { x: 1, y: 3, z: 4 } });
+  });
+
+  test('json/split separates by keys', async () => {
+    const result = await executeInline(
+      'json/split',
+      { obj: { a: 1, b: 2, c: 3 } },
+      { keys: ['a', 'c'] }
+    );
+    expect(result).toEqual({
+      selected: { a: 1, c: 3 },
+      rest: { b: 2 },
+    });
+  });
+});
+
+describe('string nodes', () => {
+  test('string/template replaces placeholders', async () => {
+    const result = await executeInline(
+      'string/template',
+      { name: 'Dan', role: 'engineer' },
+      { template: 'Hello {{name}}, you are a {{role}}!' }
+    );
+    expect(result).toEqual({ value: 'Hello Dan, you are a engineer!' });
+  });
+
+  test('string/template preserves unresolved placeholders', async () => {
+    const result = await executeInline(
+      'string/template',
+      { name: 'Dan' },
+      { template: 'Hi {{name}}, your id is {{id}}' }
+    );
+    expect(result).toEqual({ value: 'Hi Dan, your id is {{id}}' });
+  });
+});
+
+describe('flow nodes', () => {
+  test('flow/guard passes when ok', async () => {
+    const result = await executeInline('flow/guard', { ok: true });
+    expect(result.pass).toBe(true);
+    expect(result.fail).toBe(false);
+  });
+
+  test('flow/guard fails when not ok', async () => {
+    const result = await executeInline('flow/guard', { ok: false });
+    expect(result.pass).toBe(false);
+    expect(result.fail).toBe(true);
+  });
+
+  test('coerce to number', async () => {
+    const result = await executeInline('coerce', { value: '42' }, { type: 'number' });
+    expect(result).toEqual({ value: 42 });
+  });
+
+  test('coerce to string', async () => {
+    const result = await executeInline('coerce', { value: 42 }, { type: 'string' });
+    expect(result).toEqual({ value: '42' });
+  });
+
+  test('coerce to boolean', async () => {
+    const result = await executeInline('coerce', { value: 1 }, { type: 'boolean' });
+    expect(result).toEqual({ value: true });
+  });
+});

job/compute-worker/src/discovery.ts

@@ -21,7 +21,8 @@ const COLUMNS = `
     id, name, task_identifier, service_url,
     is_invocable, is_built_in, max_attempts,
     priority, queue_name, scope, namespace_id,
-    required_configs, required_secrets, description`;
+    required_configs, required_secrets, description,
+    runtime`;
 
 export class FunctionDiscovery {
   private cache: TtlCache<PlatformFunctionDefinition | null>;

job/compute-worker/src/index.ts

@@ -21,10 +21,11 @@ import type { Pool, PoolClient } from 'pg';
 import { BillingTracker } from './billing';
 import { ComputeLogTracker } from './compute-log';
 import { FunctionDiscovery } from './discovery';
+import { executeInline, getInlineImpl } from './inline';
 import { InvocationTracker } from './invocation';
 import { ComputeModuleLoader } from './module-loader';
 import { compute_request } from './req';
-import type { ComputeJobRow, ComputeWorkerOptions } from './types';
+import type { ComputeJobRow, ComputeWorkerOptions, PlatformFunctionDefinition } from './types';
 import { isGraphNodePayload } from './types';
 
 const DEFAULT_DATABASE_ID = '00000000-0000-0000-0000-000000000000';
@@ -34,6 +35,8 @@ export { TtlCache } from './cache';
 export type { ComputeLogEntry } from './compute-log';
 export { ComputeLogTracker } from './compute-log';
 export { FunctionDiscovery } from './discovery';
+export { executeInline, getInlineImpl, listInlineNodes } from './inline';
+export type { InlineImplFn, InlineNodeDef } from './inline';
 export { InvocationTracker } from './invocation';
 export { ComputeModuleLoader } from './module-loader';
 export type { ComputeRequestOptions, ComputeRequestResult } from './req';
@@ -48,6 +51,7 @@ export type {
   CreateInvocationInput,
   FunctionModuleConfig,
   FunctionRequirement,
+  FunctionRuntime,
   GraphNodePayload,
   InvocationModuleConfig,
   InvocationStatus,
@@ -300,6 +304,118 @@ export default class ComputeWorker {
       throw new Error(`Function "${fn.name}" (${functionName}) is not invocable`);
     }
 
+    // Determine dispatch mode: inline (in-process) vs HTTP
+    const isInline = fn.runtime === 'inline' || getInlineImpl(functionName) !== null;
+
+    if (isInline) {
+      await this.doWorkInline(job, fn, graphNode, payload);
+    } else {
+      await this.doWorkHttp(job, fn, graphNode, payload);
+    }
+  }
+
+  // ─── Inline dispatch (in-process, no HTTP) ──────────────────────────────
+
+  private async doWorkInline(
+    job: ComputeJobRow,
+    fn: PlatformFunctionDefinition,
+    graphNode: boolean,
+    payload: any
+  ): Promise<void> {
+    const { task_identifier } = job;
+    const functionName = fn.task_identifier;
+    const databaseId = job.database_id || this.platformDatabaseId;
+    const scope = job.entity_id ? 'org' : 'platform';
+
+    // Inline nodes use inputs directly; props come from the graph node definition
+    const inputs = graphNode ? (payload.inputs as Record<string, unknown>) : (payload as Record<string, unknown>);
+
+    await this.setJobGUCs(job);
+
+    const { id: invocationId } = await this.tracker.create({
+      task_identifier,
+      payload,
+      job_id: job.id,
+      database_id: databaseId,
+      actor_id: job.actor_id,
+      scope,
+    });
+
+    const reqStart = process.hrtime();
+    try {
+      const result = await executeInline(functionName, inputs, {});
+
+      const elapsed = process.hrtime(reqStart);
+      const ms = Math.round((elapsed[0] * 1e9 + elapsed[1]) / 1e6);
+      await this.tracker.complete(
+        invocationId, ms, undefined,
+        scope, scope === 'org' ? databaseId : undefined
+      );
+
+      await this.computeLog.log({
+        task_identifier,
+        job_id: job.id,
+        invocation_id: invocationId,
+        database_id: databaseId,
+        entity_id: job.entity_id,
+        organization_id: job.organization_id,
+        entity_type: job.entity_type,
+        actor_id: job.actor_id,
+        status: 'completed',
+        duration_ms: ms,
+      });
+
+      if (graphNode) {
+        await this.completeGraphNode(
+          payload.execution_id,
+          payload.node_name,
+          result
+        );
+      }
+    } catch (err: any) {
+      const elapsed = process.hrtime(reqStart);
+      const ms = Math.round((elapsed[0] * 1e9 + elapsed[1]) / 1e6);
+      await this.tracker.fail(
+        invocationId, ms, err.message,
+        scope, scope === 'org' ? databaseId : undefined
+      );
+
+      await this.computeLog.log({
+        task_identifier,
+        job_id: job.id,
+        invocation_id: invocationId,
+        database_id: databaseId,
+        entity_id: job.entity_id,
+        organization_id: job.organization_id,
+        entity_type: job.entity_type,
+        actor_id: job.actor_id,
+        status: 'failed',
+        duration_ms: ms,
+        error: err.message,
+      });
+
+      if (graphNode) {
+        try {
+          await this.failGraphExecution(payload.execution_id, err.message);
+        } catch (graphErr) {
+          log.error('Failed to mark graph execution as failed', graphErr);
+        }
+      }
+      throw err;
+    }
+  }
+
+  // ─── HTTP dispatch (external function service) ──────────────────────────
+
+  private async doWorkHttp(
+    job: ComputeJobRow,
+    fn: PlatformFunctionDefinition,
+    graphNode: boolean,
+    payload: any
+  ): Promise<void> {
+    const { task_identifier } = job;
+    const functionName = fn.task_identifier;
+
     const url = this.resolveUrl(fn.service_url, functionName);
     if (!url) {
       throw new Error(