fix(engine): post-delta 전송 오류를 안전하게 재시도한다

cycles/2026-09/wk1/09-01/1917-post-delta-websocket-retry-wrap.md

@@ -0,0 +1,112 @@
+---
+date: 2026-09-01
+scope: [rubato-engine, retry, cursor]
+type: fix
+---
+
+## TL;DR
+
+모델이 글을 조금 내보낸 뒤 WebSocket이 끊기면 Rubato가 모든 시도를
+`senpi:no-turn-retry:`로 끝내 다른 세션까지 대기 상태에 빠졌다.
+provider가 실제로 도구를 실행한 경우만 재시도를 막고, 나머지는 Senpi가 원래 가진
+횟수 제한 재시도로 넘기도록 경계를 좁혔다.
+
+## Keywords
+
+`withRubatoStream` `senpi:no-turn-retry:` `WebSocket error`
+`kCursorExecResolved` `AgentSession._isRetryableError`
+
+## 왜 바꿨나
+
+`senpi:no-turn-retry:WebSocket error`는 모델이 쓰는 문장이 아니다.
+`rubato-stream.mjs`가 텍스트나 toolCall delta를 하나라도 본 뒤 전송 오류가 나면
+같은 턴을 다시 보내지 말라는 뜻으로 붙이던 내부 표식이다.
+
+의도는 맞았다. Cursor는 모델 응답 도중 서버가 로컬의 bash·write 같은 도구를
+실행시킬 수 있고, 연결이 끊겼다고 같은 턴을 다시 보내면 그 도구가 두 번 실행될 수
+있다. 문제는 이 경계를 “provider가 도구를 실행했는가”가 아니라 “화면에 무언가
+나왔는가”로 잡은 데 있었다. 글만 쓰다가 끊긴 턴, 이전 toolResult를 읽고 답하던 턴,
+아직 실행되지 않은 일반 toolCall까지 전부 터미널 오류가 됐다.
+
+실제 로그에서도 8월 29일 이후 같은 표식이 여덟 번 나왔고, 수정 중인 세션도
+파일을 읽고 다음 말을 만들다가 같은 오류로 한 번 멈췄다.
+
+## 어디까지 따라가 봤나
+
+오류는 아래 순서로 굳었다.
+
+1. provider stream이 `WebSocket error`를 돌려준다.
+2. `withRubatoStream`이 post-delta라는 이유로 `senpi:no-turn-retry:`를 붙인다.
+3. Senpi의 `AgentSession._isRetryableError`가 이 접두사를 보고 재시도를 거부한다.
+4. `agent_end.willRetry`가 `false`가 되고, child runner는 해당 턴을 실패로 정착시킨다.
+5. task manager가 세션을 terminal record로 바꾸므로 부모가 새 메시지를 보내기 전까지
+   다른 세션은 그대로 멈춘다.
+
+접두사만 지우고 모든 턴을 다시 보내는 것도 답은 아니었다. provider가 이미 실행한
+도구가 있으면 텍스트 중복보다 큰 부작용이 생긴다. 반대로 일반 toolCall은
+`stopReason: "error"`인 동안 agent loop가 실행하지 않으며, 재시도 전에 실패한
+assistant message도 active branch에서 빠진다. 이전 턴의 toolResult는 남기 때문에
+같은 도구를 다시 실행하지 않고 현재 model call만 새로 보낼 수 있다.
+
+## 결정
+
+재시도 경계를 `kCursorExecResolved`로 잡았다.
+
+Cursor exec-channel은 provider가 도구를 실행하기 전에 toolCall block에 이
+module-local Symbol을 붙인다. stream을 읽는 동안 표식을 한 번이라도 보면 call state에
+latch하고, terminal message가 다른 객체로 바뀌어 block을 잃더라도 fail-open하지 않게
+했다.
+
+- text·thinking delta: 제한 재시도
+- 아직 실행되지 않은 일반 toolCall: 제한 재시도
+- Cursor가 stream 안에서 실행한 tool: `senpi:no-turn-retry:` 유지
+- 사용자가 직접 중단한 턴: 기존처럼 재시도 금지
+
+실패한 toolCall을 성공한 `toolUse`로 바꾸지는 않는다. 잘린 인자가 빈 인자 도구로
+실행되는 기존 위험도 그대로 막는다.
+
+## 바꾼 자리
+
+| 파일 | 변경 |
+|---|---|
+| `harness/rubato-pi/src/rubato-stream.mjs` | provider 실행 표식을 감지하고 latch한 뒤 그 경우에만 post-delta 재시도를 막는다. |
+| `harness/rubato-pi/test/unit/rubato-stream.test.mjs` | text·thinking·일반 toolCall과 Cursor exec-channel을 나눠 실제 AgentSession 판정까지 검사한다. |
+
+## 검증
+
+```text
+cd harness/rubato-pi
+node --test test/unit/rubato-stream.test.mjs
+34 tests, 34 pass, 0 fail
+```
+
+전체 검사도 돌렸지만 최신 `rubato/base` 자체가 이미 깨져 있었다. 변경분을 전부
+stash한 원본에서도 같은 명령이 같은 자리에서 실패하는 것을 다시 확인했다.
+
+```text
+npm --prefix harness/rubato-pi test
+exit 1 — role-prompt 생성물, 로컬 extension 경로, service-tier, TUI 기대값 실패
+
+npm --prefix harness/rubato-pi run test:integration
+exit 1 — task child/runtime와 surface memory 통합 검사 실패
+
+bun run typecheck
+exit 1 — packages/senpi-task/src/state/record.ts:24
+         packages/senpi-task/src/team/member-extension/index.ts:155
+```
+
+이번 변경 파일을 적용했을 때와 stash한 `origin/rubato/base` 원본에서 세 명령이 모두
+각각 exit 1이어서, 위 실패는 이 수정이 만든 회귀가 아니다.
+
+원래 작업본에서는 일반 Rubato Opus로 독립 검토했다. P0~P2는 없었고, stream delta마다
+완료된 block을 다시 훑던 부분과 일반 toolCall의 실제 재시도 판정 단정이 약하다는 P3를
+반영했다.
+
+## 남은 위험
+
+실제 네트워크를 강제로 끊는 live fault injection은 하지 않았다. 테스트는 설치된
+Senpi의 실제 `_isRetryableError`와 pi-ai의 실제 module-local Symbol을 사용한다.
+
+앞으로 Cursor가 아닌 provider도 model stream 안에서 로컬 도구를 실행한다면 같은
+실행 표식 계약을 채택해야 한다. 그 표식 없이 provider 내부 부작용이 생기면 Rubato가
+안전한 재시도와 위험한 재실행을 구분할 방법이 없다.

