import { generateKeyPairSync, randomUUID } from "node:crypto";
import type { PoolClient } from "pg";
import { HttpException } from "@nestjs/common";
import { beforeEach, describe, expect, it } from "vitest";
import {
  CapacityGate, CapacityQueueFullError, ChannelPendingCapacityError, DEFAULT_CHANNEL_QUOTA, FAILOVER_RESERVE,
  capacityUnits, effectiveUnits, fairShareOrder, jobResourceCaps, pendingBoundFor,
} from "./capacity.js";
import { MemoryRedis } from "../production/testing/memory-redis.js";
import { MediaControlService, MEDIA_CONTROL_KEYS as K } from "./media-control.service.js";
import { StubCredentialHandleProvider } from "./credential-handles.js";
import { Ed25519ControlSigner } from "./wire.js";
import type { RedisReply } from "./redis.js";
import type { DesiredSpec } from "./spec.js";
import type { DirectorReceipt, TenantTransaction } from "../production/journal.js";

/**
 * Capacity acceptance: per-worker
 * unit accounting, the PENDING_CAPACITY queue with fair share across
 * workspaces, admission on the take/start path (over-quota write rejected with
 * nothing persisted), the bounded pending queue (8 = 2 × the default quota 4 →
 * 429 CAPACITY_QUEUE_FULL), concurrency at quota − 1, tenant isolation, and
 * per-job resource-cap isolation at the accounting level.
 *
 * `MemoryRedis` does not implement the capacity Lua scripts, so the gate runs
 * its read-modify-write fallback under the in-process per-worker lock — the
 * same path a Redis fake without `eval` support takes.
 */

const WORKER = "media-worker-a";
const HANDLE = "vh_0123456789abcdefXYZ";
const ACTOR = { userId: null, role: "system" } as const;

function running(profile: DesiredSpec["profile"] = "720p30"): DesiredSpec {
  return {
    desiredStatus: "RUNNING",
    source: { kind: "hls", url: "https://cdn.example.com/live/index.m3u8" },
    output: { endpointUrl: "rtmps://live.example.net/app", credentialHandleId: HANDLE },
    profile, mode: "passthrough",
  };
}
const stopped: DesiredSpec = { desiredStatus: "STOPPED", source: null, output: null, profile: "720p30", mode: "passthrough" };

/** Drives the gate's read-modify-write fallback: the capacity scripts are unsupported here. */
class FakeRedis extends MemoryRedis {
  override async eval(script: string, keys: string[], args: string[]): Promise<RedisReply> {
    if (script.includes("tostring(cur)") || script.includes("tostring(v)")) throw new Error("no script support");
    return super.eval(script, keys, args);
  }
}

/** Aggregate DB fake: the director path (production channels) starts/stops channels. */
class ProductionDb {
  state = { version: 0, receipts: [] as DirectorReceipt[] };
  applied: Array<{ ch: string; status: string }> = [];
  failAfter = 0;
  readonly transaction: TenantTransaction = async <T>(_ws: string, fn: (client: PoolClient) => Promise<T>): Promise<T> => {
    const before = structuredClone(this.state);
    const query = async (sql: string, values: unknown[] = []): Promise<{ rows: unknown[]; rowCount: number }> => {
      let rows: unknown[] = [];
      if (sql.startsWith("SELECT id FROM virtual_channels")) rows = [{ id: values[1] }]; // this channel exists
      else if (sql.startsWith("SELECT desired_version")) rows = [{ desired_version: String(this.state.version), desired_status: "STOPPED" }];
      else if (sql.includes("FROM audit_log")) {
        const found = sql.includes("requestId")
          ? this.state.receipts.find(r => r.requestId === values[2])
          : this.state.receipts.at(-1);
        rows = found ? [{ diff: found }] : [];
      } else if (sql.startsWith("UPDATE virtual_channels")) {
        this.state.version = Number(values[2]);
        this.applied.push({ ch: String(values[1]), status: String(values[3]) });
      } else if (sql.startsWith("SELECT audit_append")) {
        this.state.receipts.push(values[6] as DirectorReceipt);
        if (this.failAfter > 0) { this.failAfter -= 1; throw new Error("commit refused"); }
      }
      return { rows, rowCount: rows.length };
    };
    try { return await fn({ query } as unknown as PoolClient); }
    catch (error) { this.state = before; throw error; } // PostgreSQL rolls back with the transaction
  };
}

