Chapter 23 · Capstone Project 3: Test Agent — CI Integration and Scheduled Regression Inspection

Chapter goals: Build on the cases.json artifacts from Chapter 18, implement two types of execution Agents — API automation and Web automation — aggregate test reports, and provide a production-grade integration checklist.

23.1 Execution Layer Architecture

Each case produced in Chapter 18 has a type field; route to different executors based on this:

cases.json
   │ Route by type
   ├── type = "api"   ──▶ API Execution Agent ──▶ runApiTest tool (fetch)
   ├── type = "ui"    ──▶ Web Execution Agent  ──▶ Playwright
   └── type = "mixed" ──▶ Split into api + ui steps executed separately
                │
                ▼
        TestResult[] (unified result structure)
                │
                ▼
        Report Agent ──▶ report.md (with failure analysis)

The unified result structure is the prerequisite for aggregated reporting:

// executor/types.ts —— Execution result contract
export interface TestResult {
  caseId: string;          // Corresponds to TestCase.id
  status: "passed" | "failed" | "skipped";
  durationMs: number;      // Execution duration
  error?: string;          // Failure reason (assertion diff / timeout / element not found)
  evidence?: string;       // Evidence: for API it is the response summary; for Web it is the screenshot path
}

23.2 API Automation Execution Agent

The core is a runApiTest tool: the Agent reads cases, translates steps into HTTP request parameters, and the tool internally executes and asserts with fetch.

// executor/apiTool.ts —— API test tool
import { Type } from "typebox";
import type { AgentTool } from "@earendil-works/pi-agent-core";
import type { TestResult } from "./types.js";

export function makeApiTestTool(baseUrl: string): AgentTool {
  return {
    name: "run_api_test",
    description:
      "Execute a single API test. Send an HTTP request and assert on status code and response fields." +
      "expectedJsonPath uses dot notation for nested fields (e.g. user.name);" +
      "On assertion failure return the actual value for analysis; do not retry more than once.",
    parameters: Type.Object({
      caseId: Type.String({ description: "Associated case ID, e.g. TC-001" }),
      method: Type.String({ description: "HTTP method, uppercase: GET/POST/PUT/DELETE" }),
      path: Type.String({ description: "API path, e.g. /api/cart/items" }),
      body: Type.Optional(Type.String({ description: "JSON string request body; leave empty for GET" })),
      expectStatus: Type.Number({ description: "Expected HTTP status code" }),
      expectJsonPath: Type.Optional(Type.String({ description: "Expected response field path for assertion" })),
      expectValue: Type.Optional(Type.String({ description: "Expected value for that field (stringified comparison)" })),
    }),
    execute: async (_id, params) => {
      const started = Date.now();
      const result: TestResult = {
        caseId: params.caseId, status: "passed",
        durationMs: 0,
      };
      try {
        // Send the real request
        const res = await fetch(baseUrl + params.path, {
          method: params.method,
          headers: { "Content-Type": "application/json" },
          body: params.body || undefined,
          signal: AbortSignal.timeout(10_000),   // 10-second timeout protection
        });

        // Assertion 1: status code
        if (res.status !== params.expectStatus) {
          throw new Error(`Status assertion failed: expected ${params.expectStatus}, got ${res.status}`);
        }

        // Assertion 2: response field (optional)
        if (params.expectJsonPath) {
          const json: any = await res.json();
          const actual = params.expectJsonPath
            .split(".")
            .reduce((node, key) => node?.[key], json);
          if (String(actual) !== params.expectValue) {
            throw new Error(
              `Field assertion failed ${params.expectJsonPath}: expected ${params.expectValue}, got ${actual}`);
          }
        }
        result.evidence = `HTTP ${res.status}, ${Date.now() - started}ms`;
      } catch (err) {
        // On failure do not throw to the model for retry — record a structured result instead to avoid meaningless re-runs
        result.status = "failed";
        result.error = (err as Error).message;
      }
      result.durationMs = Date.now() - started;
      return {
        content: [{ type: "text", text: JSON.stringify(result) }],
        details: result,
      };
    },
  };
}

