File size: 4,682 Bytes
4e23b01 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | import { readFile, rm, mkdtemp } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { setTimeout as delay } from 'node:timers/promises';
import type { Event } from '#/index';
export interface AgentWirePayload {
readonly type: string;
readonly [key: string]: unknown;
}
export interface AgentSessionWireRecord {
readonly type: 'agent';
readonly agentId: string;
readonly event: AgentWirePayload;
}
export async function makeTempDir(tempDirs: string[], prefix: string): Promise<string> {
const dir = await mkdtemp(join(tmpdir(), prefix));
tempDirs.push(dir);
return dir;
}
export async function removeTempDirs(tempDirs: string[]): Promise<void> {
for (const dir of tempDirs.splice(0)) {
await removeTempDir(dir);
}
}
export async function waitForAgentWireEvent(
homeDir: string,
sessionId: string,
eventType: string,
predicate: (event: AgentWirePayload) => boolean = () => true,
): Promise<AgentWirePayload> {
const deadline = Date.now() + 1_000;
while (Date.now() < deadline) {
const events = await readWireEvents(homeDir, sessionId);
for (const event of events) {
const agentEvent = toMainAgentWirePayload(event);
if (agentEvent === undefined) continue;
if (agentEvent.type !== eventType) continue;
if (predicate(agentEvent)) {
return agentEvent;
}
}
await delay(10);
}
throw new Error(`Timed out waiting for ${eventType} in ${sessionId}`);
}
export function waitForSDKEvent(
session: {
onEvent(listener: (event: Event) => void): () => void;
},
predicate: (event: Event) => boolean,
timeoutMs = 1_000,
): Promise<Event> {
return new Promise((resolve, reject) => {
const timeout = setTimeout(() => {
unsubscribe();
reject(new Error('Timed out waiting for session event'));
}, timeoutMs);
const unsubscribe = session.onEvent((event) => {
if (!predicate(event)) return;
clearTimeout(timeout);
unsubscribe();
resolve(event);
});
});
}
async function readWireEvents(homeDir: string, sessionId: string): Promise<readonly unknown[]> {
const sessionDir = await readIndexedSessionDir(homeDir, sessionId);
if (sessionDir === undefined) return [];
try {
const raw = await readFile(join(sessionDir, 'agents', 'main', 'wire.jsonl'), 'utf-8');
return raw
.split('\n')
.filter(Boolean)
.map((line) => JSON.parse(line) as unknown);
} catch (error) {
const code = (error as NodeJS.ErrnoException).code;
if (code === 'ENOENT') return [];
throw error;
}
}
function toMainAgentWirePayload(value: unknown): AgentWirePayload | undefined {
if (isAgentWirePayload(value)) return value;
if (!isAgentSessionWireRecord(value)) return undefined;
if (value.agentId !== 'main') return undefined;
return value.event;
}
async function readIndexedSessionDir(
homeDir: string,
sessionId: string,
): Promise<string | undefined> {
let raw: string;
try {
raw = await readFile(join(homeDir, 'session_index.jsonl'), 'utf-8');
} catch (error) {
const code = (error as NodeJS.ErrnoException).code;
if (code === 'ENOENT') return undefined;
throw error;
}
let sessionDir: string | undefined;
for (const line of raw.split(/\r?\n/)) {
if (line.trim().length === 0) continue;
let parsed: unknown;
try {
parsed = JSON.parse(line) as unknown;
} catch {
continue;
}
if (!isRecord(parsed)) continue;
if (parsed['sessionId'] !== sessionId) continue;
if (typeof parsed['sessionDir'] !== 'string') continue;
sessionDir = parsed['sessionDir'];
}
return sessionDir;
}
function isAgentSessionWireRecord(value: unknown): value is AgentSessionWireRecord {
if (!isRecord(value)) return false;
if (value['type'] !== 'agent') return false;
if (typeof value['agentId'] !== 'string') return false;
return isAgentWirePayload(value['event']);
}
function isAgentWirePayload(value: unknown): value is AgentWirePayload {
return isRecord(value) && typeof value['type'] === 'string';
}
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null;
}
async function removeTempDir(dir: string): Promise<void> {
for (let attempt = 0; attempt < 10; attempt += 1) {
try {
await rm(dir, { recursive: true, force: true });
return;
} catch (error) {
const code = (error as NodeJS.ErrnoException).code;
if (code !== 'ENOTEMPTY' && code !== 'EBUSY' && code !== 'EPERM') {
throw error;
}
await delay(10);
}
}
await rm(dir, { recursive: true, force: true });
}
|