Architecting an Autonomous QA Testing SaaS in 2026: Multi-Agent Test Generation with LangGraph, Node.js, and Stripe

Introduction: Why QA Is the Perfect Micro-Niche for Agentic Engineering

Every engineering team I've worked with has the same quiet problem. The test suite is either too thin to trust or too brittle to maintain. Someone writes 400 Playwright tests during a "quality sprint", half of them break after the next redesign, and six months later the CI pipeline has a habit of skipping the flaky ones.

By 2026 the industry has mostly accepted that writing and maintaining end-to-end tests is work an agent should do. "Agentic Engineering" is the umbrella term: software teams supervise autonomous agents instead of hand-writing every artifact. QA is one of the best places to apply it because the feedback loop is objective. A test passes or it fails. There's a stack trace. The DOM either has the button or it doesn't.

That objectivity is what makes an autonomous QA testing SaaS commercially viable. You aren't selling vague "AI magic". You're selling a measurable result: more coverage, fewer escaped bugs, and less time spent fixing flaky selectors.

This guide walks through the full architecture of that product:

• Self-reflecting QA agents built with LangGraph and Python, running on local LLMs through Ollama
• A Node.js + PostgreSQL REST API that manages test jobs and queues
• Real-time execution logs streamed to a Next.js + Shadcn UI dashboard over Socket.io
• Usage-based billing with Stripe Billing Meters, plus Zapier for operations automation
• Docker containers deployed to AWS through a secure CI/CD pipeline

I'll stick to decisions that matter in production and leave out toy demos. Let's build it.

The Product: What Exactly Are We Selling?

Before any code, be clear about the value proposition. A customer connects a staging URL (and optionally a repository) and gets:

1. Automatic discovery: an agent crawls the app and maps the key user flows (signup, checkout, settings, and so on).
2. Test generation: agents write Playwright end-to-end tests for each flow, including the edge cases humans tend to skip.
3. Mutation and hardening: agents deliberately vary inputs (unicode names, expired cards, double-clicks, slow networks) to find weak spots.
4. Execution and triage: tests run in isolated browsers. Failures are classified as test bug, flaky, or real product bug.
5. Live visibility: a dashboard streams every step, screenshot, and agent decision as it happens.

The billing unit is executed test minutes. Customers understand it, it tracks your real infrastructure cost, and it grows naturally as they rely on the product more.

High-Level Architecture

Here's the whole system on one page. Every arrow represents a deliberate boundary between services.

flowchart LR
    subgraph Client["Customer"]
        UI["Next.js + Shadcn Dashboard"]
        CI["Customer CI (GitHub Actions)"]
    end

    subgraph API["Node.js API Layer"]
        REST["REST API (Fastify)"]
        WS["Socket.io Gateway"]
        WH["Stripe Webhooks"]
    end

    subgraph Data["State"]
        PG[("PostgreSQL\njobs, tests, runs")]
        RD[("Redis\npub/sub + socket adapter")]
        S3[("S3\nscreenshots, traces")]
    end

    subgraph Agents["Python Agent Workers"]
        LG["LangGraph QA Graph"]
        OL["Ollama (Local LLMs)"]
        PW["Playwright Executors"]
    end

    subgraph Billing["Billing & Ops"]
        ST["Stripe Billing Meters"]
        ZP["Zapier Automations"]
    end

    UI -->|HTTPS| REST
    CI -->|API key| REST
    UI <-->|WebSocket| WS
    REST --> PG
    LG -->|poll SKIP LOCKED| PG
    LG --> OL
    LG --> PW
    PW --> S3
    LG -->|publish logs| RD
    RD --> WS
    REST -->|meter events| ST
    ST --> WH
    ST --> ZP

Why split it this way?

• Python owns intelligence. LangGraph, LangChain integrations, and most evaluation tooling live in the Python ecosystem.
• Node.js owns I/O. REST, WebSockets, and webhooks are high-concurrency, low-CPU workloads, which is what Node's event loop handles best.
• PostgreSQL is the contract. The API writes job rows and the workers claim them. No custom RPC layer is needed, and you get durability for free.
• Redis is the nervous system. Workers publish log events and the Socket.io gateway fans them out to browsers.

