feat(deployment): add behavioural workload signals with replay tooling

apps/api/src/app/console.ts

@@ -104,6 +104,28 @@ program
     });
   });
 
+program
+  .command("replay-behavioural-signals")
+  .description("Re-score recorded workload probe evidence against the behavioural signals, writing nothing")
+  .option("-s, --since <date>", "Start of the evidence window", value => z.coerce.date().parse(value))
+  .option("-u, --until <date>", "End of the evidence window", value => z.coerce.date().parse(value))
+  .option("-f, --fixture <path...>", "Replay a fixture file or a directory of fixture files")
+  .option("-b, --bundled-fixtures", "Include the fixture corpus shipped with the repo", false)
+  .option("--accel-min-vram-mb <number>", "Override the accelerator memory threshold", value => z.number({ coerce: true }).parse(value))
+  .option("--artifact-min-mb <number>", "Override the artifact size threshold", value => z.number({ coerce: true }).parse(value))
+  .action(async (options, command) => {
+    await executeCliHandler(command.name(), async () => {
+      return container.resolve(WorkloadAbuseController).replayBehaviouralSignals({
+        since: options.since,
+        until: options.until,
+        fixturePaths: options.fixture,
+        bundledFixtures: options.bundledFixtures,
+        accelMinVramMb: options.accelMinVramMb,
+        artifactMinMb: options.artifactMinMb
+      });
+    });
+  });
+
 program
   .command("cleanup-provider-deployments")
   .description("Close trial deployments for a provider")

apps/api/src/workload-abuse/config/env.config.spec.ts

@@ -27,6 +27,32 @@ describe("workload abuse env config", () => {
     ]);
   });
 
+  it("defaults behavioural signals to off with no relay endpoints", () => {
+    const config = envSchema.parse({});
+
+    expect(config.WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED).toBe(false);
+    expect(config.WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS).toEqual([]);
+    expect(config.WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB).toBe(1024);
+    expect(config.WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB).toBe(256);
+  });
+
+  it("parses the relay endpoint list", () => {
+    const config = envSchema.parse({ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: JSON.stringify(["10.0.0.7", "10.0.0.8:443"]) });
+
+    expect(config.WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS).toEqual(["10.0.0.7", "10.0.0.8:443"]);
+  });
+
+  it("falls back to no relay endpoints when the variable is set but blank", () => {
+    const config = envSchema.parse({ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: "  " });
+
+    expect(config.WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS).toEqual([]);
+  });
+
+  it("rejects a relay endpoint list that is not a JSON array of strings", () => {
+    expect(() => envSchema.parse({ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: "10.0.0.7" })).toThrow(/ip or ip:port/);
+    expect(() => envSchema.parse({ WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: JSON.stringify([443]) })).toThrow(/ip or ip:port/);
+  });
+
   it("rejects a signature document that is not JSON", () => {
     expect(() => envSchema.parse({ WORKLOAD_ABUSE_SIGNATURES: "not json" })).toThrow(/valid signature document/);
   });
@@ -46,4 +72,28 @@ describe("workload abuse env config", () => {
   it("falls back to the default schedule when the initial delay setting is blank", () => {
     expect(envSchema.parse({ WORKLOAD_ABUSE_PROBE_INITIAL_DELAYS_MIN: " " }).WORKLOAD_ABUSE_PROBE_INITIAL_DELAYS_MIN).toEqual([5, 20, 60]);
   });
+
+  it("falls back to the behavioural signal defaults when those variables are set but blank", () => {
+    const config = envSchema.parse({
+      WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED: "",
+      WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: "",
+      WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: " "
+    });
+
+    expect(config.WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED).toBe(false);
+    expect(config.WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB).toBe(1024);
+    expect(config.WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB).toBe(256);
+  });
+
+  it("still reads the behavioural signal variables when they carry a value", () => {
+    const config = envSchema.parse({
+      WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED: "true",
+      WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: "2048",
+      WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: "512"
+    });
+
+    expect(config.WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED).toBe(true);
+    expect(config.WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB).toBe(2048);
+    expect(config.WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB).toBe(512);
+  });
 });

