Fix testcase generation message flooding by using internalMessages

frontend/internal-packages/agent/README.md

@@ -234,47 +234,50 @@ The `qaAgent` node is implemented as a **LangGraph subgraph** that encapsulates
 %%{init: {'flowchart': {'curve': 'linear'}}}%%
 graph TD;
 	__start__([<p>__start__</p>]):::first
+	prepareTestcaseGeneration(prepareTestcaseGeneration)
 	testcaseGeneration(testcaseGeneration)
 	applyGeneratedSqls(applyGeneratedSqls)
 	validateSchema(validateSchema)
 	invokeRunTestTool(invokeRunTestTool)
 	analyzeTestFailures(analyzeTestFailures)
 	resetFailedSqlTests(resetFailedSqlTests)
 	__end__([<p>__end__</p>]):::last
+	__start__ --> prepareTestcaseGeneration;
 	applyGeneratedSqls --> validateSchema;
 	invokeRunTestTool --> analyzeTestFailures;
+	resetFailedSqlTests --> prepareTestcaseGeneration;
 	testcaseGeneration --> applyGeneratedSqls;
 	validateSchema --> invokeRunTestTool;
-	__start__ -.-> testcaseGeneration;
-	__start__ -.-> applyGeneratedSqls;
-	__start__ -.-> validateSchema;
-	__start__ -.-> invokeRunTestTool;
-	__start__ -.-> analyzeTestFailures;
-	__start__ -.-> resetFailedSqlTests;
-	__start__ -.-> __end__;
+	prepareTestcaseGeneration -.-> testcaseGeneration;
+	prepareTestcaseGeneration -.-> applyGeneratedSqls;
+	prepareTestcaseGeneration -.-> validateSchema;
+	prepareTestcaseGeneration -.-> invokeRunTestTool;
+	prepareTestcaseGeneration -.-> analyzeTestFailures;
+	prepareTestcaseGeneration -.-> resetFailedSqlTests;
+	prepareTestcaseGeneration -.-> __end__;
 	analyzeTestFailures -.-> resetFailedSqlTests;
 	analyzeTestFailures -.-> __end__;
-	resetFailedSqlTests -.-> testcaseGeneration;
-	resetFailedSqlTests -.-> applyGeneratedSqls;
-	resetFailedSqlTests -.-> validateSchema;
-	resetFailedSqlTests -.-> invokeRunTestTool;
-	resetFailedSqlTests -.-> analyzeTestFailures;
-	resetFailedSqlTests -.-> __end__;
 	classDef default fill:#f2f0ff,line-height:1.2;
 	classDef first fill-opacity:0;
 	classDef last fill:#bfb6fc;
 ```
 
 ### QA Agent Components
 
-#### 1. testcaseGeneration Node
+#### 1. prepareTestcaseGeneration Node
+
+- **Purpose**: Adds a start message to state before distributing requirements to parallel subgraphs
+- **Performed by**: prepareTestcaseGeneration function
+- **Output**: Updates messages state with a notification message indicating testcase generation has started
+
+#### 2. testcaseGeneration Node
 
 - **Purpose**: Implements map-reduce pattern for parallel testcase generation using a dedicated subgraph
 - **Performed by**: Multiple parallel instances of testcase generation subgraph
 - **Retry Policy**: maxAttempts: 3 (internal to subgraph)
 - **Output**: AI-generated test cases with DML operations using tool calls
 
-#### 2. applyGeneratedSqls Node
+#### 3. applyGeneratedSqls Node
 
 - **Purpose**: Maps generated SQLs from testcaseGeneration to analyzedRequirements.testcases
 - **Performed by**: applyGeneratedSqlsNode function
@@ -307,49 +310,49 @@ graph TD;
 - **generateTestcase**: Generates test cases and DML operations using GPT-5-nano with specialized prompts
 - **invokeSaveTool**: Executes saveTestcaseTool to persist generated test cases with DML operations
 
-#### 3. validateSchemaRequirements Node
+#### 4. validateSchemaRequirements Node
 
 - **Purpose**: Pre-validation node that checks if schema can fulfill requirements before test generation
 - **Performed by**: Schema validation logic with requirement analysis
 - **Retry Policy**: maxAttempts: 3 (internal to testcaseGeneration subgraph)
 - **Decision Making**: Routes to generateTestcase if sufficient, or END if schema is insufficient
 
-#### 4. generateTestcase Node
+#### 5. generateTestcase Node
 
 - **Purpose**: Generates test cases and DML operations for a single requirement
 - **Performed by**: GPT-5-nano with specialized test case generation prompts
 - **Retry Policy**: maxAttempts: 3 (internal to testcaseGeneration subgraph)
 - **Tool Integration**: Uses saveTestcaseTool for structured test case output
 
-#### 5. invokeSaveTool Node
+#### 6. invokeSaveTool Node
 
 - **Purpose**: Executes saveTestcaseTool to persist generated test cases
 - **Performed by**: ToolNode with saveTestcaseTool
 - **Retry Policy**: maxAttempts: 3 (internal to testcaseGeneration subgraph)
 - **Output**: Saves test cases with DML operations to workflow state
 
-#### 6. validateSchema Node
+#### 7. validateSchema Node
 
 - **Purpose**: Creates AI message to trigger test execution for schema validation
 - **Performed by**: validateSchemaNode function
 - **Retry Policy**: maxAttempts: 3 (internal to subgraph)
 - **Output**: Generates tool call for runTestTool execution
 
-#### 7. invokeRunTestTool Node
+#### 8. invokeRunTestTool Node
 
 - **Purpose**: Executes DML statements and validates schema functionality
 - **Performed by**: ToolNode with runTestTool
 - **Retry Policy**: maxAttempts: 3 (internal to subgraph)
 - **Validation**: Schema integrity and DML execution results
 
-#### 8. analyzeTestFailures Node
+#### 9. analyzeTestFailures Node
 
 - **Purpose**: Analyzes test execution results and identifies failed tests for retry
 - **Performed by**: analyzeTestFailuresNode function
 - **Output**: Sets `failureAnalysis` with `failedSqlTestIds` and `failedSchemaTestIds` arrays
 - **Routing**: Routes to `resetFailedSqlTests` if failures exist, otherwise to `END`
 
-#### 9. resetFailedSqlTests Node
+#### 10. resetFailedSqlTests Node
 
 - **Purpose**: Resets SQL fields for failed tests to enable regeneration
 - **Performed by**: resetFailedSqlTestsNode function

frontend/internal-packages/agent/src/qa-agent/createQaAgentGraph.test.ts

@@ -7,32 +7,29 @@ describe('createQaAgentGraph', () => {
   const expectedMermaidDiagram = `%%{init: {'flowchart': {'curve': 'linear'}}}%%
 graph TD;
 	__start__([<p>__start__</p>]):::first
