Skip to content

Commit ca5ce02

Browse files
author
Mark Pollack
committed
Close the stdio agent's input first on a graceful close, and stop it once
closeGracefully() sent SIGTERM without closing the agent's standard input, so the agent's own end-of-input drain never ran and it exited 143. It now closes stdin and waits up to two seconds for the agent to exit by itself before SIGTERM, then kill. A second close no longer stops the process again or logs its stop twice.
1 parent d4c1713 commit ca5ce02

3 files changed

Lines changed: 193 additions & 10 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -522,6 +522,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
522522

523523
### Fixed
524524

525+
- **The stdio client's graceful close lets the agent exit by itself, and logs the stop once.**
526+
`StdioAcpClientTransport.closeGracefully()` sent SIGTERM straight away without closing the
527+
agent's standard input, so an agent's end-of-input handling (an SDK stdio agent answers what it
528+
received, flushes, and exits 0) never ran and every close ended in exit 143. It now closes the
529+
agent's standard input first and waits up to `StdioAcpClientTransport.END_OF_INPUT_WAIT_MILLIS`
530+
(2 seconds) for the agent to exit; only an agent still running then is sent SIGTERM, and killed
531+
five seconds later as before. Closing again, as `closeGracefully()` followed by try-with-resources
532+
`close()` does, no longer stops the process a second time or logs "ACP agent process stopped"
533+
twice.
534+
525535
- **`prompt()` returns only after the session-update consumers have handled the turn's updates.**
526536
On the client, notifications are delivered one at a time on their own drain, while a response
527537
completed its caller as soon as it arrived. A prompt's response follows its turn's updates, so a

‎acp-core/src/main/java/com/agentclientprotocol/sdk/client/transport/StdioAcpClientTransport.java‎

