fix(managed): Align checkpoint limits and expose recovery blocks (#15094)

Co-authored-by: Qwen-Coder <[email protected]>

docs/design/2026-10-22-managed-checkpoint-recovery.md

@@ -0,0 +1,25 @@
+# Managed checkpoint storage and live recovery blocks
+
+[English](2026-10-22-managed-checkpoint-recovery.md) | [简体中文](2026-10-22-managed-checkpoint-recovery.zh-CN.md)
+
+## Problem and evidence
+
+A long Hosted Turn accumulates completed Runtime receipts in its recovery checkpoint. Those entries are the durable repeat-dispatch fence and must remain present. An isolated reproduction on main completed 73 small calls before the next `managed-checkpoint` reached 65,776 bytes and met the ordinary 65,536-byte HTTP resource limit. The existing checkpoint reader already accepts up to 1 MiB. The Harness then reported no active prompt and a recovery block, while the Java consumer kept waiting on an event stream that continued heartbeats.
+
+## Decisions
+
+Allow the `managed-checkpoint` resource kind up to the existing 1 MiB checkpoint parser bound in the TypeScript publisher, Java Session Store, tool publication references and CSI native readback grants. Preserve all receipt identities, ordinals, references, digests and existing resource/transaction bounds. Other ordinary resources retain 64 KiB. This introduces no new resource encoding, database schema, object-storage dependency or configuration option. TypeScript publisher oversize errors identify the resource kind, actual bytes and limit.
+
+Inspect the status of the exact tenant/session/generation attachment every 15 seconds, independently of opaque journal control records. Only `recoveryBlocked=true` together with `hasActivePrompt=false` establishes a live recovery block. Healthy model execution and approval waits remain active. Drain and flush already-read events, then use the existing transcript API to confirm there are no authoritative journal records beyond the consumed cursor. Recheck the local event queue after the probe so an arriving terminal event takes priority. Park only under the live dispatch lease. Store a named recovery-blocked error and prevent execution redispatch and idempotent command replays. After at least 15 seconds, the dispatcher may claim it only for read-only status and journal observation at its original epoch and cursor; this branch never warms a Runtime, takes over execution, submits input or continues a checkpoint. Observation requires the original cached attachment and captures one client for all reads; a cold cache never loads or takes over a Session. Missing, unavailable or changed-generation evidence retains the park without consuming the execution retry budget. Later authoritative terminal events settle the original Turn. Until then, retain its unsettled input, Runtime ownership, policy-change barrier, lifecycle barrier and child-result watch. A sampled block alone must not close the Session or classify a child outcome unknown. Explicit cancellation remains available.
+
+The WebShell read adapter presents this named park as `recovery_blocked`. Its existing summary polling updates the page without treating stream heartbeats as work progress. The browser disables another submission to that Session while retaining cancellation. The raw internal Turn remains active for cancellation and recovery; the change does not invent a physical stop receipt.
+
+## Scope and rollout
+
+This is a bounded checkpoint budget correction and live-block observation, not a general large-resource backend. Checkpoints above 1 MiB remain refused and must surface as blocked. Deploy the Java reader before a Harness can publish checkpoints above 64 KiB; older stores will still refuse them. Existing resources and journals remain readable, and the SQL schema is unchanged. User previews are not restarted until isolated validation passes.
+
+## Validation and acceptance
+
+Reproduce the baseline refusal with tiny tool output, then run the same 100-call fixture to completion without deleting receipts. Verify a cold reopen, unchanged call identities, and refusal above 1 MiB while ordinary resources still stop at 64 KiB. Exercise actual Java resource commit/read and reader compatibility.
+
+Verify a heartbeat-only blocked Harness leaves the Java consumer, becomes durably parked and appears blocked through the WebShell adapter and browser provider. Healthy inactive approval waits, active model work, cancellation races and lost dispatch leases must not be misclassified. Verify bounded re-observation consumes a later terminal without replay, and unavailable evidence retains the park without consuming retries. Run affected TypeScript tests, build/typecheck/bundle, Java verification, applicable MySQL/Hosted recovery checks and independent end-to-end verification. Record unavailable environment gates explicitly.

docs/design/2026-10-22-managed-checkpoint-recovery.zh-CN.md

@@ -0,0 +1,25 @@
+# Managed 检查点存储与运行中恢复阻塞
+
+[English](2026-10-22-managed-checkpoint-recovery.md) | [简体中文](2026-10-22-managed-checkpoint-recovery.zh-CN.md)
+
+## 问题与证据
+
+较长的 Hosted Turn 会将已完成 Runtime 调用的收据累积进恢复检查点。这些条目是持久化的重复执行保护，必须保留。在 main 上的隔离复现中，73 次小调用完成后，下一份 `managed-checkpoint` 达到 65,776 字节，触发普通 HTTP 资源的 65,536 字节限制。现有检查点读取器已经支持最多 1 MiB。随后 Harness 报告没有活动 prompt 且恢复受阻，但 Java 消费端仍等待持续发送心跳的事件流。
+
+## 决策
+
+在 TypeScript 发布端、Java Session Store、工具发布引用和 CSI 原生回读授权中，将 `managed-checkpoint` 资源的预算对齐现有检查点解析上限 1 MiB。保留所有收据身份、ordinal、引用、摘要，以及现有资源和事务边界。其他普通资源保持 64 KiB。无需新增资源编码、数据库表结构、对象存储依赖或配置项。TypeScript 发布端超限错误写明资源类型、实际字节数与上限。
+
+每 15 秒读取同一 tenant/session/generation 附着的状态，不受 opaque journal 控制记录续期影响。只有同时满足 `recoveryBlocked=true` 和 `hasActivePrompt=false` 才认定运行中的恢复阻塞。正常模型执行和审批等待继续保持活动。先消费并刷新已读取事件，再通过现有 transcript API 确认权威 journal 在已消费 cursor 之后没有待处理记录；探测后再次检查本地事件队列，让刚到达的完成事件优先。仅在当前 dispatch 租约约束下停放 Turn。保存具名的恢复阻塞错误，禁止重新 dispatch 执行和幂等命令重试。至少等待 15 秒后，调度器只可 claim 该 Turn，按原 epoch 和 cursor 只读观测状态与 journal；此分支不预热 Runtime、不接管执行、不提交输入，也不继续检查点。观察必须使用原有缓存附着，并让所有读取固定使用同一个 client；缓存缺失时不加载或接管 Session。缺失、不可读或世代变化的证据继续保持停放，不消耗执行重试预算。随后到达的权威终态事件可结算原 Turn。在此之前保留未结算输入、Runtime 所有权、策略变更屏障、生命周期屏障和子任务结果观察。一次阻塞采样本身不能关闭 Session，也不能将子任务结果认定为 unknown。显式取消仍可用。
+
+WebShell 读取适配器将这个具名停放呈现为 `recovery_blocked`。浏览器通过现有 summary 轮询更新页面，不把事件流心跳当作业务进展。该 Session 禁止再次提交，保留取消能力。内部 Turn 仍为可取消、可恢复的活动记录；这项变更不伪造物理停止回执。
+
+## 范围与发布顺序
+
+本次修正有界的检查点预算并补上运行中阻塞观测，不建设通用大资源后端。超过 1 MiB 的检查点仍被拒绝，并应显示阻塞。先部署 Java 读取端，再让 Harness 发布超过 64 KiB 的检查点；旧 Store 仍会拒绝这些资源。现有资源和 journal 保持可读，SQL schema 不变。隔离验证通过前不重启用户预览。
+
+## 验证与验收
+
+使用微小工具输出复现原始拒绝，再让同一个 100 次调用的 fixture 完成，且不删除收据。验证冷加载、调用身份保持、超过 1 MiB 的拒绝，以及普通资源仍限制在 64 KiB。覆盖真实 Java 资源提交、读取和 reader 兼容性。
+
+验证只有心跳的阻塞 Harness 能结束 Java 消费等待、持久化停放，并通过 WebShell 适配器和浏览器 provider 显示阻塞。正常的无活动 prompt 审批等待、活动模型执行、取消竞争与 dispatch 租约丢失均不得被误判。验证有界的重复观测能在不重放执行的情况下消费稍后到达的终态；证据不可读时继续停放且不消耗重试预算。执行相关 TypeScript 测试、build/typecheck/bundle、Java 验证、适用的 MySQL/Hosted 恢复检查和独立端到端验证，并明确记录环境中无法执行的门禁。

packages/cli/src/serve/contracts/managed-csi-native-readback-v1.fixtures.json

@@ -4087,5 +4087,19 @@
       },
       "accepted": true
     }
+  ],
+  "checkpointSizeCases": [
+    {
+      "byteLength": 65537,
+      "valid": true
+    },
+    {
+      "byteLength": 1048576,
+      "valid": true
+    },
+    {
+      "byteLength": 1048577,
+      "valid": false
+    }
   ]
 }

packages/cli/src/serve/managed-csi-native-readback.test.ts

@@ -25,6 +25,39 @@ import {
 } from './managed-csi-native-readback.js';
 
 describe('private CSI native readback', () => {
+  const checkpointSizeCases: Array<{ byteLength: number; valid: boolean }> =
+    fixtures.checkpointSizeCases;
+  it.each(checkpointSizeCases)(
+    'bounds the grant checkpoint at $byteLength bytes',
+    ({ byteLength, valid }) => {
+      const sample = fixtures.valid.find(
+        (candidate: { name: string }) => candidate.name === 'execute',
+      );
+      const response = structuredClone(sample.response);
+      response.evidence.grant.checkpointRef.byteLength = byteLength;
+      const read = () =>
+        readCsiNativeResponse(response, readCsiNativeRequest(sample.request));
+      if (valid) expect(read()).toEqual(response);
+      else expect(read).toThrow();
+    },
+  );
+
+  it.each([65536, 65537])(
+    'keeps an ordinary prepare reference bounded at %s bytes',
+    (byteLength) => {
+      const sample = fixtures.valid.find(
+        (candidate: { name: string }) => candidate.name === 'prepare',
+      );
+      const request = structuredClone(sample.request);
+      request.subject.byteLength = byteLength;
+      if (byteLength === 65536) {
+        expect(readCsiNativeRequest(request)).toEqual(request);
+      } else {
+        expect(() => readCsiNativeRequest(request)).toThrow();
+      }
+    },
+  );
+
   for (const value of fixtures.valid) {
     it(`accepts the shared ${value.name} response and original opaque bytes`, () => {
       const request = readCsiNativeRequest(value.request);

packages/cli/src/serve/managed-csi-native-readback.ts

@@ -9,6 +9,7 @@ import { isDeepStrictEqual } from 'node:util';
 import {
   assertManagedSessionDurableRef,
   assertManagedSessionStableId,
+  MANAGED_SESSION_LIMITS,
   parseManagedSessionRecordJson,
   type ManagedSessionDurableRef,
   type ManagedSessionJsonValue,
@@ -431,7 +432,10 @@ function ref(value: unknown, kind: string): ManagedSessionDurableRef {
     result.kind === kind &&
       result.schemaVersion === 1 &&
       result.byteLength > 0 &&
-      result.byteLength <= 64 * 1024,
+      result.byteLength <=
+        (kind === 'managed-checkpoint'
+          ? MANAGED_SESSION_LIMITS.maxEventBytes
+          : 64 * 1024),
   );
   return result;
 }

packages/core/src/managed-runtime/contracts/managed-tool-publication-v1.fixtures.json

@@ -1128,7 +1128,7 @@
           "resourceId": "checkpoint-1",
           "kind": "managed-checkpoint",
           "schemaVersion": 1,
-          "byteLength": 65537,
+          "byteLength": 1048577,
           "digest": "44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a"
         }
       },
@@ -1906,5 +1906,32 @@
       "requestDigest": "sha256:06dade6f696f7d0182c24280475d93c19d2efe72bcc8e7580ddf604b9b6bf0d4",
       "argsDigest": "sha256:c126dbf7eb742625012b1feb10f1be9bb504d93a135f6f253d0779d3a1130a8e"
     }
+  ],
+  "referenceSizeCases": [
+    {
+      "field": "checkpointRef",
+      "byteLength": 65537,
+      "valid": true
+    },
+    {
+      "field": "checkpointRef",
+      "byteLength": 1048576,
+      "valid": true
+    },
+    {
+      "field": "checkpointRef",
+      "byteLength": 1048577,
+      "valid": false
+    },
+    {
+      "field": "argsRef",
+      "byteLength": 65536,
+      "valid": true
+    },
+    {
+      "field": "argsRef",
+      "byteLength": 65537,
+      "valid": false
+    }
   ]
 }

