feat(orchestrator): harden zero-impact shadow rollout

Gate and bound Portal shadow observations while preserving Native execution. Add fail-closed service boundaries, terminal retention controls, Canary readiness telemetry, ops visibility, and isolated regression coverage.

services/orchestrator/README.md

@@ -197,7 +197,15 @@ MEMIND_ORCHESTRATOR_CHECKPOINT_MODE=memory pnpm dev:orchestrator
 ```
 
 The service binds to `127.0.0.1:8093` by default. Configure the same URL and
-service token in Portal, then enable `shadow` in `/ops/admin/orchestrator`.
+service token in Portal, set
+`MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED=1` on that Portal instance,
+then enable `shadow` in `/ops/admin/orchestrator`. Without the Portal gate,
+Native code runs do not construct or call the observer.
+
+The admin runtime projection distinguishes requested Shadow mode from effective
+Portal wiring. Canary readiness requires `shadow_wiring_enabled=true`; a closed
+environment gate therefore blocks promotion even when older successful samples
+are still present.
 
 memindadm exposes the projection through:
 
@@ -223,8 +231,14 @@ POST /v1/workers/jobs/:jobId/complete
 POST /v1/workers/jobs/:jobId/fail
 POST /v1/workers/recover-expired
 GET /v1/workers/stats
+DELETE /v1/runs/:runId
+POST /v1/maintenance/purge-terminal-runs
 ```
 
+Shadow RunSpecs use the `control-plane-only-v1` policy and do not copy user
+messages, user IDs, or session IDs into Orchestrator storage. Deleting a
+terminal run removes its checkpoint thread and linked terminal Executor Job.
+
 ## Colima deployment
 
 Colima is the recommended first container host on macOS because this service and
@@ -278,6 +292,23 @@ requires the repository release gates and a separately approved deployment.
 | `MEMIND_ORCHESTRATOR_CHECKPOINT_MODE` | `postgres` or explicit `memory` | `postgres` |
 | `MEMIND_ORCHESTRATOR_DATABASE_URL` | Dedicated Orchestrator PostgreSQL URL | required |
 | `MEMIND_ORCHESTRATOR_DATABASE_SCHEMA` | Checkpoint and Executor Job schema | `memind_orchestrator` |
+| `MEMIND_ORCHESTRATOR_RETENTION_DAYS` | Delete terminal workflow data older than this age; `0` disables | `0` |
+| `MEMIND_ORCHESTRATOR_RETENTION_SWEEP_INTERVAL_MS` | Retention sweep interval, minimum one hour | `86400000` |
+| `MEMIND_ORCHESTRATOR_RETENTION_SWEEP_LIMIT` | Maximum terminal candidates per sweep | `100` |
+
+Portal-side variables:
+
+| Variable | Purpose | Default |
+|---|---|---|
+| `MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED` | Construct and schedule the Shadow observer | `false` |
+| `MEMIND_ORCHESTRATOR_SHADOW_MAX_CONCURRENCY` | Maximum simultaneous observations | `2` |
+| `MEMIND_ORCHESTRATOR_SHADOW_MAX_QUEUE` | Maximum waiting observations before skip | `100` |
+| `MEMIND_ORCHESTRATOR_ALLOWED_ORIGINS` | Additional exact non-loopback Orchestrator origins | empty |
+
+An Orchestrator bound beyond loopback refuses to start without the service
+token. Execution refuses to start without both service and worker tokens.
+Retention is disabled by default; the maintenance endpoint is dry-run unless
+the authenticated caller sends `apply=true`.
 
 ## Extraction test
 

services/orchestrator/admin-config.mjs

@@ -12,6 +12,7 @@ import { listPhase5ExecutorAdapters } from './executor-gateway.mjs';
 const CONFIG_TABLE = 'h5_orchestrator_admin_config';
 const CONFIG_SCOPE = 'global';
 const EXECUTION_HANDOFF_IMPLEMENTED = true;