Then create the API execution Agent to drive this tool in batch per case:

// executor/apiAgent.ts
import { Agent } from "@earendil-works/pi-agent-core";
import { createModels } from "@earendil-works/pi-ai";
import type { TestCase, TestResult } from "./types.js";
import { makeApiTestTool } from "./apiTool.js";

const models = createModels();
const model = models.getModel("anthropic", "claude-sonnet-4-6")!;

export async function runApiCases(
  cases: TestCase[],
  baseUrl: string,
): Promise<TestResult[]> {
  const results: TestResult[] = [];
  const agent = new Agent({
    initialState: {
      systemPrompt: `You are an API test executor.
Analyze the given api-type cases one by one; call the run_api_test tool once per case.
Translate each case's steps into method/path/body/expectStatus/expectJsonPath.
After all cases are executed, output only a JSON array (TestResult structure); no other text.`,
      model,
      tools: [makeApiTestTool(baseUrl)],
      messages: [],
    },
    streamFn: models.streamSimple.bind(models),
  });

  let reply = "";
  agent.subscribe((e) => {
    // Collect structured data directly from tool results (more reliable than parsing the final reply)
    if (e.type === "tool_execution_end" && e.toolCallId) {
      const r = e.result?.details as TestResult | undefined;
      if (r?.caseId) results.push(r);
    }
    if (e.type === "message_update" && e.assistantMessageEvent.type === "text_delta") {
      reply += e.assistantMessageEvent.delta;
    }
  });

  await agent.prompt(`Execute the following cases:\n${JSON.stringify(
    cases.filter((c) => c.type !== "ui"), null, 2)}`);

  console.log(`API execution complete: ${results.filter((r) => r.status === "passed").length}/${results.length} passed`);
  return results;
}

Collecting results from tool details

The details returned by execute appears in the tool_execution_end event — collect structured results directly from the event rather than having the model re-state them in the final reply (the model occasionally drops fields when paraphrasing JSON).

23.3 Web Automation Execution Agent

Hand UI cases to Playwright. The strategy is "Agent generates a spec file → CLI execution → parse results":

// executor/webAgent.ts —— Web automation: generate and run Playwright spec
import { Agent } from "@earendil-works/pi-agent-core";
import { createModels } from "@earendil-works/pi-ai";
import { execFile } from "node:child_process";
import { promisify } from "node:util";
import { writeFile, readFile } from "node:fs/promises";
import type { TestCase, TestResult } from "./types.js";

const run = promisify(execFile);
const models = createModels();
const model = models.getModel("anthropic", "claude-sonnet-4-6")!;

