diff --git a/docs/design/2026-10-18-managed-workspace-release-ownership.md b/docs/design/2026-10-18-managed-workspace-release-ownership.md new file mode 100644 index 0000000000..80f3602d3a --- /dev/null +++ b/docs/design/2026-10-18-managed-workspace-release-ownership.md @@ -0,0 +1,505 @@ +# Managed Workspace: release ownership when the worker refuses or hangs + +[English](2026-10-18-managed-workspace-release-ownership.md) | [简体中文](2026-10-18-managed-workspace-release-ownership.zh-CN.md) + +- **Status:** Proposed +- **Date:** 2026-10-18 +- **Baseline:** `0fac9973df` +- **Issue:** [#12937](https://github.com/QwenLM/qwen-code/issues/12937) (split out of [#12868](https://github.com/QwenLM/qwen-code/pull/12868)) +- **Modules:** `packages/managed-agent-server`, `packages/cli/src/serve`, `packages/core/src/tools` + +Every claim in "Current state" was re-read at the baseline commit above. "Proposed" +marks behavior that does not exist yet; the status table in §2.6 separates the two. + +## 1. Problem + +A Workspace Runtime Session owns its storage through a durable row. The release +path clears that row only after the worker acknowledges, so a worker refusal, +timeout, or hang leaves the storage permanently owned by a Session that can +never use it again: + +- every later `acquire` on that storage fails `claim()` with `409 workspace_busy`; +- a retry of the same Runtime Session stays in `RELEASING`; +- no other automatic path clears the row. + +The fix must free the durable lease on a path that does not require a successful +worker answer, without weakening the acquire-side rule that a missing response +proves nothing. + +## 2. Current state + +### 2.1 Acquire keeps ownership on an uncertain answer + +`WorkspaceRuntimeTransport.acquire` claims the storage before it installs or +activates anything, and deliberately keeps the claim when the install result is +uncertain (`WorkspaceRuntimeTransport.java:104-105`): + +```java +ownership.claim(context.binding(), context.session()); +// Failure retains ownership: a missing response cannot prove the worker did nothing. +``` + +`acquireUncertain` then reports a retryable failure. This invariant is the reason +a blind compensating release on a transport failure is not acceptable. + +### 2.2 Release clears ownership only after the worker answers + +`WorkspaceRuntimeTransport.release` (`:409-435`) has three stages: + +```java +return delegate.release(lease, session).thenCompose(released -> { + if (!Boolean.TRUE.equals(released)) { + throw WorkspaceExecutionStore.unavailable(); // :424-426 + } + return delegate.activateWorkspace(context.runtime(), context.session(), context.binding(), false); +}) +.thenApply(ignored -> { + ownership.release(context.binding(), context.session()); // :430 + detachOwed.remove(detach); + return true; +}); +``` + +Any throw or timeout before the last stage skips `ownership.release`. The +`RELEASING` shortcut at `:416-421` does not help: it requires +`!ownership.isHeld(...)`, and the holder is exactly what is stuck. + +`ownership.release` has two call sites today, not one: `:430` (full release) and +`:455` (`releaseMount`). The second belongs to the separate mount-handback +operation and is not a compensation path for a failed full release. + +### 2.3 The Broker leaves the Session in RELEASING + +`RuntimeBrokerService.completeSessionRelease` (`:3085`) transitions the Session +to `RELEASING` before calling the transport (`:3101`), calls +`transport.release(...)` (`:3112`), and only on success calls +`finishSessionRelease` (`:3124`). On error it clears the latch and completes +exceptionally (`:3116-3121`); the durable Session row stays `RELEASING`. + +A retry rebuilds the same chain and reaches the same worker refusal. The +worker's memoized release promise (see 2.5) is what makes this permanent. + +### 2.4 The holder row is durable and exact-match + +`ma_storage_owner` (`V1__managed_schema.sql:861-891`) stores +`holder_key`, `binding_id`, `runtime_generation`, `runtime_session_id`, plus the +maintenance-hold columns. There is no holder generation, expiry, or heartbeat +column. + +- `WorkspaceExecutionStore.claim` (`:337`) takes the row `FOR UPDATE` and + refuses a different holder with `busy()` (`:703`, `409 workspace_busy`, + retryable). It has no supersede rule. +- `WorkspaceExecutionStore.release` (`:422`) clears the row only when the + binding at `(binding_id, generation)` is still `READY` or `DRAINING`, and only + for the exact `holder_key` (`:447-450`). A `LOST` or `RELEASED` binding makes + it throw `workspace_unavailable`. + +### 2.5 Worker release is memoized and unbounded + +`managed-runtime-provider-worker.ts:297-327`: + +```ts +if (session.release) return session.release; // :308 +... +entry.release = (async () => { + await entry.value?.runtime.releasePrepared(); // :316 + ... +})().catch((error) => { entry.release = undefined; throw error; }); // :323-326 +``` + +`ManagedToolRuntime.releasePrepared()` (`managed-tool-runtime.ts:919-932`) +refuses while an execution is unfinished: + +```ts +throw new Error('Managed Runtime still owns unfinished execution.'); +``` + +and otherwise awaits `entry.cancellation` promises with no deadline. Two +consequences: + +1. A refusal is a plain `Error`. The worker's response mapper + (`managed-runtime-provider-worker.ts:726-731`) turns it into `409` with + `code: managed_runtime_provider_operation_failed`, so the Broker cannot tell + a deliberate refusal from an unrelated provider failure. +2. A hang keeps `entry.release` pending forever; every retry returns the same + promise. The Broker's HTTP client times out (`HttpRuntimeTransport.java:50`, + `REQUEST_TIMEOUT = Duration.ofSeconds(30)`), but the worker never retries. + +### 2.6 Other paths that clear the holder + +| Path | Proof it requires | Reaches the stuck case? | +| ----------------------------------------------------------- | -------------------------------------------------------- | ------------------------------------- | +| `ownership.release` in `WorkspaceRuntimeTransport` (`:430`) | worker acknowledged | no | +| `releaseMount` (`:455`) | separate mount-handback operation | no | +| `WorkspaceExecutionStore.releaseLost` (clear at `:650`) | binding `LOST` + `hasStoppedWriters()` + operation lease | no (binding is still `READY`) | +| `WorkspaceMaintenanceRecovery` (`:162`) | maintenance hold | no (operator/maintenance driven only) | +| `WorkspaceOperatorRecoveryStore` | operator recovery operation | no (manual) | + +`RuntimeBrokerService.releaseStoppedSession` (`:5838`) is the one path that +durably retires a Session **without** calling the transport: it requires a +drain receipt matching the binding and no active executions, then calls +`finishSessionRelease` and removes the in-memory session. It never clears +`ma_storage_owner`, so even the successful close cascade can retire a Session +while its storage stays owned. + +`releaseUnusableSession` (`:3137`) handles a dead worker: it invalidates a +`READY` binding to `LOST` and fails with `runtime_reconciliation_required`. The +holder is then W0e's (`releaseLost`) responsibility, which requires durable stop +evidence. + +| Behavior | At baseline `0fac9973df` | +| ------------------------------------------------------------------ | ------------------------ | +| Acquire retains ownership on an uncertain answer | Implemented | +| Release clears ownership only after worker ack | Implemented | +| `RELEASING` Session with a stranded holder has an automatic escape | **Not implemented** | +| Local Session retirement clears the holder | **Not implemented** | +| Worker release is bounded and retry-safe | **Not implemented** | +| Refusal carries a distinguishable retryable code | **Not implemented** | + +## 3. Failure modes + +**F1 — refusal.** The Broker settled an execution the worker still considers +unfinished. `releasePrepared()` throws; the worker answers `409`; the Broker +fails the release; the Session stays `RELEASING`; the holder stays. The worker +may never finish, so retries never converge. + +**F2 — hang.** `releasePrepared()` awaits a cancellation promise that never +settles. The memoized `entry.release` never settles either, so every later +attempt returns the same hung promise. The Broker times out each time; the +holder stays forever. + +**F3 — crash window.** A Broker that dies between `claim` and `release` (or +mid-release) leaves the row behind. This design does not solve F3 by itself; +the existing reconciliation and `releaseLost` paths remain the owners. + +## 4. Goals and non-goals + +**Goals** + +- G1: after the Broker has durably retired a Session, its storage holder is + cleared without requiring a successful worker answer. +- G2: no compensating release on a mere transport failure; the acquire-side + invariant in §2.1 is preserved. +- G3: a hung worker release is abandoned at a bound and can be retried. +- G4: a worker refusal is distinguishable from an unrelated provider failure. + +**Non-goals** + +- Changing `claim` to supersede a holder by time or generation alone (§6, D-1). +- Compensating on `releaseUnusableSession` without stop evidence (§6, D-4). +- Changing the public/WebShell API, the OpenAPI contract, or the schema. +- Automatically draining a Workspace on N failed routine releases; that is a + control-plane policy decision, not an ownership primitive. + +## 5. Proposal + +### 5.1 Worker: bounded, distinguishable release + +File: `packages/cli/src/serve/managed-runtime-provider-worker.ts`. + +1. Give `ProviderSession` a monotonically increasing `releaseAttempt`. +2. Race the first step against a deadline: + + ```ts + const attempt = ++entry.releaseAttempt; + try { + await withDeadline( + entry.value?.runtime.releasePrepared(), + RELEASE_DEADLINE_MS, + ); + } catch (error) { + throw isDeadline(error) + ? conflict( + 'Managed Runtime release timed out.', + 'managed_runtime_release_timeout', + ) + : conflict(messageOf(error), 'managed_runtime_owner_hold_pending'); + } + ``` + + `managed_runtime_owner_hold_pending` already exists + (`managed-runtime-provider-worker.ts:312`) and already means "the worker + still owns work". `managed_runtime_release_timeout` is new and retryable. + +3. Before mutating provider state (`closeSessionAdmission`, `forgetSession`, + `closed = true`, push to `released`), require the attempt to still own the + identity: + + ```ts + if (this.sessions.get(id) !== entry || entry.releaseAttempt !== attempt) { + return true; // an abandoned attempt must not retire a re-acquired Session + } + ``` + +4. The abandoned `releasePrepared()` promise keeps running; attach a no-op + rejection handler so it cannot surface as an unhandled rejection. + +5. `entry.release` is cleared only when the failing attempt is still current, so + a timed-out attempt can be retried with a fresh one. + +`RELEASE_DEADLINE_MS` must stay below `HttpRuntimeTransport.REQUEST_TIMEOUT` +(30 s) so the Broker sees a clean retryable `409` instead of a client timeout. +The initial value is 20 s; see §11. + +Core stays unchanged: the deadline belongs to the memoizing caller, and the +core runtime keeps its synchronous refusal contract. + +### 5.2 Broker: `releaseRetired` with a durable proof predicate + +Add a transport hook. The default is a no-op because only the Workspace +transport owns durable storage: + +```java +/** + * Finishes durable ownership for a Session the Broker has already retired. + * The caller must hold durable proof that the Runtime no longer holds the + * Session: a persisted drain receipt matching the binding and no active + * executions. Never call this on a mere transport failure. + */ +default CompletionStage releaseRetired(RuntimeLease lease, RuntimeSession session) { + return CompletableFuture.completedFuture(true); +} +``` + +`WorkspaceRuntimeTransport` overrides it and re-verifies the proof against the +durable rows it already injects (`bindings`, `sessions`): + +```java +@Override +public CompletionStage releaseRetired(RuntimeLease lease, RuntimeSession session) { + if (!managed(session)) { + return CompletableFuture.completedFuture(true); + } + Context context = context(lease, session, false); + RuntimeBindingRecord binding = bindings.findById(context.binding().getBindingId()); + if (binding != null && binding.getState() == RuntimeBindingRecord.State.LOST) { + // The LOST family owns this holder; its evidence gate is stricter. + return CompletableFuture.completedFuture(true); + } + boolean proven = binding != null + && binding.getGeneration() == context.binding().getGeneration() + && binding.getState() == RuntimeBindingRecord.State.DRAINING + && binding.getDrainReceipt() != null + && binding.getDrainReceipt().matches(binding); + if (!proven) { + // Held without proof is a refusal; nothing held is already settled. + return CompletableFuture.completedFuture( + !ownership.isHeld(context.binding(), context.session())); + } + ownership.release(context.binding(), context.session()); + detachOwed.remove(detachKey(context)); + return CompletableFuture.completedFuture(true); +} +``` + +Return contract: `true` means the caller may proceed (cleared, never held, or +owned by the `LOST` evidence path); `false` means a holder exists but this path +has no proof, and the caller must fail the release rather than leave a silent +wedge. + +The proof predicate is deliberately narrow: + +- **P1** the binding at `(binding_id, generation)` is `DRAINING` — the drain + claim is the fence that also refuses a new `claim` + (`WorkspaceExecutionStore.claim` requires `!drain_requested`); +- **P2** `drain_receipt` is non-null and matches the binding — the worker was + stopped, not merely unreachable; +- **P3** the caller has already verified no active executions for + `(binding_id, generation)` — `releaseStoppedSession` does this before it + retires the Session; +- **P4** `WorkspaceExecutionStore.release` clears only the exact + `(holder_key, binding_id, runtime_generation, runtime_session_id)` tuple, so a + newer holder can never be cleared by an older retirement. + +`LOST` bindings are excluded on purpose: their stop evidence and holder clearing +belong to `releaseLost`. + +### 5.3 Broker: drain-arm escalation + +D1–D2 are not reachable in F1/F2 if the drain cascade waits for the worker +before it stops it. Today `drainClaimedBinding` releases every Session first and +only then calls `stopAcknowledgedRuntime`; a refusing worker fails the first +step, so the stop is never attempted. + +Change `drainClaimedBinding` so that one worker refusal escalates once: + +1. Run `drainSessions` as today. +2. If a session release fails with `managed_runtime_owner_hold_pending` or + `managed_runtime_release_timeout`, call `provisioner.stopDrained(binding)`, + persist the receipt with `renewal.persistDrainReceipt(...)`, and run + `drainSessions` again. +3. On the second pass every active Session takes `releaseStoppedSession` + (receipt present, no active executions), which calls + `transport.releaseRetired(...)` before `finishSessionRelease`. +4. A second refusal, or a provisioner without `supportsDrainedStop`, fails the + drain with the existing error. The holder then stays for operator recovery; + the failure is explicit rather than silent. + +This matches the documented drain semantics ("the drain arm stops what is live +first") and keeps routine release (`drain = false`) non-destructive: only the +close cascade may stop a worker. + +### 5.4 Ordering and crash windows + +`releaseStoppedSession` calls `releaseRetired` **before** +`finishSessionRelease`: + +- compensation succeeds, then the durable `RELEASED` commit — the normal case; +- compensation fails — the release fails, the Session stays `RELEASING`, the + drain claim stays retryable, and the holder is untouched; +- the Broker crashes between the two — the Session is still active with no + holder. A retry takes the existing absent-holder shortcut + (`WorkspaceRuntimeTransport.java:416-421`) and completes. No wedge, and no + storage was handed to a second writer. + +Compensating after `finishSessionRelease` would be strictly worse: a crash in +between leaves `RELEASED` plus a stranded holder, and a retry short-circuits on +the `RELEASED` row without ever reaching the compensation. + +`releaseStoppedSession` is synchronous today (`:5838`). It becomes +`CompletionStage`-returning; the two callers (`releaseSavedSession` at `:747` +and `drainSessions`) are already asynchronous compositions. + +### 5.5 Not changing + +- `claim` conflict semantics and the `409 workspace_busy` contract. +- The acquire-side retention rule and `acquireUncertain`. +- `releaseUnusableSession` and `releaseLost`. +- The `ma_storage_owner` schema. +- `RuntimeBrokerService.mapFailure` (`:5538`), which already preserves a + `RuntimeBrokerException` unchanged, so the new worker codes reach the drain + escalation intact. + +## 6. Decisions and rationale + +**D-1: proof-gated compensation, not holder expiry or generation supersede.** +An expiry or generation counter can prove that a holder is _old_, never that the +worker _stopped_. Superseding on age would let a second Session claim storage a +live worker may still be writing, which is exactly the invariant §2.1 protects. +The issue lists expiry as an alternative; this design rejects it because the +proof it needs is the same drain receipt, and storing the proof durably adds a +column without adding a capability. + +**D-2: compensation before the durable release, not after.** See §5.4. The +failure and crash windows both converge to a retryable state instead of a +`RELEASED` row with a stuck holder. + +**D-3: escalate only in the drain arm.** A routine release is required to stop +nothing. The close cascade already owns stop semantics, so the escalation is a +repair of that arm's ordering, not a new policy. + +**D-4: do not compensate in `releaseUnusableSession`.** A dead process is +stronger than a missing response, but it is not proof that orphaned children or +the storage mount stopped writing. `releaseLost` already encodes that stricter +evidence and remains the owner of the `LOST` path. + +**D-5: bound the release where it is memoized.** The hang lives in the memoized +promise, not in `releasePrepared`'s individual awaits. Bounding at the provider +keeps core free of a timeout contract and lets the existing `.catch` reset the +memo. + +**D-6: no schema change.** The drain receipt, binding state, and exact holder +columns already carry the proof. A new column would be a second source of truth. + +## 7. Invariants + +1. A holder is cleared only when the worker is proven stopped (drain receipt + matching the binding, no active executions) or has acknowledged the release. +2. A transport failure alone never clears a holder. +3. Clearing is exact-match; it can never clear a newer holder. +4. An abandoned worker release attempt never mutates provider state for an + identity it no longer owns. +5. A routine release never stops a worker. + +## 8. Risks and mitigations + +| Risk | Mitigation | +| ------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------- | +| Compensation clears a holder a live worker still writes | P1–P3; `LOST` excluded; `releaseLost` keeps its own evidence gate | +| Stale worker attempt retires a re-acquired Session | `releaseAttempt` token plus identity check after the await (§5.1) | +| Deadline fires while `releasePrepared` later succeeds | the abandoned attempt only returns; the next attempt re-runs `releasePrepared`, which is idempotent for settled entries | +| Escalation stops a shared worker serving other Sessions | the close cascade is binding-wide by design; `drainSessions` releases every active Session on the binding | +| Provisioner without `supportsDrainedStop` | explicit failure, operator recovery remains; documented as a residual gap | +| `releaseStoppedSession` signature change ripples | only two callers, both already `CompletionStage` compositions | +| Abandoned promise rejection becomes unhandled | attach a no-op rejection handler in `withDeadline` | + +## 9. Validation plan + +**Java (H2, then `mysql-integration`)** + +- `WorkspaceExecutionStoreTest`: `release` still refuses a `LOST`/`RELEASED` + binding; exact-match clearing; no newer-holder clearing. +- `WorkspaceRuntimeTransportTest` (new cases): + - a worker refusal leaves the holder intact (invariant); + - `releaseRetired` without a matching receipt returns `false` and does not + clear; + - `releaseRetired` with a matching `DRAINING` binding and receipt clears it; + - `releaseRetired` is idempotent when nothing is held. +- `RuntimeBrokerServiceTest`: + - drain cascade: refusal → `stopDrained` → second pass clears the Session and + the holder; + - routine release: refusal does not stop the worker and does not clear; + - second refusal fails the drain explicitly. +- MySQL profile required for the transaction/SQL behavior (per module AGENTS). + +**TypeScript** + +- `managed-runtime-provider-worker.test.ts`: + - refusal maps to `managed_runtime_owner_hold_pending`; + - timeout maps to `managed_runtime_release_timeout` and clears the memo; + - the next attempt runs `releasePrepared` again; + - an abandoned attempt that resolves later does not close a re-acquired + Session (identity token). +- `managed-tool-runtime.test.ts` stays as-is; core behavior is unchanged. + +**End-to-end** + +- Managed Workspace with a worker forced to refuse `release`: close the + Session; assert `ma_storage_owner.holder_key IS NULL`, the Session reaches + `RELEASED`, and a later `acquire` on the same storage succeeds. +- Negative: a transport-level failure without a receipt leaves `holder_key` + set, and the Session stays `RELEASING`. + +**Reference checks** + +- `npm run lint` and Prettier for the touched TS/Java-adjacent docs; `mvn verify` + for the Java module (delete `target/` before root lint, per module AGENTS). + +## 10. Acceptance criteria + +1. With a worker that refuses release with `managed_runtime_owner_hold_pending`, + the close cascade clears `ma_storage_owner.holder_key` for that Session, and + a later acquire on the same storage succeeds. +2. With a worker whose `releasePrepared()` never settles, the release returns a + retryable error within `RELEASE_DEADLINE_MS`; the next release is a fresh + attempt; the Broker never waits past its own request timeout. +3. A plain transport failure (connection error, client timeout) with no drain + receipt leaves the holder set and the Session `RELEASING`. +4. `releaseRetired` cannot clear a holder that does not match the exact + `(holder_key, binding_id, runtime_generation, runtime_session_id)` tuple. +5. No schema migration and no OpenAPI change. +6. New tests in §9 pass, and the existing module suites stay green. + +## 11. Open questions + +1. **Deadline value.** 20 s is chosen below the 30 s HTTP timeout. Should it be + configurable, or fixed to keep one behavior across deployments? +2. **Escalation trigger.** Should the drain escalate on the first refusal, or + after a retry budget (for example, two consecutive refusals) to avoid + stopping a worker that was about to settle? +3. **Control-plane retry budget.** Should the control plane auto-drain after N + failed routine releases? This design deliberately leaves that policy alone. +4. **`releaseMount` symmetry.** `releaseMount` already clears the holder without + a worker answer. It is safe today because the execution mount is separate + from the installation, but the two proofs should be documented together in a + follow-up. + +## 12. Implementation order + +1. Worker refusal code and bounded release (§5.1) — independently testable, + already closes F2. +2. `RuntimeTransport.releaseRetired` plus the `WorkspaceRuntimeTransport` + override (§5.2). +3. `releaseStoppedSession` compensation and its async signature (§5.4). +4. Drain-arm escalation (§5.3). +5. Tests from §9, then the MySQL profile. diff --git a/docs/design/2026-10-18-managed-workspace-release-ownership.zh-CN.md b/docs/design/2026-10-18-managed-workspace-release-ownership.zh-CN.md new file mode 100644 index 0000000000..4147e84ffc --- /dev/null +++ b/docs/design/2026-10-18-managed-workspace-release-ownership.zh-CN.md @@ -0,0 +1,367 @@ +# Managed Workspace：worker 拒绝或挂起时的释放所有权 + +[English](2026-10-18-managed-workspace-release-ownership.md) | [简体中文](2026-10-18-managed-workspace-release-ownership.zh-CN.md) + +- **状态：** 提案 +- **日期：** 2026-10-18 +- **基线：** `0fac9973df` +- **Issue：** [#12937](https://github.com/QwenLM/qwen-code/issues/12937)（拆自 [#12868](https://github.com/QwenLM/qwen-code/pull/12868)） +- **模块：** `packages/managed-agent-server`、`packages/cli/src/serve`、`packages/core/src/tools` + +「现状」一节中的每项主张都在上述基线提交上重新读过。「提案」标记尚未实现的行为；§2.6 的状态表区分两者。 + +## 1. 问题 + +一个 Workspace Runtime Session 通过一行持久记录持有其存储。释放路径只在 worker 应答之后清除该行，因此 worker 的拒绝、超时或挂起会让存储被一个再也无法使用它的 Session 永久持有： + +- 之后对该存储的每次 `acquire` 都在 `claim()` 处失败，返回 `409 workspace_busy`； +- 同一 Runtime Session 的重试停在 `RELEASING`； +- 没有任何其他自动路径会清除该行。 + +修复必须在不要求 worker 成功应答的路径上释放持久租约，同时不削弱 acquire 侧规则：响应缺失不能证明任何事。 + +## 2. 现状 + +### 2.1 acquire 在结果不确定时保留所有权 + +`WorkspaceRuntimeTransport.acquire` 在安装或激活任何东西之前先占用存储，并在安装结果不确定时有意保留占用（`WorkspaceRuntimeTransport.java:104-105`）： + +```java +ownership.claim(context.binding(), context.session()); +// Failure retains ownership: a missing response cannot prove the worker did nothing. +``` + +随后 `acquireUncertain` 报告一个可重试失败。这条不变量正是不能在 transport 失败时盲目补偿释放的原因。 + +### 2.2 release 只在 worker 应答后清除所有权 + +`WorkspaceRuntimeTransport.release`（`:409-435`）分三个阶段： + +```java +return delegate.release(lease, session).thenCompose(released -> { + if (!Boolean.TRUE.equals(released)) { + throw WorkspaceExecutionStore.unavailable(); // :424-426 + } + return delegate.activateWorkspace(context.runtime(), context.session(), context.binding(), false); +}) +.thenApply(ignored -> { + ownership.release(context.binding(), context.session()); // :430 + detachOwed.remove(detach); + return true; +}); +``` + +最后阶段之前的任何抛出或超时都会跳过 `ownership.release`。`:416-421` 的 `RELEASING` 捷径没有帮助：它要求 `!ownership.isHeld(...)`，而持有者恰恰就是卡住的东西。 + +今天 `ownership.release` 有两个调用点，不是一个：`:430`（完整释放）与 `:455`（`releaseMount`）。后者属于独立的挂载归还操作，不是完整释放失败后的补偿路径。 + +### 2.3 Broker 把 Session 留在 RELEASING + +`RuntimeBrokerService.completeSessionRelease`（`:3085`）在调用 transport 之前先把 Session 迁到 `RELEASING`（`:3101`），调用 `transport.release(...)`（`:3112`），只在成功时调用 `finishSessionRelease`（`:3124`）。出错时它清除闩锁并以异常完成（`:3116-3121`）；持久 Session 行保持 `RELEASING`。 + +重试会重建同一条链并撞上同一个 worker 拒绝。真正让它永久化的是 worker 备忘的释放 promise（见 2.5）。 + +### 2.4 holder 行是持久且精确匹配的 + +`ma_storage_owner`（`V1__managed_schema.sql:861-891`）存储 `holder_key`、`binding_id`、`runtime_generation`、`runtime_session_id`，以及维护占用列。没有 holder 代数、过期时间或心跳列。 + +- `WorkspaceExecutionStore.claim`（`:337`）以 `FOR UPDATE` 读取该行，遇到不同持有者就以 `busy()`（`:703`，`409 workspace_busy`，可重试）拒绝。没有任何取代规则。 +- `WorkspaceExecutionStore.release`（`:422`）只在 `(binding_id, generation)` 处的 binding 仍为 `READY` 或 `DRAINING` 时清除该行，且只清精确的 `holder_key`（`:447-450`）。binding 为 `LOST` 或 `RELEASED` 时它会抛 `workspace_unavailable`。 + +### 2.5 worker 释放是备忘的且无界 + +`managed-runtime-provider-worker.ts:297-327`： + +```ts +if (session.release) return session.release; // :308 +... +entry.release = (async () => { + await entry.value?.runtime.releasePrepared(); // :316 + ... +})().catch((error) => { entry.release = undefined; throw error; }); // :323-326 +``` + +`ManagedToolRuntime.releasePrepared()`（`managed-tool-runtime.ts:919-932`）在执行未完成时拒绝： + +```ts +throw new Error('Managed Runtime still owns unfinished execution.'); +``` + +否则它以无截止时间的方式等待 `entry.cancellation` promise。两个后果： + +1. 拒绝是一个普通 `Error`。worker 的响应映射（`managed-runtime-provider-worker.ts:726-731`）把它变成 `409` 加 `code: managed_runtime_provider_operation_failed`，因此 Broker 无法区分一次有意的拒绝与无关的 provider 失败。 +2. 挂起会让 `entry.release` 永远处于 pending；每次重试都返回同一个 promise。Broker 的 HTTP 客户端会超时（`HttpRuntimeTransport.java:50`，`REQUEST_TIMEOUT = Duration.ofSeconds(30)`），但 worker 从不重试。 + +### 2.6 其他会清除 holder 的路径 + +| 路径 | 要求的证据 | 能覆盖卡住场景吗 | +| -------------------------------------------------------------- | -------------------------------------------------------- | -------------------------- | +| `WorkspaceRuntimeTransport` 中的 `ownership.release`（`:430`） | worker 已应答 | 否 | +| `releaseMount`（`:455`） | 独立的挂载归还操作 | 否 | +| `WorkspaceExecutionStore.releaseLost`（清除于 `:650`） | binding `LOST` + `hasStoppedWriters()` + operation lease | 否（binding 仍为 `READY`） | +| `WorkspaceMaintenanceRecovery`（`:162`） | maintenance hold | 否（仅由维护流程驱动） | +| `WorkspaceOperatorRecoveryStore` | operator recovery 操作 | 否（人工） | + +`RuntimeBrokerService.releaseStoppedSession`（`:5838`）是唯一**不**调用 transport 就能持久退役 Session 的路径：它要求与 binding 匹配的 drain receipt 且没有活跃 execution，随后调用 `finishSessionRelease` 并移除内存中的 session。它从不清除 `ma_storage_owner`，因此即便成功的关闭级联也可能在存储仍被持有时退役 Session。 + +`releaseUnusableSession`（`:3137`）处理已死的 worker：它把 `READY` binding 置为 `LOST`，并以 `runtime_reconciliation_required` 失败。holder 随后属于 W0e（`releaseLost`）的职责，后者要求持久停止证据。 + +| 行为 | 基线 `0fac9973df` | +| -------------------------------------------------------- | ----------------- | +| acquire 在结果不确定时保留所有权 | 已实现 | +| release 只在 worker 应答后清除所有权 | 已实现 | +| 卡住的 holder 有自动逃生路径（Session 处于 `RELEASING`） | **未实现** | +| 本地退役 Session 时清除 holder | **未实现** | +| worker 释放有界且可安全重试 | **未实现** | +| 拒绝携带可区分的可重试错误码 | **未实现** | + +## 3. 失败模式 + +**F1 —— 拒绝。** Broker 已结算了一个 worker 仍认为未完成的 execution。`releasePrepared()` 抛出；worker 应答 `409`；Broker 释放失败；Session 停在 `RELEASING`；holder 残留。worker 可能永远不会完成，因此重试永不收敛。 + +**F2 —— 挂起。** `releasePrepared()` 等待一个永不 settle 的 cancellation promise。备忘的 `entry.release` 同样永不 settle，因此之后每次尝试都返回同一个挂起的 promise。Broker 每次超时；holder 永久残留。 + +**F3 —— 崩溃窗口。** Broker 在 `claim` 与 `release` 之间（或释放中途）死亡会留下该行。本设计本身不解决 F3；现有 reconcile 与 `releaseLost` 路径仍是其归属。 + +## 4. 目标与非目标 + +**目标** + +- G1：Broker 持久退役 Session 之后，其存储 holder 无需 worker 成功应答即可清除。 +- G2：单纯的 transport 失败不触发补偿释放；§2.1 的 acquire 侧不变量保持不变。 +- G3：挂起的 worker 释放在界限处被放弃，并可重试。 +- G4：worker 拒绝可以与无关的 provider 失败区分。 + +**非目标** + +- 把 `claim` 改成仅凭时间或代数取代 holder（§6，D-1）。 +- 在没有停止证据时对 `releaseUnusableSession` 做补偿（§6，D-4）。 +- 改动公共/WebShell API、OpenAPI 契约或 schema。 +- 在 N 次普通释放失败后自动 drain Workspace；那是控制面策略，不是所有权原语。 + +## 5. 方案 + +### 5.1 worker：有界且可区分的释放 + +文件：`packages/cli/src/serve/managed-runtime-provider-worker.ts`。 + +1. 给 `ProviderSession` 增加单调递增的 `releaseAttempt`。 +2. 让第一步与截止时间竞争： + + ```ts + const attempt = ++entry.releaseAttempt; + try { + await withDeadline( + entry.value?.runtime.releasePrepared(), + RELEASE_DEADLINE_MS, + ); + } catch (error) { + throw isDeadline(error) + ? conflict( + 'Managed Runtime release timed out.', + 'managed_runtime_release_timeout', + ) + : conflict(messageOf(error), 'managed_runtime_owner_hold_pending'); + } + ``` + + `managed_runtime_owner_hold_pending` 已存在（`managed-runtime-provider-worker.ts:312`），含义就是「worker 仍持有工作」。`managed_runtime_release_timeout` 是新增的可重试错误码。 + +3. 在改变 provider 状态（`closeSessionAdmission`、`forgetSession`、`closed = true`、压入 `released`）之前，要求该 attempt 仍拥有该 identity： + + ```ts + if (this.sessions.get(id) !== entry || entry.releaseAttempt !== attempt) { + return true; // an abandoned attempt must not retire a re-acquired Session + } + ``` + +4. 被放弃的 `releasePrepared()` promise 会继续运行；给它挂一个空 rejection 处理器，避免变成 unhandled rejection。 + +5. 只有失败 attempt 仍是当前 attempt 时才清除 `entry.release`，使超时的 attempt 可以用新的 attempt 重试。 + +`RELEASE_DEADLINE_MS` 必须低于 `HttpRuntimeTransport.REQUEST_TIMEOUT`（30 秒），使 Broker 看到干净的可重试 `409`，而不是客户端超时。初始值取 20 秒；见 §11。 + +core 不变：截止时间属于做备忘的调用方，core runtime 保留其同步拒绝契约。 + +### 5.2 Broker：带持久证据谓词的 `releaseRetired` + +新增一个 transport 钩子。默认是空操作，因为只有 Workspace transport 拥有持久存储： + +```java +/** + * Finishes durable ownership for a Session the Broker has already retired. + * The caller must hold durable proof that the Runtime no longer holds the + * Session: a persisted drain receipt matching the binding and no active + * executions. Never call this on a mere transport failure. + */ +default CompletionStage releaseRetired(RuntimeLease lease, RuntimeSession session) { + return CompletableFuture.completedFuture(true); +} +``` + +`WorkspaceRuntimeTransport` 覆盖它，并针对其已注入的持久行（`bindings`、`sessions`）重新验证证据： + +```java +@Override +public CompletionStage releaseRetired(RuntimeLease lease, RuntimeSession session) { + if (!managed(session)) { + return CompletableFuture.completedFuture(true); + } + Context context = context(lease, session, false); + RuntimeBindingRecord binding = bindings.findById(context.binding().getBindingId()); + if (binding != null && binding.getState() == RuntimeBindingRecord.State.LOST) { + // The LOST family owns this holder; its evidence gate is stricter. + return CompletableFuture.completedFuture(true); + } + boolean proven = binding != null + && binding.getGeneration() == context.binding().getGeneration() + && binding.getState() == RuntimeBindingRecord.State.DRAINING + && binding.getDrainReceipt() != null + && binding.getDrainReceipt().matches(binding); + if (!proven) { + // Held without proof is a refusal; nothing held is already settled. + return CompletableFuture.completedFuture( + !ownership.isHeld(context.binding(), context.session())); + } + ownership.release(context.binding(), context.session()); + detachOwed.remove(detachKey(context)); + return CompletableFuture.completedFuture(true); +} +``` + +返回契约：`true` 表示调用方可以继续（已清除、从未持有，或由 `LOST` 证据路径负责）；`false` 表示 holder 存在但本路径证据不足，调用方必须让释放失败，而不是留下静默的卡死。 + +证据谓词被刻意收窄： + +- **P1** `(binding_id, generation)` 处的 binding 为 `DRAINING` —— drain claim 同时也是拒绝新 `claim` 的栅栏（`WorkspaceExecutionStore.claim` 要求 `!drain_requested`）； +- **P2** `drain_receipt` 非空且与 binding 匹配 —— worker 是被停止了，而不仅仅是不可达； +- **P3** 调用方已确认 `(binding_id, generation)` 没有活跃 execution —— `releaseStoppedSession` 在退役 Session 之前已经检查； +- **P4** `WorkspaceExecutionStore.release` 只清除精确的 `(holder_key, binding_id, runtime_generation, runtime_session_id)` 元组，因此旧退役永远清不到新持有者。 + +`LOST` binding 被刻意排除：其停止证据与 holder 清除属于 `releaseLost`。 + +### 5.3 Broker：drain 臂的升级 + +如果 drain 级联在停止 worker 之前等待 worker，那么 F1/F2 下 D1–D2 不可达。今天 `drainClaimedBinding` 先释放每个 Session，之后才调用 `stopAcknowledgedRuntime`；拒绝的 worker 会让第一步失败，于是停止永远不会被尝试。 + +修改 `drainClaimedBinding`，使一次 worker 拒绝触发一次升级： + +1. 照旧运行 `drainSessions`。 +2. 若某个 session 释放以 `managed_runtime_owner_hold_pending` 或 `managed_runtime_release_timeout` 失败，则调用 `provisioner.stopDrained(binding)`，用 `renewal.persistDrainReceipt(...)` 持久化 receipt，并再次运行 `drainSessions`。 +3. 第二遍中每个活跃 Session 都走 `releaseStoppedSession`（有 receipt、无活跃 execution），它在 `finishSessionRelease` 之前调用 `transport.releaseRetired(...)`。 +4. 第二次拒绝，或 provisioner 不支持 `supportsDrainedStop`，则以现有错误使 drain 失败。holder 随后留给 operator recovery；失败是显式的，不是静默的。 + +这符合已记录的 drain 语义（「drain 臂先停止活跃的东西」），同时保持普通释放（`drain = false`）非破坏性：只有关闭级联可以停止 worker。 + +### 5.4 顺序与崩溃窗口 + +`releaseStoppedSession` 在 `finishSessionRelease` **之前**调用 `releaseRetired`： + +- 补偿成功，然后提交持久 `RELEASED` —— 正常路径； +- 补偿失败 —— 释放失败，Session 保持 `RELEASING`，drain claim 仍可重试，holder 未被触碰； +- Broker 在两者之间崩溃 —— Session 仍活跃且没有 holder。重试走现有的无持有者捷径（`WorkspaceRuntimeTransport.java:416-421`）并完成。不会卡死，也没有把存储交给第二个写者。 + +在 `finishSessionRelease` 之后补偿严格更差：中间崩溃会留下 `RELEASED` 加一个卡住的 holder，而重试会在 `RELEASED` 行上短路，永远到不了补偿。 + +`releaseStoppedSession` 今天是同步的（`:5838`）。它将改为返回 `CompletionStage`；两个调用方（`:747` 的 `releaseSavedSession` 与 `drainSessions`）本来就是异步组合。 + +### 5.5 不改动的部分 + +- `claim` 冲突语义与 `409 workspace_busy` 契约。 +- acquire 侧保留规则与 `acquireUncertain`。 +- `releaseUnusableSession` 与 `releaseLost`。 +- `ma_storage_owner` schema。 +- `RuntimeBrokerService.mapFailure`（`:5538`），它已经原样保留 `RuntimeBrokerException`，因此新的 worker 错误码能完整到达 drain 升级。 + +## 6. 决策与理由 + +**D-1：证据门控的补偿，而不是 holder 过期或代数取代。** +过期时间或代数计数只能证明 holder 是「旧的」，永远不能证明 worker「停了」。按时间取代会让第二个 Session 占用一个活跃 worker 可能仍在写入的存储，这正是 §2.1 保护的不变量。issue 把过期列为备选；本设计拒绝它，因为它所需证据同样是 drain receipt，而持久化该证据只增加一列、不增加能力。 + +**D-2：补偿放在持久释放之前，而不是之后。** 见 §5.4。失败与崩溃窗口都收敛到可重试状态，而不是 `RELEASED` 行加卡住的 holder。 + +**D-3：只在 drain 臂升级。** 普通释放被要求不停止任何东西。关闭级联本来就拥有停止语义，因此该升级是对这个臂顺序的修补，不是新策略。 + +**D-4：不在 `releaseUnusableSession` 中补偿。** 进程已死比响应缺失更强，但它不是「孤儿子进程或存储挂载已停止写入」的证据。`releaseLost` 已经编码了更严格的证据，仍是 `LOST` 路径的归属。 + +**D-5：在备忘的地方加界。** 挂起存在于备忘的 promise 中，而不在 `releasePrepared` 的单个 await 中。在 provider 加界让 core 不必引入超时契约，并让现有 `.catch` 能重置备忘。 + +**D-6：不改 schema。** drain receipt、binding 状态与精确 holder 列已经携带证据。新增列会成为第二个事实来源。 + +## 7. 不变量 + +1. 只有 worker 被证明已停止（与 binding 匹配的 drain receipt、无活跃 execution）或已应答释放时，才清除 holder。 +2. 单纯 transport 失败永不清除 holder。 +3. 清除是精确匹配的；永远不会清除更新持有者。 +4. 被放弃的 worker 释放 attempt 永远不会为它已不拥有的 identity 改变 provider 状态。 +5. 普通释放永不停止 worker。 + +## 8. 风险与缓解 + +| 风险 | 缓解 | +| --------------------------------------------------- | ------------------------------------------------------------------------------------ | +| 补偿清除了活跃 worker 仍在写入的 holder | P1–P3；排除 `LOST`；`releaseLost` 保留自己的证据门 | +| 过期的 worker attempt 退役了重新 acquire 的 Session | `releaseAttempt` 令牌加 await 之后的 identity 检查（§5.1） | +| 截止时间已到而 `releasePrepared` 之后成功 | 被放弃的 attempt 只返回；下一次 attempt 重新运行 `releasePrepared`，对已结算条目幂等 | +| 升级停止了服务其他 Session 的共享 worker | 关闭级联按设计就是 binding 级的；`drainSessions` 会释放该 binding 上所有活跃 Session | +| provisioner 不支持 `supportsDrainedStop` | 显式失败，保留 operator recovery；记录为残留缺口 | +| `releaseStoppedSession` 签名变化扩散 | 只有两个调用方，且都已是 `CompletionStage` 组合 | +| 被放弃 promise 的 rejection 变成 unhandled | 在 `withDeadline` 中挂空 rejection 处理器 | + +## 9. 验证计划 + +**Java（H2，然后 `mysql-integration`）** + +- `WorkspaceExecutionStoreTest`：`release` 仍拒绝 `LOST`/`RELEASED` binding；精确匹配清除；不清除更新持有者。 +- `WorkspaceRuntimeTransportTest`（新增用例）： + - worker 拒绝时 holder 保持（不变量）； + - `releaseRetired` 在没有匹配 receipt 时返回 `false` 且不清除； + - `releaseRetired` 在 `DRAINING` binding 与 receipt 匹配时清除； + - 未持有时 `releaseRetired` 幂等。 +- `RuntimeBrokerServiceTest`： + - drain 级联：拒绝 → `stopDrained` → 第二遍清除 Session 与 holder； + - 普通释放：拒绝不停止 worker 也不清除； + - 第二次拒绝使 drain 显式失败。 +- 事务/SQL 行为需要 MySQL profile（按模块 AGENTS）。 + +**TypeScript** + +- `managed-runtime-provider-worker.test.ts`： + - 拒绝映射为 `managed_runtime_owner_hold_pending`； + - 超时映射为 `managed_runtime_release_timeout` 并清除备忘； + - 下一次 attempt 重新运行 `releasePrepared`； + - 之后才 resolve 的被放弃 attempt 不会关闭重新 acquire 的 Session（identity 令牌）。 +- `managed-tool-runtime.test.ts` 不变；core 行为未改。 + +**端到端** + +- 强制 worker 拒绝 `release` 的 Managed Workspace：关闭 Session；断言 `ma_storage_owner.holder_key IS NULL`、Session 到达 `RELEASED`，之后对同一存储的 `acquire` 成功。 +- 反向：没有 receipt 的 transport 级失败保持 `holder_key` 已设置，Session 停在 `RELEASING`。 + +**参考检查** + +- 对改动的 TS/Java 相邻文档运行 `npm run lint` 与 Prettier；Java 模块运行 `mvn verify`（按模块 AGENTS，在根 lint 前删除 `target/`）。 + +## 10. 验收标准 + +1. 当 worker 以 `managed_runtime_owner_hold_pending` 拒绝释放时，关闭级联会清除该 Session 的 `ma_storage_owner.holder_key`，之后对同一存储的 acquire 成功。 +2. 当 worker 的 `releasePrepared()` 永不 settle 时，释放在 `RELEASE_DEADLINE_MS` 内返回可重试错误；下一次释放是全新 attempt；Broker 不会等待超过自己的请求超时。 +3. 没有 drain receipt 的单纯 transport 失败（连接错误、客户端超时）保持 holder 已设置，Session 停在 `RELEASING`。 +4. `releaseRetired` 无法清除与精确 `(holder_key, binding_id, runtime_generation, runtime_session_id)` 元组不匹配的 holder。 +5. 无 schema 迁移，无 OpenAPI 改动。 +6. §9 的新测试通过，现有模块套件保持绿色。 + +## 11. 未决问题 + +1. **截止时间取值。** 20 秒选在 30 秒 HTTP 超时之下。是否应可配置，还是固定以保证各部署行为一致？ +2. **升级触发条件。** drain 应在第一次拒绝时升级，还是在重试预算之后（例如连续两次拒绝），以免停止一个本将结算的 worker？ +3. **控制面重试预算。** 控制面是否应在 N 次普通释放失败后自动 drain？本设计有意不动该策略。 +4. **`releaseMount` 对称性。** `releaseMount` 已经不需要 worker 应答就清除 holder。它今天安全，因为执行挂载与安装是分开的，但两种证据应在后续文档中一并说明。 + +## 12. 实施顺序 + +1. worker 拒绝错误码与有界释放（§5.1）—— 可独立测试，已经关闭 F2。 +2. `RuntimeTransport.releaseRetired` 与 `WorkspaceRuntimeTransport` 覆盖（§5.2）。 +3. `releaseStoppedSession` 补偿及其异步签名（§5.4）。 +4. drain 臂升级（§5.3）。 +5. §9 的测试，然后 MySQL profile。 diff --git a/packages/cli/src/serve/managed-runtime-provider-worker.test.ts b/packages/cli/src/serve/managed-runtime-provider-worker.test.ts index 855ba5934d..3cdb199dfc 100644 --- a/packages/cli/src/serve/managed-runtime-provider-worker.test.ts +++ b/packages/cli/src/serve/managed-runtime-provider-worker.test.ts @@ -2324,3 +2324,54 @@ const timer = setInterval(() => { await control({ kind: 'release' }); }); }); + +describe('Managed Runtime provider release deadline', () => { + let workspace: string; + let storage: string; + let executor: ManagedToolExecutor; + let provider: ManagedRuntimeProviderWorker; + + beforeEach(() => { + workspace = fs.realpathSync( + fs.mkdtempSync(path.join(os.tmpdir(), 'qwen-provider-release-')), + ); + storage = fs.mkdtempSync( + path.join(os.tmpdir(), 'qwen-provider-release-storage-'), + ); + vi.spyOn(Storage, 'getGlobalQwenDir').mockReturnValue(storage); + executor = new ManagedToolExecutor(async () => undefined); + provider = new ManagedRuntimeProviderWorker( + executor, + async () => ({ + directory: workspace, + workspaceRoot: workspace, + preapproved: true, + }), + 50, + ); + executor.attachProvider(provider); + }); + + afterEach(async () => { + await executor.close(); + vi.restoreAllMocks(); + fs.rmSync(workspace, { recursive: true, force: true }); + fs.rmSync(storage, { recursive: true, force: true }); + }); + + it('abandons a release that outlives its deadline and retries it', async () => { + await expect(provider.control(SESSION, { kind: 'acquire' })).resolves.toBe( + true, + ); + const stuck = vi + .spyOn(ManagedToolRuntime.prototype, 'releasePrepared') + .mockImplementation(() => new Promise(() => {})); + await expect( + provider.control(SESSION, { kind: 'release' }), + ).rejects.toMatchObject({ code: 'managed_runtime_release_timeout' }); + stuck.mockRestore(); + await expect(provider.control(SESSION, { kind: 'release' })).resolves.toBe( + true, + ); + }); +}); diff --git a/packages/cli/src/serve/managed-runtime-provider-worker.ts b/packages/cli/src/serve/managed-runtime-provider-worker.ts index e5d38eadd9..badfbf4fc2 100644 --- a/packages/cli/src/serve/managed-runtime-provider-worker.ts +++ b/packages/cli/src/serve/managed-runtime-provider-worker.ts @@ -93,6 +93,7 @@ interface ProviderSession { value?: ProviderRuntime; pending: number; closed: boolean; + releaseAttempt: number; release?: Promise; retirement?: Promise; } @@ -106,6 +107,33 @@ function conflict( throw new ManagedToolConflictError(message, code); } +/** + * A release the worker gives up on must stay retryable: the Broker's own + * request timeout is 30 s, so the worker answers first and keeps the decision + * in its own hands. + */ +const RELEASE_DEADLINE_MS = 20_000; + +class ReleaseDeadlineError extends Error {} + +function withDeadline( + work: Promise | undefined, + deadlineMs: number, +): Promise { + if (!work) return Promise.resolve(undefined); + let timer: ReturnType | undefined; + return Promise.race([ + work, + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new ReleaseDeadlineError()), deadlineMs); + }), + ]).finally(() => { + if (timer !== undefined) clearTimeout(timer); + // The abandoned work keeps running; it must not surface as unhandled. + void work.catch(() => undefined); + }); +} + /** * Drops the process-global entries core registers per session — the project * dir, and the model that `new Config` publishes — so a long-lived worker does @@ -234,6 +262,7 @@ export class ManagedRuntimeProviderWorker { constructor( private readonly executor: ManagedToolExecutor, private readonly contextFor: ContextResolver, + private readonly releaseDeadlineMs = RELEASE_DEADLINE_MS, ) {} hasActiveSession(sessionId: string): boolean { @@ -302,6 +331,7 @@ export class ManagedRuntimeProviderWorker { identity, closed: true, pending: 0, + releaseAttempt: 0, }); return true; } @@ -312,8 +342,34 @@ export class ManagedRuntimeProviderWorker { 'managed_runtime_owner_hold_pending', ); const entry = session; + const attempt = ++entry.releaseAttempt; + const prepared = entry.value?.runtime.releasePrepared(); entry.release = (async () => { - await entry.value?.runtime.releasePrepared(); + try { + await withDeadline(prepared, this.releaseDeadlineMs); + } catch (error) { + if (error instanceof ReleaseDeadlineError) { + conflict( + 'Managed Runtime release timed out.', + 'managed_runtime_release_timeout', + ); + } + // Whatever kept preparation from finishing still owns work here. + conflict( + error instanceof Error + ? error.message + : 'Managed Runtime release failed.', + 'managed_runtime_owner_hold_pending', + ); + } + // A timed-out attempt can finish after another attempt or a + // re-acquire took the identity; it must not retire that Session. + if ( + this.sessions.get(identity.runtimeSessionId) !== entry || + entry.releaseAttempt !== attempt + ) { + return true; + } this.executor.closeSessionAdmission(identity.runtimeSessionId); forgetSession(identity.runtimeSessionId); entry.closed = true; @@ -321,7 +377,7 @@ export class ManagedRuntimeProviderWorker { await this.retireReleased(); return true; })().catch((error: unknown) => { - entry.release = undefined; + if (entry.releaseAttempt === attempt) entry.release = undefined; throw error; }); return entry.release; @@ -332,7 +388,7 @@ export class ManagedRuntimeProviderWorker { conflict('Managed Runtime Session is closed.'); if (!session) { this.executor.claimProviderSession(identity.runtimeSessionId); - session = { identity, closed: false, pending: 0 }; + session = { identity, closed: false, pending: 0, releaseAttempt: 0 }; this.sessions.set(identity.runtimeSessionId, session); } if (!session.ready) { diff --git a/packages/cli/src/serve/managed-runtime-tool-executor.ts b/packages/cli/src/serve/managed-runtime-tool-executor.ts index a014769f86..2957093782 100644 --- a/packages/cli/src/serve/managed-runtime-tool-executor.ts +++ b/packages/cli/src/serve/managed-runtime-tool-executor.ts @@ -186,6 +186,7 @@ export class ManagedToolConflictError extends Error { | 'managed_runtime_identity_conflict' | 'managed_runtime_provider_operation_failed' | 'managed_runtime_owner_hold_pending' + | 'managed_runtime_release_timeout' | 'managed_tool_result_conflict' = 'managed_runtime_identity_conflict', ) { super(message); diff --git a/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/service/WorkspaceRuntimeTransport.java b/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/service/WorkspaceRuntimeTransport.java index af9a0bfe0e..32d33d832c 100644 --- a/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/service/WorkspaceRuntimeTransport.java +++ b/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/service/WorkspaceRuntimeTransport.java @@ -433,6 +433,32 @@ final class WorkspaceRuntimeTransport implements RuntimeTransport { }); } + @Override + public CompletionStage releaseRetired(RuntimeLease lease, RuntimeSession session) { + if (!managed(session)) { + return CompletableFuture.completedFuture(true); + } + Context context = context(lease, session, false); + RuntimeBindingRecord binding = bindings.findById(context.runtime().getBindingId()); + if (binding != null && binding.getState() == RuntimeBindingRecord.State.LOST) { + // The LOST family owns this holder; its evidence gate is stricter. + return CompletableFuture.completedFuture(true); + } + boolean proven = binding != null + && binding.getGeneration() == context.runtime().getGeneration() + && binding.getState() == RuntimeBindingRecord.State.DRAINING + && binding.getDrainReceipt() != null + && binding.getDrainReceipt().matches(binding); + if (!proven) { + // Held without proof is a refusal; nothing held is already settled. + return CompletableFuture.completedFuture( + !ownership.isHeld(context.binding(), context.session())); + } + ownership.release(context.binding(), context.session()); + detachOwed.remove(detachKey(context)); + return CompletableFuture.completedFuture(true); + } + @Override public CompletionStage releaseMount(RuntimeLease lease, RuntimeSession session) { if (!managed(session)) { diff --git a/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransport.java b/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransport.java index edbf6e438e..a2e863d7c9 100644 --- a/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransport.java +++ b/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransport.java @@ -1294,14 +1294,17 @@ public final class HttpRuntimeTransport implements RuntimeTransport { case "managed_runtime_provider_invalid", "managed_runtime_tool_invalid" -> 400; case "managed_runtime_identity_conflict", "managed_context_unavailable", "managed_context_conflict", "managed_runtime_provider_operation_failed", - "managed_runtime_provider_incompatible", "managed_runtime_owner_hold_pending" -> 409; + "managed_runtime_provider_incompatible", "managed_runtime_owner_hold_pending", + "managed_runtime_release_timeout" -> 409; case "managed_runtime_provider_too_large" -> 413; case "managed_runtime_provider_unsupported" -> 501; default -> 0; }; if (response.statusCode() == expectedStatus) { + boolean retryable = "managed_runtime_owner_hold_pending".equals(code) + || "managed_runtime_release_timeout".equals(code); return error(expectedStatus, code, - BrokerValues.requireWellFormed(message, "provider error"), false); + BrokerValues.requireWellFormed(message, "provider error"), retryable); } } catch (RuntimeBrokerException | IllegalArgumentException ignored) { // Unrecognized responses supply no provider-specific error evidence. @@ -1328,9 +1331,15 @@ public final class HttpRuntimeTransport implements RuntimeTransport { if (fields.keySet().equals(Set.of("code", "error")) && fields.get("error") instanceof String && ("managed_context_unavailable".equals(code) - || "managed_context_conflict".equals(code))) { + || "managed_context_conflict".equals(code) + || "managed_runtime_owner_hold_pending".equals(code) + || "managed_runtime_release_timeout".equals(code))) { + boolean retryable = "managed_runtime_owner_hold_pending".equals(code) + || "managed_runtime_release_timeout".equals(code); return error(409, (String) code, - "Managed Session context is unavailable or conflicts.", false); + retryable ? "Managed Runtime Session still owns work." + : "Managed Session context is unavailable or conflicts.", + retryable); } } catch (RuntimeBrokerException ignored) { // An unrecognized error body supplies no Session-scoped evidence. diff --git a/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeBrokerService.java b/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeBrokerService.java index 8dd9368328..72569e5e42 100644 --- a/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeBrokerService.java +++ b/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeBrokerService.java @@ -589,7 +589,7 @@ public final class RuntimeBrokerService implements AutoCloseable { // The sessions release — and their sweep of provably ended // background exits — run BEFORE the unsettled gate: a row the // sweep can prove never blocks the close on a stale active. - drainSessions(draining, renewal, authority, null)).thenCompose(ignored -> { + releaseDrainedSessions(draining, renewal, authority)).thenCompose(ignored -> { if (sessionRepository.countActiveByBinding(draining.getBindingId(), draining.getGeneration()) != 0 || executionRepository.hasActiveByBinding(draining.getBindingId(), draining.getGeneration())) { throw conflict("workspace_close_execution_unsettled", "Original resources are unsettled"); @@ -597,7 +597,8 @@ public final class RuntimeBrokerService implements AutoCloseable { var receipt = renewal.current.get().getDrainReceipt(); return receipt == null ? stopAcknowledgedRuntime(draining) : CompletableFuture.completedFuture(receipt); - }).thenAccept(receipt -> { + }).thenCompose(receipt -> releaseRetiredHolders(draining, renewal) + .thenApply(ignored -> receipt)).thenAccept(receipt -> { if (authority != null) { bindingRepository.requireHarnessCleanup(draining.getRequest().getScope(), draining.getRequest().getIsolationKey(), authority); } @@ -621,6 +622,69 @@ public final class RuntimeBrokerService implements AutoCloseable { }); } + /** + * Releases the drained Sessions, and escalates a worker that refuses to + * release to the durable stop this drain arm already owns. Only the close + * cascade reaches this path: a routine release never stops a worker. + */ + private CompletionStage releaseDrainedSessions(RuntimeBindingRecord binding, + BindingRenewal renewal, RuntimeLifecycleAuthority authority) { + return drainSessions(binding, renewal, authority, null).exceptionallyCompose(error -> { + if (!workerRefusedRelease(error) || !provisioner.supportsDrainedStop(binding)) { + return CompletableFuture.failedFuture(error); + } + return safeStage(() -> provisioner.stopDrained(binding)) + .thenAccept(receipt -> renewal.persistDrainReceipt(receipt, authority)) + .thenCompose(ignored -> drainSessions(binding, renewal, authority, null)); + }); + } + + private static boolean workerRefusedRelease(Throwable error) { + Throwable cause = unwrap(error); + return cause instanceof RuntimeBrokerException failure + && ("managed_runtime_owner_hold_pending".equals(failure.getCode()) + || "managed_runtime_release_timeout".equals(failure.getCode())); + } + + /** + * Clears durable Workspace ownership for Sessions this drain has retired. + * Their worker is stopped and their rows are released, so no writer can + * still hold the storage. Runs before the binding is marked RELEASED, so a + * crash retries it; the exact-match clear cannot touch a newer holder. + */ + private CompletionStage releaseRetiredHolders(RuntimeBindingRecord binding, BindingRenewal renewal) { + var current = renewal.current.get(); + // Only a persisted receipt can prove the worker stopped. On the plain + // close path every holder was cleared by the acknowledged release, so + // there is nothing to scan and nothing to clear. + if (!binding.getRequest().isManagedContext() || binding.getLease() == null + || current == null || current.getDrainReceipt() == null) { + return CompletableFuture.completedFuture(null); + } + return releaseRetiredHolders(binding, (String) null); + } + + private CompletionStage releaseRetiredHolders(RuntimeBindingRecord binding, String cursor) { + var batch = sessionRepository.findByBinding(binding.getBindingId(), binding.getGeneration(), cursor, 100); + CompletionStage work = CompletableFuture.completedFuture(null); + for (var record : batch) { + if (record.isActive()) { + continue; + } + work = work.thenCompose(ignored -> safeStage(() -> + transport.releaseRetired(binding.getLease(), record.getSession()))) + .thenAccept(retired -> { + if (!Boolean.TRUE.equals(retired)) { + throw unavailable("runtime_session_release_failed", + "Retired Session still owns Workspace storage"); + } + }); + } + return work.thenCompose(ignored -> batch.size() < 100 + ? CompletableFuture.completedFuture(null) + : releaseRetiredHolders(binding, batch.getLast().getRuntimeSessionId())); + } + private CompletionStage drainSessions(RuntimeBindingRecord binding, BindingRenewal renewal, RuntimeLifecycleAuthority authority, String cursor) { var batch = sessionRepository.findByBinding(binding.getBindingId(), binding.getGeneration(), cursor, 100); diff --git a/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeTransport.java b/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeTransport.java index fe1f5ee910..1e310bae89 100644 --- a/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeTransport.java +++ b/packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeTransport.java @@ -209,6 +209,18 @@ public interface RuntimeTransport { false)); } + /** + * Finishes durable ownership for a Session the Broker has already retired. + * The caller must hold durable proof that the Runtime stopped holding the + * Session — a persisted drain receipt matching the binding and no active + * executions — and must never call this on a mere transport failure. A + * transport without durable ownership has nothing to clear. + */ + default CompletionStage releaseRetired(RuntimeLease lease, + RuntimeSession session) { + return CompletableFuture.completedFuture(true); + } + /** * Claims the Workspace execution mount again for an acquired Session * whose mount was handed back by {@link #releaseMount}; a mount the diff --git a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/service/WorkspaceRuntimeTransportTest.java b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/service/WorkspaceRuntimeTransportTest.java index 1b143d1a89..a043f6a1f0 100644 --- a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/service/WorkspaceRuntimeTransportTest.java +++ b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/service/WorkspaceRuntimeTransportTest.java @@ -6,6 +6,7 @@ import com.alibaba.qwen.code.runtimebroker.ManagedCsiProtocol; import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRecord; import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRepository; import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException; +import com.alibaba.qwen.code.runtimebroker.RuntimeDrainReceipt; import com.alibaba.qwen.code.runtimebroker.RuntimeLease; import com.alibaba.qwen.code.runtimebroker.RuntimeProvisionRequest; import com.alibaba.qwen.code.runtimebroker.RuntimeProvisionSeed; @@ -245,6 +246,36 @@ class WorkspaceRuntimeTransportTest { fixture.request(), fixture.capture()).toCompletableFuture().join(); } + @Test + void releaseRetiredClearsTheHolderWithAMatchingReceipt() throws Exception { + var fixture = sshAcquisitionFixture(); + var runtime = fixture.runtime(); + var receipt = new RuntimeDrainReceipt(runtime.getBindingId(), runtime.getGeneration(), + runtime.getProvisionSeed().getProvisionRequestId(), runtime.getResourceHandle(), Instant.now()); + var draining = runtime.withDrainRequested(true, Instant.now()) + .withState(RuntimeBindingRecord.State.DRAINING, runtime.getLease(), Instant.now()) + .withDrainReceipt(receipt); + when(fixture.bindings().findById(runtime.getBindingId())).thenReturn(draining); + + assertThat(fixture.transport().releaseRetired(fixture.lease(), fixture.session()) + .toCompletableFuture().join()).isTrue(); + verify(fixture.ownership()).release(any(), any()); + } + + @Test + void releaseRetiredRefusesWithoutAPersistedReceipt() throws Exception { + var fixture = sshAcquisitionFixture(); + var runtime = fixture.runtime(); + var draining = runtime.withDrainRequested(true, Instant.now()) + .withState(RuntimeBindingRecord.State.DRAINING, runtime.getLease(), Instant.now()); + when(fixture.bindings().findById(runtime.getBindingId())).thenReturn(draining); + when(fixture.ownership().isHeld(any(), any())).thenReturn(true); + + assertThat(fixture.transport().releaseRetired(fixture.lease(), fixture.session()) + .toCompletableFuture().join()).isFalse(); + verify(fixture.ownership(), never()).release(any(), any()); + } + private static void unavailable(Runnable action) { assertThatThrownBy(action::run).isInstanceOf(RuntimeBrokerException.class) .extracting(error -> ((RuntimeBrokerException) error).getCode()).isEqualTo("workspace_unavailable"); diff --git a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/runtimebroker/ProviderRuntimeTransportTest.java b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/runtimebroker/ProviderRuntimeTransportTest.java index 0e96adc049..49dbc25aad 100644 --- a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/runtimebroker/ProviderRuntimeTransportTest.java +++ b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/runtimebroker/ProviderRuntimeTransportTest.java @@ -135,17 +135,18 @@ class ProviderRuntimeTransportTest { @Test void preservesRecognizedProviderErrorReasonsAndStatusWithoutTreatingRefusalAsNotStarted() { - for (Map.Entry failure : Map.of( - "managed_runtime_tool_invalid", 400, - "managed_runtime_provider_invalid", 400, - "managed_runtime_identity_conflict", 409, - "managed_context_unavailable", 409, - "managed_context_conflict", 409, - "managed_runtime_provider_operation_failed", 409, - "managed_runtime_provider_incompatible", 409, - "managed_runtime_owner_hold_pending", 409, - "managed_runtime_provider_too_large", 413, - "managed_runtime_provider_unsupported", 501).entrySet()) { + for (Map.Entry failure : Map.ofEntries( + Map.entry("managed_runtime_tool_invalid", 400), + Map.entry("managed_runtime_provider_invalid", 400), + Map.entry("managed_runtime_identity_conflict", 409), + Map.entry("managed_context_unavailable", 409), + Map.entry("managed_context_conflict", 409), + Map.entry("managed_runtime_provider_operation_failed", 409), + Map.entry("managed_runtime_provider_incompatible", 409), + Map.entry("managed_runtime_owner_hold_pending", 409), + Map.entry("managed_runtime_release_timeout", 409), + Map.entry("managed_runtime_provider_too_large", 413), + Map.entry("managed_runtime_provider_unsupported", 501)).entrySet()) { status = failure.getValue(); response = body -> Map.of("code", failure.getKey(), "error", "Specific provider reason."); CompletionException exception = assertThrows(CompletionException.class, @@ -154,7 +155,10 @@ class ProviderRuntimeTransportTest { assertEquals(status, error.getStatusCode()); assertEquals(failure.getKey(), error.getCode()); assertEquals("Specific provider reason.", error.getMessage()); - assertEquals(false, error.isRetryable()); + assertEquals( + "managed_runtime_owner_hold_pending".equals(failure.getKey()) + || "managed_runtime_release_timeout".equals(failure.getKey()), + error.isRetryable()); } }