feat(harness): add run context and retry core

This commit is contained in:
zhuyongxin
2026-07-21 18:36:19 +08:00
parent 4274f3350b
commit 6b74990f86
47 changed files with 1910 additions and 1 deletions
@@ -0,0 +1,33 @@
package com.superbiz.agent.config;
import org.junit.jupiter.api.Test;
import org.yaml.snakeyaml.Yaml;
import java.io.InputStream;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
class SpringAiRetryConfigurationTest {
@Test
void disablesHiddenSpringAiRetries() {
try (InputStream input = getClass().getClassLoader().getResourceAsStream("application.yml")) {
assertNotNull(input);
Map<String, Object> root = new Yaml().load(input);
Map<String, Object> spring = map(root.get("spring"));
Map<String, Object> ai = map(spring.get("ai"));
Map<String, Object> retry = map(ai.get("retry"));
assertEquals(1, ((Number) retry.get("max-attempts")).intValue());
} catch (Exception e) {
throw new AssertionError("Failed to read application.yml", e);
}
}
@SuppressWarnings("unchecked")
private Map<String, Object> map(Object value) {
return (Map<String, Object>) value;
}
}
@@ -0,0 +1,92 @@
package com.superbiz.agent.harness.core;
import com.superbiz.agent.harness.retry.HarnessRetryPolicies;
import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
class DiagnosisHarnessCoreTest {
@Test
void deadlinePreventsFakeToolInvocation() {
MutableClock clock = new MutableClock(Instant.parse("2026-07-21T10:00:00Z"));
DiagnosisHarnessCore core = HarnessCoreFixtures.core(clock);
RunContext context = core.startRun("session-1", "run-1");
AtomicInteger toolCalls = new AtomicInteger();
clock.advance(Duration.ofMinutes(2));
RunAbortedException failure = assertThrows(RunAbortedException.class, () -> {
core.beforeToolCall(context, "query_logs");
toolCalls.incrementAndGet();
});
assertEquals(0, toolCalls.get());
assertEquals(RunState.TIMED_OUT, failure.termination().state());
assertEquals(RunCancellationReason.DEADLINE_EXCEEDED, context.cancellation().reason().orElseThrow());
assertThrows(RunAbortedException.class, () -> core.beforeModelCall(context));
}
@Test
void clientCancellationStopsSubsequentFakeModel() {
DiagnosisHarnessCore core = HarnessCoreFixtures.core(new MutableClock(Instant.parse("2026-07-21T10:00:00Z")));
RunContext context = core.startRun("session-1", "run-1");
AtomicBoolean resourceCancelled = new AtomicBoolean();
context.cancellation().onCancel(reason -> resourceCancelled.set(true));
assertTrue(core.cancel(context, RunCancellationReason.CLIENT_DISCONNECTED));
assertTrue(resourceCancelled.get());
assertThrows(RunAbortedException.class, () -> core.beforeModelCall(context));
assertEquals(RunState.CANCELLED, context.lifecycle().state());
}
@Test
void toolBudgetExhaustionWinsOverLaterFailure() {
MutableClock clock = new MutableClock(Instant.parse("2026-07-21T10:00:00Z"));
DiagnosisHarnessCore core = new DiagnosisHarnessCore(
clock,
() -> "run-1",
Duration.ofMinutes(1),
new RunBudgetLimits(2, 1, 1, 100, 100, 200, 100),
HarnessRetryPolicies.strict());
RunContext context = core.startRun("session-1");
core.beforeToolCall(context, "query_logs");
BudgetExceededException failure = assertThrows(
BudgetExceededException.class,
() -> core.beforeToolCall(context, "query_mysql"));
RunTermination termination = context.lifecycle().termination().orElseThrow();
assertEquals(BudgetKind.TOOL_CALLS, failure.kind());
assertEquals(RunState.BUDGET_EXHAUSTED, termination.state());
assertFalse(core.completeFailure(context, "late failure"));
assertEquals(termination, context.lifecycle().termination().orElseThrow());
}
@Test
void actualTokenExhaustionIsRecordedAndStopsNextOperation() {
MutableClock clock = new MutableClock(Instant.parse("2026-07-21T10:00:00Z"));
DiagnosisHarnessCore core = new DiagnosisHarnessCore(
clock,
() -> "run-1",
Duration.ofMinutes(1),
new RunBudgetLimits(2, 2, 2, 10, 10, 15, 100),
HarnessRetryPolicies.strict());
RunContext context = core.startRun("session-1");
core.beforeModelCall(context);
assertThrows(BudgetExceededException.class, () -> core.recordTokens(context, 9, 8));
assertEquals(17, context.budget().snapshot().totalTokens());
assertEquals(RunState.BUDGET_EXHAUSTED, context.lifecycle().state());
assertThrows(RunAbortedException.class, () -> core.beforeToolCall(context, "query_logs"));
}
}
@@ -0,0 +1,24 @@
package com.superbiz.agent.harness.core;
import com.superbiz.agent.harness.retry.HarnessRetryPolicies;
import java.time.Duration;
public final class HarnessCoreFixtures {
private HarnessCoreFixtures() {
}
public static RunBudgetLimits generousLimits() {
return new RunBudgetLimits(10, 10, 5, 10_000, 10_000, 20_000, 1_000_000);
}
public static DiagnosisHarnessCore core(MutableClock clock) {
return new DiagnosisHarnessCore(
clock,
() -> "run-generated-1",
Duration.ofMinutes(2),
generousLimits(),
HarnessRetryPolicies.strict());
}
}
@@ -0,0 +1,42 @@
package com.superbiz.agent.harness.core;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
public final class MutableClock extends Clock {
private final AtomicReference<Instant> instant;
private final ZoneId zone;
public MutableClock(Instant instant) {
this(instant, ZoneId.of("UTC"));
}
private MutableClock(Instant instant, ZoneId zone) {
this.instant = new AtomicReference<>(Objects.requireNonNull(instant));
this.zone = Objects.requireNonNull(zone);
}
public void advance(Duration duration) {
instant.updateAndGet(current -> current.plus(duration));
}
@Override
public ZoneId getZone() {
return zone;
}
@Override
public Clock withZone(ZoneId zone) {
return new MutableClock(instant(), zone);
}
@Override
public Instant instant() {
return instant.get();
}
}
@@ -0,0 +1,74 @@
package com.superbiz.agent.harness.core;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
class RunBudgetTest {
@Test
void rejectsInvalidLimits() {
assertThrows(IllegalArgumentException.class,
() -> new RunBudgetLimits(0, 1, 1, 1, 1, 1, 1));
}
@Test
void perToolExhaustionDoesNotPartiallyIncrementCounters() {
RunBudget budget = new RunBudget(new RunBudgetLimits(2, 3, 1, 100, 100, 200, 100));
budget.reserveToolCall("query_logs");
BudgetExceededException failure = assertThrows(
BudgetExceededException.class,
() -> budget.reserveToolCall("query_logs"));
RunBudgetUsage usage = budget.snapshot();
assertEquals(BudgetKind.TOOL_CALLS_PER_TOOL, failure.kind());
assertEquals(1, usage.toolCalls());
assertEquals(1, usage.toolCallsByName().get("query_logs"));
}
@Test
void tokenExhaustionKeepsActualUsage() {
RunBudget budget = new RunBudget(new RunBudgetLimits(2, 2, 2, 10, 10, 15, 100));
BudgetExceededException failure = assertThrows(
BudgetExceededException.class,
() -> budget.recordTokens(9, 8));
assertEquals(BudgetKind.TOTAL_TOKENS, failure.kind());
assertEquals(9, budget.snapshot().inputTokens());
assertEquals(8, budget.snapshot().outputTokens());
assertEquals(17, budget.snapshot().totalTokens());
}
@Test
void concurrentCapacityReservationsAreAtomic() throws Exception {
RunCapacityCounter counter = new RunCapacityCounter(100);
ExecutorService executor = Executors.newFixedThreadPool(5);
try {
List<Future<Long>> reservations = new ArrayList<>();
for (int i = 0; i < 10; i++) {
reservations.add(executor.submit(() -> counter.reserve(10)));
}
for (Future<Long> reservation : reservations) {
reservation.get();
}
} finally {
executor.shutdownNow();
}
assertEquals(100, counter.usedBytes());
BudgetExceededException failure = assertThrows(
BudgetExceededException.class,
() -> counter.reserve(1));
assertEquals(BudgetKind.RUN_BYTES, failure.kind());
assertEquals(100, counter.usedBytes());
}
}
@@ -0,0 +1,79 @@
package com.superbiz.agent.harness.core;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
class RunContextTest {
@Test
void rejectsBlankIdentityBeforePublishingContext() {
DiagnosisHarnessCore core = HarnessCoreFixtures.core(new MutableClock(Instant.parse("2026-07-21T10:00:00Z")));
assertThrows(IllegalArgumentException.class, () -> core.startRun(" ", "run-1"));
assertThrows(IllegalArgumentException.class, () -> core.startRun("session-1", " "));
}
@Test
void explicitlyPropagatesSameHandlesAcrossAsyncBoundary() throws Exception {
DiagnosisHarnessCore core = HarnessCoreFixtures.core(new MutableClock(Instant.parse("2026-07-21T10:00:00Z")));
RunContext context = core.startRun("session-1", "run-1");
AtomicReference<Thread> worker = new AtomicReference<>();
RunContext observed = CompletableFuture.supplyAsync(() -> {
worker.set(Thread.currentThread());
return fakeTool(context);
}).get(5, TimeUnit.SECONDS);
assertEquals("session-1", observed.sessionId());
assertEquals("run-1", observed.runId());
assertSame(context.cancellation(), observed.cancellation());
assertSame(context.budget(), observed.budget());
assertSame(context.lifecycle(), observed.lifecycle());
assertFalse(worker.get().equals(Thread.currentThread()));
}
@Test
void cancellationUsesFirstReasonAndNotifiesRegisteredCallbacks() {
DiagnosisHarnessCore core = HarnessCoreFixtures.core(new MutableClock(Instant.parse("2026-07-21T10:00:00Z")));
RunContext context = core.startRun("session-1", "run-1");
AtomicInteger callbacks = new AtomicInteger();
context.cancellation().onCancel(reason -> callbacks.incrementAndGet());
context.cancellation().onCancel(reason -> callbacks.incrementAndGet());
assertTrue(core.cancel(context, RunCancellationReason.CLIENT_DISCONNECTED));
assertFalse(context.cancellation().cancel(RunCancellationReason.USER_REQUESTED));
assertEquals(RunCancellationReason.CLIENT_DISCONNECTED, context.cancellation().reason().orElseThrow());
assertEquals(2, callbacks.get());
assertEquals(RunState.CANCELLED, context.lifecycle().state());
}
@Test
void firstTerminalOutcomeCannotBeOverwritten() {
DiagnosisHarnessCore core = HarnessCoreFixtures.core(new MutableClock(Instant.parse("2026-07-21T10:00:00Z")));
RunContext context = core.startRun("session-1", "run-1");
assertTrue(core.completeSuccess(context));
RunTermination success = context.lifecycle().termination().orElseThrow();
assertFalse(core.completeFailure(context, "late failure"));
assertFalse(core.cancel(context, RunCancellationReason.CLIENT_DISCONNECTED));
assertSame(success, context.lifecycle().termination().orElseThrow());
assertEquals(RunState.SUCCESS, context.lifecycle().state());
}
private RunContext fakeTool(RunContext context) {
return context;
}
}
@@ -0,0 +1,134 @@
package com.superbiz.agent.harness.retry;
import com.superbiz.agent.harness.core.DiagnosisHarnessCore;
import com.superbiz.agent.harness.core.BudgetExceededException;
import com.superbiz.agent.harness.core.BudgetKind;
import com.superbiz.agent.harness.core.HarnessCoreFixtures;
import com.superbiz.agent.harness.core.MutableClock;
import com.superbiz.agent.harness.core.RunAbortedException;
import com.superbiz.agent.harness.core.RunCancellationReason;
import com.superbiz.agent.harness.core.RunContext;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
class HarnessRetryExecutorTest {
private DiagnosisHarnessCore core;
private RunContext context;
private HarnessRetryExecutor executor;
@BeforeEach
void setUp() {
core = HarnessCoreFixtures.core(new MutableClock(Instant.parse("2026-07-21T10:00:00Z")));
context = core.startRun("session-1", "run-1");
executor = new HarnessRetryExecutor(core);
}
@Test
void retriesRouterTransportFailureOnceAndRecordsBothAttempts() {
AtomicInteger calls = new AtomicInteger();
List<RetryAttempt> attempts = new ArrayList<>();
String result = executor.execute(
context,
context.retryPolicies().intentRouter(),
() -> {
if (calls.incrementAndGet() == 1) {
throw new IOException("temporary transport failure");
}
return "DIAGNOSIS";
},
exception -> RetryFailure.TRANSPORT,
attempts::add);
assertEquals("DIAGNOSIS", result);
assertEquals(2, calls.get());
assertEquals(List.of(
RetryAttempt.failed(1, RetryFailure.TRANSPORT),
RetryAttempt.succeeded(2)), attempts);
}
@Test
void toolPolicyRunsOnlyOneAttempt() {
AtomicInteger calls = new AtomicInteger();
List<RetryAttempt> attempts = new ArrayList<>();
RetryExecutionException failure = assertThrows(RetryExecutionException.class, () -> executor.execute(
context,
context.retryPolicies().toolCall(),
() -> {
calls.incrementAndGet();
throw new IOException("tool failed");
},
exception -> RetryFailure.TRANSPORT,
attempts::add));
assertEquals(1, calls.get());
assertEquals(1, failure.attempts());
assertEquals(List.of(RetryAttempt.failed(1, RetryFailure.TRANSPORT)), attempts);
}
@Test
void cancellationBetweenAttemptsPreventsSecondInvocation() {
AtomicInteger calls = new AtomicInteger();
assertThrows(RunAbortedException.class, () -> executor.execute(
context,
context.retryPolicies().semanticGuard(),
() -> {
calls.incrementAndGet();
throw new IOException("temporary transport failure");
},
exception -> RetryFailure.TRANSPORT,
attempt -> core.cancel(context, RunCancellationReason.CLIENT_DISCONNECTED)));
assertEquals(1, calls.get());
}
@Test
void noEvidenceAndBusinessRejectionAreNotRetryable() {
AtomicInteger calls = new AtomicInteger();
RetryExecutionException failure = assertThrows(RetryExecutionException.class, () -> executor.execute(
context,
context.retryPolicies().intentRouter(),
() -> {
calls.incrementAndGet();
throw new IllegalStateException("valid empty result");
},
exception -> RetryFailure.NO_EVIDENCE,
attempt -> { }));
assertEquals(1, calls.get());
assertEquals(RetryFailure.NO_EVIDENCE, failure.failure());
assertEquals(1, context.retryPolicies().diagnosisAgent().maxAttempts());
assertEquals(1, context.retryPolicies().evidenceRepair().maxAttempts());
}
@Test
void budgetFailureCannotBeMisclassifiedAsRetryable() {
AtomicInteger calls = new AtomicInteger();
RetryExecutionException failure = assertThrows(RetryExecutionException.class, () -> executor.execute(
context,
context.retryPolicies().intentRouter(),
() -> {
calls.incrementAndGet();
throw new BudgetExceededException(BudgetKind.MODEL_CALLS, 1, 2);
},
exception -> RetryFailure.TRANSPORT,
attempt -> { }));
assertEquals(1, calls.get());
assertEquals(RetryFailure.BUDGET_EXHAUSTED, failure.failure());
}
}
@@ -0,0 +1,26 @@
package com.superbiz.agent.harness.tool.store;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
class ToolCallKeyFactoryTest {
private final ToolCallKeyFactory factory = new ToolCallKeyFactory("superbiz:harness:tool-call");
@Test
void preservesExactFrameworkToolCallId() {
assertEquals(
"superbiz:harness:tool-call:run-123:call_ABC-123",
factory.create("run-123", "call_ABC-123"));
}
@Test
void rejectsBlankUnsafeAndOversizedSegmentsWithoutFallbackId() {
assertThrows(IllegalArgumentException.class, () -> factory.create("run-1", ""));
assertThrows(IllegalArgumentException.class, () -> factory.create("run:1", "call-1"));
assertThrows(IllegalArgumentException.class, () -> factory.create("run-1", "call/1"));
assertThrows(IllegalArgumentException.class, () -> factory.create("run-1", "a".repeat(129)));
}
}