diff --git a/packages/sdk-java/runtime-broker/src/main/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransport.java b/packages/sdk-java/runtime-broker/src/main/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransport.java index 5fb534987dd..654b128cbd2 100644 --- a/packages/sdk-java/runtime-broker/src/main/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransport.java +++ b/packages/sdk-java/runtime-broker/src/main/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransport.java @@ -25,10 +25,11 @@ * HTTP client for the private Managed Runtime v2 routes. * *

Attestation plus the tool operations execute, status, and cancel, keyed - * by the original call reference. Prepare, acquire, control, and release stay - * on a later transport slice. + * by the original call reference. Status and cancel answers are projected to + * the Broker's closed state and result. Acquire, control, and release are + * not part of the v2 tool contract and fail closed. */ -public final class HttpRuntimeTransport { +public final class HttpRuntimeTransport implements RuntimeTransport { static final int BODY_LIMIT_BYTES = 16 * 1024; static final int TOOL_REQUEST_LIMIT_BYTES = 256 * 1024; static final int TOOL_RESULT_LIMIT_BYTES = 1024 * 1024; @@ -194,7 +195,8 @@ public CompletionStage> status(RuntimeLease lease, body.put("afterSequence", afterSequence); byte[] encoded = encodeToolRequest(body, BODY_LIMIT_BYTES); return post(lease, STATUS_PATH, encoded, TOOL_RESULT_LIMIT_BYTES) - .thenApply(bytes -> parseToolResponse(bytes, "status")); + .thenApply(bytes -> projectClosedStatus( + parseToolResponse(bytes, "status"), "status")); } /** Asks the Runtime to cancel one call by its original reference. */ @@ -209,7 +211,62 @@ public CompletionStage> cancel(RuntimeLease lease, body.put("reference", referenceIdentity(reference)); byte[] encoded = encodeToolRequest(body, BODY_LIMIT_BYTES); return post(lease, CANCEL_PATH, encoded, TOOL_RESULT_LIMIT_BYTES) - .thenApply(bytes -> parseToolResponse(bytes, "cancel")); + .thenApply(bytes -> projectClosedStatus( + parseToolResponse(bytes, "cancel"), "cancel")); + } + + /** + * Session verbs are not on the v2 tool contract. Fail closed instead of + * inventing a route the worker does not serve. + */ + @Override + public CompletionStage acquire(RuntimeLease lease, + RuntimeSession session) { + return unsupportedSessionVerb(); + } + + @Override + public CompletionStage control(RuntimeLease lease, + RuntimeSession session, Map operation) { + return unsupportedSessionVerb(); + } + + @Override + public CompletionStage release(RuntimeLease lease, + RuntimeSession session) { + return unsupportedSessionVerb(); + } + + private static CompletionStage unsupportedSessionVerb() { + return CompletableFuture.failedFuture(new RuntimeBrokerException(501, + "runtime_session_verb_unsupported", + "Runtime transport does not support session verbs.", false)); + } + + /** + * The Broker accepts only {@code state}, plus {@code result} when the + * state is {@code settled}. Wire fields such as {@code protocolVersion} + * and {@code lastSequence} stay on the HTTP response and are validated + * before this projection. + */ + private static Map projectClosedStatus( + Map response, String operation) { + Object state = response.get("state"); + if (!(state instanceof String text) || !TOOL_STATES.contains(text)) { + throw protocol("Managed Runtime " + operation + + " returned an invalid state."); + } + boolean settled = "settled".equals(text); + if (settled != response.containsKey("result")) { + throw protocol("Managed Runtime " + operation + + " result does not match its state."); + } + Map projected = new LinkedHashMap<>(); + projected.put("state", text); + if (settled) { + projected.put("result", response.get("result")); + } + return Map.copyOf(projected); } private static Map referenceIdentity( diff --git a/packages/sdk-java/runtime-broker/src/test/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransportTest.java b/packages/sdk-java/runtime-broker/src/test/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransportTest.java index b21a17d670f..4d108ed4f1e 100644 --- a/packages/sdk-java/runtime-broker/src/test/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransportTest.java +++ b/packages/sdk-java/runtime-broker/src/test/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransportTest.java @@ -23,6 +23,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.CompletionStage; @@ -397,9 +398,23 @@ void statusAnswersUnknownFromTheSharedFixture() throws Exception { assertEquals(statusSuite.required("canonicalRequest") .required("body"), sent); assertEquals("unknown", answer.get("state")); + assertEquals(Set.of("state"), answer.keySet()); assertNull(answer.get("result")); } + @Test + void sessionVerbsFailClosed() { + assertTrue(transport instanceof RuntimeTransport); + CompletionException thrown = assertThrows(CompletionException.class, + () -> transport.acquire(toolLease(1), toolSession()) + .toCompletableFuture().join()); + RuntimeBrokerException failure = + (RuntimeBrokerException) thrown.getCause(); + assertEquals(501, failure.getStatusCode()); + assertEquals("runtime_session_verb_unsupported", failure.getCode()); + assertFalse(failure.isRetryable()); + } + @Test void cancelSettlesAPreparedExecutionFromTheSharedFixture() throws Exception { @@ -714,7 +729,10 @@ void acceptsLastSequenceOnlyOnStatus() throws Exception { toolLease(server.getAddress().getPort()), toolSession(), toolReference(), 0).toCompletableFuture() .get(2, TimeUnit.SECONDS); - assertEquals(0, ((Number) status.get("lastSequence")).longValue()); + assertFalse(status.containsKey("lastSequence")); + assertFalse(status.containsKey("protocolVersion")); + assertEquals("settled", status.get("state")); + assertTrue(status.containsKey("result")); } @Test @@ -736,8 +754,8 @@ void validatesSequenceNumbersWithoutRoundingOrLongTruncation() toolLease(server.getAddress().getPort()), toolSession(), toolReference(), 0).toCompletableFuture() .get(2, TimeUnit.SECONDS); - assertEquals(0, new BigDecimal(valid).compareTo(new BigDecimal( - answer.get("lastSequence").toString())), valid); + assertEquals(Set.of("state"), answer.keySet(), valid); + assertEquals("unknown", answer.get("state"), valid); } reply.set(json(200, ("{\"protocolVersion\":2.000000000000000001," + "\"state\":\"unknown\"}").getBytes(StandardCharsets.UTF_8)));