feat(pipeline): add deploy-staging gate before prod deploy (ORCH-35)

tests/test_qg.py

@@ -19,6 +19,7 @@ from src.qg.checks import (
     check_tests_passed,
     check_tests_local,
     check_deploy_status,
+    check_staging_status,
 )
 from src.stages import get_qg_for_stage
 
@@ -448,3 +449,108 @@ class TestCheckTestsLocal:
         assert "../../tests/" in cmd
         assert kwargs["cwd"] == os.path.join(str(tmp_path), "src", "api")
 
+
+
+class TestCheckStagingStatus:
+    """ORCH-35: deploy-staging -> deploy gate reads machine-readable staging_status:
+    from 15-staging-log.md frontmatter. Mirrors check_deploy_status pattern."""
+
+    def _write_log(self, repo_dir, content):
+        wi_dir = repo_dir / "docs" / "work-items" / "ET-035"
+        wi_dir.mkdir(parents=True, exist_ok=True)
+        (wi_dir / "15-staging-log.md").write_text(content)
+
+    def test_success_verdict_passes(self, setup_work_item_dir):
+        self._write_log(
+            setup_work_item_dir,
+            "---\nstaging_status: SUCCESS\ntimestamp: 2026-06-05T00:00:00Z\n---\n\nAll staging tests passed.\n",
+        )
+        from src.qg.checks import check_staging_status
+        passed, reason = check_staging_status("enduro-trails", "ET-035")
+        assert passed is True
+        assert "SUCCESS" in reason
+
+    def test_failed_verdict_fails(self, setup_work_item_dir):
+        self._write_log(
+            setup_work_item_dir,
+            "---\nstaging_status: FAILED\ntimestamp: 2026-06-05T00:00:00Z\n---\n\n2 tests failed.\n",
+        )
+        from src.qg.checks import check_staging_status
+        passed, reason = check_staging_status("enduro-trails", "ET-035")
+        assert passed is False
+        assert "FAILED" in reason
+
+    def test_no_file_fails(self, setup_work_item_dir):
+        from src.qg.checks import check_staging_status
+        passed, reason = check_staging_status("enduro-trails", "ET-035")
+        assert passed is False
+        assert "not found" in reason.lower()
+
+    def test_no_field_fails(self, setup_work_item_dir):
+        # Frontmatter present but no staging_status field -> must NOT pass.
+        self._write_log(
+            setup_work_item_dir,
+            "---\nversion: v0.0.3\n---\n\nStatus: all good (prose only).\n",
+        )
+        from src.qg.checks import check_staging_status
+        passed, reason = check_staging_status("enduro-trails", "ET-035")
+        assert passed is False
+
+    def test_prose_only_no_frontmatter_fails(self, setup_work_item_dir):
+        # Prose mentioning SUCCESS but no machine-readable frontmatter -> fail.
+        self._write_log(
+            setup_work_item_dir,
+            "# Staging Log\n\nStatus: SUCCESS (prose, not frontmatter).\n",
+        )
+        from src.qg.checks import check_staging_status
+        passed, reason = check_staging_status("enduro-trails", "ET-035")
+        assert passed is False
+
+    def test_origin_main_success_passes_when_absent_in_worktree(self, monkeypatch):
+        # Deployer merged 15-staging-log.md into main; not in worktree -> recover from main.
+        monkeypatch.setattr(
+            "src.qg.checks._staging_log_from_main",
+            lambda repo, wi: "---\nstaging_status: SUCCESS\n---\n\nAll good.\n",
+        )
+        from src.qg.checks import check_staging_status
+        passed, reason = check_staging_status("enduro-trails", "ET-035-main")
+        assert passed is True
+        assert "SUCCESS" in reason
+
+    def test_origin_main_failed_fails(self, monkeypatch):
+        monkeypatch.setattr(
+            "src.qg.checks._staging_log_from_main",
+            lambda repo, wi: "---\nstaging_status: FAILED\n---\n\nboom.\n",
+        )
+        from src.qg.checks import check_staging_status
+        passed, reason = check_staging_status("enduro-trails", "ET-035-main")
+        assert passed is False
+        assert "FAILED" in reason
+
+    def test_absent_everywhere_fails(self, monkeypatch):
+        monkeypatch.setattr(
+            "src.qg.checks._staging_log_from_main", lambda repo, wi: None
+        )
+        from src.qg.checks import check_staging_status
+        passed, reason = check_staging_status("enduro-trails", "ET-035-absent")
+        assert passed is False
+        assert "not found" in reason.lower()
+
+    def test_deploy_staging_qg_is_check_staging_status(self):
+        assert get_qg_for_stage("deploy-staging") == "check_staging_status"
+
+    def test_registered_in_qg_checks(self):
+        from src.qg.checks import QG_CHECKS, check_staging_status
+        assert QG_CHECKS.get("check_staging_status") is check_staging_status
+
+    def test_deploy_stage_qg_still_check_deploy_status(self):
+        """Regression: existing deploy QG must not be broken."""
+        assert get_qg_for_stage("deploy") == "check_deploy_status"
+
+    def test_stage_chain(self):
+        """Full chain: testing->deploy-staging->deploy->done."""
+        from src.stages import get_next_stage
+        assert get_next_stage("testing") == "deploy-staging"
+        assert get_next_stage("deploy-staging") == "deploy"
+        assert get_next_stage("deploy") == "done"
+