harness/rubato-pi/src/rubato-stream.mjs

@@ -2,7 +2,7 @@
 //
 // 예전에는 계측·timing·정착이 삭제된 FX transport 구현 **안쪽**에 있었다. 그래서
 // provider 를 직결로 바꾸는 순간 그 의미가 통째로 사라졌다 — statusline 의
-// TTFT/wait/think, measurement log, 델타 후 재시도 금지, 도구를 든 사용자 중단의
+// TTFT/wait/think, measurement log, provider 실행 뒤 재시도 금지, 도구를 든 사용자 중단의
 // 정착이 전부 그 transport 전용 코드였다.
 //
 // 여기는 transport 를 구현하지 않는다. native `stream`/`streamSimple` 을 그대로
@@ -27,6 +27,7 @@ import { recordSpeedIndexCall, speedIndexStore } from "./speed-index-store.mjs";
 const { isCursorExecResolved } = await import(
   pathToFileURL(senpiNested("@earendil-works/pi-ai/dist/utils/block-symbols.js")).href
 );
+
 export { isCursorExecResolved };
 
 /** 엔진(agent-session TURN_RETRY_SUPPRESSION_PREFIX)이 재시도 금지로 읽는 접두사. */
@@ -56,6 +57,20 @@ function terminalMessage(event) {
   return event?.type === "done" ? event.message : event?.error;
 }
 
+/**
+ * provider 가 model stream 안에서 이미 tool 을 실행했는지 본다.
+ *
+ * 일반 toolCall delta 는 agent loop 가 terminal `toolUse` 뒤에 실행하므로 transport
+ * 오류 시점에는 부작용이 없다. 반면 Cursor exec-channel block 은 provider 가
+ * 요청 도중 tool 을 직접 실행하며, 시작과 함께 `kCursorExecResolved`를 붙인다.
+ * 그 턴을 다시 보내면 실행 완료 여부와 관계없이 같은 부작용을 두 번 낼 수 있다.
+ */
+function hasProviderExecutedTool(message) {
+  return (message?.content ?? []).some(
+    (part) => part?.type === "toolCall" && isCursorExecResolved(part),
+  );
+}
+
 /**
  * 이 tool block 의 인자가 실행 가능한 상태인지 본다. 아무것도 바꾸지 않는다.
  *
@@ -185,6 +200,7 @@ function createCallState(model, options, modelId) {
     firstReasoningAtMs: undefined,
     firstTextAtMs: undefined,
     emittedDelta: false,
+    providerExecutedTool: false,
     // recorder 내부 dedupe 에 기대지 않는다. 계약은 "logical call 당 최대 한 번"이고,
     // 그 계약을 지키는 주체가 이 decorator 다.
     firstOutputRecorded: false,
@@ -292,6 +308,13 @@ function isReplayableContent(event) {
 
 function observeDelta(state, event) {
   if (isReplayableContent(event)) state.emittedDelta = true;
+  if (
+    !state.providerExecutedTool &&
+    (isCursorExecResolved(event?.toolCall) ||
+      hasProviderExecutedTool(event?.partial))
+  ) {
+    state.providerExecutedTool = true;
+  }
   // TTFT 는 빈 start/end 프레임이 아니라 사용자가 실제로 볼 첫 내용이다.
   const isContentDelta = event.type === "text_delta" ||
     event.type === "thinking_delta" || event.type === "toolcall_delta";
@@ -331,7 +354,8 @@ function attachTiming(state, message) {
  * 종료 정착표(설계 문서)를 여기서 지킨다:
  * - 텍스트·도구 델타 전 오류 → 그대로 error, 재시도 허용
  * - 사고만 나간 오류 → 그대로 error, 재시도 허용 (사고는 재시도가 두 번 그리지 않는다)
- * - 텍스트·도구 델타 후 오류 → `senpi:no-turn-retry:` 로 재시도 금지
+ * - text/thinking/일반 toolCall 델타 뒤 오류 → 엔진의 제한 재시도에 위임
+ * - provider 내부 tool 실행 뒤 오류 → `senpi:no-turn-retry:` 로 재시도 금지
  * - 사용자 중단 + 완성된 미실행 tool → `toolUse` done
  * - 전송 실패는 도구가 있어도 성공으로 바꾸지 않는다
  */
