fix(managed-agent): address the #13330 R4 review findings

Probe one row past the selection window for the skip-path rotation gate
so an exactly-full window does not rotate rows that are selected again
anyway, pin the post-cap modulo retry at streak 192, and mirror all five
of the client builder's base-URL rules in the boot-time guard.

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

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

@@ -86,9 +86,14 @@ public class QwenHostedHarnessConnector implements HarnessConnector {
         if (scheme == null
                 || !(scheme.equalsIgnoreCase("http")
                         || scheme.equalsIgnoreCase("https"))
-                || baseUri.getHost() == null) {
-            // "localhost:4170" parses as an opaque URI and would only fail at
-            // the first RPC; reject it while the service is starting.
+                || baseUri.getHost() == null
+                || baseUri.getUserInfo() != null
+                || baseUri.getQuery() != null
+                || baseUri.getFragment() != null) {
+            // The credentials, query, and fragment clauses mirror the client
+            // builder's rules: "localhost:4170" parses as an opaque URI and
+            // the rest would only fail at the first RPC, so reject them
+            // while the service is starting.
             throw new IllegalStateException("Enabled Hosted Harness requires"
                     + " an absolute http(s) base URL");
         }

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

@@ -19,11 +19,11 @@ public class MessageMaterializer {
     // Selection gate for failing targets: retry on streaks 1, 2, 4, ...,
     // up to the cap, and once every MAX_BACKOFF_STREAK passes at and past
     // it. A skipped pass rotates the target behind fresher rows only when
-    // the selection window is full — below the limit the row is selected
-    // again anyway, so the write would buy nothing — and the catch already
-    // rotates after a failed attempt. Only the exponential phase warns with
-    // the stack, so a poisoned session can neither starve healthy ones nor
-    // warn at 10 Hz forever.
+    // a row sits behind the selection window — at or below the limit the
+    // same rows are selected again whatever their order, so the write would
+    // buy nothing — and the catch already rotates after a failed attempt.
+    // Only the exponential phase warns with the stack, so a poisoned
+    // session can neither starve healthy ones nor warn at 10 Hz forever.
     private static final int MAX_BACKOFF_STREAK = 64;
     private final AgentStateStore store;
     private final Map<String, Integer> failures = new ConcurrentHashMap<>();
@@ -35,10 +35,14 @@ public class MessageMaterializer {
     @Scheduled(fixedDelayString =
             "${qwen.managed-agent.events.materialize-interval:100ms}")
     public void materialize() {
+        // Saturation means a row sits behind the selection window, so
+        // probe one row past the limit rather than inferring it from an
+        // exactly-full window.
         List<MaterializationTarget> targets =
-                store.findMaterializationTargets(TARGET_LIMIT);
-        boolean saturated = targets.size() == TARGET_LIMIT;
-        for (MaterializationTarget target : targets) {
+                store.findMaterializationTargets(TARGET_LIMIT + 1);
+        boolean saturated = targets.size() > TARGET_LIMIT;
+        for (MaterializationTarget target
+                : targets.subList(0, Math.min(TARGET_LIMIT, targets.size()))) {
             String key = target.tenantId() + ":" + target.sessionId();
             int streak = failures.getOrDefault(key, 0);
             if (streak > 0) {

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

@@ -1361,6 +1361,34 @@ class QwenHostedHarnessConnectorTest {
                 mock(WorkspaceExecutionStore.class)))
                 .isInstanceOf(IllegalStateException.class)
                 .hasMessageContaining("base URL");
+
+        // Credentials, a query, and a fragment are rejected by the client
+        // builder as well, so each fails here instead of lazily inside
+        // client() as an endless transient retry.
+        ManagedAgentProperties credentialed = properties();
+        credentialed.getHarness()
+                .setBaseUrl("https://user:[email protected]:4170");
+        assertThatThrownBy(() -> new QwenHostedHarnessConnector(credentialed,
+                mock(AgentStateStore.class),
+                mock(WorkspaceExecutionStore.class)))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("base URL");
+
+        ManagedAgentProperties queried = properties();
+        queried.getHarness().setBaseUrl("http://127.0.0.1:4170/?v=1");
+        assertThatThrownBy(() -> new QwenHostedHarnessConnector(queried,
+                mock(AgentStateStore.class),
+                mock(WorkspaceExecutionStore.class)))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("base URL");
+
+        ManagedAgentProperties fragmented = properties();
+        fragmented.getHarness().setBaseUrl("http://127.0.0.1:4170/#frag");
+        assertThatThrownBy(() -> new QwenHostedHarnessConnector(fragmented,
+                mock(AgentStateStore.class),
+                mock(WorkspaceExecutionStore.class)))
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageContaining("base URL");
     }
 
     @Test

packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/service/MessageMaterializerTest.java

@@ -24,7 +24,7 @@ class MessageMaterializerTest {
     @Test
     void backsOffAndRotatesAFailingTarget() {
         AgentStateStore store = mock(AgentStateStore.class);
-        when(store.findMaterializationTargets(32))
+        when(store.findMaterializationTargets(33))
                 .thenReturn(List.of(POISON));
         when(store.materializeNextBatch(anyString(), anyString(), anyInt()))
                 .thenThrow(new IllegalStateException("gap"));
@@ -50,7 +50,7 @@ class MessageMaterializerTest {
     @Test
     void warnsWithTheStackOnlyDuringTheExponentialPhase() {
         AgentStateStore store = mock(AgentStateStore.class);
-        when(store.findMaterializationTargets(32))
+        when(store.findMaterializationTargets(33))
                 .thenReturn(List.of(POISON));
         when(store.materializeNextBatch(anyString(), anyString(), anyInt()))
                 .thenThrow(new IllegalStateException("gap"));
@@ -90,7 +90,7 @@ class MessageMaterializerTest {
     @Test
     void atTheCapRetriesOnceEveryMaxStreakPasses() {
         AgentStateStore store = mock(AgentStateStore.class);
-        when(store.findMaterializationTargets(32))
+        when(store.findMaterializationTargets(33))
                 .thenReturn(List.of(POISON));
         when(store.materializeNextBatch(anyString(), anyString(), anyInt()))
                 .thenThrow(new IllegalStateException("gap"));
@@ -114,20 +114,29 @@ class MessageMaterializerTest {
         materializer.materialize();
         verify(store, times(9)).materializeNextBatch("tenant", "poison",
                 200);
+
+        // 192 is the first post-cap due streak that is not also a power of
+        // two, so it separates the modulo rule from unbounded doubling.
+        for (int pass = 0; pass < 64; pass++) {
+            materializer.materialize();
+        }
+        verify(store, times(10)).materializeNextBatch("tenant", "poison",
+                200);
     }
 
     @Test
     void aFailingDeferralDegradesOnlyItsOwnTarget() {
         AgentStateStore store = mock(AgentStateStore.class);
-        // A saturated window (TARGET_LIMIT rows) with the poisoned row at
-        // its head: only a full window rotates on the skip path, and the
-        // faulting first row is the one that must not abort the pass.
+        // A saturated window — a 33rd row behind the TARGET_LIMIT rows —
+        // with the poisoned row at its head: only then does the skip path
+        // rotate, and the faulting first row is the one that must not abort
+        // the pass.
         List<MaterializationTarget> targets = new ArrayList<>();
         targets.add(POISON);
-        for (int i = 1; i < 32; i++) {
+        for (int i = 1; i <= 32; i++) {
             targets.add(new MaterializationTarget("tenant", "healthy-" + i));
         }
-        when(store.findMaterializationTargets(32)).thenReturn(targets);
+        when(store.findMaterializationTargets(33)).thenReturn(targets);
         when(store.materializeNextBatch(anyString(), anyString(), anyInt()))
                 .thenThrow(new IllegalStateException("gap"));
         doThrow(new DataAccessResourceFailureException("lock wait"))
@@ -192,10 +201,38 @@ class MessageMaterializerTest {
                         .isNull());
     }
 
+    @Test
+    void anExactlyFullWindowDoesNotRotateOnTheSkipPath() {
+        AgentStateStore store = mock(AgentStateStore.class);
+        // Exactly TARGET_LIMIT candidates: no row sits behind the window,
+        // so the skipped pass's rotation write would move a row that is
+        // selected again anyway. Only the failed attempts rotate.
+        List<MaterializationTarget> targets = new ArrayList<>();
+        targets.add(POISON);
+        for (int i = 1; i < 32; i++) {
+            targets.add(new MaterializationTarget("tenant", "healthy-" + i));
+        }
+        when(store.findMaterializationTargets(33)).thenReturn(targets);
+        when(store.materializeNextBatch(anyString(), anyString(), anyInt()))
+                .thenThrow(new IllegalStateException("gap"));
+        MessageMaterializer materializer = new MessageMaterializer(store);
+
+        // Four passes: attempts land on streaks 0, 1 and 2, and pass four
+        // is the skipped one.
+        for (int pass = 0; pass < 4; pass++) {
+            materializer.materialize();
+        }
+
+        verify(store, times(3)).materializeNextBatch("tenant", "healthy-31",
+                200);
+        verify(store, times(3)).deferMaterializationTarget("tenant",
+                "healthy-31");
+    }
+
     @Test
     void clearsTheStreakAfterASuccess() {
         AgentStateStore store = mock(AgentStateStore.class);
-        when(store.findMaterializationTargets(32))
+        when(store.findMaterializationTargets(33))
                 .thenReturn(List.of(POISON));
         when(store.materializeNextBatch(anyString(), anyString(), anyInt()))
                 .thenThrow(new IllegalStateException("gap"))