diff --git a/.fkst/workflows/cron-acceptance.toml b/.fkst/workflows/cron-acceptance.toml
new file mode 100644
index 000000000..dc0ac6f5b
--- /dev/null
+++ b/.fkst/workflows/cron-acceptance.toml
@@ -0,0 +1,69 @@
+description = "Milestone-24 acceptance: prove the scheduled-workflow machinery end to end."
+
+# The acceptance instance for #5846.
+#
+# It carries the SHAPE the issue specifies -- deterministic step, agentic step,
+# deterministic step that commits a cross-run ledger -- while deliberately
+# carrying NONE of the concrete instance. #5846 draws a content boundary: search
+# parameters, judgment criteria, destination-table identifiers, and
+# credential-broker service names are operator-supplied and must not enter this
+# repository. So the external API is replaced by locally generated candidates and
+# the destination table by a committed file.
+#
+# What that still proves is the bounded milestone: a due slot creating a run
+# issue, an idle session waking on it, `run` and `task` steps executing in
+# declared order, arguments substituted as data, a previously seeded cross-run
+# ledger suppressing repeats, at least one accepted row being committed, and one
+# fkst-cron-run:v1 record travelling back to release the schedule. The workflow
+# fails closed when the required prior-ledger suppression is not observable.
+#
+# What it does NOT prove is the operator's real workload -- live API pagination,
+# rate-limit backoff, and credential-broker delivery. Those need the concrete
+# instance, and this definition is the slot it drops into.
+
+[[step]]
+id = "collect"
+kind = "run"
+command = [
+  "python3",
+  ".fkst/workflows/cron-acceptance/collect.py",
+  "--topic",
+  "{{ topic }}",
+  "--count",
+  "{{ count }}",
+]
+timeout_secs = 300
+
+[[step]]
+id = "score"
+kind = "task"
+prompt = """
+Read `candidates.json` in the repository root. It is a JSON array of objects,
+each with an `id` and a `title`.
+
+Score every entry from 0 to 10 for how well its title matches this criterion:
+{{ criterion }}
+
+Write `scored.json` to the repository root: a JSON array, one object per input
+entry, each with exactly these keys:
+
+  id        the entry's id, unchanged
+  score     an integer 0-10
+  rationale one sentence, under 120 characters
+
+Emit the JSON array as the entire file content. Do not wrap it in a code fence
+and do not add commentary before or after it. Every input entry must appear
+exactly once in the output.
+"""
+timeout_secs = 900
+
+[[step]]
+id = "publish"
+kind = "run"
+command = [
+  "python3",
+  ".fkst/workflows/cron-acceptance/publish.py",
+  "--min-score",
+  "{{ min_score }}",
+]
+timeout_secs = 300
diff --git a/.fkst/workflows/cron-acceptance/collect.py b/.fkst/workflows/cron-acceptance/collect.py
new file mode 100644
index 000000000..da542589e
--- /dev/null
+++ b/.fkst/workflows/cron-acceptance/collect.py
@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Step 1 of the milestone-24 acceptance workflow: collect candidates.
+
+Deterministic. Emits `candidates.json` for the agentic scoring step, having
+first dropped every id the cross-run ledger already records as published.
+
+The ledger read is what makes a run after the seeded first run provably
+different, which is the property #5846 asks this step to demonstrate. The
+window slides by one id per run and overlaps the previous run by one. Acceptance
+therefore requires observable prior state: at least one id must be suppressed
+and at least one genuinely new id must be carried forward. A first run seeds the
+ledger but cannot, by itself, prove cross-run suppression.
+
+Candidates are generated locally rather than fetched. #5846 draws a content
+boundary around the concrete instance -- its API, search terms, and credentials
+are operator-supplied and must not enter this repository -- so this step proves
+the machinery around the fetch, and the operator swaps the fetch in.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import pathlib
+import re
+import sys
+
+import ledger
+
+LOG = logging.getLogger("collect")
+
+CANDIDATES = pathlib.Path("candidates.json")
+
+# One overlapping id per run: enough to prove the ledger suppressed something,
+# without suppressing so much that a run has nothing left to score.
+OVERLAP = 1
+
+# A slug keeps generated ids inside the character set the run record's `steps`
+# attribute can carry, and keeps a hostile `--topic` from reaching a filename.
+SLUG = re.compile(r"[^a-z0-9-]+")
+
+
+def slugify(topic: str) -> str:
+    """A filesystem- and marker-safe form of an operator-supplied topic."""
+    slug = SLUG.sub("-", topic.strip().lower()).strip("-")
+    if not slug:
+        raise ValueError(f"topic {topic!r} contains no usable characters")
+    return slug[:40]
+
+
+def build_window(slug: str, published: list[str], count: int) -> list[dict[str, str]]:
+    """The ids this run considers, overlapping the previous run by OVERLAP."""
+    start = max(0, len(published) - OVERLAP)
+    return [
+        {"id": f"{slug}-{index:04d}", "title": f"{slug} candidate {index}"}
+        for index in range(start, start + count)
+    ]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--topic", required=True, help="operator-supplied subject")
+    parser.add_argument("--count", required=True, help="window size for this run")
+    args = parser.parse_args()
+
+    try:
+        count = int(args.count)
+    except ValueError:
+        LOG.error("--count must be an integer, got %r", args.count)
+        return 2
+    if not 1 <= count <= 100:
+        LOG.error("--count must be between 1 and 100, got %d", count)
+        return 2
+
+    try:
+        slug = slugify(args.topic)
+    except ValueError as error:
+        LOG.error("%s", error)
+        return 2
+
+    try:
+        recorded, _ = ledger.read()
+    except ledger.LedgerError as error:
+        LOG.error("%s", error)
+        return 1
+    published = set(recorded)
+    window = build_window(slug, sorted(published), count)
+    fresh = [entry for entry in window if entry["id"] not in published]
+    suppressed = len(window) - len(fresh)
+
+    if suppressed < OVERLAP:
+        LOG.error(
+            "acceptance requires a seeded ledger that suppresses at least %d candidate(s); observed %d",
+            OVERLAP,
+            suppressed,
+        )
+        return 1
+    if not fresh:
+        LOG.error("acceptance requires at least one new candidate after ledger suppression")
+        return 1
+
+    CANDIDATES.write_text(json.dumps(fresh, indent=2) + "\n", encoding="utf-8")
+    LOG.info(
+        "topic=%s window=%d suppressed_by_ledger=%d carried_forward=%d -> %s",
+        slug,
+        len(window),
+        suppressed,
+        len(fresh),
+        CANDIDATES,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
+    sys.exit(main())
diff --git a/.fkst/workflows/cron-acceptance/ledger.py b/.fkst/workflows/cron-acceptance/ledger.py
new file mode 100644
index 000000000..a9a57cf14
--- /dev/null
+++ b/.fkst/workflows/cron-acceptance/ledger.py
@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""The cross-run ledger for the milestone-24 acceptance workflow.
+
+The ledger is the ONLY state that survives between runs, and #5846 requires it
+to be a committed file in the repository rather than a control-plane store --
+that is what keeps the scheduled-workflow capability stateless.
+
+It lives on its own branch, `cron-acceptance-ledger`, NOT on `develop`. Each run
+works in a fresh clone of the default branch, so a ledger committed to `develop`
+would need a pull request per run; `develop` is protected and the repository's
+own rules forbid pushing to it directly. A dedicated branch is writable, carries
+exactly two files, and keeps run bookkeeping out of the source history.
+
+Writes go through git plumbing (hash-object / mktree / commit-tree) rather than
+add+commit, so the run never switches the working tree it is executing from.
+
+The commit is IDEMPOTENT: if the resulting tree matches the branch tip's tree,
+nothing is committed. Re-running the same slot therefore cannot double-publish,
+which is the property #5846 asks for by name.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import subprocess
+
+LOG = logging.getLogger("ledger")
+
+BRANCH = "cron-acceptance-ledger"
+LEDGER_FILE = "ledger.json"
+PUBLISHED_FILE = "published.json"
+
+# Bound the ledger so a long-lived schedule cannot grow one file without limit.
+# The window only ever consults the tail, so older ids are safe to drop.
+MAX_IDS = 200
+
+
+class LedgerError(RuntimeError):
+    """A ledger operation failed in a way the run must not paper over."""
+
+
+def _git(*args: str, check: bool = True, stdin: str | None = None) -> str:
+    """Run one git command, returning stdout.
+
+    Never `shell=True`: arguments reach git as argv, so an operator-supplied
+    value cannot become shell syntax.
+    """
+    result = subprocess.run(
+        ["git", *args],
+        input=stdin,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if check and result.returncode != 0:
+        raise LedgerError(
+            f"git {' '.join(args)} failed ({result.returncode}): {result.stderr.strip()}"
+        )
+    return result.stdout
+
+
+def _tip() -> str | None:
+    """The ledger branch's current commit, or None when it does not exist yet."""
+    _git("fetch", "--quiet", "origin", BRANCH, check=False)
+    revision = _git("rev-parse", "--verify", "--quiet", "FETCH_HEAD", check=False).strip()
+    return revision or None
+
+
+def read() -> tuple[list[str], dict[str, dict]]:
+    """The published ids and the destination table, as of the branch tip.
+
+    A missing branch means the first run and yields empty state. A branch that
+    exists but whose contents will not parse is a HARD failure: treating
+    corruption as "nothing published yet" would re-publish the entire history,
+    the exact outcome the ledger exists to prevent.
+    """
+    tip = _tip()
+    if tip is None:
+        LOG.info("no %s branch yet; treating this as the first run", BRANCH)
+        return [], {}
+
+    ids = _read_json(tip, LEDGER_FILE, default=[])
+    if not isinstance(ids, list) or not all(isinstance(entry, str) for entry in ids):
+        raise LedgerError(f"{LEDGER_FILE} on {BRANCH} must be a JSON array of strings")
+
+    published = _read_json(tip, PUBLISHED_FILE, default={})
+    if not isinstance(published, dict):
+        raise LedgerError(f"{PUBLISHED_FILE} on {BRANCH} must be a JSON object keyed by id")
+
+    LOG.info("ledger at %s carries %d published id(s)", tip[:8], len(ids))
+    return ids, published
+
+
+def _read_json(tip: str, name: str, default):
+    """One file's decoded content from a commit, or `default` when absent."""
+    raw = _git("show", f"{tip}:{name}", check=False)
+    if not raw.strip():
+        return default
+    try:
+        return json.loads(raw)
+    except json.JSONDecodeError as error:
+        raise LedgerError(f"{name} on {BRANCH} is not valid JSON: {error}") from error
+
+
+def write(ids: list[str], published: dict[str, dict], message: str) -> bool:
+    """Commit the ledger and destination table. Returns False when unchanged."""
+    trimmed = ids[-MAX_IDS:]
+    entries = {
+        LEDGER_FILE: json.dumps(trimmed, indent=2) + "\n",
+        PUBLISHED_FILE: json.dumps(published, indent=2, sort_keys=True) + "\n",
+    }
+
+    lines = []
+    for name, content in sorted(entries.items()):
+        blob = _git("hash-object", "-w", "--stdin", stdin=content).strip()
+        lines.append(f"100644 blob {blob}\t{name}")
+    tree = _git("mktree", stdin="\n".join(lines) + "\n").strip()
+
+    tip = _tip()
+    if tip is not None and _git("rev-parse", f"{tip}^{{tree}}").strip() == tree:
+        LOG.info("ledger tree is unchanged; nothing to commit (idempotent re-run)")
+        return False
+
+    parents = ["-p", tip] if tip else []
+    commit = _git("commit-tree", tree, *parents, "-m", message).strip()
+    _git("push", "--quiet", "origin", f"{commit}:refs/heads/{BRANCH}")
+    LOG.info("committed ledger %s to %s (%d id(s) retained)", commit[:8], BRANCH, len(trimmed))
+    return True
diff --git a/.fkst/workflows/cron-acceptance/publish.py b/.fkst/workflows/cron-acceptance/publish.py
new file mode 100644
index 000000000..55da0bf83
--- /dev/null
+++ b/.fkst/workflows/cron-acceptance/publish.py
@@ -0,0 +1,230 @@
+#!/usr/bin/env python3
+"""Step 3 of the milestone-24 acceptance workflow: publish and record.
+
+Filters the scored entries by `--min-score`, writes the accepted ones into the
+destination table, appends their ids to the cross-run ledger, and commits both
+to the ledger branch.
+
+Two properties #5846 asks for by name:
+
+  * Keyed by entry id, so a partial failure of this step is safely retryable --
+    re-publishing an id overwrites its row rather than appending a duplicate.
+  * Idempotent commit, so re-running the same slot cannot double-publish. That
+    lives in `ledger.write`, which skips a commit whose tree is unchanged.
+
+The scored file is parsed DEFENSIVELY. The agentic step is a language model
+told to emit bare JSON, and a model that wraps its answer in a fence or adds a
+sentence of preamble is producing the right answer in the wrong envelope. So the
+envelope is stripped -- but only the envelope. Anything still unparseable,
+incomplete, duplicated, malformed, or empty fails the step LOUDLY. Acceptance
+also requires at least one published row and a new ledger commit, so a run cannot
+report success without observable publication evidence.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import pathlib
+import re
+import sys
+
+import ledger
+
+LOG = logging.getLogger("publish")
+
+CANDIDATES = pathlib.Path("candidates.json")
+SCORED = pathlib.Path("scored.json")
+
+# ```json ... ``` or ``` ... ```, the two envelopes a model actually produces.
+FENCE = re.compile(r"```(?:json)?\s*(.*?)\s*```", re.DOTALL)
+SCORE_KEYS = {"id", "score", "rationale"}
+
+
+def parse_scored(raw: str) -> list[dict]:
+    """Decode the agentic step's output, tolerating fences and prose."""
+    for candidate in _candidate_payloads(raw):
+        try:
+            decoded = json.loads(candidate)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(decoded, list):
+            return decoded
+        # A model that emitted {"entries": [...]} answered correctly in a
+        # different shape; accept the single obvious array it contains.
+        if isinstance(decoded, dict):
+            arrays = [value for value in decoded.values() if isinstance(value, list)]
+            if len(arrays) == 1:
+                return arrays[0]
+    raise SystemExit(
+        f"{SCORED} did not contain a JSON array of scores. Refusing to treat an "
+        f"unreadable payload as an empty one."
+    )
+
+
+def _candidate_payloads(raw: str):
+    """The substrings worth attempting, most-literal first."""
+    yield raw
+    for match in FENCE.finditer(raw):
+        yield match.group(1)
+    # Prose on either side of a bare array.
+    start, end = raw.find("["), raw.rfind("]")
+    if start != -1 and end > start:
+        yield raw[start : end + 1]
+
+
+def load_entries() -> tuple[dict[str, dict], list[dict]]:
+    """The candidates this run collected and the scores the model returned."""
+    if not CANDIDATES.exists():
+        raise SystemExit(f"{CANDIDATES} is missing; step 1 did not produce it")
+    if not SCORED.exists():
+        raise SystemExit(f"{SCORED} is missing; step 2 did not produce it")
+    try:
+        candidates = json.loads(CANDIDATES.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as error:
+        raise SystemExit(f"{CANDIDATES} is unreadable: {error}") from error
+    if not isinstance(candidates, list):
+        raise SystemExit(f"{CANDIDATES} must be a JSON array")
+    if not candidates:
+        raise SystemExit(f"{CANDIDATES} must contain at least one candidate")
+
+    by_id: dict[str, dict] = {}
+    for index, entry in enumerate(candidates):
+        if not isinstance(entry, dict):
+            raise SystemExit(f"{CANDIDATES} entry {index} must be a JSON object")
+        entry_id = entry.get("id")
+        if not isinstance(entry_id, str) or not entry_id:
+            raise SystemExit(
+                f"{CANDIDATES} entry {index} must have a non-empty string id"
+            )
+        if not isinstance(entry.get("title"), str):
+            raise SystemExit(
+                f"{CANDIDATES} entry {entry_id!r} must have a string title"
+            )
+        if entry_id in by_id:
+            raise SystemExit(f"{CANDIDATES} contains duplicate id {entry_id!r}")
+        by_id[entry_id] = entry
+
+    scores = parse_scored(SCORED.read_text(encoding="utf-8"))
+    validate_scores(by_id, scores)
+    return by_id, scores
+
+
+def validate_scores(by_id: dict[str, dict], scores: list[dict]) -> None:
+    """Require one well-formed integer score for every collected candidate."""
+    if not scores:
+        raise SystemExit(f"{SCORED} must contain one score for every candidate")
+    if len(scores) != len(by_id):
+        raise SystemExit(
+            f"{SCORED} contains {len(scores)} score(s) for {len(by_id)} candidate(s)"
+        )
+
+    seen: set[str] = set()
+    for index, scored in enumerate(scores):
+        if not isinstance(scored, dict):
+            raise SystemExit(f"{SCORED} entry {index} must be a JSON object")
+        if set(scored) != SCORE_KEYS:
+            raise SystemExit(
+                f"{SCORED} entry {index} must contain exactly id, score, and rationale"
+            )
+
+        entry_id = scored["id"]
+        if not isinstance(entry_id, str) or entry_id not in by_id:
+            raise SystemExit(f"{SCORED} entry {index} has an unknown id")
+        if entry_id in seen:
+            raise SystemExit(f"{SCORED} contains duplicate id {entry_id!r}")
+        seen.add(entry_id)
+
+        score = scored["score"]
+        if isinstance(score, bool) or not isinstance(score, int) or not 0 <= score <= 10:
+            raise SystemExit(
+                f"{SCORED} entry {entry_id!r} must have an integer score from 0 to 10"
+            )
+        rationale = scored["rationale"]
+        if not isinstance(rationale, str) or not rationale.strip() or len(rationale) > 120:
+            raise SystemExit(
+                f"{SCORED} entry {entry_id!r} must have a non-empty rationale "
+                "of at most 120 characters"
+            )
+
+    if set(by_id) != seen:
+        raise SystemExit(f"{SCORED} is missing score(s) for collected candidate ids")
+
+
+def accepted_rows(
+    by_id: dict[str, dict], scores: list[dict], minimum: int
+) -> dict[str, dict]:
+    """The validated rows clearing `minimum`."""
+    rows: dict[str, dict] = {}
+    for scored in scores:
+        entry_id = scored["id"]
+        raw_score = scored["score"]
+        if raw_score < minimum:
+            continue
+        rows[entry_id] = {
+            "id": entry_id,
+            "title": by_id[entry_id]["title"],
+            "score": raw_score,
+            "rationale": scored["rationale"],
+        }
+    return rows
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--min-score", required=True, help="inclusive acceptance threshold")
+    args = parser.parse_args()
+
+    try:
+        minimum = int(args.min_score)
+    except ValueError:
+        LOG.error("--min-score must be an integer, got %r", args.min_score)
+        return 2
+    if not 0 <= minimum <= 10:
+        LOG.error("--min-score must be between 0 and 10, got %d", minimum)
+        return 2
+
+    by_id, scores = load_entries()
+    rows = accepted_rows(by_id, scores, minimum)
+    if not rows:
+        LOG.error(
+            "acceptance requires at least one scored candidate at or above min_score=%d",
+            minimum,
+        )
+        return 1
+
+    try:
+        recorded, published = ledger.read()
+        published.update(rows)
+        # Order-preserving append: previously published ids keep their position,
+        # so the sliding window in step 1 stays stable across runs.
+        for entry_id in sorted(rows):
+            if entry_id not in recorded:
+                recorded.append(entry_id)
+        changed = ledger.write(
+            recorded,
+            published,
+            f"chore(cron-acceptance): publish {len(rows)} row(s), ledger at {len(recorded)}",
+        )
+    except ledger.LedgerError as error:
+        LOG.error("%s", error)
+        return 1
+    if not changed:
+        LOG.error("acceptance requires a new ledger commit proving publication")
+        return 1
+
+    LOG.info(
+        "min_score=%d accepted=%d table_rows=%d ledger=%d committed=%s",
+        minimum,
+        len(rows),
+        len(published),
+        len(recorded),
+        changed,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
+    sys.exit(main())
diff --git a/.fkst/workflows/cron-acceptance/test_acceptance.py b/.fkst/workflows/cron-acceptance/test_acceptance.py
new file mode 100644
index 000000000..9e33564a2
--- /dev/null
+++ b/.fkst/workflows/cron-acceptance/test_acceptance.py
@@ -0,0 +1,160 @@
+#!/usr/bin/env python3
+"""Fail-closed oracle tests for the cron acceptance workflow."""
+
+from __future__ import annotations
+
+import json
+import pathlib
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+
+import collect
+import publish
+
+
+class CollectTests(unittest.TestCase):
+    def run_collect(self, recorded: list[str]) -> tuple[int, list[dict] | None]:
+        with tempfile.TemporaryDirectory() as directory:
+            candidates_path = pathlib.Path(directory) / "candidates.json"
+            with mock.patch.object(collect.ledger, "read", return_value=(recorded, {})):
+                with mock.patch.object(
+                    sys,
+                    "argv",
+                    ["collect.py", "--topic", "AI Tools", "--count", "3"],
+                ):
+                    with mock.patch.object(collect, "CANDIDATES", candidates_path):
+                        result = collect.main()
+                        candidates = (
+                            json.loads(candidates_path.read_text(encoding="utf-8"))
+                            if candidates_path.exists()
+                            else None
+                        )
+        return result, candidates
+
+    def test_requires_observable_prior_ledger_suppression(self) -> None:
+        result, candidates = self.run_collect([])
+
+        self.assertEqual(result, 1)
+        self.assertIsNone(candidates)
+
+    def test_emits_only_fresh_candidates_after_suppression(self) -> None:
+        result, candidates = self.run_collect(["ai-tools-0000"])
+
+        self.assertEqual(result, 0)
+        self.assertEqual(
+            candidates,
+            [
+                {"id": "ai-tools-0001", "title": "ai-tools candidate 1"},
+                {"id": "ai-tools-0002", "title": "ai-tools candidate 2"},
+            ],
+        )
+
+
+class PublishValidationTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.candidates = {
+            "candidate-1": {"id": "candidate-1", "title": "Candidate one"},
+            "candidate-2": {"id": "candidate-2", "title": "Candidate two"},
+        }
+        self.valid_scores = [
+            {"id": "candidate-1", "score": 8, "rationale": "Strong fit."},
+            {"id": "candidate-2", "score": 4, "rationale": "Weak fit."},
+        ]
+
+    def test_accepts_complete_unique_integer_scores(self) -> None:
+        publish.validate_scores(self.candidates, self.valid_scores)
+
+    def test_rejects_empty_incomplete_duplicate_and_malformed_scores(self) -> None:
+        cases = [
+            [],
+            self.valid_scores[:1],
+            [self.valid_scores[0], self.valid_scores[0]],
+            [
+                {"id": "candidate-1", "score": 8.0, "rationale": "Not an integer."},
+                self.valid_scores[1],
+            ],
+            [
+                {
+                    "id": "candidate-1",
+                    "score": 8,
+                    "rationale": "Strong fit.",
+                    "extra": True,
+                },
+                self.valid_scores[1],
+            ],
+        ]
+
+        for scores in cases:
+            with self.subTest(scores=scores):
+                with self.assertRaises(SystemExit):
+                    publish.validate_scores(self.candidates, scores)
+
+
+class PublishMainTests(unittest.TestCase):
+    def run_publish(
+        self, scores: list[dict], *, changed: bool = True
+    ) -> tuple[int, mock.Mock]:
+        candidates = [
+            {"id": "candidate-1", "title": "Candidate one"},
+            {"id": "candidate-2", "title": "Candidate two"},
+        ]
+        with tempfile.TemporaryDirectory() as directory:
+            root = pathlib.Path(directory)
+            candidates_path = root / "candidates.json"
+            scores_path = root / "scored.json"
+            candidates_path.write_text(json.dumps(candidates), encoding="utf-8")
+            scores_path.write_text(json.dumps(scores), encoding="utf-8")
+            ledger_write = mock.Mock(return_value=changed)
+            with mock.patch.object(publish, "CANDIDATES", candidates_path):
+                with mock.patch.object(publish, "SCORED", scores_path):
+                    with mock.patch.object(publish.ledger, "read", return_value=([], {})):
+                        with mock.patch.object(publish.ledger, "write", ledger_write):
+                            with mock.patch.object(
+                                sys,
+                                "argv",
+                                ["publish.py", "--min-score", "5"],
+                            ):
+                                result = publish.main()
+        return result, ledger_write
+
+    def test_requires_at_least_one_published_row(self) -> None:
+        result, ledger_write = self.run_publish(
+            [
+                {"id": "candidate-1", "score": 4, "rationale": "Below threshold."},
+                {"id": "candidate-2", "score": 3, "rationale": "Below threshold."},
+            ]
+        )
+
+        self.assertEqual(result, 1)
+        ledger_write.assert_not_called()
+
+    def test_requires_a_ledger_mutation(self) -> None:
+        result, ledger_write = self.run_publish(
+            [
+                {"id": "candidate-1", "score": 8, "rationale": "Strong fit."},
+                {"id": "candidate-2", "score": 4, "rationale": "Weak fit."},
+            ],
+            changed=False,
+        )
+
+        self.assertEqual(result, 1)
+        ledger_write.assert_called_once()
+
+    def test_succeeds_with_valid_publication_and_commit_evidence(self) -> None:
+        result, ledger_write = self.run_publish(
+            [
+                {"id": "candidate-1", "score": 8, "rationale": "Strong fit."},
+                {"id": "candidate-2", "score": 4, "rationale": "Weak fit."},
+            ]
+        )
+
+        self.assertEqual(result, 0)
+        ledger_write.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()