export async function runWebCases(
  cases: TestCase[],
  baseUrl: string,
): Promise<TestResult[]> {
  // ① Let the Agent translate UI cases into Playwright specs (TypeScript)
  const gen = new Agent({
    initialState: {
      systemPrompt: `You are a Playwright test development expert.
Translate UI cases into @playwright/test spec files.
Rules:
1. One test() per case; prefix the test title with the case ID, e.g. "TC-003 Login failure prompt";
2. At the start of each test await page.goto(baseUrl + relativePath);
3. Auto-screenshot on failure: test.use({ screenshot: "only-on-failure" });
4. Use expect + locator for assertions; do not use bare waitForTimeout;
5. Output only the complete spec file code; do not explain.`,
      model,
      messages: [],
    },
    streamFn: models.streamSimple.bind(models),
  });

  let spec = "";
  gen.subscribe((e) => {
    if (e.type === "message_update" && e.assistantMessageEvent.type === "text_delta") {
      spec += e.assistantMessageEvent.delta;
    }
  });
  await gen.prompt(
    `baseUrl is ${baseUrl}. Case list:\n${JSON.stringify(
      cases.filter((c) => c.type !== "api"), null, 2)}`);

  // ② Write to disk and execute
  const specPath = "tests/generated.spec.ts";
  await writeFile(specPath, spec.replace(/^```(?:typescript)?\s*/i, "").replace(/```\s*$/, ""));
  try {
    await run("npx", ["playwright", "test", specPath, "--reporter=json"], {
      cwd: process.cwd(),
      timeout: 300_000,
    });
  } catch (e: any) {
    // playwright test exits with non-zero when cases fail — this is expected; parse stdout
    console.error("Some cases failed; parsing report…");
  }

  // ③ Parse the JSON report into unified TestResult
  const report = JSON.parse(await readFile("test-results/.json-report", "utf-8"));
  return report.suites.flatMap((s: any) =>
    s.specs.map((sp: any) => ({
      caseId: sp.title.split(" ")[0],                    // "TC-003 ..." take the ID
      status: sp.ok ? "passed" : "failed",
      durationMs: sp.tests?.[0]?.results?.[0]?.duration ?? 0,
      error: sp.tests?.[0]?.results?.[0]?.error?.message,
      evidence: sp.tests?.[0]?.results?.[0]?.attachments
        ?.find((a: any) => a.contentType.includes("image"))?.path,
    })),
  );
}

Generated code must be reviewed

Letting the Agent directly生成 executable scripts is powerful, but never execute in production without review. Recommend: generate in an isolated directory first → manual or lint-rule spot check → then run. The beforeToolCall hook from Chapter 6 can also be used to intercept dangerous operations.

23.4 Report Agent: Aggregation and Failure Analysis

// executor/report.ts —— Aggregate both result types and produce a Markdown report
import { Agent } from "@earendil-works/pi-agent-core";
import { createModels } from "@earendil-works/pi-ai";
import { writeFile } from "node:fs/promises";
import type { TestResult } from "./types.js";

const models = createModels();
// A cheaper small model suffices for report analysis — save costs (see 23.5)
const model = models.getModel("anthropic", "claude-haiku-4-5")!;

export async function makeReport(
  apiResults: TestResult[],
  webResults: TestResult[],
): Promise<string> {
  const all = [...apiResults, ...webResults];
  const passed = all.filter((r) => r.status === "passed").length;

  // Generate the statistics table with plain code first — deterministic data need not bother the model
  const table = all.map((r) =>
    `| ${r.caseId} | ${r.status} | ${r.durationMs}ms | ${r.error ?? "-"} |`).join("\n");

  const agent = new Agent({
    initialState: {
      systemPrompt: `You are a test report writer. Based on the given list of failed cases:
1. Group by suspected root cause (frontend defect / backend defect / environment issue / case itself issue);
2. Give defect location suggestions for each group (which log layer to check first, reproduction path);
3. Output Markdown that can be pasted directly into a defect management tool.`,
      model,
      messages: [],
    },
    streamFn: models.streamSimple.bind(models),
  });

  const failed = all.filter((r) => r.status === "failed");
  let analysis = "";
  agent.subscribe((e) => {
    if (e.type === "message_update" && e.assistantMessageEvent.type === "text_delta") {
      analysis += e.assistantMessageEvent.delta;
    }
  });
  await agent.prompt(`Failed case details:\n${JSON.stringify(failed, null, 2)}`);

  const md = [
    `# Test Execution Report`,
    ``,
    `- Total ${all.length} cases, ${passed} passed (pass rate ${Math.round(passed / all.length * 100)}%)`,
    `- API: ${apiResults.length} cases / Web: ${webResults.length} cases`,
    ``,
    `## Details`,
    ``,
    `| Case | Result | Duration | Failure Reason |`,
    `|---|---|---|---|`,
    table,
    ``,
    `## Failure Analysis`,
    ``,
    analysis,
  ].join("\n");

  await writeFile("report.md", md);
  console.log(`📄 Report generated: report.md (pass rate ${Math.round(passed / all.length * 100)}%)`);
  return md;
}

