test(acp): sync fake lifecycle modes to 1.0.x

Author: loong10k

src/test/resources/fake-acp-agent.py

@@ -1,27 +1,48 @@
 #!/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()
 
 
 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: