feat(service-ai): turn idempotency (turnId) — safe Retry without re-planning (ADR-0013 D1) (#1900)

A user turn becomes an idempotent unit. The client supplies a stable
`turnId` per turn (constant across Retry); the server dedups the inbound
user message by (conversationId, turnId) and short-circuits the stored
reply when the turn already completed, instead of writing a duplicate
user row and re-running the tool loop (which re-plans → renamed objects,
orphan drafts).

- ai_messages: + turn_id column + (conversation_id, turn_id) index
  (auto-reconciled by the SQL driver; no hand-written migration).
- contract: turnId on ToolExecutionContext; addMessage(…, turnId) +
  new getTurnState(conversationId, turnId) on IAIConversationService.
- conversation services (objectql + in-memory): persist turn_id on every
  message of a turn; getTurnState reports userExists + the final reply
  (an assistant message with no pending tool calls).
- ai-service: dedup + short-circuit in BOTH chatWithToolsImpl and
  streamChatWithTools; every turn message tagged with turnId.
- agent-routes: read body.turnId into toolExecutionContext.

Tests: +6 dedup/short-circuit (chat + stream) and +4 getTurnState cases;
full service-ai suite green (385).

Cross-repo: pairs with objectui chat client (turnId in request body) and
a cloud SHA bump, per the ADR-0012 cross-repo recipe. Refs cloud#334.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: os-zhuang

packages/services/service-ai/src/__tests__/ai-service.test.ts

@@ -518,6 +518,165 @@ describe('AIService', () => {
     );
     expect(result.content).toBe('still ok');
   });
