memind/scripts/run-headroom-active-portal-e2e.mjs

#!/usr/bin/env node
/**
 * End-to-end headroom active observation:
 * 1) sync goosed deepseek-no-think provider through headroom proxy
 * 2) run a Portal agent turn and collect fusion/headroom events
 * 3) run a direct goosed pong turn on the headroom-routed provider
 */
import { randomUUID } from 'node:crypto';
import path from 'node:path';
import { fileURLToPath } from 'node:url';

import { createDbPool } from '../db.mjs';
import {
  MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID,
  syncProfileToGoosed,
} from '../llm-providers.mjs';
import { deepseekDisableThinkingEnabled } from '../deepseek-no-think-proxy.mjs';
import {
  buildHeadroomRunObservation,
  ensureHeadroomProxyRunning,
  probeHeadroomProxyReachable,
  resolveGoosedApiUrlWithHeadroom,
  resolveHeadroomMode,
  resolveHeadroomProxyBaseUrl,
  resolveHeadroomUpstreamBaseUrl,
} from '../memind-headroom-policy.mjs';
import { prepareGooseV149CheckEnv } from './goose-v149-canary.mjs';
import { waitForAgentRunWorkerIdle } from './goose-v149-worker-idle.mjs';
import { collectReplyEvents, createV149Client } from './goose-v149-sse.mjs';
import {
  createReporter,
  loginViaApi,
  resolvePortalBase,
  waitForRunTerminal,
} from './scenario-test-lib.mjs';

const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..');
prepareGooseV149CheckEnv(process.env, root);

process.env.MEMIND_HEADROOM_MODE = process.env.MEMIND_HEADROOM_MODE ?? 'active';
process.env.MEMIND_CONTEXT_BUDGET_MODE = process.env.MEMIND_CONTEXT_BUDGET_MODE ?? 'shadow';
process.env.MEMIND_RECALL_FUSION_MODE = process.env.MEMIND_RECALL_FUSION_MODE ?? 'shadow';
if (!process.env.MEMIND_HEADROOM_PROXY_PORT) {
  process.env.MEMIND_HEADROOM_PROXY_PORT = '8789';
}
if (!process.env.MEMIND_HEADROOM_PROXY_BASE_URL) {
  process.env.MEMIND_HEADROOM_PROXY_BASE_URL = `http://127.0.0.1:${process.env.MEMIND_HEADROOM_PROXY_PORT}/v1`;
}
if (!process.env.MEMIND_DEEPSEEK_DISABLE_THINKING) {
  process.env.MEMIND_DEEPSEEK_DISABLE_THINKING = '1';
}

const baseUrl = resolvePortalBase(Number(process.env.H5_PORT ?? 8081));
const v149Target = process.env.TKMIND_API_TARGET_V149
  || process.env.GOOSE_V149_API_TARGET
  || 'https://127.0.0.1:18049';
const apiSecret = process.env.TKMIND_SERVER__SECRET_KEY_V149
  || process.env.GOOSE_SERVER__SECRET_KEY
  || process.env.TKMIND_SERVER__SECRET_KEY
  || 'local-dev-secret';
const timeoutMs = Number(process.env.GOOSE_V149_HEADROOM_PORTAL_TIMEOUT_MS ?? 180_000);

async function portalReachable() {
  try {
    const response = await fetch(`${baseUrl}/auth/status`);
    return response.ok;
  } catch {
    return false;
  }
}

async function recoverStuckRuns(pool) {
  const now = Date.now();
  const cutoff = now - Number(process.env.MEMIND_AGENT_RUN_STALE_MS ?? 600_000);
  const [result] = await pool.query(
    `UPDATE h5_agent_runs
     SET status = 'failed',
         error_message = 'headroom portal e2e preflight recovery',
         updated_at = ?,
         completed_at = ?
     WHERE status = 'running'
       AND started_at IS NOT NULL
       AND started_at <= ?`,
    [now, now, cutoff],
  );
  return Number(result?.affectedRows ?? 0);
}

