feat(external-context): Add daemon memory writes

integration-tests/cli/external-context-mem0-daemon-write.test.ts

@@ -0,0 +1,393 @@
+/**
+ * @license
+ * Copyright 2026 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { createServer } from 'node:http';
+import {
+  mkdirSync,
+  mkdtempSync,
+  readFileSync,
+  realpathSync,
+  rmSync,
+  writeFileSync,
+} from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join, resolve } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { describe, expect, it, vi } from 'vitest';
+import type { DaemonEvent, DaemonClient } from '@qwen-code/sdk';
+import { fakeToolCall, startFakeOpenAIServer } from '../fake-openai-server.js';
+import {
+  approveWorkspaceMcpServers,
+  spawnDaemon,
+  writeWorkspaceSettings,
+  type SpawnedDaemon,
+} from './_daemon-harness.js';
+
+const repo = fileURLToPath(new URL('../../', import.meta.url));
+const serverName = 'external-context-mem0-write';
+const toolName = `mcp__${serverName}__context_remember`;
+type Session = Awaited<ReturnType<DaemonClient['createOrAttachSession']>>;
+type Permission = {
+  requestId: string;
+  toolCall: { rawInput: unknown };
+  options: Array<{ optionId: string; kind: string }>;
+};
+type Attempt = {
+  session: Session;
+  events: DaemonEvent[];
+  controller: AbortController;
+  subscription: Promise<void>;
+  task: Promise<unknown>;
+};
+
+const skip =
+  process.platform === 'win32' ||
+  Boolean(
+    process.env['QWEN_SANDBOX'] && process.env['QWEN_SANDBOX'] !== 'false',
+  );
+
+describe.skipIf(skip)('daemon explicit external memory writes', () => {
+  it('binds permissions and provider writes to the owning workspace and reloads only that writer', async () => {
+    const root = realpathSync(
+      mkdtempSync(join(tmpdir(), 'mem0-daemon-write-e2e-')),
+    );
+    const home = join(root, 'home');
+    const qwenHome = join(home, '.qwen');
+    mkdirSync(qwenHome, { recursive: true });
+    const workspaces = ['A', 'B'].map((name) => join(root, name));
+    workspaces.forEach((cwd) => mkdirSync(cwd));
+    const requests: Array<{
+      body: unknown;
+      authorization: string | undefined;
+    }> = [];
+    const provider = createServer(async (req, res) => {
+      const chunks: Buffer[] = [];
+      for await (const chunk of req) chunks.push(Buffer.from(chunk));
+      requests.push({
+        body: JSON.parse(Buffer.concat(chunks).toString('utf8')) as unknown,
+        authorization: req.headers.authorization,
+      });
+      res.writeHead(200, { 'content-type': 'application/json' });
+      res.end(
+        JSON.stringify({
+          results: [{ id: `memory-${requests.length}`, event: 'ADD' }],
+        }),
+      );
+    });
+    await new Promise<void>((resolve, reject) => {
+      provider.once('error', reject);
+      provider.listen(0, '127.0.0.1', resolve);
+    });
+    const address = provider.address();
+    if (!address || typeof address === 'string')
+      throw new Error('No provider port');
+    const writeJson = (path: string, value: unknown) =>
+      writeFileSync(path, JSON.stringify(value));
+    const dialectPath = join(root, 'write-dialect.json');
+    writeJson(dialectPath, {
+      writeDialectVersion: 1,
+      id: 'synthetic-write-v1',
+      auth: 'authorization-token',
+      create: {
+        path: '/memories',
+        userIdLocation: 'json',
+        agentIdLocation: 'omit',
+        appIdLocation: 'omit',
+      },
+      response: { completion: 'records', collection: 'results', idField: 'id' },
+    });
+    const configs = workspaces.map((cwd, index) => ({
+      schemaVersion: 4,
+      repositoryRoot: cwd,
+      dialectPath,
+      endpoint: {
+        origin: `http://127.0.0.1:${address.port}`,
+        basePath: '',
+        allowInsecureHttp: true,
+      },
+      credentialEnv: 'SYNTHETIC_WRITE_TOKEN',
+      scope: { userId: `scope-${index}` },
+      timeoutMs: 5000,
+    }));
+    const configPaths = configs.map((config, index) => {
+      const path = join(root, `write-${index}.json`);
+      writeJson(path, config);
+      return path;
+    });
+    const approvals: Record<string, unknown> = {};
+    workspaces.forEach((cwd, index) => {
+      const servers = {
+        [serverName]: {
+          command: process.execPath,
+          args: [
+            resolve(
+              repo,
+              'integrations/external-context-mem0/dist/write-main.js',
+            ),
+          ],
+          cwd,
+          env: {
+            QWEN_EXTERNAL_CONTEXT_MEM0_WRITE_CONFIG: configPaths[index]!,
+            SYNTHETIC_WRITE_TOKEN: `synthetic-${index}`,
+          },
+          includeTools: ['context_remember'],
+          trust: false,
+          alwaysLoadTools: true,
+        },
+      };
+      writeWorkspaceSettings(cwd, {
+        tools: { approvalMode: 'default' },
+        permissions: { ask: [toolName] },
+        mcpServers: servers,
+      });
+      const env = approveWorkspaceMcpServers(cwd, servers);
+      Object.assign(
+        approvals,
+        JSON.parse(readFileSync(env['QWEN_CODE_MCP_APPROVALS_PATH']!, 'utf8')),
+      );
+    });
+    const approvalsPath = join(root, 'approved-mcp.json');
+    const trustPath = join(root, 'trusted-folders.json');
+    writeJson(approvalsPath, approvals);
+    writeJson(
+      trustPath,
+      Object.fromEntries(workspaces.map((cwd) => [cwd, 'TRUST_FOLDER'])),
+    );
+    writeJson(join(qwenHome, 'settings.json'), {
+      security: { folderTrust: { enabled: true } },
+    });
+    const fakeModel = await startFakeOpenAIServer(({ body }) => {
+      const messages = body['messages'] as Array<Record<string, unknown>>;
+      const index = messages.findLastIndex(
+        (message) => message['role'] === 'user',
+      );
+      const parts = messages[index]?.['content'];
+      const prompt =
+        typeof parts === 'string'
+          ? parts
+          : Array.isArray(parts)
+            ? parts.map((part: { text?: string }) => part.text ?? '').join('\n')
+            : '';
+      const match = prompt.match(/WRITE_E2E=(\{[^\n]*\})/u);
+      if (
+        match &&
+        !messages.slice(index + 1).some((message) => message['role'] === 'tool')
+      ) {
+        const args = JSON.parse(match[1]!) as { content: string };
+        return { toolCalls: [fakeToolCall(toolName, args)] };
+      }
+      return { content: 'WRITE_E2E_DONE' };
+    });
+    let daemon: SpawnedDaemon | undefined;
+    const attempts: Attempt[] = [];
+
+    async function session(workspaceCwd: string): Promise<Session> {
+      const deadline = Date.now() + 30000;
+      while (Date.now() < deadline) {
+        try {
+          const created = await daemon!.client.createOrAttachSession({
+            workspaceCwd,
+            sessionScope: 'thread',
+          });
+          expect(created.workspaceCwd).toBe(workspaceCwd);
+          return created;
+        } catch (error) {
+          if (
+            !(error instanceof Error) ||
+            !error.message.includes('daemon_runtime_starting')
+          )
+            throw error;
+        }
+        await new Promise((resolve) => setTimeout(resolve, 100));
+      }
+      throw new Error('Workspace runtime did not finish starting');
+    }
+
+    async function start(session: Session, content: string): Promise<Attempt> {
+      const run: Attempt = {
+        session,
+        events: [],
+        controller: new AbortController(),
+        subscription: Promise.resolve(),
+        task: Promise.resolve(),
+      };
+      attempts.push(run);
+      let notifyReady: (() => void) | undefined;
+      let notifyError: ((error: unknown) => void) | undefined;
+      const ready = new Promise<void>((resolve, reject) => {
+        notifyReady = resolve;
+        notifyError = reject;
+      });
+      run.subscription = (async () => {
+        try {
+          for await (const event of daemon!.client.subscribeEvents(
+            session.sessionId,
+            {
+              clientId: session.clientId,
+              signal: run.controller.signal,
+              onSseStreamAccepted: () => notifyReady?.(),
+            },
+          ))
+            run.events.push(event);
+        } catch (error) {
+          notifyError?.(error);
+          if (!run.controller.signal.aborted) throw error;
+        }
+      })();
+      void run.subscription.catch(() => undefined);
+      await ready;
+      run.task = daemon!.client.prompt(
+        session.sessionId,
+        {
+          prompt: [
+            { type: 'text', text: `WRITE_E2E=${JSON.stringify({ content })}` },
+          ],
+        },
+        undefined,
+        session.clientId,
+      );
+      void run.task.catch(() => undefined);
+      return run;
+    }
+
+    async function permission(run: Attempt): Promise<Permission> {
+      await vi.waitFor(
+        () =>
+          expect(
+            run.events.some((event) => event.type === 'permission_request'),
+          ).toBe(true),
+        { timeout: 30000 },
+      );
+      return run.events.find((event) => event.type === 'permission_request')!
+        .data as Permission;
+    }
+
+    async function vote(
+      run: Attempt,
+      permission: Permission,
+      kind = 'allow_once',
+    ) {
+      const option = permission.options.find((option) => option.kind === kind);
+      expect(option).toBeDefined();
+      return daemon!.client.respondToSessionPermission(
+        run.session.sessionId,
+        permission.requestId,
+        { outcome: { outcome: 'selected', optionId: option!.optionId } },
+        run.session.clientId,
+      );
+    }
+
+    try {
+      daemon = await spawnDaemon({
+        workspaceCwd: workspaces[0],
+        bootTimeoutMs: 30000,
+        env: {
+          HOME: home,
+          QWEN_HOME: qwenHome,
+          QWEN_RUNTIME_DIR: join(root, 'runtime'),
+          QWEN_CODE_TRUSTED_FOLDERS_PATH: trustPath,
+          QWEN_CODE_MCP_APPROVALS_PATH: approvalsPath,
+          QWEN_SANDBOX: 'false',
+          QWEN_CODE_NO_RELAUNCH: 'true',
+          QWEN_CODE_LEGACY_MCP_BLOCKING: '1',
+          OPENAI_API_KEY: 'fake-key',
+          OPENAI_BASE_URL: fakeModel.baseUrl,
+          OPENAI_MODEL: 'fake-model',
+          QWEN_MODEL: 'fake-model',
+          NO_PROXY: '127.0.0.1,localhost',
+          no_proxy: '127.0.0.1,localhost',
+        },
+      });
+      const a = await session(workspaces[0]!);
+      const workspaceB = await daemon.client.addWorkspace(workspaces[1]!);
+      expect(workspaceB.trusted).toBe(true);
+      const b = await session(workspaces[1]!);
+      const content = '  exact\n中文 😀 "quoted"\t FINAL  ';
+      const runA = await start(a, content);
+      const pA = await permission(runA);
+      const runB = await start(b, content);
+      const pB = await permission(runB);
+      expect(pA.toolCall.rawInput).toEqual({ content });
+      expect(pB.toolCall.rawInput).toEqual({ content });
+      expect(requests).toHaveLength(0);
+      expect(await vote(runA, pB)).toBe(false);
+      expect(requests).toHaveLength(0);
+      expect(await vote(runA, pA, 'reject_once')).toBe(true);
+      await runA.task;
+      expect(requests).toHaveLength(0);
+      expect(await vote(runB, pB)).toBe(true);
+      await runB.task;
+      expect(requests).toEqual([
+        {
+          body: {
+            messages: [{ role: 'user', content }],
+            infer: false,
+            user_id: 'scope-1',
+          },
+          authorization: 'Token synthetic-1',
+        },
+      ]);
+      expect(await vote(runB, pB)).toBe(false);
+
+      for (let index = 0; index < 2; index++) {
+        const run = await start(a, content);
+        const pending = await permission(run);
+        expect(requests).toHaveLength(index + 1);
+        expect(await vote(run, pending)).toBe(true);
+        await run.task;
+        expect(requests[index + 1]).toEqual({
+          body: {
+            messages: [{ role: 'user', content }],
+            infer: false,
+            user_id: 'scope-0',
+          },
+          authorization: 'Token synthetic-0',
+        });
+      }
+
+      configs[1]!.scope.userId = 'scope-B-reloaded';
+      writeJson(configPaths[1]!, configs[1]);
+      const restarted = await daemon.client
+        .workspaceById(workspaceB.id)
+        .restartMcpServer(serverName, {
+          clientId: b.clientId,
+          entryIndex: '*',
+          timeoutMs: 30000,
+        });
+      expect(restarted).toMatchObject({
+        serverName,
+        entries: [{ restarted: true }],
+      });
+      const reloaded = await start(b, content);
+      expect(await vote(reloaded, await permission(reloaded))).toBe(true);
+      await reloaded.task;
+      expect(requests.at(-1)).toMatchObject({
+        body: { user_id: 'scope-B-reloaded' },
+        authorization: 'Token synthetic-1',
+      });
+      const cancelled = await start(a, 'must not be submitted');
+      const stale = await permission(cancelled);
+      await daemon.client.cancel(a.sessionId, a.clientId);
+      await cancelled.task.catch(() => undefined);
+      expect(await vote(cancelled, stale)).toBe(false);
+      expect(requests).toHaveLength(4);
+    } finally {
+      for (const run of attempts) {
+        await daemon?.client
+          .cancel(run.session.sessionId, run.session.clientId)
+          .catch(() => undefined);
+        run.controller.abort();
+        await run.subscription.catch(() => undefined);
+        await run.task.catch(() => undefined);
+      }
+      await daemon?.dispose();
+      await fakeModel.close();
+      provider.closeAllConnections();
+      await new Promise<void>((resolve) => provider.close(() => resolve()));
+      rmSync(root, { recursive: true, force: true });
+    }
+  });
+});