+
+  // ─── Turn idempotency (ADR-0013 D1) ─────────────────────────────
+
+  it('chatWithTools re-sending the same turnId returns the stored reply without re-running the model', async () => {
+    const conversationService = new InMemoryConversationService();
+    let calls = 0;
+    const adapter: LLMAdapter = {
+      name: 'turn-idem',
+      chat: async () => {
+        calls += 1;
+        return { content: `reply #${calls}`, model: 'test' };
+      },
+      complete: async () => ({ content: '' }),
+    };
+    const service = new AIService({ adapter, conversationService, logger: silentLogger });
+    const conv = await conversationService.create();
+    const ctx = { toolExecutionContext: { conversationId: conv.id, turnId: 'turn-1' } };
+
+    const first = await service.chatWithTools([{ role: 'user', content: 'hi' }], ctx);
+    const second = await service.chatWithTools([{ role: 'user', content: 'hi' }], ctx);
+
+    // Adapter ran exactly once; the retry short-circuited the stored reply.
+    expect(calls).toBe(1);
+    expect(first.content).toBe('reply #1');
+    expect(second.content).toBe('reply #1');
+
+    // No duplicate user/assistant rows — still just the one turn.
+    const after = await conversationService.get(conv.id);
+    expect(after?.messages.map((m) => m.role)).toEqual(['user', 'assistant']);
+  });
+
+  it('chatWithTools re-sending the same turnId does NOT re-execute tools', async () => {
+    const conversationService = new InMemoryConversationService();
+    const toolRegistry = new ToolRegistry();
+    let toolCalls = 0;
+    toolRegistry.register(
+      { name: 'echo', description: 'echo', parameters: {} as any },
+      async (input: any) => {
+        toolCalls += 1;
+        return JSON.stringify({ ok: true, input });
+      },
+    );
+    let modelCalls = 0;
+    const adapter: LLMAdapter = {
+      name: 'turn-idem-tools',
+      chat: async () => {
+        modelCalls += 1;
+        if (modelCalls === 1) {
+          return {
+            content: '',
+            model: 'test',
+            toolCalls: [{ type: 'tool-call' as const, toolCallId: 'tc1', toolName: 'echo', input: { x: 1 } }],
+          };
+        }
+        return { content: 'all done', model: 'test' };
+      },
+      complete: async () => ({ content: '' }),
+    };
+    const service = new AIService({ adapter, conversationService, toolRegistry, logger: silentLogger });
+    const conv = await conversationService.create();
+    const ctx = { toolExecutionContext: { conversationId: conv.id, turnId: 'turn-1' } };
+
+    await service.chatWithTools([{ role: 'user', content: 'use echo' }], ctx);
+    expect(toolCalls).toBe(1);
+
+    // Retry the SAME completed turn → no further tool execution / planning.
+    const retry = await service.chatWithTools([{ role: 'user', content: 'use echo' }], ctx);
+    expect(retry.content).toBe('all done');
+    expect(toolCalls).toBe(1);
+    expect(modelCalls).toBe(2); // unchanged from the first turn's two rounds
+  });
+
+  it('chatWithTools dedups the inbound user message when re-running an INCOMPLETE turn', async () => {
+    const conversationService = new InMemoryConversationService();
+    // Seed an incomplete turn: a user message exists, but no assistant reply.
+    const conv = await conversationService.create();
+    await conversationService.addMessage(conv.id, { role: 'user', content: 'hi' }, undefined, 'turn-1');
+
+    let calls = 0;
+    const adapter: LLMAdapter = {
+      name: 'turn-idem-incomplete',
+      chat: async () => {
+        calls += 1;
+        return { content: 'recovered reply', model: 'test' };
+      },
+      complete: async () => ({ content: '' }),
+    };
+    const service = new AIService({ adapter, conversationService, logger: silentLogger });
+
+    const result = await service.chatWithTools(
+      [{ role: 'user', content: 'hi' }],
+      { toolExecutionContext: { conversationId: conv.id, turnId: 'turn-1' } },
+    );
+
+    // Re-ran (turn was incomplete) but did NOT write a second user row.
+    expect(calls).toBe(1);
+    expect(result.content).toBe('recovered reply');
+    const after = await conversationService.get(conv.id);
+    expect(after?.messages.map((m) => m.role)).toEqual(['user', 'assistant']);
+  });
+
+  it('chatWithTools without a turnId keeps the legacy behaviour (each call is a fresh turn)', async () => {
+    const conversationService = new InMemoryConversationService();
+    let calls = 0;
+    const adapter: LLMAdapter = {
+      name: 'no-turn',
+      chat: async () => {
+        calls += 1;
+        return { content: `reply #${calls}`, model: 'test' };
+      },
+      complete: async () => ({ content: '' }),
+    };
+    const service = new AIService({ adapter, conversationService, logger: silentLogger });
+    const conv = await conversationService.create();
+    const ctx = { toolExecutionContext: { conversationId: conv.id } };
+
+    await service.chatWithTools([{ role: 'user', content: 'hi' }], ctx);
+    await service.chatWithTools([{ role: 'user', content: 'hi' }], ctx);
+
+    // No idempotency key → both calls run and persist independently.
+    expect(calls).toBe(2);
+    const after = await conversationService.get(conv.id);
+    expect(after?.messages.map((m) => m.role)).toEqual(['user', 'assistant', 'user', 'assistant']);
+  });
+
+  it('streamChatWithTools re-sending the same turnId replays the stored reply without re-running the model', async () => {
+    const conversationService = new InMemoryConversationService();
+    let calls = 0;
+    const adapter: LLMAdapter = {
+      name: 'turn-idem-stream',
+      chat: async () => {
+        calls += 1;
+        return { content: 'streamed reply', model: 'test' };
+      },
+      complete: async () => ({ content: '' }),
+    };
+    const service = new AIService({ adapter, conversationService, logger: silentLogger });
+    const conv = await conversationService.create();
+    const ctx = { toolExecutionContext: { conversationId: conv.id, turnId: 'turn-1' } };
+
+    const drain = async () => {
+      const events: TextStreamPart<ToolSet>[] = [];
+      for await (const ev of service.streamChatWithTools([{ role: 'user', content: 'hi' }], ctx)) {
+        events.push(ev);
+      }
+      return events;
+    };
+
+    await drain();
+    const second = await drain();
+
+    expect(calls).toBe(1); // retry short-circuited; adapter not called again
+    const text = second.find((e) => (e as { type: string }).type === 'text-delta');
+    expect(text && (text as any).text).toBe('streamed reply');
+    expect(second[second.length - 1].type).toBe('finish');
+
+    const after = await conversationService.get(conv.id);
+    expect(after?.messages.map((m) => m.role)).toEqual(['user', 'assistant']);
+  });
 });
 
 // ─────────────────────────────────────────────────────────────────

packages/services/service-ai/src/__tests__/objectql-conversation-service.test.ts