tests/test_stage_engine.py

@@ -136,7 +136,7 @@ class TestHappyPathAgentSelection:
             ("architecture", "development", "developer"),
             ("development", "review", "reviewer"),
             ("review", "testing", "tester"),
-            ("testing", "deploy", "deployer"),
+            ("testing", "deploy-staging", "deployer"),
         ],
     )
     def test_advance_launches_current_stage_agent(
@@ -507,6 +507,120 @@ class TestAnalysisApprovedFlow:
         flow.assert_called_once()
 
 
+
+
+# ---------------------------------------------------------------------------
+# ORCH-35: deploy-staging gate — rollback on staging failure
+# ---------------------------------------------------------------------------
+class TestStagingGate:
+    """deploy-staging -> deploy must be gated on check_staging_status.
+    FAILED verdict rolls back to development (same as deploy БАГ-8 pattern:
+    staging failure = code is bad, needs developer fix)."""
+
+    def test_staging_success_advances_to_deploy(self, monkeypatch):
+        """Happy path: staging SUCCESS -> advance to deploy (no agent launched)."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS, "check_staging_status": _pass},
+        )
+        task_id = _make_task("deploy-staging")
+        res = advance_stage(
+            task_id, "deploy-staging", "enduro-trails", "ET-035",
+            "feature/ET-035-x", finished_agent="deployer",
+        )
+        assert res.advanced is True
+        assert res.to_stage == "deploy"
+        assert _stage(task_id) == "deploy"
+        # deploy-staging has agent=deployer, so deployer is enqueued for deploy stage
+        assert res.enqueued_agent == "deployer"
+        jobs = _jobs()
+        assert len(jobs) == 1
+        assert jobs[0]["agent"] == "deployer"
+
+    def test_staging_failed_rolls_back_to_development(self, monkeypatch):
+        """ORCH-35: staging FAILED -> roll back to development, not to testing."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS,
+             "check_staging_status": _fail("Staging status: FAILED")},
+        )
+        task_id = _make_task("deploy-staging")
+        res = advance_stage(
+            task_id, "deploy-staging", "enduro-trails", "ET-035",
+            "feature/ET-035-x", finished_agent="deployer",
+        )
+        assert res.advanced is False
+        assert res.rolled_back_to == "development"
+        assert _stage(task_id) == "development"   # NOT deploy, NOT testing
+        assert res.alerted is True
+        assert stage_engine.set_issue_blocked.called
+        assert stage_engine.send_telegram.called
+
+    def test_staging_failed_does_not_reach_deploy(self, monkeypatch):
+        """Prod deploy is unreachable if staging gate is not green."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS,
+             "check_staging_status": _fail("Staging log not found")},
+        )
+        task_id = _make_task("deploy-staging")
+        res = advance_stage(
+            task_id, "deploy-staging", "enduro-trails", "ET-035",
+            "feature/ET-035-x", finished_agent="deployer",
+        )
+        assert res.advanced is False
+        # Task must NOT be in deploy stage
+        assert _stage(task_id) != "deploy"
+
+    def test_staging_missing_log_rolls_back(self, monkeypatch):
+        """Missing 15-staging-log.md -> gate fails -> rollback to development."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS,
+             "check_staging_status": _fail("Staging log not found (15-staging-log.md)")},
+        )
+        task_id = _make_task("deploy-staging")
+        res = advance_stage(
+            task_id, "deploy-staging", "enduro-trails", "ET-035",
+            "feature/ET-035-x", finished_agent="deployer",
+        )
+        assert res.advanced is False
+        assert _stage(task_id) == "development"
+
+    def test_testing_to_deploy_staging_advance(self, monkeypatch):
+        """testing -> deploy-staging: deployer is enqueued (ORCH-35 chain check)."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS, "check_tests_passed": _pass},
+        )
+        task_id = _make_task("testing")
+        res = advance_stage(
+            task_id, "testing", "enduro-trails", "ET-035",
+            "feature/ET-035-x", finished_agent="tester",
+        )
+        assert res.advanced is True
+        assert res.to_stage == "deploy-staging"
+        assert _stage(task_id) == "deploy-staging"
+        assert res.enqueued_agent == "deployer"
+
+    def test_deploy_still_rolls_back_on_check_deploy_status_fail(self, monkeypatch):
+        """Existing БАГ-8 rollback must still work for deploy stage (regression guard)."""
+        monkeypatch.setattr(
+            stage_engine, "QG_CHECKS",
+            {**stage_engine.QG_CHECKS,
+             "check_deploy_status": _fail("Deploy status: FAILED")},
+        )
+        task_id = _make_task("deploy")
+        res = advance_stage(
+            task_id, "deploy", "enduro-trails", "ET-011",
+            "feature/ET-011-x", finished_agent="deployer",
+        )
+        assert res.advanced is False
+        assert res.rolled_back_to == "development"
+        assert _stage(task_id) == "development"
+        assert res.alerted is True
+
+
 # ---------------------------------------------------------------------------
 # launcher + plane both delegate to the engine
 # ---------------------------------------------------------------------------