Skip to content

Commit fcd98f1

Browse files
committed
fix(production): 生产就绪深审修复——ACP 生命周期与 server 启动健壮性
深审新代码定位 4 处缺陷并修复: 1. ACP connect() 握手失败不再泄露子进程——失败路径 destroy 并重置 connected,允许同 client 重试 2. ACP 双重 connect() 加 CAS 守卫——原实现会覆盖 process/stdin 引用 导致第一个子进程成为孤儿 3. KimiServerClient.start() 处理 kimi web 端口被占自动 +1 重试—— drainer 捕获子进程输出,健康循环失败时从输出提取实际 base URL 4. KimiCli.RunOptions 校验 --session 与 --continue 互斥(原会组装出 CLI 启动即拒的参数,错误延迟暴露) 测试 66 个全绿(+3:双 connect 拒绝、失败后可重试不泄露、互斥校验)
1 parent f77e2b6 commit fcd98f1

5 files changed

Lines changed: 95 additions & 10 deletions

File tree

‎src/main/java/io/github/easy4j/kimi/acp/KimiAcpClient.java‎

Lines changed: 29 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ public class KimiAcpClient implements AutoCloseable {
7979
private final Map<String, PromptStream> promptStreams = new ConcurrentHashMap<String, PromptStream>();
8080
private final AtomicLong rpcIds = new AtomicLong();
8181
private final AtomicBoolean closed = new AtomicBoolean(false);
82+
private final AtomicBoolean connected = new AtomicBoolean(false);
8283
private final ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(r -> {
8384
Thread thread = new Thread(r, "kimi-acp-timer");
8485
thread.setDaemon(true);
@@ -109,6 +110,13 @@ public KimiAcpClient(KimiAcpConfig config) {
109110
* fails or times out.
110111
*/
111112
public String connect() {
113+
if (closed.get()) {
114+
throw new IllegalStateException("kimi acp client is closed");
115+
}
116+
// CAS guard: a second connect would orphan the first child process.
117+
if (!connected.compareAndSet(false, true)) {
118+
throw new IllegalStateException("kimi acp client is already connected");
119+
}
112120
List<String> command = new ArrayList<String>();
113121
command.add(config.getLocalExecutable());
114122
if (config.getAcpSubcommand() != null) {
@@ -124,6 +132,7 @@ public String connect() {
124132
try {
125133
process = builder.start();
126134
} catch (IOException e) {
135+
connected.set(false);
127136
throw new KimiException("Failed to spawn kimi acp: " + config.getLocalExecutable(), e);
128137
}
129138
stdin = new PrintWriter(new OutputStreamWriter(process.getOutputStream(), StandardCharsets.UTF_8), true);
@@ -135,14 +144,27 @@ public String connect() {
135144
params.put("protocolVersion", Integer.valueOf(1));
136145
Map<String, Object> clientCaps = new LinkedHashMap<String, Object>();
137146
params.put("clientCapabilities", clientCaps);
138-
JsonNode result = await(request("initialize", params), config.getConnectTimeoutMillis(), "initialize");
139-
if (result.hasNonNull("protocolVersion")) {
140-
protocolVersion = result.path("protocolVersion").asText(null);
141-
}
142-
if (result.hasNonNull("agentInfo")) {
143-
agentVersion = result.path("agentInfo").path("version").asText(null);
147+
try {
148+
JsonNode result = await(request("initialize", params), config.getConnectTimeoutMillis(), "initialize");
149+
if (result.hasNonNull("protocolVersion")) {
150+
protocolVersion = result.path("protocolVersion").asText(null);
151+
}
152+
if (result.hasNonNull("agentInfo")) {
153+
agentVersion = result.path("agentInfo").path("version").asText(null);
154+
}
155+
return agentVersion;
156+
} catch (RuntimeException e) {
157+
// Handshake failure leaves the child alive — destroy it here so a
158+
// discarded client cannot leak the process, and allow a retry.
159+
Process current = process;
160+
if (current != null) {
161+
current.destroy();
162+
}
163+
process = null;
164+
stdin = null;
165+
connected.set(false);
166+
throw e;
144167
}
145-
return agentVersion;
146168
}
147169

148170
/**

‎src/main/java/io/github/easy4j/kimi/cli/KimiCli.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -253,6 +253,12 @@ public KimiCliResult prompt(RunOptions options) {
253253
}
254254

255255
private String[] buildPromptArgs(RunOptions options) {
256+
if (options.session != null && options.continueLast) {
257+
// The kimi CLI rejects this combination at startup; fail here
258+
// with a clear message instead.
259+
throw new IllegalArgumentException(
260+
"RunOptions: --session and --continue are mutually exclusive");
261+
}
256262
List<String> args = new ArrayList<String>();
257263
if (options.model != null) {
258264
args.add("--model");

‎src/main/java/io/github/easy4j/kimi/server/KimiServerClient.java‎

Lines changed: 37 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -75,8 +75,12 @@ public class KimiServerClient implements AutoCloseable {
7575
JsonMapper.builder().disable(tools.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
7676
.build();
7777

78+
private static final java.util.regex.Pattern HTTP_URL =
79+
java.util.regex.Pattern.compile("https?://\\S+");
80+
7881
private final KimiServerConfig config;
7982
private final AtomicBoolean ownsServer = new AtomicBoolean(false);
83+
private final java.util.List<String> startupOutput = new java.util.ArrayList<String>();
8084

8185
private volatile Process process;
8286
private volatile String baseUrl;
@@ -119,12 +123,20 @@ public String start() {
119123
throw new KimiException("Failed to spawn kimi web: " + config.getLocalExecutable(), e);
120124
}
121125
ownsServer.set(true);
122-
// Drain child output so the process never blocks on a full pipe.
126+
// Drain child output so the process never blocks on a full pipe, and
127+
// keep the last lines: when the configured port is busy the CLI
128+
// retries on port+1 and announces the real URL on stdout.
123129
Thread drainer = new Thread(() -> {
124130
try (BufferedReader reader =
125131
new BufferedReader(new InputStreamReader(process.getInputStream(), StandardCharsets.UTF_8))) {
126-
while (reader.readLine() != null) {
127-
// discard
132+
String line;
133+
while ((line = reader.readLine()) != null) {
134+
synchronized (startupOutput) {
135+
startupOutput.add(line);
136+
if (startupOutput.size() > 100) {
137+
startupOutput.remove(0);
138+
}
139+
}
128140
}
129141
} catch (IOException e) {
130142
log.debug("kimi web output drain ended: {}", e.getMessage());
@@ -147,6 +159,7 @@ public String start() {
147159
return baseUrl;
148160
} catch (Exception e) {
149161
lastError = e;
162+
candidate = detectBaseUrl(candidate);
150163
sleepQuietly(300);
151164
}
152165
}
@@ -372,6 +385,27 @@ private JsonNode exchange(String base, String method, String path, Object body)
372385
}
373386
}
374387

388+
/**
389+
* Returns the first {@code http://...} URL announced on the child's
390+
* stdout that differs from {@code current}, or {@code current} when none
391+
* was found. Handles the CLI's busy-port {@code +1} retry behaviour.
392+
*/
393+
private String detectBaseUrl(String current) {
394+
synchronized (startupOutput) {
395+
for (String line : startupOutput) {
396+
java.util.regex.Matcher matcher = HTTP_URL.matcher(line);
397+
if (matcher.find()) {
398+
String url = matcher.group().replaceAll("/+$", "");
399+
if (!url.equals(current)) {
400+
log.debug("kimi web announced actual base url: {}", url);
401+
return url;
402+
}
403+
}
404+
}
405+
}
406+
return current;
407+
}
408+
375409
private String readToken() {
376410
String explicit = config.getToken();
377411
if (explicit != null && !explicit.trim().isEmpty()) {

‎src/test/java/io/github/easy4j/kimi/acp/KimiAcpClientE2ETest.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,23 @@ void shouldFailConnectWhenAgentExitsPrematurely() {
110110
client.close();
111111
}
112112

113+
@Test
114+
void shouldRejectDoubleConnect() {
115+
try (KimiAcpClient client = new KimiAcpClient(config())) {
116+
client.connect();
117+
assertThrows(IllegalStateException.class, client::connect);
118+
}
119+
}
120+
121+
@Test
122+
void shouldAllowRetryAfterFailedConnectWithoutLeakingProcess() {
123+
KimiAcpConfig config = config();
124+
config.setLocalExecutable("/nonexistent/kimi");
125+
KimiAcpClient client = new KimiAcpClient(config);
126+
assertThrows(KimiException.class, client::connect);
127+
assertFalse(client.isClosed(), "failed connect must leave the client open for a retry");
128+
}
129+
113130
@Test
114131
void shouldRejectUseAfterClose() {
115132
KimiAcpClient client = new KimiAcpClient(config());

‎src/test/java/io/github/easy4j/kimi/cli/KimiCliTest.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,12 @@ void shouldDelegatePromptContinueLast() {
114114
assertTrue(out.contains("--prompt hi"));
115115
}
116116

117+
@Test
118+
void shouldRejectSessionAndContinueTogether() {
119+
KimiCli.RunOptions options = new KimiCli.RunOptions("hi", null).session("s1").continueLast(true);
120+
assertThrows(IllegalArgumentException.class, () -> echoCli().prompt(options));
121+
}
122+
117123
@Test
118124
void shouldRejectBlankPrompt() {
119125
assertThrows(IllegalArgumentException.class, () -> echoCli().prompt(" "));

0 commit comments

Comments
 (0)