diff --git a/README.md b/README.md index 33af2d8c74..5eca2d47f4 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,10 @@ resources, and `agentcore/cdk/`, the AWS CDK app that deploys them. Deployment t and region each — live in `agentcore/aws-targets.json`. Project commands find the project by walking up from the current directory. +`agentcore dev` runs runtimes locally or deploys harnesses and prints invoke guidance. +Use `--mode browser` for Agent Inspector, `--skip-deploy` to reuse deployed harnesses, +and `--agent ` to choose an agent in mixed projects. + ## Commands | Command | Purpose | diff --git a/bun.lock b/bun.lock index 49971f949b..ec339c0dc8 100644 --- a/bun.lock +++ b/bun.lock @@ -27,7 +27,7 @@ "bedrock-agentcore": "^0.4.3", "cli-truncate": "^6.1.1", "commander": "^15.0.0", - "handlebars": "^4.7.9", + "handlebars": "^4.7.10", "ink": "^7.1.0", "ink-scroll-view": "^0.3.7", "lodash": "^4.18.1", @@ -899,7 +899,7 @@ "graceful-fs": ["graceful-fs@4.2.11", "", {}, "sha512-RbJ5/jmFcNNCcDV5o9eTnBLJ/HszWV0P73bc+Ff4nS/rJj+YaS6IGyiOL0VoBYX+l1Wrl3k63h/KrH+nhJ0XvQ=="], - "handlebars": ["handlebars@4.7.9", "", { "dependencies": { "minimist": "^1.2.5", "neo-async": "^2.6.2", "source-map": "^0.6.1", "wordwrap": "^1.0.0" }, "optionalDependencies": { "uglify-js": "^3.1.4" }, "bin": { "handlebars": "bin/handlebars" } }, "sha512-4E71E0rpOaQuJR2A3xDZ+GM1HyWYv1clR58tC8emQNeQe3RH7MAzSbat+V0wG78LQBo6m6bzSG/L4pBuCsgnUQ=="], + "handlebars": ["handlebars@4.7.10", "", { "dependencies": { "minimist": "^1.2.8", "neo-async": "^2.6.2", "source-map": "^0.6.1", "wordwrap": "^1.0.0" }, "optionalDependencies": { "uglify-js": "^3.1.4" }, "bin": { "handlebars": "bin/handlebars" } }, "sha512-P5VJMVM7qgBn6vjXMw8WG9uVI+ncf2pi72j4de4yz5ZULLj2RGqLYaKOYGsgyrViQ0tePOVlN1tDCCXXtFqXKg=="], "has-bigints": ["has-bigints@1.1.0", "", {}, "sha512-R3pbpkcIqv2Pm3dUwgjclDRVmWpTJW2DcMzcIhEXEx1oh/CEMObMm3KLmRJOdvhM7o4uQBnwr8pzRK2sJWIqfg=="], diff --git a/e2eTest/project/harness.test.ts b/e2eTest/project/harness.test.ts index 71edc840ad..27e3ad2615 100644 --- a/e2eTest/project/harness.test.ts +++ b/e2eTest/project/harness.test.ts @@ -59,7 +59,7 @@ const HarnessInvokeResponseSchema = z.object({ }); const harnessTags = [TAGS.HARNESS, TAGS.CANARY]; -describe("add, deploy, and invoke harnesses", { sequential: true, tags: harnessTags }, () => { +describe("add, dev, deploy, and invoke harnesses", { sequential: true, tags: harnessTags }, () => { const cli = new CliRunner(); const projectName = `${E2E_PREFIX}${Date.now().toString(36)}`; let projectDir: string; @@ -91,34 +91,36 @@ describe("add, deploy, and invoke harnesses", { sequential: true, tags: harnessT }, ); - test("deploys all harnesses", { timeout: TIMEOUT_MS.PROJECT_DEPLOY }, async () => { - const deployment = parseResult( - DeployResponseSchema, - await cli.run(["deploy", "--yes", "--json"], projectDir), - ); - expect(deployment.message).toContain("Deployed project"); - }); - - test.each(HARNESS_TEST_CASES)( - "$name can be invoked after deployed", - { concurrent: true, timeout: TIMEOUT_MS.PROJECT_INVOKE }, - async (harness) => { - const response = parseResult( - HarnessInvokeResponseSchema, - await cli.run( - ["invoke", "--harness", harness.name, "--json", ...harness.invokeFlags], - projectDir, - ), + describe.each(["dev", "deploy"])("%s", { sequential: true }, (command) => { + test("deploys all harnesses", { timeout: TIMEOUT_MS.PROJECT_DEPLOY }, async () => { + const deployment = parseResult( + DeployResponseSchema, + await cli.run([command, "--yes", "--json"], projectDir), ); - const responseText = response.transcript - .filter((item) => item.kind === "text") - .flatMap((item) => item.text ?? []) - .join(""); + expect(deployment.message).toContain("Deployed project"); + }); - expect(responseText.trim()).not.toBe(""); - if (harness.expectedText) expect(responseText).toContain(harness.expectedText); - }, - ); + test.each(HARNESS_TEST_CASES)( + "$name can be invoked after deployed", + { concurrent: true, timeout: TIMEOUT_MS.PROJECT_INVOKE }, + async (harness) => { + const response = parseResult( + HarnessInvokeResponseSchema, + await cli.run( + ["invoke", "--harness", harness.name, "--json", ...harness.invokeFlags], + projectDir, + ), + ); + const responseText = response.transcript + .filter((item) => item.kind === "text") + .flatMap((item) => item.text ?? []) + .join(""); + + expect(responseText.trim()).not.toBe(""); + if (harness.expectedText) expect(responseText).toContain(harness.expectedText); + }, + ); + }); test( "removes all harnesses and deploys the empty project", diff --git a/package.json b/package.json index a61157fb40..91216dc792 100644 --- a/package.json +++ b/package.json @@ -88,7 +88,7 @@ "bedrock-agentcore": "^0.4.3", "cli-truncate": "^6.1.1", "commander": "^15.0.0", - "handlebars": "^4.7.9", + "handlebars": "^4.7.10", "ink": "^7.1.0", "ink-scroll-view": "^0.3.7", "lodash": "^4.18.1", diff --git a/src/core/dev/inspector/invocations.ts b/src/core/dev/inspector/invocations.ts index aa936fdc8d..f2ee81f18c 100644 --- a/src/core/dev/inspector/invocations.ts +++ b/src/core/dev/inspector/invocations.ts @@ -1,5 +1,9 @@ // Every upstream fetch carries the client abort signal, so a browser disconnect tears down the agent request. import { randomUUID } from "node:crypto"; +import type { + InvokeHarnessRequest, + InvokeHarnessStreamOutput, +} from "@aws-sdk/client-bedrock-agentcore"; import type { HttpRequest, HttpResponse } from "../../../io/httpServer"; import { apiError, @@ -18,6 +22,9 @@ export async function handleInvocations( request: HttpRequest, ): Promise { const parsed = parseJsonBody(request.body); + if (parsed && "harnessName" in parsed) { + return invokeHarness({ deps, body: parsed, signal: request.signal }); + } const agentName = asString(parsed?.agentName); // Request header, agent body, and echoed x-session-id must agree, so one session id is computed once. const sessionId = asString(parsed?.sessionId) ?? randomUUID(); @@ -46,6 +53,110 @@ export async function handleInvocations( }); } +type HarnessOverrides = Partial> & { + systemPrompt?: string; +}; + +async function invokeHarness({ + deps, + body, + signal, +}: { + deps: InspectorDeps; + body: Record; + signal: AbortSignal; +}): Promise { + const name = asString(body.harnessName); + if (!name?.trim()) return apiError(400, "harnessName is required"); + const prompt = asString(body.prompt); + if (!prompt?.trim()) return apiError(400, "prompt is required"); + if (!deps.project?.spec.harnesses.some((harness) => harness.name === name)) { + return apiError(404, `Harness "${name}" not found`); + } + if (!deps.invokeHarness) return apiError(409, "Harness invocation is not available"); + + const sessionId = asString(body.sessionId) ?? randomUUID(); + const overrides = body.harnessOverrides as HarnessOverrides | undefined; + const systemPrompt = asString(overrides?.systemPrompt); + try { + const response = await deps.invokeHarness( + name, + { + ...overrides, + qualifier: overrides?.qualifier ?? "DEFAULT", + runtimeSessionId: sessionId, + runtimeUserId: asString(body.userId), + messages: [{ role: "user", content: [{ text: prompt }] }], + systemPrompt: systemPrompt ? [{ text: systemPrompt }] : undefined, + }, + signal, + ); + return sse(transformHarnessSse(response.stream ?? []), sessionId); + } catch (error) { + return apiError(502, `Harness invocation failed: ${errorMessage(error)}`); + } +} + +async function* transformHarnessSse( + stream: AsyncIterable | Iterable, +): AsyncGenerator { + try { + for await (const event of stream) { + const payload = harnessStreamEvent(event); + if (payload) yield sseEvent(payload); + } + } catch (error) { + yield sseEvent({ type: "error", errorType: "invocationError", message: errorMessage(error) }); + } +} + +function harnessStreamEvent(event: InvokeHarnessStreamOutput): unknown { + if (event.messageStart) return { type: "messageStart", ...event.messageStart }; + if (event.contentBlockStart) { + const { start, contentBlockIndex } = event.contentBlockStart; + const type = start?.toolUse ? "toolUse" : start?.toolResult ? "toolResult" : undefined; + if (type) { + return { + type: "contentBlockStart", + contentBlockIndex, + start: { type, ...start }, + }; + } + } + if (event.contentBlockDelta) { + const { delta, contentBlockIndex } = event.contentBlockDelta; + let payload: unknown; + if (delta?.text !== undefined) payload = { type: "text", text: delta.text }; + else if (delta?.toolUse) payload = { type: "toolUse", ...delta.toolUse }; + else if (delta?.toolResult) { + payload = { + type: "toolResult", + results: delta.toolResult.map((chunk) => + chunk.json !== undefined ? { text: JSON.stringify(chunk.json) } : chunk, + ), + }; + } else if (delta?.reasoningContent) { + payload = { type: "reasoningContent", ...delta.reasoningContent }; + } + if (payload) return { type: "contentBlockDelta", contentBlockIndex, delta: payload }; + } + if (event.contentBlockStop) return { type: "contentBlockStop", ...event.contentBlockStop }; + if (event.messageStop) return { type: "messageStop", ...event.messageStop }; + if (event.metadata) return { type: "metadata", ...event.metadata }; + for (const errorType of [ + "validationException", + "internalServerException", + "runtimeClientError", + ] as const) { + const error = event[errorType]; + if (error) return { type: "error", errorType, message: error.message ?? String(error) }; + } + if (event.$unknown) { + const [errorType] = event.$unknown; + return { type: "error", errorType, message: `Unknown harness stream event: ${errorType}` }; + } +} + async function forwardInvocation( port: number, body: Buffer | string, diff --git a/src/core/dev/inspector/types.ts b/src/core/dev/inspector/types.ts index 3d8be57be5..605dfca33a 100644 --- a/src/core/dev/inspector/types.ts +++ b/src/core/dev/inspector/types.ts @@ -5,6 +5,10 @@ * implementations (DevSupervisor, the trace store, bundled assets) at the edge, * and tests inject fakes. */ +import type { + InvokeHarnessRequest, + InvokeHarnessResponse, +} from "@aws-sdk/client-bedrock-agentcore"; import type { Project } from "../../../handlers/project/types"; /** One managed agent's state, as reported by GET /api/status. */ @@ -54,4 +58,9 @@ export interface InspectorDeps { project?: Project; /** Agent name to pre-select in the UI. */ selectedAgent?: string; + invokeHarness?: ( + name: string, + input: Omit, + signal: AbortSignal, + ) => Promise; } diff --git a/src/handlers/project/dev/index.test.ts b/src/handlers/project/dev/index.test.ts index 8af08b8c63..ddaa996eae 100644 --- a/src/handlers/project/dev/index.test.ts +++ b/src/handlers/project/dev/index.test.ts @@ -1,5 +1,10 @@ import { describe, expect, test } from "bun:test"; import { join } from "node:path"; +import { + InternalServerException, + type InvokeHarnessStreamOutput, +} from "@aws-sdk/client-bedrock-agentcore"; +import type { GetHarnessResponse } from "@aws-sdk/client-bedrock-agentcore-control"; import type { ProjectRuntime } from "../../../projectSchemas/runtime"; import { InputValidationError, @@ -9,11 +14,12 @@ import { UserCancellationError, } from "../../../errors"; import type { HttpRequestHandler, PortChecker } from "../../../io"; -import { ProjectKey, ValueContext } from "../../../router"; -import { testIO } from "../../../testing"; +import { GlobalConfigAccessorKey, ProjectKey, Router, ValueContext } from "../../../router"; +import { TestCoreClient, TestGlobalConfigAccessor, testIO } from "../../../testing"; import { JsonRendererKey } from "../../../tui"; import { JsonKey, RegionKey } from "../../keys"; -import type { Project } from "../types"; +import type { DeployProjectInput, Project } from "../types"; +import { ProjectSpecSchema } from "../../../projectSchemas/project"; import { BMA_POLICY_FILE, BMA_TEMPLATE_NAME, @@ -38,7 +44,7 @@ function project(...runtimes: ProjectRuntime[]): Project { return { name: "test-project", rootPath: "/workspace/project", - spec: { runtimes } as Project["spec"], + spec: { ...ProjectSpecSchema.parse({ name: "testProject", version: 2 }), runtimes }, }; } @@ -122,8 +128,36 @@ function harness(options: HarnessOptions = {}) { const container = options.container ?? captureRunner(); const collector = fakeCollector(); const environmentInputs: DevEnvironmentInput[] = []; + const deployments: { project: Project; input: DeployProjectInput }[] = []; + const resolutions: Parameters< + DevProjectHandlerConfig["projectManager"]["resolveDeployedResource"] + >[1][] = []; + const core = new TestCoreClient(); + const harnessClient = core.harness.setGetResponse({ + harness: { arn: "arn:aws:bedrock-agentcore:eu-west-1:111122223333:harness/test" }, + } as GetHarnessResponse); + const projectManager = core.projectManager; + projectManager.deploy = async function* (configuredProject, input) { + deployments.push({ project: configuredProject, input }); + yield { type: "output", line: "Deploying harnesses" }; + return { outputs: {} }; + }; + projectManager.resolveTarget = async () => undefined; + projectManager.resolveDeployedResource = async (_project, input) => { + resolutions.push(input); + return { + resourceType: "harness", + name: input.name, + id: "harness-test", + target: { name: input.target, account: "111122223333", region: "eu-west-1" }, + credentialProvider: async () => ({ accessKeyId: "test", secretAccessKey: "test" }), + }; + }; + projectManager.resolve = async () => + options.reloadedRuntimes ? project(...options.reloadedRuntimes) : undefined; const handler = createDevProjectHandler({ io: io.io, + harness: harnessClient, runners: { CodeZip: codeZip.runner, Container: container.runner }, loadDevEnvironment: options.loadEnvironment ?? @@ -151,15 +185,13 @@ function harness(options: HarnessOptions = {}) { watchFile: (path, onChange) => { watchers.push({ path, onChange }); }, - projectManager: { - resolve: async () => - options.reloadedRuntimes ? project(...options.reloadedRuntimes) : undefined, - }, + projectManager, waitReady: async () => { await Bun.sleep(5); }, }); const ctx = ValueContext.EmptyContext() + .withValue(GlobalConfigAccessorKey, new TestGlobalConfigAccessor()) .withValue(ProjectKey, options.project ?? project(runtime())) .withValue(JsonKey, options.json ?? false) .withValue(RegionKey, "us-west-2") @@ -173,6 +205,9 @@ function harness(options: HarnessOptions = {}) { container, collector, environmentInputs, + deployments, + resolutions, + harnessClient, io, ui, watchers, @@ -182,22 +217,28 @@ function harness(options: HarnessOptions = {}) { agent?: string; port?: number; traces?: boolean; - mode?: "browser" | "headless" | "tui"; + mode?: "browser" | "headless"; + target?: string; + yes?: boolean; + "skip-deploy"?: boolean; "ui-port"?: number; } = {}, - ) => handler.handle(ctx, { traces: true, mode: "headless", ...flags }, {}), + ) => + handler.handle( + ctx, + { traces: true, yes: false, "skip-deploy": false, mode: "headless", ...flags }, + {}, + ), + route: (args: readonly string[]) => + new Router("agentcore", "test") + .handler(handler) + .route(["node", "agentcore", "dev", ...args], ctx), }; } /** Ask the captured Inspector handler for the current agent status. */ async function inspectorStatus(subject: ReturnType): Promise<{ name: string }[]> { - const response = await subject.inspectorHandler()!({ - method: "GET", - url: "/api/status", - headers: { host: "127.0.0.1:8081" }, - body: Buffer.alloc(0), - signal: new AbortController().signal, - }); + const response = await inspectorRequest({ subject, url: "/api/status" }); const status = JSON.parse(String(response.body)) as { agents: { name: string }[] }; return status.agents; } @@ -230,7 +271,13 @@ describe("project dev selection and dispatch", () => { ); test.each([ - [project(), {}, "This project has no runtimes", InputValidationError], + [project(), {}, "This project has no runtimes or harnesses", InputValidationError], + [ + harnessProject(["orders"], [runtime("orders")]), + { agent: "orders" }, + "names both a runtime and a harness", + InputValidationError, + ], [ project(runtime("orders"), runtime("support", "Container")), { port: 4567 }, @@ -240,7 +287,7 @@ describe("project dev selection and dispatch", () => { [ project(runtime("orders"), runtime("support", "Container")), { agent: "missing" }, - "Runtime 'missing' was not found. Available runtimes: orders, support", + "Agent 'missing' was not found. Available agents: orders, support", ResourceNotFoundError, ], ] as const)( @@ -445,18 +492,27 @@ describe("project dev trace collection", () => { await expect(subject.run({ agent: "orders" })).rejects.toThrow("runner failed"); expect(subject.collector.state.closed).toBe(1); }); + + test("the collector is closed when startup output fails", async () => { + const subject = harness(); + subject.io.io.stderr.write = () => { + throw new Error("status output failed"); + }; + await expect(subject.run({ agent: "orders" })).rejects.toThrow("status output failed"); + expect(subject.collector.state.closed).toBe(1); + }); }); -describe("project dev Inspector UI mode", () => { - // Wraps the run promise so awaiting this helper does not flatten it into - // "wait for the whole dev command to exit". - async function runUi(subject: ReturnType, flags: Record = {}) { - const pending = subject.run({ mode: "browser", ...flags }); - pending.catch(() => undefined); - await Bun.sleep(5); // let the handler start the UI server and block on events - return { pending }; - } +// Return the pending command inside an object so awaiting startup does not wait for dev to exit. +async function runUi(subject: ReturnType, args: readonly string[] = []) { + const pending = subject.route(["--mode", "browser", ...args]); + pending.catch(() => undefined); + await Bun.sleep(10); + expect(subject.ui.starts).toHaveLength(1); + return { pending }; +} +describe("project dev Inspector UI mode", () => { test("starts the Inspector, prints the URL, and opens the browser on a TTY", async () => { const subject = harness({ tty: true }); const { pending } = await runUi(subject); @@ -520,13 +576,28 @@ describe("project dev Inspector UI mode", () => { }); test("--agent narrows the supervised set", async () => { + const container = stayingRunner([{ type: "stdout", line: "support ready" }]); const subject = harness({ - project: project(bmaRuntime(), runtime("orders"), runtime("support", "Container")), + project: harnessProject( + ["harness"], + [bmaRuntime(), runtime("orders"), runtime("support", "Container")], + ), + container, }); - const { pending } = await runUi(subject, { agent: "support" }); + const { pending } = await runUi(subject, ["--agent", "support"]); expect((await inspectorStatus(subject)).map((agent) => agent.name)).toEqual(["support"]); + const status = await inspectorRequest({ subject, url: "/api/status" }); + expect(JSON.parse(String(status.body)).harnesses).toEqual([]); expect(subject.io.stderr()).not.toContain("Skipping runtime"); + const started = await inspectorRequest({ + subject, + url: "/api/start", + body: { agentName: "support" }, + }); + expect(started.status).toBe(200); + expect(container.inputs[0]?.port).toBe(8080); + expect(subject.io.stdout()).toContain("[support] support ready"); process.emit("SIGINT", "SIGINT"); await pending.catch(() => undefined); @@ -563,6 +634,276 @@ test("project dev renders attributed human and NDJSON output", async () => { } }); +function harnessProject( + names: readonly string[] = ["support"], + runtimes: readonly ProjectRuntime[] = [], +): Project { + const configured = project(...runtimes); + configured.spec.harnesses = names.map((name) => ({ name, path: `harnesses/${name}` })); + return configured; +} + +async function inspectorRequest({ + subject, + url, + body, +}: { + subject: ReturnType; + url: string; + body?: unknown; +}) { + return subject.inspectorHandler()!({ + method: body === undefined ? "GET" : "POST", + url, + headers: { host: "127.0.0.1:8081", "x-agentcore-local": "1" }, + body: Buffer.from(body === undefined ? "" : JSON.stringify(body)), + signal: new AbortController().signal, + }); +} + +describe("project dev harnesses", () => { + test.each([ + { names: ["support"], args: [], json: false }, + { names: ["orders", "support"], args: ["--mode", "headless"], json: true }, + { names: ["support"], args: ["--skip-deploy"], json: true }, + ])("headless deploys and returns invoke guidance (%o)", async ({ names, args, json }) => { + const subject = harness({ project: harnessProject(names), tty: true, json }); + await subject.route([...args]); + const skipDeploy = args.some((arg) => arg === "--skip-deploy"); + expect(subject.deployments).toHaveLength(skipDeploy ? 0 : 1); + expect(subject.ui.starts).toEqual([]); + expect(subject.ui.opened).toEqual([]); + expect(subject.codeZip.inputs).toEqual([]); + expect(subject.io.stderr()).toContain("Next step:\n agentcore invoke"); + }); + + test.each([ + { names: ["support"], args: [], target: "default", port: 8081 }, + { + names: ["orders", "support"], + args: ["--target", "staging", "--ui-port", "9001", "--yes"], + target: "staging", + port: 9001, + }, + ])("deploys then serves Inspector (%o)", async ({ names, args, target, port }) => { + const configuredProject = harnessProject(names); + const subject = harness({ project: configuredProject, tty: true }); + const { pending } = await runUi(subject, args); + try { + const deployment = subject.deployments[0]!; + expect(subject.deployments).toHaveLength(1); + expect(deployment.project).toBe(configuredProject); + expect(deployment.input.target).toBe(target); + const response = await inspectorRequest({ subject, url: "/api/status" }); + const status = JSON.parse(String(response.body)); + expect(status.agents).toEqual([]); + expect(status.harnesses.map(({ name }: { name: string }) => name)).toEqual(names); + expect(subject.ui.starts[0]?.port).toBe(port); + expect(subject.ui.opened).toEqual([`http://127.0.0.1:${port}`]); + expect(subject.codeZip.inputs).toHaveLength(0); + expect(subject.collector.starts).toHaveLength(0); + } finally { + process.emit("SIGTERM", "SIGTERM"); + await expect(pending).rejects.toBeInstanceOf(UserCancellationError); + } + expect(subject.ui.closed).toBe(1); + }); + + test("rejects runtime flags on harness dev", async () => { + const subject = harness({ project: harnessProject() }); + await expect(subject.route(["--port=8080", "--no-traces"])).rejects.toThrow( + "--port, --no-traces does not apply to harness dev", + ); + expect(subject.deployments).toHaveLength(0); + }); + + test("rejects harness flags on runtime dev", async () => { + const subject = harness(); + await expect(subject.route(["--target=default", "--yes", "--skip-deploy"])).rejects.toThrow( + "--target, --yes, --skip-deploy does not apply to runtime dev", + ); + expect(subject.codeZip.inputs).toHaveLength(0); + }); + + test("mixed projects require an unambiguous agent and can choose the runtime path", async () => { + const subject = harness({ project: harnessProject(["support"], [runtime("orders")]) }); + await expect(subject.route([])).rejects.toThrow("Pass --agent "); + await subject.route(["--agent", "orders", "--no-traces"]); + expect(subject.codeZip.inputs).toHaveLength(1); + expect(subject.deployments).toHaveLength(0); + }); + + test("Inspector invokes the selected harness without deploying", async () => { + const subject = harness({ + project: harnessProject(["orders", "support"], [runtime("runtime")]), + tty: true, + }); + const toolUse = { toolUseId: "search-1", name: "search" }; + const toolResult = { toolUseId: "search-1", status: "success" as const }; + subject.harnessClient.queueInvokeStream( + (async function* (): AsyncGenerator { + yield { contentBlockStart: { contentBlockIndex: 0, start: undefined } }; + yield { contentBlockDelta: { contentBlockIndex: 0, delta: { text: "Hello" } } }; + yield { contentBlockStart: { contentBlockIndex: 1, start: { toolUse } } }; + yield { contentBlockStart: { contentBlockIndex: 2, start: { toolResult } } }; + yield { + contentBlockDelta: { + contentBlockIndex: 2, + delta: { + toolResult: [ + { text: "before:" }, + { json: null }, + { json: false }, + { json: 0 }, + { json: { answer: 42 } }, + { text: ":after" }, + ], + }, + }, + }; + yield { + contentBlockDelta: { + contentBlockIndex: 3, + delta: { reasoningContent: { text: "Thinking" } }, + }, + }; + yield { + internalServerException: new InternalServerException({ + message: "internal error", + $metadata: {}, + }), + }; + yield { $unknown: ["newEventType", {}] }; + throw new Error("stream failed"); + })(), + ); + const { pending } = await runUi(subject, [ + "--agent", + "support", + "--target", + "staging", + "--skip-deploy", + ]); + try { + const status = await inspectorRequest({ subject, url: "/api/status" }); + expect(JSON.parse(String(status.body)).harnesses).toEqual([{ name: "support" }]); + const response = await inspectorRequest({ + subject, + url: "/invocations", + body: { + harnessName: "support", + prompt: "Hello", + userId: "user", + }, + }); + expect(response.status).toBe(200); + const events = Buffer.concat( + await Array.fromAsync(response.body as AsyncIterable), + ) + .toString() + .split("\n") + .filter((line) => line.startsWith("data: ")) + .map((line) => JSON.parse(line.slice(6))); + expect(events).toEqual([ + { type: "contentBlockDelta", contentBlockIndex: 0, delta: { type: "text", text: "Hello" } }, + { + type: "contentBlockStart", + contentBlockIndex: 1, + start: { type: "toolUse", toolUse }, + }, + { + type: "contentBlockStart", + contentBlockIndex: 2, + start: { type: "toolResult", toolResult }, + }, + { + type: "contentBlockDelta", + contentBlockIndex: 2, + delta: { + type: "toolResult", + results: [ + { text: "before:" }, + { text: "null" }, + { text: "false" }, + { text: "0" }, + { text: '{"answer":42}' }, + { text: ":after" }, + ], + }, + }, + { + type: "contentBlockDelta", + contentBlockIndex: 3, + delta: { type: "reasoningContent", text: "Thinking" }, + }, + { type: "error", errorType: "internalServerException", message: "internal error" }, + { + type: "error", + errorType: "newEventType", + message: "Unknown harness stream event: newEventType", + }, + { type: "error", errorType: "invocationError", message: "stream failed" }, + ]); + expect(subject.deployments).toHaveLength(0); + expect(subject.ui.opened).toEqual(["http://127.0.0.1:8081"]); + expect(subject.resolutions).toEqual([ + { target: "staging", resourceType: "harness", name: "support" }, + ]); + expect( + subject.harnessClient.calls + .find(({ method }) => method === "invokeHarness") + ?.args.slice(0, 2), + ).toMatchObject([ + { + harnessArn: "arn:aws:bedrock-agentcore:eu-west-1:111122223333:harness/test", + qualifier: "DEFAULT", + runtimeSessionId: response.headers?.["x-session-id"], + runtimeUserId: "user", + messages: [{ role: "user", content: [{ text: "Hello" }] }], + }, + { region: "eu-west-1", credentials: expect.any(Function) }, + ]); + const overridden = await inspectorRequest({ + subject, + url: "/invocations", + body: { + harnessName: "support", + prompt: "Hello", + harnessOverrides: { qualifier: "STAGING" }, + }, + }); + expect(overridden.status).toBe(200); + expect(subject.harnessClient.calls.at(-1)?.args[0]).toMatchObject({ + qualifier: "STAGING", + }); + } finally { + process.emit("SIGINT", "SIGINT"); + await pending.catch(() => undefined); + } + const signal = subject.harnessClient.calls.find(({ method }) => method === "invokeHarness") + ?.args[2] as AbortSignal; + expect(signal.aborted).toBe(true); + }); + + test.each([ + ["/invocations", { harnessName: "missing", prompt: "hi" }, 404, 'Harness "missing" not found'], + ["/invocations", { harnessName: "support", prompt: "hi" }, 502, "lookup failed"], + ["/api/start", { agentName: "runtime" }, 404, "Runtime 'runtime' was not found"], + ] as const)("Inspector rejects %s %o", async (url, body, status, error) => { + const subject = harness({ project: harnessProject() }); + subject.harnessClient.setError(new Error("lookup failed")); + const { pending } = await runUi(subject, ["--skip-deploy"]); + try { + const response = await inspectorRequest({ subject, url, body }); + expect(response.status).toBe(status); + expect(JSON.parse(String(response.body)).error).toContain(error); + } finally { + process.emit("SIGINT", "SIGINT"); + await pending.catch(() => undefined); + } + }); +}); + function heldRunner() { let start!: (input: DevServerInput) => void; let release: (() => void) | undefined; diff --git a/src/handlers/project/dev/index.ts b/src/handlers/project/dev/index.ts index de3603cdd1..836123bd17 100644 --- a/src/handlers/project/dev/index.ts +++ b/src/handlers/project/dev/index.ts @@ -14,12 +14,16 @@ import { NotImplementedError, ResourceNotFoundError, SilentCLIError, - UserCancellationError, } from "../../../errors"; import type { AppIO, BrowserOpener, FileWatcher, PortChecker, startHttpServer } from "../../../io"; -import { createHandler, flag, ProjectKey, type Middleware } from "../../../router"; +import { createHandler, flag, ProjectKey, type Context, type Middleware } from "../../../router"; +import { withUserCancellation } from "../../../runnable"; import { JsonRendererKey, type JsonRenderer } from "../../../tui"; import { JsonKey, RegionKey } from "../../keys"; +import { coreOptsFromCtx } from "../../utils"; +import type { CoreHarnessClient } from "../../harness/types"; +import { DEFAULT_TARGET_NAME } from "../../../projectSchemas/aws-targets"; +import { createDeployProjectHandler, DEPLOY_NEXT_STEP } from "../deploy"; import type { Project, ProjectManager } from "../types"; import { BMA_TEMPLATE_NAME, isBmaRuntime } from "../bma"; import type { DevEnvironmentLoader } from "./environment"; @@ -27,6 +31,8 @@ import type { DevEvent, DevRunner, DevTraceCollector, DevTraceCollectorStarter } /** The Inspector UI binds 8081 or, when that is taken, the next free port. */ const UI_DEFAULT_PORT = 8081; +const RUNTIME_OPTIONS_GROUP = "Runtime Options:"; +const HARNESS_OPTIONS_GROUP = "Harness Options:"; export type DevProjectHandlerConfig = { io: AppIO; @@ -43,7 +49,8 @@ export type DevProjectHandlerConfig = { /** Watches agentcore.json so the Inspector reflects config edits live. */ watchFile: FileWatcher; /** Re-resolves the project after a config change to pick up runtime edits. */ - projectManager: Pick; + projectManager: ProjectManager; + harness: Pick; /** Overrides how the supervisor decides an agent is ready (defaults to a real TCP poll). */ waitReady?: SupervisorConfig["waitReady"]; }; @@ -61,34 +68,18 @@ function supportsLocalDev(runtime: ProjectRuntime): boolean { return !isBmaRuntime(runtime); } -function selectRuntimes(project: Project, name?: string): ProjectRuntime[] { - if (project.spec.runtimes.length === 0) { - throw new InputValidationError( - "This project has no runtimes. Add a runtime to agentcore/agentcore.json and retry.", - ); - } - const selectedRuntimes = name - ? project.spec.runtimes.filter((runtime) => runtime.name === name) - : project.spec.runtimes; - if (selectedRuntimes.length === 0) { - const available = project.spec.runtimes.map((candidate) => candidate.name).join(", "); - throw new ResourceNotFoundError( - `Runtime '${name}' was not found. Available runtimes: ${available}.`, - ); - } - const supportedRuntimes = selectedRuntimes.filter(supportsLocalDev); - if (supportedRuntimes.length === 0) { - throw new NotImplementedError( - `Local dev is not supported for runtime '${selectedRuntimes[0]!.name}' (${BMA_TEMPLATE_NAME}). ` + - "Run agentcore deploy, then use client.py to connect through Bedrock Managed Agents.", - { source: ERROR_SOURCE.USER }, - ); - } - return supportedRuntimes; -} - /** An agent's own output, always tagged with the agent that produced it. */ -function renderAgentEvent(io: AppIO, event: DevEvent, agent: string, json?: JsonRenderer): void { +function renderAgentEvent({ + io, + event, + agent, + json, +}: { + io: AppIO; + event: DevEvent; + agent: string; + json?: JsonRenderer; +}): void { if (json) { json.renderJsonLine({ agent, ...event }); return; @@ -100,7 +91,15 @@ function renderAgentEvent(io: AppIO, event: DevEvent, agent: string, json?: Json } /** A command-level status line, not attributed to any agent. */ -function renderStatus(io: AppIO, message: string, json?: JsonRenderer): void { +function renderStatus({ + io, + message, + json, +}: { + io: AppIO; + message: string; + json?: JsonRenderer; +}): void { if (json) { json.renderJsonLine({ type: "status", message }); return; @@ -111,94 +110,273 @@ function renderStatus(io: AppIO, message: string, json?: JsonRenderer): void { export const createDevProjectHandler = (config: DevProjectHandlerConfig) => createHandler({ name: "dev", - description: "run the project locally for development", + description: + "test project changes: run Runtimes locally or deploy for Harnesses; --mode browser opens Agent Inspector.", middlewares: config.middlewares, flags: [ - flag("agent", "Runtime to run", z.string().optional()), + flag("agent", "runtime or harness to run", z.string().min(1).optional()), flag( - "port", - "port for the development server", + "ui-port", + "port for the Agent Inspector web UI (browser mode)", z.coerce.number().int().min(1).max(65535).optional(), ), - flag("traces", "disable local OTEL trace collection", z.boolean().default(true)), flag( "mode", - "how to run: browser (Agent Inspector web UI) or headless (agents stream to the terminal)", + "how to run: browser (Agent Inspector web UI) or headless (runtime output or harness invoke guidance)", z.enum(["browser", "headless"]).default("headless"), ), flag( - "ui-port", - "port for the Agent Inspector web UI (browser mode)", + "port", + "port for the development server", z.coerce.number().int().min(1).max(65535).optional(), + { group: RUNTIME_OPTIONS_GROUP }, + ), + flag("traces", "disable local OTEL trace collection", z.boolean().default(true), { + group: RUNTIME_OPTIONS_GROUP, + }), + flag( + "target", + 'name of the aws-targets.json entry to deploy; the default target is created automatically from your AWS account and region on first deploy (default: "default")', + z.string().min(1).optional(), + { group: HARNESS_OPTIONS_GROUP }, + ), + flag( + "yes", + "confirm removing the target's stack when the project declares nothing to deploy", + z.boolean().default(false), + { group: HARNESS_OPTIONS_GROUP }, + ), + flag( + "skip-deploy", + "skip deployment and use the deployed harnesses", + z.boolean().default(false), + { group: HARNESS_OPTIONS_GROUP }, ), ], handle: async (ctx, flags) => { - const controller = new AbortController(); - const json = ctx.require(JsonKey) ? ctx.require(JsonRendererKey) : undefined; - const interrupt = () => { - if (controller.signal.aborted) return; - config.io.stderr.write("Shutting down…\n"); - controller.abort(new UserCancellationError()); - }; + const selection = resolveAgentSelection(ctx.require(ProjectKey), flags.agent); + if (selection.type === "harness") { + const unsupportedFlags = [ + flags.port !== undefined && "--port", + !flags.traces && "--no-traces", + ].filter(Boolean); + if (unsupportedFlags.length) { + throw new InputValidationError( + `${unsupportedFlags.join(", ")} does not apply to harness dev.`, + ); + } + await runHarnessDev({ + config, + ctx, + input: { + harnesses: selection.harnesses, + mode: flags.mode, + uiPort: flags["ui-port"], + deploymentTarget: flags.target ?? DEFAULT_TARGET_NAME, + yes: flags.yes, + skipDeploy: flags["skip-deploy"], + }, + }); + } else { + const unsupportedFlags = (["target", "yes", "skip-deploy"] as const).filter( + (name) => flags[name], + ); + if (unsupportedFlags.length) { + throw new InputValidationError( + `${unsupportedFlags.map((name) => `--${name}`).join(", ")} does not apply to runtime dev.`, + ); + } + await runRuntimeDev({ + config, + ctx, + input: { + runtimes: selection.runtimes, + agent: flags.agent, + port: flags.port, + traces: flags.traces, + mode: flags.mode, + uiPort: flags["ui-port"], + }, + }); + } + }, + }); + +type AgentSelection = + | { type: "harness"; harnesses: Project["spec"]["harnesses"] } + | { type: "runtime"; runtimes: ProjectRuntime[] }; - const signals = ["SIGINT", "SIGTERM"] as const; - for (const signal of signals) process.on(signal, interrupt); +function resolveAgentSelection(project: Project, agent?: string): AgentSelection { + const { runtimes, harnesses } = project.spec; + if (runtimes.length === 0 && harnesses.length === 0) { + throw new InputValidationError("This project has no runtimes or harnesses."); + } + const selectedRuntimes = agent ? runtimes.filter(({ name }) => name === agent) : runtimes; + const selectedHarnesses = agent ? harnesses.filter(({ name }) => name === agent) : harnesses; + if (selectedRuntimes.length > 0 && selectedHarnesses.length > 0) { + const message = agent + ? `Agent '${agent}' names both a runtime and a harness. Give them distinct names in agentcore.json.` + : "This project has both runtimes and harnesses. Pass --agent to choose which to run."; + throw new InputValidationError(message); + } + if (selectedHarnesses.length) return { type: "harness", harnesses: selectedHarnesses }; + if (!selectedRuntimes.length) { + throw new ResourceNotFoundError( + `Agent '${agent}' was not found. Available agents: ${[...runtimes, ...harnesses].map(({ name }) => name).join(", ")}.`, + ); + } + const supportedRuntimes = selectedRuntimes.filter(supportsLocalDev); + if (!supportedRuntimes.length) { + throw new NotImplementedError( + `Local dev is not supported for runtime '${selectedRuntimes[0]!.name}' (${BMA_TEMPLATE_NAME}). ` + + "Run agentcore deploy, then use client.py to connect through Bedrock Managed Agents.", + { source: ERROR_SOURCE.USER }, + ); + } + return { type: "runtime", runtimes: supportedRuntimes }; +} + +async function runHarnessDev({ + config, + ctx, + input, +}: { + config: DevProjectHandlerConfig; + ctx: Context; + input: { + harnesses: Project["spec"]["harnesses"]; + mode: "browser" | "headless"; + uiPort?: number; + deploymentTarget: string; + yes: boolean; + skipDeploy: boolean; + }; +}): Promise { + const { harnesses, deploymentTarget } = input; + if (!input.skipDeploy) { + await createDeployProjectHandler(config).handle( + ctx, + { target: deploymentTarget, yes: input.yes }, + {}, + ); + } + if (input.mode === "headless" && (input.skipDeploy || ctx.require(JsonKey))) { + config.io.stderr.write(`Next step:\n ${DEPLOY_NEXT_STEP}\n`); + } + if (input.mode === "headless") return; + const project = ctx.require(ProjectKey); + + const inspectorProject = { + ...project, + spec: { + ...project.spec, + runtimes: [], + harnesses, + }, + }; + await withUserCancellation( + async (signal) => { + let server: Awaited> | undefined; + try { + const uiPort = (await findFreePort(UI_DEFAULT_PORT, input.uiPort, config.checkPort, signal)) + .port; + server = await config.startServer( + createInspectorHandler({ + supervisor: { + snapshot: () => [], + running: () => undefined, + start: async (name) => { + throw new ResourceNotFoundError(`Runtime '${name}' was not found.`); + }, + }, + project: inspectorProject, + assets: config.inspectorAssets, + invokeHarness: async (name, input, requestSignal) => { + const deployed = await config.projectManager.resolveDeployedResource(project, { + target: deploymentTarget, + resourceType: "harness", + name, + }); + const options = { + ...coreOptsFromCtx(ctx), + region: deployed.target.region, + credentials: deployed.credentialProvider, + }; + const detail = await config.harness.getHarness(deployed.id, options); + if (!detail.harness?.arn) + throw new ResourceNotFoundError(`Harness '${name}' has no deployed ARN.`); + return config.harness.invokeHarness( + { ...input, harnessArn: detail.harness.arn }, + options, + AbortSignal.any([requestSignal, signal]), + ); + }, + }), + { port: uiPort, signal }, + ); + const json = ctx.require(JsonKey) ? ctx.require(JsonRendererKey) : undefined; + const url = `http://127.0.0.1:${server.port}`; + renderStatus({ io: config.io, message: `Agent Inspector running at ${url}`, json }); + if (config.isInteractive() && !json) await config.openBrowser(url); + signal.throwIfAborted(); + await new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => reject(signal.reason), { + once: true, + }); + }); + } finally { + await server?.close(); + } + }, + () => config.io.stderr.write("Shutting down…\n"), + ); +} + +async function runRuntimeDev({ + config, + ctx, + input, +}: { + config: DevProjectHandlerConfig; + ctx: Context; + input: { + runtimes: ProjectRuntime[]; + agent?: string; + port?: number; + traces: boolean; + mode: "browser" | "headless"; + uiPort?: number; + }; +}): Promise { + const { runtimes } = input; + await withUserCancellation( + async (signal) => { + const json = ctx.require(JsonKey) ? ctx.require(JsonRendererKey) : undefined; let collector: DevTraceCollector | undefined; try { const project = ctx.require(ProjectKey); const region = ctx.require(RegionKey); - const runtimes = selectRuntimes(project, flags.agent); - if (runtimes.length > 1 && flags.port !== undefined) { + if (runtimes.length > 1 && input.port !== undefined) { throw new InputValidationError( "--port applies to a single runtime. Use --agent to select one.", ); } - if (!flags.agent) { + if (!input.agent) { for (const runtime of project.spec.runtimes.filter( (runtime) => !supportsLocalDev(runtime), )) { - renderStatus( - config.io, - `Skipping runtime '${runtime.name}': local dev is not supported for ${BMA_TEMPLATE_NAME}.`, + renderStatus({ + io: config.io, + message: `Skipping runtime '${runtime.name}': local dev is not supported for ${BMA_TEMPLATE_NAME}.`, json, - ); + }); } } - if ( - flags.traces && - runtimes.some((runtime) => runtime.instrumentation?.enableOtel ?? true) - ) { - const tracesDirectory = join(project.rootPath, "agentcore", ".cli", "traces", "otlp"); - let tracePersistErrorReported = false; - collector = await config.startTraceCollector({ - tracesDirectory, - // A container reaches the collector over the host bridge, which a - // 127.0.0.1 bind refuses, so bind all interfaces when any runtime - // is a container. - host: runtimes.some((runtime) => runtime.build === "Container") - ? "0.0.0.0" - : "127.0.0.1", - // Persistence can fail after startup (disk, permissions). Warn once — - // exports are still acked, so without this the loss would be silent. - onError: (error) => { - if (tracePersistErrorReported) return; - tracePersistErrorReported = true; - const detail = error instanceof Error ? error.message : String(error); - renderStatus( - config.io, - `Warning: failed to persist traces to ${tracesDirectory} (${detail}); collected traces may be incomplete.`, - json, - ); - }, - }); - renderStatus( - config.io, - `OTEL collector listening on port ${collector.port}; traces persist to ${tracesDirectory}.`, - json, - ); - } - controller.signal.throwIfAborted(); + collector = input.traces + ? await startRuntimeTraceCollector({ config, project, runtimes, json }) + : undefined; + signal.throwIfAborted(); const getDevEnvVarsForRuntime = async ( runtime: ProjectRuntime, @@ -215,22 +393,21 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => return { ...env, ...otel }; }; - if (flags.mode === "headless" && flags.agent) { - await runWithoutUi( + if (input.mode === "headless" && input.agent) { + await runWithoutUi({ config, - runtimes[0]!, + runtime: runtimes[0]!, project, - flags.port, - getDevEnvVarsForRuntime, - controller, + options: { port: input.port, environment: getDevEnvVarsForRuntime }, + signal, json, - ); + }); return; } const assignedPorts = - flags.mode === "headless" - ? await resolveDevPorts(runtimes, flags.port, config.checkPort, controller.signal) + input.mode === "headless" + ? await resolveDevPorts(runtimes, input.port, config.checkPort, signal) : undefined; const supervisor = new DevSupervisor({ runtimes, @@ -238,7 +415,7 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => runners: config.runners, getDevEnvVarsForRuntime, // The --port guard above rejects an explicit port with more than one - // runtime, so passing flags.port here only ever applies to a lone one. + // runtime, so passing input.port here only ever applies to a lone one. resolvePort: async (runtime) => { if (assignedPorts) { const assignedPort = assignedPorts.get(runtime.name); @@ -247,45 +424,38 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => } return assignedPort; } - return ( - await resolveDevPort( - runtime.protocol, - flags.port, - config.checkPort, - controller.signal, - ) - ).port; + return (await resolveDevPort(runtime.protocol, input.port, config.checkPort, signal)) + .port; }, waitReady: config.waitReady, - signal: controller.signal, + signal, }); - if (flags.mode === "headless") { + if (input.mode === "headless") { void Promise.allSettled(runtimes.map((runtime) => supervisor.start(runtime.name))); for await (const { agentName, event } of supervisor.events()) { - renderAgentEvent(config.io, event, agentName, json); + renderAgentEvent({ io: config.io, event, agent: agentName, json }); const phases = supervisor.snapshot(); if (phases.every(({ phase }) => phase !== "starting" && phase !== "running")) { if (phases.some(({ phase }) => phase === "failed")) throw new SilentCLIError(); break; } } - controller.signal.throwIfAborted(); + signal.throwIfAborted(); return; } - const uiPort = ( - await findFreePort(UI_DEFAULT_PORT, flags["ui-port"], config.checkPort, controller.signal) - ).port; + const uiPort = (await findFreePort(UI_DEFAULT_PORT, input.uiPort, config.checkPort, signal)) + .port; const server = await config.startServer( createInspectorHandler({ supervisor, traces: collector?.traces, assets: config.inspectorAssets, - project, - selectedAgent: flags.agent, + project: { ...project, spec: { ...project.spec, harnesses: [] } }, + selectedAgent: input.agent, }), - { port: uiPort, signal: controller.signal }, + { port: uiPort, signal }, ); const onConfigChange = async () => { @@ -294,68 +464,108 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => if (!reloaded) return; const runtimes = reloaded.spec.runtimes.filter(supportsLocalDev); supervisor.setRuntimes( - flags.agent ? runtimes.filter((runtime) => runtime.name === flags.agent) : runtimes, + input.agent ? runtimes.filter((runtime) => runtime.name === input.agent) : runtimes, ); - renderStatus(config.io, "Reloaded agents from agentcore.json.", json); + renderStatus({ io: config.io, message: "Reloaded agents from agentcore.json.", json }); } catch { // A half-saved config parses on the next change event. } }; - config.watchFile( - projectSpecPath(project.rootPath), - () => void onConfigChange(), - controller.signal, - ); + config.watchFile(projectSpecPath(project.rootPath), () => void onConfigChange(), signal); const url = `http://127.0.0.1:${server.port}`; - renderStatus(config.io, `Agent Inspector running at ${url}`, json); + renderStatus({ io: config.io, message: `Agent Inspector running at ${url}`, json }); if (config.isInteractive() && !json) await config.openBrowser(url); for await (const { agentName, event } of supervisor.events()) { - renderAgentEvent(config.io, event, agentName, json); + renderAgentEvent({ io: config.io, event, agent: agentName, json }); } - controller.signal.throwIfAborted(); - } catch (error) { - controller.signal.throwIfAborted(); - throw error; + signal.throwIfAborted(); } finally { - for (const signal of signals) process.removeListener(signal, interrupt); // Close only after the runner returns, which is after the child's own // shutdown grace, so the agent's final spans still reach the collector. await collector?.close(); } }, + () => config.io.stderr.write("Shutting down…\n"), + ); +} + +async function startRuntimeTraceCollector({ + config, + project, + runtimes, + json, +}: { + config: DevProjectHandlerConfig; + project: Project; + runtimes: ProjectRuntime[]; + json?: JsonRenderer; +}): Promise { + if (!runtimes.some((runtime) => runtime.instrumentation?.enableOtel ?? true)) return; + const tracesDirectory = join(project.rootPath, "agentcore", ".cli", "traces", "otlp"); + let tracePersistErrorReported = false; + const collector = await config.startTraceCollector({ + tracesDirectory, + // A container reaches the collector over the host bridge, which a loopback bind refuses. + host: runtimes.some((runtime) => runtime.build === "Container") ? "0.0.0.0" : "127.0.0.1", + onError: (error) => { + if (tracePersistErrorReported) return; + tracePersistErrorReported = true; + const detail = error instanceof Error ? error.message : String(error); + renderStatus({ + io: config.io, + message: `Warning: failed to persist traces to ${tracesDirectory} (${detail}); collected traces may be incomplete.`, + json, + }); + }, }); + try { + renderStatus({ + io: config.io, + message: `OTEL collector listening on port ${collector.port}; traces persist to ${tracesDirectory}.`, + json, + }); + return collector; + } catch (error) { + await collector.close(); + throw error; + } +} /** * Run one runtime directly. Unlike the supervised Inspector path, a crash here * fails the command (scripts and CI rely on the non-zero exit). */ -async function runWithoutUi( - config: DevProjectHandlerConfig, - runtime: ProjectRuntime, - project: Project, - explicitPort: number | undefined, - environment: (runtime: ProjectRuntime) => Promise>, - controller: AbortController, - json?: JsonRenderer, -): Promise { - const devPort = await resolveDevPort( - runtime.protocol, - explicitPort, - config.checkPort, - controller.signal, - ); +async function runWithoutUi({ + config, + runtime, + project, + options, + signal, + json, +}: { + config: DevProjectHandlerConfig; + runtime: ProjectRuntime; + project: Project; + options: { + port?: number; + environment: (runtime: ProjectRuntime) => Promise>; + }; + signal: AbortSignal; + json?: JsonRenderer; +}): Promise { + const devPort = await resolveDevPort(runtime.protocol, options.port, config.checkPort, signal); if (devPort.port !== devPort.requestedPort) { - renderStatus( - config.io, - `Port ${devPort.requestedPort} is in use; using ${devPort.port}.`, + renderStatus({ + io: config.io, + message: `Port ${devPort.requestedPort} is in use; using ${devPort.port}.`, json, - ); + }); } - const env = await environment(runtime); - controller.signal.throwIfAborted(); + const env = await options.environment(runtime); + signal.throwIfAborted(); const runner = config.runners[runtime.build]; for await (const event of runner.run({ @@ -363,8 +573,8 @@ async function runWithoutUi( projectRoot: project.rootPath, port: devPort.port, env, - signal: controller.signal, + signal, })) { - renderAgentEvent(config.io, event, runtime.name, json); + renderAgentEvent({ io: config.io, event, agent: runtime.name, json }); } } diff --git a/src/handlers/project/index.ts b/src/handlers/project/index.ts index a84c990e62..b87a46cc68 100644 --- a/src/handlers/project/index.ts +++ b/src/handlers/project/index.ts @@ -37,6 +37,7 @@ export function createProjectHandlers(core: Core, io: AppIO): Handler[] { const projectBoundHandlers = [ createDevProjectHandler({ projectManager, + harness: core.harness, io, middlewares: [withProjectMiddleware], runners: { diff --git a/src/runnable/index.test.ts b/src/runnable/index.test.ts index dbb238b7ae..bb3573dcad 100644 --- a/src/runnable/index.test.ts +++ b/src/runnable/index.test.ts @@ -76,38 +76,56 @@ test("respects custom errors codes from known errors", async () => { expect(errors).toEqual(["Error: custom failure"]); }); -test("withUserCancellation returns the result and removes its SIGINT listener", async () => { - const initialListeners = process.listenerCount("SIGINT"); +test("withUserCancellation returns the result and removes its signal listeners", async () => { + const initialListeners = ["SIGINT", "SIGTERM"].map((signal) => process.listenerCount(signal)); let signal: AbortSignal | undefined; + let cancellations = 0; - const result = await withUserCancellation(async (current) => { - signal = current; - return "done"; - }); + const result = await withUserCancellation( + async (current) => { + signal = current; + return "done"; + }, + () => cancellations++, + ); expect(result).toBe("done"); expect(signal?.aborted).toBe(true); - expect(process.listenerCount("SIGINT")).toBe(initialListeners); + expect(cancellations).toBe(0); + expect(["SIGINT", "SIGTERM"].map((signal) => process.listenerCount(signal))).toEqual( + initialListeners, + ); }); -test("withUserCancellation replaces transport aborts with the shared reason", async () => { - const initialListeners = process.listenerCount("SIGINT"); - let signal: AbortSignal | undefined; - const pending = withUserCancellation((current) => { - signal = current; - return new Promise((_, reject) => { - const abort = () => reject(new Error("transport aborted")); - if (current.aborted) abort(); - else current.addEventListener("abort", abort, { once: true }); - }); - }); +test.each(["SIGINT", "SIGTERM"] as const)( + "withUserCancellation replaces %s transport aborts with the shared reason", + async (signalName) => { + const initialListeners = ["SIGINT", "SIGTERM"].map((signal) => process.listenerCount(signal)); + let signal: AbortSignal | undefined; + let cancellations = 0; + const pending = withUserCancellation( + (current) => { + signal = current; + return new Promise((_, reject) => { + const abort = () => reject(new Error("transport aborted")); + if (current.aborted) abort(); + else current.addEventListener("abort", abort, { once: true }); + }); + }, + () => cancellations++, + ); - process.emit("SIGINT", "SIGINT"); + process.emit(signalName, signalName); + process.emit(signalName, signalName); - expect(signal?.reason).toBeInstanceOf(UserCancellationError); - await expect(pending).rejects.toBe(signal?.reason); - expect(process.listenerCount("SIGINT")).toBe(initialListeners); -}); + expect(signal?.reason).toBeInstanceOf(UserCancellationError); + await expect(pending).rejects.toBe(signal?.reason); + expect(cancellations).toBe(1); + expect(["SIGINT", "SIGTERM"].map((signal) => process.listenerCount(signal))).toEqual( + initialListeners, + ); + }, +); test("withUserCancellation preserves non-cancellation failures", async () => { const failure = new TypeError("operation failed"); diff --git a/src/runnable/index.tsx b/src/runnable/index.tsx index 1cd2401671..b37a7aa783 100644 --- a/src/runnable/index.tsx +++ b/src/runnable/index.tsx @@ -1,11 +1,19 @@ import { AgentCoreCLIError, ExitCode, SilentCLIError, UserCancellationError } from "../errors"; -/** Runs a headless operation with process SIGINT mapped to UserCancellationError. */ -export async function withUserCancellation(fn: (signal: AbortSignal) => Promise): Promise { +/** Runs an operation with SIGINT and SIGTERM mapped to UserCancellationError. */ +export async function withUserCancellation( + fn: (signal: AbortSignal) => Promise, + onCancel?: () => void, +): Promise { const controller = new AbortController(); - const interrupt = () => controller.abort(new UserCancellationError()); + const interrupt = () => { + if (controller.signal.aborted) return; + onCancel?.(); + controller.abort(new UserCancellationError()); + }; + const signals = ["SIGINT", "SIGTERM"] as const; // Ink's signal-exit handler must see our listener until cancellation cleanup settles. - process.on("SIGINT", interrupt); + for (const signal of signals) process.on(signal, interrupt); try { const result = await fn(controller.signal); controller.signal.throwIfAborted(); @@ -15,7 +23,7 @@ export async function withUserCancellation(fn: (signal: AbortSignal) => Promi throw error; } finally { controller.abort(); - process.off("SIGINT", interrupt); + for (const signal of signals) process.off(signal, interrupt); } } diff --git a/src/tui/tui.test.tsx b/src/tui/tui.test.tsx index 3a295895a1..6143176ded 100644 --- a/src/tui/tui.test.tsx +++ b/src/tui/tui.test.tsx @@ -125,7 +125,7 @@ describe("TUI handoff", () => { ({ name: "test-project", rootPath: process.cwd(), - spec: { runtimes: [] } as unknown as Project["spec"], + spec: { runtimes: [], harnesses: [] } as unknown as Project["spec"], }) as Project; const { streams, stdin } = ttyTestIO(); const root = createRootHandler(core, { @@ -142,7 +142,7 @@ describe("TUI handoff", () => { await waitFor(() => streams.stdout().includes("/ dev")); stdin.write("\r"); - // The project has no runtimes, so dev's own validation fails: proof the + // The project has no agents, so dev's own validation fails: proof the // TUI handed off to the dev command rather than showing its help. const error = await routePromise.catch((error) => error); expect(error).toBeInstanceOf(InputValidationError);