Merge pull request #1478 from pikasTech/fix/issue-1422-terminal-effects-async

fix: 异步化终态 turn status 副作用

internal/cloud/server-agent-chat.test.ts

@@ -2399,6 +2399,15 @@ test("cloud api turn status skips AgentRun refresh for complete terminal evidenc
 test("cloud api turn status records terminal side effects once (#1422)", async () => {
   const ownerCalls = [];
   const factCalls = [];
+  let releaseOwnerWrite;
+  let ownerWriteReleased = false;
+  const ownerWrite = new Promise((resolve) => {
+    releaseOwnerWrite = () => {
+      if (ownerWriteReleased) return;
+      ownerWriteReleased = true;
+      resolve();
+    };
+  });
   const traceId = "trc_issue1422_terminal_effects_once";
   const runId = "run_issue1422_terminal_effects_once";
   const commandId = "cmd_issue1422_terminal_effects_once";
@@ -2461,6 +2470,7 @@ test("cloud api turn status records terminal side effects once (#1422)", async (
       },
       async recordAgentSessionOwner(input) {
         ownerCalls.push(input);
+        await ownerWrite;
         return { ok: true, sessionId: input.sessionId };
       }
     }
@@ -2469,19 +2479,31 @@ test("cloud api turn status records terminal side effects once (#1422)", async (
 
   try {
     const { port } = server.address();
-    for (let i = 0; i < 2; i += 1) {
+    const fetchTurnStatus = async () => {
-      const response = await fetch(`http://127.0.0.1:${port}/v1/agent/turns/${traceId}`, {
+      const response = await Promise.race([
-        headers: { cookie: "hwlab_session=test-stub-session" }
+        fetch(`http://127.0.0.1:${port}/v1/agent/turns/${traceId}`, {
-      });
+          headers: { cookie: "hwlab_session=test-stub-session" }
+        }),
+        delay(250).then(() => null)
+      ]);
+      assert.ok(response, "turn status should not wait for terminal side effects to finish");
       assert.equal(response.status, 200);
       const body = await response.json();
       assert.equal(body.status, "completed");
       assert.equal(body.finalResponse.text, finalText);
+    };
+    for (let i = 0; i < 2; i += 1) {
+      await fetchTurnStatus();
     }
+    for (let i = 0; i < 20 && ownerCalls.length < 1; i += 1) await delay(10);
     assert.equal(ownerCalls.length, 1);
     assert.equal(factCalls.length, 1);
+    assert.equal(codeAgentChatResults.get(traceId).turnStatusTerminalEffects.pending, true);
+    releaseOwnerWrite();
+    for (let i = 0; i < 20 && codeAgentChatResults.get(traceId).turnStatusTerminalEffects?.recorded !== true; i += 1) await delay(10);
     assert.equal(codeAgentChatResults.get(traceId).turnStatusTerminalEffects.recorded, true);
   } finally {
+    releaseOwnerWrite?.();
     await new Promise((resolve, reject) => {
       server.close((error) => (error ? reject(error) : resolve()));
     });

internal/cloud/server-code-agent-http.ts

@@ -1222,7 +1222,7 @@ async function resolveCodeAgentTurnStatusSnapshot(traceId, options) {
       turnRefreshSatisfiedByResultSync = synced.eventsRefreshed === true || synced.resultSynced === true || synced.terminalRefreshSkipped === true;
       if (result && !canAccessOwnedResult(result, options.actor)) return forbiddenTurnSnapshot(traceId);
       if (result && isTraceCommandTerminalStatus(result.status)) {
-        await recordCodeAgentTerminalTurnStatusEffects({ payload: result, params: result, options });
+        scheduleCodeAgentTerminalTurnStatusEffects({ payload: result, params: result, options });
       }
     } catch (error) {
       resultPollError = error;
@@ -1263,7 +1263,7 @@ async function resolveCodeAgentTurnStatusSnapshot(traceId, options) {
         agentRunResult = synced.result ?? agentRunResult;
       }
       if (isTraceCommandTerminalStatus(agentRunResult?.status)) {
-        await recordCodeAgentTerminalTurnStatusEffects({ payload: agentRunResult, params: agentRunResult, options, preserveLastTraceId: true });
+        scheduleCodeAgentTerminalTurnStatusEffects({ payload: agentRunResult, params: agentRunResult, options, preserveLastTraceId: true });
       }
     } catch (error) {
       refreshError = error;
@@ -1788,20 +1788,64 @@ async function recordCodeAgentTerminalTurnStatusEffects({ payload = {}, params =
   const ownerSettled = !ownerRequired || Boolean(owner);
   const effects = {
     recorded: billingSettled && ownerSettled,
+    pending: false,
     billingSettled,
     ownerSettled,
     preserveLastTraceId: Boolean(preserveLastTraceId),
     recordedAt: new Date().toISOString(),
     valuesPrinted: false
   };
-  if (effects.recorded) {
+  payload.turnStatusTerminalEffects = effects;
-    payload.turnStatusTerminalEffects = effects;
+  const traceId = safeTraceId(payload.traceId ?? params.traceId);
-    const traceId = safeTraceId(payload.traceId ?? params.traceId);
+  if (traceId) options.codeAgentChatResults?.set?.(traceId, payload);
-    if (traceId) options.codeAgentChatResults?.set?.(traceId, payload);
-  }
   return effects;
 }
 
+function scheduleCodeAgentTerminalTurnStatusEffects({ payload = {}, params = {}, options = {}, preserveLastTraceId = false } = {}) {
+  if (!payload || typeof payload !== "object" || !isTraceCommandTerminalStatus(payload.status)) return null;
+  const existing = payload.turnStatusTerminalEffects;
+  if (existing?.recorded === true || existing?.pending === true) return existing;
+  const traceId = safeTraceId(payload.traceId ?? params.traceId);
+  const pending = {
+    recorded: false,
+    pending: true,
+    billingSettled: false,
+    ownerSettled: false,
+    preserveLastTraceId: Boolean(preserveLastTraceId),
+    scheduledAt: new Date().toISOString(),
+    valuesPrinted: false
+  };
+  payload.turnStatusTerminalEffects = pending;
+  if (traceId) options.codeAgentChatResults?.set?.(traceId, payload);
+  setImmediate(() => {
+    void recordCodeAgentTerminalTurnStatusEffects({ payload, params, options, preserveLastTraceId }).catch((error) => {
+      const failed = {
+        recorded: false,
+        pending: false,
+        billingSettled: false,
+        ownerSettled: false,
+        preserveLastTraceId: Boolean(preserveLastTraceId),
+        errorCode: error?.code ?? "terminal_turn_status_effects_failed",
+        recordedAt: new Date().toISOString(),
+        valuesPrinted: false
+      };
+      payload.turnStatusTerminalEffects = failed;
+      if (traceId) {
+        options.codeAgentChatResults?.set?.(traceId, payload);
+        (options.traceStore ?? defaultCodeAgentTraceStore).append(traceId, {
+          type: "turn-status",
+          status: "degraded",
+          label: "turn-status:terminal-effects-failed",
+          errorCode: failed.errorCode,
+          message: error?.message ?? "Terminal turn status side effects failed and will retry on the next poll.",
+          valuesPrinted: false
+        });
+      }
+    });
+  });
+  return pending;
+}
+
 function codeAgentTerminalBillingSettled({ payload = {}, params = {}, options = {}, billing = null } = {}) {
   const reservation = params.userBillingReservation ?? payload.userBillingReservation;
   const reservationId = typeof reservation?.reservationId === "string" ? reservation.reservationId : "";