memind/scripts/smoke-orchestrator-shadow.mjs

#!/usr/bin/env node
import crypto from 'node:crypto';
import { createAgentRunGateway } from '../agent-run-gateway.mjs';
import { createDbPool, isDatabaseConfigured } from '../db.mjs';
import { createOrchestratorAdminConfigService } from '../services/orchestrator/admin-config.mjs';
import { createOrchestratorObservabilityService } from '../services/orchestrator/observability.mjs';
import { createWorkflowShadowObserver } from '../services/orchestrator/shadow-observer.mjs';
import { loadMemindEnvFiles } from './memind-runtime-profile.mjs';

loadMemindEnvFiles(new URL('..', import.meta.url).pathname);

const args = new Set(process.argv.slice(2));
const cleanup = args.has('--cleanup');
const restoreConfig = args.has('--restore-config');
const serviceUrl = String(
  process.env.MEMIND_ORCHESTRATOR_URL ?? 'http://127.0.0.1:8093',
).trim().replace(/\/$/, '');
const serviceToken = String(process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN ?? '').trim();

function assertLocalServiceUrl(value) {
  const url = new URL(value);
  if (!['127.0.0.1', 'localhost', '::1'].includes(url.hostname) && !args.has('--allow-remote')) {
    throw new Error('Shadow smoke defaults to a loopback Orchestrator; pass --allow-remote explicitly');
  }
}

async function waitForShadowEvent(pool, runId, timeoutMs = 15_000) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    const [rows] = await pool.query(
      `SELECT event_type, data_json, created_at
         FROM h5_agent_run_events
        WHERE run_id = ?
          AND event_type IN ('workflow_shadow_completed', 'workflow_shadow_failed')
        ORDER BY created_at DESC
        LIMIT 1`,
      [runId],
    );
    if (rows[0]) return rows[0];
    await new Promise((resolve) => setTimeout(resolve, 100));
  }
  throw new Error(`Timed out waiting for Shadow projection for run ${runId}`);
}

async function deleteOrchestratorRun(runId) {
  const response = await fetch(`${serviceUrl}/v1/runs/${encodeURIComponent(runId)}`, {
    method: 'DELETE',
    headers: {
      authorization: `Bearer ${serviceToken}`,
      accept: 'application/json',
    },
  });
  if (response.ok) return;
  if (response.status === 404) {
    const body = await response.json().catch(() => null);
    if (body?.error?.code === 'RUN_NOT_FOUND') return;
  }
  throw new Error(`Orchestrator run cleanup failed with HTTP ${response.status}`);
}

async function main() {
  if (!isDatabaseConfigured()) {
    throw new Error('Shadow smoke requires the local Memind MySQL configuration');
  }
  if (!serviceToken) {
    throw new Error('MEMIND_ORCHESTRATOR_SERVICE_TOKEN is required');
  }
  assertLocalServiceUrl(serviceUrl);
  const ready = await fetch(`${serviceUrl}/ready`);
  if (!ready.ok) throw new Error(`Orchestrator readiness failed with HTTP ${ready.status}`);

  const pool = createDbPool();
  const configService = createOrchestratorAdminConfigService(pool);
  const previous = await configService.getAdminConfig();
  let runId = null;
  try {
    await configService.updateAdminConfig({
      ...previous.config,
      mode: 'shadow',
      serviceUrl,
      primaryEngine: 'langgraph',
      workflowAllowlist: ['code-run-v1'],
    });
    const [users] = await pool.query(
      `SELECT id
         FROM h5_users
        ORDER BY created_at ASC
        LIMIT 1`,
    );
    if (!users[0]?.id) throw new Error('Shadow smoke requires at least one local user');

    const observer = createWorkflowShadowObserver({
      configService,
      serviceToken,
    });
    const gateway = createAgentRunGateway({
      pool,
      tkmindProxy: {},
      autoDispatch: false,
      observeWorkflowRun: observer,
    });
    const requestId = `orchestrator-shadow-smoke-${Date.now()}-${crypto.randomUUID()}`;
    const run = await gateway.createRun(users[0].id, {
      requestId,
      toolMode: 'code',
      taskType: 'orchestrator_shadow_smoke',
      userMessage: {
        role: 'user',
        content: [{ type: 'text', text: 'Observe this local smoke run without executing tools.' }],
      },
    });
    runId = run.id;
    const event = await waitForShadowEvent(pool, runId);
    if (event.event_type !== 'workflow_shadow_completed') {
      throw new Error(`Shadow smoke failed: ${JSON.stringify(event.data_json)}`);
    }

    const completedAt = Date.now();
    await pool.query(
      `UPDATE h5_agent_runs
          SET status = 'succeeded', updated_at = ?, completed_at = ?
        WHERE id = ? AND status = 'queued'`,
      [completedAt, completedAt, runId],
    );

    const observability = createOrchestratorObservabilityService({
      pool,
      configService,
      serviceToken,
    });
    const [summary, detail] = await Promise.all([
      observability.listShadowRuns({ hours: 1, limit: 10 }),
      observability.getShadowRun(runId),
    ]);
    if (!detail?.remote?.available || detail.remote.state?.status !== 'succeeded') {
      throw new Error('Shadow checkpoint detail was not recoverable through the remote API');
    }

    console.log(JSON.stringify({
      ok: true,
      runId,
      mode: 'shadow',
      shadowEvent: event.event_type,
      metrics: summary.metrics,
      checkpointStatus: detail.remote.state.status,
      graphEvents: detail.remote.events.map((item) => item.type),
      cleanup,
      restoreConfig,
    }, null, 2));
  } finally {
    let cleanupError = null;
    try {
      if (cleanup && runId) {
        await deleteOrchestratorRun(runId);
        await pool.query('DELETE FROM h5_agent_runs WHERE id = ?', [runId]);
      }
    } catch (error) {
      cleanupError = error;
    }
    try {
      if (restoreConfig) {
        await configService.updateAdminConfig(previous.config);
      }
    } finally {
      await pool.end();
    }
    if (cleanupError) throw cleanupError;
  }
}

main().catch((error) => {
  console.error(error instanceof Error ? error.message : error);
  process.exitCode = 1;
});