From 14d23735fca5fad7ac136467ebcd4d531fbeb5e8 Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Thu, 30 Apr 2026 12:32:11 -0700 Subject: [PATCH 1/3] Fix #2368: synthetic-session exemption for slice integration-branch pushes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Multi-slice implement phases were stranded because two correct-in-isolation behaviours collided: * #2028 (gateway): every pipeline-session push without `consensus_push=true` is rejected unless the target is in `INFRASTRUCTURE_BRANCHES`. * #2220 (orchestrator): `create_slice_integration_branch` registers a synthetic pipeline session and pushes `parent:refs/heads/` via `/api/v1/git/push` so the slice PR's diff is non-empty before agents spawn. Result: every multi-slice pipeline 403'd on the per-slice integration push and 15 slices failed before any agent ran (latent since #2220, only unmasked once #2337 stopped silently demoting multi-slice contracts to monolithic implement). Fix: add a path + flag exemption to the gateway's infrastructure-push bypass — a session whose `synthetic` flag is `True` (only settable by the launcher, since `/api/v1/sessions/create` is gated on `require_launcher_auth`) targeting `egg//(slice|phase)-N` is treated as orchestrator infrastructure. No new orchestrator-role push surface is introduced; agent BRC enforcement is unchanged. Bonus: the slice integration-branch name now derives from `pipeline.branch` directly, so a qualifier suffix (`-v3`, `-backend`) is preserved. Two qualified pipelines for the same issue would otherwise collide in the `egg/issue-N/slice-M` namespace. Tests cover the four parent shapes (issue / qualified-issue / JIRA / legacy `phase-N`), confirm a non-synthetic session is still blocked on slice-shaped branches (the synthetic flag is load-bearing), and add an orchestrator regression assert that `create_slice_integration_branch` precedes `_run_concurrent_phase` so a future refactor can't re-introduce the spawn-then-push ordering bug. --- gateway/gateway.py | 53 ++++- gateway/tests/test_pipeline_push_block.py | 192 ++++++++++++++++ orchestrator/gateway_client.py | 17 +- orchestrator/routes/pipelines.py | 6 +- .../tests/test_slice_run_loop_integration.py | 208 ++++++++++++++++++ 5 files changed, 468 insertions(+), 8 deletions(-) diff --git a/gateway/gateway.py b/gateway/gateway.py index a79e53cc46..c66f1fd67c 100644 --- a/gateway/gateway.py +++ b/gateway/gateway.py @@ -1075,6 +1075,15 @@ def config_reload() -> Response: return jsonify({"status": "ok", "message": "Configuration reloaded"}) +# Slice integration-branch shape for the synthetic-session exemption (#2368). +# Matches ``egg//(slice|phase)-`` where ```` is the parent +# pipeline branch (issue-driven, JIRA-driven, or qualifier-suffixed). Only +# orchestrator-issued sessions can ever set ``synthetic=True`` (the launcher +# secret gates ``/api/v1/sessions/create``), so this exemption is not reachable +# from a sandboxed agent's session token. +_SLICE_INTEGRATION_BRANCH_RE = re.compile(r"^egg/[A-Za-z0-9][A-Za-z0-9_/-]*/(?:slice|phase)-\d+$") + + @app.route("/api/v1/git/push", methods=["POST"]) @require_session_or_launcher_auth def git_push() -> tuple[Response, int] | Response: @@ -1285,6 +1294,42 @@ def git_push() -> tuple[Response, int] | Response: INFRASTRUCTURE_BRANCHES = {CHECKPOINT_BRANCH, PIPELINE_STATE_BRANCH} is_infrastructure_push = branch in INFRASTRUCTURE_BRANCHES + # Slice integration-branch creation (#2368): the orchestrator pre-creates + # ``egg//(slice|phase)-N`` on origin from the parent branch via a + # synthetic, launcher-authenticated session before any agent runs. That + # push is orchestrator infrastructure — not an agent BRC propose — so it + # must bypass the pipeline-session push block introduced in #2028. The + # ``synthetic=True`` flag can only be set by the launcher (the + # ``/api/v1/sessions/create`` endpoint is gated by ``require_launcher_auth``), + # so a sandboxed agent's session token cannot reach this branch. + is_slice_integration_push = False + if not is_infrastructure_push and _SLICE_INTEGRATION_BRANCH_RE.match(branch): + # ``Session.synthetic`` is a ``bool`` (default ``False``); only an + # orchestrator-issued session can carry ``synthetic=True`` because + # ``/api/v1/sessions/create`` is gated on the launcher secret. Use + # an identity check rather than a truthiness test so a future + # surface that ever stores something other than ``True`` (and any + # MagicMock fake whose default attr is truthy) cannot accidentally + # opt into the exemption. + if hasattr(g, "session") and getattr(g.session, "synthetic", False) is True: + is_slice_integration_push = True + is_infrastructure_push = True + audit_log( + "push_slice_integration_exempt", + "git_push", + success=True, + details={ + "repo_path": repo_path, + "remote": remote, + "refspec": refspec, + "branch": branch, + "reason": ( + "Synthetic-session slice integration branch push — " + "orchestrator infrastructure (#2368)" + ), + }, + ) + repo_info = parse_owner_repo(repo) if repo_info: # Infrastructure operations — always accessible regardless of @@ -1292,6 +1337,12 @@ def git_push() -> tuple[Response, int] | Response: # infrastructure branch pushes (checkpoints, pipeline state). is_ckpt_repo = _is_checkpoint_repo_for_request(repo_info.owner, repo_info.repo) if is_infrastructure_push or is_ckpt_repo: + if is_ckpt_repo: + exempt_type = "checkpoint_repo" + elif is_slice_integration_push: + exempt_type = "slice_integration_branch" + else: + exempt_type = "infrastructure_branch" audit_log( "push_infrastructure_exempt", "git_push", @@ -1300,7 +1351,7 @@ def git_push() -> tuple[Response, int] | Response: "repo": repo, "branch": branch, "reason": "Infrastructure operation exempt from private mode policy", - "exempt_type": "checkpoint_repo" if is_ckpt_repo else "infrastructure_branch", + "exempt_type": exempt_type, }, ) else: diff --git a/gateway/tests/test_pipeline_push_block.py b/gateway/tests/test_pipeline_push_block.py index 8c90fb6c33..118d9ef6e6 100644 --- a/gateway/tests/test_pipeline_push_block.py +++ b/gateway/tests/test_pipeline_push_block.py @@ -49,6 +49,7 @@ def _make_session( role: str = "coder", pipeline_id: str | None = "issue-1669", assigned_branch: str | None = "egg/issue-1669", + synthetic: bool = False, ) -> MagicMock: """Create a mock session with the given agent role and pipeline context.""" mock_session = MagicMock() @@ -59,6 +60,7 @@ def _make_session( mock_session.phase = "implement" mock_session.pipeline_id = pipeline_id mock_session.assigned_branch = assigned_branch + mock_session.synthetic = synthetic return mock_session @@ -689,3 +691,193 @@ def test_launcher_push_missing_mode_uses_session_default(self, client): assert response.status_code == 200, ( f"Expected 200 with no mode, got {response.status_code}: {response.data!r}" ) + + +class TestSliceIntegrationBranchExemption: + """Slice integration-branch creation exemption (#2368). + + The orchestrator's ``create_slice_integration_branch`` registers a + synthetic, launcher-authed session and pushes + ``parent:refs/heads/egg//slice-N`` so the slice PR's diff is + non-empty before agents spawn. That push is orchestrator + infrastructure and must bypass the #2028 pipeline-session block. + The exemption is keyed on the session's ``synthetic`` flag — only + the launcher can set it — and the slice integration-branch shape. + """ + + def test_synthetic_session_slice_branch_push_allowed(self, client): + """Synthetic-session push to ``egg/issue-N/slice-M`` is allowed.""" + session = _make_session(synthetic=True, assigned_branch="egg/issue-2261/slice-7") + patches = _push_context(session) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patches[6], + patches[7], + ): + response = _do_push( + client, + refspec="egg/issue-2261:refs/heads/egg/issue-2261/slice-7", + ) + assert response.status_code == 200, ( + f"Expected 200 for synthetic slice integration push, " + f"got {response.status_code}: {response.data!r}" + ) + + def test_synthetic_session_qualified_slice_branch_push_allowed(self, client): + """Qualifier-suffixed branches (#2368 bonus) — ``egg/issue-N-v3/slice-M`` — pass.""" + session = _make_session(synthetic=True, assigned_branch="egg/issue-2261-v3/slice-7") + patches = _push_context(session) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patches[6], + patches[7], + ): + response = _do_push( + client, + refspec="egg/issue-2261-v3:refs/heads/egg/issue-2261-v3/slice-7", + ) + assert response.status_code == 200 + + def test_synthetic_session_jira_slice_branch_push_allowed(self, client): + """JIRA-driven branches — ``egg/KORE-1234/slice-M`` — pass.""" + session = _make_session(synthetic=True, assigned_branch="egg/KORE-1234/slice-3") + patches = _push_context(session) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patches[6], + patches[7], + ): + response = _do_push( + client, + refspec="egg/KORE-1234:refs/heads/egg/KORE-1234/slice-3", + ) + assert response.status_code == 200 + + def test_synthetic_session_legacy_phase_branch_push_allowed(self, client): + """Legacy ``phase-N`` slice IDs (pre-#2137) still flow through the loader, + so the integration-branch exemption must accept them too.""" + session = _make_session(synthetic=True, assigned_branch="egg/issue-2261/phase-1") + patches = _push_context(session) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patches[6], + patches[7], + ): + response = _do_push( + client, + refspec="egg/issue-2261:refs/heads/egg/issue-2261/phase-1", + ) + assert response.status_code == 200 + + def test_non_synthetic_session_slice_branch_push_blocked(self, client): + """Agent (non-synthetic) push to a slice integration branch is still blocked. + + Agents reach a slice's integration branch via ``mcp__brc__propose`` + (``consensus_push=true``); a direct push is the very pattern #2028 is + designed to catch. The exemption MUST gate on ``synthetic=True`` — + if the regex alone is enough, agents can use slice-shaped branch + names to bypass enforcement. + """ + session = _make_session(synthetic=False, assigned_branch="egg/issue-2261/slice-7") + patches = _push_context(session) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patches[6], + patches[7], + ): + response = _do_push( + client, + refspec="egg/issue-2261:refs/heads/egg/issue-2261/slice-7", + ) + assert response.status_code == 403, ( + "Non-synthetic session push to slice integration branch must still " + "be blocked by pipeline-session enforcement" + ) + data = json.loads(response.data) + assert "pipeline sessions" in data["message"].lower() + + def test_synthetic_session_non_slice_branch_still_blocked(self, client): + """Synthetic flag alone is not enough — branch must match the slice shape. + + Defends against future code paths that mark a session synthetic for + unrelated reasons but would bypass the agent-push block if the + branch-shape gate were missing. + """ + session = _make_session(synthetic=True, assigned_branch="egg/issue-2261") + patches = _push_context(session) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patches[6], + patches[7], + ): + response = _do_push(client, refspec="egg/issue-2261") + assert response.status_code == 403 + + def test_audit_event_records_slice_integration_exempt_type(self, client): + """Exemption emits a distinct audit event so operators can trace + synthetic-session pushes separately from checkpoint/state writes.""" + session = _make_session(synthetic=True, assigned_branch="egg/issue-2261/slice-7") + patches = _push_context(session) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patches[6], + patches[7], + ): + with patch.object(gateway, "audit_log") as mock_audit: + response = _do_push( + client, + refspec="egg/issue-2261:refs/heads/egg/issue-2261/slice-7", + ) + assert response.status_code == 200 + events = [ + (call.args[0] if call.args else None, call.kwargs.get("details") or {}) + for call in mock_audit.call_args_list + ] + slice_events = [e for e in events if e[0] == "push_slice_integration_exempt"] + assert slice_events, ( + f"Expected push_slice_integration_exempt event, got: {[e[0] for e in events]}" + ) + infra_events = [ + e + for e in events + if e[0] == "push_infrastructure_exempt" + and e[1].get("exempt_type") == "slice_integration_branch" + ] + assert infra_events, ( + "Expected push_infrastructure_exempt with exempt_type=slice_integration_branch" + ) diff --git a/orchestrator/gateway_client.py b/orchestrator/gateway_client.py index 95deac2bf9..f9593a0ef9 100644 --- a/orchestrator/gateway_client.py +++ b/orchestrator/gateway_client.py @@ -1529,12 +1529,17 @@ def create_slice_integration_branch( """Create the slice integration branch on origin from ``parent_branch``. Sends ``git push origin parent_branch:refs/heads/integration_branch`` - through the existing per-agent ``/api/v1/git/push`` endpoint so - no privileged orchestrator-role surface is introduced - (decision-15). The branch ownership check uses the - ``integration_branch`` name as the target — naming convention - ``egg/issue-N/slice-M`` is owned by the orchestrator role's - existing prefix-allowlist. + via a synthetic, launcher-authenticated session through + ``/api/v1/git/push``. The gateway treats the push as + orchestrator infrastructure: the synthetic flag (only settable + by ``/api/v1/sessions/create``, which is gated on the launcher + secret) combined with the slice integration-branch name + ``egg//(slice|phase)-N`` short-circuits the + pipeline-session push block from #2028 — see the + ``_SLICE_INTEGRATION_BRANCH_RE`` exemption in + ``gateway/gateway.py``. The branch itself still passes the + normal ``egg/`` prefix branch-ownership check, so no + orchestrator-role push surface is introduced. Returns ``True`` on success, ``False`` on any error (the caller logs and surfaces a clear error to the run loop). diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 210023aea2..459f407d27 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -10982,7 +10982,11 @@ def _run_implement_phase_slices( else f"egg/{pipeline_id}/work" ) issue_number = pipeline.issue_number - issue_branch = f"egg/issue-{issue_number}" if issue_number is not None else pipeline_branch + # Slice integration branches stack under ``pipeline_branch`` directly + # so any qualifier suffix (``-v3``, ``-backend``) is preserved — two + # qualified pipelines for the same issue would otherwise collide in + # the ``egg/issue-N/slice-M`` namespace (#2368). + issue_branch = pipeline_branch # Wrap scheduler construction so the run loop doesn't crash if the # contract bypassed plan-ingestion validation and reaches the diff --git a/orchestrator/tests/test_slice_run_loop_integration.py b/orchestrator/tests/test_slice_run_loop_integration.py index ddfeceef83..67c996fae3 100644 --- a/orchestrator/tests/test_slice_run_loop_integration.py +++ b/orchestrator/tests/test_slice_run_loop_integration.py @@ -1381,3 +1381,211 @@ def test_release_called_on_integration_branch_failure(self) -> None: ) assert global_slice_admit.snapshot()["admitted"] == 0 global_slice_admit.reset_for_testing() + + +# --------------------------------------------------------------------------- +# #2368 — slice integration-branch creation must precede agent spawn, +# and the integration branch must preserve the pipeline's qualifier. +# +# These regression guards lock in the ordering and naming invariants that +# masked the gateway/orchestrator conflict between #2028 (pipeline-session +# push enforcement) and #2220 (synthetic-session integration push). Without +# the ordering guard, a future refactor could re-introduce the pattern of +# spawning agents first and pushing the integration branch lazily — exactly +# the failure mode that #2337's silent demotion to monolithic implement was +# masking. Without the qualifier guard, two qualified pipelines for the same +# issue (e.g. ``-v3`` and ``-v4``) would collide in the ``slice-N`` namespace. +# --------------------------------------------------------------------------- + + +class TestSliceIntegrationBranchPrecedesAgentSpawn: + """Regression guards for #2368.""" + + def test_integration_branch_pushed_before_concurrent_phase(self) -> None: + """``create_slice_integration_branch`` must precede ``_run_concurrent_phase``. + + Spawning agents first and pushing lazily is the exact ordering bug + that #2220 introduced and #2337 was masking — agents would push to + a missing parent branch. + """ + pipeline = _make_pipeline() + contract = _make_contract(slices=[_make_slice("slice-1", tasks=[_make_task("task-1")])]) + + call_order: list[str] = [] + + def _track_create_branch(*args: Any, **kwargs: Any) -> bool: + call_order.append("create_slice_integration_branch") + return True + + def _track_run_phase(*args: Any, **kwargs: Any) -> tuple[int, str]: + call_order.append("_run_concurrent_phase") + return 0, "ok" + + with ( + patch("egg_contracts.loader.load_contract", return_value=contract), + patch("egg_contracts.loader.save_contract"), + patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon, + patch( + "routes.pipelines._run_concurrent_phase", + side_effect=_track_run_phase, + ), + patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), + ): + mock_start_recon.return_value = (MagicMock(), threading.Event()) + spawner = MagicMock() + spawner.gateway = MagicMock() + spawner.gateway.create_slice_integration_branch = MagicMock( + side_effect=_track_create_branch + ) + spawner.gateway.create_slice_pr = MagicMock(return_value="https://example/pr/1") + + _run_implement_phase_slices( + pipeline_id=pipeline.id, + pipeline=pipeline, + spawner=spawner, + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + store=MagicMock(), + certs_volume=None, + worktree_repo_path=Path("/tmp/x"), + ) + + assert "create_slice_integration_branch" in call_order + assert "_run_concurrent_phase" in call_order + assert call_order.index("create_slice_integration_branch") < call_order.index( + "_run_concurrent_phase" + ), ( + "create_slice_integration_branch must run BEFORE _run_concurrent_phase — " + "agents push directly to the slice integration branch, so the parent " + "ref must exist on origin first" + ) + + def test_concurrent_phase_not_invoked_when_integration_branch_creation_fails( + self, + ) -> None: + """When integration-branch push fails, agents must not spawn. + + This is the exact behaviour that surfaced #2368: the gateway 403'd + every ``create_slice_integration_branch`` call, the slice loop + correctly skipped the spawn, and the operator saw 15 slice failures + instead of a wedged pipeline running on missing parents. + """ + from orchestrator import global_slice_admit + + global_slice_admit.reset_for_testing(cap=4) + + pipeline = _make_pipeline() + contract = _make_contract(slices=[_make_slice("slice-1")]) + + with ( + patch("egg_contracts.loader.load_contract", return_value=contract), + patch("egg_contracts.loader.save_contract"), + patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon, + patch("routes.pipelines._run_concurrent_phase") as mock_run_phase, + patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), + ): + mock_start_recon.return_value = (MagicMock(), threading.Event()) + spawner = MagicMock() + spawner.gateway = MagicMock() + spawner.gateway.create_slice_integration_branch.return_value = False + + _run_implement_phase_slices( + pipeline_id=pipeline.id, + pipeline=pipeline, + spawner=spawner, + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + store=MagicMock(), + certs_volume=None, + worktree_repo_path=Path("/tmp/x"), + ) + + assert mock_run_phase.call_count == 0, ( + "_run_concurrent_phase must not be invoked when integration-branch " + "creation fails — agents would push to a missing parent" + ) + + global_slice_admit.reset_for_testing() + + +class TestSliceIntegrationBranchQualifierPreserved: + """The slice integration branch derives from ``pipeline.branch`` so the + qualifier suffix is preserved (#2368 bonus). + + Two qualified pipelines for the same issue (``egg/issue-N-v3`` and + ``egg/issue-N-v4``) must not collide in the ``slice-M`` namespace. + """ + + def test_qualified_pipeline_branch_propagates_to_slice_branches(self) -> None: + """``egg/issue-N-v3`` ⇒ slices stack under the qualified prefix.""" + config = PipelineConfig() + for key, val in { + "concurrent_execution": True, + "max_concurrent_agents": 6, + "consensus_timeout_minutes": 30, + }.items(): + try: + setattr(config, key, val) + except AttributeError, ValueError: + config.__dict__[key] = val + pipeline = Pipeline( + id="issue-2261-v3", + issue_number=2261, + repo="owner/repo", + branch="egg/issue-2261-v3", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + config=config, + ) + contract = _make_contract( + pipeline_id="issue-2261-v3", + issue_number=2261, + slices=[_make_slice("slice-1", tasks=[_make_task("task-1")])], + ) + + captured: dict[str, Any] = {} + + def _capture(*args: Any, **kwargs: Any) -> bool: + captured["parent_branch"] = kwargs.get("parent_branch") + captured["integration_branch"] = kwargs.get("integration_branch") + return True + + with ( + patch("egg_contracts.loader.load_contract", return_value=contract), + patch("egg_contracts.loader.save_contract"), + patch("routes.pipelines._start_stacked_pr_reconciler") as mock_start_recon, + patch("routes.pipelines._run_concurrent_phase", return_value=(0, "ok")), + patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), + ): + mock_start_recon.return_value = (MagicMock(), threading.Event()) + spawner = MagicMock() + spawner.gateway = MagicMock() + spawner.gateway.create_slice_integration_branch = MagicMock(side_effect=_capture) + spawner.gateway.create_slice_pr = MagicMock(return_value="https://example/pr/1") + + _run_implement_phase_slices( + pipeline_id=pipeline.id, + pipeline=pipeline, + spawner=spawner, + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + store=MagicMock(), + certs_volume=None, + worktree_repo_path=Path("/tmp/x"), + ) + + assert captured.get("parent_branch") == "egg/issue-2261-v3", ( + "parent_branch for the root slice must be the qualified pipeline branch, " + f"got {captured.get('parent_branch')!r}" + ) + assert captured.get("integration_branch") == "egg/issue-2261-v3/slice-1", ( + "integration_branch must inherit the qualifier so qualified pipelines " + "for the same issue don't collide in the slice namespace; got " + f"{captured.get('integration_branch')!r}" + ) From 47c9f56bd3af1ff5c2631e1dbc1fb5b461c04475 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Thu, 30 Apr 2026 20:19:50 +0000 Subject: [PATCH 2/3] Address #2370 review: fix qualifier-dropping orphan no-op + nits The blocking fix: stacked_pr_reconciler.find_orphaned_child_prs derived issue_branch from contract.issue.number, producing 'egg/issue-N' even when the contract carried a qualifier (e.g. pipeline_id='issue-N-v3'). Since this PR's bonus fix made create_slice_integration_branch preserve the qualifier ('egg/issue-N-v3/slice-M'), the reconciler's per-slice lookup never matched and orphan detection silently no-op'd for every qualified pipeline. Switch to 'egg/{contract.contract_key}' which returns the canonical pipeline-id for all three shapes (issue-driven, qualified, JIRA). Non-blocking nits from the review: - Tighten _SLICE_INTEGRATION_BRANCH_RE to single-segment bases (drop / from the second character class). - Replace dead Python-2-shaped 'except E1, E2' try/except in the slice-loop test helper with a direct PipelineConfig kwarg construction; refactor the new qualified-pipeline test to reuse the helper instead of duplicating the boilerplate. - Document the intentional dual audit emission (push_slice_integration_exempt + push_infrastructure_exempt with exempt_type=slice_integration_branch) inline so operators don't conclude the latter was an infra push. Tests: regression coverage for the qualifier-preservation bug (qualified-pipeline orphan detection + walk-up resolver) and for the tightened regex (multi-segment base rejected). --- gateway/gateway.py | 17 ++++- gateway/tests/test_pipeline_push_block.py | 26 +++++++ orchestrator/stacked_pr_reconciler.py | 15 +++- .../tests/test_slice_run_loop_integration.py | 51 +++++-------- .../tests/test_stacked_pr_reconciler.py | 71 ++++++++++++++++++- 5 files changed, 141 insertions(+), 39 deletions(-) diff --git a/gateway/gateway.py b/gateway/gateway.py index c66f1fd67c..3317a6ce50 100644 --- a/gateway/gateway.py +++ b/gateway/gateway.py @@ -1076,12 +1076,14 @@ def config_reload() -> Response: # Slice integration-branch shape for the synthetic-session exemption (#2368). -# Matches ``egg//(slice|phase)-`` where ```` is the parent -# pipeline branch (issue-driven, JIRA-driven, or qualifier-suffixed). Only +# Matches ``egg//(slice|phase)-`` where ```` is a single +# segment naming the parent pipeline branch (issue-driven, JIRA-driven, or +# qualifier-suffixed) — multi-segment bases are never produced by the +# orchestrator, so the second character class excludes ``/``. Only # orchestrator-issued sessions can ever set ``synthetic=True`` (the launcher # secret gates ``/api/v1/sessions/create``), so this exemption is not reachable # from a sandboxed agent's session token. -_SLICE_INTEGRATION_BRANCH_RE = re.compile(r"^egg/[A-Za-z0-9][A-Za-z0-9_/-]*/(?:slice|phase)-\d+$") +_SLICE_INTEGRATION_BRANCH_RE = re.compile(r"^egg/[A-Za-z0-9][A-Za-z0-9_-]*/(?:slice|phase)-\d+$") @app.route("/api/v1/git/push", methods=["POST"]) @@ -1343,6 +1345,15 @@ def git_push() -> tuple[Response, int] | Response: exempt_type = "slice_integration_branch" else: exempt_type = "infrastructure_branch" + # A successful slice-integration push intentionally emits BOTH + # ``push_slice_integration_exempt`` (above, the orchestrator- + # specific event) AND ``push_infrastructure_exempt`` with + # ``exempt_type="slice_integration_branch"`` (here, the generic + # exemption event). Operators grepping ``push_infrastructure_exempt`` + # for "infra pushes" should filter out the slice variant via + # ``exempt_type``; the dual emission is intentional so the + # orchestrator-specific path is also visible to operators + # filtering on the slice-integration event name (#2370 review). audit_log( "push_infrastructure_exempt", "git_push", diff --git a/gateway/tests/test_pipeline_push_block.py b/gateway/tests/test_pipeline_push_block.py index 118d9ef6e6..cbf233e95f 100644 --- a/gateway/tests/test_pipeline_push_block.py +++ b/gateway/tests/test_pipeline_push_block.py @@ -843,6 +843,32 @@ def test_synthetic_session_non_slice_branch_still_blocked(self, client): response = _do_push(client, refspec="egg/issue-2261") assert response.status_code == 403 + def test_synthetic_session_multi_segment_base_blocked(self, client): + """Multi-segment base shapes (``egg/foo/bar/slice-N``) are not produced + by the orchestrator and the regex MUST reject them. + + The documented branch shape is ``egg//(slice|phase)-N``; + accepting multi-segment bases would widen the exemption surface beyond + what the orchestrator actually emits. + """ + session = _make_session(synthetic=True, assigned_branch="egg/foo/bar/slice-1") + patches = _push_context(session) + with ( + patches[0], + patches[1], + patches[2], + patches[3], + patches[4], + patches[5], + patches[6], + patches[7], + ): + response = _do_push( + client, + refspec="egg/foo:refs/heads/egg/foo/bar/slice-1", + ) + assert response.status_code == 403 + def test_audit_event_records_slice_integration_exempt_type(self, client): """Exemption emits a distinct audit event so operators can trace synthetic-session pushes separately from checkpoint/state writes.""" diff --git a/orchestrator/stacked_pr_reconciler.py b/orchestrator/stacked_pr_reconciler.py index d4c311c189..dc65fb1a74 100644 --- a/orchestrator/stacked_pr_reconciler.py +++ b/orchestrator/stacked_pr_reconciler.py @@ -176,9 +176,18 @@ def find_orphaned_child_prs( pr_by_head[head] = pr orphans: list[OrphanedChildPR] = [] - issue_number = contract.issue.number if contract.issue is not None else None - pipeline_id = contract.contract_key - issue_branch = f"egg/issue-{issue_number}" if issue_number else f"egg/{pipeline_id}" + # ``contract_key`` returns the canonical pipeline id for all + # supported shapes — issue-driven (``issue-N``), qualified + # (``issue-N-v3``, ``issue-N-backend``), and JIRA (``ENG-1234``). + # The orchestrator's slice-integration branches preserve the + # qualifier (see ``routes/pipelines.py`` ``pipeline.branch`` + # propagation), so deriving the issue branch from ``contract_key`` + # keeps the reconciler's lookup shape aligned with the producer's + # branch shape. A pre-#2137 split here that hard-coded + # ``egg/issue-N`` for any contract with a populated ``issue`` + # field silently no-op'd orphan detection on every qualified + # pipeline. + issue_branch = f"egg/{contract.contract_key}" slices_by_id = {s.id: s for s in contract.slices} for slice_ in contract.slices: diff --git a/orchestrator/tests/test_slice_run_loop_integration.py b/orchestrator/tests/test_slice_run_loop_integration.py index 67c996fae3..93d21fe5e5 100644 --- a/orchestrator/tests/test_slice_run_loop_integration.py +++ b/orchestrator/tests/test_slice_run_loop_integration.py @@ -78,23 +78,27 @@ # --------------------------------------------------------------------------- -def _make_pipeline(pipeline_id: str = "issue-9999", issue_number: int | None = 9999) -> Pipeline: - """Pipeline with concurrent_execution enabled for slice-loop tests.""" - config = PipelineConfig() - for key, val in { - "concurrent_execution": True, - "max_concurrent_agents": 6, - "consensus_timeout_minutes": 30, - }.items(): - try: - setattr(config, key, val) - except AttributeError, ValueError: - config.__dict__[key] = val +def _make_pipeline( + pipeline_id: str = "issue-9999", + issue_number: int | None = 9999, +) -> Pipeline: + """Pipeline with concurrent_execution enabled for slice-loop tests. + + ``branch`` is derived from ``pipeline_id`` so qualified pipelines + (``issue-N-v3``, ``issue-N-backend``) propagate the qualifier into + ``pipeline.branch`` — the slice-loop's canonical source for the + integration-branch parent (#2370 review). + """ + config = PipelineConfig( + concurrent_execution=True, + max_concurrent_agents=6, + consensus_timeout_minutes=30, + ) return Pipeline( id=pipeline_id, issue_number=issue_number, repo="owner/repo", - branch=f"egg/issue-{issue_number}" if issue_number else f"egg/{pipeline_id}", + branch=f"egg/{pipeline_id}", status=PipelineStatus.RUNNING, current_phase=PipelinePhase.IMPLEMENT, config=config, @@ -1522,25 +1526,8 @@ class TestSliceIntegrationBranchQualifierPreserved: def test_qualified_pipeline_branch_propagates_to_slice_branches(self) -> None: """``egg/issue-N-v3`` ⇒ slices stack under the qualified prefix.""" - config = PipelineConfig() - for key, val in { - "concurrent_execution": True, - "max_concurrent_agents": 6, - "consensus_timeout_minutes": 30, - }.items(): - try: - setattr(config, key, val) - except AttributeError, ValueError: - config.__dict__[key] = val - pipeline = Pipeline( - id="issue-2261-v3", - issue_number=2261, - repo="owner/repo", - branch="egg/issue-2261-v3", - status=PipelineStatus.RUNNING, - current_phase=PipelinePhase.IMPLEMENT, - config=config, - ) + pipeline = _make_pipeline(pipeline_id="issue-2261-v3", issue_number=2261) + assert pipeline.branch == "egg/issue-2261-v3" # helper-derived; sanity-check contract = _make_contract( pipeline_id="issue-2261-v3", issue_number=2261, diff --git a/orchestrator/tests/test_stacked_pr_reconciler.py b/orchestrator/tests/test_stacked_pr_reconciler.py index b4629928db..f62d22ac35 100644 --- a/orchestrator/tests/test_stacked_pr_reconciler.py +++ b/orchestrator/tests/test_stacked_pr_reconciler.py @@ -65,9 +65,10 @@ def _slice( ) -def _contract(*slices: Slice) -> Contract: +def _contract(*slices: Slice, pipeline_id: str | None = None) -> Contract: return Contract( issue=IssueInfo(number=2137, title="t", url="u"), + pipeline_id=pipeline_id, slices=list(slices), ) @@ -353,6 +354,74 @@ def test_pr_with_zero_number_dropped(self) -> None: prs = [_pr(number=0, head="egg/issue-2137/slice-2", base="egg/issue-2137/slice-1")] assert find_orphaned_child_prs(contract, prs, set()) == [] + def test_qualified_pipeline_id_preserves_qualifier_in_issue_branch(self) -> None: + # Regression for #2370 review: when a contract's pipeline_id + # carries a qualifier suffix (e.g. ``issue-2137-v3``, + # ``issue-2137-backend``), the orchestrator's slice-loop + # creates branches as ``egg/issue-2137-v3/slice-N`` (qualifier + # preserved). A pre-fix bug derived ``issue_branch`` from the + # raw issue number, producing the unqualified + # ``egg/issue-2137/slice-N`` lookup key — orphan detection + # silently no-op'd for every qualified pipeline. Lock the + # producer/consumer shape here. + contract = _contract( + _slice( + "slice-2", + deps=["slice-1"], + parent_branch="egg/issue-2137-v3/slice-1", + ), + pipeline_id="issue-2137-v3", + ) + prs = [ + _pr( + number=11, + head="egg/issue-2137-v3/slice-2", + base="egg/issue-2137-v3/slice-1", + ) + ] + # Parent base no longer on origin → must surface as orphan. + orphans = find_orphaned_child_prs(contract, prs, set()) + assert len(orphans) == 1 + orphan = orphans[0] + assert orphan.slice_id == "slice-2" + assert orphan.pr_number == 11 + assert orphan.branch == "egg/issue-2137-v3/slice-2" + assert orphan.deleted_base == "egg/issue-2137-v3/slice-1" + # Walk-up fallback: parent slice missing from contract, so the + # qualified pipeline branch is the safe target — and crucially + # is NOT the unqualified ``egg/issue-2137``. + assert orphan.intended_new_base == "egg/issue-2137-v3" + + def test_qualified_pipeline_id_walks_up_to_qualified_ancestor(self) -> None: + # The walk-up resolver must also use the qualified branch + # when looking for an extant ancestor. + contract = _contract( + _slice("slice-1"), + _slice( + "slice-2", + deps=["slice-1"], + parent_branch="egg/issue-2137-v3/slice-1", + ), + _slice( + "slice-3", + deps=["slice-2"], + parent_branch="egg/issue-2137-v3/slice-2", + ), + pipeline_id="issue-2137-v3", + ) + prs = [ + _pr( + number=12, + head="egg/issue-2137-v3/slice-3", + base="egg/issue-2137-v3/slice-2", + ) + ] + # slice-2's branch deleted; slice-1's qualified branch alive. + extant: set[str] = {"egg/issue-2137-v3/slice-1"} + orphans = find_orphaned_child_prs(contract, prs, extant) + assert len(orphans) == 1 + assert orphans[0].intended_new_base == "egg/issue-2137-v3/slice-1" + # ---------- reconcile_once ---------- From cbd60fe859bcff960da47e5d3da05e3a60e7e63f Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Thu, 30 Apr 2026 20:36:27 +0000 Subject: [PATCH 3/3] Drop pre-#2137 framing from reconciler comment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reviewer noted the historical reference is misleading — the issue-number-based derivation that this comment documents was introduced and removed in the same PR (#2370), not a pre-#2137 artefact. Re-frame as describing the prior ternary directly. Cosmetic; no behaviour change. --- orchestrator/stacked_pr_reconciler.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/orchestrator/stacked_pr_reconciler.py b/orchestrator/stacked_pr_reconciler.py index dc65fb1a74..c6ee7b798d 100644 --- a/orchestrator/stacked_pr_reconciler.py +++ b/orchestrator/stacked_pr_reconciler.py @@ -183,10 +183,10 @@ def find_orphaned_child_prs( # qualifier (see ``routes/pipelines.py`` ``pipeline.branch`` # propagation), so deriving the issue branch from ``contract_key`` # keeps the reconciler's lookup shape aligned with the producer's - # branch shape. A pre-#2137 split here that hard-coded - # ``egg/issue-N`` for any contract with a populated ``issue`` - # field silently no-op'd orphan detection on every qualified - # pipeline. + # branch shape. A prior ``f"egg/issue-{issue_number}"`` ternary + # here hard-coded the unqualified shape for any contract with a + # populated ``issue`` field, silently no-op'ing orphan detection + # on every qualified pipeline. issue_branch = f"egg/{contract.contract_key}" slices_by_id = {s.id: s for s in contract.slices}