Core Concept: The Self-Reflecting QA Agent Loop

The most common mistake in AI testing tools is treating test generation as a single LLM call: "Here's a page, write a test." The model returns something plausible, it fails on the first run, and the product feels broken.

A useful QA agent is a loop with judgment:

stateDiagram-v2
    [*] --> Plan
    Plan --> Generate
    Generate --> Execute
    Execute --> Reflect
    Reflect --> Generate: test bug (fix & retry)
    Reflect --> Mutate: test passed
    Reflect --> Report: real product bug
    Reflect --> Report: budget exhausted
    Mutate --> Execute: new edge-case variant
    Mutate --> Report: mutation budget exhausted
    Report --> [*]

Each node has one job:

• Plan: turn a discovered user flow into a structured test plan (steps, assertions, test data).
• Generate: write Playwright TypeScript code from the plan and a compact DOM snapshot.
• Execute: run the test in a sandboxed browser and capture logs, screenshots, and the trace.
• Reflect: read the failure and decide whose fault it is. This node decides whether the product is useful at all.
• Mutate: take a passing test and create adversarial variants (boundary values, unusual input, timing changes).
• Report: save the results and the agent's reasoning for the dashboard.

The Reflect node is your moat. Distinguishing "my selector was wrong" from "the checkout button really does nothing" is what separates a toy from a product people will pay for.

Building the Agent Graph with LangGraph and Ollama

Defining the Graph State

LangGraph graphs pass a typed state object between nodes. Put your budgets in the state so every node can check them.

# agents/qa_graph/state.py
from typing import TypedDict, Literal, Annotated
import operator
 
Verdict = Literal["pass", "test_bug", "product_bug", "flaky", "unknown"]
 
class QAState(TypedDict):
    job_id: str
    flow_name: str
    target_url: str
    dom_snapshot: str             # compact, pruned accessibility tree
    plan: dict | None
    test_code: str | None
    last_run: dict | None         # {status, error, duration_ms, trace_url}
    verdict: Verdict | None
    variants: Annotated[list[str], operator.add]  # accumulated mutated tests
    findings: Annotated[list[dict], operator.add] # bugs and insights
    # Hard budgets: never let an agent loop without them
    reflect_iterations: int
    mutation_count: int
    max_reflect_iterations: int
    max_mutations: int

The Annotated[list, operator.add] reducers tell LangGraph to append to the lists rather than overwrite them. That matters when several nodes add findings.

Connecting to Local LLMs via Ollama

Running models locally through Ollama gives you three business advantages: predictable cost (GPU hours rather than per-token billing), data privacy (customer DOMs never leave your VPC), and no rate limits from third-party providers.

# agents/qa_graph/llm.py
from langchain_ollama import ChatOllama
import os
 
OLLAMA_URL = os.getenv("OLLAMA_BASE_URL", "http://ollama:11434")
 
# A capable coding model for generation, and a smaller, faster one for triage
codegen_llm = ChatOllama(
    model=os.getenv("CODEGEN_MODEL", "qwen2.5-coder:32b"),
    base_url=OLLAMA_URL,
    temperature=0.1,
    num_ctx=16384,
)
 
triage_llm = ChatOllama(
    model=os.getenv("TRIAGE_MODEL", "llama3.1:8b"),
    base_url=OLLAMA_URL,
    temperature=0,
    format="json",  # force structured output for routing decisions
)

Tip: Keep model names in environment variables. Open-weight models improve every few months, and you want to upgrade them without redeploying code.

The Generate Node

Generation only works when the model gets structured, pruned context. Never pass raw HTML. Pass an accessibility-tree snapshot with roles, names, and test IDs.

# agents/qa_graph/nodes/generate.py
from langchain_core.messages import SystemMessage, HumanMessage
from ..llm import codegen_llm
from ..logging import emit
 
SYSTEM = """You are a senior QA engineer writing Playwright tests in TypeScript.
Rules:
- Prefer getByRole, getByLabel, getByTestId. Never use nth-child or XPath.
- Every test must contain at least one meaningful expect() assertion.
- Use web-first assertions (toBeVisible, toHaveURL), never fixed timeouts.
- Output ONLY the test file content, no markdown fences."""
 
