fix(context): repair headroom probe routing and add portal E2E gate

Use the default headroom health probe instead of the goosed fetch adapter so active mode routes deepseek-no-think through the proxy, and add an E2E script that verifies goosed pong via headroom with fail-open portal checks. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-09 22:01:29 +08:00
parent fc4ac98e9f
commit 26ce1241c6
8 changed files with 440 additions and 12 deletions
@@ -0,0 +1,385 @@
+#!/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);
+});
@@ -40,7 +40,7 @@ async function main() {

  const child = spawn(
    'headroom',
-    ['proxy', '--port', String(port)],
+    ['proxy', '--port', String(port), '--openai-api-url', upstream.replace(/\/v1\/?$/, '')],
    {
      stdio: 'inherit',
      env: {