apps/api/src/workload-abuse/config/env.config.ts

@@ -69,6 +69,17 @@ function blankToUndefined(value: unknown): unknown {
   return typeof value === "string" && value.trim() === "" ? undefined : value;
 }
 
+function parseRelayEndpoints(raw: string, ctx: z.RefinementCtx): string[] {
+  const parsed = z.array(z.string().min(1)).safeParse(parseJson(raw));
+
+  if (!parsed.success) {
+    ctx.addIssue({ code: z.ZodIssueCode.custom, message: "WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS must be a JSON array of ip or ip:port strings" });
+    return z.NEVER;
+  }
+
+  return parsed.data;
+}
+
 export const envSchema = z.object({
   WORKLOAD_ABUSE_PROBE_ENABLED: z
     .enum(["true", "false"])
@@ -99,7 +110,18 @@ export const envSchema = z.object({
   /** A domain with an account older than this predates the attack, so it is somebody's real domain. */
   WORKLOAD_ABUSE_DOMAIN_BLOCK_MIN_ACCOUNT_AGE_DAYS: z.number({ coerce: true }).int().positive().default(30),
   /** Evidence rows feed the behavioural replay, so they must outlive its 30-day window with margin. */
-  WORKLOAD_ABUSE_EVIDENCE_RETENTION_DAYS: z.number({ coerce: true }).int().positive().default(90)
+  WORKLOAD_ABUSE_EVIDENCE_RETENTION_DAYS: z.number({ coerce: true }).int().positive().default(90),
+  WORKLOAD_ABUSE_BEHAVIOURAL_SIGNALS_ENABLED: z.preprocess(
+    blankToUndefined,
+    z
+      .enum(["true", "false"])
+      .default("false")
+      .transform(value => value === "true")
+  ),
+  WORKLOAD_ABUSE_SIGNAL_ACCEL_MIN_VRAM_MB: z.preprocess(blankToUndefined, z.number({ coerce: true }).int().positive().default(1_024)),
+  WORKLOAD_ABUSE_SIGNAL_ARTIFACT_MIN_MB: z.preprocess(blankToUndefined, z.number({ coerce: true }).int().positive().default(256)),
+  /** The list itself lives in Doppler: these are our own endpoints, so a reader learns how the exclusion works but not what it covers. */
+  WORKLOAD_ABUSE_SIGNAL_RELAY_ENDPOINTS: z.preprocess(blankToUndefined, z.string().default("[]").transform(parseRelayEndpoints))
 });
 
 export type WorkloadAbuseConfig = z.infer<typeof envSchema>;

apps/api/src/workload-abuse/controllers/workload-abuse.controller.spec.ts

@@ -1,6 +1,10 @@
 import { describe, expect, it } from "vitest";
 import { mock } from "vitest-mock-extended";
 
+import type {
+  BehaviouralReplaySummary,
+  BehaviouralSignalReplayService
+} from "@src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service";
 import type { ProbeEvidenceService } from "@src/workload-abuse/services/probe-evidence/probe-evidence.service";
 import type { TrialAbuseEnforcementJobService } from "@src/workload-abuse/services/trial-abuse-enforcement-job/trial-abuse-enforcement-job.service";
 import type { TrialWorkloadProbeJobService } from "@src/workload-abuse/services/trial-workload-probe-job/trial-workload-probe-job.service";
@@ -61,14 +65,26 @@ describe(WorkloadAbuseController.name, () => {
     expect(probeEvidenceService.purgeExpired).not.toHaveBeenCalled();
   });
 
+  it("returns the replay of the window it was asked for", async () => {
+    const { controller, behaviouralSignalReplayService } = setup();
+    const summary = mock<BehaviouralReplaySummary>({ deployments: [] });
+    behaviouralSignalReplayService.replay.mockResolvedValue(summary);
+    const options = { since: new Date("2026-08-01T00:00:00.000Z") };
+
+    await expect(controller.replayBehaviouralSignals(options)).resolves.toBe(summary);
+
+    expect(behaviouralSignalReplayService.replay).toHaveBeenCalledWith(options);
+  });
+
   function setup() {
     const probeJobService = mock<TrialWorkloadProbeJobService>();
     probeJobService.reconcile.mockResolvedValue();
     const enforcementJobService = mock<TrialAbuseEnforcementJobService>();
     enforcementJobService.reconcile.mockResolvedValue();
     const probeEvidenceService = mock<ProbeEvidenceService>();
-    const controller = new WorkloadAbuseController(probeJobService, enforcementJobService, probeEvidenceService);
+    const behaviouralSignalReplayService = mock<BehaviouralSignalReplayService>();
+    const controller = new WorkloadAbuseController(probeJobService, enforcementJobService, probeEvidenceService, behaviouralSignalReplayService);
 
-    return { controller, probeJobService, enforcementJobService, probeEvidenceService };
+    return { controller, probeJobService, enforcementJobService, probeEvidenceService, behaviouralSignalReplayService };
   }
 });