def generate_node(state: dict) -> dict:
    emit(state["job_id"], "generate", f"Writing test for '{state['flow_name']}'")
 
    feedback = ""
    if state.get("last_run") and state.get("verdict") == "test_bug":
        feedback = (
            f"\nYour previous attempt failed with:\n{state['last_run']['error']}\n"
            f"Previous code:\n{state['test_code']}\nFix the root cause."
        )
 
    response = codegen_llm.invoke([
        SystemMessage(content=SYSTEM),
        HumanMessage(content=(
            f"Target URL: {state['target_url']}\n"
            f"Test plan: {state['plan']}\n"
            f"Accessibility snapshot:\n{state['dom_snapshot']}{feedback}"
        )),
    ])
    return {"test_code": response.content.strip()}

Notice the feedback injection. On a retry, the agent sees its own previous code and the actual error. That's the "self-reflecting" part, and it routinely turns a 40% first-pass success rate into 85% or better after two iterations.

The Reflect Node: Classifying Failures

# agents/qa_graph/nodes/reflect.py
import json
from ..llm import triage_llm
from ..logging import emit
 
TRIAGE_PROMPT = """Classify this Playwright test result. Respond with JSON:
{{"verdict": "pass|test_bug|product_bug|flaky", "reason": "...", "confidence": 0-1}}
 
- test_bug: selector wrong, bad assumption, timing issue in the TEST itself
- product_bug: the app misbehaves (500 error, element missing that the flow requires,
  wrong data shown, JS exception in console)
- flaky: passes/fails inconsistently, network or animation related
 
Run status: {status}
Error: {error}
Console errors: {console}
"""
 
def reflect_node(state: dict) -> dict:
    run = state["last_run"]
    if run["status"] == "passed":
        return {"verdict": "pass"}
 
    raw = triage_llm.invoke(TRIAGE_PROMPT.format(
        status=run["status"],
        error=run.get("error", "")[:4000],
        console=run.get("console_errors", [])[:20],
    )).content
 
    result = json.loads(raw)
    emit(state["job_id"], "reflect", f"Verdict: {result['verdict']} ({result['reason']})")
 
    update = {
        "verdict": result["verdict"],
        "reflect_iterations": state["reflect_iterations"] + 1,
    }
    if result["verdict"] == "product_bug" and result["confidence"] >= 0.7:
        update["findings"] = [{
            "type": "product_bug",
            "flow": state["flow_name"],
            "reason": result["reason"],
            "trace_url": run.get("trace_url"),
        }]
    return update

Routing with Conditional Edges and Budgets

The router is ordinary Python, which keeps it easy to test and reason about.

# agents/qa_graph/graph.py
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from .state import QAState
from .nodes import plan_node, generate_node, execute_node, reflect_node, mutate_node, report_node
 
def route_after_reflect(state: QAState) -> str:
    if state["reflect_iterations"] >= state["max_reflect_iterations"]:
        return "report"                       # budget exhausted
    match state["verdict"]:
        case "pass":
            return "mutate" if state["mutation_count"] < state["max_mutations"] else "report"
        case "test_bug" | "flaky":
            return "generate"
        case _:
            return "report"                   # product bug or unknown
 
def build_graph(checkpointer: PostgresSaver):
    g = StateGraph(QAState)
    g.add_node("plan", plan_node)
    g.add_node("generate", generate_node)
    g.add_node("execute", execute_node)
    g.add_node("reflect", reflect_node)
    g.add_node("mutate", mutate_node)
    g.add_node("report", report_node)
 
    g.set_entry_point("plan")
    g.add_edge("plan", "generate")
    g.add_edge("generate", "execute")
    g.add_edge("execute", "reflect")
    g.add_conditional_edges("reflect", route_after_reflect,
                            {"generate": "generate", "mutate": "mutate", "report": "report"})
    g.add_edge("mutate", "execute")
    g.add_edge("report", END)
 
    return g.compile(checkpointer=checkpointer)

Using the Postgres checkpointer means that if a worker container crashes mid-run (and in production it will), the job resumes from the last completed node instead of starting over. Customers never pay twice for the same minutes.

