Skip to content

Commit b8bb07d

Browse files
loong10kloong10k
committed
Merge fix/2.0.x-appserver-thread-id: codex 0.154.0 真实线协议对齐（已 review）
2 parents 1e5edb8 + 63f26a6 commit b8bb07d

2 files changed

Lines changed: 129 additions & 26 deletions

File tree

‎src/main/java/io/github/easy4j/codex/appserver/CodexAppServerTurn.java‎

Lines changed: 37 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -135,14 +135,27 @@ public void onOpen(WebSocket socket) {
135135
}
136136

137137
/**
138-
* Sends the first RPC: {@code thread/resume} when the session key already
139-
* maps to a thread id, {@code thread/start} otherwise.
138+
* Sends the JSON-RPC lifecycle handshake first — real codex app-servers
139+
* (verified against 0.154.0) reject any request before
140+
* {@code initialize} with {@code -32600 Not initialized} — then starts or
141+
* resumes the thread.
140142
*
141143
* <p>Split from {@link #onOpen(WebSocket)} so contract tests can drive the
142144
* turn without a socket; requests are only recorded in
143145
* {@link #sentMessages} when no socket is attached.</p>
144146
*/
145147
void begin() {
148+
CompletableFuture<JsonNode> initRpc = newRpc("initialize", buildInitializeParams());
149+
initRpc.thenAccept(result -> {
150+
sendNotification("notifications/initialized");
151+
startOrResumeThread();
152+
}).exceptionally(error -> {
153+
completeError(unwrap(error));
154+
return null;
155+
});
156+
}
157+
158+
private void startOrResumeThread() {
146159
String sessionKey = request.normalizedSessionKey();
147160
String previousThreadId = Objects.isNull(sessionKey) ? null : threadBySession.get(sessionKey);
148161
boolean resume = hasText(previousThreadId);
@@ -319,6 +332,22 @@ Map<String, Object> buildThreadStartParams(String resumeThreadId) {
319332
return params;
320333
}
321334

335+
Map<String, Object> buildInitializeParams() {
336+
Map<String, Object> clientInfo = new LinkedHashMap<>();
337+
clientInfo.put("name", "easy4j-codex-java-sdk");
338+
clientInfo.put("version", "2.0.x");
339+
Map<String, Object> params = new LinkedHashMap<>();
340+
params.put("clientInfo", clientInfo);
341+
return params;
342+
}
343+
344+
private void sendNotification(String method) {
345+
Map<String, Object> payload = new LinkedHashMap<>();
346+
payload.put("jsonrpc", "2.0");
347+
payload.put("method", method);
348+
sendText(toJson(payload));
349+
}
350+
322351
Map<String, Object> buildTurnStartParams(String targetThreadId) {
323352
Map<String, Object> input = new LinkedHashMap<>();
324353
input.put("type", "text");
@@ -366,7 +395,12 @@ private void completeError(Throwable error) {
366395
}
367396

368397
private String extractThreadId(JsonNode result) {
369-
String threadId = firstText(result, "threadId", "thread_id");
398+
// codex ≥0.14x 将线程对象嵌套在 result.thread（实测 0.154.0 返回
399+
// result.thread.id = UUID）；旧版本为顶层 threadId/thread_id。两者都兼容。
400+
String threadId = firstText(result.path("thread"), "id", "threadId", "thread_id");
401+
if (!hasText(threadId)) {
402+
threadId = firstText(result, "threadId", "thread_id");
403+
}
370404
return hasText(threadId) ? threadId : null;
371405
}
372406

‎src/test/java/io/github/easy4j/codex/appserver/CodexAppServerTurnTest.java‎

Lines changed: 92 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,10 @@
3737
* inbound frames are fed straight into {@code handleFrame}, mirroring the
3838
* integration contract of the app-server protocol.
3939
*
40+
* <p>Real wire shape (verified against codex 0.154.0): {@code initialize} is
41+
* mandatory before any thread RPC ({@code -32600 Not initialized} otherwise),
42+
* and the thread id is nested at {@code result.thread.id}.</p>
43+
*
4044
* @since 3.0.0
4145
*/
4246
class CodexAppServerTurnTest {
@@ -55,6 +59,24 @@ private JsonNode lastFrame(List<String> sent) {
5559
}
5660
}
5761

62+
private JsonNode frameAt(List<String> sent, int index) {
63+
try {
64+
return mapper.readTree(sent.get(index));
65+
} catch (Exception ex) {
66+
throw new IllegalStateException("Invalid captured frame", ex);
67+
}
68+
}
69+
70+
/**
71+
* Drives {@code begin()} through the mandatory initialize handshake:
72+
* initialize (id=1) → notifications/initialized → thread/start|resume (id=2).
73+
*/
74+
private void handshake(CodexAppServerTurn turn) {
75+
turn.begin();
76+
turn.handleFrame(
77+
"{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"serverInfo\":{\"name\":\"codex\"}}}");
78+
}
79+
5880
@Test
5981
void shouldMapWebSocketUrls() {
6082
assertEquals("ws://host:8081", CodexAppServerTurn.toWebSocketUrl("ws://host:8081"));
@@ -69,15 +91,21 @@ void shouldMapWebSocketUrls() {
6991
}
7092

7193
@Test
72-
void shouldStartWithThreadStartWhenNoSessionKey() {
94+
void shouldInitializeBeforeThreadStartWhenNoSessionKey() {
7395
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").build(),
7496
new ThreadMappingCache(10));
7597

76-
turn.begin();
98+
handshake(turn);
7799

100+
// 倒数第二帧应为 initialize，最后一帧为 thread/start（id=2）
101+
JsonNode initializeFrame = frameAt(turn.sentMessages(), 0);
102+
assertEquals("initialize", initializeFrame.path("method").asText());
103+
assertEquals("easy4j-codex-java-sdk",
104+
initializeFrame.path("params").path("clientInfo").path("name").asText());
105+
JsonNode initializedFrame = frameAt(turn.sentMessages(), 1);
106+
assertEquals("notifications/initialized", initializedFrame.path("method").asText());
78107
JsonNode frame = lastFrame(turn.sentMessages());
79-
assertEquals("2.0", frame.path("jsonrpc").asText());
80-
assertEquals(1L, frame.path("id").asLong());
108+
assertEquals(2L, frame.path("id").asLong());
81109
assertEquals("thread/start", frame.path("method").asText());
82110
assertTrue(frame.path("params").isEmpty());
83111
}
@@ -89,7 +117,7 @@ void shouldResumeThreadWhenSessionKeyMapped() {
89117
CodexAppServerTurn turn = newTurn(
90118
AppServerTurnRequest.builder().prompt("hi").sessionKey(" chat-1 ").build(), cache);
91119

92-
turn.begin();
120+
handshake(turn);
93121

94122
JsonNode frame = lastFrame(turn.sentMessages());
95123
assertEquals("thread/resume", frame.path("method").asText());
@@ -101,8 +129,8 @@ void shouldBuildTurnStartWithTextInputItem() {
101129
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("Fix it").build(),
102130
new ThreadMappingCache(10));
103131

104-
turn.begin();
105-
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"threadId\":\"th_9\"}}");
132+
handshake(turn);
133+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"thread\":{\"id\":\"th_9\"}}}");
106134

107135
JsonNode frame = lastFrame(turn.sentMessages());
108136
assertEquals("turn/start", frame.path("method").asText());
@@ -112,21 +140,48 @@ void shouldBuildTurnStartWithTextInputItem() {
112140
assertEquals("Fix it", input.path("text").asText());
113141
}
114142

143+
@Test
144+
void shouldExtractNestedThreadIdFromRealWireShape() {
145+
// 实测 codex 0.154.0：thread id 嵌套在 result.thread.id
146+
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").build(),
147+
new ThreadMappingCache(10));
148+
149+
handshake(turn);
150+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"thread\":"
151+
+ "{\"id\":\"01a0af32-c02f-7811-ae30-e154d84cf31a\",\"sessionId\":\"01a0af32\"}}}");
152+
153+
JsonNode frame = lastFrame(turn.sentMessages());
154+
assertEquals("turn/start", frame.path("method").asText());
155+
assertEquals("01a0af32-c02f-7811-ae30-e154d84cf31a",
156+
frame.path("params").path("threadId").asText());
157+
}
158+
159+
@Test
160+
void shouldAcceptLegacyTopLevelThreadId() {
161+
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").build(),
162+
new ThreadMappingCache(10));
163+
164+
handshake(turn);
165+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"threadId\":\"th_legacy\"}}");
166+
167+
assertEquals("th_legacy", lastFrame(turn.sentMessages()).path("params").path("threadId").asText());
168+
}
169+
115170
@Test
116171
void shouldFailWhenThreadStartResultHasNoThreadId() {
117172
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").build(),
118173
new ThreadMappingCache(10));
119174