async function syncDeepseekNoThinkThroughHeadroom(client) {
  const apiKey = process.env.DEEPSEEK_API_KEY || process.env.GOOSE_V149_TEST_API_KEY || '';
  if (!apiKey) return null;
  if (!deepseekDisableThinkingEnabled()) {
    throw new Error('MEMIND_DEEPSEEK_DISABLE_THINKING must be enabled for headroom deepseek sync');
  }
  const model = process.env.GOOSE_V149_PROVIDER_MODEL ?? 'deepseek-chat';
  const goosedFetch = (url, init) => {
    const pathname = `${url.pathname}${url.search}`;
    return client.apiFetch(pathname, init);
  };
  await syncProfileToGoosed(
    v149Target,
    apiSecret,
    {
      providerId: 'custom_deepseek',
      providerKind: 'builtin',
      apiKey,
      defaultModel: model,
      models: ['deepseek-chat', 'deepseek-reasoner'],
    },
    goosedFetch,
  );
  return readGoosedProviderApiUrl(client, MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID);
}

async function readGoosedProviderApiUrl(client, providerId) {
  const response = await client.apiFetch(
    `/config/custom-providers/${encodeURIComponent(providerId)}`,
  );
  if (!response.ok) return null;
  const payload = await response.json().catch(() => ({}));
  return payload?.config?.base_url
    ?? payload?.config?.api_url
    ?? payload?.api_url
    ?? payload?.apiUrl
    ?? null;
}

async function fetchRunEvents(pool, runId) {
  const [rows] = await pool.query(
    `SELECT event_type, data_json, created_at
     FROM h5_agent_run_events
     WHERE run_id = ?
     ORDER BY created_at ASC`,
    [runId],
  );
  return rows.map((row) => ({
    eventType: row.event_type,
    data: typeof row.data_json === 'string'
      ? JSON.parse(row.data_json)
      : row.data_json,
    createdAt: Number(row.created_at),
  }));
}

async function runPortalTurn(pool, { sessionId = null } = {}) {
  const username = process.env.RELEASE_GATE_SCENARIO_USERNAME ?? 'john';
  const password =
    process.env.JOHN_PASSWORD
    ?? process.env.H5_ACCESS_PASSWORD
    ?? process.env.MEMIND_PASSWORD
    ?? '';
  if (!password) throw new Error('set JOHN_PASSWORD for portal e2e');

  const reporter = createReporter();
  const auth = await loginViaApi(baseUrl, { username, password }, reporter);

  await waitForAgentRunWorkerIdle(root, process.env, {
    logPrefix: '[headroom-portal-e2e]',
  });

  let activeSessionId = sessionId;
  if (!activeSessionId) {
    const startRes = await fetch(`${baseUrl}/api/agent/start`, {
      method: 'POST',
      headers: { Cookie: auth.cookie, 'Content-Type': 'application/json' },
      body: JSON.stringify({}),
    });
    const started = await startRes.json().catch(() => ({}));
    if (!startRes.ok || !started?.id) {
      throw new Error(`agent/start failed: ${startRes.status}`);
    }
    activeSessionId = started.id;
  }

  const warmResume = await fetch(`${baseUrl}/api/agent/resume`, {
    method: 'POST',
    headers: { Cookie: auth.cookie, 'Content-Type': 'application/json' },
    body: JSON.stringify({
      session_id: activeSessionId,
      load_model_and_extensions: true,
    }),
  });
  if (!warmResume.ok) {
    const warmBody = await warmResume.text().catch(() => '');
    throw new Error(`pre-run resume failed: ${warmResume.status} ${warmBody.slice(0, 200)}`);
  }

  const requestId = randomUUID();
  const runRes = await fetch(`${baseUrl}/api/agent/runs`, {
    method: 'POST',
    headers: { Cookie: auth.cookie, 'Content-Type': 'application/json' },
    body: JSON.stringify({
      request_id: requestId,
      session_id: activeSessionId,
      user_message: {
        id: randomUUID(),
        role: 'user',
        content: [{ type: 'text', text: `headroom portal e2e ping ${Date.now().toString(36)}` }],
        metadata: {
          userVisible: true,
          agentVisible: true,
          displayText: 'headroom portal e2e ping',
        },
      },
    }),
  });
  const runPayload = await runRes.json().catch(() => ({}));
  if (!runRes.ok) {
    throw new Error(`POST /api/agent/runs ${runRes.status}: ${JSON.stringify(runPayload).slice(0, 300)}`);
  }
  const runId = runPayload.run?.id ?? runPayload.id;
  const terminal = await waitForRunTerminal(baseUrl, auth.cookie, runId, timeoutMs);
  const events = await fetchRunEvents(pool, runId);
  return {
    sessionId: activeSessionId,
    runId,
    terminal,
    events,
  };
}