Mutation: Where the Real Edge Cases Come From

Mutation testing is the feature that makes demos impressive. Once a test passes, the mutate node asks the model to create one adversarial variant at a time:

• Input mutations: 255-character names, emoji, RTL text, SQL-looking strings, leading or trailing whitespace
• State mutations: expired sessions, empty carts, already-used coupon codes
• Timing mutations: double-submit, back button mid-flow, throttled 3G network
• Viewport mutations: 320px mobile width, 200% zoom

Real-world example: on an early pilot, a mutated checkout test that clicked "Pay" twice within 150ms produced two Stripe charges on the customer's staging site. No human-written test had ever covered it. Finding a bug like that is how the tool sells itself to a customer.

The Node.js + PostgreSQL Test Queue API

Why PostgreSQL Is Enough for the Queue

You don't need Kafka or SQS on day one. PostgreSQL's FOR UPDATE SKIP LOCKED gives you a concurrent, durable job queue where multiple workers claim jobs without stepping on each other.

-- db/migrations/001_jobs.sql
CREATE TYPE job_status AS ENUM ('queued', 'running', 'completed', 'failed', 'cancelled');
 
CREATE TABLE test_jobs (
  id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  org_id        UUID NOT NULL REFERENCES organizations(id),
  project_id    UUID NOT NULL REFERENCES projects(id),
  target_url    TEXT NOT NULL,
  flows         JSONB NOT NULL DEFAULT '[]',
  status        job_status NOT NULL DEFAULT 'queued',
  priority      SMALLINT NOT NULL DEFAULT 5,
  attempts      SMALLINT NOT NULL DEFAULT 0,
  locked_by     TEXT,
  locked_at     TIMESTAMPTZ,
  billed_seconds INTEGER NOT NULL DEFAULT 0,
  created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
  finished_at   TIMESTAMPTZ
);
 
CREATE INDEX idx_jobs_claimable
  ON test_jobs (priority DESC, created_at)
  WHERE status = 'queued';

The partial index keeps the claim query fast even after millions of completed jobs pile up in the table.

The Python worker claims a job atomically:

UPDATE test_jobs
SET status = 'running', locked_by = $1, locked_at = now(), attempts = attempts + 1
WHERE id = (
  SELECT id FROM test_jobs
  WHERE status = 'queued'
  ORDER BY priority DESC, created_at
  FOR UPDATE SKIP LOCKED
  LIMIT 1
)
RETURNING *;

The REST Endpoint

I use Fastify for the API because of its schema validation and performance, but Express works fine too.

// api/src/routes/jobs.ts
import { FastifyInstance } from "fastify";
import { z } from "zod";
import { db } from "../db";
import { assertWithinPlanLimits } from "../billing/limits";
 
const CreateJobSchema = z.object({
  projectId: z.string().uuid(),
  targetUrl: z.string().url(),
  flows: z.array(z.string()).max(50).default([]),
  priority: z.number().int().min(1).max(10).default(5),
});
 
export async function jobRoutes(app: FastifyInstance) {
  app.post("/v1/jobs", { preHandler: app.authenticate }, async (req, reply) => {
    const body = CreateJobSchema.parse(req.body);
    const orgId = req.user.orgId;
 
    // Block before queueing, not after burning GPU time
    await assertWithinPlanLimits(orgId);
 
    const { rows } = await db.query(
      `INSERT INTO test_jobs (org_id, project_id, target_url, flows, priority)
       VALUES ($1, $2, $3, $4, $5) RETURNING id, status, created_at`,
      [orgId, body.projectId, body.targetUrl, JSON.stringify(body.flows), body.priority]
    );
 
    // Wake idle workers immediately instead of waiting for the next poll
    await db.query(`NOTIFY new_job, '${rows[0].id}'`);
 
    return reply.code(202).send(rows[0]);
  });
}

Two small details matter here:

• Return 202 Accepted, not 200. The work is asynchronous, and your API should say so honestly.
• NOTIFY lets workers that LISTEN pick up jobs instantly, while polling remains as a fallback.

Recovering Stuck Jobs

Workers die. Add a reaper that requeues jobs whose lock is too old:

