From bbb43f97a35e5b163699b75f1f840289b1445ce4 Mon Sep 17 00:00:00 2001
From: john <cynell@139.com>
Date: Fri, 24 Jul 2026 23:07:46 +0800
Subject: [PATCH] feat: persist blocked executor jobs

---
 .../memind-orchestrator-boundary.md           |  22 +-
 ops/src/api/admin.ts                          |   5 +
 ops/src/pages/admin/OrchestratorPage.tsx      |   5 +-
 services/orchestrator/README.md               |  24 ++-
 services/orchestrator/admin-config.mjs        |  10 +
 services/orchestrator/admin-config.test.mjs   |  11 +
 services/orchestrator/app.test.mjs            |   8 +-
 .../orchestrator/code-run-shadow-graph.mjs    |  47 ++++-
 services/orchestrator/executor-gateway.mjs    |   3 +-
 services/orchestrator/executor-job-store.mjs  | 191 ++++++++++++++++++
 .../orchestrator/executor-job-store.test.mjs  | 183 +++++++++++++++++
 services/orchestrator/package-lock.json       |   3 +-
 services/orchestrator/package.json            |   3 +-
 services/orchestrator/runtime.mjs             |  14 +-
 services/orchestrator/runtime.test.mjs        |  42 ++++
 services/orchestrator/server.mjs              |  56 ++++-
 services/orchestrator/server.test.mjs         |  48 +++++
 17 files changed, 649 insertions(+), 26 deletions(-)
 create mode 100644 services/orchestrator/executor-job-store.mjs
 create mode 100644 services/orchestrator/executor-job-store.test.mjs
 create mode 100644 services/orchestrator/server.test.mjs

diff --git a/docs/architecture/memind-orchestrator-boundary.md b/docs/architecture/memind-orchestrator-boundary.md
index 339b283..2e4c022 100644
--- a/docs/architecture/memind-orchestrator-boundary.md
+++ b/docs/architecture/memind-orchestrator-boundary.md
@@ -102,6 +102,20 @@ not a deployment backend. The compile-time dispatch capability remains false,
 so creating a job records `blocked` state without importing an executor SDK,
 launching a process, or touching a workspace.
 
+Phase 3.3 persists that state in the Orchestrator-owned PostgreSQL database.
+LangGraph checkpoint tables and `executor_jobs` share the isolated
+`memind_orchestrator` schema and credentials, but remain behind separate
+storage interfaces and readiness probes. This is a deployment simplification,
+not a domain ownership leak: neither store can access Memind business tables,
+and the Executor Job Store can move to a separate database without changing
+the graph or the Memind-facing protocol.
+
+The `build_plan` node creates the deterministic preview job
+`<runId>:executor-preview` with idempotency key
+`<runId>:build_plan:v1`. The stored result is terminal `blocked` state with
+`executor_dispatch_not_implemented`; it is a durable audit and recovery
+boundary, not an execution claim.
+
 The implemented internal API is:
 
 ```text
@@ -123,8 +137,9 @@ validate_run -> build_plan -> finalize_run
 ```
 
 It requires `policy.executionMode=observe-only` and
-`policy.sideEffectsAllowed=false`. It projects the Native executor boundary but
-cannot call Goosed, Aider, OpenHands, Tool Gateway, or the filesystem.
+`policy.sideEffectsAllowed=false`. It projects the Native executor boundary and
+persists a blocked Executor Job, but cannot dispatch Goosed, Aider, OpenHands,
+Tool Gateway, or the filesystem.
 
 ## Failure isolation
 
@@ -139,7 +154,8 @@ failure:
    service URL boundary.
 
 Successful observations are projected as `workflow_shadow_completed`; graph
-checkpoints stay in the Orchestrator-owned PostgreSQL database.
+checkpoints and blocked Executor Jobs stay in the Orchestrator-owned PostgreSQL
+database.
 
 Both terminal projection events contain bounded observation latency. The
 Memind-owned observability service aggregates those events and may join them to
diff --git a/ops/src/api/admin.ts b/ops/src/api/admin.ts
index 4fd8d61..c3bbd7e 100644
--- a/ops/src/api/admin.ts
+++ b/ops/src/api/admin.ts
@@ -268,6 +268,11 @@ export type OrchestratorServiceHealth = {
     service: string | null;
     checkpoint: { kind?: string; durable?: boolean } | null;
     execution: string | null;
+    executorGateway: {
+      dispatchImplemented: boolean;
+      executionEnabled: boolean;
+      store: { kind: string | null; durable: boolean } | null;
+    } | null;
   } | null;
 };
 
