import { test } from "node:test";
import assert from "node:assert/strict";
import { mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import {
  computeLeafPending,
  focusHoldLine,
  laneDispatchRoot,
  makeFocusHoldReporter,
  pickMetadataScanChannel,
  priorityContextFor,
  resetPriorityContextForTest,
  sharedAutoQueueState,
  isChannelHeldForLane,
} from "./autoRunner";
import {
  defaultChannelPriority,
  isDefaultChannelPriority,
  sanitizeChannelPriority,
  type ChannelPriority,
  type FocusSummary,
} from "../lib/channelPriority";
import { LANES } from "../lib/autoQueueTypes";
import {
  emptyAutoQueueState,
  readAutoQueueState,
  writeAutoQueueState,
} from "../jobs/autoQueueState";
import type {
  AutoQueueGroup,
  AutoQueuePolicy,
  ChannelWork,
} from "../jobs/autoQueuePolicy";
import type { Paths } from "../lib/paths";
import type { SiteSettings } from "../lib/settings";
import { forgetChannelMedia, inspectChannelMedia } from "../lib/channelMedia";
import { bucketLaneOperationId } from "../lib/operations";

// THE RUNNER'S HALF OF S1 (plans/channel-priority.md), and it is here rather
// than in `jobs/channelPriorityCompile.test.ts` for one reason:
// `../architecture.test.ts` forbids `jobs/ -> controller/`, including from a
// test file, and its ALLOWED list may only shrink. The engine-level proof — a
// focus holding and releasing a real `buildPendingByLeaf` + `selectNextWork`
// pair — lives there; what lives here is the thing that can only be said about
// the runner: it writes ONE line per transition.
//
// WHY A LINE AND NOT AN IDLE REASON. A lane whose focus group holds the rest is
// not idle, it is dispatching focus work, so `AutoRunnerIdleReason` gains
// nothing and `no-pending` stays the true answer for an empty tree. The log
// line is what makes "why is only jeralyzer moving?" answerable from the job
// log, and the banner (S4) is what makes it answerable from the page.

function summary(over: Partial<FocusSummary> = {}): FocusSummary {
  return {
    kind: "channels",
    siteId: null,
    slugs: ["slow-a"],
    channelCount: 1,
    active: true,
    focusPending: 2,
    otherPending: 5,
    holding: true,
    heldChannels: 1,
    ...over,
  };
}

test("the focus-hold line fires once per state change, not once per tick", () => {
  const lines: string[] = [];
  const report = makeFocusHoldReporter("download", (line) => lines.push(line));

  // No focus at all: silence, however many ticks run.
  report(null);
  report(null);
  assert.deepEqual(lines, []);

  // Entering the hold: one line.
  report(summary());
  report(summary());
  report(summary());
  assert.equal(lines.length, 1);
  assert.match(lines[0], /Auto-download: focus \(1 channel\) is holding/);

  // THE COUNTS MOVE ON EVERY GRANT AND MUST NOT RE-FIRE IT. The state is the
  // three-valued thing; the counts are in the message only, which is why the
  // reporter keys on the state and not on the line it printed.
  report(summary({ focusPending: 1 }));
  report(summary({ focusPending: 1, heldChannels: 2, otherPending: 9 }));
  assert.equal(lines.length, 1);

  // The focus runs out: the release line, once.
  report(summary({ focusPending: 0, holding: false }));
  report(summary({ focusPending: 0, holding: false }));
  assert.equal(lines.length, 2);
  assert.match(lines[1], /has no work left in this lane/);

  // New focus work retakes the lane: the hold line again.
  report(summary());
  assert.equal(lines.length, 3);
  assert.match(lines[2], /is holding/);

  // The focus ends (or resolves to nothing): back to silence, no line.
  report(null);
  report(null);
  assert.equal(lines.length, 3);
});

test("an inactive focus is the same state as no focus", () => {
  const lines: string[] = [];
  const report = makeFocusHoldReporter("digest", (line) => lines.push(line));
  // What a `{kind:"site"}` focus naming an unknown site resolves to: a summary
  // exists, but nothing is focused, so the lane is not holding for anyone.
  report(summary({ active: false, channelCount: 0, slugs: [], holding: false }));
  assert.deepEqual(lines, []);
  assert.equal(
    focusHoldLine("digest", summary({ active: false })),
    null,
  );
});

test("the line names the lane and pluralises the focus set", () => {
  assert.match(
    focusHoldLine("transcription", summary({ channelCount: 30 })) ?? "",
    /Auto-transcription: focus \(30 channels\) is holding this lane — 2 focus unit\(s\) pending, 1 channel\(s\) held\./,
  );
  assert.match(
    focusHoldLine("backfill", summary({ holding: false, otherPending: 7 })) ??
      "",
    /Auto-backfill: focus \(1 channel\) has no work left in this lane — 7 unit\(s\) released/,
  );
  assert.equal(focusHoldLine("download", null), null);
});

// --- The two functions S1 shipped untested (the S0/S1 review, finding 6) -----
//
// `laneDispatchRoot` and `priorityContextFor` are the whole of "compile, not
// consult" on the dispatch side, and between them they carry the one promise
// that lets this plan ship without a corpus-wide gate: an absent document
// changes NOTHING. S5 exported both — the first because the status payload has
// to ask the runner which tree it dispatches from rather than compiling a
// second one (editor/app/operations/channelPriorityView.ts), the second
// because its cache key, its TTL and its "only a site focus reads the sites
// directory" rule are three claims no pure function can be asked about.

function policyWith(root: AutoQueueGroup): AutoQueuePolicy {
  return {
    enabled: true,
    maxWorkers: null,
    replaceAutoSubs: false,
    order: "listed",
    snoozeUntil: null,
    root,
  };
}

const STORED_ROOT: AutoQueueGroup = {
  id: "download-root",
  mode: "strict",
  weight: 1,
  maxWorkers: null,
  children: [
    { id: "hand-made", match: { type: "channel", value: "slow-b" } },
    { id: "download-all", match: { type: "all" } },
  ],
};

test("a default model is the byte-identical bypass: the STORED root, by identity", () => {
  const policy = policyWith(STORED_ROOT);
  const model = defaultChannelPriority();
  assert.equal(isDefaultChannelPriority(model), true);
  for (const lane of LANES) {
    const root = laneDispatchRoot(lane, policy, { model, focusSlugs: [] }, [
      "slow-a",
      "slow-b",
    ]);
    // NOT deepEqual. The promise is that the compiler never runs, so the object
    // the runner dispatches from is the one `getSettings()` returned — a
    // structural copy would mean a compile happened and merely agreed.
    assert.equal(root, policy.root, `${lane} must bypass the compiler`);
  }
});

test("one channel entry is enough to leave the bypass, and the tree is compiled", () => {
  const policy = policyWith(STORED_ROOT);
  // A `low` tier says something even though nothing is focused: the whole
  // point of `isDefaultChannelPriority` is that it is the FULL document, not
  // the focus alone.
  const model = sanitizeChannelPriority({
    focus: { kind: "none" },
    channels: { "slow-b": { tier: "low" } },
  });
  assert.equal(isDefaultChannelPriority(model), false);
  const root = laneDispatchRoot("download", policy, { model, focusSlugs: [] }, [
    "slow-a",
    "slow-b",
  ]);
  assert.notEqual(root, policy.root);
  assert.deepEqual(
    root.children.map((c) => c.id),
    ["prio-normal", "prio-low", "prio-all"],
  );
  // And the hand-made leaf is simply not in it: the stored tree is not
  // consulted at all while a model exists.
  assert.equal(JSON.stringify(root).includes("hand-made"), false);
});

test("a focus alone leaves the bypass too", () => {
  const model = sanitizeChannelPriority({
    focus: { kind: "channels", slugs: ["slow-a"] },
    channels: {},
  });
  assert.equal(isDefaultChannelPriority(model), false);
});

// --- priorityContextFor: the cache, the TTL, and who reads the sites dir ----

function sitesFixture(): Paths {
  const dir = mkdtempSync(path.join(tmpdir(), "prio-ctx-"));
  const sitesDir = path.join(dir, "sites");
  mkdirSync(path.join(sitesDir, "testsite"), { recursive: true });
  writeFileSync(
    path.join(sitesDir, "testsite", "site.json"),
    JSON.stringify({
      siteTitle: "Test site",
      channels: [{ slug: "slow-a" }, { slug: "slow-b" }],
    }),
  );
  return { sitesDir } as Paths;
}

function settingsWith(priority: ChannelPriority): SiteSettings {
  return { channelPriority: priority } as SiteSettings;
}

test("priorityContextFor resolves a site focus against the sites directory", () => {
  resetPriorityContextForTest();
  const paths = sitesFixture();
  const ctx = priorityContextFor(
    paths,
    settingsWith(
      sanitizeChannelPriority({
        focus: { kind: "site", siteId: "testsite" },
        channels: {},
      }),
    ),
  );
  assert.deepEqual(ctx.focusSlugs, ["slow-a", "slow-b"]);
});

test("only a site focus reads the sites directory", () => {
  resetPriorityContextForTest();
  // A path that does not exist: `listSites` would return [] rather than throw,
  // so the proof is the RESULT — a channel focus resolves in full from a
  // document alone, against a sites dir that could not have been read.
  const paths = { sitesDir: path.join(tmpdir(), "prio-ctx-absent") } as Paths;
  const ctx = priorityContextFor(
    paths,
    settingsWith(
      sanitizeChannelPriority({
        focus: { kind: "channels", slugs: ["slow-b", "slow-a"] },
        channels: {},
      }),
    ),
  );
  assert.deepEqual(ctx.focusSlugs, ["slow-b", "slow-a"]);
  // The same document with a SITE focus against the same absent directory
  // resolves to nothing — which is the designed answer for an unknown siteId
  // (a typo must not hold the whole corpus), and is what makes the line above
  // a statement about the read and not about the focus kind.
  resetPriorityContextForTest();
  assert.deepEqual(
    priorityContextFor(
      paths,
      settingsWith(
        sanitizeChannelPriority({
          focus: { kind: "site", siteId: "testsite" },
          channels: {},
        }),
      ),
    ).focusSlugs,
    [],
  );
});

test("the context is cached on the document, and a changed document re-resolves", () => {
  resetPriorityContextForTest();
  const paths = sitesFixture();
  const settings = settingsWith(
    sanitizeChannelPriority({
      focus: { kind: "site", siteId: "testsite" },
      channels: {},
    }),
  );
  const first = priorityContextFor(paths, settings);
  // Same document, same paths, inside the TTL: the SAME object, so the sites
  // directory was not read a second time. This is the claim that lets the
  // runner call it on every three-second tick.
  assert.equal(priorityContextFor(paths, settings), first);

  // A channel added to the focused site joins the focus WITHOUT a settings
  // write — that is why this is a TTL and not a pure memo. Rewrite the site,
  // expire the cache by hand, and the focus set moves.
  writeFileSync(
    path.join(paths.sitesDir, "testsite", "site.json"),
    JSON.stringify({
      siteTitle: "Test site",
      channels: [{ slug: "slow-a" }, { slug: "slow-b" }, { slug: "slow-c" }],
    }),
  );
  assert.deepEqual(priorityContextFor(paths, settings).focusSlugs, [
    "slow-a",
    "slow-b",
  ]);
  first.at = 0;
  assert.deepEqual(priorityContextFor(paths, settings).focusSlugs, [
    "slow-a",
    "slow-b",
    "slow-c",
  ]);

  // A DIFFERENT DOCUMENT is the other trigger, and it does not wait for the
  // TTL: the key is the document plus the sites dir.
  const ended = priorityContextFor(
    paths,
    settingsWith(defaultChannelPriority()),
  );
  assert.deepEqual(ended.focusSlugs, []);

  // And the paths are in the key, so two worktrees' runners in one process
  // cannot share a focus resolved against the other's sites directory.
  const other = sitesFixture();
  writeFileSync(
    path.join(other.sitesDir, "testsite", "site.json"),
    JSON.stringify({ siteTitle: "Other", channels: [{ slug: "only-here" }] }),
  );
  assert.deepEqual(priorityContextFor(paths, settings).focusSlugs, [
    "slow-a",
    "slow-b",
    "slow-c",
  ]);
  assert.deepEqual(priorityContextFor(other, settings).focusSlugs, [
    "only-here",
  ]);
});

// ---------------------------------------------------------------------------
// The download lane's metadata-scan pre-pick.
//
// It runs before every video pick on the lane, so its rule is the one thing
// standing between a filtered channel and 1,800 prefetch-reject-discard cycles
// against a source that is counting them.

function work(over: Partial<ChannelWork> & { slug: string }): ChannelWork {
  return { platform: "youtube", buckets: {}, ...over };
}

const noSkip = {
  scanned: new Set<string>(),
  platformSkip: new Set<string>(),
  platformOf: () => "youtube",
};

test("a filtered channel with an unscanned backlog is the pick", () => {
  const pick = pickMetadataScanChannel(
    [
      work({ slug: "plain", scanUnscanned: 500, filtered: false }),
      work({ slug: "filtered", scanUnscanned: 12, filtered: true }),
    ],
    noSkip,
  );
  assert.deepEqual(pick, { slug: "filtered", targets: 12 });
});

test("no filter, no scan — however big the backlog", () => {
  // The scan's whole output is a verdict the filter makes. Without a filter it
  // would fetch a title per listed video, decide nothing, and spend the
  // source's patience doing it.
  assert.equal(
    pickMetadataScanChannel(
      [work({ slug: "plain", scanUnscanned: 9999, filtered: false })],
      noSkip,
    ),
    null,
  );
});

test("an empty backlog is not work", () => {
  assert.equal(
    pickMetadataScanChannel(
      [work({ slug: "filtered", scanUnscanned: 0, filtered: true })],
      noSkip,
    ),
    null,
  );
  // And a projection with no scan fields at all — every lane but this one —
  // offers nothing rather than throwing or defaulting to "yes".
  assert.equal(
    pickMetadataScanChannel([work({ slug: "filtered" })], noSkip),
    null,
  );
});

test("once per runner: a channel already scanned this run is skipped", () => {
  // A scan that left the backlog above zero was stopped by something — a soft
  // block, a cooldown, a batch that ended early. Re-offering it three seconds
  // later would hammer the source that just refused us.
  assert.equal(
    pickMetadataScanChannel(
      [work({ slug: "filtered", scanUnscanned: 40, filtered: true })],
      { ...noSkip, scanned: new Set(["filtered"]) },
    ),
    null,
  );
});

test("a busy or cooling platform offers no scan either", () => {
  // The same gate the downloads honour: the scan is one more thing asking that
  // source for titles, and the per-platform cap is one at a time.
  assert.equal(
    pickMetadataScanChannel(
      [work({ slug: "filtered", scanUnscanned: 40, filtered: true })],
      { ...noSkip, platformSkip: new Set(["youtube"]) },
    ),
    null,
  );
  // A DIFFERENT platform is unaffected — the gate is per source, not global.
  assert.deepEqual(
    pickMetadataScanChannel(
      [
        work({ slug: "yt", scanUnscanned: 40, filtered: true }),
        work({ slug: "od", scanUnscanned: 7, filtered: true, platform: "odysee" }),
      ],
      {
        scanned: new Set<string>(),
        platformSkip: new Set(["youtube"]),
        platformOf: (slug) => (slug === "od" ? "odysee" : "youtube"),
      },
    ),
    { slug: "od", targets: 7 },
  );
});

test("the first candidate in list order wins, because that order is the priority", () => {
  assert.deepEqual(
    pickMetadataScanChannel(
      [
        work({ slug: "first", scanUnscanned: 1, filtered: true }),
        work({ slug: "second", scanUnscanned: 900, filtered: true }),
      ],
      noSkip,
    ),
    { slug: "first", targets: 1 },
  );
});

test("a focus holds the scan the same way it holds every download pick", () => {
  // The compiled tree makes a focus hold downloads by strict descent, and the
  // pre-pick does not go through that tree. Without this, "only jeralyzer is
  // moving" is false the moment another channel has titles to read — on the
  // same network the focus is trying to have to itself.
  const channels = [
    work({ slug: "other", scanUnscanned: 50, filtered: true }),
    work({ slug: "focused", scanUnscanned: 4, filtered: true }),
  ];
  assert.deepEqual(
    pickMetadataScanChannel(channels, {
      ...noSkip,
      focus: { slugs: new Set(["focused"]), holding: true },
    }),
    { slug: "focused", targets: 4 },
  );
  // Focus exhausted — strict descent moves on, and so does the scan.
  assert.deepEqual(
    pickMetadataScanChannel(channels, {
      ...noSkip,
      focus: { slugs: new Set(["focused"]), holding: false },
    }),
    { slug: "other", targets: 50 },
  );
  // A holding focus with no scan work of its own offers nothing, rather than
  // falling through to the channel it is holding back.
  assert.equal(
    pickMetadataScanChannel(
      [work({ slug: "other", scanUnscanned: 50, filtered: true })],
      { ...noSkip, focus: { slugs: new Set(["focused"]), holding: true } },
    ),
    null,
  );
});


// --- computeLeafPending's `shared`: the four-lane poll reads once -----------
//
// /operations asks all four lanes for their pending work on a ~3 s poll, and
// each call used to list every channel's config off disk and re-parse the
// auto-queue state — both of which the caller had already read for the rest of
// the payload, and neither of which is per-lane.

function pendingFixture(): Paths {
  const dir = mkdtempSync(path.join(tmpdir(), "leaf-pending-"));
  const channelsDir = path.join(dir, "channels");
  // A channel that IS on disk. An injected listing has to be believed over it.
  mkdirSync(path.join(channelsDir, "on-disk"), { recursive: true });
  writeFileSync(
    path.join(channelsDir, "on-disk", "config.json"),
    JSON.stringify({ url: "https://www.youtube.com/@ondisk" }),
  );
  return {
    channelsDir,
    sitesDir: path.join(dir, "sites"),
    autoQueueStateFile: path.join(dir, "auto-queue.json"),
  } as Paths;
}

test("computeLeafPending takes its channel listing from `shared`", async () => {
  const paths = pendingFixture();
  const result = await computeLeafPending(LANES[0], paths, {
    configs: [],
    state: emptyAutoQueueState(),
  });
  // The leaves the compiled root defines, all at zero: an empty listing is no
  // channels, so there is no work to attribute and nothing to pick.
  assert.deepEqual(Object.values(result.counts), [0]);
  assert.deepEqual(Object.values(result.head), [[]]);
  assert.deepEqual(result.owner, {});
  assert.equal(result.nextUp, null);
});

test("an injected auto-queue state changes nothing about the answer", async () => {
  const paths = pendingFixture();
  // The state carries SWRR weights, which only matter once there is work to
  // pick between — so what `shared.state` must never do is change the result.
  // It saves a read; it is not an input to the numbers.
  const state = emptyAutoQueueState();
  state[LANES[0]].runtime.currentWeights = { "prio-all": 17 };
  const injected = await computeLeafPending(LANES[0], paths, {
    configs: [],
    state,
  });
  const read = await computeLeafPending(LANES[0], paths, { configs: [] });
  assert.deepEqual(injected.counts, read.counts);
  assert.deepEqual(injected.head, read.head);
  assert.deepEqual(injected.owner, read.owner);
  assert.equal(injected.nextUp, read.nextUp);

  // AND THE CALLER'S OBJECT IS NOT TOUCHED. selectNextWork advances fairness by
  // mutating currentWeights in place, and this runs on a 3-second poll: the
  // deep clone that stops merely HAVING the page open from skewing a
  // round-robin group now has to protect a state the caller still holds and
  // will hand to the next lane.
  assert.deepEqual(state[LANES[0]].runtime.currentWeights, { "prio-all": 17 });
});

// ONE STATE OBJECT PER PROCESS, EVEN AT BOOT (release 8, slice S).
//
// Boot starts all four lanes' runLoops without awaiting them, so they all ask
// `sharedAutoQueueState` inside one read's await gap. When only the resolved
// value was cached each got its own copy, and on 2026-09-25 a lane persisting
// its boot-time copy of the WHOLE file erased the download lane's 13:20 backoff
// (`fails: 13` back to 12) and its video deferral. These pin the cache to the
// in-flight read: same object, mutations visible before any write, and a write
// through either caller carries the other's change to disk.

function freshSingletonForTest(): void {
  globalThis.__yttAutoRunner__ = undefined;
  globalThis.__yttAutoQueueState__ = undefined;
}

function stateTempPaths(): Paths {
  const dir = mkdtempSync(path.join(tmpdir(), "shared-aq-state-"));
  mkdirSync(path.join(dir, ".auto-queue"), { recursive: true });
  const file = path.join(dir, ".auto-queue", "state.json");
  // The boot-time content from the incident: a YouTube cooldown at fails 12.
  const seeded = emptyAutoQueueState();
  seeded.download.platformBackoff.youtube = { fails: 12, until: 1_000 };
  writeFileSync(file, JSON.stringify(seeded));
  return { autoQueueStateFile: file } as Paths;
}

test("sharedAutoQueueState: concurrent callers (the boot fan-out) get ONE object", async () => {
  freshSingletonForTest();
  const paths = stateTempPaths();
  const [a, b, c, d] = await Promise.all(
    LANES.map(() => sharedAutoQueueState(paths)),
  );
  assert.equal(a, b);
  assert.equal(a, c);
  assert.equal(a, d);
  assert.equal(a.download.platformBackoff.youtube?.fails, 12);
  // And a caller after the read resolved gets the same one too.
  assert.equal(await sharedAutoQueueState(paths), a);
  freshSingletonForTest();
});

test("sharedAutoQueueState: a mutation through one caller is visible to the other before any write", async () => {
  freshSingletonForTest();
  const paths = stateTempPaths();
  const pa = sharedAutoQueueState(paths);
  const pb = sharedAutoQueueState(paths);
  const a = await pa;
  a.download.videoDeferrals["_60iFE_FBPQ"] = {
    until: 99_000,
    channelSlug: "some-channel",
  };
  a.download.platformBackoff.youtube = { fails: 13, until: 2_000 };
  const b = await pb;
  assert.equal(b.download.platformBackoff.youtube?.fails, 13);
  assert.ok(b.download.videoDeferrals["_60iFE_FBPQ"]);
  freshSingletonForTest();
});

test("sharedAutoQueueState: another lane's persist carries the download lane's backoff to disk", async () => {
  freshSingletonForTest();
  const paths = stateTempPaths();
  // Obtained concurrently, as boot does.
  const [downloadLane, otherLane] = await Promise.all([
    sharedAutoQueueState(paths),
    sharedAutoQueueState(paths),
  ]);
  // 13:20 — the download lane escalates the cooldown and persists.
  downloadLane.download.platformBackoff.youtube = { fails: 13, until: 2_000 };
  await writeAutoQueueState(paths, downloadLane);
  // 13:33:37 — a different lane persists ITS reference to the state.
  otherLane.digest.runtime.currentWeights = { someLeaf: 1 };
  await writeAutoQueueState(paths, otherLane);
  const onDisk = await readAutoQueueState(paths);
  assert.deepEqual(onDisk.download.platformBackoff.youtube, {
    fails: 13,
    until: 2_000,
  });
  assert.deepEqual(onDisk.digest.runtime.currentWeights, { someLeaf: 1 });
  freshSingletonForTest();
});

test("sharedAutoQueueState: a reset singleton reads afresh; a different state file is its own object", async () => {
  freshSingletonForTest();
  const p1 = stateTempPaths();
  const p2 = stateTempPaths();
  const first = await sharedAutoQueueState(p1);
  freshSingletonForTest();
  const afterReset = await sharedAutoQueueState(p1);
  assert.notEqual(afterReset, first);
  // A read for another file in flight at the same time does not hand either
  // caller the other file's object, and the later file owns the singleton.
  freshSingletonForTest();
  const [x, y] = await Promise.all([
    sharedAutoQueueState(p1),
    sharedAutoQueueState(p2),
  ]);
  assert.notEqual(x, y);
  assert.equal(await sharedAutoQueueState(p2), y);
  freshSingletonForTest();
});

// --- The media hold: every lane skips a channel whose move marker stands ----
//
// Release 16 slice RM. The lanes' one channel-level chokepoint is the
// projection computeLeafPending shares with the runner (buildChannelWork): a
// channel whose `.relocating.json` stands is held (lib/channelMediaHold.ts) and
// projects no work on any lane, and the hold lifts the moment the marker goes
// — the movers forget the inspect memo on every marker write and removal, as
// this test does by hand.

async function laneTotal(
  kind: (typeof LANES)[number],
  paths: Paths,
  configs: { slug: string; config: never }[],
): Promise<number> {
  const r = await computeLeafPending(kind, paths, {
    configs,
    state: emptyAutoQueueState(),
  });
  return Object.values(r.counts).reduce((a, b) => a + b, 0);
}

test("a channel with a move marker projects no work on any lane, and the hold lifts with it", async () => {
  const paths = pendingFixture();
  const slug = "moving";
  const channelDir = path.join(paths.channelsDir, slug);
  mkdirSync(path.join(channelDir, "data", "v1"), { recursive: true });
  const config = {
    url: "https://www.youtube.com/@moving",
    handling: "transcribe",
  };
  writeFileSync(path.join(channelDir, "config.json"), JSON.stringify(config));
  const opId = bucketLaneOperationId("transcription");
  writeFileSync(
    path.join(channelDir, "snapshot.json"),
    JSON.stringify({
      generatedAt: new Date().toISOString(),
      totals: { videos: 1, downloaded: 1, transcribed: 0 },
      buckets: { downloadedNoTranscript: ["v1"], noTranscript: ["v1"] },
      ...(opId ? { backfill: { [opId]: { ids: ["v1"] } } } : {}),
    }),
  );
  const configs = [{ slug, config: config as never }];
  forgetChannelMedia(slug);
  assert.ok(
    (await laneTotal("transcription", paths, configs)) > 0,
    "in place, the channel's video is the lane's work",
  );

  writeFileSync(
    path.join(channelDir, ".relocating.json"),
    JSON.stringify({
      target: "/mnt/platter/moving/data",
      direction: "out",
      startedAt: new Date().toISOString(),
      phase: "copy",
    }),
  );
  forgetChannelMedia(slug);
  for (const lane of LANES) {
    // (Release 17: the digest lane is held only by the text hold, and a media
    // move leaves the text readable — `isChannelHeldForLane` below. This
    // fixture has no digest work, so its total is 0 either way.)
    assert.equal(
      await laneTotal(lane, paths, configs),
      0,
      `the ${lane} lane skips a channel whose media is moving`,
    );
  }

  // The move completes (or is abandoned): the marker goes, the hold lifts.
  rmSync(path.join(channelDir, ".relocating.json"));
  forgetChannelMedia(slug);
  assert.ok((await laneTotal("transcription", paths, configs)) > 0);
});


// --- Release 17: the lanes by tier ------------------------------------------
//
// A stalled, unmounted or moving MEDIA tier holds the lanes that open the big
// files (transcription, download, backfill) and lets the digest lane run: a
// digest reads and writes text, which stays on the corpus disk. Only the text
// hold — a `legacy` channel, a tier migration in flight — keeps the digest lane
// off a channel. Decided from the inspector's real answers over fixtures.

test("release 17: an unmounted media drive holds transcription/download/backfill, lets digest run; legacy holds all four", async () => {
  const paths = pendingFixture();
  const slug = "tiered";
  const channelDir = path.join(paths.channelsDir, slug);
  mkdirSync(path.join(channelDir, "data", "v1"), { recursive: true });
  const target = path.join(path.dirname(paths.channelsDir), "platter", slug, "media");
  const writeCfg = (extra: Record<string, unknown>) =>
    writeFileSync(
      path.join(channelDir, "config.json"),
      JSON.stringify({ url: "https://www.youtube.com/@t", handling: "transcribe", ...extra }),
    );
  writeCfg({ mediaDir: target });
  symlinkSync(target, path.join(channelDir, "media")); // dangling: not mounted
  const unmounted = await inspectChannelMedia(paths, slug, undefined, { fresh: true });
  assert.equal(unmounted.status, "unreachable");
  const heldBy = (loc: Parameters<typeof isChannelHeldForLane>[1]) =>
    Object.fromEntries(LANES.map((lane) => [lane, isChannelHeldForLane(lane, loc)]));
  assert.deepEqual(heldBy(unmounted), {
    transcription: true,
    download: true,
    digest: false,
    backfill: true,
  });

  // A media move in flight: the same.
  writeFileSync(
    path.join(channelDir, ".relocating.json"),
    JSON.stringify({ target, direction: "out", startedAt: "", phase: "copy", scope: "media" }),
  );
  const moving = await inspectChannelMedia(paths, slug, undefined, { fresh: true });
  assert.equal(moving.status, "in-transition");
  assert.equal(heldBy(moving).digest, false);
  assert.equal(heldBy(moving).transcription, true);
  // A tier migration rebuilds data/ itself: the digest lane is held too.
  writeFileSync(
    path.join(channelDir, ".relocating.json"),
    JSON.stringify({ target, direction: "out", startedAt: "", phase: "copy", scope: "tier-migration" }),
  );
  const migrating = await inspectChannelMedia(paths, slug, undefined, { fresh: true });
  assert.equal(heldBy(migrating).digest, true);
  rmSync(path.join(channelDir, ".relocating.json"));

  // The retired layout holds every lane.
  writeCfg({ dataDir: path.join(path.dirname(paths.channelsDir), "platter", slug, "data") });
  const legacy = await inspectChannelMedia(paths, slug, undefined, { fresh: true });
  assert.equal(legacy.status, "legacy");
  assert.deepEqual(heldBy(legacy), {
    transcription: true,
    download: true,
    digest: true,
    backfill: true,
  });
});