run-pipeline.spec.ts

run-pipeline.spec.tsapps/worker/src/processors/automations/__tests__/run-pipeline.spec.tsraw ← __tests__12,676 B← Back to the app
1/**
2 * apps/worker — TASK-0113 run-pipeline UNIT tests.
3 *
4 * Pins the four pure pieces the spec's TASK-0113 "Tests" line requires:
5 *   1. state machine  — every invalid (from→to) pair throws; every valid one passes
6 *   2. backoff        — attempt n delay = 2^(n-1) s, capped at 600 s
7 *   3. log truncation — over-cap `data` blobs get a `truncated: true` flag and
8 *                       sensitive keys are redacted to `***`
9 *   4. emission rule  — `automation.executed` fires ONLY at depth 0 (no storm)
10 * Plus the shared loop guard (self-trigger block + depth limit) that the
11 * pipeline consults before any action runs.
12 *
13 * All pure / dependency-injected: no Postgres, no Redis, no BullMQ. A fake
14 * Redis records `xadd` calls so the emission rule is observable.
15 */
16import { describe, expect, it, vi } from "vitest";
17import {
18  runBackoffSeconds,
19  AUTOMATION_MAX_DEPTH,
20  loopGuard
21} from "@factory/shared";
22import { assertTransition, canTransition, isTerminal, TERMINAL } from "../run-state.js";
23import { redact, truncateLogData, LOG_DATA_MAX_BYTES } from "../log.js";
24import { emitAutomationExecuted } from "../dead-letter.js";
25import { autoRunMaxRetries } from "../pipeline.js";
26 
27// ---------------------------------------------------------------------------
28// 1. State machine
29// ---------------------------------------------------------------------------
30 
31const ALL = ["PENDING", "RUNNING", "SUCCEEDED", "FAILED", "DEAD_LETTER"] as const;
32 
33const VALID: ReadonlyArray<readonly [typeof ALL[number], typeof ALL[number]]> = [
34  ["PENDING", "RUNNING"],
35  ["RUNNING", "SUCCEEDED"],
36  ["RUNNING", "FAILED"],
37  ["RUNNING", "DEAD_LETTER"],
38  ["FAILED", "PENDING"],
39  ["FAILED", "DEAD_LETTER"],
40  ["PENDING", "FAILED"],
41  ["DEAD_LETTER", "PENDING"]
42];
43 
44describe("run state machine", () => {
45  it("accepts exactly the spec's valid transitions", () => {
46    for (const [from, to] of VALID) {
47      expect(canTransition(from, to), `${from} → ${to} should be valid`).toBe(true);
48      expect(() => assertTransition(from, to), `${from} → ${to} should not throw`).not.toThrow();
49    }
50  });
51 
52  it("rejects every other transition (25 total − 8 valid − 5 self-loops)", () => {
53    let rejected = 0;
54    for (const from of ALL) {
55      for (const to of ALL) {
56        const isValid = VALID.some(([f, t]) => f === from && t === to);
57        if (from === to) {
58          // self-loop is never a transition
59          expect(canTransition(from, to), `${from} → ${from} should be invalid`).toBe(false);
60          expect(() => assertTransition(from, to), `${from} → ${from} should throw`).toThrow(
61            /invalid automation run transition/
62          );
63          continue;
64        }
65        if (!isValid) {
66          expect(canTransition(from, to), `${from} → ${to} should be invalid`).toBe(false);
67          expect(() => assertTransition(from, to), `${from} → ${to} should throw`).toThrow(
68            /invalid automation run transition/
69          );
70          rejected++;
71        }
72      }
73    }
74    // 5×5 = 25 pairs; 8 valid; 5 self-loops; 25 - 8 - 5 = 12 invalid non-self pairs.
75    expect(rejected).toBe(12);
76  });
77 
78  it("SUCCEEDED and DEAD_LETTER are the terminal states", () => {
79    expect(TERMINAL).toEqual(["SUCCEEDED", "DEAD_LETTER"]);
80    for (const t of ALL) {
81      expect(isTerminal(t), `${t} terminal?`).toBe(t === "SUCCEEDED" || t === "DEAD_LETTER");
82    }
83  });
84 
85  it("terminal states have no outgoing edges", () => {
86    for (const from of TERMINAL) {
87      for (const to of ALL) {
88        if (from === to) continue;
89        if (from === "DEAD_LETTER" && to === "PENDING") continue; // manual retry
90        expect(canTransition(from, to), `${from} → ${to} should be invalid`).toBe(false);
91      }
92    }
93  });
94});
95 
96// ---------------------------------------------------------------------------
97// 2. Backoff schedule
98// ---------------------------------------------------------------------------
99 
100describe("run backoff schedule (2^(n-1) s, cap 600 s)", () => {
101  it("returns 0 for sub-1 / non-integer attempts", () => {
102    expect(runBackoffSeconds(0)).toBe(0);
103    expect(runBackoffSeconds(-1)).toBe(0);
104    expect(runBackoffSeconds(1.5)).toBe(0);
105    expect(runBackoffSeconds(NaN)).toBe(0);
106  });
107 
108  it("doubles each attempt until the cap", () => {
109    // attempt 1 = first retry → 2^0 = 1 s
110    expect(runBackoffSeconds(1)).toBe(1);
111    expect(runBackoffSeconds(2)).toBe(2);
112    expect(runBackoffSeconds(3)).toBe(4);
113    expect(runBackoffSeconds(4)).toBe(8);
114    expect(runBackoffSeconds(5)).toBe(16);
115    expect(runBackoffSeconds(6)).toBe(32);
116    expect(runBackoffSeconds(7)).toBe(64);
117    expect(runBackoffSeconds(8)).toBe(128);
118    expect(runBackoffSeconds(9)).toBe(256);
119    expect(runBackoffSeconds(10)).toBe(512); // 2^9 = 512, still under the cap
120  });
121 
122  it("caps at 600 s", () => {
123    // attempt n → 2^(n-1): 2^9 = 512 (< 600), 2^10 = 1024 (→ capped 600)
124    expect(runBackoffSeconds(10)).toBe(512);
125    expect(runBackoffSeconds(11)).toBe(600);
126    expect(runBackoffSeconds(99)).toBe(600);
127    // the schedule is monotonically non-decreasing and bounded
128    let prev = 0;
129    for (let n = 1; n <= 30; n++) {
130      const d = runBackoffSeconds(n);
131      expect(d).toBeGreaterThanOrEqual(prev);
132      expect(d).toBeLessThanOrEqual(600);
133      prev = d;
134    }
135  });
136 
137  it("autoRunMaxRetries honours the env override, else 8", () => {
138    const env = (v: unknown): Parameters<typeof autoRunMaxRetries>[0] =>
139      ({ AUTO_RUN_MAX_RETRIES: v }) as Parameters<typeof autoRunMaxRetries>[0];
140    expect(autoRunMaxRetries(env(8))).toBe(8);
141    expect(autoRunMaxRetries(env(3))).toBe(3);
142    expect(autoRunMaxRetries(env(0))).toBe(8); // < 1 → default
143    expect(autoRunMaxRetries(env(undefined))).toBe(8);
144    expect(autoRunMaxRetries(env(2.5))).toBe(8); // non-integer → default
145  });
146});
147 
148// ---------------------------------------------------------------------------
149// 3. Log redaction + truncation
150// ---------------------------------------------------------------------------
151 
152describe("log redaction", () => {
153  it("masks sensitive keys (case-insensitive, exact-name) recursively", () => {
154    const out = redact({
155      Authorization: "Bearer abc",
156      "x-api-key": "secret",
157      COOKIE: "sid=1",
158      nested: { "webhook-secret": "shh", token: "t", keyboard: "a key-like NON-secret" },
159      list: [{ password: "p", public: "ok" }]
160    }) as Record<string, unknown>;
161    expect(out.Authorization).toBe("***");
162    expect(out["x-api-key"]).toBe("***");
163    expect(out.COOKIE).toBe("***");
164    const nested = out.nested as Record<string, unknown>;
165    expect(nested["webhook-secret"]).toBe("***");
166    expect(nested.token).toBe("***");
167    // exact-name match only — a key that merely CONTAINS "key" is not masked
168    expect(nested.keyboard).toBe("a key-like NON-secret");
169    // noUncheckedIndexedAccess: assert on the (possibly-undefined) element
170    expect((out.list as Array<Record<string, unknown>>)[0]).toMatchObject({
171      password: "***",
172      public: "ok"
173    });
174  });
175 
176  it("passes scalars and unknown structures through", () => {
177    expect(redact("plain")).toBe("plain");
178    expect(redact(42)).toBe(42);
179    expect(redact(null)).toBe(null);
180    expect(redact([1, 2, { k: "v" }])).toEqual([1, 2, { k: "v" }]);
181  });
182});
183 
184describe("log truncation (LOG_DATA_MAX_BYTES = 32 KB)", () => {
185  it("returns small blobs untouched and without a truncated flag", () => {
186    const small = { hello: "world" };
187    expect(truncateLogData(small)).toEqual(small);
188  });
189 
190  it("stays within the cap and sets truncated:true on an over-cap blob", () => {
191    // A single huge string pushes the JSON well past 32 KB.
192    const big = { message: "x".repeat(LOG_DATA_MAX_BYTES * 4) };
193    const out = truncateLogData(big) as Record<string, unknown>;
194    expect(Buffer.byteLength(JSON.stringify(out), "utf8")).toBeLessThanOrEqual(LOG_DATA_MAX_BYTES);
195    expect(out.truncated).toBe(true);
196  });
197 
198  it("never exceeds the cap even with many medium strings", () => {
199    // Many ~2 KB strings: shortening each to 500 chars should bring it under cap.
200    const many: Record<string, string> = {};
201    for (let i = 0; i < 40; i++) many[`k${i}`] = "y".repeat(2048);
202    const out = truncateLogData(many) as Record<string, unknown>;
203    expect(Buffer.byteLength(JSON.stringify(out), "utf8")).toBeLessThanOrEqual(LOG_DATA_MAX_BYTES);
204  });
205});
206 
207// ---------------------------------------------------------------------------
208// 4. automation.executed emission rule (depth-0 only)
209// ---------------------------------------------------------------------------
210 
211describe("automation.executed emission (depth-0 only, no storm)", () => {
212  const fakeRedis = () => {
213    const xadd = vi.fn(async () => "1-1");
214    return { xadd, redis: { xadd } as never };
215  };
216 
217  it("emits at depth 0 (human/API-originated)", async () => {
218    const { redis, xadd } = fakeRedis();
219    const emitted = await emitAutomationExecuted({
220      redis,
221      workspaceId: "ws_1",
222      automationId: "auto_1",
223      runId: "run_1",
224      status: "SUCCEEDED",
225      triggerType: "record.created",
226      durationMs: 12,
227      depth: 0
228    });
229    expect(emitted).toBe(true);
230    expect(xadd).toHaveBeenCalledTimes(1);
231    // xadd(stream, "*", "event", "automation.executed", "payload", json, "ts", iso)
232    const args = xadd.mock.calls[0] as unknown[];
233    expect(args[0]).toBe("factory:events");
234    expect(args[2]).toBe("event");
235    expect(args[3]).toBe("automation.executed");
236    const payload = JSON.parse(args[5] as string);
237    expect(payload).toMatchObject({
238      workspaceId: "ws_1",
239      automationId: "auto_1",
240      runId: "run_1",
241      status: "SUCCEEDED",
242      triggerType: "record.created"
243    });
244  });
245 
246  it("does NOT emit at depth ≥ 1 (chained runs)", async () => {
247    for (const depth of [1, 4, 5, 99]) {
248      const { redis, xadd } = fakeRedis();
249      const emitted = await emitAutomationExecuted({
250        redis,
251        workspaceId: "ws_1",
252        automationId: "auto_1",
253        runId: `run_${depth}`,
254        status: "FAILED",
255        triggerType: "record.updated",
256        durationMs: 5,
257        depth
258      });
259      expect(emitted, `depth ${depth} should not emit`).toBe(false);
260      expect(xadd).not.toHaveBeenCalled();
261    }
262  });
263 
264  it("includes the error code on a failed terminal emit", async () => {
265    const { redis, xadd } = fakeRedis();
266    await emitAutomationExecuted({
267      redis,
268      workspaceId: "ws_1",
269      automationId: "auto_1",
270      runId: "run_e",
271      status: "FAILED",
272      triggerType: "schedule",
273      durationMs: 3,
274      error: { code: "ECONNREFUSED" },
275      depth: 0
276    });
277    const args = xadd.mock.calls[0] as unknown[];
278    const payload = JSON.parse(args[5] as string);
279    expect(payload.error).toEqual({ code: "ECONNREFUSED" });
280  });
281});
282 
283// ---------------------------------------------------------------------------
284// 5. Loop guard (self-trigger block + depth limit)
285// ---------------------------------------------------------------------------
286 
287describe("loop guard (self-trigger + depth limit)", () => {
288  it("allows a human/API event (no context) at depth 0", () => {
289    expect(loopGuard(null, "auto_1")).toEqual({ blocked: false });
290    expect(loopGuard(undefined, "auto_1")).toEqual({ blocked: false });
291  });
292 
293  it("blocks an automation reacting to its OWN event (self-trigger)", () => {
294    const d = loopGuard({ automationId: "auto_1", depth: 0 }, "auto_1");
295    // full-object assertion: `reason` only exists on the blocked branch of
296    // the LoopGuardDecision union (discriminated on `blocked`)
297    expect(d).toEqual({ blocked: true, reason: "self_trigger", depth: 0 });
298  });
299 
300  it("allows an event from a DIFFERENT automation below the depth cap", () => {
301    const d = loopGuard({ automationId: "auto_OTHER", depth: 2 }, "auto_1");
302    expect(d.blocked).toBe(false);
303  });
304 
305  it("blocks at the depth limit (depth >= AUTOMATION_MAX_DEPTH)", () => {
306    const d = loopGuard({ automationId: "auto_OTHER", depth: AUTOMATION_MAX_DEPTH }, "auto_1");
307    expect(d).toEqual({ blocked: true, reason: "depth_limit", depth: AUTOMATION_MAX_DEPTH });
308    // one above the cap is also blocked
309    const d2 = loopGuard({ automationId: "auto_OTHER", depth: AUTOMATION_MAX_DEPTH + 1 }, "auto_1");
310    expect(d2).toEqual({ blocked: true, reason: "depth_limit", depth: AUTOMATION_MAX_DEPTH + 1 });
311  });
312 
313  it("self-trigger wins over depth when both apply", () => {
314    const d = loopGuard({ automationId: "auto_1", depth: AUTOMATION_MAX_DEPTH }, "auto_1");
315    expect(d).toEqual({ blocked: true, reason: "self_trigger", depth: AUTOMATION_MAX_DEPTH });
316  });
317});
318