fix(managed-agent): settle the no-tool cancellation takeover instead of refusing forever (Arm B)

The round-6 real-stack wedge: a Turn with no Runtime work parked mid
model round, its Harness generation dead, cancelled — the no-tool arm of
the takeover branch answered the baseline retriable 409 on every
cancellation-side redispatch, and with the Turn still carrying its
submission mark, the retry budget never fired. Turn 2 stayed CANCELLING
forever and every later Turn met the same wall.

The fix keeps the wire honest and the settlement explicit:

- LoadHarnessSession carries `cancellationTakeover`, set ONLY by
  recoverManagedRuntime's cancellation arm. A plain passive re-attach
  shares the passive wire shape and deliberately never sets it — nothing
  may mint a canned CANCELLED record for a wait whose owner could still
  exist.
- With the signal present, the no-tool arm of the takeover branch
  settles the ownerless park into the daemon's journal itself
  (turn_result/cancelled, landing turn.settled); the writer fence on the
  very load is the safety boundary (it proves the producing generation
  can never write again). On write failure the arm keeps the baseline
  retriable 409. Without the signal the arm keeps its merge-baseline
  answers — retriable 409 on cancel, typed model_start decline on
  drive — so the collision guard holds by construction.

Witnesses: `settles the cancellation of an ownerless parked no-tool Turn
on the takeover load` (200 plain attach + durable settle tag + cancel
reads settled-at-tail as 204 + next prompt admits; the revert flips the
verdicts red) with the collision-guard passive-only rerun still at 409,
plus wire-side `carriesTheCancellationTakeoverFlagOnlyOnTheCancellationLoad`
asserting the flag travels only with the signal.

Verification: TS hosted suites 259/259, qwencode 32/32,
QwenHostedHarnessConnectorTest 34/34, HarnessCoordinatorTest 52/52,
typecheck, eslint, prettier. Design doc updated in both languages.

docs/design/2026-10-02-hosted-replaceable-harness.md

@@ -270,6 +270,21 @@ otherwise a boot-identity adoption evicts the stale-boot entry, puts a
 snapshot-carrying attachment under the live boot, and the next dispatch's
 cached branch answers "nothing parked" (R11-3).
 
+The remaining no-tool cancellation wedge (Arm B) is closed with the same
+settlement principle, this time fed by intent rather than by state: a
+Turn with no Runtime work whose Harness generation died can never be
+driven, so its cancelled terminal can only be a journal record — and only
+an EXPLICIT cancellation may write it. `LoadHarnessSession` therefore
+carries the boolean `cancellationTakeover` (set ONLY by
+`recoverManagedRuntime`'s cancellation arm; a plain passive re-attach
+shares the passive wire shape and must never mint anything). With the
+signal present, the no-tool arm of the takeover branch settles the park
+into the journal itself (`turn_result`/`cancelled`, which lands the
+`turn.settled` event); without it, the branch keeps the baseline retriable
+refusal for cancels and the typed `model_start` decline for drives. The
+writer fence on the very load is what makes the settle safe: it proves
+the producing generation can never write again (Arm B 2026-10-06).
+
 Thrown errors stay transient, exactly as today. The load route answers
 declines with new 409 code `hosted_turn_recovery_declined` plus a `reason`
 field; inside the takeover branch, `hosted_turn_recovery_required` is

docs/design/2026-10-02-hosted-replaceable-harness.zh-CN.md

@@ -231,6 +231,18 @@ passive 重附着与恢复 load 一样携带恢复快照，所以 connector 在
 boot 条目、再在活 boot 下放入携快照的 attachment，而下一次派发的
 缓存分支会对着它回答「没有停靠」（R11-3）。
 