+	prepareTestcaseGeneration(prepareTestcaseGeneration)
 	testcaseGeneration(testcaseGeneration)
 	applyGeneratedSqls(applyGeneratedSqls)
 	validateSchema(validateSchema)
 	invokeRunTestTool(invokeRunTestTool)
 	analyzeTestFailures(analyzeTestFailures)
 	resetFailedSqlTests(resetFailedSqlTests)
 	__end__([<p>__end__</p>]):::last
+	__start__ --> prepareTestcaseGeneration;
 	applyGeneratedSqls --> validateSchema;
 	invokeRunTestTool --> analyzeTestFailures;
+	resetFailedSqlTests --> prepareTestcaseGeneration;
 	testcaseGeneration --> applyGeneratedSqls;
 	validateSchema --> invokeRunTestTool;
-	__start__ -.-> testcaseGeneration;
-	__start__ -.-> applyGeneratedSqls;
-	__start__ -.-> validateSchema;
-	__start__ -.-> invokeRunTestTool;
-	__start__ -.-> analyzeTestFailures;
-	__start__ -.-> resetFailedSqlTests;
-	__start__ -.-> __end__;
+	prepareTestcaseGeneration -.-> testcaseGeneration;
+	prepareTestcaseGeneration -.-> applyGeneratedSqls;
+	prepareTestcaseGeneration -.-> validateSchema;
+	prepareTestcaseGeneration -.-> invokeRunTestTool;
+	prepareTestcaseGeneration -.-> analyzeTestFailures;
+	prepareTestcaseGeneration -.-> resetFailedSqlTests;
+	prepareTestcaseGeneration -.-> __end__;
 	analyzeTestFailures -.-> resetFailedSqlTests;
 	analyzeTestFailures -.-> __end__;
-	resetFailedSqlTests -.-> testcaseGeneration;
-	resetFailedSqlTests -.-> applyGeneratedSqls;
-	resetFailedSqlTests -.-> validateSchema;
-	resetFailedSqlTests -.-> invokeRunTestTool;
-	resetFailedSqlTests -.-> analyzeTestFailures;
-	resetFailedSqlTests -.-> __end__;
 	classDef default fill:#f2f0ff,line-height:1.2;
 	classDef first fill-opacity:0;
 	classDef last fill:#bfb6fc;

frontend/internal-packages/agent/src/qa-agent/createQaAgentGraph.ts

@@ -1,6 +1,9 @@
 import { END, START, StateGraph } from '@langchain/langgraph'
 import { RETRY_POLICY } from '../utils/errorHandling'
-import { continueToRequirements } from './distributeRequirements'
+import {
+  continueToRequirements,
+  prepareTestcaseGeneration,
+} from './distributeRequirements'
 import { analyzeTestFailuresNode } from './nodes/analyzeTestFailuresNode'
 import { applyGeneratedSqlsNode } from './nodes/applyGeneratedSqlsNode'
 import { invokeRunTestToolNode } from './nodes/invokeRunTestToolNode'