diff --git a/ops/src/pages/admin/OrchestratorPage.tsx b/ops/src/pages/admin/OrchestratorPage.tsx
index 01e4c36..35a8f2c 100644
--- a/ops/src/pages/admin/OrchestratorPage.tsx
+++ b/ops/src/pages/admin/OrchestratorPage.tsx
@@ -173,7 +173,7 @@ export function OrchestratorPage() {
         <div>
           <h3 style={{ margin: 0 }}>Executor Gateway</h3>
           <p style={{ margin: '4px 0 0', color: '#68716c', fontSize: 13 }}>
-            Phase 3.2 仅注册版本化适配器契约；所有执行器均为 contract-only，不会启动进程或接管任务。
+            Phase 3.3 已持久化 blocked 预览任务；所有执行器仍为 contract-only，不会启动进程或接管任务。
           </p>
         </div>
         <div style={{ display: 'grid', gridTemplateColumns: 'repeat(auto-fit, minmax(220px, 1fr))', gap: 10 }}>
@@ -264,6 +264,9 @@ export function OrchestratorPage() {
             {serviceHealth?.details?.checkpoint
               ? ` · checkpoint ${serviceHealth.details.checkpoint.kind ?? 'unknown'}`
               : ''}
+            {serviceHealth?.details?.executorGateway?.store
+              ? ` · executor jobs ${serviceHealth.details.executorGateway.store.kind ?? 'unknown'}`
+              : ''}
           </span>
           <button
             type="button"
diff --git a/services/orchestrator/README.md b/services/orchestrator/README.md
index 0002b9f..7d1b4af 100644
--- a/services/orchestrator/README.md
+++ b/services/orchestrator/README.md
@@ -43,7 +43,7 @@ Phase 2 adds:
 - a real LangGraph `StateGraph` service on the versioned HTTP boundary;
 - a PostgreSQL `PostgresSaver` checkpoint owner;
 - an explicitly non-durable MemorySaver option for tests and local debugging;
-- a `code-run-v1` observe-only graph with no executor or tool access;
+- a `code-run-v1` observe-only graph with no executor dispatch or tool access;
 - a Portal fire-and-forget shadow projection;
 - live health probing from memindadm;
 - a separate Colima/Compose deployment artifact.
@@ -110,6 +110,22 @@ Phase 3.2 establishes the Executor Gateway boundary:
 `EXECUTOR_DISPATCH_IMPLEMENTED=false` is a code-level gate. Phase 3.2 can record
 blocked, idempotent job state but cannot invoke an adapter or launch a process.
 
+Phase 3.3 makes that blocked state durable:
+
+- the Orchestrator owns a PostgreSQL `executor_jobs` table beside its LangGraph
+  checkpoint tables, under the same isolated schema and database credentials;
+- the `build_plan` node atomically creates one deterministic Executor Job per
+  run and graph-node version;
+- retries recover the same job through its idempotency key, while a changed
+  semantic request fails closed;
+- `/health` and `/ready` report and probe checkpoint and Executor Job storage
+  independently;
+- memory mode keeps both stores explicitly non-durable.
+
+The persisted job is still a preview terminal state:
+`status=blocked`, `reason=executor_dispatch_not_implemented`, and
+`executionEnabled=false`. Phase 3.3 does not call Goosed, Aider, or OpenHands.
+
 The default memindadm mode remains `off`. Canary and Active routing are not
 wired to the LangGraph executor in Phase 3. Native Agent Run remains the sole
 executor in every mode.
@@ -131,7 +147,7 @@ Supported modes:
 
 ## Local process
 
-PostgreSQL is the default and required checkpoint backend:
+PostgreSQL is the default and required checkpoint and Executor Job backend:
 
 ```bash
 MEMIND_ORCHESTRATOR_DATABASE_URL='postgresql://...' \
@@ -203,8 +219,8 @@ requires the repository release gates and a separately approved deployment.
 | `MEMIND_ORCHESTRATOR_PORT` | HTTP port | `8093` |
 | `MEMIND_ORCHESTRATOR_SERVICE_TOKEN` | Bearer token for `/v1/*` | empty |
 | `MEMIND_ORCHESTRATOR_CHECKPOINT_MODE` | `postgres` or explicit `memory` | `postgres` |
-| `MEMIND_ORCHESTRATOR_DATABASE_URL` | Dedicated checkpoint PostgreSQL URL | required |
-| `MEMIND_ORCHESTRATOR_DATABASE_SCHEMA` | Checkpoint schema | `memind_orchestrator` |
+| `MEMIND_ORCHESTRATOR_DATABASE_URL` | Dedicated Orchestrator PostgreSQL URL | required |
+| `MEMIND_ORCHESTRATOR_DATABASE_SCHEMA` | Checkpoint and Executor Job schema | `memind_orchestrator` |
 
 ## Extraction test
 
diff --git a/services/orchestrator/admin-config.mjs b/services/orchestrator/admin-config.mjs
index aa00113..0622001 100644
--- a/services/orchestrator/admin-config.mjs
+++ b/services/orchestrator/admin-config.mjs
@@ -223,6 +223,16 @@ async function probeServiceHealth(config, {
           durable: Boolean(body.checkpoint.durable),
         } : null,
         execution: body.execution == null ? null : String(body.execution).slice(0, 64),
+        executorGateway: body.executorGateway && typeof body.executorGateway === 'object' ? {
+          dispatchImplemented: body.executorGateway.dispatchImplemented === true,
+          executionEnabled: body.executorGateway.executionEnabled === true,
+          store: body.executorGateway.store && typeof body.executorGateway.store === 'object' ? {
+            kind: body.executorGateway.store.kind == null
+              ? null
+              : String(body.executorGateway.store.kind).slice(0, 64),
+            durable: body.executorGateway.store.durable === true,
+          } : null,
+        } : null,
       } : null,
     };
   } catch (error) {
diff --git a/services/orchestrator/admin-config.test.mjs b/services/orchestrator/admin-config.test.mjs
index 6b37c8d..9c21dc6 100644
--- a/services/orchestrator/admin-config.test.mjs
+++ b/services/orchestrator/admin-config.test.mjs
@@ -198,6 +198,12 @@ test('orchestrator runtime probe reports sanitized service health', async () =>
         service: 'memind-langgraph-orchestrator',
         checkpoint: { kind: 'postgres', durable: true, password: 'must-not-pass-through' },
         execution: 'observe-only',
+        executorGateway: {
+          dispatchImplemented: false,
+          executionEnabled: false,
+          store: { kind: 'postgres', durable: true, password: 'must-not-pass-through' },
+          adapters: [{ id: 'goosed', secret: 'must-not-pass-through' }],
+        },
         ignoredSecret: 'must-not-pass-through',
       }), {
         status: 200,
@@ -214,6 +220,11 @@ test('orchestrator runtime probe reports sanitized service health', async () =>
   assert.equal(requestedUrl, 'http://127.0.0.1:8093/ready');
   assert.equal(runtime.serviceHealth.ok, true);
   assert.equal(runtime.serviceHealth.details.checkpoint.durable, true);
+  assert.deepEqual(runtime.serviceHealth.details.executorGateway, {
+    dispatchImplemented: false,
+    executionEnabled: false,
+    store: { kind: 'postgres', durable: true },
+  });
   assert.equal('password' in runtime.serviceHealth.details.checkpoint, false);
   assert.equal('ignoredSecret' in runtime.serviceHealth.details, false);
 });
diff --git a/services/orchestrator/app.test.mjs b/services/orchestrator/app.test.mjs
index 06f43f8..3f5b320 100644
--- a/services/orchestrator/app.test.mjs
+++ b/services/orchestrator/app.test.mjs
@@ -40,7 +40,13 @@ test('orchestrator HTTP API exposes health and authenticated run endpoints', asy
 
   const health = await fetch(`${server.baseUrl}/health`);
   assert.equal(health.status, 200);
-  assert.equal((await health.json()).execution, 'observe-only');
+  const healthBody = await health.json();
+  assert.equal(healthBody.execution, 'observe-only');
+  assert.equal(healthBody.executorGateway.executionEnabled, false);
+  assert.deepEqual(healthBody.executorGateway.store, {
+    kind: 'memory',
+    durable: false,
+  });
   const ready = await fetch(`${server.baseUrl}/ready`);
   assert.equal(ready.status, 200);
   assert.equal((await ready.json()).ready, true);
diff --git a/services/orchestrator/code-run-shadow-graph.mjs b/services/orchestrator/code-run-shadow-graph.mjs
index c00d21e..8ed2ae7 100644
--- a/services/orchestrator/code-run-shadow-graph.mjs
+++ b/services/orchestrator/code-run-shadow-graph.mjs
@@ -67,12 +67,49 @@ function validateNode(state) {
   };
 }
 
-function planNode(state) {
+async function planNode(state, executorGateway) {
   const taskType = String(state.spec.input?.taskType ?? '').trim() || 'code-change';
   const instruction = String(state.spec.input?.instruction ?? '');
+  const executorJob = await executorGateway.createJob({
+    jobId: `${state.spec.runId}:executor-preview`,
+    idempotencyKey: `${state.spec.runId}:build_plan:v1`,
+    executor: 'goosed',
+    task: {
+      type: taskType,
+      instruction,
+      workspaceRef: state.spec.input?.workspaceRef ?? null,
+      inputRefs: state.spec.input?.inputRefs ?? [],
+    },
+    subject: state.spec.subject,
+    authorization: {
+      executionAllowed: false,
+      actorId: state.spec.subject?.userId ?? null,
+    },
+    policy: {
+      sideEffectsAllowed: false,
+      networkAllowed: false,
+    },
+    controls: {
+      timeoutMs: state.spec.limits?.timeoutMs,
+      cancellationAllowed: true,
+      fallbackExecutor: 'aider',
+    },
+    metadata: {
+      source: 'langgraph-shadow-plan',
+      workflow: state.spec.workflow,
+    },
+  });
   const plan = {
     taskType,
     executorAdapter: 'native-agent-run',
+    executorJob: {
+      kind: 'executor-job',
+      id: executorJob.job.jobId,
+      executor: executorJob.job.executor,
+      status: executorJob.job.status,
+      reason: executorJob.job.reason,
+      durable: executorGateway.status().store.durable,
+    },
     instructionCharacters: instruction.length,
     steps: [
       'accept_control_plane_run',
@@ -86,6 +123,7 @@ function planNode(state) {
     events: [eventFor(state, 'workflow_planned', {
       taskType,
       executorAdapter: plan.executorAdapter,
+      executorJob: plan.executorJob,
       stepCount: plan.steps.length,
     })],
   };
@@ -106,11 +144,14 @@ function finalizeNode(state) {
   };
 }
 
-export function createCodeRunShadowGraph({ checkpointer } = {}) {
+export function createCodeRunShadowGraph({ checkpointer, executorGateway } = {}) {
   if (!checkpointer) throw new Error('Code run shadow graph requires a checkpointer');
+  if (!executorGateway?.createJob) {
+    throw new Error('Code run shadow graph requires an Executor Gateway');
+  }
   return new StateGraph(ShadowState)
     .addNode('validate_run', validateNode)
-    .addNode('build_plan', planNode)
+    .addNode('build_plan', (state) => planNode(state, executorGateway))
     .addNode('finalize_run', finalizeNode)
     .addEdge(START, 'validate_run')
     .addEdge('validate_run', 'build_plan')
diff --git a/services/orchestrator/executor-gateway.mjs b/services/orchestrator/executor-gateway.mjs
index c5b6245..7d50835 100644
--- a/services/orchestrator/executor-gateway.mjs
+++ b/services/orchestrator/executor-gateway.mjs
@@ -230,6 +230,7 @@ export function createInMemoryExecutorJobStore() {
   const byId = new Map();
   const byIdempotencyKey = new Map();
   return {
+    kind: 'memory',
     durable: false,
     async getById(jobId) {
       const value = byId.get(String(jobId ?? '').trim());
@@ -397,7 +398,7 @@ export function createExecutorGateway({
         dispatchImplemented: EXECUTOR_DISPATCH_IMPLEMENTED,
         executionEnabled: false,
         store: {
-          kind: store.durable ? 'durable' : 'memory',
+          kind: String(store.kind ?? (store.durable ? 'durable' : 'memory')),
           durable: store.durable === true,
         },
         adapters: registry.list(),
diff --git a/services/orchestrator/executor-job-store.mjs b/services/orchestrator/executor-job-store.mjs
new file mode 100644
index 0000000..e4cbdef
--- /dev/null
+++ b/services/orchestrator/executor-job-store.mjs
@@ -0,0 +1,191 @@
+import pg from 'pg';
+import { createInMemoryExecutorJobStore } from './executor-gateway.mjs';
+
+const { Pool } = pg;
+const DEFAULT_EXECUTOR_SCHEMA = 'memind_orchestrator';
+const EXECUTOR_JOB_TABLE = 'executor_jobs';
+
+function normalizeMode(value) {
+  const mode = String(value ?? 'postgres').trim().toLowerCase();
+  if (mode === 'memory' || mode === 'postgres') return mode;
+  throw new Error(`Unsupported executor job store mode: ${value}`);
+}
+
+function normalizeSchema(value) {
+  const schema = String(value ?? DEFAULT_EXECUTOR_SCHEMA).trim();
+  if (!/^[a-z_][a-z0-9_]{0,62}$/i.test(schema)) {
+    throw new Error('Invalid Executor Gateway PostgreSQL schema');
+  }
+  return schema;
+}
+
+function parseState(value) {
+  if (value == null) return null;
+  if (typeof value === 'object') return structuredClone(value);
+  try {
+    return JSON.parse(String(value));
+  } catch {
+    return null;
+  }
+}
+
+function projectRow(row) {
+  return parseState(row?.state_json);
+}
+
+export function createPostgresExecutorJobStore({
+  pool,
+  schema = DEFAULT_EXECUTOR_SCHEMA,
+} = {}) {
+  if (!pool?.query) throw new Error('PostgreSQL Executor Job Store requires a pool');
+  const normalizedSchema = normalizeSchema(schema);
+  const table = `"${normalizedSchema}"."${EXECUTOR_JOB_TABLE}"`;
+
+  return {
+    kind: 'postgres',
+    durable: true,
+
+    async ensureSchema() {
+      await pool.query(`CREATE SCHEMA IF NOT EXISTS "${normalizedSchema}"`);
+      await pool.query(`
+        CREATE TABLE IF NOT EXISTS ${table} (
+          job_id VARCHAR(128) PRIMARY KEY,
+          idempotency_key VARCHAR(200) NOT NULL UNIQUE,
+          request_fingerprint CHAR(64) NOT NULL,
+          state_json JSONB NOT NULL,
+          created_at BIGINT NOT NULL,
+          updated_at BIGINT NOT NULL
+        )
+      `);
+      await pool.query(`
+        CREATE INDEX IF NOT EXISTS executor_jobs_updated_at_idx
+          ON ${table} (updated_at DESC)
+      `);
+    },
+
+    async getById(jobId) {
+      const result = await pool.query(
+        `SELECT state_json FROM ${table} WHERE job_id = $1 LIMIT 1`,
+        [String(jobId ?? '').trim()],
+      );
+      return projectRow(result.rows?.[0]);
+    },
+
+    async getByIdempotencyKey(key) {
+      const result = await pool.query(
+        `SELECT state_json FROM ${table} WHERE idempotency_key = $1 LIMIT 1`,
+        [String(key ?? '').trim()],
+      );
+      return projectRow(result.rows?.[0]);
+    },
+
+    async createIfAbsent(record) {
+      const inserted = await pool.query(
+        `INSERT INTO ${table}
+          (job_id, idempotency_key, request_fingerprint, state_json, created_at, updated_at)
+         VALUES ($1, $2, $3, $4::jsonb, $5, $6)
+         ON CONFLICT DO NOTHING
+         RETURNING state_json`,
+        [
+          record.jobId,
+          record.idempotencyKey,
+          record.requestFingerprint,
+          JSON.stringify(record),
+          record.createdAt,
+          record.updatedAt,
+        ],
+      );
+      const created = projectRow(inserted.rows?.[0]);
+      if (created) return { created: true, record: created };
+
+      const existingResult = await pool.query(
+        `SELECT state_json FROM ${table} WHERE idempotency_key = $1 LIMIT 1`,
+        [record.idempotencyKey],
+      );
+      const existing = projectRow(existingResult.rows?.[0]);
+      if (existing) return { created: false, record: existing };
+
+      const error = new Error(`Executor job id already exists: ${record.jobId}`);
+      error.code = 'EXECUTOR_JOB_ID_CONFLICT';
+      error.status = 409;
+      throw error;
+    },
+
+    async update(record) {
+      const result = await pool.query(
+        `UPDATE ${table}
+            SET state_json = $2::jsonb,
+                updated_at = $3
+          WHERE job_id = $1
+          RETURNING state_json`,
+        [record.jobId, JSON.stringify(record), record.updatedAt],
+      );
+      return projectRow(result.rows?.[0]);
+    },
+  };
+}
+
+export async function createExecutorJobPersistence({
+  mode = process.env.MEMIND_ORCHESTRATOR_CHECKPOINT_MODE,
+  connectionString = process.env.MEMIND_ORCHESTRATOR_DATABASE_URL,
+  schema = process.env.MEMIND_ORCHESTRATOR_DATABASE_SCHEMA,
+  poolFactory = (config) => new Pool(config),
+} = {}) {
+  const normalizedMode = normalizeMode(mode);
+  if (normalizedMode === 'memory') {
+    const store = createInMemoryExecutorJobStore();
+    return {
+      kind: 'memory',
+      durable: false,
+      store,
+      async probe() {
+        return true;
+      },
+      async close() {},
+    };
+  }
+
+  const normalizedConnectionString = String(connectionString ?? '').trim();
+  if (!normalizedConnectionString) {
+    const error = new Error(
+      'MEMIND_ORCHESTRATOR_DATABASE_URL is required for PostgreSQL Executor Job Store',
+    );
+    error.code = 'ORCHESTRATOR_DATABASE_URL_REQUIRED';
+    throw error;
+  }
+
+  const normalizedSchema = normalizeSchema(schema);
+  const pool = poolFactory({ connectionString: normalizedConnectionString });
+  const store = createPostgresExecutorJobStore({ pool, schema: normalizedSchema });
+  try {
+    await store.ensureSchema();
+  } catch (error) {
+    try {
+      await pool.end?.();
+    } catch {
+      // Preserve the schema initialization failure.
+    }
+    throw error;
+  }
+  return {
+    kind: 'postgres',
+    durable: true,
+    store,
+    async probe() {
+      await pool.query('SELECT 1 AS ok');
+      return true;
+    },
+    async close() {
+      await pool.end?.();
+    },
+  };
+}
+
+export const executorJobStoreInternals = {
+  DEFAULT_EXECUTOR_SCHEMA,
+  EXECUTOR_JOB_TABLE,
+  normalizeMode,
+  normalizeSchema,
+  parseState,
+  projectRow,
+};
diff --git a/services/orchestrator/executor-job-store.test.mjs b/services/orchestrator/executor-job-store.test.mjs
new file mode 100644
index 0000000..0bf0bd1
--- /dev/null
+++ b/services/orchestrator/executor-job-store.test.mjs
@@ -0,0 +1,183 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import {
+  createExecutorJobPersistence,
+  createPostgresExecutorJobStore,
+  executorJobStoreInternals,
+} from './executor-job-store.mjs';
+
+function record(overrides = {}) {
+  return {
+    version: 'executor-job-state-v1',
+    jobId: 'job-1',
+    idempotencyKey: 'idem-1',
+    requestFingerprint: 'a'.repeat(64),
+    status: 'blocked',
+    createdAt: 1000,
+    updatedAt: 1000,
+    ...overrides,
+  };
+}
+
+test('PostgreSQL Executor Job Store creates an isolated schema and job table', async () => {
+  const queries = [];
+  const store = createPostgresExecutorJobStore({
+    schema: 'memind_orchestrator_test',
+    pool: {
+      async query(sql, params = []) {
+        queries.push({ sql, params });
+        return { rows: [] };
+      },
+    },
+  });
+  await store.ensureSchema();
+  assert.equal(store.kind, 'postgres');
+  assert.equal(store.durable, true);
+  assert.match(queries[0].sql, /CREATE SCHEMA IF NOT EXISTS "memind_orchestrator_test"/);
+  assert.match(queries[1].sql, /"memind_orchestrator_test"\."executor_jobs"/);
+  assert.match(queries[1].sql, /idempotency_key VARCHAR\(200\) NOT NULL UNIQUE/);
+  assert.match(queries[2].sql, /executor_jobs_updated_at_idx/);
+});
+
+test('PostgreSQL Executor Job Store atomically creates and reloads idempotent state', async () => {
+  const stored = record();
+  const calls = [];
+  const store = createPostgresExecutorJobStore({
+    pool: {
+      async query(sql, params = []) {
+        calls.push({ sql, params });
+        if (sql.includes('INSERT INTO')) {
+          return { rows: [{ state_json: stored }] };
+        }
+        if (sql.includes('UPDATE')) {
+          return { rows: [{ state_json: { ...stored, status: 'cancelled' } }] };
+        }
+        if (sql.includes('WHERE job_id =')) {
+          return { rows: [{ state_json: JSON.stringify(stored) }] };
+        }
+        if (sql.includes('WHERE idempotency_key =')) {
+          return { rows: [{ state_json: stored }] };
+        }
+        return { rows: [] };
+      },
+    },
+  });
+
+  const created = await store.createIfAbsent(stored);
+  assert.equal(created.created, true);
+  assert.deepEqual(created.record, stored);
+  assert.deepEqual(await store.getById('job-1'), stored);
+  assert.deepEqual(await store.getByIdempotencyKey('idem-1'), stored);
+  assert.equal((await store.update({ ...stored, status: 'cancelled' })).status, 'cancelled');
+  assert.equal(JSON.parse(calls[0].params[3]).jobId, 'job-1');
+});
+
+test('PostgreSQL Executor Job Store returns the existing idempotent record after conflict', async () => {
+  const stored = record();
+  const store = createPostgresExecutorJobStore({
+    pool: {
+      async query(sql) {
+        if (sql.includes('INSERT INTO')) return { rows: [] };
+        if (sql.includes('WHERE idempotency_key =')) {
+          return { rows: [{ state_json: stored }] };
+        }
+        return { rows: [] };
+      },
+    },
+  });
+  assert.deepEqual(
+    await store.createIfAbsent(stored),
+    { created: false, record: stored },
+  );
+});
+
+test('Executor Job persistence supports explicit non-durable memory mode', async () => {
+  const persistence = await createExecutorJobPersistence({ mode: 'memory' });
+  assert.equal(persistence.kind, 'memory');
+  assert.equal(persistence.durable, false);
+  assert.equal(persistence.store.durable, false);
+  assert.equal(persistence.store.kind, 'memory');
+  assert.equal(await persistence.probe(), true);
+  await persistence.close();
+});
+
+test('Executor Job persistence initializes and closes its PostgreSQL pool', async () => {
+  const queries = [];
+  let closed = false;
+  const pool = {
+    async query(sql) {
+      queries.push(sql);
+      return { rows: [] };
+    },
+    async end() {
+      closed = true;
+    },
+  };
+  const persistence = await createExecutorJobPersistence({
+    mode: 'postgres',
+    connectionString: 'postgresql://orchestrator:test@postgres/orchestrator',
+    schema: 'memind_orchestrator',
+    poolFactory(config) {
+      assert.equal(
+        config.connectionString,
+        'postgresql://orchestrator:test@postgres/orchestrator',
+      );
+      return pool;
+    },
+  });
+  assert.equal(persistence.kind, 'postgres');
+  assert.equal(persistence.durable, true);
+  assert.equal(persistence.store.kind, 'postgres');
+  assert.equal(await persistence.probe(), true);
+  assert.match(queries.at(-1), /SELECT 1 AS ok/);
+  await persistence.close();
+  assert.equal(closed, true);
+});
+
+test('Executor Job persistence rejects missing PostgreSQL config and unsafe schema names', async () => {
+  let poolCreated = false;
+  await assert.rejects(
+    createExecutorJobPersistence({ mode: 'postgres', connectionString: '' }),
+    (error) => error.code === 'ORCHESTRATOR_DATABASE_URL_REQUIRED',
+  );
+  await assert.rejects(
+    createExecutorJobPersistence({
+      mode: 'postgres',
+      connectionString: 'postgresql://orchestrator:test@postgres/orchestrator',
+      schema: 'public; drop schema public',
+      poolFactory() {
+        poolCreated = true;
+        return null;
+      },
+    }),
+    /Invalid Executor Gateway PostgreSQL schema/,
+  );
+  assert.equal(poolCreated, false);
+  assert.throws(
+    () => executorJobStoreInternals.normalizeSchema('public; drop schema public'),
+    /Invalid Executor Gateway PostgreSQL schema/,
+  );
+});
+
+test('Executor Job persistence closes its pool when schema initialization fails', async () => {
+  let closed = false;
+  const initializationError = new Error('schema unavailable');
+  await assert.rejects(
+    createExecutorJobPersistence({
+      mode: 'postgres',
+      connectionString: 'postgresql://orchestrator:test@postgres/orchestrator',
+      poolFactory() {
+        return {
+          async query() {
+            throw initializationError;
+          },
+          async end() {
+            closed = true;
+          },
+        };
+      },
+    }),
+    initializationError,
+  );
+  assert.equal(closed, true);
+});
diff --git a/services/orchestrator/package-lock.json b/services/orchestrator/package-lock.json
index f4dd4d1..fafb0d9 100644
--- a/services/orchestrator/package-lock.json
+++ b/services/orchestrator/package-lock.json
@@ -11,7 +11,8 @@
         "@langchain/core": "^1.2.3",
         "@langchain/langgraph": "^1.4.8",
         "@langchain/langgraph-checkpoint-postgres": "^1.0.4",
-        "express": "^4.21.2"
+        "express": "^4.21.2",
+        "pg": "^8.12.0"
       },
       "engines": {
         "node": ">=20"
diff --git a/services/orchestrator/package.json b/services/orchestrator/package.json
index 9155308..69a4bfe 100644
--- a/services/orchestrator/package.json
+++ b/services/orchestrator/package.json
@@ -14,6 +14,7 @@
     "@langchain/core": "^1.2.3",
     "@langchain/langgraph": "^1.4.8",
     "@langchain/langgraph-checkpoint-postgres": "^1.0.4",
-    "express": "^4.21.2"
+    "express": "^4.21.2",
+    "pg": "^8.12.0"
   }
 }
diff --git a/services/orchestrator/runtime.mjs b/services/orchestrator/runtime.mjs
index 17f9e9e..b798863 100644
--- a/services/orchestrator/runtime.mjs
+++ b/services/orchestrator/runtime.mjs
@@ -3,6 +3,7 @@ import {
   CODE_RUN_SHADOW_WORKFLOW,
   createCodeRunShadowGraph,
 } from './code-run-shadow-graph.mjs';
+import { createExecutorGateway } from './executor-gateway.mjs';
 
 function graphConfig(runId) {
   return {
@@ -35,14 +36,22 @@ export function createLangGraphOrchestratorRuntime({
   checkpointKind = 'unknown',
   durable = false,
   checkpointProbe = null,
+  executorJobStore = null,
+  executorJobProbe = null,
 } = {}) {
-  const graph = createCodeRunShadowGraph({ checkpointer });
+  const executorGateway = createExecutorGateway({
+    ...(executorJobStore ? { store: executorJobStore } : {}),
+  });
+  const graph = createCodeRunShadowGraph({ checkpointer, executorGateway });
   const probe = typeof checkpointProbe === 'function'
     ? checkpointProbe
     : async () => {
         await checkpointer.getTuple(graphConfig('__orchestrator_readiness__'));
         return true;
       };
+  const probeExecutorJobs = typeof executorJobProbe === 'function'
+    ? executorJobProbe
+    : async () => true;
 
   async function getSnapshot(runId) {
     return graph.getState(graphConfig(runId));
@@ -63,6 +72,7 @@ export function createLangGraphOrchestratorRuntime({
         durable: Boolean(durable),
       },
       execution: 'observe-only',
+      executorGateway: executorGateway.status(),
     };
   }
 
@@ -122,7 +132,7 @@ export function createLangGraphOrchestratorRuntime({
     health,
 
     async ready() {
-      await probe();
+      await Promise.all([probe(), probeExecutorJobs()]);
       return {
         ...health(),
         ready: true,
diff --git a/services/orchestrator/runtime.test.mjs b/services/orchestrator/runtime.test.mjs
index f759051..154ee13 100644
--- a/services/orchestrator/runtime.test.mjs
+++ b/services/orchestrator/runtime.test.mjs
@@ -1,6 +1,7 @@
 import assert from 'node:assert/strict';
 import test from 'node:test';
 import { MemorySaver } from '@langchain/langgraph';
+import { createInMemoryExecutorJobStore } from './executor-gateway.mjs';
 import { createLangGraphOrchestratorRuntime } from './runtime.mjs';
 
 function runSpec(overrides = {}) {
@@ -35,6 +36,14 @@ test('LangGraph runtime checkpoints a deterministic observe-only code workflow',
   assert.equal(state.result.observed, true);
   assert.equal(state.result.executed, false);
   assert.equal(state.plan.executorAdapter, 'native-agent-run');
+  assert.deepEqual(state.plan.executorJob, {
+    kind: 'executor-job',
+    id: 'run-shadow-1:executor-preview',
+    executor: 'goosed',
+    status: 'blocked',
+    reason: 'executor_dispatch_not_implemented',
+    durable: false,
+  });
 
   const restored = await runtime.getState('run-shadow-1');
   assert.deepEqual(restored, state);
@@ -66,6 +75,39 @@ test('LangGraph runtime is idempotent by run id', async () => {
   assert.equal((await runtime.listEvents('run-shadow-1')).events.length, 3);
 });
 
+test('LangGraph runtime persists a durable blocked Executor Job and probes both stores', async () => {
+  const checkpointProbeCalls = [];
+  const executorProbeCalls = [];
+  const executorJobStore = {
+    ...createInMemoryExecutorJobStore(),
+    kind: 'postgres',
+    durable: true,
+  };
+  const runtime = createLangGraphOrchestratorRuntime({
+    checkpointer: new MemorySaver(),
+    checkpointKind: 'postgres',
+    durable: true,
+    checkpointProbe: async () => checkpointProbeCalls.push('checkpoint'),
+    executorJobStore,
+    executorJobProbe: async () => executorProbeCalls.push('executor-jobs'),
+  });
+
+  const state = await runtime.start(runSpec());
+  assert.equal(state.plan.executorJob.durable, true);
+  assert.equal(
+    (await executorJobStore.getById('run-shadow-1:executor-preview')).status,
+    'blocked',
+  );
+
+  const ready = await runtime.ready();
+  assert.equal(ready.ready, true);
+  assert.equal(ready.executorGateway.store.kind, 'postgres');
+  assert.equal(ready.executorGateway.store.durable, true);
+  assert.equal(ready.executorGateway.executionEnabled, false);
+  assert.deepEqual(checkpointProbeCalls, ['checkpoint']);
+  assert.deepEqual(executorProbeCalls, ['executor-jobs']);
+});
+
 test('LangGraph runtime rejects execution-enabled and unsupported workflows', async () => {
   const runtime = createLangGraphOrchestratorRuntime({
     checkpointer: new MemorySaver(),
diff --git a/services/orchestrator/server.mjs b/services/orchestrator/server.mjs
index ac34bdb..d3aab1a 100644
--- a/services/orchestrator/server.mjs
+++ b/services/orchestrator/server.mjs
@@ -1,6 +1,7 @@
 import { pathToFileURL } from 'node:url';
 import { createOrchestratorApp } from './app.mjs';
 import { createOrchestratorCheckpoint } from './checkpoint.mjs';
+import { createExecutorJobPersistence } from './executor-job-store.mjs';
 import { createLangGraphOrchestratorRuntime } from './runtime.mjs';
 
 function positivePort(value, fallback = 8093) {
@@ -18,11 +19,24 @@ export async function startOrchestratorServer({
     connectionString: env.MEMIND_ORCHESTRATOR_DATABASE_URL,
     schema: env.MEMIND_ORCHESTRATOR_DATABASE_SCHEMA,
   });
+  let executorJobs;
+  try {
+    executorJobs = await createExecutorJobPersistence({
+      mode: env.MEMIND_ORCHESTRATOR_CHECKPOINT_MODE,
+      connectionString: env.MEMIND_ORCHESTRATOR_DATABASE_URL,
+      schema: env.MEMIND_ORCHESTRATOR_DATABASE_SCHEMA,
+    });
+  } catch (error) {
+    await Promise.allSettled([checkpoint.close()]);
+    throw error;
+  }
   const runtime = createLangGraphOrchestratorRuntime({
     checkpointer: checkpoint.checkpointer,
     checkpointKind: checkpoint.kind,
     durable: checkpoint.durable,
     checkpointProbe: checkpoint.probe,
+    executorJobStore: executorJobs.store,
+    executorJobProbe: executorJobs.probe,
   });
   const app = createOrchestratorApp({
     runtime,
@@ -30,20 +44,44 @@ export async function startOrchestratorServer({
   });
   const host = String(env.MEMIND_ORCHESTRATOR_HOST ?? '127.0.0.1').trim() || '127.0.0.1';
   const port = positivePort(env.MEMIND_ORCHESTRATOR_PORT, 8093);
-  const server = await new Promise((resolve, reject) => {
-    const listening = app.listen(port, host, () => resolve(listening));
-    listening.once('error', reject);
-  });
+  let server;
+  try {
+    server = await new Promise((resolve, reject) => {
+      const listening = app.listen(port, host, () => resolve(listening));
+      listening.once('error', reject);
+    });
+  } catch (error) {
+    await Promise.allSettled([executorJobs.close(), checkpoint.close()]);
+    throw error;
+  }
   logger.log(`[orchestrator] listening on http://${host}:${port} (${checkpoint.kind})`);
 
   async function close() {
-    await new Promise((resolve, reject) => {
-      server.close((error) => (error ? reject(error) : resolve()));
-    });
-    await checkpoint.close();
+    let serverCloseError = null;
+    try {
+      await new Promise((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()));
+      });
+    } catch (error) {
+      serverCloseError = error;
+    }
+    const results = await Promise.allSettled([
+      executorJobs.close(),
+      checkpoint.close(),
+    ]);
+    if (serverCloseError) throw serverCloseError;
+    const failed = results.find((result) => result.status === 'rejected');
+    if (failed) throw failed.reason;
   }
 
-  return { app, server, runtime, checkpoint, close };
+  return {
+    app,
+    server,
+    runtime,
+    checkpoint,
+    executorJobs,
+    close,
+  };
 }
 
 const isEntrypoint = process.argv[1]
diff --git a/services/orchestrator/server.test.mjs b/services/orchestrator/server.test.mjs
new file mode 100644
index 0000000..93d0fde
--- /dev/null
+++ b/services/orchestrator/server.test.mjs
@@ -0,0 +1,48 @@
+import assert from 'node:assert/strict';
+import net from 'node:net';
+import test from 'node:test';
+import { startOrchestratorServer } from './server.mjs';
+
+async function reservePort() {
+  const server = net.createServer();
+  await new Promise((resolve, reject) => {
+    server.once('error', reject);
+    server.listen(0, '127.0.0.1', resolve);
+  });
+  const { port } = server.address();
+  await new Promise((resolve, reject) => {
+    server.close((error) => (error ? reject(error) : resolve()));
+  });
+  return port;
+}
+
+test('orchestrator server owns checkpoint and Executor Job Store lifecycle', async (t) => {
+  const port = await reservePort();
+  const running = await startOrchestratorServer({
+    env: {
+      MEMIND_ORCHESTRATOR_CHECKPOINT_MODE: 'memory',
+      MEMIND_ORCHESTRATOR_HOST: '127.0.0.1',
+      MEMIND_ORCHESTRATOR_PORT: String(port),
+    },
+    logger: { log() {} },
+  });
+  let closed = false;
+  t.after(async () => {
+    if (!closed) await running.close();
+  });
+
+  assert.equal(running.checkpoint.kind, 'memory');
+  assert.equal(running.executorJobs.kind, 'memory');
+  const response = await fetch(`http://127.0.0.1:${port}/ready`);
+  assert.equal(response.status, 200);
+  const ready = await response.json();
+  assert.equal(ready.ready, true);
+  assert.deepEqual(ready.executorGateway.store, {
+    kind: 'memory',
+    durable: false,
+  });
+
+  await running.close();
+  closed = true;
+  assert.equal(running.server.listening, false);
+});