fix(serve): Fix Hosted local state and foreground Shell validation (#15670)

* fix(serve): isolate Hosted sessions from local transcript state

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

* fix(serve): Skip local PR refresh in Hosted Harness

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

* fix(serve): Match advertised foreground Shell arguments

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

* test(managed-agent): Stabilize release verification fixtures

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

* fix(serve): Normalize Hosted Shell inputs and complete isolation

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

---------

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

packages/cli/src/config/config.ts

@@ -232,10 +232,9 @@ export interface CliArgs {
   /** Specify a session ID without session resumption */
   sessionId: string | undefined;
   /**
-   * Internal: `sessionId` was generated by the daemon just now (not
-   * caller-supplied), so the caller-id occupancy check is skipped — a fresh
-   * UUID has no case-twin to protect, and the check's readdir cost and
-   * fail-closed policy must not apply to the id-less creation hot path.
+   * Internal: the caller owns Session identity independently of local
+   * transcript occupancy: a freshly daemon-generated UUID, or a Hosted-store
+   * ID used with chatRecording disabled. Skip the local occupancy check.
    */
   sessionIdGenerated?: boolean | undefined;
   /**
@@ -2424,9 +2423,8 @@ export async function loadCliConfig(
     // Check if session ID is already in use — case-insensitively: a legacy
     // mixed-case transcript still occupies the id, and creating a
     // case-only twin would make both spellings permanently unrestorable.
-    // A daemon-generated fresh UUID (`sessionIdGenerated`) skips the check:
-    // it has no case-twin to protect, and the check's readdir cost and
-    // fail-closed policy must not hit the id-less creation hot path.
+    // Daemon-generated fresh UUIDs and Hosted-store IDs with local recording
+    // disabled (`sessionIdGenerated`) skip this local occupancy check.
     if (!argv.sessionIdGenerated) {
       const sessionService = new SessionService(cwd);
       let occupied: boolean;

packages/cli/src/serve/hosted-context-compression.test.ts

@@ -187,6 +187,11 @@ describe('Hosted context compression', () => {
     const before = session.authority.committedSequence;
     await compression.admit(request);
     await compression.wait();
+    expect(vi.mocked(loadCliConfig).mock.calls[0]?.[1]).toMatchObject({
+      sessionId: session.authority.sessionHeader.sessionKey.sessionId,
+      sessionIdGenerated: true,
+      chatRecording: false,
+    });
     const result = compression.status(request.operationId)!;
     expect(result).toMatchObject({
       status: 'completed',

packages/cli/src/serve/hosted-context-compression.ts

@@ -345,6 +345,7 @@ export class HostedContextCompression {
       acp: true,
       safeMode: true,
       chatRecording: false,
+      sessionIdGenerated: true,
       sessionId:
         this.options.session.authority.sessionHeader.sessionKey.sessionId,
     } as CliArgs;

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

@@ -1080,7 +1080,12 @@ describe('Hosted Harness model boundary', () => {
     expect(hooks.setHistory).toHaveBeenCalledWith([]);
     expect(hooks.shutdown).toHaveBeenCalledOnce();
     const args = vi.mocked(loadCliConfig).mock.calls[0];
-    expect(args?.[1]).toMatchObject({ safeMode: true, chatRecording: false });
+    expect(args?.[1]).toMatchObject({
+      safeMode: true,
+      chatRecording: false,
+      sessionId: input.sessionId,
+      sessionIdGenerated: true,
+    });
     expect(args?.[9]?.toolInvocationGuard?.({} as never)).toMatchObject({
       allowed: false,
     });

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

@@ -333,6 +333,8 @@ export async function runHostedHarnessTextTurn(input: {
     safeMode: true,
     chatRecording: false,
     sessionId: input.sessionId,
+    // The Hosted store owns this ID, independently of local transcripts.
+    sessionIdGenerated: true,
   } as CliArgs;
   const config = await loadCliConfig(
     settings.merged,

packages/cli/src/serve/hosted-hook-model.test.ts

@@ -124,6 +124,11 @@ describe('Hosted Hook model boundary', () => {
       (runner) => runner(hook, HookEventName.SessionEnd, input, signal),
     );
     expect(result).toMatchObject({ success: true, outcome: 'success' });
+    expect(vi.mocked(loadCliConfig).mock.calls[0]?.[1]).toMatchObject({
+      sessionId: 'session-1',
+      sessionIdGenerated: true,
+      chatRecording: false,
+    });
     expect(config.initialize).toHaveBeenCalledWith(
       expect.objectContaining({
         skipHooks: true,

packages/cli/src/serve/hosted-hook-model.ts

@@ -149,6 +149,7 @@ export async function runHostedHookOperation<T>(
       acp: true,
       safeMode: true,
       chatRecording: false,
+      sessionIdGenerated: true,
       sessionId: input.sessionId,
     } as CliArgs,
     input.cwd,

packages/cli/src/serve/hosted-workflow-turn.test.ts

@@ -294,6 +294,11 @@ describe('runHostedWorkflowTurn', () => {
     // settings load is pinned untrusted.
     const loadCalls = vi.mocked(loadCliConfig).mock.calls;
     expect(loadCalls).toHaveLength(1);
+    expect(loadCalls[0]![1]).toMatchObject({
+      sessionId: base.sessionId,
+      sessionIdGenerated: true,
+      chatRecording: false,
+    });
     expect(loadCalls[0]).toHaveLength(9);
     expect((loadCalls[0]![1] as { safeMode?: boolean }).safeMode).toBe(true);
     // #13858: the agents' chats open without this Harness's cwd listing.

packages/cli/src/serve/hosted-workflow-turn.ts

@@ -204,6 +204,7 @@ export async function runHostedWorkflowTurn(input: {
     acp: true,
     safeMode: true,
     chatRecording: false,
+    sessionIdGenerated: true,
     sessionId: input.sessionId,
   } as CliArgs;
   // The config is the per-turn hosted shape (safeMode, untrusted

packages/cli/src/serve/hosted-workspace-tool-turn.test.ts

@@ -10729,6 +10729,83 @@ it.each([true, false])(
   },
 );
 
+it.each([
+  { enabled: true, isBackground: false },
+  { enabled: false, isBackground: false },
+  { enabled: true, isBackground: 'FaLsE' },
+  { enabled: false, isBackground: 'FaLsE' },
+])(
+  'matches its advertised explicit foreground Shell contract (background enabled: $enabled, input: $isBackground)',
+  async ({ enabled, isBackground }) => {
+    const rig = backgroundTurnRig({
+      executionStatus: 'not_started',
+      responseParts: [],
+      capture: null,
+    });
+    turn = rig.turn;
+    enablement.childRun = enabled;
+    const declarations = await turn.declarations(new AbortController().signal);
+    const shell = declarations.find(
+      (tool) => tool.name === 'run_shell_command',
+    );
+    const args = { command: 'echo hi', is_background: isBackground };
+    const call = { ...calls[0], name: 'run_shell_command', args };
+    const results = await turn.execute(
+      [call],
+      [{ functionCall: { id: call.callId, name: call.name, args } }],
+      'model',
+      new AbortController().signal,
+    );
+    if (enabled) {
+      expect(shell?.parametersJsonSchema).toHaveProperty(
+        'properties.is_background',
+      );
+      expect(broker.prepareV3).toHaveBeenCalledOnce();
+      expect(broker.executeV3).toHaveBeenCalledOnce();
+      expect(JSON.parse(broker.executeV3.mock.calls[0]![1])).toMatchObject({
+        toolName: 'run_shell_command',
+        input: { command: 'echo hi', is_background: false },
+      });
+      expect(
+        JSON.parse(broker.executeV3.mock.calls[0]![1]).input.is_background,
+      ).toBe(false);
+    } else {
+      expect(shell?.parametersJsonSchema).not.toHaveProperty(
+        'properties.is_background',
+      );
+      expect(results[0]?.functionResponse?.response?.['error']).toContain(
+        'foreground command',
+      );
+      expect(broker.prepareV3).not.toHaveBeenCalled();
+      expect(broker.executeV3).not.toHaveBeenCalled();
+    }
+    expect(rig.orchestrator.calls).toEqual([]);
+  },
+);
+
+it('reports a malformed background flag without hiding admitted capabilities', async () => {
+  const rig = backgroundTurnRig({
+    executionStatus: 'not_started',
+    responseParts: [],
+    capture: null,
+  });
+  turn = rig.turn;
+  const args = { command: 'echo hi', is_background: 'yes' };
+  const call = { ...calls[0], name: 'run_shell_command', args };
+  const results = await turn.execute(
+    [call],
+    [{ functionCall: { id: call.callId, name: call.name, args } }],
+    'model',
+    new AbortController().signal,
+  );
+  const error = results[0]?.functionResponse?.response?.['error'];
+  expect(error).toContain('is_background');
+  expect(error).not.toContain('unavailable');
+  expect(broker.prepareV3).not.toHaveBeenCalled();
+  expect(broker.executeV3).not.toHaveBeenCalled();
+  expect(rig.orchestrator.calls).toEqual([]);
+});
+
 it.each([true, false])(
   'preserves extension Shell pins with background admission enabled (%s)',
   async (enabled) => {

packages/cli/src/serve/hosted-workspace-tool-turn.ts

@@ -3952,8 +3952,9 @@ export class HostedWorkspaceToolTurn {
             backgroundIllFormed ||
             (backgroundRequested && !backgroundAdmitted)
           ) {
-            validationError =
-              'Hosted Shell requires a foreground command in the saved directory. Background jobs and Monitor are unavailable; correct the arguments before retrying.';
+            validationError = this.backgroundShellAdmitted()
+              ? 'Hosted Shell is_background must be a boolean; correct the arguments before retrying.'
+              : 'Hosted Shell requires a foreground command in the saved directory. Background jobs and Monitor are unavailable; correct the arguments before retrying.';
           } else if (
             args['description'] !== undefined &&
             typeof args['description'] !== 'string'
@@ -3969,7 +3970,7 @@ export class HostedWorkspaceToolTurn {
               'Hosted Shell timeout must be an integer from 1 to 600000 ms.';
           }
           if (
-            !backgroundAdmitted &&
+            !this.backgroundShellAdmitted() &&
             this.publication &&
             args['is_background'] !== undefined
           ) {
@@ -3979,7 +3980,7 @@ export class HostedWorkspaceToolTurn {
             hostedShellInputError(call.name, args) ?? validationError;
           input = backgroundAdmitted
             ? { ...args, is_background: true }
-            : this.publication
+            : this.publication && args['is_background'] === undefined
               ? { ...args }
               : { ...args, is_background: false };
         } else if (call.name === 'web_fetch') {

packages/cli/src/serve/run-qwen-serve.test.ts

@@ -6730,6 +6730,70 @@ describe('runQwenServe deployment profiles', () => {
     }
   });
 
+  it.each([
+    { hosted: false, deferred: false },
+    { hosted: true, deferred: false },
+    { hosted: false, deferred: true },
+    { hosted: true, deferred: true },
+  ])(
+    'scopes local session recovery to ordinary profiles (Hosted: $hosted, deferred: $deferred)',
+    async ({ hosted, deferred }) => {
+      const workspace = fs.realpathSync(
+        fs.mkdtempSync(path.join(os.tmpdir(), 'qws-hosted-local-state-')),
+      );
+      const chats = vi.spyOn(
+        qwenCore.SessionService.prototype,
+        'getWorktreeSessionPath',
+      );
+      let handle: RunHandle | undefined;
+      let restoreBridge: (() => void) | undefined;
+      try {
+        const options = {
+          port: 0,
+          hostname: '127.0.0.1',
+          mode: 'http-bridge',
+          workspace,
+          serveWebShell: false,
+          ...(hosted
+            ? {
+                profile: 'hosted-harness' as const,
+                token: 'hosted-secret',
+                hostedHarnessCapabilityDigest: `sha256:${'a'.repeat(64)}`,
+                managedRuntimeBrokerUrl: 'https://broker.internal',
+                managedRuntimeBrokerToken: 'broker-secret',
+              }
+            : {}),
+        } satisfies Parameters<typeof runQwenServe>[0];
+        const deps = {
+          trustedWorkspace: true,
+          daemonLogBaseDir: path.join(workspace, 'debug'),
+        };
+        if (deferred) {
+          const started = await startDeferredDaemon(workspace, {
+            serveOptions: options,
+            deps,
+          });
+          handle = started.handle;
+          restoreBridge = () => started.createBridge.mockRestore();
+          await fetch(`${handle.url}/health`);
+          await handle.runtimeReady;
+        } else {
+          handle = await runQwenServe(options, {
+            ...deps,
+            bridge: makeRuntimeBridge(),
+          });
+        }
+        if (hosted) expect(chats).not.toHaveBeenCalled();
+        else expect(chats).toHaveBeenCalled();
+      } finally {
+        await handle?.close();
+        restoreBridge?.();
+        chats.mockRestore();
+        fs.rmSync(workspace, { recursive: true, force: true });
+      }
+    },
+  );
+
   it('hands the update restarter the Broker pair the Hosted Harness resolved', async () => {
     const workspace = fs.realpathSync(
       fs.mkdtempSync(path.join(os.tmpdir(), 'qws-hosted-restart-')),
@@ -14246,7 +14310,7 @@ vi.mock('./server/session-pr-refresh.js', () => {
   );
 });
 
-describe('session-pr-refresh degraded load on the serve fast path', () => {
+describe('session-pr-refresh isolation on the serve fast path', () => {
   let tmpDir: string;
 
   afterEach(() => {
@@ -14256,37 +14320,55 @@ describe('session-pr-refresh degraded load on the serve fast path', () => {
     }
   });
 
-  it('degrades to no PR-state sweep instead of leaking an unhandled rejection', async () => {
-    tmpDir = fs.realpathSync(
-      fs.mkdtempSync(path.join(os.tmpdir(), 'qws-pr-refresh-degrade-')),
-    );
-    // The serve fast path installs no process-level unhandledRejection
-    // handler before the runtime builds, so record them here: without the
-    // import's .catch the rejection escapes and Node's default is to exit.
-    const rejections: unknown[] = [];
-    const recordRejection = (reason: unknown): void => {
-      rejections.push(reason);
-    };
-    process.on('unhandledRejection', recordRejection);
-    let handle: RunHandle | undefined;
-    try {
-      ({ handle } = await startDeferredDaemon(tmpDir));
-      const health = await fetch(`${handle.url}/health`);
-      expect(health.status).toBe(200);
-      // The first health schedules the runtime build; the refresh module
-      // import fires inside it and rejects on the next turns.
-      await expect(handle.runtimeReady).resolves.toBeUndefined();
-      await new Promise((resolve) => setTimeout(resolve, 150));
-      expect(rejections).toEqual([]);
-      // The intended degradation is "no PR-state sweep" — the daemon keeps
-      // serving everything else.
-      const healthAfter = await fetch(`${handle.url}/health`);
-      expect(healthAfter.status).toBe(200);
-    } finally {
-      process.off('unhandledRejection', recordRejection);
-      await handle?.close();
-    }
-  });
+  it.each([false, true])(
+    'keeps serving when the refresh chunk is unavailable (Hosted: %s)',
+    async (hosted) => {
+      tmpDir = fs.realpathSync(
+        fs.mkdtempSync(path.join(os.tmpdir(), 'qws-pr-refresh-degrade-')),
+      );
+      // The serve fast path installs no process-level unhandledRejection
+      // handler before the runtime builds, so record them here: without the
+      // import's .catch the rejection escapes and Node's default is to exit.
+      const stderr = vi.spyOn(process.stderr, 'write');
+      const rejections: unknown[] = [];
+      const recordRejection = (reason: unknown): void => {
+        rejections.push(reason);
+      };
+      process.on('unhandledRejection', recordRejection);
+      let handle: RunHandle | undefined;
+      try {
+        ({ handle } = await startDeferredDaemon(tmpDir, {
+          serveOptions: hosted
+            ? {
+                profile: 'hosted-harness',
+                serveWebShell: false,
+                hostedHarnessCapabilityDigest: `sha256:${'a'.repeat(64)}`,
+              }
+            : {},
+        }));
+        const headers = { authorization: 'Bearer secret-token' };
+        const health = await fetch(`${handle.url}/health`, { headers });
+        expect(health.status).toBe(200);
+        // The first health schedules the runtime build; the refresh module
+        // import fires inside it and rejects on the next turns.
+        await expect(handle.runtimeReady).resolves.toBeUndefined();
+        await new Promise((resolve) => setTimeout(resolve, 150));
+        expect(rejections).toEqual([]);
+        expect(
+          stderr.mock.calls.some(([chunk]) =>
+            String(chunk).includes('session-pr-refresh load failed:'),
+          ),
+        ).toBe(!hosted);
+        // The intended degradation is "no PR-state sweep" — the daemon keeps
+        // serving everything else.
+        const healthAfter = await fetch(`${handle.url}/health`, { headers });
+        expect(healthAfter.status).toBe(200);
+      } finally {
+        process.off('unhandledRejection', recordRejection);
+        await handle?.close();
+      }
+    },
+  );
 });
 
 describe('runQwenServe Web Shell signals on RunHandle', () => {

packages/cli/src/serve/run-qwen-serve.ts

@@ -6595,7 +6595,7 @@ async function runQwenServeImpl(
         trustMaterialization: primaryTrustMaterialization,
       },
     ];
-    if (trustedWorkspace) {
+    if (trustedWorkspace && opts.profile !== 'hosted-harness') {
       await recoverBranchWorktreePreparations({
         workspaceCwd: boundWorkspace,
         sessionService: await createBranchRecoverySessionService(
@@ -7461,35 +7461,37 @@ async function runQwenServeImpl(
     // closure check forbids in this pre-listen root's static closure.
     sessionPrRefreshTimer?.dispose();
     const refreshGeneration = ++sessionPrRefreshGeneration;
-    void import('./server/session-pr-refresh.js')
-      .then((mod) => {
-        if (refreshGeneration !== sessionPrRefreshGeneration) return;
-        sessionPrRefreshTimer = mod.startSessionPrRefreshTimer({
-          workspaceRegistry,
-          // The coordinator lives on the serve app (createServeApp below),
-          // which is built after this timer starts; read it per tick like
-          // the metrics sampler reads `acpHandle`.
-          getArchiveCoordinator: () =>
-            (
-              app.locals as {
-                sessionArchiveCoordinator?: SessionArchiveCoordinator;
-              }
-            ).sessionArchiveCoordinator,
+    if (opts.profile !== 'hosted-harness') {
+      void import('./server/session-pr-refresh.js')
+        .then((mod) => {
+          if (refreshGeneration !== sessionPrRefreshGeneration) return;
+          sessionPrRefreshTimer = mod.startSessionPrRefreshTimer({
+            workspaceRegistry,
+            // The coordinator lives on the serve app (createServeApp below),
+            // which is built after this timer starts; read it per tick like
+            // the metrics sampler reads `acpHandle`.
+            getArchiveCoordinator: () =>
+              (
+                app.locals as {
+                  sessionArchiveCoordinator?: SessionArchiveCoordinator;
+                }
+              ).sessionArchiveCoordinator,
+          });
+        })
+        .catch((error) => {
+          // Degrade to "no PR-state sweep" instead of leaking an unhandled
+          // rejection: the serve fast path installs no process-level
+          // unhandledRejection handler before this runs, and Node's default
+          // for one is to exit — a failed chunk load (e.g. an in-place
+          // upgrade replacing dist/ under the running daemon) would take
+          // down every runtime, session, and connection the daemon serves.
+          daemonLog.warn(
+            `session-pr-refresh load failed: ${
+              error instanceof Error ? error.message : String(error)
+            }`,
+          );
         });
-      })
-      .catch((error) => {
-        // Degrade to "no PR-state sweep" instead of leaking an unhandled
-        // rejection: the serve fast path installs no process-level
-        // unhandledRejection handler before this runs, and Node's default
-        // for one is to exit — a failed chunk load (e.g. an in-place
-        // upgrade replacing dist/ under the running daemon) would take
-        // down every runtime, session, and connection the daemon serves.
-        daemonLog.warn(
-          `session-pr-refresh load failed: ${
-            error instanceof Error ? error.message : String(error)
-          }`,
-        );
-      });
+    }
 
     // Factory for dynamically creating workspace runtimes (POST /workspaces).
     interface WorkspaceRuntimeBuildOptions {
@@ -8072,7 +8074,11 @@ async function runQwenServeImpl(
           });
         }
       }
-      if (wsRuntime.primary && wsRuntime.trusted) {
+      if (
+        wsRuntime.primary &&
+        wsRuntime.trusted &&
+        opts.profile !== 'hosted-harness'
+      ) {
         await recoverBranchWorktreePreparations({
           workspaceCwd: cwd,
           sessionService: await createBranchRecoverySessionService(wsRuntime),

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

@@ -19,7 +19,7 @@ import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMock
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.http.MediaType;
 import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.test.context.bean.override.mockito.MockitoBean;
+import org.springframework.test.context.bean.override.convention.TestBean;
 import org.springframework.test.web.servlet.MockMvc;
 
 import java.nio.charset.StandardCharsets;
@@ -30,7 +30,8 @@ import java.util.UUID;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.mockito.Mockito.when;
+import static org.mockito.Answers.RETURNS_DEFAULTS;
+import static org.mockito.Mockito.mock;
 import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@@ -51,18 +52,37 @@ class ManagedSessionCreationSkillsTest {
     @Autowired private ManagedExtensionPackageService extensionPackages;
     @Autowired private org.springframework.transaction.PlatformTransactionManager transactionManager;
     @TempDir private Path archiveDirectory;
-    @MockitoBean private HarnessConnector harness;
+    @TestBean(methodName = "createHarness") private HarnessConnector harness;
+    private static volatile boolean sessionSkillsSupported;
+    private static volatile boolean sessionExtensionsSupported;
+    private static volatile boolean mcpLifecycleSupported;
+    private static volatile boolean publicChildAgentsSupported;
     private String tenant;
     private Boolean previousShellEnabled;
     private final Map<String, Object> skill = Map.of("name", "review", "description", "Review changes",
             "body", "ORIGINAL_INSTRUCTIONS", "base_directory", "docs");
 
+    // Scheduled availability reads share this mock; configure capability answers
+    // at creation instead of mutating Mockito stubbing while those reads run.
+    static HarnessConnector createHarness() {
+        return mock(HarnessConnector.class, call -> switch (call.getMethod().getName()) {
+            case "supportsSessionSkills" -> sessionSkillsSupported;
+            case "supportsSessionExtensions" -> sessionExtensionsSupported;
+            case "supportsPublicChildAgents" -> publicChildAgentsSupported;
+            case "supportsLifecycle" -> call.getArguments().length == 1
+                    && ("hosted-workspace-shell/1".equals(call.getArgument(0))
+                    || mcpLifecycleSupported && "hosted-workspace-mcp/1".equals(call.getArgument(0)));
+            default -> RETURNS_DEFAULTS.answer(call);
+        });
+    }
+
     @BeforeEach
     void setup() {
         tenant = "skills-" + UUID.randomUUID();
-        when(harness.supportsSessionSkills()).thenReturn(true);
-        when(harness.supportsSessionExtensions()).thenReturn(true);
-        when(harness.supportsLifecycle("hosted-workspace-shell/1")).thenReturn(true);
+        sessionSkillsSupported = true;
+        sessionExtensionsSupported = true;
+        mcpLifecycleSupported = false;
+        publicChildAgentsSupported = false;
         properties.getExtensions().setEnabled(false);
         jdbc.update("INSERT INTO ma_workspace (tenant_id, workspace_id, workspace_generation, storage_id,"
                 + " display_name, config_ref, policy_ref, state) VALUES (?, 'ws', 1, 'storage', 'Test', 'config', 'policy', 'ACTIVE')", tenant);
@@ -126,7 +146,7 @@ class ManagedSessionCreationSkillsTest {
         String session = create(body, "empty", false).required("id").asText();
         assertThat(snapshots.initialSelection(tenant, session).get("extensions")).isEqualTo(List.of());
         assertThat(snapshots.snapshot(tenant, session)).isNull();
-        when(harness.supportsSessionExtensions()).thenReturn(false);
+        sessionExtensionsSupported = false;
         submit(body, "unsupported-package-harness", true).andExpect(status().isConflict());
         body.remove("workspace");
         submit(body, "unbound-package", true).andExpect(status().isBadRequest());
@@ -178,7 +198,7 @@ class ManagedSessionCreationSkillsTest {
                         .andExpect(jsonPath("$.error.code").value("unsupported_feature"));
             }
             assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_session WHERE tenant_id=?", Long.class, tenant)).isZero();
-            when(harness.supportsLifecycle("hosted-workspace-mcp/1")).thenReturn(true);
+            mcpLifecycleSupported = true;
             String session = create(body, "negotiated-mcp", true).required("sessionId").asText();
             assertThat(store.requireSession(tenant, session).toolProfile()).isEqualTo("hosted-workspace-mcp/1");
             assertThat(snapshots.initialSelection(tenant, session).get("extensions")).isEqualTo(List.of());
@@ -200,7 +220,7 @@ class ManagedSessionCreationSkillsTest {
         }
         boolean previousManaged = properties.getManagedExtensions().isEnabled();
         try {
-            when(harness.supportsPublicChildAgents()).thenReturn(true);
+            publicChildAgentsSupported = true;
             var definition = json.createObjectNode();
             definition.putObject("model");
             definition.put("instructions", "");
@@ -255,7 +275,7 @@ class ManagedSessionCreationSkillsTest {
         ((com.fasterxml.jackson.databind.node.ObjectNode) body.withArray("skills").get(0)).put("body", "CHANGED");
         submit(body, "key", true).andExpect(status().isConflict())
                 .andExpect(jsonPath("$.error.code").value("idempotency_conflict"));
-        when(harness.supportsSessionSkills()).thenReturn(false);
+        sessionSkillsSupported = false;
         create(request("qwen-code", List.of(skill)), "key", true);
         assertThat(snapshots.effective(store.requireSession(tenant, session))).containsExactly(skill);
         jdbc.update("UPDATE ma_request_receipt SET result_json = REPLACE(result_json, 'ORIGINAL_INSTRUCTIONS', 'TAMPERED')"
@@ -316,9 +336,9 @@ class ManagedSessionCreationSkillsTest {
     @Test
     void refusesUnsupportedUnauthorizedAndInvalidRequestsBeforeCreatingAnything() throws Exception {
         var body = request("qwen-code", List.of(skill));
-        when(harness.supportsSessionSkills()).thenReturn(false);
+        sessionSkillsSupported = false;
         submit(body, "unsupported", true).andExpect(status().isConflict());
-        when(harness.supportsSessionSkills()).thenReturn(true);
+        sessionSkillsSupported = true;
         submit(request("qwen-code", List.of(skill, skill)), "duplicate", true).andExpect(status().isConflict());
         var unbound = request("qwen-code", List.of(skill));
         unbound.remove("workspace");

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

@@ -1204,13 +1204,18 @@ class Issue13181QueryBudgetTest {
         fixture.ledger.reset();
         try (var verifier = new ToolPublicationVerifier(data, 1)) {
             assertThat(verifier.runOnce()).isTrue();
+            long renewals = fixture.ledger.count("update ma_operation set",
+                    "claim_until = ?, verification_next_at = case when");
+            assertThat(renewals).isGreaterThanOrEqualTo(2);
+            assertThat(fixture.ledger.count("from ma_session_journal_tx")).isZero();
+            assertThat(fixture.ledger.count("from ma_session_head", "for update"))
+                    .isGreaterThanOrEqualTo(2 * (renewals + 1));
+            await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
+                verifier.runOnce();
+                assertThat(data.operationStatus(binding.path("sessionKey"), "pub-1", PUBLICATION_TOKEN, "async-heartbeat")
+                        .path("state").asText()).isEqualTo("SUCCEEDED");
+            });
         }
-        long renewals = fixture.ledger.count("update ma_operation set",
-                "claim_until = ?, verification_next_at = case when");
-        assertThat(renewals).isGreaterThanOrEqualTo(2);
-        assertThat(fixture.ledger.count("from ma_session_journal_tx")).isZero();
-        assertThat(fixture.ledger.count("from ma_session_head", "for update"))
-                .isGreaterThanOrEqualTo(2 * (renewals + 1));
         fixture.ledger.reset();
         for (int i = 0; i < 12; i++) {
             assertThat(data.operationStatus(binding.path("sessionKey"), "pub-1", PUBLICATION_TOKEN, "async-heartbeat")

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

@@ -182,6 +182,8 @@ class ContextInstallationFaultGateTest {
         if (attester == Attester.SERVICE) {
             proxy.schedule("attest", FaultProxy.Action.PASS);
         }
+        // Neither attempt of the bounded read-only EOF retry can attest.
+        proxy.schedule("attest", FaultProxy.Action.DROP);
         proxy.schedule("attest", FaultProxy.Action.DROP);
         BrokerProcess broker = rig.broker("broker", proxy,
                 FaultGateRig.Provisioner.LOCAL_PROCESS);
@@ -191,6 +193,7 @@ class ContextInstallationFaultGateTest {
         BrokerProcess.Reply failed = broker.warm(HARNESS);
         assertFalse(failed.ok(), "an unattested managed worker became READY");
         assertFalse(failed.retryable());
+        assertEquals(attester == Attester.PROVISIONER ? 2 : 3, proxy.count("attest"));
         if (attester == Attester.PROVISIONER) {
             assertEquals("runtime_provision_failed", failed.code());
         } else {