mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-01 14:12:51 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
003ce4e248 | ||
|
|
a208ccda17 | ||
|
|
c5aeca3fc4 |
@@ -74,10 +74,7 @@ registrations remain inspectable; a changed destination requires explicit identi
|
||||
|
||||
Both daemons advertise `connectedHosts`; the shared link additionally requires `hostMultiplexing`.
|
||||
A completes capability and pinned-identity checks on each physical recovery before admitting channels. There is no local fallback. An older daemon continues to serve existing local
|
||||
operations; requesting this feature gives update guidance. The registry retries an older destination
|
||||
with capped backoff, so it comes online once updated. Self, duplicate, changed-identity, and
|
||||
authentication failures stay in `error` until an update changes the connection, `enabled`, or
|
||||
`replaceIdentity`, or A restarts.
|
||||
operations; requesting this feature gives update guidance.
|
||||
|
||||
Disposing a routed API releases its observations and session. It does not remove the saved host or
|
||||
affect sibling API scopes. Disposing the origin cascades to its derived scopes. Observations restore
|
||||
|
||||
@@ -3,7 +3,7 @@ import {
|
||||
composerLocator,
|
||||
expectComposerDraft,
|
||||
expectComposerVisible,
|
||||
submitMessage,
|
||||
fillComposerDraft,
|
||||
typeIntoFocusedComposer,
|
||||
} from "../support/helpers/composer";
|
||||
import { openAgentRoute, seedMockAgentWorkspace } from "../support/helpers/mock-agent";
|
||||
@@ -12,8 +12,14 @@ function userMessages(page: Page) {
|
||||
return page.getByTestId("user-message");
|
||||
}
|
||||
|
||||
async function sendDraftWhenReady(page: Page) {
|
||||
await expect(page.getByRole("button", { name: "Send message", exact: true })).toBeEnabled();
|
||||
await pressInComposer(page, "Enter");
|
||||
}
|
||||
|
||||
async function sendAndWaitForMessage(page: Page, text: string, expectedCount: number) {
|
||||
await submitMessage(page, text);
|
||||
await fillComposerDraft(page, text);
|
||||
await sendDraftWhenReady(page);
|
||||
await expect(userMessages(page)).toHaveCount(expectedCount, { timeout: 30_000 });
|
||||
await expectComposerDraft(page, "");
|
||||
}
|
||||
@@ -49,7 +55,7 @@ test.describe("Composer recall", () => {
|
||||
await pressInComposer(page, "ArrowUp");
|
||||
await expectComposerDraft(page, "add tests for login");
|
||||
|
||||
await pressInComposer(page, "Enter");
|
||||
await sendDraftWhenReady(page);
|
||||
await expect(userMessages(page)).toHaveCount(3, { timeout: 30_000 });
|
||||
await expect(userMessages(page).last()).toContainText("add tests for login");
|
||||
} finally {
|
||||
|
||||
@@ -717,13 +717,11 @@ describe("terminal emulator runtime in a real browser", () => {
|
||||
{ name: "DA1-zero", bytes: "\x1b[0c" },
|
||||
{ name: "DA2", bytes: "\x1b[>c" },
|
||||
{ name: "DA3", bytes: "\x1b[=c" },
|
||||
{ name: "XTVERSION", bytes: "\x1b[>q" },
|
||||
{ name: "DSR-5", bytes: "\x1b[5n" },
|
||||
{ name: "DSR-6", bytes: "\x1b[6n" },
|
||||
{ name: "DSR-?6", bytes: "\x1b[?6n" },
|
||||
{ name: "DECRQM", bytes: "\x1b[1$p" },
|
||||
{ name: "DECRQM-?", bytes: "\x1b[?1$p" },
|
||||
{ name: "DECRQSS", bytes: "\x1bP$qm\x1b\\" },
|
||||
{ name: "OSC-10-foreground-color", bytes: "\x1b]10;?\x07" },
|
||||
{ name: "OSC-11-background-color", bytes: "\x1b]11;?\x07" },
|
||||
{ name: "OSC-12-cursor-color", bytes: "\x1b]12;?\x07" },
|
||||
|
||||
@@ -647,7 +647,6 @@ export class TerminalEmulatorRuntime {
|
||||
terminal.parser.registerCsiHandler({ final: "c" }, () => true);
|
||||
terminal.parser.registerCsiHandler({ prefix: ">", final: "c" }, () => true);
|
||||
terminal.parser.registerCsiHandler({ prefix: "=", final: "c" }, () => true);
|
||||
terminal.parser.registerCsiHandler({ prefix: ">", final: "q" }, () => true);
|
||||
terminal.parser.registerCsiHandler({ final: "n" }, () => true);
|
||||
terminal.parser.registerCsiHandler({ prefix: "?", final: "n" }, () => true);
|
||||
terminal.parser.registerCsiHandler({ final: "R" }, () => true);
|
||||
@@ -656,7 +655,6 @@ export class TerminalEmulatorRuntime {
|
||||
{ prefix: "?", intermediates: "$", final: "p" },
|
||||
() => true,
|
||||
);
|
||||
terminal.parser.registerDcsHandler({ intermediates: "$", final: "q" }, () => true);
|
||||
for (const code of [10, 11, 12]) {
|
||||
terminal.parser.registerOscHandler(code, (data) => data.trim() === "?");
|
||||
}
|
||||
|
||||
@@ -204,16 +204,15 @@ export function createHostTransportFactory(
|
||||
}
|
||||
export function createHostClient(
|
||||
connection: string,
|
||||
config: Omit<DaemonClientConfig, "url"> & HostTransportOptions,
|
||||
config: Omit<DaemonClientConfig, "url" | "transportFactory"> & HostTransportOptions,
|
||||
budget?: HostTransportBudget,
|
||||
): DaemonClient {
|
||||
const target = parseHostTarget(connection);
|
||||
const { webSocketAgent, transportFactory, ...clientConfig } = config;
|
||||
const { webSocketAgent, ...clientConfig } = config;
|
||||
return new DaemonClient({
|
||||
...clientConfig,
|
||||
url: "ws://host-connector/ws",
|
||||
...(target.password ? { password: target.password } : {}),
|
||||
transportFactory:
|
||||
transportFactory ?? createHostTransportFactory(connection, budget, { webSocketAgent }),
|
||||
transportFactory: createHostTransportFactory(connection, budget, { webSocketAgent }),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { constants as bufferConstants } from "node:buffer";
|
||||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import * as path from "node:path";
|
||||
@@ -236,4 +237,64 @@ describe("importing a renamed Claude session", () => {
|
||||
|
||||
await expect(importSession(sessionId)).rejects.toThrow();
|
||||
});
|
||||
|
||||
test("reads the rename from a transcript too large to hold as one string", async () => {
|
||||
const sessionId = "oversized-transcript";
|
||||
const projectDir = claudeProjectDirSync(SESSION_CWD, { configDir });
|
||||
await fs.mkdir(projectDir, { recursive: true });
|
||||
const transcript = path.join(projectDir, `${sessionId}.jsonl`);
|
||||
// A hole stands in for the long session this used to fail on: the file only
|
||||
// has to be longer than a string can be, and the rename sits at its end.
|
||||
await fs.writeFile(transcript, "");
|
||||
await fs.truncate(transcript, bufferConstants.MAX_STRING_LENGTH + 1);
|
||||
await fs.appendFile(
|
||||
transcript,
|
||||
`\n${[
|
||||
userRecord({ sessionId, content: "Trace the slow query" }),
|
||||
{ type: "custom-title", customTitle: "Slow query hunt", sessionId },
|
||||
]
|
||||
.map((record) => JSON.stringify(record))
|
||||
.join("\n")}\n`,
|
||||
);
|
||||
|
||||
const imported = await importSession(sessionId);
|
||||
|
||||
expect(imported.config.title).toBe("Slow query hunt");
|
||||
});
|
||||
|
||||
test("reads the rename from behind more records than one read covers", async () => {
|
||||
const sessionId = "late-records-rename";
|
||||
await writeSession({
|
||||
sessionId,
|
||||
records: [
|
||||
userRecord({ sessionId, content: "Review this project" }),
|
||||
{ type: "custom-title", customTitle: "Scrollback review", sessionId },
|
||||
// One record longer than a single read, then enough records to leave the
|
||||
// rename several reads back from the end of the file.
|
||||
userRecord({ sessionId, content: "x".repeat(128 * 1024) }),
|
||||
...Array.from({ length: 64 }, (_, index) =>
|
||||
userRecord({ sessionId, content: `${"y".repeat(4 * 1024)} ${index}` }),
|
||||
),
|
||||
],
|
||||
});
|
||||
|
||||
const imported = await importSession(sessionId);
|
||||
|
||||
expect(imported.config.title).toBe("Scrollback review");
|
||||
});
|
||||
|
||||
test("imports a large record after the last rename without repeatedly copying it", async () => {
|
||||
const sessionId = "large-record-after-rename";
|
||||
await writeSession({
|
||||
sessionId,
|
||||
records: [
|
||||
{ type: "custom-title", customTitle: "Large attachment review", sessionId },
|
||||
userRecord({ sessionId, content: "x".repeat(96 * 1024 * 1024) }),
|
||||
],
|
||||
});
|
||||
|
||||
const imported = await importSession(sessionId);
|
||||
|
||||
expect(imported.config.title).toBe("Large attachment review");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,48 +0,0 @@
|
||||
import pino from "pino";
|
||||
import { describe, expect, test } from "vitest";
|
||||
|
||||
import { ClaudeAgentClient } from "./agent.js";
|
||||
|
||||
describe("Claude runtime info before a turn", () => {
|
||||
const client = new ClaudeAgentClient({ logger: pino({ level: "silent" }) });
|
||||
|
||||
test("reports the explicitly configured model before the first turn", async () => {
|
||||
const session = await client.createSession({
|
||||
provider: "claude",
|
||||
cwd: process.cwd(),
|
||||
model: "claude-haiku-5-5",
|
||||
});
|
||||
try {
|
||||
expect(await session.getRuntimeInfo()).toMatchObject({ model: "claude-haiku-5-5" });
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("reports the persisted configured model after rebuilding an idle session", async () => {
|
||||
const session = await client.resumeSession({
|
||||
provider: "claude",
|
||||
sessionId: "runtime-model-before-turn",
|
||||
metadata: { cwd: process.cwd(), model: "claude-haiku-5-5" },
|
||||
});
|
||||
try {
|
||||
expect(await session.getRuntimeInfo()).toMatchObject({
|
||||
sessionId: "runtime-model-before-turn",
|
||||
model: "claude-haiku-5-5",
|
||||
});
|
||||
expect((await session.getRuntimeInfo()).model).toBe("claude-haiku-5-5");
|
||||
expect((await session.getRuntimeInfo()).extra?.runtimeModel).toBeUndefined();
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("leaves the model unknown when Claude will choose its default", async () => {
|
||||
const session = await client.createSession({ provider: "claude", cwd: process.cwd() });
|
||||
try {
|
||||
expect((await session.getRuntimeInfo()).model).toBeNull();
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -2460,7 +2460,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
const info: AgentRuntimeInfo = {
|
||||
provider: "claude",
|
||||
sessionId: this.claudeSessionId,
|
||||
model: this.lastOptionsModel ?? this.config.model ?? null,
|
||||
model: this.lastOptionsModel,
|
||||
modeId: this.currentMode ?? null,
|
||||
...(this.lastRuntimeModel
|
||||
? {
|
||||
@@ -6853,40 +6853,79 @@ function normalizeClaudeSessionTitle(title: string | null): string | null {
|
||||
}
|
||||
|
||||
const CLAUDE_CUSTOM_TITLE_TYPE = "custom-title";
|
||||
const CLAUDE_CUSTOM_TITLE_NEEDLE = Buffer.from(`"${CLAUDE_CUSTOM_TITLE_TYPE}"`);
|
||||
const CLAUDE_TITLE_SCAN_CHUNK_BYTES = 64 * 1024;
|
||||
|
||||
function readClaudeCustomTitleFromRecord(record: Buffer): string | null {
|
||||
// The needle also matches a prompt that quotes it, so a record still has to
|
||||
// parse as a `custom-title` entry carrying a usable name. Records without the
|
||||
// needle are never decoded, so a long one costs a byte search and nothing more.
|
||||
if (!record.includes(CLAUDE_CUSTOM_TITLE_NEEDLE)) {
|
||||
return null;
|
||||
}
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(record.toString("utf8").trim());
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
const entry = toObjectRecord(parsed);
|
||||
if (!entry || entry.isSidechain || entry.type !== CLAUDE_CUSTOM_TITLE_TYPE) {
|
||||
return null;
|
||||
}
|
||||
return normalizeClaudeSessionTitle(
|
||||
typeof entry.customTitle === "string" ? entry.customTitle : null,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* The `custom-title` record is re-emitted every time the session is saved, so
|
||||
* the current name is the last usable one in the file. Searching backwards finds it without
|
||||
* walking the transcript, which matters because sessions reach tens of
|
||||
* megabytes.
|
||||
* the current name is the last usable one in the file. Reading backwards a chunk
|
||||
* at a time finds it without walking the transcript, which matters because
|
||||
* sessions reach tens of megabytes, and never holds the file as one string, so a
|
||||
* transcript past V8's maximum string length still imports.
|
||||
*/
|
||||
function readLastClaudeCustomTitle(content: string): string | null {
|
||||
const needle = `"${CLAUDE_CUSTOM_TITLE_TYPE}"`;
|
||||
let at = content.lastIndexOf(needle);
|
||||
while (at >= 0) {
|
||||
const lineStart = content.lastIndexOf("\n", at) + 1;
|
||||
const lineEnd = content.indexOf("\n", at);
|
||||
const line = lineEnd === -1 ? content.slice(lineStart) : content.slice(lineStart, lineEnd);
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(line.trim());
|
||||
} catch {
|
||||
parsed = undefined;
|
||||
}
|
||||
const entry = toObjectRecord(parsed);
|
||||
if (entry && !entry.isSidechain && entry.type === CLAUDE_CUSTOM_TITLE_TYPE) {
|
||||
const title = normalizeClaudeSessionTitle(
|
||||
typeof entry.customTitle === "string" ? entry.customTitle : null,
|
||||
async function readLastClaudeCustomTitle(filePath: string): Promise<string | null> {
|
||||
const handle = await fsPromises.open(filePath, "r");
|
||||
try {
|
||||
const stats = await handle.stat();
|
||||
if (!stats.isFile()) {
|
||||
// Opening a directory succeeds on Windows even though reading it does not,
|
||||
// so reject the way reading the transcript as a file would.
|
||||
throw Object.assign(
|
||||
new Error(`EISDIR: illegal operation on a directory, read '${filePath}'`),
|
||||
{ code: "EISDIR", path: filePath },
|
||||
);
|
||||
if (title) {
|
||||
return title;
|
||||
}
|
||||
}
|
||||
// The needle also matches a prompt that quotes it, and a record can carry no
|
||||
// usable name, so keep looking backwards for one that does.
|
||||
at = at <= 0 ? -1 : content.lastIndexOf(needle, at - 1);
|
||||
let end = stats.size;
|
||||
// Retain unfinished record pieces in reverse file order. Assemble each
|
||||
// complete record once, so a large tool/image record costs linear work.
|
||||
let pieces: Buffer[] = [];
|
||||
while (end > 0) {
|
||||
const length = Math.min(CLAUDE_TITLE_SCAN_CHUNK_BYTES, end);
|
||||
const chunk = Buffer.alloc(length);
|
||||
await handle.read(chunk, 0, length, end - length);
|
||||
let recordEnd = chunk.length;
|
||||
for (
|
||||
let at = chunk.lastIndexOf(10);
|
||||
at >= 0;
|
||||
at = at === 0 ? -1 : chunk.lastIndexOf(10, at - 1)
|
||||
) {
|
||||
pieces.push(chunk.subarray(at + 1, recordEnd));
|
||||
const record = Buffer.concat(pieces.toReversed());
|
||||
const title = readClaudeCustomTitleFromRecord(record);
|
||||
if (title) return title;
|
||||
pieces = [];
|
||||
recordEnd = at;
|
||||
}
|
||||
if (recordEnd > 0) pieces.push(chunk.subarray(0, recordEnd));
|
||||
end -= length;
|
||||
}
|
||||
// The first record in the file has no newline in front of it.
|
||||
return readClaudeCustomTitleFromRecord(Buffer.concat(pieces.toReversed()));
|
||||
} finally {
|
||||
await handle.close();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function isErrnoException(error: unknown): error is NodeJS.ErrnoException {
|
||||
@@ -6898,9 +6937,8 @@ async function readClaudeRenamedSessionTitle(input: {
|
||||
sessionId: string;
|
||||
configDir: string;
|
||||
}): Promise<string | null> {
|
||||
let content: string;
|
||||
try {
|
||||
content = await fsPromises.readFile(await claudeTranscriptPath(input), "utf8");
|
||||
return await readLastClaudeCustomTitle(await claudeTranscriptPath(input));
|
||||
} catch (error) {
|
||||
// A session with no transcript on disk has no rename to carry over. Every
|
||||
// other read failure is a real fault and stays explicit.
|
||||
@@ -6909,7 +6947,6 @@ async function readClaudeRenamedSessionTitle(input: {
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
return readLastClaudeCustomTitle(content);
|
||||
}
|
||||
|
||||
function normalizeImportablePromptPreview(text: string): string | null {
|
||||
|
||||
@@ -784,211 +784,6 @@ test("old capabilities fail before execution and authentication errors recover t
|
||||
expect((await online(paseo)).hosts[0]!.serverId).toBe(b.daemon.getServerId());
|
||||
});
|
||||
|
||||
async function updatableDestination() {
|
||||
let features: { connectedHosts: boolean; hostMultiplexing: boolean } | undefined;
|
||||
let holding = false;
|
||||
const held: (() => void)[] = [];
|
||||
let connections = 0;
|
||||
const http = createServer();
|
||||
const ws = new WebSocketServer({ server: http });
|
||||
ws.on("connection", (socket) => {
|
||||
connections++;
|
||||
socket.on("message", (data) => {
|
||||
const frame = JSON.parse(data.toString());
|
||||
if (frame.type === "ping") socket.send(JSON.stringify({ type: "pong" }));
|
||||
if (frame.type !== "hello") return;
|
||||
const reply = () =>
|
||||
socket.send(
|
||||
JSON.stringify({
|
||||
type: "session",
|
||||
message: {
|
||||
type: "status",
|
||||
payload: {
|
||||
status: "server_info",
|
||||
serverId: "srv_updated",
|
||||
hostname: "updated",
|
||||
version: "0.12.0",
|
||||
...(features ? { features } : {}),
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
if (holding) held.push(reply);
|
||||
else reply();
|
||||
});
|
||||
});
|
||||
http.listen(0, "127.0.0.1");
|
||||
await once(http, "listening");
|
||||
cleanup.push(async () => {
|
||||
for (const socket of ws.clients) socket.terminate();
|
||||
await new Promise<void>((resolve) => ws.close(() => resolve()));
|
||||
await new Promise<void>((resolve) => http.close(() => resolve()));
|
||||
});
|
||||
const address = http.address();
|
||||
if (!address || typeof address === "string") throw new Error("Missing destination port");
|
||||
return {
|
||||
port: address.port,
|
||||
connections: () => connections,
|
||||
hold: () => {
|
||||
holding = true;
|
||||
},
|
||||
update: () => {
|
||||
features = { connectedHosts: true, hostMultiplexing: true };
|
||||
holding = false;
|
||||
for (const reply of held.splice(0)) reply();
|
||||
},
|
||||
};
|
||||
}
|
||||
async function capabilityRetry(paseo: ReturnType<typeof createPaseoApi>): Promise<number> {
|
||||
let retryAt = "";
|
||||
await expect
|
||||
.poll(
|
||||
async () => {
|
||||
const row = (await paseo.hosts.list()).hosts[0];
|
||||
retryAt = row?.nextRetryAt ?? "";
|
||||
return row;
|
||||
},
|
||||
{ timeout: 5000 },
|
||||
)
|
||||
.toMatchObject({
|
||||
status: "offline",
|
||||
nextRetryAt: expect.any(String),
|
||||
error: { code: "unsupported_capability" },
|
||||
});
|
||||
return Date.parse(retryAt);
|
||||
}
|
||||
|
||||
test("a destination updated past unsupported_capability keeps its guidance while retrying and comes online, while duplicate_host stays terminal", async () => {
|
||||
const destination = await updatableDestination();
|
||||
const a = await daemon();
|
||||
const { paseo } = await api(a);
|
||||
const host = await paseo.hosts.add({ connection: `127.0.0.1:${destination.port}` });
|
||||
await capabilityRetry(paseo);
|
||||
destination.hold();
|
||||
await expect
|
||||
.poll(async () => (await paseo.hosts.list()).hosts[0], { timeout: 5000 })
|
||||
.toMatchObject({ status: "connecting", error: { code: "unsupported_capability" } });
|
||||
await expect(paseo.hosts.client("srv_updated")).rejects.toMatchObject({
|
||||
code: "unsupported_capability",
|
||||
});
|
||||
destination.update();
|
||||
expect((await online(paseo)).hosts[0]).toMatchObject({
|
||||
registrationId: host.registrationId,
|
||||
serverId: "srv_updated",
|
||||
nextRetryAt: null,
|
||||
error: null,
|
||||
});
|
||||
await paseo.hosts.add({ connection: `ws://127.0.0.1:${destination.port}/ws` });
|
||||
await expect
|
||||
.poll(async () => (await paseo.hosts.list()).hosts[1], { timeout: 5000 })
|
||||
.toMatchObject({ status: "error", nextRetryAt: null, error: { code: "duplicate_host" } });
|
||||
});
|
||||
|
||||
test("disabling or removing a host cancels its pending capability retry", async () => {
|
||||
const destination = await updatableDestination();
|
||||
const a = await daemon();
|
||||
const { paseo } = await api(a);
|
||||
const host = await paseo.hosts.add({ connection: `127.0.0.1:${destination.port}` });
|
||||
let retryAt = await capabilityRetry(paseo);
|
||||
let connections = destination.connections();
|
||||
await paseo.hosts.update(host.registrationId, { enabled: false });
|
||||
await new Promise((resolve) => setTimeout(resolve, retryAt - Date.now() + 500));
|
||||
expect(destination.connections()).toBe(connections);
|
||||
await paseo.hosts.update(host.registrationId, { enabled: true });
|
||||
retryAt = await capabilityRetry(paseo);
|
||||
connections = destination.connections();
|
||||
await paseo.hosts.remove(host.registrationId);
|
||||
await new Promise((resolve) => setTimeout(resolve, retryAt - Date.now() + 500));
|
||||
expect(destination.connections()).toBe(connections);
|
||||
});
|
||||
|
||||
import {
|
||||
createDaemonChannel,
|
||||
exportPublicKey,
|
||||
generateKeyPair,
|
||||
openDaemonSecureChannel,
|
||||
type Transport,
|
||||
} from "@getpaseo/relay/e2ee";
|
||||
test("a capability retry over a relay keeps the protected endpoint security it negotiated", async () => {
|
||||
const keys = generateKeyPair();
|
||||
const legacyReceived: unknown[] = [];
|
||||
let connections = 0;
|
||||
let retryClosed = false;
|
||||
const http = createServer();
|
||||
const ws = new WebSocketServer({ server: http });
|
||||
ws.on("connection", (socket) => {
|
||||
connections++;
|
||||
const wire: Transport = {
|
||||
send: (data) => socket.send(data),
|
||||
close: (code, reason) => socket.close(code, reason),
|
||||
onmessage: null,
|
||||
onclose: null,
|
||||
onerror: null,
|
||||
};
|
||||
socket.on("message", (data, isBinary) => {
|
||||
const bytes = Buffer.from(data as Buffer);
|
||||
wire.onmessage?.({
|
||||
data: isBinary
|
||||
? bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength)
|
||||
: bytes.toString(),
|
||||
isBinary,
|
||||
});
|
||||
});
|
||||
socket.on("close", (code, reason) => wire.onclose?.(code, reason.toString()));
|
||||
if (connections > 1) {
|
||||
socket.on("close", () => {
|
||||
retryClosed = true;
|
||||
});
|
||||
void createDaemonChannel(wire, keys, {
|
||||
onmessage: (data) => legacyReceived.push(data),
|
||||
}).catch(() => {});
|
||||
return;
|
||||
}
|
||||
const channel = openDaemonSecureChannel(wire, {
|
||||
daemonKeyPair: keys,
|
||||
events: {
|
||||
onmessage: (data) => {
|
||||
if (typeof data !== "string" || JSON.parse(data).type !== "hello") return;
|
||||
void channel.then((secure) =>
|
||||
secure.send(
|
||||
JSON.stringify({
|
||||
type: "session",
|
||||
message: {
|
||||
type: "status",
|
||||
payload: {
|
||||
status: "server_info",
|
||||
serverId: "srv_relayed",
|
||||
hostname: "relayed",
|
||||
version: "0.12.0",
|
||||
},
|
||||
},
|
||||
}),
|
||||
),
|
||||
);
|
||||
},
|
||||
},
|
||||
});
|
||||
void channel.catch(() => {});
|
||||
});
|
||||
http.listen(0, "127.0.0.1");
|
||||
await once(http, "listening");
|
||||
cleanup.push(async () => {
|
||||
for (const socket of ws.clients) socket.terminate();
|
||||
await new Promise<void>((resolve) => ws.close(() => resolve()));
|
||||
await new Promise<void>((resolve) => http.close(() => resolve()));
|
||||
});
|
||||
const address = http.address();
|
||||
if (!address || typeof address === "string") throw new Error("Missing relay port");
|
||||
const a = await daemon();
|
||||
const { paseo } = await api(a);
|
||||
await paseo.hosts.add({
|
||||
connection: `relay://127.0.0.1:${address.port}/srv_relayed?key=${encodeURIComponent(exportPublicKey(keys.publicKey))}`,
|
||||
});
|
||||
await capabilityRetry(paseo);
|
||||
await expect.poll(() => retryClosed, { timeout: 15000 }).toBe(true);
|
||||
expect(legacyReceived).toEqual([]);
|
||||
});
|
||||
|
||||
test("an unreachable saved target becomes online after its daemon starts without being added again", async () => {
|
||||
const reservation = createServer();
|
||||
reservation.listen(0, "127.0.0.1");
|
||||
|
||||
@@ -7,12 +7,10 @@ import { z } from "zod";
|
||||
import {
|
||||
createHostClient,
|
||||
createHostChannelTransport,
|
||||
createHostTransportFactory,
|
||||
parseHostTarget,
|
||||
HostTransportBudget,
|
||||
} from "@getpaseo/client/node";
|
||||
import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
|
||||
import type { DaemonTransportFactory } from "@getpaseo/client/internal/daemon-client-transport-types";
|
||||
import {
|
||||
HostSummarySchema,
|
||||
HostAddSchema,
|
||||
@@ -237,30 +235,23 @@ export class ConnectedHosts {
|
||||
if (route.registrationId === id)
|
||||
route.close(code, "Host access was disabled, replaced or removed");
|
||||
}
|
||||
private start(
|
||||
row: Saved,
|
||||
resume?: { failures: number; transports: DaemonTransportFactory },
|
||||
): void {
|
||||
private start(row: Saved): void {
|
||||
if (!row.enabled || this.closed) return;
|
||||
const transports =
|
||||
resume?.transports ??
|
||||
createHostTransportFactory(row.connection, this.budget, {
|
||||
webSocketAgent: webSocketProxyAgent,
|
||||
});
|
||||
let stopped = false;
|
||||
let attempts = resume?.failures ?? 0;
|
||||
let retry: ReturnType<typeof setTimeout> | undefined;
|
||||
const client = createHostClient(row.connection, {
|
||||
clientId: `host-link-${this.self.serverId}-${row.registrationId}`,
|
||||
connectTimeoutMs: 10000,
|
||||
transportFactory: transports,
|
||||
});
|
||||
const client = createHostClient(
|
||||
row.connection,
|
||||
{
|
||||
clientId: `host-link-${this.self.serverId}-${row.registrationId}`,
|
||||
connectTimeoutMs: 10000,
|
||||
webSocketAgent: webSocketProxyAgent,
|
||||
},
|
||||
this.budget,
|
||||
);
|
||||
const live: Live = {
|
||||
client,
|
||||
accepted: false,
|
||||
stop: () => {
|
||||
stopped = true;
|
||||
clearTimeout(retry);
|
||||
void client.close();
|
||||
},
|
||||
};
|
||||
@@ -294,7 +285,7 @@ export class ConnectedHosts {
|
||||
if (info.features?.connectedHosts !== true || info.features?.hostMultiplexing !== true)
|
||||
throw new HostError(
|
||||
"unsupported_capability",
|
||||
"Update the destination daemon to use connected hosts; retrying automatically.",
|
||||
"Update the destination daemon to use connected hosts.",
|
||||
);
|
||||
// Pin and persist under the same writer as management to prevent discovery races.
|
||||
await this.mutate(async () => {
|
||||
@@ -313,22 +304,16 @@ export class ConnectedHosts {
|
||||
if (stopped || client.getLastServerInfoMessage() !== info || !client.isConnected) return;
|
||||
Object.assign(row, next);
|
||||
live.accepted = true;
|
||||
attempts = 0;
|
||||
this.changed();
|
||||
});
|
||||
} catch (error) {
|
||||
if (stopped) return;
|
||||
row.status = "error";
|
||||
row.error = connectionFailure(error, client.authFailureReason !== null) ?? {
|
||||
code: "host_offline",
|
||||
message: "Host connection failed",
|
||||
};
|
||||
if (terminalFailures.has(row.error.code)) {
|
||||
row.status = "error";
|
||||
row.nextRetryAt = null;
|
||||
} else {
|
||||
stopped = true;
|
||||
retry = this.scheduleAcceptanceRetry(row, attempts, transports);
|
||||
}
|
||||
row.nextRetryAt = null;
|
||||
this.changed();
|
||||
await client.close();
|
||||
}
|
||||
@@ -344,17 +329,18 @@ export class ConnectedHosts {
|
||||
if (route.registrationId === row.registrationId) route.close();
|
||||
row.status = state.status === "connecting" ? "connecting" : "offline";
|
||||
row.nextRetryAt = state.status === "disconnected" ? (state.nextRetryAt ?? null) : null;
|
||||
if (state.status === "disconnected")
|
||||
row.error = connectionFailure(
|
||||
Object.assign(new Error("Check the SSH configuration and daemon credentials"), {
|
||||
code: state.failureCode,
|
||||
}),
|
||||
client.authFailureReason !== null,
|
||||
) ?? {
|
||||
code: "host_offline",
|
||||
message: "Cannot reach host. Check the connection; retrying automatically.",
|
||||
};
|
||||
else if (row.error?.code !== "unsupported_capability") row.error = null;
|
||||
row.error =
|
||||
state.status === "disconnected"
|
||||
? (connectionFailure(
|
||||
Object.assign(new Error("Check the SSH configuration and daemon credentials"), {
|
||||
code: state.failureCode,
|
||||
}),
|
||||
client.authFailureReason !== null,
|
||||
) ?? {
|
||||
code: "host_offline",
|
||||
message: "Cannot reach host. Check the connection; retrying automatically.",
|
||||
})
|
||||
: null;
|
||||
if (row.error?.code === "authentication_failed") {
|
||||
row.status = "error";
|
||||
row.nextRetryAt = null;
|
||||
@@ -365,18 +351,6 @@ export class ConnectedHosts {
|
||||
});
|
||||
void client.connect().catch(() => {});
|
||||
}
|
||||
private scheduleAcceptanceRetry(
|
||||
row: Saved,
|
||||
failures: number,
|
||||
transports: DaemonTransportFactory,
|
||||
): ReturnType<typeof setTimeout> {
|
||||
const delay = acceptanceRetryDelayMs(failures);
|
||||
row.status = "offline";
|
||||
row.nextRetryAt = new Date(Date.now() + delay).toISOString();
|
||||
const retry = setTimeout(() => this.start(row, { failures: failures + 1, transports }), delay);
|
||||
retry.unref();
|
||||
return retry;
|
||||
}
|
||||
async openRoute(input: {
|
||||
serverId: string;
|
||||
scopeId: string;
|
||||
@@ -557,20 +531,6 @@ export class ConnectedHosts {
|
||||
}
|
||||
}
|
||||
|
||||
const ACCEPTANCE_RETRY_BASE_MS = 1500;
|
||||
const ACCEPTANCE_RETRY_MAX_MS = 30_000;
|
||||
|
||||
function acceptanceRetryDelayMs(failures: number): number {
|
||||
return Math.min(ACCEPTANCE_RETRY_BASE_MS * 2 ** failures, ACCEPTANCE_RETRY_MAX_MS);
|
||||
}
|
||||
|
||||
const terminalFailures = new Set([
|
||||
"server_id_mismatch",
|
||||
"self_host",
|
||||
"duplicate_host",
|
||||
"authentication_failed",
|
||||
]);
|
||||
|
||||
function connectionFailure(error: unknown, authenticationFailed: boolean): HostSummary["error"] {
|
||||
if (error instanceof HostError) return { code: error.code, message: error.message };
|
||||
if (error instanceof Error && "code" in error && error.code === "authentication_failed")
|
||||
|
||||
Reference in New Issue
Block a user