@@ -415,6 +415,60 @@ describe('ObjectQLConversationService', () => {
     expect(fetched!.metadata).toBeUndefined();
   });
 
+  // ── Turn idempotency (ADR-0013 D1) ──────────────────────────────
+
+  it('persists turn_id on every message of a turn', async () => {
+    const conv = await service.create();
+    await service.addMessage(conv.id, { role: 'user', content: 'hi' }, undefined, 'turn-1');
+    await service.addMessage(conv.id, { role: 'assistant', content: 'reply' }, undefined, 'turn-1');
+
+    const rows = await engine.find('ai_messages', { where: { conversation_id: conv.id } });
+    expect(rows).toHaveLength(2);
+    expect(rows.every((r: any) => r.turn_id === 'turn-1')).toBe(true);
+  });
+
+  it('getTurnState reports a completed turn (user + final assistant reply)', async () => {
+    const conv = await service.create();
+    await service.addMessage(conv.id, { role: 'user', content: 'hi' }, undefined, 'turn-1');
+    await service.addMessage(
+      conv.id,
+      { role: 'assistant', content: 'final answer' },
+      { model: 'test', promptTokens: 3, completionTokens: 5, totalTokens: 8 },
+      'turn-1',
+    );
+
+    const state = await service.getTurnState(conv.id, 'turn-1');
+    expect(state.userExists).toBe(true);
+    expect(state.reply?.content).toBe('final answer');
+    expect(state.reply?.model).toBe('test');
+    expect(state.reply?.usage?.totalTokens).toBe(8);
+  });
+
+  it('getTurnState treats a tool-call-only assistant turn as NOT a final reply', async () => {
+    const conv = await service.create();
+    await service.addMessage(conv.id, { role: 'user', content: 'use a tool' }, undefined, 'turn-1');
+    // Assistant turn that only requested tools — carries tool_calls, no final text.
+    await service.addMessage(
+      conv.id,
+      {
+        role: 'assistant',
+        content: [{ type: 'tool-call', toolCallId: 'tc1', toolName: 'echo', input: {} }],
+      } as ModelMessage,
+      undefined,
+      'turn-1',
+    );
+
+    const state = await service.getTurnState(conv.id, 'turn-1');
+    expect(state.userExists).toBe(true);
+    expect(state.reply).toBeNull(); // turn never produced a final reply
+  });
+
+  it('getTurnState returns no reply / no user for an unknown turn', async () => {
+    const conv = await service.create();
+    const state = await service.getTurnState(conv.id, 'never-seen');
+    expect(state).toEqual({ userExists: false, reply: null });
+  });
+
   it('should handle invalid JSON in tool_calls gracefully', async () => {
     const conv = await service.create();
     await service.addMessage(conv.id, { role: 'assistant', content: 'checking tools' });

packages/services/service-ai/src/ai-service.ts

@@ -352,9 +352,10 @@ export class AIService implements IAIService {
     conversationId: string,
     message: ModelMessage,
     extras?: MessageObservability,
+    turnId?: string,
   ): Promise<void> {
     try {
-      await this.conversationService.addMessage(conversationId, message, extras);
+      await this.conversationService.addMessage(conversationId, message, extras, turnId);
     } catch (err) {
       this.logger.warn('[AI] persist message failed', {
         conversationId,
@@ -565,13 +566,34 @@ export class AIService implements IAIService {
     // Working copy of the conversation
     const conversation = [...messages];
 
+    // Turn idempotency (ADR-0013 D1). When the client supplies a stable
+    // turnId, dedup the inbound user message and short-circuit a turn that
+    // already completed instead of re-running the tool loop / re-planning.
+    const turnId = toolExecutionContext?.turnId;
+
     // Persist the inbound user turn when a conversationId is supplied.
     // Only the last message is written — callers are assumed to pass
     // prior history alongside the new turn; we don't diff.
     if (conversationId && messages.length > 0) {
       const last = messages[messages.length - 1];
       if (last && (last as { role?: string }).role === 'user') {
-        await this.persistMessage(conversationId, last);
+        if (turnId) {
+          const state = await this.conversationService.getTurnState(conversationId, turnId);
+          // Completed turn re-requested → return the stored reply, no tools.
+          if (state.reply) {
+            this.logger.debug('[AI] chatWithTools short-circuit completed turn', { turnId });
+            return autoCreatedConversationId
+              ? { ...state.reply, conversationId: autoCreatedConversationId }
+              : state.reply;
+          }
+          // Incomplete/failed turn re-requested → don't write a duplicate
+          // user row; fall through and re-run (D2 handles "already succeeded").
+          if (!state.userExists) {
+            await this.persistMessage(conversationId, last, undefined, turnId);
+          }
+        } else {
+          await this.persistMessage(conversationId, last);
+        }
       }
     }
 
@@ -602,6 +624,7 @@ export class AIService implements IAIService {
               content: result.content,
             } as ModelMessage,
             turnObservability,
+            turnId,
           );
           void this.summarizeConversation(conversationId);
         }
@@ -627,8 +650,9 @@ export class AIService implements IAIService {
       if (conversationId) {
         // Attribute usage / latency to the assistant turn that triggered
         // the tool calls — the subsequent role:'tool' messages have no
-        // LLM cost of their own.
-        await this.persistMessage(conversationId, assistantTurn, turnObservability);
+        // LLM cost of their own. Tagged with turnId but keeps tool_calls,
+        // so getTurnState never mistakes it for the turn's final reply.
+        await this.persistMessage(conversationId, assistantTurn, turnObservability, turnId);
       }
 
       // Execute all tool calls in parallel, threading the per-request
@@ -666,7 +690,7 @@ export class AIService implements IAIService {
         } as ModelMessage;
         conversation.push(toolTurn);
         if (conversationId) {
-          await this.persistMessage(conversationId, toolTurn);
+          await this.persistMessage(conversationId, toolTurn, undefined, turnId);
         }
       }
 
@@ -700,6 +724,7 @@ export class AIService implements IAIService {
           content: finalResult.content,
         } as ModelMessage,
         finalObservability,
+        turnId,
       );
       void this.summarizeConversation(conversationId);
     }
@@ -760,10 +785,28 @@ export class AIService implements IAIService {
       } as TextStreamPart<ToolSet>;
     }
 
+    // Turn idempotency (ADR-0013 D1) — see chatWithToolsImpl for the rationale.
+    const turnId = toolExecutionContext?.turnId;
+
     if (conversationId && messages.length > 0) {
       const last = messages[messages.length - 1];
       if (last && (last as { role?: string }).role === 'user') {
-        await this.persistMessage(conversationId, last);
+        if (turnId) {
+          const state = await this.conversationService.getTurnState(conversationId, turnId);
+          // Completed turn re-requested → replay the stored reply, no tools.
+          if (state.reply) {
+            this.logger.debug('[AI] streamChatWithTools short-circuit completed turn', { turnId });
+            yield textDeltaPart('stream', state.reply.content);
+            yield finishPart(state.reply);
+            return;
+          }
+          // Incomplete/failed turn → don't duplicate the user row; re-run.
+          if (!state.userExists) {
+            await this.persistMessage(conversationId, last, undefined, turnId);
+          }
+        } else {
+          await this.persistMessage(conversationId, last);
+        }
       }
     }
 
@@ -783,6 +826,7 @@ export class AIService implements IAIService {
               content: result.content,
             } as ModelMessage,
             turnObservability,
+            turnId,
           );
           void this.summarizeConversation(conversationId);
         }
@@ -805,7 +849,7 @@ export class AIService implements IAIService {
       } as ModelMessage;
       conversation.push(assistantTurn);
       if (conversationId) {
-        await this.persistMessage(conversationId, assistantTurn, turnObservability);
+        await this.persistMessage(conversationId, assistantTurn, turnObservability, turnId);
       }
 
       // Drain tool PROGRESS events (emitted via `ctx.onProgress` during a
@@ -872,7 +916,7 @@ export class AIService implements IAIService {
         } as ModelMessage;
         conversation.push(toolTurn);
         if (conversationId) {
-          await this.persistMessage(conversationId, toolTurn);
+          await this.persistMessage(conversationId, toolTurn, undefined, turnId);
         }
       }
 
@@ -899,6 +943,7 @@ export class AIService implements IAIService {
           content: result.content,
         } as ModelMessage,
         finalObservability,
+        turnId,
       );
       void this.summarizeConversation(conversationId);
     }