fix(agui): clear active-run marker before terminal signal propagation

[helloworldtang]

问题

AG-UI 线程的 active-run 标记在 doFinally 中清除，而 doFinally 的执行时机是终止信号传播给订阅者之后。调用方收集完一个 run 的事件后立即在同一 thread 上发起下一个 run——这正是标准的 resume 流程——此时可能观察到残留的标记，被拒绝并收到：

RunError[message=Thread already has an active run; wait for run <id> to finish before starting another run on the same thread, code=AGUI_INTERRUPT_CONTRACT_ERROR]

尽管上一个 run 从调用方视角已经完全结束。

该问题已在本仓库的 CI 上间歇性复现：AguiPermissionResumeTest 在 PR #2858 的分支（一次 Update branch 之后）以及 main 分支自身（如 2026-09-09 d20ebbe0 的 Java CI with Maven）都失败过。本地同一 commit 复跑 3/3 全部通过——这正是时序竞态的典型表现：CI runner 越慢，竞态窗口越大。

根因

AguiRequestProcessor.process 中：

• resumeCoordinator.beginRun(input) 注册 thread 的活跃 run（putIfAbsent）；
• 标记仅由 .doFinally(signal -> finishRun(threadId, runId)) 清除；
• doFinally 的清理在 onComplete/onError 向下游传播之后执行，因此 finishRun 与调用方随后的 beginRun 之间没有 happens-before 保证。

修复

改为在 doOnComplete / doOnError 中清除标记——它们在终止信号到达调用方之前执行，为 finishRun 与后续 beginRun 之间建立 happens-before 保证。finishRun 是幂等的（remove(key, value)）；doFinally 钩子保留作为 cancel 路径的兜底（cancel 无法在信号传播前拦截）。

测试

• 新增回归测试 processAllowsImmediateFollowUpRunAfterPreviousRunTerminates：同 thread 连续 20 次立即发起新 run，断言无 RunError——把传播契约钉成可执行断言；
• agentscope-extensions-agui：507 个测试全部通过；
• agentscope-core（经 -am 联动构建）：2335 个测试全部通过。

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

[oss-maintainer]

Summary

Clears the AG-UI active-run marker in doOnComplete/doOnError (before terminal-signal propagation) instead of relying solely on doFinally, fixing the resume-flow race where an immediately-following run on the same thread could be rejected with AGUI_INTERRUPT_CONTRACT_ERROR. The approach is correct: finishRun uses remove(threadId, runId), so the now-triple invocation is idempotent and cannot clear a newer run's marker, and doFinally is kept as the cancellation-path safety net. Registry keys (threadId/runId) are captured within Flux.defer, and the sync-catch path already called finishRun unconditionally.

Verdict

LGTM — well-diagnosed fix with a solid regression test. One non-blocking [Info] note inline about what the new test does and does not pin down.

---

Automated review by github-manager-bot

[oss-maintainer]

[Info] Good regression test for the observable contract (successive runs on the same thread must not be rejected). One caveat: with a synchronous Flux.just(...) publisher this test would likely also pass on the old doFinally-only code, because doFinally is registered by the upstream publisher and still fires before the downstream subscriber's onComplete. The two-argument activeRunsByThread.remove(threadId, runId) in finishRun already prevented stale markers from stealing a newer run. So this pins the behavior rather than proving the race — fine as-is, just noting it.

[oss-maintainer]

Summary

Re-review of the follow-up commit 46156613 (test-only, +76 in AguiRequestProcessorTest, production diff unchanged from the head I approved at 12aa4db3). The two new tests cover the paths the original one did not: an in-band RunError on a completing stream (doOnComplete) and a raw error signal escaping the adapter (doOnError). CI on this head is green — build (ubuntu-latest), build (windows-latest), Check License, Check Module Sync, codecov/patch all pass — CLA is signed, merge state is CLEAN, and the fix itself (idempotent finishRun on all three hooks, remove(threadId, runId) so a newer run's marker is never stolen) still reads correctly.

Verdict

LGTM — approving the new head. Two non-blocking [Info] notes inline: the new tests pin the contract rather than reproduce the cross-thread race (same caveat as my previous comment), and the "first call fails" stubs are keyed on subscription count instead of run id.

---

Automated review by github-manager-bot

[oss-maintainer]

[Info] Extending the note I left on the previous head, and it applies to both new tests: with a synchronous publisher the doFinally cleanup and the follow-up beginRun end up on the same thread inside the caller's own stack (block() only returns after the whole signal path has unwound), so these would very likely also pass on the pre-fix doFinally-only code. They are good contract documentation, just not red/green guards.

A deterministic version is to do the follow-up run from a downstream hook instead of after block() — a doOnComplete callback registered by the caller runs after the operator's own doOnComplete but before doFinally, so it fails on the old code and passes on the new:

AtomicReference<List<AguiEvent>> followUp = new AtomicReference<>();
processor.process(request(input("run-0"))).events()
        .doOnComplete(
                () ->
                        followUp.set(
                                processor.process(request(input("run-1")))
                                        .events()
                                        .collectList()
                                        .block()))
        .subscribe();
assertNotNull(followUp.get());
assertTrue(followUp.get().stream().noneMatch(AguiEvent.RunError.class::isInstance));

Non-blocking — CI is green and the production change itself is what fixes the real cross-thread flake.

[oss-maintainer]

[Info] Minor robustness nit, non-blocking: adapterRuns.getAndIncrement() == 0 keys the injected failure on the subscription count rather than on the run. process() re-executes beginRun and adapter.run on every subscription (see processCreatesIndependentRunStateForEachEventsSubscription in this same file), so a second subscription anywhere in this test would silently move the failure to the wrong run. Matching on the run id is self-documenting and immune to that:

return "run-1".equals(input.getRunId())
        ? Flux.error(new IllegalStateException("adapter stream failed"))
        : Flux.empty();

Same pattern for the calls counter in processAllowsImmediateFollowUpRunAfterInStreamRunError.