const WS = randomUUID();
const CH = randomUUID();

async function catchError(promise: Promise<unknown>): Promise<HttpException> {
  const error: unknown = await promise.then(() => null, (e: unknown) => e);
  expect(error).toBeInstanceOf(HttpException);
  return error as HttpException;
}
function codeOf(error: HttpException): unknown {
  const response = error.getResponse();
  return typeof response === "object" && response !== null ? (response as Record<string, unknown>).code : response;
}

describe("§4.4 unit accounting", () => {
  it("charges each running channel its profile's units of its worker's capacity", () => {
    expect(capacityUnits("720p30")).toBe(4);
    expect(capacityUnits("1080p30")).toBe(6);
    expect(capacityUnits("1080p60")).toBe(10);
  });

  it("reserves the 20 % failover reserve from advertised capacity", () => {
    expect(effectiveUnits(100, FAILOVER_RESERVE)).toBe(80);
    expect(effectiveUnits(25, FAILOVER_RESERVE)).toBe(20); // 4 × 720p30 channels
    expect(effectiveUnits(7, FAILOVER_RESERVE)).toBe(5); // floor
  });

  it("scales per-job caps with reserved units so Σ caps fit the worker", () => {
    expect(jobResourceCaps("1080p60")).toEqual({ capacityUnits: 10, cpuShares: 2560, memoryMiB: 5120, fdBudget: 256, egressMbps: 20 });
    // Two 720p30 jobs fit a 10-unit worker (2×4 = 8 ≤ 10) with disjoint budgets:
    // a job hitting its own cap cannot take a sibling's budget.
    const a = jobResourceCaps("720p30"), b = jobResourceCaps("720p30");
    expect(a.cpuShares + b.cpuShares).toBeLessThanOrEqual(10 * 256);
    expect(a.memoryMiB + b.memoryMiB).toBeLessThanOrEqual(10 * 512);
    expect(a.fdBudget).toBe(b.fdBudget); // per-job RLIMIT_NOFILE, not shared
  });
});

describe("fair share across workspaces", () => {
  const entry = (ws: string, seq: number, units = 4) => ({ workspaceId: ws, seq, units });

  it("interleaves workspaces so one tenant's queue cannot delay another", () => {
    const order = fairShareOrder([entry("A", 1), entry("A", 3), entry("A", 5), entry("B", 2)], () => 1).map(i => i.workspaceId);
    expect(order).toEqual(["A", "B", "A", "A"]);
  });

  it("a heavier tenant earns proportionally more service without starving the light one", () => {
    // A (weight 2) is served, then B's ratio is lower, so B runs second.
    const order = fairShareOrder([entry("A", 1), entry("A", 2), entry("B", 3), entry("A", 4), entry("A", 5)], ws => (ws === "A" ? 2 : 1));
    expect(order.map(i => i.workspaceId)).toEqual(["A", "B", "A", "A", "A"]);
    expect(order.findIndex(i => i.workspaceId === "B")).toBe(1);
  });

  it("keeps FIFO order within a workspace (admission_requested_at)", () => {
    expect(fairShareOrder([entry("A", 9), entry("A", 2), entry("A", 5)], () => 1).map(i => i.seq)).toEqual([2, 5, 9]);
  });
});

