fix(sdk-java): Harden daemon transport reliability

.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."

.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

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

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.

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();

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.
 

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<InputStream> 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);
         }