diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 1c1a9879..a31c864f 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -24,6 +24,7 @@ jobs:
       runs-on: ubuntu-24.04-arm
       timeout-minutes: 30
       python-version: "3.12"
+      node-version: "24"
       enable-cache: true
       uv-version: 0.12.10
       sync-command: uv sync --all-packages --locked
@@ -31,9 +32,134 @@ jobs:
       test-command: bash scripts/ci/python-tests.sh unit
       test-results-path: |
         reports/python-unit.xml
-        reports/coverage/python-unit/coverage.db
+        reports/coverage/
       test-results-name: python-unit-arm64-${{ github.run_attempt }}
     name: Unit / Python lint and tests (ARM64)
+  coverage-report:
+    name: Report / Backend unit coverage PR comment
+    needs: python-unit
+    if: >-
+      always() && !cancelled() && github.event_name == 'pull_request' &&
+      github.event.pull_request.head.repo.full_name == github.repository &&
+      github.actor != 'dependabot[bot]'
+    runs-on: ubuntu-24.04
+    timeout-minutes: 5
+    permissions:
+      contents: read
+      pull-requests: write
+    steps:
+      # No checkout: downloaded reports are data, never executable code.
+      - uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        continue-on-error: true
+        with:
+          pattern: python-unit-arm64-*
+          path: report
+          merge-multiple: false
+      - uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0
+        env:
+          UNIT_RESULT: ${{ needs.python-unit.result }}
+        with:
+          script: |
+            const fs = require('fs');
+            const marker = '<!-- devfeed-backend-unit-coverage -->';
+            const pr = context.payload.pull_request;
+            const current = await github.rest.pulls.get({...context.repo, pull_number: pr.number});
+            if (current.data.head.sha !== pr.head.sha) return;
+            const url = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`;
+            const workspaces = [
+              'apps/admin-api', 'apps/aggregator', 'apps/api', 'apps/article-enrichment-worker',
+              'apps/cli', 'apps/images-worker', 'apps/mcp', 'apps/notifications',
+              'apps/search-indexer', 'apps/source-discovery-worker', 'apps/user-api',
+              'packages/core', 'packages/http'
+            ];
+            const counters = ['covered_lines', 'num_statements', 'covered_branches', 'num_branches'];
+            const integer = value => Number.isSafeInteger(value) && value >= 0;
+            const metrics = value => {
+              if (!value || !counters.every(key => integer(value[key])) ||
+                  value.covered_lines > value.num_statements || value.covered_branches > value.num_branches) {
+                throw new Error('Invalid coverage counts');
+              }
+              const percent = (covered, total) => total ? 100 * covered / total : 100;
+              return [percent(value.covered_lines, value.num_statements),
+                percent(value.covered_branches, value.num_branches),
+                percent(value.covered_lines + value.covered_branches, value.num_statements + value.num_branches)];
+            };
+            let summary, measured;
+            try {
+              // A report-only rerun reuses the successful unit job's earlier artifact.
+              const attempt = Number(process.env.GITHUB_RUN_ATTEMPT);
+              // download-artifact flattens a single match, even with merge-multiple: false.
+              const flat = fs.existsSync('report/coverage/summary.json');
+              const artifacts = flat ? [] : fs.readdirSync('report')
+                .map(name => /^python-unit-arm64-(\d+)$/.exec(name))
+                .filter(Boolean).map(match => Number(match[1]))
+                .filter(value => integer(value) && value > 0 && value <= attempt);
+              const artifactAttempt = Math.max(...artifacts);
+              if (!flat && (!integer(artifactAttempt) ||
+                  (process.env.UNIT_RESULT !== 'success' && artifactAttempt !== attempt))) {
+                throw new Error('No matching unit artifact');
+              }
+              const file = flat ? 'report/coverage/summary.json' :
+                `report/python-unit-arm64-${artifactAttempt}/coverage/summary.json`;
+              const raw = fs.readFileSync(file, 'utf8');
+              if (raw.length > 30000) throw new Error('Oversized coverage report');
+              summary = JSON.parse(raw);
+              if (!integer(summary.unit_attempt) || summary.unit_attempt < 1 ||
+                  summary.unit_attempt > attempt || (!flat && summary.unit_attempt !== artifactAttempt) ||
+                  (process.env.UNIT_RESULT !== 'success' && summary.unit_attempt !== attempt)) {
+                throw new Error('Coverage belongs to a different unit attempt');
+              }
+              if (summary.schema !== 1 || summary.valid !== true ||
+                  !['passed', 'failed'].includes(summary.status) ||
+                  !Number.isFinite(summary.minimum) || summary.minimum <= 0 || summary.minimum > 100 ||
+                  !integer(summary.unit_exit_code) || !integer(summary.source_files) || !summary.source_files ||
+                  Object.keys(summary.packages).sort().join('|') !== workspaces.join('|')) {
+                throw new Error('Incomplete coverage report');
+              }
+              measured = metrics(summary.totals);
+              if (!summary.totals.num_statements) throw new Error('Empty coverage report');
+              const sum = Object.fromEntries(counters.map(key => [key, 0]));
+              for (const name of workspaces) {
+                metrics(summary.packages[name]);
+                for (const key of counters) sum[key] += summary.packages[name][key];
+              }
+              if (!counters.every(key => sum[key] === summary.totals[key])) throw new Error('Inconsistent totals');
+              const tests = summary.tests;
+              if (!tests || !['total', 'passed', 'failed', 'skipped'].every(key => integer(tests[key])) ||
+                  !tests.total || tests.total !== tests.passed + tests.failed + tests.skipped) {
+                throw new Error('Invalid unit test counts');
+              }
+            } catch {
+              summary = null;
+            }
+            const lines = [marker, '### Backend unit coverage', ''];
+            if (summary) {
+              const tests = summary.tests;
+              const passed = process.env.UNIT_RESULT === 'success' && summary.status === 'passed' &&
+                summary.unit_exit_code === 0 && tests.failed === 0 && tests.skipped === 0 &&
+                Number(measured[2].toFixed(2)) >= summary.minimum;
+              lines.push(`**${passed ? 'PASS' : 'FAIL'}** · minimum combined coverage **${summary.minimum.toFixed(2)}%**.`, '',
+                `Tests: ${tests.passed} passed, ${tests.failed} failed, ${tests.skipped} skipped. Source files: ${summary.source_files}.`, '',
+                '| Workspace | Lines | Branches | Combined |', '| --- | ---: | ---: | ---: |');
+              const row = (name, values) => `| ${name} | ${values.map(value => `${value.toFixed(2)}%`).join(' | ')} |`;
+              lines.push(row('All backend code', measured));
+              for (const name of workspaces) lines.push(row(name, metrics(summary.packages[name])));
+              lines.push('', 'Unit coverage includes every backend workspace. PostgreSQL/Redis integration tests run separately.');
+            } else {
+              lines.push('**Coverage unavailable:** no valid unit coverage artifact could be loaded. Inspect the Python unit and coverage-report logs; this does not count as a passing coverage check.');
+            }
+            lines.push('', `Commit: ${pr.head.sha.slice(0, 12)} · [HTML, JSON, XML reports and job logs](${url}) · attempt ${process.env.GITHUB_RUN_ATTEMPT}`);
+            const body = lines.join('\n');
+            const comments = await github.paginate(github.rest.issues.listComments, {...context.repo, issue_number: pr.number});
+            const previous = comments.find(comment => comment.user.type === 'Bot' && comment.body?.startsWith(marker));
+            const latest = await github.rest.pulls.get({...context.repo, pull_number: pr.number});
+            if (latest.data.head.sha !== pr.head.sha) return;
+            if (previous) {
+              await github.rest.issues.updateComment({...context.repo, comment_id: previous.id, body});
+            } else {
+              await github.rest.issues.createComment({...context.repo, issue_number: pr.number, body});
+            }
+            if (!summary) core.setFailed('Backend coverage artifact is missing or invalid');
   admin-web:
     uses: abhi1693/actions/.github/workflows/node-ci.yml@afba12afb85bf40be9328c92dc53b779f18d5c6d # v1.0.1
     with:
@@ -280,6 +406,7 @@ jobs:
     if: ${{ always() && !cancelled() }}
     needs:
       - python-unit
+      - coverage-report
       - admin-web
       - user-web
       - extensions
@@ -299,6 +426,7 @@ jobs:
         env:
           EVENT_NAME: ${{ github.event_name }}
           REF_TYPE: ${{ github.ref_type }}
+          COVERAGE_COMMENT_REQUIRED: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name == github.repository && github.actor != 'dependabot[bot]' }}
           SONAR_EXPECTED: >-
             ${{ (github.event_name == 'push' && github.ref == 'refs/heads/master') ||
                 (github.event_name == 'pull_request' &&
@@ -306,7 +434,7 @@ jobs:
                  github.actor != 'dependabot[bot]') }}
           RESULTS: ${{ toJSON(needs) }}
         run: |
-          echo "$RESULTS" | jq -e --arg event "$EVENT_NAME" --arg ref_type "$REF_TYPE" --arg sonar "$SONAR_EXPECTED" '
+          echo "$RESULTS" | jq -e --arg event "$EVENT_NAME" --arg ref_type "$REF_TYPE" --arg comment "$COVERAGE_COMMENT_REQUIRED" --arg sonar "$SONAR_EXPECTED" '
             ([.["python-unit"], .["admin-web"], .["user-web"], .extensions,
               .["python-integration"], .["migration-upgrade"], .["reader-parity"],
               .["admin-browser"], .["live-browser"], .security]
@@ -316,5 +444,7 @@ jobs:
              else .containers.result == "success" and .["release-images"].result == "skipped" end) and
             (if $event == "pull_request" then .performance.result == "success"
              else .performance.result == "skipped" end) and
+            (if $comment == "true" then .["coverage-report"].result == "success"
+             else .["coverage-report"].result == "skipped" end) and
             (if $sonar == "true" then .sonarqube.result == "success"
              else .sonarqube.result == "skipped" end)'
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 9aa0e380..56ca67b9 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -71,8 +71,8 @@ Run the checks relevant to your change:
 # Backend lint, formatting, types and version consistency
 bash .github/scripts/python-checks.sh
 
-# Backend unit tests; no dependency services are started
-bash scripts/test.sh -q -m 'not integration'
+# Backend unit tests with line and branch coverage; no services are started
+bash scripts/ci/python-tests.sh unit
 
 # Reader app
 npm run web:lint
@@ -92,6 +92,29 @@ Pre-commit runs formatting and the affected projects' lint, type and unit checks
 it does not replace CI or integration testing. If a formatter changes files,
 review and stage those changes before retrying the commit.
 
+Backend unit coverage includes every Python source file in all 13 backend
+workspaces, including modules the tests never import. The minimum combined line
+and branch coverage is **67.60%**, configured by `tool.coverage.report.fail_under` in
+`pyproject.toml`; both CI and backend pre-commit checks enforce it. Keep this
+minimum when adding code, and raise it as coverage improves. A missing workspace,
+missing report, empty suite, failure, or skipped unit test fails the CI report gate.
+
+CI updates one coverage comment on each same-repository PR, showing test counts
+and line, branch, and combined coverage for each workspace. Superseded PR commits
+cannot overwrite the current comment. Fork PRs and Dependabot runs retain the job
+summary and report artifacts because their tokens cannot write PR comments.
+Download the Python unit artifact for HTML, JSON, and Cobertura XML reports; local
+reports are in `reports/coverage/` (open `html/index.html`). Integration results do
+not inflate the unit coverage score and remain independent CI requirements.
+
+For a focused test run, omit coverage or temporarily disable its whole-suite
+minimum for that command:
+
+```sh
+uv run --locked pytest -q tests/test_mcp_oauth_unit.py
+uv run --locked pytest -q tests/test_mcp_oauth_unit.py --cov --cov-fail-under=0
+```
+
 Integration tests require explicit `DEVFEED_TEST_DATABASE_URL` and
 `DEVFEED_TEST_REDIS_URL`. The PostgreSQL database name must end in `_test`, and
 Redis must use database **15**. These tests truncate tables and flush Redis: never
diff --git a/packages/core/src/catalog_cache.py b/packages/core/src/catalog_cache.py
index d39a0b91..68773529 100644
--- a/packages/core/src/catalog_cache.py
+++ b/packages/core/src/catalog_cache.py
@@ -84,10 +84,11 @@ def snapshot(session, names, loader):
         if body is not None:
             entry = json.loads(body)
             if entry["revision"] == [list(item) for item in before]:
+                value = entry["value"]
                 record_cache_read("catalog", "hit")
                 if values is not None:
-                    values[slot] = (before, entry["value"])
-                return entry["value"]
+                    values[slot] = (before, value)
+                return value
     except CacheUnavailable:
         outcome, reason = "bypass", "cache_unavailable"
     except (ValueError, UnicodeError, KeyError, TypeError):
diff --git a/pyproject.toml b/pyproject.toml
index adcb0221..9a7999d0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -80,6 +80,7 @@ dev = [
   "pre-commit>=4.3,<5",
   "jsonschema>=4.23,<5",
   "testcontainers>=4.10,<5",
+  "pytest-cov>=7,<8",
 ]
 loadtest = [
     "locust==2.46.6",
@@ -108,7 +109,27 @@ markers = ["integration: requires a dedicated PostgreSQL database and Redis"]
 [tool.coverage.run]
 branch = true
 relative_files = true
-include = ["apps/*/src/*.py", "packages/*/src/*.py", "scripts/**/*.py"]
+data_file = "reports/coverage/.coverage-unit"
+source = [
+  "packages/core/src",
+  "packages/http/src",
+  "apps/api/src",
+  "apps/admin-api/src",
+  "apps/user-api/src",
+  "apps/mcp/src",
+  "apps/aggregator/src",
+  "apps/notifications/src",
+  "apps/cli/src",
+  "apps/search-indexer/src",
+  "apps/article-enrichment-worker/src",
+  "apps/images-worker/src",
+  "apps/source-discovery-worker/src",
+]
+
+[tool.coverage.report]
+fail_under = 67.60
+precision = 2
+show_missing = true
 
 [tool.mypy]
 python_version = "3.12"
diff --git a/scripts/ci/coverage_report.py b/scripts/ci/coverage_report.py
new file mode 100644
index 00000000..7d9f5fc6
--- /dev/null
+++ b/scripts/ci/coverage_report.py
@@ -0,0 +1,166 @@
+"""Validate complete backend unit coverage and produce a bounded PR reporting artifact."""
+
+import argparse
+import json
+import math
+import os
+import tomllib
+import xml.etree.ElementTree as ET
+from pathlib import Path
+
+COUNTERS = ("covered_lines", "num_statements", "covered_branches", "num_branches")
+
+
+def counts(value: dict) -> dict:
+    result = {name: value[name] for name in COUNTERS}
+    if any(type(number) is not int or number < 0 for number in result.values()):
+        raise ValueError("Invalid coverage counts")
+    if (
+        result["covered_lines"] > result["num_statements"]
+        or result["covered_branches"] > result["num_branches"]
+    ):
+        raise ValueError("Covered counts exceed measured code")
+    return result
+
+
+def percentages(value: dict) -> dict:
+    lines, branches = value["num_statements"], value["num_branches"]
+    return {
+        **value,
+        "line_percent": 100 * value["covered_lines"] / lines if lines else 100.0,
+        "branch_percent": 100 * value["covered_branches"] / branches if branches else 100.0,
+        "combined_percent": 100
+        * (value["covered_lines"] + value["covered_branches"])
+        / (lines + branches)
+        if lines + branches
+        else 100.0,
+    }
+
+
+def summarize(root: Path, test_exit_code: int) -> dict:
+    attempt = int(os.environ.get("GITHUB_RUN_ATTEMPT", "1"))
+    if attempt < 1:
+        raise ValueError("Unit run attempt must be positive")
+    config = tomllib.loads((root / "pyproject.toml").read_text())
+    coverage = config["tool"]["coverage"]
+    members = config["tool"]["uv"]["workspace"]["members"]
+    sources = coverage["run"]["source"]
+    if coverage["run"].get("branch") is not True or set(sources) != {
+        f"{member}/src" for member in members
+    }:
+        raise ValueError("Coverage must include every backend workspace with branches")
+    if (
+        len(sources) != len(set(sources))
+        or coverage["run"].get("omit")
+        or coverage["report"].get("omit")
+    ):
+        raise ValueError("Coverage sources must not be duplicated or omitted")
+    minimum = coverage["report"]["fail_under"]
+    if type(minimum) not in (int, float) or not math.isfinite(minimum) or not 0 < minimum <= 100:
+        raise ValueError("Coverage minimum must be a finite percentage")
+    document = json.loads((root / "reports/coverage/unit.json").read_text())
+    if document["meta"].get("branch_coverage") is not True:
+        raise ValueError("Branch coverage was not measured")
+    expected = {
+        path.relative_to(root).as_posix()
+        for source in sources
+        for path in (root / source).rglob("*.py")
+    }
+    if not expected or set(document["files"]) != expected:
+        raise ValueError("Coverage inventory differs from backend source files")
+    packages = {member: dict.fromkeys(COUNTERS, 0) for member in sorted(members)}
+    for filename, entry in document["files"].items():
+        member = next(member for member in members if filename.startswith(f"{member}/src/"))
+        for name, count in counts(entry["summary"]).items():
+            packages[member][name] += count
+    total = {name: sum(package[name] for package in packages.values()) for name in COUNTERS}
+    if counts(document["totals"]) != total or total["num_statements"] == 0:
+        raise ValueError("Coverage totals do not match measured source")
+    cases = list(ET.parse(root / "reports/python-unit.xml").getroot().iter("testcase"))
+    if not cases:
+        raise ValueError("No unit tests were reported")
+    failed = sum(
+        case.find("failure") is not None or case.find("error") is not None for case in cases
+    )
+    skipped = sum(case.find("skipped") is not None for case in cases)
+    tests = {
+        "total": len(cases),
+        "failed": failed,
+        "skipped": skipped,
+        "passed": len(cases) - failed - skipped,
+    }
+    total = percentages(total)
+    passed = (
+        test_exit_code == 0
+        and failed == 0
+        and skipped == 0
+        and round(total["combined_percent"], 2) >= minimum
+    )
+    return {
+        "schema": 1,
+        "valid": True,
+        "status": "passed" if passed else "failed",
+        "minimum": minimum,
+        "unit_exit_code": test_exit_code,
+        "unit_attempt": attempt,
+        "source_files": len(expected),
+        "totals": total,
+        "packages": {member: percentages(value) for member, value in packages.items()},
+        "tests": tests,
+    }
+
+
+def markdown(summary: dict) -> str:
+    if not summary["valid"]:
+        return (
+            "### Backend unit coverage\n\nCoverage report is missing or invalid. "
+            "Inspect the Python unit job logs.\n"
+        )
+    total, tests = summary["totals"], summary["tests"]
+    rows = [
+        "### Backend unit coverage",
+        "",
+        f"**{summary['status'].upper()}** — minimum combined coverage "
+        f"**{summary['minimum']:.2f}%**.",
+        "",
+        f"Tests: {tests['passed']} passed, {tests['failed']} failed, {tests['skipped']} skipped. "
+        f"Source files: {summary['source_files']}.",
+        "",
+        "| Workspace | Lines | Branches | Combined |",
+        "| --- | ---: | ---: | ---: |",
+    ]
+    for name, value in [("All backend code", total), *summary["packages"].items()]:
+        rows.append(
+            f"| {name} | {value['line_percent']:.2f}% | {value['branch_percent']:.2f}% | "
+            f"{value['combined_percent']:.2f}% |"
+        )
+    rows.extend(
+        [
+            "",
+            "Unit coverage includes every backend workspace. "
+            "PostgreSQL/Redis integration tests run separately.",
+            "",
+        ]
+    )
+    return "\n".join(rows)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[2])
+    parser.add_argument("--test-exit-code", type=int, required=True)
+    args = parser.parse_args()
+    try:
+        summary = summarize(args.root, args.test_exit_code)
+    except (OSError, ValueError, KeyError, TypeError, ET.ParseError, StopIteration):
+        summary = {"schema": 1, "valid": False, "status": "incomplete"}
+    destination = args.root / "reports/coverage"
+    destination.mkdir(parents=True, exist_ok=True)
+    (destination / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
+    (destination / "summary.md").write_text(markdown(summary))
+    print(markdown(summary))
+    return 0 if summary["status"] == "passed" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/ci/python-tests.sh b/scripts/ci/python-tests.sh
index 156b9f70..9b4328ad 100644
--- a/scripts/ci/python-tests.sh
+++ b/scripts/ci/python-tests.sh
@@ -18,9 +18,34 @@ mkdir -p reports
 export DEVFEED_DATABASE_URL=postgresql+psycopg://ci@database.invalid/ci
 export DEVFEED_REDIS_URL=redis://redis.invalid/15
 if [ "$suite" != integration ]; then
+  rm -rf reports/coverage
+  rm -f reports/python-unit.xml
   mkdir -p reports/coverage/python-unit
-  COVERAGE_FILE=reports/coverage/python-unit/coverage.db uv run --locked --no-build coverage run -m pytest \
-    -q -m 'not integration' --junitxml=reports/python-unit.xml
+  unit_coverage=reports/coverage/python-unit/coverage.db
+  backend_include='apps/*/src/**,packages/*/src/**'
+  backend_sources=$(uv run --locked --no-build python -c \
+    'import tomllib; print(",".join(tomllib.load(open("pyproject.toml", "rb"))["tool"]["coverage"]["run"]["source"]))')
+  unit_status=0
+  # Keep CI utility measurements for Sonar; the unit floor/report covers every backend file.
+  COVERAGE_FILE="$unit_coverage" uv run --locked --no-build coverage run \
+    --source="$backend_sources,scripts" -m pytest -q -m 'not integration' \
+    --junitxml=reports/python-unit.xml || unit_status=$?
+  report_status=0
+  uv run --locked --no-build coverage report --data-file="$unit_coverage" \
+    --include="$backend_include" --skip-covered || report_status=$?
+  uv run --locked --no-build coverage json --data-file="$unit_coverage" \
+    --include="$backend_include" -o reports/coverage/unit.json || report_status=$?
+  uv run --locked --no-build coverage xml --data-file="$unit_coverage" \
+    --include="$backend_include" -o reports/coverage/unit.xml || report_status=$?
+  uv run --locked --no-build coverage html --data-file="$unit_coverage" \
+    --include="$backend_include" -d reports/coverage/html || report_status=$?
+  uv run --locked --no-build python scripts/ci/coverage_report.py --test-exit-code "$unit_status" \
+    || report_status=$?
+  if [[ -n "${GITHUB_STEP_SUMMARY:-}" ]]; then
+    cat reports/coverage/summary.md >> "$GITHUB_STEP_SUMMARY"
+  fi
+  if [[ "$unit_status" -ne 0 ]]; then exit "$unit_status"; fi
+  if [[ "$report_status" -ne 0 ]]; then exit "$report_status"; fi
   uv run --locked python scripts/ci/check_reports.py junit reports/python-unit.xml
 fi
 if [ "$suite" = unit ]; then exit 0; fi
diff --git a/scripts/ci/sonar_coverage.py b/scripts/ci/sonar_coverage.py
index 31b873b8..cb4663df 100644
--- a/scripts/ci/sonar_coverage.py
+++ b/scripts/ci/sonar_coverage.py
@@ -56,6 +56,9 @@ def prepare_reports(root: Path = ROOT) -> None:
     coverage.combine(data_paths=[str(path) for path in databases], strict=True, keep=True)
     if not coverage.get_data().measured_files():
         raise ValueError("No Python coverage was measured")
+    # Workspace roots are useful for the unit inventory, but XML must distinguish
+    # identically named files across services and retain measured CI utilities.
+    coverage.set_option("run:source", [str(root)])
     coverage.xml_report(outfile=str(reports / "python.xml"))
 
 
diff --git a/scripts/pre_commit_checks.py b/scripts/pre_commit_checks.py
index a6b475e0..dbabca89 100644
--- a/scripts/pre_commit_checks.py
+++ b/scripts/pre_commit_checks.py
@@ -44,7 +44,17 @@ def checks_for(paths: set[str]) -> list[tuple[str, ...]]:
                 ("uv", "run", "--locked", "ruff", "check", "."),
                 ("uv", "run", "--locked", "ruff", "format", "--check", "."),
                 ("uv", "run", "--locked", "mypy"),
-                ("uv", "run", "--locked", "pytest", "-q", "-m", "not integration"),
+                (
+                    "uv",
+                    "run",
+                    "--locked",
+                    "pytest",
+                    "-q",
+                    "-m",
+                    "not integration",
+                    "--cov",
+                    "--cov-report=term:skip-covered",
+                ),
             ]
         )
     admin = shared_web or any(
diff --git a/tests/test_catalog_snapshot_unit.py b/tests/test_catalog_snapshot_unit.py
new file mode 100644
index 00000000..ef47df79
--- /dev/null
+++ b/tests/test_catalog_snapshot_unit.py
@@ -0,0 +1,116 @@
+"""Revision-cache safety and fallback policy independent of PostgreSQL/Redis I/O."""
+
+import json
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from devfeed_core import catalog_cache
+from devfeed_core.cache import CacheUnavailable
+
+REVISIONS = [("tags", "tag-revision"), ("topics", "topic-revision")]
+NAMES = ("topics", "tags")
+
+
+@pytest.fixture
+def snapshot(monkeypatch):
+    session = Mock()
+    session.execute.side_effect = [REVISIONS, REVISIONS]
+    cache = SimpleNamespace(namespace="test", redis=Mock(), _run=lambda call: call())
+    cache.redis.get.return_value = None
+    loader = Mock(return_value=[{"name": "Python"}])
+    metrics = Mock()
+    monkeypatch.setattr(catalog_cache, "get_settings", lambda: SimpleNamespace(cache_enabled=True))
+    monkeypatch.setattr(catalog_cache, "get_cache", lambda: cache)
+    monkeypatch.setattr(catalog_cache, "record_cache_read", metrics)
+    return SimpleNamespace(session=session, cache=cache, loader=loader, metrics=metrics)
+
+
+def test_matching_revisions_reuse_cached_catalog_without_loading_database_rows(snapshot):
+    snapshot.cache.redis.get.return_value = json.dumps({"revision": REVISIONS, "value": []})
+    assert catalog_cache.snapshot(snapshot.session, NAMES, snapshot.loader) == []
+    snapshot.loader.assert_not_called()
+    snapshot.cache.redis.set.assert_not_called()
+    assert snapshot.session.execute.call_count == 1
+    snapshot.metrics.assert_called_once_with("catalog", "hit")
+
+
+@pytest.mark.parametrize(
+    "body,reason",
+    [
+        (None, None),
+        (json.dumps({"revision": [["tags", "old"], ["topics", "old"]], "value": []}), None),
+        (b"\xff", "invalid_entry"),
+        ("not json", "invalid_entry"),
+        ("{}", "invalid_entry"),
+        ("[]", "invalid_entry"),
+        (json.dumps({"revision": REVISIONS}), "invalid_entry"),
+    ],
+)
+def test_misses_and_invalid_entries_reload_and_publish_only_the_current_revision(
+    snapshot, body, reason
+):
+    snapshot.cache.redis.get.return_value = body
+    assert (
+        catalog_cache.snapshot(snapshot.session, NAMES, snapshot.loader)
+        == snapshot.loader.return_value
+    )
+    snapshot.loader.assert_called_once_with()
+    snapshot.session.flush.assert_called_once_with()
+    snapshot.metrics.assert_called_once_with("catalog", "miss", reason)
+    args, kwargs = snapshot.cache.redis.set.call_args
+    assert args[0] == "test:catalog:v1:tags,topics"
+    assert json.loads(args[1]) == {
+        "revision": [list(row) for row in REVISIONS],
+        "value": snapshot.loader.return_value,
+    }
+    assert kwargs == {"ex": 600}
+
+
+def test_concurrent_catalog_edit_never_publishes_a_mixed_snapshot(snapshot):
+    snapshot.session.execute.side_effect = [REVISIONS, [("tags", "new"), REVISIONS[1]]]
+    assert (
+        catalog_cache.snapshot(snapshot.session, NAMES, snapshot.loader)
+        == snapshot.loader.return_value
+    )
+    snapshot.cache.redis.set.assert_not_called()
+
+
+@pytest.mark.parametrize("failed_operation", ["get", "set"])
+def test_cache_outage_keeps_database_catalog_available(snapshot, failed_operation):
+    getattr(snapshot.cache.redis, failed_operation).side_effect = CacheUnavailable
+    assert (
+        catalog_cache.snapshot(snapshot.session, NAMES, snapshot.loader)
+        == snapshot.loader.return_value
+    )
+    snapshot.loader.assert_called_once_with()
+    if failed_operation == "get":
+        snapshot.metrics.assert_called_once_with("catalog", "bypass", "cache_unavailable")
+
+
+def test_missing_revisions_bypass_storage_without_guessing_cache_freshness(snapshot):
+    snapshot.session.execute.side_effect = [REVISIONS[:1]]
+    assert (
+        catalog_cache.snapshot(snapshot.session, NAMES, snapshot.loader)
+        == snapshot.loader.return_value
+    )
+    snapshot.cache.redis.get.assert_not_called()
+    snapshot.cache.redis.set.assert_not_called()
+
+
+def test_disabled_cache_uses_loader_without_flushing_or_querying_revisions(snapshot, monkeypatch):
+    monkeypatch.setattr(catalog_cache, "get_settings", lambda: SimpleNamespace(cache_enabled=False))
+    assert (
+        catalog_cache.snapshot(snapshot.session, NAMES, snapshot.loader)
+        == snapshot.loader.return_value
+    )
+    snapshot.session.flush.assert_not_called()
+    snapshot.session.execute.assert_not_called()
+    snapshot.metrics.assert_called_once_with("catalog", "bypass", "disabled")
+
+
+def test_database_failure_is_not_converted_to_a_successful_cache_response(snapshot):
+    snapshot.loader.side_effect = RuntimeError("database unavailable")
+    with pytest.raises(RuntimeError, match="database unavailable"):
+        catalog_cache.snapshot(snapshot.session, NAMES, snapshot.loader)
+    snapshot.cache.redis.set.assert_not_called()
diff --git a/tests/test_ci_gates.py b/tests/test_ci_gates.py
index bc94b5b9..b5adf853 100644
--- a/tests/test_ci_gates.py
+++ b/tests/test_ci_gates.py
@@ -107,13 +107,17 @@ def test_ci_shards_reject_invalid_partition(tmp_path, index, count):
         ("schedule", "branch", False),
     ],
 )
-def test_ci_required_accepts_only_the_expected_successes(event, ref_type, sonar_expected):
+@pytest.mark.parametrize("comment_required", [False, True])
+def test_ci_required_accepts_only_the_expected_successes(
+    event, ref_type, sonar_expected, comment_required
+):
     # Execute the actual gate, including the mutually exclusive release/check jobs.
     workflow = (ROOT / ".github/workflows/ci.yml").read_text()
     gate = textwrap.dedent(workflow.split("  required:\n", 1)[1].split("        run: |\n", 1)[1])
     release = event == "push" and ref_type == "tag"
     results = {
         "python-unit": {"result": "success"},
+        "coverage-report": {"result": "success" if comment_required else "skipped"},
         "admin-web": {"result": "success"},
         "user-web": {"result": "success"},
         "extensions": {"result": "success"},
@@ -128,6 +132,12 @@ def test_ci_required_accepts_only_the_expected_successes(event, ref_type, sonar_
         "performance": {"result": "success" if event == "pull_request" else "skipped"},
         "sonarqube": {"result": "success" if sonar_expected else "skipped"},
     }
+    declared_needs = (
+        workflow.split("  required:\n", 1)[1]
+        .split("    needs:\n", 1)[1]
+        .split("    runs-on:", 1)[0]
+    )
+    assert {line.strip().removeprefix("- ") for line in declared_needs.splitlines()} == set(results)
 
     def evaluate():
         return subprocess.run(
@@ -136,6 +146,7 @@ def evaluate():
                 **os.environ,
                 "EVENT_NAME": event,
                 "REF_TYPE": ref_type,
+                "COVERAGE_COMMENT_REQUIRED": str(comment_required).lower(),
                 "SONAR_EXPECTED": str(sonar_expected).lower(),
                 "RESULTS": json.dumps(results),
             },
diff --git a/tests/test_coverage_report.py b/tests/test_coverage_report.py
new file mode 100644
index 00000000..dc34d45e
--- /dev/null
+++ b/tests/test_coverage_report.py
@@ -0,0 +1,341 @@
+"""Fail-closed coverage inventories, measured counters, and PR comment rendering."""
+
+import importlib.util
+import json
+import subprocess
+import textwrap
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+SPEC = importlib.util.spec_from_file_location(
+    "coverage_report", ROOT / "scripts/ci/coverage_report.py"
+)
+report = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(report)
+
+
+@pytest.fixture
+def measurement(tmp_path):
+    (tmp_path / "pyproject.toml").write_text(
+        '[tool.uv.workspace]\nmembers = ["apps/api", "packages/core"]\n'
+        '[tool.coverage.run]\nbranch = true\nsource = ["apps/api/src", "packages/core/src"]\n'
+        "[tool.coverage.report]\nfail_under = 65.0\n"
+    )
+    paths = ["apps/api/src/main.py", "packages/core/src/unimported.py"]
+    for name in paths:
+        path = tmp_path / name
+        path.parent.mkdir(parents=True)
+        path.write_text("def value():\n    return 1\n")
+    covered = {"covered_lines": 3, "num_statements": 4, "covered_branches": 1, "num_branches": 2}
+    document = {
+        "meta": {"branch_coverage": True},
+        "files": {name: {"summary": dict(covered)} for name in paths},
+        "totals": {name: value * 2 for name, value in covered.items()},
+    }
+    (tmp_path / "reports/coverage").mkdir(parents=True)
+    (tmp_path / "reports/python-unit.xml").write_text(
+        '<testsuites><testsuite tests="2"><testcase/><testcase/></testsuite></testsuites>'
+    )
+
+    def save():
+        (tmp_path / "reports/coverage/unit.json").write_text(json.dumps(document))
+
+    save()
+    return tmp_path, document, save
+
+
+def test_report_counts_all_backend_files_and_separates_lines_from_branches(measurement):
+    root, _, _ = measurement
+    value = report.summarize(root, 0)
+    assert value["status"] == "passed" and value["source_files"] == 2
+    assert value["totals"]["line_percent"] == 75
+    assert value["totals"]["branch_percent"] == 50
+    assert value["totals"]["combined_percent"] == pytest.approx(66.6666667)
+    assert value["tests"] == {"total": 2, "passed": 2, "failed": 0, "skipped": 0}
+    assert set(value["packages"]) == {"apps/api", "packages/core"}
+    assert "66.67%" in report.markdown(value)
+
+
+@pytest.mark.parametrize("exit_code", [1, 2, 5])
+def test_successful_testcases_cannot_hide_pytest_or_coverage_failure(measurement, exit_code):
+    root, _, _ = measurement
+    assert report.summarize(root, exit_code)["status"] == "failed"
+
+
+@pytest.mark.parametrize("tag", ["failure", "error", "skipped"])
+def test_nonpassing_testcases_fail_the_report(measurement, tag):
+    root, _, _ = measurement
+    (root / "reports/python-unit.xml").write_text(
+        f"<testsuite><testcase><{tag}/></testcase></testsuite>"
+    )
+    assert report.summarize(root, 0)["status"] == "failed"
+
+
+def test_coverage_below_the_configured_minimum_fails(measurement):
+    root, _, _ = measurement
+    path = root / "pyproject.toml"
+    path.write_text(path.read_text().replace("65.0", "67.0"))
+    assert report.summarize(root, 0)["status"] == "failed"
+
+
+@pytest.mark.parametrize("attempt", ["0", "invalid"])
+def test_unit_artifact_metadata_rejects_invalid_run_attempt(measurement, monkeypatch, attempt):
+    root, _, _ = measurement
+    monkeypatch.setenv("GITHUB_RUN_ATTEMPT", attempt)
+    with pytest.raises(ValueError):
+        report.summarize(root, 0)
+
+
+@pytest.mark.parametrize(
+    "change",
+    [
+        "missing-file",
+        "extra-file",
+        "totals",
+        "negative",
+        "boolean",
+        "overcount",
+        "branches",
+        "empty",
+    ],
+)
+def test_incomplete_or_falsified_measurements_are_rejected(measurement, change):
+    root, document, save = measurement
+    if change == "missing-file":
+        document["files"].pop("packages/core/src/unimported.py")
+    elif change == "extra-file":
+        document["files"]["tests/test_example.py"] = next(iter(document["files"].values()))
+    elif change == "branches":
+        document["meta"]["branch_coverage"] = False
+    elif change == "empty":
+        for entry in document["files"].values():
+            entry["summary"] = dict.fromkeys(report.COUNTERS, 0)
+        document["totals"] = dict.fromkeys(report.COUNTERS, 0)
+    else:
+        document["totals"]["covered_lines"] = {
+            "totals": 1,
+            "negative": -1,
+            "boolean": True,
+            "overcount": 99,
+        }[change]
+    save()
+    with pytest.raises(ValueError):
+        report.summarize(root, 0)
+
+
+@pytest.mark.parametrize(
+    "change", ["source", "duplicate", "branch", "omit", "minimum", "empty-tests"]
+)
+def test_configuration_cannot_hide_workspaces_or_accept_empty_suites(measurement, change):
+    root, _, _ = measurement
+    path = root / "pyproject.toml"
+    value = path.read_text()
+    if change == "source":
+        value = value.replace(
+            'source = ["apps/api/src", "packages/core/src"]', 'source = ["apps/api/src"]'
+        )
+    elif change == "duplicate":
+        value = value.replace(
+            'source = ["apps/api/src", "packages/core/src"]',
+            'source = ["apps/api/src", "packages/core/src", "apps/api/src"]',
+        )
+    elif change == "branch":
+        value = value.replace("branch = true", "branch = false")
+    elif change == "omit":
+        value += 'omit = ["*/low_coverage.py"]\n'
+    elif change == "minimum":
+        value = value.replace("65.0", "nan")
+    else:
+        (root / "reports/python-unit.xml").write_text('<testsuite tests="100"/>')
+    path.write_text(value)
+    with pytest.raises(ValueError):
+        report.summarize(root, 0)
+
+
+@pytest.mark.parametrize(
+    "broken", ["missing-coverage", "invalid-json", "missing-junit", "invalid-xml"]
+)
+def test_cli_always_writes_an_incomplete_report_on_invalid_input(measurement, broken):
+    root, _, _ = measurement
+    path = root / (
+        "reports/coverage/unit.json"
+        if broken in {"missing-coverage", "invalid-json"}
+        else "reports/python-unit.xml"
+    )
+    if broken.startswith("missing"):
+        path.unlink()
+    else:
+        path.write_text("invalid input")
+    result = subprocess.run(
+        [
+            "uv",
+            "run",
+            "--locked",
+            "python",
+            str(ROOT / "scripts/ci/coverage_report.py"),
+            "--root",
+            str(root),
+            "--test-exit-code",
+            "0",
+        ],
+        cwd=ROOT,
+        capture_output=True,
+        text=True,
+    )
+    assert result.returncode == 1
+    assert json.loads((root / "reports/coverage/summary.json").read_text()) == {
+        "schema": 1,
+        "valid": False,
+        "status": "incomplete",
+    }
+    assert "missing or invalid" in (root / "reports/coverage/summary.md").read_text()
+
+
+def run_comment(tmp_path, *, scenario="success"):
+    workflow = (ROOT / ".github/workflows/ci.yml").read_text()
+    script = textwrap.dedent(
+        workflow.split("          script: |\n", 1)[1].split("  admin-web:\n", 1)[0]
+    )
+    wrapper = """
+      const fs = require('fs');
+      const vm = require('vm');
+      const input = JSON.parse(fs.readFileSync(process.argv[2], 'utf8'));
+      const calls = [];
+      const head = input.stale ? 'new-head' : 'abcdef1234567890';
+      let reads = 0;
+      const github = {rest: {
+        pulls: {get: async () => ({data: {head: {
+          sha: input.lateStale && reads++ ? 'new-head' : head}}})},
+        issues: {listComments: {}, createComment: async value => calls.push(['create', value]),
+          updateComment: async value => calls.push(['update', value])}
+      }, paginate: async () => input.previous ? [{id: 7,
+        user: {type: input.human ? 'User' : 'Bot'},
+        body: '<!-- devfeed-backend-unit-coverage -->\\nold'}] : []};
+      const context = {repo: {owner: 'owner', repo: 'repo'},
+        serverUrl: 'https://github.com', runId: 123,
+        payload: {pull_request: {number: 4, head: {sha: 'abcdef1234567890'}}}};
+      const sandbox = {require: () => ({readdirSync: () => input.artifacts,
+        existsSync: () => input.flat,
+        readFileSync: path => {
+          const expected = input.flat ? 'report/coverage/summary.json' :
+            `report/${input.expectedArtifact}/coverage/summary.json`;
+          if (path !== expected) {
+            throw new Error('Wrong artifact selected');
+          }
+          return JSON.stringify(input.summary);
+        }}),
+        github, context,
+        core: {setFailed: message => calls.push(['failed', message])},
+        process: {env: {UNIT_RESULT: input.result, GITHUB_RUN_ATTEMPT: '2'}}};
+      vm.runInNewContext('(async () => {' + input.script + '})()', sandbox)
+        .then(() => console.log(JSON.stringify(calls)))
+        .catch(error => {console.error(error); process.exit(1);});
+    """
+    workspaces = [
+        "apps/admin-api",
+        "apps/aggregator",
+        "apps/api",
+        "apps/article-enrichment-worker",
+        "apps/cli",
+        "apps/images-worker",
+        "apps/mcp",
+        "apps/notifications",
+        "apps/search-indexer",
+        "apps/source-discovery-worker",
+        "apps/user-api",
+        "packages/core",
+        "packages/http",
+    ]
+    counters = {"covered_lines": 3, "num_statements": 4, "covered_branches": 1, "num_branches": 2}
+    summary = {
+        "schema": 1,
+        "valid": True,
+        "status": "passed",
+        "minimum": 65,
+        "unit_exit_code": 0,
+        "unit_attempt": 1
+        if scenario in {"report-rerun", "stale-failure", "flat-rerun", "flat-stale-failure"}
+        else 2,
+        "source_files": 270,
+        "totals": {key: value * 13 for key, value in counters.items()},
+        "packages": dict.fromkeys(workspaces, counters),
+        "tests": {"total": 2, "passed": 2, "failed": 0, "skipped": 0},
+    }
+    if scenario == "malformed":
+        summary["totals"]["covered_lines"] = "@everyone <script>"
+    elif scenario == "missing":
+        summary = {"schema": 1, "valid": False, "status": "incomplete"}
+    elif scenario == "below-floor":
+        summary["minimum"] = 80
+    elif scenario == "pytest-failed":
+        summary["unit_exit_code"] = 1
+    elif scenario == "mismatched":
+        summary["totals"]["covered_lines"] = 1
+    input_path, wrapper_path = tmp_path / "input.json", tmp_path / "comment.cjs"
+    input_path.write_text(
+        json.dumps(
+            {
+                "script": script,
+                "summary": summary,
+                "result": "failure"
+                if scenario in {"job-failed", "stale-failure", "flat-stale-failure"}
+                else "success",
+                "flat": scenario.startswith("flat-"),
+                "artifacts": ["python-unit-arm64-1"]
+                if scenario in {"report-rerun", "stale-failure"}
+                else [
+                    "python-unit-arm64-1",
+                    "python-unit-arm64-2",
+                    "python-unit-arm64-invalid",
+                    "python-unit-arm64-3",
+                ],
+                "expectedArtifact": "python-unit-arm64-1"
+                if scenario == "report-rerun"
+                else "python-unit-arm64-2",
+                "stale": scenario == "stale",
+                "lateStale": scenario == "late-stale",
+                "previous": scenario in {"previous", "human"},
+                "human": scenario == "human",
+            }
+        )
+    )
+    wrapper_path.write_text(textwrap.dedent(wrapper))
+    result = subprocess.run(
+        ["node", str(wrapper_path), str(input_path)], capture_output=True, text=True, check=True
+    )
+    return json.loads(result.stdout)
+
+
+@pytest.mark.parametrize(
+    "scenario", ["success", "previous", "human", "report-rerun", "flat-single", "flat-rerun"]
+)
+def test_pr_report_uses_measured_counts_and_updates_one_bot_comment(tmp_path, scenario):
+    calls = run_comment(tmp_path, scenario=scenario)
+    assert len(calls) == 1 and calls[0][0] == ("update" if scenario == "previous" else "create")
+    body = calls[0][1]["body"]
+    assert "**PASS**" in body and "66.67%" in body and "65.00%" in body
+    assert "2 passed" in body and "abcdef123456" in body
+    assert "packages/core" in body and "actions/runs/123" in body
+
+
+@pytest.mark.parametrize("scenario", ["below-floor", "pytest-failed", "job-failed"])
+def test_pr_report_never_claims_success_for_a_failed_gate(tmp_path, scenario):
+    assert "**FAIL**" in run_comment(tmp_path, scenario=scenario)[0][1]["body"]
+
+
+@pytest.mark.parametrize(
+    "scenario", ["missing", "malformed", "mismatched", "stale-failure", "flat-stale-failure"]
+)
+def test_invalid_reports_have_fixed_text_and_cannot_inject_comment_content(tmp_path, scenario):
+    calls = run_comment(tmp_path, scenario=scenario)
+    body = calls[0][1]["body"]
+    assert "Coverage unavailable" in body and "**PASS**" not in body
+    assert "@everyone" not in body and "<script>" not in body
+    assert calls[1] == ["failed", "Backend coverage artifact is missing or invalid"]
+
+
+@pytest.mark.parametrize("scenario", ["stale", "late-stale"])
+def test_superseded_pr_head_receives_no_comment(tmp_path, scenario):
+    assert run_comment(tmp_path, scenario=scenario) == []
diff --git a/tests/test_job_retry_unit.py b/tests/test_job_retry_unit.py
new file mode 100644
index 00000000..45632b75
--- /dev/null
+++ b/tests/test_job_retry_unit.py
@@ -0,0 +1,220 @@
+"""Retry selection and dispatch policy without network or production services."""
+
+import uuid
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from devfeed_core import job_retries, models
+from devfeed_core.services import OperationConflict, RecordNotFound
+from sqlalchemy import Column, MetaData, Table, create_engine, select
+
+NOW = datetime(2026, 10, 4, tzinfo=UTC)
+ID = uuid.UUID(int=10)
+ACTOR = {"subject": "reviewer"}
+
+
+@pytest.mark.parametrize(
+    "model,subject",
+    [
+        (models.IngestionJob, "source_id"),
+        (models.ArticleAnalysisJob, "article_id"),
+        (models.ArticleEnrichmentJob, "article_id"),
+        (models.SourceEnrichmentJob, "source_id"),
+        (models.ArticleImageJob, "article_id"),
+        (models.ArticleImageJob, "topic_id"),
+        (models.TopicAnalysisJob, "proposal_id"),
+        (models.TopicAnalysisJob, "topic_id"),
+    ],
+)
+def test_retry_selection_keeps_only_unresolved_latest_failure_even_with_out_of_order_active_run(
+    model, subject
+):
+    # Execute the selection expression against minimal in-memory tables. Existing
+    # PostgreSQL integration tests still verify real row locks and transactions.
+    metadata = MetaData()
+    names = ["id", "status", "created_at", "source_id", "article_id", "topic_id", "proposal_id"]
+    table = Table(
+        model.__tablename__,
+        metadata,
+        *[
+            Column(name, model.__table__.c[name].type.copy())
+            for name in names
+            if name in model.__table__.c
+        ],
+    )
+    engine = create_engine("sqlite://")
+    metadata.create_all(engine)
+    base = {column.name: None for column in table.c}
+
+    def row(number, owner, status="failed", age=0):
+        return {
+            **base,
+            "id": uuid.UUID(int=number),
+            subject: uuid.UUID(int=owner),
+            "status": status,
+            "created_at": NOW + timedelta(seconds=age),
+        }
+
+    rows = [
+        row(1, 1),
+        row(2, 1),  # ID breaks equal-timestamp ties.
+        row(3, 2),
+        row(4, 2, "succeeded", 1),
+        row(5, 3),
+        row(6, 3, "running", -1),  # Earlier active job still supersedes.
+        row(7, 4),
+        row(8, 4, "queued", -1),
+        row(9, 5),  # Different subject remains retryable.
+    ]
+    with engine.begin() as connection:
+        connection.execute(table.insert(), rows)
+        eligible = set(
+            connection.scalars(select(model.id).where(job_retries.retry_candidate(model)))
+        )
+        display = dict(
+            connection.execute(select(model.id, job_retries.job_display_status(model))).all()
+        )
+    engine.dispose()
+    assert eligible == {uuid.UUID(int=2), uuid.UUID(int=9)}
+    assert [display[uuid.UUID(int=index)] for index in (1, 3, 5, 7)] == ["retried"] * 4
+    assert display[uuid.UUID(int=6)] == "running"
+    assert display[uuid.UUID(int=8)] == "queued"
+
+
+def test_notification_retry_selection_reuses_each_failed_delivery():
+    expression = job_retries.retry_candidate(models.NotificationDelivery)
+    assert expression.compare(models.NotificationDelivery.status == "failed")
+
+
+@pytest.mark.parametrize(
+    "age,eligible",
+    [
+        (timedelta(), True),
+        (timedelta(days=28) - timedelta(microseconds=1), True),
+        (timedelta(days=28), False),
+        (timedelta(days=29), False),
+    ],
+)
+def test_notification_retry_keeps_identity_and_resets_the_lease_only_within_safe_window(
+    monkeypatch, age, eligible
+):
+    monkeypatch.setattr(job_retries, "utcnow", lambda: NOW)
+    delivery = models.NotificationDelivery(
+        id=ID,
+        status="failed",
+        created_at=NOW - age,
+        attempts=7,
+        error="upstream failure",
+        dispatched_at=NOW,
+        finished_at=NOW,
+        lease_until=NOW,
+        lease_token=uuid.uuid4(),
+    )
+    session = Mock()
+    session.scalar.return_value = delivery
+    if not eligible:
+        with pytest.raises(OperationConflict, match="safe retry window"):
+            job_retries.retry_notification(session, ID)
+        assert delivery.status == "failed" and delivery.attempts == 7
+        session.flush.assert_not_called()
+        return
+    assert job_retries.retry_notification(session, ID) is delivery
+    assert delivery.id == ID and delivery.status == "queued" and delivery.attempts == 0
+    assert delivery.available_at == NOW
+    assert all(
+        getattr(delivery, field) is None
+        for field in [
+            "dispatched_at",
+            "finished_at",
+            "lease_until",
+            "lease_token",
+            "error",
+        ]
+    )
+    session.flush.assert_called_once()
+
+
+@pytest.mark.parametrize("status", [None, "queued", "running", "succeeded"])
+def test_notification_retry_rejects_missing_or_nonfailed_delivery(status):
+    session = Mock()
+    session.scalar.return_value = None if status is None else SimpleNamespace(status=status)
+    with pytest.raises(RecordNotFound if status is None else OperationConflict):
+        job_retries.retry_notification(session, ID)
+    session.flush.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "kind,model,function,field",
+    [
+        ("ingestion", models.IngestionJob, "retry_job", "id"),
+        ("article-enrichment", models.ArticleEnrichmentJob, "retry_article", "id"),
+        ("images", models.ArticleImageJob, "retry_image", "id"),
+        ("source-enrichment", models.SourceEnrichmentJob, "request_enrichment", "source_id"),
+        ("analysis", models.ArticleAnalysisJob, "request_analysis", "article_id"),
+        ("topic-analysis", models.TopicAnalysisJob, "request_topic_analysis", "proposal_id"),
+        ("notifications", models.NotificationDelivery, "retry_notification", "id"),
+    ],
+)
+def test_failed_job_dispatches_to_the_correct_pipeline_and_subject(
+    monkeypatch, kind, model, function, field
+):
+    monkeypatch.setattr(job_retries, "get_settings", lambda: SimpleNamespace(ai_enabled=True))
+    values = {"id": ID, "status": "failed"}
+    if field != "id":
+        values[field] = uuid.UUID(int=20)
+    job = model(**values)
+    session = Mock()
+    session.scalar.return_value = ID
+    dispatch = Mock(return_value=object())
+    monkeypatch.setattr(job_retries, function, dispatch)
+    assert job_retries.retry_failed_job(session, job, kind, ACTOR) is dispatch.return_value
+    args = (session, getattr(job, field))
+    dispatch.assert_called_once_with(*args, *([ACTOR] if field == "proposal_id" else []))
+
+
+def test_relationship_retry_retains_the_original_target(monkeypatch):
+    monkeypatch.setattr(job_retries, "get_settings", lambda: SimpleNamespace(ai_enabled=True))
+    target = uuid.UUID(int=21)
+    job = models.TopicAnalysisJob(
+        id=ID,
+        status="failed",
+        topic_id=uuid.UUID(int=20),
+        input_snapshot={"related_topic_id": str(target)},
+    )
+    session = Mock()
+    session.scalar.return_value = ID
+    dispatch = Mock(return_value=object())
+    monkeypatch.setattr(job_retries, "request_relationship_analysis", dispatch)
+    assert (
+        job_retries.retry_failed_job(session, job, "topic-analysis", ACTOR) is dispatch.return_value
+    )
+    args = dispatch.call_args.args
+    assert args[0] is session and args[1] == job.topic_id and args[3] == ACTOR
+    assert args[2].related_topic_id == target
+
+
+@pytest.mark.parametrize(
+    "reason", ["not-failed", "superseded", "ai-disabled", "image-present", "unsupported"]
+)
+def test_retry_does_not_admit_unsafe_or_unnecessary_work(monkeypatch, reason):
+    monkeypatch.setattr(job_retries, "get_settings", lambda: SimpleNamespace(ai_enabled=False))
+    job = models.ArticleImageJob(id=ID, status="queued" if reason == "not-failed" else "failed")
+    session = Mock()
+    session.scalar.return_value = None if reason == "superseded" else ID
+    image_retry = Mock(return_value=None)
+    monkeypatch.setattr(job_retries, "retry_image", image_retry)
+    kind = (
+        "analysis"
+        if reason == "ai-disabled"
+        else "unknown"
+        if reason == "unsupported"
+        else "images"
+    )
+    with pytest.raises(OperationConflict):
+        job_retries.retry_failed_job(session, job, kind, ACTOR)
+    if reason == "not-failed":
+        session.scalar.assert_not_called()
+    if reason != "image-present":
+        image_retry.assert_not_called()
diff --git a/tests/test_mcp_oauth_unit.py b/tests/test_mcp_oauth_unit.py
new file mode 100644
index 00000000..3c759c1f
--- /dev/null
+++ b/tests/test_mcp_oauth_unit.py
@@ -0,0 +1,404 @@
+"""OAuth policy boundaries; Redis atomicity remains covered by integration tests."""
+
+import asyncio
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from urllib.parse import parse_qs, urlsplit
+
+import pytest
+from devfeed_mcp import oauth
+from mcp.server.auth.provider import (
+    AccessToken,
+    AuthorizationCode,
+    AuthorizationParams,
+    AuthorizeError,
+    RefreshToken,
+    RegistrationError,
+    TokenError,
+)
+from mcp.shared.auth import OAuthClientInformationFull
+
+RESOURCE = "https://mcp.example/mcp"
+NOW = 1_000_000
+
+
+def run(awaitable):
+    return asyncio.run(awaitable)
+
+
+@pytest.fixture
+def provider(monkeypatch):
+    records = {}
+
+    async def put(identifier, value, **_):
+        records[identifier] = value
+
+    async def remove(*identifiers):
+        for identifier in identifiers:
+            records.pop(identifier, None)
+
+    store = SimpleNamespace(
+        records=records,
+        get=AsyncMock(side_effect=records.get),
+        set=AsyncMock(side_effect=put),
+        getdel=AsyncMock(side_effect=lambda identifier: records.pop(identifier, None)),
+        delete=AsyncMock(side_effect=remove),
+        incr=AsyncMock(return_value=1),
+        expire=AsyncMock(),
+        eval=AsyncMock(),
+        zrem=AsyncMock(),
+    )
+    monkeypatch.setattr(oauth.time, "time", lambda: NOW)
+    return oauth.OAuthProvider(store, RESOURCE, "https://reader.example/", 900)
+
+
+def client(**changes):
+    return OAuthClientInformationFull(
+        **{
+            "client_id": "client",
+            "client_name": "Reader agent",
+            "token_endpoint_auth_method": "none",
+            "redirect_uris": ["https://agent.example/callback"],
+            **changes,
+        }
+    )
+
+
+def record(provider, kind, token, value):
+    provider.redis.records[oauth.key(kind, token)] = json.dumps(value)
+
+
+def grant(**changes):
+    return {
+        "client_id": "client",
+        "expires_at": NOW + 3600,
+        "absolute_expires_at": NOW + 90 * 86400,
+        "resource": RESOURCE,
+        "identity": {"user_id": "reader"},
+        "scopes": oauth.SCOPES,
+        **changes,
+    }
+
+
+def token_record(**changes):
+    return {
+        "client_id": "client",
+        "scopes": ["devfeed:read"],
+        "grant_id": "grant",
+        "expires_at": NOW + 900,
+        "resource": RESOURCE,
+        "subject": "reader",
+        **changes,
+    }
+
+
+@pytest.mark.parametrize(
+    "redirect",
+    ["https://agent.example/cb", "http://localhost/cb", "http://127.0.0.1/cb", "http://[::1]/cb"],
+)
+def test_public_registration_round_trips_and_hashes_storage_keys(provider, redirect):
+    account = client(redirect_uris=[redirect])
+    assert run(provider.get_client(account.client_id)) is None
+    run(provider.register_client(account))
+    assert run(provider.get_client(account.client_id)) == account
+    assert provider.redis.set.await_args.kwargs["ex"] == 90 * 86400
+    assert "client" not in oauth.key("access", "client").split(":")[-1]
+    assert provider.redis.expire.await_args.args[1] == 120
+
+
+@pytest.mark.parametrize(
+    "changes,error",
+    [
+        ({"token_endpoint_auth_method": "client_secret_post"}, "invalid_client_metadata"),
+        ({"redirect_uris": []}, "invalid_redirect_uri"),
+        ({"redirect_uris": ["https://agent.example/cb"] * 11}, "invalid_redirect_uri"),
+        ({"redirect_uris": ["http://agent.example/cb"]}, "invalid_redirect_uri"),
+        ({"redirect_uris": ["https://name:password@agent.example/cb"]}, "invalid_redirect_uri"),
+        ({"redirect_uris": ["https://agent.example/cb#fragment"]}, "invalid_redirect_uri"),
+        ({"redirect_uris": ["https://agent.example/" + "x" * 2000]}, "invalid_redirect_uri"),
+        ({"redirect_uris": ["custom:/callback"]}, "invalid_redirect_uri"),
+    ],
+)
+def test_registration_rejects_unsafe_or_unbounded_metadata(provider, changes, error):
+    with pytest.raises(RegistrationError) as caught:
+        run(provider.register_client(client(**changes)))
+    assert caught.value.error == error
+    provider.redis.set.assert_not_awaited()
+
+
+def test_registration_rate_limit_does_not_store_a_client(provider):
+    provider.redis.incr.return_value = 101
+    with pytest.raises(RegistrationError):
+        run(provider.register_client(client()))
+    provider.redis.set.assert_not_awaited()
+
+
+def parameters(**changes):
+    return AuthorizationParams(
+        **{
+            "redirect_uri": "https://agent.example/callback",
+            "redirect_uri_provided_explicitly": True,
+            "code_challenge": "challenge",
+            "state": "opaque state",
+            "scopes": None,
+            **changes,
+        }
+    )
+
+
+@pytest.mark.parametrize("named", [True, False])
+def test_authorization_stores_bounded_consent_request_with_approved_resource(provider, named):
+    account = client(client_name="Reader agent" if named else None)
+    location = run(provider.authorize(account, parameters()))
+    request_id = parse_qs(urlsplit(location).query)["request"][0]
+    pending = run(provider.get_record("pending", request_id))
+    assert urlsplit(location).path == "/mcp/authorize"
+    assert pending["resource"] == RESOURCE
+    assert pending["scopes"] == oauth.SCOPES
+    assert pending["client_name"] == ("Reader agent" if named else "MCP client")
+    assert pending["state"] == "opaque state"
+    assert provider.redis.set.await_args.kwargs["ex"] == 600
+
+
+@pytest.mark.parametrize(
+    "changes,error",
+    [
+        ({"resource": "https://other.example/mcp"}, "invalid_target"),
+        ({"scopes": ["devfeed:write"]}, "invalid_scope"),
+        ({"scopes": ["devfeed:read", "admin"]}, "invalid_scope"),
+    ],
+)
+def test_authorization_rejects_audience_and_scope_escalation(provider, changes, error):
+    with pytest.raises(AuthorizeError) as caught:
+        run(provider.authorize(client(), parameters(**changes)))
+    assert caught.value.error == error
+    provider.redis.set.assert_not_awaited()
+
+
+def test_authorization_rate_limit_is_per_client(provider):
+    provider.redis.incr.return_value = 21
+    with pytest.raises(AuthorizeError):
+        run(provider.authorize(client(), parameters()))
+    assert ":client:" in provider.redis.incr.await_args.args[0]
+    provider.redis.set.assert_not_awaited()
+
+
+def code_record(**changes):
+    return {
+        **parameters(scopes=["devfeed:read"]).model_dump(mode="json"),
+        "client_id": "client",
+        "grant_id": "grant",
+        "expires_at": NOW + 60,
+        **changes,
+    }
+
+
+def test_authorization_codes_are_client_bound_and_consumed_once(provider):
+    assert run(provider.load_authorization_code(client(), "code")) is None
+    record(provider, "code", "code", code_record(client_id="other"))
+    assert run(provider.load_authorization_code(client(), "code")) is None
+    record(provider, "code", "code", code_record())
+    code = run(provider.load_authorization_code(client(), "code"))
+    assert isinstance(code, AuthorizationCode)
+    provider.issue = AsyncMock(return_value="issued")
+    assert run(provider.exchange_authorization_code(client(), code)) == "issued"
+    provider.issue.assert_awaited_once_with("grant", "client", ["devfeed:read"])
+    with pytest.raises(TokenError):
+        run(provider.exchange_authorization_code(client(), code))
+
+
+@pytest.mark.parametrize(
+    "changes,error",
+    [
+        ({"client_id": "other"}, "invalid_grant"),
+        ({"expires_at": NOW}, "invalid_grant"),
+        ({"resource": "https://other.example"}, "invalid_scope"),
+        ({"scopes": []}, "invalid_scope"),
+    ],
+)
+def test_issuance_validates_grant_before_requesting_atomic_renewal(provider, changes, error):
+    record(provider, "grant", "grant", grant(**changes))
+    with pytest.raises(TokenError) as caught:
+        run(provider.issue("grant", "client", ["devfeed:read"]))
+    assert caught.value.error == error
+    provider.redis.eval.assert_not_awaited()
+
+
+@pytest.mark.parametrize("present", [False, True])
+def test_missing_or_concurrently_revoked_grants_never_issue_tokens(provider, present):
+    if present:
+        record(provider, "grant", "grant", grant())
+        provider.redis.eval.return_value = None
+    with pytest.raises(TokenError):
+        run(provider.issue("grant", "client", ["devfeed:read"]))
+    provider.redis.set.assert_not_awaited()
+
+
+@pytest.mark.parametrize("remaining,expected", [(3600, 900), (30, 30)])
+def test_access_token_lifetime_is_bounded_by_renewed_grant(provider, remaining, expected):
+    value = grant(expires_at=NOW + remaining)
+    record(provider, "grant", "grant", value)
+    provider.redis.eval.return_value = json.dumps(value)
+    issued = run(provider.issue("grant", "client", ["devfeed:read"]))
+    assert issued.expires_in == expected
+    assert issued.scope == "devfeed:read"
+    assert issued.access_token != issued.refresh_token
+    access = run(provider.get_record("access", issued.access_token))
+    refresh = run(provider.get_record("refresh", issued.refresh_token))
+    assert access["expires_at"] == NOW + expected
+    assert refresh["expires_at"] == value["expires_at"]
+    assert access["subject"] == "reader"
+    assert access["resource"] == RESOURCE
+    assert provider.redis.expire.await_args.args[1] >= 90 * 86400
+
+
+@pytest.mark.parametrize(
+    "present,owner", [(False, "client"), (False, "other"), (True, "other"), (True, "client")]
+)
+def test_refresh_loading_is_client_bound_and_replay_revokes_only_its_owner(
+    provider, present, owner
+):
+    record(provider, "grant", "grant", grant())
+    record(provider, "used-refresh", "token", token_record(client_id=owner))
+    if present:
+        record(provider, "refresh", "token", token_record(client_id=owner))
+    loaded = run(provider.load_refresh_token(client(), "token"))
+    if present and owner == "client":
+        assert isinstance(loaded, RefreshToken)
+    else:
+        assert loaded is None
+    assert (oauth.key("grant", "grant") not in provider.redis.records) == (
+        not present and owner == "client"
+    )
+
+
+@pytest.mark.parametrize(
+    "stage", ["missing", "replay", "missing-grant", "rotation-race", "success"]
+)
+def test_refresh_exchange_handles_revocation_and_single_use_rotation(provider, stage):
+    value = token_record()
+    token = RefreshToken.model_validate({**value, "token": "token"})
+    if stage != "missing":
+        record(provider, "grant", "grant", grant())
+    if stage == "replay":
+        record(provider, "used-refresh", "token", value)
+    elif stage in {"missing-grant", "rotation-race", "success"}:
+        record(provider, "refresh", "token", value)
+    if stage == "missing-grant":
+        provider.redis.records.pop(oauth.key("grant", "grant"))
+    provider.redis.eval.return_value = json.dumps(value) if stage == "success" else None
+    provider.issue = AsyncMock(return_value="renewed")
+    if stage == "success":
+        assert run(provider.exchange_refresh_token(client(), token, ["devfeed:read"])) == "renewed"
+        provider.issue.assert_awaited_once_with("grant", "client", ["devfeed:read"])
+    else:
+        with pytest.raises(TokenError):
+            run(provider.exchange_refresh_token(client(), token, ["devfeed:read"]))
+        provider.issue.assert_not_awaited()
+    if stage in {"replay", "rotation-race"}:
+        assert oauth.key("grant", "grant") not in provider.redis.records
+
+
+@pytest.mark.parametrize(
+    "stage", ["missing", "expired", "wrong-resource", "missing-grant", "expired-grant", "valid"]
+)
+def test_access_loading_requires_a_live_grant_and_matching_resource(provider, stage):
+    if stage != "missing":
+        record(
+            provider,
+            "access",
+            "token",
+            token_record(
+                expires_at=NOW if stage == "expired" else NOW + 900,
+                resource="https://other.example" if stage == "wrong-resource" else RESOURCE,
+            ),
+        )
+    if stage != "missing-grant":
+        record(
+            provider,
+            "grant",
+            "grant",
+            grant(expires_at=NOW if stage == "expired-grant" else NOW + 3600),
+        )
+    result = run(provider.load_access_token("token"))
+    assert isinstance(result, AccessToken) if stage == "valid" else result is None
+
+
+@pytest.mark.parametrize("kind,model", [("access", AccessToken), ("refresh", RefreshToken)])
+def test_token_revocation_removes_the_connection_without_affecting_other_grants(
+    provider, kind, model
+):
+    token = model.model_validate({**token_record(), "token": "token"})
+    record(provider, "grant", "grant", grant())
+    record(provider, "grant", "other", grant())
+    record(provider, kind, "token", token_record())
+    run(provider.revoke_token(token))
+    assert oauth.key("grant", "grant") not in provider.redis.records
+    assert oauth.key("grant", "other") in provider.redis.records
+    provider.redis.zrem.assert_awaited_once_with(oauth.key("connections", "reader"), "grant")
+    provider.redis.zrem.reset_mock()
+    run(provider.revoke_grant("missing"))
+    provider.redis.zrem.assert_not_awaited()
+    provider.redis.records.pop(oauth.key(kind, "token"))
+    run(provider.revoke_token(token))
+
+
+@pytest.mark.parametrize(
+    "method,path,body,status",
+    [
+        ("GET", "/token", b"", 200),
+        ("POST", "/token", f"resource={RESOURCE}".encode(), 200),
+        ("POST", "/token", b"grant_type=refresh_token", 200),
+        ("POST", "/token", b"resource=https://other.example", 400),
+        ("POST", "/token", f"resource={RESOURCE}&resource={RESOURCE}".encode(), 400),
+        ("POST", "/token", b"\xff", 400),
+        ("POST", "/register", b"x" * 32768, 200),
+        ("POST", "/register", b"x" * 32769, 413),
+    ],
+)
+def test_oauth_transport_bounds_chunked_bodies_and_preserves_downstream_receive(
+    method, path, body, status
+):
+    forwarded = []
+
+    async def app(scope, receive, send):
+        forwarded.append(await receive())
+        forwarded.append(await receive())
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+
+    split = len(body) // 2
+    messages = [
+        {"type": "http.request", "body": body[:split], "more_body": True},
+        {"type": "http.request", "body": body[split:], "more_body": False},
+        {"type": "http.disconnect"},
+    ]
+    receive = AsyncMock(side_effect=messages)
+    send = AsyncMock()
+    scope = {"type": "http", "method": method, "path": path}
+    run(oauth.OAuthPolicyMiddleware(app, RESOURCE)(scope, receive, send))
+    response = send.await_args_list[0].args[0]
+    assert response["status"] == status
+    if status != 200:
+        assert not forwarded
+    elif method == "POST":
+        assert forwarded == [
+            {"type": "http.request", "body": body, "more_body": False},
+            {"type": "http.disconnect"},
+        ]
+
+
+def test_oauth_disconnect_aborts_processing_and_non_http_scopes_pass_through():
+    app = AsyncMock()
+    middleware = oauth.OAuthPolicyMiddleware(app, RESOURCE)
+    run(
+        middleware(
+            {"type": "http", "method": "POST", "path": "/token"},
+            AsyncMock(return_value={"type": "http.disconnect"}),
+            AsyncMock(),
+        )
+    )
+    app.assert_not_awaited()
+    receive, send = AsyncMock(), AsyncMock()
+    run(middleware({"type": "lifespan"}, receive, send))
+    app.assert_awaited_once_with({"type": "lifespan"}, receive, send)
diff --git a/tests/test_pre_commit_checks.py b/tests/test_pre_commit_checks.py
index 7d3bf48e..92a48bb1 100644
--- a/tests/test_pre_commit_checks.py
+++ b/tests/test_pre_commit_checks.py
@@ -46,7 +46,17 @@ def test_shared_frontend_changes_test_both_apps(path):
 def test_python_changes_run_types_and_only_unit_tests(path):
     commands = hooks.checks_for({path})
     assert ("uv", "run", "--locked", "mypy") in commands
-    assert ("uv", "run", "--locked", "pytest", "-q", "-m", "not integration") in commands
+    assert (
+        "uv",
+        "run",
+        "--locked",
+        "pytest",
+        "-q",
+        "-m",
+        "not integration",
+        "--cov",
+        "--cov-report=term:skip-covered",
+    ) in commands
     assert not any(command[0] == "npm" for command in commands)
 
 
diff --git a/tests/test_publication_policy_unit.py b/tests/test_publication_policy_unit.py
new file mode 100644
index 00000000..2a6a4d1d
--- /dev/null
+++ b/tests/test_publication_policy_unit.py
@@ -0,0 +1,291 @@
+"""Publication authority, freshness, deterministic rejection, and decision idempotency."""
+
+import uuid
+from types import SimpleNamespace as NS
+from unittest.mock import Mock
+
+import pytest
+from devfeed_core import publication_policy as policy
+from devfeed_core.analysis import snapshot_hash, source_snapshot
+from devfeed_core.models import ArticlePublicationDecision
+
+ID, JOB = uuid.UUID(int=1), uuid.UUID(int=2)
+
+
+@pytest.fixture
+def scenario(monkeypatch):
+    settings = NS(full_automation=False)
+    monkeypatch.setattr(policy, "get_settings", lambda: settings)
+    monkeypatch.setattr(policy, "analysis_catalog_current", lambda *_: True)
+    monkeypatch.setattr(policy, "catalog", lambda _: {"topics": [], "tags": []})
+    article = NS(
+        id=ID,
+        canonical_url="https://example.com/article",
+        title="Python typing",
+        summary="Useful engineering evidence",
+        ai_summary=None,
+        metadata_source_type="feed",
+        content_type="tutorial",
+        content_format="article",
+        language="en",
+        editorial_revision=4,
+        review_status="pending",
+        publication_status="unpublished",
+        classification_provenance={
+            "analysis_id": str(JOB),
+            "developer_relevance": "relevant",
+            "page_kind": "article",
+        },
+        topic_links=[NS(role="primary", topic=NS(status="active"))],
+        origins=[
+            NS(
+                source=NS(
+                    id=ID,
+                    approval_status="approved",
+                    enabled=True,
+                    publication_policy="auto",
+                    publication_policy_revision=3,
+                )
+            )
+        ],
+    )
+    job = NS(
+        id=JOB,
+        status="succeeded",
+        outcome="applied",
+        editorial_revision=4,
+        input_hash=snapshot_hash(source_snapshot(article, None)),
+        result={"outcome": "ready", "developer_relevance": "relevant", "page_kind": "article"},
+    )
+    session = Mock()
+    session.get.return_value = None
+    session.scalar.return_value = None
+    return settings, article, job, session
+
+
+def test_current_ready_analysis_and_approved_auto_source_are_publishable(scenario):
+    _, article, job, session = scenario
+    value = policy.evaluate_publication(session, article, job, taxonomy={})
+    assert value["status"] == "would_publish" and value["reasons"] == []
+    assert value["mode"] == "auto" and value["source_policy_revision"] == 3
+    assert value["analysis_id"] == str(JOB) and value["policy_version"] == policy.POLICY_VERSION
+
+
+@pytest.mark.parametrize("change", ["disabled", "unapproved", "manual", "absent"])
+def test_ineligible_sources_cannot_authorize_publication(scenario, change):
+    _, article, job, session = scenario
+    source = article.origins[0].source
+    if change == "absent":
+        article.origins = []
+    else:
+        setattr(
+            source,
+            {
+                "disabled": "enabled",
+                "unapproved": "approval_status",
+                "manual": "publication_policy",
+            }[change],
+            {"disabled": False, "unapproved": "pending", "manual": "manual"}[change],
+        )
+    value = policy.evaluate_publication(session, article, job)
+    assert value["mode"] == "manual" and value["source_id"] is None
+    assert "source_policy_manual" in value["reasons"]
+
+
+def test_auto_source_wins_over_preview_in_a_stable_order(scenario):
+    _, article, job, session = scenario
+    preview = NS(
+        **{
+            **vars(article.origins[0].source),
+            "id": uuid.UUID(int=0),
+            "publication_policy": "preview",
+        }
+    )
+    article.origins.insert(0, NS(source=preview))
+    assert policy.evaluate_publication(session, article, job)["source_id"] == str(ID)
+
+
+@pytest.mark.parametrize(
+    "change,reason",
+    [
+        ("missing", "current_analysis_required"),
+        ("failed", "current_analysis_required"),
+        ("outcome", "current_analysis_required"),
+        ("hash", "current_analysis_required"),
+        ("revision", "current_analysis_required"),
+        ("provenance", "current_analysis_required"),
+        ("evidence", "insufficient_analysis_evidence"),
+        ("result", "analysis_uncertain"),
+        ("relevance", "analysis_uncertain"),
+        ("page", "analysis_uncertain"),
+        ("reviewed", "editorial_decision_exists"),
+        ("published", "editorial_decision_exists"),
+    ],
+)
+def test_stale_uncertain_and_previously_reviewed_articles_are_blocked(scenario, change, reason):
+    _, article, job, session = scenario
+    if change == "missing":
+        job = None
+    elif change == "provenance":
+        article.classification_provenance["analysis_id"] = "old"
+    elif change == "reviewed":
+        article.review_status = "approved"
+    elif change == "published":
+        article.publication_status = "published"
+    elif change in {"result", "relevance", "page"}:
+        job.result[
+            {"result": "outcome", "relevance": "developer_relevance", "page": "page_kind"}[change]
+        ] = "unknown"
+    else:
+        field, value = {
+            "failed": ("status", "failed"),
+            "outcome": ("outcome", "cancelled"),
+            "hash": ("input_hash", "stale"),
+            "revision": ("editorial_revision", 2),
+            "evidence": ("outcome", "insufficient_evidence"),
+        }[change]
+        setattr(job, field, value)
+    assert reason in policy.evaluate_publication(session, article, job)["reasons"]
+
+
+def test_empty_source_text_and_changed_catalog_block_automation(scenario, monkeypatch):
+    _, article, job, session = scenario
+    article.summary = "12345"
+    monkeypatch.setattr(policy, "analysis_catalog_current", lambda *_: False)
+    reasons = policy.evaluate_publication(session, article, job)["reasons"]
+    assert {"insufficient_source_text", "current_catalog_required"} <= set(reasons)
+
+
+def test_extracted_text_can_supply_evidence_when_feed_summary_is_empty(scenario):
+    _, article, job, session = scenario
+    article.summary = ""
+    content = NS(text="Extracted engineering evidence", method="readability")
+    session.get.return_value = content
+    job.input_hash = snapshot_hash(source_snapshot(article, content))
+    assert (
+        "insufficient_source_text"
+        not in policy.evaluate_publication(session, article, job)["reasons"]
+    )
+
+
+@pytest.mark.parametrize("full", [False, True])
+def test_manual_review_is_respected_unless_full_automation_is_explicit(scenario, full):
+    settings, article, job, session = scenario
+    settings.full_automation = full
+    session.scalar.side_effect = [None] if full else [None, ID]
+    value = policy.evaluate_publication(session, article, job)
+    assert ("human_review_required" in value["reasons"]) is not full
+    assert value["full_automation"] is full
+
+
+@pytest.mark.parametrize(
+    "mode,status,actions",
+    [("auto", "published", ["approve", "publish"]), ("preview", "would_publish", [])],
+)
+def test_preview_records_decision_while_auto_approves_before_publishing(
+    scenario, monkeypatch, mode, status, actions
+):
+    _, article, job, session = scenario
+    article.origins[0].source.publication_policy = mode
+    decide = Mock()
+    monkeypatch.setattr(policy, "decide_article", decide)
+    value = policy.apply_publication_policy(session, article, job)
+    assert value["status"] == status
+    assert [call.args[2].action for call in decide.call_args_list] == actions
+    assert all(call.args[2].expected_revision == 4 for call in decide.call_args_list)
+    saved = session.add.call_args.args[0]
+    assert isinstance(saved, ArticlePublicationDecision) and saved.decision == value
+    assert job.result["publication_policy"] == value
+
+
+def test_existing_fingerprint_reuses_decision_without_repeating_editorial_actions(
+    scenario, monkeypatch
+):
+    _, article, job, session = scenario
+    previous = {"status": "published"}
+    session.scalar.side_effect = [None, None, NS(decision=previous)]
+    decide = Mock()
+    monkeypatch.setattr(policy, "decide_article", decide)
+    assert policy.apply_publication_policy(session, article, job) is previous
+    decide.assert_not_called()
+    session.add.assert_not_called()
+
+
+@pytest.mark.parametrize("paywalled", [True, False, None])
+def test_only_explicit_successful_paywall_evidence_rejects_even_without_analysis(
+    scenario, monkeypatch, paywalled
+):
+    _, article, _, session = scenario
+    session.scalar.side_effect = [
+        NS(
+            id=JOB,
+            status="succeeded",
+            result={"paywalled": paywalled, "paywall_reason": "subscriber"},
+        ),
+        None,
+        None,
+    ]
+    decide = Mock()
+    monkeypatch.setattr(policy, "decide_article", decide)
+    value = policy.apply_publication_policy(session, article, None)
+    assert value["status"] == ("rejected" if paywalled is True else "blocked")
+    assert decide.call_count == int(paywalled is True)
+    if paywalled is True:
+        assert value["reasons"] == ["paywalled_content"]
+        assert "paywall" in decide.call_args.args[2].note
+
+
+@pytest.mark.parametrize("negative", ["unrelated", "non_article", "both"])
+def test_full_automation_rejects_current_explicit_negative_classification(
+    scenario, monkeypatch, negative
+):
+    settings, article, job, session = scenario
+    settings.full_automation = True
+    if negative in {"unrelated", "both"}:
+        job.result["developer_relevance"] = "unrelated"
+    if negative in {"non_article", "both"}:
+        job.result["page_kind"] = "non_article"
+    session.scalar.side_effect = [None, JOB, None]
+    decide = Mock()
+    monkeypatch.setattr(policy, "decide_article", decide)
+    value = policy.apply_publication_policy(session, article, job)
+    assert value["status"] == "rejected"
+    assert value["reasons"] == (
+        ["unrelated_content", "non_article_content"]
+        if negative == "both"
+        else ["unrelated_content" if negative == "unrelated" else "non_article_content"]
+    )
+    assert decide.call_args.args[2].action == "reject"
+
+
+@pytest.mark.parametrize(
+    "guard", ["stale", "superseded", "empty", "reviewed", "published", "manual", "catalog"]
+)
+def test_negative_results_cannot_override_freshness_or_existing_decisions(
+    scenario, monkeypatch, guard
+):
+    settings, article, job, session = scenario
+    settings.full_automation = True
+    job.result["developer_relevance"] = "unrelated"
+    if guard == "stale":
+        job.input_hash = "old"
+    elif guard == "empty":
+        article.summary = "123"
+        job.input_hash = snapshot_hash(source_snapshot(article, None))
+    elif guard in {"reviewed", "published"}:
+        setattr(
+            article,
+            "review_status" if guard == "reviewed" else "publication_status",
+            "approved" if guard == "reviewed" else "published",
+        )
+    elif guard == "manual":
+        article.origins = []
+    elif guard == "catalog":
+        monkeypatch.setattr(policy, "analysis_catalog_current", lambda *_: False)
+    session.scalar.side_effect = (
+        [None, None] if guard == "stale" else [None, ID if guard == "superseded" else JOB, None]
+    )
+    decide = Mock()
+    monkeypatch.setattr(policy, "decide_article", decide)
+    assert policy.apply_publication_policy(session, article, job)["status"] == "blocked"
+    decide.assert_not_called()
diff --git a/tests/test_reading_streaks_unit.py b/tests/test_reading_streaks_unit.py
new file mode 100644
index 00000000..4bc1ac55
--- /dev/null
+++ b/tests/test_reading_streaks_unit.py
@@ -0,0 +1,125 @@
+"""UTC streak projection policies; database uniqueness is covered by integration tests."""
+
+import uuid
+from datetime import UTC, datetime, timedelta, timezone
+from unittest.mock import Mock
+
+import pytest
+from devfeed_core import reading_streaks as ledger
+from devfeed_core.models import UserReadingDay, UserReadingStreak
+
+NOW = datetime(2026, 10, 4, 12, tzinfo=UTC)
+USER, ARTICLE = uuid.UUID(int=1), uuid.UUID(int=2)
+
+
+@pytest.fixture(autouse=True)
+def fixed_clock(monkeypatch):
+    monkeypatch.setattr(ledger, "utcnow", lambda: NOW)
+
+
+@pytest.mark.parametrize("days_ago,active", [(0, True), (1, True), (2, False), (-1, False)])
+def test_only_today_and_yesterday_keep_a_streak_active(days_ago, active):
+    streak = UserReadingStreak(
+        current_days=3,
+        longest_days=8,
+        total_days=15,
+        last_read_date=(NOW - timedelta(days=days_ago)).date(),
+    )
+    value = ledger.reading_streak_value(streak)
+    assert value == {
+        "current_days": 3 if active else 0,
+        "longest_days": 8,
+        "total_days": 15,
+        "last_read_date": streak.last_read_date,
+    }
+    assert ledger.reading_streak_value(streak, NOW.date()) == value
+
+
+def test_absent_streak_has_zero_counts():
+    assert ledger.reading_streak_value(None) == {
+        "current_days": 0,
+        "longest_days": 0,
+        "total_days": 0,
+        "last_read_date": None,
+    }
+
+
+@pytest.mark.parametrize("moment", [NOW.replace(tzinfo=None), NOW + timedelta(seconds=1)])
+def test_invalid_event_times_are_rejected_before_database_work(moment):
+    session = Mock()
+    with pytest.raises(ValueError, match="aware timestamp"):
+        ledger.record_reading_day(session, USER, ARTICLE, moment)
+    session.scalar.assert_not_called()
+
+
+def test_deleted_user_cannot_create_reading_history():
+    session = Mock()
+    session.scalar.return_value = None
+    with pytest.raises(ValueError, match="unavailable"):
+        ledger.record_reading_day(session, USER, ARTICLE)
+    assert session.scalar.call_count == 1
+
+
+def test_duplicate_open_does_not_change_streak_or_daily_counts():
+    session = Mock()
+    session.scalar.side_effect = [USER, None]
+    assert not ledger.record_reading_day(session, USER, ARTICLE)
+    session.get.assert_not_called()
+    session.execute.assert_not_called()
+
+
+@pytest.mark.parametrize("days_ago,expected", [(0, 4), (1, 5), (3, 1), (None, 1)])
+def test_distinct_articles_count_days_once_and_reset_after_a_gap(days_ago, expected):
+    streak = UserReadingStreak(
+        user_id=USER,
+        current_days=4,
+        longest_days=4,
+        total_days=10,
+        last_read_date=(NOW - timedelta(days=days_ago)).date() if days_ago is not None else None,
+    )
+    session = Mock()
+    session.scalar.side_effect = [USER, ARTICLE]
+    session.get.return_value = streak
+    assert ledger.record_reading_day(session, USER, ARTICLE)
+    assert streak.current_days == expected
+    assert streak.longest_days == max(4, expected)
+    assert streak.total_days == (10 if days_ago == 0 else 11)
+    assert streak.last_read_date == NOW.date() and streak.updated_at == NOW
+    session.flush.assert_called_once()
+
+
+@pytest.mark.parametrize("existing", [False, True])
+def test_missing_projection_or_out_of_order_event_rebuilds_history(monkeypatch, existing):
+    session = Mock()
+    session.scalar.side_effect = [USER, ARTICLE]
+    session.get.return_value = UserReadingStreak(last_read_date=NOW.date()) if existing else None
+    rebuild = Mock()
+    monkeypatch.setattr(ledger, "rebuild_reading_history", rebuild)
+    moment = datetime(2026, 10, 4, 1, tzinfo=timezone(timedelta(hours=5, minutes=30)))
+    assert ledger.record_reading_day(session, USER, ARTICLE, moment)
+    rebuild.assert_called_once_with(session, USER)
+    values = session.scalar.call_args.args[0].compile().params
+    assert values["read_date"].isoformat() == "2026-10-03"
+    assert values["occurred_at"].tzinfo == UTC
+
+
+@pytest.mark.parametrize(
+    "existing,offsets,counts",
+    [(False, [], (0, 0, 0)), (False, [0, 1, 2, 5], (1, 3, 4)), (True, [0, 1], (2, 2, 2))],
+)
+def test_rebuild_uses_retained_events_and_replaces_obsolete_projection(existing, offsets, counts):
+    session = Mock()
+    session.scalar.return_value = USER
+    rows = [((NOW + timedelta(days=i)).date(), i + 2, NOW + timedelta(days=i)) for i in offsets]
+    session.execute.return_value.all.return_value = rows
+    streak = UserReadingStreak(user_id=USER, current_days=99) if existing else None
+    session.get.return_value = streak
+    ledger.rebuild_reading_history(session, USER)
+    added = [call.args[0] for call in session.add.call_args_list]
+    projection = streak or next(value for value in added if isinstance(value, UserReadingStreak))
+    assert (projection.current_days, projection.longest_days, projection.total_days) == counts
+    assert projection.last_read_date == (rows[-1][0] if rows else None)
+    days = [value for value in added if isinstance(value, UserReadingDay)]
+    assert [(value.read_date, value.article_count, value.last_read_at) for value in days] == rows
+    assert "DELETE FROM user_reading_days" in str(session.execute.call_args.args[0])
+    session.flush.assert_called_once()
diff --git a/tests/test_relationship_coverage_unit.py b/tests/test_relationship_coverage_unit.py
new file mode 100644
index 00000000..dad2608e
--- /dev/null
+++ b/tests/test_relationship_coverage_unit.py
@@ -0,0 +1,210 @@
+"""Relationship coverage cursor safety, bounded retries, and scheduling admission."""
+
+import uuid
+from contextlib import nullcontext
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from devfeed_core import relationship_coverage as coverage
+from devfeed_core import topic_decision_budget
+from devfeed_core.analysis import snapshot_hash
+from devfeed_core.models import TopicRelationshipScan
+from devfeed_core.services import OperationConflict
+
+NOW = datetime(2026, 10, 4, tzinfo=UTC)
+ID, THROUGH = uuid.UUID(int=1), uuid.UUID(int=2)
+
+
+def scan(**changes):
+    return TopicRelationshipScan(
+        **{
+            "topic_id": ID,
+            "generation": 7,
+            "topic_snapshot": {"name": "Python"},
+            "failures": 0,
+            "job_id": uuid.UUID(int=3),
+            **changes,
+        }
+    )
+
+
+@pytest.mark.parametrize("failures,hours", [(0, 1), (1, 2), (4, 16), (5, 24), (20, 24)])
+def test_failed_batches_back_off_without_advancing_the_cursor(failures, hours):
+    value = scan(failures=failures, after_topic_id=ID)
+    coverage._completed_job(value, SimpleNamespace(status="failed", error="quota"), NOW)
+    assert value.failures == failures + 1 and value.last_error == "quota"
+    assert value.next_run_at == NOW + timedelta(hours=hours)
+    assert value.after_topic_id == ID and value.job_id is None
+
+
+@pytest.mark.parametrize(
+    "outcome,stale,full,proposals,advance",
+    [
+        ("cancelled", False, False, True, False),
+        ("enriched", True, False, True, False),
+        ("enriched", False, True, True, False),
+        ("enriched", False, True, False, True),
+        ("no_additions", False, False, False, True),
+    ],
+)
+def test_only_current_complete_batches_advance_the_cursor(outcome, stale, full, proposals, advance):
+    value = scan(failures=4, last_error="previous failure", after_topic_id=ID)
+    job = SimpleNamespace(
+        status="succeeded",
+        outcome=outcome,
+        input_snapshot={
+            "topic": {"name": "Old"} if stale else value.topic_snapshot,
+            "coverage": {"through_topic_id": str(THROUGH)},
+        },
+        result={
+            "relationships": list(range(20 if full else 2)),
+            "proposal_ids": ["proposal"] if proposals else [],
+        },
+    )
+    coverage._completed_job(value, job, NOW)
+    assert value.after_topic_id == (THROUGH if advance else ID)
+    assert value.job_id is None and value.next_run_at == NOW
+    assert value.failures == (4 if stale or outcome == "cancelled" else 0)
+
+
+@pytest.mark.parametrize("existing", [False, True])
+def test_identity_change_resets_progress_and_allocates_a_new_generation(monkeypatch, existing):
+    session, topic = Mock(), SimpleNamespace(id=ID)
+    session.scalar.return_value = 42
+    monkeypatch.setattr(coverage, "topic_snapshot", lambda _: {"name": "New"})
+    monkeypatch.setattr(coverage, "utcnow", lambda: NOW)
+    value = coverage.reset_scan(
+        session, topic, scan(failures=5, after_topic_id=THROUGH) if existing else None
+    )
+    assert value.generation == 42 and value.topic_snapshot == {"name": "New"}
+    assert value.after_topic_id is value.job_id is value.finished_at is value.last_error is None
+    assert value.failures == 0 and value.next_run_at == NOW
+    assert session.add.call_count == (0 if existing else 1)
+
+
+def test_oversized_batches_shrink_until_the_model_can_accept_them(monkeypatch):
+    request = Mock(
+        side_effect=[
+            OperationConflict("keep within the model limit"),
+            OperationConflict("within the model limit"),
+            "job",
+        ]
+    )
+    monkeypatch.setattr(coverage, "request_relationship_analysis", request)
+    peers = [uuid.UUID(int=i) for i in range(1, 6)]
+    assert coverage._request_batch(Mock(), SimpleNamespace(id=ID), peers) == ("job", peers[0])
+    assert [call.kwargs["candidate_ids"] for call in request.call_args_list] == [
+        peers,
+        peers[:2],
+        peers[:1],
+    ]
+
+
+@pytest.mark.parametrize(
+    "peers,message", [([ID], "within the model limit"), ([ID, THROUGH], "active job")]
+)
+def test_unrecoverable_admission_conflicts_propagate(monkeypatch, peers, message):
+    monkeypatch.setattr(
+        coverage, "request_relationship_analysis", Mock(side_effect=OperationConflict(message))
+    )
+    with pytest.raises(OperationConflict, match=message):
+        coverage._request_batch(Mock(), SimpleNamespace(id=ID), peers)
+
+
+@pytest.mark.parametrize(
+    "ai,research,allowance", [(False, True, True), (True, False, True), (True, True, False)]
+)
+def test_disabled_or_exhausted_automation_does_not_schedule(monkeypatch, ai, research, allowance):
+    session, factory = Mock(), Mock()
+    factory.begin.return_value = nullcontext(session)
+    monkeypatch.setattr(
+        coverage,
+        "get_settings",
+        lambda: SimpleNamespace(ai_enabled=ai, auto_research_relationships=research),
+    )
+    monkeypatch.setattr(topic_decision_budget, "relationship_allowance", lambda _: allowance)
+    assert coverage.schedule_relationship_coverage(factory) == {
+        "relationship_jobs_scheduled": 0,
+        "relationship_scans_completed": 0,
+    }
+    session.scalars.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "state",
+    [
+        "inactive",
+        "complete",
+        "no-slots",
+        "conflict",
+        "scheduled",
+        "new-scan",
+        "completed-batch",
+        "no-cursor",
+    ],
+)
+def test_scheduler_preserves_cursor_and_only_counts_admitted_work(monkeypatch, state):
+    session, factory = Mock(), Mock()
+    factory.begin.return_value = nullcontext(session)
+    value = scan(job_id=None, after_topic_id=THROUGH)
+    if state == "no-cursor":
+        value.after_topic_id = None
+    topic = SimpleNamespace(id=ID, status="inactive" if state == "inactive" else "active")
+    job = SimpleNamespace(id=uuid.UUID(int=9), input_snapshot={"topic": value.topic_snapshot})
+    settings = SimpleNamespace(
+        ai_enabled=True,
+        auto_research_relationships=True,
+        automation_batch_size=10,
+        relationship_research_max_pending=2,
+        relationship_research_batch_size=5,
+    )
+    monkeypatch.setattr(coverage, "get_settings", lambda: settings)
+    monkeypatch.setattr(coverage, "utcnow", lambda: NOW)
+    monkeypatch.setattr(topic_decision_budget, "relationship_allowance", lambda _: True)
+    monkeypatch.setattr(coverage, "lock_topics", Mock())
+    reset = Mock()
+    monkeypatch.setattr(coverage, "reset_scan", reset)
+    session.scalars.side_effect = [
+        SimpleNamespace(all=lambda: [topic] if state == "new-scan" else []),
+        SimpleNamespace(all=lambda: [value]),
+        [] if state == "complete" else [ID],
+    ]
+    completed_job = SimpleNamespace(
+        status="succeeded",
+        outcome="no_additions",
+        input_snapshot={
+            "topic": value.topic_snapshot,
+            "coverage": {"through_topic_id": str(THROUGH)},
+        },
+        result={},
+    )
+    session.execute.return_value.all.return_value = (
+        [(value, completed_job)] if state == "completed-batch" else []
+    )
+    session.scalar.return_value = 2 if state == "no-slots" else 0
+    session.get.return_value = topic
+    request = (
+        Mock(side_effect=OperationConflict("quota"))
+        if state == "conflict"
+        else Mock(return_value=(job, ID))
+    )
+    monkeypatch.setattr(coverage, "_request_batch", request)
+    result = coverage.schedule_relationship_coverage(factory)
+    assert result == {
+        "relationship_jobs_scheduled": int(
+            state in {"scheduled", "new-scan", "completed-batch", "no-cursor"}
+        ),
+        "relationship_scans_completed": int(state == "complete"),
+    }
+    assert value.after_topic_id == (None if state == "no-cursor" else THROUGH)
+    assert reset.call_count == int(state == "new-scan")
+    if state in {"scheduled", "new-scan", "completed-batch", "no-cursor"}:
+        assert value.job_id == job.id
+        assert job.input_snapshot["coverage"] == {"generation": 7, "through_topic_id": str(ID)}
+        assert job.input_hash == snapshot_hash(job.input_snapshot)
+    elif state == "conflict":
+        assert value.failures == 1 and value.next_run_at == NOW + timedelta(hours=1)
+    else:
+        request.assert_not_called()
diff --git a/tests/test_scheduler_unit.py b/tests/test_scheduler_unit.py
new file mode 100644
index 00000000..e399bd49
--- /dev/null
+++ b/tests/test_scheduler_unit.py
@@ -0,0 +1,263 @@
+"""Scheduler isolation, queue cleanup, recovery routing, and heartbeat safety."""
+
+import logging
+import sys
+import uuid
+from contextlib import contextmanager, nullcontext
+from datetime import UTC, datetime
+from types import SimpleNamespace as NS
+from unittest.mock import Mock
+
+import pytest
+from devfeed_aggregator import quota_monitor, scheduler
+from devfeed_core import feed_notifications, job_retention
+from devfeed_notifications import delivery
+
+NOW = datetime(2026, 10, 4, tzinfo=UTC)
+
+
+@pytest.mark.parametrize("batch", [1, 5, 10])
+@pytest.mark.parametrize("fail", [False, True])
+def test_article_lanes_reserve_capacity_and_close_connections_on_failure(monkeypatch, batch, fail):
+    queues = [Mock(), Mock()]
+    lookup = Mock(side_effect=queues)
+    dispatch = Mock(side_effect=RuntimeError("queue unavailable")) if fail else Mock(return_value=2)
+    monkeypatch.setattr(scheduler, "get_queue", lookup)
+    monkeypatch.setattr(scheduler, "dispatch_jobs", dispatch)
+    if fail:
+        with pytest.raises(RuntimeError):
+            scheduler.dispatch_article_lanes("factory", "analysis", batch, NOW)
+    else:
+        assert scheduler.dispatch_article_lanes("factory", "analysis", batch, NOW) == 4
+        assert [call.args[0] for call in lookup.call_args_list] == [
+            "article-analysis-fresh",
+            "article-analysis",
+        ]
+    assert dispatch.call_args.args[2] == max(1, batch // 2)
+    queues[0].connection.close.assert_called_once()
+    assert queues[1].connection.close.call_count == int(not fail)
+
+
+@pytest.mark.parametrize(
+    "field,level",
+    [
+        (None, logging.DEBUG),
+        ("scheduled", logging.INFO),
+        ("recovered", logging.WARNING),
+        ("images_recovered", logging.WARNING),
+        ("profiles_recovered", logging.WARNING),
+        ("articles_recovered", logging.WARNING),
+    ],
+)
+def test_tick_logs_work_and_recovery_at_the_correct_severity(monkeypatch, caplog, field, level):
+    counts = {"recovered": 0, "images_recovered": 0, "scheduled": 0}
+    if field:
+        counts[field] = 1
+    monkeypatch.setattr(scheduler, "_tick", lambda: counts)
+    monkeypatch.setattr(scheduler, "background_cycle", lambda _: nullcontext())
+    with caplog.at_level(logging.DEBUG, logger=scheduler.__name__):
+        assert scheduler.tick() == counts
+    record = next(
+        record for record in caplog.records if record.message == "scheduler_tick_completed"
+    )
+    assert record.levelno == level and record.duration_ms >= 0
+
+
+def test_failed_tick_is_logged_and_propagates(monkeypatch, caplog):
+    monkeypatch.setattr(scheduler, "_tick", Mock(side_effect=RuntimeError("tick failed")))
+    monkeypatch.setattr(scheduler, "background_cycle", lambda _: nullcontext())
+    with pytest.raises(RuntimeError), caplog.at_level(logging.ERROR):
+        scheduler.tick()
+    assert any(record.message == "scheduler_tick_failed" for record in caplog.records)
+
+
+@pytest.mark.parametrize(
+    "kind",
+    [
+        "analysis",
+        "topic-analysis",
+        "research-verification",
+        "images",
+        "source-enrichment",
+        "article-enrichment",
+    ],
+)
+def test_expired_jobs_use_their_pipeline_recovery_policy(monkeypatch, kind):
+    job = NS(id=uuid.UUID(int=1), status="running")
+    session, factory = Mock(), Mock()
+    session.scalars.return_value.all.return_value = [job]
+    factory.begin.return_value = nullcontext(session)
+    handlers = {name: Mock() for name in ("fail_analysis", "fail_verification", "fail_or_retry")}
+    for name, handler in handlers.items():
+        monkeypatch.setattr(scheduler, name, handler)
+    assert scheduler.recover_jobs(factory, 10, NOW, kind=kind) == 1
+    selected = (
+        "fail_verification"
+        if kind == "research-verification"
+        else "fail_analysis"
+        if kind in {"analysis", "topic-analysis"}
+        else "fail_or_retry"
+    )
+    assert {name: handler.call_count for name, handler in handlers.items()} == {
+        name: int(name == selected) for name in handlers
+    }
+    assert handlers[selected].call_args.args[0] is job
+
+
+def test_unsupported_recovery_requires_an_explicit_policy():
+    factory = Mock()
+    with pytest.raises(ValueError, match="own recovery policy"):
+        scheduler.recover_jobs(factory, 10, NOW, kind="ingestion")
+    factory.begin.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "mode",
+    [
+        "minimal",
+        "all",
+        "retention-failed",
+        "overview-failed",
+        "dispatch-failed",
+        "topics-only",
+        "relationships-only",
+    ],
+)
+def test_tick_dispatches_outside_transactions_and_preserves_cleanup(monkeypatch, mode):
+    session, factory = Mock(), Mock()
+    opened = 0
+
+    @contextmanager
+    def transaction():
+        nonlocal opened
+        opened += 1
+        try:
+            yield session
+        finally:
+            opened -= 1
+
+    factory.begin.side_effect = transaction
+    expired = NS(id=uuid.UUID(int=2), source_id=uuid.UUID(int=3), status="running", attempts=1)
+    source = NS(id=uuid.UUID(int=4))
+    session.scalars.side_effect = [
+        NS(all=lambda: [uuid.UUID(int=1), expired.id]),
+        NS(all=lambda: [source]),
+    ]
+    session.scalar.side_effect = [None, expired]
+    settings = NS(
+        scheduler_batch_size=5,
+        notifications_enabled=mode == "all",
+        solver_queue_enabled=mode == "all",
+        ai_enabled=mode != "minimal",
+        auto_approve_topics=mode in {"all", "topics-only"},
+        auto_approve_topic_relationships=mode in {"all", "relationships-only"},
+    )
+    monkeypatch.setattr(scheduler, "get_settings", lambda: settings)
+    monkeypatch.setattr(scheduler, "session_factory", lambda: factory)
+    monkeypatch.setattr(scheduler, "utcnow", lambda: NOW)
+    monkeypatch.setattr(quota_monitor, "refresh_quota", Mock())
+    refresh = Mock(side_effect=RuntimeError("overview")) if mode == "overview-failed" else Mock()
+    monkeypatch.setattr(scheduler, "refresh_overview_daily", refresh)
+    prune = Mock()
+    monkeypatch.setattr(scheduler, "prune_article_opens", prune)
+    monkeypatch.setattr(
+        job_retention,
+        "prune_job_payloads",
+        Mock(side_effect=RuntimeError("retention")) if mode == "retention-failed" else Mock(),
+    )
+    monkeypatch.setattr(scheduler, "expand_recommendation_events", Mock(return_value=3))
+    monkeypatch.setattr(
+        scheduler, "schedule_automation", Mock(return_value={"automation_scheduled": 4})
+    )
+    monkeypatch.setattr(scheduler, "schedule_verification", Mock(return_value=2))
+    monkeypatch.setattr(scheduler, "fail_job", Mock())
+    monkeypatch.setattr(scheduler, "request_ingestion", Mock(return_value=NS(id=uuid.UUID(int=5))))
+    monkeypatch.setattr(scheduler, "recover_jobs", Mock(return_value=1))
+    monkeypatch.setattr(feed_notifications, "expand_feed_notifications", Mock())
+    monkeypatch.setattr(delivery, "recover_notifications", Mock(return_value=2))
+    monkeypatch.setattr(scheduler, "dispatch_discovery", Mock())
+    monkeypatch.setattr(scheduler, "dispatch_recommendations", Mock(return_value=3))
+    queues = []
+
+    def queue(*_):
+        result = Mock()
+        queues.append(result)
+        return result
+
+    def dispatch(*_, **kwargs):
+        assert opened == 0, "Redis dispatch must not hold database transactions"
+        if mode == "dispatch-failed":
+            raise RuntimeError("dispatch")
+        return 1
+
+    def lanes(_, __, ___, ____, definitions):
+        assert opened == 0
+        return {lane.kind: 1 for lane in definitions}
+
+    monkeypatch.setattr(scheduler, "get_queue", queue)
+    monkeypatch.setattr(scheduler, "dispatch_jobs", dispatch)
+    monkeypatch.setattr(scheduler, "dispatch_lanes", lanes)
+    if mode == "dispatch-failed":
+        with pytest.raises(RuntimeError, match="dispatch"):
+            scheduler._tick()
+    else:
+        result = scheduler._tick()
+        assert result["recovered"] == result["scheduled"] == result["dispatched"] == 1
+        assert result["automation_scheduled"] == 4 and result["recommendation_users_queued"] == 3
+        assert result["analyses_dispatched"] == (0 if mode == "minimal" else 2)
+        assert result["notifications_dispatched"] == int(mode == "all")
+        assert result["notifications_recovered"] == (2 if mode == "all" else 0)
+        assert result["verifications_dispatched"] == int(
+            mode in {"all", "topics-only", "relationships-only"}
+        )
+        heartbeat_queue = next(item for item in queues if item.connection.set.call_count)
+        heartbeat_queue.connection.set.assert_called_once_with(
+            "devfeed:scheduler:heartbeat", NOW.isoformat(), ex=120
+        )
+    assert prune.call_count == int(mode != "overview-failed")
+    assert all(item.connection.close.call_count == 1 for item in queues)
+
+
+@pytest.mark.parametrize("fails", [False, True])
+def test_scheduler_heartbeat_only_advances_after_successful_cycles(monkeypatch, fails):
+    stop, health, telemetry = Mock(), Mock(), object()
+    stop.is_set.side_effect = [False, True]
+    health_context = Mock()
+    health_context.return_value = nullcontext(health)
+    monkeypatch.setattr(scheduler, "get_settings", lambda: NS(log_level="INFO", log_format="json"))
+    monkeypatch.setattr(scheduler, "configure_logging", Mock())
+    monkeypatch.setattr(scheduler, "start_runtime", lambda _: telemetry)
+    finish = Mock()
+    monkeypatch.setattr(scheduler, "stop_runtime", finish)
+    monkeypatch.setattr(scheduler.threading, "Event", lambda: stop)
+    signal_handler = Mock()
+    monkeypatch.setattr(scheduler.signal, "signal", signal_handler)
+    monkeypatch.setattr(scheduler, "scheduler_health", health_context)
+    monkeypatch.setattr(scheduler, "session_factory", Mock())
+    monkeypatch.setattr(scheduler, "recommendation_dispatcher", lambda *_: nullcontext())
+    monkeypatch.setattr(
+        scheduler, "tick", Mock(side_effect=RuntimeError("tick")) if fails else Mock()
+    )
+    scheduler.run()
+    assert health.completed.call_count == int(not fails)
+    stop.wait.assert_called_once_with(15)
+    finish.assert_called_once_with(telemetry)
+    signal_handler.call_args.args[1](None, None)
+    stop.set.assert_called_once()
+
+
+@pytest.mark.parametrize("mode", ["once", "failed", "daemon"])
+def test_scheduler_cli_reports_failed_once_runs(monkeypatch, mode):
+    monkeypatch.setattr(sys, "argv", ["scheduler"] + ([] if mode == "daemon" else ["--once"]))
+    monkeypatch.setattr(scheduler, "get_settings", lambda: NS(log_level="INFO", log_format="json"))
+    monkeypatch.setattr(scheduler, "configure_logging", Mock())
+    tick, run = Mock(side_effect=RuntimeError("tick")) if mode == "failed" else Mock(), Mock()
+    monkeypatch.setattr(scheduler, "tick", tick)
+    monkeypatch.setattr(scheduler, "run", run)
+    if mode == "failed":
+        with pytest.raises(SystemExit) as exc:
+            scheduler.main()
+        assert exc.value.code == 1
+    else:
+        scheduler.main()
+    assert run.call_count == int(mode == "daemon") and tick.call_count == int(mode != "daemon")
diff --git a/tests/test_search_indexer.py b/tests/test_search_indexer.py
index 8511ec22..a7d3edff 100644
--- a/tests/test_search_indexer.py
+++ b/tests/test_search_indexer.py
@@ -1,7 +1,9 @@
 from contextlib import nullcontext
 from threading import Event
 from types import SimpleNamespace
+from unittest.mock import Mock
 
+import pytest
 from devfeed_search_indexer import runtime
 
 
@@ -36,3 +38,76 @@ def fail(*_):
 
     assert runtime._consume_index(Event(), object(), True, output.append) == 1
     assert output == []
+
+
+@pytest.mark.parametrize("once", [False, True])
+@pytest.mark.parametrize("fails", [False, True])
+def test_runtime_stops_telemetry_and_registers_shutdown_signals(monkeypatch, once, fails):
+    engine, telemetry, stop = Mock(), object(), Mock()
+    monkeypatch.setattr(
+        runtime, "get_settings", lambda: SimpleNamespace(log_level="INFO", log_format="json")
+    )
+    monkeypatch.setattr(runtime, "configure_logging", Mock())
+    monkeypatch.setattr(runtime.threading, "Event", lambda: stop)
+    handlers = Mock()
+    monkeypatch.setattr(runtime.signal, "signal", handlers)
+    monkeypatch.setattr(runtime, "Typesense", lambda **_: engine)
+    start, finish = Mock(return_value=telemetry), Mock()
+    monkeypatch.setattr(runtime, "start_runtime", start)
+    monkeypatch.setattr(runtime, "stop_runtime", finish)
+    monkeypatch.setattr(
+        runtime,
+        "_consume_index",
+        Mock(side_effect=RuntimeError("sync")) if fails else Mock(return_value=0),
+    )
+    if fails:
+        with pytest.raises(RuntimeError):
+            runtime.run_indexer(once=once)
+    else:
+        assert runtime.run_indexer(once=once) == 0
+    engine.setup.assert_called_once()
+    assert start.call_count == int(not once)
+    finish.assert_called_once_with(None if once else telemetry)
+    handlers.call_args.args[1](None, None)
+    stop.set.assert_called_once()
+
+
+@pytest.mark.parametrize("fails", [False, True])
+@pytest.mark.parametrize("count", [0, 7])
+def test_reconciliation_failure_does_not_stop_sync_and_interval_is_respected(
+    monkeypatch, tmp_path, fails, count
+):
+    stop = Mock()
+    stop.is_set.side_effect = [False, False, True]
+    monkeypatch.setattr(
+        runtime, "get_settings", lambda: SimpleNamespace(search_reconcile_interval_seconds=60)
+    )
+    monkeypatch.setattr(runtime, "current", lambda: object())
+    monkeypatch.setattr(runtime.time, "monotonic", Mock(return_value=10))
+    monkeypatch.setattr(runtime, "background_cycle", lambda _: nullcontext())
+    monkeypatch.setattr(runtime, "session_factory", Mock())
+    reconcile = Mock(side_effect=RuntimeError("reconcile")) if fails else Mock()
+    monkeypatch.setattr(runtime, "reconcile", reconcile)
+    sync = Mock(return_value=count)
+    monkeypatch.setattr(runtime, "sync_batch", sync)
+    heartbeat = tmp_path / "heartbeat"
+    monkeypatch.setattr(runtime, "Path", lambda _: heartbeat)
+    assert runtime._consume_index(stop, object(), False, Mock()) == 0
+    assert reconcile.call_count == 1 and sync.call_count == 2
+    assert [call.args for call in stop.wait.call_args_list] == ([(1,), (1,)] if count == 0 else [])
+    assert heartbeat.is_file()
+
+
+def test_sync_failure_backs_off_without_writing_a_heartbeat(monkeypatch, tmp_path):
+    stop = Mock()
+    stop.is_set.side_effect = [False, True]
+    monkeypatch.setattr(runtime, "get_settings", lambda: SimpleNamespace())
+    monkeypatch.setattr(runtime, "current", lambda: None)
+    monkeypatch.setattr(runtime, "background_cycle", lambda _: nullcontext())
+    monkeypatch.setattr(runtime, "session_factory", Mock())
+    monkeypatch.setattr(runtime, "sync_batch", Mock(side_effect=RuntimeError("sync")))
+    heartbeat = tmp_path / "heartbeat"
+    monkeypatch.setattr(runtime, "Path", lambda _: heartbeat)
+    assert runtime._consume_index(stop, object(), False, Mock()) == 0
+    stop.wait.assert_called_once_with(5)
+    assert not heartbeat.exists()
diff --git a/tests/test_search_records_unit.py b/tests/test_search_records_unit.py
new file mode 100644
index 00000000..7cbe12fd
--- /dev/null
+++ b/tests/test_search_records_unit.py
@@ -0,0 +1,219 @@
+"""Public search projection, bounded documents, and suggestion admission policies."""
+
+import hashlib
+import uuid
+from datetime import UTC, datetime
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from devfeed_core import search_records, search_suggestions, topic_logos
+
+ID = uuid.UUID(int=1)
+SECOND = uuid.UUID(int=2)
+NOW = datetime(2026, 10, 4, tzinfo=UTC)
+
+
+@pytest.mark.parametrize("kind", search_records.MODELS)
+def test_empty_record_requests_perform_no_database_work(kind):
+    session = Mock()
+    assert search_records.public_records(session, kind, []) == {}
+    session.execute.assert_not_called()
+
+
+@pytest.mark.parametrize("kind", search_records.MODELS)
+def test_record_hydration_is_keyed_by_identity_and_returns_plain_values(kind):
+    session = Mock()
+    session.execute.return_value.mappings.return_value = [{"id": ID, "name": "Python"}]
+    assert search_records.public_records(session, kind, [ID]) == {ID: {"id": ID, "name": "Python"}}
+    assert session.execute.call_count == 1
+
+
+@pytest.mark.parametrize("kind", ["articles", "topics", "sources", "tags"])
+def test_public_hits_strip_markup_encode_paths_and_preserve_available_metadata(monkeypatch, kind):
+    monkeypatch.setattr(
+        topic_logos,
+        "get_settings",
+        lambda: SimpleNamespace(image_public_url="https://images.example"),
+    )
+    row = {
+        "id": ID,
+        "name": "C++",
+        "slug": "c++/guide",
+        "title": "Readable title",
+        "description": "<b>Original</b>",
+        "ai_description": "<b>Preferred</b>",
+        "feed_at": NOW,
+        "content_type": "tutorial",
+        "logo_url": "https://source.example/logo",
+        "managed_logo": {"variants": [{"key": "64.png", "width": 64}]},
+    }
+    value = search_records.hit(kind, row)
+    assert value["id"] == str(ID) and value["title"] == "Readable title"
+    assert value["description"] == "Preferred" and value["label"] == "tutorial"
+    assert value["href"] == f"/{kind}/c%2B%2B%2Fguide"
+    assert value["published_at"] == NOW.isoformat()
+    assert value["image_url"] == (
+        "https://images.example/64.png" if kind == "topics" else row["logo_url"]
+    )
+
+
+def test_sparse_hits_have_safe_fallbacks_and_bounded_descriptions():
+    value = search_records.hit(
+        "sources", {"id": ID, "name": "Source", "summary": "<p>" + "x" * 500 + "</p>"}
+    )
+    assert value["title"] == "Source" and len(value["description"]) == 320
+    assert value["href"] == f"/sources/{ID}"
+    assert value["published_at"] is None and value["image_url"] is None
+    assert value["logo_variants"] == []
+
+
+def test_article_documents_deduplicate_bound_terms_and_sort_related_identifiers(monkeypatch):
+    row = {
+        "id": ID,
+        "title": "Article",
+        "slug": "article",
+        "summary": "<b>Summary</b>",
+        "aliases": ["Python", "Python", ""],
+        "keywords": ["typing"],
+        "feed_at": NOW,
+        "author": "Reader",
+        "content_type": "tutorial",
+    }
+    monkeypatch.setattr(search_records, "public_records", lambda *_: {ID: row})
+    session = Mock()
+    session.execute.side_effect = [
+        [(ID, SECOND, "Python", "python", ["Py", "Python"])],
+        [(ID, SECOND, "Engineering")],
+        [(ID, SECOND, "Typing", "typing", ["static typing"])],
+    ]
+    documents, removed = search_records.documents(session, "articles", [ID, SECOND])
+    assert removed == {SECOND}
+    assert len(documents) == 1
+    value = documents[0]
+    assert value["description"] == "Summary"
+    assert value["terms"] == [
+        "Python",
+        "python",
+        "Py",
+        "Engineering",
+        "Typing",
+        "typing",
+        "static typing",
+        "article",
+    ]
+    assert value["topics"] == value["sources"] == value["tags"] == [str(SECOND)]
+    assert value["published_at"] == int(NOW.timestamp())
+    assert value["author"] == "Reader" and value["content_type"] == "tutorial"
+
+
+def test_non_article_documents_are_bounded_and_do_not_request_article_links(monkeypatch):
+    monkeypatch.setattr(
+        search_records,
+        "public_records",
+        lambda *_: {
+            ID: {
+                "id": ID,
+                "name": "Topic",
+                "description": "d" * 14000,
+                "aliases": [f"term-{index}-" + "x" * 250 for index in range(400)],
+            }
+        },
+    )
+    session = Mock()
+    documents, removed = search_records.documents(session, "topics", [ID])
+    value = documents[0]
+    assert removed == set() and value["title"] == "Topic"
+    assert len(value["description"]) == 12000
+    assert len(value["terms"]) == 300 and all(len(term) <= 200 for term in value["terms"])
+    assert value["published_at"] == 0 and value["author"] == ""
+    assert value["content_type"] == "article"
+    assert value["topics"] == value["sources"] == value["tags"] == []
+    session.execute.assert_not_called()
+
+
+def test_absent_documents_return_tombstones_without_loading_relations(monkeypatch):
+    monkeypatch.setattr(search_records, "public_records", lambda *_: {})
+    session = Mock()
+    assert search_records.documents(session, "articles", [ID]) == ([], {ID})
+    session.execute.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "query,expected",
+    [
+        ("  Python\tAsync ", "python async"),
+        ("ＰＹＴＨＯＮ", "python"),
+        ("Straße", "strasse"),
+        ("x" * 200, "x" * 200),
+    ],
+)
+def test_query_normalization_handles_unicode_and_exact_length_boundary(query, expected):
+    assert search_suggestions.normalize_query(query) == expected
+    assert search_suggestions.query_hash(expected) == hashlib.sha256(expected.encode()).hexdigest()
+
+
+@pytest.mark.parametrize("query", ["", " \t ", "!!!", "x" * 201, "word " * 21])
+def test_invalid_search_queries_never_write_aggregates(query):
+    session = Mock()
+    with pytest.raises(ValueError):
+        search_suggestions.record_successful_query(session, query)
+    session.execute.assert_not_called()
+
+
+def test_aggregate_write_has_only_normalized_query_and_count_fields():
+    session = Mock()
+    digest = search_suggestions.record_successful_query(session, "  Ｐython ")
+    assert digest == search_suggestions.query_hash("python")
+    values = session.execute.call_args.args[0].compile().params
+    assert values["query"] == "python" and values["query_hash"] == digest
+    assert values["successful_count"] == 1
+    assert not {"user_id", "ip", "session_id"}.intersection(values)
+
+
+@pytest.mark.parametrize("prefix", ["", " \t "])
+def test_empty_suggestion_prefix_avoids_database_reads(prefix):
+    session = Mock()
+    assert search_suggestions.approved_suggestions(session, prefix, limit=5, minimum=2) == []
+    session.scalars.assert_not_called()
+
+
+def test_suggestions_escape_literal_wildcards_and_retain_the_volume_limit():
+    session = Mock()
+    session.scalars.return_value = ["python_%", "python_guide"]
+    assert search_suggestions.approved_suggestions(session, " PYTHON_% ", limit=5, minimum=2) == [
+        "python_%",
+        "python_guide",
+    ]
+    values = session.scalars.call_args.args[0].compile().params
+    assert "python/_/%" in values.values() and 5 in values.values() and 2 in values.values()
+
+
+@pytest.mark.parametrize("digest", ["x" * 64, "A" * 64, "a" * 63, "a" * 65])
+def test_moderation_rejects_invalid_hashes_before_lookup(digest):
+    session = Mock()
+    with pytest.raises(ValueError, match="Invalid query hash"):
+        search_suggestions.review_query(
+            session, digest, status="approved", reviewer="admin", note=None
+        )
+    session.get.assert_not_called()
+
+
+def test_moderation_rejects_missing_queries_and_records_review_actor_and_note():
+    session = Mock()
+    digest = "a" * 64
+    session.get.return_value = None
+    with pytest.raises(LookupError):
+        search_suggestions.review_query(
+            session, digest, status="approved", reviewer="admin", note=None
+        )
+    row = SimpleNamespace()
+    session.get.return_value = row
+    assert (
+        search_suggestions.review_query(
+            session, digest, status="rejected", reviewer="admin", note="private"
+        )
+        is row
+    )
+    assert row.status == "rejected" and row.reviewed_by == "admin" and row.review_note == "private"
+    assert row.reviewed_at.tzinfo == UTC
diff --git a/tests/test_sonar_coverage.py b/tests/test_sonar_coverage.py
index fb005670..ae06e286 100644
--- a/tests/test_sonar_coverage.py
+++ b/tests/test_sonar_coverage.py
@@ -81,11 +81,20 @@ def test_prepare_rejects_missing_shards(tmp_path):
         prepare_reports(tmp_path)
 
 
-def test_prepare_combines_complementary_shard_coverage(tmp_path):
-    (tmp_path / "pyproject.toml").write_text("[tool.coverage.run]\nrelative_files = true\n")
+@pytest.mark.parametrize("backend_roots", [False, True])
+def test_prepare_combines_complementary_shard_coverage(tmp_path, backend_roots):
+    config = "[tool.coverage.run]\nrelative_files = true\n"
+    if backend_roots:
+        config += 'source = ["apps/api/src", "apps/user-api/src"]\n'
+    (tmp_path / "pyproject.toml").write_text(config)
     source = tmp_path / "apps/api/src/main.py"
     source.parent.mkdir(parents=True)
     source.write_text("def choose(value):\n    if value:\n        return 1\n    return 0\n")
+    additional = ("apps/user-api/src/main.py", "scripts/ci/helper.py")
+    for filename in additional:
+        source = tmp_path / filename
+        source.parent.mkdir(parents=True)
+        source.write_text("value = 1\n")
     arcs = [
         {(-1, 1), (1, -1)},
         {(-1, 2), (2, 3), (3, -1)},
@@ -96,7 +105,10 @@ def test_prepare_combines_complementary_shard_coverage(tmp_path):
         path = tmp_path / "reports/coverage" / suite / "coverage.db"
         path.parent.mkdir(parents=True)
         data = CoverageData(basename=str(path))
-        data.add_arcs({"apps/api/src/main.py": covered})
+        measured = {"apps/api/src/main.py": covered}
+        if suite == "python-unit":
+            measured.update({filename: {(-1, 1), (1, -1)} for filename in additional})
+        data.add_arcs(measured)
         data.write()
     for frontend in FRONTENDS:
         source = tmp_path / "apps" / frontend / "src/page.ts"
@@ -119,8 +131,9 @@ def test_prepare_combines_complementary_shard_coverage(tmp_path):
         text=True,
     )
     report = ET.parse(tmp_path / "reports/coverage/python.xml")
-    measured = report.find(".//class")
-    assert measured.get("filename") == "apps/api/src/main.py"
+    classes = {entry.get("filename"): entry for entry in report.findall(".//class")}
+    assert set(classes) == {"apps/api/src/main.py", *additional}
+    measured = classes["apps/api/src/main.py"]
     assert {line.get("number") for line in measured.findall("./lines/line")} == {"1", "2", "3", "4"}
     assert all(line.get("hits") == "1" for line in measured.findall("./lines/line"))
     assert measured.find("./lines/line[@number='2']").get("condition-coverage") == "100% (2/2)"
@@ -150,5 +163,5 @@ def test_prepare_imports_absolute_sources_outside_the_current_working_directory(
 
     report = ET.parse(tmp_path / "reports/coverage/python.xml")
     measured = report.find(".//class")
-    assert measured.get("filename") == str(source)
+    assert measured.get("filename") == "apps/api/src/main.py"
     assert measured.find("./lines/line").attrib == {"number": "1", "hits": "1"}
diff --git a/tests/test_topic_decision_worker_unit.py b/tests/test_topic_decision_worker_unit.py
new file mode 100644
index 00000000..77c70f3e
--- /dev/null
+++ b/tests/test_topic_decision_worker_unit.py
@@ -0,0 +1,289 @@
+"""Decision-worker lease ownership, persisted budgets, and unlocked model transport."""
+
+import uuid
+from contextlib import contextmanager
+from datetime import UTC, datetime
+from types import SimpleNamespace as NS
+from unittest.mock import Mock
+
+import pytest
+from devfeed_aggregator import topic_decision_tasks as worker
+from devfeed_core.analysis import snapshot_hash
+from devfeed_core.config import get_settings
+from devfeed_core.models import TopicAnalysisJob, TopicDecisionRun, TopicProposal
+from devfeed_core.topic_decision_budget import initial_state
+from devfeed_core.topic_decisions import DecisionDeferred
+
+NOW = datetime(2026, 10, 4, tzinfo=UTC)
+ID, PROPOSAL, TOKEN = (uuid.UUID(int=i) for i in (1, 2, 3))
+TOPIC = {"name": "Python", "slug": "python"}
+
+
+@pytest.fixture
+def harness(monkeypatch):
+    settings = get_settings().model_copy(
+        update={"codex_model": "unit-model", "full_automation": False}
+    )
+    proposal = TopicProposal(id=PROPOSAL, proposed=dict(TOPIC), status="pending", evidence=[])
+    job = TopicAnalysisJob(
+        id=ID,
+        proposal_id=PROPOSAL,
+        input_hash=snapshot_hash(TOPIC),
+        status="running",
+        attempts=2,
+        lease_token=TOKEN,
+        result={},
+        usage={},
+        duration_ms=0,
+    )
+    initial_run = TopicDecisionRun(
+        proposal_id=PROPOSAL,
+        input_hash=job.input_hash,
+        status="active",
+        state=initial_state(settings),
+    )
+    state = NS(
+        run=initial_run,
+        proposal=proposal,
+        owned=True,
+        in_transaction=0,
+        phase="start",
+        complete_error=None,
+        settled_elsewhere=False,
+    )
+    session, factory = Mock(), Mock()
+
+    @contextmanager
+    def transaction():
+        state.in_transaction += 1
+        try:
+            yield session
+        finally:
+            state.in_transaction -= 1
+
+    factory.begin.side_effect = transaction
+    session.get.side_effect = lambda model, *_args, **_kwargs: (
+        state.proposal if model is TopicProposal else state.run
+    )
+    session.scalar.return_value = job
+
+    def added(value):
+        if isinstance(value, TopicDecisionRun):
+            state.run = value
+
+    session.add.side_effect = added
+    monkeypatch.setattr(worker, "get_settings", lambda: settings)
+    monkeypatch.setattr(worker, "utcnow", lambda: NOW)
+    monkeypatch.setattr(worker, "owned_job", lambda *_: job if state.owned else None)
+    monkeypatch.setattr(worker, "lock_topics", Mock())
+    clients = []
+
+    def client(config):
+        result = Mock(usage={"inputTokens": 100, "outputTokens": 30}, web_search_count=1)
+        result.config = config
+
+        def complete(*_args, **_kwargs):
+            assert state.in_transaction == 0, "Model transport must release database locks"
+            if state.settled_elsewhere:
+                ledger = state.run.state
+                ledger["calls"][-1].update(status="returned", seconds=0.01)
+            if state.complete_error:
+                raise state.complete_error
+            return {"outcome": "ready"}
+
+        result.complete.side_effect = complete
+        clients.append(result)
+        return result
+
+    monkeypatch.setattr(worker, "CodexClient", client)
+    fetch = Mock(return_value={"pages": [], "failures": []})
+    monkeypatch.setattr(worker, "fetch_bundle", fetch)
+
+    def approve(_, __):
+        proposal.status = "approved"
+
+    approve = Mock(side_effect=approve)
+    monkeypatch.setattr(worker, "auto_approve_research", approve)
+    review = Mock(side_effect=lambda *_: setattr(proposal, "status", "rejected"))
+    monkeypatch.setattr(worker, "review_proposal", review)
+    pause = Mock(return_value=60)
+    monkeypatch.setattr(worker, "safe_pause", pause)
+
+    def execute(decision):
+        monkeypatch.setattr(worker, "run_decision", decision)
+        worker.execute_decision(factory, ID, TOKEN, PROPOSAL, TOPIC)
+
+    return NS(
+        settings=settings,
+        proposal=proposal,
+        job=job,
+        state=state,
+        session=session,
+        clients=clients,
+        fetch=fetch,
+        approve=approve,
+        review=review,
+        pause=pause,
+        execute=execute,
+    )
+
+
+def exercise_callbacks(_id, topic, state, *, call, save, fetch):
+    assert fetch(["https://example.com/"]) == {"pages": [], "failures": []}
+    save("discovery", {"sources": []})
+    assert call("discovery", "Discover exact identity", {}, web=True) == {"outcome": "ready"}
+    call("draft", "Draft from source evidence", {}, escalated=True)
+    call("verification", "Independently verify", {})
+    return {
+        "decision": "approved",
+        "topic": {**topic, "description": "Developer tool"},
+        "sources": [],
+    }
+
+
+@pytest.mark.parametrize("new_run", [False, True])
+def test_worker_persists_each_call_budget_and_charges_without_holding_locks(harness, new_run):
+    if new_run:
+        harness.state.run = None
+    harness.execute(exercise_callbacks)
+    run, job = harness.state.run, harness.job
+    assert [call["status"] for call in run.state["calls"]] == ["returned"] * 3
+    assert [client.operation for client in harness.clients] == [
+        "topic_discovery",
+        "topic_draft",
+        "topic_verification",
+    ]
+    assert harness.clients[0].search_limit > 0 and harness.clients[1].search_limit == 0
+    assert harness.clients[1].reason == "validation_escalation"
+    assert all(
+        client.token_limit > 0 and client.config.codex_timeout_seconds >= 10
+        for client in harness.clients
+    )
+    assert (
+        job.usage["inputTokens"] == 300
+        and job.usage["outputTokens"] == 90
+        and job.usage["totalTokens"] == 390
+    )
+    assert job.status == "succeeded" and job.outcome == "enriched"
+    assert job.lease_token is job.lease_until is None
+    assert run.status == "decided" and run.reason is None
+    assert "evidence_seconds" in run.state and "discovery" in run.state
+    assert harness.proposal.evidence[0]["before"] == TOPIC
+    assert harness.proposal.evidence[0]["analysis_id"] == str(ID)
+    assert job.result["applied_input_hash"] == snapshot_hash(harness.proposal.proposed)
+
+
+def test_settlement_is_idempotent_when_another_transaction_already_charged_the_call(harness):
+    harness.state.settled_elsewhere = True
+    harness.execute(exercise_callbacks)
+    assert harness.job.usage == {} and harness.job.duration_ms == 0
+    assert all(call["status"] == "returned" for call in harness.state.run.state["calls"])
+
+
+@pytest.mark.parametrize("full", [False, True])
+def test_out_of_scope_result_only_reviews_the_proposal_in_full_automation(harness, full):
+    harness.settings.full_automation = full
+    harness.execute(
+        lambda *_args, **_kwargs: {"decision": "rejected", "topic": TOPIC, "sources": []}
+    )
+    assert harness.job.outcome == "out_of_scope"
+    assert harness.review.call_count == int(full) and harness.approve.call_count == 0
+    assert harness.state.run.status == ("decided" if full else "review")
+    assert harness.state.run.reason == (None if full else "manual_review_required")
+
+
+def test_unapproved_enrichment_remains_available_for_manual_review(harness):
+    harness.approve.side_effect = None
+    harness.execute(exercise_callbacks)
+    assert harness.state.run.status == "review" and harness.proposal.status == "pending"
+    assert harness.state.run.reason == "manual_review_required"
+
+
+@pytest.mark.parametrize(
+    "change", ["missing", "reviewed", "hash", "run-hash", "inactive", "inactive-reason"]
+)
+def test_invalid_initial_state_defers_without_spending_model_budget(harness, change):
+    if change == "missing":
+        harness.state.proposal = None
+    elif change == "reviewed":
+        harness.proposal.status = "approved"
+    elif change == "hash":
+        harness.proposal.proposed = {"name": "Changed"}
+    elif change == "run-hash":
+        harness.state.run.input_hash = "old"
+    else:
+        harness.state.run.status = "deferred"
+        harness.state.run.reason = "manual_hold" if change == "inactive-reason" else None
+    harness.execute(exercise_callbacks)
+    assert harness.job.outcome == "decision_deferred"
+    assert harness.job.result["deferred_reason"] == (
+        "manual_hold" if change == "inactive-reason" else "proposal_changed"
+    )
+    assert harness.clients == []
+
+
+@pytest.mark.parametrize("phase", ["initial", "final", "error"])
+def test_lost_lease_never_applies_or_overwrites_a_job(harness, phase):
+    def decision(*_args, **_kwargs):
+        harness.state.owned = False
+        if phase == "error":
+            raise RuntimeError("dependency")
+        return {"decision": "approved", "topic": TOPIC, "sources": []}
+
+    if phase == "initial":
+        harness.state.owned = False
+    harness.execute(decision)
+    assert harness.job.status == "running" and harness.job.result == {}
+    assert harness.proposal.proposed == TOPIC and harness.proposal.evidence == []
+
+
+@pytest.mark.parametrize("change", ["reviewed", "changed"])
+def test_proposal_changed_during_transport_is_superseded(harness, change):
+    def decision(*_args, **_kwargs):
+        if change == "reviewed":
+            harness.proposal.status = "approved"
+        else:
+            harness.proposal.proposed = {"name": "Other"}
+        return {"decision": "approved", "topic": TOPIC, "sources": []}
+
+    harness.execute(decision)
+    assert harness.job.outcome == "superseded"
+    assert harness.state.run.status == "deferred" and harness.state.run.reason == "proposal_changed"
+    harness.approve.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "error",
+    [
+        DecisionDeferred("token_budget_exhausted"),
+        worker.AnalysisError("codex_server_overloaded"),
+        RuntimeError("private dependency detail"),
+    ],
+)
+def test_failed_calls_keep_charges_and_dependency_details_are_not_exposed(harness, error):
+    harness.state.complete_error = error
+    harness.execute(exercise_callbacks)
+    run, job = harness.state.run, harness.job
+    assert len(run.state["calls"]) == 1 and run.state["calls"][0]["status"] == "failed"
+    assert run.state["calls"][0]["charged_tokens"] == run.state["calls"][0]["reserved_tokens"]
+    assert job.usage["totalTokens"] == 130
+    if str(error) == "codex_server_overloaded":
+        assert job.status == "queued" and run.status == "active"
+        harness.pause.assert_called_once()
+    else:
+        assert job.outcome == "decision_deferred" and run.status == "deferred"
+        assert job.result["deferred_reason"] == (
+            str(error) if isinstance(error, DecisionDeferred) else "decision_dependency_failure"
+        )
+        assert "private dependency detail" not in str(job.result)
+
+
+@pytest.mark.parametrize("reason", [None, "operator_hold"])
+def test_run_paused_between_calls_cannot_reserve_additional_budget(harness, reason):
+    def decision(_id, _topic, _state, *, call, **_kwargs):
+        harness.state.run.status, harness.state.run.reason = "review", reason
+        call("discovery", "prompt", {})
+
+    harness.execute(decision)
+    assert harness.clients == [] and harness.state.run.state["calls"] == []
+    assert harness.job.result["deferred_reason"] == (reason or "decision_not_active")
diff --git a/uv.lock b/uv.lock
index 52c6b097..abf465c3 100644
--- a/uv.lock
+++ b/uv.lock
@@ -811,6 +811,7 @@ dev = [
     { name = "mypy" },
     { name = "pre-commit" },
     { name = "pytest" },
+    { name = "pytest-cov" },
     { name = "ruff" },
     { name = "testcontainers" },
 ]
@@ -840,6 +841,7 @@ dev = [
     { name = "mypy", specifier = ">=1.15,<3" },
     { name = "pre-commit", specifier = ">=4.3,<5" },
     { name = "pytest", specifier = ">=8.3,<10" },
+    { name = "pytest-cov", specifier = ">=7,<8" },
     { name = "ruff", specifier = ">=0.12,<1" },
     { name = "testcontainers", specifier = ">=4.10,<5" },
 ]
@@ -2811,6 +2813,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" },
 ]
 
+[[package]]
+name = "pytest-cov"
+version = "7.1.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "coverage" },
+    { name = "pluggy" },
+    { name = "pytest" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/b1/51/a849f96e117386044471c8ec2bd6cfebacda285da9525c9106aeb28da671/pytest_cov-7.1.0.tar.gz", hash = "sha256:30674f2b5f6351aa09702a9c8c364f6a01c27aae0c1366ae8016160d1efc56b2", size = 55592, upload-time = "2026-03-21T20:11:16.284Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/9d/7a/d968e294073affff457b041c2be9868a40c1c71f4a35fcc1e45e5493067b/pytest_cov-7.1.0-py3-none-any.whl", hash = "sha256:a0461110b7865f9a271aa1b51e516c9a95de9d696734a2f71e3e78f46e1d4678", size = 22876, upload-time = "2026-03-21T20:11:14.438Z" },
+]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"