admin/ClawX
1
0
Fork 0
mirror of https://github.com/ValueCell-ai/ClawX.git synced 2026-08-14 08:53:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(cc-connect): preserve isolated reconnect state

Browse Source
This commit is contained in:
ashione
2026-07-27 17:49:17 +08:00
parent f058ccf28f
commit 2bc9d2740e
5 changed files with 182 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
electron/runtime/cc-connect-bridge-adapter.ts
-1
View File
@@ -848,7 +848,6 @@ export class CcConnectBridgeAdapter {
if (this.socket === socket) {
this.socket = null;
this.clearHeartbeat();
this.failPendingRuns('cc-connect bridge disconnected before the run completed');
this.scheduleReconnect();
}
finish(new Error('cc-connect bridge connection closed'));
electron/runtime/cc-connect-codex-transcript.ts
+27 -5
View File
@@ -7,6 +7,11 @@ const MAX_TOOL_OUTPUT_CHARS = 16_000;
const TRANSCRIPT_TURN_MATCH_WINDOW_MS = 2 * 60_000;
const MAX_TRANSCRIPT_FILE_CACHE_ENTRIES = 512;
const MAX_TRANSCRIPT_PATH_CACHE_ENTRIES = 2_048;
const MAX_FALLBACK_TURN_HINTS = 20;
const MAX_FALLBACK_DIRECTORIES = 12;
const MAX_FALLBACK_CANDIDATE_FILES = 64;
const MAX_FALLBACK_FILE_BYTES = 8 * 1024 * 1024;
const MAX_FALLBACK_TOTAL_BYTES = 32 * 1024 * 1024;
type CachedTranscriptFile = {
mtimeMs: number;
size: number;
@@ -211,20 +216,35 @@ async function findTurnTranscriptFiles(
): Promise<string[]> {
const sessionRoot = join(codexHomeDir, 'sessions');
const directories = new Map<string, string>();
for (const hint of hints) {
const recentHints = [...hints]
.sort((left, right) => right.timestamp - left.timestamp)
.slice(0, MAX_FALLBACK_TURN_HINTS);
for (const hint of recentHints) {
for (const parts of transcriptCandidateDateParts(hint.timestamp)) {
const directory = join(sessionRoot, ...parts);
directories.set(directory, directory);
if (directories.size >= MAX_FALLBACK_DIRECTORIES) break;
}
if (directories.size >= MAX_FALLBACK_DIRECTORIES) break;
}
const matches: string[] = [];
let candidateFiles = 0;
let candidateBytes = 0;
for (const directory of directories.values()) {
const entries = await readdir(directory, { withFileTypes: true }).catch(() => []);
for (const entry of entries) {
if (!entry.isFile() || !entry.name.endsWith('.jsonl')) continue;
const transcriptEntries = entries
.filter((entry) => entry.isFile() && entry.name.endsWith('.jsonl'))
.sort((left, right) => right.name.localeCompare(left.name));
for (const entry of transcriptEntries) {
if (candidateFiles >= MAX_FALLBACK_CANDIDATE_FILES) return matches;
candidateFiles += 1;
const path = join(directory, entry.name);
const metadata = await stat(path).catch(() => null);
if (!metadata || metadata.size > MAX_FALLBACK_FILE_BYTES) continue;
if (candidateBytes + metadata.size > MAX_FALLBACK_TOTAL_BYTES) return matches;
candidateBytes += metadata.size;
const file = await readTranscriptFile(path);
if (file?.jsonl && transcriptMatchesTurn(file, hints, expectedWorkDir)) matches.push(path);
if (file?.jsonl && transcriptMatchesTurn(file, recentHints, expectedWorkDir)) matches.push(path);
}
}
return matches;
@@ -401,7 +421,9 @@ export async function loadCcConnectCodexTranscriptTools(
MAX_TRANSCRIPT_PATH_CACHE_ENTRIES,
);
const file = await readTranscriptFile(transcriptPath);
if (file?.jsonl && transcriptMatchesWorkDir(file, expectedWorkDir)) {
const matchesPublicTurn = turnHints.length === 0
|| (file !== null && transcriptMatchesTurn(file, turnHints, expectedWorkDir));
if (file?.jsonl && transcriptMatchesWorkDir(file, expectedWorkDir) && matchesPublicTurn) {
idMatchedPaths.add(transcriptPath);
}
}
electron/runtime/cc-connect-provider.ts
+4 -7
View File
@@ -843,12 +843,9 @@ export class CcConnectRuntimeProvider extends EventEmitter implements RuntimePro
const channel = ccConnectSessionChannel(session.logicalKey);
let transcriptMessages: RawMessage[] = [];
if (channel) {
const codexHomeDirs = new Set([
this.currentProjectProfileByAgent.get(session.agentId)?.codexHomeDir,
this.currentProviderProfile?.codexHomeDir,
...this.currentProjectProfiles.map((profile) => profile.codexHomeDir),
getCcConnectCodexHomeDir(),
].filter((value): value is string => Boolean(value)));
const owningProfile = this.currentProjectProfileByAgent.get(session.agentId)
?? this.currentProviderProfile;
const codexHomeDir = owningProfile?.codexHomeDir ?? getCcConnectCodexHomeDir();
const transcriptTurnHints = publicMessages.slice(-limit).flatMap((message) => {
const content = message.role === 'user' ? runtimeMessageText(message.content) : '';
return content && typeof message.timestamp === 'number'
@@ -857,7 +854,7 @@ export class CcConnectRuntimeProvider extends EventEmitter implements RuntimePro
});
if (session.agentSessionId || transcriptTurnHints.length > 0) {
transcriptMessages = await loadCcConnectCodexTranscriptTools(
codexHomeDirs,
codexHomeDir,
session.agentSessionId ?? '',
transcriptTurnHints,
this.currentProjectWorkDirByAgent.get(session.agentId),
tests/unit/cc-connect-bridge-adapter.test.ts
+33 -41
View File
@@ -945,6 +945,7 @@ describe('cc-connect BridgePlatform adapter', () => {
const received: Record<string, unknown>[] = [];
const emitted: Array<[string, unknown]> = [];
const progressPrefix = '__cc_connect_progress_card_v1__:';
let droppedMessage: Record<string, unknown> | undefined;
server.on('connection', (socket) => {
sockets.push(socket);
@@ -953,29 +954,31 @@ describe('cc-connect BridgePlatform adapter', () => {
received.push(parsed);
if (parsed.type === 'register') {
socket.send(JSON.stringify({ type: 'register_ack', ok: true }));
if (sockets.indexOf(socket) === 1 && droppedMessage) {
setTimeout(() => {
socket.send(JSON.stringify({
type: 'reply',
session_key: droppedMessage?.session_key,
reply_ctx: droppedMessage?.reply_ctx,
content: 'original answer after reconnect',
}));
}, 10);
}
return;
}
if (parsed.type === 'message') {
if (sockets.indexOf(socket) === 0) {
socket.send(JSON.stringify({
type: 'preview_start',
ref_id: 'dropped-progress',
session_key: parsed.session_key,
reply_ctx: parsed.reply_ctx,
content: `${progressPrefix}${JSON.stringify({
version: 2,
state: 'running',
items: [{ kind: 'tool_use', tool: 'Bash', text: 'sleep 30' }],
})}`,
}));
} else {
socket.send(JSON.stringify({
type: 'reply',
session_key: parsed.session_key,
content: 'reply after dropped connection',
}));
}
}
if (parsed.type !== 'message' || sockets.indexOf(socket) !== 0) return;
droppedMessage = parsed;
socket.send(JSON.stringify({
type: 'preview_start',
ref_id: 'dropped-progress',
session_key: parsed.session_key,
reply_ctx: parsed.reply_ctx,
content: `${progressPrefix}${JSON.stringify({
version: 2,
state: 'running',
items: [{ kind: 'tool_use', tool: 'Bash', text: 'sleep 30' }],
})}`,
}));
});
});
@@ -1019,10 +1022,9 @@ describe('cc-connect BridgePlatform adapter', () => {
type: 'tool.completed',
runId: droppedRun.runId,
toolCallId: `${droppedRun.runId}:progress:0`,
isError: true,
meta: expect.objectContaining({
status: 'failed',
success: false,
status: 'completed',
success: true,
inferredFromRunCompletion: true,
}),
})],
@@ -1030,27 +1032,17 @@ describe('cc-connect BridgePlatform adapter', () => {
type: 'run.ended',
runId: droppedRun.runId,
sessionKey: 'agent:coder:feishu:chat-1:user-1',
status: 'error',
error: 'cc-connect bridge disconnected before the run completed',
})],
]));
});
const replacementRun = await adapter.send({
sessionKey: 'agent:coder:feishu:chat-1:user-1',
message: 'answer after reconnect',
idempotencyKey: 'idem-after-drop',
});
await vi.waitFor(() => {
expect(emitted).toEqual(expect.arrayContaining([
['chat:runtime-event', expect.objectContaining({
type: 'run.ended',
runId: replacementRun.runId,
sessionKey: 'agent:coder:feishu:chat-1:user-1',
status: 'completed',
})],
]));
});
expect(emitted).not.toEqual(expect.arrayContaining([
['chat:runtime-event', expect.objectContaining({
type: 'run.ended',
runId: droppedRun.runId,
status: 'error',
})],
]));
await adapter.close();
const connectionCountAfterClose = sockets.length;
tests/unit/cc-connect-runtime-provider.test.ts
+118 -5
View File
@@ -2490,11 +2490,10 @@ describe('CcConnectRuntimeProvider', () => {
it('merges Codex transcript tool calls into cc-connect channel history', async () => {
const agentSessionId = '019fa23b-1dad-76b1-9910-c47608ebf367';
const staleAgentSessionId = '019fa000-0000-7000-8000-000000000000';
const ownerCodexHome = join(tempDir, 'runtimes', 'cc-connect', 'codex-home');
const otherAgentCodexHome = join(tempDir, 'credentials', 'other-agent', 'codex-home');
const transcriptDir = join(
tempDir,
'runtimes',
'cc-connect',
'codex-home',
ownerCodexHome,
'sessions',
'2026',
'07',
@@ -2508,6 +2507,7 @@ describe('CcConnectRuntimeProvider', () => {
payload: {
id: agentSessionId,
timestamp: '2026-07-27T06:20:10.931Z',
cwd: '/tmp/workspace',
},
}),
JSON.stringify({
@@ -2539,10 +2539,42 @@ describe('CcConnectRuntimeProvider', () => {
},
}),
].join('\n'), 'utf8');
const otherTranscriptDir = join(otherAgentCodexHome, 'sessions', '2026', '07', '27');
await mkdir(otherTranscriptDir, { recursive: true });
await writeFile(join(otherTranscriptDir, `rollout-${staleAgentSessionId}.jsonl`), [
JSON.stringify({
timestamp: '2026-07-27T06:20:10.931Z',
type: 'session_meta',
payload: {
id: staleAgentSessionId,
timestamp: '2026-07-27T06:20:10.931Z',
cwd: '/tmp/workspace',
},
}),
JSON.stringify({
timestamp: '2026-07-27T06:20:10.950Z',
type: 'response_item',
payload: {
type: 'message',
role: 'user',
content: [{ type: 'input_text', text: '你在哪' }],
},
}),
JSON.stringify({
timestamp: '2026-07-27T06:20:17.195Z',
type: 'response_item',
payload: {
type: 'function_call',
name: 'exec_command',
arguments: '{"cmd":"must-not-cross-agent"}',
call_id: 'call-other-agent',
},
}),
].join('\n'), 'utf8');
const { loadCcConnectCodexTranscriptTools } = await import('@electron/runtime/cc-connect-codex-transcript');
await expect(loadCcConnectCodexTranscriptTools([
join(tempDir, 'credentials', 'current-provider', 'codex-home'),
join(tempDir, 'runtimes', 'cc-connect', 'codex-home'),
ownerCodexHome,
], agentSessionId)).resolves.toHaveLength(2);
const channelSession = {
@@ -2604,6 +2636,19 @@ describe('CcConnectRuntimeProvider', () => {
skillSyncer: vi.fn(async () => ({ skills: [] })),
providerProfileLoader: vi.fn(async () => createProviderProfile()) as never,
});
Reflect.set(provider, 'currentProjectProfileByAgent', new Map([
['project-manager', createProviderProfile({ codexHomeDir: ownerCodexHome })],
]));
Reflect.set(provider, 'currentProjectProfiles', [
createProviderProfile({ codexHomeDir: ownerCodexHome }),
createProviderProfile({ codexHomeDir: otherAgentCodexHome }),
]);
Reflect.set(provider, 'currentProviderProfile', createProviderProfile({
codexHomeDir: otherAgentCodexHome,
}));
Reflect.set(provider, 'currentProjectWorkDirByAgent', new Map([
['project-manager', '/tmp/workspace'],
]));
await expect(provider.loadHistory({
sessionKey: guiSession.logicalKey,
@@ -2640,6 +2685,10 @@ describe('CcConnectRuntimeProvider', () => {
expect.objectContaining({ id: 'channel-assistant', role: 'assistant' }),
],
});
expect(JSON.stringify(await provider.loadHistory({
sessionKey: channelSession.logicalKey,
limit: 20,
}))).not.toContain('must-not-cross-agent');
});
it('rejects stale cross-workspace session IDs and matches later Web Search and MCP turns', async () => {
@@ -2825,6 +2874,70 @@ describe('CcConnectRuntimeProvider', () => {
expect(JSON.stringify(idMatchedMessages)).not.toContain('second-call');
});
it('bounds fallback transcript scanning before reading older candidates', async () => {
const transcriptDir = join(
tempDir,
'runtimes',
'cc-connect',
'codex-home',
'sessions',
'2026',
'07',
'27',
);
await mkdir(transcriptDir, { recursive: true });
const turnTimestamp = Date.parse('2026-07-27T06:30:10.100Z');
const transcript = (content: string, callId: string) => [
JSON.stringify({
timestamp: '2026-07-27T06:00:10.000Z',
type: 'session_meta',
payload: {
timestamp: '2026-07-27T06:00:10.000Z',
cwd: '/workspace/coder',
},
}),
JSON.stringify({
timestamp: '2026-07-27T06:30:10.100Z',
type: 'response_item',
payload: {
type: 'message',
role: 'user',
content: [{ type: 'input_text', text: content }],
},
}),
JSON.stringify({
timestamp: '2026-07-27T06:30:11.000Z',
type: 'response_item',
payload: {
type: 'function_call',
call_id: callId,
name: 'exec_command',
arguments: '{"cmd":"bounded"}',
},
}),
].join('\n');
await Promise.all([
writeFile(
join(transcriptDir, 'rollout-2026-07-27T00-00-00-target.jsonl'),
transcript('bounded fallback target', 'target-call'),
'utf8',
),
...Array.from({ length: 64 }, (_, index) => writeFile(
join(transcriptDir, `rollout-2026-07-27T12-${String(index).padStart(2, '0')}-00-decoy.jsonl`),
transcript('different prompt', `decoy-${index}`),
'utf8',
)),
]);
const { loadCcConnectCodexTranscriptTools } = await import('@electron/runtime/cc-connect-codex-transcript');
await expect(loadCcConnectCodexTranscriptTools(
join(tempDir, 'runtimes', 'cc-connect', 'codex-home'),
'',
[{ content: 'bounded fallback target', timestamp: turnTimestamp }],
'/workspace/coder',
)).resolves.toEqual([]);
});
it('aborts active cc-connect chat runs through Bridge without restarting the runtime', async () => {
const binaryPath = join(tempDir, 'cc-connect');
await writeFile(binaryPath, '#!/bin/sh\n', { mode: 0o755 });