fix(sdk-java): Close Hosted Harness review criticals from #12654 (#13314)

* fix(sdk-java): Close Hosted Harness review criticals from #12654 #12654 merged before its second review round was processed. This closes the 11 Critical and 2 Minor findings, each re-verified against main and pinned by a new HostedHarnessClientTest case whose mutation witness was run (reverting the fix reddens the test): - HarnessEventStream close() no longer parks behind a next() blocked in the socket read: the closed flag is atomic, the raw input is closed without any stream monitor, and HostedHarnessClient.close() completes; the blocked consumer fails with DaemonTransportException (R1-1). - The hosted SSE transport gains the idle watchdog its stdio sibling has: Builder.sseIdleTimeout defaults to 45s, and a silent peer makes next() fail with "Hosted Harness SSE idle timeout" (R1-43). - submitTurn releases the active-prompt slot on a definitive 409 and throws PromptAlreadyActiveException; retaining a marker keyed by the rejected promptId wedged the session client-side (R1-2). A definitive non-409 failure on a same-identity retry releases that slot too (F2). - getStatus() snapshots the entry before the remote read and removes only that registration, so a stale hasActivePrompt:false cannot delete a different prompt admitted in flight (R1-4). - Responses missing the X-Qwen-Harness-Boot-Id fence header on mutation paths are classified outcome-unknown (create/prompt/mutation/delete) instead of a bare DaemonProtocolException (R1-5). - A local executor rejection (saturated or terminated pool) surfaces as DaemonTransportException instead of claiming the request may have reached the daemon (R1-45). - closeSession(ref) and detachSession treat a definitive 404 as done and tear down the heartbeat timer (R1-3, R1-46). - streamEvents falls back to the session ref's validated watermark when the builder omits the resume pair, so the epoch fence actually applies (R1-44). - The shared SSE parser accepts the id-less session_snapshot frame, so StreamHarnessEvents.snapshot(true) works (R1-47). - computePayloadDigest hashes a sorted-key canonical form, so the wire dedup key is reproducible across JVM processes (R1-6). - HarnessRuntimeExecutionRecovery requires status for known outcomes at construction instead of NPE-ing in isContinuationReady() (F1). Test: HostedHarnessClientTest 34/34 (15 new); full module 181 green with 9 env-gated skips on JDK 17 and clean test green on JDK 11; 14 mutation witness arms all red. The remaining 49 Suggestion-level findings are dispositioned in their review threads and tracked in #13313. * fix: Close second-round review criticals on the Hosted Harness client Addresses the CHANGES_REQUESTED round on #13314, which execution-proved one real regression in the first commit plus five more critical defects around it: - R1-1 (the regression CI confirmed on the fault-gates lane): the digest was canonicalized but the wire `prompt` member was not, and the hosted route re-derives the digest from the received bytes — every addText submission drew `400 invalid_hosted_prompt`. toJson() now emits the canonical form, the digest reuses the JsonSupport.encode facade over that same value, and the test asserts the wire bytes and the digest are literally the same value. - R1-3: the 45s idle watchdog was inherited from the stdio transport, whose server heartbeats; the hosted events route wrote nothing while no event was committed, so healthy silent streams (tool calls, the 10-minute approval wait) were aborted. The hosted route now mirrors the primary route's keepalive comment every 15s, and Builder.sseIdleTimeout accepts Duration.ZERO for callers that own the deadline. A keepalive-only peer is covered by a new test (consumer parked in read for three idle budgets). - R1-4: an idle abort while the consumer was between next() calls surfaced IllegalStateException (not the DaemonException family). The closed guard now distinguishes who closed the stream: idle abort -> DaemonTransportException("Hosted Harness SSE idle timeout"), caller close -> IllegalStateException, matching the javadoc. - R1-5: the cause-chain scan certified any rejection-shaped IOException as "never dispatched", but delivery also runs on the caller executor, so a rejection after the server accepted the prompt dropped the active-prompt marker for an admitted turn. Only the bare RejectedExecutionException (submit-time) and the terminated-executor case now classify as saturated/known; everything mid-flight stays outcome-unknown. A deterministic delivery-phase test drives this via a package-private executor seam (park the worker, fill the queue, then answer 202). - R1-6: the missing-boot-header wrap reclassified every unfenced response as outcome-unknown, including definitive 4xx the server emits itself before the fencing middleware (413, 429). All four sites now re-derive: ambiguous (408/5xx) keeps outcome-unknown, definitive unfenced statuses surface as DaemonHttpException with their status preserved. - R1-7 (suggestion folded in, same hunk family): detachSession tears the local attachment down for every definitive answer, not just 204 and 404 — a 403/400/410 no longer leaks a heartbeat timer. - R1-13 (suggestion, adjacent): a caller queued on cursorLock when the stream closes re-checks inside the lock and gets the closed-stream signal instead of a transport error. Also: the snapshot frame comment now names the actual producer (the bridge's withSnapshot on the primary events route — never the hosted route); the StreamHarnessEvents builder javadoc documents the lastEventId fallback precedence and that snapshot(true) is a primary-transport option; the README retry section and both design-doc languages now state the canonical digest as a cross-language wire contract and record that changing the digest algorithm requires quiescing or rewriting in-flight turn rows (R1-2's deployment note); the two 10ms heartbeat tests assert liveness instead of comparing zeros; and an explicit-cursor-wins case plus a definitive boot-headerless DELETE case close the two remaining unpinned arms. Test: HostedHarnessClientTest 42/42 (full module 189 green, 9 env-gated skips) on JDK 17; clean test green on JDK 11 (2 additional version-gated skips: JDK 11 parks such sends instead of surfacing the rejection); hosted-harness-session.test.ts 181/181. Nine mutation witness arms all red (revert toJson canonicalization, forget idle signal in the closed guard, drop the in-lock re-check, unbounded cause-walk, blanket boot-header wrap at prompt and DELETE sites, detach without finally, watchdog removed with the keepalive arm staying green, ref-only resume fallback). * fix(sdk-java): close third-round review criticals on the Hosted Harness client (#13314) - submitTurn: a definitive unfenced refusal (a response without a valid boot header that is not an ambiguous 408/5xx) now releases only the active-prompt marker the call owns, in both the validateGeneration arm and the post-consume arm; a refusal produced before the route's promptId lookup can no longer delete an earlier same-identity entry whose outcome is unknown, and a refused fresh submission no longer wedges the session. - send: snapshot the executor termination state before dispatch so a close() racing an in-flight send keeps its outcome-unknown classification instead of certifying non-dispatch. - StreamHarnessEvents: an explicit lastEventId cursor now always wins, including a deliberate 0 (replay from the beginning), which HarnessCoordinator passes for turn rows without a recorded watermark; only an omitted cursor falls back to the session ref's watermark. - HarnessEventStream: the idle watchdog now measures peer silence only while a consumer is parked in next(), so time between calls (a blocking queue handoff, a slow event handler) no longer aborts a healthy keepalived stream. - docs: the payload digest is re-derived by the server from its own re-serialization of the parsed prompt member (JSON.stringify form), not from the received bytes; the README qualifies cross-release digest stability and the keepalive floor for a custom idle timeout. Co-authored-by: Qwen-Coder <[email protected]> * fix(sdk-java): close fourth-round review findings on the Hosted Harness client (#13314) - Carry the wire status and machine-readable code through PromptAlreadyActiveException instead of collapsing every 409 into a fixed message - Keep a prompt marker whose submission is still in flight from being cleared by a status answer that predates the admission - Map a watchdog scheduling rejection to DaemonTransportException and drop the stranded stream registration - Document the enforced idle-abort window (budget to 1.5x) and the peer-side backpressure limit on consumer pauses - Pin the fenced definitive-refusal release, the Duration.ZERO watchdog opt-out, the next() entry stamp, and the terminated-executor release with dedicated regression tests * fix(sdk-java): close fifth-round review findings on the Hosted Harness client (#13314) - Retain the active-prompt marker when a fenced 409 carries hosted_prompt_recovery_required: that code proves the promptId is in the durable input log, so releasing the marker would admit a second identity into a session with two unsettled inputs and fail every recovery entrance closed; every other 409 code still releases it - Clear the hosted events keepalive from stop() as well as from the close handler, and send X-Accel-Buffering: no so a buffering intermediary cannot swallow the keepalives the client watchdog relies on - Pin the settled status-read clearing, the failed-close unregister, and the disabled watchdog's empty scheduler queue with positive witnesses, and extract the shared fenced-SSE response prelude in the client tests - Correct the documented idle-abort window with the 100 ms poll floor, state the digest's sha256: wire encoding in the README, and narrow the PromptAlreadyActiveException javadoc to what the code does * fix(sdk-java): lifecycle teardown runs only on definitive answers Close the corridor the next-round review's close/detach receipts queued: local attachments (and their heartbeat timers) must not survive a definitive refusal, and must NOT be torn down on outcome-unknown. - detachSession: the unconditional `finally { removeAttachment }` also ran when requireMutationStatus surfaced unexpected-success as MutationOutcomeUnknownException — a state where the server's answer is not decided and the lease must keep beating. Teardown now runs on the definitive limb (204 or tolerated 404) and inside the DaemonHttpException arm only; outcome-unknown surfaces keep the attachment (anUnexpectedDetachAnswerKeepsTheAttachment). - closeSession(HarnessSessionRef) and closeSession(String): mirror the same try/catch — a definitive DELETE refusal (e.g. 403) now also retires the attachment and the active-prompt ledger instead of leaking the timer until close(), while ambiguous answers keep them (aRefusedCloseStillTearsDownTheHeartbeat ×2 overloads, anUnexpectedCloseAnswerKeepsTheAttachment, aBootHeaderlessCloseKeepsTheAttachment for the gateway-502 shape). - closeSession(String) gains the ensureOpen() guard every other public entry has (a deferral receipt plus #13313's R1-50): the closed-client path is JVM-independent again, so aLocalExecutorRejectionIsNotReported AsOutcomeUnknown becomes closeSessionByIdFailsFastAfterClientClose — IllegalStateException on every JDK — and the saturated/terminated taxonomy stays pinned by aTerminatedHttpExecutorReleasesTheOwnedMarker. - closeByRefTreatsAnAbsentHarnessSessionAsClosed now also counts the DELETE request (dropping the remote call reddens assertEquals(1)). - Housekeeping of the takeover branch: aRecoveryRequiredRefusalKeepsThe OwnedMarker's stub never incremented its promptCalls counter, so its trailing assertion could never pass once it ran (CI never reached it through the verify-checkout race) — fixed by counting in the handler. Witnesses: full module 214/214 green on JDK 17 (clean test green on JDK 11, 13 skips = 9 env + 4 version-gated); corridor mutation arms all red — detach with an unconditional finally flattens the keep-beating assertion, removing either close-mirror makes refused-close heartbeats keep growing (9→14, 10→16), and removing ensureOpen flips the fail-fast test from IllegalStateException back to the transport exception. * fix(sdk-java): settle registered admission markers and align lifecycle teardown Co-authored-by: Qwen-Coder <[email protected]> * fix(sdk-java): retire hosted admission state on proven restart and scope the settle latch per attempt (#13314) A proven harness generation change now retires every local record for that peer at all three entrances: the submitTurn fence catch releases the admission ledger entry, the scheduled heartbeat retires the attachment and ledger instead of swallowing the mismatch, and detachSession mirrors closeSession for the generation branch while keeping the ledger across definitive refusals. The admission settle latch is per-attempt and outcome-scoped: only the owning call's concluded send settles it, every attempt re-arms it, and a same-identity retry's 202 publishes a fresh settled entry so a status read holding a stale snapshot cannot clear the marker that is live now. Pre-dispatch unchecked escapes release the owned marker via a dispatch boundary flag in send() instead of settle-and-keep. The SSE idle watchdog merges the parked flag and activity stamp into one volatile waitingSince write, so a tick can never pair a fresh flag with a stale stamp; keepalive refills still refresh the budget. Docs correct the persisted-digest claim: a form change invalidates rows whose stored key order was not already canonical, not every row. * fix(cli): drop ambient QWEN_HOME from the cli test setup (#13314) A run from inside a Qwen Code session always exports QWEN_HOME, which outranks the homedir() mocks the settings suite anchors on: the module-load getUserSettingsPath() then resolves to the ambient home, the mocked user-settings file is never found, and 13 settings.test.ts tests fail. The verification gate runs in exactly such an environment. Delete the variable in test-setup.ts, mirroring the existing QWEN_RUNTIME_DIR sanitization; tests that exercise QWEN_HOME set it in-body. Co-authored-by: Qwen-Coder <[email protected]> * fix(sdk-java): close hosted harness ledger wedges from review (#13314) Unfenced 2xx/3xx mutation answers are now outcome-unknown instead of definitive, unfenced refusals carry the peer's buffered body, the per-attempt settle latch is owner-gated, close/detach retire only the records they observed, and wire 409s stay reachable through DaemonHttpException. Co-authored-by: Qwen-Coder <[email protected]> * fix(sdk-java): close hosted harness review criticals on digest recovery, refusal typing, and ledger identity (#13314) - SubmitHarnessTurn accepts a persisted pre-canonicalization payload digest and re-sends the prompt in the exact as-built key order that digest covers, so recovering an older turn row reaches the route's replay path instead of failing the self-check before any request is sent. - An unfenced 4xx on the generic mutation channel surfaces as the protocol failure it is rather than DaemonHttpException, whose status consumers key terminal verdicts on; only submitTurn keeps the typed refusal its admission-marker release keys on. - ActivePrompt records the submitting client id and replaces the shared settle latch with a per-attempt send counter: lifecycle teardowns retire only markers whose identity proved stale, and a status read clears a marker only while no send attempt is in flight. Co-authored-by: Qwen-Coder <[email protected]> * fix(sdk-java): close hosted harness round-10 review criticals on close/detach retirement (#13314) - README: correct the digest-migration paragraph — the client re-sends the shape its digest covers and accepts a pre-canonicalization as-built digest, so upgrades do not quiesce or rewrite in-flight turn rows - closeSession(String): a successful headerless DELETE deleted the session outright, so retire every attachment for the id, including one registered while the DELETE was in flight - submitTurn: gate all six owning-release arms on no send attempt still open, so a definitive refusal cannot strand a same-identity retry's 202 with no ledger entry - detachLifecycle: retire the observed admission marker even when no attachment remains (a refused close drops the attachment first), in both overloads - closeSession(HarnessSessionRef): a 204 proves identity-matched deletion, so retire every marker for the id; keep identity-matched retirement only for the tolerated identity-scoped 404 * fix(cli): wait out the record watcher's backoff in the unrecorded-reply squad test (#13314) The deterministic verification gate rejected the previous commit on "wakes the leader with the error when a member reply is never recorded" (default 1s vi.waitFor timed out waiting for the leader's wake). The give-up that wakes the leader is eligible the moment the test bumps its virtual clock, but it fires on the watcher's next real-timer check, and that chain backs off geometrically (recordWatchMs doubling to a 15s cap) — under full-suite load the next tick lands past the 1s default. The sibling restart-recovery test already guards the identical pattern with timeout: 3_000 and the comment "The watcher backs off; its next check may be a while."; apply the same bound here. Assertion-preserving. Probe: with the watcher's initial interval raised to 1.5s, the unmodified test fails with the gate's exact assertion at the same line; with the timeout it passes. * fix(sdk-java): close HostedHarnessClient admission-ledger fences found in review (#13314) Round-14 review found the admission-ledger fencing still open in eight places: - detachLifecycle(HarnessSessionRef) retired the pre-send snapshot even on a tolerated 404, wiping the marker the session's current owner planted through a re-issued identity; it now retires through the same identity-matched retireLedger detachSession uses, which also covers a marker this identity replanted while the detach was in flight. - detachSession collapsed a deleting 204 and a tolerated 404 into one identity-matched retirement, so a proven deletion kept a marker from an older identity; the 204 arm now retires unconditionally. - Both closeSession overloads applied the observed-snapshot rule to a generation change, so a marker published while the DELETE was in flight survived a proven restart; both now retire unconditionally on HostedHarnessGenerationException. - The sendMutation channel and the streamEvents fence observed a generation change without retiring local state; both now discard it, and sendMutation threads the session id from every caller. - The unfenced definitive-4xx arm released without reading the refusal code; an unfenced 409 is now narrowed by code like the fenced arm, so a boot-header-stripped hosted_prompt_recovery_required retains the marker and surfaces as PromptAlreadyActiveException. - When both the owning attempt and a joined same-identity retry were definitively refused, neither released the marker; an owning release refused by an open sibling send is now owed and discharged at the last concluded attempt, while outcome-unknown or durably admitted conclusions block it. - A non-owning retry's send window closed before its 202 republish, so an owning release landing in between dropped the joined entry and the replace silently no-oped; the window now closes after the ledger decision. - The dispatch guard probed the executor, so a saturated-but-running pool was misclassified outcome-unknown; send() now dispatches with sendAsync and maps the synchronous RejectedExecutionException to a known-outcome DaemonTransportException, keeping post-dispatch rejections outcome-unknown. Each fix carries a pinning test, and the three pre-dispatch escape tests move their injection seam from the removed executor-state probe to the executor's execute(). * test(cli): settle fresh process rows before identity-judged ledger sweeps (#13314) The deterministic gate rejected the previous round on managed-runtime-ledger.test.ts: on some hosts ps reports a just-born process with the negative-elapsed wraparound (e.g. 441077234-00:18:40, which parsePsElapsed rightly rejects) until it ages past the glitch, so an unwitnessed sweep judged the live fixture 'unknown' and held it unproven instead of killing it. settledGroupTable now also waits for the leader's row to carry a datable elapsed column, and the five live-spawn sweep tests that never settled now settle before sweeping. Assertions unchanged. Co-authored-by: Qwen-Coder <[email protected]> * fix(sdk-java): report a close-aborted parked SSE read as a transport failure (#13314) Co-authored-by: Qwen-Coder <[email protected]> * chore(sdk-java): re-verify #13314 on a tree clean of generated Maven artifacts The deterministic gate rejected 253326ce58 because repo-wide 'npm run lint' (eslint flat config lints .js repo-wide) swept up the JaCoCo report resources that local 'mvn test' generates under packages/sdk-java/*/target/site/jacoco/ — 139 errors in jacoco-resources/prettify.js and sort.js. Those files are git-ignored build output, not committed content; the committed tree is lint-clean. This commit carries no tree change: the repair is the cleaned working tree, with 'npm run lint' now passing, and build/typecheck plus the focused R15-1 contract tests (HarnessEventStreamCloseTest and the four HostedHarnessClientTest idle-timeout/close cases) re-verified green. Co-authored-by: Qwen-Coder <[email protected]> * fix(sdk-java): close hosted harness ledger and SSE close races - Classify a parked SSE read's raced close as DaemonTransportException, matching the interrupted-read signal, with the idle-timeout half kept distinct; the programming-error type stays reserved for callers arriving after the close. - Keep the admission ledger on every refused close, on a settled release whose entry is releaseBlocked, and for a code-less unfenced 409, which is outcome-unknown rather than a definitive refusal. - Retire the marker an owning same-identity replay just planted when the promptId's terminal event was already observed, via the settled-checked release path. - Read the getStatus marker clock before settledness and re-check both inside the bin-locked clear, so a retry's beginSend landing between the snapshot reads cannot be cleared under its open send window. - Retire only the observed attachment on a by-id close, and self-retire an attachment whose heartbeat meets a definitive hosted_session_not_found 404 instead of beating against the dead session forever. * chore(sdk-java): repair round verification on a clean working tree The deterministic gate rejected 4884dc8197 on npm run lint: the previous round's mvn test had generated the git-ignored JaCoCo HTML report under packages/sdk-java/qwencode/target/site/jacoco/jacoco-resources/*.js, and the flat eslint config lints git-ignored-but-present .js files (it does not import .gitignore), failing on the vendored prettify.js/sort.js. The root eslint config is outside this PR's footprint, so the repair removes the untracked build output from the working tree. No source change; the rejected commit's content is preserved as-is. * fix(sdk-java): retire hosted admission state on journal replays and deleting lifecycle detaches (#13314) * docs(sdk-java): state the digest canonical key order the code implements (#13314) The published digest contract claimed object keys are sorted recursively, but JsonSupport.ownKeyOrder and the route's sha256(JSON.stringify(prompt)) re-derivation use ECMAScript own-key order: array-index keys ascend numerically first, then the remaining keys in insertion order. A second-language client built on the published sentence computes a different digest and is refused with invalid_hosted_prompt. * fix(sdk-java): keep teardown retirement off live state and durable history (#13314) A detach or a DELETE closes only the live Hosted session: the durable journal survives it, and after a re-load the prompt route answers a settled promptId's same-identity replay with a 202 that owes no terminal event. Retiring the terminal-event record on teardown therefore left that replay's marker vetoing every later prompt. No teardown retires the record any more; the map is bounded instead (least recently used past 4096 sessions), and an evicted session's replay marker is cleared by the busy-turn status probe. The headerless lifecycle detach also removed the admission marker unconditionally after its attachment snapshot, so a re-load that registered a new attachment and admitted a turn inside that window lost its marker and a different prompt reached the wire. Every definitive teardown answer (detach 204 on all three entrances, the headerless tolerated 404, DELETE 204 on both overloads) now retires through one bin-locked decision that reads the current attachment under the marker's lock: a marker the current attachment's identity planted, or joined with a same-identity retry, stays; an abandoned identity's marker retires. The identity-scoped tolerated-404 retirement keeps a marker another identity's retry joined, under the same lock. Tests: the teardown cases are parameterized over all five entrances; fixtures whose vetoed submit hit an unrouted /status now answer the busy-turn probe instead of waiting out the request timeout (the class drops from ~320 s to ~20 s), and the probe's generation-change and send-clock behavior are pinned. --------- Co-authored-by: qwen-code-dev-bot <[email protected]> Co-authored-by: Qwen-Coder <[email protected]> Co-authored-by: qwen-code-ci-bot <[email protected]> Co-authored-by: qwen-code-dev-bot <[email protected]> Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: qwen-code[bot] <qwen-code[bot]@users.noreply.github.com>
2026-11-07 06:19:43 -10:00 · 2026-10-16 21:06:22 +08:00
1 parent 5929e276ed
commit d565d7463d
@@ -65,6 +65,8 @@ This foundation exports contract creation, digest validation, and an Express req

 The `qwen` Java SDK ships `HostedHarnessClient` for this protocol. Construction performs capability negotiation, validates the advertised capability digest against the configured deployment digest, and pins the process boot ID; every later request then carries the protocol and boot-ID fencing automatically, and a `hosted_harness_generation_mismatch` surfaces as a non-retryable conflict rather than a retry. Sessions are created, loaded, heartbeated, prompted, streamed, and cancelled through typed request objects, turn submissions carry a precomputed payload digest, and event streams reuse the daemon SSE reader with replay cursors. `HostedHarnessClientTest` drives the client against a stub HTTP server for negotiation, fencing, streaming, and error mapping; `ManagedHostedRuntimeE2ETest`, gated on `QWEN_MANAGED_HOSTED_E2E_BASE_URL`, exercises a real hosted profile.

