The Legend of AWS Warrior (Part 3 - Gen Question): Autonomous AWS Question Bank Generation with Loop Engineering, Strands Multi-Agent Graphs, Harness Plugins and Live OpenTelemetry Tracing

// ANTI-PATTERN: Blind imperative retry loop (NOT Loop Engineering)
for (let attempt = 0; attempt < 3; attempt++) {
  const raw = await callModel(giantMonolithicPrompt);
  if (isValid(raw)) return raw; // Wipes memory on failure; re-runs Web Search & MCP from scratch!
}

// src/VoiceAgent/question-generator.mjs
import { BedrockRuntimeClient, ConverseCommand } from "@aws-sdk/client-bedrock-runtime";
import { McpClient, tool } from "@strands-agents/sdk";

const defaultClient = new BedrockRuntimeClient({ maxAttempts: 3, retryMode: "adaptive" });

export function createNovaWebSearchTool({
  send = command => defaultClient.send(command),
  modelId = process.env.VOICE_MODEL_ID ?? "us.amazon.nova-2-lite-v1:0",
} = {}) {
  return tool({
    name: "nova_web_search",
    description: "Search the live web using Amazon Nova Web Grounding (nova_grounding) for common AWS interview questions and recent service changes.",
    inputSchema: {
      type: "object",
      properties: { query: { type: "string" } },
      required: ["query"],
    },
    callback: async input => {
      const query = String(input?.query ?? "").trim();
      const response = await send(new ConverseCommand({
        modelId,
        messages: [{ role: "user", content: [{ text: query }] }],
        toolConfig: { tools: [{ systemTool: { name: "nova_grounding" } }] },
        inferenceConfig: { maxTokens: 1000, temperature: 0 },
      }));
      return extractNovaGroundingResult(response);
    },
  });
}

export async function loadAwsKnowledgeMcpTools({
  url = "https://knowledge-mcp.global.api.aws",
  allowed = ["aws___search_documentation", "aws___read_documentation"],
} = {}) {
  const [mcpClient] = await McpClient.loadServers({
    awsKnowledge: { url, toolFilters: { allowed } },
  });
  return { tools: await mcpClient.listTools(), disconnect: () => mcpClient.disconnect() };
}

// src/VoiceAgent/question-generator.mjs
import { createHarness } from "@strands-agents/harness";
import { BedrockModel, BeforeInvocationEvent, BeforeModelCallEvent } from "@strands-agents/sdk";
import { ContextOffloader, InMemoryStorage } from "@strands-agents/sdk/vended-plugins/context-offloader";
import { GoalLoop } from "@strands-agents/sdk/vended-plugins/goal";

async function createStrandsNodeAgent({
  id, description, systemPrompt, tools = [], send, modelId, maxTokens = 2200, maxTurns = 4, traceAttributes,
}) {
  const boundedMaxTurns = Math.min(10, Math.max(1, Math.trunc(Number(maxTurns) || 4)));
  const model = new BedrockModel({
    modelId, maxTokens, temperature: 0, stream: false,
    clientConfig: { maxAttempts: 3, retryMode: "adaptive" },
  });

  // 1. Intra-Node GoalLoop Plugin: Self-corrects malformed non-JSON output before graph handoff
  const jsonGoalLoop = new GoalLoop({
    name: `${id}:json-goal-loop`,
    maxAttempts: Math.min(2, boundedMaxTurns),
    preserveContext: true,
    goal: response => {
      const text = (response?.content ?? []).map(b => b?.text ?? "").join("\n").trim();
      try {
        parseJsonFromText(text);
        return { passed: true };
      } catch (error) {
        return {
          passed: false,
          feedback: `Output from ${id} was not valid JSON (${error.message}). Return ONLY valid JSON matching your node schema.`,
        };
      }
    },
  });

  // 2. ContextOffloader Plugin: Offloads large MCP documentation reads (>2500 tokens) to InMemoryStorage
  const plugins = [jsonGoalLoop];
  if (tools.length > 0) {
    plugins.unshift(new ContextOffloader({
      storage: new InMemoryStorage(), maxResultTokens: 2500, previewTokens: 1000, includeRetrievalTool: false,
    }));
  }

  // 3. Strands Harness with Large-Tool Offloading and Overflow Protection
  const agent = await createHarness({
    id, name: id, description, model, effort: "auto", caching: false,
    systemPrompt, tools, plugins, builtinTools: [], builtinPlugins: [],
    contextManager: { strategies: ["largeToolOffloading", "overflowProtection"], stash: false },
    session: false, memory: false, skills: false, backgroundTasks: false, printer: false, retryStrategy: null,
    ...(traceAttributes ? { traceAttributes } : {}),
  });

  // 4. Hard Inner Turn Cap (Default 4 turns, shared across tool calls and GoalLoop resumes)
  let turnsInCurrentInvocation = 0;
  agent.addHook(BeforeInvocationEvent, event => {
    if (!event?.resumed) turnsInCurrentInvocation = 0;
  });
  agent.addHook(BeforeModelCallEvent, () => {
    if (++turnsInCurrentInvocation > boundedMaxTurns) {
      throw new Error(`Agent "${id}" exceeded maximum inner loop turns (${boundedMaxTurns})`);
    }
  });
  return agent;
}