@@ -15,6 +18,7 @@ export const createQaAgentGraph = () => {
 
   qaAgentGraph
     // Add nodes for map-reduce pattern
+    .addNode('prepareTestcaseGeneration', prepareTestcaseGeneration)
     .addNode('testcaseGeneration', testcaseGeneration)
 
     .addNode('applyGeneratedSqls', applyGeneratedSqlsNode)
@@ -29,9 +33,12 @@ export const createQaAgentGraph = () => {
     .addNode('resetFailedSqlTests', resetFailedSqlTestsNode)
 
     // Define edges for map-reduce flow
-    // Use conditional edge with Send API for parallel execution from START
-    // Send targets the testcaseGeneration
-    .addConditionalEdges(START, continueToRequirements)
+    // START → prepareTestcaseGeneration to add start message to state
+    .addEdge(START, 'prepareTestcaseGeneration')
+
+    // Use conditional edge with Send API for parallel execution
+    // Send targets the testcaseGeneration subgraph
+    .addConditionalEdges('prepareTestcaseGeneration', continueToRequirements)
 
     // After all parallel subgraph executions complete, apply generated SQLs
     .addEdge('testcaseGeneration', 'applyGeneratedSqls')
@@ -51,8 +58,8 @@ export const createQaAgentGraph = () => {
       [END]: END,
     })
 
-    // After resetting failed SQL tests, go back to testcaseGeneration to regenerate
-    .addConditionalEdges('resetFailedSqlTests', continueToRequirements)
+    // After resetting failed SQL tests, prepare again before regenerating
+    .addEdge('resetFailedSqlTests', 'prepareTestcaseGeneration')
 
   return qaAgentGraph.compile()
 }

frontend/internal-packages/agent/src/qa-agent/distributeRequirements/index.ts

@@ -1,12 +1,34 @@
+import { dispatchCustomEvent } from '@langchain/core/callbacks/dispatch'
+import { AIMessage } from '@langchain/core/messages'
 import { Send } from '@langchain/langgraph'
 import type { QaAgentState } from '../shared/qaAgentAnnotation'
 import { getUnprocessedRequirements } from './getUnprocessedRequirements'
 
 export type { TestCaseData } from './types'
 
+/**
+ * Prepare testcase generation by adding a start message to state
+ * This node runs before distributing requirements to parallel subgraphs
+ */
+export async function prepareTestcaseGeneration(state: QaAgentState) {
+  const targetTestcases = getUnprocessedRequirements(state)
+
+  const message = new AIMessage({
+    id: crypto.randomUUID(),
+    name: 'qa',
+    content: `Generating test cases (processing ${targetTestcases.length} requirements)...`,
+  })
+
+  await dispatchCustomEvent('messages', message)
+
+  return {
+    messages: [message],
+  }
+}
+
 /**
  * Conditional edge function to create Send objects for parallel processing
- * This is called directly from START node
+ * Distributes each requirement to testcaseGeneration subgraph
  */
 export function continueToRequirements(state: QaAgentState) {
   const targetTestcases = getUnprocessedRequirements(state)
@@ -20,7 +42,7 @@ export function continueToRequirements(state: QaAgentState) {
         currentTestcase: testcaseData,
         schemaData: state.schemaData,
         goal: state.analyzedRequirements.goal,
-        messages: [], // Start with empty messages for isolation
+        internalMessages: [], // Start with empty messages for isolation
       }),
   )
 }

frontend/internal-packages/agent/src/qa-agent/testcaseGeneration/generateTestcaseNode.ts

@@ -7,7 +7,6 @@ import { ChatOpenAI } from '@langchain/openai'
 import { fromAsyncThrowable } from '@liam-hq/neverthrow'
 import { yamlSchemaDeparser } from '@liam-hq/schema'
 import { removeReasoningFromMessages } from '../../utils/messageCleanup'
