From 70e299b1e607fcffd3374dd23095651cbcc45e48 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Wed, 19 Aug 2026 15:42:47 -0400 Subject: [PATCH 1/5] feat(dev): collect local OTEL traces in project dev MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An in-process OTLP/HTTP receiver (protobuf via the pinned otlp-transformer decoders, or JSON) persists agent traces through the storage layer. project dev starts it unless --no-traces or the runtime disables instrumentation, points every spawned agent at it (signal-specific OTEL env), rewrites the endpoint to host.docker.internal for containers — with an explicit host-gateway mapping so Linux Docker Engine resolves it — and keeps uvicorn --reload workers instrumented via sitecustomize on PYTHONPATH. Oversized collector requests get a 413 before the connection closes so exporters do not retry them as transient failures. Rebuilt from explicit paths: the previous tree-snapshot commit accidentally reverted unrelated merged work (cdk target guard, config-bundle TUI, error classification). --- bun.lock | 54 +++++- package.json | 4 + .../templates/shared/env.local.template | 6 +- src/core/dev/codezip.test.ts | 81 ++++++++- src/core/dev/codezip.ts | 53 +++++- src/core/dev/container.test.ts | 2 + src/core/dev/container.ts | 5 + src/core/dev/otel/collector.test.ts | 172 ++++++++++++++++++ src/core/dev/otel/collector.ts | 152 ++++++++++++++++ src/handlers/project/dev/index.test.ts | 94 +++++++++- src/handlers/project/dev/index.ts | 38 +++- src/handlers/project/dev/types.ts | 13 ++ src/handlers/project/index.ts | 2 + src/io/httpServer.test.ts | 73 ++++++++ src/io/httpServer.ts | 119 ++++++++++++ src/io/index.ts | 7 + src/router/flags.tsx | 8 +- src/router/router.test.ts | 21 +++ 18 files changed, 883 insertions(+), 21 deletions(-) create mode 100644 src/core/dev/otel/collector.test.ts create mode 100644 src/core/dev/otel/collector.ts create mode 100644 src/io/httpServer.test.ts create mode 100644 src/io/httpServer.ts diff --git a/bun.lock b/bun.lock index 779b6bc4c..2d059282b 100644 --- a/bun.lock +++ b/bun.lock @@ -11,6 +11,7 @@ "@aws-sdk/client-iam": "^3.1080.0", "@opentelemetry/api": "^1.9.1", "@opentelemetry/exporter-metrics-otlp-http": "^0.221.0", + "@opentelemetry/otlp-transformer": "0.213.0", "@opentelemetry/resources": "^2.10.0", "@opentelemetry/sdk-metrics": "^2.10.0", "@smithy/core": "3.29.3", @@ -43,6 +44,9 @@ }, }, }, + "overrides": { + "@opentelemetry/core": "^2.10.0", + }, "packages": { "@alcalzone/ansi-tokenize": ["@alcalzone/ansi-tokenize@0.3.0", "", { "dependencies": { "ansi-styles": "^6.2.1", "is-fullwidth-code-point": "^5.0.0" } }, "sha512-p+CMKJ93HFmLkjXKlXiVGlMQEuRb6H0MokBSwUsX+S6BRX8eV5naFZpQJFfJHjRZY0Hmnqy1/r6UWl3x+19zYA=="], @@ -90,7 +94,7 @@ "@opentelemetry/api": ["@opentelemetry/api@1.9.1", "", {}, "sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q=="], - "@opentelemetry/api-logs": ["@opentelemetry/api-logs@0.221.0", "", { "dependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-OlanaW1vv7ufTqQ3/fPLI4arGt5ZoM+P8abOMki6uEYnpRazepSWDwDnnw+la7kE26SHVC18//SMccrDvLKOXQ=="], + "@opentelemetry/api-logs": ["@opentelemetry/api-logs@0.213.0", "", { "dependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-zRM5/Qj6G84Ej3F1yt33xBVY/3tnMxtL1fiDIxYbDWYaZ/eudVw3/PBiZ8G7JwUxXxjW8gU4g6LnOyfGKYHYgw=="], "@opentelemetry/core": ["@opentelemetry/core@2.10.0", "", { "dependencies": { "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.0.0 <1.10.0" } }, "sha512-/wNZ8twnEQQA4HoHu22+vcsdru6pWPWxW+7w+FlxT6Id7PE/WIbZmVKkte+PF72e0F2dnImFeHD2syyE1Mw6MQ=="], @@ -98,16 +102,18 @@ "@opentelemetry/otlp-exporter-base": ["@opentelemetry/otlp-exporter-base@0.221.0", "", { "dependencies": { "@opentelemetry/core": "2.10.0", "@opentelemetry/otlp-transformer": "0.221.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-UFPIq80OH3Ns/oPFHRj14d4DTOxUo+MUFU8hUiCq5jTqFhdeJnfVSANHT+xp92409cA+oxzvlZCe6NM1wvCuBA=="], - "@opentelemetry/otlp-transformer": ["@opentelemetry/otlp-transformer@0.221.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.221.0", "@opentelemetry/core": "2.10.0", "@opentelemetry/resources": "2.10.0", "@opentelemetry/sdk-logs": "0.221.0", "@opentelemetry/sdk-metrics": "2.10.0", "@opentelemetry/sdk-trace": "2.10.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-lg6lkOU08Az23jVcn/0Els9HP+V8PnR4Km6p0KgpTggS0n/WuhnmY64rSh83Of9iR9nD+dpWr6adlcX8KzAwjg=="], + "@opentelemetry/otlp-transformer": ["@opentelemetry/otlp-transformer@0.213.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.213.0", "@opentelemetry/core": "2.6.0", "@opentelemetry/resources": "2.6.0", "@opentelemetry/sdk-logs": "0.213.0", "@opentelemetry/sdk-metrics": "2.6.0", "@opentelemetry/sdk-trace-base": "2.6.0", "protobufjs": "^7.0.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-RSuAlxFFPjeK4d5Y6ps8L2WhaQI6CXWllIjvo5nkAlBpmq2XdYWEBGiAbOF4nDs8CX4QblJDv5BbMUft3sEfDw=="], "@opentelemetry/resources": ["@opentelemetry/resources@2.10.0", "", { "dependencies": { "@opentelemetry/core": "2.10.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-q6MMm2zhggzsHVNbabYwut+a6nbuQQe3URUoxaojM/8K1IBfwwPzvxIjNi2/lI1TFe+fMHMW9MWhrtDLEXEnkA=="], - "@opentelemetry/sdk-logs": ["@opentelemetry/sdk-logs@0.221.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.221.0", "@opentelemetry/core": "2.10.0", "@opentelemetry/resources": "2.10.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.4.0 <1.10.0" } }, "sha512-FaDcazjyMp7TZZZAsqbo4IkovP0UegoCu0EBkiNt+qCqvUf7FPAsfcrZ3+ZEkKgXZ/jHafop+JoGPDk3A0SmLg=="], + "@opentelemetry/sdk-logs": ["@opentelemetry/sdk-logs@0.213.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.213.0", "@opentelemetry/core": "2.6.0", "@opentelemetry/resources": "2.6.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.4.0 <1.10.0" } }, "sha512-00xlU3GZXo3kXKve4DLdrAL0NAFUaZ9appU/mn00S/5kSUdAvyYsORaDUfR04Mp2CLagAOhrzfUvYozY/EZX2g=="], "@opentelemetry/sdk-metrics": ["@opentelemetry/sdk-metrics@2.10.0", "", { "dependencies": { "@opentelemetry/core": "2.10.0", "@opentelemetry/resources": "2.10.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.9.0 <1.10.0" } }, "sha512-t6r1VSvXNtSDnPXU1FbZeetJb7yyovHmgu0wRSoftxtE0g2rSNhQZQUy69sRUCL+iioJpX8SN/S6wq6ZtvLySQ=="], "@opentelemetry/sdk-trace": ["@opentelemetry/sdk-trace@2.10.0", "", { "dependencies": { "@opentelemetry/core": "2.10.0", "@opentelemetry/resources": "2.10.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-MfQGq3GRmTh5fM/y+OjaO0vj6+luCB1XO2gfXCalKCfgKw0eHL++sm75DNweC6ohlp+aFvACqeE0fYayqdRaoQ=="], + "@opentelemetry/sdk-trace-base": ["@opentelemetry/sdk-trace-base@2.6.0", "", { "dependencies": { "@opentelemetry/core": "2.6.0", "@opentelemetry/resources": "2.6.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-g/OZVkqlxllgFM7qMKqbPV9c1DUPhQ7d4n3pgZFcrnrNft9eJXZM2TNHTPYREJBrtNdRytYyvwjgL5geDKl3EQ=="], + "@opentelemetry/semantic-conventions": ["@opentelemetry/semantic-conventions@1.43.0", "", {}, "sha512-eSYWTm620tTk45EKSedaUL8MFYI8hW164hIXsgIHyxu3VobUB3fFCu5t0hQby6OoWRPsG1KkKUG2M5UadiLiVg=="], "@oxlint/binding-android-arm-eabi": ["@oxlint/binding-android-arm-eabi@1.74.0", "", { "os": "android", "cpu": "arm" }, "sha512-+gHd12muVI9ZLBaWLPkHt3Fj7jihFjgQ1MGtBaRL8vWrWrI0P7dLUty/cHrHS0oqPYIRgQUJsPu2CExQuMcwNw=="], @@ -148,6 +154,24 @@ "@oxlint/binding-win32-x64-msvc": ["@oxlint/binding-win32-x64-msvc@1.74.0", "", { "os": "win32", "cpu": "x64" }, "sha512-VTC9IYTIMrVUk/i6Ms1ohzzDKZFkWn0KU2OBbPBzgmVZ2V30165T/zK4LztTr0Xgp9fZ1qQZ1rsZAu/rEmySlA=="], + "@protobufjs/aspromise": ["@protobufjs/aspromise@1.1.2", "", {}, "sha512-j+gKExEuLmKwvz3OgROXtrJ2UG2x8Ch2YZUxahh+s1F2HZ+wAceUNLkvy6zKCPVRkU++ZWQrdxsUeQXmcg4uoQ=="], + + "@protobufjs/base64": ["@protobufjs/base64@1.1.2", "", {}, "sha512-AZkcAA5vnN/v4PDqKyMR5lx7hZttPDgClv83E//FMNhR2TMcLUhfRUBHCmSl0oi9zMgDDqRUJkSxO3wm85+XLg=="], + + "@protobufjs/codegen": ["@protobufjs/codegen@2.0.5", "", {}, "sha512-zgXFLzW3Ap33e6d0Wlj4MGIm6Ce8O89n/apUaGNB/jx+hw+ruWEp7EwGUshdLKVRCxZW12fp9r40E1mQrf/34g=="], + + "@protobufjs/eventemitter": ["@protobufjs/eventemitter@1.1.1", "", {}, "sha512-vW1GmwMZNnL+gMRaovlh9yZX74kc+TTU3FObkkurpMaRtBfLP3ldjS9KQWlwZgraRE0+dheEEoAxdzcJQ8eXZg=="], + + "@protobufjs/fetch": ["@protobufjs/fetch@1.1.1", "", { "dependencies": { "@protobufjs/aspromise": "^1.1.1" } }, "sha512-GpptLrs57adMSuHi3VNj0mAF8dwh36LMaYF6XyJ6JMWlVsc+t42tm1HSEDmOs3A8fC9yyeisgLhsTVQokOZ0zw=="], + + "@protobufjs/float": ["@protobufjs/float@1.0.2", "", {}, "sha512-Ddb+kVXlXst9d+R9PfTIxh1EdNkgoRe5tOX6t01f1lYWOvJnSPDBlG241QLzcyPdoNTsblLUdujGSE4RzrTZGQ=="], + + "@protobufjs/path": ["@protobufjs/path@1.1.2", "", {}, "sha512-6JOcJ5Tm08dOHAbdR3GrvP+yUUfkjG5ePsHYczMFLq3ZmMkAD98cDgcT2iA1lJ9NVwFd4tH/iSSoe44YWkltEA=="], + + "@protobufjs/pool": ["@protobufjs/pool@1.1.0", "", {}, "sha512-0kELaGSIDBKvcgS4zkjz1PeddatrjYcmMWOlAuAPwAeccUrPHdUqo/J6LiymHHEiJT5NrF1UVwxY14f+fy4WQw=="], + + "@protobufjs/utf8": ["@protobufjs/utf8@1.1.2", "", {}, "sha512-b1UQwcEZ4yCnMCD8DAL1VlbvBJE9/IX4FTIp7BG1xYpf29SLazLSrqUkj4w7Y5y7cCVP6E5tcqqcI0xemPkHug=="], + "@smithy/core": ["@smithy/core@3.29.8", "", { "dependencies": { "@smithy/types": "^4.16.1", "tslib": "^2.6.2" } }, "sha512-rpCbCV+TimOBi3VLNBMmtTvgfOWcFIEAru3+TFlG87SL2F+te4jOnnNR+cf3uR4eJ5Qf4LnT80fqnBKgPRS6zA=="], "@smithy/credential-provider-imds": ["@smithy/credential-provider-imds@4.4.13", "", { "dependencies": { "@smithy/core": "^3.29.8", "@smithy/types": "^4.16.1", "tslib": "^2.6.2" } }, "sha512-X+2HNZhWi5i3rJsCas0LPf6fTQUaKyJ40zd8aTO/bwpRfpU3biYaqLr7C1WMibL7PVKJalpi1PyybjGPNoHC8Q=="], @@ -266,6 +290,8 @@ "logform": ["logform@2.7.0", "", { "dependencies": { "@colors/colors": "1.6.0", "@types/triple-beam": "^1.3.2", "fecha": "^4.2.0", "ms": "^2.1.1", "safe-stable-stringify": "^2.3.1", "triple-beam": "^1.3.0" } }, "sha512-TFYA4jnP7PVbmlBIfhlSe+WKxs9dklXMTEGcBCIvLhE/Tn3H6Gk1norupVW7m5Cnd4bLcr08AytbyV/xj7f/kQ=="], + "long": ["long@5.3.2", "", {}, "sha512-mNAgZ1GmyNhD7AuqnTG3/VQ26o760+ZYBPKjPvugO8+nLbYfX6TVpJPseBvopbdY+qpZ/lKUnmEc1LeZYS3QAA=="], + "mimic-fn": ["mimic-fn@2.1.0", "", {}, "sha512-OqbOk5oEQeAZ8WXWydlu9HJjz9WVdEIvamMCcXmuqUYjTknH/sqsWvhQ3vgwKFRR1HpjvNBKQ37nbJgYzGqGcg=="], "mimic-function": ["mimic-function@5.0.1", "", {}, "sha512-VP79XUPxV2CigYP3jWwAUFSku2aKqBH7uTAapFWCBqutsbmDo96KY5o8uh6U+/YSIn5OxJnXp73beVkpqMIGhA=="], @@ -292,6 +318,8 @@ "prettier": ["prettier@3.9.5", "", { "bin": { "prettier": "bin/prettier.cjs" } }, "sha512-/FVl766LpUfB5vXgCYOYa0MeV/441Ia99AeICQIQFTY/Nw0roZwULcXpku5i1/m5kt/baz+s4Zogspd839HSMg=="], + "protobufjs": ["protobufjs@7.6.5", "", { "dependencies": { "@protobufjs/aspromise": "^1.1.2", "@protobufjs/base64": "^1.1.2", "@protobufjs/codegen": "^2.0.5", "@protobufjs/eventemitter": "^1.1.1", "@protobufjs/fetch": "^1.1.1", "@protobufjs/float": "^1.0.2", "@protobufjs/path": "^1.1.2", "@protobufjs/pool": "^1.1.0", "@protobufjs/utf8": "^1.1.1", "@types/node": ">=13.7.0", "long": "^5.3.2" } }, "sha512-/FPD0nUc9jH6rfFjji9IBqOz4pcSE3CsT1m7Ep6Mdb0LxSUMj8hgl6GomOvZzpNpAqqGaXA0P3VSrZLFzIhQrw=="], + "react": ["react@19.2.7", "", {}, "sha512-HNe9WslTbXmFK8o8cmwgAeJFSBvt1bPdHCVKtaaV+WlAN36mpT4hcRpwbf3fY56ar2oIXzsBpOAiIRHAdY0OlQ=="], "react-devtools-core": ["react-devtools-core@7.0.1", "", { "dependencies": { "shell-quote": "^1.6.1", "ws": "^7" } }, "sha512-C3yNvRHaizlpiASzy7b9vbnBGLrhvdhl1CbdU6EnZgxPNbai60szdLtl+VL76UNOt5bOoVTOz5rNWZxgGt+Gsw=="], @@ -406,6 +434,18 @@ "@aws-sdk/client-iam/@smithy/node-http-handler": ["@smithy/node-http-handler@4.9.5", "", { "dependencies": { "@smithy/core": "^3.29.3", "@smithy/types": "^4.16.1", "tslib": "^2.6.2" } }, "sha512-bNqdxTQTxmLbomSmlkZFz8L6B/feQ2HHzw4L2zY7Ecp2XffYAZq2uzdWDdxJHJFbEvqd+SRuluJso0P8+xPdbw=="], + "@opentelemetry/exporter-metrics-otlp-http/@opentelemetry/otlp-transformer": ["@opentelemetry/otlp-transformer@0.221.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.221.0", "@opentelemetry/core": "2.10.0", "@opentelemetry/resources": "2.10.0", "@opentelemetry/sdk-logs": "0.221.0", "@opentelemetry/sdk-metrics": "2.10.0", "@opentelemetry/sdk-trace": "2.10.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-lg6lkOU08Az23jVcn/0Els9HP+V8PnR4Km6p0KgpTggS0n/WuhnmY64rSh83Of9iR9nD+dpWr6adlcX8KzAwjg=="], + + "@opentelemetry/otlp-exporter-base/@opentelemetry/otlp-transformer": ["@opentelemetry/otlp-transformer@0.221.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.221.0", "@opentelemetry/core": "2.10.0", "@opentelemetry/resources": "2.10.0", "@opentelemetry/sdk-logs": "0.221.0", "@opentelemetry/sdk-metrics": "2.10.0", "@opentelemetry/sdk-trace": "2.10.0" }, "peerDependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-lg6lkOU08Az23jVcn/0Els9HP+V8PnR4Km6p0KgpTggS0n/WuhnmY64rSh83Of9iR9nD+dpWr6adlcX8KzAwjg=="], + + "@opentelemetry/otlp-transformer/@opentelemetry/resources": ["@opentelemetry/resources@2.6.0", "", { "dependencies": { "@opentelemetry/core": "2.6.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-D4y/+OGe3JSuYUCBxtH5T9DSAWNcvCb/nQWIga8HNtXTVPQn59j0nTBAgaAXxUVBDl40mG3Tc76b46wPlZaiJQ=="], + + "@opentelemetry/otlp-transformer/@opentelemetry/sdk-metrics": ["@opentelemetry/sdk-metrics@2.6.0", "", { "dependencies": { "@opentelemetry/core": "2.6.0", "@opentelemetry/resources": "2.6.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.9.0 <1.10.0" } }, "sha512-CicxWZxX6z35HR83jl+PLgtFgUrKRQ9LCXyxgenMnz5A1lgYWfAog7VtdOvGkJYyQgMNPhXQwkYrDLujk7z1Iw=="], + + "@opentelemetry/sdk-logs/@opentelemetry/resources": ["@opentelemetry/resources@2.6.0", "", { "dependencies": { "@opentelemetry/core": "2.6.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-D4y/+OGe3JSuYUCBxtH5T9DSAWNcvCb/nQWIga8HNtXTVPQn59j0nTBAgaAXxUVBDl40mG3Tc76b46wPlZaiJQ=="], + + "@opentelemetry/sdk-trace-base/@opentelemetry/resources": ["@opentelemetry/resources@2.6.0", "", { "dependencies": { "@opentelemetry/core": "2.6.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-D4y/+OGe3JSuYUCBxtH5T9DSAWNcvCb/nQWIga8HNtXTVPQn59j0nTBAgaAXxUVBDl40mG3Tc76b46wPlZaiJQ=="], + "listr2/cli-truncate": ["cli-truncate@5.2.0", "", { "dependencies": { "slice-ansi": "^8.0.0", "string-width": "^8.2.0" } }, "sha512-xRwvIOMGrfOAnM1JYtqQImuaNtDEv9v6oIYAs4LIHwTiKee8uwvIi363igssOC0O5U04i4AlENs79LQLu9tEMw=="], "log-update/cli-cursor": ["cli-cursor@5.0.0", "", { "dependencies": { "restore-cursor": "^5.0.0" } }, "sha512-aCj4O5wKyszjMmDT4tZj93kxyydN/K5zPWSCe6/0AV/AA1pqe5ZBIw0a2ZfPQV7lL5/yb5HsUreJ6UFAF1tEQw=="], @@ -470,6 +510,14 @@ "@aws-sdk/client-iam/@aws-sdk/credential-provider-node/@smithy/credential-provider-imds": ["@smithy/credential-provider-imds@4.4.8", "", { "dependencies": { "@smithy/core": "^3.29.3", "@smithy/types": "^4.16.1", "tslib": "^2.6.2" } }, "sha512-q9J7JTiXrAhB8sDp4px97uEPT7CwKH61Co78grdNQvU8QZAdiuaSRhP0tUVf2ogy36RZTrlMU1rBmDEH+cnkiA=="], + "@opentelemetry/exporter-metrics-otlp-http/@opentelemetry/otlp-transformer/@opentelemetry/api-logs": ["@opentelemetry/api-logs@0.221.0", "", { "dependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-OlanaW1vv7ufTqQ3/fPLI4arGt5ZoM+P8abOMki6uEYnpRazepSWDwDnnw+la7kE26SHVC18//SMccrDvLKOXQ=="], + + "@opentelemetry/exporter-metrics-otlp-http/@opentelemetry/otlp-transformer/@opentelemetry/sdk-logs": ["@opentelemetry/sdk-logs@0.221.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.221.0", "@opentelemetry/core": "2.10.0", "@opentelemetry/resources": "2.10.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.4.0 <1.10.0" } }, "sha512-FaDcazjyMp7TZZZAsqbo4IkovP0UegoCu0EBkiNt+qCqvUf7FPAsfcrZ3+ZEkKgXZ/jHafop+JoGPDk3A0SmLg=="], + + "@opentelemetry/otlp-exporter-base/@opentelemetry/otlp-transformer/@opentelemetry/api-logs": ["@opentelemetry/api-logs@0.221.0", "", { "dependencies": { "@opentelemetry/api": "^1.3.0" } }, "sha512-OlanaW1vv7ufTqQ3/fPLI4arGt5ZoM+P8abOMki6uEYnpRazepSWDwDnnw+la7kE26SHVC18//SMccrDvLKOXQ=="], + + "@opentelemetry/otlp-exporter-base/@opentelemetry/otlp-transformer/@opentelemetry/sdk-logs": ["@opentelemetry/sdk-logs@0.221.0", "", { "dependencies": { "@opentelemetry/api-logs": "0.221.0", "@opentelemetry/core": "2.10.0", "@opentelemetry/resources": "2.10.0", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.4.0 <1.10.0" } }, "sha512-FaDcazjyMp7TZZZAsqbo4IkovP0UegoCu0EBkiNt+qCqvUf7FPAsfcrZ3+ZEkKgXZ/jHafop+JoGPDk3A0SmLg=="], + "listr2/cli-truncate/slice-ansi": ["slice-ansi@8.0.0", "", { "dependencies": { "ansi-styles": "^6.2.3", "is-fullwidth-code-point": "^5.1.0" } }, "sha512-stxByr12oeeOyY2BlviTNQlYV5xOj47GirPr4yA1hE9JCtxfQN0+tVbkxwCtYDQWhEKWFHsEK48ORg5jrouCAg=="], "log-update/cli-cursor/restore-cursor": ["restore-cursor@5.1.0", "", { "dependencies": { "onetime": "^7.0.0", "signal-exit": "^4.1.0" } }, "sha512-oMA2dcrw6u0YfxJQXm342bFKX/E4sG9rbTzO9ptUcR/e8A33cHuvStiYOwH7fszkZlZ1z/ta9AAoPk2F4qIOHA=="], diff --git a/package.json b/package.json index 8786ec2a3..42fdb3f57 100644 --- a/package.json +++ b/package.json @@ -58,6 +58,7 @@ "@aws-sdk/client-iam": "^3.1080.0", "@opentelemetry/api": "^1.9.1", "@opentelemetry/exporter-metrics-otlp-http": "^0.221.0", + "@opentelemetry/otlp-transformer": "0.213.0", "@opentelemetry/resources": "^2.10.0", "@opentelemetry/sdk-metrics": "^2.10.0", "@smithy/core": "3.29.3", @@ -75,5 +76,8 @@ "winston": "^3.19.0", "winston-daily-rotate-file": "^5.0.0", "zod": "^4.4.3" + }, + "overrides": { + "@opentelemetry/core": "^2.10.0" } } diff --git a/src/assets/templates/shared/env.local.template b/src/assets/templates/shared/env.local.template index cc9e703e7..fb931a12b 100644 --- a/src/assets/templates/shared/env.local.template +++ b/src/assets/templates/shared/env.local.template @@ -1,7 +1,11 @@ # Environment variables for local development. # `agentcore project dev` loads this file into your agent's process. Values here # override injected values except PORT, FASTMCP_PORT, and LOCAL_DEV, which the -# CLI owns. This file is gitignored — keep secrets out of version control. +# CLI owns. While trace collection is on (the default), the CLI also owns the +# OTEL_* and AGENT_OBSERVABILITY_ENABLED variables so traces reach its local +# collector — pass --no-traces (or set instrumentation.enableOtel to false in +# agentcore.json) to disable collection and set your own. +# This file is gitignored — keep secrets out of version control. # # Example: # MY_API_KEY=... diff --git a/src/core/dev/codezip.test.ts b/src/core/dev/codezip.test.ts index 24b83f9f3..da366d59d 100644 --- a/src/core/dev/codezip.test.ts +++ b/src/core/dev/codezip.test.ts @@ -1,11 +1,11 @@ import { afterEach, describe, expect, test } from "bun:test"; import { mkdir, mkdtemp, rm, symlink, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; -import { join, relative } from "node:path"; +import { delimiter, join, relative } from "node:path"; import { InputValidationError } from "../../errors"; import type { ProjectRuntime } from "../../projectSchemas/runtime"; import type { DevEvent, DevServerInput } from "../../handlers/project/dev/types"; -import type { ProcessEvent, ProcessStreamer, StreamProcessOptions } from "../../io"; +import type { ProcessEvent, ProcessRunner, ProcessStreamer, StreamProcessOptions } from "../../io"; import { CodeZipDevRunner } from "./codezip"; type ProcessCall = { @@ -54,15 +54,22 @@ async function projectRoot(withNodeModules = false): Promise { return root; } -function harness(output: ProcessEvent[] = []) { +function harness(output: ProcessEvent[] = [], probe: { dir?: string; fail?: boolean } = {}) { const calls: ProcessCall[] = []; + const probeCalls: string[][] = []; const fakeStreamProcess: ProcessStreamer = async function* (command, options) { calls.push({ command, options }); yield* output; }; + const fakeRunProcess: ProcessRunner = async (command, options) => { + probeCalls.push(command); + if (probe.fail) throw new Error("probe failed"); + options.onOutput?.(`${probe.dir ?? ""}\n`); + }; return { calls, - runner: new CodeZipDevRunner({ streamProcess: fakeStreamProcess }), + probeCalls, + runner: new CodeZipDevRunner({ streamProcess: fakeStreamProcess, runProcess: fakeRunProcess }), }; } @@ -193,3 +200,69 @@ describe("CodeZipDevRunner", () => { ]); }); }); + +describe("CodeZipDevRunner OTEL instrumentation", () => { + async function sitecustomizeDir(): Promise { + const directory = await mkdtemp(join(tmpdir(), "otel-site-")); + tempDirectories.push(directory); + await writeFile(join(directory, "sitecustomize.py"), ""); + return directory; + } + + function otelInput(root: string, extraEnv: Record = {}): DevServerInput { + const base = input(root, runtime()); + return { + ...base, + env: { ...base.env, OTEL_EXPORTER_OTLP_ENDPOINT: "http://127.0.0.1:4318", ...extraEnv }, + }; + } + + test("prepends the sitecustomize directory to PYTHONPATH when instrumentation is installed", async () => { + const root = await projectRoot(); + const directory = await sitecustomizeDir(); + const { calls, probeCalls, runner } = harness([], { dir: directory }); + + await collect(runner.run(otelInput(root))); + + expect(probeCalls[0]?.slice(0, 4)).toEqual(["uv", "run", "python", "-c"]); + expect(calls[0]?.options.env?.PYTHONPATH).toBe(directory); + }); + + test("preserves an existing PYTHONPATH", async () => { + const root = await projectRoot(); + const directory = await sitecustomizeDir(); + const { calls, runner } = harness([], { dir: directory }); + + await collect(runner.run(otelInput(root, { PYTHONPATH: "/existing" }))); + + expect(calls[0]?.options.env?.PYTHONPATH).toBe(`${directory}${delimiter}/existing`); + }); + + test("does not probe without an OTEL endpoint or for Node entrypoints", async () => { + const root = await projectRoot(true); + const { probeCalls, runner } = harness(); + + await collect(runner.run(input(root, runtime()))); + await collect(runner.run({ ...otelInput(root), runtime: runtime({ entrypoint: "index.js" }) })); + + expect(probeCalls).toEqual([]); + }); + + test.each([ + ["probe failure", { fail: true }], + ["missing sitecustomize.py", { dir: "/nonexistent" }], + ] as const)("warns and starts untraced on %s", async (_case, probe) => { + const root = await projectRoot(); + const { calls, probeCalls, runner } = harness([], probe); + + const events = await collect(runner.run(otelInput(root))); + + expect(probeCalls).toHaveLength(1); + expect(calls).toHaveLength(1); + expect(calls[0]?.options.env?.PYTHONPATH).toBeUndefined(); + expect(events).toContainEqual({ + type: "status", + message: expect.stringContaining("traces will not be collected"), + }); + }); +}); diff --git a/src/core/dev/codezip.ts b/src/core/dev/codezip.ts index 777656023..1c84c9317 100644 --- a/src/core/dev/codezip.ts +++ b/src/core/dev/codezip.ts @@ -1,19 +1,28 @@ import { existsSync } from "node:fs"; -import { join, resolve } from "node:path"; +import { delimiter, join, resolve } from "node:path"; import { InputValidationError } from "../../errors"; import type { DevEvent, DevRunner, DevServerInput } from "../../handlers/project/dev/types"; -import { streamProcess, type ProcessStreamer, type StreamProcessOptions } from "../../io"; +import { + runProcess, + streamProcess, + type ProcessRunner, + type ProcessStreamer, + type StreamProcessOptions, +} from "../../io"; import { isDirectory, isFile, resolvePathWithinProject } from "./path"; type CodeZipDevRunnerConfig = { streamProcess?: ProcessStreamer; + runProcess?: ProcessRunner; }; export class CodeZipDevRunner implements DevRunner { private readonly streamProcess: ProcessStreamer; + private readonly runProcess: ProcessRunner; constructor(config: CodeZipDevRunnerConfig = {}) { this.streamProcess = config.streamProcess ?? streamProcess; + this.runProcess = config.runProcess ?? runProcess; } public async *run(input: DevServerInput): AsyncGenerator { @@ -41,8 +50,48 @@ export class CodeZipDevRunner implements DevRunner { yield { type: "status", message: "Starting development server" }; const serverProcess = commandForRuntime(entrypoint!, directory, input); + if (entrypoint!.endsWith(".py") && input.env?.OTEL_EXPORTER_OTLP_ENDPOINT) { + const sitecustomizeDir = await this.findOtelSitecustomizeDir(directory); + if (sitecustomizeDir) { + const existing = serverProcess.options.env?.PYTHONPATH; + serverProcess.options.env = { + ...serverProcess.options.env, + PYTHONPATH: existing ? `${sitecustomizeDir}${delimiter}${existing}` : sitecustomizeDir, + }; + } else { + yield { + type: "status", + message: + "OTEL auto-instrumentation is not installed in the agent environment; traces will not be collected. Add aws-opentelemetry-distro to the agent's dependencies to enable them.", + }; + } + } yield* this.streamProcess(serverProcess.command, serverProcess.options); } + + /** + * Locate the auto-instrumentation sitecustomize.py directory inside the agent's + * uv environment. Prepending it to PYTHONPATH instruments every Python process — + * an `opentelemetry-instrument` wrapper would only instrument uvicorn's reloader + * parent, leaving the re-spawned worker processes untraced. + */ + private async findOtelSitecustomizeDir(directory: string): Promise { + const output: string[] = []; + const probe = + "import opentelemetry.instrumentation.auto_instrumentation as m; import os; print(os.path.dirname(m.__file__))"; + try { + await this.runProcess(["uv", "run", "python", "-c", probe], { + cwd: directory, + onOutput: (chunk) => output.push(chunk), + }); + } catch { + return undefined; + } + const sitecustomizeDir = output.join("").trim().split("\n").at(-1)?.trim(); + if (!sitecustomizeDir || !existsSync(join(sitecustomizeDir, "sitecustomize.py"))) + return undefined; + return sitecustomizeDir; + } } function commandForRuntime( diff --git a/src/core/dev/container.test.ts b/src/core/dev/container.test.ts index 1b36e71d4..accbc223c 100644 --- a/src/core/dev/container.test.ts +++ b/src/core/dev/container.test.ts @@ -207,6 +207,8 @@ describe("ContainerDevRunner", () => { containerName(root), "-p", `127.0.0.1:3000:${containerPort}`, + "--add-host", + "host.docker.internal:host-gateway", "--env-file", run.envFile!.path, imageTag(root), diff --git a/src/core/dev/container.ts b/src/core/dev/container.ts index 12da02b54..fdb9fd323 100644 --- a/src/core/dev/container.ts +++ b/src/core/dev/container.ts @@ -183,6 +183,11 @@ export class ContainerDevRunner implements DevRunner { containerName, "-p", `127.0.0.1:${input.port}:${containerPort}`, + // Docker Engine on Linux does not define host.docker.internal (Desktop, + // Podman, and Finch do); the mapping makes the OTLP endpoint rewrite + // resolve everywhere and is harmless where the name already exists. + "--add-host", + "host.docker.internal:host-gateway", ...awsMount, "--env-file", envFile, diff --git a/src/core/dev/otel/collector.test.ts b/src/core/dev/otel/collector.test.ts new file mode 100644 index 000000000..eb4fc16e7 --- /dev/null +++ b/src/core/dev/otel/collector.test.ts @@ -0,0 +1,172 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + ExportLogsServiceRequest, + ExportTraceServiceRequest, + type OtelCollector, + startOtelCollector, +} from "./collector"; + +const TRACE_ID_HEX = "0123456789abcdef0123456789abcdef"; + +function protobufTracePayload(): Uint8Array { + const message = ExportTraceServiceRequest.fromObject({ + resourceSpans: [ + { + resource: { attributes: [{ key: "service.name", value: { stringValue: "proto-agent" } }] }, + scopeSpans: [ + { + scope: { name: "test" }, + spans: [ + { + traceId: Buffer.from(TRACE_ID_HEX, "hex"), + spanId: Buffer.from("0123456789abcdef", "hex"), + name: "invoke_agent strands", + kind: 1, + startTimeUnixNano: `${BigInt(Date.now()) * 1_000_000n}`, + endTimeUnixNano: `${BigInt(Date.now()) * 1_000_000n}`, + }, + ], + }, + ], + }, + ], + }); + return ExportTraceServiceRequest.encode(message).finish(); +} + +function protobufLogsPayload(): Uint8Array { + const message = ExportLogsServiceRequest.fromObject({ + resourceLogs: [ + { + resource: { attributes: [{ key: "service.name", value: { stringValue: "proto-agent" } }] }, + scopeLogs: [ + { + scope: { name: "test" }, + logRecords: [ + { + traceId: Buffer.from(TRACE_ID_HEX, "hex"), + timeUnixNano: `${BigInt(Date.now()) * 1_000_000n}`, + body: { stringValue: "a log line" }, + }, + ], + }, + ], + }, + ], + }); + return ExportLogsServiceRequest.encode(message).finish(); +} + +let directory: string; +let collector: OtelCollector; + +beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), "otel-collector-")); + collector = await startOtelCollector({ tracesDirectory: directory }); +}); + +afterEach(async () => { + await collector.close(); + await rm(directory, { recursive: true, force: true }); +}); + +function post( + path: string, + body: string | Uint8Array, + contentType = "application/x-protobuf", +): Promise { + return fetch(`http://127.0.0.1:${collector.port}${path}`, { + method: "POST", + headers: { "Content-Type": contentType }, + body, + }); +} + +describe("startOtelCollector", () => { + test("ingests protobuf trace exports and serves them back through the store", async () => { + const response = await post("/v1/traces", protobufTracePayload()); + expect(response.status).toBe(200); + + const traces = await collector.store.list(); + expect(traces).toHaveLength(1); + expect(traces[0]!.traceId).toBe(TRACE_ID_HEX); + }); + + test("ingests protobuf log exports into the same trace", async () => { + await post("/v1/traces", protobufTracePayload()); + const response = await post("/v1/logs", protobufLogsPayload()); + expect(response.status).toBe(200); + + const detail = await collector.store.get(TRACE_ID_HEX); + expect(detail?.resourceLogs).toBeDefined(); + }); + + test("ingests JSON trace exports", async () => { + const body = JSON.stringify({ + resourceSpans: [ + { + resource: { attributes: [{ key: "service.name", value: { stringValue: "json-agent" } }] }, + scopeSpans: [ + { + scope: { name: "test" }, + spans: [ + { + traceId: TRACE_ID_HEX, + spanId: "0123456789abcdef", + name: "invoke_agent strands", + kind: 1, + startTimeUnixNano: `${BigInt(Date.now()) * 1_000_000n}`, + endTimeUnixNano: `${BigInt(Date.now()) * 1_000_000n}`, + }, + ], + }, + ], + }, + ], + }); + const response = await post("/v1/traces", body, "application/json"); + expect(response.status).toBe(200); + expect((await collector.store.list()).map((trace) => trace.traceId)).toEqual([TRACE_ID_HEX]); + }); + + test("rejects malformed payloads with 400", async () => { + expect((await post("/v1/traces", "not json", "application/json")).status).toBe(400); + expect((await post("/v1/traces", Buffer.from([0xff, 0xff, 0xff]))).status).toBe(400); + expect(await collector.store.list()).toEqual([]); + }); + + test("health check responds ok and unknown routes 404", async () => { + const health = await fetch(`http://127.0.0.1:${collector.port}/`); + expect(await health.json()).toEqual({ status: "ok" }); + expect( + (await fetch(`http://127.0.0.1:${collector.port}/v1/metrics`, { method: "POST" })).status, + ).toBe(404); + }); + + test("envVars point the SDK at the collector, including signal-specific overrides", () => { + const endpoint = `http://127.0.0.1:${collector.port}`; + expect(collector.envVars).toMatchObject({ + OTEL_EXPORTER_OTLP_ENDPOINT: endpoint, + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: `${endpoint}/v1/traces`, + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT: `${endpoint}/v1/logs`, + OTEL_EXPORTER_OTLP_PROTOCOL: "http/protobuf", + OTEL_EXPORTER_OTLP_TRACES_PROTOCOL: "http/protobuf", + OTEL_EXPORTER_OTLP_LOGS_PROTOCOL: "http/protobuf", + OTEL_METRICS_EXPORTER: "none", + }); + }); + + test("abort signal closes the receiver", async () => { + const controller = new AbortController(); + const aborted = await startOtelCollector({ + tracesDirectory: directory, + signal: controller.signal, + }); + controller.abort(); + await Bun.sleep(20); + expect(fetch(`http://127.0.0.1:${aborted.port}/`)).rejects.toThrow(); + }); +}); diff --git a/src/core/dev/otel/collector.ts b/src/core/dev/otel/collector.ts new file mode 100644 index 000000000..c19a3375a --- /dev/null +++ b/src/core/dev/otel/collector.ts @@ -0,0 +1,152 @@ +// Decodes OTLP/HTTP protobuf payloads (the only protocol Python and Node OTEL +// SDKs export over HTTP) with the generated types from @opentelemetry/otlp-transformer. +// The version is pinned: newer releases dropped the generated request decoders. +import root from "@opentelemetry/otlp-transformer/build/src/generated/root"; +import { type HttpRequest, type HttpResponse, startHttpServer } from "../../../io"; +import { TraceStore } from "./store"; +import type { OtlpPayload } from "./types"; + +/** The slice of a generated protobufjs message type the collector (and its tests) use. */ +export interface OtlpMessageType { + decode(data: Uint8Array): unknown; + fromObject(object: object): unknown; + encode(message: unknown): { finish(): Uint8Array }; +} + +// The generated root's declaration file types it as an opaque protobufjs Root, +// so the real static-message shape is asserted once, here. +const { trace, logs } = ( + root as unknown as { + opentelemetry: { + proto: { + collector: { + trace: { v1: { ExportTraceServiceRequest: OtlpMessageType } }; + logs: { v1: { ExportLogsServiceRequest: OtlpMessageType } }; + }; + }; + }; + } +).opentelemetry.proto.collector; + +export const ExportTraceServiceRequest = trace.v1.ExportTraceServiceRequest; +export const ExportLogsServiceRequest = logs.v1.ExportLogsServiceRequest; +type OtlpDecoder = Pick; + +export interface OtelCollector { + /** The loopback port the OTLP/HTTP receiver listens on. */ + port: number; + /** Reads the traces this collector persists. */ + store: TraceStore; + /** Environment variables that point an agent's OTEL SDK at this collector. */ + envVars: Record; + /** Stops the receiver. Also invoked by the start signal, if one was given. */ + close(): Promise; +} + +export interface StartOtelCollectorOptions { + /** Directory to persist OTLP JSON Lines files into. */ + tracesDirectory: string; + /** Closes the collector when aborted. */ + signal?: AbortSignal; +} + +/** + * Starts an in-process OTLP/HTTP receiver for dev mode on an OS-assigned + * loopback port. Accepts `POST /v1/traces` and `POST /v1/logs` in protobuf or + * JSON encoding and appends the raw payloads to a TraceStore. + */ +export async function startOtelCollector( + options: StartOtelCollectorOptions, +): Promise { + const store = new TraceStore(options.tracesDirectory); + const server = await startHttpServer((request) => route(request, store), { + signal: options.signal, + }); + + return { port: server.port, store, envVars: otelEnvVars(server.port), close: server.close }; +} + +async function route(request: HttpRequest, store: TraceStore): Promise { + if (request.method === "POST" && request.url === "/v1/traces") { + return ingest(request, store, ExportTraceServiceRequest); + } + if (request.method === "POST" && request.url === "/v1/logs") { + return ingest(request, store, ExportLogsServiceRequest); + } + if (request.method === "GET" && request.url === "/") { + return json(200, { status: "ok" }); + } + return { status: 404 }; +} + +async function ingest( + request: HttpRequest, + store: TraceStore, + decoder: OtlpDecoder, +): Promise { + let payload: OtlpPayload; + try { + payload = decodePayload(request.body, String(request.headers["content-type"] ?? ""), decoder); + } catch { + return json(400, { error: "Invalid OTLP payload" }); + } + await store.append(payload); + return json(200, {}); +} + +/** + * Decode an OTLP payload. The JSON round-trip on the protobuf path converts the + * message to plain objects (protobufjs renders Long as string and bytes as base64). + */ +function decodePayload(body: Buffer, contentType: string, decoder: OtlpDecoder): OtlpPayload { + if (contentType.includes("application/json")) { + return JSON.parse(body.toString()) as OtlpPayload; + } + return JSON.parse(JSON.stringify(decoder.decode(new Uint8Array(body)))) as OtlpPayload; +} + +/** + * Environment for a spawned agent so its OTEL SDK exports to the collector at + * `port`. Signal-specific variables are set alongside the generic ones because + * they take precedence in the SDK — a stray OTEL_EXPORTER_OTLP_TRACES_ENDPOINT + * from the shell or .env.local must not silently redirect traces elsewhere. + * Per the OTEL spec, signal-specific endpoints are full URLs (the signal path + * is only appended to the generic endpoint). + */ +export function otelEnvVars(port: number): Record { + const endpoint = `http://127.0.0.1:${port}`; + return { + OTEL_EXPORTER_OTLP_ENDPOINT: endpoint, + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: `${endpoint}/v1/traces`, + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT: `${endpoint}/v1/logs`, + OTEL_EXPORTER_OTLP_PROTOCOL: "http/protobuf", + OTEL_EXPORTER_OTLP_TRACES_PROTOCOL: "http/protobuf", + OTEL_EXPORTER_OTLP_LOGS_PROTOCOL: "http/protobuf", + OTEL_METRICS_EXPORTER: "none", + AGENT_OBSERVABILITY_ENABLED: "true", + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: "true", + OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED: "true", + }; +} + +/** + * Rewrite loopback OTLP endpoints so a containerized agent can reach the + * collector on the host. host.docker.internal resolves on Docker Desktop, + * Finch, and Podman; bare-metal Linux Docker would additionally need + * `--add-host=host.docker.internal:host-gateway` (matches the reference CLI). + */ +export function rewriteOtelEndpointForContainer( + env: Record, +): Record { + const rewritten = { ...env }; + for (const [key, value] of Object.entries(rewritten)) { + if (key.startsWith("OTEL_EXPORTER_OTLP") && key.endsWith("_ENDPOINT")) { + rewritten[key] = value.replace(/127\.0\.0\.1|localhost/, "host.docker.internal"); + } + } + return rewritten; +} + +function json(status: number, body: unknown): HttpResponse { + return { status, headers: { "Content-Type": "application/json" }, body: JSON.stringify(body) }; +} diff --git a/src/handlers/project/dev/index.test.ts b/src/handlers/project/dev/index.test.ts index e345d0eb8..44ca9a302 100644 --- a/src/handlers/project/dev/index.test.ts +++ b/src/handlers/project/dev/index.test.ts @@ -1,4 +1,5 @@ import { describe, expect, test } from "bun:test"; +import { join } from "node:path"; import type { ProjectRuntime } from "../../../projectSchemas/runtime"; import { InputValidationError, @@ -13,7 +14,7 @@ import { JsonKey, RegionKey } from "../../keys"; import type { Project } from "../types"; import { createDevProjectHandler, type DevProjectHandlerConfig } from "."; import type { DevEnvironmentInput } from "./environment"; -import type { DevEvent, DevRunner, DevServerInput } from "./types"; +import type { DevEvent, DevRunner, DevServerInput, DevTraceCollector } from "./types"; function runtime(name = "orders", build: ProjectRuntime["build"] = "CodeZip"): ProjectRuntime { return { @@ -44,6 +45,27 @@ function captureRunner(events: DevEvent[] = []) { return { runner, inputs }; } +function fakeCollector() { + const starts: { tracesDirectory: string; signal?: AbortSignal }[] = []; + const state = { closed: 0 }; + const collector: DevTraceCollector = { + port: 43180, + envVars: { + OTEL_EXPORTER_OTLP_ENDPOINT: "http://127.0.0.1:43180", + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "http://127.0.0.1:43180/v1/traces", + OTEL_EXPORTER_OTLP_PROTOCOL: "http/protobuf", + }, + close: async () => { + state.closed++; + }, + }; + const start: DevProjectHandlerConfig["startTraceCollector"] = async (options) => { + starts.push(options); + return collector; + }; + return { start, starts, state }; +} + type HarnessOptions = { project?: Project; codeZip?: ReturnType; @@ -57,6 +79,7 @@ function harness(options: HarnessOptions = {}) { const io = testIO(); const codeZip = options.codeZip ?? captureRunner(); const container = options.container ?? captureRunner(); + const collector = fakeCollector(); const environmentInputs: DevEnvironmentInput[] = []; const handler = createDevProjectHandler({ io: io.io, @@ -68,6 +91,7 @@ function harness(options: HarnessOptions = {}) { return { env: { FROM_LOADER: "yes" } }; }), checkPort: options.checkPort ?? (async () => true), + startTraceCollector: collector.start, }); const ctx = ValueContext.EmptyContext() .withValue(ProjectKey, options.project ?? project(runtime())) @@ -81,9 +105,11 @@ function harness(options: HarnessOptions = {}) { return { codeZip, container, + collector, environmentInputs, io, - run: (flags: { agent?: string; port?: number } = {}) => handler.handle(ctx, flags, {}), + run: (flags: { agent?: string; port?: number; traces?: boolean } = {}) => + handler.handle(ctx, { traces: true, ...flags }, {}), }; } @@ -128,7 +154,12 @@ describe("project dev selection and dispatch", () => { expect(subject.container.inputs[0]).toMatchObject({ projectRoot: "/workspace/project", port: 4567, - env: { FROM_LOADER: "yes" }, + env: { + FROM_LOADER: "yes", + OTEL_EXPORTER_OTLP_ENDPOINT: "http://host.docker.internal:43180", + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "http://host.docker.internal:43180/v1/traces", + OTEL_SERVICE_NAME: "support", + }, runtime: { name: "support", build: "Container" }, }); }); @@ -145,7 +176,56 @@ describe("project dev selection and dispatch", () => { expect(checked).toEqual([8080, 8081]); expect(subject.codeZip.inputs[0]?.port).toBe(8081); - expect(subject.io.stderr()).toBe("Port 8080 is in use; using 8081."); + expect(subject.io.stderr()).toContain("Port 8080 is in use; using 8081."); + }); +}); + +describe("project dev trace collection", () => { + test("starts the collector, announces it, and points a CodeZip agent at loopback", async () => { + const subject = harness(); + await subject.run(); + + expect(subject.collector.starts).toEqual([ + { + tracesDirectory: join("/workspace/project", "agentcore", ".cli", "traces", "otlp"), + signal: expect.any(AbortSignal), + }, + ]); + expect(subject.io.stderr()).toContain("OTEL collector listening on port 43180"); + expect(subject.codeZip.inputs[0]?.env).toMatchObject({ + OTEL_EXPORTER_OTLP_ENDPOINT: "http://127.0.0.1:43180", + OTEL_SERVICE_NAME: "orders", + }); + expect(subject.collector.state.closed).toBe(1); + }); + + test("--no-traces skips the collector entirely", async () => { + const subject = harness(); + await subject.run({ traces: false }); + + expect(subject.collector.starts).toHaveLength(0); + expect(subject.codeZip.inputs[0]?.env).toEqual({ FROM_LOADER: "yes" }); + }); + + test("a runtime with instrumentation disabled skips the collector", async () => { + const disabled = { ...runtime(), instrumentation: { enableOtel: false } } as ProjectRuntime; + const subject = harness({ project: project(disabled) }); + await subject.run(); + + expect(subject.collector.starts).toHaveLength(0); + expect(subject.codeZip.inputs[0]?.env).toEqual({ FROM_LOADER: "yes" }); + }); + + test("the collector is closed when the runner fails", async () => { + const codeZip = captureRunner(); + codeZip.runner.run = async function* () { + yield* []; + throw new InputValidationError("runner failed"); + }; + const subject = harness({ codeZip }); + + await expect(subject.run()).rejects.toThrow("runner failed"); + expect(subject.collector.state.closed).toBe(1); }); }); @@ -158,7 +238,7 @@ test("project dev renders human and NDJSON output", async () => { for (const json of [false, true]) { const subject = harness({ codeZip: captureRunner(events), json }); - await subject.run(); + await subject.run({ traces: false }); expect(subject.io.stdout()).toBe( json ? events.map((event) => JSON.stringify(event)).join("\n") : "agent output", ); @@ -199,7 +279,9 @@ describe("project dev interruption", () => { expect(input.signal.reason).toBeInstanceOf(UserCancellationError); await expect(pending).rejects.toBe(input.signal.reason); expect((input.signal.reason as UserCancellationError).exitCode).toBe(130); - expect(subject.io.stderr()).toBe("Shutting down…"); + // Traces are on by default, so the collector's "listening" line precedes this. + expect(subject.io.stderr()).toContain("Shutting down…"); + expect(subject.collector.state.closed).toBe(1); expect(process.listenerCount(signal)).toBe(before); }, ); diff --git a/src/handlers/project/dev/index.ts b/src/handlers/project/dev/index.ts index 2492db672..c1c237b7c 100644 --- a/src/handlers/project/dev/index.ts +++ b/src/handlers/project/dev/index.ts @@ -1,4 +1,6 @@ +import { join } from "node:path"; import z from "zod"; +import { rewriteOtelEndpointForContainer } from "../../../core/dev/otel/collector"; import { resolveDevPort } from "../../../core/dev/port"; import type { ProjectRuntime } from "../../../projectSchemas/runtime"; import { @@ -12,15 +14,25 @@ import { JsonRendererKey, type JsonRenderer } from "../../../tui"; import { JsonKey, RegionKey } from "../../keys"; import type { Project } from "../types"; import type { DevEnvironmentLoader } from "./environment"; -import type { DevEvent, DevRunner } from "./types"; +import type { DevEvent, DevRunner, DevTraceCollector, DevTraceCollectorStarter } from "./types"; export type DevProjectHandlerConfig = { io: AppIO; runners: { CodeZip: DevRunner; Container: DevRunner }; loadDevEnvironment: DevEnvironmentLoader; checkPort: PortChecker; + startTraceCollector: DevTraceCollectorStarter; }; +/** Env for a spawned agent so its OTEL SDK reports to the collector as this runtime. */ +function otelEnvForRuntime( + collector: DevTraceCollector, + runtime: ProjectRuntime, +): Record { + const env = { ...collector.envVars, OTEL_SERVICE_NAME: runtime.name }; + return runtime.build === "Container" ? rewriteOtelEndpointForContainer(env) : env; +} + function selectRuntime(project: Project, name?: string): ProjectRuntime { if (project.spec.runtimes.length === 0) { throw new InputValidationError( @@ -64,6 +76,7 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => "port for the development server", z.coerce.number().int().min(1).max(65535).optional(), ), + flag("traces", "disable local OTEL trace collection", z.boolean().default(true)), ], handle: async (ctx, flags) => { const controller = new AbortController(); @@ -76,6 +89,7 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => const signals = ["SIGINT", "SIGTERM"] as const; for (const signal of signals) process.on(signal, interrupt); + let collector: DevTraceCollector | undefined; try { const project = ctx.require(ProjectKey); const runtime = selectRuntime(project, flags.agent); @@ -103,12 +117,31 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => }); controller.signal.throwIfAborted(); + let otelEnv: Record = {}; + if (flags.traces && (runtime.instrumentation?.enableOtel ?? true)) { + const tracesDirectory = join(project.rootPath, "agentcore", ".cli", "traces", "otlp"); + collector = await config.startTraceCollector({ + tracesDirectory, + signal: controller.signal, + }); + otelEnv = otelEnvForRuntime(collector, runtime); + renderEvent( + config.io, + { + type: "status", + message: `OTEL collector listening on port ${collector.port}; traces persist to ${tracesDirectory}.`, + }, + json, + ); + } + controller.signal.throwIfAborted(); + const runner = config.runners[runtime.build]; for await (const event of runner.run({ runtime, projectRoot: project.rootPath, port: devPort.port, - env, + env: { ...env, ...otelEnv }, signal: controller.signal, })) { renderEvent(config.io, event, json); @@ -118,6 +151,7 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => throw error; } finally { for (const signal of signals) process.removeListener(signal, interrupt); + await collector?.close(); } }, }); diff --git a/src/handlers/project/dev/types.ts b/src/handlers/project/dev/types.ts index 933c0c452..289c1b5c9 100644 --- a/src/handlers/project/dev/types.ts +++ b/src/handlers/project/dev/types.ts @@ -16,3 +16,16 @@ export type DevServerInput = { export interface DevRunner { run(input: DevServerInput): AsyncGenerator; } + +/** A local OTLP receiver that spawned agents export traces to. */ +export interface DevTraceCollector { + port: number; + /** Environment variables that point an agent's OTEL SDK at the receiver. */ + envVars: Record; + close(): Promise; +} + +export type DevTraceCollectorStarter = (options: { + tracesDirectory: string; + signal?: AbortSignal; +}) => Promise; diff --git a/src/handlers/project/index.ts b/src/handlers/project/index.ts index 232879425..2ac86e40e 100644 --- a/src/handlers/project/index.ts +++ b/src/handlers/project/index.ts @@ -2,6 +2,7 @@ import { Router } from "../../router"; import { checkPort, type AppIO } from "../../io"; import { CodeZipDevRunner } from "../../core/dev/codezip"; import { ContainerDevRunner } from "../../core/dev/container"; +import { startOtelCollector } from "../../core/dev/otel/collector"; import { withProject } from "../../middleware"; import { createCreateProjectHandler } from "./create"; import { createRemoveProjectHandler } from "./remove"; @@ -40,6 +41,7 @@ export function createProjectHandler(config: ProjectHandlerConfig): Router { }, loadDevEnvironment, checkPort, + startTraceCollector: startOtelCollector, }), ), ); diff --git a/src/io/httpServer.test.ts b/src/io/httpServer.test.ts new file mode 100644 index 000000000..a96b23fb0 --- /dev/null +++ b/src/io/httpServer.test.ts @@ -0,0 +1,73 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { type HttpServerHandle, startHttpServer } from "./httpServer"; + +let handle: HttpServerHandle | undefined; + +afterEach(async () => { + await handle?.close(); + handle = undefined; +}); + +describe("startHttpServer", () => { + test("serves requests on an OS-assigned loopback port", async () => { + handle = await startHttpServer((request) => ({ + status: 200, + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + method: request.method, + url: request.url, + body: request.body.toString(), + }), + })); + + expect(handle.port).toBeGreaterThan(0); + const response = await fetch(`http://127.0.0.1:${handle.port}/v1/traces`, { + method: "POST", + body: "ping", + }); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ method: "POST", url: "/v1/traces", body: "ping" }); + }); + + test("handler errors become 500s without crashing the server", async () => { + handle = await startHttpServer(() => { + throw new Error("boom"); + }); + + const response = await fetch(`http://127.0.0.1:${handle.port}/`); + expect(response.status).toBe(500); + + const again = await fetch(`http://127.0.0.1:${handle.port}/`); + expect(again.status).toBe(500); + }); + + test("oversized bodies get a 413 response, not a connection reset", async () => { + handle = await startHttpServer(() => ({ status: 200 })); + + const response = await fetch(`http://127.0.0.1:${handle.port}/v1/traces`, { + method: "POST", + body: Buffer.alloc(51 * 1024 * 1024), + }); + expect(response.status).toBe(413); + }); + + test("aborting the signal closes the server", async () => { + const controller = new AbortController(); + const server = await startHttpServer(() => ({ status: 200 }), { signal: controller.signal }); + + controller.abort(); + await Bun.sleep(20); + expect(fetch(`http://127.0.0.1:${server.port}/`)).rejects.toThrow(); + }); + + test("close is idempotent", async () => { + const server = await startHttpServer(() => ({ status: 200 })); + await server.close(); + await server.close(); + }); + + test("listen failure rejects instead of hanging", async () => { + handle = await startHttpServer(() => ({ status: 200 })); + expect(startHttpServer(() => ({ status: 200 }), { port: handle.port })).rejects.toThrow(); + }); +}); diff --git a/src/io/httpServer.ts b/src/io/httpServer.ts new file mode 100644 index 000000000..9a26757fe --- /dev/null +++ b/src/io/httpServer.ts @@ -0,0 +1,119 @@ +// Uses node:http rather than Bun.serve because the npm bundle targets Node, +// where Bun APIs are absent (same constraint as exec.ts). +import { + type IncomingHttpHeaders, + type IncomingMessage, + type Server, + type ServerResponse, + createServer, +} from "node:http"; + +/** Cap request bodies so a runaway local client cannot exhaust memory. */ +const MAX_BODY_BYTES = 50 * 1024 * 1024; + +export interface HttpRequest { + method: string; + url: string; + headers: IncomingHttpHeaders; + body: Buffer; +} + +export interface HttpResponse { + status: number; + headers?: Record; + body?: string | Buffer; +} + +export type HttpRequestHandler = (request: HttpRequest) => HttpResponse | Promise; + +export interface HttpServerHandle { + /** The port the server is listening on. */ + port: number; + /** Stops accepting connections and closes active ones. Idempotent. */ + close(): Promise; +} + +/** + * Starts a loopback-only HTTP server for local dev tooling. Binds 127.0.0.1 on + * the given port (0 lets the OS assign one). Handler errors become plain 500s; + * oversized bodies become 413s. Aborting the signal closes the server. + */ +export async function startHttpServer( + handler: HttpRequestHandler, + options: { port?: number; signal?: AbortSignal } = {}, +): Promise { + const server = createServer((request, response) => { + void respond(handler, request, response); + }); + + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(options.port ?? 0, "127.0.0.1", resolve); + }); + + const address = server.address(); + const port = typeof address === "object" && address !== null ? address.port : 0; + + const close = () => closeServer(server); + options.signal?.addEventListener("abort", () => void close(), { once: true }); + + return { port, close }; +} + +async function respond( + handler: HttpRequestHandler, + request: IncomingMessage, + response: ServerResponse, +): Promise { + let body: Buffer; + try { + body = await readBody(request); + } catch (error) { + // Answer before closing: destroying the socket first would surface as a + // connection reset, which OTLP exporters treat as transient and retry. + const status = error instanceof BodyTooLargeError ? 413 : 400; + response.writeHead(status, { Connection: "close" }).end(() => request.destroy()); + return; + } + + try { + const result = await handler({ + method: request.method ?? "GET", + url: request.url ?? "/", + headers: request.headers, + body, + }); + response.writeHead(result.status, result.headers); + response.end(result.body); + } catch { + response.writeHead(500, { "Content-Type": "application/json" }); + response.end(JSON.stringify({ error: "internal error" })); + } +} + +class BodyTooLargeError extends Error {} + +function readBody(request: IncomingMessage): Promise { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + let size = 0; + request.on("data", (chunk: Buffer) => { + size += chunk.length; + if (size > MAX_BODY_BYTES) { + request.pause(); + reject(new BodyTooLargeError()); + return; + } + chunks.push(chunk); + }); + request.on("end", () => resolve(Buffer.concat(chunks))); + request.on("error", reject); + }); +} + +function closeServer(server: Server): Promise { + return new Promise((resolve) => { + server.close(() => resolve()); + server.closeAllConnections(); + }); +} diff --git a/src/io/index.ts b/src/io/index.ts index fb0da0b0e..23eb4a03d 100644 --- a/src/io/index.ts +++ b/src/io/index.ts @@ -37,3 +37,10 @@ export { export type { AppIO, ReadWriteJson } from "./types"; export { warn } from "./warn"; export { checkPort, type PortChecker } from "./port"; +export { + startHttpServer, + type HttpRequest, + type HttpRequestHandler, + type HttpResponse, + type HttpServerHandle, +} from "./httpServer"; diff --git a/src/router/flags.tsx b/src/router/flags.tsx index 2b690d094..5b26a8978 100644 --- a/src/router/flags.tsx +++ b/src/router/flags.tsx @@ -5,15 +5,17 @@ import type { Flag, GlobalFlag } from "./handler"; import { coerce, formatZodError, inspect } from "./schema"; // toOption builds a Commander Option from a flag's schema. Booleans become value-less -// toggles; everything else takes a value (`` / variadic ``). A -// required, non-boolean flag is made mandatory; defaults are forwarded. +// toggles — declared as `--no-` when they default to true, so the flag turns +// the behavior off (Commander's negation stores the value under the positive name). +// Everything else takes a value (`` / variadic ``). A required, +// non-boolean flag is made mandatory; defaults are forwarded. export function toOption(flag: Flag): Option { const info = inspect(flag.schema); const long = `--${flag.name}`; let token: string; if (info.boolean) { - token = long; + token = info.hasDefault && info.defaultValue === true ? `--no-${flag.name}` : long; } else if (info.variadic) { token = `${long} <${flag.name}...>`; } else { diff --git a/src/router/router.test.ts b/src/router/router.test.ts index 5b17f96d9..dba80053b 100644 --- a/src/router/router.test.ts +++ b/src/router/router.test.ts @@ -284,6 +284,27 @@ test("boolean flags default to false when omitted", async () => { expect(seen).toEqual({ verbose: false }); }); +test("a boolean flag defaulting to true is declared as its --no- negation", async () => { + const seen: { traces: boolean }[] = []; + + const run = createHandler({ + name: "run", + description: "", + flags: [flag("traces", "collect traces", z.boolean().default(true))], + handle: async (_ctx, flags) => { + seen.push(flags); + }, + }); + + const root = new Router("app"); + root.handler(run); + + await root.route(["node", "app", "run"]); + await root.route(["node", "app", "run", "--no-traces"]); + + expect(seen).toEqual([{ traces: true }, { traces: false }]); +}); + test("applies a schema default for an omitted flag", async () => { let seen: { count: number } | undefined; From 9b90471105cca4051d6e5d6cad629af52c4f4953 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Fri, 21 Aug 2026 11:32:05 -0400 Subject: [PATCH 2/5] fix(dev): ack OTLP exports and report trace-persistence failures A batch that can't be persisted (disk full, permissions) was being turned into a 500, which the OTEL SDK exporter retries forever while the user sees nothing. Ack the export (200) so retries stop, and surface the failure once via an onError sink threaded from the collector to the dev handler, which owns the IO to warn the user. Addresses Gitika's review on store.ts:45 (catch in one place; don't let persistence faults read as a silent, retried loss). --- src/core/dev/otel/collector.test.ts | 27 +++++++++++++++++++++++++- src/core/dev/otel/collector.ts | 24 ++++++++++++++++++----- src/handlers/project/dev/index.test.ts | 17 +++++++++++++++- src/handlers/project/dev/index.ts | 16 +++++++++++++++ src/handlers/project/dev/types.ts | 2 ++ 5 files changed, 79 insertions(+), 7 deletions(-) diff --git a/src/core/dev/otel/collector.test.ts b/src/core/dev/otel/collector.test.ts index eb4fc16e7..a3785cf48 100644 --- a/src/core/dev/otel/collector.test.ts +++ b/src/core/dev/otel/collector.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { @@ -138,6 +138,31 @@ describe("startOtelCollector", () => { expect(await collector.store.list()).toEqual([]); }); + test("acks with 200 and reports onError when persistence fails", async () => { + // A traces dir nested under a regular file makes mkdir (and thus append) fail. + const blocker = join(directory, "blocker"); + await writeFile(blocker, "x"); + const errors: unknown[] = []; + const failing = await startOtelCollector({ + tracesDirectory: join(blocker, "otlp"), + onError: (error) => errors.push(error), + }); + try { + const response = await fetch(`http://127.0.0.1:${failing.port}/v1/traces`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + resourceSpans: [{ scopeSpans: [{ spans: [{ traceId: TRACE_ID_HEX, name: "x" }] }] }], + }), + }); + // Exporter must see success so it stops retrying; the failure is reported instead. + expect(response.status).toBe(200); + expect(errors).toHaveLength(1); + } finally { + await failing.close(); + } + }); + test("health check responds ok and unknown routes 404", async () => { const health = await fetch(`http://127.0.0.1:${collector.port}/`); expect(await health.json()).toEqual({ status: "ok" }); diff --git a/src/core/dev/otel/collector.ts b/src/core/dev/otel/collector.ts index c19a3375a..8e725ae80 100644 --- a/src/core/dev/otel/collector.ts +++ b/src/core/dev/otel/collector.ts @@ -48,6 +48,8 @@ export interface StartOtelCollectorOptions { tracesDirectory: string; /** Closes the collector when aborted. */ signal?: AbortSignal; + /** Called when a batch can't be persisted; the export is still acked to stop retries. */ + onError?: (error: unknown) => void; } /** @@ -59,19 +61,23 @@ export async function startOtelCollector( options: StartOtelCollectorOptions, ): Promise { const store = new TraceStore(options.tracesDirectory); - const server = await startHttpServer((request) => route(request, store), { + const server = await startHttpServer((request) => route(request, store, options.onError), { signal: options.signal, }); return { port: server.port, store, envVars: otelEnvVars(server.port), close: server.close }; } -async function route(request: HttpRequest, store: TraceStore): Promise { +async function route( + request: HttpRequest, + store: TraceStore, + onError?: (error: unknown) => void, +): Promise { if (request.method === "POST" && request.url === "/v1/traces") { - return ingest(request, store, ExportTraceServiceRequest); + return ingest(request, store, ExportTraceServiceRequest, onError); } if (request.method === "POST" && request.url === "/v1/logs") { - return ingest(request, store, ExportLogsServiceRequest); + return ingest(request, store, ExportLogsServiceRequest, onError); } if (request.method === "GET" && request.url === "/") { return json(200, { status: "ok" }); @@ -83,6 +89,7 @@ async function ingest( request: HttpRequest, store: TraceStore, decoder: OtlpDecoder, + onError?: (error: unknown) => void, ): Promise { let payload: OtlpPayload; try { @@ -90,7 +97,14 @@ async function ingest( } catch { return json(400, { error: "Invalid OTLP payload" }); } - await store.append(payload); + try { + await store.append(payload); + } catch (error) { + // A persistence failure (disk full, permissions) is the collector's problem, + // not the agent's: ack the export so the SDK exporter stops retrying the batch + // forever, and report it once so the user isn't left with silently missing traces. + onError?.(error); + } return json(200, {}); } diff --git a/src/handlers/project/dev/index.test.ts b/src/handlers/project/dev/index.test.ts index 44ca9a302..cde3b11a3 100644 --- a/src/handlers/project/dev/index.test.ts +++ b/src/handlers/project/dev/index.test.ts @@ -46,7 +46,7 @@ function captureRunner(events: DevEvent[] = []) { } function fakeCollector() { - const starts: { tracesDirectory: string; signal?: AbortSignal }[] = []; + const starts: Parameters[0][] = []; const state = { closed: 0 }; const collector: DevTraceCollector = { port: 43180, @@ -189,6 +189,7 @@ describe("project dev trace collection", () => { { tracesDirectory: join("/workspace/project", "agentcore", ".cli", "traces", "otlp"), signal: expect.any(AbortSignal), + onError: expect.any(Function), }, ]); expect(subject.io.stderr()).toContain("OTEL collector listening on port 43180"); @@ -199,6 +200,20 @@ describe("project dev trace collection", () => { expect(subject.collector.state.closed).toBe(1); }); + test("reports a trace-persistence failure once, not per failed export", async () => { + const subject = harness(); + await subject.run(); + + const onError = subject.collector.starts[0]?.onError; + onError?.(new Error("disk full")); + onError?.(new Error("disk full")); + + const stderr = subject.io.stderr(); + expect(stderr).toContain("failed to persist traces"); + expect(stderr).toContain("disk full"); + expect(stderr.match(/failed to persist traces/g)).toHaveLength(1); + }); + test("--no-traces skips the collector entirely", async () => { const subject = harness(); await subject.run({ traces: false }); diff --git a/src/handlers/project/dev/index.ts b/src/handlers/project/dev/index.ts index c1c237b7c..9cc3ab731 100644 --- a/src/handlers/project/dev/index.ts +++ b/src/handlers/project/dev/index.ts @@ -120,9 +120,25 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => let otelEnv: Record = {}; if (flags.traces && (runtime.instrumentation?.enableOtel ?? true)) { const tracesDirectory = join(project.rootPath, "agentcore", ".cli", "traces", "otlp"); + let tracePersistErrorReported = false; collector = await config.startTraceCollector({ tracesDirectory, signal: controller.signal, + // 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); + renderEvent( + config.io, + { + type: "status", + message: `Warning: failed to persist traces to ${tracesDirectory} (${detail}); collected traces may be incomplete.`, + }, + json, + ); + }, }); otelEnv = otelEnvForRuntime(collector, runtime); renderEvent( diff --git a/src/handlers/project/dev/types.ts b/src/handlers/project/dev/types.ts index 289c1b5c9..ccbdcc886 100644 --- a/src/handlers/project/dev/types.ts +++ b/src/handlers/project/dev/types.ts @@ -28,4 +28,6 @@ export interface DevTraceCollector { export type DevTraceCollectorStarter = (options: { tracesDirectory: string; signal?: AbortSignal; + /** Reports a trace-persistence failure (the export is still acked to stop retries). */ + onError?: (error: unknown) => void; }) => Promise; From b34bd04b5631806b270e1a96f45d5fb553458627 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Fri, 21 Aug 2026 13:57:50 -0400 Subject: [PATCH 3/5] test(dev): cover spanCount noise-filtering; clarify hex-id helper name Follow-ups from #2043 review (Gitika, Harrison): - Add a TraceStore.list spanCount test with transport-noise spans (1 agent + 4 http-send -> "1"), guarding the post-filter count. - Rename hexFromB64OrString -> hexFromBase64OrHex; both reviewers misread B64. --- src/core/dev/otel/store.test.ts | 10 ++++++++++ src/core/dev/otel/transforms.test.ts | 10 +++++----- src/core/dev/otel/transforms.ts | 20 ++++++++++---------- 3 files changed, 25 insertions(+), 15 deletions(-) diff --git a/src/core/dev/otel/store.test.ts b/src/core/dev/otel/store.test.ts index 53d04cc2a..8017f2bd0 100644 --- a/src/core/dev/otel/store.test.ts +++ b/src/core/dev/otel/store.test.ts @@ -70,6 +70,16 @@ describe("TraceStore", () => { expect(traces[0]!.spanCount).toBe("2"); }); + test("spanCount reflects the rendered waterfall, not filtered transport noise", async () => { + const trace = payload(TRACE_A); + const spans = trace.resourceSpans![0]!.scopeSpans![0]!.spans!; + // 1 meaningful agent span + 4 "http send" spans the inspector filters out. + for (let i = 0; i < 4; i++) spans.push({ ...spans[0]!, name: "GET / http send" }); + await store.append(trace); + + expect((await store.list())[0]!.spanCount).toBe("1"); + }); + test("payloads without a trace id are dropped", async () => { await store.append({ resourceSpans: [] }); expect(await store.list()).toEqual([]); diff --git a/src/core/dev/otel/transforms.test.ts b/src/core/dev/otel/transforms.test.ts index 180391284..caad88a36 100644 --- a/src/core/dev/otel/transforms.test.ts +++ b/src/core/dev/otel/transforms.test.ts @@ -4,7 +4,7 @@ import { extractAnyValue, extractTraceMeta, flattenAttributes, - hexFromB64OrString, + hexFromBase64OrHex, nanoToMs, partitionByTraceId, } from "./transforms"; @@ -199,10 +199,10 @@ describe("helpers", () => { expect(nanoToMs(undefined)).toBe(0); }); - test("hexFromB64OrString accepts hex, base64, and empty", () => { - expect(hexFromB64OrString(TRACE_ID_HEX.toUpperCase())).toBe(TRACE_ID_HEX); - expect(hexFromB64OrString(TRACE_ID_B64)).toBe(TRACE_ID_HEX); - expect(hexFromB64OrString(undefined)).toBe(""); + test("hexFromBase64OrHex accepts hex, base64, and empty", () => { + expect(hexFromBase64OrHex(TRACE_ID_HEX.toUpperCase())).toBe(TRACE_ID_HEX); + expect(hexFromBase64OrHex(TRACE_ID_B64)).toBe(TRACE_ID_HEX); + expect(hexFromBase64OrHex(undefined)).toBe(""); }); test("flattenAttributes handles typed values, arrays, and kvlist, empty for none", () => { diff --git a/src/core/dev/otel/transforms.ts b/src/core/dev/otel/transforms.ts index cef0f5ca1..0369c8bca 100644 --- a/src/core/dev/otel/transforms.ts +++ b/src/core/dev/otel/transforms.ts @@ -28,7 +28,7 @@ export function extractTraceMeta( if (service) services.add(service); for (const scopeSpan of resourceSpan.scopeSpans ?? []) { for (const span of scopeSpan.spans ?? []) { - meta.traceId ??= hexFromB64OrString(span.traceId) || undefined; + meta.traceId ??= hexFromBase64OrHex(span.traceId) || undefined; widenTimeBounds(meta, nanoToMs(span.startTimeUnixNano)); widenTimeBounds(meta, nanoToMs(span.endTimeUnixNano)); meta.sessionId ??= @@ -43,7 +43,7 @@ export function extractTraceMeta( if (service) services.add(service); for (const scopeLog of resourceLog.scopeLogs ?? []) { for (const record of scopeLog.logRecords ?? []) { - meta.traceId ??= hexFromB64OrString(record.traceId) || undefined; + meta.traceId ??= hexFromBase64OrHex(record.traceId) || undefined; widenTimeBounds( meta, nanoToMs(record.timeUnixNano) || nanoToMs(record.observedTimeUnixNano), @@ -79,7 +79,7 @@ export function partitionByTraceId(payload: OtlpPayload): Map hexFromB64OrString(span.traceId)); + const byTrace = groupBy(scopeSpan.spans ?? [], (span) => hexFromBase64OrHex(span.traceId)); for (const [traceId, spans] of byTrace) { (partition(traceId).resourceSpans ??= []).push({ resource: resourceSpan.resource, @@ -92,7 +92,7 @@ export function partitionByTraceId(payload: OtlpPayload): Map - hexFromB64OrString(record.traceId), + hexFromBase64OrHex(record.traceId), ); for (const [traceId, logRecords] of byTrace) { (partition(traceId).resourceLogs ??= []).push({ @@ -137,9 +137,9 @@ export function buildTraceDetail( spans: scopeSpan.spans ?.map((span) => ({ ...span, - traceId: hexFromB64OrString(span.traceId), - spanId: hexFromB64OrString(span.spanId), - parentSpanId: hexFromB64OrString(span.parentSpanId), + traceId: hexFromBase64OrHex(span.traceId), + spanId: hexFromBase64OrHex(span.spanId), + parentSpanId: hexFromBase64OrHex(span.parentSpanId), attributes: flattenAttributes(span.attributes), })) .filter((span) => isMeaningfulSpan(span)), @@ -157,8 +157,8 @@ export function buildTraceDetail( scope: scopeLog.scope, logRecords: scopeLog.logRecords?.map((record) => ({ ...record, - traceId: hexFromB64OrString(record.traceId), - spanId: hexFromB64OrString(record.spanId), + traceId: hexFromBase64OrHex(record.traceId), + spanId: hexFromBase64OrHex(record.spanId), body: record.body === undefined ? undefined : extractAnyValue(record.body), attributes: flattenAttributes(record.attributes), })), @@ -224,7 +224,7 @@ export function nanoToMs(nano: string | undefined): number { * Normalize a trace/span id that may be base64 (protobuf JSON conversion) or * already hex (JSON ingest) into lowercase hex. */ -export function hexFromB64OrString(value: string | undefined): string { +export function hexFromBase64OrHex(value: string | undefined): string { if (!value) return ""; if (/^[0-9a-f]+$/i.test(value) && (value.length === 32 || value.length === 16)) return value.toLowerCase(); From 8cabff76d67317314a1ed01e32b44843ddf37908 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Fri, 21 Aug 2026 14:14:22 -0400 Subject: [PATCH 4/5] docs(dev): clarify collector/httpServer/flag comments per review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Harrison review on #1980: - httpServer: generalize the answer-before-close comment (drop OTLP specificity) — the module is a shared io primitive (the Inspector server reuses it). - flags: explain why a default-true boolean is exposed as --no-. - collector: reword the onError comment to state the collector's guarantee (ack + hand to onError) rather than the caller's report-once behavior; drop the volatile "matches the reference CLI" aside. --- src/core/dev/otel/collector.ts | 6 +++--- src/io/httpServer.ts | 4 ++-- src/router/flags.tsx | 9 +++++---- 3 files changed, 10 insertions(+), 9 deletions(-) diff --git a/src/core/dev/otel/collector.ts b/src/core/dev/otel/collector.ts index 8e725ae80..addb760da 100644 --- a/src/core/dev/otel/collector.ts +++ b/src/core/dev/otel/collector.ts @@ -101,8 +101,8 @@ async function ingest( await store.append(payload); } catch (error) { // A persistence failure (disk full, permissions) is the collector's problem, - // not the agent's: ack the export so the SDK exporter stops retrying the batch - // forever, and report it once so the user isn't left with silently missing traces. + // not the agent's: ack the export anyway so the SDK exporter stops retrying the + // batch forever, and hand the error to onError for the caller to surface. onError?.(error); } return json(200, {}); @@ -147,7 +147,7 @@ export function otelEnvVars(port: number): Record { * Rewrite loopback OTLP endpoints so a containerized agent can reach the * collector on the host. host.docker.internal resolves on Docker Desktop, * Finch, and Podman; bare-metal Linux Docker would additionally need - * `--add-host=host.docker.internal:host-gateway` (matches the reference CLI). + * `--add-host=host.docker.internal:host-gateway`. */ export function rewriteOtelEndpointForContainer( env: Record, diff --git a/src/io/httpServer.ts b/src/io/httpServer.ts index 9a26757fe..d9df97214 100644 --- a/src/io/httpServer.ts +++ b/src/io/httpServer.ts @@ -69,8 +69,8 @@ async function respond( try { body = await readBody(request); } catch (error) { - // Answer before closing: destroying the socket first would surface as a - // connection reset, which OTLP exporters treat as transient and retry. + // Answer before closing: destroying the socket first surfaces as a connection + // reset, which many clients treat as transient and silently retry. const status = error instanceof BodyTooLargeError ? 413 : 400; response.writeHead(status, { Connection: "close" }).end(() => request.destroy()); return; diff --git a/src/router/flags.tsx b/src/router/flags.tsx index 5b26a8978..6314cf223 100644 --- a/src/router/flags.tsx +++ b/src/router/flags.tsx @@ -4,10 +4,11 @@ import type { Context } from "./context"; import type { Flag, GlobalFlag } from "./handler"; import { coerce, formatZodError, inspect } from "./schema"; -// toOption builds a Commander Option from a flag's schema. Booleans become value-less -// toggles — declared as `--no-` when they default to true, so the flag turns -// the behavior off (Commander's negation stores the value under the positive name). -// Everything else takes a value (`` / variadic ``). A required, +// toOption builds a Commander Option from a flag's schema. A boolean that defaults +// to true is exposed as `--no-`: the behavior is already on, so the only useful +// action is turning it off, which Commander stores under the positive name (e.g. +// `--no-traces` sets `traces=false`). A boolean that defaults off stays `--`. +// Everything else takes a value (`` / variadic ``); a required // non-boolean flag is made mandatory; defaults are forwarded. export function toOption(flag: Flag): Option { const info = inspect(flag.schema); From 10be0a69341306c70204ae8f0cfa6f46e83eb87f Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Fri, 21 Aug 2026 14:50:19 -0400 Subject: [PATCH 5/5] fix(dev): harden OTEL collector, container reachability, and dev lifecycle Address reviewer findings on the collector and dev wiring: - Validate top-level OTLP shape and return 400 instead of mislabeling a bad payload as a persistence error. - Guard the shared HTTP server against a client that disconnects mid-response so it can no longer crash project dev; add an optional bind host. - Bind the collector to 0.0.0.0 for container runtimes so a container can reach it over the host bridge on Linux. - Run the container template under opentelemetry-instrument so it emits traces. - Keep the collector alive through the child's shutdown grace so final spans are not lost. - Force the OTEL settings that would otherwise let shell or .env.local values disable or break local collection. - Make the uv sitecustomize discovery abortable and read its path from a marker rather than the last merged output line. --- .../hello-world-python-container/Dockerfile | 4 +- src/core/dev/codezip.test.ts | 47 +++++++---- src/core/dev/codezip.ts | 24 ++++-- src/core/dev/otel/collector.test.ts | 34 ++++++++ src/core/dev/otel/collector.ts | 77 +++++++++++++------ src/handlers/project/dev/index.test.ts | 11 ++- src/handlers/project/dev/index.ts | 6 +- src/handlers/project/dev/types.ts | 3 +- src/io/exec.test.ts | 19 +++++ src/io/exec.ts | 16 +++- src/io/httpServer.test.ts | 20 +++++ src/io/httpServer.ts | 21 +++-- 12 files changed, 227 insertions(+), 55 deletions(-) diff --git a/src/assets/templates/hello-world-python-container/Dockerfile b/src/assets/templates/hello-world-python-container/Dockerfile index c56f8d35b..08a64f467 100644 --- a/src/assets/templates/hello-world-python-container/Dockerfile +++ b/src/assets/templates/hello-world-python-container/Dockerfile @@ -34,4 +34,6 @@ USER bedrock_agentcore # 9000: A2A Mode EXPOSE 8080 8000 9000 -CMD ["python", "-m", "main"] +# opentelemetry-instrument (from aws-opentelemetry-distro) starts a real +# TracerProvider; plain `python -m main` would export nothing. +CMD ["opentelemetry-instrument", "python", "-m", "main"] diff --git a/src/core/dev/codezip.test.ts b/src/core/dev/codezip.test.ts index da366d59d..4b5af17d8 100644 --- a/src/core/dev/codezip.test.ts +++ b/src/core/dev/codezip.test.ts @@ -54,21 +54,25 @@ async function projectRoot(withNodeModules = false): Promise { return root; } -function harness(output: ProcessEvent[] = [], probe: { dir?: string; fail?: boolean } = {}) { +function harness( + output: ProcessEvent[] = [], + site: { dir?: string; fail?: boolean; noise?: string[] } = {}, +) { const calls: ProcessCall[] = []; - const probeCalls: string[][] = []; + const discoverCalls: string[][] = []; const fakeStreamProcess: ProcessStreamer = async function* (command, options) { calls.push({ command, options }); yield* output; }; const fakeRunProcess: ProcessRunner = async (command, options) => { - probeCalls.push(command); - if (probe.fail) throw new Error("probe failed"); - options.onOutput?.(`${probe.dir ?? ""}\n`); + discoverCalls.push(command); + if (site.fail) throw new Error("discovery failed"); + for (const line of site.noise ?? []) options.onOutput?.(`${line}\n`); + if (site.dir !== undefined) options.onOutput?.(`AGENTCORE_OTEL_SITECUSTOMIZE=${site.dir}\n`); }; return { calls, - probeCalls, + discoverCalls, runner: new CodeZipDevRunner({ streamProcess: fakeStreamProcess, runProcess: fakeRunProcess }), }; } @@ -220,11 +224,24 @@ describe("CodeZipDevRunner OTEL instrumentation", () => { test("prepends the sitecustomize directory to PYTHONPATH when instrumentation is installed", async () => { const root = await projectRoot(); const directory = await sitecustomizeDir(); - const { calls, probeCalls, runner } = harness([], { dir: directory }); + const { calls, discoverCalls, runner } = harness([], { dir: directory }); + + await collect(runner.run(otelInput(root))); + + expect(discoverCalls[0]?.slice(0, 4)).toEqual(["uv", "run", "python", "-c"]); + expect(calls[0]?.options.env?.PYTHONPATH).toBe(directory); + }); + + test("reads the marked path even when uv writes progress to the merged output", async () => { + const root = await projectRoot(); + const directory = await sitecustomizeDir(); + const { calls, runner } = harness([], { + dir: directory, + noise: ["Resolved 12 packages", "Installed 12 packages"], + }); await collect(runner.run(otelInput(root))); - expect(probeCalls[0]?.slice(0, 4)).toEqual(["uv", "run", "python", "-c"]); expect(calls[0]?.options.env?.PYTHONPATH).toBe(directory); }); @@ -238,26 +255,26 @@ describe("CodeZipDevRunner OTEL instrumentation", () => { expect(calls[0]?.options.env?.PYTHONPATH).toBe(`${directory}${delimiter}/existing`); }); - test("does not probe without an OTEL endpoint or for Node entrypoints", async () => { + test("does not run discovery without an OTEL endpoint or for Node entrypoints", async () => { const root = await projectRoot(true); - const { probeCalls, runner } = harness(); + const { discoverCalls, runner } = harness(); await collect(runner.run(input(root, runtime()))); await collect(runner.run({ ...otelInput(root), runtime: runtime({ entrypoint: "index.js" }) })); - expect(probeCalls).toEqual([]); + expect(discoverCalls).toEqual([]); }); test.each([ - ["probe failure", { fail: true }], + ["discovery failure", { fail: true }], ["missing sitecustomize.py", { dir: "/nonexistent" }], - ] as const)("warns and starts untraced on %s", async (_case, probe) => { + ] as const)("warns and starts untraced on %s", async (_case, site) => { const root = await projectRoot(); - const { calls, probeCalls, runner } = harness([], probe); + const { calls, discoverCalls, runner } = harness([], site); const events = await collect(runner.run(otelInput(root))); - expect(probeCalls).toHaveLength(1); + expect(discoverCalls).toHaveLength(1); expect(calls).toHaveLength(1); expect(calls[0]?.options.env?.PYTHONPATH).toBeUndefined(); expect(events).toContainEqual({ diff --git a/src/core/dev/codezip.ts b/src/core/dev/codezip.ts index 1c84c9317..9764f4f3f 100644 --- a/src/core/dev/codezip.ts +++ b/src/core/dev/codezip.ts @@ -16,6 +16,8 @@ type CodeZipDevRunnerConfig = { runProcess?: ProcessRunner; }; +const SITECUSTOMIZE_MARKER = "AGENTCORE_OTEL_SITECUSTOMIZE="; + export class CodeZipDevRunner implements DevRunner { private readonly streamProcess: ProcessStreamer; private readonly runProcess: ProcessRunner; @@ -51,7 +53,7 @@ export class CodeZipDevRunner implements DevRunner { yield { type: "status", message: "Starting development server" }; const serverProcess = commandForRuntime(entrypoint!, directory, input); if (entrypoint!.endsWith(".py") && input.env?.OTEL_EXPORTER_OTLP_ENDPOINT) { - const sitecustomizeDir = await this.findOtelSitecustomizeDir(directory); + const sitecustomizeDir = await this.findOtelSitecustomizeDir(directory, input.signal); if (sitecustomizeDir) { const existing = serverProcess.options.env?.PYTHONPATH; serverProcess.options.env = { @@ -75,19 +77,29 @@ export class CodeZipDevRunner implements DevRunner { * an `opentelemetry-instrument` wrapper would only instrument uvicorn's reloader * parent, leaving the re-spawned worker processes untraced. */ - private async findOtelSitecustomizeDir(directory: string): Promise { + private async findOtelSitecustomizeDir( + directory: string, + signal: AbortSignal, + ): Promise { const output: string[] = []; - const probe = - "import opentelemetry.instrumentation.auto_instrumentation as m; import os; print(os.path.dirname(m.__file__))"; + // uv writes sync progress to stderr, which merges into onOutput, so the path + // is printed behind a marker and read from that line rather than the last one. + const script = `import opentelemetry.instrumentation.auto_instrumentation as m, os; print("${SITECUSTOMIZE_MARKER}" + os.path.dirname(m.__file__))`; try { - await this.runProcess(["uv", "run", "python", "-c", probe], { + await this.runProcess(["uv", "run", "python", "-c", script], { cwd: directory, onOutput: (chunk) => output.push(chunk), + signal, }); } catch { return undefined; } - const sitecustomizeDir = output.join("").trim().split("\n").at(-1)?.trim(); + const marked = output + .join("") + .split("\n") + .map((line) => line.trim()) + .find((line) => line.startsWith(SITECUSTOMIZE_MARKER)); + const sitecustomizeDir = marked?.slice(SITECUSTOMIZE_MARKER.length); if (!sitecustomizeDir || !existsSync(join(sitecustomizeDir, "sitecustomize.py"))) return undefined; return sitecustomizeDir; diff --git a/src/core/dev/otel/collector.test.ts b/src/core/dev/otel/collector.test.ts index a3785cf48..b9ae3fa66 100644 --- a/src/core/dev/otel/collector.test.ts +++ b/src/core/dev/otel/collector.test.ts @@ -138,6 +138,29 @@ describe("startOtelCollector", () => { expect(await collector.store.list()).toEqual([]); }); + test.each(["null", "[]", "42", '{"resourceSpans":5}'])( + "rejects structurally invalid JSON %s with 400 instead of a persistence error", + async (body) => { + const errors: unknown[] = []; + const strict = await startOtelCollector({ + tracesDirectory: directory, + onError: (error) => errors.push(error), + }); + try { + const response = await fetch(`http://127.0.0.1:${strict.port}/v1/traces`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body, + }); + expect(response.status).toBe(400); + expect(errors).toEqual([]); + expect(await strict.store.list()).toEqual([]); + } finally { + await strict.close(); + } + }, + ); + test("acks with 200 and reports onError when persistence fails", async () => { // A traces dir nested under a regular file makes mkdir (and thus append) fail. const blocker = join(directory, "blocker"); @@ -184,6 +207,17 @@ describe("startOtelCollector", () => { }); }); + test("envVars force the settings that would otherwise break local collection", () => { + expect(collector.envVars).toMatchObject({ + OTEL_SDK_DISABLED: "false", + OTEL_TRACES_EXPORTER: "otlp", + OTEL_LOGS_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_COMPRESSION: "none", + OTEL_EXPORTER_OTLP_TRACES_COMPRESSION: "none", + OTEL_EXPORTER_OTLP_LOGS_COMPRESSION: "none", + }); + }); + test("abort signal closes the receiver", async () => { const controller = new AbortController(); const aborted = await startOtelCollector({ diff --git a/src/core/dev/otel/collector.ts b/src/core/dev/otel/collector.ts index addb760da..8775aeca1 100644 --- a/src/core/dev/otel/collector.ts +++ b/src/core/dev/otel/collector.ts @@ -33,7 +33,7 @@ export const ExportLogsServiceRequest = logs.v1.ExportLogsServiceRequest; type OtlpDecoder = Pick; export interface OtelCollector { - /** The loopback port the OTLP/HTTP receiver listens on. */ + /** The port the OTLP/HTTP receiver listens on. */ port: number; /** Reads the traces this collector persists. */ store: TraceStore; @@ -46,6 +46,8 @@ export interface OtelCollector { export interface StartOtelCollectorOptions { /** Directory to persist OTLP JSON Lines files into. */ tracesDirectory: string; + /** Address to bind. Defaults to 127.0.0.1; use 0.0.0.0 to reach it from a container. */ + host?: string; /** Closes the collector when aborted. */ signal?: AbortSignal; /** Called when a batch can't be persisted; the export is still acked to stop retries. */ @@ -53,15 +55,16 @@ export interface StartOtelCollectorOptions { } /** - * Starts an in-process OTLP/HTTP receiver for dev mode on an OS-assigned - * loopback port. Accepts `POST /v1/traces` and `POST /v1/logs` in protobuf or - * JSON encoding and appends the raw payloads to a TraceStore. + * Starts an in-process OTLP/HTTP receiver for dev mode on an OS-assigned port. + * Accepts `POST /v1/traces` and `POST /v1/logs` in protobuf or JSON encoding and + * appends the raw payloads to a TraceStore. */ export async function startOtelCollector( options: StartOtelCollectorOptions, ): Promise { const store = new TraceStore(options.tracesDirectory); const server = await startHttpServer((request) => route(request, store, options.onError), { + host: options.host, signal: options.signal, }); @@ -74,10 +77,10 @@ async function route( onError?: (error: unknown) => void, ): Promise { if (request.method === "POST" && request.url === "/v1/traces") { - return ingest(request, store, ExportTraceServiceRequest, onError); + return ingest(request, store, ExportTraceServiceRequest, "resourceSpans", onError); } if (request.method === "POST" && request.url === "/v1/logs") { - return ingest(request, store, ExportLogsServiceRequest, onError); + return ingest(request, store, ExportLogsServiceRequest, "resourceLogs", onError); } if (request.method === "GET" && request.url === "/") { return json(200, { status: "ok" }); @@ -89,16 +92,20 @@ async function ingest( request: HttpRequest, store: TraceStore, decoder: OtlpDecoder, + field: "resourceSpans" | "resourceLogs", onError?: (error: unknown) => void, ): Promise { - let payload: OtlpPayload; + let decoded: unknown; try { - payload = decodePayload(request.body, String(request.headers["content-type"] ?? ""), decoder); + decoded = decodePayload(request.body, String(request.headers["content-type"] ?? ""), decoder); } catch { return json(400, { error: "Invalid OTLP payload" }); } + if (!isOtlpPayload(decoded, field)) { + return json(400, { error: "Invalid OTLP payload" }); + } try { - await store.append(payload); + await store.append(decoded); } catch (error) { // A persistence failure (disk full, permissions) is the collector's problem, // not the agent's: ack the export anyway so the SDK exporter stops retrying the @@ -108,24 +115,44 @@ async function ingest( return json(200, {}); } -/** - * Decode an OTLP payload. The JSON round-trip on the protobuf path converts the - * message to plain objects (protobufjs renders Long as string and bytes as base64). - */ -function decodePayload(body: Buffer, contentType: string, decoder: OtlpDecoder): OtlpPayload { +/** Decode an OTLP export body by its content type into a plain, unvalidated object. */ +function decodePayload(body: Buffer, contentType: string, decoder: OtlpDecoder): unknown { if (contentType.includes("application/json")) { - return JSON.parse(body.toString()) as OtlpPayload; + return JSON.parse(body.toString()); } - return JSON.parse(JSON.stringify(decoder.decode(new Uint8Array(body)))) as OtlpPayload; + return decodeProtobufToPlainObject(body, decoder); +} + +/** + * Decode a protobuf export and flatten it to plain objects. The JSON round-trip + * is what does the flattening: protobufjs renders Long as string and bytes as + * base64, which is exactly the wire shape the rest of the code reads. + */ +function decodeProtobufToPlainObject(body: Buffer, decoder: OtlpDecoder): unknown { + return JSON.parse(JSON.stringify(decoder.decode(new Uint8Array(body)))); +} + +/** + * A payload is only valid when it is a plain object whose export field, if + * present, is an array. This rejects non-objects and shapes like + * `{ resourceSpans: 5 }` at the 400 boundary instead of letting them fail later + * inside the store as a mislabeled persistence error. + */ +function isOtlpPayload( + value: unknown, + field: "resourceSpans" | "resourceLogs", +): value is OtlpPayload { + if (typeof value !== "object" || value === null || Array.isArray(value)) return false; + const records = (value as Record)[field]; + return records === undefined || Array.isArray(records); } /** - * Environment for a spawned agent so its OTEL SDK exports to the collector at - * `port`. Signal-specific variables are set alongside the generic ones because - * they take precedence in the SDK — a stray OTEL_EXPORTER_OTLP_TRACES_ENDPOINT - * from the shell or .env.local must not silently redirect traces elsewhere. - * Per the OTEL spec, signal-specific endpoints are full URLs (the signal path - * is only appended to the generic endpoint). + * Env that points a spawned agent's OTEL SDK at the collector on `port`. While + * tracing is on the CLI owns these settings, so nothing from the shell or + * .env.local can turn collection off or break it: compression is off (the + * collector reads bodies undecompressed) and the SDK and exporters stay on. + * Signal-specific endpoints are full URLs and win over the generic one. */ export function otelEnvVars(port: number): Record { const endpoint = `http://127.0.0.1:${port}`; @@ -136,6 +163,12 @@ export function otelEnvVars(port: number): Record { OTEL_EXPORTER_OTLP_PROTOCOL: "http/protobuf", OTEL_EXPORTER_OTLP_TRACES_PROTOCOL: "http/protobuf", OTEL_EXPORTER_OTLP_LOGS_PROTOCOL: "http/protobuf", + OTEL_EXPORTER_OTLP_COMPRESSION: "none", + OTEL_EXPORTER_OTLP_TRACES_COMPRESSION: "none", + OTEL_EXPORTER_OTLP_LOGS_COMPRESSION: "none", + OTEL_SDK_DISABLED: "false", + OTEL_TRACES_EXPORTER: "otlp", + OTEL_LOGS_EXPORTER: "otlp", OTEL_METRICS_EXPORTER: "none", AGENT_OBSERVABILITY_ENABLED: "true", OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: "true", diff --git a/src/handlers/project/dev/index.test.ts b/src/handlers/project/dev/index.test.ts index cde3b11a3..21c1b87f7 100644 --- a/src/handlers/project/dev/index.test.ts +++ b/src/handlers/project/dev/index.test.ts @@ -188,7 +188,7 @@ describe("project dev trace collection", () => { expect(subject.collector.starts).toEqual([ { tracesDirectory: join("/workspace/project", "agentcore", ".cli", "traces", "otlp"), - signal: expect.any(AbortSignal), + host: "127.0.0.1", onError: expect.any(Function), }, ]); @@ -200,6 +200,15 @@ describe("project dev trace collection", () => { expect(subject.collector.state.closed).toBe(1); }); + test("binds the collector to all interfaces so a container can reach it", async () => { + const subject = harness({ + project: project(runtime("support", "Container")), + }); + await subject.run(); + + expect(subject.collector.starts[0]?.host).toBe("0.0.0.0"); + }); + test("reports a trace-persistence failure once, not per failed export", async () => { const subject = harness(); await subject.run(); diff --git a/src/handlers/project/dev/index.ts b/src/handlers/project/dev/index.ts index 9cc3ab731..b37ae7b7a 100644 --- a/src/handlers/project/dev/index.ts +++ b/src/handlers/project/dev/index.ts @@ -123,7 +123,9 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => let tracePersistErrorReported = false; collector = await config.startTraceCollector({ tracesDirectory, - signal: controller.signal, + // A container reaches the collector over the host bridge, which a + // 127.0.0.1 bind refuses, so the container path binds all interfaces. + host: 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) => { @@ -167,6 +169,8 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) => throw error; } 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(); } }, diff --git a/src/handlers/project/dev/types.ts b/src/handlers/project/dev/types.ts index ccbdcc886..15b51cb96 100644 --- a/src/handlers/project/dev/types.ts +++ b/src/handlers/project/dev/types.ts @@ -27,7 +27,8 @@ export interface DevTraceCollector { export type DevTraceCollectorStarter = (options: { tracesDirectory: string; - signal?: AbortSignal; + /** Address to bind. Defaults to 127.0.0.1; 0.0.0.0 lets a container reach it. */ + host?: string; /** Reports a trace-persistence failure (the export is still acked to stop retries). */ onError?: (error: unknown) => void; }) => Promise; diff --git a/src/io/exec.test.ts b/src/io/exec.test.ts index d20a8a090..d4ab96454 100644 --- a/src/io/exec.test.ts +++ b/src/io/exec.test.ts @@ -76,6 +76,25 @@ describe("runProcess", () => { runProcess(["definitely-not-a-real-tool-xyz"], { cwd: process.cwd() }), ).rejects.toBeInstanceOf(ProcessFailedError); }); + + test("terminates and rejects when the signal aborts", async () => { + const waiting = await script("wait.js", "setTimeout(() => {}, 60_000)"); + const controller = new AbortController(); + const promise = runProcess(["node", waiting], { + cwd: process.cwd(), + signal: controller.signal, + }); + controller.abort(new Error("cancelled")); + + await expect(promise).rejects.toThrow("cancelled"); + }); + + test("rejects immediately when the signal is already aborted", async () => { + const succeeding = await script("noop.js", ""); + await expect( + runProcess(["node", succeeding], { cwd: process.cwd(), signal: AbortSignal.abort() }), + ).rejects.toThrow(); + }); }); async function collect(events: AsyncIterable): Promise { diff --git a/src/io/exec.ts b/src/io/exec.ts index 94ad19f59..4496975df 100644 --- a/src/io/exec.ts +++ b/src/io/exec.ts @@ -55,6 +55,8 @@ export type RunProcessOptions = { cwd: string; /** Receives each chunk of combined stdout/stderr as it streams (e.g. into a logger). */ onOutput?: (chunk: string) => void; + /** Terminates the process and rejects when aborted, so callers can cancel a slow run. */ + signal?: AbortSignal; }; /** Runs a subprocess to completion. Injectable so tests never spawn real processes. */ @@ -81,8 +83,12 @@ export type ProcessStreamer = ( * Runs a subprocess, streaming combined stdout/stderr to `onOutput` while also * capturing it; rejects with {@link ProcessFailedError} on a non-zero exit. */ -export const runProcess: ProcessRunner = ([executable, ...args], { cwd, onOutput }) => { +export const runProcess: ProcessRunner = ([executable, ...args], { cwd, onOutput, signal }) => { return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(abortReason(signal)); + return; + } const child = spawn(executable!, args, { cwd, stdio: ["ignore", "pipe", "pipe"], @@ -98,11 +104,17 @@ export const runProcess: ProcessRunner = ([executable, ...args], { cwd, onOutput child.stdout.on("data", collect); child.stderr.on("data", collect); + const onAbort = () => killTree(child, "SIGTERM"); + signal?.addEventListener("abort", onAbort, { once: true }); + child.on("error", (error) => { + signal?.removeEventListener("abort", onAbort); reject(new ProcessFailedError([executable!, ...args], cwd, null, String(error))); }); child.on("close", (exitCode) => { - if (exitCode === 0) resolve(); + signal?.removeEventListener("abort", onAbort); + if (signal?.aborted) reject(abortReason(signal)); + else if (exitCode === 0) resolve(); else reject(new ProcessFailedError([executable!, ...args], cwd, exitCode, output)); }); }); diff --git a/src/io/httpServer.test.ts b/src/io/httpServer.test.ts index a96b23fb0..8325451e6 100644 --- a/src/io/httpServer.test.ts +++ b/src/io/httpServer.test.ts @@ -51,6 +51,26 @@ describe("startHttpServer", () => { expect(response.status).toBe(413); }); + test("binds the given host", async () => { + handle = await startHttpServer(() => ({ status: 200 }), { host: "0.0.0.0" }); + expect((await fetch(`http://127.0.0.1:${handle.port}/`)).status).toBe(200); + }); + + test("a client that disconnects mid-response does not take down the server", async () => { + handle = await startHttpServer(async () => { + await Bun.sleep(50); + return { status: 200, body: "late" }; + }); + + const controller = new AbortController(); + const aborted = fetch(`http://127.0.0.1:${handle.port}/`, { signal: controller.signal }); + controller.abort(); + await expect(aborted).rejects.toThrow(); + await Bun.sleep(80); + + expect((await fetch(`http://127.0.0.1:${handle.port}/`)).status).toBe(200); + }); + test("aborting the signal closes the server", async () => { const controller = new AbortController(); const server = await startHttpServer(() => ({ status: 200 }), { signal: controller.signal }); diff --git a/src/io/httpServer.ts b/src/io/httpServer.ts index d9df97214..142ae6fd0 100644 --- a/src/io/httpServer.ts +++ b/src/io/httpServer.ts @@ -34,21 +34,24 @@ export interface HttpServerHandle { } /** - * Starts a loopback-only HTTP server for local dev tooling. Binds 127.0.0.1 on - * the given port (0 lets the OS assign one). Handler errors become plain 500s; - * oversized bodies become 413s. Aborting the signal closes the server. + * Starts an HTTP server for local dev tooling. Binds `host` (default 127.0.0.1) + * on the given port (0 lets the OS assign one). Handler errors become plain 500s; + * oversized bodies become 413s. Aborting the signal closes the server. A wider + * bind such as 0.0.0.0 is only for reaching the server from a container. */ export async function startHttpServer( handler: HttpRequestHandler, - options: { port?: number; signal?: AbortSignal } = {}, + options: { port?: number; host?: string; signal?: AbortSignal } = {}, ): Promise { const server = createServer((request, response) => { - void respond(handler, request, response); + // A dropped connection mid-response can reject here; swallow it so a client + // that disconnects can never take down the whole dev command. + void respond(handler, request, response).catch(() => {}); }); await new Promise((resolve, reject) => { server.once("error", reject); - server.listen(options.port ?? 0, "127.0.0.1", resolve); + server.listen(options.port ?? 0, options.host ?? "127.0.0.1", resolve); }); const address = server.address(); @@ -86,6 +89,12 @@ async function respond( response.writeHead(result.status, result.headers); response.end(result.body); } catch { + // Once any byte is written, writeHead throws, so only send the 500 when the + // response has not started; otherwise just close what is already open. + if (response.headersSent) { + response.end(); + return; + } response.writeHead(500, { "Content-Type": "application/json" }); response.end(JSON.stringify({ error: "internal error" })); }