23.5 Production Integration Checklist

To turn these three scripts into an operable service, the following elements还需补齐:

① HTTP Service Wrapper — expose the trigger entry with Fastify:

// server.ts —— HTTP entry that triggers the full pipeline
import Fastify from "fastify";
import { readFile } from "node:fs/promises";

const app = Fastify({ logger: true });

app.post<{ Body: { requirementPath: string; mockups?: string[]; baseUrl: string } }>(
  "/qa/run",
  async (req, reply) => {
    const { requirementPath, mockups = [], baseUrl } = req.body;
    // Idempotent: use the requirement document content hash as runId; repeat submissions return the cached report directly
    const runId = createHash("sha256")
      .update(await readFile(requirementPath))
      .digest("hex")
      .slice(0, 12);
    // In a real project this should be an async task queue; HTTP returns runId immediately
    const report = await runPipeline(requirementPath, mockups, baseUrl);
    return { runId, report };
  },
);

app.listen({ port: 3000 });

② Session Persistence for Re-runs — the SQLite backend from Chapter 17 shines here: persist each stage's output (features / reviews / plans / cases / results) per runId. On re-run, if a stage's input has not changed, read the cache and skip the LLM call.

③ Error Recovery and Retries — distinguish two kinds of failures:

• Infrastructure failures (network timeout, 429 rate limit): exponential backoff retry; agent.continue() can恢复 from the breakpoint mid-round;
• Logic failures (review not passed, JSON parse failed): do not blindly retry; record and hand to a human. For JSON parse failures, retry once and attach the specific position of the last parse error in the prompt.

④ Token Cost Control — allocate model tiers per stage:

Stage	Task Characteristics	Recommended Tier
Requirement parsing / review	Long-text understanding + judgment	Mid-tier (sonnet-level)
Test plan	Strategy design	Mid-high tier
Case generation (with mockups)	Multimodal + structured	Mid-high tier
API execution translation	Mechanical step translation	Low-tier (haiku-level)
Report analysis	Pattern归纳	Low-tier

Pair with AbortSignal.timeout to prevent runaway rounds, and batch cases.json (≤ 20 cases per batch) to avoid context explosion.

Chapter Summary

• Cases are routed by type to API/Web executors; the unified TestResult structure is the prerequisite for aggregation;
• API execution: run_api_test tool internally uses fetch for assertions; failures are recorded in results rather than thrown to the model for retry;
• Web execution: Agent generates a Playwright spec → npx playwright test runs → JSON report is parsed; failures auto-screenshot as evidence;
• Report = deterministic statistics (calculated by code) + failure root-cause analysis (written by a low-tier model), generating Markdown;
• Production four-piece set: Fastify service entry, SQLite session cache, retry strategy distinguishing infrastructure/logic failures, and model cost control by tiered allocation.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. Why collect test results from the details of tool_execution_end events instead of parsing the model's final text reply?

2. For Web automation, why adopt "Agent generates spec → CLI execution" instead of letting the Agent incrementally call browser tools?

3. In API test execution, if a case's assertion fails, how should the tool handle it internally?

4. Regarding model tier allocation across stages, which distribution is most reasonable?

🛠️ Hands-on Practice

1. Add a mock API service (e.g. json-server) to the cases.json from Chapter 18 and end-to-end run API execution and report generation.
2. Add a "retry failed cases once automatically" mechanism to the Web executor; compare pass rates before and after retry and ponder which failures are flaky.
3. Implement the idempotent design from 23.5: on a second submission of the same requirement document, return the cached report directly without triggering any LLM call.

The three-chapter capstone project ends here. Looking back at the entire course, you now have the complete chain from unified LLM API to multi-Agent production systems — next chapter: CI Integration and Scheduled Regression Inspection connects the pipeline to continuous integration.