fleetd #581: cover completion CAS removals

2026-09-12 20:19:20 +07:00
parent 204da67d66
commit dfeb9340b4
1 changed files with 143 additions and 0 deletions
@@ -625,6 +625,149 @@ class CompletionResolverTest {
        assertEquals(Rendezvous.Kind.REPLY, waiterB.getNow(null).kind());
    }

+    @Test
+    void aSupersededDoneTurnMustNotEvictItsSuccessorsRegistration() {
+        Rendezvous rendezvous = new Rendezvous();
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(new FakeHerdr()), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none());
+
+        var waiterA = rendezvous.open("term_a");
+        var turnA = new CompletionResolver.InFlight(waiterA, null);
+        assertTrue(rendezvous.resolve("term_a", "A replied"));
+        rendezvous.close("term_a", waiterA);
+        var waiterB = rendezvous.open("term_a");
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB));
+
+        resolver.resolve("term_a", turnA);
+
+        assertSuccessorRegistrationSurvives(resolver, rendezvous, waiterB,
+                "a done turn must not evict B from resolve()'s early return");
+    }
+
+    @Test
+    void aSupersededExhaustedTurnMustNotEvictItsSuccessorsRegistration() {
+        FakeHerdr herdr = new FakeHerdr().readText("⏺ usage limit has been reached\n❯ ");
+        Rendezvous rendezvous = new Rendezvous();
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                target -> Pattern.compile("usage limit has been reached"), ExhaustionSink.none());
+
+        var waiterA = rendezvous.open("term_a");
+        var turnA = new CompletionResolver.InFlight(waiterA, null);
+        rendezvous.close("term_a", waiterA);
+        var waiterB = rendezvous.open("term_a");
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB));
+
+        resolver.resolve("term_a", turnA);
+
+        assertSuccessorRegistrationSurvives(resolver, rendezvous, waiterB,
+                "an exhausted turn must not evict B from resolve()'s exhausted branch");
+    }
+
+    @Test
+    void aSupersededBackendErrorTurnMustNotEvictItsSuccessorsRegistration() {
+        FakeHerdr herdr = new FakeHerdr().readText("⏺ API Error: 400 invalid request body\n❯ ");
+        Rendezvous rendezvous = new Rendezvous();
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none());
+
+        var waiterA = rendezvous.open("term_a");
+        var turnA = new CompletionResolver.InFlight(waiterA, null);
+        rendezvous.close("term_a", waiterA);
+        var waiterB = rendezvous.open("term_a");
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB));
+
+        resolver.resolve("term_a", turnA);
+
+        assertSuccessorRegistrationSurvives(resolver, rendezvous, waiterB,
+                "a backend-error turn must not evict B from resolve()'s error branch");
+    }
+
+    @Test
+    void aSupersededRawExhaustedTurnMustNotEvictItsSuccessorsRegistration() {
+        FakeHerdr herdr = new FakeHerdr().readText("╭────\nusage limit has been reached");
+        Rendezvous rendezvous = new Rendezvous();
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                target -> Pattern.compile("usage limit has been reached"), ExhaustionSink.none());
+
+        var waiterA = rendezvous.open("term_a");
+        var turnA = new CompletionResolver.InFlight(waiterA, null);
+        rendezvous.close("term_a", waiterA);
+        var waiterB = rendezvous.open("term_a");
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB));
+
+        resolver.resolve("term_a", turnA);
+
+        assertSuccessorRegistrationSurvives(resolver, rendezvous, waiterB,
+                "a raw exhausted turn must not evict B from the raw-scrape exhausted branch");
+    }
+
+    @Test
+    void aSupersededRawBackendErrorTurnMustNotEvictItsSuccessorsRegistration() {
+        FakeHerdr herdr = new FakeHerdr().readText("╭────\nAPI Error: 400 invalid request body");
+        Rendezvous rendezvous = new Rendezvous();
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none());
+
+        var waiterA = rendezvous.open("term_a");
+        var turnA = new CompletionResolver.InFlight(waiterA, null);
+        rendezvous.close("term_a", waiterA);
+        var waiterB = rendezvous.open("term_a");
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB));
+
+        resolver.resolve("term_a", turnA);
+
+        assertSuccessorRegistrationSurvives(resolver, rendezvous, waiterB,
+                "a raw backend-error turn must not evict B from the raw-scrape error branch");
+    }
+
+    @Test
+    void aSupersededDoneFailedTurnMustNotEvictItsSuccessorsRegistration() {
+        Rendezvous rendezvous = new Rendezvous();
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(new FakeHerdr()), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none());
+
+        var waiterA = rendezvous.open("term_a");
+        var turnA = new CompletionResolver.InFlight(waiterA, null);
+        assertTrue(rendezvous.resolve("term_a", "A replied"));
+        rendezvous.close("term_a", waiterA);
+        var waiterB = rendezvous.open("term_a");
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB));
+
+        resolver.fail("term_a", turnA);
+
+        assertSuccessorRegistrationSurvives(resolver, rendezvous, waiterB,
+                "a done failed turn must not evict B from fail()'s early return");
+    }
+
+    @Test
+    void aSupersededTooFastBackendErrorTurnMustNotEvictItsSuccessorsRegistration() {
+        FakeHerdr herdr = new FakeHerdr().readText("⏺ API Error: 400 invalid request body\n❯ ");
+        Rendezvous rendezvous = new Rendezvous();
+        long[] clock = {10_000_000_000L};
+        CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
+                ExhaustedPatternLookup.none(), ExhaustionSink.none(), () -> clock[0]);
+
+        var waiterA = rendezvous.open("term_a");
+        var turnA = new CompletionResolver.InFlight(waiterA, null, clock[0]);
+        rendezvous.close("term_a", waiterA);
+        var waiterB = rendezvous.open("term_a");
+        resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB));
+        clock[0] += CompletionResolver.MIN_TURN_NANOS - 1;
+
+        resolver.resolve("term_a", turnA);
+
+        assertSuccessorRegistrationSurvives(resolver, rendezvous, waiterB,
+                "a too-fast backend-error turn must not evict B from failTooFast()");
+    }
+
+    private static void assertSuccessorRegistrationSurvives(CompletionResolver resolver, Rendezvous rendezvous,
+                                                             Object waiterB, String message) {
+        CompletionResolver.InFlight afterA = resolver.inFlight("term_a");
+        assertNotNull(afterA, message + " — a one-arg remove(target) would remove B");
+        assertEquals(waiterB, afterA.waiter(), message + " — the surviving record must belong to B");
+        assertTrue(rendezvous.resolve("term_a", "B replied"), message + " — B must still resolve normally");
+    }
+
    // --- CB-578 stage A: backend-exhausted classification ---------------------------------

    @Test