qwen-code/integration-tests/interactive/workflow-completion.test.ts

/**
 * @license
 * Copyright 2026 Qwen Team
 * SPDX-License-Identifier: Apache-2.0
 */

import { mkdirSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { afterEach, describe, expect, it } from 'vitest';
import {
  startFakeOpenAIServer,
  type FakeOpenAIServer,
} from '../fake-openai-server.js';
import {
  applyContainerSandboxNoProxy,
  fakeServerHostOptions,
  TestRig,
} from '../test-helper.js';
import { pickE2eRenderer } from '../renderer-matrix.js';
import {
  InteractiveSession,
  sendAboutUntilRendered,
} from './interactive-session.js';

// A model reply to a workflow completion is a full roundtrip that starts only
// after the run settles and the notification drains, which the previous 30s
// budget was too tight for on a loaded runner (cf. the cron suite's #10904).
// This widening is hardening, not the #14191 fix: #14191 was a docker boot
// stall upstream of these waits — run 38524609091 job 115744720088 burned all
// three retries' 90s ready-prompt budget in InteractiveSession.start with the
// sandbox image already local — addressed by the self-hosted docker tier of
// readyPromptBudgetMs in integration-tests/ready-prompt-budget.ts.
const MODEL_REPLY_SCREEN_TIMEOUT = 120_000;

// Per-case ceiling with slack over a loaded self-hosted docker-leg run
// (~556s): 180s ready prompt + ~110s /about window (its 90s deadline is
// checked only after a trailing 20s waitForScreen) + 2 x ~123s reply waits
// (3s pre-poll sleep, then a 120s deadline) + ~20s of idles, paging, and
// sends. The slack keeps a breach attributable to the wait that overran
// instead of surfacing as a bare vitest timeout. retry: 0 because six cases
// at this ceiling would multiply past the 20-minute no-AK PR gate that runs
// this file, and a genuine break fails at its first waitForScreen either way
// — retrying would only re-burn the same budget.
const WORKFLOW_CASE_OPTIONS = { timeout: 600_000, retry: 0 };

// OpenTUI's schedule_tool handler is a separate, unwired entry point.
describe.skipIf(pickE2eRenderer() === 'opentui')(
  'saved workflow slash-command completion',
  () => {
    const usedCaseDirs = new Set<string>();
    let rig: TestRig;
    let server: FakeOpenAIServer | undefined;
    let session: InteractiveSession | undefined;
    let restoreNoProxy: (() => void) | undefined;

    afterEach(async () => {
      try {
        await session?.close();
      } finally {
        session = undefined;
        await server?.close();
        server = undefined;
        restoreNoProxy?.();
        await rig?.cleanup();
      }
    });

    it.each([
      {
        source: 'slash',
        status: 'completed',
        script:
          'return { marker: "WORKFLOW_PARTIAL_RESULT_12176", failed: ["fr"] };',
        marker: 'WORKFLOW_PARTIAL_RESULT_12176',
      },
      {
        source: 'slash',
        status: 'failed',
        script: 'throw new Error("WORKFLOW_RUN_ERROR_12176");',
        marker: 'WORKFLOW_RUN_ERROR_12176',
      },
      {
        source: 'slash',
        status: 'completed',
        // Exceeds the wire cap while keeping interactive rendering bounded.
        script:
          'return { marker: "WORKFLOW_LARGE_RESULT_12176", rows: "&".repeat(30_000), failed: ["fr"] };',
        marker: 'WORKFLOW_LARGE_RESULT_12176',
        largeResult: true,
      },
      {
        source: 'model',
        status: 'completed',
        script:
          'return { marker: "WORKFLOW_MODEL_RESULT_12176", error: new Error("MODEL_ERROR_REASON_12176") };',
        marker: 'WORKFLOW_MODEL_RESULT_12176',
        resultDetails: ['Error: MODEL_ERROR_REASON_12176'],
      },
      {
        source: 'slash',
        status: 'completed',
        script:
          'const error = new Error("WORKFLOW_VM_ERROR_12176"); return { marker: "WORKFLOW_ERROR_RESULT_12176", failed: ["fr"], errors: [error, new Error("AGENT_B_TIMEOUT_12176"), new Error("AGENT_C_OOM_12176")], error };',
        marker: 'WORKFLOW_ERROR_RESULT_12176',
        reportedError: 'WORKFLOW_VM_ERROR_12176',
        resultDetails: [
          'Error: AGENT_B_TIMEOUT_12176',
          'Error: AGENT_C_OOM_12176',
        ],
      },
      {
        source: 'slash',
        status: 'completed',
        script:
          'return { marker: "WORKFLOW_COLLECTION_RESULT_12176", failed: ["fr"], errors: new Map([["agent-map", new Error("MAP_RATE_LIMIT_12176")]]), error: new Set(["SET_TIMEOUT_12176"]) };',
        marker: 'WORKFLOW_COLLECTION_RESULT_12176',
        resultDetails: [
          'agent-map',
          'Error: MAP_RATE_LIMIT_12176',
          'SET_TIMEOUT_12176',
        ],
      },
    ])('delivers $marker once', WORKFLOW_CASE_OPTIONS, async (testCase) => {
      const isSlash = testCase.source === 'slash';
      rig = new TestRig();
      restoreNoProxy = applyContainerSandboxNoProxy();
      // The marker keeps same-status cases from sharing a directory: the
      // post-merge E2E legs set KEEP_OUTPUT (the no-AK PR gate that runs
      // this file does not), so setup() there would remove the previous
      // case's directory while its CLI is still writing, failing with
      // ENOTEMPTY before the body runs.
      await rig.setup(
        `workflow-completion-${testCase.status}-${testCase.marker}`,
        {
          settings: { telemetry: { enabled: false } },
        },
      );
      // sanitizeTestName is lossy, so distinct markers are not by themselves
      // proof of distinct directories; nothing else catches a duplicate or
      // dropped marker before a KEEP_OUTPUT leg hits ENOTEMPTY.
      expect(usedCaseDirs.has(rig.testDir!)).toBe(false);
      usedCaseDirs.add(rig.testDir!);
      const qwenHome = join(rig.testDir!, '.qwen-home');
      mkdirSync(qwenHome, { recursive: true });
      writeFileSync(
        join(qwenHome, 'settings.json'),
        JSON.stringify({
          general: { enableAutoUpdate: false },
          tools: { workflowsEnabled: true },
          model: { skipWorkflowUsageWarning: true },
          memory: {
            enableManagedAutoMemory: false,
            enableManagedAutoDream: false,
          },
          ui: { enableFollowupSuggestions: false },
          security: {
            auth: { selectedType: 'openai' },
            folderTrust: { enabled: false },
          },
        }),
      );
      const workflowDir = join(rig.testDir!, '.qwen', 'workflows');
      mkdirSync(workflowDir, { recursive: true });
      writeFileSync(join(workflowDir, 'report-probe.js'), testCase.script);

      const reply = `MODEL_RECEIVED_WORKFLOW_${testCase.status.toUpperCase()}`;
      server = await startFakeOpenAIServer(({ body }) => {
        if (body['stream'] !== true) return { content: '{}' };
        const messages = JSON.stringify(body['messages']);
        if (!isSlash && !messages.includes(testCase.marker)) {
          return {
            toolCalls: [
              {
                id: 'model-workflow-call',
                type: 'function',
                function: {
                  name: 'workflow',
                  arguments: JSON.stringify({ name: 'report-probe' }),
                },
              },
            ],
          };
        }
        const expectedReply = messages.includes(
          'What did that workflow return?',
        )
          ? `${reply}_FOLLOWUP`
          : reply;
        return {
          content:
            (!isSlash || messages.includes('<kind>workflow</kind>')) &&
            messages.includes(testCase.marker)
              ? expectedReply
              : 'WORKFLOW_RESULT_MISSING',
        };
      }, fakeServerHostOptions());

      session = await InteractiveSession.start({
        cwd: rig.testDir!,
        env: {
          QWEN_HOME: qwenHome,
          QWEN_RUNTIME_DIR: join(rig.testDir!, '.runtime'),
          QWEN_CODE_LANG: 'en',
          QWEN_CODE_DISABLE_WORKFLOWS: '0',
        },
        args: [
          '--auth-type',
          'openai',
          '--openai-api-key',
          'fake-key',
          '--openai-base-url',
          server.baseUrl,
          '--model',
          'fake-model',
          '--approval-mode',
          'yolo',
        ],
      });

      await sendAboutUntilRendered(session);
      expect(
        server.requests.filter(({ body }) => body['stream'] === true),
      ).toHaveLength(0);

      await session.idle(500);
      await session.send(
        isSlash ? '/report-probe' : 'Run the saved report-probe workflow.',
      );
      await session.waitForScreen(
        (screen) => screen.includes(reply),
        'model reply based on the workflow completion, without a follow-up prompt',
        MODEL_REPLY_SCREEN_TIMEOUT,
      );
      await session.idle(1_000);

      let screen = await session.screen();
      if (testCase.largeResult) {
        // Ink's virtualized history only exposes the current viewport, not
        // terminal scrollback. Read preceding pages of the long completion.
        for (
          let page = 0;
          page < 20 &&
          (!screen.includes(testCase.marker) ||
            !screen.includes('Run ID: wf_') ||
            !screen.includes('Reported failed: ["fr"]'));
          page++
        ) {
          session.pressKey('\x1b[5~'); // PageUp
          await session.idle(500);
          screen += `\n${await session.screen()}`;
        }
        session.pressKey('\x1b[1;5F'); // Ctrl+End
        await session.idle(500);
      }
      expect(screen).not.toContain('started in the background');
      expect(screen).toContain(testCase.marker);
      for (const detail of testCase.resultDetails ?? []) {
        expect(screen).toContain(detail);
      }
      if (isSlash) expect(screen).toContain('Run ID: wf_');
      if (isSlash && testCase.status === 'completed') {
        expect(screen).toContain('Reported failed: ["fr"]');
      }
      if (testCase.reportedError) {
        expect(screen).toContain(
          `Reported error: Error: ${testCase.reportedError}`,
        );
      }
      const requests = server.requests.filter(
        ({ body }) => body['stream'] === true,
      );
      expect(requests).toHaveLength(isSlash ? 1 : 2);
      const messages = JSON.stringify(requests.at(-1)!.body['messages']);
      if (isSlash)
        expect(messages).toContain(`<status>${testCase.status}</status>`);
      expect(messages.match(/<kind>workflow<\/kind>/g) ?? []).toHaveLength(
        isSlash ? 1 : 0,
      );
      expect(messages).toContain(testCase.marker);
      if (testCase.reportedError) {
        const reported = messages.match(
          /<reported-failures>([\s\S]*?)<\/reported-failures>/,
        )?.[1];
        expect(reported).toContain(
          `Reported error: Error: ${testCase.reportedError}`,
        );
        expect(reported).not.toContain(' at ');
        expect(reported).toContain(
          'Reported errors: [\\"Error: WORKFLOW_VM_ERROR_12176\\",\\"Error: AGENT_B_TIMEOUT_12176\\",\\"Error: AGENT_C_OOM_12176\\"]',
        );
      }
      for (const detail of testCase.resultDetails ?? []) {
        const payload = isSlash
          ? messages.match(
              /<reported-failures>([\s\S]*?)<\/reported-failures>/,
            )?.[1]
          : messages;
        expect(payload).toContain(detail);
      }
      if (testCase.largeResult) {
        expect(
          messages.match(/<result>([\s\S]*?)<\/result>/)?.[1].length,
        ).toBeLessThanOrEqual(25_000);
        expect(messages).toContain('<result-truncated>');
        expect(messages).toContain(
          'Error, Map, and Set contents are not preserved there',
        );
        expect(messages).toMatch(
          /its resultPreview field keeps the result as bounded display text, shown by \/workflows wf_[a-f0-9]+\./,
        );
        expect(messages).toContain(
          'reported-failure previews may also be truncated',
        );
        expect(messages).toMatch(/[/\\]workflows[/\\]+wf_[a-f0-9]+\.json/);
        expect(messages).toContain(
          '<reported-failures>Reported failed: [\\"fr\\"]</reported-failures>',
        );
        expect(messages.match(/<result>([\s\S]*?)<\/result>/)?.[1]).not.toMatch(
          /&[^;]*$/,
        );
      } else if (isSlash && testCase.status === 'completed') {
        expect(messages).toContain('&quot;failed&quot;:[&quot;fr&quot;]');
      }

      await session.idle(500);
      await session.send('What did that workflow return?');
      await session.waitForScreen(
        (text) => text.includes(`${reply}_FOLLOWUP`),
        'follow-up result',
        MODEL_REPLY_SCREEN_TIMEOUT,
      );
      await session.idle(1_000);
      const followUpRequests = server.requests.filter(
        ({ body }) => body['stream'] === true,
      );
      expect(followUpRequests).toHaveLength(isSlash ? 2 : 3);
      const followUp = JSON.stringify(
        followUpRequests.at(-1)!.body['messages'],
      );
      expect(followUp).toContain(testCase.marker);
      for (const detail of testCase.resultDetails ?? []) {
        const payload = isSlash
          ? followUp.match(
              /<reported-failures>([\s\S]*?)<\/reported-failures>/,
            )?.[1]
          : followUp;
        expect(payload).toContain(detail);
      }
      if (testCase.reportedError) {
        expect(
          followUp.match(
            /<reported-failures>([\s\S]*?)<\/reported-failures>/,
          )?.[1],
        ).toContain(testCase.reportedError);
      }
      if (testCase.largeResult) {
        expect(followUp).toContain('<result-truncated>');
        expect(followUp).toContain(
          'Error, Map, and Set contents are not preserved there',
        );
        expect(followUp).toMatch(
          /its resultPreview field keeps the result as bounded display text, shown by \/workflows wf_[a-f0-9]+\./,
        );
        expect(followUp).toContain(
          '<reported-failures>Reported failed: [\\"fr\\"]</reported-failures>',
        );
      }
      expect(followUp.match(/<kind>workflow<\/kind>/g) ?? []).toHaveLength(
        isSlash ? 1 : 0,
      );
    });
  },
);