async function runPortalTurnWithRetry(pool) {
  const maxAttempts = Number(process.env.GOOSE_V149_HEADROOM_PORTAL_ATTEMPTS ?? 2);
  let lastResult = null;
  let lastError = null;
  let sessionId = null;
  for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
    try {
      const result = await runPortalTurn(pool, { sessionId });
      lastResult = { ...result, attempt };
      if (result.terminal.status === 'succeeded') {
        return lastResult;
      }
      sessionId = result.sessionId;
    } catch (err) {
      lastError = err;
    }
    if (attempt < maxAttempts) {
      console.log(`HEADROOM_PORTAL_E2E_RETRY: attempt=${attempt}`);
      await recoverStuckRuns(pool);
      await waitForAgentRunWorkerIdle(root, process.env, {
        logPrefix: '[headroom-portal-e2e]',
      });
    }
  }
  if (lastResult) return lastResult;
  throw lastError ?? new Error('portal run failed');
}

async function runGoosedPong(client, providerId, model) {
  const session = await client.apiJson('/agent/start', {
    working_dir: process.env.GOOSE_V149_WORKING_DIR || root,
  });
  await client.apiJson('/agent/update_provider', {
    session_id: session.id,
    provider: providerId,
    model,
  });
  const result = await collectReplyEvents({
    client,
    sessionId: session.id,
    requestId: randomUUID(),
    userMessage: {
      role: 'user',
      created: Date.now(),
      content: [{ type: 'text', text: 'Reply with exactly one word: pong' }],
      metadata: { userVisible: true, agentVisible: true, displayText: 'headroom goosed pong' },
    },
    timeoutMs: Number(process.env.GOOSE_V149_REPLY_TIMEOUT_MS ?? 90_000),
  });
  return {
    sessionId: session.id,
    outcome: result.outcome,
    tokenState: result.finishEvent?.token_state ?? null,
  };
}