packages/core/src/managed-runtime/contracts/managed-tool-publication-v1.schema.json

@@ -170,7 +170,7 @@
             "byteLength": {
               "type": "integer",
               "minimum": 1,
-              "maximum": 65536
+              "maximum": 1048576
             },
             "digest": {
               "type": "string",

packages/core/src/managed-runtime/http-managed-session-store.test.ts

@@ -29,6 +29,7 @@ import type { HookExecution, HookRegistration } from './managed-hook-record.js';
 import {
   MANAGED_SESSION_EVENT_SUBTYPE,
   MANAGED_SESSION_HEADER_SUBTYPE,
+  MANAGED_SESSION_LIMITS,
   ManagedSessionAppendUnsettledError,
   managedSessionEventRefs,
   managedSessionEventsDigest,
@@ -1444,6 +1445,19 @@ describe('HTTP Managed Session store', () => {
       await expect(
         stores.resourceStore.publish('ordinary', bytes),
       ).rejects.toThrow('inline limit');
+      const checkpoint = await stores.resourceStore.publish(
+        'managed-checkpoint',
+        Buffer.alloc(MANAGED_SESSION_LIMITS.maxEventBytes, 0x61),
+      );
+      expect((await stores.resourceStore.read(checkpoint)).byteLength).toBe(
+        MANAGED_SESSION_LIMITS.maxEventBytes,
+      );
+      await expect(
+        stores.resourceStore.publish(
+          'managed-checkpoint',
+          Buffer.alloc(MANAGED_SESSION_LIMITS.maxEventBytes + 1),
+        ),
+      ).rejects.toThrow('kind=managed-checkpoint, bytes=1048577');
       const attachment = await stores.resourceStore.publish(
         'managed-input',
         Buffer.alloc(7 * 1024 * 1024, 0x61),

packages/core/src/managed-runtime/http-managed-session-store.ts

@@ -72,6 +72,8 @@ export const HTTP_MANAGED_SESSION_STORE_CONTRACT = {
 const MAX_ATTACHMENT_INPUT_BYTES = 8 * 1024 * 1024;
 
 export function managedSessionInlineResourceBytes(kind: string): number {
+  if (kind === 'managed-checkpoint')
+    return MANAGED_SESSION_LIMITS.maxEventBytes;
   if (kind === 'managed-input') return MAX_ATTACHMENT_INPUT_BYTES;
   if (kind === 'managed-agent-skills') return 2 * 1024 * 1024;
   if (kind === 'hosted-model-request') return HOSTED_MODEL_REQUEST_BYTES;
@@ -569,7 +571,7 @@ class HttpManagedSessionResourceStore implements ManagedSessionResourceStore {
     const maxBytes = managedSessionInlineResourceBytes(safeKind);
     if (source.byteLength > maxBytes) {
       throw new ManagedSessionRecordError(
-        `resource bytes exceed the ${maxBytes}-byte inline limit; OSS storage is not enabled.`,
+        `resource bytes exceed the ${maxBytes}-byte inline limit (kind=${safeKind}, bytes=${source.byteLength}); OSS storage is not enabled.`,
       );
     }
     const bytes = Buffer.from(source);

packages/core/src/managed-runtime/managed-tool-publication.test.ts

@@ -46,6 +46,11 @@ const fixtures = JSON.parse(
   digestVectors: Array<{ id: string; binding: unknown; digest: string }>;
   tokenVector: { token: string; hash: string };
   bindingDigest: string;
+  referenceSizeCases: Array<{
+    field: 'argsRef' | 'checkpointRef';
+    byteLength: number;
+    valid: boolean;
+  }>;
   cases: Array<{
     id: string;
     kind: 'binding' | 'request' | 'grant';
@@ -82,6 +87,21 @@ describe('managed-tool-publication/1', () => {
     });
   }
 
+  it.each(fixtures.referenceSizeCases)(
+    'bounds $field at $byteLength bytes',
+    ({ field, byteLength, valid }) => {
+      const binding = parseToolPublicationBinding(fixtures.cases[0].value);
+      const value = {
+        ...binding,
+        [field]: { ...binding[field], byteLength },
+      };
+      expect(ajv.getSchema('publication#/$defs/binding')!(value)).toBe(valid);
+      const parse = () => parseToolPublicationBinding(value);
+      if (valid) expect(parse).not.toThrow();
+      else expect(parse).toThrow();
+    },
+  );
+
   it('matches independent Unicode, key-order and integer digest vectors', () => {
     for (const vector of fixtures.digestVectors) {
       expect(

packages/core/src/managed-runtime/managed-tool-publication.ts

@@ -13,6 +13,7 @@ import {
   assertManagedSessionSequence,
   assertManagedSessionStableId,
   assertManagedSessionTime,
+  MANAGED_SESSION_LIMITS,
   managedSessionKeysEqual,
   parseManagedSessionRecordJson,
   type ManagedSessionDurableRef,
@@ -173,7 +174,10 @@ function ref(
     result.kind === kind &&
       result.schemaVersion === 1 &&
       result.byteLength > 0 &&
-      result.byteLength <= TOOL_PUBLICATION_LIMITS.maxBodyBytes,
+      result.byteLength <=
+        (kind === 'managed-checkpoint'
+          ? MANAGED_SESSION_LIMITS.maxEventBytes
+          : TOOL_PUBLICATION_LIMITS.maxBodyBytes),
   );
   return Object.freeze(result);
 }

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

@@ -176,6 +176,12 @@ public interface HarnessConnector extends AutoCloseable {
     SourceStream stream(String tenantId, String sessionId, long lastEventId,
             String eventEpoch);
 
+    /** Reads only an existing original attachment; it must never load or take over execution. */
+    default SourceStream observeStream(String tenantId, String sessionId, String bootId,
+            long lastEventId, String eventEpoch) {
+        throw new UnsupportedOperationException("Read-only Harness observation is unavailable");
+    }
+
     default void resolveAction(
             String tenantId,
             String sessionId,
@@ -319,6 +325,10 @@ public interface HarnessConnector extends AutoCloseable {
 
         SourceEvent next();
 
+        default boolean recoveryBlocked(long afterEventId) {
+            return false;
+        }
+
         @Override
         void close();
     }

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

@@ -5,6 +5,7 @@ import com.alibaba.qwen.code.daemon.CreateHarnessSession;
 import com.alibaba.qwen.code.daemon.DaemonApprovalMode;
 import com.alibaba.qwen.code.daemon.DaemonEvent;
 import com.alibaba.qwen.code.daemon.DaemonHttpException;
+import com.alibaba.qwen.code.daemon.GetHarnessTranscript;
 import com.alibaba.qwen.code.daemon.HarnessEventStream;
 import com.alibaba.qwen.code.daemon.HarnessSessionRef;
 import com.alibaba.qwen.code.daemon.HostedHarnessClient;
@@ -701,7 +702,27 @@ public class QwenHostedHarnessConnector implements HarnessConnector {
     private SourceStream doStream(String tenantId, String sessionId,
             long lastEventId, String eventEpoch) {
         HarnessSessionRef ref = attachment(tenantId, sessionId, false);
-        HarnessEventStream stream = client().streamEvents(
+        return openStream(client(), ref, lastEventId, eventEpoch);
+    }
+
+    @Override
+    public SourceStream observeStream(String tenantId, String sessionId, String bootId,
+            long lastEventId, String eventEpoch) {
+        requireShellApproval(sessions.requireSession(tenantId, sessionId));
+        HarnessSessionRef ref = attachments.get(new AttachmentKey(tenantId, sessionId));
+        if (bootId == null || ref == null || !bootId.equals(ref.getHarnessBootId())) {
+            throw new IllegalStateException("Original Harness attachment is unavailable");
+        }
+        HostedHarnessClient source = client();
+        if (!bootId.equals(source.capabilities().getBootId())) {
+            throw new IllegalStateException("Original Harness generation is unavailable");
+        }
+        return openStream(source, ref, lastEventId, eventEpoch);
+    }
+
+    private SourceStream openStream(HostedHarnessClient source, HarnessSessionRef ref,
+            long lastEventId, String eventEpoch) {
+        HarnessEventStream stream = source.streamEvents(
                 StreamHarnessEvents.builder()
                         .session(ref)
                         .lastEventId(lastEventId)
@@ -721,6 +742,20 @@ public class QwenHostedHarnessConnector implements HarnessConnector {
                         event.getPromptId(), event.getMetadata());
             }
 
+            @Override
+            public boolean recoveryBlocked(long afterEventId) {
+                var status = source.getStatus(ref);
+                Object blocked = status.getRaw().get("recoveryBlocked");
+                if (blocked != null && !(blocked instanceof Boolean)) {
+                    throw new IllegalStateException(
+                            "Hosted Harness recoveryBlocked is not a boolean");
+                }
+                return !status.hasActivePrompt() && Boolean.TRUE.equals(blocked)
+                        && source.getTranscript(GetHarnessTranscript.builder()
+                                .session(ref).cursor(Long.toString(afterEventId)).limit(1).build())
+                                .getEvents().isEmpty();
+            }
+
             @Override
             public void close() {
                 stream.close();

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/persistence/ManagedAgentStore.java

@@ -4764,8 +4764,9 @@ public class ManagedAgentStore implements AgentStateStore {
                 || !live && hasRetainedRuntimeSession(tenantId, sessionId)
                 || !jdbc.queryForList("SELECT 1 FROM ma_turn WHERE tenant_id = ? AND session_id = ?"
                         + " AND (status IN ('RECOVERY_BLOCKED', 'CANCELLING') OR status IN ('ACCEPTED', 'RUNNING')"
-                        + " AND error_code IN (?, ?)) LIMIT 1", tenantId, sessionId,
-                        AgentStateStore.RUNTIME_RECOVERY_BLOCKED, AgentStateStore.WORKFLOW_RECOVERY_BLOCKED).isEmpty()
+                        + " AND error_code IN (?, ?, ?)) LIMIT 1", tenantId, sessionId,
+                        AgentStateStore.RUNTIME_RECOVERY_BLOCKED, AgentStateStore.WORKFLOW_RECOVERY_BLOCKED,
+                        AgentStateStore.HARNESS_RECOVERY_BLOCKED).isEmpty()
                 || !jdbc.queryForList("SELECT 1 FROM ma_session_head WHERE tenant_id = ?"
                         + " AND session_id = ? AND head_kind = 'MANAGED_SESSION' AND"
                         + " (state <> 'ACTIVE' OR recovery_status <> 'READY')", tenantId, sessionId).isEmpty()) {
@@ -5174,12 +5175,9 @@ public class ManagedAgentStore implements AgentStateStore {
     }
 
     public List<DispatchTarget> findDispatchable(long now, int limit) {
-        // #14387 / D7-2: a Turn marked recovery-blocked (its workflow's
-        // generation was lost, or its parked Runtime execution's acceptance
-        // is still unknown) is parked, never re-dispatched — without the
-        // carve-out the sweep would claim → refuse → release it once per
-        // tick forever. A cancel still scans it: settling the Turn
-        // cancelled is the cancel's own act, not a re-dispatch.
+        // Workflow/Runtime parks have separate settlement paths. A due
+        // Harness park enters only the coordinator's read-only observation
+        // branch; cancellation still uses the normal dispatch lane.
         return jdbc.query(
                 "SELECT tenant_id, session_id, turn_id FROM ma_turn WHERE status IN ('ACCEPTED', 'RUNNING', 'CANCELLING') AND (dispatch_lease_until IS NULL OR dispatch_lease_until < ?) AND (retry_after IS NULL OR retry_after <= ?) AND (status = 'CANCELLING' OR error_code IS NULL OR error_code NOT IN ('" + AgentStateStore.WORKFLOW_RECOVERY_BLOCKED + "', '" + AgentStateStore.RUNTIME_RECOVERY_BLOCKED + "')) AND EXISTS (SELECT 1 FROM ma_session private_scope WHERE private_scope.tenant_id = ma_turn.tenant_id AND private_scope.session_id = ma_turn.session_id AND private_scope.csi_guard = FALSE) ORDER BY gmt_modified ASC LIMIT ?",
                 (result, row) -> new DispatchTarget(
@@ -5250,19 +5248,20 @@ public class ManagedAgentStore implements AgentStateStore {
             String errorCode, String errorMessage) {
         requireLegacyMutation(tenantId, sessionId);
         long now = clock.millis();
-        // The verdict parks the Turn out of the dispatch sweep, so it is
-        // fenced by the live dispatch lease like every terminal decision:
-        // an owner whose lease already expired holds nothing to park with.
+        Long retryAfter = AgentStateStore.HARNESS_RECOVERY_BLOCKED.equals(errorCode)
+                ? now + 15_000L : null;
+        // A sampled Harness block can lift. Retain the unsettled outcome
+        // and back off its read-only recheck without consuming retry budget.
         return jdbc.update("UPDATE ma_turn SET error_code = ?,"
-                        + " error_message = ?, dispatch_owner = NULL,"
+                        + " error_message = ?, retry_after = ?, dispatch_owner = NULL,"
                         + " dispatch_lease_until = NULL, gmt_modified = ?,"
                         + " version = version + 1 WHERE tenant_id = ? AND"
                         + " session_id = ? AND turn_id = ? AND"
                         + " dispatch_owner = ? AND dispatch_lease_until >= ?"
                         + " AND status IN ('ACCEPTED', 'RUNNING',"
-                        + " 'CANCELLING')",
-                errorCode, errorMessage, now, tenantId, sessionId, turnId,
-                owner, now) == 1;
+                        + " 'CANCELLING') AND (? <> ? OR status <> 'CANCELLING')",
+                errorCode, errorMessage, retryAfter, now, tenantId, sessionId, turnId,
+                owner, now, errorCode, AgentStateStore.HARNESS_RECOVERY_BLOCKED) == 1;
     }
 
     @Override
@@ -6860,11 +6859,9 @@ public class ManagedAgentStore implements AgentStateStore {
     }
 
     /**
-     * #14387: the lifecycle gate's active-Turn count. A Turn marked
-     * recovery-blocked is parked out of the dispatch sweep and never
-     * drives again — it is evidence of an unwitnessed execution, not live
-     * work, so it does not hold a lifecycle operation open. A CANCELLING
-     * Turn still blocks: its cancel is in flight.
+     * Only a permanently refused workflow leaves this lifecycle gate.
+     * Runtime and Harness parks still await authoritative settlement;
+     * closing them could discard a later result. Cancellation also waits.
      */
     private boolean hasLiveTurn(String tenantId, String sessionId) {
         Integer count = jdbc.queryForObject("SELECT COUNT(*) FROM"

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/persistence/ManagedSessionStoreModels.java

@@ -71,6 +71,7 @@ public final class ManagedSessionStoreModels {
 
     public static int inlineResourceBytes(String kind) {
         return switch (kind) {
+            case "managed-checkpoint" -> MAX_EVENT_BYTES;
             case "managed-input" -> MAX_ATTACHMENT_INPUT_BYTES;
             case "managed-agent-skills" -> MAX_AGENT_SKILLS_BYTES;
             case "hosted-model-request" -> MAX_MODEL_REQUEST_BYTES;

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/persistence/ToolPublicationContract.java

@@ -359,7 +359,7 @@ public final class ToolPublicationContract {
         id(r.get("resourceId"));
         require(kind.equals(text(r, "kind")) && count(r.get("schemaVersion"), 1) == 1,
                 "Invalid publication resource kind");
-        count(r.get("byteLength"), MAX_BODY_BYTES);
+        count(r.get("byteLength"), ManagedSessionStoreModels.inlineResourceBytes(kind));
         digest(r.get("digest"));
     }
 

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/persistence/ToolPublicationStore.java

@@ -675,7 +675,7 @@ public final class ToolPublicationStore {
             var current = sessions.readResource(text(key, "tenantId"), text(key, "workspaceId"),
                     text(key, "sessionId"), writer.checkpointId(), writerToken);
             require("managed-checkpoint".equals(current.kind()) && current.schemaVersion() == 1
-                    && current.byteLength() <= ToolPublicationContract.MAX_BODY_BYTES, "Current checkpoint is invalid");
+                    && current.byteLength() <= ManagedSessionStoreModels.MAX_EVENT_BYTES, "Current checkpoint is invalid");
             requireCheckpoint(b, ToolPublicationContract.readJson(current.bytes()), writer);
         }
         JsonNode args = readResource(b, b.get("argsRef"), writerToken);

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/session/AgentStateStore.java

@@ -64,6 +64,8 @@ public interface AgentStateStore {
      */
     String RUNTIME_RECOVERY_BLOCKED = "hosted_runtime_recovery_blocked";
 
+    String HARNESS_RECOVERY_BLOCKED = "hosted_harness_recovery_blocked";
+
     // The annotation must sit on the default itself: the delegating body runs
     // on the target instance, so without it the self-call bypasses the proxy.
     @org.springframework.transaction.annotation.Transactional
@@ -798,7 +800,9 @@ public interface AgentStateStore {
      * the error code stays on the still-active row for the child result
      * relay to classify from, and the dispatch lease is released so the
      * drive ends without settling the Turn. Never a terminal status: the
-     * unwitnessed execution's outcome was never observed.
+     * unwitnessed execution's outcome was never observed. A Harness park
+     * backs off before read-only journal observation; it may still settle
+     * from later authoritative evidence without replaying execution.
      */
     boolean markTurnRecoveryBlocked(String tenantId, String sessionId,
             String turnId, String owner, String errorCode,

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/session/HarnessCoordinator.java

@@ -2,6 +2,7 @@ package com.alibaba.qwen.code.managedagent.session;
 
 import com.alibaba.qwen.code.daemon.DaemonHttpException;
 import com.alibaba.qwen.code.daemon.DaemonProtocolException;
+import com.alibaba.qwen.code.daemon.DaemonTransportException;
 import com.alibaba.qwen.code.daemon.HarnessRuntimeRecovery;
 import com.alibaba.qwen.code.daemon.HarnessSessionRefusedException;
 import com.alibaba.qwen.code.daemon.HostedHarnessCapabilityMismatchException;
@@ -440,6 +441,10 @@ public class HarnessCoordinator {
             AtomicBoolean recoveryPath) {
         SessionRecord session = store.requireSession(claimed.tenantId(),
                 claimed.sessionId());
+        if (!"CANCELLING".equals(claimed.status())
+                && AgentStateStore.HARNESS_RECOVERY_BLOCKED.equals(claimed.errorCode())) {
+            return observeBlockedHarness(claimed, session, leaseLost);
+        }
         // A bound Session re-attaches through the takeover load: failures
         // here wait on the prior generation's lease, a wait bounded by that
         // lease itself rather than by the pre-admission retry budget.
@@ -848,6 +853,32 @@ public class HarnessCoordinator {
         }
     }
 
+    private boolean observeBlockedHarness(TurnRecord turn, SessionRecord session,
+            AtomicBoolean leaseLost) {
+        long lastEventId = turn.harnessLastEventId() == null ? 0 : turn.harnessLastEventId();
+        try (SourceStream stream = harness.observeStream(turn.tenantId(), turn.sessionId(),
+                session.harnessBootId(), lastEventId, turn.harnessEventEpoch())) {
+            requireLease(leaseLost);
+            if (!Objects.equals(turn.harnessEventEpoch(), stream.eventEpoch())) {
+                throw new IllegalStateException("Hosted Harness event epoch changed");
+            }
+            if (!stream.recoveryBlocked(lastEventId)) {
+                return consumeStream(turn, session.harnessBootId(), stream, leaseLost);
+            }
+        } catch (RuntimeException error) {
+            LOG.warn("Managed Harness recovery observation deferred tenant={} session={} turn={}",
+                    turn.tenantId(), turn.sessionId(), turn.turnId(), error);
+        }
+        // No execution takeover, submit or continuation belongs to this
+        // lane. Unavailable evidence retains the original unsettled park.
+        if (leaseLost.get()) {
+            return false;
+        }
+        return store.markTurnRecoveryBlocked(turn.tenantId(), turn.sessionId(),
+                turn.turnId(), owner, AgentStateStore.HARNESS_RECOVERY_BLOCKED,
+                "The Hosted Harness outcome remains unconfirmed; awaiting its journal.");
+    }
+
     private boolean consumeStream(TurnRecord turn, String bootId,
             SourceStream stream, AtomicBoolean leaseLost) {
         int capacity = Math.max(128, batchMaxEvents * 2);
@@ -858,10 +889,43 @@ public class HarnessCoordinator {
         List<HarnessEvent> batch = new ArrayList<>();
         int batchBytes = 0;
         long flushAt = 0;
+        long statusAt = System.nanoTime() + TimeUnit.SECONDS.toNanos(15);
+        long lastEventId = turn.harnessLastEventId() == null ? 0 : turn.harnessLastEventId();
         boolean flushFirstVisibleText = true;
         try {
             while (true) {
-                StreamItem item = take(incoming, batch, flushAt);
+                if (System.nanoTime() >= statusAt) {
+                    flush(turn, stream.eventEpoch(), batch, leaseLost);
+                    batchBytes = 0;
+                    flushAt = 0;
+                    // An unavailable advisory read must not replace the stream's
+                    // authoritative terminal event or its bounded idle failure.
+                    boolean recoveryBlocked = false;
+                    try {
+                        recoveryBlocked = incoming.isEmpty() && stream.recoveryBlocked(lastEventId);
+                    } catch (DaemonTransportException error) {
+                        LOG.warn("Managed Harness recovery probe is unavailable"
+                                        + " tenant={} session={} turn={} failure={}",
+                                turn.tenantId(), turn.sessionId(), turn.turnId(),
+                                error.getClass().getSimpleName());
+                    }
+                    if (recoveryBlocked && incoming.isEmpty()) {
+                        requireLease(leaseLost);
+                        boolean parked = store.markTurnRecoveryBlocked(turn.tenantId(),
+                                turn.sessionId(), turn.turnId(), owner,
+                                AgentStateStore.HARNESS_RECOVERY_BLOCKED,
+                                "The Hosted Harness stopped with recovery blocked;"
+                                        + " reconcile its saved execution before continuing.");
+                        if (!parked) {
+                            LOG.warn("Managed Harness recovery-blocked mark lost to a lease CAS"
+                                            + " tenant={} session={} turn={}",
+                                    turn.tenantId(), turn.sessionId(), turn.turnId());
+                        }
+                        return parked;
+                    }
+                    statusAt = System.nanoTime() + TimeUnit.SECONDS.toNanos(15);
+                }
+                StreamItem item = take(incoming, batch, flushAt, statusAt);
                 if (item == null) {
                     flush(turn, stream.eventEpoch(), batch, leaseLost);
                     batchBytes = 0;
@@ -879,6 +943,9 @@ public class HarnessCoordinator {
                 }
                 SourceEvent source = item.event();
                 requireLease(leaseLost);
+                if (source.id() != null) {
+                    lastEventId = Math.max(lastEventId, source.id());
+                }
                 if (source.id() == null
                         || source.promptId() != null
                         && !turn.promptId().equals(source.promptId())) {
@@ -973,12 +1040,10 @@ public class HarnessCoordinator {
     }
 
     private StreamItem take(BlockingQueue<StreamItem> incoming,
-            List<HarnessEvent> batch, long flushAt) {
+            List<HarnessEvent> batch, long flushAt, long statusAt) {
         try {
-            if (batch.isEmpty()) {
-                return incoming.take();
-            }
-            long remaining = Math.max(1, flushAt - System.nanoTime());
+            long deadline = batch.isEmpty() ? statusAt : Math.min(flushAt, statusAt);
+            long remaining = Math.max(1, deadline - System.nanoTime());
             return incoming.poll(remaining, TimeUnit.NANOSECONDS);
         } catch (InterruptedException error) {
             Thread.currentThread().interrupt();

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/session/ManagedAgentService.java

@@ -1432,7 +1432,9 @@ public class ManagedAgentService {
 
     private static WebShellTurn webShellTurn(TurnSummary turn) {
         return new WebShellTurn(turn.turnId(), turn.sessionId(),
-                turn.status().toLowerCase(Locale.ROOT), turn.createdAt(),
+                AgentStateStore.HARNESS_RECOVERY_BLOCKED.equals(turn.errorCode())
+                        && ("RUNNING".equals(turn.status()) || "ACCEPTED".equals(turn.status()))
+                        ? "recovery_blocked" : turn.status().toLowerCase(Locale.ROOT), turn.createdAt(),
                 turn.completedAt(), turn.errorCode());
     }
 

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/runtimebroker/transport/CsiNativeReadbackProtocol.java

@@ -293,7 +293,8 @@ public final class CsiNativeReadbackProtocol {
         closed(reference, Set.of("resourceId", "kind", "schemaVersion", "byteLength", "digest"));
         id(reference.get("resourceId"));
         require(kind.equals(reference.get("kind")) && counter(reference.get("schemaVersion")) == 1
-                && counter(reference.get("byteLength")) > 0 && counter(reference.get("byteLength")) <= 64 * 1024);
+                && counter(reference.get("byteLength")) > 0
+                && counter(reference.get("byteLength")) <= ("managed-checkpoint".equals(kind) ? 1024 * 1024 : 64 * 1024));
         digest(reference.get("digest"), false);
         return reference;
     }

packages/managed-agent-server/src/main/resources/openapi/managed-agent-public-api.openapi.json

@@ -2,8 +2,8 @@
   "openapi": "3.1.0",
   "info": {
     "title": "Qwen Managed Agent Public and WebShell API",
-    "version": "1.83.0",
-    "description": "Canonical contract for the public Managed Agent resources and the WebShell adapter. partial means that the route exists but generated types and full contract tests are pending; planned means that the route or field is not implemented. Workspace context and cwd operations are v1.9 target contracts, not current server capabilities. v1.10 adds planned command admission, Actions, lifecycle and actor authorization. v1.12 refines planned W0 creation discovery, actor-specific create hints and explicit default selection. Filter planned fields as well as routes from production SDKs. v1.13 moves this contract into the Qwen Code repository as its single source and records the shipped rename, archive, unarchive and delete routes as partial. v1.14 integrates W0d workspace discovery, the WebShell workspace lookup, and empty-session binding as partial; workspace_binding/workspaceBinding advertises this narrow flow without enabling workspace execution or context switching. v1.15 adds the archived, archiving and deleting Session statuses and the cancelling Turn status, requires request_id in error envelopes, declares the error responses that the server returns, and marks Session create, list and get on both surfaces implemented. v1.16 adds the planned Stage H task contract (SessionTaskView as PublicTask, task list, detail, events and cancel) and names the MCP catalog, hook catalog, automation and channel resources as planned; their shapes arrive with Stages H1 to H6. v1.17 implements event replay: events carry their schema and projection versions and a top-level Item and Part identity, JSON event pages return has_more and next_cursor and accept limits up to 1000, and a persisted replay floor answers expired cursors with 409 cursor_expired in JSON and one agent.session.resync_required frame over SSE; it marks the public event query and stream, the WebShell event stream and the WebShell transcript implemented. v1.18 implements the durable Session lifecycle (Stage D4): close, archive and delete answer 202 with a command operation on both surfaces, an archive requires a closed Session, a deleted Session leaves a tombstone whose operations stay readable, and the operation query serves them; unarchive restores a closed Session, and Session capabilities advertise session_lifecycle. v1.19 serves the task list and detail on both surfaces as partial (Stage H0c): the Session store projects them from the Stage H records the Session authority commits, a task.updated Session event announces each change of a task's view, and Session capabilities advertise tasks. Task events and cancel stay planned until a slice's tasks produce output and accept a cancel. v1.20 implements the Turn read model (Stage D5): the public Turn list and detail read a Session's Turns without their input, newest first by creation time in milliseconds and then Turn ID, with an opaque cursor; a deleted Session's Turns are not readable, and the durable admission fields of a Turn stay planned until D7. v1.21 declares on the task list, detail and event routes of both surfaces the 403 actor_scope_mismatch that the tenant filter answers on every /v1/agents/ and WebShell route, as the Session and Turn reads already do, and the shared Forbidden response names it; a caller that cannot read a task still gets 404. v1.22 adds deployment-opted-in creation with an initial Workspace file-tool Turn (G0) on the public and WebShell surfaces; the fixed server-owned profile does not enable later Workspace operations. v1.23 settles the planned task event and cancel semantics: a durable retention floor, committed-prefix publication, stable cursors, Artifact visibility before expiry, bounded backlog under archival failure, ordered idempotent cancellation and command outcomes. Clients tolerate unknown optional event fields across minor versions. Task events and cancel remain planned. v1.24 serves the read-only Session MCP catalog as partial (Stage H1), projecting display metadata and schemas from committed records without Runtime identities, credentials or production profile enablement. v1.25 implements permission Actions (Stage D6b) on both surfaces: list, get and respond, projected from the Session's action.changed records; a WebShell permission Action carries inputRevision, policyRevision, functionCallId, toolName and expiresAt, its option ids allow and deny are stable, a Turn has at most one requested approval at a time, and a response is a durable command operation carrying requestId. v1.26 declares the tenant filter's 403 actor_scope_mismatch on every covered public and WebShell route, including planned routes, and pins the refusal code in contract traffic. v1.27 adds O3 durable tool-result projection, artifact metadata, immutable range/streaming content and WebShell result reads. Artifact capability requires a configured O3 reader and a Workspace-bound Session; execution admission remains independent. v1.28 admits later Turns of a Workspace-bound Session for the SLine truncated
+    "version": "1.84.0",
+    "description": "Canonical contract for the public Managed Agent resources and the WebShell adapter. partial means that the route exists but generated types and full contract tests are pending; planned means that the route or field is not implemented. Workspace context and cwd operations are v1.9 target contracts, not current server capabilities. v1.10 adds planned command admission, Actions, lifecycle and actor authorization. v1.12 refines planned W0 creation discovery, actor-specific create hints and explicit default selection. Filter planned fields as well as routes from production SDKs. v1.13 moves this contract into the Qwen Code repository as its single source and records the shipped rename, archive, unarchive and delete routes as partial. v1.14 integrates W0d workspace discovery, the WebShell workspace lookup, and empty-session binding as partial; workspace_binding/workspaceBinding advertises this narrow flow without enabling workspace execution or context switching. v1.15 adds the archived, archiving and deleting Session statuses and the cancelling Turn status, requires request_id in error envelopes, declares the error responses that the server returns, and marks Session create, list and get on both surfaces implemented. v1.16 adds the planned Stage H task contract (SessionTaskView as PublicTask, task list, detail, events and cancel) and names the MCP catalog, hook catalog, automation and channel resources as planned; their shapes arrive with Stages H1 to H6. v1.17 implements event replay: events carry their schema and projection versions and a top-level Item and Part identity, JSON event pages return has_more and next_cursor and accept limits up to 1000, and a persisted replay floor answers expired cursors with 409 cursor_expired in JSON and one agent.session.resync_required frame over SSE; it marks the public event query and stream, the WebShell event stream and the WebShell transcript implemented. v1.18 implements the durable Session lifecycle (Stage D4): close, archive and delete answer 202 with a command operation on both surfaces, an archive requires a closed Session, a deleted Session leaves a tombstone whose operations stay readable, and the operation query serves them; unarchive restores a closed Session, and Session capabilities advertise session_lifecycle. v1.19 serves the task list and detail on both surfaces as partial (Stage H0c): the Session store projects them from the Stage H records the Session authority commits, a task.updated Session event announces each change of a task's view, and Session capabilities advertise tasks. Task events and cancel stay planned until a slice's tasks produce output and accept a cancel. v1.20 implements the Turn read model (Stage D5): the public Turn list and detail read a Session's Turns without their input, newest first by creation time in milliseconds and then Turn ID, with an opaque cursor; a deleted Session's Turns are not readable, and the durable admission fields of a Turn stay planned until D7. v1.21 declares on the task list, detail and event routes of both surfaces the 403 actor_scope_mismatch that the tenant filter answers on every /v1/agents/ and WebShell route, as the Session and Turn reads already do, and the shared Forbidden response names it; a caller that cannot read a task still gets 404. v1.22 adds deployment-opted-in creation with an initial Workspace file-tool Turn (G0) on the public and WebShell surfaces; the fixed server-owned profile does not enable later Workspace operations. v1.23 settles the planned task event and cancel semantics: a durable retention floor, committed-prefix publication, stable cursors, Artifact visibility before expiry, bounded backlog under archival failure, ordered idempotent cancellation and command outcomes. Clients tolerate unknown optional event fields across minor versions. Task events and cancel remain planned. v1.24 serves the read-only Session MCP catalog as partial (Stage H1), projecting display metadata and schemas from committed records without Runtime identities, credentials or production profile enablement. v1.25 implements permission Actions (Stage D6b) on both surfaces: list, get and respond, projected from the Session's action.changed records; a WebShell permission Action carries inputRevision, policyRevision, functionCallId, toolName and expiresAt, its option ids allow and deny are stable, a Turn has at most one requested approval at a time, and a response is a durable command operation carrying requestId. v1.26 declares the tenant filter's 403 actor_scope_mismatch on every covered public and WebShell route, including planned routes, and pins the refusal code in contract traffic. v1.27 adds O3 durable tool-result projection, artifact metadata, immutable range/streaming content and WebShell result reads. Artifact capability requires a configured O3 reader and a Workspace-bound Session; execution admission remains independent. v1.28 admits later Turns of a Workspace-bound Session for the SLine truncated
   },
   "servers": [
     {
@@ -6596,7 +6596,8 @@
             "format": "uuid"
           },
           "status": {
-            "type": "string"
+            "type": "string",
+            "description": "recovery_blocked with hosted_harness_recovery_blocked means the Harness stopped without a safely settled Turn. The input remains active internally. Execution redispatch is parked, while bounded read-only observation of the original Harness journal may settle a later terminal result. Policy changes and lifecycle close wait for settlement; explicit cancellation remains available."
           },
           "submittedAt": {
             "type": "integer",

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ManagedAgentServerIntegrationTest.java

@@ -1430,8 +1430,9 @@ class ManagedAgentServerIntegrationTest {
     // #14387: the recovery-blocked mark is durable on the still-active
     // row — the status the claim saw stands, no retry counter moves, the
     // lease releases — and a stale owner never re-marks.
-    @Test
-    void marksATurnRecoveryBlockedWithoutSettlingIt() {
+    @org.junit.jupiter.params.ParameterizedTest
+    @org.junit.jupiter.params.provider.ValueSource(strings = {"hosted_workflow_recovery_blocked", "hosted_harness_recovery_blocked"})
+    void marksATurnRecoveryBlockedWithoutSettlingIt(String recoveryCode) throws Exception {
         pauseRecoveryScanning();
         String tenant = "tenant-turn-blocked-" + UUID.randomUUID();
         Admission session = store.insertSessionCommand(tenant,
@@ -1448,14 +1449,14 @@ class ManagedAgentServerIntegrationTest {
 
         assertThat(store.markTurnRecoveryBlocked(tenant,
                 session.sessionId(), turn.turnId(), owner,
-                "hosted_workflow_recovery_blocked",
+                recoveryCode,
                 "the Harness generation was lost")).isTrue();
 
         assertThat(store.findTurn(tenant, session.sessionId(),
                 turn.turnId())).get().satisfies(record -> {
                     assertThat(record.status()).isEqualTo(claimed.status());
                     assertThat(record.errorCode()).isEqualTo(
-                            "hosted_workflow_recovery_blocked");
+                            recoveryCode);
                     assertThat(record.errorMessage()).isEqualTo(
                             "the Harness generation was lost");
                     assertThat(record.dispatchOwner()).isNull();
@@ -1465,17 +1466,34 @@ class ManagedAgentServerIntegrationTest {
                 });
         assertThat(store.markTurnRecoveryBlocked(tenant,
                 session.sessionId(), turn.turnId(), owner,
-                "hosted_workflow_recovery_blocked",
+                recoveryCode,
                 "the Harness generation was lost")).isFalse();
 
-        // The parked Turn leaves the dispatch sweep entirely — no
-        // claim-refuse-release spin — until a cancel: settling the Turn
-        // cancelled is the cancel's own act, so a CANCELLING row is
-        // scanned despite the mark.
+        if ("hosted_harness_recovery_blocked".equals(recoveryCode)) {
+            mvc.perform(post("/api/agent/web-shell/v1/sessions/get")
+                    .header(TenantContextFilter.HEADER, tenant)
+                    .contentType(MediaType.APPLICATION_JSON)
+                    .content("{\"sessionId\":\"" + session.sessionId() + "\"}"))
+                    .andExpect(status().isOk())
+                    .andExpect(jsonPath("$.activeTurn.status").value("recovery_blocked"))
+                    .andExpect(jsonPath("$.activeTurn.errorCode").value(recoveryCode));
+        }
+
         DispatchTarget target = new DispatchTarget(tenant,
                 session.sessionId(), turn.turnId());
-        assertThat(store.findDispatchable(Long.MAX_VALUE, 100))
-                .doesNotContain(target);
+        if ("hosted_harness_recovery_blocked".equals(recoveryCode)) {
+            long due = store.findTurn(tenant, session.sessionId(), turn.turnId()).orElseThrow().retryAfter();
+            assertThat(store.findDispatchable(due - 1, 100)).doesNotContain(target);
+            assertThat(store.findDispatchable(due, 100)).contains(target);
+            assertThat(store.claimTurn(tenant, session.sessionId(), turn.turnId(),
+                    "too-soon", Duration.ofMinutes(1))).isEmpty();
+            assertThatThrownBy(() -> store.beginOperation(tenant, session.sessionId(),
+                    OperationKind.CLOSE, "", "blocked-close", "close-digest"))
+                    .isInstanceOfSatisfying(ManagedAgentException.class,
+                            error -> assertThat(error.getCode()).isEqualTo("turn_active"));
+        } else {
+            assertThat(store.findDispatchable(Long.MAX_VALUE, 100)).doesNotContain(target);
+        }
         store.insertCancelCommand(tenant, "CANCEL_TURN", "blocked-cancel",
                 "sha256:" + "4".repeat(64), session.sessionId(),
                 turn.turnId());
@@ -1484,6 +1502,15 @@ class ManagedAgentServerIntegrationTest {
         assertThat(store.findTurn(tenant, session.sessionId(),
                 turn.turnId()).get().status()).isEqualTo("CANCELLING");
 
+        if ("hosted_harness_recovery_blocked".equals(recoveryCode)) {
+            assertThat(store.claimTurn(tenant, session.sessionId(), turn.turnId(),
+                    "cancel-owner", Duration.ofMinutes(1))).isPresent();
+            assertThat(store.markTurnRecoveryBlocked(tenant, session.sessionId(),
+                    turn.turnId(), "cancel-owner", recoveryCode, "late status probe")).isFalse();
+            assertThat(store.findTurn(tenant, session.sessionId(), turn.turnId()).orElseThrow()
+                    .dispatchOwner()).isEqualTo("cancel-owner");
+        }
+
         // The mark is lease-fenced: an owner whose lease already expired
         // parks nothing — the CAS misses and the row stays unmarked.
         Admission session2 = store.insertSessionCommand(tenant,
@@ -1498,12 +1525,46 @@ class ManagedAgentServerIntegrationTest {
                 turn2.turnId(), owner, Duration.ofMillis(-1))).isPresent();
         assertThat(store.markTurnRecoveryBlocked(tenant,
                 session2.sessionId(), turn2.turnId(), owner,
-                "hosted_workflow_recovery_blocked",
+                recoveryCode,
                 "the Harness generation was lost")).isFalse();
         assertThat(store.findTurn(tenant, session2.sessionId(),
                 turn2.turnId()).get().errorCode()).isNull();
     }
 
+    @Test
+    void aHarnessParkSettlesOnlyFromItsOriginalLeasedJournal() {
+        pauseRecoveryScanning();
+        String tenant = "tenant-harness-recheck-" + UUID.randomUUID();
+        Admission session = store.insertSessionCommand(tenant, "CREATE_SESSION", "create",
+                "sha256:" + "1".repeat(64), "qwen-code", null, null, List.of(), null);
+        Admission turn = store.insertTurnCommand(tenant, "SUBMIT_TURN", "turn", "sha256:" + "2".repeat(64),
+                session.sessionId(), List.of(), "sha256:" + "3".repeat(64));
+        store.claimTurn(tenant, session.sessionId(), turn.turnId(), "first", Duration.ofMinutes(1)).orElseThrow();
+        assertThat(store.bindHarness(tenant, session.sessionId(), turn.turnId(), "first", "boot")).isTrue();
+        store.recordAdmission(tenant, session.sessionId(), turn.turnId(), "first", "epoch", 1);
+        assertThat(store.markTurnRecoveryBlocked(tenant, session.sessionId(), turn.turnId(), "first",
+                "hosted_harness_recovery_blocked", "Awaiting original result")).isTrue();
+        jdbc.update("UPDATE ma_turn SET retry_after=0 WHERE tenant_id=?", tenant);
+        var recheck = store.claimTurn(tenant, session.sessionId(), turn.turnId(), "observer", Duration.ofMinutes(1))
+                .orElseThrow();
+        assertThat(recheck.retryCount()).isZero();
+        var terminal = List.of(new HarnessEvent(2, "boot:epoch:2",
+                new ProjectedEvent("turn.completed", Map.of(), true, "COMPLETED", null, null)));
+        assertThatThrownBy(() -> store.recordHarnessEvents(tenant, session.sessionId(), turn.turnId(),
+                "first", "epoch", terminal)).hasMessageContaining("lease was lost");
+        assertThatThrownBy(() -> store.recordHarnessEvents(tenant, session.sessionId(), turn.turnId(),
+                "observer", "other-epoch", terminal)).hasMessageContaining("epoch changed");
+        store.recordHarnessEvents(tenant, session.sessionId(), turn.turnId(), "observer", "epoch", terminal);
+        assertThat(store.findTurn(tenant, session.sessionId(), turn.turnId()).orElseThrow())
+                .satisfies(record -> {
+                    assertThat(record.status()).isEqualTo("COMPLETED");
+                    assertThat(record.errorCode()).isNull();
+                    assertThat(record.completedAt()).isNotNull();
+                });
+        assertThat(store.beginOperation(tenant, session.sessionId(), OperationKind.CLOSE,
+                "", "close", "close-digest").replayed()).isFalse();
+    }
+
     // The terminal retry budget counts consecutive failures without
     // journaled progress: every fenced cursor update — admission, recovery
     // admission, and the journaled-event cursor — restarts it (review round

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ManagedApprovalChangeTest.java

@@ -162,9 +162,14 @@ class ManagedApprovalChangeTest {
             assertCode(() -> change("actor", "blocked-" + state, "yolo", 0), "session_context_busy");
         }
         for (String code : List.of(com.alibaba.qwen.code.managedagent.session.AgentStateStore.RUNTIME_RECOVERY_BLOCKED,
-                com.alibaba.qwen.code.managedagent.session.AgentStateStore.WORKFLOW_RECOVERY_BLOCKED)) {
+                com.alibaba.qwen.code.managedagent.session.AgentStateStore.WORKFLOW_RECOVERY_BLOCKED,
+                com.alibaba.qwen.code.managedagent.session.AgentStateStore.HARNESS_RECOVERY_BLOCKED)) {
             jdbc.update("UPDATE ma_turn SET status='RUNNING', error_code=? WHERE tenant_id=?", code, tenant);
             assertCode(() -> change("actor", "blocked-" + code, "yolo", 0), "session_context_busy");
+            assertThat(jdbc.queryForObject("SELECT approval_revision FROM ma_session WHERE tenant_id=? AND session_id=?",
+                    Long.class, tenant, session)).isZero();
+            assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_request_receipt WHERE tenant_id=?"
+                    + " AND operation='SET_APPROVAL'", Integer.class, tenant)).isZero();
         }
     }
 

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ManagedSessionStoreIntegrationTest.java

@@ -88,6 +88,40 @@ class ManagedSessionStoreIntegrationTest {
     @Autowired
     private ManagedSessionStore sessionStore;
 
+    @Test
+    void checkpointResourcesUseTheReaderBudgetWhileOrdinaryResourcesStaySmall() throws Exception {
+        for (String kind : List.of("managed-checkpoint", "managed-context", "oversized-checkpoint")) {
+            String session = "checkpoint-budget-" + UUID.randomUUID();
+            String base = "/internal/managed-session-store/v1/sessions/" + session;
+            mvc.perform(post(base + "/writers:acquire")
+                    .header(TenantContextFilter.HEADER, TENANT)
+                    .header(ManagedSessionStoreModels.WRITER_TOKEN_HEADER, TOKEN_A)
+                    .contentType(MediaType.APPLICATION_JSON)
+                    .content(writerRequest(WRITER_A).toString())).andExpect(status().isOk());
+            int length = ManagedSessionStoreModels.MAX_EVENT_BYTES
+                    + ("oversized-checkpoint".equals(kind) ? 1 : 0);
+            byte[] bytes = new byte[length];
+            Arrays.fill(bytes, (byte) 'x');
+            String records = "{\"subtype\":\"session_execution_engine\"}\n"
+                    + "{\"subtype\":\"managed_session_header_v1\"}\n";
+            ObjectNode request = genesisRequest(records, bytes);
+            ((ObjectNode) request.withArray("resources").get(0)).put("kind",
+                    "managed-context".equals(kind) ? kind : "managed-checkpoint");
+            var result = mvc.perform(post(base + "/transactions:commit")
+                    .header(TenantContextFilter.HEADER, TENANT)
+                    .header(ManagedSessionStoreModels.WRITER_TOKEN_HEADER, TOKEN_A)
+                    .contentType(MediaType.APPLICATION_JSON).content(request.toString()));
+            if ("managed-checkpoint".equals(kind)) {
+                result.andExpect(status().isOk());
+                assertThat(sessionStore.readResource(TENANT, WORKSPACE, session,
+                        "resource-context", TOKEN_A).bytes()).isEqualTo(bytes);
+            } else {
+                result.andExpect(status().isNotImplemented())
+                        .andExpect(jsonPath("$.error.code").value("managed_session_oss_disabled"));
+            }
+        }
+    }
+
     @Test
     void acceptsClientBuiltCensusAfterTheDefinitionIsCommitted(@TempDir Path directory) throws Exception {
         ArrayNode fixtures = objectMapper.createArrayNode();

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ToolPublicationContractTest.java

@@ -38,7 +38,16 @@ class ToolPublicationContractTest {
         JsonNode schema = JSON.readTree(contracts().resolve(
                 "managed-tool-publication-v1.schema.json").toFile());
         var factory = JsonSchemaFactory.getInstance(VersionFlag.V202012);
-        for (JsonNode example : suite.required("cases")) {
+        var cases = (com.fasterxml.jackson.databind.node.ArrayNode) suite.required("cases").deepCopy();
+        for (JsonNode size : suite.required("referenceSizeCases")) {
+            ObjectNode value = (ObjectNode) cases.get(0).required("value").deepCopy();
+            ((ObjectNode) value.get(size.required("field").asText())).put("byteLength",
+                    size.required("byteLength").asLong());
+            cases.addObject().put("id", size.toString()).put("kind", "binding")
+                    .put("schemaValid", size.required("valid").asBoolean())
+                    .put("valid", size.required("valid").asBoolean()).set("value", value);
+        }
+        for (JsonNode example : cases) {
             String kind = example.required("kind").asText();
             JsonNode value = example.required("value");
             var selected = schema.deepCopy();

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ToolPublicationStoreTest.java

@@ -2751,6 +2751,29 @@ class ToolPublicationStoreTest {
                 .isEqualTo(original.path("bindingDigest"));
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void acceptsLargeCurrentCheckpointsWithFreshOrOriginalBindings(boolean freshBinding) {
+        ObjectNode next = checkpoint.deepCopy();
+        var items = (com.fasterxml.jackson.databind.node.ArrayNode) next.path("tools").path("items");
+        ObjectNode first = (ObjectNode) items.get(0);
+        for (int i = 2; i <= 400; i++) {
+            items.add(first.deepCopy().put("executionCallId", "execution-" + i)
+                    .put("functionCallId", "model-" + i));
+        }
+        JsonNode nextRef = ref("checkpoint-2", "managed-checkpoint", next);
+        assertThat(nextRef.path("byteLength").asLong()).isBetween(65_537L, 1_048_576L);
+        ObjectNode payload = checkpointPayload("checkpoint-2");
+        payload.put("previousCheckpointId", "checkpoint-1");
+        payload.set("stateRef", nextRef);
+        append("checkpoint.commit", event(3, "checkpoint.committed", payload) + MARKER, 1,
+                List.of(resource(nextRef, next)), "checkpoint-2");
+        if (freshBinding) {
+            binding.set("checkpointRef", nextRef);
+        }
+        assertThat(reserve().path("state").asText()).isEqualTo("OPEN");
+    }
+
     @Test
     void renewsOriginalBindingWhenAnotherToolAdvancesTheWaitCheckpoint() {
         JsonNode original = reserve();

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/harness/QwenHostedHarnessConnectorTest.java

@@ -2852,7 +2852,7 @@ class QwenHostedHarnessConnectorTest {
                     ReflectionTestUtils.setField(connector, "client", alive);
                     return ref;
                 });
-        return new Object[] {connector, dead, alive};
+        return new Object[] {connector, dead, alive, ref};
     }
 
     @Test
@@ -2909,6 +2909,95 @@ class QwenHostedHarnessConnectorTest {
         verify(alive).streamEvents(any());
     }
 
+    @ParameterizedTest
+    @CsvSource({"false,true,false,true", "true,true,false,false", "false,false,false,false", "false,true,true,false"})
+    void streamRecoveryStatusUsesTheSameInactiveAttachment(boolean active,
+            boolean blocked, boolean pendingTranscript, boolean expected) {
+        Object[] fixture = refetchFixture();
+        QwenHostedHarnessConnector connector = (QwenHostedHarnessConnector) fixture[0];
+        HostedHarnessClient client = (HostedHarnessClient) fixture[2];
+        when(client.streamEvents(any())).thenReturn(mock(HarnessEventStream.class));
+        var status = mock(com.alibaba.qwen.code.daemon.HarnessSessionStatus.class);
+        when(status.hasActivePrompt()).thenReturn(active);
+        when(status.getRaw()).thenReturn(Map.of("recoveryBlocked", blocked));
+        when(client.getStatus(any())).thenReturn(status);
+        var page = mock(com.alibaba.qwen.code.daemon.HarnessTranscriptPage.class);
+        when(page.getEvents()).thenReturn(pendingTranscript
+                ? List.<Object>of(Map.of("type", "turn_complete")) : List.of());
+        when(client.getTranscript(any())).thenReturn(page);
+        try (var stream = connector.stream("tenant-a", SESSION_ID, 0L, "epoch")) {
+            assertThat(stream.recoveryBlocked(42L)).isEqualTo(expected);
+        }
+        verify(client).getStatus((HarnessSessionRef) fixture[3]);
+        if (!active && blocked) {
+            var query = org.mockito.ArgumentCaptor.forClass(com.alibaba.qwen.code.daemon.GetHarnessTranscript.class);
+            verify(client).getTranscript(query.capture());
+            assertThat(ReflectionTestUtils.getField(query.getValue(), "session")).isSameAs(fixture[3]);
+            assertThat(ReflectionTestUtils.getField(query.getValue(), "cursor")).isEqualTo("42");
+            assertThat(ReflectionTestUtils.getField(query.getValue(), "limit")).isEqualTo(1);
+        } else {
+            verify(client, never()).getTranscript(any());
+        }
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"cold", "attachment-generation", "client-generation"})
+    void readOnlyObservationNeverLoadsAMissingOrChangedAttachment(String mode) {
+        Object[] fixture = refetchFixture();
+        QwenHostedHarnessConnector connector = (QwenHostedHarnessConnector) fixture[0];
+        HostedHarnessClient original = (HostedHarnessClient) fixture[1];
+        HostedHarnessClient client = (HostedHarnessClient) fixture[2];
+        if (!"cold".equals(mode)) {
+            connector.createOrLoad("tenant-a", SESSION_ID, true);
+            clearInvocations(original, client);
+        }
+        if ("client-generation".equals(mode)) {
+            when(client.capabilities().getBootId()).thenReturn(NEW_BOOT_ID);
+        }
+        String boot = "attachment-generation".equals(mode) ? NEW_BOOT_ID : BOOT_ID;
+        assertThatThrownBy(() -> connector.observeStream("tenant-a", SESSION_ID, boot, 42, "epoch"))
+                .isInstanceOf(IllegalStateException.class).hasMessageContaining("unavailable");
+        for (HostedHarnessClient candidate : List.of(original, client)) {
+            verify(candidate, never()).loadSession(any());
+            verify(candidate, never()).createSession(any());
+            verify(candidate, never()).streamEvents(any());
+        }
+    }
+
+    @Test
+    void readOnlyObservationKeepsItsOriginalClientAndAttachment() {
+        Object[] fixture = refetchFixture();
+        QwenHostedHarnessConnector connector = (QwenHostedHarnessConnector) fixture[0];
+        HostedHarnessClient original = (HostedHarnessClient) fixture[1];
+        HostedHarnessClient client = (HostedHarnessClient) fixture[2];
+        HarnessSessionRef ref = (HarnessSessionRef) fixture[3];
+        connector.createOrLoad("tenant-a", SESSION_ID, true);
+        clearInvocations(original, client);
+        HarnessEventStream events = mock(HarnessEventStream.class);
+        when(events.getEventEpoch()).thenReturn("epoch");
+        when(client.streamEvents(any())).thenReturn(events);
+        var status = mock(com.alibaba.qwen.code.daemon.HarnessSessionStatus.class);
+        when(status.getRaw()).thenReturn(Map.of("recoveryBlocked", true));
+        when(client.getStatus(ref)).thenReturn(status);
+        var page = mock(com.alibaba.qwen.code.daemon.HarnessTranscriptPage.class);
+        when(page.getEvents()).thenReturn(List.of());
+        when(client.getTranscript(any())).thenReturn(page);
+        try (var stream = connector.observeStream("tenant-a", SESSION_ID, BOOT_ID, 42, "epoch")) {
+            ReflectionTestUtils.setField(connector, "client", original);
+            assertThat(stream.recoveryBlocked(42)).isTrue();
+        }
+        var request = ArgumentCaptor.forClass(com.alibaba.qwen.code.daemon.StreamHarnessEvents.class);
+        verify(client).streamEvents(request.capture());
+        assertThat(ReflectionTestUtils.getField(request.getValue(), "session")).isSameAs(ref);
+        assertThat(ReflectionTestUtils.getField(request.getValue(), "eventEpoch")).isEqualTo("epoch");
+        assertThat(ReflectionTestUtils.getField(request.getValue(), "lastEventId")).isEqualTo(42L);
+        verify(client).getStatus(ref);
+        verify(events).close();
+        verifyNoInteractions(original);
+        verify(client, never()).loadSession(any());
+        verify(client, never()).createSession(any());
+    }
+
     @Test
     void resolveActionRefetchesTheClientAfterResolvingAttachment()
             throws Exception {

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/persistence/WorkspaceCsiCheckpointSnapshotStoreTest.java

@@ -279,7 +279,7 @@ class WorkspaceCsiCheckpointSnapshotStoreTest {
     @ValueSource(strings = {"inline", "transaction", "rows", "total"})
     void refusesFixedCapacityOverflow(String limit) {
         if ("inline".equals(limit)) {
-            byte[] bytes = new byte[ManagedSessionStoreModels.MAX_INLINE_RESOURCE_BYTES + 1];
+            byte[] bytes = new byte[ManagedSessionStoreModels.MAX_EVENT_BYTES + 1];
             jdbc.update("UPDATE ma_resource SET inline_bytes = ?, byte_length = ?, sha256 = ?"
                     + " WHERE resource_id = X'636865636b706f696e74'", bytes, bytes.length, hash(bytes));
         } else {

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/persistence/WorkspaceCsiSqlWorkerMySqlIT.java

@@ -446,7 +446,12 @@ class WorkspaceCsiSqlWorkerMySqlIT {
                 assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_request_receipt WHERE operation = 'CSI_RESOURCE_READ'"
                         + " AND state = 'RETURNED'", Integer.class)).isPositive();
                 assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_request_receipt WHERE operation = 'CSI_RESOURCE_READ'"
-                        + " AND state = 'OPEN'", Integer.class)).isZero();
+                        + " AND state = 'OPEN'", Integer.class))
+                        .as(() -> "Unfinished original CSI reads: " + jdbc.queryForList(
+                                "SELECT result_json FROM ma_request_receipt WHERE operation = 'CSI_RESOURCE_READ' AND state = 'OPEN'")
+                                + "; session heads: " + jdbc.queryForList(
+                                "SELECT writer_generation, writer_lease_until, activation_epoch, activation_expires_at, journal_revision FROM ma_session_head"))
+                        .isZero();
                 assertThat(nativeReads.get()).isGreaterThanOrEqualTo(7);
                 String schema;
                 try (var connection = source.getConnection()) {

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/session/HarnessCoordinatorTest.java

@@ -5,6 +5,7 @@ import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
 import com.alibaba.qwen.code.daemon.DaemonHttpException;
 import com.alibaba.qwen.code.daemon.DaemonProtocolException;
+import com.alibaba.qwen.code.daemon.DaemonTransportException;
 import com.alibaba.qwen.code.daemon.HarnessRuntimeRecovery;
 import com.alibaba.qwen.code.daemon.HarnessSessionRefusedException;
 import com.alibaba.qwen.code.daemon.HostedHarnessGenerationException;
@@ -54,6 +55,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.argThat;
@@ -1166,8 +1168,10 @@ class HarnessCoordinatorTest {
     // — the mark cannot park a CANCELLING row (findDispatchable re-admits
     // it by design), so writing it would re-open the claim-refuse-release
     // spin this fix exists to close (P1 review finding on #14461).
-    @Test
-    void aCancellingWorkflowTurnKeepsItsBoundedPathWhenTheTakeoverDeclines() {
+    @ParameterizedTest
+    @org.junit.jupiter.params.provider.NullSource
+    @ValueSource(strings = AgentStateStore.HARNESS_RECOVERY_BLOCKED)
+    void aCancellingWorkflowTurnKeepsItsBoundedPathWhenTheTakeoverDeclines(String recoveryCode) {
         AgentStateStore store = mock(AgentStateStore.class);
         HarnessConnector harness = mock(HarnessConnector.class);
         RuntimeWarmer runtimeWarmer = mock(RuntimeWarmer.class);
@@ -1177,7 +1181,7 @@ class HarnessCoordinatorTest {
                 List.of(Map.of("type", "workflow_launch", "script",
                         "return 1;")),
                 "sha256:" + "a".repeat(64), "CANCELLING", true, "epoch-old",
-                1L, "previous-owner", Long.MAX_VALUE, 0, 0, 0, null, null,
+                1L, "previous-owner", Long.MAX_VALUE, 0, 0, 0, null, recoveryCode,
                 null, 1, 1, null, 1);
         when(store.claimTurn(eq("tenant"), eq("session"), eq("turn"),
                 anyString(), any(Duration.class)))
@@ -1199,6 +1203,8 @@ class HarnessCoordinatorTest {
         } finally {
             coordinator.close();
         }
+        verify(harness).recoverManagedRuntime("tenant", "session", true);
+        verify(harness, never()).observeStream(anyString(), anyString(), anyString(), anyLong(), anyString());
         // The cancel keeps its bounded retry path — the mark that cannot
         // park a CANCELLING row is never written, and no terminal is forged.
         verify(store, never()).markTurnRecoveryBlocked(anyString(),
@@ -1324,17 +1330,17 @@ class HarnessCoordinatorTest {
     // Session's close settles the Turn.
     @Test
     void aMarkedWorkflowTurnEndsItsDriveBeforeAnyAttach() {
+        String errorCode = AgentStateStore.WORKFLOW_RECOVERY_BLOCKED;
         AgentStateStore store = mock(AgentStateStore.class);
         HarnessConnector harness = mock(HarnessConnector.class);
         RuntimeWarmer runtimeWarmer = mock(RuntimeWarmer.class);
         when(runtimeWarmer.isEnabled()).thenReturn(false);
         TurnRecord claimed = new TurnRecord("tenant", "session", "turn",
                 "prompt",
-                List.of(Map.of("type", "workflow_launch", "script",
-                        "return 1;")),
+                List.of(Map.of("type", "workflow_launch", "script", "return 1;")),
                 "sha256:" + "a".repeat(64), "RUNNING", true, "epoch-old", 1L,
                 "previous-owner", Long.MAX_VALUE, 0, 0, 0, null,
-                AgentStateStore.WORKFLOW_RECOVERY_BLOCKED,
+                errorCode,
                 "The Harness generation running this workflow was lost.",
                 1, 1, null, 1);
         when(store.claimTurn(eq("tenant"), eq("session"), eq("turn"),
@@ -1360,6 +1366,66 @@ class HarnessCoordinatorTest {
                 anyString(), anyString(), anyString(), anyString());
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = {"blocked", "terminal", "unavailable", "epoch-changed", "park-cas-lost", "transport-error"})
+    void aHarnessParkOnlyObservesTheOriginalExecution(String mode) {
+        AgentStateStore store = boundCancellingStore();
+        TurnRecord parked = new TurnRecord("tenant", "session", "turn", "prompt",
+                List.of(Map.of("type", "text", "text", "recover")),
+                "sha256:" + "a".repeat(64), "RUNNING", true, "epoch", 1L,
+                "previous-owner", Long.MAX_VALUE, 0, 0, 0, null,
+                AgentStateStore.HARNESS_RECOVERY_BLOCKED, "Awaiting journal", 1, 1, null, 1);
+        when(store.claimTurn(eq("tenant"), eq("session"), eq("turn"),
+                anyString(), any(Duration.class))).thenReturn(Optional.of(parked));
+        when(store.markTurnRecoveryBlocked(anyString(), anyString(), anyString(),
+                anyString(), anyString(), anyString())).thenReturn(true);
+        HarnessConnector harness = mock(HarnessConnector.class);
+        SourceStream stream = mock(SourceStream.class);
+        when(stream.eventEpoch()).thenReturn("epoch-changed".equals(mode) ? "other" : "epoch");
+        when(stream.recoveryBlocked(1)).thenReturn("blocked".equals(mode));
+        if ("park-cas-lost".equals(mode)) {
+            when(stream.recoveryBlocked(1)).thenThrow(new IllegalStateException("journal unavailable"));
+            when(store.markTurnRecoveryBlocked(anyString(), anyString(), anyString(),
+                    anyString(), anyString(), anyString())).thenReturn(false);
+        } else if ("transport-error".equals(mode)) {
+            when(stream.recoveryBlocked(1)).thenThrow(mock(DaemonTransportException.class));
+        }
+        when(stream.next()).thenReturn(new SourceEvent(2L, "turn_complete",
+                Map.of("stopReason", "end_turn"), "prompt", Map.of()), null);
+        if ("unavailable".equals(mode)) {
+            when(harness.observeStream("tenant", "session", "boot", 1, "epoch"))
+                    .thenThrow(new IllegalStateException("journal unavailable"));
+        } else {
+            when(harness.observeStream("tenant", "session", "boot", 1, "epoch")).thenReturn(stream);
+        }
+        HarnessCoordinator coordinator = new HarnessCoordinator(store, harness,
+                new HarnessEventProjector(), mock(RuntimeWarmer.class), directExecutor(),
+                Clock.systemUTC(), new ManagedAgentProperties());
+        try {
+            coordinator.dispatch("tenant", "session", "turn");
+        } finally {
+            coordinator.close();
+        }
+        verify(harness).observeStream("tenant", "session", "boot", 1, "epoch");
+        verifyNoMoreInteractions(harness);
+        if ("terminal".equals(mode)) {
+            verify(store).recordHarnessEvents(eq("tenant"), eq("session"), eq("turn"),
+                    anyString(), eq("epoch"), argThat(events -> events.size() == 1
+                            && events.get(0).sourceId() == 2 && events.get(0).projection().terminal()));
+            verify(store, never()).markTurnRecoveryBlocked(anyString(), anyString(), anyString(),
+                    anyString(), anyString(), anyString());
+        } else {
+            verify(store).markTurnRecoveryBlocked(eq("tenant"), eq("session"), eq("turn"),
+                    anyString(), eq(AgentStateStore.HARNESS_RECOVERY_BLOCKED), anyString());
+            verify(store, never()).recordHarnessEvents(anyString(), anyString(), anyString(),
+                    anyString(), anyString(), anyList());
+        }
+        verify(store, never()).failTurn(anyString(), anyString(), anyString(),
+                anyString(), anyString(), anyString());
+        verify(store, never()).scheduleTurnRetry(anyString(), anyString(), anyString(),
+                anyString(), anyLong(), anyBoolean());
+    }
+
     // `managed_session_writer_conflict` is the only wait that ends on its
     // own when the fenced predecessor's lease lapses, so a bound Session's
     // recovery attach escapes the pre-admission budget on that code.
@@ -3437,6 +3503,91 @@ class HarnessCoordinatorTest {
         });
     }
 
+    @org.junit.jupiter.params.ParameterizedTest
+    @org.junit.jupiter.params.provider.ValueSource(strings = {"idle", "controls", "terminal", "transport-error"})
+    void anInactiveRecoveryBlockedStreamParksWithoutSettlingOrReplaying(String mode) throws Exception {
+        boolean terminalAtProbe = "terminal".equals(mode) || "transport-error".equals(mode);
+        AgentStateStore store = boundCancellingStore();
+        TurnRecord running = turn("tenant", "session", "turn", "prompt", "epoch", 1);
+        when(store.claimTurn(eq("tenant"), eq("session"), eq("turn"),
+                anyString(), any(Duration.class))).thenReturn(Optional.of(running));
+        when(store.findTurn("tenant", "session", "turn")).thenReturn(Optional.of(running));
+        when(store.renewTurn(anyString(), anyString(), anyString(), anyString(), any(Duration.class)))
+                .thenReturn(true);
+        when(store.bindHarness(eq("tenant"), eq("session"), eq("turn"), anyString(), eq("boot")))
+                .thenReturn(true);
+        HarnessConnector harness = mock(HarnessConnector.class);
+        when(harness.isWorkspaceFilesAvailable()).thenReturn(true);
+        when(harness.recoverManagedRuntime("tenant", "session", false)).thenReturn(new Attachment("boot"));
+        SourceStream stream = mock(SourceStream.class);
+        when(stream.eventEpoch()).thenReturn("epoch");
+        CountDownLatch closed = new CountDownLatch(1);
+        CountDownLatch emitTerminal = new CountDownLatch(1);
+        CountDownLatch terminalQueued = new CountDownLatch(1);
+        var sequence = new java.util.concurrent.atomic.AtomicLong(1);
+        when(stream.next()).thenAnswer(ignored -> {
+            if ("controls".equals(mode) && !closed.await(2, TimeUnit.SECONDS)) {
+                return new SourceEvent(sequence.incrementAndGet(), "managed_journal_event",
+                        Map.of(), null, Map.of());
+            }
+            if (terminalAtProbe) {
+                if (sequence.getAndIncrement() == 1) {
+                    emitTerminal.await();
+                    return new SourceEvent(2L, "turn_complete", Map.of("stopReason", "end_turn"),
+                            "prompt", Map.of());
+                }
+                terminalQueued.countDown();
+            }
+            closed.await();
+            return null;
+        });
+        when(stream.recoveryBlocked(anyLong())).thenAnswer(ignored -> {
+            if (terminalAtProbe) {
+                emitTerminal.countDown();
+                assertThat(terminalQueued.await(5, TimeUnit.SECONDS)).isTrue();
+            }
+            if ("transport-error".equals(mode)) {
+                throw mock(DaemonTransportException.class);
+            }
+            return true;
+        });
+        doAnswer(ignored -> {
+            closed.countDown();
+            return null;
+        }).when(stream).close();
+        when(harness.stream("tenant", "session", 1, "epoch")).thenReturn(stream);
+        when(store.markTurnRecoveryBlocked(anyString(), anyString(), anyString(),
+                anyString(), anyString(), anyString())).thenReturn(true);
+        ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
+        HarnessCoordinator coordinator = new HarnessCoordinator(store, harness,
+                new HarnessEventProjector(), mock(RuntimeWarmer.class), executor,
+                Clock.systemUTC(), new ManagedAgentProperties());
+        try {
+            coordinator.dispatch("tenant", "session", "turn");
+            if (terminalAtProbe) {
+                verify(store, timeout(20_000)).recordHarnessEvents(eq("tenant"), eq("session"),
+                        eq("turn"), anyString(), eq("epoch"), argThat(events -> events.size() == 1
+                                && events.get(0).sourceId() == 2 && events.get(0).projection().terminal()));
+                verify(store, never()).markTurnRecoveryBlocked(anyString(), anyString(), anyString(),
+                        anyString(), anyString(), anyString());
+            } else {
+                verify(store, timeout(20_000)).markTurnRecoveryBlocked(eq("tenant"), eq("session"),
+                        eq("turn"), anyString(), eq(AgentStateStore.HARNESS_RECOVERY_BLOCKED), anyString());
+            }
+        } finally {
+            coordinator.close();
+            executor.shutdownNow();
+        }
+        verify(stream).close();
+        verify(store, never()).scheduleTurnRetry(anyString(), anyString(), anyString(),
+                anyString(), anyLong(), anyBoolean());
+        verify(store, never()).failTurn(anyString(), anyString(), anyString(),
+                anyString(), anyString(), anyString());
+        verify(harness, never()).cancel(anyString(), anyString());
+        verify(harness, never()).submit(anyString(), anyString(), anyString(), any(), anyString());
+        verify(harness, never()).continueManagedRuntime(anyString(), anyString(), anyString(), anyString(), anyString());
+    }
+
     private static void verifyWorkspaceCancellationIsTargeted(HarnessConnector harness) {
         verify(harness, never()).cancel(anyString(), anyString());
         verify(harness, never()).recoverManagedCancellation(anyString(), anyString());

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/workspace/ChildResultRelayTest.java

@@ -483,12 +483,13 @@ class ChildResultRelayTest {
                 .containsExactly("commit_result", "accept", "mark_accepted");
     }
 
-    @Test
-    void aRunningChildWaitsWithoutEatingAttempts() {
+    @org.junit.jupiter.params.ParameterizedTest
+    @org.junit.jupiter.params.provider.ValueSource(strings = {"", AgentStateStore.HARNESS_RECOVERY_BLOCKED})
+    void aRunningChildWaitsWithoutEatingAttempts(String recoveryCode) {
         row.set(new RelayRow(TENANT, PARENT, RUN, "creation-key", CHILD,
                 "watching", "owner", now + 30_000, 0, 0, null, now, now));
         when(store.latestTurn(TENANT, CHILD)).thenReturn(
-                new TurnLine("turn-1", "RUNNING", now + 1L, null, true,
+                new TurnLine("turn-1", "RUNNING", now + 1L, recoveryCode.isEmpty() ? null : recoveryCode, true,
                         null));
         relay.scan();
         // A running Turn is a wait, not a failure: no attempt, no error,

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/runtimebroker/transport/CsiNativeReadbackProtocolTest.java

@@ -115,6 +115,43 @@ public class CsiNativeReadbackProtocolTest {
         assertThrows(IllegalArgumentException.class, () -> CsiNativeReadbackProtocol.executeResult(JsonCodec.encode(response), request));
     }
 
+    @Test
+    void appliesTheCheckpointReaderLimitWithoutWideningOrdinaryReferences() throws Exception {
+        var fixture = JsonCodec.parseObject(Files.readAllBytes(Path.of(
+                "../cli/src/serve/contracts/managed-csi-native-readback-v1.fixtures.json")), "fixture");
+        var valid = ((List<?>) fixture.get("valid")).stream().map(ManagedCsiFilesProtocolTest::map).toList();
+        var execute = valid.stream().filter(row -> "execute".equals(row.get("name"))).findFirst().orElseThrow();
+        var request = CsiNativeReadbackProtocol.request(ManagedCsiFilesProtocolTest.map(execute.get("request")));
+        for (Object candidate : (List<?>) fixture.get("checkpointSizeCases")) {
+            var size = ManagedCsiFilesProtocolTest.map(candidate);
+            var response = new LinkedHashMap<>(ManagedCsiFilesProtocolTest.map(execute.get("response")));
+            var evidence = new LinkedHashMap<>(ManagedCsiFilesProtocolTest.map(response.get("evidence")));
+            var grant = new LinkedHashMap<>(ManagedCsiFilesProtocolTest.map(evidence.get("grant")));
+            var checkpoint = new LinkedHashMap<>(ManagedCsiFilesProtocolTest.map(grant.get("checkpointRef")));
+            checkpoint.put("byteLength", size.get("byteLength"));
+            grant.put("checkpointRef", checkpoint);
+            evidence.put("grant", grant);
+            response.put("evidence", evidence);
+            if (Boolean.TRUE.equals(size.get("valid"))) {
+                assertTrue(BrokerValues.sameJsonMap(response, CsiNativeReadbackProtocol.response(response, request)));
+            } else {
+                assertThrows(IllegalArgumentException.class, () -> CsiNativeReadbackProtocol.response(response, request));
+            }
+        }
+        var prepare = valid.stream().filter(row -> "prepare".equals(row.get("name"))).findFirst().orElseThrow();
+        for (int bytes : List.of(65536, 65537)) {
+            var ordinary = new LinkedHashMap<>(ManagedCsiFilesProtocolTest.map(prepare.get("request")));
+            var subject = new LinkedHashMap<>(ManagedCsiFilesProtocolTest.map(ordinary.get("subject")));
+            subject.put("byteLength", bytes);
+            ordinary.put("subject", subject);
+            if (bytes == 65536) {
+                assertTrue(BrokerValues.sameJsonMap(ordinary, CsiNativeReadbackProtocol.request(ordinary)));
+            } else {
+                assertThrows(IllegalArgumentException.class, () -> CsiNativeReadbackProtocol.request(ordinary));
+            }
+        }
+    }
+
     public static Map<String, Object> executionResult(String text) {
         return Map.of("executionStatus", "success", "responseParts", List.of(Map.of("text", text)));
     }

packages/web-shell/client/components/managed/ManagedSessionsPage.test.tsx

@@ -1740,29 +1740,32 @@ describe('ManagedSessionsPage', () => {
     },
   );
 
-  it('uses the standard composer stop action with Managed cancel authority', async () => {
-    mocks.client.getSession.mockImplementation(async () =>
-      summary('s1', {
-        phase: 'agent_running',
-        capabilities: { canSend: false, canCancel: true },
-      }),
-    );
-    await render('s1', 'en', provider, null);
-    const stop = [...container.querySelectorAll('button')].find(
-      (button) => button.textContent === 'Stop managed task',
-    );
-    expect(stop?.disabled).toBe(false);
-    await act(async () => {
-      stop!.click();
-      await flush();
-    });
-    expect(mocks.client.cancel).toHaveBeenCalledWith(
-      's1',
-      'p1',
-      expect.objectContaining({ idempotencyKey: expect.any(String) }),
-    );
-    expect(mocks.client.submitPrompt).not.toHaveBeenCalled();
-  });
+  it.each(['agent_running', 'failed'] as const)(
+    'uses the standard composer stop action with Managed cancel authority in phase %s',
+    async (phase) => {
+      mocks.client.getSession.mockImplementation(async () =>
+        summary('s1', {
+          phase,
+          capabilities: { canSend: false, canCancel: true },
+        }),
+      );
+      await render('s1', 'en', provider, null);
+      const stop = [...container.querySelectorAll('button')].find(
+        (button) => button.textContent === 'Stop managed task',
+      );
+      expect(stop?.disabled).toBe(false);
+      await act(async () => {
+        stop!.click();
+        await flush();
+      });
+      expect(mocks.client.cancel).toHaveBeenCalledWith(
+        's1',
+        'p1',
+        expect.objectContaining({ idempotencyKey: expect.any(String) }),
+      );
+      expect(mocks.client.submitPrompt).not.toHaveBeenCalled();
+    },
+  );
 
   it('handles Escape only for the visible Managed conversation after composition and other controls', async () => {
     mocks.client.getSession.mockImplementation(async () =>

packages/web-shell/client/components/managed/ManagedSessionsPage.tsx

@@ -2054,7 +2054,9 @@ function ManagedSessionsContent({
                   approvalChange.busy
                 }
                 dialogOpen={!interactive}
-                isRunning={home && Boolean(active)}
+                isRunning={
+                  home && Boolean(active || summary?.capabilities.canCancel)
+                }
                 onCancel={
                   home &&
                   cancellationEnabled &&

packages/web-shell/client/components/managed/generated/managed-agent-api.ts

@@ -958,6 +958,7 @@ export interface components {
             turnId: string;
             /** Format: uuid */
             sessionId: string;
+            /** @description recovery_blocked with hosted_harness_recovery_blocked means the Harness stopped without a safely settled Turn. The input remains active internally. Execution redispatch is parked, while bounded read-only observation of the original Harness journal may settle a later terminal result. Policy changes and lifecycle close wait for settlement; explicit cancellation remains available. */
             status: string;
             /** Format: int64 */
             submittedAt: number;

packages/web-shell/client/components/managed/java-managed-agent-provider.test.ts

@@ -126,6 +126,50 @@ describe('createJavaManagedAgentProvider', () => {
     ).toBeUndefined();
   });
 
+  it.each([undefined, true, false])(
+    'keeps a recovery-blocked Turn parked with server cancel permission %s',
+    async (cancel) => {
+      const provider = createJavaManagedAgentProvider({
+        baseUrl: 'https://product.example',
+        fetch: vi.fn<typeof fetch>().mockResolvedValue(
+          jsonResponse({
+            sessionId: 's1',
+            title: 'Blocked',
+            status: 'active',
+            createdAt: 1,
+            updatedAt: 2,
+            ...(cancel === undefined
+              ? {}
+              : {
+                  recovery: {
+                    state: 'awaiting_evidence',
+                    reasonCode: 'recovery_evidence_unavailable',
+                    turnId: 't1',
+                    nextTurnAllowed: false,
+                    actions: {
+                      recheck: false,
+                      cancel,
+                      continueInNewSession: true,
+                    },
+                  },
+                }),
+            activeTurn: {
+              turnId: 't1',
+              status: 'recovery_blocked',
+              submittedAt: 1,
+              errorCode: 'hosted_harness_recovery_blocked',
+            },
+          }),
+        ),
+      });
+      const summary = await provider.getSession('s1', { clientId: 'client' });
+      expect(summary.phase).toBe('failed');
+      expect(summary.capabilities.canSend).toBe(false);
+      expect(summary.capabilities.canCancel).toBe(cancel !== false);
+      expect(summary.failure?.code).toBe('hosted_harness_recovery_blocked');
+    },
+  );
+
   it.each([true, false, undefined])(
     'uses the advertised rename capability %s independently of a running Turn',
     async (rename) => {

packages/web-shell/client/components/managed/java-managed-agent-provider.ts

@@ -783,6 +783,7 @@ function toSessionSummary(
   session: JavaAgentSession,
 ): ManagedAgentSessionSummary {
   const turnStatus = session.activeTurn?.status.toLowerCase();
+  const recoveryBlocked = turnStatus === 'recovery_blocked';
   const runtimeState = toRuntimeState(session.environment?.state);
   const phase = toPhase(turnStatus, runtimeState);
   const active =
@@ -859,6 +860,7 @@ function toSessionSummary(
         sessionActive &&
         !active &&
         !session.contextCompressionOperationId &&
+        !recoveryBlocked &&
         session.recovery?.nextTurnAllowed !== false &&
         (!session.workspace ||
           (workspaceTurns && session.capabilities?.foregroundShell !== false)),
@@ -867,7 +869,7 @@ function toSessionSummary(
       // server refuses anyone else.
       canCancel:
         sessionActive &&
-        active &&
+        (active || recoveryBlocked) &&
         turnStatus !== 'cancelling' &&
         session.recovery?.actions.cancel !== false,
       ...(session.capabilities?.durableOperations === true