diff --git a/openspec/changes/kimi-acp-lifecycle-hardening/evidence.md b/openspec/changes/kimi-acp-lifecycle-hardening/evidence.md new file mode 100644 index 0000000..a089dd3 --- /dev/null +++ b/openspec/changes/kimi-acp-lifecycle-hardening/evidence.md @@ -0,0 +1,199 @@ +# kimi-acp-lifecycle-hardening 验证证据 + +## 1. OpenSpec Strict Validation + +Workflow commit: `1a810365662c9c445ea1ff554ee211c396c82bbb` + +Workflow run: `35519997399` + +Pinned CLI: `@fission-ai/openspec@1.13.1` + +全部 8 个 Change 均执行： + +```text +openspec validate --type change --strict --no-interactive +``` + +结果： + +| Change | Job | Result | +|---|---:|---| +| kimi-runtime-foundation | 106102324518 | PASS | +| kimi-acp-lifecycle-hardening | 106102324513 | PASS | +| kimi-typed-protocol | 106102324502 | PASS | +| kimi-unified-events | 106102324343 | PASS | +| kimi-server-websocket | 106102324517 | PASS | +| kimi-process-hardening | 106102324484 | PASS | +| kimi-observability | 106102324551 | PASS | +| kimi-branch-parity | 106102324475 | PASS | + +每个 job 日志均记录 `openspec --version = 1.13.1` 和 `Change '' is valid`。 + +## 2. CodeGraph Baseline + +实现前使用 CodeGraph v1.6.0 实际索引： + +| Branch | Files | Nodes | Edges | +|---|---:|---:|---:| +| feature/1.0.x | 25 | 609 | 1,283 | +| feature/2.0.x | 25 | 608 | 1,278 | +| feature/3.0.x | 25 | 608 | 1,278 | + +关键调用基线已检查： +- `KimiAcpClient.promptAsync` → request / scheduleTimeout / PromptStream / KimiAcpTurnResult +- `KimiAcpClient.request` ← initialize/session/prompt/session management callers +- ACP reader / pending RPC / promptStreams 为生命周期核心资源。 + +**Post-hardening CodeGraph impact：NOT_RUN**。当前执行环境没有可用本地 Runner，因此不得把 baseline 冒充最终 impact review。 + +## 3. TDD Cycle 1 + +### RED + +Commit: `e247907b37c727b644941e81f517e4fed7e83e35` + +CI run/job: `35520185628 / 106102809432` + +Result: **BUILD FAILURE** + +```text +Tests run: 72, Failures: 4, Errors: 0 +``` + +失败事实： +- callback exception 杀死 reader，future 变为 TimeoutException； +- malformed JSON 被忽略，future 变为 TimeoutException； +- write-before-connect 后 pendingRpcs expected 0 but was 1； +- same-session 第二个 prompt 未被拒绝。 + +### GREEN + +Commit: `24a51acd038f554438f115ce2d124149f432bd2f` + +CI run/job: `35520278533 / 106103048600` + +```text +KimiAcpLifecycleHardeningTest: 5/5 PASS +Tests run: 72, Failures: 0, Errors: 0 +BUILD SUCCESS +``` + +## 4. TDD Cycle 2 + +### RED + +Commit: `8d1e8685962e62a4fd86e9d3bb2b627ac840170a` + +CI run/job: `35520389264 / 106103337398` + +```text +Tests run: 75, Failures: 1 +``` + +唯一失败：`cancel` 只发送 notification，没有终结 active prompt；测试等待后得到 TimeoutException。 + +### GREEN + +Commit: `173d3f70be51822472d83c4c230b0c9ec826bb8f` + +CI run/job: `35520462943 / 106103531738` + +Result: **PASS** + +实现 cancel → active turn exceptional completion，并由既有 completion cleanup 移除 pending RPC / prompt stream。 + +## 5. TDD Cycle 3 + +### RED + +Commit: `4ceb2460f522a056edae62135a5b237dfadb48df` + +CI run/job: `35520544960 / 106103751629` + +```text +Tests run: 78, Failures: 0, Errors: 3 +``` + +三个 error 全部为： +`NoSuchMethodException: KimiAcpClient.getState()` + +### GREEN + +Commit: `409b7cc314237bfe1df4955a57ff90388e83197b` + +CI run/job: `35520636076 / 106103986308` + +```text +KimiAcpLifecycleHardeningTest: 11/11 PASS +Tests run: 78, Failures: 0, Errors: 0, Skipped: 0 +BUILD SUCCESS +``` + +新增 `KimiAcpState`： +NEW → CONNECTING → INITIALIZING → READY → CLOSING → CLOSED，transport fatal failure → FAILED。 + +FAILED transport 会在注册新 RPC 前拒绝调用。 + +## 6. Compatibility Line Evidence + +### feature/3.0.x implementation line + +JDK: 21 +Branch: `hardening/3.0.x-acp-lifecycle` +Head validated: `409b7cc314237bfe1df4955a57ff90388e83197b` + +```text +KimiAcpLifecycleHardeningTest: 11/11 PASS +Full suite: 78/78 PASS +BUILD SUCCESS +``` + +### feature/2.0.x compatibility line + +PR: #2 +Head: `9361053e3a65fae325182a9e51247ba27e1e5640` +CI run/job: `35520872348 / 106104609921` +JDK: Temurin 17.0.20 + +```text +KimiAcpLifecycleHardeningTest: 11/11 PASS +Full suite: 78/78 PASS +BUILD SUCCESS +``` + +### feature/1.0.x compatibility line + +PR: #3 +Head: `2606afe39e882aaed2f6f2e1f22bc52c35c4def5` +CI run/job: `35520900971 / 106104684383` +JDK: Temurin 8u504 + +```text +KimiAcpLifecycleHardeningTest: 11/11 PASS +Full suite: 78/78 PASS +BUILD SUCCESS +``` + +## 7. Current Semantic Changes + +已验证： +- failed write 不残留 pending RPC； +- same-session active prompt 不被覆盖； +- callback exception 不再杀死 ACP reader； +- malformed JSON 变为明确 protocol failure； +- process exit 使 pending prompt 失败； +- timeout/close 清理 pending registries； +- cancel 立即给 active prompt 单一异常终态； +- lifecycle state 可观察； +- transport fatal failure 进入 FAILED，并拒绝新 RPC； +- recoverable connect/spawn failure 回到 NEW； +- close 幂等。 + +## 8. Evidence Gaps + +仍为 NOT_RUN / NOT_PROVEN： +- 高并发 barrier race； +- stubborn process force-kill； +- 多轮 thread/timer/process leak baseline； +- post-change CodeGraph impact； +- dependency vulnerability/static analysis（本 Change 最终 production gate）。 diff --git a/openspec/changes/kimi-acp-lifecycle-hardening/review.md b/openspec/changes/kimi-acp-lifecycle-hardening/review.md new file mode 100644 index 0000000..7eb781c --- /dev/null +++ b/openspec/changes/kimi-acp-lifecycle-hardening/review.md @@ -0,0 +1,52 @@ +# kimi-acp-lifecycle-hardening 评审记录 + +## Decision + +**APPROVED FOR IMPLEMENTATION / IN PROGRESS** + +用户于 2026-09-20 明确要求进入 OpenSpec 评审、先执行 8 个 Change 的 strict validate，并从本 Change 开始严格 TDD 实现。 + +本记录只批准 `kimi-acp-lifecycle-hardening` 已书面定义的范围，不扩展到 Runtime Foundation、Typed Protocol、WebSocket 或其他 Change。 + +## Specification Gate + +- OpenSpec CLI: `1.13.1` +- GitHub Actions run: `35519997399` +- Job: `106102324513` +- Command: + `openspec validate kimi-acp-lifecycle-hardening --type change --strict --no-interactive` +- Result: `Change 'kimi-acp-lifecycle-hardening' is valid` +- Status: **PASS** + +其余 7 个 Change 在同一 matrix run 中也均 strict validate PASS；详见 evidence.md。 + +## TDD Review + +实现严格分为测试提交与生产代码提交，RED 失败证据在修复前由 CI 产生： + +1. RED-1 `e247907b...` → 4 个预期失败。 +2. GREEN-1 `24a51acd...` → 72/72 tests PASS。 +3. RED-2 `8d1e8685...` → cancel 终态 1 个预期失败。 +4. GREEN-2 `173d3f70...` → CI PASS。 +5. RED-3 `4ceb2460...` → lifecycle state 3 个预期 error。 +6. GREEN-3 `409b7cc3...` → 78/78 tests PASS。 + +三条兼容线随后运行相同 hardening tests： +- Java 21 / Jackson 3：PASS +- Java 17 / Jackson 2：PASS +- Java 8 / Jackson 2：PASS + +## Remaining Gates + +以下内容尚未完成，因此本 Change 仍为 **IN PROGRESS**，不得宣称全部 production-ready： + +- response/timeout/cancel/close 的 barrier 级高并发 race 压测； +- 不同 session 的并发 turn isolation 压测； +- stubborn child process 的有界 graceful/force shutdown； +- 线程、timer、process 的重复循环 leak baseline； +- hardening 后重新运行 CodeGraph impact review； +- 最终三线合并后的 CI 与 workspace/release hygiene。 + +## Approval Boundary + +当前允许继续完成上述剩余任务，并在证据齐全后把三个兼容线 PR 转 Ready/合并。 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 daa88ff..0623ec6 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(); } 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..53a5701 --- /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 3.0.0 + */ +public enum KimiAcpState { + NEW, + CONNECTING, + INITIALIZING, + READY, + CLOSING, + CLOSED, + FAILED +} 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..97a7270 --- /dev/null +++ b/src/test/java/io/github/easy4j/kimi/acp/KimiAcpLifecycleHardeningTest.java @@ -0,0 +1,302 @@ +/* + * 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.lang.reflect.Modifier; +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.CountDownLatch; +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"); + } + } + + + @Test + void shouldCleanPromptRegistriesAfterPromptTimeout() throws Exception { + KimiAcpConfig config = config("hang-prompt"); + config.setReadTimeoutMillis(200); + try (KimiAcpClient client = new KimiAcpClient(config)) { + client.connect(); + String sessionId = client.newSession("/tmp"); + + CompletableFuture future = + client.promptAsync(sessionId, "timeout", null); + + ExecutionException failure = assertThrows(ExecutionException.class, + () -> future.get(2, TimeUnit.SECONDS)); + assertTrue(failure.getCause() instanceof KimiException); + assertTrue(failure.getCause().getMessage().toLowerCase().contains("timed out")); + assertEquals(0, privateMapSize(client, "pendingRpcs")); + assertEquals(0, privateMapSize(client, "promptStreams")); + } + } + + @Test + void shouldKeepConcurrentSessionsIsolated() throws Exception { + try (KimiAcpClient client = new KimiAcpClient(config("concurrent-prompts"))) { + client.connect(); + + List firstDeltas = java.util.Collections.synchronizedList(new ArrayList()); + List secondDeltas = java.util.Collections.synchronizedList(new ArrayList()); + + CompletableFuture first = + client.promptAsync("session-A", "A", firstDeltas::add); + CompletableFuture second = + client.promptAsync("session-B", "B", secondDeltas::add); + + KimiAcpTurnResult firstResult = first.get(2, TimeUnit.SECONDS); + KimiAcpTurnResult secondResult = second.get(2, TimeUnit.SECONDS); + + assertEquals("A-1A-2", firstResult.getContent()); + assertEquals("B-1B-2", secondResult.getContent()); + assertEquals(java.util.Arrays.asList("A-1", "A-2"), firstDeltas); + assertEquals(java.util.Arrays.asList("B-1", "B-2"), secondDeltas); + assertEquals(0, privateMapSize(client, "promptStreams")); + } + } + + @Test + void shouldForceTerminateStubbornOwnedProcessOnClose() throws Exception { + KimiAcpClient client = new KimiAcpClient(config("stubborn-close")); + client.connect(); + Process child = privateProcess(client); + assertTrue(child.isAlive()); + + client.close(); + + assertTrue(child.waitFor(1, TimeUnit.SECONDS), + "close must force-terminate an owned ACP process that ignores graceful termination"); + assertTrue(!child.isAlive()); + assertEquals("CLOSED", lifecycleState(client)); + } + + private static Process privateProcess(KimiAcpClient client) throws Exception { + Field field = KimiAcpClient.class.getDeclaredField("process"); + field.setAccessible(true); + return (Process) field.get(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(); + } +} diff --git a/src/test/resources/fake-acp-agent.py b/src/test/resources/fake-acp-agent.py index 281ba19..9dfc59f 100755 --- a/src/test/resources/fake-acp-agent.py +++ b/src/test/resources/fake-acp-agent.py @@ -1,28 +1,77 @@ #!/usr/bin/env python3 -"""Fake Kimi ACP agent for end-to-end 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. +"""Fake Kimi ACP agent for end-to-end and lifecycle-hardening tests. + +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 + hang-prompt never answer session/prompt + concurrent-prompts answer two sessions from background threads + stubborn-close ignore SIGTERM and stay alive after stdin EOF """ import json import sys +import time +import signal +import threading + + +MODE = sys.argv[1] if len(sys.argv) > 1 else "normal" +WRITE_LOCK = threading.Lock() + +if MODE == "stubborn-close": + signal.signal(signal.SIGTERM, signal.SIG_IGN) def send(payload): - sys.stdout.write(json.dumps(payload) + "\n") - sys.stdout.flush() + with WRITE_LOCK: + sys.stdout.write(json.dumps(payload, ensure_ascii=False) + "\n") + sys.stdout.flush() 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 send_concurrent_prompt(session_id, req_id): + if session_id.endswith("A"): + time.sleep(0.08) + pieces = ["A-1", "A-2"] + else: + time.sleep(0.02) + pieces = ["B-1", "B-2"] + for piece in pieces: + send({"jsonrpc": "2.0", "method": "session/update", "params": { + "sessionId": session_id, + "update": {"sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": piece}}, + }}) + time.sleep(0.01) + reply(req_id, {"stopReason": "end_turn"}) + + def main(): + workers = [] for line in sys.stdin: line = line.strip() if not line: @@ -31,6 +80,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,32 +98,45 @@ 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) + elif MODE == "hang-prompt": + time.sleep(5) + elif MODE == "concurrent-prompts": + worker = threading.Thread(target=send_concurrent_prompt, + args=(session_id, req_id)) + worker.daemon = True + worker.start() + workers.append(worker) + else: + send_normal_prompt(session_id, req_id) elif method == "session/cancel": pass elif req_id is not None: send({"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": "method not found: " + method}}) + for worker in workers: + worker.join(timeout=1) + + if MODE == "stubborn-close": + while True: + time.sleep(1) + if __name__ == "__main__": main()