// api/src/cron/reaper.ts
export async function reapStaleJobs() {
  await db.query(`
    UPDATE test_jobs
    SET status = CASE WHEN attempts >= 3 THEN 'failed'::job_status ELSE 'queued'::job_status END,
        locked_by = NULL, locked_at = NULL
    WHERE status = 'running' AND locked_at < now() - interval '15 minutes'
  `);
}

Streaming Real-Time Logs with Socket.io

Watching an AI agent think and click through your app live is the most persuasive part of the product. Customers keep the dashboard open on a second monitor just to watch it.

The Data Flow

sequenceDiagram
    participant W as Python Worker
    participant R as Redis
    participant G as Socket.io Gateway (Node)
    participant B as Browser (Next.js)

    B->>G: connect (JWT) + join room job:123
    G->>G: verify org owns job 123
    W->>R: PUBLISH job:123 {node: "reflect", msg: "..."}
    R->>G: message on job:123
    G->>B: emit "log" to room job:123
    W->>R: PUBLISH job:123 {type: "screenshot", url: "s3://..."}
    G->>B: emit "artifact"

The Python Emitter

# agents/qa_graph/logging.py
import json, time, os
import redis
 
r = redis.Redis.from_url(os.environ["REDIS_URL"])
 
def emit(job_id: str, node: str, message: str, **extra) -> None:
    r.publish(f"job:{job_id}", json.dumps({
        "ts": time.time(), "node": node, "message": message, **extra
    }))

The Socket.io Gateway with Room Authorization

// api/src/realtime/gateway.ts
import { Server } from "socket.io";
import { createAdapter } from "@socket.io/redis-adapter";
import { createClient } from "redis";
import { verifyJwt } from "../auth";
import { db } from "../db";
 
export async function createGateway(httpServer: any) {
  const pub = createClient({ url: process.env.REDIS_URL });
  const sub = pub.duplicate();
  const logSub = pub.duplicate();
  await Promise.all([pub.connect(), sub.connect(), logSub.connect()]);
 
  const io = new Server(httpServer, { cors: { origin: process.env.APP_ORIGIN } });
  io.adapter(createAdapter(pub, sub)); // horizontal scaling across API pods
 
  io.use(async (socket, next) => {
    try {
      socket.data.user = await verifyJwt(socket.handshake.auth.token);
      next();
    } catch {
      next(new Error("unauthorized"));
    }
  });
 
  io.on("connection", (socket) => {
    socket.on("job:subscribe", async (jobId: string) => {
      const { rowCount } = await db.query(
        "SELECT 1 FROM test_jobs WHERE id = $1 AND org_id = $2",
        [jobId, socket.data.user.orgId]
      );
      if (rowCount) socket.join(`job:${jobId}`);
    });
  });
 
  // Relay worker events into Socket.io rooms
  await logSub.pSubscribe("job:*", (message, channel) => {
    io.to(channel).emit("log", JSON.parse(message));
  });
 
  return io;
}

Never skip the ownership check in job:subscribe. Without it, anyone who guesses a job UUID can watch another company's staging environment. That's a security incident and a lost customer at the same time.

The Next.js + Shadcn Live Log Panel

// web/components/live-log-panel.tsx
"use client";
import { useEffect, useRef, useState } from "react";
import { io, Socket } from "socket.io-client";
import { ScrollArea } from "@/components/ui/scroll-area";
import { Badge } from "@/components/ui/badge";
import { Card, CardHeader, CardTitle, CardContent } from "@/components/ui/card";
 
type LogEvent = { ts: number; node: string; message: string };
 
const nodeColor: Record<string, "default" | "secondary" | "destructive" | "outline"> = {
  generate: "secondary",
  execute: "default",
  reflect: "outline",
  report: "destructive",
};
 