apps/api/src/workload-abuse/controllers/workload-abuse.controller.ts

@@ -1,6 +1,11 @@
 import { singleton } from "tsyringe";
 
 import type { DryRunOptions } from "@src/core/types/console";
+import {
+  type BehaviouralReplayOptions,
+  type BehaviouralReplaySummary,
+  BehaviouralSignalReplayService
+} from "@src/workload-abuse/services/behavioural-signal-replay/behavioural-signal-replay.service";
 import { ProbeEvidenceService } from "@src/workload-abuse/services/probe-evidence/probe-evidence.service";
 import { TrialAbuseEnforcementJobService } from "@src/workload-abuse/services/trial-abuse-enforcement-job/trial-abuse-enforcement-job.service";
 import { TrialWorkloadProbeJobService } from "@src/workload-abuse/services/trial-workload-probe-job/trial-workload-probe-job.service";
@@ -10,7 +15,8 @@ export class WorkloadAbuseController {
   constructor(
     private readonly probeJobService: TrialWorkloadProbeJobService,
     private readonly enforcementJobService: TrialAbuseEnforcementJobService,
-    private readonly probeEvidenceService: ProbeEvidenceService
+    private readonly probeEvidenceService: ProbeEvidenceService,
+    private readonly behaviouralSignalReplayService: BehaviouralSignalReplayService
   ) {}
 
   /** Each sweep runs whether or not the other fails, and retention runs whether or not the sweeps do, so nothing waits another run. */
@@ -24,4 +30,8 @@ export class WorkloadAbuseController {
     if (failures.length === 1) throw failures[0];
     if (failures.length > 1) throw new AggregateError(failures, "Both the probe sweep and the enforcement sweep failed");
   }
+
+  async replayBehaviouralSignals(options: BehaviouralReplayOptions): Promise<BehaviouralReplaySummary> {
+    return await this.behaviouralSignalReplayService.replay(options);
+  }
 }

apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.spec.ts