describe("CapacityGate admission", () => {
  let redis: MemoryRedis;
  let gate: CapacityGate;

  beforeEach(() => {
    redis = new MemoryRedis();
    gate = new CapacityGate({ redis, now: () => Date.parse("2026-10-07T12:00:00.000Z") });
  });

  const start = (channelId: string, spec: DesiredSpec = running(), key = channelId) =>
    gate.admit(WS, channelId, WORKER, spec, ACTOR, key, "desired-v1", null);
  const queued = (promise: Promise<unknown>) => catchError(promise).then(error => {
    expect(error).toBeInstanceOf(ChannelPendingCapacityError);
    expect(error.getStatus()).toBe(202);
    expect(codeOf(error)).toBe("PENDING_CAPACITY");
    return error as ChannelPendingCapacityError;
  });

  it("4 placed channels; the 5th queues as PENDING_CAPACITY while running channels are unaffected", async () => {
    await gate.advertiseWorker(WORKER, 25); // effective 20 → four 4-unit channels
    const placed = Array.from({ length: 4 }, () => randomUUID());
    for (const channelId of placed) await expect(start(channelId)).resolves.toMatchObject({ placed: true, units: 4 });
    for (const channelId of placed) expect(await gate.channelOf(WS, channelId)).toMatchObject({ state: "PLACED", units: 4 });

    const fifth = randomUUID();
    const error = await queued(start(fifth));
    expect((error.getResponse() as Record<string, unknown>).position).toBe(1);
    expect(await gate.channelOf(WS, fifth)).toMatchObject({ state: "PENDING_CAPACITY", units: 4 });
    // The four running channels keep their placements and reservations.
    const snapshot = await gate.snapshot(WS, [...placed, fifth]);
    expect(snapshot.placedChannels).toBe(4);
    expect(snapshot.pendingChannels).toBe(1);
    expect(snapshot.workers[0]).toMatchObject({ usedUnits: 16, queued: 1 });
  });

  it("freeing capacity never places more than the quota", async () => {
    await gate.advertiseWorker(WORKER, 25);
    const placed = Array.from({ length: 4 }, () => randomUUID());
    for (const channelId of placed) await start(channelId);
    const waiting = Array.from({ length: 5 }, () => randomUUID());
    for (const channelId of waiting) await queued(start(channelId));
    const all = [...placed, ...waiting];

    for (const channelId of placed) {
      await gate.release(WS, channelId);
      const admitted = await gate.drain(WORKER);
      expect(admitted.length).toBeLessThanOrEqual(1);
      const snapshot = await gate.snapshot(WS, all);
      expect(snapshot.placedChannels).toBeLessThanOrEqual(DEFAULT_CHANNEL_QUOTA);
      expect(snapshot.placedChannels + snapshot.pendingChannels).toBe(9 - placed.indexOf(channelId) - 1);
      expect(snapshot.workers[0]?.usedUnits).toBeLessThanOrEqual(20);
    }
    const final = await gate.snapshot(WS, all);
    expect(final.placedChannels).toBe(DEFAULT_CHANNEL_QUOTA); // exactly the quota, never more
    expect(final.pendingChannels).toBe(1);
  });

  it("drives a workspace to the pending bound and rejects the 9th with 429 CAPACITY_QUEUE_FULL, nothing persisted", async () => {
    await gate.advertiseWorker(WORKER, 25);
    const placed = Array.from({ length: 4 }, () => randomUUID());
    for (const channelId of placed) await start(channelId);
    const waiting: string[] = [];
    for (let i = 0; i < pendingBoundFor(DEFAULT_CHANNEL_QUOTA); i += 1) {
      const channelId = randomUUID();
      waiting.push(channelId);
      await queued(start(channelId));
    }
    expect(pendingBoundFor(DEFAULT_CHANNEL_QUOTA)).toBe(8); // 2 × the default quota of 4

    const ninth = randomUUID();
    const error = await catchError(start(ninth));
    expect(error).toBeInstanceOf(CapacityQueueFullError);
    expect(error.getStatus()).toBe(429);
    expect(codeOf(error)).toBe("CAPACITY_QUEUE_FULL");
    // Nothing persisted for the rejected start.
    expect(await gate.channelOf(WS, ninth)).toBeNull();
    const snapshot = await gate.snapshot(WS, [ninth]);
    expect(snapshot.placedChannels).toBe(0);
    expect(snapshot.pendingChannels).toBe(0);
    expect((await gate.snapshot(WS, [...placed, ...waiting])).pendingChannels).toBe(8);
  });

  it("two concurrent starts at quota − 1 admit exactly one", async () => {
    await gate.advertiseWorker(WORKER, 25); // effective 20: 4 channels by capacity, quota 4
    for (let i = 0; i < DEFAULT_CHANNEL_QUOTA - 1; i += 1) await start(randomUUID());
    const a = randomUUID(), b = randomUUID();
    const results = await Promise.allSettled([start(a), start(b)]);
    expect(results.filter(r => r.status === "fulfilled")).toHaveLength(1);
    const rejected = results.filter(r => r.status === "rejected").map(r => (r as PromiseRejectedResult).reason);
    expect(rejected).toHaveLength(1);
    expect(rejected[0]).toBeInstanceOf(ChannelPendingCapacityError);
    // One placed, one queued — both accounted, never two placements.
    expect((await gate.snapshot(WS, [a, b])).placedChannels).toBe(1);
    const states = await Promise.all([gate.channelOf(WS, a), gate.channelOf(WS, b)]);
    expect(states.filter(c => c?.state === "PLACED")).toHaveLength(1);
    expect(states.filter(c => c?.state === "PENDING_CAPACITY")).toHaveLength(1);
  });

  it("a tenant saturating its quota cannot delay another tenant's placement", async () => {
    await gate.advertiseWorker(WORKER, 100); // effective 80: room for both tenants
    const other = randomUUID();
    // Tenant A saturates its quota and fills its pending queue.
    for (let i = 0; i < DEFAULT_CHANNEL_QUOTA; i += 1) await start(randomUUID());
    for (let i = 0; i < pendingBoundFor(DEFAULT_CHANNEL_QUOTA); i += 1) await queued(start(randomUUID()));
    // Tenant B places immediately, ahead of A's queued channels.
    const b1 = randomUUID(), b2 = randomUUID();
    await expect(gate.admit(other, b1, WORKER, running(), ACTOR, b1, "desired-v1", null)).resolves.toMatchObject({ placed: true });
    await expect(gate.admit(other, b2, WORKER, running(), ACTOR, b2, "desired-v1", null)).resolves.toMatchObject({ placed: true });
    expect((await gate.snapshot(other, [b1, b2])).placedChannels).toBe(2);
  });

  it("re-admitting under the same idempotency key is a replay counted once", async () => {
    await gate.advertiseWorker(WORKER, 25);
    const channelId = randomUUID();
    await start(channelId, running(), "same-key");
    await expect(start(channelId, running(), "same-key")).resolves.toMatchObject({ replay: true });
    expect((await gate.snapshot(WS, [channelId])).placedChannels).toBe(1);
  });

  it("a profile change beyond free worker capacity queues instead of overcommitting", async () => {
    await gate.advertiseWorker(WORKER, 10); // effective 8: one 6-unit channel fits
    const channelId = randomUUID();
    await start(channelId, running("1080p30"));
    const second = randomUUID();
    await queued(start(second, running("1080p60"))); // 10 units > 2 free
    expect(await gate.channelOf(WS, second)).toMatchObject({ state: "PENDING_CAPACITY" });
    expect(await gate.channelOf(WS, channelId)).toMatchObject({ state: "PLACED", units: 6 }); // unaffected
  });

  it("undo reverses reservations so a failed desired-state write leaks no capacity", async () => {
    await gate.advertiseWorker(WORKER, 25);
    const channelId = randomUUID();
    const admitted = await start(channelId);
    await admitted.undo();
    expect(await gate.channelOf(WS, channelId)).toBeNull();
    expect((await gate.snapshot(WS, [channelId])).placedChannels).toBe(0);
    const refill = randomUUID();
    await expect(start(refill)).resolves.toMatchObject({ placed: true }); // units are reusable
    expect((await gate.snapshot(WS, [refill])).workers[0]).toMatchObject({ usedUnits: 4 });
  });

  it("records per-job throttle/kill events without touching sibling reservations", async () => {
    await gate.advertiseWorker(WORKER, 25);
    const a = randomUUID(), b = randomUUID();
    await start(a);
    await start(b);
    await gate.recordResourceEvent(WS, a, "killed", "memory.max exceeded");
    expect(await gate.channelOf(WS, a)).toMatchObject({ state: "PLACED", units: 4, lastEvent: { kind: "killed", reason: "memory.max exceeded" } });
    // b's placement and budget are untouched: the kill is isolated to a's own job.
    const sibling = await gate.channelOf(WS, b);
    expect(sibling).toMatchObject({ state: "PLACED", units: 4 });
    expect(sibling?.lastEvent).toBeUndefined();
    const snapshot = await gate.snapshot(WS, [a, b]);
    expect(snapshot.placedChannels).toBe(2);
    expect(snapshot.workers[0]?.usedUnits).toBe(8);
    expect(snapshot.channels.find(c => c.channelId === a)?.caps).toEqual(jobResourceCaps("720p30"));
  });

  it("without an advertised capacity nothing is accounted (opt-in per worker)", async () => {
    const channelId = randomUUID();
    const admitted = await start(channelId);
    expect(admitted.channel).toBeNull();
    expect(await gate.channelOf(WS, channelId)).toBeNull();
  });
});

