Skip to content

Commit 3dad2b6

Browse files
author
Mark Pollack
committed
Report the end of the stdio agent process to the client session
StdioAcpClientTransport did not implement awaitTermination(), whose interface default never completes, so when the agent process exited or crashed, AcpClientSession left pending requests to wait out the request timeout. Complete it when the transport is closed, and fail it with an AcpConnectionException naming the exit code (and the signal, 128 + n on POSIX) once the agent's standard output has ended and the process has exited on its own.
1 parent 1d95932 commit 3dad2b6

3 files changed

Lines changed: 157 additions & 2 deletions

File tree

‎CHANGELOG.md‎

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

152152
### Fixed
153153

154+
- **A stdio client did not notice its agent process exiting.** `StdioAcpClientTransport` did not
155+
implement `awaitTermination()`, so when the agent exited or crashed, pending requests waited out the
156+
request timeout. It now completes when the transport is closed, and errors with an
157+
`AcpConnectionException` such as `ACP agent process exited with code 137 (signal 9)` once the
158+
agent's output has ended and the process has exited on its own; the session fails pending requests
159+
with it at once.
154160
- **A peer's error answer to a message it could not read was logged as an SDK bug.** An error
155161
response with `"id": null` (what JSON-RPC 2.0 prescribes, and what this SDK and its peers send, for
156162
a message whose id could not be read) was logged at ERROR as "a bug in the request sender code".

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

Lines changed: 72 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import java.util.function.Function;
1919
import java.util.function.Predicate;
2020