// src/VoiceAgent/question-generator.mjs
export function findDuplicateOrOverlappingQuestion(question, avoidQuestions = [], avoidIds = []) {
  const blockedIds = new Set(avoidIds.map(id => String(id).trim()).filter(Boolean));
  if (blockedIds.has(question.id)) return { reason: "duplicate_id", matchId: question.id };
  const promptTokens = tokenizeTopicText(`${question.title ?? ""} ${question.prompts?.["en-US"] ?? ""}`, question.serviceId);
  const conceptTokens = tokenizeTopicText((question.concepts ?? []).map(c => `${c.id} ${c.description}`).join(" "), question.serviceId);

  for (const rawExisting of avoidQuestions) {
    const existing = summarizeExistingQuestion(rawExisting);
    if (!existing) continue;
    if (existing.id === question.id) return { reason: "duplicate_id", matchId: existing.id, matchTitle: existing.title };
    const promptSim = jaccardSimilarity(promptTokens, tokenizeTopicText(`${existing.title} ${existing.enPrompt}`, question.serviceId));
    if (promptSim >= 0.55) return { reason: "overlapping_prompt", similarity: promptSim, matchId: existing.id };
    const conceptSim = jaccardSimilarity(conceptTokens, tokenizeTopicText(existing.concepts.join(" "), question.serviceId));
    if (conceptSim >= 0.6) return { reason: "overlapping_concepts", similarity: conceptSim, matchId: existing.id };
  }
  return null;
}

export function classifyFailureStage(errorOrMessage) {
  const msg = String(errorOrMessage?.message ?? errorOrMessage ?? "");
  if (/already used|overlaps too closely|duplicate_|overlapping_|Question id is required/i.test(msg)) return "topic";
  if (/Mandarin prompt \(zh-CN\) contains Cantonese-only characters|Synthetic goodZhHk failed/i.test(msg)) return "localization";
  return "rubric";
}

// Inside verifyAndCalibrate(candidate, targetMeta): 3x Parallel Live Grader Calibration
const [enRaw, hkRaw, wrongRaw] = await Promise.all([
  grader({ operation: "grade_answer", language: "en-US", feedbackLanguage: "English", question: rubricFor("en-US"), transcript: transcripts.goodEn }),
  grader({ operation: "grade_answer", language: "zh-HK", feedbackLanguage: "Traditional Chinese (Cantonese)", question: rubricFor("zh-HK"), transcript: transcripts.goodZhHk }),
  grader({ operation: "grade_answer", language: "en-US", feedbackLanguage: "English", question: rubricFor("en-US"), transcript: transcripts.wrongOutdated }),
]);
const enGrade = gradeAnswer(enRaw, question, transcripts.goodEn);
if (!enGrade.passed || enGrade.score < 80) throw new Error(`Synthetic goodEn failed calibration (score=${enGrade.score}, expected >=80)`);
const hkGrade = gradeAnswer(hkRaw, question, transcripts.goodZhHk);
if (!hkGrade.passed || hkGrade.score < 80) throw new Error(`Synthetic goodZhHk failed calibration (score=${hkGrade.score}, expected >=80)`);
const wrongGrade = gradeAnswer(wrongRaw, question, transcripts.wrongOutdated);
if (wrongGrade.passed || wrongGrade.score >= 60) throw new Error(`Synthetic wrongOutdated passed unexpectedly (score=${wrongGrade.score}, expected <60)`);

// Strands Tool bridging inner agent execution with outer Graph invocationState
const calibrationTool = tool({
  name: "verify_and_calibrate_question",
  description: "Run schema validation, semantic duplicate detection, and 3-way live student grading.",
  inputSchema: {
    type: "object",
    properties: {
      question: { type: "object" },
      syntheticTranscripts: {
        type: "object",
        properties: { goodEn: { type: "string" }, goodZhHk: { type: "string" }, wrongOutdated: { type: "string" } },
        required: ["goodEn", "goodZhHk", "wrongOutdated"],
      },
    },
    required: ["question", "syntheticTranscripts"],
  },
  callback: async (input, toolContext) => {
    const state = toolContext?.invocationState ?? {};
    try {
      const verified = await verifyAndCalibrate(input, state.targetMeta);
      state.calibratedResult = verified;
      state.lastCritique = null;
      state.lastFailureStage = null;
      return { passed: true, questionId: verified.question.id, calibration: verified.calibration };
    } catch (error) {
      state.calibratedResult = null;
      state.lastCritique = error.message;
      state.lastFailureStage = classifyFailureStage(error);
      return { passed: false, failureStage: state.lastFailureStage, critique: error.message };
    }
  },
});

