Merge pull request #44 from MaxLinCode/codex/turn-routing-policy-split

Land conversation state and routing layers

Author: MaxLinCode

apps/web/src/app/api/telegram/webhook/route.test.ts

@@ -1,5 +1,5 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import type { InboxPlanningOutput } from "@atlas/core";
+import type { InboxPlanningOutput, RoutedTurn, TurnPolicyAction, TurnInterpretationType } from "@atlas/core";
 import {
   getDefaultGoogleCalendarConnectionStore,
   getDefaultFollowUpRuntimeStore,
@@ -108,6 +108,37 @@ function buildRequest(body: unknown, secret = "test-webhook-secret") {
   });
 }
 
+function buildRoutedTurn(input: {
+  turnType: TurnInterpretationType;
+  action: TurnPolicyAction;
+  confidence?: number;
+  ambiguity?: "none" | "low" | "high";
+  resolvedProposalId?: string;
+}) : RoutedTurn {
+  return {
+    interpretation: {
+      turnType: input.turnType,
+      confidence: input.confidence ?? 0.9,
+      resolvedEntityIds: [],
+      ...(input.resolvedProposalId ? { resolvedProposalId: input.resolvedProposalId } : {}),
+      ambiguity: input.ambiguity ?? "none"
+    },
+    policy: {
+      action: input.action,
+      reason: "test",
+      requiresWrite: input.action === "execute_mutation" || input.action === "recover_and_execute",
+      requiresConfirmation: input.action === "present_proposal",
+      useMutationPipeline: input.action === "execute_mutation" || input.action === "recover_and_execute",
+      ...(input.resolvedProposalId ? { targetProposalId: input.resolvedProposalId } : {}),
+      ...(input.action === "recover_and_execute"
+        ? { mutationInputSource: "recovered_proposal" as const }
+        : input.action === "execute_mutation"
+          ? { mutationInputSource: "direct_user_turn" as const }
+          : {})
+    }
+  };
+}
+
 async function seedOutstandingFollowUpBundle(titles: string[]) {
   const store = getDefaultInboxProcessingStore();
   const followUpStore = getDefaultFollowUpRuntimeStore();
@@ -749,10 +780,9 @@ describe("telegram webhook route", () => {
         store: getDefaultInboxProcessingStore(),
         calendar: getDefaultCalendarAdapter(),
         primeProcessingStore: seedInboxItemForProcessingTests,
-        turnRouter: async () => ({
-          route: "mutation",
-          reason: "Direct scheduling request.",
-          writesAllowed: true
+        turnRouter: async () => buildRoutedTurn({
+          turnType: "planning_request",
+          action: "execute_mutation"
         })
       }
     );
@@ -863,10 +893,10 @@ describe("telegram webhook route", () => {
           summary: "The assistant proposed a concrete 3pm schedule and the user is now confirming it."
         }),
         confirmedMutationRecoverer,
-        turnRouter: async () => ({
-          route: "confirmed_mutation",
-          reason: "The user is confirming a recent concrete proposal.",
-          writesAllowed: true
+        turnRouter: async () => buildRoutedTurn({
+          turnType: "confirmation",
+          action: "recover_and_execute",
+          resolvedProposalId: "proposal-1"
         })
       }
     );
@@ -1019,10 +1049,10 @@ describe("telegram webhook route", () => {
           reason: "The user refined the recent concrete proposal.",
           userReplyMessage: "Done - I've moved it to 4pm."
         }),
-        turnRouter: async () => ({
-          route: "confirmed_mutation",
-          reason: "The user is refining a recent concrete proposal.",
-          writesAllowed: true
+        turnRouter: async () => buildRoutedTurn({
+          turnType: "confirmation",
+          action: "recover_and_execute",
+          resolvedProposalId: "proposal-1"
         })
       }
     );
@@ -1155,10 +1185,10 @@ describe("telegram webhook route", () => {
           reason: "The latest turn clearly reports completion of the recent journaling task.",
           userReplyMessage: "Got it."
         }),
-        turnRouter: async () => ({
-          route: "confirmed_mutation",
-          reason: "The latest turn clearly completes one recent task.",
-          writesAllowed: true
+        turnRouter: async () => buildRoutedTurn({
+          turnType: "confirmation",
+          action: "recover_and_execute",
+          resolvedProposalId: "proposal-1"
         })
       }
     );
@@ -1223,10 +1253,10 @@ describe("telegram webhook route", () => {
           reason: "I have two recent proposals in view. Which one do you want me to apply?",
           userReplyMessage: "I have two recent proposals in view. Which one do you want me to apply?"
         }),
-        turnRouter: async () => ({
-          route: "confirmed_mutation",
-          reason: "This looks like a confirmation but the target is ambiguous.",
-          writesAllowed: true
+        turnRouter: async () => buildRoutedTurn({
+          turnType: "confirmation",
+          action: "recover_and_execute",
+          resolvedProposalId: "proposal-1"
         })
       }
     );
@@ -1405,10 +1435,9 @@ describe("telegram webhook route", () => {
           primeProcessingStore: seedInboxItemForProcessingTests,
           conversationMemorySummarizer,
           conversationResponder,
-          turnRouter: async () => ({
-            route: "conversation",
-            reason: "Planning dialogue request.",
-            writesAllowed: false
+          turnRouter: async () => buildRoutedTurn({
+            turnType: "informational",
+            action: "reply_only"
           })
         }
       );
@@ -1469,6 +1498,7 @@ describe("telegram webhook route", () => {
       });
       expect(conversationResponder).toHaveBeenCalledWith({
         route: "conversation",
+        rawText: "How should I prioritize this week?",
         normalizedText: "How should I prioritize this week?",
         recentTurns: [
           {
@@ -1557,10 +1587,11 @@ describe("telegram webhook route", () => {
           summary: "The user is discussing a possible move to tomorrow morning."
         }),
         conversationResponder,
-        turnRouter: async () => ({
-          route: "conversation_then_mutation",
-          reason: "Mixed turn should discuss first.",
-          writesAllowed: false
+        turnRouter: async () => buildRoutedTurn({
+          turnType: "planning_request",
+          action: "present_proposal",
+          confidence: 0.68,
+          ambiguity: "low"
         })
       }
     );
@@ -1586,6 +1617,7 @@ describe("telegram webhook route", () => {
     expect(listOutgoingBotEventsForTests()).toHaveLength(1);
     expect(conversationResponder).toHaveBeenCalledWith({
       route: "conversation_then_mutation",
+      rawText: "Could we move it to tomorrow morning?",
       normalizedText: "Could we move it to tomorrow morning?",
       recentTurns: [
         {
@@ -1647,17 +1679,17 @@ describe("telegram webhook route", () => {
           throw new Error("summary unavailable");
         },
         conversationResponder,
-        turnRouter: async () => ({
-          route: "conversation",
-          reason: "Discussion turn.",
-          writesAllowed: false
+        turnRouter: async () => buildRoutedTurn({
+          turnType: "informational",
+          action: "reply_only"
         })
       }
     );
 
     expect(response.status).toBe(200);
     expect(conversationResponder).toHaveBeenCalledWith({
       route: "conversation",
+      rawText: "Should we talk this through first?",
       normalizedText: "Should we talk this through first?",
       recentTurns: [
         {
@@ -1713,10 +1745,9 @@ describe("telegram webhook route", () => {
         conversationResponder: async () => ({
           reply: "From our recent exchange, it sounds like you mean the dentist reminder, but I am not treating that as confirmed state."
         }),
-        turnRouter: async () => ({
-          route: "conversation",
-          reason: "Write-adjacent question still needs cautious conversational handling.",
-          writesAllowed: false
+        turnRouter: async () => buildRoutedTurn({
+          turnType: "informational",
+          action: "reply_only"
         })
       }
     );

apps/web/src/lib/server/conversation-response.test.ts

@@ -10,6 +10,7 @@ describe("conversation response service", () => {
     const result = await buildConversationResponse(
       {
         route: "conversation",
+        rawText: "How should I prioritize this week?",
         normalizedText: "How should I prioritize this week?",
         recentTurns: [
           {
@@ -28,6 +29,7 @@ describe("conversation response service", () => {
     expect(result.reply).toContain("deadline-driven work");
     expect(respond).toHaveBeenCalledWith({
       route: "conversation",
+      rawText: "How should I prioritize this week?",
       normalizedText: "How should I prioritize this week?",
       recentTurns: [
         {
@@ -44,6 +46,7 @@ describe("conversation response service", () => {
     const result = await buildConversationResponse(
       {
         route: "conversation_then_mutation",
+        rawText: "Could we move it to tomorrow morning?",
         normalizedText: "Could we move it to tomorrow morning?",
         recentTurns: [
           {
@@ -68,10 +71,11 @@ describe("conversation response service", () => {
     await expect(
       buildConversationResponse({
         route: "conversation",
+        rawText: "   ",
         normalizedText: "   ",
         recentTurns: [],
         memorySummary: null
       })
-    ).rejects.toThrow("must include non-empty normalizedText");
+    ).rejects.toThrow("must include non-empty rawText and normalizedText");
   });
 });

apps/web/src/lib/server/conversation-response.ts

@@ -22,8 +22,8 @@ export async function buildConversationResponse(
 }
 
 function parseConversationResponseInput(input: BuildConversationResponseInput): BuildConversationResponseInput {
-  if (!input.normalizedText.trim()) {
-    throw new Error("Conversation response input must include non-empty normalizedText.");
+  if (!input.rawText.trim() || !input.normalizedText.trim()) {
+    throw new Error("Conversation response input must include non-empty rawText and normalizedText.");
   }
 
   return input;

apps/web/src/lib/server/conversation-state.ts

@@ -10,14 +10,17 @@ import {
   type PresentedItem,
   type ScheduleBlock,
   type Task,
-  type TurnRoute
+  type TurnInterpretation,
+  type TurnPolicyAction
 } from "@atlas/core";
 import { type ProcessedInboxResult } from "@atlas/db";
 
 type DeriveConversationReplyStateInput = {
   snapshot: ConversationStateSnapshot;
-  route: Extract<TurnRoute, "conversation" | "conversation_then_mutation">;
+  policyAction: Extract<TurnPolicyAction, "reply_only" | "ask_clarification" | "present_proposal">;
+  interpretation: TurnInterpretation;
   reply: string;
+  userTurnText: string;
   summaryText: string | null;
   occurredAt?: string;
 };
@@ -36,16 +39,22 @@ export function deriveConversationReplyState(input: DeriveConversationReplyState
   const newClarifications: PendingClarification[] = [];
   let nextFocusEntityId: string | null | undefined;
 
-  if (input.route === "conversation_then_mutation") {
+  if (input.policyAction === "present_proposal") {
     const proposalEntity = buildConversationEntity(input.snapshot.conversation.id, {
       kind: "proposal_option",
       label: summarizeLabel(input.reply),
       status: "active",
       createdAt: occurredAt,
       updatedAt: occurredAt,
       data: {
-        route: input.route,
-        replyText: input.reply
+        route: "conversation_then_mutation",
+        replyText: input.reply,
+        policyAction: input.policyAction,
+        targetEntityId: input.interpretation.resolvedEntityIds[0] ?? null,
+        mutationInputSource: null,
+        confirmationRequired: true,
+        originatingTurnText: input.userTurnText,
+        missingSlots: input.interpretation.missingSlots
       }
     });
 
@@ -60,7 +69,7 @@ export function deriveConversationReplyState(input: DeriveConversationReplyState
     nextFocusEntityId = proposalEntity.id;
   }
 
-  if (looksLikeClarification(input.reply)) {
+  if (input.policyAction === "ask_clarification") {
     const clarificationEntity = buildConversationEntity(input.snapshot.conversation.id, {
       kind: "clarification",
       label: summarizeLabel(input.reply),
@@ -90,13 +99,13 @@ export function deriveConversationReplyState(input: DeriveConversationReplyState
     ...(presentedItems.length > 0 ? { presentedItems } : {}),
     ...(newClarifications.length > 0 ? { newClarifications } : {}),
     ...(nextFocusEntityId !== undefined ? { focusEntityId: nextFocusEntityId } : {}),
-    pendingConfirmation: input.route === "conversation_then_mutation",
+    pendingConfirmation: input.policyAction === "present_proposal",
     validEntityIds: entityRegistry.map((entity) => entity.id)
   }).state;
 
   return {
     summaryText: input.summaryText,
-    mode: input.route as ConversationRecordMode,
+    mode: getConversationModeForPolicy(input.policyAction),
     entityRegistry: trimEntityRegistry(entityRegistry),
     discourseState: nextDiscourseState
   };
@@ -254,14 +263,16 @@ function isTaskResolved(task: Task) {
   return task.lifecycleState === "done" || task.lifecycleState === "archived";
 }
 
-function looksLikeClarification(reply: string) {
-  return reply.trim().endsWith("?");
-}
-
 function summarizeLabel(text: string) {
   return text.trim().slice(0, 120);
 }
 
+function getConversationModeForPolicy(
+  policyAction: DeriveConversationReplyStateInput["policyAction"]
+): ConversationRecordMode {
+  return policyAction === "reply_only" ? "conversation" : "conversation_then_mutation";
+}
+
 function buildConversationEntity(
   conversationId: string,
   input: {