120-
turn.begin();
121-
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}");
175+
handshake(turn);
176+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{}}");
122177

123178
assertTrue(turn.future().isCompletedExceptionally());
124179
CompletionException ex = assertThrows(CompletionException.class, () -> turn.future().join());
125180
assertInstanceOf(CodexAppServerException.class, ex.getCause());
126181
}
127182

128183
@Test
129-
void shouldFailWhenRpcRespondsWithError() {
184+
void shouldFailWhenInitializeRespondsWithError() {
130185
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").build(),
131186
new ThreadMappingCache(10));
132187

@@ -138,6 +193,19 @@ void shouldFailWhenRpcRespondsWithError() {
138193
assertInstanceOf(CodexAppServerException.class, ex.getCause());
139194
}
140195

196+
@Test
197+
void shouldFailWhenRpcRespondsWithError() {
198+
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").build(),
199+
new ThreadMappingCache(10));
200+
201+
handshake(turn);
202+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"error\":{\"code\":-1,\"message\":\"boom\"}}");
203+
204+
assertTrue(turn.future().isCompletedExceptionally());
205+
CompletionException ex = assertThrows(CompletionException.class, () -> turn.future().join());
206+
assertInstanceOf(CodexAppServerException.class, ex.getCause());
207+
}
208+
141209
@Test
142210
void shouldIgnoreUnknownRpcResponses() {
143211
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").build(),
@@ -154,7 +222,8 @@ void shouldCollectAgentMessageItemsAndIgnoreOthers() {
154222
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").onDelta(deltas::add).build(),
155223
new ThreadMappingCache(10));
156224

157-
turn.begin();
225+
handshake(turn);
226+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"thread\":{\"id\":\"th_1\"}}}");
158227
turn.handleFrame("{\"method\":\"item/completed\",\"params\":{\"item\":{\"type\":\"commandExecution\",\"text\":\"rm\"}}}");
159228
turn.handleFrame("{\"method\":\"item/completed\",\"params\":{\"item\":{\"type\":\"agentMessage\",\"text\":\"你好\"}}}");
160229
turn.handleFrame("{\"method\":\"item/completed\",\"params\":{\"item\":{\"itemType\":\"agent_message\",\"content\":\"世界\"}}}");
@@ -171,8 +240,8 @@ void shouldFallBackToTurnCompletedMessageWithoutItems() {
171240
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").build(),
172241
new ThreadMappingCache(10));
173242

