From 34067b2510e453b63357cb8c83e64074e80cc7a3 Mon Sep 17 00:00:00 2001 From: jwbron <8340608+jwbron@users.noreply.github.com> Date: Thu, 2 Jul 2026 00:28:57 +0000 Subject: [PATCH 1/6] docs: fix stale .egg-state/contracts/ security-boundary note --- docs/guides/sdlc-pipeline.md | 2 +- docs/reference/agent-roles.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/guides/sdlc-pipeline.md b/docs/guides/sdlc-pipeline.md index e8a13ff91c..ef7f4e3418 100644 --- a/docs/guides/sdlc-pipeline.md +++ b/docs/guides/sdlc-pipeline.md @@ -961,7 +961,7 @@ All three keys (`tests_globs`, `code_globs`, `docs_globs`) are optional. Unset k The conflict-resolver role's allow list is the union of all three glob lists, so any of these keys also widens what the conflict-resolver can write. -**Security boundary:** Security-relevant blocklists (`.egg-state/contracts/`, `.github/`) are hard-coded and cannot be relaxed by repo config. Only the language-convention globs are configurable. +**Security boundary:** Security-relevant blocklists (the whole `.egg-state/` tree, `.github/`) are hard-coded and cannot be relaxed by repo config. Only the language-convention globs are configurable. The orchestrator pre-resolves the override at spawn time and passes it to sandbox containers via the `EGG_PIPELINE_REPO_PATTERNS_JSON` environment variable. The gateway reads the override directly from `repositories.yaml` at push time. Validation behavior differs slightly between paths: `config/repo_config.py::get_repo_role_patterns` (the orchestrator/gateway path that reads `repositories.yaml`) emits a WARNING log on invalid root type, unknown keys, non-list values, and non-string list entries; `shared/egg_restrictions/patterns.py::load_repo_pattern_override` (the env-var path used inside the sandbox) only logs on invalid JSON and silently filters the rest. In practice the operator still sees diagnostic warnings at orchestrator spawn time because the orchestrator runs `get_repo_role_patterns` before serializing into the env var. diff --git a/docs/reference/agent-roles.md b/docs/reference/agent-roles.md index 8346120f2d..211ad43ffd 100644 --- a/docs/reference/agent-roles.md +++ b/docs/reference/agent-roles.md @@ -717,7 +717,7 @@ The gateway's `get_attributed_changed_files_in_push()` walks the unpushed range For the exact allowed and blocked patterns per role, see `shared/egg_restrictions/patterns.py` (canonical source). The gateway imports from this shared package for push-time validation. -**Per-repo overrides (#2528):** The test/code/docs glob lists described above are the *defaults*. Repositories with non-Python file conventions (Go, JS/TS, etc.) can override them via a `role_patterns:` block in `repositories.yaml`. Only the language-convention globs (`tests_globs`, `code_globs`, `docs_globs`) are configurable; security-relevant blocklists (`.egg-state/contracts/`, `.github/`) are hard-coded and cannot be relaxed. See [Per-Repository Role Patterns](../guides/sdlc-pipeline.md#per-repository-role-patterns) for the configuration schema. +**Per-repo overrides (#2528):** The test/code/docs glob lists described above are the *defaults*. Repositories with non-Python file conventions (Go, JS/TS, etc.) can override them via a `role_patterns:` block in `repositories.yaml`. Only the language-convention globs (`tests_globs`, `code_globs`, `docs_globs`) are configurable; security-relevant blocklists (the whole `.egg-state/` tree, `.github/`) are hard-coded and cannot be relaxed. See [Per-Repository Role Patterns](../guides/sdlc-pipeline.md#per-repository-role-patterns) for the configuration schema. ## Per-Agent Git Identity From 1ef5c1ffb54dc1f3aef45c86287a8ad27e0543ff Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:08:13 +0000 Subject: [PATCH 2/6] docs: sync patterns.py security-boundary comments with .egg-state widening --- shared/egg_restrictions/patterns.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/shared/egg_restrictions/patterns.py b/shared/egg_restrictions/patterns.py index 5da4fb75c4..2c8928ea14 100644 --- a/shared/egg_restrictions/patterns.py +++ b/shared/egg_restrictions/patterns.py @@ -162,9 +162,10 @@ def _normalize_path(file_path: str) -> str: # are the load-bearing language conventions. Repos can override them via # ``role_patterns:`` in ``repositories.yaml`` so non-Python conventions # (Go ``*_test.go``, JS ``__tests__/``, etc.) get correct role boundaries. -# Security-relevant blocklists (``.egg-state/contracts/``, ``.github/``) -# are NOT overridable — they enforce the policy boundary independent of -# repo-specific conventions. +# Security-relevant blocklists (the whole ``.egg-state/`` tree for the +# coder/tester tier, ``.egg-state/contracts/`` for the other roles, and +# ``.github/``) are NOT overridable — they enforce the policy boundary +# independent of repo-specific conventions. # Default test-file conventions: directory patterns + file-name patterns. # fnmatch does NOT support brace expansion, so each suffix is spelled out. @@ -952,8 +953,9 @@ def build_agent_patterns( A fresh dict mapping role name to ``AgentFilePattern``. Security note: - Security-relevant blocklists (``.egg-state/contracts/``, - ``.github/``, etc.) are sourced from this module's hard-coded + Security-relevant blocklists (the whole ``.egg-state/`` tree for + the coder/tester tier, ``.egg-state/contracts/`` for the other + roles, and ``.github/``) are sourced from this module's hard-coded builders and CANNOT be overridden by a repo's config. The per-repo knobs only widen the language-convention lists. """ From 27095061457b026b1c4fd700899e2f41edda2782 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:18:54 +0000 Subject: [PATCH 3/6] docs: note plan-agent .egg-state/reviews/ block in security-boundary comments --- shared/egg_restrictions/patterns.py | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/shared/egg_restrictions/patterns.py b/shared/egg_restrictions/patterns.py index 2c8928ea14..b0c8515b97 100644 --- a/shared/egg_restrictions/patterns.py +++ b/shared/egg_restrictions/patterns.py @@ -163,9 +163,10 @@ def _normalize_path(file_path: str) -> str: # ``role_patterns:`` in ``repositories.yaml`` so non-Python conventions # (Go ``*_test.go``, JS ``__tests__/``, etc.) get correct role boundaries. # Security-relevant blocklists (the whole ``.egg-state/`` tree for the -# coder/tester tier, ``.egg-state/contracts/`` for the other roles, and -# ``.github/``) are NOT overridable — they enforce the policy boundary -# independent of repo-specific conventions. +# coder/tester tier, ``.egg-state/contracts/`` -- plus ``.egg-state/reviews/`` +# for plan agents -- for the other roles, and ``.github/``) are NOT +# overridable -- they enforce the policy boundary independent of +# repo-specific conventions. # Default test-file conventions: directory patterns + file-name patterns. # fnmatch does NOT support brace expansion, so each suffix is spelled out. @@ -954,10 +955,11 @@ def build_agent_patterns( Security note: Security-relevant blocklists (the whole ``.egg-state/`` tree for - the coder/tester tier, ``.egg-state/contracts/`` for the other - roles, and ``.github/``) are sourced from this module's hard-coded - builders and CANNOT be overridden by a repo's config. The - per-repo knobs only widen the language-convention lists. + the coder/tester tier, ``.egg-state/contracts/`` -- plus + ``.egg-state/reviews/`` for plan agents -- for the other roles, and + ``.github/``) are sourced from this module's hard-coded builders and + CANNOT be overridden by a repo's config. The per-repo knobs only + widen the language-convention lists. """ if repo is not None and tests_globs is None and code_globs is None and docs_globs is None: override = load_repo_pattern_override(repo) From 2c807d521f761755022d29b43c0b909328716db8 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:32:20 +0000 Subject: [PATCH 4/6] docs: restore em-dash convention in security-boundary comments --- shared/egg_restrictions/patterns.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/shared/egg_restrictions/patterns.py b/shared/egg_restrictions/patterns.py index b0c8515b97..685d4b8b5d 100644 --- a/shared/egg_restrictions/patterns.py +++ b/shared/egg_restrictions/patterns.py @@ -163,9 +163,9 @@ def _normalize_path(file_path: str) -> str: # ``role_patterns:`` in ``repositories.yaml`` so non-Python conventions # (Go ``*_test.go``, JS ``__tests__/``, etc.) get correct role boundaries. # Security-relevant blocklists (the whole ``.egg-state/`` tree for the -# coder/tester tier, ``.egg-state/contracts/`` -- plus ``.egg-state/reviews/`` -# for plan agents -- for the other roles, and ``.github/``) are NOT -# overridable -- they enforce the policy boundary independent of +# coder/tester tier, ``.egg-state/contracts/`` — plus ``.egg-state/reviews/`` +# for plan agents — for the other roles, and ``.github/``) are NOT +# overridable — they enforce the policy boundary independent of # repo-specific conventions. # Default test-file conventions: directory patterns + file-name patterns. @@ -955,8 +955,8 @@ def build_agent_patterns( Security note: Security-relevant blocklists (the whole ``.egg-state/`` tree for - the coder/tester tier, ``.egg-state/contracts/`` -- plus - ``.egg-state/reviews/`` for plan agents -- for the other roles, and + the coder/tester tier, ``.egg-state/contracts/`` — plus + ``.egg-state/reviews/`` for plan agents — for the other roles, and ``.github/``) are sourced from this module's hard-coded builders and CANNOT be overridden by a repo's config. The per-repo knobs only widen the language-convention lists. From e89909535818f9954b4068fe8155acd9dc7bca6e Mon Sep 17 00:00:00 2001 From: "james-in-a-box[bot]" <246424927+james-in-a-box[bot]@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:32:23 +0000 Subject: [PATCH 5/6] Fix Lint CI failures: ruff-format + file-size allowlist Apply ruff format to the 6 files flagged by the Python lint job (exception-tuple parens dropped per PEP 758 under target-version py314, plus long-line wrapping). Also add orchestrator/models.py to scripts/file-size-allowlist.yaml (tracked by #3450) since it exceeds the 1500-line hard cap and decomposing it is out of scope for this doc-only PR. --- config/repo_config.py | 31 + gateway/gateway.py | 182 +- gateway/tests/test_repo_visibility.py | 153 ++ orchestrator/mcp_tools/_submit.py | 59 +- orchestrator/routes/pipelines.py | 1871 ++++++++++++++++- orchestrator/tests/test_kubernetes_spawner.py | 245 +++ scripts/file-size-allowlist.yaml | 6 +- 7 files changed, 2503 insertions(+), 44 deletions(-) diff --git a/config/repo_config.py b/config/repo_config.py index 1d65dc4e59..c68b4dff96 100644 --- a/config/repo_config.py +++ b/config/repo_config.py @@ -610,6 +610,37 @@ def get_auth_mode(repo: str) -> str: return cast(str, auth_mode) +def assert_uniform_auth(repos: list[str]) -> None: + """Require a single uniform auth mode across a run's repos (#3393). + + Multi-repo pipelines v1 require every repo in one run to share one auth + mode — all ``bot`` or all ``user``. Mixed auth is where the gateway session + model gets genuinely complex and is deferred to a later phase (see issue + #3393 "Hard parts"). This is the canonical per-run auth-uniformity guard, + living beside :func:`get_auth_mode`; it is importable from both the gateway + and the orchestrator (``config/repo_config.py`` is bundled into both + images). + + Args: + repos: Repositories in ``owner/name`` format. A single repo (or an + empty list) is trivially uniform. + + Raises: + ValueError: when the repos resolve to more than one auth mode. The + message names the offending repos grouped by mode so the operator + can see exactly which side diverges. + """ + modes: dict[str, list[str]] = {} + for repo in repos: + modes.setdefault(get_auth_mode(repo), []).append(repo) + if len(modes) > 1: + groups = "; ".join(f"{mode}: {', '.join(sorted(rs))}" for mode, rs in sorted(modes.items())) + raise ValueError( + "Mixed auth modes across the pipeline's repos are not supported in v1 " + "(a run must be uniformly 'bot' or 'user'). Diverging repos — " + groups + "." + ) + + def is_user_mode_repo(repo: str) -> bool: """ Check if a repository is configured to use user mode. diff --git a/gateway/gateway.py b/gateway/gateway.py index 66c34c4e41..8b2e48d97f 100644 --- a/gateway/gateway.py +++ b/gateway/gateway.py @@ -4532,6 +4532,176 @@ def gh_list_open_prs() -> tuple[Response, int] | Response: return make_success("Open PR list complete", {"prs": prs}) +@app.route("/api/v1/gh/pr/merge_state", methods=["POST"]) +@require_launcher_auth +def gh_pr_merge_state() -> tuple[Response, int] | Response: + """Control-plane PR merge-state read: return ``state`` + ``mergedAt`` (#3393). + + An **orchestrator-only** route gated by ``@require_launcher_auth`` + rather than ``@require_session_auth`` — the caller is the control + plane (the orchestrator holds the launcher secret), not a sandboxed + agent. It is the read half of the cq-1 cross-repo merge-sequencing + gate: the orchestrator polls an upstream slice PR's merge state to + decide when to mark a downstream draft PR ready. Modelled on + ``gh_find_open_pr`` / ``gh_list_open_prs`` (#2925): the orchestrator + is the server that manages pipelines, not an ``AgentRole``, so it + authenticates as the control plane and uses a purpose-built, + fixed-argv read-only endpoint (no general gh surface here). + + Merge detection deliberately keys off the PR's ``mergedAt`` / + ``state`` — NOT head-SHA equality: a squash/rebase merge produces a + merge-commit SHA that differs from the PR head, so a SHA comparison + would misfire (#3393 task-5-1 pin (a)). + + Request body: + {"repo": "owner/name", "pr_number": } + + Returns: + ``{"state": "OPEN|CLOSED|MERGED"|null, "mergedAt": ""|null}``. + """ + data = request.get_json() + if not data or not isinstance(data, dict): + return make_error("Invalid body: must be a JSON object") + + repo = data.get("repo") + if not isinstance(repo, str) or not repo.strip(): + return make_error("Missing or invalid repo: must be a non-empty string") + repo = repo.strip() + if OWNER_REPO_PATTERN.match(repo) is None: + return make_error("Invalid repo: must be 'owner/name'") + + # ``bool`` is a subclass of ``int``; reject it explicitly so ``True`` + # cannot slip through as ``pr_number=1``. + pr_number = data.get("pr_number") + if isinstance(pr_number, bool) or not isinstance(pr_number, int) or pr_number < 1: + return make_error("Invalid pr_number: must be a positive integer") + + args = [ + "pr", + "view", + str(pr_number), + "--repo", + repo, + "--json", + "state,mergedAt", + ] + + auth_mode = get_auth_mode(repo) + github = get_github_client(mode=auth_mode) + result = github.execute(args, timeout=60, mode=auth_mode) + + if not result.success: + stderr_excerpt = (result.stderr or "")[:500] + audit_log( + "gh_pr_merge_state_failed", + "gh_pr_merge_state", + success=False, + details={"repo": repo, "pr_number": pr_number, "stderr": stderr_excerpt}, + ) + return make_error( + f"Command failed: {result.stderr}", + status_code=500, + details=result.to_dict(), + ) + + state_val: Any = None + merged_at: Any = None + stdout = (result.stdout or "").strip() + if stdout: + try: + parsed = json.loads(stdout) + except ValueError, TypeError: + parsed = None + if isinstance(parsed, dict): + state_val = parsed.get("state") + merged_at = parsed.get("mergedAt") + + audit_log( + "gh_pr_merge_state", + "gh_pr_merge_state", + success=True, + details={"repo": repo, "pr_number": pr_number, "state": state_val}, + ) + return make_success( + "PR merge-state lookup complete", + {"state": state_val, "mergedAt": merged_at}, + ) + + +@app.route("/api/v1/gh/pr/ready", methods=["POST"]) +@require_launcher_auth +def gh_pr_ready() -> tuple[Response, int] | Response: + """Control-plane PR draft→ready transition: wrap ``gh pr ready`` (#3393). + + An **orchestrator-only** route gated by ``@require_launcher_auth`` — + the write half of the cq-1 cross-repo merge-sequencing gate. When the + upstream slice PR merges, the orchestrator transitions the downstream + cross-repo dependent PR from draft to ready. Like the sibling + control-plane PR routes (``gh_find_open_pr`` / ``gh_list_open_prs``), + the caller is the control plane, so it authenticates with the + launcher secret and this route constructs a **fixed, narrow argv** + server-side (``pr ready --repo ``) — there is no arbitrary + gh-command surface on the launcher-auth path, only this single + ready-transition. ``pr ready`` is already on ``ALLOWED_GH_COMMANDS`` + (github_client.py) so the underlying ``gh`` invocation re-validates + through the same allowlist floor. + + Request body: + {"repo": "owner/name", "pr_number": } + + Returns: + ``{"stdout": ""}`` on success. + """ + data = request.get_json() + if not data or not isinstance(data, dict): + return make_error("Invalid body: must be a JSON object") + + repo = data.get("repo") + if not isinstance(repo, str) or not repo.strip(): + return make_error("Missing or invalid repo: must be a non-empty string") + repo = repo.strip() + if OWNER_REPO_PATTERN.match(repo) is None: + return make_error("Invalid repo: must be 'owner/name'") + + pr_number = data.get("pr_number") + if isinstance(pr_number, bool) or not isinstance(pr_number, int) or pr_number < 1: + return make_error("Invalid pr_number: must be a positive integer") + + args = [ + "pr", + "ready", + str(pr_number), + "--repo", + repo, + ] + + auth_mode = get_auth_mode(repo) + github = get_github_client(mode=auth_mode) + result = github.execute(args, timeout=60, mode=auth_mode) + + if not result.success: + stderr_excerpt = (result.stderr or "")[:500] + audit_log( + "gh_pr_ready_failed", + "gh_pr_ready", + success=False, + details={"repo": repo, "pr_number": pr_number, "stderr": stderr_excerpt}, + ) + return make_error( + f"Failed to mark PR ready: {result.stderr}", + status_code=500, + details=result.to_dict(), + ) + + audit_log( + "gh_pr_ready", + "gh_pr_ready", + success=True, + details={"repo": repo, "pr_number": pr_number}, + ) + return make_success("PR marked ready", {"stdout": result.stdout}) + + # ============================================================================= # Jira REST Endpoints # ============================================================================= @@ -7763,8 +7933,16 @@ def worktree_create() -> tuple[Response, int] | Response: assigned_branch=assigned_branch, repo_slug=repo, ) - # Translate container path to host path for egg launcher mount sources - worktrees[repo_name] = translate_to_host_path(str(info.worktree_path)) + # Translate container path to host path for egg launcher mount sources. + # Key by the full ``owner/repo`` slug (#3393 slice-3, operator + # ruling #6) so two repos with the same short name under different + # owners (``ownerA/foo`` vs ``ownerB/foo``) no longer collide on a + # single map entry. When the caller passed a bare repo name (no + # ``/``), ``repo`` equals ``repo_name`` so bare-name callers are + # unaffected. The on-disk worktree directory (and the container + # mount target) stays the bare ``repo_name`` — only the map KEY + # carries the owner prefix. + worktrees[repo] = translate_to_host_path(str(info.worktree_path)) except (ValueError, RuntimeError) as e: # Capture full traceback so operators can diagnose without # re-instrumenting the gateway. See #2186. diff --git a/gateway/tests/test_repo_visibility.py b/gateway/tests/test_repo_visibility.py index 1641491ea2..e87b648a45 100644 --- a/gateway/tests/test_repo_visibility.py +++ b/gateway/tests/test_repo_visibility.py @@ -519,3 +519,156 @@ def test_no_retry_on_other_request_exception(self, mock_get, mock_sleep): assert result is None assert mock_get.call_count == 1 mock_sleep.assert_not_called() + + +# --- Slice-2 (#3393) uniform visibility / auth-mode validation --------------- +# +# multi-repo pipelines require every repo in one run to share a single +# visibility posture (all private/internal or all public) and a single auth +# mode (all bot or all user) — private mode is a pipeline-wide posture, so a +# mixed set would leak content across the private/public boundary. These +# helpers resolve each repo's visibility (``get_repo_visibility``) and auth mode +# (``config.repo_config.get_auth_mode``) and REJECT a mixed set with an +# actionable ``ValueError`` that NAMES the offending repos. A same-name / +# different-owner set is NOT rejected — uniformity is a property of the +# visibility/auth *bucket*, not the bare name (operator ruling #6). A single +# repo (N=1) is trivially uniform. +# +# The helpers are added by the slice-2 *coder* (a parallel BRC producer). Until +# that lands in this tester worktree the import below fails and this whole +# section skips with an explicit reason — it activates automatically at +# convergence, when the coder and tester branches merge. The exact interface +# the tester expects is handed to the coder via the task-2-3 contract gap so +# the two halves converge on the same shape: +# +# validate_visibility_uniformity(repos: list[str]) -> None +# validate_auth_mode_uniformity(repos: list[str]) -> None +# * ``repos`` are ``owner/name`` slugs. +# * no-op when the set is uniform or has < 2 repos. +# * raise ValueError naming the offending repos + their buckets on a +# mixed set. ``internal`` shares the private posture. + +import pytest + +try: + from repo_visibility import ( # type: ignore[attr-defined] + validate_auth_mode_uniformity, + validate_visibility_uniformity, + ) + + _UNIFORMITY_AVAILABLE = True + _UNIFORMITY_IMPORT_ERR: str | None = None +except Exception as _exc: # noqa: BLE001 + _UNIFORMITY_AVAILABLE = False + _UNIFORMITY_IMPORT_ERR = repr(_exc) + + +_skip_uniformity = pytest.mark.skipif( + not _UNIFORMITY_AVAILABLE, + reason=( + "slice-2 coder uniformity helpers (validate_visibility_uniformity / " + "validate_auth_mode_uniformity) not yet integrated into the tester " + "worktree (parallel producer); activates at convergence. import error: " + f"{_UNIFORMITY_IMPORT_ERR}" + ), +) + + +def _patch_visibility(monkeypatch, mapping): + """Route visibility resolution to an ``{'owner/repo': visibility}`` map. + + Patches both the module-level convenience function and the checker + accessor, so the helper resolves correctly whichever seam it calls through. + """ + + def _fake(owner, repo, **_): + return mapping[f"{owner}/{repo}"] + + monkeypatch.setattr("repo_visibility.get_repo_visibility", _fake, raising=False) + + checker = MagicMock() + checker.get_visibility.side_effect = lambda owner, repo, **_: mapping[f"{owner}/{repo}"] + checker.is_private.side_effect = lambda owner, repo, **_: ( + mapping[f"{owner}/{repo}"] + in ( + "private", + "internal", + ) + ) + monkeypatch.setattr("repo_visibility.get_visibility_checker", lambda: checker, raising=False) + + +def _patch_auth_mode(monkeypatch, mapping): + """Route auth-mode resolution to a ``{'owner/repo': mode}`` map.""" + + def _fake(repo, **_): + return mapping[repo] + + for target in ("repo_visibility.get_auth_mode", "repo_config.get_auth_mode"): + monkeypatch.setattr(target, _fake, raising=False) + + +@_skip_uniformity +class TestVisibilityUniformity: + """Uniform-visibility submission validation (AC-2).""" + + def test_uniform_private_accepted(self, monkeypatch): + _patch_visibility(monkeypatch, {"jwbron/a": "private", "jwbron/b": "private"}) + validate_visibility_uniformity(["jwbron/a", "jwbron/b"]) # no raise + + def test_uniform_public_accepted(self, monkeypatch): + _patch_visibility(monkeypatch, {"jwbron/a": "public", "jwbron/b": "public"}) + validate_visibility_uniformity(["jwbron/a", "jwbron/b"]) # no raise + + def test_mixed_visibility_rejected_names_offenders(self, monkeypatch): + _patch_visibility(monkeypatch, {"jwbron/priv": "private", "jwbron/pub": "public"}) + with pytest.raises(ValueError) as excinfo: + validate_visibility_uniformity(["jwbron/priv", "jwbron/pub"]) + msg = str(excinfo.value) + # The error is actionable: it names the repos across the split. + assert "jwbron/priv" in msg + assert "jwbron/pub" in msg + + def test_internal_shares_private_posture(self, monkeypatch): + # internal is on the private side of the boundary; internal+private is + # uniform and must NOT be rejected. + _patch_visibility(monkeypatch, {"jwbron/a": "internal", "jwbron/b": "private"}) + validate_visibility_uniformity(["jwbron/a", "jwbron/b"]) # no raise + + def test_same_name_different_owner_not_rejected(self, monkeypatch): + # ruling #6: identity is the owner/name slug, not the bare name — a + # same-name set with a uniform bucket is accepted. + _patch_visibility(monkeypatch, {"ownerA/foo": "private", "ownerB/foo": "private"}) + validate_visibility_uniformity(["ownerA/foo", "ownerB/foo"]) # no raise + + def test_single_repo_is_trivially_uniform(self, monkeypatch): + _patch_visibility(monkeypatch, {"jwbron/only": "public"}) + validate_visibility_uniformity(["jwbron/only"]) # no raise + + +@_skip_uniformity +class TestAuthModeUniformity: + """Uniform-auth-mode submission validation (AC-2).""" + + def test_uniform_bot_accepted(self, monkeypatch): + _patch_auth_mode(monkeypatch, {"jwbron/a": "bot", "jwbron/b": "bot"}) + validate_auth_mode_uniformity(["jwbron/a", "jwbron/b"]) # no raise + + def test_uniform_user_accepted(self, monkeypatch): + _patch_auth_mode(monkeypatch, {"jwbron/a": "user", "jwbron/b": "user"}) + validate_auth_mode_uniformity(["jwbron/a", "jwbron/b"]) # no raise + + def test_mixed_auth_rejected_names_offenders(self, monkeypatch): + _patch_auth_mode(monkeypatch, {"jwbron/bot": "bot", "jwbron/user": "user"}) + with pytest.raises(ValueError) as excinfo: + validate_auth_mode_uniformity(["jwbron/bot", "jwbron/user"]) + msg = str(excinfo.value) + assert "jwbron/user" in msg + + def test_same_name_different_owner_not_rejected(self, monkeypatch): + _patch_auth_mode(monkeypatch, {"ownerA/foo": "bot", "ownerB/foo": "bot"}) + validate_auth_mode_uniformity(["ownerA/foo", "ownerB/foo"]) # no raise + + def test_single_repo_is_trivially_uniform(self, monkeypatch): + _patch_auth_mode(monkeypatch, {"jwbron/only": "user"}) + validate_auth_mode_uniformity(["jwbron/only"]) # no raise diff --git a/orchestrator/mcp_tools/_submit.py b/orchestrator/mcp_tools/_submit.py index e3127b2ee5..774ce4bdca 100644 --- a/orchestrator/mcp_tools/_submit.py +++ b/orchestrator/mcp_tools/_submit.py @@ -75,8 +75,65 @@ def _handle_submit_task(self, args: dict[str, Any]) -> dict[str, Any]: base_id = f"{base_id}-{qualifier}" data["pipeline_id"] = base_id data["branch"] = args.get("branch") or f"egg/{base_id}" - if args.get("repo"): + # Repo(s): accept either the single ``repo`` (back-compat) or a + # ``repos`` list of {repo, base_branch, primary} entries (#3393, + # multi-repo pipelines). Exactly one of the two must be supplied. + # A ``repos`` list is normalised to the wire shape the route expects + # (``data["repos"]``) and its primary entry is mirrored onto the + # legacy ``repo``/``base_branch`` scalars so the single-repo plumbing + # (pipeline naming, primary base-branch resolution) keeps working. + repos_arg = args.get("repos") + if repos_arg is not None: + if args.get("repo"): + return { + "error": ( + "Pass either 'repo' (single-repo) or 'repos' (multi-repo list), not both." + ) + } + if isinstance(repos_arg, str): + try: + repos_arg = json.loads(repos_arg) + except json.JSONDecodeError as e: + return {"error": f"Invalid repos JSON: {e}"} + if not isinstance(repos_arg, list) or not repos_arg: + return {"error": "repos must be a non-empty list of {repo, base_branch} entries"} + normalized: list[dict[str, Any]] = [] + primary_index = 0 + seen_primary = False + for idx, entry in enumerate(repos_arg): + if isinstance(entry, str): + entry = {"repo": entry} + if not isinstance(entry, dict) or not entry.get("repo"): + return {"error": f"repos[{idx}] must be an object with a 'repo' field"} + norm: dict[str, Any] = {"repo": entry["repo"]} + if entry.get("base_branch"): + norm["base_branch"] = entry["base_branch"] + normalized.append(norm) + if entry.get("primary"): + if seen_primary: + return {"error": "At most one repos entry may set 'primary'"} + seen_primary = True + primary_index = idx + # Emit the wire list canonically primary-first (index 0 == primary) and + # drop the transient ``primary`` flag: the route treats ``repos[0]`` as + # primary, and the Pipeline model mirrors ``repos[0]`` onto the legacy + # singleton. Keeping the two in agreement avoids a spurious + # repo/primary conflict rejection when a non-first entry is flagged. + if primary_index != 0: + normalized.insert(0, normalized.pop(primary_index)) + data["repos"] = normalized + primary = normalized[0] + # Mirror the primary onto the legacy scalars for downstream plumbing + # that still reads the singleton (route naming, base-branch detection). + data["repo"] = primary["repo"] + if primary.get("base_branch"): + data["base_branch"] = primary["base_branch"] + elif args.get("repo"): data["repo"] = args["repo"] + if not data.get("repo"): + return { + "error": "Missing repo: pass either 'repo' (single-repo) or 'repos' (multi-repo list)" + } if args.get("config"): config = args["config"] if isinstance(config, str): diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 07a0189a8b..4b109a31c0 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -369,6 +369,7 @@ def get_repo_checks(repo: str) -> list[dict[str, str]]: # type: ignore[misc] PipelineMode, PipelinePhase, PipelineStatus, + RepoSpec, ReviewVerdict, ) from ..slice_id_validation import SLICE_ID_PATTERN, extract_slice_id @@ -428,6 +429,7 @@ def get_repo_checks(repo: str) -> list[dict[str, str]]: # type: ignore[misc] PipelineMode, PipelinePhase, PipelineStatus, + RepoSpec, ReviewVerdict, ) from slice_id_validation import SLICE_ID_PATTERN, extract_slice_id # type: ignore @@ -729,7 +731,11 @@ def _spawn_overseer_agent( except ImportError: from models import AgentRole # type: ignore[no-redef] - overseer_repo = pipeline_repos[0] if pipeline_repos else None + # The overseer resolves its model from the pipeline's PRIMARY repo. + # ``pipeline_repos`` is canonically primary-first (#3393 slices 1-2), so + # take the first (primary) entry via ``next(iter(...))`` rather than a + # positional ``[0]`` collapse (#3393 slice-3). + overseer_repo = next(iter(pipeline_repos or []), None) try: overseer_decision = resolve_overseer_model( "adversarial", @@ -1210,6 +1216,15 @@ def _corrective_open_operator_hitl( if not result.success: raise RuntimeError(f"failed to open operator HITL decision: {result.message}") save_contract(contract, resolved_repo) + # NOTE(#3427): like ``route_impasses``, this overseer-corrective writer + # lands the ``cq-N`` decision with a bare ``save_contract`` and no + # write-time ``persist_contract_statefiles`` — so a HITL opened between + # checkpoints shares the same phase-restart volatility window (the + # ``git reset --hard origin/`` can revert it). The append-only + # guard protects it from id reuse, but not from reversion. Not persisted + # here because the corrective seam runs against ``get_repo_path()`` (the + # base repo), not a pushable pipeline worktree — wiring a worktree-scoped + # persist through the CorrectiveExecutor is the residual follow-up. return decision_id @@ -2068,6 +2083,149 @@ def get_pipeline(pipeline_id: str) -> tuple[Response, int]: ) +def _normalize_submission_repos( + repos_arg: Any, +) -> tuple[str | None, list[dict[str, str | None]], str | None, str | None]: + """Validate + normalize a multi-repo submission list (#3393). + + Accepts the ``repos`` payload from ``POST /api/v1/pipelines`` — a list of + ``{repo, base_branch?, primary?}`` entries (a bare ``"owner/name"`` string + is tolerated as ``{repo: ...}``). Returns + ``(error, entries, primary_repo, primary_base_branch)``: + + * ``error`` — a human-readable message when validation fails (the other + fields are meaningless in that case), else ``None``. + * ``entries`` — normalized ``{"repo", "base_branch"}`` dicts, reordered so + the primary is ``entries[0]`` (the ``Pipeline`` validator mirrors + ``repos[0]`` onto the legacy singleton and ``primary_repo``). + + Per-entry repo/base_branch formats are validated with the same regexes the + single-repo path uses. Same-name repos under different owners are NOT + rejected here — they are distinct full ``owner/name`` slugs (operator + ruling #6; the owner/repo re-key lands in slice 3). + """ + if not isinstance(repos_arg, list) or not repos_arg: + return ("repos must be a non-empty list of {repo, base_branch} entries", [], None, None) + entries: list[dict[str, str | None]] = [] + primary_index = 0 + seen_primary = False + for idx, raw in enumerate(repos_arg): + entry = {"repo": raw} if isinstance(raw, str) else raw + if not isinstance(entry, dict) or not entry.get("repo"): + return (f"repos[{idx}] must be an object with a 'repo' field", [], None, None) + repo_val = entry["repo"] + if not re.match(r"^[a-zA-Z0-9._-]+/[a-zA-Z0-9._-]+$", repo_val): + return ( + f"Invalid repo format in repos[{idx}]: {repo_val!r} (expected owner/name)", + [], + None, + None, + ) + base_val = entry.get("base_branch") + if base_val is not None and ( + not re.match(r"^[a-zA-Z0-9_./-]+$", base_val) or ".." in base_val + ): + return (f"Invalid base_branch in repos[{idx}]: {base_val!r}", [], None, None) + entries.append({"repo": repo_val, "base_branch": base_val}) + if entry.get("primary"): + if seen_primary: + return ("At most one repos entry may set 'primary'", [], None, None) + seen_primary = True + primary_index = idx + # Reorder so the primary is first: the Pipeline model mirrors repos[0] + # onto the legacy repo/base_branch singleton and exposes it as + # ``primary_repo``. + if primary_index != 0: + entries.insert(0, entries.pop(primary_index)) + primary = entries[0] + return (None, entries, primary["repo"], primary["base_branch"]) + + +def _assert_repo_set_uniform(repos: list[str]) -> str | None: + """Reject mixed-visibility / mixed-auth repo sets at submission (#3393, task-2-2). + + A pipeline-wide private-mode posture (context filtering, egress rules) + requires every repo in one run to be uniformly private or uniformly public, + and — for v1 — to share a single auth mode. Returns an actionable, + repo-naming error string when the set diverges on either dimension, or + ``None`` when it is uniform. A single repo (after de-duplication) is + trivially uniform and short-circuits before any lookup, so N=1 pipelines + pay no cost and make no gateway round-trip. + + Runtime note (container boundary): the orchestrator image bundles + ``config/repo_config.py`` but NOT ``gateway/``, so the per-repo lookups are + reached the way the orchestrator already reaches them — auth via + ``repo_config.assert_uniform_auth`` (imported directly, the same callable the + gateway's ``validate_auth_mode_uniformity`` delegates to) and visibility via + ``GatewayClient.get_repo_visibility`` over HTTP (the gateway holds the + tokens; mirrors ``_compute_gateway_mode``). ``internal`` counts as private. + The visibility comparison below is the HTTP-boundary twin of + ``gateway.repo_visibility.validate_visibility_uniformity`` (which the + orchestrator cannot import); keep the two in step. + """ + unique = list(dict.fromkeys(repos)) + if len(unique) <= 1: + return None + + # Auth-mode uniformity — repo_config is bundled into the orchestrator image. + try: + from repo_config import assert_uniform_auth + + assert_uniform_auth(unique) + except ValueError as exc: + return str(exc) + except Exception as exc: # pragma: no cover - defensive (config read failure) + # Fail CLOSED for consistency with the visibility boundary below + # (reviewer_security v1): a config-read failure means we cannot prove a + # uniform auth mode, so we must not admit the set. repo_config is a + # local, bundled read — this path is genuinely exceptional, not a + # transient network hiccup. + logger.warning("Auth-mode uniformity check errored; failing closed", error=str(exc)) + return ( + "Could not determine the auth mode for the pipeline's repos, so a " + "uniform bot/user auth mode cannot be verified. Resubmit once repo " + "configuration is resolvable." + ) + + # Visibility uniformity — resolved via the gateway (the orchestrator's only + # visibility source). FAIL CLOSED on an indeterminate lookup (reviewer_security + # v1): for a multi-repo set (we only reach here when len(unique) > 1) a repo + # whose visibility cannot be resolved to a known bucket means the uniform + # private/public posture cannot be PROVEN — and this is a confidentiality + # boundary (a mixed set that slips through would let private-repo content + # flow through shared plan/contract/PR surfaces into a public repo, with no + # downstream re-check: _compute_gateway_mode derives the network mode from + # the PRIMARY repo only). N=1 short-circuits above, so the common case pays + # nothing. This mirrors gateway.repo_visibility.validate_visibility_uniformity; + # keep the two in step. Unrecognized (non-None) labels are treated as + # indeterminate too — only the known {public|private|internal} contract admits. + gw = get_gateway_client() + posture: dict[str, list[str]] = {} + for repo in unique: + vis = gw.get_repo_visibility(repo) + if vis in ("private", "internal"): + bucket = "private" + elif vis == "public": + bucket = "public" + else: + return ( + f"Could not determine repository visibility for {repo!r}; cannot " + "verify a uniform private/public posture across the pipeline's " + "repos (a run must be uniformly private or uniformly public so " + "private-repo content cannot leak through shared plan/contract/PR " + "surfaces). Resubmit once the repo's visibility is resolvable." + ) + posture.setdefault(bucket, []).append(repo) + if len(posture) > 1: + groups = "; ".join(f"{b}: {', '.join(sorted(rs))}" for b, rs in sorted(posture.items())) + return ( + "Mixed repository visibility across the pipeline's repos is not allowed " + "(a run must be uniformly private or uniformly public, so private-repo " + f"content cannot leak through shared plan/PR surfaces). Diverging repos — {groups}." + ) + return None + + @pipelines_bp.route("", methods=["POST"]) @require_lifecycle_secret def create_pipeline() -> tuple[Response, int]: @@ -2108,6 +2266,35 @@ def create_pipeline() -> tuple[Response, int]: branch = data.get("branch") base_branch = data.get("base_branch") prompt = data.get("prompt") + + # #3393 (multi-repo): a submission may carry a ``repos`` list instead of + # (or in addition to) the single ``repo``. Normalize it up front and derive + # the primary onto the legacy ``repo``/``base_branch`` scalars so the + # single-repo plumbing below (naming, base-branch detection, branch checks) + # keeps working and a direct HTTP submission — one that bypasses the + # submit_task MCP tool that would otherwise mirror the primary — is + # supported. ``repos_entries`` is None for a single-repo submission. + repos_entries: list[dict[str, str | None]] | None = None + repos_arg = data.get("repos") + if repos_arg is not None: + _repos_err, repos_entries, _primary_repo, _primary_base = _normalize_submission_repos( + repos_arg + ) + if _repos_err: + return make_error_response( + _repos_err, status_code=400, details={"reason": "invalid_repos"} + ) + if repo and _primary_repo and repo != _primary_repo: + return make_error_response( + f"Conflicting repo {repo!r} and repos primary {_primary_repo!r}; " + "pass one or the other.", + status_code=400, + details={"reason": "repo_repos_conflict"}, + ) + if not repo: + repo = _primary_repo + if not base_branch: + base_branch = _primary_base mode = data.get("mode", "issue") analysis = data.get("analysis") plan = data.get("plan") @@ -2468,6 +2655,37 @@ def create_pipeline() -> tuple[Response, int]: }, ) + # #3393 (multi-repo): enforce uniform visibility + auth across the run's + # repos before creating the pipeline. Single-repo submissions are trivially + # uniform and short-circuit without a gateway round-trip. Runs after the + # gateway-ready gate above so the visibility lookup can reach the gateway. + _uniform_repos = ( + [e["repo"] for e in repos_entries] if repos_entries else ([repo] if repo else []) + ) + _uniformity_err = _assert_repo_set_uniform([r for r in _uniform_repos if r]) + if _uniformity_err: + return make_error_response( + _uniformity_err, + status_code=400, + details={"reason": "non_uniform_repo_set"}, + ) + + # Assemble the full list-shaped repo set persisted onto the Pipeline. The + # primary (entries[0]) carries the resolved ``base_branch`` (detected above + # when absent); secondary repos keep their submitted base_branch (None ⇒ + # auto-detected downstream). For a single-repo submission we leave + # ``repos_specs`` as None and let the Pipeline validator synthesize a + # one-element list from the legacy singleton (N=1 back-compat). + repos_specs: list[RepoSpec] | None = None + if repos_entries is not None: + repos_specs = [ + RepoSpec( + repo=entry["repo"], + base_branch=(base_branch if idx == 0 else entry["base_branch"]), + ) + for idx, entry in enumerate(repos_entries) + ] + try: store = get_state_store(repo_path) pipeline = store.create_pipeline( @@ -2475,6 +2693,7 @@ def create_pipeline() -> tuple[Response, int]: repo=repo, branch=branch, base_branch=base_branch, + repos=repos_specs, config=config, prompt=prompt, network_mode=network_mode, @@ -5609,16 +5828,35 @@ def _get_refine_review_criteria() -> str: "- Is there a clear recommended approach?\n" "- Is the recommendation justified with specific reasons?\n" "- Does the recommendation align with the analysis findings?\n\n" - "### 7. HITL Decision Registration\n" - "- Run `egg-contract show` and verify that contract decisions or feedback " - "items exist for every open question in the analysis.\n" - "- If open questions appear as prose text without corresponding " - "`` or `` " - "markers (generated by `egg-contract`), flag as `needs_revision` — " - "the agent must re-run `egg-contract add-decision` or " - "`egg-contract add-feedback` for each question.\n" - "- If there are zero open questions, verify that the requirements are " - "genuinely unambiguous and no assumptions were made silently.\n\n" + "### 7. HITL Decision Registration & Un-surfaced Decisions (#3390)\n" + "- Run `egg-contract show` and verify a contract decision or feedback " + "item exists for every open question in the analysis, and that each " + "decision-bearing section cites its `cq-N` (the `--format markdown` " + "output of `egg-contract add-decision` embeds it). Open questions as " + "bare prose with no registered `cq-N` ⇒ **NACK** — the producer must " + "register each via `egg-contract add-decision` / " + "`egg-contract add-feedback` and re-propose. (Deterministic " + "propose-time checks already validate the producer's *attested* ids; " + "your job is the judgment half the validators cannot do.)\n" + "- **Un-surfaced decisions — NACK.** Read the draft for choices it " + "quietly *commits to* that should be the operator's call — e.g. " + '"we will drop the legacy filter", a scope narrowing/widening, a ' + "user-visible behavior change, abandoning a stated requirement — " + "with no registered `cq-N` backing the choice. Consensus must not " + "close on a draft that bakes in a human-grade decision outside the " + "HITL channel: the producer either registers the decision (the gate " + "then surfaces it) or rewrites the draft to remove the unilateral " + "commitment.\n" + "- **Calibration — do not over-NACK.** An implementation choice the " + "planner can make from the analysis (API shape, migration approach, " + "fallback design, detector shape) is NOT a human-grade decision; do " + "not force registration of those. The bar is the same as §5: answers " + "only the operator owns (product intent, scope boundaries, external " + "commitments, user-visible behavior).\n" + "- If the ledger is deliberately empty (the producer attested " + "`no_decisions_rationale`), verify the rationale holds: requirements " + "genuinely unambiguous, no assumptions made silently. NACK if you " + "find a hidden operator-grade choice.\n\n" + _human_companion_review_criteria( companion="`*-analysis-human.md`", parent="the refine analysis", @@ -6001,6 +6239,30 @@ def _get_plan_review_criteria() -> str: '`python3 -c "from egg_contracts.plan_parser import parse_plan_file, ' "validate_slice_file_overlap as v; r = parse_plan_file(''); " "print('\\n'.join(v(r.to_contract_slices())))\"`.\n\n" + "### 13. HITL Decision Registration & Un-surfaced Decisions (#3390)\n" + "- Run `egg-contract show` and check the plan-phase decision ledger: " + "every plan-phase open question must be a registered contract " + "decision (`cq-N`), and the plan draft must cite the id where the " + "question is raised. A plan-grade question living only in prose ⇒ " + "**NACK** the producer that owns it (task_planner for the plan " + "draft, architect for slice-shape questions, risk_analyst for " + "risk-acceptance questions).\n" + "- **Un-surfaced decisions — NACK.** A plan that silently commits to " + "a choice only the operator owns — dropping a requirement, changing " + "user-visible behavior, accepting a risk the operator never saw, " + "de-scoping acceptance criteria — without a registered `cq-N` bakes " + "a human-grade decision into the pipeline outside the HITL channel. " + "NACK: the producer registers the decision or removes the " + "unilateral commitment.\n" + "- **Calibration — do not over-NACK.** Design calls the plan phase " + "legitimately owns (task decomposition, API shape, migration " + "approach, slice ordering within the architect's constraints) are " + "NOT operator decisions — do not force registration of those. The " + "bar is answers only the operator owns (product intent, scope " + "boundaries, external commitments, user-visible behavior).\n" + "- A deliberately empty ledger arrives as a producer's " + "`no_decisions_rationale` attestation — verify it holds; NACK if " + "the plan hides an operator-grade choice.\n\n" + _human_companion_review_criteria( companion="`*-plan-human.md`", parent="the implementation plan", @@ -7977,12 +8239,21 @@ class WorktreeSyncOutcome(NamedTuple): summaries) that are on HEAD but not yet on origin. Empty when the rev-list itself failed; the divergence is still reported, but the operator can't be given the exact commit list inline. + + ``rebase_category`` / ``rebase_detail`` carry the failing rebase's + ``PushResult.category`` / ``detail`` (conflicting paths, the rebase + argv, and a git-output excerpt) when ``diverged_unreconciled`` is + True. They exist so the reconcile HITL can show the operator *what* + failed instead of an unfalsifiable generic claim (#3416) — the log + lines carry the same data but roll; the decision persists. """ case: str diverged_unreconciled: bool = False backup_ref: str | None = None local_only_commit_shas: tuple[str, ...] = () + rebase_category: str | None = None + rebase_detail: str | None = None def _build_sync_recovery_backup_ref(pipeline_id: str, unix_ts: int) -> str: @@ -8546,6 +8817,8 @@ def _sync_worktree_with_remote( diverged_unreconciled=True, backup_ref=backup_ref if backup_ok else None, local_only_commit_shas=local_only, + rebase_category=rebase_outcome.category, + rebase_detail=rebase_outcome.detail, ) # Step 4: Reset local branch to remote. @@ -9497,6 +9770,78 @@ def _commit_statefiles_to_worktree( return True +def persist_contract_statefiles( + pipeline_id: str, + worktree_path: Path, + message: str, + *, + pipeline: Pipeline | None = None, +) -> bool: + """Durably persist a contract decision write: commit + push to the work branch. + + Contract HITL decisions (``cq-N`` registrations and resolutions) are + written to the shared pipeline worktree's contract file with no git + commit; the file was only serialized to the work branch at slice/phase + checkpoints. Both phase-(re)start syncs — the gateway's worktree-reuse + reset and ``_sync_worktree_with_remote`` step 4 — run + ``git reset --hard origin/``, so any decision write that had not + been committed AND pushed by then was silently reverted, letting the + bootstrap reconciler re-mint the same ``cq-N`` ids and clobber + just-resolved operator decisions (#3427). Committing and pushing at + write time makes the reset target already contain the decision. + + Best-effort by design: failures are logged and swallowed — the write is + still live on the worktree file and the next checkpoint commit retries. + Returns ``True`` only when the state was committed and pushed (or there + was nothing new to commit). + """ + try: + if pipeline is None: + _, pipeline = _resolve_pipeline(pipeline_id, get_repo_path()) + identifier = _pipeline_identifier(getattr(pipeline, "issue_number", None), pipeline_id) + committed = _commit_statefiles_to_worktree( + worktree_path, + message, + identifier, + pipeline_id=pipeline_id, + ) + if not committed: + return True # Nothing new on disk — already durable. + branch = getattr(pipeline, "branch", None) + if not branch: + logger.warning( + "Contract decision write committed but pipeline has no work " + "branch to push to; the commit is local-only and a worktree " + "reset may still discard it (#3427)", + pipeline_id=pipeline_id, + ) + return False + gateway_mode, _ = _compute_gateway_mode(pipeline) + _get_spawner().gateway.push_worktree_branch( + pipeline_id=pipeline_id, + repo_path=str(worktree_path), + branch=branch, + mode=gateway_mode, + base_branch=getattr(pipeline, "base_branch", None), + ) + logger.info( + "Contract decision write persisted to work branch (#3427)", + pipeline_id=pipeline_id, + branch=branch, + commit_message=message, + ) + return True + except Exception as persist_err: # noqa: BLE001 — best-effort durability + logger.warning( + "Failed to durably persist contract decision write; the decision " + "is live on the worktree file but will not survive a worktree " + "reset until the next checkpoint commit (#3427)", + pipeline_id=pipeline_id, + error=str(persist_err), + ) + return False + + def _ensure_statefiles_on_branch( worktree_repo_path: Path, pipeline: Pipeline, @@ -10393,6 +10738,47 @@ def _resolve_pipeline_worktree_path(pipeline: Pipeline, fallback: Path) -> Path: return fallback +def _resolve_slice_gate_repo(slice_obj, pipeline: Pipeline) -> str | None: + """The repo every implement-phase gate for *slice_obj* is scoped to (#3393). + + Single source of truth for slice → gate-repo resolution (task-6-1): the + test gate, the reviewer diff base, the per-repo check/lint commands, and + the slice agent's cwd all key off this one accessor. It is exactly + :func:`models.resolve_slice_repo` — the slice's own ``repo`` when set, + else the pipeline's primary repo (so a repoless slice, or any slice in an + N=1 pipeline, scopes to the single/primary repo). Returns ``None`` only + for a genuinely repoless pipeline (test scaffolds with no repo at all). + """ + try: + from models import resolve_slice_repo # type: ignore[no-redef] + except ImportError: + from ..models import resolve_slice_repo # type: ignore[no-redef] + return resolve_slice_repo(slice_obj, pipeline) + + +def _resolve_slice_worktree_path( + pipeline: Pipeline, slice_repo: str | None, fallback: Path +) -> Path: + """Resolve the on-disk worktree path for a slice's repo (#3393 task-6-1). + + A multi-repo pipeline materialises one worktree per participating repo + under ``WORKTREE_BASE_DIR / pipeline.id / `` — the same + owner/repo-keyed layout as :func:`_resolve_pipeline_worktree_path`, one + directory per repo. Given a slice's resolved repo (``owner/name``), this + returns that repo's worktree when it exists on disk, else *fallback* + (the pipeline-primary worktree). For an N=1 pipeline the slice's repo IS + the primary, so ``slice_repo`` matches ``pipeline.repo`` and the answer + is byte-identical to the pipeline-primary worktree — callers therefore + only reach here for a genuine secondary-repo slice. + """ + repo_short = slice_repo.split("/")[-1] if slice_repo else None + if repo_short: + candidate = WORKTREE_BASE_DIR / pipeline.id / repo_short + if candidate.exists(): + return candidate + return fallback + + def _persist_phase_brc_history( pipeline: Pipeline, store: StateStore, @@ -10441,6 +10827,10 @@ def _persist_phase_brc_history( worktree_path, f"Persist statefiles after {phase} phase", pipeline_identifier=_pipeline_identifier(pipeline.issue_number, pipeline.id), + # Contract files are keyed by pipeline_id, not the issue-number + # prefix; without this the restart-time persist skipped the + # contract entirely (#1829 gap, observed in #3427). + pipeline_id=pipeline.id, ) except subprocess.CalledProcessError as git_err: logger.warning( @@ -10639,6 +11029,8 @@ def _compose_context_pr_body( pipeline, worktree_repo_path: Path, identifier: int | str, + context_repo: str | None = None, + sibling_context_prs: list[dict[str, Any]] | None = None, ) -> str: """Compose the context-PR body from contract + pipeline state (#3115). @@ -10699,6 +11091,20 @@ def _compose_context_pr_body( body_lines.append(f"- Issue: #{pipeline.issue_number}") has_meaningful_content = True + # #3393 slice-4 / task-4-2: the repo this context PR lives in. A + # slice PR in this same repo cross-links as a bare ``#N`` autolink; + # a slice PR in a DIFFERENT repo of the pipeline must be qualified + # as ``owner/repo#N`` (a bare ``#N`` would resolve against the wrong + # repo). Defaults to the pipeline primary — the repo the up-front + # opener composes the primary context PR for. For an N=1 pipeline + # every slice resolves to the primary, so every link stays bare and + # the body is byte-identical to the single-repo shape. + this_context_repo = context_repo or getattr(pipeline, "primary_repo", None) or pipeline.repo + try: + from models import resolve_slice_repo # type: ignore[no-redef] + except ImportError: + from ..models import resolve_slice_repo # type: ignore[no-redef] + slices = list(contract.slices or []) if slices: body_lines.append(f"- Slices ({len(slices)}):") @@ -10712,10 +11118,17 @@ def _compose_context_pr_body( line = f" {number}. {name} (`{s.id}`)" # Cross-link the stack (#3122): once the slice's PR is open # its number is persisted on the contract and the run loop - # re-composes this body, so the entry gains a link. Bare - # ``#N`` autolinks within the repo the context PR lives in. + # re-composes this body, so the entry gains a link. if getattr(s, "pr_number", None): - line += f" — #{s.pr_number}" + s_repo = resolve_slice_repo(s, pipeline) + if s_repo and this_context_repo and s_repo != this_context_repo: + # Cross-repo sibling — repo-qualify so GitHub resolves + # the autolink to the right repo (#3393 slice-4). + line += f" — {s_repo}#{s.pr_number}" + else: + # Same-repo (or repo unknown): bare ``#N`` autolinks + # within the repo this context PR lives in. + line += f" — #{s.pr_number}" body_lines.append(line) has_meaningful_content = True @@ -10748,6 +11161,35 @@ def _compose_context_pr_body( if has_meaningful_content: sections.append("\n".join(["## Pipeline context", "", *body_lines])) + + # #3393 slice-4 / task-4-2: cross-reference the pipeline's context + # PRs in OTHER repos. Rendered only for a multi-repo pipeline (the + # opener passes ``sibling_context_prs`` when it coordinates >1 + # repo); an N=1 pipeline passes ``None`` and this section is + # omitted, keeping the body byte-identical to the single-repo shape. + coord_lines: list[str] = [] + for ref in sibling_context_prs or []: + ref_repo = (ref.get("repo") or "").strip() + ref_number = ref.get("number") + if not ref_repo or not isinstance(ref_number, int) or isinstance(ref_number, bool): + continue + if ref_number < 1: + continue + # ``owner/repo#N`` autolinks cross-repo (a bare ``#N`` would + # resolve against the repo this body lives in). + coord_lines.append(f"- {ref_repo}#{ref_number}") + if coord_lines: + sections.append( + "\n".join( + [ + "## Coordinated repos", + "", + "This pipeline coordinates PRs across multiple repos (#3393):", + "", + *coord_lines, + ] + ) + ) return "\n\n".join(sections) @@ -11212,6 +11654,16 @@ def _open_context_pr_at_implement_start( head=pipeline.branch, base=effective_base, ) + _maybe_open_secondary_context_prs( + pipeline_id, + pipeline=pipeline, + primary_pr_number=existing_pr_number, + work_branch=pipeline.branch, + worktree_repo_path=worktree_repo_path, + identifier=identifier, + gateway_mode=gateway_mode, + spawner=spawner, + ) return existing_pr_number # Step 4: open a new context PR. Read title/description from the @@ -11314,9 +11766,284 @@ def _open_context_pr_at_implement_start( base=effective_base, url=pr_url, ) + _maybe_open_secondary_context_prs( + pipeline_id, + pipeline=pipeline, + primary_pr_number=new_pr_number, + work_branch=pipeline.branch, + worktree_repo_path=worktree_repo_path, + identifier=identifier, + gateway_mode=gateway_mode, + spawner=spawner, + ) return new_pr_number +def _repos_with_slices(contract, pipeline) -> list[str]: + """Repos that own ≥1 slice — the lazy-per-repo participation set (#3393, slice-4). + + A repo *participates* (gets its own ``egg//work`` branch + context + PR) iff at least one slice resolves to it via + :func:`models.resolve_slice_repo`. The result is ordered by + ``pipeline.repos`` and de-duplicated; a submitted repo that ends up + owning no slices is excluded (operator ruling #1). For an N=1 pipeline + this returns the single repo. This is the invariant the context-PR + opener's per-repo iteration honours (task-4-2). + """ + try: + from models import resolve_slice_repo # type: ignore[no-redef] + except ImportError: + from ..models import resolve_slice_repo # type: ignore[no-redef] + + slices = getattr(contract, "slices", None) or [] + owning = {resolve_slice_repo(s, pipeline) for s in slices} + return [spec.repo for spec in (pipeline.repos or []) if spec.repo in owning] + + +def _maybe_open_secondary_context_prs( + pipeline_id: str, + *, + pipeline: Any, + primary_pr_number: int, + work_branch: str | None, + worktree_repo_path: Path, + identifier: int | str, + gateway_mode: str, + spawner: Any, +) -> None: + """Guarded, never-raising entry to the lazy per-repo context opener (#3393). + + No-op unless the pipeline coordinates more than one repo, so the N=1 + single-repo path in :func:`_open_context_pr_at_implement_start` + performs zero extra work (no contract load, no gateway calls) and is + byte-for-byte unchanged. Requires a resolvable primary repo + work + branch; both are guaranteed set on the multi-repo remote path that + reaches here (the opener already returned for local-mode pipelines). + """ + if len(getattr(pipeline, "repos", None) or []) <= 1: + return + primary_repo = pipeline.primary_repo + if not primary_repo or not work_branch: + return + try: + _open_secondary_context_prs( + pipeline_id, + pipeline=pipeline, + primary_repo=primary_repo, + primary_pr_number=primary_pr_number, + work_branch=work_branch, + worktree_repo_path=worktree_repo_path, + identifier=identifier, + gateway_mode=gateway_mode, + spawner=spawner, + ) + except Exception as sec_err: # noqa: BLE001 + logger.warning( + "Lazy per-repo context PRs raised (continuing — primary context PR unaffected) (#3393)", + pipeline_id=pipeline_id, + error=str(sec_err), + ) + + +def _open_secondary_context_prs( + pipeline_id: str, + *, + pipeline: Any, + primary_repo: str, + primary_pr_number: int, + work_branch: str, + worktree_repo_path: Path, + identifier: int | str, + gateway_mode: str, + spawner: Any, +) -> dict[str, int]: + """Open the lazy per-repo context PRs for a multi-repo pipeline (#3393, slice-4 / task-4-2). + + :func:`_open_context_pr_at_implement_start` opens the PRIMARY repo's + context PR (``egg//work → base``) exactly as it always has. This + helper adds the *other* repos: it iterates the set of repos that own + ≥1 slice (via ``resolve_slice_repo`` over the contract's slices), + drops the primary, and for each remaining repo opens that repo's own + ``egg//work`` context PR (same branch naming, per repo). A + submitted repo with NO slices is skipped — lazy-per-repo, operator + ruling #1. Every opened context PR (primary + secondaries) then has + its body refreshed to cross-reference the sibling context PRs in the + other repos (``## Coordinated repos``). + + It is only invoked when ``len(pipeline.repos) > 1``; for an N=1 + pipeline the caller never reaches here, so the single-repo path is + byte-for-byte unchanged. + + Prerequisite / current limit (honest scope note): opening a context + PR in a secondary repo requires that repo's ``egg//work`` branch + to exist on its remote, which in turn needs a secondary-repo worktree + to push it. Threading the full repo set into worktree CREATION was + explicitly deferred by slice-3 (the worktree map is owner/repo-keyed + and list-shaped, but only the primary repo is materialised today), so + until that later wiring lands the secondary ``create_pr`` will + typically fail on a missing head branch. This helper therefore: + + * uses the launcher-auth ``lookup_open_pr`` idempotency primitive + (which works per-repo with no worktree) to ADOPT an already-open + secondary context PR, and + * ATTEMPTS ``create_pr`` otherwise, soft-failing (log, continue) so a + missing secondary branch never strands the pipeline. + + The iteration + cross-referencing structure is therefore complete and + forward-compatible: once secondary-repo worktree/branch creation is + wired, secondary context PRs open with no further change here. + + Every failure is caught and logged; the helper never raises. Returns + the ``{repo: pr_number}`` map of context PRs known after the pass + (always including the primary), for logging / tests. + """ + opened: dict[str, int] = {primary_repo: primary_pr_number} + + try: + from egg_contracts.loader import load_contract + except ImportError: + logger.warning( + "Secondary context PRs: egg_contracts.loader unavailable (skipping) (#3393)", + pipeline_id=pipeline_id, + ) + return opened + + try: + contract = load_contract(identifier, worktree_repo_path) + except Exception as load_err: # noqa: BLE001 + logger.warning( + "Secondary context PRs: contract load failed (skipping) (#3393)", + pipeline_id=pipeline_id, + error=str(load_err), + ) + return opened + + # Repos owning ≥1 slice (ordered by ``pipeline.repos``), minus the + # primary — the lazy-per-repo participation set (task-4-2). + secondary_repos = [r for r in _repos_with_slices(contract, pipeline) if r != primary_repo] + + if not secondary_repos: + # Multi-repo pipeline whose slices all resolve to the primary + # (e.g. no slice pinned a secondary repo). Nothing lazy to open. + return opened + + base_by_repo = {spec.repo: spec.base_branch for spec in (pipeline.repos or [])} + context_title = ( + contract.pr.title.strip() + if contract.pr and (contract.pr.title or "").strip() + else f"{identifier} context" + ) + + for repo in secondary_repos: + # ``base_branch=None`` ⇒ the repo's default branch. Without a + # secondary worktree we cannot run ``_detect_default_branch`` + # here, so fall back to ``main`` (the create call resolves the + # real default server-side when base is omitted anyway). + base = base_by_repo.get(repo) or "main" + try: + existing = spawner.gateway.lookup_open_pr( + pipeline_id=pipeline_id, + repo=repo, + head=work_branch, + base=base, + ) + if existing is not None: + opened[repo] = existing + logger.info( + "Secondary context PR: adopted existing PR (#3393)", + pipeline_id=pipeline_id, + repo=repo, + pr_number=existing, + ) + continue + + body = _compose_context_pr_body( + contract=contract, + pipeline=pipeline, + worktree_repo_path=worktree_repo_path, + identifier=identifier, + context_repo=repo, + sibling_context_prs=[ + {"repo": r, "number": n} for r, n in opened.items() if r != repo + ], + ) + pr_url = spawner.gateway.create_pr( + pipeline_id=pipeline_id, + repo=repo, + title=context_title, + body=body, + head=work_branch, + base=base, + issue_number=pipeline.issue_number, + mode=gateway_mode, # type: ignore[arg-type] + ) + match = re.search(r"/pull/(\d+)(?:[/?#]|$)", pr_url or "") + if match: + opened[repo] = int(match.group(1)) + logger.info( + "Secondary context PR: opened new PR (#3393)", + pipeline_id=pipeline_id, + repo=repo, + pr_number=opened[repo], + head=work_branch, + base=base, + ) + else: + logger.warning( + "Secondary context PR: create returned no parseable URL (#3393)", + pipeline_id=pipeline_id, + repo=repo, + url=pr_url, + ) + except Exception as sec_err: # noqa: BLE001 + # Best-effort: a missing secondary ``egg//work`` branch + # (the deferred-worktree limit above) surfaces here as a + # gateway create failure. Log + continue so the primary + # context PR + slice stack are unaffected. + logger.warning( + "Secondary context PR deferred (continuing) — secondary-repo " + "work branch likely absent until secondary worktree creation " + "is wired (#3393)", + pipeline_id=pipeline_id, + repo=repo, + error=str(sec_err), + ) + + # Cross-reference pass: refresh every opened context PR body so each + # links the sibling context PRs in the other repos. Best-effort and + # cosmetic — a failed refresh never affects the slice stack. + if len(opened) > 1: + for repo, number in opened.items(): + try: + body = _compose_context_pr_body( + contract=contract, + pipeline=pipeline, + worktree_repo_path=worktree_repo_path, + identifier=identifier, + context_repo=repo, + sibling_context_prs=[ + {"repo": r, "number": n} for r, n in opened.items() if r != repo + ], + ) + spawner.gateway.update_pr_body( + pipeline_id=pipeline_id, + repo=repo, + pr_number=number, + body=body, + issue_number=pipeline.issue_number, + mode=gateway_mode, # type: ignore[arg-type] + ) + except Exception as refresh_err: # noqa: BLE001 + logger.warning( + "Coordinated-repos cross-reference refresh failed (continuing) (#3393)", + pipeline_id=pipeline_id, + repo=repo, + error=str(refresh_err), + ) + + return opened + + def _is_slice_dag_mode(contract) -> bool: """Return True when the contract represents a multi-slice DAG (#2777, cq-10). @@ -11833,12 +12560,18 @@ def _escalate_layer_c_hitl( one is added in a follow-up. """ try: - from egg_contracts.decisions import next_cq_id + from egg_contracts.decisions import ( + find_duplicate_open_question, + find_resolved_question, + next_cq_id, + ) from egg_contracts.loader import load_contract, save_contract from egg_contracts.models import Decision, DecisionOption, DecisionType except ImportError: try: from orchestrator.egg_contracts.decisions import ( # type: ignore[no-redef] + find_duplicate_open_question, + find_resolved_question, next_cq_id, ) from orchestrator.egg_contracts.loader import ( # type: ignore[no-redef] @@ -11861,6 +12594,35 @@ def _escalate_layer_c_hitl( try: with get_pipeline_state_lock(pipeline_id): contract_local = load_contract(pipeline_id, worktree_repo_path) + existing_decisions = contract_local.decisions or [] + decision_phase = current_phase or PipelinePhase.IMPLEMENT + # Dedupe/carry-forward — parity with ``register_open_question`` + # (#3374/#3392). The Layer-C question text is deterministic per + # (case, slice, pipeline), so every bootstrap re-run after a + # ``restart_phase`` re-derives the identical question. Without + # this guard each re-run minted a fresh ``cq-N`` (or, against a + # reset-stale contract, re-minted an existing one), making the + # operator re-answer questions they had already answered (#3427). + duplicate = find_duplicate_open_question(existing_decisions, question, decision_phase) + if duplicate is not None: + logger.info( + "Layer-C HITL escalation adopted existing open decision (slice-4 TASK-4-4)", + pipeline_id=pipeline_id, + slice_id=slice_id, + decision_id=getattr(duplicate, "id", None), + ) + return + carried = find_resolved_question(existing_decisions, question, decision_phase) + if carried is not None: + logger.info( + "Layer-C HITL escalation skipped: identical question " + "already resolved by the operator (slice-4 TASK-4-4)", + pipeline_id=pipeline_id, + slice_id=slice_id, + decision_id=getattr(carried, "id", None), + resolution=str(getattr(carried, "resolution", None))[:200], + ) + return # Use the canonical ``cq-N`` allocator from # ``shared/egg_contracts/decisions.py``. Orchestrator-side # HITL escalations write to the ``cq-N`` namespace; the @@ -11880,7 +12642,8 @@ def _escalate_layer_c_hitl( # bootstrap which can run before any phase walk, and # future slice-DAG topologies may span phases. # - # The ``or PipelinePhase.IMPLEMENT`` arm is defensive: the + # The ``or PipelinePhase.IMPLEMENT`` arm (folded into + # ``decision_phase`` above) is defensive: the # ``Pipeline.current_phase`` field is non-Optional with a # default at the schema layer (``models.py:1032``), so # in-tree callers should always populate it. The fallback @@ -11893,7 +12656,7 @@ def _escalate_layer_c_hitl( id=decision_id, question=question, type=DecisionType.HITL, - phase=current_phase or PipelinePhase.IMPLEMENT, + phase=decision_phase, options=options, ) ) @@ -11904,6 +12667,13 @@ def _escalate_layer_c_hitl( slice_id=slice_id, decision_id=decision_id, ) + # Durably land the new decision on the work branch so the next + # phase-(re)start worktree reset cannot revert it (#3427). + persist_contract_statefiles( + pipeline_id, + worktree_repo_path, + f"Persist Layer-C HITL escalation {decision_id} (#3427)", + ) except Exception as escalate_err: # noqa: BLE001 logger.warning( "Layer-C HITL escalation failed (slice-4 TASK-4-4); slice will " @@ -11973,6 +12743,219 @@ def _escalate_blocked_slice_to_hitl( ) +# --- #3393 slice-5: cross-repo merge-sequencing HITL holds ------------------- +# Stable discriminator prefix on the cross-repo-hold Decision question so +# (a) the poll can idempotently detect an already-registered hold for a +# gate across reconciler ticks / orchestrator restarts, and (b) a future +# dispatch handler in ``routes/decisions.py`` can route on the literal +# substring without a separate context field on the contract Decision. +_CROSS_REPO_HOLD_MARKER_PREFIX = "[#3393 cross-repo-hold" + + +def _cross_repo_hold_marker(slice_id: str) -> str: + """Return the stable per-gate discriminator embedded in the hold question.""" + return f"{_CROSS_REPO_HOLD_MARKER_PREFIX} slice={slice_id}]" + + +_CROSS_REPO_HOLD_REASON_TEXT = { + "closed_unmerged": ( + "the upstream cross-repo PR was CLOSED without merging, so the " + "automated merge-state hold cannot auto-ready this slice's PR" + ), + "timeout": ( + "the upstream cross-repo PR did not merge within the poll bound, so " + "the automated merge-state hold timed out rather than leaving this " + "slice's PR draft indefinitely" + ), + "beyond_merge_state": ( + "the plan declared this cross-repo dependency a beyond-merge-state " + "condition (release/publish, version-pin, or cannot-continue block), " + "which is released by human decision, never automated detection" + ), +} + + +# The two operator-selectable options on a cross-repo hold Decision. The +# RELEASE option readies the PR; the KEEP option leaves it draft for manual +# handling. Kept as constants so the registration (options list) and the +# resolution reader agree on one shape. +_CROSS_REPO_HOLD_RELEASE_OPTION_ID = "opt-release" +_CROSS_REPO_HOLD_RELEASE_OPTION_LABEL = "Release the hold and mark the PR ready" +_CROSS_REPO_HOLD_KEEP_OPTION_ID = "opt-keep" +_CROSS_REPO_HOLD_KEEP_OPTION_LABEL = "Keep the PR held for manual handling" + + +def _cross_repo_hold_resolution(contract: Any, slice_id: str) -> str | None: + """Return the human's verdict on the cross-repo hold Decision for a slice. + + Scans the (freshly-loaded) contract for the Decision carrying this gate's + :func:`_cross_repo_hold_marker` and, when it is resolved, maps the + operator's SELECTED option to a gate verdict: + + * :data:`cross_repo_merge_gate.RELEASE` — the release option was chosen + (mark the PR ready), else + * :data:`cross_repo_merge_gate.KEEP` — the keep-held option was chosen, OR + the resolution is present but unrecognized (fail-safe: an ambiguous + resolution must NOT auto-ready — cq-1 "human owns the release"). + + Returns ``None`` when the Decision is absent or not yet resolved (keep + waiting). The stored ``Decision.resolution`` may be the option label, the + option id, or a ``{"action":"select","selected":