diff --git a/src/jvm/com/tjclp/scalagent/a2a/A2AServerLive.scala b/src/jvm/com/tjclp/scalagent/a2a/A2AServerLive.scala index d129fef..dc9a7bc 100644 --- a/src/jvm/com/tjclp/scalagent/a2a/A2AServerLive.scala +++ b/src/jvm/com/tjclp/scalagent/a2a/A2AServerLive.scala @@ -136,7 +136,16 @@ private[a2a] final class A2AServerLiveImpl( scope <- Scope.make _ <- (for - serverEnv <- (ZLayer.succeed(Server.Config.default.binding(config.host, config.port)) >>> Server.live) + // zio-http's Server.Config.default caps request bodies at 100 KiB + // (RequestStreaming.Disabled(1024*100)); A2A messages carry base64 + // file uploads that exceed that, so apply the configured limit or + // they'd 413. disableRequestStreaming keeps full-body aggregation + // (the routes parse the whole JSON-RPC body) with the larger cap. + serverEnv <- (ZLayer.succeed( + Server.Config.default + .binding(config.host, config.port) + .disableRequestStreaming(config.maxRequestBodyBytes) + ) >>> Server.live) .build(scope) _ <- Server.install(a2aRoutes).provideEnvironment(serverEnv) _ <- startGrpcServer diff --git a/src/shared/com/tjclp/scalagent/a2a/A2AServerDefaults.scala b/src/shared/com/tjclp/scalagent/a2a/A2AServerDefaults.scala index 8883280..9182460 100644 --- a/src/shared/com/tjclp/scalagent/a2a/A2AServerDefaults.scala +++ b/src/shared/com/tjclp/scalagent/a2a/A2AServerDefaults.scala @@ -10,7 +10,10 @@ object A2AServerDefaults: val EventReplayLimit: Int = 1000 val EventStoreAppendTimeout: Duration = 2.seconds val EventStoreLoadTimeout: Duration = 5.seconds - val MaxRequestBodyBytes: Int = 1024 * 1024 + // A2A messages carry file uploads as base64 inside the JSON-RPC body + // (~33% inflation), so this must comfortably exceed the largest raw + // upload a client allows. 64 MiB covers rtalk's 32 MB raw cap encoded. + val MaxRequestBodyBytes: Int = 64 * 1024 * 1024 val PushNotificationPostTimeout: Duration = 20.seconds val PushNotificationMaxRetries: Int = 2 val PushNotificationRetryBaseDelay: Duration = 250.millis diff --git a/test/js/com/tjclp/scalagent/a2a/A2ARestTransportSpec.scala b/test/js/com/tjclp/scalagent/a2a/A2ARestTransportSpec.scala index 86feb00..2990a47 100644 --- a/test/js/com/tjclp/scalagent/a2a/A2ARestTransportSpec.scala +++ b/test/js/com/tjclp/scalagent/a2a/A2ARestTransportSpec.scala @@ -552,6 +552,44 @@ class A2ARestTransportSpec extends FunSuite: assert(response.error.exists(_.message.contains("Request body exceeds 8 byte limit"))) } + // Parity with the JVM #58 fix: the JS server enforces maxRequestBodyBytes in + // readBodyLimited (not via a framework default), so the raised 64 MiB shared + // default must let it accept the same large inline uploads. Uses the default + // limit (no override) and a body well above zio-http's old 100 KiB cap. + test("accepts message/send bodies larger than the old 100 KiB cap with the default limit"): + val config = A2AServer.Config( + name = "LargeBodyTest", + description = "Large body test server", + host = "127.0.0.1", + port = 0, + executionOverride = Some(completedExecution), + ) + // ~200 KiB: above zio-http's old 100 KiB cap, well below the 64 MiB default. + val request = A2ARequest + .MessageSend(A2AMessage.userText("x" * (200 * 1024)), configuration = Some(MessageSendConfiguration.default)) + .toJson + val headers = Map( + "Content-Type" -> A2AContentType.A2AJson, + A2AHeader.Version -> A2AProtocol.Version, + ) + + val program = + ZIO.scoped { + for + server <- A2AServer.create(config) + sent <- fetchText(server.url + "/message:send", method = "POST", body = Some(request), headers = headers) + result <- ZIO.fromEither(sent._3.fromJson[A2AResponse.SendMessageResult].left.map(new RuntimeException(_))) + task = result match + case A2AResponse.SendMessageResult.TaskResult(task) => task + case other => throw new RuntimeException(s"Expected task, got $other") + yield (sent._1, task) + } + + runTask(program).map { case (status, task) => + assertEquals(status, 200) + assertEquals(task.status.state, TaskState.Completed) + } + test("REST errors use google.rpc-style bodies"): val config = A2AServer.Config( name = "RestErrorTest", diff --git a/test/jvm/com/tjclp/scalagent/a2a/A2AServerLiveSpec.scala b/test/jvm/com/tjclp/scalagent/a2a/A2AServerLiveSpec.scala index c92ae30..549aacf 100644 --- a/test/jvm/com/tjclp/scalagent/a2a/A2AServerLiveSpec.scala +++ b/test/jvm/com/tjclp/scalagent/a2a/A2AServerLiveSpec.scala @@ -59,6 +59,17 @@ class A2AServerLiveSpec extends FunSuite: response.statusCode() -> response.body() } + private def post(url: String, body: String, requestHeaders: Map[String, String]): Task[(Int, String)] = + ZIO.attemptBlocking { + val builder = HttpRequest + .newBuilder(URI.create(url)) + .timeout(JavaDuration.ofSeconds(10)) + .POST(HttpRequest.BodyPublishers.ofString(body)) + requestHeaders.foreach { case (name, value) => builder.header(name, value) } + val response = client.send(builder.build(), HttpResponse.BodyHandlers.ofString()) + response.statusCode() -> response.body() + } + private def grpcStreamingResponse( port: Int, message: A2AMessage, @@ -688,6 +699,49 @@ class A2AServerLiveSpec extends FunSuite: assert(response.error.exists(_.message.contains("Request body exceeds 8 byte limit"))) } + // Regression for #58: zio-http's Server.Config.default caps request streaming at + // 100 KiB, so unless config.maxRequestBodyBytes is wired into the server the bound + // server 413s on larger bodies (e.g. base64 file uploads) before the handler runs. + // The handleHttp-based body-limit test above bypasses the Server.Config cap, so this + // exercises a real bound zio-http server end to end. + test("JVM bound server accepts JSON-RPC bodies larger than zio-http's default 100 KiB cap"): + // ~200 KiB: above zio-http's 100 KiB default, well below the 64 MiB configured limit. + val largeText = "x" * (200 * 1024) + val body = rpc( + A2AMethod.MessageSend, + A2ARequest.MessageSend(A2AMessage.userText(largeText)).toJsonAST.toOption.get, + 58, + ).toJson + + val program = + ZIO.scoped { + for + port <- freeLocalPort + config = A2AServerLive.Config( + name = "LargeBodyJvmTest", + description = "Large body JVM test server", + host = "127.0.0.1", + port = port, + executionOverride = Some(completedExecution), + ) + _ <- A2AServerLive + .create(config) + .timeoutFail(new RuntimeException("server did not become ready"))(10.seconds) + result <- post( + s"http://127.0.0.1:$port/", + body, + Map("Content-Type" -> A2AContentType.Json, A2AHeader.Version -> A2AProtocol.Version), + ) + response <- ZIO.fromEither(result._2.fromJson[JsonRpcResponse].left.map(new RuntimeException(_))) + task <- sendTask(response) + yield (result._1, task) + } + + runTask(program).map { case (status, task) => + assertEquals(status, 200) + assertEquals(task.status.state, TaskState.Completed) + } + test("JVM start fails when the configured port is already bound"): val program = ZIO.scoped {