feat(task-019): add deterministic end-to-end demo stack flow

apps/backend/Dockerfile

@@ -5,6 +5,10 @@ ENV PYTHONUNBUFFERED=1
 
 WORKDIR /app
 
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends git nodejs npm \
+    && rm -rf /var/lib/apt/lists/*
+
 COPY apps/backend/requirements.txt ./requirements.txt
 RUN pip install --no-cache-dir -r requirements.txt
 

apps/backend/src/application/agent_jobs.py

@@ -18,6 +18,10 @@ from src.domain.contracts import (
     Role,
 )
 from src.application.observability import evaluate_retry_policy, project_job_for_view
+from src.application.demo_runtime import (
+    build_demo_section_completion_payload,
+    should_auto_retry_section_job,
+)
 
 
 VALID_FAKE_AGENT_PROFILE = "fake-codex"
@@ -91,6 +95,21 @@ def retry_agent_job(
             "job_type": retry.job_type.value,
         },
     )
+    if should_auto_retry_section_job(retry):
+        completion = build_demo_section_completion_payload(
+            payload=retry.payload,
+            failed=False,
+        )
+        return complete_agent_job(
+            repository,
+            job_id=retry.id,
+            workspace_path=f"/tmp/demo/{retry.id}",
+            stdout="deterministic demo section retry succeeded\n",
+            stderr="",
+            exit_code=0,
+            duration_ms=8,
+            output=completion,
+        )
     return AgentJobResponse(job=_project_admin_view(repository, retry))
 
 

apps/backend/src/application/demo_runtime.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from src.domain.contracts import AgentJobSummary
+
+
+_DEMO_MODE_ENV = "PIPELINE_DEMO_MODE"
+_DEMO_REPOSITORY_ENV = "PIPELINE_DEMO_PUBLISH_REPO_PATH"
+_DEFAULT_DEMO_REPOSITORY_PATH = "/tmp/pipeline-demo-site.git"
+
+
+def is_demo_mode() -> bool:
+    value = os.environ.get(_DEMO_MODE_ENV, "").strip().lower()
+    return value in {"1", "true", "yes", "on"}
+
+
+def resolve_demo_repository_path() -> Path:
+    configured = os.environ.get(_DEMO_REPOSITORY_ENV, _DEFAULT_DEMO_REPOSITORY_PATH)
+    return Path(configured).expanduser().resolve()
+
+
+def ensure_demo_bare_repository(path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if not path.exists():
+        _run_git(["init", "--bare", str(path)])
+    if _has_main_branch(path):
+        return
+
+    with tempfile.TemporaryDirectory(prefix="demo-site-seed-") as temp_dir:
+        worktree = Path(temp_dir) / "seed"
+        worktree.mkdir(parents=True, exist_ok=True)
+        _run_git(["init"], cwd=worktree)
+        _run_git(["config", "user.name", "Pipeline Demo Bot"], cwd=worktree)
+        _run_git(["config", "user.email", "pipeline-demo-bot@example.com"], cwd=worktree)
+        (worktree / "README.md").write_text("# Pipeline Demo Site\n", encoding="utf-8")
+        _run_git(["add", "README.md"], cwd=worktree)
+        _run_git(["commit", "-m", "seed demo repository"], cwd=worktree)
+        _run_git(["branch", "-M", "main"], cwd=worktree)
+        _run_git(["remote", "add", "origin", str(path)], cwd=worktree)
+        _run_git(["push", "origin", "main"], cwd=worktree)
+
+
+def should_auto_run_section_jobs() -> bool:
+    return is_demo_mode()
+
+
+def should_auto_retry_section_job(job: AgentJobSummary) -> bool:
+    if not is_demo_mode():
+        return False
+    if job.payload.get("demo_fail_once") is True and job.attempt > 1:
+        return True
+    return False
+
+
+def build_demo_section_completion_payload(
+    *,
+    payload: dict[str, Any],
+    failed: bool,
+) -> dict[str, Any]:
+    heading = _string_value(payload.get("heading")) or "Section"
+    section_id = _string_value(payload.get("section_id")) or "section"
+    draft_markdown = (
+        f"## {heading}\n\n"
+        f"This demo section is generated deterministically for `{section_id}`.\n\n"
+        "Read the related architecture guide: [internal reference](/guides/pipeline-demo).\n"
+    )
+    if failed:
+        return {
+            "status": "FAILED",
+            "error_category": "CLI_EXIT_CODE_FAILURE",
+            "error_message": "Deterministic demo failure for retry walkthrough.",
+            "payload": {
+                "section_id": section_id,
+                "heading": heading,
+                "artifact_label": heading,
+                "last_successful_step": "outline generated",
+                "unsupported_claims": [],
+                "draft_markdown": "",
+            },
+        }
+
+    return {
+        "status": "SUCCEEDED",
+        "output_files": [{"path": f"outputs/{section_id}.md"}],
+        "payload": {
+            "section_id": section_id,
+            "heading": heading,
+            "artifact_label": heading,
+            "unsupported_claims": [],
+            "last_successful_step": "section scaffold complete",
+            "suggested_visuals": [f"{heading} diagram"],
+            "draft_markdown": draft_markdown,
+        },
+    }
+
+
+def slugify(value: str) -> str:
+    normalized = re.sub(r"[^a-zA-Z0-9]+", "-", value.strip().lower()).strip("-")
+    return normalized or "article"
+
+
+def _has_main_branch(path: Path) -> bool:
+    result = subprocess.run(
+        ["git", "--git-dir", str(path), "rev-parse", "--verify", "refs/heads/main"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    return result.returncode == 0
+
+
+def _run_git(command: list[str], *, cwd: Path | None = None) -> None:
+    subprocess.run(
+        ["git", *command] if command[0] != "git" else command,
+        cwd=str(cwd) if cwd is not None else None,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+
+
+def _string_value(value: Any) -> str:
+    if isinstance(value, str):
+        return value.strip()
+    return ""

apps/backend/src/application/evidence.py

@@ -3,6 +3,11 @@ from __future__ import annotations
 from datetime import UTC, datetime
 from uuid import UUID
 
+from src.application.agent_jobs import complete_agent_job
+from src.application.demo_runtime import (
+    build_demo_section_completion_payload,
+    should_auto_run_section_jobs,
+)
 from src.domain.contracts import (
     AgentJobListResponse,
     AgentJobStatus,
@@ -189,6 +194,7 @@ def start_draft(repository: object, *, article_id: UUID) -> AgentJobListResponse
     jobs = []
     for section in approved_plan.sections:
         used_evidence_ids = claim_evidence_by_section.get(str(section.id), [])
+        demo_fail_once = should_auto_run_section_jobs() and len(jobs) == 0
         jobs.append(
             repository.agent_jobs.create(
                 article_id=article_id,
@@ -212,11 +218,16 @@ def start_draft(repository: object, *, article_id: UUID) -> AgentJobListResponse
                     "unsupported_claims": [],
                     "suggested_visuals": [],
                     "draft_markdown": "",
+                    "demo_fail_once": demo_fail_once,
                 },
                 queued_at=now,
             )
         )
 
+    if should_auto_run_section_jobs():
+        _auto_complete_demo_section_jobs(repository, jobs)
+        jobs = repository.agent_jobs.list_for_article(article_id)
+
     return AgentJobListResponse(jobs=jobs)
 
 
@@ -313,3 +324,29 @@ def _insufficient_reasons(claims: list[object]) -> list[str]:
 
 def _now() -> datetime:
     return datetime.now(UTC)
+
+
+def _auto_complete_demo_section_jobs(
+    repository: object,
+    jobs: list[object],
+) -> None:
+    for index, job in enumerate(jobs):
+        failed = index == 0
+        output = build_demo_section_completion_payload(
+            payload=job.payload,
+            failed=failed,
+        )
+        complete_agent_job(
+            repository,
+            job_id=job.id,
+            workspace_path=f"/tmp/demo/{job.id}",
+            stdout=(
+                "deterministic demo failure; retry to continue\n"
+                if failed
+                else "deterministic demo section scaffold succeeded\n"
+            ),
+            stderr="" if not failed else "exit code 1 simulated\n",
+            exit_code=1 if failed else 0,
+            duration_ms=12,
+            output=output,
+        )

apps/backend/src/application/seed_data.py

@@ -4,6 +4,11 @@ from datetime import UTC, datetime
 from hashlib import sha256
 from uuid import NAMESPACE_URL, UUID, uuid5
 
+from src.application.demo_runtime import (
+    ensure_demo_bare_repository,
+    is_demo_mode,
+    resolve_demo_repository_path,
+)
 from src.domain.contracts import (
     PublishingRules,
     Role,
@@ -44,8 +49,9 @@ def seed_reference_data(repository: object) -> None:
         updated_at=SEED_TIMESTAMP,
     )
 
+    repository_url = _repository_url_for_seed()
     publishing_rules = PublishingRules(
-        repository_url="git@github.com:example/site.git",
+        repository_url=repository_url,
         production_branch="main",
         content_format="mdx",
         content_path_template="content/articles/{slug}.mdx",
@@ -85,15 +91,15 @@ def seed_reference_data(repository: object) -> None:
         updated_at=SEED_TIMESTAMP,
     )
 
-    publishing_yaml = """target: b2b_saas_blog
+    publishing_yaml = f"""target: b2b_saas_blog
 repository:
-  url: git@github.com:example/site.git
+  url: {repository_url}
   branch: main
 content:
   format: mdx
-  path_template: content/articles/{slug}.mdx
+  path_template: content/articles/{{slug}}.mdx
 assets:
-  path_template: public/articles/{slug}/{filename}
+  path_template: public/articles/{{slug}}/{{filename}}
 """
     transform_script = """export function transformArticle(article) {
   return {
@@ -121,6 +127,14 @@ assets:
     )
 
 
+def _repository_url_for_seed() -> str:
+    if not is_demo_mode():
+        return "git@github.com:example/site.git"
+    demo_repo_path = resolve_demo_repository_path()
+    ensure_demo_bare_repository(demo_repo_path)
+    return str(demo_repo_path)
+
+
 def _stable_uuid(value: str) -> UUID:
     return uuid5(NAMESPACE_URL, f"ai-content-pipeline:{value}")
 

apps/backend/tests/integration/test_end_to_end_demo_stack_smoke_public_api.py

@@ -0,0 +1,357 @@
+from __future__ import annotations
+
+import base64
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+
+
+BACKEND_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(BACKEND_ROOT))
+
+from src.application.seed_data import seed_reference_data  # noqa: E402
+from src.infrastructure.repositories import open_backend_repository  # noqa: E402
+from src.presentation.dependencies import get_repository  # noqa: E402
+from src.presentation.main import app  # noqa: E402
+
+
+DEMO_EDITOR_EMAIL = "editor@example.com"
+DEMO_ADMIN_EMAIL = "admin@example.com"
+DEMO_USER_EMAIL_HEADER = "X-Demo-User-Email"
+
+
+class EndToEndDemoStackSmokePublicApiTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp_dir = tempfile.TemporaryDirectory()
+        self.object_root = Path(self.tmp_dir.name) / "objects"
+        self.demo_repo_path = Path(self.tmp_dir.name) / "demo-site.git"
+        os.environ["OBJECT_STORAGE_LOCAL_ROOT"] = str(self.object_root)
+        os.environ["PIPELINE_DEMO_MODE"] = "1"
+        os.environ["PIPELINE_DEMO_PUBLISH_REPO_PATH"] = str(self.demo_repo_path)
+        dsn = f"sqlite:///{Path(self.tmp_dir.name) / 'demo-stack-smoke.db'}"
+        self.repository = open_backend_repository(dsn)
+        self.repository.setup()
+        seed_reference_data(self.repository)
+        app.dependency_overrides[get_repository] = lambda: self.repository
+        self.client = TestClient(app)
+
+    def tearDown(self) -> None:
+        app.dependency_overrides.clear()
+        os.environ.pop("OBJECT_STORAGE_LOCAL_ROOT", None)
+        os.environ.pop("PIPELINE_DEMO_MODE", None)
+        os.environ.pop("PIPELINE_DEMO_PUBLISH_REPO_PATH", None)
+        self.tmp_dir.cleanup()
+
+    def test_demo_happy_path_reaches_publish_commit_created_with_failure_retry(self) -> None:
+        site_id = self._fetch_first_site_id()
+        article_id = self._create_article(site_id)
+
+        self._generate_submit_boundary_answers(article_id)
+        plan = self._generate_and_approve_plan(article_id)
+        self.assertGreaterEqual(len(plan["sections"]), 1)
+
+        research = self.client.post(
+            f"/api/articles/{article_id}/research/start",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, research.status_code, research.text)
+
+        research_listing = self.client.get(
+            f"/api/articles/{article_id}/research",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, research_listing.status_code, research_listing.text)
+        manifests = research_listing.json()["manifests"]
+        self.assertTrue(manifests)
+        self._assert_research_artifacts_exist(manifests)
+
+        evidence = self.client.get(
+            f"/api/articles/{article_id}/evidence",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, evidence.status_code, evidence.text)
+        self.assertEqual("EVIDENCE_MATRIX_READY", evidence.json()["article"]["status"])
+        self._approve_all_evidence(article_id, evidence.json()["evidence"])
+
+        start_draft = self.client.post(
+            f"/api/articles/{article_id}/draft/start",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(202, start_draft.status_code, start_draft.text)
+        started_jobs = start_draft.json()["jobs"]
+        section_jobs = [job for job in started_jobs if job["job_type"] == "SECTION_SCAFFOLD"]
+        self.assertTrue(section_jobs)
+        failed_jobs = [job for job in section_jobs if job["status"] == "FAILED"]
+        self.assertTrue(failed_jobs, "Demo path must include visible failure before retry.")
+
+        retry = self.client.post(
+            f"/api/agent-jobs/{failed_jobs[0]['id']}/retry",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
+        )
+        self.assertEqual(201, retry.status_code, retry.text)
+        self.assertEqual("SUCCEEDED", retry.json()["job"]["status"])
+
+        assemble = self.client.post(
+            f"/api/articles/{article_id}/draft/assemble",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, assemble.status_code, assemble.text)
+        draft = assemble.json()["draft"]
+
+        assets = self.client.post(
+            f"/api/articles/{article_id}/assets/generate-specs",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, assets.status_code, assets.text)
+        generated_assets = assets.json()["assets"]
+        self.assertTrue(generated_assets)
+        self.assertTrue(
+            any(asset["asset_type"] in {"diagram", "table", "flowchart", "comparison_matrix", "architecture_diagram"} for asset in generated_assets)
+        )
+        self._approve_assets(article_id, generated_assets)
+
+        seo = self.client.post(
+            f"/api/articles/{article_id}/seo/review",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, seo.status_code, seo.text)
+        language = self.client.post(
+            f"/api/articles/{article_id}/language/review",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, language.status_code, language.text)
+        self._accept_all_review_suggestions(article_id, seo.json()["report"]["issues"], kind="seo")
+        self._accept_all_review_suggestions(article_id, language.json()["report"]["issues"], kind="language")
+
+        issues = self.client.get(
+            f"/api/articles/{article_id}/final-review/issues",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, issues.status_code, issues.text)
+        self.assertEqual(0, issues.json()["unresolved_count"])
+
+        drafts = self.client.get(
+            f"/api/articles/{article_id}/drafts",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, drafts.status_code, drafts.text)
+        latest_draft = max(drafts.json()["drafts"], key=lambda item: item["version"])
+
+        approval = self.client.post(
+            f"/api/articles/{article_id}/final-approval",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            json={
+                "draft_version": latest_draft["version"],
+                "publishing_settings": {
+                    "content_path": "/guides/task-019-demo-stack",
+                    "author": "Demo Editor",
+                    "publishing_mode": "MANUAL",
+                    "frontmatter": {
+                        "title": latest_draft["title"],
+                        "category": "Demo",
+                    },
+                },
+            },
+        )
+        self.assertEqual(200, approval.status_code, approval.text)
+        self.assertEqual("PUBLISH_DRY_RUN_REQUIRED", approval.json()["article"]["status"])
+
+        dry_run = self.client.post(
+            f"/api/articles/{article_id}/publishing/dry-run",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, dry_run.status_code, dry_run.text)
+        self.assertTrue(dry_run.json()["content_shape_valid"])
+        self.assertEqual("PUBLISH_COMMIT_READY", dry_run.json()["article"]["status"])
+
+        commit = self.client.post(
+            f"/api/articles/{article_id}/publishing/create-commit",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, commit.status_code, commit.text)
+        publish_commit = commit.json()["publish_commit"]
+        self.assertEqual("PUBLISH_COMMIT_CREATED", publish_commit["status"])
+        self.assertEqual("PUBLISH_COMMIT_CREATED", commit.json()["article"]["status"])
+        self.assertEqual(str(self.demo_repo_path.resolve()), publish_commit["repository_url"])
+
+        self._assert_commit_payload_valid(publish_commit)
+        self._assert_timeline_contains_major_actions(article_id)
+
+    def _fetch_first_site_id(self) -> str:
+        site_response = self.client.get(
+            "/api/sites",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, site_response.status_code, site_response.text)
+        site = site_response.json()[0]["site"]
+        self.assertEqual(
+            str(self.demo_repo_path.resolve()),
+            site["publishing_rules"]["repository_url"],
+        )
+        return site["id"]
+
+    def _create_article(self, site_id: str) -> str:
+        article_response = self.client.post(
+            "/api/articles",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            json={
+                "target_site_id": site_id,
+                "brief_description": "Demo-stack shortest public happy path toward publish flow.",
+                "working_title": "Task 019 Demo Stack Smoke",
+                "content_type": "longform_guide",
+                "primary_keyword": "demo stack smoke path",
+            },
+        )
+        self.assertEqual(201, article_response.status_code, article_response.text)
+        return article_response.json()["article"]["id"]
+
+    def _generate_submit_boundary_answers(self, article_id: str) -> None:
+        questions_response = self.client.post(
+            f"/api/articles/{article_id}/boundary-questions/generate",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, questions_response.status_code, questions_response.text)
+        questions = questions_response.json()["questions"]
+        for question in questions:
+            if question["is_required"]:
+                patch_response = self.client.patch(
+                    f"/api/articles/{article_id}/boundary-questions/{question['id']}",
+                    headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+                    json={"answer": f"Answer for {question['category']}"},
+                )
+                self.assertEqual(200, patch_response.status_code, patch_response.text)
+        submit_response = self.client.post(
+            f"/api/articles/{article_id}/boundary-questions/submit",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, submit_response.status_code, submit_response.text)
+
+    def _generate_and_approve_plan(self, article_id: str) -> dict[str, object]:
+        plan_response = self.client.post(
+            f"/api/articles/{article_id}/plan/generate",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, plan_response.status_code, plan_response.text)
+        plan = plan_response.json()["plan"]
+        approve_response = self.client.post(
+            f"/api/articles/{article_id}/plans/{plan['id']}/approve",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, approve_response.status_code, approve_response.text)
+        return plan
+
+    def _approve_assets(self, article_id: str, assets: list[dict[str, object]]) -> None:
+        fake_image = base64.b64encode(b"fake-image-bytes").decode("utf-8")
+        for asset in assets:
+            upload = self.client.post(
+                f"/api/articles/{article_id}/assets/{asset['id']}/upload",
+                headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+                json={
+                    "filename": f"{asset['id']}.png",
+                    "content_base64": fake_image,
+                    "content_type": "image/png",
+                },
+            )
+            self.assertEqual(200, upload.status_code, upload.text)
+            approve = self.client.post(
+                f"/api/articles/{article_id}/assets/{asset['id']}/approve",
+                headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            )
+            self.assertEqual(200, approve.status_code, approve.text)
+
+    def _approve_all_evidence(self, article_id: str, evidence_rows: list[dict[str, object]]) -> None:
+        for evidence in evidence_rows:
+            response = self.client.patch(
+                f"/api/articles/{article_id}/evidence/{evidence['id']}",
+                headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+                json={"review_status": "APPROVED"},
+            )
+            self.assertEqual(200, response.status_code, response.text)
+
+    def _accept_all_review_suggestions(
+        self,
+        article_id: str,
+        issues: list[dict[str, object]],
+        *,
+        kind: str,
+    ) -> None:
+        for issue in issues:
+            suggestion_id = issue["suggestion_id"]
+            endpoint = f"/api/articles/{article_id}/{kind}/suggestions/{suggestion_id}/accept"
+            response = self.client.post(
+                endpoint,
+                headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            )
+            self.assertEqual(200, response.status_code, response.text)
+
+    def _assert_research_artifacts_exist(self, manifests: list[dict[str, object]]) -> None:
+        for manifest in manifests:
+            artifacts = manifest.get("artifacts", [])
+            self.assertTrue(artifacts)
+            for artifact in artifacts:
+                object_key = artifact["object_key"]
+                path = self.object_root / object_key
+                self.assertTrue(path.exists(), f"Missing research artifact: {path}")
+
+    def _assert_commit_payload_valid(self, publish_commit: dict[str, object]) -> None:
+        manifest = publish_commit["content_bundle_manifest"]
+        content_path = manifest["content"]["path"]
+        assets = manifest["assets"]
+        commit_sha = publish_commit["commit_sha"]
+
+        markdown = self._git_show(commit_sha, content_path)
+        self.assertIn("---", markdown)
+        self.assertIn("title:", markdown)
+        self.assertIn("author:", markdown)
+        self.assertIn("##", markdown)
+
+        self.assertTrue(assets)
+        for asset in assets:
+            target = asset["target_path"]
+            blob = self._git_show(commit_sha, target)
+            self.assertTrue(blob)
+
+    def _assert_timeline_contains_major_actions(self, article_id: str) -> None:
+        detail = self.client.get(
+            f"/api/articles/{article_id}",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, detail.status_code, detail.text)
+        timeline = detail.json()["timeline"]
+        event_types = [event["event_type"] for event in timeline]
+        for required in (
+            "ARTICLE_CREATED",
+            "BOUNDARY_QUESTIONS_GENERATED",
+            "BOUNDARY_ANSWERS_SUBMITTED",
+            "PLAN_APPROVED",
+            "EVIDENCE_MATRIX_READY",
+            "PARALLEL_PRODUCTION_STARTED",
+            "AGENT_JOB_FAILED",
+            "AGENT_JOB_RETRIED",
+            "DRAFT_ASSEMBLED",
+            "SEO_REVIEW_COMPLETED",
+            "LANGUAGE_REVIEW_COMPLETED",
+            "FINAL_APPROVAL_GRANTED",
+            "PUBLISH_DRY_RUN_SUCCEEDED",
+            "PUBLISH_COMMIT_CREATED",
+        ):
+            self.assertIn(required, event_types)
+
+    def _git_show(self, commit_sha: str, relative_path: str) -> str:
+        result = subprocess.run(
+            ["git", "--git-dir", str(self.demo_repo_path), "show", f"{commit_sha}:{relative_path}"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        self.assertEqual(0, result.returncode, result.stderr or result.stdout)
+        return result.stdout
+
+
+if __name__ == "__main__":
+    unittest.main()