+残余的无工具取消楔死（Arm B）按同一结算原则关闭，只是这次喂它的是
+**意图**而不是状态：没有 Runtime 工作的 Turn 在其 Harness 代死后
+再也无法被驱动，所以它的取消终态只能是 journal 里的一条记录——而
+只有**显式的取消**才有资格写它。`LoadHarnessSession` 因此增加布尔
+`cancellationTakeover`（只由 `recoverManagedRuntime` 的取消臂设
+置；普通 passive 重附着共享 passive 线型却绝不能代它铸任何记录）。
+带旗信号时，接管分支的无工具臂自己把停靠写进 journal
+（`turn_result`/`cancelled`，该记录随即带来 `turn.settled` 事件）；
+不带时，该臂保持基线：取消侧可重试 409、drive 侧类型化 `model_start`
+decline。这次写入之所以安全，正是同一次 load 自带的 writer fence：
+它证明生产代再也写不了一字（Arm B 2026-10-06）。
+
 抛出的错误保持瞬时，与今天完全一致。load 路由对 decline 回答新的
 409 code `hosted_turn_recovery_declined` 并带 `reason` 字段；在接管
 分支内部，`hosted_turn_recovery_required` 此后只为瞬态发出（路由上

packages/cli/src/serve/hosted-harness-session.test.ts

@@ -9222,6 +9222,110 @@ describe('Hosted Harness Runtime turn takeover', () => {
     ).toBe(true);
   });
 