describe("admission wired into the start path (setDesired)", () => {
  let redis: FakeRedis;
  let db: ProductionDb;
  let service: MediaControlService;

  beforeEach(async () => {
    redis = new FakeRedis();
    db = new ProductionDb();
    service = new MediaControlService({
      redis, transaction: db.transaction, assignments: [],
      signer: new Ed25519ControlSigner("test", generateKeyPairSync("ed25519").privateKey),
      handles: new StubCredentialHandleProvider(),
      now: () => Date.parse("2026-10-07T12:00:00.000Z"), commandTtlMs: 30_000,
      capacity: { workerUnits: { [WORKER]: 100 } },
    });
    await service.advertiseWorkerCapacity(WORKER, 100); // effective 80: room for both tenants
  });

  const start = (channelId: string, spec: DesiredSpec = running(), key = channelId) =>
    service.setDesired(WS, channelId, spec, { idempotencyKey: key, actor: ACTOR });
  const assign = (channelId: string) => service.assign(WS, channelId, WORKER, 1);
  const desiredKey = (channelId: string) => redis.data.get(K.desired(WS, channelId)) ?? null;
  const versionKey = (channelId: string) => redis.data.get(K.version(WS, channelId)) ?? null;

  it("4 placed channels; the 5th stays PENDING_CAPACITY while running channels are unaffected", async () => {
    const placed = [CH, ...Array.from({ length: 3 }, () => randomUUID())];
    for (const channelId of placed) { await assign(channelId); await expect(start(channelId)).resolves.toBeTruthy(); }
    for (const channelId of placed) expect(JSON.parse(desiredKey(channelId) ?? "null")).toMatchObject({ desiredStatus: "RUNNING" });

    const fifth = randomUUID();
    await assign(fifth);
    const error = await catchError(start(fifth));
    expect(codeOf(error)).toBe("PENDING_CAPACITY");
    expect((await service.channelCapacity(WS, fifth))?.state).toBe("PENDING_CAPACITY");
    // Nothing persisted for the queued start...
    expect(versionKey(fifth)).toBeNull();
    expect(desiredKey(fifth)).toBeNull();
    // ...and the four running channels keep their committed desired state.
    for (const channelId of placed) expect(JSON.parse(desiredKey(channelId) ?? "null")).toMatchObject({ desiredStatus: "RUNNING" });
    const capacity = await service.capacityGate.snapshot(WS, [...placed, fifth]);
    expect(capacity).toMatchObject({ placedChannels: 4, pendingChannels: 1 });
    expect(capacity.workers[0]).toMatchObject({ usedUnits: 16 });
  });

  it("an over-quota desired-state write is rejected with nothing persisted", async () => {
    const placed = [CH, ...Array.from({ length: 3 }, () => randomUUID())];
    for (const channelId of placed) { await assign(channelId); await start(channelId); }
    const before = redis.published.length;
    const fifth = randomUUID();
    await assign(fifth);
    await expect(start(fifth)).rejects.toThrow();
    expect(versionKey(fifth)).toBeNull();
    expect(desiredKey(fifth)).toBeNull();
    expect(db.applied.filter(a => a.ch === fifth)).toEqual([]);
    expect(redis.published.length).toBe(before); // no command issued
  });

  it("freeing worker capacity never places more than the quota", async () => {
    const placed = [CH, ...Array.from({ length: 3 }, () => randomUUID())];
    for (const channelId of placed) { await assign(channelId); await start(channelId); }
    const waiting = Array.from({ length: 4 }, () => randomUUID());
    for (const channelId of waiting) { await assign(channelId); await expect(start(channelId)).rejects.toThrow(); }

    await start(placed[0]!, stopped, `${placed[0]}:stop`); // stop → free 4 units → the queue drains one
    const capacity = await service.capacityGate.snapshot(WS, [...placed, ...waiting]);
    expect(capacity.placedChannels).toBe(DEFAULT_CHANNEL_QUOTA);
    expect(capacity.pendingChannels).toBe(3);
    expect(capacity.workers[0]?.usedUnits).toBe(16);
    // The admitted channel's start was re-applied: a committed desired snapshot exists.
    const waitingIds: readonly string[] = waiting;
    const admitted = capacity.channels.find(c => c.state === "PLACED" && waitingIds.includes(c.channelId));
    expect(admitted).toBeDefined();
    const admittedId = admitted?.channelId;
    expect(typeof admittedId).toBe("string");
    expect(JSON.parse(desiredKey(admittedId ?? "") ?? "null")).toMatchObject({ desiredStatus: "RUNNING" });
    // The stopped channel no longer holds capacity.
    expect(await service.channelCapacity(WS, placed[0]!)).toBeNull();
  });

  it("drives a workspace to the pending bound (8) and 429s the 9th with nothing persisted", async () => {
    const placed = [CH, ...Array.from({ length: 3 }, () => randomUUID())];
    for (const channelId of placed) { await assign(channelId); await start(channelId); }
    const waiting: string[] = [];
    for (let i = 0; i < pendingBoundFor(DEFAULT_CHANNEL_QUOTA); i += 1) {
      const channelId = randomUUID();
      waiting.push(channelId);
      await assign(channelId);
      await expect(start(channelId)).rejects.toThrow("PENDING_CAPACITY");
    }
    const before = redis.published.length;
    const ninth = randomUUID();
    await assign(ninth);
    const error = await catchError(start(ninth));
    expect(error).toBeInstanceOf(CapacityQueueFullError);
    expect(error.getStatus()).toBe(429);
    expect(codeOf(error)).toBe("CAPACITY_QUEUE_FULL");
    // Nothing persisted: no Redis desired state, no capacity record, no DB write, no command.
    expect(versionKey(ninth)).toBeNull();
    expect(desiredKey(ninth)).toBeNull();
    expect(await service.channelCapacity(WS, ninth)).toBeNull();
    expect(db.applied.filter(a => a.ch === ninth)).toEqual([]);
    expect(redis.published.length).toBe(before);
    expect((await service.capacityGate.snapshot(WS, [...placed, ...waiting])).pendingChannels).toBe(8);
  });

  it("two concurrent starts at quota − 1 admit exactly one", async () => {
    const placed = [CH, ...Array.from({ length: DEFAULT_CHANNEL_QUOTA - 2 }, () => randomUUID())];
    for (const channelId of placed) { await assign(channelId); await start(channelId); }
    const a = randomUUID(), b = randomUUID();
    await assign(a); await assign(b);
    const results = await Promise.allSettled([start(a), start(b)]);
    expect(results.filter(r => r.status === "fulfilled")).toHaveLength(1);
    expect(results.filter(r => r.status === "rejected")).toHaveLength(1);
    const capacity = await service.capacityGate.snapshot(WS, [...placed, a, b]);
    expect(capacity.placedChannels).toBe(DEFAULT_CHANNEL_QUOTA); // never one more
    expect(capacity.pendingChannels).toBe(1);
    expect(capacity.workers[0]?.usedUnits).toBe(16);
  });

  it("a tenant saturating its quota cannot delay another tenant's placement", async () => {
    const tenantB = randomUUID();
    const b1 = randomUUID(), b2 = randomUUID();
    await service.assign(tenantB, b1, WORKER, 1);
    await service.assign(tenantB, b2, WORKER, 1);
    // Tenant A: quota saturated and its pending queue full.
    const placed = [CH, ...Array.from({ length: 3 }, () => randomUUID())];
    for (const channelId of placed) { await assign(channelId); await start(channelId); }
    for (let i = 0; i < pendingBoundFor(DEFAULT_CHANNEL_QUOTA); i += 1) {
      const channelId = randomUUID();
      await assign(channelId);
      await expect(start(channelId)).rejects.toThrow("PENDING_CAPACITY");
    }
    // Tenant B still places immediately.
    await expect(service.setDesired(tenantB, b1, running(), { idempotencyKey: b1, actor: ACTOR })).resolves.toBeTruthy();
    await expect(service.setDesired(tenantB, b2, running(), { idempotencyKey: b2, actor: ACTOR })).resolves.toBeTruthy();
    expect((await service.capacityGate.snapshot(tenantB, [b1, b2])).placedChannels).toBe(2);
    expect(await service.channelCapacity(tenantB, b1)).toMatchObject({ state: "PLACED" });
    expect(await service.channelCapacity(tenantB, b2)).toMatchObject({ state: "PLACED" });
  });

  it("per-job cap throttle/kill isolation is visible in the accounting", async () => {
    await assign(CH);
    await start(CH);
    await service.recordResourceEvent(WS, CH, "throttled", "cpu.weight reached");
    expect(await service.channelCapacity(WS, CH)).toMatchObject({
      state: "PLACED", units: 4, lastEvent: { kind: "throttled", reason: "cpu.weight reached" },
    });
    const status = await service.getStatus(WS);
    expect(status.capacity?.channels[0]?.caps).toEqual(jobResourceCaps("720p30"));
  });

  it("a mid-transaction failure after admission leaves no capacity leak", async () => {
    const other = randomUUID();
    await assign(other);
    db.failAfter = 1; // audit_append throws → the whole transaction rolls back
    await expect(start(other)).rejects.toThrow("commit refused");
    expect(await service.channelCapacity(WS, other)).toBeNull();
    const snapshot = await service.capacityGate.snapshot(WS, [other]);
    expect(snapshot.placedChannels).toBe(0);
    expect(versionKey(other)).toBeNull();
    // The leaked-free units are reusable by a later start.
    await expect(start(other)).resolves.toBeTruthy();
  });
});

