test/integration/sessionExecution.test.ts
Rama main · réplica de lectura del repositorio original.
// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
import {
cancelSessionExecution,
createSessionMessageIntent,
fetchExecutionArtifacts,
fetchSessionExecutionRoute,
fetchSessionExecutions,
readExecutionArtifact,
sendSessionExecutionNow,
} from '@/service/executionApi';
import { hydrateProjectEventStore } from '@/service/projectEventStoreHydration';
import {
createWorkspaceSessionDraft,
sessionMessageConfiguration,
submitSessionMessage,
submitWorkspaceSessionDraft,
} from '@/service/sessionMessage';
import { useAuthStore } from '@/store/authStore';
import { setConnectionConfig } from '@/store/connectionStore';
import { getProjectEventStore } from '@/store/projectEventStore';
import { useProjectRuntimeStore } from '@/store/projectRuntimeStore';
import { useProjectStore } from '@/store/projectStore';
import {
getSessionExecutionState,
observeSessionExecution,
requireLegacyExecution,
resetSessionExecutionStore,
} from '@/store/sessionExecutionStore';
import { SPACE_SCHEMA_VERSION, useSpaceStore } from '@/store/spaceStore';
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process';
import path from 'node:path';
import { createInterface } from 'node:readline';
import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
vi.mock('@/host/createHost', () => ({
createHost: () => ({
electronAPI: {
getLocalControlCapability: async () => 'synthetic-local-control',
},
ipcRenderer: null,
}),
}));
vi.mock('@/lib/notifyError', () => ({
reportError: () => 'task',
notifyError: vi.fn(),
}));
vi.mock('@/components/Toast/storageToast', () => ({
showStorageToast: vi.fn(),
}));
interface Stats {
wire: Array<{
run: string;
url: string;
body: { messages: Array<{ role: string; content: unknown }> };
}>;
counts: Record<string, number>;
finals: Array<{ run_id: string; state: string }>;
requests: Array<{ request_id: string; status: string }>;
}
interface Reply {
status: number;
headers: Record<string, string>;
body: string;
}
let child: ChildProcessWithoutNullStreams;
let sequence = 0;
const waiting = new Map<
number,
{ resolve: (value: any) => void; reject: (error: Error) => void }
>();
let stderr = '';
let loseAcknowledgement = false;
let pauseNextSave: Promise<void> | null = null;
const calls: Array<{ method: string; path: string; body?: string }> = [];
const ipc = <T = unknown>(value: Record<string, unknown>): Promise<T> =>
new Promise((resolve, reject) => {
const id = ++sequence;
waiting.set(id, { resolve, reject });
child.stdin.write(JSON.stringify({ ...value, id }) + '\n');
});
const stats = () => ipc<Stats>({ kind: 'stats' });
async function until(check: () => Promise<boolean>, timeout = 20000) {
const deadline = Date.now() + timeout;
while (Date.now() < deadline) {
if (await check()) return;
await new Promise((resolve) => setTimeout(resolve, 30));
}
throw new Error('IPC condition timed out: ' + stderr.slice(-10000));
}
const settled = (id: string) =>
until(async () =>
(await stats()).finals.some(
(row) => row.run_id === id && row.state === 'settled'
)
);
async function space(id: string, git: boolean) {
await ipc({ kind: 'space', space: id, git });
useSpaceStore.getState().upsertSpaces([
{
id,
name: id,
sourceType: 'folder',
rootPath: '/synthetic/' + id,
status: 'active',
schemaVersion: SPACE_SCHEMA_VERSION,
createdAt: 1,
updatedAt: 1,
},
]);
}
beforeAll(async () => {
const root = process.cwd();
child = spawn(
path.join(root, 'backend/.venv/bin/python'),
[
'-u',
'-B',
path.join(root, 'test/integration/fixtures/session_execution_bridge.py'),
],
{
cwd: path.join(root, 'backend'),
env: {
PATH: '/usr/bin:/bin',
PYTHONPATH: '.',
PYTHONDONTWRITEBYTECODE: '1',
PYTEST_DISABLE_PLUGIN_AUTOLOAD: '1',
GIT_CONFIG_GLOBAL: '/dev/null',
GIT_CONFIG_SYSTEM: '/dev/null',
GIT_TERMINAL_PROMPT: '0',
},
}
);
child.stderr.on('data', (data) => {
stderr += data.toString();
});
child.on('exit', (code) => {
waiting.forEach((promise) =>
promise.reject(new Error(`Bridge exited ${code}: ${stderr}`))
);
waiting.clear();
});
createInterface({ input: child.stdout }).on('line', (line) => {
if (!line.startsWith('@IPC ')) return;
const value = JSON.parse(line.slice(5));
const pending = waiting.get(value.id);
waiting.delete(value.id);
if (value.error)
pending?.reject(new Error(value.error + '\n' + stderr.slice(-5000)));
else pending?.resolve(value.result);
});
useAuthStore.setState({ token: 'synthetic-account-secret', user_id: 1 });
setConnectionConfig({
brainEndpoint: 'http://synthetic-brain',
channel: 'desktop',
});
vi.stubGlobal(
'fetch',
async (input: string | URL | Request, options: RequestInit = {}) => {
const url = new URL(String(input), 'http://synthetic-account');
const request = {
method: options.method ?? 'GET',
path: url.pathname + url.search,
body: options.body as string | undefined,
};
calls.push(request);
if (request.method === 'PATCH' && pauseNextSave) {
const gate = pauseNextSave;
pauseNextSave = null;
await gate;
}
const result = await ipc<Reply>({
kind: 'http',
...request,
headers: Object.fromEntries(new Headers(options.headers).entries()),
});
if (
loseAcknowledgement &&
request.method === 'POST' &&
/\/executions$/.test(request.path) &&
result.status === 202
) {
loseAcknowledgement = false;
throw new TypeError('Synthetic lost acknowledgement');
}
return new Response(Uint8Array.from(Buffer.from(result.body, 'base64')), {
status: result.status,
headers: result.headers,
});
}
);
await ipc({ kind: 'history', enabled: true });
}, 30000);
afterAll(async () => {
resetSessionExecutionStore();
if (child && child.exitCode === null) {
child.stdin.write(JSON.stringify({ kind: 'close' }) + '\n');
await new Promise<void>((resolve) => {
const timer = setTimeout(() => child.kill(), 10000);
child.once('exit', () => {
clearTimeout(timer);
resolve();
});
});
}
vi.unstubAllGlobals();
});
describe('production TS transport → real Python ASGI execution', () => {
it.each([
[false, false],
[false, true],
[true, false],
[true, true],
])(
'overlaps across Sessions; Git=%s cross-Space=%s',
async (git, cross) => {
const name = `space-${git}-${cross}`;
await space(name, git);
if (cross) await space(name + '-other', git);
const a = createWorkspaceSessionDraft(name, 'write A', 'medium');
const b = createWorkspaceSessionDraft(
cross ? name + '-other' : name,
'write B',
'medium'
);
// Lost response reuses both the Session and request, resolving GET before retry.
loseAcknowledgement = true;
await expect(submitWorkspaceSessionDraft(a)).rejects.toThrow(
'lost acknowledgement'
);
const beforeRetry = calls.length;
await submitWorkspaceSessionDraft(a);
expect(
calls
.slice(beforeRetry)
.some(
(call) => call.method === 'POST' && /\/executions$/.test(call.path)
)
).toBe(false);
await submitWorkspaceSessionDraft(b);
const scope = a.intent.scope;
const stop = observeSessionExecution(scope);
await until(
async () => getSessionExecutionState(scope).requests.length === 1
);
stop();
await ipc({ kind: 'start' });
await until(async () => {
const current = await stats();
return Boolean(
current.counts[a.intent.requestId] &&
current.counts[b.intent.requestId]
);
});
await ipc({ kind: 'release', run: a.intent.requestId });
await ipc({ kind: 'release', run: b.intent.requestId });
await settled(a.intent.requestId);
await settled(b.intent.requestId);
const remount = observeSessionExecution(scope);
await until(
async () =>
getSessionExecutionState(scope).requests[0]?.settlement === 'settled'
);
remount();
expect(
(await stats()).requests.filter(
(row) => row.request_id === a.intent.requestId
)
).toHaveLength(1);
await expect(requireLegacyExecution(scope)).rejects.toThrow(
'managed_execution_required'
);
const firstFile = (
await fetchExecutionArtifacts(scope, a.intent.requestId)
).artifacts[0];
expect(
await (
await readExecutionArtifact(
scope,
a.intent.requestId,
firstFile.artifact_id
)
).text()
).toBe(a.intent.requestId);
const first = createSessionMessageIntent(
scope,
'write next',
'follow_up'
);
const second = createSessionMessageIntent(
scope,
'write last',
'follow_up'
);
await submitSessionMessage(
first,
sessionMessageConfiguration(scope.projectId)
);
await submitSessionMessage(
second,
sessionMessageConfiguration(scope.projectId)
);
await until(async () => Boolean((await stats()).counts[first.requestId]));
expect((await stats()).counts[second.requestId]).toBeUndefined();
await ipc({ kind: 'release', run: first.requestId });
await settled(first.requestId);
await until(async () =>
Boolean((await stats()).counts[second.requestId])
);
await ipc({ kind: 'release', run: second.requestId });
await settled(second.requestId);
const lastFile = (await fetchExecutionArtifacts(scope, second.requestId))
.artifacts[0];
expect(firstFile.filename).toBe(lastFile.filename);
expect(firstFile.content_digest).not.toBe(lastFile.content_digest);
expect(
await (
await readExecutionArtifact(
scope,
a.intent.requestId,
firstFile.artifact_id
)
).text()
).toBe(a.intent.requestId);
expect(
await (
await readExecutionArtifact(
scope,
second.requestId,
lastFile.artifact_id
)
).text()
).toBe(second.requestId);
const wire = (await stats()).wire;
const request = wire.find((item) => item.run === first.requestId)!;
expect(request.url).toBe(
'https://model.example.test/authorized/v1/chat/completions'
);
expect(JSON.stringify(request.body.messages)).toContain(
'Saved ' + a.intent.requestId
);
expect(
request.body.messages.filter(
(message) =>
message.role === 'user' && message.content === 'write next'
)
).toHaveLength(1);
await hydrateProjectEventStore({
projectId: scope.projectId,
expectedAccountKey: scope.accountKey,
});
const snapshot = getProjectEventStore(scope.projectId).getSnapshot();
await hydrateProjectEventStore({
projectId: scope.projectId,
expectedAccountKey: scope.accountKey,
});
expect(getProjectEventStore(scope.projectId).getSnapshot().chat).toEqual(
snapshot.chat
);
},
60000
);
it('awaits the real settings save before registration and freezes the Run configuration', async () => {
await space('configuration-race', false);
const draft = createWorkspaceSessionDraft(
'configuration-race',
'initial settings',
'medium'
);
await submitWorkspaceSessionDraft(draft);
await until(async () =>
Boolean((await stats()).counts[draft.intent.requestId])
);
await ipc({ kind: 'release', run: draft.intent.requestId });
await settled(draft.intent.requestId);
let release!: () => void;
pauseNextSave = new Promise<void>((resolve) => {
release = resolve;
});
useProjectStore
.getState()
.setProjectThinkingEffort(draft.intent.scope.projectId, 'high');
const next = createSessionMessageIntent(
draft.intent.scope,
'new settings',
'follow_up'
);
const before = calls.length;
const submission = submitSessionMessage(
next,
sessionMessageConfiguration(draft.intent.scope.projectId)
);
await new Promise((resolve) => setTimeout(resolve, 30));
expect(calls.slice(before).some((call) => call.method === 'POST')).toBe(
false
);
expect(
(await fetchSessionExecutions(draft.intent.scope)).items
).toHaveLength(1);
release();
await submission;
expect(next.body?.envelope.thinking_effort).toBe('high');
expect(draft.intent.body?.envelope.thinking_effort).toBe('medium');
expect(next.body?.envelope.configuration_revision).not.toBe(
draft.intent.body?.envelope.configuration_revision
);
await until(async () => Boolean((await stats()).counts[next.requestId]));
await ipc({ kind: 'release', run: next.requestId });
await settled(next.requestId);
}, 30000);
it('Send now and Stop drain through settlement; cancelled pending never starts; flag never downgrades', async () => {
await space('control', false);
const draft = createWorkspaceSessionDraft('control', 'hold', 'medium');
await submitWorkspaceSessionDraft(draft);
const scope = draft.intent.scope;
await until(async () =>
Boolean((await stats()).counts[draft.intent.requestId])
);
const next = createSessionMessageIntent(
scope,
'interrupt and write',
'follow_up'
);
const cancelled = createSessionMessageIntent(
scope,
'cancel before start',
'follow_up'
);
await submitSessionMessage(
next,
sessionMessageConfiguration(scope.projectId)
);
await submitSessionMessage(
cancelled,
sessionMessageConfiguration(scope.projectId)
);
await cancelSessionExecution(scope, cancelled.requestId);
const operation = crypto.randomUUID();
await sendSessionExecutionNow(scope, next.requestId, operation);
await sendSessionExecutionNow(scope, next.requestId, operation);
await until(async () => Boolean((await stats()).counts[next.requestId]));
expect(
(await stats()).finals.find(
(row) => row.run_id === draft.intent.requestId
)?.state
).toBe('settled');
await cancelSessionExecution(scope, next.requestId);
await settled(next.requestId);
expect((await stats()).counts[cancelled.requestId]).toBeUndefined();
await ipc({ kind: 'flag', enabled: false });
expect(await fetchSessionExecutionRoute(scope)).toMatchObject({
route: 'managed',
entry_enabled: false,
});
await expect(requireLegacyExecution(scope)).rejects.toThrow(
'managed_execution_required'
);
useAuthStore.setState({ user_id: 2 });
await expect(fetchSessionExecutions(scope)).rejects.toThrow(
'account_changed'
);
useAuthStore.setState({ user_id: 1 });
await ipc({ kind: 'flag', enabled: true });
expect(
useProjectRuntimeStore.getState().getProjectById(scope.projectId)
).toBeDefined();
}, 30000);
});