async function main() {
  if (!(await portalReachable())) {
    throw new Error(`Portal not reachable at ${baseUrl}`);
  }

  const headroomBoot = await ensureHeadroomProxyRunning({ env: process.env });
  let headroomReachable = headroomBoot.reachable;
  const headroomBase = resolveHeadroomProxyBaseUrl();
  if (!headroomReachable) {
    const deadline = Date.now() + 30_000;
    while (Date.now() < deadline) {
      await new Promise((resolve) => setTimeout(resolve, 2000));
      headroomReachable = await probeHeadroomProxyReachable({ baseUrl: headroomBase, env: process.env });
      if (headroomReachable) break;
    }
  }
  if (!headroomReachable) {
    throw new Error(`headroom proxy unreachable at ${headroomBase}`);
  }

  const routing = resolveGoosedApiUrlWithHeadroom({
    apiUrl: resolveHeadroomUpstreamBaseUrl(),
    mode: resolveHeadroomMode(),
    headroomReachable,
    eligible: true,
  });
  if (!routing.routed) {
    throw new Error(`headroom active routing not applied: ${JSON.stringify(routing)}`);
  }

  const pool = createDbPool();
  const recovered = await recoverStuckRuns(pool);
  if (recovered > 0) {
    console.log(`HEADROOM_PORTAL_E2E: recovered ${recovered} stale running row(s)`);
  }

  const client = createV149Client();
  const routedApiUrl = await syncDeepseekNoThinkThroughHeadroom(client);
  const providerApiUrl = routedApiUrl
    ?? await readGoosedProviderApiUrl(client, MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID);
  const headroomHostPort = headroomBase.replace(/\/v1\/?$/, '');
  const providerUsesHeadroom = String(providerApiUrl ?? '').startsWith(headroomHostPort);

  const portal = await runPortalTurnWithRetry(pool);
  const headroomEvents = portal.events.filter((event) => event.eventType === 'headroom_context_observed');
  const expectedHeadroomObservation = buildHeadroomRunObservation({ skillId: null });
  const budgetEvents = portal.events.filter((event) => event.eventType === 'context_budget_resolved');
  const fusionEvents = portal.events.filter((event) => event.eventType === 'recall_fusion_resolved');

  let goosed = null;
  if (providerUsesHeadroom) {
    goosed = await runGoosedPong(
      client,
      MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID,
      process.env.GOOSE_V149_PROVIDER_MODEL ?? 'deepseek-chat',
    );
  }

  console.log('HEADROOM_PORTAL_E2E:');
  console.log(`  headroom_started=${headroomBoot.started}`);
  console.log(`  headroom_proxy=${headroomBase}`);
  console.log(`  upstream=${resolveHeadroomUpstreamBaseUrl()}`);
  console.log(`  goosed_sync=${JSON.stringify({ providerId: MEMIND_DEEPSEEK_NO_THINK_PROVIDER_ID, mode: routing.mode })}`);
  console.log(`  deepseek_no_think_api_url=${providerApiUrl ?? 'missing'}`);
  console.log(`  provider_uses_headroom=${providerUsesHeadroom}`);
  console.log(`  portal_run=${portal.runId} status=${portal.terminal.status}`);
  console.log(`  portal_headroom_events=${headroomEvents.length}`);
  console.log(`  portal_budget_events=${budgetEvents.length}`);
  console.log(`  portal_fusion_events=${fusionEvents.length}`);
  console.log(`  expected_headroom_observation=${JSON.stringify(expectedHeadroomObservation)}`);
  if (headroomEvents[0]?.data) {
    console.log(`  portal_headroom_observation=${JSON.stringify(headroomEvents[0].data)}`);
  } else {
    console.log('  portal_headroom_note=restart Portal with MEMIND_HEADROOM_MODE=active to emit live events');
  }
  if (goosed) {
    console.log(`  goosed_pong_outcome=${goosed.outcome}`);
    console.log(`  goosed_token_state=${JSON.stringify(goosed.tokenState)}`);
  }

  if (portal.terminal.status !== 'succeeded') {
    console.warn(
      `HEADROOM_PORTAL_E2E_WARN: portal run ${portal.terminal.status}`
      + `${portal.terminal.error ? `: ${portal.terminal.error}` : ''}`,
    );
  }
  if (!providerUsesHeadroom) {
    throw new Error(`deepseek-no-think provider not routed through headroom: ${providerApiUrl ?? 'missing'}`);
  }
  if (!goosed || goosed.outcome !== 'finish') {
    throw new Error(`goosed pong through headroom provider failed: ${goosed?.outcome ?? 'skipped'}`);
  }
  if (headroomEvents.length === 0) {
    console.warn('HEADROOM_PORTAL_E2E_WARN: live portal events missing; restart Portal with MEMIND_HEADROOM_MODE=active');
  }

  console.log(
    `HEADROOM_PORTAL_E2E_OK: session=${portal.sessionId} run=${portal.runId} `
    + `headroomEvents=${headroomEvents.length} goosed=${goosed.outcome} base=${baseUrl}`,
  );

  await pool.end();
}

main().catch(async (error) => {
  console.error(`HEADROOM_PORTAL_E2E_FAIL: ${error.message}`);
  process.exit(1);
});