+The payload digest is a cross-language wire contract, not an implementation detail: it is `sha256:<64 lowercase hex>` over a canonical form of the prompt content (object members appear in JavaScript `JSON.stringify` own-key order — array-index keys (`"0"` through `"4294967294"`, no leading zero) ascending numerically first, then the remaining keys in insertion order — applied recursively, array order preserved; the canonical form is what the client emits as the wire `prompt`), and the server re-derives the identical value by re-serializing the parsed `prompt` member the way JavaScript `JSON.stringify` does — compact separators, literal non-ASCII characters — so a client in any language must emit that member in exactly that form and hash those bytes. Changing the digest algorithm therefore changes every persisted turn identity whose stored key order was not already canonical, so the client accepts a persisted digest computed over the as-built key order when the canonical recomputation does not match, and re-sends the prompt in exactly the shape that digest covers — the route re-derives the digest over the received key order, so a legacy row still reaches its replay path instead of failing the client-side self-check before any request is sent (which would reschedule the turn without bound and leave the session refusing every new prompt as already active, with nothing naming the digest mismatch as the cause). Event streams additionally keep the daemon transport's bound on silence — a 45-second idle watchdog, configurable and disableable through `Builder.sseIdleTimeout`, that measures peer silence while a consumer is parked in `next()` and never charges the time between calls, checked every half-budget but no more often than every 100 ms, so an idle abort lands between the configured budget and the budget plus max(100 ms, half the budget) (45–67.5 s at the default) and any outer supervisor deadline should sit above that upper bound — and the hosted events route emits keepalive comments every 15 seconds so a healthy but quiet stream (long tool calls, approval waits) survives the bound, which a custom idle timeout should therefore stay comfortably above.
+
 ## Validation

 Focused unit tests verify envelope creation, random boot IDs, strict digest formatting, version negotiation, malformed and stale boot IDs, successful matching requests, response headers, and isolation from routes outside the middleware mount point.
