From faeef155053d6655543d7cdad8b7c216dbe672b7 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 6 Jun 2026 13:02:14 +0000 Subject: [PATCH] feat: quality-gate finished articles before publishing Add a "write -> gate -> publish" path so blogger publishes finished articles (e.g. from claude-blog) only after they clear a quality bar, instead of posting raw briefs unconditionally. - quality.py: score an article via claude-blog's analyze_blog.py and gate on score.total >= min_score. Fails CLOSED (missing path, non-zero exit, unparseable output, or low score all block publishing). - content/ingest.py: parse a finished Markdown article (YAML front-matter + body) into a SourcePost; resolves title/summary/tags/link and enforces banned-phrase guardrails. - agent.run_campaign_from_article: ingest -> gate -> render -> publish; CampaignResult gains quality/blocked so a blocked run publishes nothing. - cli: `post --from-file` (with --min-score) and a standalone `score` command. - config: MNEMONIK_MIN_SCORE (default 80) and MNEMONIK_CLAUDE_BLOG_PATH. - docs: .env.example + AGENTS.md document the gated flow. - tests: 12 new tests covering ingest parsing and the gate (pass, block, and every fail-closed branch). --- .env.example | 7 ++ AGENTS.md | 25 +++++- src/mnemonik_blogger/agent.py | 62 ++++++++++++- src/mnemonik_blogger/cli.py | 62 ++++++++++--- src/mnemonik_blogger/config.py | 6 ++ src/mnemonik_blogger/content/generate.py | 5 +- src/mnemonik_blogger/content/ingest.py | 86 +++++++++++++++++++ src/mnemonik_blogger/quality.py | 104 ++++++++++++++++++++++ tests/test_ingest.py | 59 +++++++++++++ tests/test_quality.py | 105 +++++++++++++++++++++++ 10 files changed, 501 insertions(+), 20 deletions(-) create mode 100644 src/mnemonik_blogger/content/ingest.py create mode 100644 src/mnemonik_blogger/quality.py create mode 100644 tests/test_ingest.py create mode 100644 tests/test_quality.py diff --git a/.env.example b/.env.example index d7059dc..50983bb 100644 --- a/.env.example +++ b/.env.example @@ -9,6 +9,13 @@ MNEMONIK_DRY_RUN=true # Optional: local protocol-facts file used when the Mnemonik MCP is unavailable. # MNEMONIK_FACTS_PATH=./facts.yaml +# --- Quality gate (for `post --from-file` / `score`) ----------------------- +# A finished article is scored by claude-blog's analyzer and must reach this +# score (0-100) before it can be published. The gate fails CLOSED: if the path +# below is unset or the analyzer errors, nothing publishes. +# MNEMONIK_CLAUDE_BLOG_PATH=../claude-blog +# MNEMONIK_MIN_SCORE=80 + # --- Telegram channel ------------------------------------------------------ # Bot must be an ADMIN of the target channel. TELEGRAM_BOT_TOKEN= diff --git a/AGENTS.md b/AGENTS.md index 292f1a6..cf206da 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,8 +18,29 @@ attestation of what went out. - `src/mnemonik_blogger/grounding/mnemonik.py` — `Grounding` protocol; static / file fallbacks; wire the live Mnemonik MCP here. - `src/mnemonik_blogger/publish/` — one adapter per platform + a registry. -- `src/mnemonik_blogger/agent.py` — ground → generate → render → publish → attest. -- `src/mnemonik_blogger/cli.py` — `platforms`, `preview`, `post`. +- `src/mnemonik_blogger/agent.py` — ground → generate → render → publish → attest; + plus `run_campaign_from_article` (ingest → quality gate → publish). +- `src/mnemonik_blogger/content/ingest.py` — parse a finished Markdown article + (front-matter + body) into a `SourcePost`. +- `src/mnemonik_blogger/quality.py` — score an article via claude-blog's + `analyze_blog.py` and gate publishing on `score.total >= min_score`. Fails closed. +- `src/mnemonik_blogger/cli.py` — `platforms`, `preview`, `post` (`--from-file`), `score`. + +## Write → gate → publish + +For finished articles (e.g. written by claude-blog), publishing is **quality-gated**: + +```bash +# Score only (no publish); exits non-zero below threshold. +MNEMONIK_CLAUDE_BLOG_PATH=../claude-blog mnemonik-blogger score article.md + +# Publish only if score.total >= min_score (default 80); dry-run unless --live. +MNEMONIK_CLAUDE_BLOG_PATH=../claude-blog \ + mnemonik-blogger post --from-file article.md --platforms auto +``` + +The gate **fails closed**: if `MNEMONIK_CLAUDE_BLOG_PATH` is unset or the analyzer +errors, the article is blocked and nothing publishes. ## Rules diff --git a/src/mnemonik_blogger/agent.py b/src/mnemonik_blogger/agent.py index cc97d4d..f75e9dd 100644 --- a/src/mnemonik_blogger/agent.py +++ b/src/mnemonik_blogger/agent.py @@ -6,25 +6,38 @@ from __future__ import annotations from dataclasses import dataclass, field +from pathlib import Path from .config import Platform, Settings from .content.generate import build_source_post, render_all +from .content.ingest import ingest_article from .grounding.mnemonik import Grounding, default_grounding from .models import PublishResult, SourcePost from .publish import build_publisher +from .quality import QualityReport, score_article @dataclass class CampaignResult: post: SourcePost results: list[PublishResult] = field(default_factory=list) + # Set for article-based campaigns; None for direct brief-based campaigns. + quality: QualityReport | None = None + # True when the quality gate blocked publishing (results is then empty). + blocked: bool = False @property def ok(self) -> bool: - return all(r.ok for r in self.results) + return not self.blocked and all(r.ok for r in self.results) def summary(self) -> str: lines = [f"# {self.post.title}"] + if self.quality is not None: + lines.append(self.quality.summary()) + if self.blocked: + lines.append("BLOCKED: quality gate not met - nothing was published.") + for issue in self.quality.issues if self.quality else []: + lines.append(f" - {issue}") for r in self.results: status = "DRY-RUN" if r.dry_run else ("OK" if r.ok else "FAIL") extra = r.primary_url or r.error or "" @@ -47,8 +60,49 @@ def run_campaign( grounding = grounding or default_grounding(settings.facts_path) post = build_source_post(title=title, brief=brief, link=link, tags=tags, grounding=grounding) - rendered = render_all(post, platforms) + results = _publish_post(post, platforms, settings, grounding, attest=attest) + return CampaignResult(post=post, results=results) + + +def run_campaign_from_article( + *, + article: str | Path, + platforms: list[Platform], + settings: Settings, + grounding: Grounding | None = None, + attest: bool = True, + min_score: int | None = None, +) -> CampaignResult: + """Publish a finished Markdown article, but only if it clears the quality gate. + + The article is scored by claude-blog's analyzer. If ``score.total`` is below + the threshold (or the gate cannot be evaluated), nothing is published and the + result is marked ``blocked``. The gate fails closed. + """ + threshold = settings.min_score if min_score is None else min_score + report = score_article(article, claude_blog_path=settings.claude_blog_path, threshold=threshold) + # ingest_article enforces guardrails; do it before any publish. + post = ingest_article(article) + + if not report.passed: + return CampaignResult(post=post, results=[], quality=report, blocked=True) + + grounding = grounding or default_grounding(settings.facts_path) + results = _publish_post(post, platforms, settings, grounding, attest=attest) + return CampaignResult(post=post, results=results, quality=report) + + +def _publish_post( + post: SourcePost, + platforms: list[Platform], + settings: Settings, + grounding: Grounding, + *, + attest: bool, +) -> list[PublishResult]: + """Render and publish a post to each platform, then optionally attest.""" + rendered = render_all(post, platforms) results: list[PublishResult] = [] for platform in platforms: publisher = build_publisher(platform, settings) @@ -57,9 +111,9 @@ def run_campaign( if attest and not settings.dry_run: published = [r for r in results if r.ok and not r.dry_run] if published: - record = f"Published '{title}' to: " + ", ".join( + record = f"Published '{post.title}' to: " + ", ".join( f"{r.platform.value} ({r.primary_url or ', '.join(r.ids)})" for r in published ) grounding.attest(record, tags=["mnemonik-blogger", "published"]) - return CampaignResult(post=post, results=results) + return results diff --git a/src/mnemonik_blogger/cli.py b/src/mnemonik_blogger/cli.py index 44fe9f5..38d0442 100644 --- a/src/mnemonik_blogger/cli.py +++ b/src/mnemonik_blogger/cli.py @@ -12,11 +12,12 @@ import typer -from .agent import run_campaign +from .agent import run_campaign, run_campaign_from_article from .config import Platform, Settings, load_settings from .content.formatters import render from .content.generate import build_source_post from .grounding.mnemonik import default_grounding +from .quality import score_article app = typer.Typer(add_completion=False, help="Mnemonik multi-platform blogger agent.") @@ -70,30 +71,67 @@ def preview( @app.command() def post( - title: str = typer.Option(..., help="Post title."), - brief: str = typer.Option(..., help="The angle / body of the post."), + title: str = typer.Option(None, help="Post title (with --brief)."), + brief: str = typer.Option(None, help="The angle / body of the post (with --title)."), + from_file: str = typer.Option( + None, + "--from-file", + help="Publish a finished Markdown article instead of a brief (quality-gated).", + ), link: str = typer.Option(None, help="Canonical URL to include."), tags: str = typer.Option(None, help="Comma-separated hashtags."), target: str = typer.Option("auto", "--platforms", help="Comma list or 'auto'."), live: bool = typer.Option(False, "--live", help="Actually publish (overrides dry-run)."), + min_score: int = typer.Option( + None, "--min-score", help="Quality threshold for --from-file (default: config)." + ), ) -> None: - """Generate and publish a post across platforms.""" + """Publish from a brief (--title/--brief) or a finished article (--from-file).""" settings = load_settings() if live: settings.dry_run = False chosen = _parse_platforms(target, settings) - result = run_campaign( - title=title, - brief=brief, - link=link, - tags=[t.strip() for t in tags.split(",")] if tags else None, - platforms=chosen, - settings=settings, - ) + + if from_file: + result = run_campaign_from_article( + article=from_file, + platforms=chosen, + settings=settings, + min_score=min_score, + ) + else: + if not title or not brief: + raise typer.BadParameter("provide --title and --brief, or --from-file") + result = run_campaign( + title=title, + brief=brief, + link=link, + tags=[t.strip() for t in tags.split(",")] if tags else None, + platforms=chosen, + settings=settings, + ) typer.echo(result.summary()) if not result.ok: raise typer.Exit(code=1) +@app.command() +def score( + article: str = typer.Argument(..., help="Path to a finished Markdown article."), + min_score: int = typer.Option( + None, "--min-score", help="Threshold to gate against (default: config)." + ), +) -> None: + """Score an article with claude-blog's analyzer (no publishing).""" + settings = load_settings() + threshold = settings.min_score if min_score is None else min_score + report = score_article(article, claude_blog_path=settings.claude_blog_path, threshold=threshold) + typer.echo(report.summary()) + for issue in report.issues: + typer.echo(f" - {issue}") + if not report.passed: + raise typer.Exit(code=1) + + if __name__ == "__main__": app() diff --git a/src/mnemonik_blogger/config.py b/src/mnemonik_blogger/config.py index a0d3ea9..6817b96 100644 --- a/src/mnemonik_blogger/config.py +++ b/src/mnemonik_blogger/config.py @@ -108,6 +108,12 @@ class Settings(_Base): dry_run: bool = True # Path to a local protocol-facts file used when the Mnemonik MCP is unavailable. facts_path: str | None = None + # Quality gate: a finished article must score >= min_score (claude-blog's + # analyzer) before --from-file publishing is allowed. 0-100. + min_score: int = 80 + # Path to a claude-blog checkout; its scripts/analyze_blog.py scores articles. + # Required for --from-file; absence fails the gate closed (nothing publishes). + claude_blog_path: str | None = None telegram: TelegramConfig = Field(default_factory=TelegramConfig) discord: DiscordConfig = Field(default_factory=DiscordConfig) diff --git a/src/mnemonik_blogger/content/generate.py b/src/mnemonik_blogger/content/generate.py index ac05e13..9f34a7a 100644 --- a/src/mnemonik_blogger/content/generate.py +++ b/src/mnemonik_blogger/content/generate.py @@ -14,7 +14,8 @@ from . import formatters, voice -def _enforce_guardrails(text: str) -> None: +def enforce_guardrails(text: str) -> None: + """Raise if `text` contains a banned phrase. Used by both brief- and file-based flows.""" lowered = text.lower() for banned in voice.BANNED_PHRASES: if banned in lowered: @@ -44,7 +45,7 @@ def build_source_post( proof = " ".join(f.text for f in facts[:2]) body = f"{body}\n\nWhy it holds up: {proof}" - _enforce_guardrails(title + " " + body) + enforce_guardrails(title + " " + body) return SourcePost( title=title, diff --git a/src/mnemonik_blogger/content/ingest.py b/src/mnemonik_blogger/content/ingest.py new file mode 100644 index 0000000..cddb3e1 --- /dev/null +++ b/src/mnemonik_blogger/content/ingest.py @@ -0,0 +1,86 @@ +"""Ingest a finished article (Markdown + optional YAML front-matter) into a SourcePost. + +claude-blog produces articles as Markdown files, optionally led by a YAML +front-matter block delimited by ``---``. This module turns such a file into the +blogger's canonical :class:`SourcePost` so the publish pipeline can repurpose it +per platform. + +Resolution rules: + * title - front-matter ``title``; else the first H1; else the file stem. + * summary - front-matter ``summary``/``description``; else the first paragraph. + * tags - front-matter ``tags`` (list or comma string). + * link - front-matter ``link``/``canonical``/``url``. + +The article is treated as already-written and reviewed, so its body is not +re-grounded or mutated; only banned-phrase guardrails are enforced. +""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any + +import yaml + +from ..models import SourcePost +from .generate import enforce_guardrails + +_FRONTMATTER = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL) +_H1 = re.compile(r"^#\s+(.+?)\s*$", re.MULTILINE) + + +def _split_frontmatter(text: str) -> tuple[dict[str, Any], str]: + match = _FRONTMATTER.match(text) + if not match: + return {}, text + parsed = yaml.safe_load(match.group(1)) + meta: dict[str, Any] = parsed if isinstance(parsed, dict) else {} + return meta, text[match.end() :] + + +def _first_paragraph(body: str) -> str: + for block in body.split("\n\n"): + stripped = block.strip() + if stripped and not stripped.startswith("#"): + return " ".join(stripped.split()) + return "" + + +def _coerce_tags(raw: Any) -> list[str]: + if isinstance(raw, str): + parts = raw.split(",") + elif isinstance(raw, list): + parts = [str(t) for t in raw] + else: + return [] + return [t.strip() for t in parts if t.strip()] + + +def ingest_article(path: str | Path) -> SourcePost: + """Parse a Markdown article file into a SourcePost (raises on guardrail violation).""" + path = Path(path) + meta, body = _split_frontmatter(path.read_text(encoding="utf-8")) + body = body.strip() + + fm_title = meta.get("title") + if fm_title: + title = str(fm_title) + else: + h1 = _H1.search(body) + title = h1.group(1).strip() if h1 else path.stem + # Drop the leading H1 so it isn't duplicated in the social copy. + body = _H1.sub("", body, count=1).strip() + + summary = meta.get("summary") or meta.get("description") or _first_paragraph(body) + link = meta.get("link") or meta.get("canonical") or meta.get("url") + + enforce_guardrails(f"{title} {body}") + + return SourcePost( + title=title, + body=body, + summary=str(summary)[:200], + tags=_coerce_tags(meta.get("tags")), + link=str(link) if link else None, + ) diff --git a/src/mnemonik_blogger/quality.py b/src/mnemonik_blogger/quality.py new file mode 100644 index 0000000..de72285 --- /dev/null +++ b/src/mnemonik_blogger/quality.py @@ -0,0 +1,104 @@ +"""Quality gate: score an article with claude-blog's analyzer before publishing. + +The blogger does not write or score articles itself - that is claude-blog's job. +This module shells out to claude-blog's ``scripts/analyze_blog.py``, parses the +0-100 ``score.total``, and decides whether the article clears the configured +threshold. + +It fails **closed**: any error (analyzer missing, non-zero exit, unparseable +output, missing/invalid score) yields a non-passing ``QualityReport`` so a broken +gate can never let unreviewed content through to a real publish. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from dataclasses import dataclass, field +from pathlib import Path + +# Hard cap on the analyzer subprocess so a hung scorer can't wedge a publish. +_TIMEOUT_S = 120 + + +@dataclass +class QualityReport: + """Outcome of scoring one article against the quality threshold.""" + + score: int + threshold: int + rating: str = "" + issues: list[str] = field(default_factory=list) + # Set when the gate could not be evaluated; ``passed`` is then always False. + error: str | None = None + + @property + def passed(self) -> bool: + return self.error is None and self.score >= self.threshold + + def summary(self) -> str: + if self.error: + return f"quality gate ERROR (fail-closed): {self.error}" + verdict = "PASS" if self.passed else "BLOCK" + return ( + f"quality gate {verdict}: scored {self.score}/100 " + f"(threshold {self.threshold}){f', rating {self.rating}' if self.rating else ''}" + ) + + +def _fail(threshold: int, error: str) -> QualityReport: + return QualityReport(score=0, threshold=threshold, error=error) + + +def _format_issue(item: object) -> str: + """Render an analyzer issue (a dict or a bare string) as one readable line.""" + if isinstance(item, dict): + text = str(item.get("issue") or item.get("message") or "").strip() + label = "/".join(str(item[k]) for k in ("severity", "category") if item.get(k)) + return f"[{label}] {text}" if label else text + return str(item).strip() + + +def score_article( + article: str | Path, + *, + claude_blog_path: str | Path | None, + threshold: int, +) -> QualityReport: + """Score ``article`` with claude-blog's analyzer; never raises (fail-closed).""" + if not claude_blog_path: + return _fail(threshold, "claude_blog_path is not configured") + + analyzer = Path(claude_blog_path) / "scripts" / "analyze_blog.py" + if not analyzer.is_file(): + return _fail(threshold, f"analyzer not found at {analyzer}") + + article = Path(article) + if not article.is_file(): + return _fail(threshold, f"article not found: {article}") + + try: + proc = subprocess.run( + [sys.executable, str(analyzer), str(article)], + capture_output=True, + text=True, + timeout=_TIMEOUT_S, + ) + except (OSError, subprocess.SubprocessError) as exc: + return _fail(threshold, f"analyzer failed to run: {exc}") + + if proc.returncode != 0: + detail = (proc.stderr or proc.stdout).strip().splitlines() + tail = detail[-1] if detail else "" + return _fail(threshold, f"analyzer exited {proc.returncode}: {tail[:200]}") + + try: + block = json.loads(proc.stdout)["score"] + total = int(block["total"]) + except (json.JSONDecodeError, KeyError, TypeError, ValueError) as exc: + return _fail(threshold, f"could not parse analyzer output: {exc}") + + rating = str(block.get("rating", "")) + issues = [s for s in (_format_issue(i) for i in block.get("issues", [])) if s][:10] + return QualityReport(score=total, threshold=threshold, rating=rating, issues=issues) diff --git a/tests/test_ingest.py b/tests/test_ingest.py new file mode 100644 index 0000000..eeb2843 --- /dev/null +++ b/tests/test_ingest.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from mnemonik_blogger.content.ingest import ingest_article + + +def test_ingest_uses_frontmatter(tmp_path: Path) -> None: + path = tmp_path / "a.md" + path.write_text( + "---\n" + "title: Provable Agent Memory\n" + "summary: Agents forget; we make recall verifiable.\n" + "tags: [Mnemonik, AIagents]\n" + "link: https://mnemonik.xyz\n" + "---\n" + "# A Different Heading\n\nBody text here.\n", + encoding="utf-8", + ) + post = ingest_article(path) + assert post.title == "Provable Agent Memory" + assert post.summary == "Agents forget; we make recall verifiable." + assert post.tags == ["Mnemonik", "AIagents"] + assert post.link == "https://mnemonik.xyz" + # Front-matter title present, so the H1 is kept in the body. + assert "A Different Heading" in post.body + + +def test_ingest_falls_back_to_h1_and_first_paragraph(tmp_path: Path) -> None: + path = tmp_path / "b.md" + path.write_text( + "# Verifiable Memory\n\nThe first paragraph becomes the summary.\n\nMore body.\n", + encoding="utf-8", + ) + post = ingest_article(path) + assert post.title == "Verifiable Memory" + assert post.summary == "The first paragraph becomes the summary." + # H1 stripped from body when it was the title source. + assert not post.body.startswith("# Verifiable Memory") + assert "More body." in post.body + + +def test_ingest_comma_string_tags(tmp_path: Path) -> None: + path = tmp_path / "c.md" + path.write_text( + "---\ntags: Mnemonik, Solana , Arweave\n---\n# T\n\nBody.\n", + encoding="utf-8", + ) + post = ingest_article(path) + assert post.tags == ["Mnemonik", "Solana", "Arweave"] + + +def test_ingest_enforces_guardrails(tmp_path: Path) -> None: + path = tmp_path / "d.md" + path.write_text("# Buy now\n\nGuaranteed returns, 100x.\n", encoding="utf-8") + with pytest.raises(ValueError): + ingest_article(path) diff --git a/tests/test_quality.py b/tests/test_quality.py new file mode 100644 index 0000000..b267fed --- /dev/null +++ b/tests/test_quality.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from mnemonik_blogger.agent import run_campaign_from_article +from mnemonik_blogger.config import Platform, Settings +from mnemonik_blogger.quality import score_article + +ARTICLE = """\ +# Verifiable Memory for AI Agents + +Agents forget, and you cannot prove what they recalled. Mnemonik makes every +memory a signed, verifiable attestation anchored on a public ledger. +""" + + +def _write_article(tmp_path: Path) -> Path: + path = tmp_path / "article.md" + path.write_text(ARTICLE, encoding="utf-8") + return path + + +def _fake_claude_blog(tmp_path: Path, *, body: str) -> str: + """Create a claude-blog checkout whose analyze_blog.py runs `body`.""" + root = tmp_path / "claude-blog" + (root / "scripts").mkdir(parents=True) + (root / "scripts" / "analyze_blog.py").write_text(body, encoding="utf-8") + return str(root) + + +def _emit_score(score: int) -> str: + payload = json.dumps({"score": {"total": score, "rating": "Publish", "issues": ["fix intro"]}}) + return f"print({payload!r})\n" + + +def test_score_passes_at_or_above_threshold(tmp_path: Path) -> None: + cb = _fake_claude_blog(tmp_path, body=_emit_score(85)) + report = score_article(_write_article(tmp_path), claude_blog_path=cb, threshold=80) + assert report.passed + assert report.score == 85 + assert report.rating == "Publish" + + +def test_score_blocks_below_threshold(tmp_path: Path) -> None: + cb = _fake_claude_blog(tmp_path, body=_emit_score(40)) + report = score_article(_write_article(tmp_path), claude_blog_path=cb, threshold=80) + assert not report.passed + assert report.error is None # evaluated fine, just too low + + +def test_missing_claude_blog_path_fails_closed(tmp_path: Path) -> None: + report = score_article(_write_article(tmp_path), claude_blog_path=None, threshold=80) + assert not report.passed + assert report.error is not None + + +def test_missing_analyzer_fails_closed(tmp_path: Path) -> None: + empty = tmp_path / "empty-checkout" + empty.mkdir() + report = score_article(_write_article(tmp_path), claude_blog_path=str(empty), threshold=80) + assert not report.passed + assert "analyzer not found" in (report.error or "") + + +def test_bad_json_fails_closed(tmp_path: Path) -> None: + cb = _fake_claude_blog(tmp_path, body="print('not json')\n") + report = score_article(_write_article(tmp_path), claude_blog_path=cb, threshold=80) + assert not report.passed + assert "parse" in (report.error or "") + + +def test_analyzer_nonzero_exit_fails_closed(tmp_path: Path) -> None: + cb = _fake_claude_blog(tmp_path, body="import sys\nsys.exit(2)\n") + report = score_article(_write_article(tmp_path), claude_blog_path=cb, threshold=80) + assert not report.passed + assert "exited 2" in (report.error or "") + + +def test_campaign_publishes_when_gate_passes(tmp_path: Path) -> None: + cb = _fake_claude_blog(tmp_path, body=_emit_score(90)) + settings = Settings(dry_run=True, claude_blog_path=cb, min_score=80) + result = run_campaign_from_article( + article=_write_article(tmp_path), + platforms=[Platform.TELEGRAM, Platform.TWITTER], + settings=settings, + ) + assert result.ok + assert not result.blocked + assert len(result.results) == 2 + assert all(r.dry_run for r in result.results) + + +def test_campaign_blocks_and_publishes_nothing_when_gate_fails(tmp_path: Path) -> None: + cb = _fake_claude_blog(tmp_path, body=_emit_score(10)) + settings = Settings(dry_run=True, claude_blog_path=cb, min_score=80) + result = run_campaign_from_article( + article=_write_article(tmp_path), + platforms=list(Platform), + settings=settings, + ) + assert result.blocked + assert not result.ok + assert result.results == [] + assert "BLOCKED" in result.summary()