From fa09e8706f81cbc6af95759cd1002f304791b119 Mon Sep 17 00:00:00 2001 From: Saurav Date: Sat, 17 Oct 2026 13:40:37 +0530 Subject: [PATCH 1/3] sdk/js: a join that hangs says where; the probe's programs print each stage The SDK probe's JS programs hung silently for 120s in-cluster, and nothing said which await they were in (#732). Each step of a join already has a timeout of its own, but a join takes them in turn over every router it is offered, after up to four control plane requests of 30s each, so the whole of it had no bound. JoinOptions .signal, documented as bounding the whole join, reached only the router dials. It now bounds the control plane sync, and the dial, auth handshake and relay reservation with every router; without one the join gets JOIN_TIMEOUT_MS (60s). A join that stops names each router it tried and the step it was in, or the control plane it was waiting on. A directory request no router answered names every router and what each said, not only the last error. call.js, a2a-call.js and a2a-agent.js print each stage to stderr as they reach it (enrolling, enrolled, joined via, provider, dialing), so a run that hangs shows the last one in the probe's own output. stdout is unchanged; the docs that embed the examples are regenerated. Refs #732 --- sdk/js/README.md | 13 +++- sdk/js/examples/a2a-agent.ts | 8 ++- sdk/js/examples/a2a-call.ts | 9 ++- sdk/js/examples/call.ts | 13 +++- sdk/js/src/discovery.test.ts | 22 ++++++ sdk/js/src/discovery.ts | 6 +- sdk/js/src/index.ts | 1 + sdk/js/src/session.test.ts | 80 ++++++++++++++++++++- sdk/js/src/session.ts | 93 +++++++++++++++++++++---- site/content/docs/guides/native-sdks.md | 30 +++++++- 10 files changed, 251 insertions(+), 24 deletions(-) diff --git a/sdk/js/README.md b/sdk/js/README.md index 1e067d4f..db2bc683 100644 --- a/sdk/js/README.md +++ b/sdk/js/README.md @@ -61,15 +61,21 @@ if (argv[0] !== undefined && !argv[0].includes("://")) { } const [service = peerId !== undefined ? "a2a://agent" : "mcp://everything", toolOrPath = peerId !== undefined ? "/card" : "echo", args = '{"message": "hi"}'] = argv; +// Each stage goes to stderr as it is reached, so a run that hangs shows the +// last one it got to; stdout carries only what callers parse. +const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; +console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", + controlPlaneUrl, bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/caller`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); +console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); +console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); console.log(`on the mesh as ${session.peerId}`); // The peer to call: an agent named by ID, or a provider of a published service. @@ -77,9 +83,13 @@ let provider: DiscoveredProvider = { peerId: peerId ?? "", addrs: [] }; if (peerId !== undefined) { // An agent is not in the discovery table; the SDK finds the path to its // peer ID through the routers. + console.error(`dialing ${peerId}`); await session.connect(peerId); } else { const providers = await session.discover(service); + for (const p of providers) { + console.error(`provider ${p.peerId}`); + } if (providers.length === 0) { throw new Error(`no member of the mesh serves ${service}`); } @@ -87,6 +97,7 @@ if (peerId !== undefined) { let reached = false; for (const candidate of providers) { try { + console.error(`dialing ${candidate.peerId}`); await session.connect(candidate); provider = candidate; reached = true; diff --git a/sdk/js/examples/a2a-agent.ts b/sdk/js/examples/a2a-agent.ts index 3d2f17f1..c3c87280 100644 --- a/sdk/js/examples/a2a-agent.ts +++ b/sdk/js/examples/a2a-agent.ts @@ -20,15 +20,21 @@ import { AgentEvent, DefaultRequestHandler, InMemoryTaskStore, type AgentExecuto import { UserBuilder, agentCardHandler, jsonRpcHandler } from "@a2a-js/sdk/server/express"; import { AgentMesh, MeshSession } from "@agentmesh-p2p/sdk"; +// Each stage goes to stderr as it is reached, so a run that hangs shows the +// last one it got to; stdout carries only what callers parse. +const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; +console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", + controlPlaneUrl, bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/a2a-agent`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); +console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); +console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); // The card names the agent as the mesh reaches it: the URL an A2A client on // the mesh gives its fetch, http://mesh/mesh//a2a/agent. diff --git a/sdk/js/examples/a2a-call.ts b/sdk/js/examples/a2a-call.ts index 8890f6f1..f66eeb88 100644 --- a/sdk/js/examples/a2a-call.ts +++ b/sdk/js/examples/a2a-call.ts @@ -21,15 +21,21 @@ if (peerId === undefined) { throw new Error("usage: a2a-call.js [text]"); } +// Each stage goes to stderr as it is reached, so a run that hangs shows the +// last one it got to; stdout carries only what callers parse. +const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; +console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", + controlPlaneUrl, bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/a2a-caller`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); +console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); +console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); console.log(`on the mesh as ${session.peerId}`); // The agent's URL on the mesh, and a fetch that carries requests to it. @@ -40,6 +46,7 @@ const factory = new ClientFactory({ cardResolver: new DefaultAgentCardResolver({ fetchImpl }), }); // The card lives under the agent's URL; the trailing slash keeps the last segment when the resolver appends the well-known path. +console.error(`dialing ${peerId}`); const client = await factory.createFromUrl(agentURL + "/"); const card = await client.getAgentCard(); console.log(`agent: ${card.name}, ${card.description}`); diff --git a/sdk/js/examples/call.ts b/sdk/js/examples/call.ts index 324667cd..6a88ce77 100644 --- a/sdk/js/examples/call.ts +++ b/sdk/js/examples/call.ts @@ -22,15 +22,21 @@ if (argv[0] !== undefined && !argv[0].includes("://")) { } const [service = peerId !== undefined ? "a2a://agent" : "mcp://everything", toolOrPath = peerId !== undefined ? "/card" : "echo", args = '{"message": "hi"}'] = argv; +// Each stage goes to stderr as it is reached, so a run that hangs shows the +// last one it got to; stdout carries only what callers parse. +const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; +console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", + controlPlaneUrl, bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/caller`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); +console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); +console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); console.log(`on the mesh as ${session.peerId}`); // The peer to call: an agent named by ID, or a provider of a published service. @@ -38,9 +44,13 @@ let provider: DiscoveredProvider = { peerId: peerId ?? "", addrs: [] }; if (peerId !== undefined) { // An agent is not in the discovery table; the SDK finds the path to its // peer ID through the routers. + console.error(`dialing ${peerId}`); await session.connect(peerId); } else { const providers = await session.discover(service); + for (const p of providers) { + console.error(`provider ${p.peerId}`); + } if (providers.length === 0) { throw new Error(`no member of the mesh serves ${service}`); } @@ -48,6 +58,7 @@ if (peerId !== undefined) { let reached = false; for (const candidate of providers) { try { + console.error(`dialing ${candidate.peerId}`); await session.connect(candidate); provider = candidate; reached = true; diff --git a/sdk/js/src/discovery.test.ts b/sdk/js/src/discovery.test.ts index 1f2d4826..10bd289d 100644 --- a/sdk/js/src/discovery.test.ts +++ b/sdk/js/src/discovery.test.ts @@ -99,3 +99,25 @@ test("a lookup one router answers with nothing is put to the next; an announceme await assert.rejects(request([{ peerId: "nobody", connection: () => undefined }], find), DirectoryError); }); + +test("a request no router answered names every router and what each said", async () => { + const key = await serviceKey("mcp", "calculator"); + const find = create(DirectoryRequestSchema, { request: { case: "findProviders", value: { key, limit: 20 } } }); + answers.set("a", () => create(DirectoryResponseSchema, { error: "routing table empty" })); + answers.set("b", () => create(DirectoryResponseSchema, { error: "rate limited" })); + try { + await assert.rejects(request([held(routerA), held(routerB)], find), (err: Error) => { + assert.ok(err instanceof DirectoryError, `got ${err.name}, want DirectoryError`); + for (const [router, said] of [ + [routerA, "routing table empty"], + [routerB, "rate limited"], + ] as const) { + assert.ok(err.message.includes(`${router.peerId.toString()}: ${said}`), `got ${JSON.stringify(err.message)}, want ${router.peerId.toString()} and what it said`); + } + return true; + }); + } finally { + answers.delete("a"); + answers.delete("b"); + } +}); diff --git a/sdk/js/src/discovery.ts b/sdk/js/src/discovery.ts index 711a7784..37dbf94a 100644 --- a/sdk/js/src/discovery.ts +++ b/sdk/js/src/discovery.ts @@ -110,7 +110,7 @@ async function ask(conn: Connection, req: DirectoryRequest, signal: AbortSignal) */ export async function request(routers: DirectoryRouter[], req: DirectoryRequest, signal?: AbortSignal): Promise { const lookup = req.request.case === "findProviders" || req.request.case === "findPeer"; - let last: unknown; + const failures: string[] = []; let asked = 0; let empty: DirectoryResponse | undefined; for (const router of routers) { @@ -128,7 +128,7 @@ export async function request(routers: DirectoryRouter[], req: DirectoryRequest, } return resp; } catch (err) { - last = err; + failures.push(`${router.peerId}: ${err instanceof Error ? err.message : String(err)}`); if (signal?.aborted === true) { break; } @@ -140,7 +140,7 @@ export async function request(routers: DirectoryRouter[], req: DirectoryRequest, if (asked === 0) { throw new DirectoryError("no held router to ask"); } - throw new DirectoryError(`no held router answered: ${last instanceof Error ? last.message : String(last)}`); + throw new DirectoryError(`no held router answered: ${failures.join("; ")}`); } /** diff --git a/sdk/js/src/index.ts b/sdk/js/src/index.ts index 947fd3ba..3d32b898 100644 --- a/sdk/js/src/index.ts +++ b/sdk/js/src/index.ts @@ -54,6 +54,7 @@ export { } from "./challenges.ts"; export { DEFAULT_ROUTERS, + JOIN_TIMEOUT_MS, MeshSession, ROUTER_REDIAL_BACKOFFS_MS, candidatesFromRouterInfos, diff --git a/sdk/js/src/session.test.ts b/sdk/js/src/session.test.ts index db5442ed..fb6204b9 100644 --- a/sdk/js/src/session.test.ts +++ b/sdk/js/src/session.test.ts @@ -36,7 +36,7 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { after, before, test } from "node:test"; -import { AUTH_HANDLER_OPTIONS, AUTH_PROTOCOL, AuthRejectedError, GOAWAY_MAX_RETRY_AFTER_MS, GOAWAY_PROTOCOL, ROUTER_SHUN_DURATION_MS, authenticateWithPeer, authStreamHandler, clampGoAwayRetryAfterMs } from "./auth.ts"; +import { AUTH_HANDLER_OPTIONS, AUTH_HANDSHAKE_TIMEOUT_MS, AUTH_PROTOCOL, AuthRejectedError, GOAWAY_MAX_RETRY_AFTER_MS, GOAWAY_PROTOCOL, ROUTER_SHUN_DURATION_MS, authenticateWithPeer, authStreamHandler, clampGoAwayRetryAfterMs } from "./auth.ts"; import { BiscuitVerificationError, ROLE_ROUTER, loadBiscuit, verifyPeerBiscuit } from "./biscuit.ts"; import { ROLE_NODE } from "./controlplane.ts"; import { BootstrapEnrollRequestSchema, BootstrapEnrollResponseSchema, ControlPlaneInfoResponseSchema, EnrollmentStatus, KeysResponseSchema, AuthFrameSchema, DirectoryRequestSchema, DirectoryResponseSchema, PeerRecordSchema, RouterGoAwaySchema, RouterGoAway_Reason, RouterInfoSchema, TokenRefreshRequestSchema, TokenRefreshResponseSchema, type RouterInfo } from "./gen/agentmesh_pb.ts"; @@ -44,7 +44,7 @@ import { Identity } from "./identity.ts"; import { HTTP_HANDLER_OPTIONS, HTTP_PROTOCOL, a2aEndpoint, httpIngressHandler } from "./libp2p-http.ts"; import { LabelsNotSatisfiedError } from "./mcp.ts"; import { AgentMesh } from "./mesh.ts"; -import { MeshSession, candidatesFromRouterInfos, selectRouters, type JoinOptions } from "./session.ts"; +import { MeshSession, candidatesFromRouterInfos, joinDeadline, selectRouters, type JoinOptions } from "./session.ts"; type Wasm = Awaited>; type KeyPair = InstanceType; @@ -605,6 +605,82 @@ test("join reports a router that refuses the handshake", async () => { } }); +test("a join bounded by its signal names the router whose handshake never answered", async () => { + const silent = await createLibp2p({ + addresses: { listen: ["/ip4/127.0.0.1/tcp/0"] }, + transports: [tcp()], + connectionEncrypters: [tls()], + streamMuxers: [yamux()], + services: { identify: identify() }, + }); + try { + // Takes the auth stream and never writes back, as a router whose + // handshake neither completes nor fails. + await silent.handle(AUTH_PROTOCOL, () => {}); + const addr = (silent.getMultiaddrs()[0] as ReturnType).toString(); + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane([addr]) }); + const started = Date.now(); + await assert.rejects(mesh.join({ signal: AbortSignal.timeout(500) }), (err: Error) => { + assert.match(err.message, new RegExp(`${silent.peerId.toString()}: auth handshake: `), `got ${JSON.stringify(err.message)}, want the router and the step it hung in`); + return true; + }); + const took = Date.now() - started; + assert.ok(took < AUTH_HANDSHAKE_TIMEOUT_MS, `join took ${took}ms, want the join's signal to end it before the ${AUTH_HANDSHAKE_TIMEOUT_MS}ms handshake timeout`); + } finally { + await silent.stop(); + } +}); + +test("a join bounded by its signal says when the control plane never answered", async () => { + const pending: Array<() => void> = []; + let hang = false; + const routes = fakeControlPlane([routerAddr]); + // Once enrolled, /keys and /info take the request and never answer, until + // the request is aborted or the test ends. + const stalling = (async (input: Parameters[0], init?: RequestInit) => { + const path = new URL(new Request(input, init).url).pathname; + if (!hang || (path !== "/keys" && path !== "/info")) { + return routes(input, init); + } + return new Promise((_, reject) => { + const stop = () => reject(init?.signal?.reason ?? new Error("test over")); + pending.push(stop); + init?.signal?.addEventListener("abort", stop, { once: true }); + }); + }) as typeof fetch; + try { + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: stalling }); + hang = true; + const started = Date.now(); + await assert.rejects(mesh.join({ signal: AbortSignal.timeout(500) }), (err: Error) => { + assert.match(err.message, /syncing with the control plane at http:\/\/127\.0\.0\.1:1/, `got ${JSON.stringify(err.message)}, want the control plane named`); + return true; + }); + const took = Date.now() - started; + assert.ok(took < 10_000, `join took ${took}ms, want the join's signal to end it, not the control plane client's 30s request timeout`); + } finally { + // The sync the join gave up on goes on to its next request; let it + // through so no request is left holding the client's 30s timer. + hang = false; + for (const stop of pending) { + stop(); + } + } +}); + +test("a join without a signal of its own still has a deadline", async () => { + const given = AbortSignal.timeout(60_000); + const own = joinDeadline(given, 20); + assert.equal(own.signal, given, "a caller's signal bounds the join as given"); + own.clear(); + + const deadline = joinDeadline(undefined, 20); + await new Promise((resolve) => deadline.signal.addEventListener("abort", resolve, { once: true })); + const reason = deadline.signal.reason as Error; + assert.match(reason.message, /^join timed out after 20ms$/, `abort reason ${JSON.stringify(reason.message)}`); + deadline.clear(); +}); + test("an explicit rejection is reported with its reason", async () => { const reason = new AuthRejectedError("12D3KooWtest", "peer is revoked"); assert.match(reason.message, /12D3KooWtest.*peer is revoked/); diff --git a/sdk/js/src/session.ts b/sdk/js/src/session.ts index 3e3fe5d1..d7476c03 100644 --- a/sdk/js/src/session.ts +++ b/sdk/js/src/session.ts @@ -125,7 +125,12 @@ export interface JoinOptions extends MeshHostOptions { * requiredLabels. Held on every call, MCP and HTTP alike; no call waives it. */ egressRequireLabels?: Record; - /** Bounds the whole join. */ + /** + * Bounds the whole join: the control plane sync, and the dial, auth + * handshake and relay reservation with every router tried. Without one the + * join is bounded by JOIN_TIMEOUT_MS (60s). An aborted join throws an error + * naming each router tried and the step it was in. + */ signal?: AbortSignal; } @@ -198,6 +203,11 @@ const DEFAULT_RELAY_CHECK_MS = 30 * 1000; const DEFAULT_ANNOUNCE_MS = 5 * 60 * 1000; /** What close() waits for the member's record to be withdrawn. */ const WITHDRAW_TIMEOUT_MS = 2 * 1000; +/** + * Bounds a join whose caller passed no signal: each step has a timeout of + * its own, but a join tries them in turn over every router it is offered. + */ +export const JOIN_TIMEOUT_MS = 60_000; /** How long a provider's positive egress verdict is kept; agentmesh-node's labelGateTTL. */ const EGRESS_VERDICT_TTL_MS = 5 * 60 * 1000; @@ -394,7 +404,7 @@ export class MeshSession { * the mesh's record TTL. A member with no relayed address announces * nothing: there is nothing to reach it at. */ - async announce(): Promise { + async announce(signal?: AbortSignal): Promise { if (this.#closed || this.#isTaskView) { return; } @@ -402,7 +412,7 @@ export class MeshSession { if (addrs.length === 0) { return; } - await provide(this.#directoryRouters(), await peerKey(this.peerId), addrs); + await provide(this.#directoryRouters(), await peerKey(this.peerId), addrs, signal); this.#announced = addrs; } @@ -699,20 +709,28 @@ export class MeshSession { let progress = false; for (const cand of picks) { tried.add(cand.peerId); + // Each step has a timeout of its own; the failure names the one the + // router was in, so a join that hangs says where. + let step = `dial ${cand.addresses.map(String).join(", ")}`; try { const conn = await this.node.dial(cand.addresses, signal !== undefined ? { signal } : {}); - const credential = await this.#authenticateRouter(conn); + step = "auth handshake"; + const credential = await untilAborted(this.#authenticateRouter(conn), signal); requireRole(credential, ROLE_ROUTER); const addr = connectedAddress(conn, cand.peerId); if (this.#reserveRelay) { - await this.#reserveOnRouter(cand.peerId, addr); + step = "relay reservation"; + await untilAborted(this.#reserveOnRouter(cand.peerId, addr), signal); } this.#upsertHeldRouter({ peerId: cand.peerId, addr, credential }); progress = true; } catch (err) { this.#shunRouter(cand.peerId, ROUTER_SHUN_DURATION_MS); await this.node.hangUp(peerIdFromString(cand.peerId)).catch(() => {}); - failures.push(`${cand.peerId}: ${err instanceof Error ? err.message : String(err)}`); + failures.push(`${cand.peerId}: ${step}: ${errorText(err)}`); + if (signal?.aborted === true) { + return failures; + } } } if (!progress && picks.length < need) { @@ -1408,8 +1426,31 @@ export function selectRouters(candidates: readonly RouterCandidate[], k: number, return scored.slice(0, k).map((s) => s.candidate); } +/** + * The signal a join runs under: the caller's, or one that aborts after ms + * with an error saying the join timed out. clear() stops the timer once the + * join is over. + */ +export function joinDeadline(signal: AbortSignal | undefined, ms = JOIN_TIMEOUT_MS): { signal: AbortSignal; clear(): void } { + if (signal !== undefined) { + return { signal, clear: () => {} }; + } + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(new Error(`join timed out after ${ms}ms`)), ms); + return { signal: controller.signal, clear: () => clearTimeout(timer) }; +} + /** Implements AgentMesh.join(); lives here to keep mesh.ts free of libp2p. */ export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}): Promise { + const deadline = joinDeadline(options.signal); + try { + return await joinUnder(mesh, options, deadline.signal); + } finally { + deadline.clear(); + } +} + +async function joinUnder(mesh: AgentMesh, options: JoinOptions, signal: AbortSignal): Promise { // The pull agentmesh-node makes before it starts: a member resuming from its // state directory after a key rotation would otherwise verify the routers, // which already hold credentials under the new key, against the keys it @@ -1417,9 +1458,13 @@ export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}): Prom // once joined. Best effort; the stored credential serves when the control // plane cannot be reached. try { - await mesh.syncControlPlane(); + await untilAborted(mesh.syncControlPlane(), signal); } catch { - // Joining goes on with what the credential holds. + // Joining goes on with what the credential holds, unless the join itself + // ran out while the control plane kept it waiting. + if (signal.aborted) { + throw new Error(`join stopped while syncing with the control plane at ${mesh.controlPlane.url}: ${errorText(signal.reason)}`); + } } const rawAddrs = options.routerAddresses ?? mesh.credential.routerAddresses; if (rawAddrs.length === 0 && mesh.routers.length === 0) { @@ -1455,14 +1500,14 @@ export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}): Prom ); session = new MeshSession(mesh, node, [], authenticatedPeers, banned, options); - const failures = await session.topUpRouters(options.signal); - if (session.routers.length === 0) { + const failures = await session.topUpRouters(signal); + if (signal.aborted || session.routers.length === 0) { const detail = failures.length > 0 ? `:\n ${failures.join("\n ")}` : ""; - throw new Error(`no router admitted this member${detail}`); + throw new Error(signal.aborted ? `join stopped: ${errorText(signal.reason)}${detail}` : `no router admitted this member${detail}`); } // Best effort: a router that does not take the record is a warning the // next announcement retries; the member is on the mesh regardless. - await session.announce().catch(() => {}); + await session.announce(signal).catch(() => {}); return session; } catch (err) { if (session !== undefined) { @@ -1474,6 +1519,30 @@ export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}): Prom } } +/** + * Settles as p does, or rejects with the signal's reason as soon as it + * aborts. p is not cancelled: it ends on its own timeout, and the caller + * tears down what it was using. + */ +function untilAborted(p: Promise, signal: AbortSignal | undefined): Promise { + if (signal === undefined) { + return p; + } + if (signal.aborted) { + p.catch(() => {}); + return Promise.reject(signal.reason); + } + return new Promise((resolve, reject) => { + const onAbort = () => reject(signal.reason); + signal.addEventListener("abort", onAbort, { once: true }); + void p.then(resolve, reject).finally(() => signal.removeEventListener("abort", onAbort)); + }); +} + +function errorText(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + /** The peer a multiaddr ends at, in canonical form: its trailing `/p2p/`, or undefined for a relay address with no target yet. */ function targetPeerOf(ma: Multiaddr): string | undefined { const last = ma.getComponents().at(-1); diff --git a/site/content/docs/guides/native-sdks.md b/site/content/docs/guides/native-sdks.md index 6035e758..85743f36 100644 --- a/site/content/docs/guides/native-sdks.md +++ b/site/content/docs/guides/native-sdks.md @@ -286,15 +286,21 @@ if (argv[0] !== undefined && !argv[0].includes("://")) { } const [service = peerId !== undefined ? "a2a://agent" : "mcp://everything", toolOrPath = peerId !== undefined ? "/card" : "echo", args = '{"message": "hi"}'] = argv; +// Each stage goes to stderr as it is reached, so a run that hangs shows the +// last one it got to; stdout carries only what callers parse. +const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; +console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", + controlPlaneUrl, bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/caller`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); +console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); +console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); console.log(`on the mesh as ${session.peerId}`); // The peer to call: an agent named by ID, or a provider of a published service. @@ -302,9 +308,13 @@ let provider: DiscoveredProvider = { peerId: peerId ?? "", addrs: [] }; if (peerId !== undefined) { // An agent is not in the discovery table; the SDK finds the path to its // peer ID through the routers. + console.error(`dialing ${peerId}`); await session.connect(peerId); } else { const providers = await session.discover(service); + for (const p of providers) { + console.error(`provider ${p.peerId}`); + } if (providers.length === 0) { throw new Error(`no member of the mesh serves ${service}`); } @@ -312,6 +322,7 @@ if (peerId !== undefined) { let reached = false; for (const candidate of providers) { try { + console.error(`dialing ${candidate.peerId}`); await session.connect(candidate); provider = candidate; reached = true; @@ -525,15 +536,21 @@ import { AgentEvent, DefaultRequestHandler, InMemoryTaskStore, type AgentExecuto import { UserBuilder, agentCardHandler, jsonRpcHandler } from "@a2a-js/sdk/server/express"; import { AgentMesh, MeshSession } from "@agentmesh-p2p/sdk"; +// Each stage goes to stderr as it is reached, so a run that hangs shows the +// last one it got to; stdout carries only what callers parse. +const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; +console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", + controlPlaneUrl, bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/a2a-agent`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); +console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); +console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); // The card names the agent as the mesh reaches it: the URL an A2A client on // the mesh gives its fetch, http://mesh/mesh//a2a/agent. @@ -736,15 +753,21 @@ if (peerId === undefined) { throw new Error("usage: a2a-call.js [text]"); } +// Each stage goes to stderr as it is reached, so a run that hangs shows the +// last one it got to; stdout carries only what callers parse. +const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; +console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", + controlPlaneUrl, bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/a2a-caller`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); +console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); +console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); console.log(`on the mesh as ${session.peerId}`); // The agent's URL on the mesh, and a fetch that carries requests to it. @@ -755,6 +778,7 @@ const factory = new ClientFactory({ cardResolver: new DefaultAgentCardResolver({ fetchImpl }), }); // The card lives under the agent's URL; the trailing slash keeps the last segment when the resolver appends the well-known path. +console.error(`dialing ${peerId}`); const client = await factory.createFromUrl(agentURL + "/"); const card = await client.getAgentCard(); console.log(`agent: ${card.name}, ${card.description}`); From e39dfb2bcb183b185c5d18bda49127a802adf2e1 Mon Sep 17 00:00:00 2001 From: Saurav Date: Sat, 17 Oct 2026 13:59:25 +0530 Subject: [PATCH 2/3] sdk/js: an aborted top-up tries no router it has not dialed yet topUpRouters checked its signal only after a router failed. A signal that aborted while one router was being admitted sent the next one to a dial that failed at once, so the router was shunned and listed in the join's error without having been tried. The signal is checked before each router as well. --- sdk/js/src/session.test.ts | 37 +++++++++++++++++++++++++++++++++++++ sdk/js/src/session.ts | 9 ++++++++- 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/sdk/js/src/session.test.ts b/sdk/js/src/session.test.ts index fb6204b9..153728b8 100644 --- a/sdk/js/src/session.test.ts +++ b/sdk/js/src/session.test.ts @@ -668,6 +668,43 @@ test("a join bounded by its signal says when the control plane never answered", } }); +test("a top-up whose signal has aborted tries no router and shuns none", async () => { + const second = await createLibp2p({ + addresses: { listen: ["/ip4/127.0.0.1/tcp/0"] }, + transports: [tcp()], + connectionEncrypters: [tls()], + streamMuxers: [yamux()], + services: { identify: identify(), relay: circuitRelayServer() }, + connectionManager: { inboundConnectionThreshold: Infinity }, + }); + const secondId = second.peerId.toString(); + const secondBiscuit = mint(secondId, ROLE_ROUTER); + await second.handle(AUTH_PROTOCOL, authStreamHandler({ ownBiscuit: () => secondBiscuit, trustedKeys: () => [cpKey] })); + const secondAddr = (second.getMultiaddrs()[0] as ReturnType).toString(); + const listed = [routerAddr]; + try { + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane(listed) }); + const session = await mesh.join({ refreshLeadMs: 0, reserveRelay: false, controlPlaneSyncIntervalMs: 0 }); + try { + // The control plane now lists a router the session has not tried. + listed.push(secondAddr); + await mesh.syncControlPlane(); + + const failures = await session.topUpRouters(AbortSignal.abort(new Error("join over"))); + assert.deepEqual(failures, [], "an aborted top-up reports no router it did not try"); + assert.equal(session.shunnedRouters.has(secondId), false, `${secondId} was shunned by a top-up that never tried it`); + + // The router was eligible all along: the same top-up without the signal holds it. + assert.deepEqual(await session.topUpRouters(), []); + assert.ok(session.isHeldRouter(secondId), `held routers ${session.routers.map((r) => r.peerId).join(",")}, want ${secondId} among them`); + } finally { + await session.close(); + } + } finally { + await second.stop(); + } +}); + test("a join without a signal of its own still has a deadline", async () => { const given = AbortSignal.timeout(60_000); const own = joinDeadline(given, 20); diff --git a/sdk/js/src/session.ts b/sdk/js/src/session.ts index d7476c03..cdafda10 100644 --- a/sdk/js/src/session.ts +++ b/sdk/js/src/session.ts @@ -691,6 +691,8 @@ export class MeshSession { */ async topUpRouters(signal?: AbortSignal): Promise { const failures: string[] = []; + // A function, not a narrowed value: the signal aborts across the awaits below. + const aborted = () => signal?.aborted === true; const want = this.#wantRouters(); const candidates = this.#candidatesForHolding(); const tried = new Set(); @@ -708,6 +710,11 @@ export class MeshSession { } let progress = false; for (const cand of picks) { + // A router not dialed yet is not tried under a signal that has + // already aborted: it would fail at once and be shunned for nothing. + if (aborted()) { + return failures; + } tried.add(cand.peerId); // Each step has a timeout of its own; the failure names the one the // router was in, so a join that hangs says where. @@ -728,7 +735,7 @@ export class MeshSession { this.#shunRouter(cand.peerId, ROUTER_SHUN_DURATION_MS); await this.node.hangUp(peerIdFromString(cand.peerId)).catch(() => {}); failures.push(`${cand.peerId}: ${step}: ${errorText(err)}`); - if (signal?.aborted === true) { + if (aborted()) { return failures; } } From 349983a210b5e34e8f5e3ffe98e0df89ac297b71 Mon Sep 17 00:00:00 2001 From: Saurav Date: Sat, 17 Oct 2026 17:12:44 +0530 Subject: [PATCH 3/3] sdk/js: cancel an abandoned join's work instead of waiting past it From review on #739. The examples go back to what they were: they are embedded in the user guides, and printing the control plane URL could leak credentials or query parameters in it. A join that stops already says where in the error Node prints, so the probe's output shows it without them. The error no longer names the control plane URL either. untilAborted only stopped waiting: a control plane sync the join gave up on went on in the background, could still call /info and refresh(), and write credential.json. The signal now goes into the requests themselves. ControlPlaneClient.info() and keys() and authenticateWithPeer take one and combine it with their own timeout, as discovery does, and syncControlPlane stops before its next request and before saving once it aborts. A /refresh already sent is left to finish, since the control plane spends the old biscuit when it answers. The sync before a join is best effort, so it gets JOIN_SYNC_TIMEOUT_MS (5s) rather than the whole join: an unreachable control plane leaves the join to go on with the stored credential. A join that holds a router succeeds even if its own timeout fires while a second router is still being tried; only the caller's signal turns that into a failure. A router cut off by the signal is no longer shunned. Refs #732 --- sdk/js/README.md | 13 +- sdk/js/examples/a2a-agent.ts | 8 +- sdk/js/examples/a2a-call.ts | 9 +- sdk/js/examples/call.ts | 13 +- sdk/js/src/auth.ts | 5 +- sdk/js/src/controlplane.ts | 32 +++-- sdk/js/src/mesh.ts | 29 +++-- sdk/js/src/session.test.ts | 150 +++++++++++++++--------- sdk/js/src/session.ts | 96 ++++++--------- site/content/docs/guides/native-sdks.md | 30 +---- 10 files changed, 184 insertions(+), 201 deletions(-) diff --git a/sdk/js/README.md b/sdk/js/README.md index db2bc683..1e067d4f 100644 --- a/sdk/js/README.md +++ b/sdk/js/README.md @@ -61,21 +61,15 @@ if (argv[0] !== undefined && !argv[0].includes("://")) { } const [service = peerId !== undefined ? "a2a://agent" : "mcp://everything", toolOrPath = peerId !== undefined ? "/card" : "echo", args = '{"message": "hi"}'] = argv; -// Each stage goes to stderr as it is reached, so a run that hangs shows the -// last one it got to; stdout carries only what callers parse. -const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; -console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl, + controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/caller`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); -console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); -console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); console.log(`on the mesh as ${session.peerId}`); // The peer to call: an agent named by ID, or a provider of a published service. @@ -83,13 +77,9 @@ let provider: DiscoveredProvider = { peerId: peerId ?? "", addrs: [] }; if (peerId !== undefined) { // An agent is not in the discovery table; the SDK finds the path to its // peer ID through the routers. - console.error(`dialing ${peerId}`); await session.connect(peerId); } else { const providers = await session.discover(service); - for (const p of providers) { - console.error(`provider ${p.peerId}`); - } if (providers.length === 0) { throw new Error(`no member of the mesh serves ${service}`); } @@ -97,7 +87,6 @@ if (peerId !== undefined) { let reached = false; for (const candidate of providers) { try { - console.error(`dialing ${candidate.peerId}`); await session.connect(candidate); provider = candidate; reached = true; diff --git a/sdk/js/examples/a2a-agent.ts b/sdk/js/examples/a2a-agent.ts index c3c87280..3d2f17f1 100644 --- a/sdk/js/examples/a2a-agent.ts +++ b/sdk/js/examples/a2a-agent.ts @@ -20,21 +20,15 @@ import { AgentEvent, DefaultRequestHandler, InMemoryTaskStore, type AgentExecuto import { UserBuilder, agentCardHandler, jsonRpcHandler } from "@a2a-js/sdk/server/express"; import { AgentMesh, MeshSession } from "@agentmesh-p2p/sdk"; -// Each stage goes to stderr as it is reached, so a run that hangs shows the -// last one it got to; stdout carries only what callers parse. -const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; -console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl, + controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/a2a-agent`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); -console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); -console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); // The card names the agent as the mesh reaches it: the URL an A2A client on // the mesh gives its fetch, http://mesh/mesh//a2a/agent. diff --git a/sdk/js/examples/a2a-call.ts b/sdk/js/examples/a2a-call.ts index f66eeb88..8890f6f1 100644 --- a/sdk/js/examples/a2a-call.ts +++ b/sdk/js/examples/a2a-call.ts @@ -21,21 +21,15 @@ if (peerId === undefined) { throw new Error("usage: a2a-call.js [text]"); } -// Each stage goes to stderr as it is reached, so a run that hangs shows the -// last one it got to; stdout carries only what callers parse. -const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; -console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl, + controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/a2a-caller`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); -console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); -console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); console.log(`on the mesh as ${session.peerId}`); // The agent's URL on the mesh, and a fetch that carries requests to it. @@ -46,7 +40,6 @@ const factory = new ClientFactory({ cardResolver: new DefaultAgentCardResolver({ fetchImpl }), }); // The card lives under the agent's URL; the trailing slash keeps the last segment when the resolver appends the well-known path. -console.error(`dialing ${peerId}`); const client = await factory.createFromUrl(agentURL + "/"); const card = await client.getAgentCard(); console.log(`agent: ${card.name}, ${card.description}`); diff --git a/sdk/js/examples/call.ts b/sdk/js/examples/call.ts index 6a88ce77..324667cd 100644 --- a/sdk/js/examples/call.ts +++ b/sdk/js/examples/call.ts @@ -22,21 +22,15 @@ if (argv[0] !== undefined && !argv[0].includes("://")) { } const [service = peerId !== undefined ? "a2a://agent" : "mcp://everything", toolOrPath = peerId !== undefined ? "/card" : "echo", args = '{"message": "hi"}'] = argv; -// Each stage goes to stderr as it is reached, so a run that hangs shows the -// last one it got to; stdout carries only what callers parse. -const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; -console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl, + controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/caller`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); -console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); -console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); console.log(`on the mesh as ${session.peerId}`); // The peer to call: an agent named by ID, or a provider of a published service. @@ -44,13 +38,9 @@ let provider: DiscoveredProvider = { peerId: peerId ?? "", addrs: [] }; if (peerId !== undefined) { // An agent is not in the discovery table; the SDK finds the path to its // peer ID through the routers. - console.error(`dialing ${peerId}`); await session.connect(peerId); } else { const providers = await session.discover(service); - for (const p of providers) { - console.error(`provider ${p.peerId}`); - } if (providers.length === 0) { throw new Error(`no member of the mesh serves ${service}`); } @@ -58,7 +48,6 @@ if (peerId !== undefined) { let reached = false; for (const candidate of providers) { try { - console.error(`dialing ${candidate.peerId}`); await session.connect(candidate); provider = candidate; reached = true; diff --git a/sdk/js/src/auth.ts b/sdk/js/src/auth.ts index 835deb09..15aea2c6 100644 --- a/sdk/js/src/auth.ts +++ b/sdk/js/src/auth.ts @@ -60,9 +60,10 @@ function framed(stream: Stream) { /** * Client side: presents frame on a new /mesh/auth/1.0.0 stream and returns * the peer's verified credential. trustedKeys are the control plane keys. + * The handshake has a timeout of its own; outer, when given, can end it sooner. */ -export async function authenticateWithPeer(conn: Connection, frame: Uint8Array, trustedKeys: Uint8Array[]): Promise { - const signal = AbortSignal.timeout(AUTH_HANDSHAKE_TIMEOUT_MS); +export async function authenticateWithPeer(conn: Connection, frame: Uint8Array, trustedKeys: Uint8Array[], outer?: AbortSignal): Promise { + const signal = outer === undefined ? AbortSignal.timeout(AUTH_HANDSHAKE_TIMEOUT_MS) : AbortSignal.any([outer, AbortSignal.timeout(AUTH_HANDSHAKE_TIMEOUT_MS)]); // A peer behind a router is reached over a relayed, limited connection. const stream = await conn.newStream(AUTH_PROTOCOL, { signal, runOnLimitedConnection: true }); try { diff --git a/sdk/js/src/controlplane.ts b/sdk/js/src/controlplane.ts index 636ed346..fd52126f 100644 --- a/sdk/js/src/controlplane.ts +++ b/sdk/js/src/controlplane.ts @@ -257,8 +257,8 @@ export class ControlPlaneClient { } /** GET /info: OIDC settings, router addresses and the ban list. Unauthenticated. */ - async info(): Promise { - const body = await this.#request("GET", "/info"); + async info(signal?: AbortSignal): Promise { + const body = await this.#request("GET", "/info", undefined, {}, signal); return fromBinary(ControlPlaneInfoResponseSchema, body); } @@ -266,8 +266,8 @@ export class ControlPlaneClient { * GET /keys: every signing key the control plane currently trusts, * verified against a key the caller already trusts (the enrollment key). */ - async keys(trusted: Uint8Array[]): Promise { - const body = await this.#request("GET", "/keys"); + async keys(trusted: Uint8Array[], signal?: AbortSignal): Promise { + const body = await this.#request("GET", "/keys", undefined, {}, signal); return verifyKeysResponse(fromBinary(KeysResponseSchema, body), trusted, this.#now()); } @@ -426,14 +426,15 @@ export class ControlPlaneClient { path: string, body?: Uint8Array, headers: Record = {}, + signal?: AbortSignal, ): Promise<{ ok: boolean; status: number; buf: Uint8Array; dateHeader: string | null }> { - const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(new Error(`control plane ${path}: timed out after ${this.#timeoutMs}ms`)), this.#timeoutMs); + // The caller's signal cancels the request itself, not only the wait for it. + const timeout = AbortSignal.timeout(this.#timeoutMs); try { const init: RequestInit = { method, headers: { Accept: PROTOBUF_CONTENT_TYPE, ...headers }, - signal: controller.signal, + signal: signal === undefined ? timeout : AbortSignal.any([signal, timeout]), }; if (body !== undefined) { init.body = new Uint8Array(body); @@ -445,13 +446,22 @@ export class ControlPlaneClient { throw new Error(`control plane ${path}: response of ${buf.length} bytes exceeds the ${MAX_RESPONSE_BYTES} byte limit`); } return { ok: resp.ok, status: resp.status, buf, dateHeader: resp.headers.get("date") }; - } finally { - clearTimeout(timer); + } catch (err) { + if (timeout.aborted && signal?.aborted !== true) { + throw new Error(`control plane ${path}: timed out after ${this.#timeoutMs}ms`); + } + throw err; } } - async #request(method: "GET" | "POST", path: string, body?: Uint8Array, headers: Record = {}): Promise { - const outcome = await this.#fetchOnce(method, path, body, headers); + async #request( + method: "GET" | "POST", + path: string, + body?: Uint8Array, + headers: Record = {}, + signal?: AbortSignal, + ): Promise { + const outcome = await this.#fetchOnce(method, path, body, headers, signal); if (!outcome.ok) { throw new ControlPlaneError(path, outcome.status, new TextDecoder().decode(outcome.buf)); } diff --git a/sdk/js/src/mesh.ts b/sdk/js/src/mesh.ts index 606ccc21..7567bd00 100644 --- a/sdk/js/src/mesh.ts +++ b/sdk/js/src/mesh.ts @@ -310,12 +310,13 @@ export class AgentMesh { * Trades the current biscuit for a fresh one and persists it. The control * plane redeems only the last biscuit it issued, so a lost refresh result * means re-enrolling; persisting before returning keeps that rare. - * Concurrent calls wait for each other. + * Concurrent calls wait for each other. A refresh whose signal has + * aborted by the time its turn comes does not start. */ - async refresh(): Promise { + async refresh(signal?: AbortSignal): Promise { const run = this.#refreshChain.then( - () => this.#refreshOnce(), - () => this.#refreshOnce(), + () => this.#refreshOnce(signal), + () => this.#refreshOnce(signal), ); this.#refreshChain = run.then( () => undefined, @@ -324,7 +325,8 @@ export class AgentMesh { return run; } - async #refreshOnce(): Promise { + async #refreshOnce(signal?: AbortSignal): Promise { + signal?.throwIfAborted(); let jwt: string | undefined; if (this.#jwtSource !== undefined) { try { @@ -336,6 +338,10 @@ export class AgentMesh { // Best effort: fall back to session-only refresh if the token source is temporarily unavailable. } } + // The last point to stop. The /refresh request itself is not cut short: + // the control plane spends the old biscuit when it answers, so only a + // refresh that runs to the end can save the new one. + signal?.throwIfAborted(); const result = await this.controlPlane.refresh({ identity: this.identity, biscuit: this.#credential.biscuit, @@ -379,25 +385,29 @@ export class AgentMesh { * addresses and the ban set. Each part is attempted even when another * fails; the errors are reported together. Gossip events only bring this * forward; they are never the only way state arrives. + * + * An aborted signal cancels the request in flight and throws its reason: + * nothing more is sent and nothing is written to the state directory. */ - async syncControlPlane(): Promise { + async syncControlPlane(signal?: AbortSignal): Promise { const errors: string[] = []; let keysChanged = false; let refreshed = false; try { - const keys = await this.controlPlane.keys(this.#credential.controlPlaneKeys); + const keys = await this.controlPlane.keys(this.#credential.controlPlaneKeys, signal); if (keys.length === 0) { throw new Error("/keys returned no keys"); } keysChanged = !sameKeySet(keys, this.#credential.controlPlaneKeys); this.#credential = withCredentialMethods({ ...this.#credential, controlPlaneKeys: keys }); if (credentialPredatesRotation(this.#credential)) { - await this.refresh(); + await this.refresh(signal); refreshed = true; } } catch (err) { errors.push(`keys: ${err instanceof Error ? err.message : String(err)}`); } + signal?.throwIfAborted(); let bannedPeerIds: string[] | undefined; let routers: RouterInfo[] | undefined; @@ -405,7 +415,7 @@ export class AgentMesh { // in the answer, so its absence must not be read as an unban. const fetchedAt = new Date(); try { - const info = await this.controlPlane.info(); + const info = await this.controlPlane.info(signal); routers = info.routers; this.#routers = info.routers; const discovered = @@ -419,6 +429,7 @@ export class AgentMesh { } catch (err) { errors.push(`info: ${err instanceof Error ? err.message : String(err)}`); } + signal?.throwIfAborted(); if (keysChanged || bannedPeerIds !== undefined) { await this.save(); } diff --git a/sdk/js/src/session.test.ts b/sdk/js/src/session.test.ts index 153728b8..9f3633b4 100644 --- a/sdk/js/src/session.test.ts +++ b/sdk/js/src/session.test.ts @@ -44,7 +44,7 @@ import { Identity } from "./identity.ts"; import { HTTP_HANDLER_OPTIONS, HTTP_PROTOCOL, a2aEndpoint, httpIngressHandler } from "./libp2p-http.ts"; import { LabelsNotSatisfiedError } from "./mcp.ts"; import { AgentMesh } from "./mesh.ts"; -import { MeshSession, candidatesFromRouterInfos, joinDeadline, selectRouters, type JoinOptions } from "./session.ts"; +import { JOIN_SYNC_TIMEOUT_MS, JOIN_TIMEOUT_MS, MeshSession, candidatesFromRouterInfos, joinMesh, selectRouters, type JoinOptions } from "./session.ts"; type Wasm = Awaited>; type KeyPair = InstanceType; @@ -127,6 +127,52 @@ function fakeControlPlane(routerAddresses: string[], routers: RouterInfo[] = []) }) as typeof fetch; } +/** + * A control plane whose /keys and /info, once stall() is called, take the + * request and never answer until it is aborted. paths lists every request + * made after stall(), in order. + */ +function stallingControlPlane(routes: typeof fetch): { fetch: typeof fetch; stall(): void; paths: string[] } { + let stalled = false; + const paths: string[] = []; + const stalling = (async (input: Parameters[0], init?: RequestInit) => { + const path = new URL(new Request(input, init).url).pathname; + if (!stalled) { + return routes(input, init); + } + paths.push(path); + if (path !== "/keys" && path !== "/info") { + return routes(input, init); + } + return new Promise((_, reject) => { + const signal = init?.signal; + if (signal === undefined || signal === null) { + reject(new Error(`${path} was sent without a signal and would never end`)); + return; + } + if (signal.aborted) { + reject(signal.reason); + return; + } + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + }) as typeof fetch; + return { fetch: stalling, stall: () => { stalled = true; }, paths }; +} + +/** A libp2p host that takes the auth stream and never writes back: a router whose handshake neither completes nor fails. */ +async function silentRouter(): Promise { + const node = await createLibp2p({ + addresses: { listen: ["/ip4/127.0.0.1/tcp/0"] }, + transports: [tcp()], + connectionEncrypters: [tls()], + streamMuxers: [yamux()], + services: { identify: identify() }, + }); + await node.handle(AUTH_PROTOCOL, () => {}); + return node; +} + before(async () => { wasm = await loadBiscuit(); cpKeyPair = biscuitKeyPair(cpIdentity); @@ -606,17 +652,8 @@ test("join reports a router that refuses the handshake", async () => { }); test("a join bounded by its signal names the router whose handshake never answered", async () => { - const silent = await createLibp2p({ - addresses: { listen: ["/ip4/127.0.0.1/tcp/0"] }, - transports: [tcp()], - connectionEncrypters: [tls()], - streamMuxers: [yamux()], - services: { identify: identify() }, - }); + const silent = await silentRouter(); try { - // Takes the auth stream and never writes back, as a router whose - // handshake neither completes nor fails. - await silent.handle(AUTH_PROTOCOL, () => {}); const addr = (silent.getMultiaddrs()[0] as ReturnType).toString(); const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane([addr]) }); const started = Date.now(); @@ -631,40 +668,34 @@ test("a join bounded by its signal names the router whose handshake never answer } }); -test("a join bounded by its signal says when the control plane never answered", async () => { - const pending: Array<() => void> = []; - let hang = false; - const routes = fakeControlPlane([routerAddr]); - // Once enrolled, /keys and /info take the request and never answer, until - // the request is aborted or the test ends. - const stalling = (async (input: Parameters[0], init?: RequestInit) => { - const path = new URL(new Request(input, init).url).pathname; - if (!hang || (path !== "/keys" && path !== "/info")) { - return routes(input, init); - } - return new Promise((_, reject) => { - const stop = () => reject(init?.signal?.reason ?? new Error("test over")); - pending.push(stop); - init?.signal?.addEventListener("abort", stop, { once: true }); - }); - }) as typeof fetch; +test("a join stopped by its signal while syncing sends nothing more to the control plane", async () => { + const cp = stallingControlPlane(fakeControlPlane([routerAddr])); + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: cp.fetch }); + cp.stall(); + const started = Date.now(); + await assert.rejects(mesh.join({ signal: AbortSignal.timeout(500) }), (err: Error) => { + assert.match(err.message, /^join stopped while syncing with the control plane: /, `got ${JSON.stringify(err.message)}, want the step the join stopped in`); + assert.doesNotMatch(err.message, /127\.0\.0\.1:1/, `got ${JSON.stringify(err.message)}, want no control plane URL in the error`); + return true; + }); + const took = Date.now() - started; + assert.ok(took < JOIN_SYNC_TIMEOUT_MS, `join took ${took}ms, want the caller's signal to end it before the ${JOIN_SYNC_TIMEOUT_MS}ms sync timeout`); + // The aborted /keys request ends the sync: no /info, no refresh. + assert.deepEqual(cp.paths, ["/keys"], "requests after the control plane stalled"); +}); + +test("a join goes on with the stored credential when the control plane does not answer", async () => { + const cp = stallingControlPlane(fakeControlPlane([routerAddr])); + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: cp.fetch }); + cp.stall(); + const started = Date.now(); + const session = await mesh.join({ refreshLeadMs: 0, reserveRelay: false, controlPlaneSyncIntervalMs: 0 }); try { - const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: stalling }); - hang = true; - const started = Date.now(); - await assert.rejects(mesh.join({ signal: AbortSignal.timeout(500) }), (err: Error) => { - assert.match(err.message, /syncing with the control plane at http:\/\/127\.0\.0\.1:1/, `got ${JSON.stringify(err.message)}, want the control plane named`); - return true; - }); const took = Date.now() - started; - assert.ok(took < 10_000, `join took ${took}ms, want the join's signal to end it, not the control plane client's 30s request timeout`); + assert.ok(took < JOIN_TIMEOUT_MS, `join took ${took}ms, want the sync to give up after about ${JOIN_SYNC_TIMEOUT_MS}ms`); + assert.ok(session.isHeldRouter(router.peerId.toString()), `held routers ${session.routers.map((r) => r.peerId).join(",")}, want the router from the stored credential`); } finally { - // The sync the join gave up on goes on to its next request; let it - // through so no request is left holding the client's 30s timer. - hang = false; - for (const stop of pending) { - stop(); - } + await session.close(); } }); @@ -705,17 +736,30 @@ test("a top-up whose signal has aborted tries no router and shuns none", async ( } }); -test("a join without a signal of its own still has a deadline", async () => { - const given = AbortSignal.timeout(60_000); - const own = joinDeadline(given, 20); - assert.equal(own.signal, given, "a caller's signal bounds the join as given"); - own.clear(); - - const deadline = joinDeadline(undefined, 20); - await new Promise((resolve) => deadline.signal.addEventListener("abort", resolve, { once: true })); - const reason = deadline.signal.reason as Error; - assert.match(reason.message, /^join timed out after 20ms$/, `abort reason ${JSON.stringify(reason.message)}`); - deadline.clear(); +test("a join that runs out of its own time keeps the router that admitted it", async () => { + const silent = await silentRouter(); + try { + const silentId = silent.peerId.toString(); + const silentAddr = (silent.getMultiaddrs()[0] as ReturnType).toString(); + // The idle router is tried first, the silent one second. + const routers = [ + create(RouterInfoSchema, { peerId: router.peerId.toString(), addresses: [routerAddr], connections: 0, connectionLimit: 100 }), + create(RouterInfoSchema, { peerId: silentId, addresses: [silentAddr], connections: 90, connectionLimit: 100 }), + ]; + const mesh = await AgentMesh.enroll({ controlPlaneUrl: "http://127.0.0.1:1", bootstrapToken: "sbt", fetch: fakeControlPlane([routerAddr, silentAddr], routers) }); + const started = Date.now(); + const session = await joinMesh(mesh, { refreshLeadMs: 0, reserveRelay: false, controlPlaneSyncIntervalMs: 0 }, 2_000); + try { + const took = Date.now() - started; + assert.ok(took < AUTH_HANDSHAKE_TIMEOUT_MS, `join took ${took}ms, want its own timeout to end it before the ${AUTH_HANDSHAKE_TIMEOUT_MS}ms handshake timeout`); + assert.deepEqual(session.routers.map((r) => r.peerId), [router.peerId.toString()], "held routers"); + assert.equal(session.shunnedRouters.has(silentId), false, `${silentId} was shunned for a handshake the join's timeout cut off`); + } finally { + await session.close(); + } + } finally { + await silent.stop(); + } }); test("an explicit rejection is reported with its reason", async () => { diff --git a/sdk/js/src/session.ts b/sdk/js/src/session.ts index cdafda10..54c04377 100644 --- a/sdk/js/src/session.ts +++ b/sdk/js/src/session.ts @@ -128,8 +128,8 @@ export interface JoinOptions extends MeshHostOptions { /** * Bounds the whole join: the control plane sync, and the dial, auth * handshake and relay reservation with every router tried. Without one the - * join is bounded by JOIN_TIMEOUT_MS (60s). An aborted join throws an error - * naming each router tried and the step it was in. + * join is bounded by JOIN_TIMEOUT_MS (60s). A join that stops throws an + * error naming each router tried and the step it was in. */ signal?: AbortSignal; } @@ -208,6 +208,12 @@ const WITHDRAW_TIMEOUT_MS = 2 * 1000; * its own, but a join tries them in turn over every router it is offered. */ export const JOIN_TIMEOUT_MS = 60_000; +/** + * Bounds the control plane sync a join starts with. The sync is best effort, + * so a control plane that does not answer in time leaves the join to go on + * with the stored credential instead of using up the whole join. + */ +export const JOIN_SYNC_TIMEOUT_MS = 5_000; /** How long a provider's positive egress verdict is kept; agentmesh-node's labelGateTTL. */ const EGRESS_VERDICT_TTL_MS = 5 * 60 * 1000; @@ -564,14 +570,15 @@ export class MeshSession { /** * Runs the handshake on a router connection once: the connection:open * listener and an explicit admission both call this, and the second waits - * for the first instead of running a second handshake. + * for the first instead of running a second handshake. The signal applies + * to a handshake this call starts; one already running keeps its own. */ - #authenticateRouter(conn: Connection): Promise { + #authenticateRouter(conn: Connection, signal?: AbortSignal): Promise { const pending = this.#routerHandshakes.get(conn.id); if (pending !== undefined) { return pending; } - const run = authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys).finally(() => { + const run = authenticateWithPeer(conn, this.mesh.authFrame(), this.mesh.credential.controlPlaneKeys, signal).finally(() => { this.#routerHandshakes.delete(conn.id); }); this.#routerHandshakes.set(conn.id, run); @@ -722,22 +729,24 @@ export class MeshSession { try { const conn = await this.node.dial(cand.addresses, signal !== undefined ? { signal } : {}); step = "auth handshake"; - const credential = await untilAborted(this.#authenticateRouter(conn), signal); + const credential = await this.#authenticateRouter(conn, signal); requireRole(credential, ROLE_ROUTER); const addr = connectedAddress(conn, cand.peerId); if (this.#reserveRelay) { step = "relay reservation"; - await untilAborted(this.#reserveOnRouter(cand.peerId, addr), signal); + await this.#reserveOnRouter(cand.peerId, addr); } this.#upsertHeldRouter({ peerId: cand.peerId, addr, credential }); progress = true; } catch (err) { - this.#shunRouter(cand.peerId, ROUTER_SHUN_DURATION_MS); - await this.node.hangUp(peerIdFromString(cand.peerId)).catch(() => {}); failures.push(`${cand.peerId}: ${step}: ${errorText(err)}`); + await this.node.hangUp(peerIdFromString(cand.peerId)).catch(() => {}); + // A router cut off by the signal did not fail on its own, so it is + // not shunned. if (aborted()) { return failures; } + this.#shunRouter(cand.peerId, ROUTER_SHUN_DURATION_MS); } } if (!progress && picks.length < need) { @@ -1434,30 +1443,13 @@ export function selectRouters(candidates: readonly RouterCandidate[], k: number, } /** - * The signal a join runs under: the caller's, or one that aborts after ms - * with an error saying the join timed out. clear() stops the timer once the - * join is over. + * Implements AgentMesh.join(); lives here to keep mesh.ts free of libp2p. + * timeoutMs bounds a join whose caller passed no signal. */ -export function joinDeadline(signal: AbortSignal | undefined, ms = JOIN_TIMEOUT_MS): { signal: AbortSignal; clear(): void } { - if (signal !== undefined) { - return { signal, clear: () => {} }; - } - const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(new Error(`join timed out after ${ms}ms`)), ms); - return { signal: controller.signal, clear: () => clearTimeout(timer) }; -} - -/** Implements AgentMesh.join(); lives here to keep mesh.ts free of libp2p. */ -export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}): Promise { - const deadline = joinDeadline(options.signal); - try { - return await joinUnder(mesh, options, deadline.signal); - } finally { - deadline.clear(); - } -} - -async function joinUnder(mesh: AgentMesh, options: JoinOptions, signal: AbortSignal): Promise { +export async function joinMesh(mesh: AgentMesh, options: JoinOptions = {}, timeoutMs = JOIN_TIMEOUT_MS): Promise { + const signal = options.signal ?? AbortSignal.timeout(timeoutMs); + // Why the join stopped: the caller's reason, or its own timeout. + const stopReason = () => (options.signal !== undefined ? errorText(signal.reason) : `timed out after ${timeoutMs}ms`); // The pull agentmesh-node makes before it starts: a member resuming from its // state directory after a key rotation would otherwise verify the routers, // which already hold credentials under the new key, against the keys it @@ -1465,12 +1457,12 @@ async function joinUnder(mesh: AgentMesh, options: JoinOptions, signal: AbortSig // once joined. Best effort; the stored credential serves when the control // plane cannot be reached. try { - await untilAborted(mesh.syncControlPlane(), signal); + await mesh.syncControlPlane(AbortSignal.any([signal, AbortSignal.timeout(JOIN_SYNC_TIMEOUT_MS)])); } catch { - // Joining goes on with what the credential holds, unless the join itself - // ran out while the control plane kept it waiting. + // Joining goes on with what the credential holds, unless the caller + // stopped the join while the control plane kept it waiting. if (signal.aborted) { - throw new Error(`join stopped while syncing with the control plane at ${mesh.controlPlane.url}: ${errorText(signal.reason)}`); + throw new Error(`join stopped while syncing with the control plane: ${stopReason()}`); } } const rawAddrs = options.routerAddresses ?? mesh.credential.routerAddresses; @@ -1508,13 +1500,17 @@ async function joinUnder(mesh: AgentMesh, options: JoinOptions, signal: AbortSig session = new MeshSession(mesh, node, [], authenticatedPeers, banned, options); const failures = await session.topUpRouters(signal); - if (signal.aborted || session.routers.length === 0) { + // One admitted router is enough to be on the mesh; only the caller's own + // signal turns that into a failure. + if (options.signal?.aborted === true || session.routers.length === 0) { const detail = failures.length > 0 ? `:\n ${failures.join("\n ")}` : ""; - throw new Error(signal.aborted ? `join stopped: ${errorText(signal.reason)}${detail}` : `no router admitted this member${detail}`); + throw new Error(signal.aborted ? `join stopped: ${stopReason()}${detail}` : `no router admitted this member${detail}`); } // Best effort: a router that does not take the record is a warning the - // next announcement retries; the member is on the mesh regardless. - await session.announce(signal).catch(() => {}); + // next announcement retries; the member is on the mesh regardless. Under + // the caller's signal only: a join that ran out of its own time still + // announces the routers it holds. + await session.announce(options.signal).catch(() => {}); return session; } catch (err) { if (session !== undefined) { @@ -1526,26 +1522,6 @@ async function joinUnder(mesh: AgentMesh, options: JoinOptions, signal: AbortSig } } -/** - * Settles as p does, or rejects with the signal's reason as soon as it - * aborts. p is not cancelled: it ends on its own timeout, and the caller - * tears down what it was using. - */ -function untilAborted(p: Promise, signal: AbortSignal | undefined): Promise { - if (signal === undefined) { - return p; - } - if (signal.aborted) { - p.catch(() => {}); - return Promise.reject(signal.reason); - } - return new Promise((resolve, reject) => { - const onAbort = () => reject(signal.reason); - signal.addEventListener("abort", onAbort, { once: true }); - void p.then(resolve, reject).finally(() => signal.removeEventListener("abort", onAbort)); - }); -} - function errorText(err: unknown): string { return err instanceof Error ? err.message : String(err); } diff --git a/site/content/docs/guides/native-sdks.md b/site/content/docs/guides/native-sdks.md index 85743f36..6035e758 100644 --- a/site/content/docs/guides/native-sdks.md +++ b/site/content/docs/guides/native-sdks.md @@ -286,21 +286,15 @@ if (argv[0] !== undefined && !argv[0].includes("://")) { } const [service = peerId !== undefined ? "a2a://agent" : "mcp://everything", toolOrPath = peerId !== undefined ? "/card" : "echo", args = '{"message": "hi"}'] = argv; -// Each stage goes to stderr as it is reached, so a run that hangs shows the -// last one it got to; stdout carries only what callers parse. -const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; -console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl, + controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/caller`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); -console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); -console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); console.log(`on the mesh as ${session.peerId}`); // The peer to call: an agent named by ID, or a provider of a published service. @@ -308,13 +302,9 @@ let provider: DiscoveredProvider = { peerId: peerId ?? "", addrs: [] }; if (peerId !== undefined) { // An agent is not in the discovery table; the SDK finds the path to its // peer ID through the routers. - console.error(`dialing ${peerId}`); await session.connect(peerId); } else { const providers = await session.discover(service); - for (const p of providers) { - console.error(`provider ${p.peerId}`); - } if (providers.length === 0) { throw new Error(`no member of the mesh serves ${service}`); } @@ -322,7 +312,6 @@ if (peerId !== undefined) { let reached = false; for (const candidate of providers) { try { - console.error(`dialing ${candidate.peerId}`); await session.connect(candidate); provider = candidate; reached = true; @@ -536,21 +525,15 @@ import { AgentEvent, DefaultRequestHandler, InMemoryTaskStore, type AgentExecuto import { UserBuilder, agentCardHandler, jsonRpcHandler } from "@a2a-js/sdk/server/express"; import { AgentMesh, MeshSession } from "@agentmesh-p2p/sdk"; -// Each stage goes to stderr as it is reached, so a run that hangs shows the -// last one it got to; stdout carries only what callers parse. -const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; -console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl, + controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/a2a-agent`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); -console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); -console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); // The card names the agent as the mesh reaches it: the URL an A2A client on // the mesh gives its fetch, http://mesh/mesh//a2a/agent. @@ -753,21 +736,15 @@ if (peerId === undefined) { throw new Error("usage: a2a-call.js [text]"); } -// Each stage goes to stderr as it is reached, so a run that hangs shows the -// last one it got to; stdout carries only what callers parse. -const controlPlaneUrl = process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com"; -console.error(`enrolling with ${controlPlaneUrl}`); const mesh = await AgentMesh.enroll({ - controlPlaneUrl, + controlPlaneUrl: process.env.AGENTMESH_CONTROL_PLANE_URL ?? "https://mesh.example.com", bootstrapTokenPath: process.env.AGENTMESH_BOOTSTRAP_TOKEN_PATH, jwtPath: process.env.AGENTMESH_JWT_PATH, stateDir: process.env.AGENTMESH_STATE_DIR ?? `${homedir()}/.config/agentmesh/a2a-caller`, // A plaintext http:// control plane is otherwise accepted only on loopback. allowInsecure: process.env.AGENTMESH_INSECURE_CONTROL_PLANE === "true", }); -console.error(`enrolled ${mesh.peerId}`); const session = await mesh.join(); -console.error(`joined via ${session.routers.map((r) => r.peerId).join(", ")}`); console.log(`on the mesh as ${session.peerId}`); // The agent's URL on the mesh, and a fetch that carries requests to it. @@ -778,7 +755,6 @@ const factory = new ClientFactory({ cardResolver: new DefaultAgentCardResolver({ fetchImpl }), }); // The card lives under the agent's URL; the trailing slash keeps the last segment when the resolver appends the well-known path. -console.error(`dialing ${peerId}`); const client = await factory.createFromUrl(agentURL + "/"); const card = await client.getAgentCard(); console.log(`agent: ${card.name}, ${card.description}`);