Repository navigation
fix(a2a): map HITL confirmation events to A2A - #2577
mikemikimike wants to merge 31 commits into
Conversation
|
|
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
|
oss-maintainer
left a comment
oss-maintainer
left a comment
There was a problem hiding this comment.
Summary
This PR modifies 3 files with 254 lines of diff.
Observations
- Files changed: 3
- Test coverage: ✅ Tests included
- CLA: ✅ Signed
Suggestions
- Please ensure all public API changes are backward compatible
- Consider adding unit tests for new logic paths
Automated review by github-manager-bot
|
Addressed the executor-side wiring noted in the review comment:
Validation at
The previous required Ubuntu run failed in the unrelated |
AgentScopeJavaBot
added
enhancement
Signed-off-by: mikemikimike <13286568797@163.com>
|
I rechecked this branch against the current base commit cdf1e5a. Upstream commit d588528 (merged as PR #2222) already migrates AgentRunner, BaseReActAgentRunner, and AgentScopeAgentExecutor to the fine-grained AgentEvent/streamEvents API described by issue #2552. The extra compatibility layer in this PR reintroduces the deprecated stream() method as a required abstract method; because the previous AgentRunner contract only required streamEvents(), that change can break existing custom AgentRunner implementations. The A2A module clean verify passes locally, but the current branch is duplicating the upstream migration and its Codecov patch check is still red. Could a maintainer please review whether this PR is superseded by #2222 and should be closed or reduced to any remaining delta? |
oss-maintainer
left a comment
oss-maintainer
left a comment
There was a problem hiding this comment.
Summary
Delta re-review after the 8 author commits since 59a23db9. The compatibility story is now much better: streamEvents gained a default, stream is back for legacy runners, BaseReActAgentRunner factors both through one streamWithAgent helper, and the executor wraps legacy Events into AgentEvents so the handler path stays single-model. Cache/error coverage is solid (AgentRunnerTest +5 cases, AgentScopeAgentExecutorTest +3). What is left is about how the fallback is triggered and when state is registered/suppressed — three inline points below.
Verdict
COMMENT — the sentinel-vs-real-error ambiguity (AgentRunner.java:74) is the one I would like resolved or explicitly accepted; the assembly-time cache registration (BaseReActAgentRunner.java:69) is the riskier of the two mechanically. CLA signed.
| * @return Flux of fine-grained events emitted during execution, or a Flux that terminates with | ||
| * {@link UnsupportedOperationException} when the runner does not support fine-grained events | ||
| */ | ||
| default Flux<AgentEvent> streamEvents(List<Msg> requestMessages, AgentRequestOptions options) { |
There was a problem hiding this comment.
UnsupportedOperationException is doing double duty here: it is the capability sentinel for not-yet-migrated runners and simultaneously a plausible genuine failure from a migrated runner's agent pipeline (unsupported modality, an SPI that refuses an operation, a codec path). The executor's onErrorResume(UnsupportedOperationException.class, ...) cannot tell the two apart before the first onNext, and if it mistakes a real error for "not supported" it re-runs the request through stream() — duplicate model calls, duplicate tool side effects, and the original error disappears from the response.
Suggest making the capability explicit instead of inferring it from an exception type: either a dedicated marker (e.g. UnsupportedAgentEventStreamException extends UnsupportedOperationException thrown only by this default), or a default boolean supportsFineGrainedEvents() { return false; } probe that AgentScopeAgentExecutor.streamAgentEvents checks up front. The executor's emitted flag only narrows the window, it does not close it.
| agentCache.put(options.getTaskId(), agent); | ||
| return agent.streamEvents(requestMessages) | ||
| .doFinally(signal -> agentCache.remove(options.getTaskId())); | ||
| agentCache.put(taskId, agent); |
There was a problem hiding this comment.
streamWithAgent registers the agent in agentCache at assembly time but only unregisters in doFinally, i.e. at subscribe-terminal time. The fallback path added in this PR can now assemble two streams for the same taskId in one request: streamEvents(...) is assembled (entry created), it terminates, and onErrorResume subscribes stream(...) — whose assembly can race the upstream doFinally cleanup and trip IllegalStateException("Agent already exists for taskId: ...") at line 65. The never-subscribed case is worse: the fineGrainedStream == null branch and any assembly-time throw leave the entry in the cache permanently, after which every later request for that taskId fails.
Could the check + put move inside Flux.defer { ... } (or doOnSubscribe) so registration happens exactly when the stream is subscribed and is always paired with the doFinally removal? That also makes the idempotency guarantee independent of the new fallback path.
| || !legacyEvent.legacyEvent.isLast()) { | ||
| return false; | ||
| } | ||
| return Objects.equals(lastLegacyEventMsgId, legacyEvent.legacyEvent.getMessageId()); |
There was a problem hiding this comment.
Dedup compares the incoming legacy message id against the single previously-seen id (lastLegacyEventMsgId, written at line 374). If a legacy runner interleaves message ids (chunk X, chunk Y, terminal X), the terminal X no longer matches lastLegacyEventMsgId (Y), so the completed message is streamed again after the deltas — duplicated text at the A2A client. Conversely, if REASONING and SUMMARY share one reply id, a terminal event can be suppressed even though it carries the authoritative final text.
An id-keyed set of "already streamed non-terminal" ids (rather than the last id only) would make the rule order-independent. A test with two interleaved ids would pin it.
|
Two follow-ups to go with my review, one of them raised by @mikemikimike in the last comment and worth an explicit ⬆️ for maintainers. 1. This PR is largely superseded by #2222 (scope question for a maintainer). On 2. The shim as written is source-breaking in the opposite direction. Making No action needed from the bot side; flagging for maintainer judgement on scope. |
oss-maintainer
left a comment
oss-maintainer
left a comment
There was a problem hiding this comment.
Summary
Delta re-review at d8e23bf5 (previous pass at 86f870a0). The single commit since then does exactly what the last round asked for, and all three open findings are resolved:
- F1 capability sentinel (
AgentRunner.java:81) - resolved. ThestreamEventsdefault now fails with a dedicatedUnsupportedAgentEventStreamException(a final subclass ofUnsupportedOperationException), and the executor catches only that marker, both synchronously (AgentScopeAgentExecutor.java:190-195) and viaonErrorResume(177-189), with an emitted-flag guard (174-182) so the fallback cannot start after anyonNext. A genuineUnsupportedOperationExceptionfrom a migrated runner's pipeline now propagates instead of causing a second execution.testDoesNotFallBackForAgentUnsupportedOperation,testDoesNotFallBackAfterFineGrainedOutputandtestFallsBackForSynchronousCapabilityMarkerpin all three states. - F2 assembly-time cache registration (
BaseReActAgentRunner.java:66) - resolved.streamWithAgentis wrapped inFlux.defer, so the agent is built and registered only at subscription time throughagentCache.compute(69-79); an existing mapping now raisesIllegalStateExceptioninstead of silently rebuilding (80-84), unregistration uses the two-arg conditionalremove(taskId, agent)(88) so a staledoFinallycan no longer evict a newer registration for the same taskId, and a synchronous stream-function failure releases the entry before surfacing the error (85-92). A marker-failing fine-grained stream now registers nothing, so the fallback path can no longer double-register or race the upstreamdoFinally. - F3 legacy dedup (
AgentScopeAgentExecutor.java:415) - resolved. Dedup records every legacy chunk that actually produced a response as aLegacyEventKey(EventType, messageId)(372-379) and suppresses anisLastevent only when that exact pair was already streamed, so interleaved message ids no longer re-emit a completed message, and the old over-suppression (anAGENT_RESULTclobbering the single remembered id) is gone.testDeduplicatesInterleavedLegacyChunksis a genuine regression test for the previous logic.
Verdict
APPROVE - the reactive semantics are unambiguous now, the cached-agent lifecycle is atomic, and the delta is backed by targeted tests. Two non-blocking notes are pinned inline: one assertion lost its teeth when registration became lazy (AgentRunnerTest.java:371), and the marker-type change deserves a migration note (UnsupportedAgentEventStreamException.java:20).
CI is green on this head (Check License, Check Module Sync, ubuntu + windows builds, codecov) and CLA is signed; mergeState=BLOCKED reflects the missing-review gate only.
Automated review by github-manager-bot
| when(mockAgent.streamEvents(messages)).thenReturn(mockFlux); | ||
| when(mockFlux.doFinally(any())).thenReturn(mockFlux); | ||
| runner.streamEvents(messages, requestOptions); | ||
| runner.streamEvents(messages, requestOptions).subscribe(); |
There was a problem hiding this comment.
Test vacuity introduced by the deferral change: the sibling tests were fixed by adding blockLast (lines 313/332/354), and .subscribe() was added to the first call on this line, but the re-stream check at the end of the test (Flux<AgentEvent> result = runner.streamEvents(messages, requestOptions);, line 377) is still only assembled, never subscribed. Now that registration is deferred to subscription time, that call cannot fail even if stop() left the entry in agentCache - pre-delta the assembly-time containsKey is what made the assert meaningful. Restub streamEvents to Flux.empty() after runner.stop(taskId) and blockLast the second call so the test actually proves the entry was released.
| package io.agentscope.core.a2a.server.executor.runner; | ||
|
|
||
| /** Signals that an {@link AgentRunner} has not implemented fine-grained event streaming. */ | ||
| public final class UnsupportedAgentEventStreamException extends UnsupportedOperationException { |
There was a problem hiding this comment.
Compatibility narrowing worth a changelog line: the documented opt-out signal changed from plain UnsupportedOperationException to this marker, and the executor's synchronous catch was narrowed to the marker type (AgentScopeAgentExecutor.java:190). Catching code is safe (the marker is a subclass), but a third-party runner that manually threw or returned Flux.error(new UnsupportedOperationException(...)) from streamEvents per the previous javadoc no longer triggers the legacy fallback - the request now fails with the raw error. Acceptable while 2.0 is unreleased; please state it in the migration notes and document on AgentRunner.streamEvents that only UnsupportedAgentEventStreamException (or not overriding the method) selects the fallback.
|
Addressed the remaining review follow-ups in commit
Validation:
|
oss-maintainer
left a comment
oss-maintainer
left a comment
There was a problem hiding this comment.
Summary
Delta re-review at b5a58287 (previous pass approved d8e23bf5). The single new commit is exactly the follow-up the last round asked for and touches no production behavior:
AgentRunner.streamEvents(...)javadoc now states the legacy-fallback contract explicitly (default implementation / marker exception vs. a realUnsupportedOperationException).AgentRunnerTest.testStopAgentAndRemoveFromCachenow subscribes to the second stream (blockLast()) instead of asserting on an un-subscribedFlux, so the "agent was removed from the cache" claim is actually exercised.- Both
change-log.mdcopies (en + zh) document the same rule.
Implementation side is unchanged and still correct: AgentScopeAgentExecutor.streamAgentEvents(...) narrows both the eager catch and the onErrorResume to the dedicated UnsupportedAgentEventStreamException subclass, so a genuine UnsupportedOperationException from the agent pipeline propagates rather than silently re-executing the agent through the legacy path. CLA signed, 6/6 checks green, mergeable=true / clean → re-approving on the current head.
Findings
- [Info]
AgentRunner.java:74— the new sentence omits the "before any element is emitted" guard that the executor really applies (anchored inline). - [Info]
docs/v2/en/docs/change-log.md:210— the A2A note is inserted between the colon lead-in and the deprecation bullet list; same layout in the zh copy (anchored inline).
Both are documentation wording nits, non-blocking.
Automated review by github-manager-bot
| * custom runners remain source-compatible whether they implement this method or only the | ||
| * legacy {@link #stream(List, AgentRequestOptions)} contract. | ||
| * The A2A executor selects the legacy stream only when this default implementation is used or | ||
| * the returned Flux terminates with {@link UnsupportedAgentEventStreamException}; a plain |
There was a problem hiding this comment.
[Info] Minor accuracy gap in the new fallback contract sentence. AgentScopeAgentExecutor.streamAgentEvents(...) also guards on whether anything has already gone downstream (emitted flag + onErrorResume(UnsupportedAgentEventStreamException.class, ...)): if the fine-grained stream emits at least one element and then fails with the marker, the error is propagated instead of re-running through the legacy stream(...). Worth saying "...terminates with UnsupportedAgentEventStreamException before any element is emitted", otherwise a custom-runner author could reasonably assume late fallback is available. Not blocking — the behavior itself is the right one (falling back after partial output would duplicate text/tool results).
|
|
||
| Python 2.0's `agent.reply_stream()` exposes a single streaming signature (`AsyncGenerator[AgentEvent, None]`) that maps directly to Java's fine-grained `io.agentscope.core.event.AgentEvent` hierarchy. To match it, the coarse-grained `Flux<Event> stream(...)` API on the Java side is `@Deprecated` as of 2.0.0: | ||
|
|
||
| For A2A `AgentRunner` implementations, leave `streamEvents(...)` unoverridden or return `UnsupportedAgentEventStreamException` when the fine-grained stream is unavailable to select the legacy `stream(...)` fallback. A plain `UnsupportedOperationException` from an overridden `streamEvents(...)` is treated as an execution failure and is not retried through the legacy stream. |
There was a problem hiding this comment.
[Info] This A2A paragraph is inserted between the lead-in sentence ("... is @Deprecated as of 2.0.0:") and the bullet list it introduces, so the colon now reads as if it introduces the A2A note. Same layout in docs/v2/zh/docs/change-log.md. Suggest moving the note below the list (or as a **Note** block after it) so the deprecation list stays attached to its intro. Docs only, not blocking.
oss-maintainer
left a comment
oss-maintainer
left a comment
There was a problem hiding this comment.
Summary
Delta re-review at 4c5e1a3f (previous pass approved b5a58287). The single new commit is a javadoc/documentation clarification of the A2A stream-fallback contract, and it is accurate against the shipped code:
AgentRunner.streamEvents(...)now states the legacy fallback is selected only when the default implementation is used or the stream fails withUnsupportedAgentEventStreamExceptionbefore any element is emitted. That is exactly whatAgentScopeAgentExecutor.streamAgentEvents(...)does: theAtomicBoolean emittedguard re-emits the error instead of falling back once anything went downstream, so a mid-stream failure can never cause a second agent execution.- The A2A paragraph in
docs/v2/en/docs/change-log.mdanddocs/v2/zh/docs/change-log.mdmoved below theforRemovalbullet list; wording is unchanged and en/zh stay in sync.
No production behavior is touched. CI on this head is green (Linux + Windows build, license, module sync, codecov, validate) and license/cla is signed. Re-approving — nothing further to ask.
Automated review by github-manager-bot
mikemikimike
changed the title
oss-maintainer
left a comment
oss-maintainer
left a comment
There was a problem hiding this comment.
Summary
Delta re-review at 9bad5d2a (previous pass approved 4c5e1a3f). The branch was rebased onto current main and then gained a substantial new commit: A2A now maps RequireUserConfirmEvent to an INPUT_REQUIRED task carrying an agentscope.confirmation_request DataPart, and accepts an agentscope.confirmation_response DataPart to resume. Net new production code on top of the previously approved state is AgentScopeAgentExecutor (+340) and BaseReActAgentRunner (+101/−14).
What looks genuinely good:
- Resume validation is strict in the right places —
replyIdmust match the pending request,toolCallIdmust reference a known pending call, duplicates are rejected,confirmedmust be a real boolean,result.toolCall.nameis pinned to the requested name, and at most one control DataPart per message. Rejectingconfirmation_responseon non-INPUT_REQUIREDtasks closes the replay hole. - The
concatWithdeferral (emit the request only after the agent has actually finished pausing, plus the "Agent was stopped before confirmation could be requested" check) is the correct ordering fix, and clearing thepausedflag insidecomputeis atomic. - The earlier threads on this PR are resolved by the rewrite:
UnsupportedAgentEventStreamExceptionand the capability-sentinel double duty are gone, and registration moved intoFlux.defer, so it is no longer done at assembly time. - Tests track the new contract closely (
AgentScopeAgentExecutorTest+410,AgentRunnerTest+86/−13), CI is green on this head (7/7),license/clais signed, and the rebase landed clean (MERGEABLE/CLEAN) with no conflict residue.
Findings (5, no blockers) are about error surfacing and lifecycle rather than the happy path:
- [Warning] Protocol-validation failures in
convertInputMessagebypass the task lifecycle — the client gets a bare text message and the task staysINPUT_REQUIRED, so a rejected resume is indistinguishable from a hung one (AgentScopeAgentExecutor.java:148). - [Warning] "confirmation more than once" fails a run the core explicitly allows:
ReActAgent.validateAndAcceptConfirmResultspermits confirming a subset of ASKING calls, and combined with the point above the second pause strands the task (AgentScopeAgentExecutor.java:550). Fine to keep the single-pause contract, but it should be documented and enforced on the resume side. - [Warning]
SignalType.ON_COMPLETEretention gives paused agents no expiry, so abandoned confirmations leak a liveReActAgenteach;CachedAgent.pausedis additionally a plain non-volatilebooleanread outside the map lock (BaseReActAgentRunner.java:127). - [Info] Client-edited tool arguments are executed by design — worth one explicit change-log line, plus a pinned contract for clearing
input(AgentScopeAgentExecutor.java:299). - [Info] The resume handshake requires the task store to preserve
status.message(), which should be stated as a durability requirement (AgentScopeAgentExecutor.java:229).
Holding this pass at COMMENT rather than re-approving, since the previous APPROVED was recorded against 4c5e1a3f and the multi-round / cache-lifecycle contract above is worth an explicit author decision first. Happy to re-approve on the next push.
Automated review by github-manager-bot
| @@ -134,6 +148,242 @@ public void execute(RequestContext context, EventQueue eventQueue) throws JSONRP | |||
| } | |||
There was a problem hiding this comment.
[Warning] Protocol-validation failures never reach the task lifecycle.
convertInputMessage(context) runs before the task is looked up/created, and the outer catch (Exception e) only enqueues a plain text message (A2A.createAgentTextMessage(...)). For the new rejection paths in convertInputMessage (missing / duplicate confirmation_response DataPart, replyId mismatch, unknown toolCallId, non-boolean confirmed, ...) the client therefore gets:
- a message that is not a
TaskStatusUpdateEvent, so a task-watching client cannot tell that the resume was rejected; - the task left in
INPUT_REQUIREDwith no state update — the client will just retry the same payload; - an error string built from
e.getMessage(), which for these new validations echoes field names and (indirectly) values from the client payload into the log and the returned message.
Since every one of these is a caller error, it would be nicer to map them to InvalidParamsError (or whatever the JSON-RPC error type for the a2a SDK is) and/or emit taskUpdater.failed(...) so the task terminates in a well-defined state. A cheap middle ground: run convertInputMessage after the task resolution and reuse the processTaskNonBlocking catch style (taskUpdater.fail(...)) for the resume path.
| LoggerUtil.debug( | ||
| log, "[{}] Handle Agent execute output event: {}", context.getTaskId(), output); | ||
| if (output instanceof RequireUserConfirmEvent request) { | ||
| if (!waitingForInput.compareAndSet(false, true)) { |
There was a problem hiding this comment.
[Warning] A second HITL pause aborts a run that the core explicitly allows to continue.
Both guards here and in BaseReActAgentRunner.streamEvents ("requested user confirmation more than once") make the A2A layer fail hard on a second RequireUserConfirmEvent within one stream. ReActAgent.validateAndAcceptConfirmResults documents that "Confirmations may cover a subset of ASKING calls", so a client that confirms 1 of 2 ASKING tool calls (or that approves a batch which then hits another gated tool) reaches this branch. Combined with finding #1 the outcome is a task stuck in INPUT_REQUIRED: the stream errors, doOnError writes a text message, and no terminal task state is ever set, so the agent progress made before the second pause is lost.
Two options, either is fine, but the chosen contract should be explicit:
- per-stream single pause (current): say so in the
AgentRunner.streamEventsjavadoc +change-log.mdentry, and make the resume side reject partial confirmation batches up front (parseConfirmationResultscan requireresultsto cover everytoolCallsid) so the second event is genuinely unreachable; or - multi-round: keep the pending
confirmation_requestDataPart in the task status message and allow N pauses per task by re-armingwaitingForInputwhen a resume is accepted (the runner already re-uses the cached agent across resumes, so the plumbing is half-way there).
Also worth noting: because BaseReActAgentRunner already filters the event and emits at most one, the guard here is only reachable for custom AgentRunner implementations — the two layers fail differently (Mono.error vs. a raw throw inside doOnNext, which in the blocking path escapes blockLast()). Aligning them would keep the error surface predictable.
| })) | ||
| .doFinally( | ||
| signal -> { | ||
| if (signal != SignalType.ON_COMPLETE |
There was a problem hiding this comment.
[Warning] Paused agents are retained with no expiry, and SignalType.CANCEL keeps them too.
doFinally keeps the CachedAgent on ON_COMPLETE when a confirmation was requested — correct for resume — but nothing bounds its lifetime afterwards:
- a client that receives
INPUT_REQUIREDand never resumes leaks one liveReActAgent(context, memory, model/tool clients) per abandoned task, forever; signal != ON_COMPLETEcoversERRORbut a cancelled subscription after the request was emitted also removes/keeps the entry depending on where the cancel lands, which makes the follow-up resume either fail with "Agent already exists for taskId" (a new, unpaused entry is built) or reuse a stale agent non-deterministically.
The old code had the same shape, but the pause semantics make the leak much easier to hit in production (users routinely walk away from a confirmation prompt). Suggested direction: store pausedAt in CachedAgent and evict expired entries on streamEvents/stop (a small Map sweep is enough — this runs per request anyway), and/or drain abandoned tasks through cancel(RequestContext, EventQueue).
Minor, same class: CachedAgent.paused is a plain mutable boolean written inside compute/computeIfPresent and read in doFinally, i.e. outside the map's bin lock. AtomicBoolean (or making the cached value an immutable record with a paused copy) removes the data race.
| throw new IllegalArgumentException( | ||
| "result.toolCall.name must match the requested tool call."); | ||
| } | ||
| if (modifiedToolCall.containsKey("input")) { |
There was a problem hiding this comment.
[Info] Client-supplied tool-call edits are passed through verbatim.
The requested tool call is copied and then input is replaced with whatever the client sent, and the resulting ToolUseBlock is handed to ConfirmResult for execution. name is pinned to the original, which is the important part, but the arguments are fully attacker-controlled by design here (that is the point of "edit before approving"), so this only needs to be explicit rather than implicit:
- a short note in the
change-log.mdentry stating thatresults[].toolCall.inputreplaces the model's arguments, so reviewers of the A2A surface know the permission decision covers edited inputs; if (modifiedToolCall.containsKey("input"))cannot clear the input (input: nullfalls throughtoStringMapas an error). Treating an explicit"input": {}as the clear-path is enough, but worth a test so the contract is pinned.
No action needed if this is already the documented AG-UI/permission-engine behaviour — the AG-UI converter does the same (editedArgs overrides both input and content). Keeping the two surfaces consistent is the ask.
|
|
||
| private static PendingConfirmation findPendingConfirmation(Task task) { | ||
| Message statusMessage = task.getStatus().message(); | ||
| DataPart requestPart = findControlPart(statusMessage, CONFIRMATION_REQUEST_TYPE); |
There was a problem hiding this comment.
[Info] The resume handshake depends on the task store keeping status.message() intact.
findPendingConfirmation reads the confirmation_request DataPart back out of task.getStatus().message(). That works with an in-memory / message-preserving store, but a task store that trims or rewrites the status message (persistence to a DB with a message column dropped on upsert, or a store that replaces INPUT_REQUIRED with a newer status update) turns a legitimate resume into "The INPUT_REQUIRED task does not contain an AgentScope confirmation request."
Worth one line in the change-log/docs on the durability requirement for the confirmation request payload (or a fallback to rebuild the pending batch from the agent state, which the runner can already reach through the cached agent).
|
Addressed the five findings from the latest review in commit I chose the single-pause-per-stream contract: each confirmation response must cover every pending tool call exactly once. A later resume uses a new stream and can request another complete batch. Invalid resume data now emits a final The English and Chinese changelogs now state that edited Please re-review the updated head when convenient. |
oss-maintainer
left a comment
oss-maintainer
left a comment
There was a problem hiding this comment.
Summary
This PR maps AgentScope RequireUserConfirmEvent to A2A INPUT_REQUIRED tasks, validates confirmation responses, and resumes paused ReActAgent instances while adding a 30-minute paused-agent retention. The core logic is sound: reply validation is strict, invalid resumes fail the task and stop the agent, and the runner defers registration and handles expiry. However, there are two remaining correctness/cleanup issues around blocking resumes and agent lifecycle cleanup.
Findings
- [Warning]
agentscope-extensions/.../AgentScopeAgentExecutor.java:141— Blocking confirmation resumes are forced through the streaming/non-blocking path because the branch requirescontext.getTask() == null. A blockingtasks/sendresume for anINPUT_REQUIREDtask should be handled byprocessTaskBlockingso the client receives the expected single-response events (Message or Task), not streamingTaskStatusUpdateEvents. Remove thecontext.getTask() == nullguard from the blocking branch or add an explicit blocking-resume handler. - [Warning]
agentscope-extensions/.../BaseReActAgentRunner.java:178—stop()only callsagent.close()when the cached agent is paused. For a running (non-paused) agent,stop()removes it from the cache and interrupts it but never closes it. BecausedoFinallylater callsremoveAgent(taskId, cachedAgent)and the cache no longer contains that entry,removeAgentreturns false and also skipsclose(). Running agents stopped viastop()therefore leak the resources released byReActAgent.close(). Always close the agent instop()(in addition to interrupting), or ensuredoFinallycloses even when the cache entry is already gone. - [Info]
agentscope-extensions/.../AgentScopeAgentExecutorTest.java:478— The new resume tests all use streaming contexts. Add a test that resumes anINPUT_REQUIREDtask with a blocking request configuration to verify the blocking response path.
Suggestions
- Add a blocking-resume test and route blocking resumes through
processTaskBlocking. - Make
BaseReActAgentRunner.stop()always close the agent (or makeremoveAgentclose the captured agent regardless of cache state) so stopped running agents release resources deterministically.
Automated review by github-manager-bot
| log.info("[{}] Using existing task.", task.getId()); | ||
| } | ||
| if (isBlockRequest(context)) { | ||
| if (isBlockRequest(context) && context.getTask() == null) { |
There was a problem hiding this comment.
Blocking confirmation resumes are forced through the streaming/non-blocking path because the branch requires context.getTask() == null. A blocking tasks/send resume for an INPUT_REQUIRED task should be handled by processTaskBlocking so the client receives the expected single-response events (Message or Task), not streaming TaskStatusUpdateEvents. Remove the context.getTask() == null guard from the blocking branch or add an explicit blocking-resume handler.
| } | ||
|
|
||
| @Override | ||
| public void stop(String taskId) { |
There was a problem hiding this comment.
stop() only calls agent.close() when the cached agent is paused. For a running (non-paused) agent, stop() removes it from the cache and interrupts it but never closes it. Because doFinally later calls removeAgent(taskId, cachedAgent) and the cache no longer contains that entry, removeAgent returns false and also skips close(). Running agents stopped via stop() therefore leak the resources released by ReActAgent.close(). Always close the agent in stop() (in addition to interrupting), or ensure doFinally closes even when the cache entry is already gone.
| @ParameterizedTest | ||
| @ValueSource(booleans = {false, true}) | ||
| @DisplayName("Should convert a confirmation response into validated AgentScope metadata") | ||
| void testStreamingRequestConvertsConfirmationResponse(boolean clearInput) |
There was a problem hiding this comment.
The new resume tests all use streaming contexts (doMockForContext(false, false, true) or doMockForContext(true, false, false)). Add a test that resumes an INPUT_REQUIRED task with a blocking request configuration to verify the blocking response path (currently routed to processTaskNonBlocking at line 141).
|
Addressed the blocking-resume and running-agent cleanup findings in
The new regression assertions fail against the previous production logic (three failures) and pass with the fix. |
oss-maintainer
left a comment
oss-maintainer
left a comment
There was a problem hiding this comment.
Summary
Re-reviewed head b256cf81 (one commit since my 2a053c72 review). Both open items from last round are resolved: blocking requests on an existing task now go through processTaskBlocking instead of being forced onto the streaming path, and BaseReActAgentRunner.stop() closes the cached agent unconditionally, so a stopped running agent no longer leaks the resources released by ReActAgent.close(). The resume test was also converted to a @CsvSource matrix over streaming/blocking with assertions on the emitted event shape, and AgentRunnerTest now verifies close() — exactly the coverage gap I had flagged. Approving; the two inline notes are follow-ups, not blockers.
Findings
- [Info]
BaseReActAgentRunner.java:185— double-close safety relies on cache-removal ordering; worth a comment. - [Info]
AgentScopeAgentExecutor.java:141— blocking resumes now wait for the full run on the request thread; add a change-log entry.
Checks
- CI:
validate,Check License,Check Module Sync,build (ubuntu-latest),build (windows-latest),codecov/patch,license/cla— all passing at review time. - CLA: signed.
Automated review by github-manager-bot
Findings
- [Info]
agentscope-extensions/.../runner/BaseReActAgentRunner.java:185— Unconditionalclose()instop()fixes the running-agent leak. Double-close is avoided becausestop()removes the entry withagentCache.remove(taskId)before the stream'sdoFinallyreachesremoveAgent(taskId, cachedAgent), whose conditionalremove(key, value)then returns false. That safety depends on the cache-removal ordering rather than onclose()being idempotent — one comment stating so would keep the next refactor from reordering it. - [Info]
agentscope-extensions/.../executor/AgentScopeAgentExecutor.java:141— Dropping thecontext.getTask() == nullguard makes blocking resumes go throughprocessTaskBlocking, which is what the previous round asked for.processTaskBlockingends inblockLast()on the request thread, so a blocking resume of anINPUT_REQUIREDtask now waits for the whole agent run instead of returning once the confirmation is accepted. That matches A2A blocking semantics, but it is a latency-visible change for existing callers and deserves a line indocs/v2/{en,zh}/docs/change-log.mdnext to the HITL mapping entry.
Inline anchoring was rejected by the review API on this commit, so both notes are listed here instead of as line comments.
Signed-off-by: mikemikimike <13286568797@163.com>
Preview approval requiredThis pull request comes from a fork, so Mintlify won't build a preview until someone with write access approves commit |
Summary
RequireUserConfirmEventto an A2AINPUT_REQUIREDtask with the pending tool calls.ReActAgentwhile awaiting confirmation, validate the reply against the pending task, and resume execution with the confirmed results.Addresses #2552
Closes #2552.
Tests
mvn -B -T1 -pl agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-a2a/agentscope-extensions-a2a-server -am clean verify(passed; 2,462 core tests, 111 A2A client tests, and 140 A2A server tests; 9 core tests skipped)