learning_ai_common_plat/services/cowork-service/e2e/ecosystem-integration.test.ts

import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { createServer, type Server } from 'node:http';

const integrationEmail = process.env.INTEGRATION_TEST_EMAIL;
const integrationPassword = process.env.INTEGRATION_TEST_PASSWORD;

const runIntegration = Boolean(
  process.env.INTEGRATION_TESTS && integrationEmail && integrationPassword
);

const platformBaseUrl = process.env.PLATFORM_SERVICE_URL ?? 'http://localhost:4003/api';
const coworkBaseUrl = process.env.COWORK_SERVICE_URL ?? 'http://localhost:4009';
const integrationProductId = 'clawcowork';

type JsonRecord = Record<string, unknown>;

function requireIntegrationCredentials(): { email: string; password: string } {
  if (!integrationEmail || !integrationPassword) {
    throw new Error(
      'Set INTEGRATION_TEST_EMAIL and INTEGRATION_TEST_PASSWORD before running integration tests'
    );
  }
  return { email: integrationEmail, password: integrationPassword };
}

async function requestJson<T = JsonRecord>(
  url: string,
  init: RequestInit = {},
  token?: string
): Promise<{ status: number; data: T }> {
  const headers: Record<string, string> = {
    'Content-Type': 'application/json',
    ...(init.headers as Record<string, string> | undefined),
  };
  if (token) headers.authorization = `Bearer ${token}`;
  if (!headers['x-product-id'] && url.startsWith(platformBaseUrl)) {
    headers['x-product-id'] = integrationProductId;
  }

  const response = await fetch(url, { ...init, headers });
  const text = await response.text();
  const data = text ? (JSON.parse(text) as T) : ({} as T);
  return { status: response.status, data };
}

async function waitFor(
  label: string,
  callback: () => Promise<boolean>,
  timeoutMs = 45_000,
  intervalMs = 2_500
): Promise<void> {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if (await callback()) return;
    await new Promise(resolve => setTimeout(resolve, intervalMs));
  }
  throw new Error(`Timed out waiting for ${label}`);
}

