From 6cbfb528578fd376410083c3a0dd58350b4cf9a2 Mon Sep 17 00:00:00 2001 From: Loong Wan Date: Sun, 20 Sep 2026 23:51:23 +0800 Subject: [PATCH 1/4] fix(acp): port lifecycle hardening to 1.0.x --- .../github/easy4j/kimi/acp/KimiAcpClient.java | 124 ++++++++++++++++-- 1 file changed, 112 insertions(+), 12 deletions(-) diff --git a/src/main/java/io/github/easy4j/kimi/acp/KimiAcpClient.java b/src/main/java/io/github/easy4j/kimi/acp/KimiAcpClient.java index 1d7037f..84c2585 100644 --- a/src/main/java/io/github/easy4j/kimi/acp/KimiAcpClient.java +++ b/src/main/java/io/github/easy4j/kimi/acp/KimiAcpClient.java @@ -34,6 +34,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import org.slf4j.Logger; @@ -80,6 +81,8 @@ public class KimiAcpClient implements AutoCloseable { private final AtomicLong rpcIds = new AtomicLong(); private final AtomicBoolean closed = new AtomicBoolean(false); private final AtomicBoolean connected = new AtomicBoolean(false); + private final AtomicReference state = + new AtomicReference(KimiAcpState.NEW); private final ScheduledExecutorService timer = Executors.newSingleThreadScheduledExecutor(r -> { Thread thread = new Thread(r, "kimi-acp-timer"); thread.setDaemon(true); @@ -117,6 +120,7 @@ public String connect() { if (!connected.compareAndSet(false, true)) { throw new IllegalStateException("kimi acp client is already connected"); } + state.set(KimiAcpState.CONNECTING); List command = new ArrayList(); command.add(config.getLocalExecutable()); if (config.getAcpSubcommand() != null) { @@ -131,8 +135,10 @@ public String connect() { builder.redirectErrorStream(false); try { process = builder.start(); + state.set(KimiAcpState.INITIALIZING); } catch (IOException e) { connected.set(false); + state.set(KimiAcpState.NEW); throw new KimiException("Failed to spawn kimi acp: " + config.getLocalExecutable(), e); } stdin = new PrintWriter(new OutputStreamWriter(process.getOutputStream(), StandardCharsets.UTF_8), true); @@ -152,6 +158,7 @@ public String connect() { if (result.hasNonNull("agentInfo")) { agentVersion = result.path("agentInfo").path("version").asText(null); } + state.set(KimiAcpState.READY); return agentVersion; } catch (RuntimeException e) { // Handshake failure leaves the child alive — destroy it here so a @@ -163,6 +170,7 @@ public String connect() { process = null; stdin = null; connected.set(false); + state.set(KimiAcpState.NEW); throw e; } } @@ -299,7 +307,10 @@ public CompletableFuture promptAsync(String sessionId, String throw new IllegalStateException("kimi acp client is closed"); } PromptStream stream = new PromptStream(sessionId, onDelta); - promptStreams.put(sessionId, stream); + PromptStream active = promptStreams.putIfAbsent(sessionId, stream); + if (active != null) { + throw new KimiException("kimi acp session already has an active prompt: " + sessionId); + } Map content = new LinkedHashMap(); content.put("type", "text"); content.put("text", text); @@ -308,14 +319,25 @@ public CompletableFuture promptAsync(String sessionId, String Map params = new LinkedHashMap(); params.put("sessionId", sessionId); params.put("prompt", blocks); - CompletableFuture response = request("session/prompt", params); + final CompletableFuture response; + try { + response = request("session/prompt", params); + } catch (RuntimeException e) { + promptStreams.remove(sessionId, stream); + throw e; + } CompletableFuture future = response.thenApply(node -> { + RuntimeException callbackFailure = stream.callbackFailure(); + if (callbackFailure != null) { + throw new KimiException("kimi acp prompt callback failed", callbackFailure); + } String stopReason = firstText(node, "stopReason", "stop_reason"); return new KimiAcpTurnResult(sessionId, stopReason, stream.content()); }); + stream.bind(future); scheduleTimeout(future, config.getReadTimeoutMillis(), "session/prompt turn"); future.whenComplete((r, error) -> { - promptStreams.remove(sessionId); + promptStreams.remove(sessionId, stream); if (error != null) { response.completeExceptionally(error); } @@ -359,6 +381,10 @@ public void cancel(String sessionId) { Map params = new LinkedHashMap(); params.put("sessionId", sessionId); notify("session/cancel", params); + PromptStream stream = promptStreams.get(sessionId); + if (stream != null) { + stream.cancel(); + } } /** @@ -418,6 +444,15 @@ public boolean isClosed() { return closed.get(); } + /** + * Returns the current ACP lifecycle state. + * + * @return the lifecycle state; never {@code null}. + */ + public KimiAcpState getState() { + return state.get(); + } + /** * Terminates the {@code kimi acp} child process and releases the timer. * Idempotent. @@ -427,12 +462,21 @@ public void close() { if (!closed.compareAndSet(false, true)) { return; } + state.set(KimiAcpState.CLOSING); + connected.set(false); timer.shutdownNow(); + PrintWriter writer = stdin; + stdin = null; + if (writer != null) { + writer.close(); + } Process current = process; + process = null; if (current != null) { current.destroy(); } failAllPending(new KimiException("kimi acp client closed")); + state.set(KimiAcpState.CLOSED); } // ============================================================ @@ -440,6 +484,10 @@ public void close() { // ============================================================ private CompletableFuture request(String method, Map params) { + KimiAcpState currentState = state.get(); + if (!"initialize".equals(method) && currentState != KimiAcpState.READY) { + throw new KimiException("kimi acp client is not ready: state=" + currentState); + } long id = rpcIds.incrementAndGet(); Map payload = new LinkedHashMap(); payload.put("jsonrpc", "2.0"); @@ -447,8 +495,16 @@ private CompletableFuture request(String method, Map p payload.put("method", method); payload.put("params", params); CompletableFuture future = new CompletableFuture(); - pendingRpcs.put(Long.valueOf(id), future); - writeJson(payload); + Long rpcId = Long.valueOf(id); + pendingRpcs.put(rpcId, future); + future.whenComplete((result, error) -> pendingRpcs.remove(rpcId, future)); + try { + writeJson(payload); + } catch (RuntimeException e) { + pendingRpcs.remove(rpcId, future); + future.completeExceptionally(e); + throw e; + } return future; } @@ -497,17 +553,29 @@ private void readLoop() { KimiException error = new KimiException( "kimi acp frame exceeded maxFrameChars=" + config.getMaxFrameChars()); log.warn("kimi acp frame over cap, tearing transport down"); - failAllPending(error); + failTransport(error); process.destroy(); return; } handleFrame(line); } - failAllPending(new KimiException("kimi acp stdout closed (child exited)")); + if (!closed.get()) { + failTransport(new KimiException("kimi acp stdout closed (child exited)")); + } } catch (IOException e) { if (!closed.get()) { - failAllPending(new KimiException("kimi acp stdout read failed", e)); + failTransport(new KimiException("kimi acp stdout read failed", e)); + } + } catch (KimiException e) { + if (!closed.get()) { + failTransport(e); + Process current = process; + if (current != null) { + current.destroy(); + } } + } finally { + connected.set(false); } } @@ -516,8 +584,7 @@ private void handleFrame(String frame) { try { node = mapper.readTree(frame); } catch (Exception ex) { - log.warn("Ignored non-JSON frame from kimi acp"); - return; + throw new KimiException("kimi acp protocol received invalid JSON frame", ex); } if (node.hasNonNull("id")) { CompletableFuture pending = pendingRpcs.remove(Long.valueOf(node.get("id").asLong())); @@ -589,6 +656,14 @@ private JsonNode await(CompletableFuture future, long timeoutMillis, S } } + private void failTransport(KimiException error) { + if (!closed.get()) { + state.set(KimiAcpState.FAILED); + } + connected.set(false); + failAllPending(error); + } + private void failAllPending(KimiException error) { for (Map.Entry> entry : pendingRpcs.entrySet()) { CompletableFuture future = pendingRpcs.remove(entry.getKey()); @@ -628,14 +703,30 @@ private String firstText(JsonNode node, String... fields) { */ private final class PromptStream { + private final String sessionId; private final StringBuilder content = new StringBuilder(); private final Consumer onDelta; private boolean truncationWarned; + private volatile RuntimeException callbackFailure; + private volatile CompletableFuture turnFuture; PromptStream(String sessionId, Consumer onDelta) { + this.sessionId = sessionId; this.onDelta = onDelta; } + void bind(CompletableFuture future) { + this.turnFuture = future; + } + + void cancel() { + CompletableFuture future = turnFuture; + if (future != null) { + future.completeExceptionally( + new KimiException("kimi acp prompt cancelled: " + sessionId)); + } + } + void append(String text) { int cap = config.getMaxContentChars(); String applied = text; @@ -653,11 +744,20 @@ void append(String text) { } } content.append(applied); - if (!applied.isEmpty() && onDelta != null) { - onDelta.accept(applied); + if (!applied.isEmpty() && onDelta != null && callbackFailure == null) { + try { + onDelta.accept(applied); + } catch (RuntimeException e) { + callbackFailure = e; + log.warn("kimi acp prompt callback failed; transport remains active"); + } } } + RuntimeException callbackFailure() { + return callbackFailure; + } + String content() { return content.toString(); } From 2993fc481dcb5be38985f8a775a9fa4312c05b24 Mon Sep 17 00:00:00 2001 From: Loong Wan Date: Sun, 20 Sep 2026 23:51:34 +0800 Subject: [PATCH 2/4] feat(acp): add lifecycle state to 1.0.x --- .../github/easy4j/kimi/acp/KimiAcpState.java | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) create mode 100644 src/main/java/io/github/easy4j/kimi/acp/KimiAcpState.java diff --git a/src/main/java/io/github/easy4j/kimi/acp/KimiAcpState.java b/src/main/java/io/github/easy4j/kimi/acp/KimiAcpState.java new file mode 100644 index 0000000..a3460ec --- /dev/null +++ b/src/main/java/io/github/easy4j/kimi/acp/KimiAcpState.java @@ -0,0 +1,22 @@ +/* + * Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java). + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + */ +package io.github.easy4j.kimi.acp; + +/** + * Observable lifecycle states of a {@link KimiAcpClient}. + * + * @since 1.0.x + */ +public enum KimiAcpState { + NEW, + CONNECTING, + INITIALIZING, + READY, + CLOSING, + CLOSED, + FAILED +} From 2c3495d2c9f0b10f8c010538c59f5457ce47e754 Mon Sep 17 00:00:00 2001 From: Loong Wan Date: Sun, 20 Sep 2026 23:51:36 +0800 Subject: [PATCH 3/4] test(acp): sync lifecycle hardening cases to 1.0.x --- .../acp/KimiAcpLifecycleHardeningTest.java | 234 ++++++++++++++++++ 1 file changed, 234 insertions(+) create mode 100644 src/test/java/io/github/easy4j/kimi/acp/KimiAcpLifecycleHardeningTest.java diff --git a/src/test/java/io/github/easy4j/kimi/acp/KimiAcpLifecycleHardeningTest.java b/src/test/java/io/github/easy4j/kimi/acp/KimiAcpLifecycleHardeningTest.java new file mode 100644 index 0000000..185b2c8 --- /dev/null +++ b/src/test/java/io/github/easy4j/kimi/acp/KimiAcpLifecycleHardeningTest.java @@ -0,0 +1,234 @@ +/* + * Copyright (c) 2018-present, easy-4-java (https://github.com/easy-4-java). + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + */ +package io.github.easy4j.kimi.acp; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import io.github.easy4j.kimi.KimiException; + +/** + * RED tests for the ACP lifecycle hardening OpenSpec change. + * + *

These tests intentionally target failure/race/resource cases that the + * current implementation does not yet satisfy. They must be committed before + * the production fix so CI provides evidence for the TDD RED phase.

+ */ +class KimiAcpLifecycleHardeningTest { + + private static final String FAKE_AGENT = Paths + .get("src", "test", "resources", "fake-acp-agent.py") + .toAbsolutePath().toString(); + + private static KimiAcpConfig config(String mode) { + KimiAcpConfig config = new KimiAcpConfig(); + config.setLocalExecutable("python3"); + config.setAcpSubcommand(null); + config.setAcpArgs(new String[] {FAKE_AGENT, mode}); + config.setConnectTimeoutMillis(2_000); + config.setReadTimeoutMillis(5_000); + return config; + } + + @Test + void shouldRemovePendingRpcWhenRequestCannotWrite() throws Exception { + try (KimiAcpClient client = new KimiAcpClient(config("normal"))) { + assertThrows(KimiException.class, () -> client.newSession("/tmp")); + assertEquals(0, privateMapSize(client, "pendingRpcs"), + "failed write before connect must not leave an orphaned RPC"); + } + } + + @Test + void shouldRejectSecondPromptForSameSessionBeforeSending() throws Exception { + try (KimiAcpClient client = new KimiAcpClient(config("delay-prompt"))) { + client.connect(); + String sessionId = client.newSession("/tmp"); + + CompletableFuture first = + client.promptAsync(sessionId, "first", null); + + assertThrows(KimiException.class, + () -> client.promptAsync(sessionId, "second", null), + "same session must not overwrite an active prompt stream"); + + client.cancel(sessionId); + first.cancel(true); + } + } + + @Test + void shouldIsolateThrowingDeltaCallbackAndKeepTransportUsable() throws Exception { + try (KimiAcpClient client = new KimiAcpClient(config("normal"))) { + client.connect(); + String sessionId = client.newSession("/tmp"); + + CompletableFuture first = + client.promptAsync(sessionId, "callback-fails", delta -> { + throw new IllegalStateException("listener boom"); + }); + + ExecutionException failure = assertThrows(ExecutionException.class, + () -> first.get(2, TimeUnit.SECONDS)); + assertTrue(failure.getCause() instanceof KimiException); + assertTrue(failure.getCause().getMessage().contains("callback")); + + List deltas = new ArrayList(); + KimiAcpTurnResult second = client.prompt("sess_after_callback", "still-alive", deltas::add); + assertEquals("你好世界", second.getContent()); + assertEquals(2, deltas.size()); + } + } + + @Test + void shouldFailPendingPromptOnMalformedJsonFrameInsteadOfTimingOut() throws Exception { + try (KimiAcpClient client = new KimiAcpClient(config("malformed-prompt"))) { + client.connect(); + String sessionId = client.newSession("/tmp"); + + CompletableFuture future = + client.promptAsync(sessionId, "bad-frame", null); + + ExecutionException failure = assertThrows(ExecutionException.class, + () -> future.get(2, TimeUnit.SECONDS)); + assertTrue(failure.getCause() instanceof KimiException); + assertTrue(failure.getCause().getMessage().toLowerCase().contains("json") + || failure.getCause().getMessage().toLowerCase().contains("protocol"), + "malformed ACP frame must become a protocol failure"); + } + } + + @Test + void shouldFailPendingPromptWhenAgentProcessExits() throws Exception { + try (KimiAcpClient client = new KimiAcpClient(config("exit-on-prompt"))) { + client.connect(); + String sessionId = client.newSession("/tmp"); + + CompletableFuture future = + client.promptAsync(sessionId, "exit", null); + + ExecutionException failure = assertThrows(ExecutionException.class, + () -> future.get(2, TimeUnit.SECONDS)); + assertTrue(failure.getCause() instanceof KimiException); + } + } + + + @Test + void shouldCompletePromptAsCancelledAndReleaseRegistries() throws Exception { + try (KimiAcpClient client = new KimiAcpClient(config("delay-prompt"))) { + client.connect(); + String sessionId = client.newSession("/tmp"); + + CompletableFuture future = + client.promptAsync(sessionId, "cancel-me", null); + client.cancel(sessionId); + + ExecutionException failure = assertThrows(ExecutionException.class, + () -> future.get(1, TimeUnit.SECONDS)); + assertTrue(failure.getCause() instanceof KimiException); + assertTrue(failure.getCause().getMessage().toLowerCase().contains("cancel")); + assertEquals(0, privateMapSize(client, "pendingRpcs")); + assertEquals(0, privateMapSize(client, "promptStreams")); + } + } + + @Test + void shouldRemoveTimedOutRpcFromPendingRegistry() throws Exception { + KimiAcpConfig config = config("hang-list"); + config.setConnectTimeoutMillis(250); + try (KimiAcpClient client = new KimiAcpClient(config)) { + client.connect(); + + assertThrows(KimiException.class, client::listSessions); + assertEquals(0, privateMapSize(client, "pendingRpcs"), + "timed out RPC must be removed from the pending registry"); + } + } + + @Test + void shouldFailAndCleanPendingPromptWhenClientCloses() throws Exception { + KimiAcpClient client = new KimiAcpClient(config("delay-prompt")); + client.connect(); + String sessionId = client.newSession("/tmp"); + CompletableFuture future = + client.promptAsync(sessionId, "close-me", null); + + client.close(); + + assertThrows(ExecutionException.class, () -> future.get(1, TimeUnit.SECONDS)); + assertEquals(0, privateMapSize(client, "pendingRpcs")); + assertEquals(0, privateMapSize(client, "promptStreams")); + client.close(); + } + + + @Test + void shouldExposeLifecycleStateTransitions() throws Exception { + KimiAcpClient client = new KimiAcpClient(config("normal")); + assertEquals("NEW", lifecycleState(client)); + + client.connect(); + assertEquals("READY", lifecycleState(client)); + + client.close(); + assertEquals("CLOSED", lifecycleState(client)); + } + + @Test + void shouldMarkTransportFailedAndRejectNewRpcAfterMalformedFrame() throws Exception { + try (KimiAcpClient client = new KimiAcpClient(config("malformed-prompt"))) { + client.connect(); + String sessionId = client.newSession("/tmp"); + + CompletableFuture future = + client.promptAsync(sessionId, "bad-frame-state", null); + assertThrows(ExecutionException.class, () -> future.get(2, TimeUnit.SECONDS)); + + assertEquals("FAILED", lifecycleState(client)); + assertThrows(KimiException.class, client::listSessions, + "a failed transport must reject new RPCs before registration"); + assertEquals(0, privateMapSize(client, "pendingRpcs")); + } + } + + @Test + void shouldReturnToNewAfterRecoverableConnectFailure() throws Exception { + KimiAcpConfig bad = config("normal"); + bad.setLocalExecutable("/nonexistent/kimi"); + try (KimiAcpClient client = new KimiAcpClient(bad)) { + assertThrows(KimiException.class, client::connect); + assertEquals("NEW", lifecycleState(client), + "spawn/initialize failure remains retryable on the same client"); + } + } + + private static String lifecycleState(KimiAcpClient client) throws Exception { + Method method = KimiAcpClient.class.getMethod("getState"); + return String.valueOf(method.invoke(client)); + } + + @SuppressWarnings("unchecked") + private static int privateMapSize(KimiAcpClient client, String fieldName) throws Exception { + Field field = KimiAcpClient.class.getDeclaredField(fieldName); + field.setAccessible(true); + return ((Map) field.get(client)).size(); + } +} From 2606afe39e882aaed2f6f2e1f22bc52c35c4def5 Mon Sep 17 00:00:00 2001 From: Loong Wan Date: Sun, 20 Sep 2026 23:51:39 +0800 Subject: [PATCH 4/4] test(acp): sync fake lifecycle modes to 1.0.x --- src/test/resources/fake-acp-agent.py | 68 ++++++++++++++++++---------- 1 file changed, 44 insertions(+), 24 deletions(-) diff --git a/src/test/resources/fake-acp-agent.py b/src/test/resources/fake-acp-agent.py index 281ba19..f8b53e4 100755 --- a/src/test/resources/fake-acp-agent.py +++ b/src/test/resources/fake-acp-agent.py @@ -1,20 +1,23 @@ #!/usr/bin/env python3 -"""Fake Kimi ACP agent for end-to-end tests. +"""Fake Kimi ACP agent for end-to-end and lifecycle-hardening tests. -Speaks newline-delimited JSON-RPC on stdio exactly like `kimi acp`: -answers `initialize`, `session/new`, `session/list`, `session/fork`, -`session/load`, `session/resume`, `session/close`, `session/delete`, -`authenticate`, `logout`, `session/set_model`, `session/set_mode`; on -`session/prompt` streams a few `session/update` notifications (one unknown -kind, one agent_message_chunk, another agent_message_chunk) and answers with -stop_reason `end_turn`. `session/cancel` notifications are ignored. +Optional first argument selects behavior: + normal normal ACP replies + delay-prompt hold a prompt open long enough to test same-session admission + malformed-prompt emit malformed JSON then stay alive + exit-on-prompt exit the process while a prompt is pending + hang-list never answer session/list """ import json import sys +import time + + +MODE = sys.argv[1] if len(sys.argv) > 1 else "normal" def send(payload): - sys.stdout.write(json.dumps(payload) + "\n") + sys.stdout.write(json.dumps(payload, ensure_ascii=False) + "\n") sys.stdout.flush() @@ -22,6 +25,24 @@ def reply(req_id, result): send({"jsonrpc": "2.0", "id": req_id, "result": result}) +def send_normal_prompt(session_id, req_id): + send({"jsonrpc": "2.0", "method": "session/update", "params": { + "sessionId": session_id, + "update": {"sessionUpdate": "tool_call", "title": "ignored"}, + }}) + send({"jsonrpc": "2.0", "method": "session/update", "params": { + "sessionId": session_id, + "update": {"sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": "你好"}}, + }}) + send({"jsonrpc": "2.0", "method": "session/update", "params": { + "sessionId": session_id, + "update": {"sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": "世界"}}, + }}) + reply(req_id, {"stopReason": "end_turn"}) + + def main(): for line in sys.stdin: line = line.strip() @@ -31,6 +52,7 @@ def main(): frame = json.loads(line) except ValueError: continue + method = frame.get("method", "") req_id = frame.get("id") params = frame.get("params") or {} @@ -48,26 +70,24 @@ def main(): reply(req_id, {"sessionId": params.get("sessionId", "sess_fake")}) elif method == "session/fork": reply(req_id, {"sessionId": "sess_forked"}) + elif method == "session/list" and MODE == "hang-list": + time.sleep(5) elif method in ("session/list", "session/set_mode", "session/set_model", "authenticate", "logout", "session/close", "session/delete"): reply(req_id, {}) elif method == "session/prompt": session_id = params.get("sessionId", "sess_fake") - send({"jsonrpc": "2.0", "method": "session/update", "params": { - "sessionId": session_id, - "update": {"sessionUpdate": "tool_call", "title": "ignored"}, - }}) - send({"jsonrpc": "2.0", "method": "session/update", "params": { - "sessionId": session_id, - "update": {"sessionUpdate": "agent_message_chunk", - "content": {"type": "text", "text": "你好"}}, - }}) - send({"jsonrpc": "2.0", "method": "session/update", "params": { - "sessionId": session_id, - "update": {"sessionUpdate": "agent_message_chunk", - "content": {"type": "text", "text": "世界"}}, - }}) - reply(req_id, {"stopReason": "end_turn"}) + if MODE == "delay-prompt": + time.sleep(3) + send_normal_prompt(session_id, req_id) + elif MODE == "malformed-prompt": + sys.stdout.write("{not-json\n") + sys.stdout.flush() + time.sleep(5) + elif MODE == "exit-on-prompt": + sys.exit(7) + else: + send_normal_prompt(session_id, req_id) elif method == "session/cancel": pass elif req_id is not None: