mirror of
https://github.com/hansjone/dsh-im-ops.git
synced 2026-10-08 23:13:17 +08:00
8c6c31a1b6
Add shared /history parsing, read-only session access, message filtering, and bounded replies. Default to three messages, cap at five, and exclude saved history commands. Include channel integration tests, translations, command help, and the Issue #62 implementation plan.
352 lines
14 KiB
JavaScript
352 lines
14 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import test from 'node:test';
|
|
import { setTimeout as delay } from 'node:timers/promises';
|
|
|
|
import {
|
|
HarnessClient,
|
|
HarnessHealthError,
|
|
HarnessInteractionError,
|
|
HarnessRpcError,
|
|
HarnessTransportError,
|
|
} from '../../../src/channels/shared/harness-client.mjs';
|
|
|
|
function localClient(apiProxy, options = {}) {
|
|
return new HarnessClient({
|
|
apiProxy,
|
|
workspace: '/tmp/dsh-im-test',
|
|
interactionReconnectDelayMs: 0,
|
|
fetchImpl: () => assert.fail('in-process calls must not use HTTP'),
|
|
createWebSocket: () => assert.fail('in-process events must not use WebSocket'),
|
|
...options,
|
|
});
|
|
}
|
|
|
|
async function eventually(predicate) {
|
|
for (let i = 0; i < 200; i += 1) {
|
|
if (predicate()) return;
|
|
await delay(5);
|
|
}
|
|
assert.fail('condition was not met');
|
|
}
|
|
|
|
// The real Host registers eagerly, then exposes a signal-controlled iterator.
|
|
// Reopening replays pending questions; since is deliberately not used.
|
|
function hostFixture() {
|
|
const streams = new Set();
|
|
const pending = new Map();
|
|
const history = [];
|
|
const prompts = [];
|
|
const responses = [];
|
|
let historyReads = 0;
|
|
let opened = 0;
|
|
const host = { streams, pending, history, prompts, responses, onPrompt: () => {} };
|
|
const success = (rpcId, value) => ({ rpcId, result: { ok: true, value } });
|
|
host.emit = (frame) => { for (const stream of streams) stream.push(frame); };
|
|
host.append = (event) => {
|
|
history.push({ event });
|
|
host.emit({ rpcId: `event-${event.seq}`, payload: { type: 'session/event', sessionId: 'session', event } });
|
|
};
|
|
host.question = () => {
|
|
const frame = {
|
|
rpcId: 'question-rpc',
|
|
payload: { type: 'question/requested', sessionId: 'session', questions: [] },
|
|
};
|
|
pending.set(frame.rpcId, frame);
|
|
host.emit(frame);
|
|
};
|
|
host.apiProxy = {
|
|
host: { describe: ({ rpcId }) => success(rpcId, {}) },
|
|
sessions: {
|
|
history: ({ rpcId }) => {
|
|
historyReads += 1;
|
|
return success(rpcId, { events: [...history] });
|
|
},
|
|
prompt: ({ rpcId, payload }) => {
|
|
assert.equal(streams.size > 0, true, 'subscribe before submitting a prompt');
|
|
prompts.push({ rpcId, payload });
|
|
host.onPrompt(rpcId, payload);
|
|
return success(rpcId, {});
|
|
},
|
|
},
|
|
events: {
|
|
mux(request, signal) {
|
|
assert.deepEqual(request.payload, {});
|
|
opened += 1;
|
|
const queue = [...pending.values()];
|
|
let wake;
|
|
let ended = false;
|
|
const stream = {
|
|
push(frame) { queue.push(frame); wake?.(); },
|
|
end() { ended = true; wake?.(); },
|
|
};
|
|
streams.add(stream);
|
|
return (async function* () {
|
|
const abort = () => stream.end();
|
|
signal.addEventListener('abort', abort, { once: true });
|
|
try {
|
|
while (!ended && !signal.aborted) {
|
|
if (queue.length) yield queue.shift();
|
|
else await new Promise((resolve) => { wake = resolve; });
|
|
}
|
|
} finally {
|
|
signal.removeEventListener('abort', abort);
|
|
streams.delete(stream);
|
|
}
|
|
})();
|
|
},
|
|
},
|
|
respond(envelope) {
|
|
responses.push(envelope);
|
|
pending.delete(envelope.rpcId);
|
|
return { accepted: true };
|
|
},
|
|
};
|
|
Object.defineProperties(host, {
|
|
historyReads: { get: () => historyReads },
|
|
opened: { get: () => opened },
|
|
});
|
|
return host;
|
|
}
|
|
|
|
test('in-process RPC preserves IDs, payloads, namespace receivers and errors', async () => {
|
|
const calls = [];
|
|
const apiProxy = {};
|
|
const methods = ['host.describe', 'workspace.list', 'workspace.create', 'session.list',
|
|
'session.create', 'session.history', 'session.prompt', 'session.cancel',
|
|
'session.models', 'session.selectModel', 'llm.models'];
|
|
for (const method of methods) {
|
|
const [domain, action] = method.split('.');
|
|
const namespace = domain === 'session' ? 'sessions' : domain;
|
|
const api = apiProxy[namespace] ??= {};
|
|
api[action] = function (request, signal) {
|
|
assert.equal(this, api);
|
|
assert.ok(signal instanceof AbortSignal);
|
|
calls.push({ method, request });
|
|
return { rpcId: request.rpcId, result: { ok: true, value: method } };
|
|
};
|
|
}
|
|
const client = localClient(apiProxy);
|
|
const payload = { sessionId: 'session', content: [{ type: 'text', text: 'hello' }] };
|
|
for (const method of methods) {
|
|
assert.equal(await client.rpc(method, payload, 1000, { rpcId: `fixed-${method}` }), method);
|
|
}
|
|
assert.deepEqual(calls, methods.map((method) => ({
|
|
method, request: { rpcId: `fixed-${method}`, payload },
|
|
})));
|
|
|
|
apiProxy.sessions.prompt = ({ rpcId }) => ({
|
|
rpcId, result: { ok: false, error: { code: 'queue-full', message: 'queue full' } },
|
|
});
|
|
await assert.rejects(client.rpc('session.prompt'), (error) => (
|
|
error instanceof HarnessRpcError && error.code === 'queue-full'
|
|
));
|
|
apiProxy.host.describe = ({ rpcId }) => ({ rpcId, result: { ok: false, error: { code: 'internal' } } });
|
|
await assert.rejects(client.health(), HarnessHealthError);
|
|
for (const response of [null, { rpcId: 'wrong' }, { rpcId: 'fixed', result: {} }]) {
|
|
apiProxy.host.describe = () => response;
|
|
await assert.rejects(client.rpc('host.describe', {}, 1000, { rpcId: 'fixed' }), (error) => (
|
|
error instanceof HarnessTransportError && error.code === 'harness-response-invalid'
|
|
));
|
|
}
|
|
await assert.rejects(client.rpc('session.missing'), { code: 'harness-api-not-found' });
|
|
});
|
|
|
|
test('in-process RPC bounds waits without retrying accepted calls or falling back to HTTP', async () => {
|
|
let calls = 0;
|
|
const client = localClient({
|
|
sessions: {
|
|
prompt: async ({ rpcId }) => {
|
|
calls += 1;
|
|
await delay(40); // Deliberately ignores the caller's AbortSignal.
|
|
return { rpcId, result: { ok: true, value: {} } };
|
|
},
|
|
},
|
|
});
|
|
await assert.rejects(client.rpc('session.prompt', {}, 2), { code: 'harness-timeout' });
|
|
assert.equal(calls, 1);
|
|
const controller = new AbortController();
|
|
const reason = new Error('caller cancelled');
|
|
const pending = client.rpc('session.prompt', {}, 1000, { signal: controller.signal });
|
|
await eventually(() => calls === 2);
|
|
controller.abort(reason);
|
|
await assert.rejects(pending, (error) => error === reason);
|
|
await assert.rejects(client.rpc('session.prompt', {}, 1000, { signal: controller.signal }), (error) => error === reason);
|
|
await delay(50);
|
|
assert.equal(calls, 2, 'late success must not trigger another prompt');
|
|
});
|
|
|
|
test('explicit baseUrl still selects the existing HTTP transport', async () => {
|
|
let requests = 0;
|
|
const client = localClient({ get host() { assert.fail('must not call local Host'); } }, {
|
|
baseUrl: 'http://127.0.0.1:1234',
|
|
fetchImpl: async (url, options) => {
|
|
requests += 1;
|
|
assert.equal(url.href, 'http://127.0.0.1:1234/api/host.describe');
|
|
return { ok: true, json: async () => ({
|
|
type: 'server-response', rpcId: JSON.parse(options.body).rpcId, result: { ok: true, value: {} },
|
|
}) };
|
|
},
|
|
});
|
|
assert.equal(await client.health(), true);
|
|
assert.equal(requests, 1);
|
|
});
|
|
|
|
test('history reading uses only the existing read RPC in both Host connection modes', async () => {
|
|
const page = { events: [], hasMore: false };
|
|
for (const mode of ['local', 'http']) {
|
|
const calls = [];
|
|
const client = localClient({
|
|
sessions: {
|
|
history({ rpcId, payload }, signal) {
|
|
assert.equal(mode, 'local');
|
|
assert.ok(signal instanceof AbortSignal);
|
|
calls.push(payload);
|
|
return { rpcId, result: { ok: true, value: page } };
|
|
},
|
|
},
|
|
}, mode === 'http' ? {
|
|
baseUrl: 'http://127.0.0.1:1234',
|
|
fetchImpl: async (url, options) => {
|
|
assert.equal(url.pathname, '/api/session.history');
|
|
const request = JSON.parse(options.body);
|
|
calls.push(request.payload);
|
|
return { ok: true, json: async () => ({
|
|
type: 'server-response', rpcId: request.rpcId, result: { ok: true, value: page },
|
|
}) };
|
|
},
|
|
} : {});
|
|
assert.deepEqual(await client.readSessionHistory('cold-session'), page);
|
|
assert.deepEqual(await client.readSessionHistory('cold-session', { beforeSeq: 40, maxMessages: 10 }), page);
|
|
assert.deepEqual(calls, [
|
|
{ sessionId: 'cold-session', maxMessages: 50 },
|
|
{ sessionId: 'cold-session', beforeSeq: 40, maxMessages: 10 },
|
|
]);
|
|
await assert.rejects(client.readSessionHistory('', {}), TypeError);
|
|
await assert.rejects(client.readSessionHistory('cold-session', { maxMessages: 0 }), TypeError);
|
|
await assert.rejects(client.readSessionHistory('cold-session', { beforeSeq: -1 }), TypeError);
|
|
assert.equal(calls.length, 2, 'invalid input must not make another RPC');
|
|
}
|
|
});
|
|
|
|
test('history reads preserve cancellation, timeout and missing-session errors without retrying', async () => {
|
|
let calls = 0;
|
|
const client = localClient({ sessions: {
|
|
async history({ rpcId, payload }) {
|
|
calls += 1;
|
|
if (payload.sessionId === 'missing') {
|
|
return { rpcId, result: { ok: false, error: { code: 'session-not-found', message: 'missing' } } };
|
|
}
|
|
await delay(30);
|
|
return { rpcId, result: { ok: true, value: { events: [], hasMore: false } } };
|
|
},
|
|
} });
|
|
await assert.rejects(client.readSessionHistory('missing'), { code: 'session-not-found' });
|
|
await assert.rejects(client.readSessionHistory('slow', { timeoutMs: 2 }), { code: 'harness-timeout' });
|
|
const signal = AbortSignal.abort(new DOMException('Cancelled', 'AbortError'));
|
|
await assert.rejects(client.readSessionHistory('cancelled', { signal }), { name: 'AbortError' });
|
|
assert.equal(calls, 2);
|
|
});
|
|
|
|
test('in-process interaction responses preserve the full envelope and rejection receipts', async () => {
|
|
let received;
|
|
let receipt = { accepted: true };
|
|
const client = localClient({ respond: (envelope) => { received = envelope; return receipt; } });
|
|
const result = { ok: true, value: { sessionId: 'session', answer: { answers: [] } } };
|
|
assert.deepEqual(await client.respondInteraction('question', result), receipt);
|
|
assert.deepEqual(received, { type: 'client-response', rpcId: 'question', result });
|
|
for (const reason of ['not-pending', 'bad-response']) {
|
|
receipt = { accepted: false, reason };
|
|
await assert.rejects(client.respondInteraction('question', result), (error) => (
|
|
error instanceof HarnessInteractionError && error.code === `interaction-${reason}`
|
|
));
|
|
}
|
|
receipt = {};
|
|
await assert.rejects(client.respondInteraction('question', result), /invalid interaction response receipt/);
|
|
});
|
|
|
|
test('ask uses an initially empty in-process mux and correlates replies with its original prompt ID', async () => {
|
|
const host = hostFixture();
|
|
host.onPrompt = (rpcId) => {
|
|
host.append({ type: 'turn/start', seq: 0, data: { turn: 1 } });
|
|
host.append({ type: 'user/message', seq: 1, data: { turn: 1, source: { rpcId } } });
|
|
host.append({ type: 'assistant/message', seq: 2, data: {
|
|
turn: 1, step: 1, message: { content: [{ type: 'text', text: 'answer' }] },
|
|
} });
|
|
host.append({ type: 'turn/end', seq: 3, data: { turn: 1, reason: { kind: 'completed' } } });
|
|
};
|
|
const client = localClient(host.apiProxy, { rpcIdPrefix: 'local-test' });
|
|
assert.equal(await client.ask('session', 'hello', { onInteraction: () => {}, timeoutMs: 1000 }), 'answer');
|
|
assert.equal(host.prompts.length, 1);
|
|
assert.match(host.prompts[0].rpcId, /^local-test-/);
|
|
assert.equal(host.prompts[0].payload.mode, 'queue');
|
|
assert.equal(host.streams.size, 0, 'ask completion must dispose its mux subscription');
|
|
});
|
|
|
|
test('clients on one Host share interaction ownership across context wrappers and reconnect safely', async () => {
|
|
const host = hostFixture();
|
|
const scope = {};
|
|
const abort = new AbortController();
|
|
const received = [[], []];
|
|
const clients = [0, 1].map(() => localClient({ ...host.apiProxy }, { interactionScope: scope }));
|
|
host.onPrompt = (rpcId) => {
|
|
if (host.prompts.length !== 1) return; // Second queued prompt has not started a Turn.
|
|
host.append({ type: 'turn/start', seq: 0, data: { turn: 1 } });
|
|
host.append({ type: 'user/message', seq: 1, data: { turn: 1, source: { rpcId } } });
|
|
};
|
|
const asks = [];
|
|
try {
|
|
for (let index = 0; index < clients.length; index += 1) {
|
|
const ask = clients[index].ask('session', 'hello', {
|
|
signal: abort.signal,
|
|
onInteraction: (interaction) => { received[index].push(interaction); },
|
|
});
|
|
asks.push(ask.catch(() => {}));
|
|
await eventually(() => host.prompts.length === index + 1);
|
|
}
|
|
host.question();
|
|
await eventually(() => received[0].length === 1);
|
|
assert.equal(received[1].length, 0, 'queued client must not claim the active client question');
|
|
const historyReads = host.historyReads;
|
|
received[0][0].reconnect();
|
|
await eventually(() => received[0].length === 2);
|
|
assert.ok(host.historyReads > historyReads, 'reconnect must refresh ownership history');
|
|
assert.equal(received[0][1].rpcId, 'question-rpc', 'replay keeps the pending request ID');
|
|
assert.equal(received[1].length, 0);
|
|
await received[0][1].respond({ ok: true, value: { sessionId: 'session', answer: { answers: [] } } });
|
|
assert.equal(host.responses.length, 1);
|
|
assert.equal(host.responses[0].rpcId, 'question-rpc');
|
|
} finally {
|
|
abort.abort();
|
|
await Promise.all(asks);
|
|
}
|
|
assert.equal(host.streams.size, 0);
|
|
});
|
|
|
|
test('different Hosts never share interaction claims, even with identical session/request IDs', async () => {
|
|
const hosts = [hostFixture(), hostFixture()];
|
|
const controllers = hosts.map(() => new AbortController());
|
|
const received = [[], []];
|
|
const asks = hosts.map((host, index) => {
|
|
host.onPrompt = (rpcId) => {
|
|
host.append({ type: 'turn/start', seq: 0, data: { turn: 1 } });
|
|
host.append({ type: 'user/message', seq: 1, data: { turn: 1, source: { rpcId } } });
|
|
};
|
|
return localClient(host.apiProxy).ask('session', 'hello', {
|
|
signal: controllers[index].signal,
|
|
onInteraction: (interaction) => { received[index].push(interaction); },
|
|
}).catch(() => {});
|
|
});
|
|
try {
|
|
await eventually(() => hosts.every((host) => host.prompts.length === 1));
|
|
hosts.forEach((host) => host.question());
|
|
await eventually(() => received.every((items) => items.length === 1));
|
|
controllers[0].abort();
|
|
await asks[0];
|
|
assert.equal(hosts[0].streams.size, 0);
|
|
assert.equal(hosts[1].streams.size, 1, 'closing one Host client must not stop another');
|
|
} finally {
|
|
controllers.forEach((controller) => controller.abort());
|
|
await Promise.all(asks);
|
|
}
|
|
assert.ok(hosts.every((host) => host.streams.size === 0));
|
|
});
|