// src/VoiceAgent/question-generator.mjs
const evaluateCriticOutcome = async state => {
  const criticResults = state.node("calibration_critic")?.results ?? [];
  if (invocationState.evaluatedCriticTurn === criticResults.length) return;
  invocationState.evaluatedCriticTurn = criticResults.length;

  if (!invocationState.calibratedResult) {
    const finalText = (state.node("calibration_critic")?.content ?? []).map(b => b?.text ?? "").join("\n").trim();
    try {
      invocationState.calibratedResult = await verifyAndCalibrate(parseJsonFromText(finalText), invocationState.targetMeta);
      invocationState.lastCritique = null;
      invocationState.lastFailureStage = null;
    } catch (error) {
      invocationState.lastCritique = error.message;
      invocationState.lastFailureStage = classifyFailureStage(error);
    }
  }
  if (invocationState.calibratedResult) {
    invocationState.shouldRetry = false;
    return;
  }
  invocationState.cycleCount = (invocationState.cycleCount ?? 0) + 1;
  invocationState.lastFailureStage = invocationState.lastFailureStage ?? classifyFailureStage(invocationState.lastCritique);
  invocationState.shouldRetry = invocationState.cycleCount < boundedMaxAttempts;

  const targetByStage = { localization: "trilingual_localizer", rubric: "aws_docs_fact_checker", topic: "interview_miner" };
  criticResults.at(-1)?.content?.push(new TextBlock(JSON.stringify({
    previousAttemptCritique: invocationState.lastCritique,
    failureStage: invocationState.lastFailureStage,
    targetNode: targetByStage[invocationState.lastFailureStage] ?? "interview_miner",
    cycle: invocationState.cycleCount,
  })));
};

const routeBackFor = (stage, forwardParent = null) => async state => {
  const lastNodeId = state.results.at(-1)?.nodeId;
  if (forwardParent && lastNodeId === forwardParent) return true;
  if (lastNodeId !== "calibration_critic") return false;
  await evaluateCriticOutcome(state);
  return Boolean(invocationState.shouldRetry && invocationState.lastFailureStage === stage);
};

const graph = new Graph({
  id: "aws-interview-question-pipeline",
  sources: ["interview_miner"],
  ...(traceAttributes ? { traceAttributes } : {}),
  nodes: [
    { agent: interviewMiner, preserveContext: true },
    { agent: awsDocsFactChecker, preserveContext: true },
    { agent: trilingualLocalizer, preserveContext: true },
    { agent: calibrationCritic, preserveContext: true },
  ],
  edges: [
    ["interview_miner", "aws_docs_fact_checker"],
    ["aws_docs_fact_checker", "trilingual_localizer"],
    ["trilingual_localizer", "calibration_critic"],
    { source: "calibration_critic", target: "trilingual_localizer", handler: routeBackFor("localization", "aws_docs_fact_checker") },
    { source: "calibration_critic", target: "aws_docs_fact_checker", handler: routeBackFor("rubric", "interview_miner") },
    { source: "calibration_critic", target: "interview_miner", handler: routeBackFor("topic") },
  ],
  maxConcurrency: 1,
  maxSteps: boundedMaxSteps,
});

graph.addHook(BeforeMultiAgentInvocationEvent, event => {
  const criticNodeState = event.state.node("calibration_critic");
  if (criticNodeState && criticNodeState.results.length === 0) criticNodeState.status = Status.COMPLETED;
});

graph.addHook(BeforeNodeCallEvent, event => {
  if (event.nodeId !== "calibration_critic") {
    const lastCriticResult = (event.state.node("calibration_critic")?.results ?? []).at(-1);
    if (lastCriticResult?.content?.length > 0) lastCriticResult.content = [];
  }
});

// 1. Streaming live Strands Graph events (src/VoiceAgent/question-generator.mjs)
const streamGen = graph.stream(taskPrompt, { invocationState });
for (let next = await streamGen.next(); !next.done; next = await streamGen.next()) {
  const event = next.value;
  if (event.type === "multiAgentHandoffEvent") {
    const isLoopBack = event.source === "calibration_critic" && (event.targets ?? []).length > 0;
    handoffs.push({ source: event.source, targets: event.targets, isLoopBack, failureStage: isLoopBack ? invocationState.lastFailureStage : null });
  }
}