@@ -0,0 +1,101 @@
+import { describe, expect, it } from "vitest";
+
+import type { ProbeEvidenceAccelerator, ProbeEvidenceArtifact } from "@src/workload-abuse/model-schemas";
+import { evaluateAccelWithoutArtifacts } from "./accel-without-artifacts";
+import type { BehaviouralSignalParams, ProbeEvidenceSnapshot } from "./types";
+
+describe("evaluateAccelWithoutArtifacts", () => {
+  it("fires when a process holds the accelerator and nothing large sits on disk", () => {
+    const { snapshot, params } = setup({ vramMb: 18_000, artifacts: [{ path: "/opt/run", sizeBytes: 4_194_304 }] });
+
+    expect(evaluateAccelWithoutArtifacts(snapshot, params)).toEqual({
+      signal: "accel_without_artifacts",
+      detail: { heaviestVramMb: 18_000, largestArtifactMb: 4 }
+    });
+  });
+
+  it("reports the heaviest process when several share the accelerator", () => {
+    const { snapshot, params } = setup({
+      accelerator: [
+        {
+          name: "accelerator-0",
+          utilPct: 99,
+          memUsedMb: 20_000,
+          memTotalMb: 24_576,
+          processes: [
+            { pid: 10, name: "worker", vramMb: 2_000 },
+            { pid: 11, name: "worker", vramMb: 16_000 }
+          ]
+        }
+      ],
+      artifacts: []
+    });
+
+    expect(evaluateAccelWithoutArtifacts(snapshot, params)?.detail).toEqual({ heaviestVramMb: 16_000, largestArtifactMb: 0 });
+  });
+
+  it("stays silent when the workload carries large files", () => {
+    const { snapshot, params } = setup({ vramMb: 18_000, artifacts: [{ path: "/models/weights", sizeBytes: 8_589_934_592 }] });
+
+    expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull();
+  });
+
+  it("stays silent when no accelerator was reported", () => {
+    const { snapshot, params } = setup({ accelerator: null, artifacts: [] });
+
+    expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull();
+  });
+
+  it("stays silent when accelerator memory stays under the threshold", () => {
+    const { snapshot, params } = setup({ vramMb: 200, artifacts: [], accelMinVramMb: 1_024 });
+
+    expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull();
+  });
+
+  it("fires when the largest artifact sits just under the threshold", () => {
+    const { snapshot, params } = setup({ vramMb: 18_000, artifacts: [{ path: "/opt/run", sizeBytes: 268_016_025 }], artifactMinMb: 256 });
+
+    expect(evaluateAccelWithoutArtifacts(snapshot, params)?.detail).toEqual({ heaviestVramMb: 18_000, largestArtifactMb: 255 });
+  });
+
+  it("stays silent when the largest artifact reaches the threshold exactly", () => {
+    const { snapshot, params } = setup({ vramMb: 18_000, artifacts: [{ path: "/opt/run", sizeBytes: 268_435_456 }], artifactMinMb: 256 });
+
+    expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull();
+  });
+
+  it("stays silent when the probe collected no disk section", () => {
+    const { snapshot, params } = setup({ vramMb: 18_000, artifacts: null });
+
+    expect(evaluateAccelWithoutArtifacts(snapshot, params)).toBeNull();
+  });
+
+  function setup(input: {
+    vramMb?: number;
+    accelerator?: ProbeEvidenceAccelerator[] | null;
+    artifacts?: ProbeEvidenceArtifact[] | null;
+    accelMinVramMb?: number;
+    artifactMinMb?: number;
+  }) {
+    const accelerator =
+      input.accelerator !== undefined
+        ? input.accelerator
+        : [
+            {
+              name: "accelerator-0",
+              utilPct: 98,
+              memUsedMb: 20_480,
+              memTotalMb: 24_576,
+              processes: [{ pid: 1234, name: "worker", vramMb: input.vramMb ?? 18_000 }]
+            }
+          ];
+    const snapshot: ProbeEvidenceSnapshot = { shellStatus: "completed", accelerator, artifacts: input.artifacts ?? null, netShape: null };
+    const params: BehaviouralSignalParams = {
+      accelMinVramMb: input.accelMinVramMb ?? 1_024,
+      artifactMinMb: input.artifactMinMb ?? 256,
+      relayEndpoints: []
+    };
+
+    return { snapshot, params };
+  }
+});

apps/api/src/workload-abuse/lib/behavioural-signals/accel-without-artifacts.ts