describe("admission on the director (take/start) path", () => {
  let redis: FakeRedis;
  let db: ProductionDb;
  let service: MediaControlService;
  let channel: string;

  beforeEach(async () => {
    redis = new FakeRedis();
    db = new ProductionDb();
    service = new MediaControlService({
      redis, transaction: db.transaction, assignments: [],
      signer: new Ed25519ControlSigner("test", generateKeyPairSync("ed25519").privateKey),
      handles: new StubCredentialHandleProvider(),
      now: () => Date.parse("2026-10-07T12:00:00.000Z"), commandTtlMs: 30_000,
      capacity: { workerUnits: { [WORKER]: 25 } },
    });
    channel = randomUUID();
    await service.advertiseWorkerCapacity(WORKER, 25);
  });

  const desired = (channelId: string, status: "RUNNING" | "STOPPED") =>
    service.director(WS, channelId, { type: "desired", spec: status === "RUNNING" ? running() : stopped },
      `${channelId}:${status}`, ACTOR);

  it("a start over quota queues as PENDING_CAPACITY with nothing persisted", async () => {
    const placed = [channel, ...Array.from({ length: 3 }, () => randomUUID())];
    for (const channelId of placed) { await service.assign(WS, channelId, WORKER, 1); await expect(desired(channelId, "RUNNING")).resolves.toBeTruthy(); }
    const fifth = randomUUID();
    await service.assign(WS, fifth, WORKER, 1);
    const error = await catchError(desired(fifth, "RUNNING"));
    expect(codeOf(error)).toBe("PENDING_CAPACITY");
    expect(db.state.version).toBe(4); // four committed writes; the fifth rolled back
    expect(db.state.receipts).toHaveLength(4);
    expect(db.applied.filter(a => a.ch === fifth)).toEqual([]);
    expect(await service.channelCapacity(WS, fifth)).toMatchObject({ state: "PENDING_CAPACITY" });
  });

  it("a stop frees capacity and the queue re-applies exactly one admitted start", async () => {
    const placed = [channel, ...Array.from({ length: 3 }, () => randomUUID())];
    for (const channelId of placed) { await service.assign(WS, channelId, WORKER, 1); await desired(channelId, "RUNNING"); }
    const waiting: string[] = [];
    for (let i = 0; i < 2; i += 1) {
      const channelId = randomUUID();
      waiting.push(channelId);
      await service.assign(WS, channelId, WORKER, 1);
      await expect(desired(channelId, "RUNNING")).rejects.toThrow("PENDING_CAPACITY");
    }

    await desired(placed[0]!, "STOPPED");
    const capacity = await service.capacityGate.snapshot(WS, [...placed, ...waiting]);
    expect(capacity.placedChannels).toBe(DEFAULT_CHANNEL_QUOTA); // never more than the quota
    expect(capacity.pendingChannels).toBe(1);
    const waitingIds: readonly string[] = waiting;
    const admitted = capacity.channels.find(c => c.state === "PLACED" && waitingIds.includes(c.channelId));
    expect(admitted).toBeDefined();
    expect(JSON.parse(redis.data.get(K.desired(WS, admitted!.channelId)) ?? "null")).toMatchObject({ desiredStatus: "RUNNING" });
  });

  it("a mid-transaction failure after admission leaks nothing", async () => {
    await service.assign(WS, channel, WORKER, 1);
    db.failAfter = 1;
    await expect(desired(channel, "RUNNING")).rejects.toThrow("commit refused");
    expect(await service.channelCapacity(WS, channel)).toBeNull();
    expect((await service.capacityGate.snapshot(WS, [channel])).placedChannels).toBe(0);
    expect(db.state.receipts).toHaveLength(0);
  });
});

describe("error shape", () => {
  it("PENDING_CAPACITY is 202, CAPACITY_QUEUE_FULL is 429, both carry their code", () => {
    const pending = new ChannelPendingCapacityError({ workspaceId: WS, channelId: CH, workerId: WORKER, units: 4,
      profile: "720p30", requestedAt: "now", position: 1, pendingChannels: 1, pendingBound: 8 });
    expect(pending).toBeInstanceOf(HttpException);
    expect(pending.getStatus()).toBe(202);
    expect((pending.getResponse() as Record<string, unknown>).code).toBe("PENDING_CAPACITY");

    const full = new CapacityQueueFullError({ workspaceId: WS, channelId: CH, workerId: WORKER, pendingChannels: 8, pendingBound: 8 });
    expect(full).toBeInstanceOf(HttpException);
    expect(full.getStatus()).toBe(429);
    expect((full.getResponse() as Record<string, unknown>).code).toBe("CAPACITY_QUEUE_FULL");
  });
});