21+
import com.agentclientprotocol.sdk.error.AcpConnectionException;
2122
import com.agentclientprotocol.sdk.json.AcpJsonMapper;
2223
import com.agentclientprotocol.sdk.json.TypeRef;
2324
import com.agentclientprotocol.sdk.spec.AcpClientTransport;
@@ -82,13 +83,25 @@ public class StdioAcpClientTransport implements AcpClientTransport {
8283

8384
private volatile boolean isClosing = false;
8485

86+
/** Set by {@link #closeGracefully}: the end of the agent process is then expected. */
87+
private volatile boolean closedLocally = false;
88+
89+
/**
90+
* Terminates when the transport can no longer deliver: empty once closed locally, with
91+
* an error naming the exit when the agent process ended on its own.
92+
*/
93+
private final Sinks.One<Void> terminationSink = Sinks.one();
94+
8595
/**
8696
* A transport instance carries exactly one session. {@link #connect} subscribes the
8797
* unicast inbound and outbound sinks and starts the agent process; a second call would
8898
* subscribe them again and start a second process, so it is refused up front.
8999
*/
90100
private final AtomicBoolean isConnected = new AtomicBoolean(false);
91101

102+
/** How long to wait for the agent process to exit after its standard output ends. */
103+
private static final Duration EXIT_WAIT = Duration.ofSeconds(5);
104+
92105
private volatile Consumer<Throwable> exceptionHandler = t -> logger.error("Transport error", t);
93106

94107
// visible for tests
@@ -318,8 +331,52 @@ public Mono<Void> sendMessage(JSONRPCMessage message) {
318331
* process's input stream. Messages are deserialized and emitted to the inbound sink.
319332
*/
320333
private void startInboundProcessing(Process process) {
321-
readLines(this.inboundScheduler, process.getInputStream(), "input stream", this::emitInbound,
322-
this.inboundSink::tryEmitComplete);
334+
readLines(this.inboundScheduler, process.getInputStream(), "input stream", this::emitInbound, () -> {
335+
this.inboundSink.tryEmitComplete();
336+
terminated(process);
337+
});
338+
}
339+
340+
/**
341+
* The agent's standard output has ended: once the process has exited, reports the end of
342+
* the transport, naming the exit, unless it was closed locally.
343+
*/
344+
private void terminated(Process process) {
345+
if (this.closedLocally) {
346+
this.terminationSink.tryEmitEmpty();
347+
return;
348+
}
349+
String reason;
350+
try {
351+
// The agent normally exits right after its output ends; one that closed its
352+
// output but goes on running can deliver nothing more either.
353+
reason = process.waitFor(EXIT_WAIT.toMillis(), java.util.concurrent.TimeUnit.MILLISECONDS)
354+
? exitDescription(process.exitValue())
355+
: "ACP agent process closed its standard output but has not exited";
356+
}
357+
catch (InterruptedException e) {
358+
Thread.currentThread().interrupt();
359+
reason = "ACP agent process closed its standard output";
360+
}
361+
if (this.closedLocally) {
362+
this.terminationSink.tryEmitEmpty();
363+
return;
364+
}
365+
logger.info("{}", reason);
366+
this.terminationSink.tryEmitError(new AcpConnectionException(reason));
367+
}
368+
369+
/**
370+
* Names how the agent process ended. On POSIX systems Java reports a process killed by
371+
* signal {@code n} as exit code {@code 128 + n}.
372+
*/
373+
static String exitDescription(int exitCode) {
374+
String description = "ACP agent process exited with code " + exitCode;
375+
boolean windows = System.getProperty("os.name", "").startsWith("Windows");
376+
if (!windows && exitCode > 128 && exitCode < 128 + 65) {
377+
description += " (signal " + (exitCode - 128) + ")";
378+
}
379+
return description;
323380
}
324381

325382
/**
@@ -433,6 +490,7 @@ protected void handleOutbound(Function<Flux<JSONRPCMessage>, Flux<JSONRPCMessage
433490
@Override
434491
public Mono<Void> closeGracefully() {
435492
return Mono.fromRunnable(() -> {
493+
closedLocally = true;
436494
isClosing = true;
437495
logger.debug("Initiating graceful shutdown");
438496

@@ -457,6 +515,7 @@ public Mono<Void> closeGracefully() {
457515
catch (Exception e) {
458516
logger.error("Error during graceful shutdown", e);
459517
}
518+
terminationSink.tryEmitEmpty();
460519
});
461520
}
462521

@@ -493,6 +552,17 @@ private static void logExit(int exitCode) {
493552
}
494553
}
495554

555+
/**
556+
* Completes when the transport is closed; errors with an {@link AcpConnectionException}
557+
* naming the exit code (and, on POSIX systems, the signal) when the agent process ends
558+
* on its own, once its standard output has been read to the end.
559+
* @return a Mono that terminates when the transport does
560+
*/
561+
@Override
562+
public Mono<Void> awaitTermination() {
563+
return this.terminationSink.asMono();
564+
}
565+
496566
public Sinks.Many<String> getErrorSink() {
497567
return this.errorSink;
498568
}

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

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,13 +14,18 @@
1414
import java.util.concurrent.CopyOnWriteArrayList;
1515
import java.util.concurrent.CountDownLatch;
1616
import java.util.concurrent.TimeUnit;
17+
import java.util.function.Function;
1718

1819
import com.agentclientprotocol.sdk.QuietLoggers;
20+
import com.agentclientprotocol.sdk.json.TypeRef;
21+
import com.agentclientprotocol.sdk.spec.AcpClientSession;
1922
import com.agentclientprotocol.sdk.spec.AcpSchema;
2023
import com.agentclientprotocol.sdk.spec.AcpSchema.JSONRPCMessage;
2124
import com.agentclientprotocol.sdk.spec.AcpSchema.JSONRPCNotification;
2225
import org.junit.jupiter.api.AfterEach;
2326
import org.junit.jupiter.api.Test;
27+
import org.junit.jupiter.api.condition.DisabledOnOs;
28+
import org.junit.jupiter.api.condition.OS;
2429
import reactor.core.publisher.Mono;
2530

2631
import static org.assertj.core.api.Assertions.assertThat;
@@ -131,6 +136,65 @@ void awaitForExitReturnsWhenTheAgentExits() {
131136
Mono.fromRunnable(transport::awaitForExit).block(TIMEOUT);
132137
}
133138

139+
/**
140+
* The agent exits while a request is waiting: the request fails at once, naming the
141+
* exit code, instead of waiting out the request timeout; closing still works.
142+
*/
143+
@Test
144+
void anAgentThatExitsMidRequestFailsThePendingRequestPromptly() {
145+
transport = new StdioAcpClientTransport(echoAgent());
146+
transport.setStdErrorHandler(line -> {
147+
});
148+
Duration requestTimeout = Duration.ofMinutes(5);
149+
AcpClientSession session = new AcpClientSession(requestTimeout, transport, Map.of(), Map.of(),
150+
Function.identity());
151+
152+
long start = System.nanoTime();
153+
try (QuietLoggers quiet = QuietLoggers.of(StdioAcpClientTransport.class, AcpClientSession.class)) {
154+
assertThatThrownBy(() -> session.sendRequest(EchoAgent.EXIT_3, Map.of(), new TypeRef<Map<String, Object>>() {
155+
}).block(TIMEOUT)).hasMessageContaining("terminated")
156+
.rootCause()
157+
.hasMessageContaining("ACP agent process exited with code 3");
158+
}
159+
assertThat(Duration.ofNanos(System.nanoTime() - start)).isLessThan(Duration.ofSeconds(10));
160+
161+
session.closeGracefully().block(TIMEOUT);
162+
}
163+
164+
/** An agent killed by a signal: the reason names the signal Java reports as 128 + n. */
165+
@Test
166+
@DisabledOnOs(OS.WINDOWS)
167+
void terminationNamesTheSignalThatKilledTheAgent() {
168+
transport = new StdioAcpClientTransport(echoAgent());
169+
transport.setStdErrorHandler(line -> {
170+
});
171+
transport.connect(message -> message.then(Mono.empty())).block(TIMEOUT);
172+
173+
try (QuietLoggers quiet = QuietLoggers.of(StdioAcpClientTransport.class)) {
174+
transport.sendMessage(new JSONRPCNotification(EchoAgent.KILL, null)).block(TIMEOUT);
175+
176+
assertThatThrownBy(() -> transport.awaitTermination().block(TIMEOUT))
177+
.hasMessageContaining("ACP agent process exited with code 137 (signal 9)");
178+
}
179+
}
180+
181+
/** Closing the transport ends {@code awaitTermination} without an error. */
182+
@Test
183+
void closingTheTransportCompletesAwaitTermination() {
184+
transport = new StdioAcpClientTransport(echoAgent());
185+
transport.setStdErrorHandler(line -> {
186+
});
187+
transport.connect(message -> message.then(Mono.empty())).block(TIMEOUT);
188+
Mono<Void> termination = transport.awaitTermination().cache();
189+
termination.subscribe(v -> {
190+
}, e -> {
191+
});
192+
193+
transport.closeGracefully().block(TIMEOUT);
194+
195+
termination.block(TIMEOUT);
196+
}
197+
134198
@Test
135199
void anAgentThatCannotBeStartedFailsTheConnect() {
136200
transport = new StdioAcpClientTransport(AgentParameters.builder("/nonexistent/acp-agent-binary").build());
@@ -147,6 +211,12 @@ public static final class EchoAgent {
147211

148212
static final String EXIT = "test/exit";
149213

214+
/** Exits with status 3 instead of echoing. */
215+
static final String EXIT_3 = "test/exit3";
216+
217+
/** Kills itself with SIGKILL instead of echoing. */
218+
static final String KILL = "test/kill";
219+
150220
/** Answered with a line that is not JSON instead of its echo. */
151221
static final String MALFORMED = "test/malformed";
152222

@@ -162,6 +232,15 @@ public static void main(String[] args) throws Exception {
162232
if (line.contains("\"" + EXIT + "\"")) {
163233
return;
164234
}
235+
if (line.contains("\"" + EXIT_3 + "\"")) {
236+
System.exit(3);
237+
}
238+
if (line.contains("\"" + KILL + "\"")) {
239+
// A process may not destroy itself through ProcessHandle; ask kill(1).
240+
new ProcessBuilder("kill", "-KILL", Long.toString(ProcessHandle.current().pid())).start()
241+
.waitFor();
242+
Thread.sleep(Long.MAX_VALUE);
243+
}
165244
String echo = line.contains("\"" + MALFORMED + "\"") ? "{not json" : line;
166245
System.out.write((echo + "\n").getBytes(StandardCharsets.UTF_8));
167246
System.out.flush();

0 commit comments

Comments
 (0)