From 590c777eb96deb8d0b8a724a75f196c3aa0b753c Mon Sep 17 00:00:00 2001 From: ibrog Date: Thu, 17 Sep 2026 13:07:57 +0300 Subject: [PATCH] feat(pipeline): validate pre-review sync handoff Generated-with: Codex --- README.md | 13 +++ promptpilot/fallback_handoff.py | 72 ++++++++++-- promptpilot/pipeline_insights.py | 19 +++- promptpilot/project_pipeline.py | 8 ++ tests/fixtures/fallback-handoff-v1.json | 58 ++++++++++ tests/test_fallback_handoff.py | 140 +++++++++++++++++++++++- 6 files changed, 295 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index effb227..69d2876 100644 --- a/README.md +++ b/README.md @@ -1151,6 +1151,19 @@ review-depth, HEAD и два одинаковых полных GraphQL snapshot сохранении точной цели в content allowlist. Pending MERGE cleanup закрывает gate и возвращается в recovery следующим запуском. +Для cross-repo sync проект может выдать content-кандидат со stage +`pre-review-validation` и объектом `pre_review_sync`: положительные int64 +`intent_comment_id`/`done_comment_id`, lowercase SHA-1 `from`/`to`/`base`, +lowercase SHA-256 `identity_sha256` и RFC3339 `intent_created_at`/ +`done_created_at`. Набор из восьми полей точный, `head == to`, а done должен +быть позже intent. Такой кандидат никогда не бывает `integration_owner` или +`merge_executable` и требует `fallback_handoff: "target-v1"`: HMAC lease +связывает все поля, но не заменяет проверку происхождения. До gate полный скилл, +оставаясь read-only, обязан сначала доказать provenance стабильными GraphQL- +снимками, затем провести обычное полное ревью всего diff и тестов. Свежий gate +выполняется ровно один раз, когда аудит закончен и агент готов к первой мутации; +быстрый `action=audit`/`complete review` для этого stage запрещён. + `action=validated` — только read-only scheduling proof; ответ явно содержит `mutation_authorized=false`. Все GraphQL, ship, CI, base-sync и CAS-проверки остаются в полном скилле. Handoff обслуживает один PR и при отказе не переходит diff --git a/promptpilot/fallback_handoff.py b/promptpilot/fallback_handoff.py index 37146b7..0f81871 100644 --- a/promptpilot/fallback_handoff.py +++ b/promptpilot/fallback_handoff.py @@ -8,15 +8,39 @@ import re import time +from datetime import datetime PROTOCOL = "promptpilot-fallback-target-v1" -REVIEW_STAGES = {"review", "integration-review", "legacy-integration-review"} +PRE_REVIEW_VALIDATION_STAGE = "pre-review-validation" +PRE_REVIEW_SYNC_FIELDS = ( + "intent_comment_id", "done_comment_id", "from", "to", "base", + "identity_sha256", "intent_created_at", "done_created_at", +) +CONTENT_REVIEW_STAGES = {"review", PRE_REVIEW_VALIDATION_STAGE} +INTEGRATION_REVIEW_STAGES = {"integration-review", "legacy-integration-review"} +REVIEW_STAGES = CONTENT_REVIEW_STAGES | INTEGRATION_REVIEW_STAGES MERGE_STAGES = {"merge", "integration-merge-ready", "legacy-integration-merge-ready", "integration-merge-recovery"} +INTEGRATION_MERGE_STAGES = MERGE_STAGES - {"merge"} +INTEGRATION_STAGES = INTEGRATION_REVIEW_STAGES | INTEGRATION_MERGE_STAGES TTL = 7200 +def _rfc3339(value: object) -> datetime | None: + if (not isinstance(value, str) + or not re.fullmatch( + r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})", + value, + )): + return None + try: + return datetime.fromisoformat( + value[:-1] + "+00:00" if value.endswith("Z") else value) + except ValueError: + return None + + def identity(value: dict) -> dict: from .project_pipeline import PipelineError @@ -26,7 +50,35 @@ def identity(value: dict) -> dict: or not isinstance(value.get("stage"), str) or value.get("stage") not in REVIEW_STAGES | MERGE_STAGES): raise PipelineError("fallback target requires exact PR, HEAD and stage") - return {key: value[key] for key in ("number", "head", "stage")} + target = {key: value[key] for key in ("number", "head", "stage")} + metadata = value.get("pre_review_sync") + if value["stage"] != PRE_REVIEW_VALIDATION_STAGE: + if metadata is not None: + raise PipelineError("fallback target has pre-review metadata on another stage") + return target + if not isinstance(metadata, dict) or set(metadata) != set(PRE_REVIEW_SYNC_FIELDS): + raise PipelineError("pre-review validation requires exact sync metadata") + if any(type(metadata[field]) is not int + or not 0 < metadata[field] <= 2**63 - 1 + for field in ("intent_comment_id", "done_comment_id")): + raise PipelineError("pre-review validation requires positive comment IDs") + if any(not isinstance(metadata[field], str) + or not re.fullmatch(r"[0-9a-f]{40}", metadata[field]) + for field in ("from", "to", "base")): + raise PipelineError("pre-review validation requires exact commit identities") + if (not isinstance(metadata["identity_sha256"], str) + or not re.fullmatch(r"[0-9a-f]{64}", metadata["identity_sha256"])): + raise PipelineError("pre-review validation requires an exact identity digest") + intent_created = _rfc3339(metadata["intent_created_at"]) + done_created = _rfc3339(metadata["done_created_at"]) + if intent_created is None or done_created is None: + raise PipelineError("pre-review validation requires RFC3339 timestamps") + if done_created <= intent_created: + raise PipelineError("pre-review sync completion must follow its intent") + if value["head"] != metadata["to"]: + raise PipelineError("pre-review validation HEAD does not match sync destination") + target["pre_review_sync"] = {field: metadata[field] for field in PRE_REVIEW_SYNC_FIELDS} + return target def validate_health(health: dict) -> None: @@ -44,7 +96,7 @@ def validate_health(health: dict) -> None: barriers = [item for item in findings if item.get("code") == "single_flight_barrier"] if owner is not None: owner = identity(owner) - if (owner["stage"] in {"review", "merge"} or len(barriers) != 1 + if (owner["stage"] not in INTEGRATION_STAGES or len(barriers) != 1 or type(barriers[0].get("pr")) is not int or barriers[0]["pr"] != owner["number"]): raise PipelineError("fallback owner contradicts the single-flight barrier") @@ -53,7 +105,7 @@ def validate_health(health: dict) -> None: queues = {} for field, stages in (("review_candidates", REVIEW_STAGES), - ("content_review_candidates", {"review"}), + ("content_review_candidates", CONTENT_REVIEW_STAGES), ("merge_executable", MERGE_STAGES)): values = health.get(field) if not isinstance(values, list): @@ -66,7 +118,7 @@ def validate_health(health: dict) -> None: reviewing = queues["review_candidates"] content = queues["content_review_candidates"] merging = queues["merge_executable"] - if owner is not None and owner["stage"] in REVIEW_STAGES: + if owner is not None and owner["stage"] in INTEGRATION_REVIEW_STAGES: if reviewing != [owner] or merging: raise PipelineError("fallback integration REVIEW contradicts executable queues") else: @@ -100,7 +152,7 @@ def rest_only_review_owner(health: dict) -> bool: owner = identity(compatible.get("integration_owner")) except PipelineError: return False - return owner["stage"] in {"integration-review", "legacy-integration-review"} + return owner["stage"] in INTEGRATION_REVIEW_STAGES def health_gate(health: dict, stage: str, target: dict, *, election: bool) -> None: @@ -115,7 +167,7 @@ def health_gate(health: dict, stage: str, target: dict, *, election: bool) -> No owner = identity(owner) if owner is not None else None field = "review_candidates" if stage == "review" else "merge_executable" - if stage == "review" and expected["stage"] == "review" and not election: + if stage == "review" and expected["stage"] in CONTENT_REVIEW_STAGES and not election: field = "content_review_candidates" candidates = health.get(field) if not isinstance(candidates, list) or not candidates: @@ -125,18 +177,18 @@ def health_gate(health: dict, stage: str, target: dict, *, election: bool) -> No raise PipelineError("fallback allowlist contains duplicate PRs") if expected not in actual or ((election or stage == "merge") and actual[0] != expected): raise PipelineError("fallback target no longer matches the exact executable candidate") - if expected["stage"] not in {"review", "merge"}: + if expected["stage"] in INTEGRATION_STAGES: if owner != expected or actual != [expected]: raise PipelineError("fallback integration target is not the sole executable owner") if stage == "review" and health.get("merge_executable") != []: raise PipelineError("fallback integration REVIEW contradicts merge_executable") elif stage == "merge" and owner is not None: raise PipelineError("fallback ordinary merge is blocked by an integration owner") - if stage == "review" and expected["stage"] == "review": + if stage == "review" and expected["stage"] in CONTENT_REVIEW_STAGES: content = health.get("content_review_candidates") if not isinstance(content, list) or expected not in [identity(item) for item in content]: raise PipelineError("fallback content target was not proved by health") - if election and owner is not None and owner["stage"] in REVIEW_STAGES: + if election and owner is not None and owner["stage"] in INTEGRATION_REVIEW_STAGES: raise PipelineError("fallback content election bypasses integration REVIEW") diff --git a/promptpilot/pipeline_insights.py b/promptpilot/pipeline_insights.py index fee1d7d..58043c0 100644 --- a/promptpilot/pipeline_insights.py +++ b/promptpilot/pipeline_insights.py @@ -2359,7 +2359,7 @@ def execution_route(task, fallback_prompt: str, working_dir: str | None = None, from .project_pipeline import PipelineError try: - validate(preflight, stage) + fallback_lease = validate(preflight, stage) if command[-2:] != ["next", stage]: raise PipelineError("fallback handoff command must end with next and the exact stage") except (PipelineError, TypeError, ValueError) as exc: @@ -2374,10 +2374,27 @@ def execution_route(task, fallback_prompt: str, working_dir: str | None = None, envelope = {"protocol": "promptpilot-fallback-target-v1", "next_already_run": True, "command": command, "gate_command": gate_command, "preflight": preflight} + pre_review_guard = "" + if fallback_lease["target"]["stage"] == "pre-review-validation": + pre_review_guard = ( + "Это специальный content-lane этап pre-review-validation, а не " + "integration review. Подписанный envelope только фиксирует все " + "восемь полей pre_review_sync и HEAD; он не доказывает provenance " + "и не разрешает быстрый результат. До gate_command, оставаясь полностью " + "read-only, сначала выполни предусмотренную скиллом полную стабильную " + "GraphQL-проверку происхождения sync-коммита. Только после её успеха " + "проверь весь diff PR как обычное содержательное ревью и выполни все " + "уместные полные тесты. Этот target нельзя завершать через быстрый " + "action=audit или `complete review`. Лишь когда provenance и аудит " + "полностью завершены и ты готов к первой мутации, переходи к описанному " + "ниже одноразовому gate_command непосредственно перед этой мутацией. " + "При любой ошибке provenance остановись без gate и без мутаций.\n\n" + ) prompt = ( "PromptPilot уже выполнил election next. Не запускай next повторно " "и не выбирай другую цель. Полностью прочитай канонический скилл " "и его legacy-протокол. Используй только exact target из envelope. " + f"{pre_review_guard}" "Непосредственно перед первой мутацией выполни gate_command ровно один " "раз: он заново " "запускает полный pipelinehealth и проверяет ту же цель. Требуется " diff --git a/promptpilot/project_pipeline.py b/promptpilot/project_pipeline.py index 139d209..012e267 100644 --- a/promptpilot/project_pipeline.py +++ b/promptpilot/project_pipeline.py @@ -1247,6 +1247,14 @@ def next_review(gh: GitHub, config: dict, *, config_path: str | None = None) -> if not candidates: return {"action": "empty", "verdict": "ПУСТО", "reason": review_empty_reason(health)} item = candidates[0] + if item.get("stage") == "pre-review-validation": + if config.get("fallback_handoff") != "target-v1": + raise PipelineError( + "pre-review validation requires the exact-target fallback protocol") + return fallback_target( + config, health, "review", item, + "pre-review sync provenance requires validation and a full content review", + ) if item.get("stage") != "review": return fallback_target(config, health, "review", item, "integration/base-sync state requires the full skill") diff --git a/tests/fixtures/fallback-handoff-v1.json b/tests/fixtures/fallback-handoff-v1.json index d8c38c2..a515b5a 100644 --- a/tests/fixtures/fallback-handoff-v1.json +++ b/tests/fixtures/fallback-handoff-v1.json @@ -210,6 +210,64 @@ } ] } + }, + { + "stage": "review", + "target": { + "number": 84, + "head": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee", + "stage": "pre-review-validation", + "pre_review_sync": { + "intent_comment_id": 1001, + "done_comment_id": 1002, + "from": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "to": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee", + "base": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "identity_sha256": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", + "intent_created_at": "2026-09-17T08:00:00Z", + "done_created_at": "2026-09-17T08:02:00Z" + } + }, + "health": { + "state": "green", + "findings": [], + "integration_owner": null, + "review_candidates": [ + { + "number": 84, + "head": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee", + "stage": "pre-review-validation", + "pre_review_sync": { + "intent_comment_id": 1001, + "done_comment_id": 1002, + "from": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "to": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee", + "base": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "identity_sha256": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", + "intent_created_at": "2026-09-17T08:00:00Z", + "done_created_at": "2026-09-17T08:02:00Z" + } + } + ], + "content_review_candidates": [ + { + "number": 84, + "head": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee", + "stage": "pre-review-validation", + "pre_review_sync": { + "intent_comment_id": 1001, + "done_comment_id": 1002, + "from": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "to": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee", + "base": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "identity_sha256": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", + "intent_created_at": "2026-09-17T08:00:00Z", + "done_created_at": "2026-09-17T08:02:00Z" + } + } + ], + "merge_executable": [] + } } ] } diff --git a/tests/test_fallback_handoff.py b/tests/test_fallback_handoff.py index 3e275a9..4c737aa 100644 --- a/tests/test_fallback_handoff.py +++ b/tests/test_fallback_handoff.py @@ -19,6 +19,21 @@ HEAD = "a" * 40 +def pre_review_sync(**changes): + value = { + "intent_comment_id": 101, + "done_comment_id": 102, + "from": "b" * 40, + "to": HEAD, + "base": "c" * 40, + "identity_sha256": "d" * 64, + "intent_created_at": "2026-09-17T08:00:00Z", + "done_created_at": "2026-09-17T08:02:00Z", + } + value.update(changes) + return value + + @pytest.fixture def config(monkeypatch, tmp_path): monkeypatch.setenv("PP_PIPELINE_LEASE_KEY_FILE", str(tmp_path / "lease.key")) @@ -29,14 +44,18 @@ def config(monkeypatch, tmp_path): def health(target_stage="integration-review"): target = {"number": 42, "head": HEAD, "stage": target_stage, "review_depth": 2} - integration = target_stage not in {"review", "merge"} + if target_stage == handoff.PRE_REVIEW_VALIDATION_STAGE: + target["pre_review_sync"] = pre_review_sync() + integration = target_stage in handoff.INTEGRATION_STAGES reviewing = target_stage in handoff.REVIEW_STAGES return {"state": "yellow" if integration else "green", "findings": ([{"code": "single_flight_barrier", "severity": "yellow", "pr": 42}] if integration else []), "integration_owner": target if integration else None, "review_candidates": [target] if reviewing else [], - "content_review_candidates": [target] if target_stage == "review" else [], + "content_review_candidates": ([target] + if target_stage in handoff.CONTENT_REVIEW_STAGES + else []), "merge_executable": [] if reviewing else [target]} @@ -55,7 +74,7 @@ def json(self, *args, **kwargs): @pytest.mark.parametrize("target_stage", [ - "review", "integration-review", "legacy-integration-review", + "review", "pre-review-validation", "integration-review", "legacy-integration-review", "integration-merge-ready", "legacy-integration-merge-ready", "integration-merge-recovery", ]) def test_cli_dispatch_handoff_runs_exactly_election_and_fresh_gate(config, monkeypatch, target_stage): @@ -97,6 +116,17 @@ def preflight(_execution, actual, _directory): assert "не повторяй gate_command и next" in route["prompt"] assert "ИТОГ: НЕ СМОГ (gate-fallback: <точный error" in route["prompt"] assert "Один envelope — один PR" in route["prompt"] + if target_stage == handoff.PRE_REVIEW_VALIDATION_STAGE: + assert "все восемь полей pre_review_sync" in route["prompt"] + assert "полную стабильную GraphQL-проверку" in route["prompt"] + assert "проверь весь diff" in route["prompt"] + assert "нельзя завершать через быстрый action=audit" in route["prompt"] + assert "До gate_command, оставаясь полностью read-only" in route["prompt"] + assert "После gate_command сначала" not in route["prompt"] + assert route["prompt"].index("До gate_command") < route["prompt"].index( + "Непосредственно перед первой мутацией") + else: + assert "все восемь полей pre_review_sync" not in route["prompt"] assert route["preflight"]["target"] == route["target"] assert len(scans) == 1 code, gated = invoke(route["gate_command"][1:]) @@ -229,6 +259,108 @@ def test_content_fallback_keeps_target_lease_across_unrelated_owner(config): handoff.health_gate(fresh, "review", target, election=True) +def test_pre_review_fallback_keeps_metadata_bound_across_unrelated_owner(config): + source = health(handoff.PRE_REVIEW_VALIDATION_STAGE) + target = source["review_candidates"][0] + preflight = handoff.create(config, source, "review", target, "provenance") + lease = handoff.validate(preflight, "review") + assert lease["target"] == handoff.identity(target) + assert lease["target"]["pre_review_sync"] == pre_review_sync() + + fresh = health("integration-review") + fresh["integration_owner"]["number"] = 90 + fresh["findings"][0]["pr"] = 90 + fresh["content_review_candidates"] = [target] + handoff.health_gate(fresh, "review", lease["target"], election=False) + with pytest.raises(pp.PipelineError): + handoff.health_gate(fresh, "review", lease["target"], election=True) + + +@pytest.mark.parametrize("field,value", [ + ("intent_comment_id", 201), + ("done_comment_id", 202), + ("from", "e" * 40), + ("to", "e" * 40), + ("base", "e" * 40), + ("identity_sha256", "e" * 64), + ("intent_created_at", "2026-09-17T08:00:01Z"), + ("done_created_at", "2026-09-17T08:02:01Z"), +]) +def test_pre_review_fresh_gate_compares_every_metadata_field(config, field, value): + source = health(handoff.PRE_REVIEW_VALIDATION_STAGE) + target = source["review_candidates"][0] + preflight = handoff.create(config, source, "review", target, "provenance") + expected = handoff.validate(preflight, "review")["target"] + fresh = copy.deepcopy(source) + for candidate in fresh["review_candidates"] + fresh["content_review_candidates"]: + candidate["pre_review_sync"][field] = value + if field == "to": + candidate["head"] = value + with pytest.raises(pp.PipelineError, match="exact executable"): + handoff.health_gate(fresh, "review", expected, election=False) + + +@pytest.mark.parametrize("mutation", [ + lambda target: target.pop("pre_review_sync"), + lambda target: target["pre_review_sync"].pop("base"), + lambda target: target["pre_review_sync"].update(extra="x"), + lambda target: target["pre_review_sync"].update(intent_comment_id=True), + lambda target: target["pre_review_sync"].update(done_comment_id=0), + lambda target: target["pre_review_sync"].update(done_comment_id=2**63), + lambda target: target["pre_review_sync"].update(**{"from": "B" * 40}), + lambda target: target["pre_review_sync"].update(identity_sha256="d" * 63), + lambda target: target["pre_review_sync"].update(intent_created_at="not-a-time"), + lambda target: target["pre_review_sync"].update(done_created_at="2026-09-17 08:02:00Z"), + lambda target: target["pre_review_sync"].update( + done_created_at="2026-09-17T07:59:59Z"), + lambda target: target["pre_review_sync"].update(to="e" * 40), +]) +def test_pre_review_target_schema_fails_closed(mutation): + target = health(handoff.PRE_REVIEW_VALIDATION_STAGE)["review_candidates"][0] + mutation(target) + with pytest.raises(pp.PipelineError): + handoff.identity(target) + + +def test_pre_review_stage_cannot_be_integration_owner_or_merge_executable(): + source = health(handoff.PRE_REVIEW_VALIDATION_STAGE) + target = source["review_candidates"][0] + as_owner = copy.deepcopy(source) + as_owner["integration_owner"] = target + as_owner["findings"] = [ + {"code": "single_flight_barrier", "severity": "yellow", "pr": 42}, + ] + with pytest.raises(pp.PipelineError, match="owner"): + handoff.validate_health(as_owner) + + in_merge = copy.deepcopy(source) + in_merge["merge_executable"] = [target] + with pytest.raises(pp.PipelineError, match="merge_executable"): + handoff.validate_health(in_merge) + + +def test_next_review_forces_targeted_full_skill_for_pre_review(config, monkeypatch): + source = health(handoff.PRE_REVIEW_VALIDATION_STAGE) + monkeypatch.setattr(pp, "run_health", lambda *_, **__: source) + monkeypatch.setattr( + pp, "stable_timeline", + lambda *_: pytest.fail("pre-review validation must not enter fast audit"), + ) + + result = pp.next_review(object(), config) + + assert result["action"] == "fallback" + assert result["target"] == handoff.identity(source["review_candidates"][0]) + assert result["handoff"]["protocol"] == handoff.PROTOCOL + + +def test_next_review_pre_review_fails_closed_without_target_protocol(monkeypatch): + source = health(handoff.PRE_REVIEW_VALIDATION_STAGE) + monkeypatch.setattr(pp, "run_health", lambda *_, **__: source) + with pytest.raises(pp.PipelineError, match="exact-target"): + pp.next_review(object(), {}) + + def test_legacy_config_does_not_silently_opt_in(monkeypatch): monkeypatch.setattr(pp, "run_health", lambda *_, **__: health()) assert pp.next_review(object(), {}) == { @@ -253,7 +385,7 @@ def test_opted_in_red_health_never_routes_to_manual_mutations(config, monkeypatc def test_onebase_producer_parity_fixture(config): corpus = json.loads((Path(__file__).parent / "fixtures" / "fallback-handoff-v1.json").read_text(encoding="utf-8")) - assert corpus["protocol"] == handoff.PROTOCOL and len(corpus["cases"]) == 7 + assert corpus["protocol"] == handoff.PROTOCOL and len(corpus["cases"]) == 8 for case in corpus["cases"]: result = handoff.create(config, case["health"], case["stage"], case["target"], "parity") assert handoff.validate(result, case["stage"])["target"] == case["target"]