feat(appserver): align 3.x protocol runtime with 2.x

Author: loong10k

src/main/java/io/github/easy4j/codex/appserver/AppServerTurnRequest.java

@@ -59,6 +59,12 @@ public class AppServerTurnRequest {
      */
     private Consumer<String> onTurnStarted;
 
+    /**
+     * Optional structured listener for turn lifecycle and streaming events.
+     * Existing callbacks remain supported for source compatibility.
+     */
+    private CodexAppServerListener listener;
+
     /**
      * Returns the session key in a comparable form, or {@code null} when unset.
      *

src/main/java/io/github/easy4j/codex/appserver/CodexAppServerClient.java

@@ -77,7 +77,8 @@ public class CodexAppServerClient implements AutoCloseable {
     private final CodexAppServerConfig config;
     private final ObjectMapper objectMapper =
             JsonMapper.builder().disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES).build();
-    private final ThreadMappingCache threadBySession;
+    private final ThreadMappingStore threadBySession;
+    private final SessionExecutionCoordinator sessionCoordinator = new SessionExecutionCoordinator();
     private final Object httpClientLock = new Object();
     private final ExecutorService clientExecutor = Executors.newCachedThreadPool(r -> {
         Thread thread = new Thread(r, "codex-app-server-client");
@@ -95,8 +96,15 @@ public class CodexAppServerClient implements AutoCloseable {
      * @throws NullPointerException if {@code config} is {@code null}.
      */
     public CodexAppServerClient(CodexAppServerConfig config) {
+        this(config, new ThreadMappingCache(
+                Objects.requireNonNull(config, "config").getMaxSessionMappings()));
+    }
+
+    public CodexAppServerClient(
+            CodexAppServerConfig config,
+            ThreadMappingStore threadMappingStore) {
         this.config = Objects.requireNonNull(config, "config");
-        this.threadBySession = new ThreadMappingCache(config.getMaxSessionMappings());
+        this.threadBySession = Objects.requireNonNull(threadMappingStore, "threadMappingStore");
     }
 
     /**
@@ -142,6 +150,19 @@ public CompletableFuture<AppServerTurnResult> runTurnAsync(AppServerTurnRequest
         if (Objects.isNull(request.getPrompt()) || request.getPrompt().trim().isEmpty()) {
             throw new IllegalArgumentException("Codex prompt must not be blank");
         }
+        String sessionKey = request.normalizedSessionKey();
+        if (Objects.isNull(sessionKey)) {
+            return startTurn(request);
+        }
+        return sessionCoordinator.submit(sessionKey, () -> startTurn(request));
+    }
+
+    private CompletableFuture<AppServerTurnResult> startTurn(AppServerTurnRequest request) {
+        if (closed) {
+            CompletableFuture<AppServerTurnResult> failed = new CompletableFuture<>();
+            failed.completeExceptionally(new IllegalStateException("Codex app-server client is closed"));
+            return failed;
+        }
         CodexAppServerTurn turn =
                 new CodexAppServerTurn(request, config, objectMapper, threadBySession, httpClient());
         return turn.start();
@@ -187,7 +208,7 @@ public List<AppServerThread> listThreads(int limit, String cursor) {
         if (limit > 0) {
             params.put("limit", limit);
         }
-        JsonNode result = execRpcNode("thread/list", params);
+        JsonNode result = execRpcNode(CodexAppServerProtocol.THREAD_LIST, params);
         List<AppServerThread> threads = new ArrayList<>();
         for (JsonNode node : result.path("threads")) {
             threads.add(parseThread(node));
@@ -205,7 +226,7 @@ public List<AppServerThread> listThreads(int limit, String cursor) {
     public AppServerThread readThread(String threadId) {
         Map<String, Object> params = new LinkedHashMap<>();
         params.put("threadId", Objects.requireNonNull(threadId, "threadId").trim());
-        JsonNode result = execRpcNode("thread/read", params);
+        JsonNode result = execRpcNode(CodexAppServerProtocol.THREAD_READ, params);
         return parseThread(result.path("thread"));
     }
 
@@ -234,26 +255,26 @@ public String readThreadRaw(String threadId) {
     public AppServerThread forkThread(String threadId) {
         Map<String, Object> params = new LinkedHashMap<>();
         params.put("threadId", Objects.requireNonNull(threadId, "threadId").trim());
-        JsonNode result = execRpcNode("thread/fork", params);
+        JsonNode result = execRpcNode(CodexAppServerProtocol.THREAD_FORK, params);
         return parseThread(result.path("thread"));
     }
 
     /** Archives a thread via {@code thread/archive}. */
     public void archiveThread(String threadId) {
-        simpleThreadCall("thread/archive", threadId);
+        simpleThreadCall(CodexAppServerProtocol.THREAD_ARCHIVE, threadId);
     }
 
     /** Unarchives a thread via {@code thread/unarchive}. */
     public void unarchiveThread(String threadId) {
-        simpleThreadCall("thread/unarchive", threadId);
+        simpleThreadCall(CodexAppServerProtocol.THREAD_UNARCHIVE, threadId);
     }
 
     /**
      * Permanently deletes a thread and its spawned descendants via
      * {@code thread/delete}. Ephemeral roots cannot be deleted.
      */
     public void deleteThread(String threadId) {
-        simpleThreadCall("thread/delete", threadId);
+        simpleThreadCall(CodexAppServerProtocol.THREAD_DELETE, threadId);
     }
 
     /**
@@ -269,7 +290,7 @@ public void interruptTurn(String threadId, String turnId) {
         Map<String, Object> params = new LinkedHashMap<>();
         params.put("threadId", Objects.requireNonNull(threadId, "threadId").trim());
         params.put("turnId", Objects.requireNonNull(turnId, "turnId").trim());
-        execRpcNode("turn/interrupt", params);
+        execRpcNode(CodexAppServerProtocol.TURN_INTERRUPT, params);
     }
 
     /**
@@ -290,7 +311,7 @@ public void steerTurn(String threadId, String expectedTurnId, String prompt) {
         params.put("threadId", Objects.requireNonNull(threadId, "threadId").trim());
         params.put("expectedTurnId", Objects.requireNonNull(expectedTurnId, "expectedTurnId").trim());
         params.put("input", List.of(input));
-        execRpcNode("turn/steer", params);
+        execRpcNode(CodexAppServerProtocol.TURN_STEER, params);
     }
 
     /**

src/main/java/io/github/easy4j/codex/appserver/CodexAppServerListener.java

@@ -0,0 +1,31 @@
+/*
+ * 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.codex.appserver;
+
+/**
+ * Listener for high-value Codex app-server turn events.
+ *
+ * <p>All methods are optional. Implementations should return quickly because
+ * callbacks are invoked on the WebSocket event path.</p>
+ */
+public interface CodexAppServerListener {
+
+    default void onTurnStarted(String turnId) {
+    }
+
+    default void onTextDelta(String delta) {
+    }
+
+    default void onItemCompleted(String itemType, String content) {
+    }
+
+    default void onTokenUsage(String rawJson) {
+    }
+
+    default void onWarning(String rawJson) {
+    }
+}

src/main/java/io/github/easy4j/codex/appserver/CodexAppServerProtocol.java

@@ -0,0 +1,43 @@
+/*
+ * 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.codex.appserver;
+
+/**
+ * Canonical method and notification names used by the Codex app-server
+ * integration. Keeping protocol strings in one place prevents the turn and
+ * generic-RPC transports from drifting apart.
+ */
+final class CodexAppServerProtocol {
+
+    static final String INITIALIZE = "initialize";
+    static final String INITIALIZED = "notifications/initialized";
+
+    static final String THREAD_START = "thread/start";
+    static final String THREAD_RESUME = "thread/resume";
+    static final String THREAD_LIST = "thread/list";
+    static final String THREAD_READ = "thread/read";
+    static final String THREAD_FORK = "thread/fork";
+    static final String THREAD_ARCHIVE = "thread/archive";
+    static final String THREAD_UNARCHIVE = "thread/unarchive";
+    static final String THREAD_DELETE = "thread/delete";
+
+    static final String TURN_START = "turn/start";
+    static final String TURN_STEER = "turn/steer";
+    static final String TURN_INTERRUPT = "turn/interrupt";
+
+    static final String TURN_STARTED = "turn/started";
+    static final String TURN_COMPLETED = "turn/completed";
+    static final String TURN_FAILED = "turn/failed";
+    static final String ITEM_STARTED = "item/started";
+    static final String ITEM_COMPLETED = "item/completed";
+    static final String AGENT_MESSAGE_DELTA = "item/agentMessage/delta";
+    static final String TOKEN_USAGE_UPDATED = "thread/tokenUsage/updated";
+    static final String ERROR = "error";
+
+    private CodexAppServerProtocol() {
+    }
+}

src/main/java/io/github/easy4j/codex/appserver/CodexAppServerRpc.java

@@ -125,7 +125,7 @@ public void onOpen(WebSocket socket) {
         clientInfo.put("version", "3.0.x");
         Map<String, Object> initParams = new java.util.LinkedHashMap<>();
         initParams.put("clientInfo", clientInfo);
-        sendRpc(INIT_REQUEST_ID, "initialize", initParams);
+        sendRpc(INIT_REQUEST_ID, CodexAppServerProtocol.INITIALIZE, initParams);
     }
 
     @Override
@@ -236,7 +236,7 @@ private void handleInitializeSettled() {
         // both back-to-back from inside the onText callback.
         Map<String, Object> payload = new java.util.LinkedHashMap<>();
         payload.put("jsonrpc", "2.0");
-        payload.put("method", "initialized");
+        payload.put("method", CodexAppServerProtocol.INITIALIZED);
         payload.put("params", java.util.Collections.emptyMap());
         socket.sendText(toJson(payload), true).whenComplete((w, error) -> {
             if (Objects.nonNull(error)) {