@@ -0,0 +1,27 @@
+import { BEHAVIOURAL_SIGNALS, type BehaviouralFinding, type BehaviouralSignalParams, type ProbeEvidenceSnapshot } from "./types";
+
+const BYTES_PER_MB = 1024 * 1024;
+
+/** A snapshot without a disk section reports nothing rather than an empty disk, so it cannot support this signal. */
+export function evaluateAccelWithoutArtifacts(snapshot: ProbeEvidenceSnapshot, params: BehaviouralSignalParams): BehaviouralFinding | null {
+  const heaviestVramMb = findHeaviestVramMb(snapshot);
+
+  if (heaviestVramMb < params.accelMinVramMb) return null;
+  if (!snapshot.artifacts) return null;
+
+  const largestArtifactMb = findLargestArtifactMb(snapshot);
+
+  if (largestArtifactMb >= params.artifactMinMb) return null;
+
+  return { signal: BEHAVIOURAL_SIGNALS.accelWithoutArtifacts, detail: { heaviestVramMb, largestArtifactMb: Math.floor(largestArtifactMb) } };
+}
+
+function findHeaviestVramMb(snapshot: ProbeEvidenceSnapshot): number {
+  const vramPerProcess = (snapshot.accelerator ?? []).flatMap(accelerator => accelerator.processes.map(process => process.vramMb));
+  return vramPerProcess.reduce((heaviest, vramMb) => Math.max(heaviest, vramMb), 0);
+}
+
+function findLargestArtifactMb(snapshot: ProbeEvidenceSnapshot): number {
+  const sizes = (snapshot.artifacts ?? []).map(artifact => artifact.sizeBytes / BYTES_PER_MB);
+  return sizes.reduce((largest, sizeMb) => Math.max(largest, sizeMb), 0);
+}

apps/api/src/workload-abuse/lib/behavioural-signals/evaluate-behavioural-signals.spec.ts

@@ -0,0 +1,65 @@
+import { describe, expect, it } from "vitest";
+
+import { evaluateBehaviouralSignals, isBehaviouralCandidate } from "./evaluate-behavioural-signals";
+import type { BehaviouralSignalParams, ProbeEvidenceSnapshot } from "./types";
+
+describe("evaluateBehaviouralSignals", () => {
+  it("returns both findings when the snapshot matches both signals", () => {
+    const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, connections: [] });
+
+    expect(evaluateBehaviouralSignals(snapshot, params).map(finding => finding.signal)).toEqual(["accel_without_artifacts", "network_isolated"]);
+  });
+
+  it("returns one finding when the workload carries weights but talks to nobody", () => {
+    const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 8_589_934_592, connections: [] });
+
+    expect(evaluateBehaviouralSignals(snapshot, params).map(finding => finding.signal)).toEqual(["network_isolated"]);
+  });
+
+  it("returns nothing when the snapshot matches neither signal", () => {
+    const { snapshot, params } = setup({
+      vramMb: 18_000,
+      artifactBytes: 8_589_934_592,
+      connections: [{ localPort: 8_080, remoteIp: "203.0.113.9", remotePort: 51_000, count: 2, state: "established" }]
+    });
+
+    expect(evaluateBehaviouralSignals(snapshot, params)).toEqual([]);
+  });
+
+  it("returns nothing when the shell that collected the snapshot was cut short", () => {
+    const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, connections: [], shellStatus: "output_capped" });
+
+    expect(evaluateBehaviouralSignals(snapshot, params)).toEqual([]);
+  });
+
+  it("treats the two signals together as a candidate", () => {
+    const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 1_024, connections: [] });
+
+    expect(isBehaviouralCandidate(evaluateBehaviouralSignals(snapshot, params))).toBe(true);
+  });
+
+  it("treats a single signal as no candidate", () => {
+    const { snapshot, params } = setup({ vramMb: 18_000, artifactBytes: 8_589_934_592, connections: [] });
+
+    expect(isBehaviouralCandidate(evaluateBehaviouralSignals(snapshot, params))).toBe(false);
+  });
+
+  function setup(input: {
+    vramMb: number;
+    artifactBytes: number;
+    connections: NonNullable<ProbeEvidenceSnapshot["netShape"]>["connections"];
+    shellStatus?: string;
+  }) {
+    const snapshot: ProbeEvidenceSnapshot = {
+      shellStatus: input.shellStatus ?? "completed",
+      accelerator: [
+        { name: "accelerator-0", utilPct: 97, memUsedMb: 20_480, memTotalMb: 24_576, processes: [{ pid: 1234, name: "worker", vramMb: input.vramMb }] }
+      ],
+      artifacts: [{ path: "/opt/payload", sizeBytes: input.artifactBytes }],
+      netShape: { listenPorts: [8_080], connections: input.connections }
+    };
+    const params: BehaviouralSignalParams = { accelMinVramMb: 1_024, artifactMinMb: 256, relayEndpoints: [] };
+
+    return { snapshot, params };
+  }
+});