export function LiveLogPanel({ jobId, token }: { jobId: string; token: string }) {
  const [logs, setLogs] = useState<LogEvent[]>([]);
  const bottomRef = useRef<HTMLDivElement>(null);
 
  useEffect(() => {
    const socket: Socket = io(process.env.NEXT_PUBLIC_WS_URL!, { auth: { token } });
    socket.on("connect", () => socket.emit("job:subscribe", jobId));
    socket.on("log", (e: LogEvent) => setLogs((prev) => [...prev.slice(-499), e]));
    return () => { socket.disconnect(); };
  }, [jobId, token]);
 
  useEffect(() => bottomRef.current?.scrollIntoView({ behavior: "smooth" }), [logs]);
 
  return (
    <Card className="h-[520px]">
      <CardHeader><CardTitle>Agent Activity</CardTitle></CardHeader>
      <CardContent>
        <ScrollArea className="h-[420px] font-mono text-sm">
          {logs.map((l, i) => (
            <div key={i} className="flex gap-3 py-1">
              <span className="text-muted-foreground">
                {new Date(l.ts * 1000).toLocaleTimeString()}
              </span>
              <Badge variant={nodeColor[l.node] ?? "outline"}>{l.node}</Badge>
              <span>{l.message}</span>
            </div>
          ))}
          <div ref={bottomRef} />
        </ScrollArea>
      </CardContent>
    </Card>
  );
}

The prev.slice(-499) cap keeps the browser from freezing during long runs. A mutation-heavy job can easily emit thousands of events.

Usage-Based Billing with Stripe and Zapier

Choosing the Right Billing Metric

Pick a metric that is understandable to customers, correlated with your costs, and hard to game. Here's how the common options compare:

The pricing model I recommend is a platform fee (for example, a monthly base covering 3 projects and a minute allowance) plus metered overage per test minute.

Reporting Usage with Stripe Billing Meters

Stripe's Billing Meters accept usage events in near real time and aggregate them onto the customer's invoice. Report usage when a job finishes, from the Node API, which is the single source of truth for billing.

// api/src/billing/report-usage.ts
import Stripe from "stripe";
import { db } from "../db";
 
const stripe = new Stripe(process.env.STRIPE_SECRET_KEY!);
 
export async function reportJobUsage(jobId: string) {
  const { rows } = await db.query(
    `SELECT j.id, j.billed_seconds, o.stripe_customer_id
     FROM test_jobs j JOIN organizations o ON o.id = j.org_id
     WHERE j.id = $1 AND j.status = 'completed' AND j.usage_reported_at IS NULL`,
    [jobId]
  );
  if (!rows.length) return; // already reported or not billable
 
  const job = rows[0];
  const minutes = Math.ceil(job.billed_seconds / 60);
 
  await stripe.billing.meterEvents.create({
    event_name: "test_minutes",
    payload: {
      stripe_customer_id: job.stripe_customer_id,
      value: String(minutes),
    },
    identifier: `job_${job.id}`, // idempotency: Stripe dedupes on this
  });
 
  await db.query("UPDATE test_jobs SET usage_reported_at = now() WHERE id = $1", [jobId]);
}

The identifier field is critical. If your process retries after a network blip, Stripe deduplicates the event and you won't double-bill the customer.

Handling Webhooks Safely

// api/src/routes/stripe-webhook.ts
app.post("/webhooks/stripe", { config: { rawBody: true } }, async (req, reply) => {
  const sig = req.headers["stripe-signature"] as string;
  let event: Stripe.Event;
  try {
    event = stripe.webhooks.constructEvent(req.rawBody!, sig, process.env.STRIPE_WEBHOOK_SECRET!);
  } catch {
    return reply.code(400).send("invalid signature");
  }
 
  switch (event.type) {
    case "customer.subscription.updated":
    case "customer.subscription.deleted":
      await syncSubscription(event.data.object as Stripe.Subscription);
      break;
    case "invoice.payment_failed":
      await flagOrgPastDue((event.data.object as Stripe.Invoice).customer as string);
      break;
  }
  return reply.code(200).send({ received: true });
});

Where Zapier Fits In

Zapier is not in your critical billing path. That belongs to your code and Stripe. Use it for operations glue that a small team shouldn't spend engineering time on:

• New paid subscription in Stripe → post to the #wins Slack channel and create a deal in your CRM
• invoice.payment_failed → send a friendly dunning email and open a task for the founder
• Usage spike (via a webhook from your API) → notify customer success that an account is ready for an upsell conversation
• Critical product bug found by an agent → create a ticket in the customer's Jira or Linear (a feature customers love)