@@ -65,6 +65,8 @@ Boot ID 是进程代际 fence，不是认证 secret 或公共标识。后续 Hos

 `qwen` Java SDK 为本协议提供 `HostedHarnessClient`。构造时完成 capability 协商，把对端发布的 capability digest 与配置的部署 digest 校验，并固定进程 boot ID；之后的每个请求自动携带协议与 boot ID fence，`hosted_harness_generation_mismatch` 以不可重试冲突而非可重试错误呈现。Session 的创建、加载、心跳、提交、事件流与取消都通过类型化请求对象进行，提交轮次携带预先计算的 payload digest，事件流复用 daemon SSE 读取器并支持续传游标。`HostedHarnessClientTest` 在 stub HTTP 服务器上验证协商、fence、流式与错误映射；`ManagedHostedRuntimeE2ETest` 以 `QWEN_MANAGED_HOSTED_E2E_BASE_URL` 为开关，验证真实 hosted profile。

+payload digest 是跨语言的线上契约，而非实现细节：它是 prompt 内容规范形式的 `sha256:<64 位小写十六进制>`（对象成员按 JavaScript `JSON.stringify` 的 own-key 顺序排列——数组索引键（`"0"` 到 `"4294967294"`，无前导零）先按数值升序排在最前，其余键随后按插入顺序排列——递归应用，数组顺序保持；客户端作为线上 `prompt` 发出的正是该规范形式），服务端会把解析后的 `prompt` 成员按 JavaScript `JSON.stringify` 的语义重新序列化（紧凑分隔符、非 ASCII 字符保持字面形式）并对这些字节取哈希，从而重算出完全相同的值——任何语言的客户端都必须以该形式发出这个成员并对这些字节取哈希。因此更改 digest 算法会改变所有存储键序本非规范形式的已持久化轮次身份；为此客户端在规范重算与存储值不匹配时，也接受按构建时键序计算的 digest，并按该 digest 所覆盖的原始键序原样重发 prompt——路由按接收到的键序重算 digest，因此旧行仍会进入重放路径，而不是在发出任何请求之前就在客户端 digest 自检处失败（那样该 turn 会被无界地重新调度，session 也会以“已有活跃 turn”为由拒绝每一个新 prompt，且没有任何信息指明 digest 不匹配是根因）。事件流还保留 daemon 传输的静默上界——45 秒空闲看门狗，可通过 `Builder.sseIdleTimeout` 配置与关闭；它度量的是消费者停在 `next()` 内期间的对端静默，两次调用之间的时间绝不计入；看门狗每半个预算检查一次，但频率不高于每 100 ms 一次，因此空闲中止发生在所配置预算与预算加 max(100 ms，半个预算) 之间（默认 45–67.5 秒），任何外层监督时限都应高于该上限——且 hosted 事件路由每 15 秒写一次 keepalive 注释，使健康但静默的流（长工具调用、审批等待）不触发该上界，因此自定义空闲上界应明显高于该间隔。
+
 ## 验证

 聚焦单元测试覆盖协议信封创建、随机 boot ID、严格 digest 格式、版本协商、非法和过期 boot ID、成功匹配请求、响应 header，以及 middleware mount point 之外路由的隔离性。