+  it('settles the cancellation of an ownerless parked no-tool Turn on the takeover load', async () => {
+    // Arm B: a no-tool Turn parked mid model round (the settlement write
+    // failed, then its Harness generation died for good). The takeover
+    // load carrying the explicit cancellation signal settles the park
+    // itself — the writer fence on the load proves the producer
+    // generation is dead — instead of refusing forever (round-6 wedge).
+    const server = await app(true);
+    const created = await headers(supertest(server).post('/session')).send({
+      sessionId: SESSION_ID,
+      sessionScope: 'thread',
+      managedSessionStore: store(),
+    });
+    expect(created.status).toBe(200);
+    const clientId = created.body.clientId as string;
+    const originalWrite = ManagedSessionRecordSink.prototype.write;
+    const write = vi
+      .spyOn(ManagedSessionRecordSink.prototype, 'write')
+      .mockImplementation(async function (
+        this: ManagedSessionRecordSink,
+        record,
+      ) {
+        if (record.subtype === 'turn_result')
+          throw new Error('settlement unavailable');
+        return originalWrite.call(this, record);
+      });
+    const prompt = [{ type: 'text', text: 'hello' }];
+    const payloadDigest = `sha256:${createHash('sha256').update(JSON.stringify(prompt)).digest('hex')}`;
+    await headers(supertest(server).post(`/session/${SESSION_ID}/prompt`))
+      .set('X-Qwen-Client-Id', clientId)
+      .send({ prompt, promptId: PROMPT_ID, payloadDigest })
+      .expect(202);
+    await vi.waitFor(
+      async () => {
+        const status = await headers(
+          supertest(server).get(`/session/${SESSION_ID}/status`),
+        ).set('X-Qwen-Client-Id', clientId);
+        expect(status.body.hasActivePrompt).toBe(false);
+        expect(status.body.recoveryBlocked).toBe(true);
+      },
+      { timeout: 10_000 },
+    );
+    write.mockRestore();
+    await headers(supertest(server).delete(`/session/${SESSION_ID}`));
+    const log = vi
+      .spyOn(stdio, 'writeStderrLineSafe')
+      .mockImplementation(() => {});
+    // Without the explicit signal the very same shape keeps the baseline
+    // retriable refusal first: nothing may mint a canned CANCELLED record
+    // for a wait whose owner could still exist (the collision guard).
+    const parkedRefused = await replacementHeaders(
+      supertest(replacementApp()).post(`/session/${SESSION_ID}/load`),
+    ).send({
+      managedSessionStore: storeFor(BOOT_ID_2),
+      passiveManagedRuntimeRecovery: true,
+    });
+    expect(parkedRefused.status).toBe(409);
+    expect(parkedRefused.body.code).toBe('hosted_turn_recovery_required');
+    const replacement = replacementApp();
+    const taken = await replacementHeaders(
+      supertest(replacement).post(`/session/${SESSION_ID}/load`),
+    ).send({
+      managedSessionStore: storeFor(BOOT_ID_2),
+      passiveManagedRuntimeRecovery: true,
+      cancellationTakeover: true,
+    });
+    expect(taken.status).toBe(200);
+    expect(taken.body.recoveryRequired).toBeUndefined();
+    expect(
+      log.mock.calls
+        .map(([line]) => line)
+        .some((line) => line.includes('settles the cancelled park on load')),
+    ).toBe(true);
+    // The cancelled terminal is durable: the plain cancel route reads it
+    // back as settled-at-tail (204, nothing left to abort), and the next
+    // prompt admits — the park is closed, not hidden.
+    const cancelled = await replacementHeaders(
+      supertest(replacement).post(`/session/${SESSION_ID}/cancel`),
+    ).set('X-Qwen-Client-Id', taken.body.clientId as string);
+    expect(cancelled.status).toBe(204);
+    const fresh = [
+      { type: 'text', text: 'second turn after the cancelled park' },
+    ];
+    const admitted = await replacementHeaders(
+      supertest(replacement).post(`/session/${SESSION_ID}/prompt`),
+    )
+      .set('X-Qwen-Client-Id', taken.body.clientId as string)
+      .send({
+        prompt: fresh,
+        promptId: randomUUID(),
+        payloadDigest: `sha256:${createHash('sha256').update(JSON.stringify(fresh)).digest('hex')}`,
+      });
+    expect(admitted.status).toBe(202);
+    await vi.waitFor(
+      async () => {
+        const status = await replacementHeaders(
+          supertest(replacement).get(`/session/${SESSION_ID}/status`),
+        ).set('X-Qwen-Client-Id', taken.body.clientId as string);
+        expect(status.body.hasActivePrompt).toBe(false);
+        expect(status.body.recoveryBlocked).toBe(false);
+      },
+      { timeout: 10_000 },
+    );
+  });
+
   it('keeps the held Runtime lease when a redriven load fails transiently', async () => {
     await parkToolTurn();
     vi.spyOn(HostedWorkspaceBroker.prototype, 'execute').mockResolvedValue({

packages/cli/src/serve/hosted-harness-session.ts

@@ -533,6 +533,33 @@ async function recoverCancelledPreToolHook(
   return true;
 }
 
+// The cancellation-side settle of a provably ownerless parked Turn: the
+// LoadHarnessSession.cancellationTakeover signal names the user's CANCEL
+// intent, and the load's own writer fence proves the producing generation
+// is dead — no execution anywhere can carry it again, so the cancelled
+// terminal can only be a journal record, written here through the same
+// sink translation the hook settle uses (turn.settled follows from the
+// record). On write failure the caller keeps the baseline retriable
+// refusal: nothing terminal is claimed about what could not be proven.
+async function settleCancelledHarnessTurn(
+  managed: ManagedSession,
+  session: HostedSession,
+  sessionId: string,
+  promptId: string,
+): Promise<void> {
+  await managed.sink.write(
+    record(session, sessionId, 'system', null, {
+      subtype: 'turn_result',
+      systemPayload: {
+        promptId,
+        state: 'cancelled',
+        stopReason: 'cancelled',
+        endedAt: Date.now(),
+      },
+    }),
+  );
+}
+
 export async function settleCancelledHookTurn(
   session: HostedSession,
 ): Promise<void> {
@@ -2287,85 +2314,120 @@ export function registerHostedHarnessSessionRoutes(
         // the tools to settle exist. Attaching one here would stand up a
         // Session whose parked Turn no route may resolve (R5-2' round).
         if (toolProfile === undefined || !brokerOptions) {
-          writeStderrLineSafe(
-            `qwen serve: Hosted Session ${sessionId} load refused (takeover_unavailable): profile=${toolProfile ?? 'none'} broker=${brokerOptions ? 'ready' : 'none'}`,
-          );
-          await managed.close();
-          if (body?.['passiveManagedRuntimeRecovery'] === true)
-            error(res, 409, 'hosted_turn_recovery_required');
-          else recoveryDeclined(res, 'model_start');
-          return;
-        }
-        try {
-          const outcome = await recoverHostedRuntimeTurn({
-            session: managed,
-            sessionId,
-            cwd,
-            promptId: unsettled,
-            brokerOptions,
-            passive: body?.['passiveManagedRuntimeRecovery'] === true,
-          });
-          if (outcome.kind === 'inapplicable') {
-            // R11-2: inapplicable pays only where a settlement route can.
-            // A requested approval keeps the plain attach (the resolve
-            // route writes the decision durably); turn_settled is settled
-            // HERE — the bare branch's projection never ran on this arm,
-            // so the missing terminal record is written by this load
-            // itself — or the load keeps the retriable refusal when the
-            // projection cannot pay either.
-            const inapplicableVerdict = await managed.authority
-              .harnessRunAuthorization()
-              .catch(() => undefined);
-            const approvalPending =
-              inapplicableVerdict?.status === 'runnable' &&
-              inapplicableVerdict.checkpoint.approval?.state === 'requested';
-            if (approvalPending) {
-              inapplicableAnswer = true;
-            } else {
-              const settle = await settleProjectablePromptId(
+          // An explicit cancellation taking over is the one caller that
+          // CAN be paid here: the cancelled terminal of the provably
+          // ownerless park lives nowhere else, so this load writes it
+          // itself (the writer fence on this very load proves the
+          // producing generation is dead), then continues as the plain
+          // attach the replay answers with. Without that signal the arm
+          // keeps its baseline answers — re-issue on cancel, model_start
+          // on drive — because nothing may mint a canned CANCELLED record
+          // for a wait whose owner could still exist (Arm B, R-round 6).
+          if (body?.['cancellationTakeover'] === true) {
+            try {
+              await settleCancelledHarnessTurn(
                 managed,
                 session,
-                fileHistory,
+                sessionId,
                 unsettled,
               );
-              if (settle === null) {
-                await managed.close();
-                writeStderrLineSafe(
-                  `qwen serve: Hosted Session ${sessionId} load refused (takeover_inapplicable_unpayable): prompt=${unsettled}`,
-                );
-                error(res, 409, 'hosted_turn_recovery_required');
-                return;
-              }
-              settlePromptId = settle;
+              writeStderrLineSafe(
+                `qwen serve: Hosted Session ${sessionId} settles the cancelled park on load: prompt=${unsettled}`,
+              );
+              // The journal owes nothing more: the cancelled record is
+              // the plain attach's whole answer, replayed home from the
+              // kept watermark.
               inapplicableAnswer = true;
+            } catch (cause) {
+              writeStderrLineSafe(
+                `qwen serve: Hosted Session ${sessionId} load refused (takeover_unavailable): profile=${toolProfile ?? 'none'} broker=${brokerOptions ? 'ready' : 'none'} settle=${String(cause)}`,
+              );
+              await managed.close();
+              error(res, 409, 'hosted_turn_recovery_required');
+              return;
             }
-          } else if (outcome.kind === 'declined') {
+          } else {
             writeStderrLineSafe(
-              `qwen serve: Hosted Session ${sessionId} load refused (takeover_unrecovered): prompt=${unsettled} reason=${outcome.reason}`,
+              `qwen serve: Hosted Session ${sessionId} load refused (takeover_unavailable): profile=${toolProfile ?? 'none'} broker=${brokerOptions ? 'ready' : 'none'}`,
             );
             await managed.close();
-            recoveryDeclined(res, outcome.reason);
+            if (body?.['passiveManagedRuntimeRecovery'] === true)
+              error(res, 409, 'hosted_turn_recovery_required');
+            else recoveryDeclined(res, 'model_start');
+            return;
+          }
+        } else {
+          try {
+            const outcome = await recoverHostedRuntimeTurn({
+              session: managed,
+              sessionId,
+              cwd,
+              promptId: unsettled,
+              brokerOptions,
+              passive: body?.['passiveManagedRuntimeRecovery'] === true,
+            });
+            if (outcome.kind === 'inapplicable') {
+              // R11-2: inapplicable pays only where a settlement route can.
+              // A requested approval keeps the plain attach (the resolve
+              // route writes the decision durably); turn_settled is settled
+              // HERE — the bare branch's projection never ran on this arm,
+              // so the missing terminal record is written by this load
+              // itself — or the load keeps the retriable refusal when the
+              // projection cannot pay either.
+              const inapplicableVerdict = await managed.authority
+                .harnessRunAuthorization()
+                .catch(() => undefined);
+              const approvalPending =
+                inapplicableVerdict?.status === 'runnable' &&
+                inapplicableVerdict.checkpoint.approval?.state === 'requested';
+              if (approvalPending) {
+                inapplicableAnswer = true;
+              } else {
+                const settle = await settleProjectablePromptId(
+                  managed,
+                  session,
+                  fileHistory,
+                  unsettled,
+                );
+                if (settle === null) {
+                  await managed.close();
+                  writeStderrLineSafe(
+                    `qwen serve: Hosted Session ${sessionId} load refused (takeover_inapplicable_unpayable): prompt=${unsettled}`,
+                  );
+                  error(res, 409, 'hosted_turn_recovery_required');
+                  return;
+                }
+                settlePromptId = settle;
+                inapplicableAnswer = true;
+              }
+            } else if (outcome.kind === 'declined') {
+              writeStderrLineSafe(
+                `qwen serve: Hosted Session ${sessionId} load refused (takeover_unrecovered): prompt=${unsettled} reason=${outcome.reason}`,
+              );
+              await managed.close();
+              recoveryDeclined(res, outcome.reason);
+              return;
+            } else if (outcome.kind === 'recovered') {
+              recovery = outcome.turn.report;
+              if (outcome.turn.acquiredRuntime)
+                session.runtimeLeaseHeld =
+                  outcome.turn.report.executions[0]?.runtimeSessionId ??
+                  outcome.turn.promptId;
+            }
+            // inapplicable: nothing a takeover owes this payload — the load
+            // continues as the plain attach it was before G3, so a requested
+            // approval or a cancellation-only load meets its own path.
+          } catch (cause) {
+            await managed.close();
+            writeStderrLineSafe(
+              `qwen serve: Hosted Harness recovery of session ${sessionId} failed: ${String(cause)}`,
+            );
+            // A failed takeover keeps the turn parked for the next attempt:
+            // refuse exactly like a plain recovery refusal so the coordinator
+            // retries instead of failing the Turn.
+            error(res, 409, 'hosted_turn_recovery_required');
             return;
-          } else if (outcome.kind === 'recovered') {
-            recovery = outcome.turn.report;
-            if (outcome.turn.acquiredRuntime)
-              session.runtimeLeaseHeld =
-                outcome.turn.report.executions[0]?.runtimeSessionId ??
-                outcome.turn.promptId;
           }
-          // inapplicable: nothing a takeover owes this payload — the load
-          // continues as the plain attach it was before G3, so a requested
-          // approval or a cancellation-only load meets its own path.
-        } catch (cause) {
-          await managed.close();
-          writeStderrLineSafe(
-            `qwen serve: Hosted Harness recovery of session ${sessionId} failed: ${String(cause)}`,
-          );
-          // A failed takeover keeps the turn parked for the next attempt:
-          // refuse exactly like a plain recovery refusal so the coordinator
-          // retries instead of failing the Turn.
-          error(res, 409, 'hosted_turn_recovery_required');
-          return;
         }
       } else if (
         restore.recoveryStatus === 'ok' &&

packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/harness/QwenHostedHarnessConnector.java

@@ -486,11 +486,18 @@ public class QwenHostedHarnessConnector implements HarnessConnector {
     private HarnessSessionRef load(SessionRecord session,
             boolean passiveManagedRuntimeRecovery,
             boolean driveRuntimeRecovery) {
+        return load(session, passiveManagedRuntimeRecovery,
+                driveRuntimeRecovery, false);
+    }
+
+    private HarnessSessionRef load(SessionRecord session,
+            boolean passiveManagedRuntimeRecovery,
+            boolean driveRuntimeRecovery, boolean cancellationTakeover) {
         String profile = toolProfile(session);
         ManagedSessionStoreConnection store = managedSessionStore(session);
         return client().loadSession(new LoadHarnessSession(session.sessionId(), store,
                 passiveManagedRuntimeRecovery, profile,
-                driveRuntimeRecovery));
+                driveRuntimeRecovery, cancellationTakeover));
     }
 
     @Override
@@ -536,7 +543,7 @@ public class QwenHostedHarnessConnector implements HarnessConnector {
         HarnessSessionRef cached = attachments.get(key);
         if (cached == null) {
             HarnessSessionRef attached = load(session, cancellation,
-                    !cancellation);
+                    !cancellation, cancellation);
             attachments.put(key, attached);
             if (attached.getRuntimeRecovery() != null) {
                 pendingRecovery.add(key);

packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/LoadHarnessSession.java

@@ -10,6 +10,7 @@ public final class LoadHarnessSession {
     private final boolean passiveManagedRuntimeRecovery;
     private final String toolProfile;
     private final boolean driveRuntimeRecovery;
+    private final boolean cancellationTakeover;
 
     public LoadHarnessSession(String harnessSessionId) {
         this(harnessSessionId, null, false);
@@ -37,12 +38,27 @@ public final class LoadHarnessSession {
             ManagedSessionStoreConnection managedSessionStore,
             boolean passiveManagedRuntimeRecovery, String toolProfile,
             boolean driveRuntimeRecovery) {
+        this(harnessSessionId, managedSessionStore,
+                passiveManagedRuntimeRecovery, toolProfile,
+                driveRuntimeRecovery, false);
+    }
+
+    /** The cancellation flag is deliberately carried separately from
+     * {@code passiveManagedRuntimeRecovery}: a plain passive re-attach and
+     * a cancellation takeover share that wire shape today, and only an
+     * explicit cancellation may let the daemon settle a parked Turn whose
+     * producer is proven dead — anything else would stamp a live wait. */
+    public LoadHarnessSession(String harnessSessionId,
+            ManagedSessionStoreConnection managedSessionStore,
+            boolean passiveManagedRuntimeRecovery, String toolProfile,
+            boolean driveRuntimeRecovery, boolean cancellationTakeover) {
         this.harnessSessionId = HostedHarnessClient.requireUuid(
                 harnessSessionId, "harnessSessionId");
         this.managedSessionStore = managedSessionStore;
         this.passiveManagedRuntimeRecovery = passiveManagedRuntimeRecovery;
         this.toolProfile = toolProfile;
         this.driveRuntimeRecovery = driveRuntimeRecovery;
+        this.cancellationTakeover = cancellationTakeover;
     }
 
     String getHarnessSessionId() {
@@ -67,6 +83,9 @@ public final class LoadHarnessSession {
         if (driveRuntimeRecovery) {
             result.put("driveRuntimeRecovery", true);
         }
+        if (cancellationTakeover) {
+            result.put("cancellationTakeover", true);
+        }
         return result;
     }
 }

packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/HostedHarnessClientTest.java

@@ -448,6 +448,32 @@ class HostedHarnessClientTest {
                 "\"passiveManagedRuntimeRecovery\":true"));
     }
 
+    // The cancellation flag rides the wire separately: a plain passive
+    // re-attach omits it, and only an explicit cancellation takeover adds
+    // the field — anything else would let the daemon mint a canned
+    // CANCELLED record for a wait whose owner could still exist (Arm B).
+    @Test
+    void carriesTheCancellationTakeoverFlagOnlyOnTheCancellationLoad() {
+        AtomicReference<String> loadBody = new AtomicReference<>();
+        server.createContext("/session/" + SESSION_ID + "/load",
+                exchange -> {
+                    loadBody.set(new String(exchange.getRequestBody()
+                            .readAllBytes(), StandardCharsets.UTF_8));
+                    sendSessionJson(exchange, 200, sessionJson());
+                });
+        try (HostedHarnessClient client = newClient()) {
+            client.loadSession(new LoadHarnessSession(SESSION_ID, null,
+                    true));
+        }
+        assertFalse(loadBody.get().contains("cancellationTakeover"));
+        try (HostedHarnessClient client = newClient()) {
+            client.loadSession(new LoadHarnessSession(SESSION_ID, null,
+                    true, null, false, true));
+        }
+        assertTrue(loadBody.get().contains(
+                "\"cancellationTakeover\":true"));
+    }
+
     @Test
     void parsesAResultsReadyRuntimeRecovery() {
         AtomicReference<String> continuationBody = new AtomicReference<>();