+const CONFIG_SCHEMA_PROMISES = new WeakMap();
 
 function normalizeBoolean(value, fallback = false) {
   if (value == null || value === '') return fallback;
@@ -39,15 +40,45 @@ function normalizeServiceUrl(value) {
   try {
     const url = new URL(raw);
     if (!['http:', 'https:'].includes(url.protocol)) return '';
+    if ((url.pathname && url.pathname !== '/') || url.search) return '';
     url.username = '';
     url.password = '';
     url.hash = '';
-    return url.toString().replace(/\/$/, '').slice(0, 2048);
+    return url.origin.slice(0, 2048);
   } catch {
     return '';
   }
 }
 
+function isLoopbackServiceUrl(value) {
+  try {
+    const hostname = new URL(value).hostname.toLowerCase();
+    return hostname === 'localhost'
+      || hostname === '::1'
+      || hostname === '[::1]'
+      || /^127(?:\.\d{1,3}){3}$/.test(hostname);
+  } catch {
+    return false;
+  }
+}
+
+function allowedServiceOrigins(env = process.env) {
+  const configured = [
+    env.MEMIND_ORCHESTRATOR_URL,
+    ...String(env.MEMIND_ORCHESTRATOR_ALLOWED_ORIGINS ?? '').split(','),
+  ]
+    .map(normalizeServiceUrl)
+    .filter(Boolean);
+  return new Set(configured);
+}
+
+function isServiceUrlAllowed(value, env = process.env) {
+  const normalized = normalizeServiceUrl(value);
+  if (!normalized) return false;
+  return isLoopbackServiceUrl(normalized)
+    || allowedServiceOrigins(env).has(normalized);
+}
+
 function parseJsonLike(value, fallback) {
   if (value == null || value === '') return fallback;
   if (typeof value === 'object') return value;
@@ -112,6 +143,10 @@ export function normalizeOrchestratorConfig(input = {}, fallback = defaultOrches
 
 function runtimeState(config, env = process.env) {
   const killSwitch = normalizeBoolean(env.MEMIND_ORCHESTRATOR_KILL_SWITCH, false);
+  const environmentShadowObservationGate = normalizeBoolean(
+    env.MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED,
+    false,
+  );
   const environmentExecutionGate = normalizeBoolean(
     env.MEMIND_ORCHESTRATOR_EXECUTION_HANDOFF_ENABLED,
     false,
@@ -121,6 +156,10 @@ function runtimeState(config, env = process.env) {
   if (killSwitch) reason = 'kill_switch';
   else if (config.mode === ORCHESTRATOR_MODE.OFF) reason = 'mode_off';
   else if (!configured) reason = 'service_url_missing';
+  else if (
+    config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH
+    && !isServiceUrlAllowed(config.serviceUrl, env)
+  ) reason = 'service_url_not_allowed';
   const executionModeSelected = [ORCHESTRATOR_MODE.CANARY, ORCHESTRATOR_MODE.ACTIVE]
     .includes(config.mode);
   const executionHandoffRequested = config.executionEnabled
@@ -131,6 +170,11 @@ function runtimeState(config, env = process.env) {
     && environmentExecutionGate
     && executionHandoffRequested
     && config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH;
+  const shadowObservationRequested = config.mode === ORCHESTRATOR_MODE.SHADOW
+    && config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH;
+  const shadowObservationEnabled = !reason
+    && environmentShadowObservationGate
+    && shadowObservationRequested;
   return {
     killSwitch,
     configured,
@@ -146,9 +190,19 @@ function runtimeState(config, env = process.env) {
     plansLangGraph: !reason
       && executionModeSelected
       && config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH,
-    shadowsLangGraph: !reason
-      && config.mode === ORCHESTRATOR_MODE.SHADOW
-      && config.primaryEngine === WORKFLOW_ENGINE.LANGGRAPH,
+    shadowsLangGraph: shadowObservationEnabled,
+    shadowObservation: {
+      requested: shadowObservationRequested,
+      enabled: shadowObservationEnabled,
+      reason: shadowObservationEnabled
+        ? null
+        : !shadowObservationRequested
+          ? 'shadow_observation_not_requested'
+          : reason
+            ? reason
+            : 'environment_shadow_observation_gate_disabled',
+      environmentGate: environmentShadowObservationGate,
+    },
     executionHandoff: {
       implemented: EXECUTION_HANDOFF_IMPLEMENTED,
       requested: executionHandoffRequested,
@@ -261,7 +315,9 @@ async function probeServiceHealth(config, {
 }
 
 async function ensureConfigTable(pool) {
-  await pool.query(`
+  const existing = CONFIG_SCHEMA_PROMISES.get(pool);
+  if (existing) return existing;
+  const initializing = pool.query(`
     CREATE TABLE IF NOT EXISTS ${CONFIG_TABLE} (
       config_scope VARCHAR(32) PRIMARY KEY,
       config_json JSON NOT NULL,
@@ -269,7 +325,12 @@ async function ensureConfigTable(pool) {
       updated_by CHAR(36) NULL,
       updated_at BIGINT NOT NULL
     ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
-  `);
+  `).catch((error) => {
+    CONFIG_SCHEMA_PROMISES.delete(pool);
+    throw error;
+  });
+  CONFIG_SCHEMA_PROMISES.set(pool, initializing);
+  return initializing;
 }
 
 async function loadStoredState(pool) {
@@ -450,6 +511,8 @@ export const orchestratorAdminConfigInternals = {
   CONFIG_TABLE,
   engineCatalog,
   EXECUTION_HANDOFF_IMPLEMENTED,
+  isServiceUrlAllowed,
+  normalizeServiceUrl,
   probeServiceHealth,
   runtimeState,
 };

services/orchestrator/admin-config.test.mjs

@@ -8,9 +8,16 @@ import {
 
 function createPool() {
   let row = null;
+  let createTableCalls = 0;
   return {
+    get createTableCalls() {
+      return createTableCalls;
+    },
     async query(sql, params = []) {
-      if (sql.includes('CREATE TABLE')) return [[], []];
+      if (sql.includes('CREATE TABLE')) {
+        createTableCalls += 1;
+        return [[], []];
+      }
       if (sql.includes('SELECT config_json')) return [[...(row ? [row] : [])], []];
       if (sql.includes('INSERT INTO')) {
         row = {
@@ -33,6 +40,12 @@ test('orchestrator config defaults to a disabled native-safe runtime', async ()
   assert.equal(state.config.primaryEngine, 'langgraph');
   assert.equal(state.runtime.effective, false);
   assert.equal(state.runtime.reason, 'mode_off');
+  assert.deepEqual(state.runtime.shadowObservation, {
+    requested: false,
+    enabled: false,
+    reason: 'shadow_observation_not_requested',
+    environmentGate: false,
+  });
   assert.equal(state.engines.find((engine) => engine.id === 'native').configured, true);
   assert.equal(state.engines.find((engine) => engine.id === 'langgraph').configured, false);
   assert.deepEqual(
@@ -59,6 +72,48 @@ test('orchestrator config normalizes unsafe values and keeps a workflow allowlis
   assert.equal(normalized.requestTimeoutMs, 500);
   assert.equal(normalized.rolloutPercent, 100);
   assert.deepEqual(normalized.workflowAllowlist, ['code-run-v1']);
+  assert.equal(
+    normalizeOrchestratorConfig({
+      serviceUrl: 'https://orchestrator.example/internal?token=secret',
+    }).serviceUrl,
+    '',
+  );
+});
+
+test('orchestrator config initializes its schema once per shared pool', async () => {
+  const pool = createPool();
+  const first = createOrchestratorAdminConfigService(pool, { env: {} });
+  const second = createOrchestratorAdminConfigService(pool, { env: {} });
+  await first.ensureSchema();
+  await first.getRuntimeState();
+  await second.getAdminConfig();
+  assert.equal(pool.createTableCalls, 1);
+});
+
+test('orchestrator blocks non-loopback service origins unless explicitly allowed', async () => {
+  const pool = createPool();
+  const blockedService = createOrchestratorAdminConfigService(pool, { env: {} });
+  const blocked = await blockedService.updateAdminConfig({
+    mode: 'shadow',
+    serviceUrl: 'https://orchestrator.example',
+  });
+  assert.equal(blocked.runtime.effective, false);
+  assert.equal(blocked.runtime.reason, 'service_url_not_allowed');
+  assert.deepEqual(blocked.runtime.shadowObservation, {
+    requested: true,
+    enabled: false,
+    reason: 'service_url_not_allowed',
+    environmentGate: false,
+  });
+
+  const allowedService = createOrchestratorAdminConfigService(pool, {
+    env: {
+      MEMIND_ORCHESTRATOR_ALLOWED_ORIGINS: 'https://orchestrator.example',
+    },
+  });
+  const allowed = await allowedService.getRuntimeState();
+  assert.equal(allowed.runtime.effective, true);
+  assert.equal(allowed.runtime.reason, null);
 });
 
 test('orchestrator config persists versioned admin updates and selects canary users', async () => {
@@ -194,6 +249,32 @@ test('Shadow mode evaluates the Canary rollout while Native remains effective',
   assert.equal(native.candidateEngine, 'native');
   assert.equal(native.candidateReason, 'canary_not_selected');
   assert.equal(native.dryRun, true);
+  const runtime = await service.getRuntimeState();
+  assert.equal(runtime.runtime.shadowsLangGraph, false);
+  assert.deepEqual(runtime.runtime.shadowObservation, {
+    requested: true,
+    enabled: false,
+    reason: 'environment_shadow_observation_gate_disabled',
+    environmentGate: false,
+  });
+});
+
+test('Shadow runtime reports the independent Portal observation wiring gate', async () => {
+  const service = createOrchestratorAdminConfigService(createPool(), {
+    env: { MEMIND_ORCHESTRATOR_SHADOW_OBSERVATION_ENABLED: '1' },
+  });
+  await service.updateAdminConfig({
+    mode: 'shadow',
+    serviceUrl: 'http://127.0.0.1:8093',
+  });
+  const runtime = await service.getRuntimeState();
+  assert.equal(runtime.runtime.shadowsLangGraph, true);
+  assert.deepEqual(runtime.runtime.shadowObservation, {
+    requested: true,
+    enabled: true,
+    reason: null,
+    environmentGate: true,
+  });
 });
 
 test('orchestrator emergency kill switch always forces native selection', async () => {

services/orchestrator/app.mjs

@@ -154,6 +154,28 @@ export function createOrchestratorApp({
     }
   });
 
+  app.delete('/v1/runs/:runId', async (request, response, next) => {
+    try {
+      const result = await runtime.deleteRun(request.params.runId);
+      if (!result) return response.status(404).json({ error: { code: 'RUN_NOT_FOUND' } });
+      return response.json(result);
+    } catch (error) {
+      return next(error);
+    }
+  });
+
+  app.post('/v1/maintenance/purge-terminal-runs', async (request, response, next) => {
+    try {
+      return response.json(await runtime.purgeTerminalRuns({
+        before: request.body?.before,
+        limit: request.body?.limit,
+        dryRun: request.body?.apply !== true,
+      }));
+    } catch (error) {
+      return next(error);
+    }
+  });
+
   app.get('/v1/executor-jobs/:jobId', async (request, response, next) => {
     try {
       const job = await runtime.getExecutorJob(request.params.jobId);

services/orchestrator/app.test.mjs

@@ -102,6 +102,39 @@ test('orchestrator HTTP API exposes health and authenticated run endpoints', asy
     ['executor_job_blocked'],
   );
   assert.equal(executorEventBody.nextCursor, 1);
+
+  const purgePreview = await fetch(
+    `${server.baseUrl}/v1/maintenance/purge-terminal-runs`,
+    {
+      method: 'POST',
+      headers: {
+        authorization: 'Bearer test-service-token',
+        'content-type': 'application/json',
+      },
+      body: JSON.stringify({ before: Date.now() + 1000 }),
+    },
+  );
+  assert.equal(purgePreview.status, 200);
+  const purgePreviewBody = await purgePreview.json();
+  assert.equal(purgePreviewBody.dryRun, true);
+  assert.equal(purgePreviewBody.candidates.length, 1);
+
+  const deleted = await fetch(`${server.baseUrl}/v1/runs/http-shadow-1`, {
+    method: 'DELETE',
+    headers: { authorization: 'Bearer test-service-token' },
+  });
+  assert.equal(deleted.status, 200);
+  assert.deepEqual(await deleted.json(), {
+    runId: 'http-shadow-1',
+    deleted: true,
+    executorJobDeleted: true,
+  });
+  assert.equal(
+    (await fetch(`${server.baseUrl}/v1/runs/http-shadow-1`, {
+      headers: { authorization: 'Bearer test-service-token' },
+    })).status,
+    404,
+  );
 });
 
 test('orchestrator HTTP API reports missing runs and observe-only violations', async (t) => {

services/orchestrator/code-run-shadow-graph.mjs

@@ -127,6 +127,7 @@ async function planNode(state, executorGateway) {
     metadata: {
       source: 'langgraph-shadow-plan',
       workflow: state.spec.workflow,
+      workflowRunId: state.spec.runId,
     },
   });
   const plan = {

services/orchestrator/executor-gateway.mjs

@@ -79,6 +79,11 @@ function gatewayError(code, message, status = 409) {
   return error;
 }
 
+function normalizeWorkflowRunId(value) {
+  const runId = String(value ?? '').trim();
+  return /^[a-zA-Z0-9_-]{1,128}$/.test(runId) ? runId : null;
+}
+
 export function buildExecutorJobEvent({
   eventId = crypto.randomUUID(),
   jobId,
@@ -366,6 +371,36 @@ export function createInMemoryExecutorJobStore() {
       const value = jobId ? byId.get(jobId) : null;
       return value ? clone(value) : null;
     },
+    async deleteById(jobId) {
+      const normalizedJobId = String(jobId ?? '').trim();
+      const existing = byId.get(normalizedJobId);
+      if (!existing) return false;
+      byId.delete(normalizedJobId);
+      byIdempotencyKey.delete(existing.idempotencyKey);
+      eventsByJobId.delete(normalizedJobId);
+      return true;
+    },
+    async listTerminalBefore({ before, limit = 100 } = {}) {
+      const cutoff = Number(before);
+      if (!Number.isFinite(cutoff)) return [];
+      return [...byId.values()]
+        .filter((record) => (
+          TERMINAL_JOB_STATUSES.has(record.status)
+          && record.workflowRunId
+          && Number(record.completedAt ?? 0) > 0
+          && Number(record.completedAt) <= cutoff
+        ))
+        .sort((left, right) => (
+          Number(left.completedAt) - Number(right.completedAt)
+          || String(left.jobId).localeCompare(String(right.jobId))
+        ))
+        .slice(0, clampInteger(limit, 100, 1, 500))
+        .map((record) => ({
+          jobId: record.jobId,
+          workflowRunId: record.workflowRunId,
+          completedAt: Number(record.completedAt),
+        }));
+    },
     async createIfAbsent(record, { initialEvent = null } = {}) {
       const existingId = byIdempotencyKey.get(record.idempotencyKey);
       if (existingId) return { created: false, record: clone(byId.get(existingId)) };
@@ -553,6 +588,8 @@ function validateJobStore(store) {
     'getById',
     'getByIdempotencyKey',
     'createIfAbsent',
+    'deleteById',
+    'listTerminalBefore',
     'update',
     'listEvents',
   ]) {
@@ -674,6 +711,7 @@ export function createExecutorGateway({
       fallbackRegistered: decision.fallbackRegistered,
       fallbackAvailable: decision.fallbackAvailable,
       decision,
+      workflowRunId: normalizeWorkflowRunId(request.metadata?.workflowRunId),
       request: dispatchAllowed ? request : null,
       createdAt: now,
       updatedAt: now,
@@ -741,10 +779,28 @@ export function createExecutorGateway({
     return store.update(updated, { event: eventForCancellation(updated) });
   }
 
+  async function deleteJob(jobId) {
+    const current = await store.getById(jobId);
+    if (!current) return false;
+    if (!TERMINAL_JOB_STATUSES.has(current.status)) {
+      throw gatewayError(
+        'EXECUTOR_JOB_NOT_TERMINAL',
+        `Executor job ${current.jobId} must be terminal before deletion`,
+      );
+    }
+    return store.deleteById(current.jobId);
+  }
+
+  async function listRetentionCandidates(options = {}) {
+    return store.listTerminalBefore(options);
+  }
+
   return {
     preview,
     createJob,
     cancel,
+    deleteJob,
+    listRetentionCandidates,
     async getJob(jobId, { includeRequest = false } = {}) {
       const record = await store.getById(jobId);
       return includeRequest ? record : projectExecutorJobForRead(record);
@@ -793,6 +849,7 @@ export const executorGatewayInternals = {
   PHASE3_EXECUTOR_DESCRIPTORS,
   TERMINAL_JOB_STATUSES,
   fingerprint,
+  normalizeWorkflowRunId,
   projectExecutorJobForRead,
   stableValue,
 };

services/orchestrator/executor-gateway.test.mjs

@@ -162,6 +162,53 @@ test('Executor Gateway cancellation is idempotent and never invokes a disabled a
   assert.equal(await gateway.listEvents('missing-job'), null);
 });
 
+test('Executor Gateway deletes only terminal jobs and their private request state', async () => {
+  const gateway = createExecutorGateway();
+  await gateway.createJob(request());
+  assert.equal(await gateway.deleteJob('job-1'), true);
+  assert.equal(await gateway.getJob('job-1'), null);
+  assert.equal(await gateway.listEvents('job-1'), null);
+  assert.equal(await gateway.deleteJob('job-1'), false);
+
+  const activeGateway = createExecutorGateway({
+    registry: createPhase5ExecutorAdapterRegistry({ enabledExecutors: ['aider'] }),
+    executionEnabled: true,
+  });
+  await activeGateway.createJob(request());
+  await assert.rejects(
+    () => activeGateway.deleteJob('job-1'),
+    (error) => error.code === 'EXECUTOR_JOB_NOT_TERMINAL' && error.status === 409,
+  );
+});
+
+test('Executor Gateway exposes only terminal workflow-linked retention candidates', async () => {
+  let now = 1000;
+  const gateway = createExecutorGateway({ nowMs: () => now });
+  await gateway.createJob(request({
+    metadata: { workflowRunId: 'run-retention-1' },
+  }));
+  assert.deepEqual(
+    await gateway.listRetentionCandidates({ before: 999 }),
+    [],
+  );
+  assert.deepEqual(
+    await gateway.listRetentionCandidates({ before: 1000 }),
+    [{
+      jobId: 'job-1',
+      workflowRunId: 'run-retention-1',
+      completedAt: 1000,
+    }],
+  );
+
+  now = 2000;
+  await gateway.createJob(request({
+    jobId: 'job-without-link',
+    idempotencyKey: 'without-link',
+    metadata: {},
+  }));
+  assert.equal((await gateway.listRetentionCandidates({ before: 3000 })).length, 1);
+});
+
 test('Executor Gateway reports implemented capability while execution remains disabled', () => {
   const status = createExecutorGateway().status();
   assert.equal(status.version, 'executor-gateway-status-v1');

services/orchestrator/executor-job-store.mjs

@@ -192,6 +192,41 @@ export function createPostgresExecutorJobStore({
       return projectRow(result.rows?.[0]);
     },
 
+    async deleteById(jobId) {
+      const result = await pool.query(
+        `DELETE FROM ${table}
+          WHERE job_id = $1
+          RETURNING job_id`,
+        [String(jobId ?? '').trim()],
+      );
+      return Boolean(result.rows?.[0]);
+    },
+
+    async listTerminalBefore({ before, limit = 100 } = {}) {
+      const cutoff = Number(before);
+      if (!Number.isFinite(cutoff)) return [];
+      const pageSize = Math.min(500, Math.max(1, Number(limit) || 100));
+      const result = await pool.query(
+        `SELECT
+           state_json->>'jobId' AS job_id,
+           state_json->>'workflowRunId' AS workflow_run_id,
+           (state_json->>'completedAt')::BIGINT AS completed_at
+         FROM ${table}
+         WHERE state_json->>'status' = ANY($1::text[])
+           AND state_json->>'workflowRunId' IS NOT NULL
+           AND COALESCE((state_json->>'completedAt')::BIGINT, 0) > 0
+           AND (state_json->>'completedAt')::BIGINT <= $2
+         ORDER BY (state_json->>'completedAt')::BIGINT ASC, job_id ASC
+         LIMIT $3`,
+        [['succeeded', 'failed', 'cancelled', 'timed_out', 'blocked'], cutoff, pageSize],
+      );
+      return (result.rows ?? []).map((row) => ({
+        jobId: row.job_id,
+        workflowRunId: row.workflow_run_id,
+        completedAt: Number(row.completed_at),
+      }));
+    },
+
     async createIfAbsent(record, { initialEvent = null } = {}) {
       const created = await withTransaction(pool, async (client) => {
         const inserted = await client.query(

services/orchestrator/executor-job-store.test.mjs

@@ -137,6 +137,51 @@ test('PostgreSQL Executor Job Store returns the existing idempotent record after
   );
 });
 
+test('PostgreSQL Executor Job Store deletes a job and cascades its events', async () => {
+  const calls = [];
+  const store = createPostgresExecutorJobStore({
+    pool: {
+      async query(sql, params) {
+        calls.push({ sql, params });
+        return sql.includes('DELETE FROM')
+          ? { rows: [{ job_id: 'job-1' }] }
+          : { rows: [] };
+      },
+    },
+  });
+  assert.equal(await store.deleteById('job-1'), true);
+  assert.match(calls[0].sql, /DELETE FROM .*executor_jobs/);
+  assert.deepEqual(calls[0].params, ['job-1']);
+});
+
+test('PostgreSQL Executor Job Store lists bounded terminal retention candidates', async () => {
+  const calls = [];
+  const store = createPostgresExecutorJobStore({
+    pool: {
+      async query(sql, params) {
+        calls.push({ sql, params });
+        return {
+          rows: [{
+            job_id: 'job-1',
+            workflow_run_id: 'run-1',
+            completed_at: '1000',
+          }],
+        };
+      },
+    },
+  });
+  assert.deepEqual(
+    await store.listTerminalBefore({ before: 2000, limit: 10 }),
+    [{ jobId: 'job-1', workflowRunId: 'run-1', completedAt: 1000 }],
+  );
+  assert.match(calls[0].sql, /workflowRunId/);
+  assert.deepEqual(calls[0].params, [
+    ['succeeded', 'failed', 'cancelled', 'timed_out', 'blocked'],
+    2000,
+    10,
+  ]);
+});
+
 test('PostgreSQL Executor Job Store rolls back job creation when its initial event fails', async () => {
   const calls = [];
   const eventError = new Error('event insert unavailable');

services/orchestrator/observability.mjs

@@ -4,6 +4,7 @@ import { WORKFLOW_ENGINE } from './contracts.mjs';
 const SHADOW_EVENT_TYPES = Object.freeze([
   'workflow_shadow_completed',
   'workflow_shadow_failed',
+  'workflow_shadow_skipped',
 ]);
 const EXECUTION_PLAN_EVENT_TYPE = 'workflow_execution_planned';
 const MAX_METRIC_EVENTS = 5000;
@@ -11,6 +12,7 @@ const CANARY_READINESS_THRESHOLDS = Object.freeze({
   hours: 24 * 7,
   minObservations: 20,
   minSuccessRate: 0.95,
+  maxSkipRate: 0,
   maxP95LatencyMs: 2000,
   minLatencyCoverageRate: 0.95,
   minNativeSettledRate: 0.95,
@@ -56,6 +58,7 @@ function normalizeExecutorJobId(value) {
 function projectShadowEventRow(row) {
   const data = parseJsonColumn(row.data_json, {}) ?? {};
   const succeeded = row.event_type === 'workflow_shadow_completed';
+  const skipped = row.event_type === 'workflow_shadow_skipped';
   const latency = Number(data.latencyMs);
   const taskType = data.taskType ?? null;
   return {
@@ -66,7 +69,7 @@ function projectShadowEventRow(row) {
     sessionId: row.agent_session_id ?? null,
     nativeStatus: row.native_status,
     nativeAttempts: Number(row.native_attempts ?? 0),
-    shadowStatus: succeeded ? 'succeeded' : 'failed',
+    shadowStatus: succeeded ? 'succeeded' : skipped ? 'skipped' : 'failed',
     engine: data.engine ?? 'langgraph',
     configVersion: data.configVersion == null ? null : Number(data.configVersion),
     phase: data.phase ?? null,
@@ -74,10 +77,11 @@ function projectShadowEventRow(row) {
     synthetic: data.synthetic === true || SYNTHETIC_TASK_TYPES.has(String(taskType ?? '')),
     executorAdapter: data.executorAdapter ?? null,
     latencyMs: Number.isFinite(latency) && latency >= 0 ? latency : null,
-    error: succeeded ? null : {
+    error: succeeded || skipped ? null : {
       code: data.code ?? 'WORKFLOW_SHADOW_FAILED',
       message: data.message ?? 'Shadow observation failed',
     },
+    skipReason: skipped ? data.reason ?? 'shadow_queue_full' : null,
     observedAt: Number(row.event_created_at ?? 0),
     nativeCompletedAt: row.native_completed_at == null
       ? null
@@ -146,6 +150,8 @@ function buildCanaryReadiness(loaded, runtimeState, now) {
   const thresholds = CANARY_READINESS_THRESHOLDS;
   const eligibleRuns = loaded.runs.filter((run) => !run.synthetic);
   const successes = eligibleRuns.filter((run) => run.shadowStatus === 'succeeded').length;
+  const failures = eligibleRuns.filter((run) => run.shadowStatus === 'failed').length;
+  const skipped = eligibleRuns.filter((run) => run.shadowStatus === 'skipped').length;
   const latencies = eligibleRuns
     .map((run) => run.latencyMs)
     .filter((value) => Number.isFinite(value));
@@ -159,6 +165,7 @@ function buildCanaryReadiness(loaded, runtimeState, now) {
     ? null
     : Math.max(0, (now - lastObservedAt) / (60 * 60 * 1000));
   const successRate = ratio(successes, eligibleRuns.length);
+  const skipRate = ratio(skipped, eligibleRuns.length);
   const latencyCoverageRate = ratio(latencies.length, eligibleRuns.length);
   const nativeSettledRate = ratio(nativeSettled, eligibleRuns.length);
   const latencyP95Ms = percentile(latencies, 0.95);
@@ -173,6 +180,12 @@ function buildCanaryReadiness(loaded, runtimeState, now) {
       actual: runtimeState?.config?.mode ?? null,
       target: 'shadow',
     },
+    {
+      id: 'shadow_wiring_enabled',
+      passed: runtimeState?.runtime?.shadowObservation?.enabled === true,
+      actual: runtimeState?.runtime?.shadowObservation?.enabled ?? null,
+      target: true,
+    },
     {
       id: 'service_healthy',
       passed: serviceHealth?.ok === true,
@@ -209,6 +222,12 @@ function buildCanaryReadiness(loaded, runtimeState, now) {
       actual: successRate,
       target: thresholds.minSuccessRate,
     },
+    {
+      id: 'shadow_skip_rate',
+      passed: skipRate != null && skipRate <= thresholds.maxSkipRate,
+      actual: skipRate,
+      target: thresholds.maxSkipRate,
+    },
     {
       id: 'latency_coverage',
       passed: latencyCoverageRate != null
@@ -271,8 +290,10 @@ function buildCanaryReadiness(loaded, runtimeState, now) {
       eligibleObservations: eligibleRuns.length,
       excludedSynthetic: loaded.runs.length - eligibleRuns.length,
       successes,
-      failures: eligibleRuns.length - successes,
+      failures,
+      skipped,
       successRate,
+      skipRate,
       latencyCoverageRate,
       latencyP95Ms,
       nativeSettledRate,
@@ -301,7 +322,8 @@ function buildCanaryReadiness(loaded, runtimeState, now) {
 
 function summarizeRuns(runs, { capped = false } = {}) {
   const successes = runs.filter((run) => run.shadowStatus === 'succeeded').length;
-  const failures = runs.length - successes;
+  const failures = runs.filter((run) => run.shadowStatus === 'failed').length;
+  const skipped = runs.filter((run) => run.shadowStatus === 'skipped').length;
   const latencies = runs
     .map((run) => run.latencyMs)
     .filter((value) => Number.isFinite(value));
@@ -309,8 +331,10 @@ function summarizeRuns(runs, { capped = false } = {}) {
     observations: runs.length,
     successes,
     failures,
+    skipped,
     successRate: runs.length ? successes / runs.length : null,
     failureRate: runs.length ? failures / runs.length : null,
+    skipRate: runs.length ? skipped / runs.length : null,
     latencyP50Ms: percentile(latencies, 0.5),
     latencyP95Ms: percentile(latencies, 0.95),
     nativeSucceeded: runs.filter((run) => run.nativeStatus === 'succeeded').length,
@@ -391,7 +415,7 @@ export function createOrchestratorObservabilityService({
          r.completed_at AS native_completed_at
        FROM h5_agent_run_events e
        INNER JOIN h5_agent_runs r ON r.id = e.run_id
-       WHERE e.event_type IN (?, ?)
+       WHERE e.event_type IN (${SHADOW_EVENT_TYPES.map(() => '?').join(', ')})
          AND e.created_at >= ?
        ORDER BY e.created_at DESC, e.id DESC
        LIMIT ?`,
@@ -452,7 +476,9 @@ export function createOrchestratorObservabilityService({
     } = {}) {
       const loaded = await loadShadowEvents({ hours });
       const normalizedLimit = clampInteger(limit, 50, 1, 200);
-      const normalizedStatus = ['succeeded', 'failed'].includes(status) ? status : 'all';
+      const normalizedStatus = ['succeeded', 'failed', 'skipped'].includes(status)
+        ? status
+        : 'all';
       const filtered = normalizedStatus === 'all'
         ? loaded.runs
         : loaded.runs.filter((run) => run.shadowStatus === normalizedStatus);
@@ -510,7 +536,8 @@ export function createOrchestratorObservabilityService({
       const [eventRows] = await pool.query(
         `SELECT id AS event_id, event_type, data_json, created_at
            FROM h5_agent_run_events
-          WHERE run_id = ? AND event_type IN (?, ?)
+          WHERE run_id = ?
+            AND event_type IN (${SHADOW_EVENT_TYPES.map(() => '?').join(', ')})
           ORDER BY created_at ASC, id ASC`,
         [normalizedRunId, ...SHADOW_EVENT_TYPES],
       );

services/orchestrator/observability.test.mjs

@@ -50,6 +50,18 @@ test('observability aggregates shadow outcomes and latency percentiles', async (
       native_attempts: 2,
       native_completed_at: 1500,
     },
+    {
+      event_id: 'event-skipped',
+      run_id: 'run-skipped',
+      event_type: 'workflow_shadow_skipped',
+      data_json: { reason: 'shadow_queue_full' },
+      event_created_at: 900,
+      request_id: 'request-skipped',
+      user_id: 'user-skipped',
+      native_status: 'succeeded',
+      native_attempts: 1,
+      native_completed_at: 1200,
+    },
   ];
   const service = createOrchestratorObservabilityService({
     pool: {
@@ -61,16 +73,20 @@ test('observability aggregates shadow outcomes and latency percentiles', async (
     configService: { getRuntimeState() {} },
   });
 
-  const result = await service.listShadowRuns({ hours: 12, limit: 2 });
+  const result = await service.listShadowRuns({ hours: 12, limit: 4 });
   assert.equal(result.window.hours, 12);
-  assert.equal(result.metrics.observations, 3);
+  assert.equal(result.metrics.observations, 4);
   assert.equal(result.metrics.successes, 2);
   assert.equal(result.metrics.failures, 1);
+  assert.equal(result.metrics.skipped, 1);
+  assert.equal(result.metrics.skipRate, 0.25);
   assert.equal(result.metrics.latencyP50Ms, 200);
   assert.equal(result.metrics.latencyP95Ms, 300);
-  assert.equal(result.metrics.nativeSucceeded, 2);
-  assert.equal(result.runs.length, 2);
+  assert.equal(result.metrics.nativeSucceeded, 3);
+  assert.equal(result.runs.length, 4);
   assert.equal(result.runs[1].error.code, 'TIMEOUT');
+  assert.equal(result.runs[3].shadowStatus, 'skipped');
+  assert.equal(result.runs[3].skipReason, 'shadow_queue_full');
 });
 
 test('observability aggregates complete dry-run routing decisions and Native outcomes', async () => {
@@ -339,6 +355,9 @@ test('canary readiness excludes smoke runs and reports explicit blockers', async
         assert.deepEqual(options, { probe: true });
         return {
           config: { mode: 'shadow' },
+          runtime: {
+            shadowObservation: { enabled: true },
+          },
           serviceHealth: {
             ok: true,
             status: 'healthy',
@@ -370,6 +389,7 @@ test('canary readiness excludes smoke runs and reports explicit blockers', async
 
 test('canary readiness passes only when operational and sample gates all pass', async () => {
   const now = 20_000_000;
+  let shadowObservationEnabled = true;
   const rows = Array.from({ length: 20 }, (_, index) => ({
     event_id: `event-${index}`,
     run_id: `run-${index}`,
@@ -399,6 +419,9 @@ test('canary readiness passes only when operational and sample gates all pass',
       async getRuntimeState() {
         return {
           config: { mode: 'shadow' },
+          runtime: {
+            shadowObservation: { enabled: shadowObservationEnabled },
+          },
           serviceHealth: {
             ok: true,
             status: 'healthy',
@@ -423,10 +446,67 @@ test('canary readiness passes only when operational and sample gates all pass',
   assert.equal(readiness.window.hours, 168);
   assert.equal(readiness.samples.eligibleObservations, 20);
   assert.equal(readiness.samples.successRate, 0.95);
+  assert.equal(readiness.samples.skipped, 0);
+  assert.equal(readiness.samples.skipRate, 0);
   assert.equal(readiness.samples.latencyCoverageRate, 1);
   assert.equal(readiness.samples.nativeSettledRate, 1);
   assert.equal(readiness.samples.distinctSessions, 5);
   assert.equal(readiness.service.executorJobStoreDurable, true);
   assert.equal(readiness.blockers.length, 0);
   assert.deepEqual(readiness.failureCodes, [{ code: 'TRANSIENT', count: 1 }]);
+
+  shadowObservationEnabled = false;
+  const blockedReadiness = await service.getCanaryReadiness();
+  assert.equal(blockedReadiness.ready, false);
+  assert.ok(blockedReadiness.blockers.includes('shadow_wiring_enabled'));
+});
+
+test('canary readiness blocks when any Shadow observation was skipped', async () => {
+  const now = 30_000_000;
+  const rows = Array.from({ length: 20 }, (_, index) => ({
+    event_id: `event-skip-${index}`,
+    run_id: `run-skip-${index}`,
+    event_type: index === 0
+      ? 'workflow_shadow_skipped'
+      : 'workflow_shadow_completed',
+    data_json: index === 0
+      ? { reason: 'shadow_queue_full' }
+      : { latencyMs: 100, taskType: 'code_task' },
+    event_created_at: now - index * 1000,
+    request_id: `request-skip-${index}`,
+    user_id: `user-${index % 3}`,
+    agent_session_id: `session-${index % 5}`,
+    native_status: 'succeeded',
+    native_attempts: 1,
+    native_completed_at: now,
+  }));
+  const service = createOrchestratorObservabilityService({
+    pool: { async query() { return [rows]; } },
+    configService: {
+      async getRuntimeState() {
+        return {
+          config: { mode: 'shadow' },
+          runtime: {
+            shadowObservation: { enabled: true },
+          },
+          serviceHealth: {
+            ok: true,
+            status: 'healthy',
+            details: {
+              checkpoint: { kind: 'postgres', durable: true },
+              executorGateway: { store: { kind: 'postgres', durable: true } },
+              execution: 'observe-only',
+            },
+          },
+        };
+      },
+    },
+    nowMs: () => now,
+  });
+
+  const readiness = await service.getCanaryReadiness();
+  assert.equal(readiness.ready, false);
+  assert.equal(readiness.samples.skipped, 1);
+  assert.equal(readiness.samples.skipRate, 0.05);
+  assert.ok(readiness.blockers.includes('shadow_skip_rate'));
 });

services/orchestrator/runtime.mjs

@@ -11,6 +11,8 @@ import {
 import { createExecutorWorkerCoordinator } from './executor-worker-protocol.mjs';
 import { createExecutorAdmissionPolicy } from './executor-admission-policy.mjs';
 
+const TERMINAL_RUN_STATUSES = new Set(['succeeded', 'failed', 'cancelled']);
+
 function graphConfig(runId) {
   return {
     configurable: {
@@ -134,6 +136,35 @@ export function createLangGraphOrchestratorRuntime({
     );
   }
 
+  async function deleteRun(runId) {
+    const normalizedRunId = String(runId ?? '').trim();
+    if (!normalizedRunId) return null;
+    const state = await getState(normalizedRunId);
+    if (!state) return null;
+    if (!TERMINAL_RUN_STATUSES.has(state.status)) {
+      const error = new Error('Workflow run must be terminal before deletion');
+      error.code = 'WORKFLOW_RUN_NOT_TERMINAL';
+      error.status = 409;
+      throw error;
+    }
+    if (typeof checkpointer.deleteThread !== 'function') {
+      const error = new Error('Checkpoint backend does not support run deletion');
+      error.code = 'WORKFLOW_DELETE_UNSUPPORTED';
+      error.status = 501;
+      throw error;
+    }
+    const executorJobId = state.plan?.executorJob?.id ?? null;
+    const executorJobDeleted = executorJobId
+      ? await executorGateway.deleteJob(executorJobId)
+      : false;
+    await checkpointer.deleteThread(normalizedRunId);
+    return {
+      runId: normalizedRunId,
+      deleted: true,
+      executorJobDeleted,
+    };
+  }
+
   function health() {
     return {
       status: 'ok',
@@ -215,6 +246,77 @@ export function createLangGraphOrchestratorRuntime({
       throw error;
     },
 
+    deleteRun,
+
+    async purgeTerminalRuns({
+      before,
+      limit = 100,
+      dryRun = true,
+    } = {}) {
+      const cutoff = Number(before);
+      if (!Number.isFinite(cutoff) || cutoff <= 0) {
+        const error = new Error('Retention purge requires a positive before timestamp');
+        error.code = 'WORKFLOW_PURGE_CUTOFF_REQUIRED';
+        error.status = 422;
+        throw error;
+      }
+      const pageSize = Math.min(500, Math.max(1, Number(limit) || 100));
+      const candidates = await executorGateway.listRetentionCandidates({
+        before: cutoff,
+        limit: pageSize,
+      });
+      const uniqueCandidates = [...new Map(
+        candidates.map((candidate) => [candidate.workflowRunId, candidate]),
+      ).values()];
+      if (dryRun) {
+        return {
+          version: 'orchestrator-retention-purge-v1',
+          dryRun: true,
+          before: cutoff,
+          candidates: uniqueCandidates,
+          deleted: [],
+          failures: [],
+        };
+      }
+      const deleted = [];
+      const failures = [];
+      for (const candidate of uniqueCandidates) {
+        try {
+          const result = await deleteRun(candidate.workflowRunId);
+          if (result) {
+            deleted.push({
+              runId: candidate.workflowRunId,
+              jobId: candidate.jobId,
+            });
+            continue;
+          }
+          const executorJobDeleted = await executorGateway.deleteJob(candidate.jobId);
+          if (executorJobDeleted) {
+            deleted.push({
+              runId: candidate.workflowRunId,
+              jobId: candidate.jobId,
+              orphanedCheckpoint: true,
+            });
+          }
+        } catch (error) {
+          failures.push({
+            runId: candidate.workflowRunId,
+            jobId: candidate.jobId,
+            code: String(error?.code ?? 'WORKFLOW_PURGE_FAILED').slice(0, 128),
+            message: String(error instanceof Error ? error.message : error).slice(0, 500),
+          });
+        }
+      }
+      return {
+        version: 'orchestrator-retention-purge-v1',
+        dryRun: false,
+        before: cutoff,
+        candidates: uniqueCandidates,
+        deleted,
+        failures,
+      };
+    },
+
     getExecutorJob(jobId) {
       return executorGateway.getJob(jobId);
     },
@@ -294,4 +396,5 @@ export function createLangGraphOrchestratorRuntime({
 export const orchestratorRuntimeInternals = {
   graphConfig,
   projectSnapshot,
+  TERMINAL_RUN_STATUSES,
 };

services/orchestrator/runtime.test.mjs

@@ -87,6 +87,45 @@ test('LangGraph runtime is idempotent by run id', async () => {
   assert.equal((await runtime.listEvents('run-shadow-1')).events.length, 3);
 });
 
+test('LangGraph runtime deletes terminal checkpoints and linked Executor Jobs', async () => {
+  const runtime = createLangGraphOrchestratorRuntime({
+    checkpointer: new MemorySaver(),
+  });
+  await runtime.start(runSpec());
+  assert.deepEqual(await runtime.deleteRun('run-shadow-1'), {
+    runId: 'run-shadow-1',
+    deleted: true,
+    executorJobDeleted: true,
+  });
+  assert.equal(await runtime.getState('run-shadow-1'), null);
+  assert.equal(await runtime.getExecutorJob('run-shadow-1:executor-preview'), null);
+  assert.equal(await runtime.deleteRun('run-shadow-1'), null);
+});
+
+test('LangGraph runtime retention purge is dry-run by default and deletes only on apply', async () => {
+  const runtime = createLangGraphOrchestratorRuntime({
+    checkpointer: new MemorySaver(),
+  });
+  await runtime.start(runSpec({ runId: 'run-retention-1' }));
+  const before = Date.now() + 1000;
+  const preview = await runtime.purgeTerminalRuns({ before });
+  assert.equal(preview.dryRun, true);
+  assert.equal(preview.candidates.length, 1);
+  assert.equal(await runtime.getState('run-retention-1') != null, true);
+
+  const applied = await runtime.purgeTerminalRuns({
+    before,
+    dryRun: false,
+  });
+  assert.equal(applied.deleted.length, 1);
+  assert.equal(applied.failures.length, 0);
+  assert.equal(await runtime.getState('run-retention-1'), null);
+  await assert.rejects(
+    () => runtime.purgeTerminalRuns({ before: 0 }),
+    (error) => error.code === 'WORKFLOW_PURGE_CUTOFF_REQUIRED',
+  );
+});
+
 test('LangGraph runtime persists a durable blocked Executor Job and probes both stores', async () => {
   const checkpointProbeCalls = [];
   const executorProbeCalls = [];
@@ -179,6 +218,10 @@ test('LangGraph active workflow waits on a leased Executor Job and converges ter
   assert.equal(state.status, 'waiting');
   assert.equal(state.phase, 'executor_queued');
   assert.equal(state.plan.executorJob.status, 'queued');
+  await assert.rejects(
+    () => runtime.deleteRun('run-active-1'),
+    (error) => error.code === 'WORKFLOW_RUN_NOT_TERMINAL' && error.status === 409,
+  );
 
   const claim = await runtime.claimExecutorJob({
     workerId: 'worker-1',

services/orchestrator/server.mjs

@@ -25,10 +25,112 @@ function csvList(value) {
   )];
 }
 
+function isLoopbackHost(value) {
+  const host = String(value ?? '').trim().toLowerCase();
+  return host === 'localhost'
+    || host === '::1'
+    || host === '[::1]'
+    || /^127(?:\.\d{1,3}){3}$/.test(host);
+}
+
+function assertSecureServerConfig({
+  host,
+  executionEnabled,
+  serviceToken,
+  workerToken,
+} = {}) {
+  const hasServiceToken = Boolean(String(serviceToken ?? '').trim());
+  const hasWorkerToken = Boolean(String(workerToken ?? '').trim());
+  if (!isLoopbackHost(host) && !hasServiceToken) {
+    const error = new Error(
+      'MEMIND_ORCHESTRATOR_SERVICE_TOKEN is required for non-loopback binding',
+    );
+    error.code = 'ORCHESTRATOR_SERVICE_TOKEN_REQUIRED';
+    throw error;
+  }
+  if (executionEnabled && (!hasServiceToken || !hasWorkerToken)) {
+    const error = new Error(
+      'Execution requires both MEMIND_ORCHESTRATOR_SERVICE_TOKEN and MEMIND_ORCHESTRATOR_WORKER_TOKEN',
+    );
+    error.code = 'ORCHESTRATOR_EXECUTION_TOKENS_REQUIRED';
+    throw error;
+  }
+  if (
+    executionEnabled
+    && String(serviceToken).trim() === String(workerToken).trim()
+  ) {
+    const error = new Error('Execution service and worker tokens must be distinct');
+    error.code = 'ORCHESTRATOR_EXECUTION_TOKENS_NOT_DISTINCT';
+    throw error;
+  }
+}
+
+function startRetentionSweep(runtime, {
+  retentionDays,
+  intervalMs = 24 * 60 * 60 * 1000,
+  limit = 100,
+  nowMs = Date.now,
+  setIntervalFn = setInterval,
+  clearIntervalFn = clearInterval,
+  logger = console,
+} = {}) {
+  const days = Number(retentionDays);
+  if (!Number.isFinite(days) || days <= 0) return null;
+  const boundedDays = Math.min(3650, Math.max(1, Math.floor(days)));
+  const configuredInterval = Number(intervalMs);
+  const boundedInterval = Number.isFinite(configuredInterval)
+    ? Math.max(60 * 60 * 1000, configuredInterval)
+    : 24 * 60 * 60 * 1000;
+  let active = false;
+  const sweep = async () => {
+    if (active) return null;
+    active = true;
+    try {
+      const result = await runtime.purgeTerminalRuns({
+        before: nowMs() - boundedDays * 24 * 60 * 60 * 1000,
+        limit,
+        dryRun: false,
+      });
+      if (result.deleted.length || result.failures.length) {
+        logger.log(
+          `[orchestrator-retention] deleted=${result.deleted.length} failures=${result.failures.length}`,
+        );
+      }
+      return result;
+    } catch (error) {
+      logger.warn(
+        '[orchestrator-retention] sweep failed:',
+        error instanceof Error ? error.message : error,
+      );
+      return null;
+    } finally {
+      active = false;
+    }
+  };
+  const timer = setIntervalFn(() => void sweep(), boundedInterval);
+  timer?.unref?.();
+  return {
+    retentionDays: boundedDays,
+    intervalMs: boundedInterval,
+    sweep,
+    stop() {
+      clearIntervalFn(timer);
+    },
+  };
+}
+
 export async function startOrchestratorServer({
   env = process.env,
   logger = console,
 } = {}) {
+  const host = String(env.MEMIND_ORCHESTRATOR_HOST ?? '127.0.0.1').trim() || '127.0.0.1';
+  const executionEnabled = envFlag(env.MEMIND_ORCHESTRATOR_EXECUTION_ENABLED, false);
+  assertSecureServerConfig({
+    host,
+    executionEnabled,
+    serviceToken: env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN,
+    workerToken: env.MEMIND_ORCHESTRATOR_WORKER_TOKEN,
+  });
   const checkpoint = await createOrchestratorCheckpoint({
     mode: env.MEMIND_ORCHESTRATOR_CHECKPOINT_MODE,
     connectionString: env.MEMIND_ORCHESTRATOR_DATABASE_URL,
@@ -52,7 +154,7 @@ export async function startOrchestratorServer({
     checkpointProbe: checkpoint.probe,
     executorJobStore: executorJobs.store,
     executorJobProbe: executorJobs.probe,
-    executionEnabled: envFlag(env.MEMIND_ORCHESTRATOR_EXECUTION_ENABLED, false),
+    executionEnabled,
     enabledExecutors: csvList(env.MEMIND_ORCHESTRATOR_ENABLED_EXECUTORS),
     admissionConfig: {
       tenantAllowlist: csvList(env.MEMIND_ORCHESTRATOR_TENANT_ALLOWLIST),
@@ -72,8 +174,8 @@ export async function startOrchestratorServer({
     serviceToken: env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN,
     workerToken: env.MEMIND_ORCHESTRATOR_WORKER_TOKEN,
   });
-  const host = String(env.MEMIND_ORCHESTRATOR_HOST ?? '127.0.0.1').trim() || '127.0.0.1';
   const port = positivePort(env.MEMIND_ORCHESTRATOR_PORT, 8093);
+  let retentionSweep = null;
   let server;
   try {
     server = await new Promise((resolve, reject) => {
@@ -84,9 +186,16 @@ export async function startOrchestratorServer({
     await Promise.allSettled([executorJobs.close(), checkpoint.close()]);
     throw error;
   }
+  retentionSweep = startRetentionSweep(runtime, {
+    retentionDays: env.MEMIND_ORCHESTRATOR_RETENTION_DAYS,
+    intervalMs: env.MEMIND_ORCHESTRATOR_RETENTION_SWEEP_INTERVAL_MS,
+    limit: env.MEMIND_ORCHESTRATOR_RETENTION_SWEEP_LIMIT,
+    logger,
+  });
   logger.log(`[orchestrator] listening on http://${host}:${port} (${checkpoint.kind})`);
 
   async function close() {
+    retentionSweep?.stop();
     let serverCloseError = null;
     try {
       await new Promise((resolve, reject) => {
@@ -110,6 +219,7 @@ export async function startOrchestratorServer({
     runtime,
     checkpoint,
     executorJobs,
+    retentionSweep,
     close,
   };
 }
@@ -129,7 +239,10 @@ if (isEntrypoint) {
 }
 
 export const orchestratorServerInternals = {
+  assertSecureServerConfig,
   csvList,
   envFlag,
+  isLoopbackHost,
   positivePort,
+  startRetentionSweep,
 };

services/orchestrator/server.test.mjs

@@ -1,7 +1,10 @@
 import assert from 'node:assert/strict';
 import net from 'node:net';
 import test from 'node:test';
-import { startOrchestratorServer } from './server.mjs';
+import {
+  orchestratorServerInternals,
+  startOrchestratorServer,
+} from './server.mjs';
 
 async function reservePort() {
   const server = net.createServer();
@@ -46,3 +49,79 @@ test('orchestrator server owns checkpoint and Executor Job Store lifecycle', asy
   closed = true;
   assert.equal(running.server.listening, false);
 });
+
+test('orchestrator server fails closed for exposed or executable configurations without tokens', () => {
+  assert.throws(
+    () => orchestratorServerInternals.assertSecureServerConfig({
+      host: '0.0.0.0',
+      executionEnabled: false,
+    }),
+    (error) => error.code === 'ORCHESTRATOR_SERVICE_TOKEN_REQUIRED',
+  );
+  assert.throws(
+    () => orchestratorServerInternals.assertSecureServerConfig({
+      host: '127.0.0.1',
+      executionEnabled: true,
+      serviceToken: 'service-token',
+    }),
+    (error) => error.code === 'ORCHESTRATOR_EXECUTION_TOKENS_REQUIRED',
+  );
+  assert.throws(
+    () => orchestratorServerInternals.assertSecureServerConfig({
+      host: '127.0.0.1',
+      executionEnabled: true,
+      serviceToken: 'shared-token',
+      workerToken: 'shared-token',
+    }),
+    (error) => error.code === 'ORCHESTRATOR_EXECUTION_TOKENS_NOT_DISTINCT',
+  );
+  assert.doesNotThrow(
+    () => orchestratorServerInternals.assertSecureServerConfig({
+      host: '0.0.0.0',
+      executionEnabled: true,
+      serviceToken: 'service-token',
+      workerToken: 'worker-token',
+    }),
+  );
+});
+
+test('orchestrator retention sweep is disabled by default and applies an explicit cutoff', async () => {
+  const calls = [];
+  let intervalCallback = null;
+  let cleared = false;
+  const runtime = {
+    async purgeTerminalRuns(options) {
+      calls.push(options);
+      return { deleted: [], failures: [] };
+    },
+  };
+  assert.equal(
+    orchestratorServerInternals.startRetentionSweep(runtime, {
+      retentionDays: 0,
+    }),
+    null,
+  );
+  const sweep = orchestratorServerInternals.startRetentionSweep(runtime, {
+    retentionDays: 30,
+    intervalMs: 60 * 60 * 1000,
+    nowMs: () => 40 * 24 * 60 * 60 * 1000,
+    setIntervalFn(callback) {
+      intervalCallback = callback;
+      return { unref() {} };
+    },
+    clearIntervalFn() {
+      cleared = true;
+    },
+    logger: { log() {}, warn() {} },
+  });
+  assert.equal(sweep.retentionDays, 30);
+  intervalCallback();
+  await new Promise((resolve) => setImmediate(resolve));
+  assert.deepEqual(calls, [{
+    before: 10 * 24 * 60 * 60 * 1000,
+    limit: 100,
+    dryRun: false,
+  }]);
+  sweep.stop();
+  assert.equal(cleared, true);
+});

services/orchestrator/shadow-dispatcher.mjs

@@ -0,0 +1,93 @@
+function positiveInteger(value, fallback) {
+  const parsed = Number(value);
+  return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
+}
+
+export function createShadowObservationDispatcher({
+  observe,
+  onCompleted = async () => {},
+  onFailed = async () => {},
+  onSkipped = async () => {},
+  maxConcurrent = 2,
+  maxQueued = 100,
+  schedule = setImmediate,
+  now = Date.now,
+  logger = console,
+} = {}) {
+  const enabled = typeof observe === 'function';
+  const concurrency = positiveInteger(maxConcurrent, 2);
+  const queueLimit = positiveInteger(maxQueued, 100);
+  const queue = [];
+  let active = 0;
+  let drainScheduled = false;
+
+  function reportCallbackFailure(label, error) {
+    logger.warn(
+      `[orchestrator-shadow] ${label} callback failed:`,
+      error instanceof Error ? error.message : error,
+    );
+  }
+
+  function scheduleDrain() {
+    if (!enabled || drainScheduled || !queue.length || active >= concurrency) return;
+    drainScheduled = true;
+    schedule(drain);
+  }
+
+  function runObservation(entry) {
+    active += 1;
+    const context = {
+      enqueuedAt: entry.enqueuedAt,
+      startedAt: now(),
+    };
+    void Promise.resolve()
+      .then(() => observe(entry.input))
+      .then((result) => onCompleted(entry.input, result, context))
+      .catch((error) => onFailed(entry.input, error, context))
+      .catch((error) => reportCallbackFailure('failure', error))
+      .finally(() => {
+        active -= 1;
+        scheduleDrain();
+      });
+  }
+
+  function drain() {
+    drainScheduled = false;
+    while (active < concurrency && queue.length) {
+      runObservation(queue.shift());
+    }
+  }
+
+  return {
+    dispatch(input) {
+      if (!enabled) return false;
+      if (queue.length >= queueLimit) {
+        const context = {
+          enqueuedAt: now(),
+          reason: 'shadow_queue_full',
+        };
+        void Promise.resolve()
+          .then(() => onSkipped(input, context))
+          .catch((error) => reportCallbackFailure('skip', error));
+        return false;
+      }
+      queue.push({ input, enqueuedAt: now() });
+      scheduleDrain();
+      return true;
+    },
+
+    status() {
+      return {
+        enabled,
+        active,
+        queued: queue.length,
+        maxConcurrent: concurrency,
+        maxQueued: queueLimit,
+      };
+    },
+  };
+}
+
+export const shadowDispatcherInternals = {
+  positiveInteger,
+};

services/orchestrator/shadow-dispatcher.test.mjs

@@ -0,0 +1,71 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { createShadowObservationDispatcher } from './shadow-dispatcher.mjs';
+
+async function flush() {
+  await new Promise((resolve) => setImmediate(resolve));
+  await new Promise((resolve) => setImmediate(resolve));
+}
+
+test('shadow dispatcher is a no-op when observation is disabled', () => {
+  const dispatcher = createShadowObservationDispatcher();
+  assert.equal(dispatcher.dispatch({ runId: 'run-disabled' }), false);
+  assert.deepEqual(dispatcher.status(), {
+    enabled: false,
+    active: 0,
+    queued: 0,
+    maxConcurrent: 2,
+    maxQueued: 100,
+  });
+});
+
+test('shadow dispatcher bounds concurrency and drops overflow without blocking callers', async () => {
+  const releases = [];
+  const started = [];
+  const skipped = [];
+  const dispatcher = createShadowObservationDispatcher({
+    maxConcurrent: 1,
+    maxQueued: 2,
+    observe: async ({ runId }) => {
+      started.push(runId);
+      await new Promise((resolve) => releases.push(resolve));
+      return { runId };
+    },
+    onSkipped: async ({ runId }, context) => {
+      skipped.push([runId, context.reason]);
+    },
+    logger: { warn() {} },
+  });
+
+  assert.equal(dispatcher.dispatch({ runId: 'run-1' }), true);
+  assert.equal(dispatcher.dispatch({ runId: 'run-2' }), true);
+  assert.equal(dispatcher.dispatch({ runId: 'run-3' }), false);
+  await flush();
+  assert.deepEqual(started, ['run-1']);
+  assert.deepEqual(skipped, [['run-3', 'shadow_queue_full']]);
+
+  releases.shift()();
+  await flush();
+  assert.deepEqual(started, ['run-1', 'run-2']);
+  releases.shift()();
+  await flush();
+  assert.equal(dispatcher.status().active, 0);
+});
+
+test('shadow dispatcher isolates observer and callback failures', async () => {
+  const failures = [];
+  const dispatcher = createShadowObservationDispatcher({
+    observe: async () => {
+      throw Object.assign(new Error('remote unavailable'), { code: 'REMOTE_DOWN' });
+    },
+    onFailed: async ({ runId }, error) => {
+      failures.push([runId, error.code]);
+    },
+    logger: { warn() {} },
+  });
+
+  assert.equal(dispatcher.dispatch({ runId: 'run-failed' }), true);
+  await flush();
+  assert.deepEqual(failures, [['run-failed', 'REMOTE_DOWN']]);
+  assert.equal(dispatcher.status().active, 0);
+});

services/orchestrator/shadow-observer.mjs

@@ -4,19 +4,6 @@ import {
 } from './contracts.mjs';
 import { createRemoteWorkflowEngine } from './engine-registry.mjs';
 
-const MAX_INSTRUCTION_CHARACTERS = 16_000;
-
-function extractMessageText(message) {
-  if (typeof message === 'string') return message;
-  if (!message || typeof message !== 'object') return '';
-  if (typeof message.content === 'string') return message.content;
-  if (!Array.isArray(message.content)) return '';
-  return message.content
-    .filter((part) => part?.type === 'text')
-    .map((part) => String(part.text ?? ''))
-    .join('\n');
-}
-
 function safeError(error) {
   return {
     code: String(error?.code ?? 'WORKFLOW_SHADOW_FAILED').slice(0, 128),
@@ -38,9 +25,7 @@ export function createWorkflowShadowObserver({
     runId,
     requestId,
     userId,
-    sessionId = null,
     workflowName = 'code-run-v1',
-    userMessage,
     taskType = null,
   } = {}) {
     const selection = await configService.selectEngine({
@@ -81,19 +66,15 @@ export function createWorkflowShadowObserver({
       timeoutMs: state.config.requestTimeoutMs,
       fetchImpl,
     });
-    const instruction = extractMessageText(userMessage).slice(0, MAX_INSTRUCTION_CHARACTERS);
     const spec = normalizeRunSpec({
       runId,
       requestId,
       workflow: { name: workflowName, version: 1 },
-      subject: { userId },
+      subject: {},
       input: {
-        instruction,
+        instruction: '[shadow-control-plane-only]',
         taskType,
         toolMode: 'code',
-        sessionRef: sessionId
-          ? { kind: 'goose-session', id: String(sessionId) }
-          : null,
       },
       policy: {
         executionMode: 'observe-only',
@@ -102,6 +83,7 @@ export function createWorkflowShadowObserver({
       metadata: {
         source: 'memind-agent-run',
         configVersion: selection.configVersion,
+        dataPolicy: 'control-plane-only-v1',
       },
     });
     try {
@@ -126,7 +108,5 @@ export function createWorkflowShadowObserver({
 }
 
 export const workflowShadowObserverInternals = {
-  MAX_INSTRUCTION_CHARACTERS,
-  extractMessageText,
   safeError,
 };

services/orchestrator/shadow-observer.test.mjs

@@ -1,5 +1,8 @@
 import assert from 'node:assert/strict';
 import test from 'node:test';
+import { MemorySaver } from '@langchain/langgraph';
+import { createOrchestratorApp } from './app.mjs';
+import { createLangGraphOrchestratorRuntime } from './runtime.mjs';
 import { createWorkflowShadowObserver } from './shadow-observer.mjs';
 
 function jsonResponse(body, status = 200) {
@@ -9,6 +12,19 @@ function jsonResponse(body, status = 200) {
   });
 }
 
+async function listen(app) {
+  const server = await new Promise((resolve) => {
+    const candidate = app.listen(0, '127.0.0.1', () => resolve(candidate));
+  });
+  const address = server.address();
+  return {
+    baseUrl: `http://127.0.0.1:${address.port}`,
+    close: () => new Promise((resolve, reject) => {
+      server.close((error) => (error ? reject(error) : resolve()));
+    }),
+  };
+}
+
 test('shadow observer skips without creating a remote client when mode does not select shadow', async () => {
   let fetchCalls = 0;
   const observer = createWorkflowShadowObserver({
@@ -127,7 +143,7 @@ test('shadow boundary records non-selected canary decisions for a complete rate
   assert.equal(result.executionPlan.taskType, 'code_analysis');
 });
 
-test('shadow observer sends a bounded observe-only RunSpec to LangGraph service', async () => {
+test('shadow observer sends a control-plane-only RunSpec without user content or identity', async () => {
   let capturedUrl = null;
   let capturedInit = null;
   const observer = createWorkflowShadowObserver({
@@ -175,6 +191,98 @@ test('shadow observer sends a bounded observe-only RunSpec to LangGraph service'
   assert.equal(body.version, 'orchestrator-run-v1');
   assert.equal(body.policy.executionMode, 'observe-only');
   assert.equal(body.policy.sideEffectsAllowed, false);
-  assert.deepEqual(body.input.sessionRef, { kind: 'goose-session', id: 'session-2' });
+  assert.equal(body.input.instruction, '[shadow-control-plane-only]');
+  assert.equal('sessionRef' in body.input, false);
+  assert.deepEqual(body.subject, { tenantId: null, userId: null });
   assert.equal(body.metadata.configVersion, 7);
+  assert.equal(body.metadata.dataPolicy, 'control-plane-only-v1');
+  assert.equal(
+    capturedInit.body.includes('Implement the service boundary'),
+    false,
+  );
+  assert.equal(capturedInit.body.includes('user-2'), false);
+  assert.equal(capturedInit.body.includes('session-2'), false);
+});
+
+test('isolated Portal shadow flow reaches LangGraph over HTTP and removes all terminal state', async (t) => {
+  const checkpointer = new MemorySaver();
+  const runtime = createLangGraphOrchestratorRuntime({
+    checkpointer,
+    checkpointKind: 'memory',
+    durable: false,
+  });
+  const server = await listen(createOrchestratorApp({
+    runtime,
+    serviceToken: 'isolated-shadow-token',
+  }));
+  t.after(server.close);
+
+  const observer = createWorkflowShadowObserver({
+    configService: {
+      async selectEngine() {
+        return {
+          shadowEngine: 'langgraph',
+          reason: 'shadow',
+          mode: 'shadow',
+          configVersion: 11,
+        };
+      },
+      async getRuntimeState() {
+        return {
+          config: {
+            serviceUrl: server.baseUrl,
+            requestTimeoutMs: 2000,
+          },
+        };
+      },
+    },
+    serviceToken: 'isolated-shadow-token',
+  });
+
+  const secretUserContent = 'private user content must never cross the shadow boundary';
+  const result = await observer({
+    runId: 'isolated-shadow-run-1',
+    requestId: 'isolated-shadow-request-1',
+    userId: 'private-user-id',
+    sessionId: 'private-session-id',
+    taskType: 'repo_refactor',
+    userMessage: {
+      role: 'user',
+      content: [{ type: 'text', text: secretUserContent }],
+    },
+  });
+
+  assert.equal(result.observed, true);
+  assert.equal(result.shadowRun.status, 'succeeded');
+  assert.equal(result.shadowRun.result.executed, false);
+  const tuple = await checkpointer.getTuple({
+    configurable: {
+      thread_id: 'isolated-shadow-run-1',
+      checkpoint_ns: '',
+    },
+  });
+  assert.equal(
+    tuple.checkpoint.channel_values.spec.input.instruction,
+    '[shadow-control-plane-only]',
+  );
+  const checkpointJson = JSON.stringify(tuple.checkpoint.channel_values.spec);
+  assert.equal(checkpointJson.includes(secretUserContent), false);
+  assert.equal(checkpointJson.includes('private-user-id'), false);
+  assert.equal(checkpointJson.includes('private-session-id'), false);
+
+  const deleted = await fetch(`${server.baseUrl}/v1/runs/isolated-shadow-run-1`, {
+    method: 'DELETE',
+    headers: { authorization: 'Bearer isolated-shadow-token' },
+  });
+  assert.equal(deleted.status, 200);
+  assert.deepEqual(await deleted.json(), {
+    runId: 'isolated-shadow-run-1',
+    deleted: true,
+    executorJobDeleted: true,
+  });
+  assert.equal(await runtime.getState('isolated-shadow-run-1'), null);
+  assert.equal(
+    await runtime.getExecutorJob('isolated-shadow-run-1:executor-preview'),
+    null,
+  );
 });