A good rule: if it breaking would cost you money or trust, write it in code. If it breaking would only cost you convenience, put it in Zapier.

Containerization and Deployment on AWS

Docker Compose for Local Development

# docker-compose.yml
services:
  postgres:
    image: postgres:17
    environment:
      POSTGRES_PASSWORD: dev
    volumes: [pgdata:/var/lib/postgresql/data]
  redis:
    image: redis:7-alpine
  ollama:
    image: ollama/ollama:latest
    volumes: [ollama:/root/.ollama]
    deploy:
      resources:
        reservations:
          devices: [{ driver: nvidia, count: 1, capabilities: [gpu] }]
  api:
    build: ./api
    env_file: .env
    depends_on: [postgres, redis]
    ports: ["4000:4000"]
  agent-worker:
    build: ./agents
    env_file: .env
    depends_on: [postgres, redis, ollama]
    deploy:
      replicas: 2
  web:
    build: ./web
    ports: ["3000:3000"]
volumes:
  pgdata:
  ollama:

Production Topology

flowchart TB
    U["Users"] --> CF["CloudFront + WAF"]
    CF --> ALB["Application Load Balancer"]
    ALB --> WEB["ECS Fargate: Next.js"]
    ALB --> API["ECS Fargate: Node API + Socket.io"]
    API --> RDS[("RDS PostgreSQL Multi-AZ")]
    API --> EC[("ElastiCache Redis")]
    subgraph Private["Private Subnets"]
        WK["ECS on EC2: Agent Workers"]
        GPU["EC2 GPU (g6/g5): Ollama"]
        SBX["Isolated Playwright Sandboxes"]
    end
    WK --> RDS
    WK --> EC
    WK --> GPU
    WK --> SBX
    SBX --> S3[("S3: traces & screenshots")]

Key decisions:

• Fargate for stateless services (web, API). There are no servers to patch.
• GPU EC2 instances for Ollama, inside private subnets with no public IP. Scale them with an Auto Scaling group driven by queue depth.
• Playwright sandboxes in a separate security group with egress allowed only to the customer's target domains. An agent should never be able to reach your internal metadata endpoint.

CI/CD with GitHub Actions and OIDC

Don't store long-lived AWS keys in GitHub. Use OIDC role assumption:

# .github/workflows/deploy.yml
name: deploy
on:
  push:
    branches: [main]
permissions:
  id-token: write
  contents: read
jobs:
  build-and-deploy:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        service: [api, agents, web]
    steps:
      - uses: actions/checkout@v4
      - uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ secrets.AWS_DEPLOY_ROLE_ARN }}
          aws-region: us-east-1
      - uses: aws-actions/amazon-ecr-login@v2
        id: ecr
      - name: Build, scan, and push
        run: |
          IMAGE=${{ steps.ecr.outputs.registry }}/qa-${{ matrix.service }}:${{ github.sha }}
          docker build -t $IMAGE ./${{ matrix.service }}
          docker run --rm aquasec/trivy image --exit-code 1 --severity CRITICAL $IMAGE
          docker push $IMAGE
      - name: Deploy to ECS
        run: |
          aws ecs update-service --cluster qa-prod \
            --service qa-${{ matrix.service }} --force-new-deployment

The Trivy scan step fails the build on critical vulnerabilities. Enterprise buyers will ask about this in their security questionnaires, so it pays to have it in place early.

Best Practices for Production-Grade QA Agents

• Constrain output with schemas. Use JSON mode or structured output for every routing decision. Free-form text should only appear in generated test code and human-readable reasons.
• Prune the DOM aggressively. Accessibility trees with roles, names, and data-testid attributes are 10 to 50 times smaller than raw HTML and give much better results.
• Make every budget explicit. Set limits on reflection iterations, mutations, tokens, and wall-clock time, and record which one stopped the run.
• Store agent reasoning, not just results. Customers trust a verdict far more when they can read why the agent called something a product bug.
• Version your prompts like code. Keep prompts in the repo, tag them, and record the prompt version on every run so you can debug regressions.
• Build a golden evaluation set. Maintain 50 to 100 known flows with known bugs. Run the agent against them on every prompt or model change before you ship.
• Isolate execution completely. Generated code is untrusted code. Run it in ephemeral containers with network egress allowlists.