// 2. Accumulating avoidQuestions across multi-slot generation (src/VoicePractice/question-generator.mjs)
for (let i = 0; i < countPerService; i++) {
  const { question, trace } = await generator.generateOrRefreshQuestion({
    serviceId, label: meta.label, passage: meta.passage,
    avoidIds: [...existingIds],
    avoidQuestions: [...existingQuestionsById.values()],
    slotIndex: i + 1, status,
  });
  existingIds.add(question.id);
  existingQuestionsById.set(question.id, question);
  generatedForService.push(question);
}

let tracerProvider = null;

export async function enableStrandsTelemetry({
  otlp = Boolean(process.env.OTEL_EXPORTER_OTLP_ENDPOINT || process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT || process.env.LANGFUSE_BASE_URL),
  console: consoleExporter = process.env.STRANDS_TRACE_CONSOLE === "1",
} = {}) {
  if (tracerProvider) return true;
  if (!otlp && !consoleExporter) return false;
  const telemetry = await import("@strands-agents/sdk/telemetry");
  tracerProvider = telemetry.setupTracer({ exporters: { otlp, console: consoleExporter } });
  return true;
}

# 1. Start the Official Jaeger All-in-One OTLP Collector and Web UI
docker run -d --rm --name strands-jaeger \
  -e COLLECTOR_OTLP_ENABLED=true \
  -p 16686:16686 -p 4317:4317 -p 4318:4318 \
  jaegertracing/all-in-one:latest

# 2. Run Single-Service s3 Generation with Strands OTLP Export
OTEL_TRACES_EXPORTER=otlp \
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 \
OTEL_SERVICE_NAME=aws-warrior-question-generator \
node src/VoicePractice/question-bank.mjs generate \
  /tmp/s3-single-question.json s3 \
  --count=1 --no-publish --force --otel --trace-console \
  --trace-json=/tmp/s3-single-trace.json

invoke_graph aws-interview-question-pipeline (34,286.7ms | 35,711 tokens)
├── node interview_miner (7,934.9ms)
│   └── invoke_agent interview_miner -> cycle-1: nova_web_search (4,833.2ms) -> cycle-2
├── node aws_docs_fact_checker (12,997.6ms)
│   └── invoke_agent aws_docs_fact_checker -> search_documentation + 2x read_documentation
├── node trilingual_localizer (3,414.3ms)
│   └── invoke_agent trilingual_localizer (3,412.8ms)
└── node calibration_critic (9,937.3ms)
    └── invoke_agent calibration_critic -> verify_and_calibrate_question (2,007.5ms)
        ├── invoke_agent Strands Agent [goodEn -> 100/100] (2,001.8ms)
        ├── invoke_agent Strands Agent [goodZhHk -> 100/100] (1,706.8ms)
        └── invoke_agent Strands Agent [wrongOutdated -> 50/100] (1,742.7ms)

{
  "id": "s3-lifecycle-optimization",
  "serviceId": "s3",
  "version": 1,
  "status": "approved",
  "title": "S3 lifecycle cost optimization",
  "prompts": {
    "en-US": "Your application stores user-generated content in Amazon S3 and needs to reduce storage costs while maintaining quick access to recently used files. How would you configure S3 lifecycle policies to achieve this?",
    "zh-HK": "你嘅應用程序將用戶產生嘅內容儲存喺 Amazon S3，而需要減少儲存成本，同時保持近期使用嘅文件快速存取。你點樣設定 S3 生命週期策略嚟達到呢個目標？",
    "zh-CN": "您的应用程序将用户生成的内容存储在 Amazon S3 中，需要降低存储成本，同时保持最近使用文件的快速访问。您将如何配置 S3 生命周期策略来实现这一目标？"
  },
  "concepts": [
    {
      "id": "lifecycle-transition-rules",
      "description": "Create lifecycle rules that transition objects to lower-cost storage classes like S3 Standard-IA, S3 One Zone-IA, or S3 Glacier after a defined period of inactivity, balancing cost savings with access performance requirements.",
      "weight": 50
    },
    {
      "id": "storage-class-selection",
      "description": "Select appropriate storage classes based on access patterns: S3 Standard for frequently accessed data, S3 Standard-IA for infrequent access with lower costs, and S3 Glacier for long-term archival with retrieval fees and time constraints.",
      "weight": 30
    },
    {
      "id": "rule-scope-and-prefixes",
      "description": "Define precise prefixes or tags in lifecycle rules to target specific object sets, avoiding unintended transitions and ensuring only appropriate data moves to lower-cost tiers.",
      "weight": 20
    }
  ]
}