-import { streamLLMResponse } from '../../utils/streamingLlmUtils'
 import { saveTestcaseTool } from '../tools/saveTestcaseTool'
 import { formatPreviousFailures } from '../utils/formatPreviousFailures'
 import {
@@ -34,8 +33,8 @@ const model = new ChatOpenAI({
  */
 export async function generateTestcaseNode(
   state: typeof testcaseAnnotation.State,
-): Promise<{ messages: BaseMessage[] }> {
-  const { currentTestcase, schemaData, goal, messages } = state
+): Promise<{ internalMessages: BaseMessage[] }> {
+  const { currentTestcase, schemaData, goal, internalMessages } = state
 
   const schemaContextResult = yamlSchemaDeparser(schemaData)
   if (schemaContextResult.isErr()) {
@@ -54,39 +53,32 @@ export async function generateTestcaseNode(
     previousFailures,
   })
 
-  const cleanedMessages = removeReasoningFromMessages(messages)
+  const cleanedMessages = removeReasoningFromMessages(internalMessages)
 
-  const streamModel = fromAsyncThrowable(() => {
-    return model.stream(
+  const invokeModel = fromAsyncThrowable(() => {
+    return model.invoke(
       [
         new SystemMessage(SYSTEM_PROMPT_FOR_TESTCASE_GENERATION),
         new HumanMessage(contextMessage),
         // Include all previous messages in this subgraph's scope
         ...cleanedMessages,
       ],
       {
-        options: {
-          timeout: 120000, // 120s
-        },
+        timeout: 120000, // 120s
       },
     )
   })
 
-  const streamResult = await streamModel()
+  const result = await invokeModel()
 
-  if (streamResult.isErr()) {
+  if (result.isErr()) {
     // eslint-disable-next-line no-throw-error/no-throw-error -- Required for LangGraph retry mechanism
     throw new Error(
-      `Failed to generate SQL for ${currentTestcase.category}/${currentTestcase.testcase.title}: ${streamResult.error.message}`,
+      `Failed to generate SQL for ${currentTestcase.category}/${currentTestcase.testcase.title}: ${result.error.message}`,
     )
   }
 
-  const response = await streamLLMResponse(streamResult.value, {
-    agentName: 'qa',
-    eventType: 'messages',
-  })
-
   return {
-    messages: [response],
+    internalMessages: [result.value],
   }
 }

frontend/internal-packages/agent/src/qa-agent/testcaseGeneration/index.integration.test.ts

@@ -42,7 +42,7 @@ describe('testcaseGeneration Integration', () => {
     })
 
     const state: TestcaseState = {
-      messages: [],
+      internalMessages: [],
       currentTestcase: {
         category: 'tasks',
         testcase: {

frontend/internal-packages/agent/src/qa-agent/testcaseGeneration/routeAfterGenerate.ts

@@ -10,9 +10,9 @@ import type { testcaseAnnotation } from './testcaseAnnotation'
 export const routeAfterGenerate = (
   state: typeof testcaseAnnotation.State,
 ): 'invokeSaveTool' | typeof END => {
-  const { messages } = state
+  const { internalMessages } = state
 
-  const lastMessage = messages[messages.length - 1]
+  const lastMessage = internalMessages[internalMessages.length - 1]
 
   // Check if the last message has tool calls
   if (lastMessage && hasToolCalls(lastMessage)) {

frontend/internal-packages/agent/src/qa-agent/testcaseGeneration/saveToolNode.ts

@@ -5,20 +5,28 @@ import type { testcaseAnnotation } from './testcaseAnnotation'
 
 /**
  * Save Tool Node for testcase generation
- * Executes the saveTestcaseTool within the isolated subgraph context with streaming support
+ * Executes the saveTestcaseTool within the isolated subgraph context
+ * Maps internalMessages to messages for ToolNode compatibility
  */
 export const saveToolNode = async (
   state: typeof testcaseAnnotation.State,
   config?: RunnableConfig,
 ) => {
   const toolNode = new ToolNode([saveTestcaseTool])
 
-  const stream = await toolNode.stream(state, config)
+  const toolNodeInput = {
+    ...state,
+    messages: state.internalMessages,
+  }
 
-  let result = {}
+  const result = await toolNode.invoke(toolNodeInput, config)
 
-  for await (const chunk of stream) {
-    result = chunk
+  // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access -- ToolNode result type is not well-typed
+  if ('messages' in result && Array.isArray(result.messages)) {
+    return {
+      // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access -- ToolNode result type is not well-typed
+      internalMessages: result.messages,
+    }
   }
 
   return result

frontend/internal-packages/agent/src/qa-agent/testcaseGeneration/testcaseAnnotation.ts

@@ -1,4 +1,5 @@
-import { Annotation, MessagesAnnotation } from '@langchain/langgraph'
+import type { BaseMessage } from '@langchain/core/messages'
+import { Annotation } from '@langchain/langgraph'
 import type { Schema } from '@liam-hq/schema'
 import type { SchemaIssue } from '../../workflowSchemaIssuesAnnotation'
 import type { TestCaseData } from '../distributeRequirements'
@@ -30,7 +31,10 @@ export const generatedSqlsAnnotation = Annotation<Array<GeneratedSql>>({
 })
 
 export const testcaseAnnotation = Annotation.Root({
-  ...MessagesAnnotation.spec,
+  internalMessages: Annotation<BaseMessage[]>({
+    reducer: (x, y) => x.concat(y),
+    default: () => [],
+  }),
   currentTestcase: Annotation<TestCaseData>,
   schemaData: Annotation<Schema>,
   goal: Annotation<string>,