Common Mistakes to Avoid

1. Unbounded agent loops. A reflect-generate cycle with no cap will happily burn 40 minutes of GPU time on a test that can never pass. Always route to report when a budget runs out.
2. Billing from the worker. If Python workers report usage directly to Stripe, retries and crashes cause double billing. Centralize billing in the API with idempotency keys.
3. Trusting the first verdict. Before reporting a "product bug", re-run the failing test once. Flakiness masquerading as a real bug destroys customer trust faster than anything else.
4. Sending raw HTML to the model. It wastes context, slows inference, and confuses the model with irrelevant markup.
5. Skipping WebSocket authorization. Joining a room must always be checked against the user's organization.
6. Over-engineering the queue on day one. PostgreSQL with SKIP LOCKED comfortably handles thousands of jobs per minute. Move to SQS or a dedicated broker when metrics tell you to, not before.
7. Generating tests nobody maintains. Re-run the healing loop automatically when the UI changes, so the suite stays green without anyone filing tickets.

🚀 Pro Tips

• Offer a GitHub Action from day one. uses: your-org/autonomous-qa@v1 on every pull request is the fastest path into a team's workflow, and it drives daily metered usage.
• Let customers export the generated tests. It feels counterintuitive, but plain Playwright files that teams own outright remove the lock-in objection and speed up enterprise deals.
• Route models by difficulty. Use a small, fast model for triage and a larger coding model only for generation and hard fixes. This alone can cut GPU costs by about half.
• Cache DOM snapshots by content hash. When the page hasn't changed between runs, skip discovery and planning entirely.
• Show the "bugs found" counter prominently. One number on the dashboard ("17 real bugs found this month") does more for renewals than any feature page.
• Sell to the pain, not the tech. Engineering managers buy "fewer escaped bugs and no flaky suite". They don't buy "LangGraph multi-agent orchestration".

📌 Key Takeaways

• Autonomous QA works because testing produces objective feedback, which lets agents correct themselves reliably.
• The plan → generate → execute → reflect → mutate loop in LangGraph, with explicit budgets, is the core of the product.
• Local LLMs on Ollama give you predictable costs and a strong privacy story that enterprise buyers value.
• A Node.js + PostgreSQL API with SKIP LOCKED queues is simple, durable, and scales further than most people expect.
• Socket.io with a Redis adapter makes the agent's work visible in real time, and that visibility is one of your best sales tools.
• Stripe Billing Meters with idempotent usage events handle metered billing cleanly. Zapier handles the non-critical operations work.
• Docker, ECS, isolated sandboxes, and OIDC-based CI/CD give you a deployment that's ready for a security review.

Conclusion

The shift to Agentic Engineering doesn't mean replacing QA engineers with a chatbot. It means giving teams a tireless, self-correcting system that writes, hardens, and maintains the tests they never had time for, and that explains its reasoning along the way.

The architecture in this guide is deliberately practical. Python and LangGraph handle the intelligence, Node.js handles I/O, PostgreSQL holds the source of truth, Redis carries the real-time signal, and Stripe turns usage into revenue. Each piece is well understood, so you can focus your effort on the part that actually differentiates the product: agents that tell a broken test apart from a broken product.

Start small. Pick one framework (Playwright), one flow type (authentication and checkout), and ten design-partner customers. Get the reflect loop excellent before you add anything else. The market for "tests that maintain themselves" is large, the buyers already have budget for it, and the pain is real. Good luck with the build.

References

• LangGraph Documentation
• LangGraph Persistence and Checkpointers
• Ollama Documentation
• LangChain Ollama Integration
• Playwright Documentation: Best Practices
• PostgreSQL: SELECT ... FOR UPDATE SKIP LOCKED
• Socket.io Redis Adapter
• Stripe Usage-Based Billing and Meters
• Stripe Webhook Signature Verification
• shadcn/ui Components
• Zapier Stripe Integrations
• GitHub Actions: Configuring OpenID Connect in AWS
• Amazon ECS Developer Guide