174-
turn.begin();
175-
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"threadId\":\"th_1\"}}");
243+
handshake(turn);
244+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"thread\":{\"id\":\"th_1\"}}}");
176245
turn.handleFrame("{\"method\":\"turn/completed\",\"params\":{\"message\":\"fallback text\"}}");
177246

178247
assertEquals("fallback text", turn.future().join().getContent());
@@ -184,8 +253,8 @@ void shouldRememberThreadMappingOnTurnCompleted() {
184253
CodexAppServerTurn turn = newTurn(
185254
AppServerTurnRequest.builder().prompt("hi").sessionKey("chat-7").build(), cache);
186255

187-
turn.begin();
188-
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"threadId\":\"th_7\"}}");
256+
handshake(turn);
257+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"thread\":{\"id\":\"th_7\"}}}");
189258
turn.handleFrame("{\"method\":\"turn/completed\",\"params\":{}}");
190259

191260
assertEquals("th_7", turn.future().join().getThreadId());
@@ -197,8 +266,8 @@ void shouldNotRememberMappingWithoutSessionKey() {
197266
ThreadMappingCache cache = new ThreadMappingCache(10);
198267
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").build(), cache);
199268

200-
turn.begin();
201-
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"threadId\":\"th_1\"}}");
269+
handshake(turn);
270+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"thread\":{\"id\":\"th_1\"}}}");
202271
turn.handleFrame("{\"method\":\"turn/completed\",\"params\":{}}");
203272

204273
assertEquals(0, cache.size());
@@ -253,8 +322,8 @@ void shouldCompleteWithoutErrorAfterTurnCompletedWhenSocketClosesLate() throws E
253322
CodexAppServerTurn turn = newTurn(AppServerTurnRequest.builder().prompt("hi").build(),
254323
new ThreadMappingCache(10));
255324

256-
turn.begin();
257-
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"threadId\":\"th_1\"}}");
325+
handshake(turn);
326+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"thread\":{\"id\":\"th_1\"}}}");
258327
turn.handleFrame("{\"method\":\"turn/completed\",\"params\":{}}");
259328
turn.onClose(null, 1000, "bye");
260329

@@ -286,8 +355,8 @@ void shouldTruncateContentAtCapWithoutFailingTurn() {
286355
AppServerTurnRequest.builder().prompt("hi").onDelta(deltas::add).build(),
287356
config, mapper, new ThreadMappingCache(10), null);
288357

289-
turn.begin();
290-
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"threadId\":\"th_1\"}}");
358+
handshake(turn);
359+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"thread\":{\"id\":\"th_1\"}}}");
291360
turn.handleFrame("{\"method\":\"item/completed\",\"params\":{\"item\":{\"type\":\"agentMessage\",\"text\":\"12345\"}}}");
292361
turn.handleFrame("{\"method\":\"item/completed\",\"params\":{\"item\":{\"type\":\"agentMessage\",\"text\":\"67890\"}}}");
293362
turn.handleFrame("{\"method\":\"item/completed\",\"params\":{\"item\":{\"type\":\"agentMessage\",\"text\":\"ABCDE\"}}}");
@@ -307,8 +376,8 @@ void shouldTreatNonPositiveCapsAsUnbounded() {
307376
CodexAppServerTurn turn = new CodexAppServerTurn(
308377
AppServerTurnRequest.builder().prompt("hi").build(), config, mapper, new ThreadMappingCache(10), null);
309378

310-
turn.begin();
311-
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"threadId\":\"th_1\"}}");
379+
handshake(turn);
380+
turn.handleFrame("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{\"thread\":{\"id\":\"th_1\"}}}");
312381
turn.onText(null, "x".repeat(4096), false);
313382
turn.handleFrame("{\"method\":\"turn/completed\",\"params\":{}}");
314383

0 commit comments

Comments
 (0)