@@ -374,11 +398,16 @@ function settleTerminal(state, options, event) {
       message.errorMessage = reason.startsWith(NO_TURN_RETRY_PREFIX) ? reason : `${NO_TURN_RETRY_PREFIX}${reason}`;
       return event;
     }
-    // 이미 화면에 나간 텍스트·도구가 있으면 같은 턴을 다시 보낼 수 없다 —
-    // 업스트림은 이미 토큰을 태웠고, 재시도하면 같은 텍스트가 두 번 나온다.
-    // 사고만 나간 경우는 여기 오지 않는다: 재시도가 사고를 다시 그리더라도
-    // 실패한 턴은 엔진이 걷어내고, Codex 는 SSE 로 넘어간다.
-    if (state.emittedDelta && rawError && !rawError.startsWith(NO_TURN_RETRY_PREFIX)) {
+    // 실패한 assistant 는 AgentSession 이 active branch 에서 걷어낸 뒤 같은
+    // model call 만 제한 재시도한다. 앞 턴의 toolResult 는 남고, 아직 terminal
+    // `toolUse`가 되지 않은 일반 toolCall 은 실행된 적이 없으므로 재시도해도 안전하다.
+    // provider 가 stream 안에서 tool 을 직접 실행한 경우만 같은 부작용을 막는다.
+    if (
+      state.emittedDelta &&
+      (state.providerExecutedTool || hasProviderExecutedTool(message)) &&
+      rawError &&
+      !rawError.startsWith(NO_TURN_RETRY_PREFIX)
+    ) {
       message.errorMessage = `${NO_TURN_RETRY_PREFIX}${rawError}`;
     }
     return event;

harness/rubato-pi/test/unit/rubato-stream.test.mjs

@@ -1,12 +1,13 @@
 // `withRubatoStream` 의 계약.
 //
 // 이 decorator 는 transport 가 아니다. native stream 을 그대로 위임하면서 Rubato
-// 고유 의미(계측 한 번, timing, 델타 후 재시도 금지, 중단 정착)만 얹는다. 그래서
+// 고유 의미(계측 한 번, timing, provider 실행 뒤 재시도 금지, 중단 정착)만 얹는다. 그래서
 // 여기서 지키는 것은 "무엇을 더 하는가"보다 **무엇을 잃지 않는가**다.
 import assert from "node:assert/strict";
+import { join } from "node:path";
 import test from "node:test";
 import { pathToFileURL } from "node:url";
-import { senpiNested } from "../../src/engine-paths.mjs";
+import { senpiDir, senpiNested } from "../../src/engine-paths.mjs";
 import {
   kRubatoStream,
   measurementBodyFromContext,
@@ -25,6 +26,9 @@ const { createAssistantMessageEventStream } = await import(
 const { kCursorExecResolved } = await import(
   pathToFileURL(senpiNested("@earendil-works/pi-ai/dist/utils/block-symbols.js")).href
 );
+const { AgentSession } = await import(
+  pathToFileURL(join(senpiDir, "dist/core/agent-session.js")).href
+);
 
 const model = { provider: "openai-codex", id: "gpt-5.6-sol" };
 const context = { messages: [] };
@@ -180,7 +184,7 @@ test("timing 은 성공 턴에만 붙고 벽시계/단조시계를 주입받는
   });
 });
 
