From 1e329b182f17b42664aed26b459d878391fd30ae Mon Sep 17 00:00:00 2001 From: jasper blues Date: Tue, 11 Aug 2026 10:11:25 +1000 Subject: [PATCH 1/6] Restore the AgentProcess a thread already held, rather than clearing it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ExecutorAsyncer set the AgentProcess thread local on the worker and cleared it in a finally. Clearing is right for a pooled worker that arrived empty, and wrong the moment a task runs on a thread that is already inside a process. An Executor is free to do exactly that. A direct executor always does, and a ThreadPoolExecutor with CallerRunsPolicy does once its queue fills — so this appears under load and not in development. The submitting thread comes back from async() holding no process. Nothing throws at that point. The next AgentProcess.get() returns null, usually a blackboard read some distance away, so it presents as "the blackboard lost my object" with the cause several frames back. That is an expensive afternoon. AgentProcessAccessor gains with(), which saves and restores. reset() stays for the callers that want the old semantics. Tests: the direct-executor shape (plain, throwing, nested, repeated), the CallerRunsPolicy pool that reaches it under queue saturation, and — unchanged and still asserted — the pooled-thread isolation that clearing was protecting, so the fix cannot be "stop cleaning up". Five of the ten fail before this change and pass after. Co-Authored-By: Claude Opus 5 (1M context) --- .../agent/spi/support/AgentProcessAccessor.kt | 27 ++ .../agent/spi/support/ExecutorAsyncer.kt | 12 +- .../ExecutorAsyncerCallerThreadTest.kt | 238 ++++++++++++++++++ 3 files changed, 269 insertions(+), 8 deletions(-) create mode 100644 embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/AgentProcessAccessor.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/AgentProcessAccessor.kt index e1ff0360e..e31d3dc8c 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/AgentProcessAccessor.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/AgentProcessAccessor.kt @@ -43,4 +43,31 @@ object AgentProcessAccessor { fun reset() { AgentProcess.remove() } + + /** + * Run [block] with [value] as the current process, restoring whatever the thread held before - + * which is not always nothing. + * + * [reset] clears the slot outright. That is right for a pooled worker that arrived empty, and + * wrong the moment a task runs on a thread that is already inside a process. An [java.util.concurrent.Executor] + * is free to do exactly that: a direct executor always does, and a [java.util.concurrent.ThreadPoolExecutor] + * with [java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy] does once its queue fills - so + * the behaviour appears under load and not in development. + * + * The submitting thread then comes back from the task holding no process. Nothing throws there. + * The next [AgentProcess.get] returns null, typically a blackboard read some distance away, so + * the symptom is "the blackboard lost my object" and the cause is several frames back. + */ + fun with(value: AgentProcess?, block: () -> T): T { + if (value == null) { + return block() + } + val previous = getValue() + setValue(value) + return try { + block() + } finally { + if (previous != null) setValue(previous) else reset() + } + } } diff --git a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/ExecutorAsyncer.kt b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/ExecutorAsyncer.kt index 7ce67b1ed..14717d4a5 100644 --- a/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/ExecutorAsyncer.kt +++ b/embabel-agent-api/src/main/kotlin/com/embabel/agent/spi/support/ExecutorAsyncer.kt @@ -42,14 +42,10 @@ class ExecutorAsyncer( return CompletableFuture.supplyAsync({ contextSnapshot.setThreadLocals().use { - if (agentProcess != null) { - AgentProcessAccessor.setValue(agentProcess) - try { - block() - } finally { - AgentProcessAccessor.reset() // cleanup - } - } else { + // Restores rather than clears: nothing leaks between tasks on a pooled thread, and + // nothing is wiped when the executor runs the task on the submitting thread, which + // is already inside a process. + AgentProcessAccessor.with(agentProcess) { block() } } diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt new file mode 100644 index 000000000..fde4343cd --- /dev/null +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt @@ -0,0 +1,238 @@ +/* + * Copyright 2024-2026 Embabel Pty Ltd. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.embabel.agent.spi.support + +import com.embabel.agent.core.AgentProcess +import com.embabel.agent.core.AgentProcess.Companion.withCurrent +import io.mockk.mockk +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertSame +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Nested +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.Timeout +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executor +import java.util.concurrent.ExecutionException +import java.util.concurrent.Executors +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.ThreadPoolExecutor +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicReference + +/** + * An [Executor] may run a task on the thread that submitted it. The [AgentProcess] thread local + * therefore has to be RESTORED on the way out, not cleared: the submitting thread is already + * inside a process, and clearing leaves it holding nothing for the rest of its work. + * + * Nothing throws when that happens. The next [AgentProcess.get] returns null - typically a + * blackboard read, arbitrarily far from the async() that emptied the slot - so the symptom is + * "the blackboard lost my object" and the cause is several frames away. + */ +@Timeout(30) +class ExecutorAsyncerCallerThreadTest { + + /** + * The shape in miniature: the task runs on the caller's thread. + */ + @Nested + inner class DirectExecutor { + + private val asyncer = ExecutorAsyncer { it.run() } + + @Test + fun `the caller still holds its process after the task returns`() { + val outer = mockk() + outer.withCurrent { + asyncer.async { "work" }.get(5, TimeUnit.SECONDS) + assertSame(outer, AgentProcess.get()) + } + } + + @Test + fun `the task itself sees the process`() { + val outer = mockk() + val seen = outer.withCurrent { + asyncer.async { AgentProcess.get() }.get(5, TimeUnit.SECONDS) + } + assertSame(outer, seen) + } + + @Test + fun `a caller holding no process is left holding none`() { + asyncer.async { "work" }.get(5, TimeUnit.SECONDS) + assertNull(AgentProcess.get()) + } + + @Test + fun `the process is restored even when the task throws`() { + val outer = mockk() + outer.withCurrent { + assertThrows(ExecutionException::class.java) { + asyncer.async { error("boom") }.get(5, TimeUnit.SECONDS) + } + assertSame(outer, AgentProcess.get()) + } + } + + @Test + fun `nesting restores each level, not just the innermost`() { + val outer = mockk() + val inner = mockk() + + outer.withCurrent { + asyncer.async { + assertSame(outer, AgentProcess.get()) + inner.withCurrent { + asyncer.async { AgentProcess.get() }.get(5, TimeUnit.SECONDS) + }.also { assertSame(inner, it) } + // The inner async must not have taken the outer process with it + assertSame(outer, AgentProcess.get()) + }.get(5, TimeUnit.SECONDS) + assertSame(outer, AgentProcess.get()) + } + } + + @Test + fun `repeated asyncs do not erode the caller's process`() { + val outer = mockk() + outer.withCurrent { + repeat(50) { asyncer.async { "work" }.get(5, TimeUnit.SECONDS) } + assertSame(outer, AgentProcess.get()) + } + } + } + + /** + * How this is actually reached in production. A saturated [ThreadPoolExecutor] configured with + * [ThreadPoolExecutor.CallerRunsPolicy] runs the overflow task on the submitting thread - so + * the behaviour appears under load and not before. + */ + @Nested + inner class CallerRunsUnderLoad { + + @Test + fun `an overflowing pool does not empty the submitting thread`() { + // One worker, queue of one: the third task overflows and runs on the caller. + val pool = ThreadPoolExecutor( + 1, 1, 0L, TimeUnit.MILLISECONDS, + LinkedBlockingQueue(1), + ThreadPoolExecutor.CallerRunsPolicy(), + ) + val asyncer = ExecutorAsyncer(pool) + val release = CountDownLatch(1) + val ranOnCaller = AtomicReference(false) + val outer = mockk() + + try { + outer.withCurrent { + val callerThread = Thread.currentThread() + // Occupy the worker, then fill the queue. + val blocking = asyncer.async { release.await(10, TimeUnit.SECONDS) } + val queued = asyncer.async { "queued" } + // This one has nowhere to go, so the caller runs it. + val overflow = asyncer.async { + ranOnCaller.set(Thread.currentThread() === callerThread) + AgentProcess.get() + } + + assertSame(outer, overflow.get(10, TimeUnit.SECONDS), "the task must see the process") + assertTrue(ranOnCaller.get(), "precondition: the overflow task must have run on the caller") + assertSame(outer, AgentProcess.get(), "the caller must still hold its process") + + release.countDown() + blocking.get(10, TimeUnit.SECONDS) + queued.get(10, TimeUnit.SECONDS) + } + } finally { + release.countDown() + pool.shutdown() + pool.awaitTermination(10, TimeUnit.SECONDS) + } + } + } + + /** + * The behaviour clearing was there to protect: a pooled thread must not carry one task's + * process into the next task to land on it. + */ + @Nested + inner class PooledThreadIsolation { + + @Test + fun `a task leaves no process behind for the next task on the same thread`() { + val single = Executors.newSingleThreadExecutor() + try { + val asyncer = ExecutorAsyncer(single) + val outer = mockk() + + outer.withCurrent { + asyncer.async { AgentProcess.get() }.get(5, TimeUnit.SECONDS) + } + + // Same pooled thread, no process on the caller this time. + val seen = asyncer.async { AgentProcess.get() }.get(5, TimeUnit.SECONDS) + assertNull(seen, "the pooled thread carried a process into an unrelated task") + } finally { + single.shutdown() + single.awaitTermination(5, TimeUnit.SECONDS) + } + } + + @Test + fun `two processes do not bleed into each other across reuse`() { + val single = Executors.newSingleThreadExecutor() + try { + val asyncer = ExecutorAsyncer(single) + val first = mockk() + val second = mockk() + + val a = first.withCurrent { asyncer.async { AgentProcess.get() }.get(5, TimeUnit.SECONDS) } + val b = second.withCurrent { asyncer.async { AgentProcess.get() }.get(5, TimeUnit.SECONDS) } + + assertSame(first, a) + assertSame(second, b) + } finally { + single.shutdown() + single.awaitTermination(5, TimeUnit.SECONDS) + } + } + + @Test + fun `parallelMap gives every worker the caller's process and leaves none behind`() { + val pool = Executors.newCachedThreadPool() + try { + val asyncer = ExecutorAsyncer(pool) + val outer = mockk() + + val seen = outer.withCurrent { + asyncer.parallelMap((1..8).toList(), 4) { AgentProcess.get() } + } + assertEquals(8, seen.size) + seen.forEach { assertSame(outer, it) } + + // Threads are recycled; an unrelated task must not inherit + val after = asyncer.parallelMap((1..8).toList(), 4) { AgentProcess.get() } + after.forEach { assertNull(it) } + } finally { + pool.shutdown() + pool.awaitTermination(10, TimeUnit.SECONDS) + } + } + } +} From caf1e2fa42cd48a932ad6f7280f9ecbfdac8294b Mon Sep 17 00:00:00 2001 From: jasper blues Date: Wed, 12 Aug 2026 07:32:02 +1000 Subject: [PATCH 2/6] Mark the boundary: neither executor Embabel owns can run on the caller MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review question worth answering in the suite rather than in a thread: does this affect virtual threads, given the default is a pooled executor? It does not, and neither does the platform default. `threading.shared` is false by default, so AsyncConfiguration builds Embabel's own executor — a cached platform pool, or a thread-per-task executor on virtual threads. A cached pool always hands off to a worker and a thread-per-task executor always starts a new thread, so in neither case is the submitting thread the one that runs the task, and the clearing this PR replaces was harmless there. Two tests, one per executor, each asserting the task really did run on another thread before asserting the caller kept its process. Both pass BEFORE this PR's fix as well as after — which is the claim being pinned. Reaching the defect needs threading.shared=true AND an application executor that can run a task on the submitting thread: CallerRunsPolicy with a bounded queue, or a direct executor. Co-Authored-By: Claude Opus 5 (1M context) --- .../ExecutorAsyncerCallerThreadTest.kt | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt index fde4343cd..b692b16a8 100644 --- a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt @@ -19,6 +19,7 @@ import com.embabel.agent.core.AgentProcess import com.embabel.agent.core.AgentProcess.Companion.withCurrent import io.mockk.mockk import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotSame import org.junit.jupiter.api.Assertions.assertNull import org.junit.jupiter.api.Assertions.assertSame import org.junit.jupiter.api.Assertions.assertThrows @@ -167,6 +168,57 @@ class ExecutorAsyncerCallerThreadTest { } } + /** + * The two executors [com.embabel.agent.spi.config.spring.AsyncConfiguration] builds when + * Embabel owns its executor - which is the default, since `threading.shared` is false. + * + * Neither can run a task on the submitting thread: a cached pool always hands off to a worker, + * and a thread-per-task executor always starts a new virtual thread. So neither is affected by + * the defect this class is about, before or after the fix. They are here to mark that boundary, + * and to answer "does this happen on virtual threads" with a test rather than an argument. + */ + @Nested + inner class OwnedExecutorsNeverRunOnTheCaller { + + @Test + fun `virtual thread per task - the task runs elsewhere and the caller keeps its process`() { + val pool = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory()) + try { + assertRunsElsewhereAndPreservesCaller(ExecutorAsyncer(pool)) + } finally { + pool.shutdown() + pool.awaitTermination(10, TimeUnit.SECONDS) + } + } + + @Test + fun `cached platform pool - the task runs elsewhere and the caller keeps its process`() { + val pool = Executors.newCachedThreadPool(Thread.ofPlatform().factory()) + try { + assertRunsElsewhereAndPreservesCaller(ExecutorAsyncer(pool)) + } finally { + pool.shutdown() + pool.awaitTermination(10, TimeUnit.SECONDS) + } + } + + private fun assertRunsElsewhereAndPreservesCaller(asyncer: ExecutorAsyncer) { + val outer = mockk() + outer.withCurrent { + val callerThread = Thread.currentThread() + val ranOn = AtomicReference() + val seen = asyncer.async { + ranOn.set(Thread.currentThread()) + AgentProcess.get() + }.get(10, TimeUnit.SECONDS) + + assertNotSame(callerThread, ranOn.get(), "this executor is not supposed to run on the caller") + assertSame(outer, seen, "the worker must still see the caller's process") + assertSame(outer, AgentProcess.get(), "and the caller must still hold it") + } + } + } + /** * The behaviour clearing was there to protect: a pooled thread must not carry one task's * process into the next task to land on it. From 203753f1fcf3083a38ec29fddc029e047a307af8 Mon Sep 17 00:00:00 2001 From: jasper blues Date: Wed, 12 Aug 2026 17:54:37 +1000 Subject: [PATCH 3/6] Cover the caller-runs case as a deployment actually reaches it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ExecutorAsyncerCallerThreadTest proves the propagation behaviour on an executor it constructs itself. That leaves one thing unproven: that a deployment can arrive at that executor through configuration at all. Reaching it needs threading.shared=true AND an application executor that can run a task on the submitting thread — two settings in different places, neither of which mentions the other. SharedExecutorCallerRunsWiringIT boots the real AsyncConfiguration against a bounded CallerRunsPolicy pool and saturates it, so the wiring and the load shape are covered together. It asserts the precondition — the overflow task genuinely ran on the submitting thread — before asserting the outcome, so it cannot pass vacuously. Against the unfixed code it fails with "expected: but was: ", after that precondition has passed. A second case pins the other side: same application executor, sharing left at its default of false, and the task runs on a worker instead. That is what makes the first case's title true rather than incidental. Named as an IT and run with the suite from #1577 (mvn -Dtest='*IT,!LLMOllama*IT' -Dsurefire.failIfNoSpecifiedTests=false test). It needs no LLM keys, so it is safe in any environment. Co-Authored-By: Claude Opus 5 (1M context) --- .../SharedExecutorCallerRunsWiringIT.kt | 145 ++++++++++++++++++ 1 file changed, 145 insertions(+) create mode 100644 embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/config/spring/SharedExecutorCallerRunsWiringIT.kt diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/config/spring/SharedExecutorCallerRunsWiringIT.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/config/spring/SharedExecutorCallerRunsWiringIT.kt new file mode 100644 index 000000000..7092787fa --- /dev/null +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/config/spring/SharedExecutorCallerRunsWiringIT.kt @@ -0,0 +1,145 @@ +/* + * Copyright 2024-2026 Embabel Pty Ltd. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.embabel.agent.spi.config.spring + +import com.embabel.agent.api.common.Asyncer +import com.embabel.agent.core.AgentProcess +import com.embabel.agent.core.AgentProcess.Companion.withCurrent +import io.mockk.mockk +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.Assertions.assertSame +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.Timeout +import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration +import org.springframework.boot.context.properties.EnableConfigurationProperties +import org.springframework.boot.test.context.runner.ApplicationContextRunner +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executor +import java.util.concurrent.LinkedBlockingQueue +import java.util.concurrent.ThreadPoolExecutor +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicReference + +/** + * The caller-thread case wired the way a deployment reaches it, rather than by handing an + * [Executor] straight to the asyncer. + * + * [com.embabel.agent.spi.support.ExecutorAsyncerCallerThreadTest] proves the propagation + * behaviour on an executor it constructs itself. That leaves one thing unproven: that a + * deployment can actually ARRIVE at that executor through configuration. Reaching it needs + * `embabel.agent.platform.threading.shared=true` AND an application executor that can run a + * task on the submitting thread - two settings in different places, neither of which mentions + * the other. This boots the real [AsyncConfiguration] against a real + * [ThreadPoolExecutor.CallerRunsPolicy] pool and saturates it, so the wiring and the load + * shape are covered together. + * + * Runs with the IT suite, which surefire excludes from the normal build: + * `mvn -Dtest='*IT,!LLMOllama*IT' -Dsurefire.failIfNoSpecifiedTests=false test`. Needs no LLM + * keys, so it is safe in any environment. + */ +@Timeout(60) +class SharedExecutorCallerRunsWiringIT { + + /** One worker, queue of one: the third task submitted has nowhere to go but the caller. */ + @Configuration + @EnableConfigurationProperties(AgentPlatformProperties::class) + class SaturatedAppExecutorConfiguration { + + @Bean(TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME, destroyMethod = "shutdownNow") + fun applicationTaskExecutor(): ThreadPoolExecutor = + ThreadPoolExecutor( + 1, 1, 0L, TimeUnit.MILLISECONDS, + LinkedBlockingQueue(1), + ThreadPoolExecutor.CallerRunsPolicy(), + ) + } + + private val contextRunner = ApplicationContextRunner() + .withUserConfiguration(SaturatedAppExecutorConfiguration::class.java, AsyncConfiguration::class.java) + + @AfterEach + fun cleanup() { + AgentProcess.remove() + } + + @Test + fun `a saturated shared pool runs the overflow on the caller and leaves its process intact`() { + contextRunner + .withPropertyValues("embabel.agent.platform.threading.shared=true") + .run { context -> + val asyncer = context.getBean(Asyncer::class.java) + val release = CountDownLatch(1) + val ranOnCaller = AtomicReference(false) + val outer = mockk() + + try { + outer.withCurrent { + val callerThread = Thread.currentThread() + // Occupy the single worker, then fill the single queue slot. + val blocking = asyncer.async { release.await(30, TimeUnit.SECONDS) } + val queued = asyncer.async { "queued" } + // Nowhere left to put this one, so the caller runs it itself. + val overflow = asyncer.async { + ranOnCaller.set(Thread.currentThread() === callerThread) + AgentProcess.get() + } + + assertSame(outer, overflow.get(30, TimeUnit.SECONDS), "the task must see the process") + assertTrue( + ranOnCaller.get(), + "precondition: the shared pool must have overflowed onto the submitting thread", + ) + assertSame( + outer, AgentProcess.get(), + "the caller must still hold its process after the overflow task returns", + ) + + release.countDown() + blocking.get(30, TimeUnit.SECONDS) + queued.get(30, TimeUnit.SECONDS) + } + } finally { + release.countDown() + } + } + } + + @Test + fun `sharing is what exposes the caller thread - the default isolates the app's pool`() { + // Same application executor, sharing left at its default of false. Embabel builds its own + // cached pool instead, which always hands off, so the overflow shape is unreachable. + contextRunner.run { context -> + val asyncer = context.getBean(Asyncer::class.java) + val ranOnCaller = AtomicReference(true) + val outer = mockk() + + outer.withCurrent { + val callerThread = Thread.currentThread() + val seen = asyncer.async { + ranOnCaller.set(Thread.currentThread() === callerThread) + AgentProcess.get() + }.get(30, TimeUnit.SECONDS) + + assertSame(outer, seen) + assertTrue(!ranOnCaller.get(), "an isolated Embabel executor must not run on the caller") + assertSame(outer, AgentProcess.get()) + } + } + } +} From 39ea5650401c7ecb4970ec17dd9a4bace9e879d9 Mon Sep 17 00:00:00 2001 From: jasper blues Date: Thu, 13 Aug 2026 06:21:33 +1000 Subject: [PATCH 4/6] State the restore contract on `with` itself, and say when sharing costs you Review feedback from @igordayen. `AgentProcessAccessorTest` covered getValue/setValue/reset and nothing at all on `with`, so the guarantee this PR exists for was pinned only through the executor - where it fails against the behaviour of a saturated thread pool several layers away rather than against its own rules. Six tests now state it directly: restore after return, restore after a throw, restore at every level on the way out of a nested `with`, and clear when the thread arrived empty. The null case is the one worth having written down. `with(null)` does not clear: a null value means the caller had nothing to propagate, which is what ExecutorAsyncer passes for a task submitted from outside any process, so the block runs against whatever the thread already holds. Intended, easy to read as an oversight, now pinned in both directions. Mutation-checked rather than assumed. Reverting the finally to a bare reset() fails four of them; making with(null) clear fails the two null-value tests. The docs note is Igor's text, next to the threading.shared table. The point it makes is not the AgentProcess bug - that is fixed here - but that sharing couples agent execution to the application executor's backpressure, which survives the fix. Co-Authored-By: Claude Opus 5 (1M context) --- .../AgentProcessContextPropagationTest.kt | 101 ++++++++++++++++++ .../asciidoc/reference/asynch-mode/page.adoc | 2 + 2 files changed, 103 insertions(+) diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/AgentProcessContextPropagationTest.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/AgentProcessContextPropagationTest.kt index fcc862e8e..b4d53f811 100644 --- a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/AgentProcessContextPropagationTest.kt +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/AgentProcessContextPropagationTest.kt @@ -17,13 +17,16 @@ package com.embabel.agent.spi.support import com.embabel.agent.core.AgentProcess import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.Nested import org.junit.jupiter.api.Test import org.mockito.Mockito.mock import java.util.concurrent.Executors import java.util.concurrent.TimeUnit import kotlin.test.assertEquals +import kotlin.test.assertFailsWith import kotlin.test.assertNotNull import kotlin.test.assertNull +import kotlin.test.assertSame class AgentProcessAccessorTest { @@ -64,6 +67,104 @@ class AgentProcessAccessorTest { assertNull(AgentProcess.get()) } + + /** + * [AgentProcessAccessor.with] stated on its own terms. + * + * [ExecutorAsyncerCallerThreadTest] covers the same guarantee through the executor, which is + * where it is reached and why it matters. These pin the contract itself, so a change to `with` + * fails against its own rules rather than against the behaviour of a saturated thread pool + * several layers away. + */ + @Nested + inner class With { + + @AfterEach + fun cleanup() { + AgentProcess.remove() + } + + @Test + fun `restores the previous process after the block returns`() { + val previous = mock(AgentProcess::class.java) + val inner = mock(AgentProcess::class.java) + accessor.setValue(previous) + + val seen = accessor.with(inner) { AgentProcess.get() } + + assertSame(inner, seen, "the block must run with the value it was given") + assertSame(previous, AgentProcess.get(), "the thread must be left holding what it held before") + } + + @Test + fun `restores the previous process when the block throws`() { + val previous = mock(AgentProcess::class.java) + val inner = mock(AgentProcess::class.java) + accessor.setValue(previous) + + assertFailsWith { + accessor.with(inner) { error("boom") } + } + + assertSame(previous, AgentProcess.get()) + } + + @Test + fun `restores the outer value after a nested with, not just the innermost`() { + val outer = mock(AgentProcess::class.java) + val middle = mock(AgentProcess::class.java) + val inner = mock(AgentProcess::class.java) + + accessor.with(outer) { + accessor.with(middle) { + accessor.with(inner) { + assertSame(inner, AgentProcess.get()) + } + assertSame(middle, AgentProcess.get(), "unwinding one level must land on the middle value") + } + assertSame(outer, AgentProcess.get(), "unwinding again must land on the outer value") + } + assertNull(AgentProcess.get(), "and the outermost frame started from nothing") + } + + @Test + fun `clears the slot when the thread held nothing to begin with`() { + val inner = mock(AgentProcess::class.java) + assertNull(AgentProcess.get(), "precondition: this thread starts empty") + + accessor.with(inner) { } + + // The case clearing was written for: a pooled worker that arrived empty must not carry + // this task's process into whatever lands on it next. + assertNull(AgentProcess.get()) + } + + /** + * A null value is not "clear the process for the duration". It means the caller had nothing + * to propagate, which is what [ExecutorAsyncer] passes when a task is submitted from + * outside any process - so the block runs against whatever the running thread already + * holds, and `with` neither sets nor clears anything. + */ + @Test + fun `a null value leaves a process the thread already holds in place`() { + val current = mock(AgentProcess::class.java) + accessor.setValue(current) + + val seen = accessor.with(null) { AgentProcess.get() } + + assertSame(current, seen, "with(null) does not clear for the duration of the block") + assertSame(current, AgentProcess.get(), "nor afterwards") + } + + @Test + fun `a null value leaves an empty thread empty, and still runs the block`() { + val seen = accessor.with(null) { AgentProcess.get() to "result" } + + assertNull(seen.first) + assertEquals("result", seen.second) + assertNull(AgentProcess.get()) + } + } } class ExecutorAsyncerContextPropagationTest { diff --git a/embabel-agent-docs/src/main/asciidoc/reference/asynch-mode/page.adoc b/embabel-agent-docs/src/main/asciidoc/reference/asynch-mode/page.adoc index 05cbaf19f..600176ea6 100644 --- a/embabel-agent-docs/src/main/asciidoc/reference/asynch-mode/page.adoc +++ b/embabel-agent-docs/src/main/asciidoc/reference/asynch-mode/page.adoc @@ -89,6 +89,8 @@ embabel.agent.platform.threading.shared=false **Note**: Virtual threads require Java 21+. On Java 17-20, Embabel automatically falls back to platform threads even if virtual threads are requested. +NOTE: `embabel.agent.platform.threading.shared=true` means Embabel may use the application's `applicationTaskExecutor` when the application and Embabel use the same threading model. If that executor is bounded and uses `CallerRunsPolicy`, saturated executor workers and a full queue can cause Embabel async work to run on the submitting application thread. Embabel preserves `AgentProcess` context in this case, but this configuration still couples agent execution to application executor backpressure and may affect request latency. Use executor sharing deliberately. + Framework employs _Asyncer_ abstraction consistently across Agent Invocation (see <> ), Agent Actions, Tool Loop, ShellCommands, and other scenarios. === Java 25 Implications From a205e508f5502adbcb2f2185978049c02e375c50 Mon Sep 17 00:00:00 2001 From: jasper blues Date: Fri, 14 Aug 2026 01:35:18 +1000 Subject: [PATCH 5/6] Cover the erosion shape, and the symptom in the words it gets reported in Four cases, all of which fail on the unfixed code, taking that count from five of twelve to nine of sixteen. parallelMap is how an action fans out, and it is where this defect looks least like a bug. Every item goes through async(), so on an executor that can run on the submitter the FIRST item empties the thread and later ones run with no process - one call, partly correct results, nothing thrown. Unfixed, the direct-executor case fails at item 2 and the saturated CallerRunsPolicy case at item 4. A single-async test cannot show that shape, and it is the shape a real deployment meets. Both parallelMap branches are covered because maxConcurrency < size takes the semaphore path, which is separate code. The blackboard case states the symptom rather than the mechanism. Nothing throws when the process is lost; the next read is typically a blackboard access some frames away, so what gets reported is "the blackboard lost my object". Unfixed, that read returns null. The suite now says that in the terms someone would use to report it. One of these asserted outside the withCurrent block on first writing and failed against the FIXED code, which is worth recording: a probe confirmed the process survives all three paths, so the test was wrong rather than the code. Assertions now sit inside the scope they are about. Co-Authored-By: Claude Opus 5 (1M context) --- .../ExecutorAsyncerCallerThreadTest.kt | 105 ++++++++++++++++++ 1 file changed, 105 insertions(+) diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt index b692b16a8..72bb341da 100644 --- a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt @@ -35,6 +35,9 @@ import java.util.concurrent.LinkedBlockingQueue import java.util.concurrent.ThreadPoolExecutor import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicReference +import com.embabel.agent.core.Blackboard +import io.mockk.every +import java.util.Collections /** * An [Executor] may run a task on the thread that submitted it. The [AgentProcess] thread local @@ -109,6 +112,65 @@ class ExecutorAsyncerCallerThreadTest { } } + /** + * The symptom the defect actually presents as, rather than the thread local behind it. + * + * Nothing throws when the process is lost. The next read through [AgentProcess.get] is + * typically a blackboard access some distance away, so what a user sees is "the blackboard + * lost my object" with the cause several frames back. Asserted here so the failure this PR + * removes is described in the suite in the terms someone would actually report it. + */ + @Test + fun `the blackboard is still reachable after an async on the caller's thread`() { + val blackboard = mockk(relaxed = true) + val outer = mockk() + every { outer.blackboard } returns blackboard + + outer.withCurrent { + asyncer.async { "work" }.get(5, TimeUnit.SECONDS) + + // What an interceptor further along the same request does. + val reachable = AgentProcess.get()?.blackboard + assertSame(blackboard, reachable, "the blackboard read that would have returned null") + } + } + + /** + * parallelMap on a caller-running executor, which erodes rather than fails. + * + * Every item goes through async(), so on this executor every item runs on the caller. With + * a clear on the way out, the FIRST item empties the thread and items two onward see + * nothing - one call, partially correct results, and no error anywhere. A single-async test + * cannot show that shape. + */ + @Test + fun `every item of a parallelMap sees the process, not just the first`() { + val outer = mockk() + + outer.withCurrent { + val seen = asyncer.parallelMap((1..5).toList(), maxConcurrency = 5) { AgentProcess.get() } + + assertEquals(5, seen.size) + seen.forEachIndexed { i, p -> assertSame(outer, p, "item ${i + 1} ran without the process") } + assertSame(outer, AgentProcess.get(), "and the caller kept it afterwards") + } + } + + @Test + fun `a concurrency-limited parallelMap does not erode the process either`() { + // The semaphore branch: maxConcurrency < size takes a different path through + // ExecutorAsyncer, and on this executor still runs every item on the caller. + val outer = mockk() + + outer.withCurrent { + val seen = asyncer.parallelMap((1..5).toList(), maxConcurrency = 2) { AgentProcess.get() } + + assertEquals(5, seen.size) + seen.forEachIndexed { i, p -> assertSame(outer, p, "item ${i + 1} ran without the process") } + assertSame(outer, AgentProcess.get(), "and the caller kept it afterwards") + } + } + @Test fun `repeated asyncs do not erode the caller's process`() { val outer = mockk() @@ -166,6 +228,49 @@ class ExecutorAsyncerCallerThreadTest { pool.awaitTermination(10, TimeUnit.SECONDS) } } + + /** + * The same executor a deployment actually configures, driving parallelMap rather than a + * single async. + * + * `parallelMap` is how an action fans out, so this is the realistic way to meet the defect: + * more items than the pool can take, the overflow running on the submitting thread, and - + * with a clear on the way out - the caller losing its process partway through its own + * fan-out while some items still succeed. + */ + @Test + fun `parallelMap over a saturated pool keeps the process for the overflow and the caller`() { + val pool = ThreadPoolExecutor( + 1, 1, 0L, TimeUnit.MILLISECONDS, + LinkedBlockingQueue(1), + ThreadPoolExecutor.CallerRunsPolicy(), + ) + val asyncer = ExecutorAsyncer(pool) + val outer = mockk() + + try { + outer.withCurrent { + val callerThread = Thread.currentThread() + val ranOn = Collections.synchronizedList(mutableListOf()) + + val seen = asyncer.parallelMap((1..12).toList(), maxConcurrency = 12) { + ranOn += Thread.currentThread() + AgentProcess.get() + } + + assertTrue( + ranOn.any { it === callerThread }, + "precondition: the pool must have overflowed onto the caller", + ) + assertEquals(12, seen.size) + seen.forEachIndexed { i, p -> assertSame(outer, p, "item ${i + 1} ran without the process") } + assertSame(outer, AgentProcess.get(), "the caller must still hold its process") + } + } finally { + pool.shutdown() + pool.awaitTermination(10, TimeUnit.SECONDS) + } + } } /** From d31c62880a5cc97f957b70c8b4c6a94b5d28019e Mon Sep 17 00:00:00 2001 From: jasper blues Date: Fri, 14 Aug 2026 01:48:06 +1000 Subject: [PATCH 6/6] Cover the fan-out through real wiring, a failing item, and a leaking task Three more cases that fail on the unfixed code, taking the executor-level count to 13 of 21. The IT gained the primitive actions actually fan out with. It drove a single async through the real AsyncConfiguration, which proves a deployment can reach the caller thread but not what happens when it reaches it via parallelMap - the path OperationContext.parallelMap and parallel actions take. Sixteen items over a one-worker pool, so some are certain to run on the submitter. A failing item in a fan-out is the ordinary case, not an exotic one: one tool call of several throwing does not abandon the request. Unfixed, the caller comes out of that holding no process, so the error handling that runs next is the code that discovers the blackboard is empty. Failure and loss of context arriving together is the worst version of this, because the exception looks like the whole story. The leak case pins the direction of the restore. Putting back what the CALLER had is not the same as leaving whatever the task last set, and a test asserting only "not null" would pass on an implementation that handed the task's process upward - work attributed to, and billed against, the wrong process. Co-Authored-By: Claude Opus 5 (1M context) --- .../SharedExecutorCallerRunsWiringIT.kt | 46 +++++++++++++++++++ .../ExecutorAsyncerCallerThreadTest.kt | 43 +++++++++++++++++ 2 files changed, 89 insertions(+) diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/config/spring/SharedExecutorCallerRunsWiringIT.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/config/spring/SharedExecutorCallerRunsWiringIT.kt index 7092787fa..d52481123 100644 --- a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/config/spring/SharedExecutorCallerRunsWiringIT.kt +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/config/spring/SharedExecutorCallerRunsWiringIT.kt @@ -20,6 +20,7 @@ import com.embabel.agent.core.AgentProcess import com.embabel.agent.core.AgentProcess.Companion.withCurrent import io.mockk.mockk import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertSame import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test @@ -29,6 +30,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.boot.test.context.runner.ApplicationContextRunner import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration +import java.util.Collections import java.util.concurrent.CountDownLatch import java.util.concurrent.Executor import java.util.concurrent.LinkedBlockingQueue @@ -120,6 +122,50 @@ class SharedExecutorCallerRunsWiringIT { } } + /** + * The same wiring, driving the primitive an action actually fans out with. + * + * `parallelMap` is how `OperationContext.parallelMap` and parallel actions reach the executor, + * and it fails differently from a single `async`: every item goes through the same path, so a + * clear on the way out empties the submitting thread partway through its OWN fan-out. Later + * items then run with no process while earlier ones succeeded - a partly correct result from + * one call, with nothing thrown. + * + * More items than the pool can hold, so some are certain to run on the caller. + */ + @Test + fun `a parallelMap over a saturated shared pool keeps the process for every item`() { + contextRunner + .withPropertyValues("embabel.agent.platform.threading.shared=true") + .run { context -> + val asyncer = context.getBean(Asyncer::class.java) + val outer = mockk() + + outer.withCurrent { + val callerThread = Thread.currentThread() + val ranOn = Collections.synchronizedList(mutableListOf()) + + val seen = asyncer.parallelMap((1..16).toList(), maxConcurrency = 16) { + ranOn += Thread.currentThread() + AgentProcess.get() + } + + assertTrue( + ranOn.any { it === callerThread }, + "precondition: the shared pool must have overflowed onto the submitting thread", + ) + assertEquals(16, seen.size) + seen.forEachIndexed { i, p -> + assertSame(outer, p, "item ${i + 1} of the fan-out ran without the process") + } + assertSame( + outer, AgentProcess.get(), + "the caller must still hold its process after its own fan-out", + ) + } + } + } + @Test fun `sharing is what exposes the caller thread - the default isolates the app's pool`() { // Same application executor, sharing left at its default of false. Embabel builds its own diff --git a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt index 72bb341da..8f1f4bf12 100644 --- a/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt +++ b/embabel-agent-api/src/test/kotlin/com/embabel/agent/spi/support/ExecutorAsyncerCallerThreadTest.kt @@ -171,6 +171,49 @@ class ExecutorAsyncerCallerThreadTest { } } + /** + * A fan-out where one item fails, which is the ordinary case rather than an exotic one - + * one tool call of several throwing does not abandon the request. + * + * The caller has to come out of that still inside its process, or the error handling that + * runs next is the code that discovers the blackboard is empty. Failure and loss of context + * arriving together is the worst version of this bug, because the exception looks like the + * whole story. + */ + @Test + fun `a failing item in a fan-out does not take the caller's process with it`() { + val outer = mockk() + + outer.withCurrent { + assertThrows(RuntimeException::class.java) { + asyncer.parallelMap((1..5).toList(), maxConcurrency = 5) { item -> + if (item == 3) error("item 3 failed") else AgentProcess.get() + } + } + assertSame(outer, AgentProcess.get(), "the caller must still be inside its process to handle the failure") + } + } + + /** + * A task that establishes its own process must not hand it back to the caller. + * + * Restoring means putting back what the CALLER had, not leaving whatever the task last set. + * A test that only asserts "not null" would pass on an implementation that leaked the + * task's process upward, which would be a worse bug than the one being fixed - work + * attributed to, and billed against, the wrong process. + */ + @Test + fun `a task that sets its own process does not leak it to the caller`() { + val outer = mockk() + val other = mockk() + + outer.withCurrent { + asyncer.async { AgentProcessAccessor.setValue(other) }.get(5, TimeUnit.SECONDS) + + assertSame(outer, AgentProcess.get(), "the caller's own process, not the one the task set") + } + } + @Test fun `repeated asyncs do not erode the caller's process`() { val outer = mockk()