describe.runIf(runIntegration)('ecosystem integration (H.1-H.12)', () => {
  let token = '';
  let webhookServer: Server | null = null;
  const webhookDeliveries: JsonRecord[] = [];
  let webhookUrl = '';
  let webhookSubscriptionId = '';
  let marketplaceListingId = '';
  let sandboxFlagInitialEnabled = true;

  beforeAll(async () => {
    const creds = requireIntegrationCredentials();
    const login = await requestJson<{
      accessToken: string;
      user: { id: string; email: string };
    }>(`${platformBaseUrl}/auth/login`, {
      method: 'POST',
      body: JSON.stringify({
        email: creds.email,
        password: creds.password,
        productId: integrationProductId,
      }),
    });

    expect(login.status).toBe(200);
    expect(login.data.accessToken).toBeTruthy();
    token = login.data.accessToken;

    webhookServer = createServer((req, res) => {
      const chunks: Buffer[] = [];
      req.on('data', chunk => chunks.push(Buffer.from(chunk)));
      req.on('end', () => {
        try {
          webhookDeliveries.push(JSON.parse(Buffer.concat(chunks).toString('utf8')));
        } catch {
          webhookDeliveries.push({ raw: Buffer.concat(chunks).toString('utf8') });
        }
        res.writeHead(200, { 'Content-Type': 'application/json' });
        res.end(JSON.stringify({ ok: true }));
      });
    });

    await new Promise<void>((resolve, reject) => {
      webhookServer!.listen(0, '127.0.0.1', err => (err ? reject(err) : resolve()));
    });
    const address = webhookServer.address();
    if (!address || typeof address === 'string') {
      throw new Error('Failed to bind webhook test server');
    }
    webhookUrl = `http://127.0.0.1:${address.port}/webhook`;
  });

  afterAll(async () => {
    if (webhookSubscriptionId) {
      await requestJson(
        `${platformBaseUrl}/webhooks/subscriptions/${webhookSubscriptionId}`,
        {
          method: 'DELETE',
        },
        token
      ).catch(() => undefined);
    }

    if (!sandboxFlagInitialEnabled) {
      await requestJson(
        `${platformBaseUrl}/flags/sandbox_enabled/toggle`,
        {
          method: 'POST',
        },
        token
      ).catch(() => undefined);
    }

    if (webhookServer) {
      await new Promise<void>((resolve, reject) =>
        webhookServer!.close(err => (err ? reject(err) : resolve()))
      );
    }
  });

  it('H.1: product registry contains clawcowork and flags are seeded', async () => {
    const products = await requestJson<{ products: Array<{ productId: string }> }>(
      `${platformBaseUrl}/products`
    );
    expect(products.status).toBe(200);
    expect(products.data.products.some(product => product.productId === integrationProductId)).toBe(
      true
    );

    const flags = await requestJson<{ flags: Array<{ key: string }> }>(
      `${platformBaseUrl}/flags`,
      {},
      token
    );
    expect(flags.status).toBe(200);
    expect(flags.data.flags.length).toBeGreaterThanOrEqual(13);
  });

  it('H.3: login obtains JWT and cowork-service accepts task submission', async () => {
    const task = await requestJson<{ id: string }>(
      `${coworkBaseUrl}/api/tasks`,
      {
        method: 'POST',
        body: JSON.stringify({
          goal: 'Integration smoke test task',
          folder: process.cwd(),
          model: 'claude-sonnet-4-20250514',
        }),
      },
      token
    );

    expect(task.status).toBe(201);
    expect(task.data.id).toBeTruthy();
  });

  it('H.3: enterprise SSO lookup endpoint is available', async () => {
    const email = process.env.INTEGRATION_SSO_EMAIL ?? integrationEmail!;
    const lookup = await requestJson<{ found: boolean; idp: { protocol?: string } | null }>(
      `${platformBaseUrl}/auth/enterprise/lookup?email=${encodeURIComponent(email)}`
    );

    expect(lookup.status).toBe(200);
    expect(typeof lookup.data.found).toBe('boolean');
    if (lookup.data.found) {
      expect(['saml', 'oidc']).toContain(lookup.data.idp?.protocol);
    }
  });

  it('H.4: sandbox_enabled propagates through platform flag polling', async () => {
    const before = await requestJson<{ flags: Record<string, boolean> }>(
      `${platformBaseUrl}/flags/poll`,
      {},
      token
    );
    expect(before.status).toBe(200);
    sandboxFlagInitialEnabled = Boolean(before.data.flags.sandbox_enabled);

    if (sandboxFlagInitialEnabled) {
      const toggle = await requestJson(
        `${platformBaseUrl}/flags/sandbox_enabled/toggle`,
        {
          method: 'POST',
        },
        token
      );
      expect(toggle.status).toBe(200);
    }

    await waitFor('sandbox_enabled=false', async () => {
      const polled = await requestJson<{ flags: Record<string, boolean> }>(
        `${platformBaseUrl}/flags/poll`,
        {},
        token
      );
      return polled.status === 200 && polled.data.flags.sandbox_enabled === false;
    });
  });

  it('H.5: audit entries appear after task submission flush', async () => {
    const task = await requestJson<{ id: string }>(
      `${coworkBaseUrl}/api/tasks`,
      {
        method: 'POST',
        body: JSON.stringify({
          goal: 'Trigger audit flush integration test',
          folder: process.cwd(),
          model: 'claude-sonnet-4-20250514',
        }),
      },
      token
    );
    expect(task.status).toBe(201);

    await waitFor(
      'audit entry for clawcowork',
      async () => {
        const audit = await requestJson<{ records?: JsonRecord[]; entries?: JsonRecord[] }>(
          `${platformBaseUrl}/audit?days=7&limit=25`,
          {},
          token
        );
        const entries = audit.data.records ?? audit.data.entries ?? [];
        return audit.status === 200 && entries.length > 0;
      },
      50_000,
      5_000
    );
  });

  it('H.6: usage limit check returns a verdict for a small daily budget', async () => {
    const verdict = await requestJson<JsonRecord>(
      `${coworkBaseUrl}/api/usage/check-limits`,
      {
        method: 'POST',
        body: JSON.stringify({ plan: 'free' }),
      },
      token
    );

    expect(verdict.status).toBe(200);
    expect('allowed' in verdict.data || 'withinLimits' in verdict.data).toBe(true);
  });

  it('H.8: telemetry metrics are queryable after submitting activity', async () => {
    await requestJson(
      `${coworkBaseUrl}/api/tasks`,
      {
        method: 'POST',
        body: JSON.stringify({
          goal: 'Trigger telemetry integration test',
          folder: process.cwd(),
          model: 'claude-sonnet-4-20250514',
        }),
      },
      token
    );

    await waitFor(
      'telemetry metrics',
      async () => {
        const metrics = await requestJson<JsonRecord>(
          `${platformBaseUrl}/telemetry/metrics`,
          {},
          token
        );
        return metrics.status === 200;
      },
      50_000,
      5_000
    );
  });

  it('H.9: webhook subscription receives a test delivery', async () => {
    const created = await requestJson<{ id: string }>(
      `${coworkBaseUrl}/api/webhooks`,
      {
        method: 'POST',
        body: JSON.stringify({
          url: webhookUrl,
          events: ['user.created', 'task.completed'],
        }),
      },
      token
    );
    expect(created.status).toBe(201);
    webhookSubscriptionId = created.data.id;

    const testDelivery = await requestJson(
      `${platformBaseUrl}/webhooks/subscriptions/${webhookSubscriptionId}/test`,
      { method: 'POST' },
      token
    );
    expect(testDelivery.status).toBe(200);

    await waitFor('webhook delivery', async () => webhookDeliveries.length > 0, 20_000, 1_000);
    expect(webhookDeliveries.length).toBeGreaterThan(0);
  });

  it('H.10: cowork-service proxies extraction requests', async () => {
    const models = await requestJson<{ models: string[] }>(`${coworkBaseUrl}/api/extract/models`);
    expect(models.status).toBe(200);
    expect(models.data.models.length).toBeGreaterThan(0);

    const extraction = await requestJson<JsonRecord>(`${coworkBaseUrl}/api/extract`, {
      method: 'POST',
      body: JSON.stringify({
        text:
          process.env.INTEGRATION_EXTRACTION_TEXT ??
          'Sample extraction payload for Claw Cowork integration testing.',
        task: 'entity-extraction',
      }),
    });

    expect(extraction.status).toBe(200);
    expect(extraction.data).toBeTruthy();
  });

  it('H.12: marketplace listing install is recorded', async () => {
    const listings = await requestJson<{
      items?: Array<{ id: string }>;
      listings?: Array<{ id: string }>;
    }>(`${coworkBaseUrl}/api/marketplace?limit=10`, {}, token);
    expect(listings.status).toBe(200);

    const items = listings.data.items ?? listings.data.listings ?? [];
    expect(items.length).toBeGreaterThan(0);
    marketplaceListingId = items[0].id;

    const install = await requestJson<JsonRecord>(
      `${coworkBaseUrl}/api/marketplace/${marketplaceListingId}/install`,
      { method: 'POST' },
      token
    );
    expect([200, 201]).toContain(install.status);

    await waitFor('marketplace install record', async () => {
      const installs = await requestJson<{
        items?: Array<{ listingId: string }>;
        installs?: Array<{ listingId: string }>;
      }>(`${coworkBaseUrl}/api/marketplace/installs`, {}, token);
      const records = installs.data.items ?? installs.data.installs ?? [];
      return records.some(record => record.listingId === marketplaceListingId);
    });
  });
});