-test("델타 전 오류는 재시도 가능하고, 델타 후 오류는 재시도를 막는다", async () => {
+test("provider tool 실행이 없는 오류는 text delta 뒤에도 재시도 가능하다", async () => {
   const before = assistant({ stopReason: "error", errorMessage: "terminated" });
   const beforeLast = (await drain(withRubatoStream(scriptedStream([
     { type: "start", partial: before },
@@ -195,10 +199,10 @@ test("델타 전 오류는 재시도 가능하고, 델타 후 오류는 재시
     { type: "text_delta", contentIndex: 0, delta: "안녕" },
     { type: "error", reason: "error", error: after },
   ]))(model, context, { env: {} }))).at(-1);
-  assert.equal(afterLast.error.errorMessage, "senpi:no-turn-retry:terminated");
+  assert.equal(afterLast.error.errorMessage, "terminated");
 });
 
-test("사고 델타만 있으면 전송 오류는 재시도하고, 텍스트·도구가 나간 뒤에만 막는다", async () => {
+test("사고나 text delta 뒤 WebSocket 오류는 실제 AgentSession 제한 재시도로 넘어간다", async () => {
   const thinking = assistant({ stopReason: "error", errorMessage: "WebSocket error" });
   const thinkingLast = (await drain(withRubatoStream(scriptedStream([
     { type: "start", partial: thinking },
@@ -214,7 +218,54 @@ test("사고 델타만 있으면 전송 오류는 재시도하고, 텍스트·
     { type: "text_delta", contentIndex: 1, delta: "안녕" },
     { type: "error", reason: "error", error: afterText },
   ]))(model, context, { env: {} }))).at(-1);
-  assert.equal(afterTextLast.error.errorMessage, "senpi:no-turn-retry:WebSocket error");
+  assert.equal(afterTextLast.error.errorMessage, "WebSocket error");
+  assert.equal(
+    AgentSession.prototype._isRetryableError.call(
+      { model: { contextWindow: 200_000 } },
+      {
+        ...afterTextLast.error,
+        usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0 },
+      },
+    ),
+    true,
+    "decorator가 접두사를 빼면 실제 AgentSession이 제한 재시도를 맡아야 한다",
+  );
+});
+
+test("Cursor tool 실행 표지는 future projector가 terminal block을 잃어도 latch에 남는다", async () => {
+  // pinned Cursor는 partial과 terminal에 같은 message 객체를 쓴다. split-object는
+  // 앞으로 stream projector가 block을 다시 만들더라도 fail-open하지 않게 하는 계약이다.
+  const executed = {
+    type: "toolCall",
+    id: "t1",
+    name: "read_file",
+    arguments: { path: "/tmp/a" },
+    [kCursorExecResolved]: true,
+  };
+  const partial = assistant({ content: [executed] });
+  const message = assistant({
+    stopReason: "error",
+    errorMessage: "WebSocket error",
+  });
+  const last = (await drain(withRubatoStream(scriptedStream([
+    { type: "start", partial },
+    { type: "toolcall_start", contentIndex: 0, partial },
+    { type: "toolcall_end", contentIndex: 0, toolCall: executed, partial },
+    { type: "error", reason: "error", error: message },
+  ]))(model, context, { env: {} }))).at(-1);
+
+  assert.equal(last.error.errorMessage, "senpi:no-turn-retry:WebSocket error");
+  assert.equal(
+    AgentSession.prototype._isRetryableError.call(
+      { model: { contextWindow: 200_000 } },
+      {
+        ...last.error,
+        usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0 },
+      },
+    ),
+    false,
+    "provider가 이미 실행한 tool은 실제 AgentSession에서도 재시도하면 안 된다",
+  );
 });
 
 test("사용자 중단 + 완성된 미실행 tool 은 toolUse 로 정착한다", async () => {
@@ -247,7 +298,19 @@ test("중단이 아닌 전송 실패는 도구가 있어도 성공으로 바꾸
     { type: "error", reason: "error", error: message },
   ]))(model, context, { env: {} }))).at(-1);
   assert.equal(last.type, "error");
-  assert.ok(last.error.errorMessage.startsWith("senpi:no-turn-retry:"));
+  assert.equal(last.error.errorMessage, "terminated");
+  assert.equal(last.error.content[0].partialJson, '{"path":"/tm', "잘린 tool을 성공으로 정착시키면 안 된다");
+  assert.equal(
+    AgentSession.prototype._isRetryableError.call(
+      { model: { contextWindow: 200_000 } },
+      {
+        ...last.error,
+        usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0 },
+      },
+    ),
+    true,
+    "실행되지 않은 일반 toolCall은 실제 AgentSession의 제한 재시도로 넘어가야 한다",
+  );
 });
 
 test("pi-ai 가 표지한 exec-resolved block 은 실행할 tool 로 세지 않는다", () => {