Lines changed: 52 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,17 @@
5757
* {@link #setStdErrorHandler}.
5858
* </p>
5959
*
60+
* <p>
61+
* {@link #closeGracefully()} ends the agent process in steps. It closes the agent's standard
62+
* input and waits up to {@value #END_OF_INPUT_WAIT_MILLIS} ms for the agent to exit by itself,
63+
* as an ACP stdio agent does at the end of its input (an agent built with this SDK answers what
64+
* it received, then exits 0). An agent still running then is sent SIGTERM ({@link
65+
* Process#destroy()}; on POSIX systems it exits 143 unless it handles the signal) and given
66+
* five more seconds, and is then killed (exit 137). The end is logged once, at INFO, however
67+
* often the transport is closed: {@code closeGracefully()} followed by {@code close()}, as
68+
* try-with-resources does, does not stop the agent twice.
69+
* </p>
70+
*
6071
* @author Mark Pollack
6172
* @author Christian Tzolov (MCP Java SDK)
6273
* @author Dariusz Jędrzejczyk (MCP Java SDK)
@@ -114,6 +125,18 @@ public class StdioAcpClientTransport implements AcpClientTransport {
114125
*/
115126
private final AtomicBoolean isConnected = new AtomicBoolean(false);
116127

128+
/**
129+
* How long {@link #closeGracefully()} waits, after closing the agent's standard input, for
130+
* the agent to exit by itself before it sends SIGTERM: {@value} ms.
131+
*/
132+
public static final long END_OF_INPUT_WAIT_MILLIS = 2_000;
133+
134+
/** How long to wait for the agent process to exit after SIGTERM before killing it. */
135+
private static final Duration TERM_WAIT = Duration.ofSeconds(5);
136+
137+
/** Set by the first {@link #closeGracefully}: the agent process is stopped once. */
138+
private final AtomicBoolean stopping = new AtomicBoolean(false);
139+
117140
/** How long to wait for the agent process to exit after its standard output ends. */
118141
private static final Duration EXIT_WAIT = Duration.ofSeconds(5);
119142

@@ -495,14 +518,19 @@ protected void handleOutbound(Function<Flux<JSONRPCMessage>, Flux<JSONRPCMessage
495518
}
496519

497520
/**
498-
* Gracefully closes the transport by destroying the process and disposing of the
499-
* schedulers. This method sends a TERM signal to the process and waits for it to exit
500-
* before cleaning up resources.
521+
* Gracefully closes the transport: closes the agent's standard input, waits up to
522+
* {@value #END_OF_INPUT_WAIT_MILLIS} ms for the agent to exit by itself, then sends it
523+
* SIGTERM and, five seconds later, kills it; then disposes of the schedulers. Closing an
524+
* already closed transport does nothing more.
501525
* @return A Mono that completes when the transport is closed
502526
*/
503527
@Override
504528
public Mono<Void> closeGracefully() {
505529
return Mono.fromRunnable(() -> {
530+
if (!this.stopping.compareAndSet(false, true)) {
531+
logger.debug("Already closed");
532+
return;
533+
}
506534
closedLocally = true;
507535
isClosing = true;
508536
logger.debug("Initiating graceful shutdown");
@@ -512,7 +540,7 @@ public Mono<Void> closeGracefully() {
512540
outboundSink.tryEmitComplete();
513541
errorSink.tryEmitComplete();
514542

515-
// Destroy process FIRST - this closes streams and unblocks readLine()
543+
// Stop the process FIRST - its end closes the streams and unblocks readLine()
516544
Process process = this.process;
517545
if (process != null) {
518546
stop(process);
@@ -533,15 +561,29 @@ public Mono<Void> closeGracefully() {
533561
}
534562

535563
/**
536-
* Sends TERM to the process and waits up to five seconds for it to exit, then kills it.
537-
* Waits with a blocking waitFor() rather than Mono.fromFuture(process.onExit()), which
538-
* would run on ForkJoinPool.commonPool.
564+
* Closes the process's standard input and waits for it to exit by itself; then sends TERM
565+
* and waits up to five seconds, then kills it. Waits with a blocking waitFor() rather than
566+
* Mono.fromFuture(process.onExit()), which would run on ForkJoinPool.commonPool.
539567
*/
540568
private static void stop(Process process) {
541-
logger.debug("Sending TERM to process");
542-
process.destroy();
543569
try {
544-
if (process.waitFor(5, java.util.concurrent.TimeUnit.SECONDS)) {
570+
var stdin = process.getOutputStream();
571+
synchronized (stdin) {
572+
stdin.close();
573+
}
574+
}
575+
catch (IOException e) {
576+
logger.debug("Closing the agent's standard input failed: {}", e.getMessage());
577+
}
578+
try {
579+
if (process.waitFor(END_OF_INPUT_WAIT_MILLIS, java.util.concurrent.TimeUnit.MILLISECONDS)) {
580+
logExit(process.exitValue());
581+
return;
582+
}
583+
logger.debug("Agent still running {} ms after the end of its input; sending TERM",
584+
END_OF_INPUT_WAIT_MILLIS);
585+
process.destroy();
586+
if (process.waitFor(TERM_WAIT.toMillis(), java.util.concurrent.TimeUnit.MILLISECONDS)) {
545587
logExit(process.exitValue());
546588
}
547589
else {

‎acp-core/src/test/java/com/agentclientprotocol/sdk/client/transport/StdioAcpClientTransportCloseTest.java‎

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,131 @@
1+
/*
2+
* Copyright 2026-2026 the original author or authors.
3+
*/
4+
5+
package com.agentclientprotocol.sdk.client.transport;
6+
7+
import java.nio.file.Path;
8+
import java.time.Duration;
9+
import java.util.List;
10+
11+
import ch.qos.logback.classic.Logger;
12+
import ch.qos.logback.classic.spi.ILoggingEvent;
13+
import ch.qos.logback.core.read.ListAppender;
14+
import com.agentclientprotocol.sdk.agent.transport.EndOfInputAgent;
15+
import com.agentclientprotocol.sdk.client.AcpClient;
16+
import com.agentclientprotocol.sdk.client.AcpSyncClient;
17+
import com.agentclientprotocol.sdk.spec.AcpSchema;
18+
import org.junit.jupiter.api.AfterEach;
19+
import org.junit.jupiter.api.BeforeEach;
20+
import org.junit.jupiter.api.Test;
21+
import org.junit.jupiter.api.condition.DisabledOnOs;
22+
import org.junit.jupiter.api.condition.OS;
23+
import org.slf4j.LoggerFactory;
24+
import reactor.core.publisher.Mono;
25+
26+
import static org.assertj.core.api.Assertions.assertThat;
27+
28+
/**
29+
* How {@link StdioAcpClientTransport#closeGracefully()} ends the agent process: it closes the
30+
* agent's standard input and lets an agent that exits by itself at the end of its input do so,
31+
* exit code 0, before it falls back to SIGTERM; and it reports the end once, however often the
32+
* transport is closed.
33+
*/
34+
class StdioAcpClientTransportCloseTest {
35+
36+
private static final Duration TIMEOUT = Duration.ofSeconds(30);
37+
38+
private final Logger logger = (Logger) LoggerFactory.getLogger(StdioAcpClientTransport.class);
39+
40+
private final ListAppender<ILoggingEvent> logs = new ListAppender<>();
41+
42+
@BeforeEach
43+
void captureLogs() {
44+
this.logs.start();
45+
this.logger.addAppender(this.logs);
46+
}
47+
48+
@AfterEach
49+
void releaseLogs() {
50+
this.logger.detachAppender(this.logs);
51+
this.logs.stop();
52+
}
53+
54+
private List<String> stops() {
55+
return this.logs.list.stream()
56+
.map(ILoggingEvent::getFormattedMessage)
57+
.filter(message -> message.startsWith("ACP agent process stopped")
58+
|| message.startsWith("Process terminated"))
59+
.toList();
60+
}
61+
62+
private static AgentParameters javaAgent(Class<?> main) {
63+
String java = Path.of(System.getProperty("java.home"), "bin", "java").toString();
64+
return AgentParameters.builder(java)
65+
.arg("-Dlogback.configurationFile=logback-stdio-agent.xml")
66+
.arg("-cp")
67+
.arg(System.getProperty("java.class.path"))
68+
.arg(main.getName())
69+
.build();
70+
}
71+
72+
/**
73+
* An SDK stdio agent, written as the README shows, answers what it received once its input
74+
* ends and exits 0: the graceful close lets it, rather than SIGTERM it (exit 143).
75+
*/
76+
@Test
77+
void aJavaStdioAgentExitsByItselfAfterAGracefulClose() {
78+
StdioAcpClientTransport transport = new StdioAcpClientTransport(javaAgent(EndOfInputAgent.class));
79+
transport.setStdErrorHandler(line -> {
80+
});
81+
AcpSyncClient client = AcpClient.sync(transport).requestTimeout(TIMEOUT).build();
82+
client.initialize();
83+
client.newSession(new AcpSchema.NewSessionRequest("/tmp", List.of()));
84+
client.prompt(new AcpSchema.PromptRequest("s-1", List.of(new AcpSchema.TextContent("hi"))));
85+
86+
assertThat(client.closeGracefully()).isTrue();
87+
88+
assertThat(stops()).containsExactly("ACP agent process stopped (exit code 0)");
89+
}
90+
91+
/** An agent that goes on running after its input ends is sent SIGTERM. */
92+
@Test
93+
@DisabledOnOs(OS.WINDOWS)
94+
void anAgentThatIgnoresTheEndOfItsInputIsTerminated() {
95+
StdioAcpClientTransport transport = new StdioAcpClientTransport(javaAgent(StubbornAgent.class));
96+
transport.connect(message -> message.then(Mono.empty())).block(TIMEOUT);
97+
98+
transport.closeGracefully().block(TIMEOUT);
99+
100+
assertThat(stops()).containsExactly("ACP agent process stopped (exit code 143)");
101+
}
102+
103+
/** {@code closeGracefully()} then {@code close()}, as try-with-resources does: one report. */
104+
@Test
105+
void closingTwiceReportsTheStopOnce() {
106+
StdioAcpClientTransport transport = new StdioAcpClientTransport(javaAgent(EndOfInputAgent.class));
107+
transport.setStdErrorHandler(line -> {
108+
});
109+
AcpSyncClient client = AcpClient.sync(transport).requestTimeout(TIMEOUT).build();
110+
client.initialize();
111+
112+
client.closeGracefully();
113+
client.close();
114+
transport.closeGracefully().block(TIMEOUT);
115+
116+
assertThat(stops()).hasSize(1);
117+
}
118+
119+
/** Never reads its input, and never exits by itself. */
120+
public static final class StubbornAgent {
121+
122+
private StubbornAgent() {
123+
}
124+
125+
public static void main(String[] args) throws InterruptedException {
126+
Thread.sleep(Long.MAX_VALUE);
127+
}
128+
129+
}
130+
131+
}

0 commit comments

Comments
 (0)