From dd243bd0197d8933543f7163eebb38df072610da Mon Sep 17 00:00:00 2001 From: jinye Date: Tue, 27 Oct 2026 15:48:06 +0800 Subject: [PATCH] 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 * fix(serve): Skip local PR refresh in Hosted Harness Co-authored-by: Qwen-Coder * fix(serve): Match advertised foreground Shell arguments Co-authored-by: Qwen-Coder * test(managed-agent): Stabilize release verification fixtures Co-authored-by: Qwen-Coder * fix(serve): Normalize Hosted Shell inputs and complete isolation Co-authored-by: Qwen-Coder --------- Co-authored-by: Qwen-Coder --- packages/cli/src/config/config.ts | 12 +- .../serve/hosted-context-compression.test.ts | 5 + .../src/serve/hosted-context-compression.ts | 1 + .../src/serve/hosted-harness-model.test.ts | 7 +- .../cli/src/serve/hosted-harness-model.ts | 2 + .../cli/src/serve/hosted-hook-model.test.ts | 5 + packages/cli/src/serve/hosted-hook-model.ts | 1 + .../src/serve/hosted-workflow-turn.test.ts | 5 + .../cli/src/serve/hosted-workflow-turn.ts | 1 + .../serve/hosted-workspace-tool-turn.test.ts | 77 +++++++++ .../src/serve/hosted-workspace-tool-turn.ts | 9 +- packages/cli/src/serve/run-qwen-serve.test.ts | 146 ++++++++++++++---- packages/cli/src/serve/run-qwen-serve.ts | 66 ++++---- .../ManagedSessionCreationSkillsTest.java | 44 ++++-- .../session/Issue13181QueryBudgetTest.java | 17 +- .../ContextInstallationFaultGateTest.java | 3 + 16 files changed, 309 insertions(+), 92 deletions(-) diff --git a/packages/cli/src/config/config.ts b/packages/cli/src/config/config.ts index 941b15c7fa2..e03eebe8343 100755 --- a/packages/cli/src/config/config.ts +++ b/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; diff --git a/packages/cli/src/serve/hosted-context-compression.test.ts b/packages/cli/src/serve/hosted-context-compression.test.ts index 42534ce6c36..c257c415dfd 100644 --- a/packages/cli/src/serve/hosted-context-compression.test.ts +++ b/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', diff --git a/packages/cli/src/serve/hosted-context-compression.ts b/packages/cli/src/serve/hosted-context-compression.ts index a1cd35d6840..4b6c9f9d431 100644 --- a/packages/cli/src/serve/hosted-context-compression.ts +++ b/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; diff --git a/packages/cli/src/serve/hosted-harness-model.test.ts b/packages/cli/src/serve/hosted-harness-model.test.ts index e5877a1a902..efca476e58c 100644 --- a/packages/cli/src/serve/hosted-harness-model.test.ts +++ b/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, }); diff --git a/packages/cli/src/serve/hosted-harness-model.ts b/packages/cli/src/serve/hosted-harness-model.ts index 537197a74d6..d4c91ed3e37 100644 --- a/packages/cli/src/serve/hosted-harness-model.ts +++ b/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, diff --git a/packages/cli/src/serve/hosted-hook-model.test.ts b/packages/cli/src/serve/hosted-hook-model.test.ts index ae59c2f8def..3dd57da60ea 100644 --- a/packages/cli/src/serve/hosted-hook-model.test.ts +++ b/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, diff --git a/packages/cli/src/serve/hosted-hook-model.ts b/packages/cli/src/serve/hosted-hook-model.ts index 706a7bb8e4c..bd82b8bebc7 100644 --- a/packages/cli/src/serve/hosted-hook-model.ts +++ b/packages/cli/src/serve/hosted-hook-model.ts @@ -149,6 +149,7 @@ export async function runHostedHookOperation( acp: true, safeMode: true, chatRecording: false, + sessionIdGenerated: true, sessionId: input.sessionId, } as CliArgs, input.cwd, diff --git a/packages/cli/src/serve/hosted-workflow-turn.test.ts b/packages/cli/src/serve/hosted-workflow-turn.test.ts index c3c5d76140c..c002b9ec73e 100644 --- a/packages/cli/src/serve/hosted-workflow-turn.test.ts +++ b/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. diff --git a/packages/cli/src/serve/hosted-workflow-turn.ts b/packages/cli/src/serve/hosted-workflow-turn.ts index fb8bb5f47c8..980b977b4e5 100644 --- a/packages/cli/src/serve/hosted-workflow-turn.ts +++ b/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 diff --git a/packages/cli/src/serve/hosted-workspace-tool-turn.test.ts b/packages/cli/src/serve/hosted-workspace-tool-turn.test.ts index 644ab59e6f9..e271df6ef9b 100644 --- a/packages/cli/src/serve/hosted-workspace-tool-turn.test.ts +++ b/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) => { diff --git a/packages/cli/src/serve/hosted-workspace-tool-turn.ts b/packages/cli/src/serve/hosted-workspace-tool-turn.ts index 2070cc66f40..053380caef6 100644 --- a/packages/cli/src/serve/hosted-workspace-tool-turn.ts +++ b/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') { diff --git a/packages/cli/src/serve/run-qwen-serve.test.ts b/packages/cli/src/serve/run-qwen-serve.test.ts index 973ec4bb87a..b4cc41741df 100644 --- a/packages/cli/src/serve/run-qwen-serve.test.ts +++ b/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[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', () => { diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index 705fbdb64ad..6763835818e 100644 --- a/packages/cli/src/serve/run-qwen-serve.ts +++ b/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), diff --git a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ManagedSessionCreationSkillsTest.java b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ManagedSessionCreationSkillsTest.java index b6e8c1afef1..1d42e62c1c9 100644 --- a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ManagedSessionCreationSkillsTest.java +++ b/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 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"); diff --git a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/session/Issue13181QueryBudgetTest.java b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/session/Issue13181QueryBudgetTest.java index 89a20260479..54179c1dce8 100644 --- a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/session/Issue13181QueryBudgetTest.java +++ b/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") diff --git a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/runtimebroker/ContextInstallationFaultGateTest.java b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/runtimebroker/ContextInstallationFaultGateTest.java index fab8fe514ac..bd347c65812 100644 --- a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/runtimebroker/ContextInstallationFaultGateTest.java +++ b/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 {