@@ -16995,6 +16995,119 @@ describe('Hosted Harness no-tool session', () => {
    }
  });

+  it('clears the keepalive timer when backpressure stops the stream', async () => {
+    const server = await app();
+    const created = await headers(supertest(server).post('/session')).send({
+      sessionId: SESSION_ID,
+      sessionScope: 'thread',
+      managedSessionStore: store(),
+    });
+    const clientId = created.body.clientId as string;
+    const prompt = [{ type: 'text', text: 'hello' }];
+    await headers(supertest(server).post(`/session/${SESSION_ID}/prompt`))
+      .set('X-Qwen-Client-Id', clientId)
+      .send({
+        prompt,
+        promptId: PROMPT_ID,
+        payloadDigest: `sha256:${createHash('sha256').update(JSON.stringify(prompt)).digest('hex')}`,
+      });
+    await vi.waitFor(async () => {
+      const transcript = await headers(
+        supertest(server).get(`/session/${SESSION_ID}/transcript`),
+      ).set('X-Qwen-Client-Id', clientId);
+      expect(transcript.body.events.length).toBeGreaterThan(2);
+    });
+
+    // Record every interval the route arms so the assertion can name the
+    // keepalive one (15s) rather than the pump one (250ms). The end() spy
+    // is a no-op so the response stays open and no 'close' event fires —
+    // the leak shape from the finding: stop() is the only teardown that
+    // runs, so it must clear both timers.
+    const armed: Array<{ handle: unknown; timeout: number | undefined }> = [];
+    const originalSetInterval = globalThis.setInterval;
+    const intervalSpy = vi
+      .spyOn(globalThis, 'setInterval')
+      .mockImplementation(function (
+        this: unknown,
+        ...args: Parameters<typeof setInterval>
+      ) {
+        const handle = originalSetInterval.apply(this, args);
+        armed.push({ handle, timeout: args[1] });
+        return handle;
+      } as typeof setInterval);
+    const clearSpy = vi.spyOn(globalThis, 'clearInterval');
+    const originalWrite = ServerResponse.prototype.write;
+    const write = vi
+      .spyOn(ServerResponse.prototype, 'write')
+      .mockImplementation(function (
+        this: ServerResponse,
+        ...args: Parameters<ServerResponse['write']>
+      ) {
+        if (typeof args[0] === 'string' && args[0].startsWith('id: ')) {
+          originalWrite.apply(this, args);
+          return false;
+        }
+        return originalWrite.apply(this, args);
+      });
+    const end = vi
+      .spyOn(ServerResponse.prototype, 'end')
+      .mockImplementation(function (this: ServerResponse) {
+        return this;
+      });
+    const listener = server;
+    const abort = new AbortController();
+    try {
+      const address = listener.address();
+      if (!address || typeof address === 'string') throw new Error('No port');
+      const stream = await fetch(
+        `http://127.0.0.1:${address.port}/session/${SESSION_ID}/events`,
+        {
+          headers: {
+            'X-Qwen-Harness-Protocol-Version': '1',
+            'X-Qwen-Harness-Boot-Id': BOOT_ID,
+            'X-Qwen-Client-Id': clientId,
+          },
+          signal: abort.signal,
+        },
+      );
+      expect(stream.status).toBe(200);
+      expect(stream.headers.get('x-accel-buffering')).toBe('no');
+      await vi.waitFor(() => {
+        expect(end).toHaveBeenCalledTimes(1);
+      });
+      const keepalive = armed.filter((entry) => entry.timeout === 15_000);
+      expect(keepalive).toHaveLength(1);
+      expect(clearSpy.mock.calls.flat()).toContain(keepalive[0]!.handle);
+    } finally {
+      write.mockRestore();
+      end.mockRestore();
+      intervalSpy.mockRestore();
+      clearSpy.mockRestore();
+      abort.abort();
+      listener.closeAllConnections();
+      // Settle the turn this test started before tearing down: while
+      // session.active is set the DELETE below answers 409
+      // hosted_turn_active and cannot close the session, so a late
+      // resource write from the still-draining turn would race
+      // afterEach's recursive rm of the scratch tree. The poll runs only
+      // after end() is restored — while the spy swallows it, a status
+      // response could never complete.
+      const settleDeadline = Date.now() + 10_000;
+      for (;;) {
+        const status = await headers(
+          supertest(server).get(`/session/${SESSION_ID}/status`),
+        ).set('X-Qwen-Client-Id', clientId);
+        if (status.body.hasActivePrompt === false) break;
+        if (Date.now() > settleDeadline) break;
+        await new Promise((resolve) => setTimeout(resolve, 25));
+      }
+      await headers(supertest(server).delete(`/session/${SESSION_ID}`)).set(
+        'X-Qwen-Client-Id',
+        clientId,
+      );
+    }
+  });
+
  it('logs the failure cause while keeping the public turn error generic', async () => {
    const log = vi
      .spyOn(stdio, 'writeStderrLineSafe')
@@ -9299,15 +9299,16 @@ export function registerHostedHarnessSessionRoutes(
      return error(res, 400, 'invalid_event_cursor');
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-store');
+    // Disable proxy buffering (nginx): the keepalives below exist to put
+    // bytes on the wire during quiet phases, and a buffering intermediary
+    // would hold them until the client's idle watchdog aborts a healthy
+    // stream. The event-stream content type alone doesn't always reach
+    // the client through every proxy.
+    res.setHeader('X-Accel-Buffering', 'no');
    res.setHeader('X-Qwen-Event-Epoch', epoch);
    res.flushHeaders();
    let cursor = after;
    let busy = false;
-    const stop = (): void => {
-      clearInterval(timer);
-      if (!res.destroyed && !res.writableEnded) res.end();
-    };
-    session.streams.add(stop);
    // Seeded once per stream, then updated as deltas flow: a committed
    // message whose text streamed must not project a second chunk.
    const streamedDeltaIds = new Set(
@@ -9353,8 +9354,28 @@ export function registerHostedHarnessSessionRoutes(
      void pump();
    }, 250);
    timer.unref();
+    // Mirror the primary route's keepalive: while no event is committed
+    // the pump writes nothing, and clients bound that silence with an
+    // idle watchdog (the Java SDK defaults to 45s), so a healthy but
+    // quiet stream must still put bytes on the wire.
+    const keepaliveTimer = setInterval(() => {
+      if (res.destroyed || res.writableEnded) return;
+      res.write(': keepalive\n\n');
+    }, 15_000);
+    keepaliveTimer.unref();
+    // stop funnels every teardown path — backpressure, pump failure,
+    // session close — through both timer clears; it is declared after
+    // the timers so it can clear them unconditionally, and the
+    // synchronous route body registers it before any path can fire.
+    const stop = (): void => {
+      clearInterval(timer);
+      clearInterval(keepaliveTimer);
+      if (!res.destroyed && !res.writableEnded) res.end();
+    };
+    session.streams.add(stop);
    res.on('close', () => {
      clearInterval(timer);
+      clearInterval(keepaliveTimer);
      session.streams.delete(stop);
    });
    void pump();
@@ -199,11 +199,23 @@ async function settledGroupTable(pgid: number): Promise<void> {
  // a busy host; an old stranger member would read the young leader's record
  // as 'ours'. Wait until the table shows this group alone before an age
  // judgement runs against it.
+  //
+  // The leader's elapsed column must also be datable: on some hosts ps
+  // reports a just-born process with the negative-elapsed wraparound (e.g.
+  // 441077234-00:18:40, which parsePsElapsed rightly rejects) until it ages
+  // past the glitch, and an undatable row judges 'unknown' — never 'ours',
+  // never 'recycled' — so a sweep that must identify the process holds it
+  // instead of killing it.
  await waitFor(() => {
    const members = [...queryProcessTable().values()].filter(
      (row) => row.pgid === pgid,
    );
-    return members.length === 1 && members[0]!.pid === pgid ? true : undefined;
+    const leader = members[0];
+    return members.length === 1 &&
+      leader!.pid === pgid &&
+      leader!.runningMs !== undefined
+      ? true
+      : undefined;
  });
 }

@@ -972,6 +984,9 @@ describe('Managed Runtime ledger', () => {
        if (!ours.pid) throw new Error('spawn failed');
        strays.add(ours);
        const oursPid = ours.pid;
+        // The sweep must date the worker's row to judge it ours; see
+        // settledGroupTable.
+        await settledGroupTable(oursPid);
        const ourFile = path.join(root, 'ours.json');
        makeLedgerFile(ourFile, { pid: oursPid });
        await sweepWorkerLedger(ourFile, {});
@@ -1088,6 +1103,9 @@ describe('Managed Runtime ledger', () => {
        worker.on('exit', () => undefined);
        if (!worker.pid) throw new Error('spawn failed');
        strays.add(worker);
+        // The sweep must date the worker's row to judge it ours; see
+        // settledGroupTable.
+        await settledGroupTable(worker.pid);
        const workFile = path.join(root, 'ledger.json');
        testInternals.writeLedgerDocument(
          workFile,
@@ -2755,6 +2773,9 @@ describe('Managed Runtime ledger', () => {
        const proc = spawnGroupLeader();
        strays.add(proc);
        const shell = proc.pid!;
+        // The sweep must date the group's row to judge it the record's; see
+        // settledGroupTable.
+        await settledGroupTable(shell);
        makeLedgerFile(path.join(directory, 'good.json'), { pid: 42424241 }, [
          { pgid: shell, startedAt: Date.now() },
        ]);
@@ -2800,6 +2821,9 @@ describe('Managed Runtime ledger', () => {
        const proc = spawnGroupLeader();
        strays.add(proc);
        const shell = proc.pid!;
+        // The sweep must date the group's row to judge it the record's; see
+        // settledGroupTable.
+        await settledGroupTable(shell);
        makeLedgerFile(
          path.join(directory, 'remaining.json'),
          { pid: 42424242 },
@@ -2937,6 +2961,9 @@ describe('Managed Runtime ledger', () => {
        const proc = spawnGroupLeader();
        strays.add(proc);
        const live = proc.pid!;
+        // The second sweep below must date the group's row to judge it the
+        // record's; see settledGroupTable.
+        await settledGroupTable(live);
        const workFile = path.join(directory, 'own.json');
        makeLedgerFile(workFile, { pid: 42424244 }, [
          { pgid: live, startedAt: Date.now() },
@@ -2080,7 +2080,10 @@ describe('SessionAgentOrchestrator squads', () => {
    expect((await h.engagement())?.active).toBe(true);

    clock += RECORD_WATCH_MAX_MS;
-    await vi.waitFor(() => expect(h.turnOf('ag_lead', 1)).toBeDefined());
+    // The watcher backs off; its next check may be a while.
+    await vi.waitFor(() => expect(h.turnOf('ag_lead', 1)).toBeDefined(), {
+      timeout: 3_000,
+    });
    const wake = h.turnOf('ag_lead', 1)!.input.prompt;
    expect(wake).toContain('Fixed in auth.ts.');
    expect(wake).toContain('disk full');
@@ -159,7 +159,7 @@ try (HostedHarnessClient harness = HostedHarnessClient.builder()
 }
 ```

-The client permits only one active prompt identity per Session while allowing different Sessions to run concurrently. A retry after `PromptAdmissionUnknownException` must reuse the same `promptId`, payload digest, prompt content, and deadline. SSE boot, epoch, or sequence changes fail closed for transcript/status reconciliation. `close()` releases local HTTP, heartbeat, timer, and stream resources only; it never destroys remote Sessions implicitly.
+The client permits only one active prompt identity per Session while allowing different Sessions to run concurrently. A retry after `PromptAdmissionUnknownException` must reuse the same `promptId`, payload digest, prompt content, and deadline. The payload digest is a cross-language contract: it is the SHA-256 of a canonical form of the prompt content, sent on the wire as `sha256:<64 lowercase hex>` (object members appear in JavaScript `JSON.stringify` own-key order — array-index keys (`"0"` through `"4294967294"`, no leading zero) ascending numerically first, then the remaining keys in insertion order — applied recursively, array order preserved), the client emits as the wire `prompt` exactly the key order its digest covers, and the server re-derives the identical value by re-serializing the parsed `prompt` member with JavaScript `JSON.stringify` semantics (compact separators, literal non-ASCII). The canonical form is versioned by the SDK: the digest is reproducible across JVM processes for a given release, but a release that changes it changes every persisted digest whose stored key order was not already canonical, so the client accepts a persisted digest computed over the as-built key order when the canonical recomputation does not match, and re-sends the prompt in exactly the shape that digest covers — a legacy row reaches its replay path instead of failing the client-side self-check. That compatibility arm covers one generation: it rescues a row written before the form changed, not a row this release writes, so a second consecutive form change still needs the stored rows quiesced. Do not rewrite a stored digest or a stored input to make it match. SSE boot, epoch, or sequence changes fail closed for transcript/status reconciliation. `close()` releases local HTTP, heartbeat, timer, and stream resources only; it never destroys remote Sessions implicitly. Event streams are bounded by a 45-second idle watchdog that measures peer silence while a consumer is parked in `next()` — time between `next()` calls is not charged (see `Builder.sseIdleTimeout`; `Duration.ZERO` disables it for callers that own the deadline). The watchdog checks every half-budget but no more often than every 100 ms, so an idle abort lands between the budget and the budget plus max(100 ms, half the budget) — 45 to ~67.5 seconds at the default — and any outer supervisor deadline should sit above that upper bound. The hosted events route emits keepalive comments every 15 seconds so legitimately silent phases — long tool calls, slow model phases, approval waits — do not trip the bound, which a custom `sseIdleTimeout` should therefore stay comfortably above.

 ## Legacy stdio API

@@ -57,6 +57,13 @@ public final class DaemonEvent {
        if (!("turn_complete".equals(type) || "turn_error".equals(type))) {
            return expectedPromptId.equals(promptId);
        }
+        return expectedPromptId.equals(terminalPromptId());
+    }
+
+    // The promptId a terminal event settles: the envelope field when
+    // present, else the data object's — null when the event carries
+    // neither.
+    String terminalPromptId() {
        Map<String, Object> object = dataObject();
        String dataPromptId = object == null ? null
                : JsonSupport.optionalString(object, "promptId");
@@ -65,8 +72,7 @@ public final class DaemonEvent {
            throw new DaemonProtocolException(
                    "SSE envelope promptId conflicts with data.promptId");
        }
-        String canonicalPromptId = promptId == null ? dataPromptId : promptId;
-        return expectedPromptId.equals(canonicalPromptId);
+        return promptId == null ? dataPromptId : promptId;
    }

    String updateKind() {
@@ -3,7 +3,7 @@ package com.alibaba.qwen.code.daemon;
 import java.util.Map;

 /** A received non-success HTTP response from the daemon or an intermediary. */
-public final class DaemonHttpException extends DaemonException {
+public class DaemonHttpException extends DaemonException {
    private final int statusCode;
    private final String responseBody;

@@ -14,6 +14,13 @@ public final class DaemonHttpException extends DaemonException {
        this.responseBody = responseBody;
    }

+    // A refusal subclass that carries no wire response (a local veto).
+    DaemonHttpException(String message) {
+        super(message);
+        this.statusCode = 0;
+        this.responseBody = "";
+    }
+
    public int getStatusCode() {
        return statusCode;
    }
@@ -34,9 +34,15 @@ public final class DaemonSessionClient implements AutoCloseable {
    private static final Set<String> SESSION_FAILURE_EVENTS = Set.of(
            "client_evicted", "session_closed", "session_died",
            "state_resync_required", "stream_error");
+    // No id is burned on these by design (see sse-events.ts in packages/cli):
+    // they are synthesized outside the replayable sequence. session_snapshot
+    // is the id-less synthetic frame the bridge's withSnapshot wrapper emits
+    // for a ?snapshot=1 subscription on the primary events route; the hosted
+    // events route never reads that query parameter and stamps every frame
+    // with an id.
    private static final Set<String> IDLESS_SYNTHETIC_EVENTS = Set.of(
            "client_evicted", "slow_client_warning", "stream_error",
-            "state_resync_required", "replay_complete");
+            "state_resync_required", "replay_complete", "session_snapshot");
    private static final Set<String> OBSERVABLE_STREAM_EVENTS = Set.of(
            "slow_client_warning", "replay_complete");
    private static final Set<Integer> RETRYABLE_SSE_STATUS = Set.of(
@@ -2,6 +2,10 @@ package com.alibaba.qwen.code.daemon;

 /** A transport failure for an operation whose outcome is otherwise known. */
 public class DaemonTransportException extends DaemonException {
+    DaemonTransportException(String message) {
+        super(message);
+    }
+
    DaemonTransportException(String message, Throwable cause) {
        super(message, cause);
    }
@@ -2,7 +2,11 @@ package com.alibaba.qwen.code.daemon;

 import java.io.IOException;
 import java.io.InputStream;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;

 /** One generation- and epoch-fenced Hosted Harness SSE stream. */
@@ -12,15 +16,20 @@ public final class HarnessEventStream implements AutoCloseable {
    private final InputStream input;
    private final SseReader reader;
    private final String eventEpoch;
-    private volatile long lastEventId;
-    // close() never takes the reader's lock: next() holds it across the
-    // blocking SSE read, and closing the body is what ends that read. A
-    // half-open connection never answers, so waiting for the lock first
-    // wedged every close — a generation adoption's client close included.
    private final AtomicBoolean closed = new AtomicBoolean();
-    // A monitor held across the blocking SSE read pins a virtual thread to
-    // its carrier on JDK 21; a ReentrantLock lets the reader unmount.
-    private final ReentrantLock lock = new ReentrantLock();
+    private final AtomicBoolean idleTimedOut = new AtomicBoolean();
+    // The park stamp IS the waiting flag: 0 while no consumer is parked,
+    // otherwise the nanoTime of the park, refreshed by every frame the
+    // reader pulls while parked. One volatile write publishes both facts,
+    // so a watchdog tick can never pair a fresh flag with a stale stamp.
+    private final AtomicLong waitingSince = new AtomicLong();
+    // Serializes concurrent next() callers so frames cannot interleave;
+    // close() never takes this lock, so a blocked read stays abortable. A
+    // monitor held across the blocking SSE read would pin a virtual thread
+    // to its carrier on JDK 21; a ReentrantLock lets the reader unmount.
+    private final ReentrantLock cursorLock = new ReentrantLock();
+    private volatile ScheduledFuture<?> idleWatchdog;
+    private volatile long lastEventId;

    HarnessEventStream(HostedHarnessClient client, HarnessSessionRef session,
            InputStream input, int maximumFrameBytes, long lastEventId,
@@ -28,9 +37,9 @@ public final class HarnessEventStream implements AutoCloseable {
        this.client = client;
        this.session = session;
        this.input = input;
-        this.reader = new SseReader(input, maximumFrameBytes, () -> {
-            // This transport does not own an idle watchdog.
-        });
+        this.reader = new SseReader(input, maximumFrameBytes,
+                () -> waitingSince.updateAndGet(
+                        v -> v == 0 ? 0 : System.nanoTime()));
        this.lastEventId = lastEventId;
        this.eventEpoch = eventEpoch;
    }
@@ -44,15 +53,40 @@ public final class HarnessEventStream implements AutoCloseable {
    }

    public DaemonEvent next() {
-        lock.lock();
+        if (closed.get()) {
+            throw closedFailure();
+        }
+        cursorLock.lock();
        try {
-            ensureOpen();
+            // A queued caller re-checks inside the lock: the stream may
+            // have been closed (or idle-aborted) while it waited.
+            if (closed.get()) {
+                throw closedFailure();
+            }
+            // The idle budget measures peer silence while a consumer is
+            // parked here; time between next() calls is not charged. The
+            // single waitingSince write publishes the park and its stamp
+            // together, so a watchdog tick either skips (0) or sees this
+            // fresh stamp — never one stale by the whole pause since the
+            // previous next().
+            waitingSince.set(System.nanoTime());
            try {
                SseReader.Frame frame = reader.next();
                // A close() that raced this read (a buffered frame, or the
-                // end of the body) reports as closed, not as an event or
-                // as the server ending the stream.
-                ensureOpen();
+                // end of the body) reports the same signal the interrupted
+                // read would have raised — one cause, one exception type
+                // however it raced — never as an event or the server
+                // ending the stream. The closed guard's programming-error
+                // type stays reserved for a caller that arrives after the
+                // close (the pre-park guards above).
+                if (closed.get()) {
+                    if (idleTimedOut.get()) {
+                        throw new DaemonTransportException(
+                                "Hosted Harness SSE idle timeout");
+                    }
+                    throw new DaemonTransportException(
+                            "HarnessEventStream is closed");
+                }
                if (frame == null) {
                    close();
                    return null;
@@ -75,9 +109,19 @@ public final class HarnessEventStream implements AutoCloseable {
                client.observeEvent(session, event);
                return event;
            } catch (IOException e) {
+                if (idleTimedOut.get()) {
+                    // The watchdog aborted the stream: the parked consumer
+                    // gets the idle-timeout signal whether or not the
+                    // watchdog's close() has landed yet.
+                    throw new DaemonTransportException(
+                            "Hosted Harness SSE idle timeout", e);
+                }
                if (closed.get()) {
-                    // close() ended the blocked read.
-                    throw new IllegalStateException(
+                    // A caller-initiated close() ended the blocked read.
+                    // The parked consumer gets a transport failure; a
+                    // caller arriving after the close gets the closed
+                    // guard's signal instead.
+                    throw new DaemonTransportException(
                            "HarnessEventStream is closed", e);
                }
                closeQuietly();
@@ -86,9 +130,11 @@ public final class HarnessEventStream implements AutoCloseable {
            } catch (RuntimeException e) {
                closeQuietly();
                throw e;
+            } finally {
+                waitingSince.set(0);
            }
        } finally {
-            lock.unlock();
+            cursorLock.unlock();
        }
    }

@@ -97,12 +143,17 @@ public final class HarnessEventStream implements AutoCloseable {
        if (!closed.compareAndSet(false, true)) {
            return;
        }
-        client.unregisterStream(this);
+        ScheduledFuture<?> watchdog = idleWatchdog;
+        if (watchdog != null) {
+            watchdog.cancel(false);
+        }
        try {
            input.close();
        } catch (IOException e) {
            throw new DaemonTransportException(
                    "Hosted Harness SSE stream could not be closed", e);
+        } finally {
+            client.unregisterStream(this);
        }
    }

@@ -114,10 +165,57 @@ public final class HarnessEventStream implements AutoCloseable {
        }
    }

-    private void ensureOpen() {
-        if (closed.get()) {
-            throw new IllegalStateException(
-                    "HarnessEventStream is closed");
+    private RuntimeException closedFailure() {
+        if (idleTimedOut.get()) {
+            return new DaemonTransportException(
+                    "Hosted Harness SSE idle timeout");
+        }
+        return new IllegalStateException(
+                "HarnessEventStream is closed");
+    }
+
+    void startIdleWatchdog() {
+        if (client.sseIdleTimeout().isZero()) {
+            // Builder.sseIdleTimeout(Duration.ZERO): the caller owns the
+            // deadline, so no watchdog is scheduled.
+            return;
+        }
+        long idleMillis = HostedHarnessClient.saturatedMillis(
+                client.sseIdleTimeout());
+        long intervalMillis = Math.max(100L, idleMillis / 2L);
+        long idleNanos = TimeUnit.MILLISECONDS.toNanos(idleMillis);
+        try {
+            idleWatchdog = client.scheduler().scheduleAtFixedRate(() -> {
+                if (closed.get()) {
+                    ScheduledFuture<?> watchdog = idleWatchdog;
+                    if (watchdog != null) {
+                        watchdog.cancel(false);
+                    }
+                    return;
+                }
+                // Only a consumer parked in next() is waiting on the peer;
+                // a stream nobody is pulling stays open until close().
+                long since = waitingSince.get();
+                if (since == 0) {
+                    return;
+                }
+                if (System.nanoTime() - since >= idleNanos
+                        && idleTimedOut.compareAndSet(false, true)) {
+                    // Closes the raw input without any stream monitor, so
+                    // the single-thread scheduler never blocks behind the
+                    // reader.
+                    closeQuietly();
+                }
+            }, intervalMillis, intervalMillis, TimeUnit.MILLISECONDS);
+        } catch (RejectedExecutionException e) {
+            // A racing client.close() already shut the scheduler down: map
+            // the local rejection the way send() does, and drop the
+            // registration streamEvents just added so the unarmed stream
+            // is not stranded in the client's set.
+            client.unregisterStream(this);
+            throw new DaemonTransportException(
+                    "Hosted Harness SSE idle watchdog could not be scheduled",
+                    e);
        }
    }
 }
@@ -1,6 +1,7 @@
 package com.alibaba.qwen.code.daemon;

 import java.util.Map;
+import java.util.Objects;

 /** One durable Runtime execution observed while loading a Hosted Harness. */
 public final class HarnessRuntimeExecutionRecovery {
@@ -16,6 +17,10 @@ public final class HarnessRuntimeExecutionRecovery {
            String executionCallId, String runtimeSessionId,
            String progressCursor, String outcome,
            Map<String, Object> status) {
+        if ("known".equals(outcome)) {
+            Objects.requireNonNull(status,
+                    "status required for known outcome");
+        }
        this.functionCallId = functionCallId;
        this.toolName = toolName;
        this.executionCallId = executionCallId;
@@ -1,8 +1,32 @@
 package com.alibaba.qwen.code.daemon;

-/** This session client already owns an unfinished local prompt call. */
-public final class PromptAlreadyActiveException extends DaemonException {
+/**
+ * Thrown when the Hosted Harness definitively refuses a prompt submission
+ * with HTTP 409. The route's 409 vocabulary is wider than a turn conflict —
+ * see {@link #getCode()}, which is {@code null} when the response body
+ * carried no recognisable code. The wire shape extends {@link
+ * DaemonHttpException} so a consumer that matches {@code catch
+ * (DaemonHttpException)} still reads the status and the peer's response
+ * body (including the machine-readable {@code code} field through
+ * {@link #getErrorCode()}); {@link DaemonSessionClient}'s local veto is the
+ * only producer that carries no status, and {@link HostedHarnessClient}
+ * vetoes a second prompt identity with a bare {@link DaemonException}.
+ */
+public final class PromptAlreadyActiveException extends DaemonHttpException {
+    private final String code;
+
    PromptAlreadyActiveException() {
        super("DaemonSessionClient permits only one local prompt at a time");
+        this.code = null;
+    }
+
+    PromptAlreadyActiveException(String operation, int statusCode,
+            String responseBody, String code) {
+        super(operation, statusCode, responseBody);
+        this.code = code;
+    }
+
+    public String getCode() {
+        return code;
    }
 }
@@ -3,7 +3,7 @@ package com.alibaba.qwen.code.daemon;
 /** Cursor and epoch used to open one Hosted Harness SSE stream. */
 public final class StreamHarnessEvents {
    private final HarnessSessionRef session;
-    private final long lastEventId;
+    private final Long lastEventId;
    private final String eventEpoch;
    private final boolean snapshot;

@@ -11,7 +11,7 @@ public final class StreamHarnessEvents {
        if (builder.session == null) {
            throw new IllegalStateException("session must be provided");
        }
-        if (builder.lastEventId < 0) {
+        if (builder.lastEventId != null && builder.lastEventId < 0) {
            throw new IllegalArgumentException(
                    "lastEventId must be non-negative");
        }
@@ -30,7 +30,7 @@ public final class StreamHarnessEvents {
        return session;
    }

-    long getLastEventId() {
+    Long getLastEventId() {
        return lastEventId;
    }

@@ -44,7 +44,7 @@ public final class StreamHarnessEvents {

    public static final class Builder {
        private HarnessSessionRef session;
-        private long lastEventId;
+        private Long lastEventId;
        private String eventEpoch;
        private boolean snapshot;

@@ -56,6 +56,12 @@ public final class StreamHarnessEvents {
            return this;
        }

+        /**
+         * Resumes the stream after this event id. An explicit cursor always
+         * wins — including 0, which asks for a replay from the beginning;
+         * only an omitted cursor falls back to the session ref's watermark,
+         * so the epoch fence stays on by default.
+         */
        public Builder lastEventId(long lastEventId) {
            this.lastEventId = lastEventId;
            return this;
@@ -66,6 +72,12 @@ public final class StreamHarnessEvents {
            return this;
        }

+        /**
+         * Adds the snapshot query parameter to the stream request. The
+         * snapshot frame is produced only by the primary daemon transport;
+         * the hosted events route does not read this parameter, so the
+         * switch is parsed-compatibly ignored there.
+         */
        public Builder snapshot(boolean snapshot) {
            this.snapshot = snapshot;
            return this;
@@ -9,6 +9,7 @@ import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeMap;

 /** Caller-owned prompt admission request for a Hosted Harness session. */
 public final class SubmitHarnessTurn {
@@ -18,6 +19,7 @@ public final class SubmitHarnessTurn {
    private final String promptId;
    private final List<Map<String, Object>> promptContent;
    private final String payloadDigest;
+    private final Object wirePromptContent;
    private final Long deadlineMillis;
    private final Map<String, Object> modelSelection;
    private final Map<String, Object> approvalSelection;
@@ -40,8 +42,22 @@ public final class SubmitHarnessTurn {
        this.promptContent = Collections.unmodifiableList(content);
        this.payloadDigest = HostedHarnessClient.requireDigest(
                builder.payloadDigest, "payloadDigest");
-        String computed = computePayloadDigest(this.promptContent);
-        if (!computed.equals(payloadDigest)) {
+        // The hosted route recomputes the digest over the received prompt's
+        // key order before any replay check, so the digest must cover the
+        // exact bytes the wire carries. New digests are minted over the
+        // canonical (key-sorted) form; a digest persisted by a build that
+        // predates canonicalization covers the as-built key order instead,
+        // and neither the stored digest nor the stored input may be
+        // rewritten. Accept that legacy shape and keep it on the wire, or
+        // recovering such a turn row fails this self-check before any
+        // request is sent.
+        if (computePayloadDigest(this.promptContent).equals(
+                this.payloadDigest)) {
+            this.wirePromptContent = canonicalForDigest(this.promptContent);
+        } else if (digestOver(this.promptContent).equals(
+                this.payloadDigest)) {
+            this.wirePromptContent = this.promptContent;
+        } else {
            throw new IllegalArgumentException(
                    "payloadDigest does not match promptContent");
        }
@@ -60,14 +76,25 @@ public final class SubmitHarnessTurn {
            throw new IllegalArgumentException(
                    "promptContent must not be empty");
        }
+        // The hosted route recomputes this value by re-serializing the
+        // parsed `prompt` member (JavaScript `JSON.stringify`
+        // semantics: compact separators, literal non-ASCII) and hashing
+        // those bytes, so the digest must cover exactly the canonical
+        // form toJson() also emits. Iteration order is only
+        // canonicalized recursively because fastjson2 does not sort
+        // maps nested in a collection.
+        return digestOver(canonicalForDigest(promptContent));
+    }
+
+    private static String digestOver(Object value) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
-            byte[] bytes = JsonSupport.encode(promptContent)
+            byte[] bytes = JsonSupport.encode(value)
                    .getBytes(StandardCharsets.UTF_8);
            byte[] hashed = digest.digest(bytes);
            StringBuilder result = new StringBuilder("sha256:");
-            for (byte value : hashed) {
-                result.append(String.format("%02x", value & 0xff));
+            for (byte b : hashed) {
+                result.append(String.format("%02x", b & 0xff));
            }
            return result.toString();
        } catch (NoSuchAlgorithmException e) {
@@ -75,6 +102,26 @@ public final class SubmitHarnessTurn {
        }
    }

+    private static Object canonicalForDigest(Object value) {
+        if (value instanceof Map<?, ?>) {
+            Map<String, Object> sorted = new TreeMap<>();
+            for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet()) {
+                sorted.put(String.valueOf(entry.getKey()),
+                        canonicalForDigest(entry.getValue()));
+            }
+            return sorted;
+        }
+        if (value instanceof List<?>) {
+            List<?> source = (List<?>) value;
+            List<Object> ordered = new ArrayList<>(source.size());
+            for (Object item : source) {
+                ordered.add(canonicalForDigest(item));
+            }
+            return ordered;
+        }
+        return value;
+    }
+
    HarnessSessionRef getSession() {
        return session;
    }
@@ -93,7 +140,7 @@ public final class SubmitHarnessTurn {

    Map<String, Object> toJson() {
        Map<String, Object> result = new LinkedHashMap<>();
-        result.put("prompt", promptContent);
+        result.put("prompt", wirePromptContent);
        result.put("promptId", promptId);
        result.put("payloadDigest", payloadDigest);
        if (deadlineMillis != null) {
@@ -10,6 +10,8 @@ import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
@@ -64,7 +66,8 @@ class HarnessEventStreamCloseTest {

    // A close that lands while next() already holds a whole frame: the
    // frame is dropped rather than delivered off a closed stream, so the
-    // read position does not move.
+    // read position does not move — and the parked consumer gets the same
+    // transport signal as a read the close interrupted on the socket.
    @Test
    @Timeout(60)
    void aCloseRacingABufferedFrameReportsTheStreamClosed()
@@ -72,8 +75,9 @@ class HarnessEventStreamCloseTest {
        assertCloseRacingTheReadReportsClosed(FRAME);
    }

-    // A close that lands as the body ends: the reader sees the close, not
-    // the server ending the stream, which a consumer reads as an early end.
+    // A close that lands as the body ends: the reader sees the close with
+    // the same transport signal, not the server ending the stream, which a
+    // consumer reads as an early end.
    @Test
    @Timeout(60)
    void aCloseRacingTheEndOfTheBodyReportsTheStreamClosed()
@@ -81,6 +85,39 @@ class HarnessEventStreamCloseTest {
        assertCloseRacingTheReadReportsClosed("");
    }

+    // A frame the reader pulls in the tick the idle watchdog commits to
+    // close is discarded with the idle-timeout signal the interrupted read
+    // would have raised — not the programming-error type — so the
+    // consumer's idle-reconnect branch still fires.
+    @Test
+    @Timeout(60)
+    void anIdleAbortRacingABufferedFrameReportsTheIdleTimeout()
+            throws Exception {
+        HostedHarnessClient client = HostedHarnessClient.builder()
+                .baseUri(harness.baseUri())
+                .bearerToken("harness-token")
+                .capabilityDigest(HeldStreamHarness.DIGEST)
+                .sseIdleTimeout(Duration.ofMillis(200))
+                .build();
+        try {
+            HarnessSessionRef session = client.createSession(
+                    CreateHarnessSession.builder()
+                            .harnessSessionId(HeldStreamHarness.SESSION_ID)
+                            .build());
+            HarnessEventStream stream = new HarnessEventStream(client,
+                    session, new DeliverOnClose(FRAME), 1 << 20, 0,
+                    HeldStreamHarness.EVENT_EPOCH);
+            stream.startIdleWatchdog();
+            DaemonTransportException idle = assertThrows(
+                    DaemonTransportException.class, stream::next);
+            assertEquals("Hosted Harness SSE idle timeout",
+                    idle.getMessage());
+            assertEquals(0, stream.getLastEventId());
+        } finally {
+            client.close();
+        }
+    }
+
    private void assertCloseRacingTheReadReportsClosed(String body)
            throws Exception {
        HostedHarnessClient client = harness.client();
@@ -95,8 +132,12 @@ class HarnessEventStreamCloseTest {
                    session, new CloseOnFirstRead(body, opened), 1 << 20, 0,
                    HeldStreamHarness.EVENT_EPOCH);
            opened.set(stream);
-            IllegalStateException closed = assertThrows(
-                    IllegalStateException.class, stream::next);
+            // The close raced the read, so the parked consumer gets the
+            // same transport signal the interrupted read would have raised
+            // — never the closed guard's programming-error type, which
+            // stays reserved for a caller arriving after the close.
+            DaemonTransportException closed = assertThrows(
+                    DaemonTransportException.class, stream::next);
            assertEquals("HarnessEventStream is closed", closed.getMessage());
            assertEquals(0, stream.getLastEventId());
        } finally {
@@ -149,9 +190,13 @@ class HarnessEventStreamCloseTest {
        reader.join(TimeUnit.SECONDS.toMillis(10));
        assertFalse(reader.isAlive(),
                "the blocked read did not end when its stream closed");
-        // A local close is not the server ending the stream.
-        IllegalStateException closed = assertInstanceOf(
-                IllegalStateException.class, outcome.get());
+        // A local close is not the server ending the stream: the read
+        // blocked on the socket is cut at the transport layer, so the
+        // parked consumer gets a DaemonTransportException, while a
+        // caller arriving after the close still gets the closed guard's
+        // IllegalStateException.
+        DaemonTransportException closed = assertInstanceOf(
+                DaemonTransportException.class, outcome.get());
        assertEquals("HarnessEventStream is closed", closed.getMessage());
        client.close();
    }
@@ -214,4 +259,46 @@ class HarnessEventStreamCloseTest {
            }
        }
    }
+
+    // A body whose first read blocks until the stream is closed — the idle
+    // watchdog's closeQuietly — and then delivers its bytes: the shape of a
+    // frame landing in the reader's buffer inside the abort tick.
+    private static final class DeliverOnClose extends InputStream {
+        private final CountDownLatch closedLatch = new CountDownLatch(1);
+        private final InputStream bytes;
+
+        DeliverOnClose(String body) {
+            this.bytes = new ByteArrayInputStream(
+                    body.getBytes(StandardCharsets.UTF_8));
+        }
+
+        @Override
+        public int read() throws IOException {
+            awaitClosed();
+            return bytes.read();
+        }
+
+        @Override
+        public int read(byte[] buffer, int offset, int length)
+                throws IOException {
+            awaitClosed();
+            return bytes.read(buffer, offset, length);
+        }
+
+        @Override
+        public void close() {
+            closedLatch.countDown();
+        }
+
+        private void awaitClosed() throws IOException {
+            try {
+                if (!closedLatch.await(30, TimeUnit.SECONDS)) {
+                    throw new IOException("the stream was never closed");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new IOException(e);
+            }
+        }
+    }
 }