diff --git a/.github/workflows/release-sdk-java.yml b/.github/workflows/release-sdk-java.yml index e387aaaa163..7127298bbe8 100644 --- a/.github/workflows/release-sdk-java.yml +++ b/.github/workflows/release-sdk-java.yml @@ -17,6 +17,7 @@ jobs: release: if: "${{ github.repository == 'QwenLM/qwen-code' }}" runs-on: 'ubuntu-latest' + timeout-minutes: 60 environment: name: "${{ inputs.dry_run && 'sdk-java-dry-run' || 'production-release' }}" url: '${{ github.server_url }}/${{ github.repository }}/releases' @@ -126,6 +127,25 @@ jobs: echo "tag_state=${tag_state}" >> "${GITHUB_OUTPUT}" echo "artifact_state=${artifact_state}" >> "${GITHUB_OUTPUT}" + - name: 'Require publishing credentials' + if: "${{ !inputs.dry_run && steps.preflight.outputs.artifact_state == 'missing' }}" + env: + CENTRAL_USERNAME: '${{ secrets.CENTRAL_USERNAME }}' + CENTRAL_PASSWORD: '${{ secrets.CENTRAL_PASSWORD }}' + MAVEN_GPG_PRIVATE_KEY: '${{ secrets.MAVEN_GPG_PRIVATE_KEY }}' + MAVEN_GPG_PASSPHRASE: '${{ secrets.MAVEN_GPG_PASSPHRASE }}' + run: |- + set -euo pipefail + missing=() + [[ -n "${CENTRAL_USERNAME}" ]] || missing+=('CENTRAL_USERNAME') + [[ -n "${CENTRAL_PASSWORD}" ]] || missing+=('CENTRAL_PASSWORD') + [[ -n "${MAVEN_GPG_PRIVATE_KEY}" ]] || missing+=('MAVEN_GPG_PRIVATE_KEY') + [[ -n "${MAVEN_GPG_PASSPHRASE}" ]] || missing+=('MAVEN_GPG_PASSPHRASE') + if (( ${#missing[@]} > 0 )); then + printf 'Missing production-release environment secret: %s\n' "${missing[@]}" >&2 + exit 1 + fi + - name: 'Verify release build' working-directory: '${{ env.SDK_DIRECTORY }}' run: 'mvn --batch-mode --no-transfer-progress clean verify -Dgpg.skip=true' @@ -155,6 +175,7 @@ jobs: env: GH_TOKEN: '${{ github.token }}' run: |- + set -euo pipefail git config user.name 'github-actions[bot]' git config user.email '41898282+github-actions[bot]@users.noreply.github.com' git tag -a "${RELEASE_TAG}" -m "Java SDK ${SDK_VERSION}" @@ -187,13 +208,14 @@ jobs: - name: 'Verify Maven Central availability' if: '${{ !inputs.dry_run }}' run: |- + set -euo pipefail artifact_url="https://repo1.maven.org/maven2/com/alibaba/qwencode-sdk/${SDK_VERSION}/qwencode-sdk-${SDK_VERSION}.pom" - for attempt in {1..30}; do - if curl --fail --silent --show-error --location --output /dev/null "${artifact_url}"; then + for attempt in {1..40}; do + if curl --fail --silent --show-error --location --connect-timeout 10 --max-time 10 --output /dev/null "${artifact_url}"; then exit 0 fi - if [[ "${attempt}" -lt 30 ]]; then - sleep 10 + if [[ "${attempt}" -lt 40 ]]; then + sleep 30 fi done echo "Published artifact is not available at ${artifact_url}." >&2 @@ -204,6 +226,7 @@ jobs: env: GH_TOKEN: '${{ github.token }}' run: |- + set -euo pipefail git fetch origin "refs/tags/${RELEASE_TAG}:refs/tags/${RELEASE_TAG}" --force if gh release view "${RELEASE_TAG}" >/dev/null 2>&1; then echo "GitHub Release ${RELEASE_TAG} already exists." diff --git a/.github/workflows/sdk-java.yml b/.github/workflows/sdk-java.yml index ba99ba31134..071c522bcf7 100644 --- a/.github/workflows/sdk-java.yml +++ b/.github/workflows/sdk-java.yml @@ -45,6 +45,7 @@ jobs: test: name: '${{ matrix.os }} / Java ${{ matrix.java }}' runs-on: '${{ matrix.os }}' + timeout-minutes: 30 strategy: fail-fast: false matrix: @@ -88,6 +89,7 @@ jobs: daemon-e2e: name: 'Real daemon E2E / Java 11' runs-on: 'ubuntu-latest' + timeout-minutes: 30 steps: - name: 'Checkout' uses: 'actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10' # v6.0.3 diff --git a/docs/design/java-daemon-sdk-alpha.md b/docs/design/java-daemon-sdk-alpha.md index e7370ca25d3..fd0b059d453 100644 --- a/docs/design/java-daemon-sdk-alpha.md +++ b/docs/design/java-daemon-sdk-alpha.md @@ -91,8 +91,9 @@ independent and is always enforced by the SDK. ## Wire flow 1. Send one non-retried `POST /session/:id/prompt`. -2. Require `202` and validate `{promptId,lastEventId}`. -3. Open `GET /session/:id/events` with `Last-Event-ID` set to the watermark. +2. Require `202` and validate `{promptId,lastEventId,eventEpoch?}`. +3. Open `GET /session/:id/events` with `Last-Event-ID` set to the watermark + and `X-Qwen-Event-Epoch` set when the daemon supplied an epoch. 4. Replay and observe only events correlated with that prompt, while treating session-level failure frames as fatal. 5. Stop only on matching `turn_complete` or `turn_error`. @@ -107,7 +108,11 @@ The JDK `HttpClient` uses HTTP/1.1 and never follows redirects. Every request sends JSON or event-stream `Accept` headers, bearer authentication when configured, and the daemon-issued `X-Qwen-Client-Id` after session creation. SSE additionally sends `Accept-Encoding: identity`, `Cache-Control: no-cache`, -and `Last-Event-ID`. +and `Last-Event-ID`. When available, `X-Qwen-Event-Epoch` travels with that +cursor. The client seeds it from the prompt admission, learns it from a +validated SSE response header for compatibility, retains a known value when a +response omits the header, and fails closed if the value changes during prompt +observation. Finite JSON and error bodies are consumed by a bounded subscriber and raced against the request deadline through `sendAsync`; receiving response headers @@ -166,10 +171,11 @@ provider because Logback is test-only. The compatible daemon is the qwen-code build released from the same source revision as the SDK. It contains the per-client detach ledger from #7386, the -per-epoch terminal guarantee from #7400, and this release's acknowledged -admission cancellation plus FIFO cancel-drain fence. The #7400 commit alone can -still acknowledge cancel before agent dispatch without stopping the admitted -prompt, or let an unacknowledged session-scoped cancel reach a queued successor. +per-epoch terminal guarantee from #7400, restart-safe event cursor epochs from +#7458, and this release's acknowledged admission cancellation plus FIFO +cancel-drain fence. The #7400 commit alone can still acknowledge cancel before +agent dispatch without stopping the admitted prompt, or let an unacknowledged +session-scoped cancel reach a queued successor. The bundled ACP child handles the daemon's internal cancellation request through one acknowledged admission-aware handshake. A custom standards-compliant ACP child that does not implement that extension receives one standard @@ -187,7 +193,8 @@ outcome unknown and the session unusable. Reclaiming a wedged shared ACP child without terminating sibling sessions requires stronger runtime isolation and is outside this alpha. -The alpha does not promise exactly-once execution across daemon restarts, +The alpha detects an event-epoch change during an observed prompt and fails +closed, but does not promise exactly-once execution across daemon restarts, automatic epoch recovery, snapshot/resync, persisted cursors, or true prompt-ID-targeted cancellation. It also does not expose creation-time model selection until the daemon can return a definitive result or the SDK owns a @@ -200,8 +207,9 @@ reaping. Those cases require stronger daemon contracts. Unit tests use an in-process HTTP server to inject SSE fragmentation, slow single-line delivery, replay, duplicates, gaps, conflicting prompt IDs, opaque future event data, watermark replay, disconnects, compressed responses, -stalled finite bodies, resync, observer failures, terminal absence, and -ambiguous mutation responses. Lifecycle tests cover one-local-prompt admission, +stalled finite bodies, event-epoch propagation and mismatch, resync, observer +failures, terminal absence, and ambiguous mutation responses. Lifecycle tests +cover one-local-prompt admission, admission/close serialization, deadline terminal followed by session reuse, cancelled completion, teardown terminal ordering, bounded text, automatic heartbeat, idempotent close, detach client identity, detach-once, and explicit diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 07a7f792c46..20d418da954 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -5836,6 +5836,56 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('deduplicates repeated cancellation broadcasts while idle', async () => { + const events: BridgeEvent[] = []; + const handle = makeChannel(); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const abort = new AbortController(); + const collecting = (async () => { + for await (const event of bridge.subscribeEvents(session.sessionId, { + signal: abort.signal, + })) { + if (event.type === 'prompt_cancelled') events.push(event); + } + })(); + + await bridge.cancelSession(session.sessionId); + await bridge.cancelSession(session.sessionId); + await vi.waitFor(() => expect(events).toHaveLength(1)); + + abort.abort(); + await collecting; + await bridge.shutdown(); + }); + + it('allows a new idle cancellation after another prompt starts', async () => { + const events: BridgeEvent[] = []; + const handle = makeChannel(); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const abort = new AbortController(); + const collecting = (async () => { + for await (const event of bridge.subscribeEvents(session.sessionId, { + signal: abort.signal, + })) { + if (event.type === 'prompt_cancelled') events.push(event); + } + })(); + + await bridge.cancelSession(session.sessionId); + await bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'reset idle cancel latch' }], + }); + await bridge.cancelSession(session.sessionId); + await vi.waitFor(() => expect(events).toHaveLength(2)); + + abort.abort(); + await collecting; + await bridge.shutdown(); + }); + it('broadcasts prompt_cancelled to peers when the originator SSE aborts mid-prompt', async () => { // Cross-client sync: client disconnect (tab close / network drop / // laptop sleep) is the most common cancel trigger in production. diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index b09b86a2c3b..4d02df15a92 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -593,6 +593,8 @@ interface SessionEntry { retryAllowed: boolean; /** Prompt id whose `prompt_cancelled` event has already been broadcast. */ cancelBroadcastPromptId?: string; + /** Whether an id-less idle cancellation has already been broadcast. */ + cancelBroadcastWithoutPrompt?: boolean; /** * Count of times `spawnOrAttach` has returned `attached: true` for * this entry — i.e. a second-or-subsequent client claimed this @@ -986,13 +988,18 @@ function broadcastPromptCancelledOnce( originatorClientId: string | undefined, reason?: 'forward_failed', ): void { - if (promptId !== undefined && entry.cancelBroadcastPromptId === promptId) { + if ( + (promptId !== undefined && entry.cancelBroadcastPromptId === promptId) || + (promptId === undefined && entry.cancelBroadcastWithoutPrompt === true) + ) { writeStderrLine( - `broadcastPromptCancelledOnce: suppressed duplicate cancel for session ${sessionId} prompt=${promptId}`, + `broadcastPromptCancelledOnce: suppressed duplicate cancel for session ${sessionId} prompt=${promptId ?? 'none'}`, ); return; } - if (promptId !== undefined) { + if (promptId === undefined) { + entry.cancelBroadcastWithoutPrompt = true; + } else { entry.cancelBroadcastPromptId = promptId; } broadcastPromptCancelled( @@ -5431,6 +5438,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { })(); entry.promptActive = true; entry.activePromptId = pendingEntry.promptId; + delete entry.cancelBroadcastWithoutPrompt; delete entry.turnError; activePromptCounter++; entry.sessionLastSeenAt = Date.now(); diff --git a/packages/sdk-java/qwencode/README.md b/packages/sdk-java/qwencode/README.md index 44eae3d10c4..bd55730ca64 100644 --- a/packages/sdk-java/qwencode/README.md +++ b/packages/sdk-java/qwencode/README.md @@ -67,7 +67,7 @@ npx tsx scripts/run-java-daemon-sdk-e2e.ts Start `qwen serve`, then create an independent thread-scoped session. `promptText` returns only after a matching `turn_complete`; incomplete streams fail with `PromptOutcomeIndeterminateException` rather than returning partial text as success. -For the lifecycle guarantees assumed by `0.1.0-alpha`, use the qwen-code build released from the same source revision as the SDK. The daemon must contain the idempotent per-client detach ledger from [#7386](https://github.com/QwenLM/qwen-code/pull/7386), the per-epoch terminal guarantee from [#7400](https://github.com/QwenLM/qwen-code/pull/7400), and this release's acknowledged admission cancellation plus FIFO cancel-drain fence. The #7400 commit alone is not sufficient: a same-wire daemon can acknowledge cancel before agent dispatch without stopping the admitted prompt, or let an unacknowledged session-scoped cancel reach a queued successor. The bundled ACP child uses one acknowledged admission-aware cancellation handshake; a custom standards-compliant ACP child without that extension receives one standard `session/cancel` notification. Feature negotiation cannot distinguish older same-wire daemon builds, so the SDK fails closed rather than reporting partial output as success. +For the lifecycle guarantees assumed by `0.1.0-alpha`, use the qwen-code build released from the same source revision as the SDK. The daemon must contain the idempotent per-client detach ledger from [#7386](https://github.com/QwenLM/qwen-code/pull/7386), the per-epoch terminal guarantee from [#7400](https://github.com/QwenLM/qwen-code/pull/7400), restart-safe event cursor epochs from [#7458](https://github.com/QwenLM/qwen-code/pull/7458), and this release's acknowledged admission cancellation plus FIFO cancel-drain fence. The #7400 commit alone is not sufficient: a same-wire daemon can acknowledge cancel before agent dispatch without stopping the admitted prompt, or let an unacknowledged session-scoped cancel reach a queued successor. The bundled ACP child uses one acknowledged admission-aware cancellation handshake; a custom standards-compliant ACP child without that extension receives one standard `session/cancel` notification. Feature negotiation cannot distinguish older same-wire daemon builds, so the SDK fails closed rather than reporting partial output as success. The bundled cancellation handshake deliberately waits for the targeted prompt call to settle before the daemon dispatches its queued successor. It has no timeout that merely acknowledges cancellation: doing so could let a late session-scoped cancel reach the next prompt. If a provider, tool, or custom integration ignores its `AbortSignal` indefinitely, the cancel mutation can therefore remain outcome-unknown and that session must not be reused. Treat a formal prompt terminal received within the caller's observation boundary as authoritative; otherwise close or destroy the session after observation fails. Recovering a wedged shared ACP child without disturbing its sibling sessions requires stronger runtime isolation and is outside this alpha contract. @@ -95,7 +95,7 @@ Use `startPrompt` with a `PromptObserver` when you need ordered text, thought, t When cancellation, deadline, teardown, or agent settlement race, the daemon's exactly-once latch publishes the first formal terminal and suppresses later candidates. Always branch on the received terminal itself; the last control mutation sent by the client does not determine the terminal kind or error code. -The SSE transport sends `Accept-Encoding: identity` and `Last-Event-ID`, validates framing and event IDs, deduplicates replay, and reconnects only the SSE GET. Prompt and other mutation requests are never retried automatically. HTTP 408 and 5xx responses to prompt admission, session creation, permission, cancel, heartbeat, detach, or delete are reported as outcome-unknown because they do not prove that the daemon rejected the mutation. Finite response bodies and SSE observation have independent deadlines. +The SSE transport sends `Accept-Encoding: identity` and `Last-Event-ID`, pairs the cursor with `X-Qwen-Event-Epoch` when the daemon supplies an epoch, validates framing and event IDs, deduplicates replay, and reconnects only the SSE GET. It learns an epoch from the validated prompt admission or SSE response headers and fails closed if the epoch changes during prompt observation. Older daemons that omit both surfaces remain compatible but retain their numeric-only stale-cursor detection. Prompt and other mutation requests are never retried automatically. HTTP 408 and 5xx responses to prompt admission, session creation, permission, cancel, heartbeat, detach, or delete are reported as outcome-unknown because they do not prove that the daemon rejected the mutation. Finite response bodies and SSE observation have independent deadlines. Creation-time model selection is intentionally not exposed by the Java daemon SDK API in this alpha. The daemon reports a rejected `modelServiceId` only as an SSE event emitted before the create response, while this SDK opens its stream from the later prompt-admission watermark. Until the daemon returns a definitive create result or the SDK owns a separate session-event subscription from `Last-Event-ID: 0`, use the daemon's configured default model. @@ -184,7 +184,7 @@ other examples see src/test/java/com/alibaba/qwen/code/cli/example `0.1.0-alpha` raises the minimum Java version for the whole artifact from 8 to 11. Java 8 applications must remain on `0.0.3-alpha`. Logback is no longer a runtime dependency; add the SLF4J provider your application uses. -This alpha deliberately fails closed when it cannot prove a prompt terminal. It does not guarantee exactly-once execution across daemon restarts, automatic epoch recovery, snapshot/resync, persisted cursors, or true prompt-ID-targeted cancellation. `prompt_cancelled` and queue events are advisory; only matching `turn_complete` and `turn_error` are terminal. +This alpha deliberately fails closed when it cannot prove a prompt terminal. It detects a daemon event-epoch change during an observed prompt but does not automatically recover from it. It does not guarantee exactly-once execution across daemon restarts, automatic epoch recovery, snapshot/resync, persisted cursors, or true prompt-ID-targeted cancellation. `prompt_cancelled` and queue events are advisory; only matching `turn_complete` and `turn_error` are terminal. If session creation has an ambiguous transport outcome, the daemon may retain a session whose ID never reached the caller. The SDK does not retry creation and cannot detach that unknown session; daemon-side lifecycle reaping is the recovery boundary. diff --git a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/DaemonClient.java b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/DaemonClient.java index f5f928fa97c..d0b933c557a 100644 --- a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/DaemonClient.java +++ b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/DaemonClient.java @@ -37,6 +37,8 @@ /** Java 11 client for the {@code qwen serve} REST and SSE transport. */ public final class DaemonClient implements AutoCloseable { + static final String EVENT_EPOCH_HEADER = "X-Qwen-Event-Epoch"; + private static final AtomicLong CLIENT_SEQUENCE = new AtomicLong(); private final String baseUrl; @@ -303,18 +305,23 @@ HttpSupport.Response sendDelete(String path, String clientId) } HttpResponse openSse(String path, String clientId, long lastEventId, - Duration observationRemaining) + String eventEpoch, Duration observationRemaining) throws IOException, InterruptedException { - HttpRequest request = requestBuilder(path, clientId) + HttpRequest.Builder request = requestBuilder(path, clientId) .header("Accept", "text/event-stream") .header("Accept-Encoding", "identity") .header("Cache-Control", "no-cache") - .header("Last-Event-ID", Long.toString(lastEventId)) + .header("Last-Event-ID", Long.toString(lastEventId)); + if (eventEpoch != null) { + request.header(EVENT_EPOCH_HEADER, eventEpoch); + } + HttpRequest builtRequest = request .timeout(shorter(requestTimeout, observationRemaining)) .GET() .build(); try { - return httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream()); + return httpClient.send(builtRequest, + HttpResponse.BodyHandlers.ofInputStream()); } catch (RejectedExecutionException e) { throw new IOException("HTTP executor is saturated", e); } diff --git a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/DaemonSessionClient.java b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/DaemonSessionClient.java index 1f95b9408a9..0410060f32c 100644 --- a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/DaemonSessionClient.java +++ b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/DaemonSessionClient.java @@ -9,6 +9,7 @@ import java.time.format.DateTimeFormatter; import java.time.format.DateTimeParseException; import java.util.Collections; +import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; @@ -24,6 +25,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; +import java.util.regex.Pattern; /** One independently attached daemon session. */ public final class DaemonSessionClient implements AutoCloseable { @@ -39,6 +41,8 @@ public final class DaemonSessionClient implements AutoCloseable { "slow_client_warning", "replay_complete"); private static final Set RETRYABLE_SSE_STATUS = Set.of( 408, 429, 500, 502, 503, 504); + private static final Pattern EVENT_EPOCH_PATTERN = Pattern.compile( + "[A-Za-z0-9_-]{1,64}"); private final DaemonClient client; private final DaemonSession session; @@ -641,7 +645,10 @@ private PromptAcceptance admit() { return new PromptAcceptance( JsonSupport.requiredString(json, "promptId", "prompt admission"), JsonSupport.requiredNonNegativeLong(json, "lastEventId", - "prompt admission")); + "prompt admission"), + validateEventEpoch( + JsonSupport.optionalString(json, "eventEpoch"), + "prompt admission.eventEpoch")); } catch (DaemonProtocolException e) { throw new PromptAdmissionUnknownException(e); } @@ -652,6 +659,7 @@ private PromptTerminal observe(PromptAcceptance admitted) { client.promptObservationTimeout()); long deadline = deadlineAfter(timeout); long cursor = admitted.getLastEventId(); + String eventEpoch = admitted.getEventEpoch(); int consecutiveFailures = 0; Duration serverRetry = null; while (true) { @@ -661,7 +669,7 @@ private PromptTerminal observe(PromptAcceptance admitted) { HttpResponse response; try { response = client.openSse(sessionPath() + "/events", - session.getClientId(), cursor, + session.getClientId(), cursor, eventEpoch, Duration.ofMillis(remainingMillis(deadline))); } catch (IOException e) { reconnectOrThrow(++consecutiveFailures, @@ -719,6 +727,15 @@ private PromptTerminal observe(PromptAcceptance admitted) { try { checkStoppedOrExpired(deadline); validateSseHeaders(response.headers()); + String responseEpoch = responseEventEpoch(response.headers()); + if (eventEpoch != null && responseEpoch != null + && !eventEpoch.equals(responseEpoch)) { + throw new DaemonProtocolException( + "SSE event epoch changed during prompt observation"); + } + if (responseEpoch != null) { + eventEpoch = responseEpoch; + } watchdog = scheduleIdleWatchdog(stream, lastActivity, idleClosed); deadlineWatchdog = scheduleDeadlineWatchdog(deadline); @@ -1059,6 +1076,30 @@ private static void validateSseHeaders(HttpHeaders headers) { } } + private static String responseEventEpoch(HttpHeaders headers) { + List values = headers.allValues(DaemonClient.EVENT_EPOCH_HEADER); + if (values.isEmpty()) { + return null; + } + if (values.size() != 1) { + throw new DaemonProtocolException( + "SSE response contained multiple event epoch headers"); + } + return validateEventEpoch(values.get(0), + "SSE response " + DaemonClient.EVENT_EPOCH_HEADER); + } + + private static String validateEventEpoch(String epoch, String context) { + if (epoch == null) { + return null; + } + if (!EVENT_EPOCH_PATTERN.matcher(epoch).matches()) { + throw new DaemonProtocolException(context + + " must match [A-Za-z0-9_-]{1,64}"); + } + return epoch; + } + private static Duration retryAfter(HttpHeaders headers) { String value = headers.firstValue("Retry-After").orElse(null); if (value == null) { diff --git a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/HttpSupport.java b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/HttpSupport.java index 515a3671ca0..67bf7bce083 100644 --- a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/HttpSupport.java +++ b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/HttpSupport.java @@ -28,10 +28,26 @@ static HttpResponse.BodyHandler bodyHandler() { } static Response consume(HttpResponse response, String operation) { - Body body = response.body(); - byte[] bytes = body.getBytes(); boolean success = response.statusCode() >= 200 && response.statusCode() < 300; + List contentEncodings = response.headers() + .allValues("Content-Encoding"); + if (contentEncodings.size() > 1 + || (contentEncodings.size() == 1 + && !"identity".equalsIgnoreCase( + contentEncodings.get(0).trim()))) { + String diagnostic = operation + + " response used unsupported Content-Encoding: " + + String.join(", ", contentEncodings); + if (!success) { + return new Response(response.statusCode(), + "Response body unavailable: " + diagnostic); + } + throw new DaemonProtocolException( + diagnostic); + } + Body body = response.body(); + byte[] bytes = body.getBytes(); if (body.isOverflow()) { if (success) { throw new DaemonProtocolException("JSON response exceeds " diff --git a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/JsonSupport.java b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/JsonSupport.java index 366d19dc67d..e2667ee3b00 100644 --- a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/JsonSupport.java +++ b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/JsonSupport.java @@ -5,6 +5,7 @@ import com.fasterxml.jackson.core.JsonParser; import com.fasterxml.jackson.core.JsonToken; import com.fasterxml.jackson.core.StreamReadFeature; +import com.fasterxml.jackson.core.io.JsonEOFException; import java.io.IOException; import java.math.BigDecimal; import java.math.BigInteger; @@ -45,34 +46,46 @@ static Map parseObject(String json, String context) { private static Map readObject(JsonParser parser, String context) throws IOException { Map result = new LinkedHashMap<>(); - while (parser.nextToken() != JsonToken.END_OBJECT) { - if (parser.currentToken() != JsonToken.FIELD_NAME) { + while (true) { + JsonToken fieldToken = nextContainerToken(parser, context, "object"); + if (fieldToken == JsonToken.END_OBJECT) { + return result; + } + if (fieldToken != JsonToken.FIELD_NAME) { throw new DaemonProtocolException(context + " contains a malformed JSON object"); } String field = parser.currentName(); - JsonToken valueToken = parser.nextToken(); - if (valueToken == null) { - throw new DaemonProtocolException(context - + " contains an incomplete JSON object"); - } + JsonToken valueToken = nextContainerToken(parser, context, "object"); result.put(field, readValue(parser, valueToken, context)); } - return result; } private static List readArray(JsonParser parser, String context) throws IOException { List result = new ArrayList<>(); - JsonToken token; - while ((token = parser.nextToken()) != JsonToken.END_ARRAY) { - if (token == null) { - throw new DaemonProtocolException(context - + " contains an incomplete JSON array"); + while (true) { + JsonToken token = nextContainerToken(parser, context, "array"); + if (token == JsonToken.END_ARRAY) { + return result; } result.add(readValue(parser, token, context)); } - return result; + } + + private static JsonToken nextContainerToken(JsonParser parser, String context, + String container) throws IOException { + try { + JsonToken token = parser.nextToken(); + if (token != null) { + return token; + } + } catch (JsonEOFException e) { + throw new DaemonProtocolException(context + + " contains an incomplete JSON " + container, e); + } + throw new DaemonProtocolException(context + + " contains an incomplete JSON " + container); } private static Object readValue(JsonParser parser, JsonToken token, diff --git a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/PromptAcceptance.java b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/PromptAcceptance.java index c3259501e74..98c80478d49 100644 --- a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/PromptAcceptance.java +++ b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/PromptAcceptance.java @@ -4,10 +4,12 @@ public final class PromptAcceptance { private final String promptId; private final long lastEventId; + private final String eventEpoch; - PromptAcceptance(String promptId, long lastEventId) { + PromptAcceptance(String promptId, long lastEventId, String eventEpoch) { this.promptId = promptId; this.lastEventId = lastEventId; + this.eventEpoch = eventEpoch; } public String getPromptId() { @@ -17,4 +19,12 @@ public String getPromptId() { public long getLastEventId() { return lastEventId; } + + /** + * Returns the event-bus epoch paired with {@link #getLastEventId()}, or + * {@code null} when connected to an older daemon. + */ + public String getEventEpoch() { + return eventEpoch; + } } diff --git a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/SseReader.java b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/SseReader.java index a6893a652e8..af45a9932d2 100644 --- a/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/SseReader.java +++ b/packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/SseReader.java @@ -14,6 +14,7 @@ final class SseReader { private final BufferedInputStream input; private final int maximumFrameBytes; private Long retryMillis; + private boolean firstLine = true; SseReader(InputStream input, int maximumFrameBytes, Runnable activity) { this.input = new BufferedInputStream(new ActivityInputStream(input, activity)); @@ -27,16 +28,22 @@ Frame next() throws IOException { boolean hasData = false; int frameBytes = 0; while (true) { - byte[] line = readLine(); + Line line = readLine(); if (line == null) { return null; } - frameBytes += line.length + 1; + frameBytes += line.wireBytes; if (frameBytes > maximumFrameBytes) { throw new DaemonProtocolException("SSE frame exceeds " + maximumFrameBytes + " bytes"); } - String decoded = decode(line); + String decoded = decode(line.bytes); + if (firstLine) { + firstLine = false; + if (decoded.startsWith("\uFEFF")) { + decoded = decoded.substring(1); + } + } if (decoded.isEmpty()) { if (!hasData) { event = null; @@ -115,22 +122,24 @@ public int read(byte[] bytes, int offset, int length) throws IOException { } } - private byte[] readLine() throws IOException { + private Line readLine() throws IOException { ByteArrayOutputStream line = new ByteArrayOutputStream(); while (true) { int next = input.read(); if (next < 0) { return null; } - if (next == '\n') { - byte[] bytes = line.toByteArray(); - if (bytes.length > 0 && bytes[bytes.length - 1] == '\r') { - byte[] withoutCarriageReturn = new byte[bytes.length - 1]; - System.arraycopy(bytes, 0, withoutCarriageReturn, 0, - withoutCarriageReturn.length); - return withoutCarriageReturn; + if (next == '\r') { + input.mark(1); + int following = input.read(); + if (following >= 0 && following != '\n') { + input.reset(); } - return bytes; + return new Line(line.toByteArray(), line.size() + + (following == '\n' ? 2 : 1)); + } + if (next == '\n') { + return new Line(line.toByteArray(), line.size() + 1); } line.write(next); if (line.size() > maximumFrameBytes) { @@ -140,6 +149,16 @@ private byte[] readLine() throws IOException { } } + private static final class Line { + private final byte[] bytes; + private final int wireBytes; + + Line(byte[] bytes, int wireBytes) { + this.bytes = bytes; + this.wireBytes = wireBytes; + } + } + private static String decode(byte[] bytes) { try { return StandardCharsets.UTF_8.newDecoder() @@ -152,7 +171,7 @@ private static String decode(byte[] bytes) { } private static Long parseId(String value) { - if (value.isEmpty() || !value.chars().allMatch(Character::isDigit)) { + if (value.isEmpty() || !isAsciiDigits(value)) { throw new DaemonProtocolException("SSE id must be a positive integer"); } try { @@ -174,7 +193,7 @@ private static Long parseRetry(String value) { if (value.isEmpty()) { return null; } - if (!value.chars().allMatch(Character::isDigit)) { + if (!isAsciiDigits(value)) { throw new DaemonProtocolException("SSE retry must be a non-negative integer"); } try { @@ -184,6 +203,11 @@ private static Long parseRetry(String value) { } } + private static boolean isAsciiDigits(String value) { + return value.chars().allMatch(character -> + character >= '0' && character <= '9'); + } + static final class Frame { private final String event; private final Long id; diff --git a/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/DaemonServeE2ETest.java b/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/DaemonServeE2ETest.java index 895fe264c82..b7ff615f711 100644 --- a/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/DaemonServeE2ETest.java +++ b/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/DaemonServeE2ETest.java @@ -62,9 +62,13 @@ public void onEvent(DaemonEvent event) { } }; - PromptTerminal terminal = session.startPrompt(PromptRequest.text( + PromptCall call = session.startPrompt(PromptRequest.text( "Create the requested test file, then report completion."), - observer).completionFuture() + observer); + PromptAcceptance acceptance = call.acceptanceFuture() + .orTimeout(5, TimeUnit.SECONDS).join(); + assertTrue(acceptance.getEventEpoch() != null); + PromptTerminal terminal = call.completionFuture() .orTimeout(30, TimeUnit.SECONDS).join(); assertEquals(PromptTerminal.Kind.COMPLETE, terminal.getKind()); assertTrue(text.toString().contains(expectedText)); diff --git a/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/DaemonSessionClientTest.java b/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/DaemonSessionClientTest.java index cd964d11b87..f8a5b93f6d6 100644 --- a/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/DaemonSessionClientTest.java +++ b/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/DaemonSessionClientTest.java @@ -2,6 +2,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -131,6 +132,26 @@ void exposesCapabilitiesWithoutFastjsonTypes() { } } + @Test + void rejectsCompressedRestResponse() { + server.removeContext("/capabilities"); + server.createContext("/capabilities", exchange -> { + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.getResponseHeaders().set("Content-Encoding", "gzip"); + byte[] body = "{\"v\":1}".getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + + try (DaemonClient daemon = newClient()) { + DaemonProtocolException failure = assertThrows( + DaemonProtocolException.class, daemon::capabilities); + assertTrue(failure.getMessage().contains( + "unsupported Content-Encoding")); + } + } + @Test void rejectsCredentialsInBaseUri() { IllegalArgumentException failure = assertThrows( @@ -342,6 +363,196 @@ public void onText(String text, DaemonEvent event) { assertEquals(List.of("0", "1"), cursors); } + @Test + void sendsAdmissionEpochOnEverySseConnection() { + AtomicInteger subscriptions = new AtomicInteger(); + List cursors = new ArrayList<>(); + List epochs = new ArrayList<>(); + server.createContext("/session/session-1/prompt", exchange -> + sendJson(exchange, 202, + "{\"promptId\":\"prompt-1\",\"lastEventId\":0," + + "\"eventEpoch\":\"epoch-admission\"}")); + server.createContext("/session/session-1/events", exchange -> { + cursors.add(exchange.getRequestHeaders().getFirst("Last-Event-ID")); + epochs.add(exchange.getRequestHeaders() + .getFirst("X-Qwen-Event-Epoch")); + if (subscriptions.incrementAndGet() == 1) { + sendSseWithEpoch(exchange, textEvent(1, "one"), + "epoch-admission"); + } else { + sendSseWithEpoch(exchange, terminalEvent(2), + "epoch-admission"); + } + }); + server.createContext("/session/session-1/detach", noContent()); + + try (DaemonClient daemon = newClient(); + DaemonSessionClient session = daemon.createSession()) { + PromptCall call = session.startPrompt(PromptRequest.text("go"), + PromptObserver.NOOP); + PromptAcceptance acceptance = call.acceptanceFuture().join(); + assertEquals("epoch-admission", acceptance.getEventEpoch()); + assertEquals(PromptTerminal.Kind.COMPLETE, + call.completionFuture().join().getKind()); + } + assertEquals(List.of("0", "1"), cursors); + assertEquals(List.of("epoch-admission", "epoch-admission"), epochs); + } + + @Test + void learnsEpochFromSseResponseAndSendsItOnReconnect() { + AtomicInteger subscriptions = new AtomicInteger(); + List epochs = new ArrayList<>(); + server.createContext("/session/session-1/prompt", exchange -> + sendJson(exchange, 202, + "{\"promptId\":\"prompt-1\",\"lastEventId\":0}")); + server.createContext("/session/session-1/events", exchange -> { + epochs.add(exchange.getRequestHeaders() + .getFirst("X-Qwen-Event-Epoch")); + if (subscriptions.incrementAndGet() == 1) { + sendSseWithEpoch(exchange, textEvent(1, "one"), + "epoch-learned"); + } else { + sendSseWithEpoch(exchange, terminalEvent(2), + "epoch-learned"); + } + }); + server.createContext("/session/session-1/detach", noContent()); + + try (DaemonClient daemon = newClient(); + DaemonSessionClient session = daemon.createSession()) { + PromptCall call = session.startPrompt(PromptRequest.text("go"), + PromptObserver.NOOP); + assertNull(call.acceptanceFuture().join().getEventEpoch()); + assertEquals(PromptTerminal.Kind.COMPLETE, + call.completionFuture().join().getKind()); + } + assertEquals(2, epochs.size()); + assertNull(epochs.get(0)); + assertEquals("epoch-learned", epochs.get(1)); + } + + @Test + void retainsKnownEpochWhenSseResponseOmitsHeader() { + AtomicInteger subscriptions = new AtomicInteger(); + List epochs = new ArrayList<>(); + server.createContext("/session/session-1/prompt", exchange -> + sendJson(exchange, 202, + "{\"promptId\":\"prompt-1\",\"lastEventId\":0," + + "\"eventEpoch\":\"epoch-known\"}")); + server.createContext("/session/session-1/events", exchange -> { + epochs.add(exchange.getRequestHeaders() + .getFirst("X-Qwen-Event-Epoch")); + if (subscriptions.incrementAndGet() == 1) { + sendSse(exchange, textEvent(1, "one")); + } else { + sendSse(exchange, terminalEvent(2)); + } + }); + server.createContext("/session/session-1/detach", noContent()); + + try (DaemonClient daemon = newClient(); + DaemonSessionClient session = daemon.createSession()) { + assertEquals(PromptTerminal.Kind.COMPLETE, + session.promptText("go").getTerminal().getKind()); + } + assertEquals(List.of("epoch-known", "epoch-known"), epochs); + } + + @Test + void preventsStaleCursorFromAcceptingNewEpochSuffix() { + AtomicReference requestEpoch = new AtomicReference<>(); + server.createContext("/session/session-1/prompt", exchange -> + sendJson(exchange, 202, + "{\"promptId\":\"prompt-1\",\"lastEventId\":2," + + "\"eventEpoch\":\"epoch-old\"}")); + server.createContext("/session/session-1/events", exchange -> { + requestEpoch.set(exchange.getRequestHeaders() + .getFirst("X-Qwen-Event-Epoch")); + String events = "epoch-old".equals(requestEpoch.get()) + ? "event: state_resync_required\ndata: {\"v\":1," + + "\"type\":\"state_resync_required\",\"data\":{" + + "\"reason\":\"epoch_reset\"," + + "\"detail\":\"epoch_mismatch\"}}\n\n" + : terminalEvent(3); + sendSseWithEpoch(exchange, events, "epoch-new"); + }); + server.createContext("/session/session-1/detach", noContent()); + + try (DaemonClient daemon = newClient(); + DaemonSessionClient session = daemon.createSession()) { + PromptOutcomeIndeterminateException failure = assertThrows( + PromptOutcomeIndeterminateException.class, + () -> session.promptText("go")); + assertTrue(failure.getMessage().contains("event epoch changed")); + } + assertEquals("epoch-old", requestEpoch.get()); + } + + @Test + void failsClosedOnMalformedSseResponseEpoch() { + server.createContext("/session/session-1/prompt", exchange -> + sendJson(exchange, 202, + "{\"promptId\":\"prompt-1\",\"lastEventId\":0}")); + server.createContext("/session/session-1/events", exchange -> + sendSseWithEpoch(exchange, terminalEvent(1), "invalid epoch")); + server.createContext("/session/session-1/detach", noContent()); + + try (DaemonClient daemon = newClient(); + DaemonSessionClient session = daemon.createSession()) { + PromptOutcomeIndeterminateException failure = assertThrows( + PromptOutcomeIndeterminateException.class, + () -> session.promptText("go")); + assertTrue(failure.getMessage() + .contains("X-Qwen-Event-Epoch must match")); + } + } + + @Test + void failsClosedOnDuplicateSseResponseEpochHeaders() { + server.createContext("/session/session-1/prompt", exchange -> + sendJson(exchange, 202, + "{\"promptId\":\"prompt-1\",\"lastEventId\":0}")); + server.createContext("/session/session-1/events", exchange -> { + byte[] bytes = ("retry: 0\n\n" + terminalEvent(1)) + .getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", + "text/event-stream"); + exchange.getResponseHeaders().add("X-Qwen-Event-Epoch", "epoch-a"); + exchange.getResponseHeaders().add("X-Qwen-Event-Epoch", "epoch-b"); + exchange.sendResponseHeaders(200, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + }); + server.createContext("/session/session-1/detach", noContent()); + + try (DaemonClient daemon = newClient(); + DaemonSessionClient session = daemon.createSession()) { + PromptOutcomeIndeterminateException failure = assertThrows( + PromptOutcomeIndeterminateException.class, + () -> session.promptText("go")); + assertTrue(failure.getMessage() + .contains("multiple event epoch headers")); + } + } + + @Test + void rejectsMalformedAdmissionEpochAsUnknownAdmission() { + server.createContext("/session/session-1/prompt", exchange -> + sendJson(exchange, 202, + "{\"promptId\":\"prompt-1\",\"lastEventId\":0," + + "\"eventEpoch\":\"invalid epoch\"}")); + server.createContext("/session/session-1/detach", noContent()); + + try (DaemonClient daemon = newClient(); + DaemonSessionClient session = daemon.createSession()) { + PromptAdmissionUnknownException failure = assertThrows( + PromptAdmissionUnknownException.class, + () -> session.promptText("go")); + assertInstanceOf(DaemonProtocolException.class, failure.getCause()); + } + } + @Test void retainsServerRetryDelayAcrossSseConnections() { AtomicInteger subscriptions = new AtomicInteger(); @@ -2490,6 +2701,76 @@ void retryableHttpPromptFailureIsAdmissionUnknownAndNotRetried() { assertEquals(1, prompts.get()); } + @Test + void compressedRetryableHttpPromptFailureIsAdmissionUnknownAndNotRetried() { + AtomicInteger prompts = new AtomicInteger(); + server.createContext("/session/session-1/prompt", exchange -> { + prompts.incrementAndGet(); + sendEncodedJson(exchange, 502, "gzip", + "{\"error\":\"bad gateway\"}"); + }); + server.createContext("/session/session-1/detach", noContent()); + + try (DaemonClient daemon = newClient(); + DaemonSessionClient session = daemon.createSession()) { + PromptCall call = session.startPrompt(PromptRequest.text("go"), + PromptObserver.NOOP); + CompletionException failure = assertThrows(CompletionException.class, + () -> call.acceptanceFuture().join()); + PromptAdmissionUnknownException admissionFailure = assertInstanceOf( + PromptAdmissionUnknownException.class, failure.getCause()); + DaemonHttpException cause = assertInstanceOf(DaemonHttpException.class, + admissionFailure.getCause()); + assertEquals(502, cause.getStatusCode()); + assertThrows(PromptAlreadyActiveException.class, + () -> session.startPrompt( + PromptRequest.text("unsafe-reuse"), + PromptObserver.NOOP)); + } + assertEquals(1, prompts.get()); + } + + @Test + void compressedPromptAdmissionIsOutcomeUnknown() { + server.createContext("/session/session-1/prompt", exchange -> + sendEncodedJson(exchange, 202, "gzip", + "{\"promptId\":\"prompt-1\",\"lastEventId\":0}")); + server.createContext("/session/session-1/detach", noContent()); + + try (DaemonClient daemon = newClient(); + DaemonSessionClient session = daemon.createSession()) { + PromptCall call = session.startPrompt(PromptRequest.text("go"), + PromptObserver.NOOP); + CompletionException failure = assertThrows(CompletionException.class, + () -> call.acceptanceFuture().join()); + PromptAdmissionUnknownException admissionFailure = assertInstanceOf( + PromptAdmissionUnknownException.class, failure.getCause()); + assertInstanceOf(DaemonProtocolException.class, + admissionFailure.getCause()); + } + } + + @Test + void compressedDefinitivePromptRejectionRemainsHttpError() { + server.createContext("/session/session-1/prompt", exchange -> + sendEncodedJson(exchange, 409, "gzip", + "{\"error\":\"conflict\"}")); + server.createContext("/session/session-1/detach", noContent()); + + try (DaemonClient daemon = newClient(); + DaemonSessionClient session = daemon.createSession()) { + PromptCall call = session.startPrompt(PromptRequest.text("go"), + PromptObserver.NOOP); + CompletionException failure = assertThrows(CompletionException.class, + () -> call.acceptanceFuture().join()); + DaemonHttpException cause = assertInstanceOf(DaemonHttpException.class, + failure.getCause()); + assertEquals(409, cause.getStatusCode()); + assertTrue(cause.getResponseBody().contains( + "unsupported Content-Encoding")); + } + } + @Test void definitivePromptRejectionRemainsHttpError() { AtomicInteger prompts = new AtomicInteger(); @@ -2581,10 +2862,28 @@ private static void sendJson(HttpExchange exchange, int status, String body) exchange.close(); } + private static void sendEncodedJson(HttpExchange exchange, int status, + String contentEncoding, String body) throws IOException { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.getResponseHeaders().set("Content-Encoding", contentEncoding); + exchange.sendResponseHeaders(status, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + } + private static void sendSse(HttpExchange exchange, String events) throws IOException { + sendSseWithEpoch(exchange, events, null); + } + + private static void sendSseWithEpoch(HttpExchange exchange, String events, + String epoch) throws IOException { byte[] bytes = ("retry: 0\n\n" + events).getBytes(StandardCharsets.UTF_8); exchange.getResponseHeaders().set("Content-Type", "text/event-stream"); + if (epoch != null) { + exchange.getResponseHeaders().set("X-Qwen-Event-Epoch", epoch); + } exchange.sendResponseHeaders(200, bytes.length); exchange.getResponseBody().write(bytes); exchange.close(); diff --git a/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/JsonSupportTest.java b/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/JsonSupportTest.java index dee9dd84d19..0cfdd4c38a7 100644 --- a/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/JsonSupportTest.java +++ b/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/JsonSupportTest.java @@ -30,6 +30,15 @@ void parsesStrictJsonWithoutLosingIntegerPrecision() { assertEquals("9223372036854775808", parsed.get("value").toString()); } + @Test + void identifiesTruncatedObjectAsIncomplete() { + DaemonProtocolException failure = assertThrows( + DaemonProtocolException.class, + () -> JsonSupport.parseObject("{\"value\":1", "test response")); + assertEquals("test response contains an incomplete JSON object", + failure.getMessage()); + } + @Test void rejectsNonFiniteJsonNumbersRecursively() { for (Number value : new Number[] { diff --git a/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/SseReaderTest.java b/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/SseReaderTest.java index 4a02b169825..a22906dc09a 100644 --- a/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/SseReaderTest.java +++ b/packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/SseReaderTest.java @@ -28,6 +28,29 @@ void parsesCrLfCommentsAndMultiLineData() throws Exception { assertEquals(1, activity.get()); } + @Test + void parsesBareCarriageReturnLineEndings() throws Exception { + String input = "id: 7\revent: update\rdata: {}\r\r"; + SseReader reader = new SseReader(new ByteArrayInputStream( + input.getBytes(StandardCharsets.UTF_8)), 1024, () -> { + }); + + SseReader.Frame frame = reader.next(); + assertEquals(7L, frame.getId()); + assertEquals("update", frame.getEvent()); + assertEquals("{}", frame.getData()); + } + + @Test + void ignoresLeadingUtf8Bom() throws Exception { + String input = "\uFEFFid: 7\ndata: {}\n\n"; + SseReader reader = new SseReader(new ByteArrayInputStream( + input.getBytes(StandardCharsets.UTF_8)), 1024, () -> { + }); + + assertEquals(7L, reader.next().getId()); + } + @Test void rejectsMalformedUtf8() { byte[] invalid = new byte[] {'d', 'a', 't', 'a', ':', ' ', (byte) 0xC3, @@ -37,6 +60,19 @@ void rejectsMalformedUtf8() { assertThrows(DaemonProtocolException.class, reader::next); } + @Test + void rejectsUnicodeDigitsInIdAndRetry() { + for (String input : new String[] { + "id: \u0661\ndata: {}\n\n", + "retry: \u0661\n\nid: 1\ndata: {}\n\n" + }) { + SseReader reader = new SseReader(new ByteArrayInputStream( + input.getBytes(StandardCharsets.UTF_8)), 1024, () -> { + }); + assertThrows(DaemonProtocolException.class, reader::next); + } + } + @Test void rejectsOversizedFrame() { String input = "data: " + "x".repeat(1024) + "\n\n"; @@ -45,6 +81,15 @@ void rejectsOversizedFrame() { assertThrows(DaemonProtocolException.class, reader::next); } + @Test + void countsBothBytesOfCrLfTowardFrameLimit() { + SseReader reader = new SseReader(new ByteArrayInputStream( + "data: x\r\n\r\n".getBytes(StandardCharsets.UTF_8)), + 10, () -> { + }); + assertThrows(DaemonProtocolException.class, reader::next); + } + @Test void retainsRetryDirectiveAcrossRetryOnlyFrame() throws Exception { SseReader reader = new SseReader(new ByteArrayInputStream( diff --git a/scripts/run-java-daemon-sdk-e2e.ts b/scripts/run-java-daemon-sdk-e2e.ts index 6e783016a49..d336e5344de 100644 --- a/scripts/run-java-daemon-sdk-e2e.ts +++ b/scripts/run-java-daemon-sdk-e2e.ts @@ -1,4 +1,4 @@ -import { spawn } from 'node:child_process'; +import { spawn, type ChildProcess } from 'node:child_process'; import { existsSync, mkdirSync, @@ -32,6 +32,7 @@ const cancelPrompt = 'java daemon e2e cancel sentinel'; const teardownPrompt = 'java daemon e2e teardown sentinel'; const directPrompt = 'java daemon e2e direct response'; const token = 'java-daemon-e2e-token'; +const javaTestTimeoutMs = 5 * 60_000; mkdirSync(workspace, { recursive: true }); mkdirSync(path.join(testHome, '.qwen'), { recursive: true }); writeFileSync( @@ -110,6 +111,7 @@ const daemon = spawn( ], { cwd: root, + detached: process.platform !== 'win32', env: { ...cleanEnvironment, HOME: testHome, @@ -125,54 +127,106 @@ const daemon = spawn( stdio: ['ignore', 'pipe', 'pipe'], }, ); -const daemonClosed = new Promise((resolve) => { - daemon.once('close', () => resolve()); -}); let stderr = ''; +let javaTest: ChildProcess | undefined; +let receivedSignal: NodeJS.Signals | undefined; let succeeded = false; daemon.stderr?.on('data', (chunk) => { stderr += chunk.toString(); }); -async function stopDaemon(): Promise { - if (daemon.exitCode === null && daemon.signalCode === null) { - daemon.kill('SIGTERM'); +let rejectSignal: (error: Error) => void = () => {}; +const signalFailure = new Promise((_resolve, reject) => { + rejectSignal = reject; +}); +const handleSignal = (signal: NodeJS.Signals) => { + if (receivedSignal === undefined) { + receivedSignal = signal; + rejectSignal(new Error(`Java daemon E2E interrupted by ${signal}`)); + } +}; +const handleSigint = () => handleSignal('SIGINT'); +const handleSigterm = () => handleSignal('SIGTERM'); +process.once('SIGINT', handleSigint); +process.once('SIGTERM', handleSigterm); + +async function stopChild(child: ChildProcess, name: string): Promise { + if (!processTreeExists(child)) return; + signalProcessTree(child, 'SIGTERM'); + if (await waitForProcessTreeExit(child, 5_000)) return; + signalProcessTree(child, 'SIGKILL'); + if (!(await waitForProcessTreeExit(child, 5_000))) { + throw new Error(`${name} process tree did not exit after SIGKILL`); + } +} + +function processTreeExists(child: ChildProcess): boolean { + if (child.pid === undefined) return false; + if (process.platform === 'win32') { + return child.exitCode === null && child.signalCode === null; } - const forceTimer = setTimeout(() => daemon.kill('SIGKILL'), 5_000); try { - await daemonClosed; - } finally { - clearTimeout(forceTimer); + process.kill(-child.pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code !== 'ESRCH'; } } +async function waitForProcessTreeExit( + child: ChildProcess, + timeoutMs: number, +): Promise { + const deadline = Date.now() + timeoutMs; + while (processTreeExists(child)) { + if (Date.now() >= deadline) return false; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + return true; +} + +function signalProcessTree(child: ChildProcess, signal: NodeJS.Signals): void { + if (process.platform !== 'win32' && child.pid !== undefined) { + try { + process.kill(-child.pid, signal); + return; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ESRCH') return; + } + } + child.kill(signal); +} + +let runFailure: unknown; try { - const port = await new Promise((resolve, reject) => { - const timer = setTimeout( - () => reject(new Error(`daemon startup timed out\n${stderr}`)), - 15_000, - ); - let output = ''; - daemon.stdout?.on('data', (chunk) => { - output += chunk.toString(); - const match = output.match(/listening on http:\/\/127\.0\.0\.1:(\d+)/); - if (match) { + const port = await Promise.race([ + new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error(`daemon startup timed out\n${stderr}`)), + 15_000, + ); + let output = ''; + daemon.stdout?.on('data', (chunk) => { + output += chunk.toString(); + const match = output.match(/listening on http:\/\/127\.0\.0\.1:(\d+)/); + if (!match) return; clearTimeout(timer); resolve(Number(match[1])); - } - }); - daemon.once('error', (error) => { - clearTimeout(timer); - reject(error); - }); - daemon.once('exit', (code) => { - clearTimeout(timer); - reject(new Error(`daemon exited with ${code}\n${stderr}`)); - }); - }); + }); + daemon.once('error', (error) => { + clearTimeout(timer); + reject(error); + }); + daemon.once('exit', (code) => { + clearTimeout(timer); + reject(new Error(`daemon exited with ${code}\n${stderr}`)); + }); + }), + signalFailure, + ]); - const javaTest = spawn( + javaTest = spawn( 'mvn', [ '--batch-mode', @@ -184,6 +238,7 @@ try { ], { cwd: path.join(root, 'packages', 'sdk-java', 'qwencode'), + detached: process.platform !== 'win32', env: { ...cleanEnvironment, QWEN_DAEMON_E2E_BASE_URL: `http://127.0.0.1:${port}`, @@ -198,9 +253,26 @@ try { stdio: 'inherit', }, ); - const result = await new Promise((resolve, reject) => { - javaTest.once('error', reject); - javaTest.once('exit', resolve); + let timeout: NodeJS.Timeout | undefined; + const result = await Promise.race([ + new Promise((resolve, reject) => { + javaTest?.once('error', reject); + javaTest?.once('exit', resolve); + }), + new Promise((_resolve, reject) => { + timeout = setTimeout( + () => + reject( + new Error( + `Java daemon E2E Maven test timed out after ${javaTestTimeoutMs}ms`, + ), + ), + javaTestTimeoutMs, + ); + }), + signalFailure, + ]).finally(() => { + if (timeout !== undefined) clearTimeout(timeout); }); if (result !== 0) { const logPath = path.join( @@ -226,12 +298,47 @@ try { ); } succeeded = true; -} finally { - await stopDaemon(); - await fake.close(); - if (succeeded) { - rmSync(temporary, { recursive: true, force: true }); - } else { - console.error(`Retained Java daemon E2E state at ${temporary}`); +} catch (error) { + runFailure = error; +} + +const cleanupFailures: unknown[] = []; +if (javaTest !== undefined) { + try { + await stopChild(javaTest, 'Maven test'); + } catch (error) { + cleanupFailures.push(error); } } +try { + await stopChild(daemon, 'daemon'); +} catch (error) { + cleanupFailures.push(error); +} +try { + await fake.close(); +} catch (error) { + cleanupFailures.push(error); +} +process.off('SIGINT', handleSigint); +process.off('SIGTERM', handleSigterm); +if (succeeded && cleanupFailures.length === 0) { + rmSync(temporary, { recursive: true, force: true }); +} else { + console.error(`Retained Java daemon E2E state at ${temporary}`); +} +if (receivedSignal !== undefined) { + process.kill(process.pid, receivedSignal); +} +if (runFailure !== undefined && cleanupFailures.length > 0) { + throw new AggregateError( + [runFailure, ...cleanupFailures], + 'Java daemon E2E and cleanup both failed', + ); +} +if (runFailure !== undefined) { + throw runFailure; +} +if (cleanupFailures.length > 0) { + throw new AggregateError(cleanupFailures, 'Java daemon E2E cleanup failed'); +}