diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index b93db1244..5c3c4e9f0 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -905,6 +905,8 @@ export async function createHub(config: HubConfig) { }, }, }); + const isSidecarRoutable = (address: string) => + sidecarRouter.getRoutableAddresses().includes(address); // A finalized turn's persisted-artifact tool-call results become // delivery file parts (CL-6000) via `createArtifactDeliveryHandler`, // built once `chatStore`/`chatPlatform` exist further down this @@ -2487,8 +2489,7 @@ export async function createHub(config: HubConfig) { workflowAllocationService, credentialCipher, eventCollectors, - isRoutable: (address) => - sidecarRouter.getRoutableAddresses().includes(address), + isRoutable: isSidecarRoutable, cryptoProviderCache: cryptoProviders, persistLaunch: async (input) => { await workbenchLaunchPersistExtra(input)(db); @@ -3303,6 +3304,7 @@ export async function createHub(config: HubConfig) { workflowAllocationService, sessionService, eventCollectors, + isRoutable: isSidecarRoutable, }, runnerInput, ), diff --git a/apps/hub/src/routability-wiring.test.ts b/apps/hub/src/routability-wiring.test.ts new file mode 100644 index 000000000..d06dbbf02 --- /dev/null +++ b/apps/hub/src/routability-wiring.test.ts @@ -0,0 +1,54 @@ +// CL-7543 follow-up: the planner's one-shot runner and the webhook +// launch path must both consult the hub's live sidecar routing table +// through the shared `isSidecarRoutable` local. A regression replacing +// either wiring with a constant (`() => true`) or dropping the hoisted +// predicate would still typecheck; this source scan is the pin, in the +// same style as `crypto-provider-cache-wiring.test.ts`. +import { describe, expect, test } from "bun:test"; +import { readFileSync } from "node:fs"; +import path from "node:path"; + +const HUB_INDEX = path.join(import.meta.dir, "index.ts"); + +/** The full `callee(...)` text of the first call in `source`. */ +function firstCall(source: string, callee: string): string { + const token = `${callee}(`; + const start = source.indexOf(token); + if (start < 0) { + throw new Error(`expected ${callee}(...) in hub index.ts`); + } + let depth = 1; + let i = start + token.length; + while (i < source.length && depth > 0) { + const ch = source[i]; + if (ch === "(") depth += 1; + else if (ch === ")") depth -= 1; + i += 1; + } + return source.slice(start, i); +} + +describe("hub sidecar routability wiring", () => { + test("one shared isSidecarRoutable predicate feeds both launch paths", () => { + const source = readFileSync(HUB_INDEX, "utf8"); + + const assigned = + /const\s+isSidecarRoutable\s*=\s*\(address:\s*string\)\s*=>\s*\n?\s*sidecarRouter\.getRoutableAddresses\(\)\.includes\(address\)/.exec( + source, + ); + expect(assigned).not.toBeNull(); + + // Exactly one definition: no second, diverging routability source. + const definitionCount = source.match( + /getRoutableAddresses\(\)\.includes\(/g, + ); + expect(definitionCount).toHaveLength(1); + + expect(firstCall(source, "runOneShotPrompt")).toContain( + "isRoutable: isSidecarRoutable", + ); + expect(firstCall(source, "launchWebhookTrigger")).toContain( + "isRoutable: isSidecarRoutable", + ); + }); +}); diff --git a/packages/agent-directory/src/agent-definition-draft-routes.ts b/packages/agent-directory/src/agent-definition-draft-routes.ts index 177c2a6f3..1ab3b82dc 100644 --- a/packages/agent-directory/src/agent-definition-draft-routes.ts +++ b/packages/agent-directory/src/agent-definition-draft-routes.ts @@ -17,6 +17,7 @@ import type { TenantEnv, RequireGrant } from "@intx/hub-api"; import { OneShotRunFailedError, OneShotRunTimedOutError, + OneShotRunUnreachableError, } from "./one-shot-prompt"; import { makeErrorEnvelope, reportError } from "@corbits/error-sink"; import { @@ -64,6 +65,7 @@ function isDraftingFailure(err: unknown): boolean { err instanceof MyraAgentDefinitionDraftingUnavailableError || err instanceof OneShotRunTimedOutError || err instanceof OneShotRunFailedError || + err instanceof OneShotRunUnreachableError || err instanceof AgentDefinitionDraftReplyUnparseableError || err instanceof AgentDefinitionDraftReferenceOutOfInventoryError ); diff --git a/packages/agent-directory/src/index.ts b/packages/agent-directory/src/index.ts index 19e11d3ab..cba857b86 100644 --- a/packages/agent-directory/src/index.ts +++ b/packages/agent-directory/src/index.ts @@ -123,6 +123,7 @@ export { OneShotDefinitionNotFoundError, OneShotRunTimedOutError, OneShotRunFailedError, + OneShotRunUnreachableError, type OneShotReply, type OneShotRunnerDeps, type OneShotPromptInput, diff --git a/packages/agent-directory/src/one-shot-prompt.test.ts b/packages/agent-directory/src/one-shot-prompt.test.ts index 4bc4ceb5b..434e109f5 100644 --- a/packages/agent-directory/src/one-shot-prompt.test.ts +++ b/packages/agent-directory/src/one-shot-prompt.test.ts @@ -3,10 +3,11 @@ import { describe, expect, test } from "bun:test"; import { - runOneShotPrompt, OneShotDefinitionNotFoundError, OneShotRunFailedError, OneShotRunTimedOutError, + OneShotRunUnreachableError, + runOneShotPrompt, } from "./one-shot-prompt"; /** Asserts a fake's call list recorded at least one call and returns the @@ -107,8 +108,10 @@ function createFakeProvision() { }; } -/** A fake `sendMail`: records every call and either succeeds or throws. */ -function createFakeSend(behavior: "ok" | "throws" = "ok"): { +/** A fake `sendMail`: records every call. A test scripts per-attempt + * failures by 1-based attempt number in `throwOn`; every other attempt + * succeeds. */ +function createFakeSend(throwOn: Record = {}): { calls: number; sendMail: (...args: unknown[]) => Promise; } { @@ -119,9 +122,8 @@ function createFakeSend(behavior: "ok" | "throws" = "ok"): { }, sendMail: async () => { calls++; - if (behavior === "throws") { - throw new Error("cipher unavailable"); - } + const thrown = throwOn[calls]; + if (thrown !== undefined) throw thrown; }, }; } @@ -166,6 +168,7 @@ function createBaseDeps() { repoStore: { resolveRef: async () => "sha_test" }, workflowAllocationService: {}, sessionService: {}, + isRoutable: () => true, cryptoProviders: { async get() { return {}; @@ -186,7 +189,7 @@ describe("runOneShotPrompt", () => { test("happy path resolves with accumulated reply content, tears the run down, and untracks it", async () => { const fake = createFakeEmitter(); const { provision, calls: launchCalls } = createFakeProvision(); - const fakeSend = createFakeSend("ok"); + const fakeSend = createFakeSend(); const { sendMail } = fakeSend; const { undeploy, calls: undeployCalls } = createFakeUndeploy(); const { lifecycle, tracked, activity, untracked } = createFakeLifecycle(); @@ -240,7 +243,7 @@ describe("runOneShotPrompt", () => { test("a failed run rejects with OneShotRunFailedError, unsubscribes, and tears the run down", async () => { const fake = createFakeEmitter(); const { provision, calls: launchCalls } = createFakeProvision(); - const { sendMail } = createFakeSend("ok"); + const { sendMail } = createFakeSend(); const { undeploy, calls: undeployCalls } = createFakeUndeploy(); const deps = { ...createBaseDeps(), @@ -272,7 +275,7 @@ describe("runOneShotPrompt", () => { test("an unknown definition throws OneShotDefinitionNotFoundError", async () => { const fake = createFakeEmitter(); const { provision } = createFakeProvision(); - const { sendMail } = createFakeSend("ok"); + const { sendMail } = createFakeSend(); const { undeploy } = createFakeUndeploy(); const deps = { ...createBaseDeps(), @@ -298,7 +301,7 @@ describe("send-path throw (not an !ok result)", () => { test("a throwing cryptoProviders.get is caught, torn down, and rejects promptly with the real cause", async () => { const fake = createFakeEmitter(); const { provision, calls: launchCalls } = createFakeProvision(); - const { sendMail } = createFakeSend("ok"); + const { sendMail } = createFakeSend(); const { undeploy, calls: undeployCalls } = createFakeUndeploy(); const deps = { ...createBaseDeps(), @@ -338,7 +341,7 @@ describe("timeout tears the launched run down", () => { test("a timeout unsubscribes AND undeploys the run it launched, before rejecting", async () => { const fake = createFakeEmitter(); const { provision, calls: launchCalls } = createFakeProvision(); - const { sendMail } = createFakeSend("ok"); + const { sendMail } = createFakeSend(); const { undeploy, calls: undeployCalls } = createFakeUndeploy(); const deps = { ...createBaseDeps(), @@ -360,3 +363,203 @@ describe("timeout tears the launched run down", () => { ]); }); }); + +// The bounded-wait mechanics these tests exercise are documented on +// `OneShotRunnerDeps.isRoutable` — that JSDoc is the authoritative +// telling; the tests below pin the observable behavior. +describe("routable wait on the opening send", () => { + const UNREACHABLE = new Error("agent is unreachable: run_1@acme.example"); + + test("retries the opening send right away when the address is routable after the first failure", async () => { + const fake = createFakeEmitter(); + const { provision, calls: launchCalls } = createFakeProvision(); + const send = createFakeSend({ 1: UNREACHABLE }); + const { undeploy, calls: undeployCalls } = createFakeUndeploy(); + const deps = { + ...createBaseDeps(), + events: fake.emitter, + provision, + sendMail: send.sendMail, + undeploy, + // Routable from the first consult on — no polling, one retry. + isRoutable: () => send.calls >= 1, + } as never; + + const promise = runOneShotPrompt(deps, INPUT); + await new Promise((r) => setTimeout(r, 10)); + const triggerAddress = firstCall(launchCalls).address; + + fake.emit("agent.event", { + agentAddress: triggerAddress, + event: { type: "connector.reply", data: { content: "Hello" } }, + }); + fake.emit("agent.event", { + agentAddress: triggerAddress, + event: { type: "message.run.ended", data: { status: "completed" } }, + }); + + const result = await promise; + expect(result.content).toBe("Hello"); + expect(send.calls).toBe(2); + expect(undeployCalls).toEqual([ + { address: triggerAddress, reason: "planning-run-complete" }, + ]); + }); + + test("rejects OneShotRunUnreachableError when the routable wait expires", async () => { + const fake = createFakeEmitter(); + const { provision, calls: launchCalls } = createFakeProvision(); + const send = createFakeSend({ 1: UNREACHABLE, 2: UNREACHABLE }); + const { undeploy, calls: undeployCalls } = createFakeUndeploy(); + const deps = { + ...createBaseDeps(), + events: fake.emitter, + provision, + sendMail: send.sendMail, + undeploy, + isRoutable: () => false, + deliverWait: { + deadlineMs: 50, + pollIntervalMs: 5, + sleep: (ms: number) => new Promise((r) => setTimeout(r, ms)), + }, + } as never; + + let caught: unknown; + try { + await runOneShotPrompt(deps, INPUT); + } catch (err) { + caught = err; + } + const triggerAddress = firstCall(launchCalls).address; + + expect(caught).toBeInstanceOf(OneShotRunUnreachableError); + expect((caught as OneShotRunUnreachableError).cause).toBe(UNREACHABLE); + // `deliverWhenRoutable` sends once, polls, and gives up at the + // deadline without ever reaching its second send. + expect(send.calls).toBe(1); + expect(undeployCalls).toEqual([ + { address: triggerAddress, reason: "planning-run-send-unreachable" }, + ]); + expect(fake.listenerCount("agent.event")).toBe(0); + }); + + test("routable mid-wait: flips true after several polls and the retried send completes", async () => { + const fake = createFakeEmitter(); + const { provision, calls: launchCalls } = createFakeProvision(); + const send = createFakeSend({ 1: UNREACHABLE }); + const { undeploy, calls: undeployCalls } = createFakeUndeploy(); + let routabilityChecks = 0; + const routabilityAddresses: string[] = []; + const deps = { + ...createBaseDeps(), + events: fake.emitter, + provision, + sendMail: send.sendMail, + undeploy, + isRoutable: (address: string) => { + routabilityChecks++; + routabilityAddresses.push(address); + return routabilityChecks > 3; + }, + deliverWait: { + deadlineMs: 5_000, + pollIntervalMs: 1, + sleep: (ms: number) => new Promise((r) => setTimeout(r, ms)), + }, + } as never; + + const promise = runOneShotPrompt(deps, INPUT); + await new Promise((r) => setTimeout(r, 10)); + const triggerAddress = firstCall(launchCalls).address; + + fake.emit("agent.event", { + agentAddress: triggerAddress, + event: { type: "connector.reply", data: { content: "Hello" } }, + }); + fake.emit("agent.event", { + agentAddress: triggerAddress, + event: { type: "message.run.ended", data: { status: "completed" } }, + }); + + const result = await promise; + expect(result.content).toBe("Hello"); + expect(routabilityChecks).toBeGreaterThan(3); + // The launched run's address, not some placeholder, is what the + // runner consults. + expect(routabilityAddresses).toContain(triggerAddress); + expect(send.calls).toBe(2); + expect(undeployCalls).toEqual([ + { address: triggerAddress, reason: "planning-run-complete" }, + ]); + }); + + test("the reply timer wins over the routable wait and no send follows the teardown", async () => { + const fake = createFakeEmitter(); + const { provision, calls: launchCalls } = createFakeProvision(); + const send = createFakeSend({ 1: UNREACHABLE }); + const { undeploy, calls: undeployCalls } = createFakeUndeploy(); + const deps = { + ...createBaseDeps(), + events: fake.emitter, + provision, + sendMail: send.sendMail, + undeploy, + isRoutable: () => false, + deliverWait: { + deadlineMs: 5_000, + pollIntervalMs: 5, + // Cooperative: yields to macrotasks so the reply timer can fire + // mid-wait (a noop sleep would starve it). + sleep: (ms: number) => new Promise((r) => setTimeout(r, ms)), + }, + } as never; + + let caught: unknown; + try { + await runOneShotPrompt(deps, { ...INPUT, timeoutMs: 50 }); + } catch (err) { + caught = err; + } + + expect(caught).toBeInstanceOf(OneShotRunTimedOutError); + const triggerAddress = firstCall(launchCalls).address; + expect(undeployCalls).toEqual([ + { address: triggerAddress, reason: "planning-run-timed-out" }, + ]); + // Let any post-teardown send attempt surface before pinning the count. + await new Promise((r) => setTimeout(r, 20)); + expect(send.calls).toBeLessThanOrEqual(1); + expect(fake.listenerCount("agent.event")).toBe(0); + }); + + test("a non-unreachable send error is not retried and rejects with the original cause", async () => { + const fake = createFakeEmitter(); + const { provision, calls: launchCalls } = createFakeProvision(); + const original = new Error("cipher unavailable"); + const send = createFakeSend({ 1: original }); + const { undeploy, calls: undeployCalls } = createFakeUndeploy(); + const deps = { + ...createBaseDeps(), + events: fake.emitter, + provision, + sendMail: send.sendMail, + undeploy, + } as never; + + let caught: unknown; + try { + await runOneShotPrompt(deps, INPUT); + } catch (err) { + caught = err; + } + const triggerAddress = firstCall(launchCalls).address; + + expect(caught).toBe(original); + expect(send.calls).toBe(1); + expect(undeployCalls).toEqual([ + { address: triggerAddress, reason: "planning-run-send-failed" }, + ]); + expect(fake.listenerCount("agent.event")).toBe(0); + }); +}); diff --git a/packages/agent-directory/src/one-shot-prompt.ts b/packages/agent-directory/src/one-shot-prompt.ts index b03351e45..85dab5a3f 100644 --- a/packages/agent-directory/src/one-shot-prompt.ts +++ b/packages/agent-directory/src/one-shot-prompt.ts @@ -8,7 +8,9 @@ import type { AgentLifecycle } from "@corbits/agent-lifecycle"; import { connectorReplyContent, messageRunEnded } from "@corbits/agent-events"; import { reportError } from "@corbits/error-sink"; import { + deliverWhenRoutable, endAgentSessionForRun, + isAgentUnreachableError, readDefinitionProjection, readFoldedBody, recordAgentSessionAtProvision, @@ -59,6 +61,24 @@ export type OneShotRunnerDeps = { >; readonly sessionService: Pick; readonly eventCollectors: EventCollectorPort; + /** + * Reads the hub's live sidecar routing table (the same source the + * webhook launch path uses). A freshly provisioned run's sidecar + * takes several seconds to register, so the opening send fires once; + * if it fails as unreachable, this is polled — bounded — until the + * address is routable and the send is retried exactly once. + */ + readonly isRoutable: (address: string) => boolean; + /** + * Test seam only. Overrides `deliverWhenRoutable`'s default wait + * budget/poll so the expiry path stays fast under test; production + * never sets it and the defaults apply. + */ + readonly deliverWait?: { + readonly deadlineMs?: number; + readonly pollIntervalMs?: number; + readonly sleep?: (ms: number) => Promise; + }; /** * Test seam only. Production never sets these; they default to * Interchange `prepareProvisionedDeployment` and `sendUserMessage`. @@ -113,6 +133,18 @@ export class OneShotRunFailedError extends Error { } } +export class OneShotRunUnreachableError extends Error { + constructor(cause: unknown) { + super( + cause instanceof Error + ? `the one-shot run's opening send kept failing while its sidecar was unreachable: ${cause.message}` + : "the one-shot run's opening send kept failing while its sidecar was unreachable", + { cause }, + ); + this.name = "OneShotRunUnreachableError"; + } +} + async function provisionOnAsset( deps: OneShotRunnerDeps, input: { @@ -304,33 +336,52 @@ export async function runOneShotPrompt( void (async () => { try { const cryptoProvider = await deps.cryptoProviders.get(launched.runId); - if (deps.sendMail !== undefined) { - await deps.sendMail({ - tenantId: input.tenantId, - sessionId: launched.sessionId, - agentAddress: launched.address, - from: `${input.principalId}@${tenantRow.domain}`, - content: input.prompt, - cryptoProvider, - }); - } else { - await deps.sessionService.sendUserMessage({ - agentAddress: launched.address, - from: `${input.principalId}@${tenantRow.domain}`, - messageId: `<${crypto.randomUUID()}@${tenantRow.domain}>`, - date: new Date(), - content: input.prompt, - sessionId: launched.sessionId, - tenantId: input.tenantId, - cryptoProvider, - }); - } + await deliverWhenRoutable({ + send: async () => { + if (deps.sendMail !== undefined) { + await deps.sendMail({ + tenantId: input.tenantId, + sessionId: launched.sessionId, + agentAddress: launched.address, + from: `${input.principalId}@${tenantRow.domain}`, + content: input.prompt, + cryptoProvider, + }); + } else { + await deps.sessionService.sendUserMessage({ + agentAddress: launched.address, + from: `${input.principalId}@${tenantRow.domain}`, + messageId: `<${crypto.randomUUID()}@${tenantRow.domain}>`, + date: new Date(), + content: input.prompt, + sessionId: launched.sessionId, + tenantId: input.tenantId, + cryptoProvider, + }); + } + }, + isRoutable: () => deps.isRoutable(launched.address), + ...deps.deliverWait, + }); // A run's session and event collector are ensured lazily now, at // the hub seams that actually see the run become mail-routable // (`apps/hub/src/mailbox-persist.ts`, the wrapped // `eventCollectors` dispatch in `apps/hub/src/index.ts`) — // CL-7480. Nothing here needs to record it. } catch (cause) { + // The reply timer may have settled the run mid-wait; a late + // failure is then a phantom for an already-torn-down run. + if (settled) return; + if (isAgentUnreachableError(cause)) { + reportError(cause, { + operation: "agent-directory.one-shot.send-unreachable", + extra: { address: launched.address }, + }); + void settle("planning-run-send-unreachable", () => { + reject(new OneShotRunUnreachableError(cause)); + }); + return; + } reportError(cause, { operation: "agent-directory.one-shot.send", extra: { address: launched.address }, diff --git a/packages/agent-directory/test/agent-definition-draft-routes.test.ts b/packages/agent-directory/test/agent-definition-draft-routes.test.ts index a60eecf06..7e3f866d3 100644 --- a/packages/agent-directory/test/agent-definition-draft-routes.test.ts +++ b/packages/agent-directory/test/agent-definition-draft-routes.test.ts @@ -13,6 +13,7 @@ import { makeErrorEnvelope, parseErrorEnvelope } from "@corbits/error-sink"; import { MyraAgentDefinitionDraftingUnavailableError } from "../src/agent-definition-drafting"; import { createAgentDefinitionDraftRoutes } from "../src/agent-definition-draft-routes"; +import { OneShotRunUnreachableError } from "../src/one-shot-prompt"; const TENANT = { id: "tnt_1", @@ -133,6 +134,22 @@ describe("agent-definition draft route envelope", () => { expect(records[0]?.properties.tenantId).toBe("tnt_1"); }); + test("an OneShotRunUnreachableError maps to the 422 drafting_failed envelope", async () => { + const app = buildApp(() => + Promise.reject( + new OneShotRunUnreachableError(new Error("agent is unreachable")), + ), + ); + + const res = await postDraft(app, { name: "Research Buddy" }); + + expect(res.status).toBe(422); + const body: unknown = await res.json(); + const envelope = parseErrorEnvelope(body); + expect(envelope?.error.code).toBe("drafting_failed"); + expect(envelope?.error.userMessage).toBe(DRAFT_FAILED_MESSAGE); + }); + test("a malformed body answers 400 makeErrorEnvelope, not {code, message}", async () => { const app = buildApp(() => Promise.resolve({ diff --git a/packages/workflows/test/unreachable-message-contract.test.ts b/packages/workflows/test/unreachable-message-contract.test.ts new file mode 100644 index 000000000..d132f5ab1 --- /dev/null +++ b/packages/workflows/test/unreachable-message-contract.test.ts @@ -0,0 +1,34 @@ +// Contract pin between the two halves of unreachable detection: the +// vendor session service throws a plain Error whose message contains +// "agent is unreachable", and `isAgentUnreachableError` (which the +// one-shot drafting wait and the webhook launch both rely on) sniffs +// exactly that substring. Either side drifting silently disables the +// bounded retry and the typed 422 drafting failure; this scan fails +// loudly instead. The vendor file itself is never modified. +import { describe, expect, test } from "bun:test"; +import { readFileSync } from "node:fs"; +import path from "node:path"; + +const HELPER_SOURCE = path.join( + import.meta.dir, + "../src/deliver-when-routable.ts", +); +const VENDOR_SESSION_SERVICE = path.join( + import.meta.dir, + "../../../vendor/intx/hub-sessions/src/session-service.ts", +); + +describe("vendor unreachable message contract", () => { + test("the vendor throw and the substring classifier agree on 'agent is unreachable'", () => { + const helper = readFileSync(HELPER_SOURCE, "utf8"); + expect(helper).toContain('"agent is unreachable"'); + + const vendor = readFileSync(VENDOR_SESSION_SERVICE, "utf8"); + // Anchored to the throw expression itself, not any mention in a + // comment: only the live throw's message keeps the classifier's + // sniff honest, so only the throw may satisfy this pin. + expect(vendor).toMatch( + /throw new Error\(\s*`Failed to deliver message to \$\{agentAddress\}: agent is unreachable`/, + ); + }); +});