diff --git a/orchestrator/cli.py b/orchestrator/cli.py index 3ad20b556f..27241c42e1 100644 --- a/orchestrator/cli.py +++ b/orchestrator/cli.py @@ -420,16 +420,23 @@ def cmd_pipelines_list(args: argparse.Namespace) -> int: def cmd_pipelines_create(args: argparse.Namespace) -> int: """Create a new pipeline.""" + from routes.pipelines import _ensure_pipeline_work_ref from state_store import get_state_store repo_path = Path(args.repo_path) if args.repo_path else Path.cwd() store = get_state_store(repo_path) + # Route through the same normalisation as the HTTP `create_pipeline` + # endpoint so a CLI-provisioned pipeline gets the ``/work`` + # shape and slice integration branches at ``/slice-N`` can + # coexist as siblings (#2399). + branch = _ensure_pipeline_work_ref(args.branch or f"egg/issue-{args.issue}") + try: pipeline = store.create_pipeline( issue_number=args.issue, repo=args.repo, - branch=args.branch or f"egg/issue-{args.issue}", + branch=branch, ) if args.json: @@ -956,7 +963,10 @@ def create_parser() -> argparse.ArgumentParser: create_parser = pipelines_subparsers.add_parser("create", help="Create a pipeline") create_parser.add_argument("--issue", type=int, required=True, help="Issue number") create_parser.add_argument("--repo", required=True, help="Repository (owner/repo)") - create_parser.add_argument("--branch", help="Branch name (default: egg/issue-N)") + create_parser.add_argument( + "--branch", + help="Branch name (default: egg/issue-N; normalised to egg/issue-N/work)", + ) create_parser.add_argument("--repo-path", help="Repository path") create_parser.add_argument("--json", action="store_true", help="Output as JSON") create_parser.set_defaults(func=cmd_pipelines_create) diff --git a/orchestrator/concurrent_executor.py b/orchestrator/concurrent_executor.py index 54b76af45e..4c2894bfce 100644 --- a/orchestrator/concurrent_executor.py +++ b/orchestrator/concurrent_executor.py @@ -41,6 +41,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] get_peer_consensus_tracker, ) from review_graph import ReviewGraph, get_review_graph_for_phase +from slice_id_validation import SLICE_ID_PATTERN logger = get_logger("orchestrator.concurrent_executor") @@ -270,8 +271,20 @@ def get_worktree_branch( # cannot see in the slice PR's diff. We honour the # pipeline's existing branch as the issue prefix when set, # otherwise fall back to the issue-number / pipeline id. + # + # The pipeline tip is pushed to ``egg//work`` (#2399), so + # the slice integration branch lives as a sibling of ``/work`` + # under ``egg//`` — strip the trailing ``/work`` from the + # pipeline branch to get the namespace root. issue = self.pipeline.issue_number or self.pipeline.id issue_branch = self.pipeline.branch or f"egg/issue-{issue}" + # Structural check (≥2 slashes, last segment ``work``) — see + # ``_slice_namespace_root`` in ``routes/pipelines.py`` for + # the matching helper. A degenerate single-segment input + # like ``egg/work`` is treated as the root itself rather + # than collapsing to ``egg``. + if issue_branch.count("/") >= 2 and issue_branch.rsplit("/", 1)[1] == "work": + issue_branch = issue_branch.rsplit("/", 1)[0] normalised_slice = slice_id if slice_id.startswith("slice-") else f"slice-{slice_id}" # Defense-in-depth: re-validate the normalised slice id # shape before embedding it in a git ref. The contract- @@ -281,10 +294,10 @@ def get_worktree_branch( # validation must not be able to smuggle path separators # or shell metacharacters in via this seam (per the # security reviewer's defense-in-depth suggestion on the - # v1 BRC review). - import re - - if not re.fullmatch(r"slice-[0-9]+", normalised_slice): + # v1 BRC review). The pattern is the canonical one shared + # with the signal handlers (#2403) and the operator restart + # route (#2410) — see ``slice_id_validation``. + if not SLICE_ID_PATTERN.fullmatch(normalised_slice): raise ValueError( f"slice_id={slice_id!r} does not match the canonical shape ``slice-``" ) @@ -298,21 +311,28 @@ def get_worktree_branch( def get_slice_integration_branch(self, slice_id: str) -> str: """Return the shared integration branch for a slice's BRC. - Each slice has its own integration branch under the pipeline - branch — ``egg/issue-N/slice-M`` — that the per-role work - branches rebase onto. Roots base off the pipeline branch - directly; child slices base off their parent slice's - integration branch. + Each slice has its own integration branch as a sibling of the + pipeline tip under ``egg//`` — ``egg/issue-N/slice-M`` — + that the per-role work branches rebase onto. Roots base off the + pipeline branch directly (``egg/issue-N/work``); child slices + base off their parent slice's integration branch. + + The pipeline tip is pushed to ``egg//work`` (#2399), so the + slice integration branch lives as a sibling of ``/work`` under + ``egg//`` — strip the trailing ``/work`` from the pipeline + branch to get the namespace root. The slice id is regex-validated for defense-in-depth (see ``get_worktree_branch``). """ issue = self.pipeline.issue_number or self.pipeline.id issue_branch = self.pipeline.branch or f"egg/issue-{issue}" + # Structural check (≥2 slashes, last segment ``work``) — see + # ``_slice_namespace_root`` in ``routes/pipelines.py``. + if issue_branch.count("/") >= 2 and issue_branch.rsplit("/", 1)[1] == "work": + issue_branch = issue_branch.rsplit("/", 1)[0] normalised_slice = slice_id if slice_id.startswith("slice-") else f"slice-{slice_id}" - import re - - if not re.fullmatch(r"slice-[0-9]+", normalised_slice): + if not SLICE_ID_PATTERN.fullmatch(normalised_slice): raise ValueError( f"slice_id={slice_id!r} does not match the canonical shape ``slice-``" ) diff --git a/orchestrator/kubernetes_spawner.py b/orchestrator/kubernetes_spawner.py index 3839d87e18..35ac9cde46 100644 --- a/orchestrator/kubernetes_spawner.py +++ b/orchestrator/kubernetes_spawner.py @@ -11,6 +11,7 @@ """ import os +import re import sys import threading import time @@ -93,6 +94,14 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] # /resolve. Blocking the key here is defense in depth; the base # spawner env below never sets it to begin with. "EGG_LIFECYCLE_SECRET", + # Slice scope (#2410, v2 review follow-up). The spawner is the + # single source of truth: ``EGG_SLICE_ID`` is derived from the + # ``slice_id`` parameter that already drives Job naming and + # worktree id. Protecting the key prevents a future caller from + # silently shipping a mismatched value via ``extra_env`` — + # without this, the agent's signals could land on a different + # slice than its Job/worktree, with no warning. + "EGG_SLICE_ID", } ) @@ -235,12 +244,14 @@ class KubernetesSpawner: DEFAULT_SANDBOX_IMAGE = os.environ.get("EGG_SANDBOX_IMAGE", "egg:latest") JOB_NAME_FORMAT = "egg-agent-{pipeline_id}-{role}" + JOB_NAME_FORMAT_SLICE = "egg-agent-{pipeline_id}-{slice_id}-{role}" @classmethod def _build_k8s_job_names( cls, pipeline_id: str, agent_role: AgentRole, + slice_id: str | None = None, ) -> tuple[str, str]: """Build the two identifiers an agent Job is known by. @@ -248,22 +259,54 @@ def _build_k8s_job_names( - ``job_name`` is the unprefixed identifier used as the gateway session ``container_id`` and in labels (e.g. - ``egg-agent-issue-1962-task-planner``). + ``egg-agent-issue-1962-task-planner`` or, for slice-scoped + spawns, ``egg-agent-issue-2261-v7-slice-2-coder``). - ``actual_k8s_job_name`` is the real k8s Job name after ``KubernetesClient`` prepends ``JOB_PREFIX`` during - ``create_container`` (e.g. - ``egg-sandbox-egg-agent-issue-1962-task-planner``). + ``create_container``. Underscores in ``agent_role.value`` (``task_planner``, ``reviewer_refine``, …) are converted to hyphens because k8s resource names are RFC-1123 labels and reject underscores. + + Slice scope (#2403): when ``slice_id`` is supplied, it is + embedded between the pipeline id and the role so concurrent + slices in the same pipeline don't collide on a single Job name + (which would cause ``spawn_agent_job``'s pre-spawn cleanup to + delete the in-flight sibling slice's Job — see line 405). """ - job_name = cls.JOB_NAME_FORMAT.format( - pipeline_id=pipeline_id, - role=agent_role.value.replace("_", "-"), - ) + if slice_id: + job_name = cls.JOB_NAME_FORMAT_SLICE.format( + pipeline_id=pipeline_id, + slice_id=slice_id, + role=agent_role.value.replace("_", "-"), + ) + else: + job_name = cls.JOB_NAME_FORMAT.format( + pipeline_id=pipeline_id, + role=agent_role.value.replace("_", "-"), + ) return job_name, f"{KubernetesClient.JOB_PREFIX}{job_name}" + @staticmethod + def _build_agent_worktree_id( + pipeline_id: str, + agent_role: AgentRole, + slice_id: str | None = None, + ) -> str: + """Build the per-agent worktree identifier. + + The id is the gateway worktree key (``container_id`` for + ``create_worktrees`` / ``delete_worktrees``) and the agent's + ``CONTAINER_ID`` env var. For slice-scoped spawns it embeds the + slice id (#2403) so concurrent slices don't share a worktree — + otherwise slice-N's coder would inherit slice-(N-1)'s worktree + contents (or step on them mid-flight). + """ + if slice_id: + return f"{pipeline_id}-{slice_id}-{agent_role.value}" + return f"{pipeline_id}-{agent_role.value}" + def __init__( self, k8s_client: KubernetesClient | None = None, @@ -288,11 +331,15 @@ def __init__( self._k8s = k8s_client self._gateway = gateway_client self._namespace = namespace - # Track restart counts per (pipeline_id, agent_role) pair - self._restart_counts: dict[tuple[str, str], int] = {} - # Per-(pipeline_id, agent_role) locks for serialising concurrent restarts. - # Protected by _restart_locks_lock (same pattern as state_store.py). - self._restart_locks: dict[tuple[str, str], threading.Lock] = {} + # Track restart counts per (pipeline_id, agent_role, slice_id) tuple. + # ``slice_id`` is ``None`` for pipeline-level agents and + # ``"slice-"`` for slice-scoped agents (#2410), so concurrent + # slices each get an independent budget. + self._restart_counts: dict[tuple[str, str, str | None], int] = {} + # Per-(pipeline_id, agent_role, slice_id) locks for serialising + # concurrent restarts. Protected by _restart_locks_lock (same + # pattern as state_store.py). + self._restart_locks: dict[tuple[str, str, str | None], threading.Lock] = {} self._restart_locks_lock = threading.Lock() @property @@ -321,8 +368,8 @@ def gateway(self) -> GatewayClient: self._gateway = get_gateway_client() return self._gateway - def _get_restart_lock(self, key: tuple[str, str]) -> threading.Lock: - """Get or create a per-(pipeline_id, agent_role) restart lock.""" + def _get_restart_lock(self, key: tuple[str, str, str | None]) -> threading.Lock: + """Get or create a per-(pipeline_id, agent_role, slice_id) restart lock.""" with self._restart_locks_lock: if key not in self._restart_locks: self._restart_locks[key] = threading.Lock() @@ -368,6 +415,7 @@ def spawn_agent_job( spawn_max_retries: int = DEFAULT_SPAWN_MAX_RETRIES, spawn_retry_initial_backoff_seconds: float = (DEFAULT_SPAWN_RETRY_INITIAL_BACKOFF_SECONDS), jira_ticket: str | None = None, + slice_id: str | None = None, ) -> SpawnedContainer: """Spawn a Kubernetes Job for an agent. @@ -398,7 +446,9 @@ def spawn_agent_job( Raises: KubernetesSpawnError: If spawning fails """ - job_name, actual_k8s_job_name = self._build_k8s_job_names(pipeline_id, agent_role) + job_name, actual_k8s_job_name = self._build_k8s_job_names( + pipeline_id, agent_role, slice_id=slice_id + ) # Clean up any existing Job with the same name. try: @@ -441,8 +491,16 @@ def spawn_agent_job( host_uid = int(os.environ.get("HOST_UID", 1000)) host_gid = int(os.environ.get("HOST_GID", 1000)) - # Per-agent worktree isolation: create a dedicated worktree - agent_worktree_id = f"{pipeline_id}-{agent_role.value}" + # Per-agent worktree isolation: create a dedicated worktree. + # Slice scope (#2403): concurrent slices in the same pipeline + # MUST get distinct worktree ids — otherwise slice-N's coder + # spawns onto slice-(N-1)'s already-mounted worktree (or + # races with it during cleanup). The id is also the agent's + # ``CONTAINER_ID`` env and the gateway worktree key, so the + # whole gateway / agent / orchestrator triangle agrees on it. + agent_worktree_id = self._build_agent_worktree_id( + pipeline_id, agent_role, slice_id=slice_id + ) worktree_created_this_call = False if repos: @@ -685,6 +743,23 @@ def spawn_agent_job( elif pipeline_id: environment["EGG_BRANCH"] = f"egg/{pipeline_id}/work" + # Slice scope (#2403, #2410): when this spawn is for a per-slice + # agent, propagate ``EGG_SLICE_ID`` so the agent's BRC handlers + # tag CONSENSUS_* signals with the slice and the orchestrator + # routes them to the per-slice tracker. Without this, the + # ``slice_id`` parameter only drove naming + worktree id and the + # restarted Job came up with no slice scope in its env — its + # signals would land on the pipeline-level tracker, which has + # no record of the agent (failure mode #3 from #2410). + # + # Single source of truth (v2 review follow-up): the spawner is + # the only writer; ``EGG_SLICE_ID`` is in ``_PROTECTED_ENV_KEYS`` + # so any ``extra_env`` value is logged and dropped, guaranteeing + # the env stays consistent with the Job name + worktree id that + # are also derived from this same ``slice_id`` parameter. + if slice_id is not None: + environment["EGG_SLICE_ID"] = slice_id + # Caller's extra_env overrides defaults, except protected keys if extra_env: for key, value in extra_env.items(): @@ -916,15 +991,17 @@ def cleanup_pipeline( worktree_ids_to_clean.add(f"{pipeline_id}-{role_label}") # Also scan filesystem for any per-agent worktrees. Only match - # entries that are either the pipeline-level worktree or a - # "{pipeline_id}-{role}" directory where {role} is a known - # AgentRole value. A naive `startswith(f"{pipeline_id}-")` + # entries that are either the pipeline-level worktree, a + # "{pipeline_id}-{role}" directory, or a slice-scoped + # "{pipeline_id}-slice-{N}-{role}" directory where {role} is a + # known AgentRole value (#2403). A naive `startswith(f"{pipeline_id}-")` # collides with longer pipeline IDs that share the prefix — e.g. # cleanup of `issue-1758` would match active worktrees of # `issue-1758-worktree-fix-tester`, wiping another pipeline's # state mid-phase (#1865). if WORKTREE_BASE_DIR.exists(): - valid_suffixes = {f"-{role.value}" for role in AgentRole} + valid_role_suffixes = {f"-{role.value}" for role in AgentRole} + slice_segment_re = re.compile(r"^-slice-[0-9]+(-.+)$") try: for entry in WORKTREE_BASE_DIR.iterdir(): if not entry.is_dir(): @@ -936,7 +1013,15 @@ def cleanup_pipeline( if not name.startswith(pipeline_id): continue suffix = name[len(pipeline_id) :] - if suffix in valid_suffixes: + if suffix in valid_role_suffixes: + worktree_ids_to_clean.add(name) + continue + # Slice-scoped: "{pipeline_id}-slice-{N}-{role}". + # The trailing "-{role}" inside the captured group + # is matched against the role allowlist so this + # branch can't sweep an unrelated sibling worktree. + slice_match = slice_segment_re.match(suffix) + if slice_match and slice_match.group(1) in valid_role_suffixes: worktree_ids_to_clean.add(name) except Exception as e: logger.warning( @@ -988,6 +1073,7 @@ def restart_agent_job( reason: str = "", spawn_max_retries: int = DEFAULT_SPAWN_MAX_RETRIES, spawn_retry_initial_backoff_seconds: float = (DEFAULT_SPAWN_RETRY_INITIAL_BACKOFF_SECONDS), + slice_id: str | None = None, ) -> SpawnedContainer: """Restart an agent Job: delete and respawn preserving worktree. @@ -1011,6 +1097,13 @@ def restart_agent_job( during worktree creation (forwarded to ``spawn_agent_job``). spawn_retry_initial_backoff_seconds: Initial backoff for spawn retries (forwarded to ``spawn_agent_job``). + slice_id: Optional slice scope (#2410). When supplied, the + slice-scoped Job name (``egg-agent-{pid}-{slice_id}-{role}``) + is the one deleted and respawned, the slice-scoped worktree + id is preserved, and ``EGG_SLICE_ID`` is propagated so the + restarted agent re-enters the per-slice consensus tracker. + The restart-budget key includes the slice scope so each + slice gets an independent budget. Returns: SpawnedContainer with new Job info. @@ -1022,7 +1115,11 @@ def restart_agent_job( if mode is None: raise ValueError("mode must be explicitly provided ('public' or 'private')") - restart_key = (pipeline_id, agent_role.value) + # Slice scope is part of the restart key so concurrent slice-N + # and slice-M agents of the same role each get an independent + # restart budget and lock. ``reset_restart_counts(pipeline_id)`` + # still clears all of them because it filters on ``k[0]``. + restart_key = (pipeline_id, agent_role.value, slice_id) lock = self._get_restart_lock(restart_key) # Timeout prevents indefinite blocking if a concurrent restart of the @@ -1050,7 +1147,12 @@ def restart_agent_job( # spawn time (hyphenated, no JOB_PREFIX); ``actual_k8s_job_name`` # is the real k8s Job name. Using the wrong form for either side # broke restart for every role with an underscore — see #2070. - job_name, actual_k8s_job_name = self._build_k8s_job_names(pipeline_id, agent_role) + # Slice scope (#2410) must be threaded through here so the + # delete + respawn target the slice-scoped Job name, not the + # pipeline-level one. + job_name, actual_k8s_job_name = self._build_k8s_job_names( + pipeline_id, agent_role, slice_id=slice_id + ) logger.info( "Restarting agent Job", @@ -1092,7 +1194,10 @@ def restart_agent_job( error=str(e), ) - # Respawn — gateway's create_worktrees() is idempotent + # Respawn — gateway's create_worktrees() is idempotent. + # ``slice_id`` is forwarded so spawn_agent_job builds the + # slice-scoped Job + worktree id and sets ``EGG_SLICE_ID`` + # on the new Job (#2410). spawned = self.spawn_agent_job( pipeline_id=pipeline_id, agent_role=agent_role, @@ -1111,6 +1216,7 @@ def restart_agent_job( preserve_worktree_on_failure=True, spawn_max_retries=spawn_max_retries, spawn_retry_initial_backoff_seconds=spawn_retry_initial_backoff_seconds, + slice_id=slice_id, ) logger.info( @@ -1125,17 +1231,26 @@ def restart_agent_job( finally: lock.release() - def get_restart_count(self, pipeline_id: str, agent_role: str) -> int: + def get_restart_count( + self, + pipeline_id: str, + agent_role: str, + slice_id: str | None = None, + ) -> int: """Get the current restart count for an agent. Args: pipeline_id: Pipeline ID. agent_role: Agent role value string. + slice_id: Optional slice scope (#2410). Pipeline-level callers + pass ``None``; slice-aware callers pass the same + ``slice-`` string they used at restart time so each + slice's budget is reported independently. Returns: Number of times the agent has been restarted. """ - key = (pipeline_id, agent_role) + key = (pipeline_id, agent_role, slice_id) lock = self._get_restart_lock(key) with lock: return self._restart_counts.get(key, 0) @@ -1162,19 +1277,33 @@ def detect_uncommitted_changes( self, pipeline_id: str, agent_role: str, + slice_id: str | None = None, ) -> dict | None: """Detect uncommitted changes in an agent's worktree after Job exit. Checks the agent's worktree directly on the filesystem for uncommitted changes. Per-agent worktrees are at: - /home/egg/.egg-worktrees/{pipeline_id}-{role}/{repo}/ + /home/egg/.egg-worktrees/{pipeline_id}-{role}/{repo}/ (pipeline-level) + /home/egg/.egg-worktrees/{pipeline_id}-{slice_id}-{role}/{repo}/ + (slice-scoped, #2410). + + Args: + pipeline_id: Pipeline ID. + agent_role: Agent role value string. + slice_id: Optional slice scope. When supplied, the slice-scoped + worktree id is inspected; pipeline-level callers omit this. Returns: Dict with change info if uncommitted changes found, None otherwise. """ import subprocess - agent_worktree_id = f"{pipeline_id}-{agent_role}" + # Mirrors ``_build_agent_worktree_id`` so a slice-scoped restart + # path can detect uncommitted work in the slice's worktree, not + # the (possibly absent) pipeline-level one. + agent_worktree_id = ( + f"{pipeline_id}-{slice_id}-{agent_role}" if slice_id else f"{pipeline_id}-{agent_role}" + ) worktree_base = WORKTREE_BASE_DIR / agent_worktree_id if not worktree_base.exists(): @@ -1213,6 +1342,7 @@ def detect_uncommitted_changes( event_type="agent_uncommitted_changes", pipeline_id=pipeline_id, agent_role=agent_role, + slice_id=slice_id, worktree_path=str(repo_dir), file_count=len(files), changed_files=files[:20], @@ -1220,6 +1350,7 @@ def detect_uncommitted_changes( return { "pipeline_id": pipeline_id, "agent_role": agent_role, + "slice_id": slice_id, "worktree_id": agent_worktree_id, "worktree_path": str(repo_dir), "file_count": len(files), @@ -1325,6 +1456,7 @@ def create_concurrent_spawn_fn( certs_volume: str | None = None, # noqa: ARG002 — Docker-era compat spawn_max_retries: int = DEFAULT_SPAWN_MAX_RETRIES, spawn_retry_initial_backoff_seconds: float = (DEFAULT_SPAWN_RETRY_INITIAL_BACKOFF_SECONDS), + slice_id: str | None = None, ): """Create a spawn callable compatible with ConcurrentPhaseExecutor. @@ -1341,6 +1473,12 @@ def create_concurrent_spawn_fn( sandbox_env: Base environment variables. image: Container image override. base_branch: Branch to base worktrees on. + slice_id: Optional slice scope (#2403). When supplied, every + spawn (including ``spawn_specific_roles`` retries) is + tagged with this slice so concurrent slices in the same + pipeline get distinct Job names and worktree ids. Without + this, slice-N spawning ``coder`` would delete slice-(N-1)'s + still-running ``coder`` Job during the pre-spawn cleanup. Returns: Callable suitable for ConcurrentPhaseExecutor.spawn_fn. @@ -1368,6 +1506,7 @@ def _spawn( command=command, spawn_max_retries=spawn_max_retries, spawn_retry_initial_backoff_seconds=(spawn_retry_initial_backoff_seconds), + slice_id=slice_id, ) return _spawn diff --git a/orchestrator/routes/contracts.py b/orchestrator/routes/contracts.py index 6bb944f344..f285e816a9 100644 --- a/orchestrator/routes/contracts.py +++ b/orchestrator/routes/contracts.py @@ -200,7 +200,12 @@ def _branch_read_contract( ) return None - branch = pipeline.branch or f"egg/{pipeline_id}" + # The pipeline tip is pushed to ``egg//work`` so slice integration + # branches can coexist as siblings — see + # :func:`routes.pipelines._ensure_pipeline_work_ref` for the rationale + # (#2399). The fallback shape mirrors the actual remote ref the + # contract was committed to. + branch = pipeline.branch or f"egg/{pipeline_id}/work" return contract_store.load_contract_from_branch(identifier, store.repo_path, branch) diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index aaaeeca1f1..8006c9b2e0 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -107,6 +107,7 @@ def get_repo_checks(repo: str) -> list[dict[str, str]]: # type: ignore[misc] PipelineStatus, ReviewVerdict, ) + from ..slice_id_validation import extract_slice_id from ..state_store import ( InvalidPipelineIdError, PipelineNotFoundError, @@ -153,6 +154,7 @@ def get_repo_checks(repo: str) -> list[dict[str, str]]: # type: ignore[misc] PipelineStatus, ReviewVerdict, ) + from slice_id_validation import extract_slice_id # type: ignore from state_store import ( # type: ignore InvalidPipelineIdError, PipelineNotFoundError, @@ -677,6 +679,73 @@ def _teardown_phase_overseer( TESTER_FINDINGS_HEADER = "### tester findings" +def _ensure_pipeline_work_ref(branch: str | None) -> str | None: + """Return the actual remote ref for an orchestrator-managed pipeline branch. + + The orchestrator pushes the pipeline tip to ``/work`` so the + ``/`` namespace can hold slice integration branches as + siblings (``/slice-N``) without git's ``directory file + conflict`` rejection — see #2399. A leaf ref at ```` and a + child at ``/slice-N`` cannot coexist on origin, so the + pipeline tip is moved one level deeper into the namespace. + + Idempotent and bounded to ``egg/``-shaped branches: + + * ``None`` → ``None`` (prompt-driven; the caller generates a + ``/work``-shaped branch later). + * ``egg/`` → ``egg//work`` (issue / CUSTOM submissions). + * ``egg//work`` → unchanged (resubmission, internal callers). + * non-``egg/`` (passed unchanged) — primarily babysit PR head refs; + the route-level caller already skips BABYSIT before reaching this + helper, so the only non-``egg/`` branch that lands here is a + CUSTOM-mode pipeline pointed at a foreign branch (e.g. + ``feature/foo``). CUSTOM-with-slices on a non-``egg/`` branch is + not a guaranteed-safe shape and is intentionally not normalised + here — the conflict would resurface at the slice push and is + tracked separately. + + The trailing-``/work`` check is structural rather than a plain + suffix match (``branch.count("/") >= 2 and branch.rsplit("/", 1)[1] + == "work"``) so a degenerate input like ``egg/work`` — a single + segment that *happens* to end in ``/work`` — gets normalised to + ``egg/work/work`` (siblings ``egg/work/slice-N``) rather than + treated as already-normalised. Trailing slashes are stripped first + so ``egg/`` does not collapse to a double-slash ``egg//work``. + """ + if branch is None: + return None + branch = branch.rstrip("/") + if not branch.startswith("egg/"): + return branch + # Structural check: only treat ``egg//work`` (≥2 slashes, last + # segment is ``work``) as already-normalised. ``egg/work`` looks + # like a suffix match but is a single-segment id and still needs the + # ``/work`` namespace deepening. + if branch.count("/") >= 2 and branch.rsplit("/", 1)[1] == "work": + return branch + return f"{branch}/work" + + +def _slice_namespace_root(pipeline_branch: str) -> str: + """Return the slice-integration-branch namespace root for a pipeline branch. + + Slice integration branches live as siblings of the pipeline tip + under ``egg//`` (see :func:`_ensure_pipeline_work_ref`). The + namespace root is the pipeline branch with the trailing ``/work`` + stripped — that's the prefix slice paths (``/slice-N``) are + built from. For legacy / non-normalised branches that do not end in + ``/work``, the branch itself is the root. + + The trailing-``/work`` check mirrors the structural check in + :func:`_ensure_pipeline_work_ref` (≥2 slashes, last segment is + ``work``) so a degenerate single-segment input like ``egg/work`` + is treated as the root itself rather than collapsing to ``egg``. + """ + if pipeline_branch.count("/") >= 2 and pipeline_branch.rsplit("/", 1)[1] == "work": + return pipeline_branch.rsplit("/", 1)[0] + return pipeline_branch + + def _pipeline_identifier( issue_number: int | None, pipeline_id: str, @@ -1376,6 +1445,15 @@ def create_pipeline() -> tuple[Response, int]: ): return make_error_response("Missing branch") + # #2399 — push the pipeline tip to ``/work`` so slice + # integration branches at ``/slice-N`` can coexist as + # siblings under the same namespace (git rejects a leaf ref and + # children of that ref's path with ``directory file conflict``). + # Skipped for BABYSIT (the branch is an existing PR head we don't + # own) and for non-``egg/`` branches. + if mode != PipelineMode.BABYSIT: + branch = _ensure_pipeline_work_ref(branch) + # Wait for the gateway to be ready before any gateway-dependent work. # On fresh deploys / pod restarts the orchestrator can accept requests # while the gateway HTTP listener is still coming up; without this gate @@ -2056,9 +2134,12 @@ def _cleanup_remote_branches( """Best-effort cleanup of remote branches for a pipeline. Deletes the pipeline's shared branch (``pipeline.branch``, typically - ``egg/{pipeline_id}``) and every per-container worktree branch - (``egg/{container_id}/work``). Failures are logged as warnings and do - not block pipeline deletion. + ``egg/{pipeline_id}/work`` since #2399) and every per-container + worktree branch (``egg/{container_id}/work``). Slice integration + branches at ``egg/{pipeline_id}/slice-N`` are siblings of the + pipeline tip and are NOT deleted here — see follow-up tracking on + #2399 for full namespace cleanup. Failures are logged as warnings + and do not block pipeline deletion. """ branches: set[str] = set() if pipeline.branch: @@ -2191,9 +2272,17 @@ def restart_agent(pipeline_id: str, agent_role: str) -> tuple[Response, int]: pipeline_id: Pipeline ID agent_role: Agent role to restart (e.g. "coder", "tester") + Query string (optional): + slice_id: Slice scope (``slice-``). When supplied, the + slice-scoped Job and worktree are restarted, ``EGG_SLICE_ID`` + is propagated to the new Job, and consensus reset targets + the per-slice tracker. Pipeline-level agents omit it. + ``slice_id`` may also be supplied via the JSON body. + Request body (optional): { - "reason": "Human-readable reason for the restart" + "reason": "Human-readable reason for the restart", + "slice_id": "slice-2" } Response: @@ -2238,6 +2327,16 @@ def restart_agent(pipeline_id: str, agent_role: str) -> tuple[Response, int]: body = request.get_json(silent=True) or {} reason = body.get("reason", "Manual restart via API") + # Slice scope (#2410): query param wins over body so the URL + # form is unambiguous; both forms validate against the canonical + # ``slice-`` shape via ``extract_slice_id``. + raw_slice_id = request.args.get("slice_id") + slice_payload = {"slice_id": raw_slice_id} if raw_slice_id is not None else body + try: + slice_id = extract_slice_id(slice_payload) + except ValueError as e: + return make_error_response(str(e), status_code=400) + # Restart the container via spawner spawner = _get_spawner() @@ -2344,6 +2443,7 @@ def restart_agent(pipeline_id: str, agent_role: str) -> tuple[Response, int]: reason=reason, spawn_max_retries=pipeline.config.spawn_max_retries, spawn_retry_initial_backoff_seconds=pipeline.config.spawn_retry_initial_backoff_seconds, + slice_id=slice_id, ) except (ContainerSpawnError, KubernetesSpawnError) as e: # Revert early status update — the agent is not actually running. @@ -2363,6 +2463,8 @@ def restart_agent(pipeline_id: str, agent_role: str) -> tuple[Response, int]: # Spawn succeeded — now reset consensus state for this agent. # If consensus reset fails, log a warning but don't fail the restart: # the restarted agent will re-enter consensus on its own. + # Slice-scoped restarts (#2410) target the per-slice tracker; the + # pipeline-level tracker has no record of the slice agent. try: try: from peer_consensus import get_peer_consensus_tracker @@ -2371,13 +2473,14 @@ def restart_agent(pipeline_id: str, agent_role: str) -> tuple[Response, int]: get_peer_consensus_tracker, # type: ignore[import-not-found] ) - tracker = get_peer_consensus_tracker(pipeline_id) + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if tracker: tracker.remove_agent(agent_role) logger.info( "Reset consensus state for agent", pipeline_id=pipeline_id, agent_role=agent_role, + slice_id=slice_id, ) except ImportError: pass @@ -2386,6 +2489,7 @@ def restart_agent(pipeline_id: str, agent_role: str) -> tuple[Response, int]: "Failed to reset consensus state (agent will re-enter consensus)", pipeline_id=pipeline_id, agent_role=agent_role, + slice_id=slice_id, error=str(e), ) @@ -2475,7 +2579,13 @@ def restart_agent(pipeline_id: str, agent_role: str) -> tuple[Response, int]: pipeline.updated_at = datetime.now(UTC) store.update_pipeline(pipeline_id, pipeline.model_dump(mode="json")) - restart_count = spawner.get_restart_count(pipeline_id, agent_role) + # Slice-scoped restarts (#2410) bumped the per-slice budget bucket + # ``(pipeline_id, agent_role, slice_id)``; the pipeline-level + # ``(pipeline_id, agent_role, None)`` bucket is untouched. Reading + # without ``slice_id`` here would return the pipeline-level count + # (typically zero) and the audit log + JSON response below would + # misreport the operator's "you've burned N of M restarts" telemetry. + restart_count = spawner.get_restart_count(pipeline_id, agent_role, slice_id=slice_id) logger.info( "Agent restarted", @@ -5248,10 +5358,13 @@ def _sync_worktree_with_remote( worktree so that all downstream code (contract loading, draft reading, populator, etc.) sees the full pipeline state. - ``pipeline_branch`` is the **remote** branch name to reconcile against - (e.g. ``egg/``). The orchestrator-side worktree is checked out - on a ``/work``-suffixed local branch (``egg//work``) that does - not exist on origin — agents push to ``egg/``. Without an + ``pipeline_branch`` is the **remote** branch name to reconcile against. + Since #2399 the pipeline tip lives at ``egg//work`` on origin so + slice integration branches at ``egg//slice-N`` can coexist as + siblings; ``pipeline.branch`` already carries that ``/work`` suffix + (set by :func:`_ensure_pipeline_work_ref` at submission time), so + callers should pass ``pipeline_branch=pipeline.branch`` directly — + the local worktree branch and the remote ref now match. Without an explicit ``pipeline_branch``, the function reads ``git branch --show-current`` and looks up ``origin/``, which always misses on real pipelines and exits at @@ -10957,16 +11070,20 @@ def _run_implement_phase_slices( return 1, "no slices in contract" pipeline_branch = pipeline.branch or ( - f"egg/issue-{pipeline.issue_number}" + f"egg/issue-{pipeline.issue_number}/work" if pipeline.issue_number is not None else f"egg/{pipeline_id}/work" ) issue_number = pipeline.issue_number - # 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 + # Slice integration branches stack as siblings of the pipeline tip + # under ``egg//`` (see :func:`_ensure_pipeline_work_ref` for the + # ``/work`` namespace decision in #2399). The namespace root drops the + # trailing ``/work`` so slice paths build to ``/slice-M`` rather + # than ``/work/slice-M``. The qualifier suffix (``-v3``, + # ``-backend``) is preserved through ``pipeline.branch`` so two + # qualified pipelines for the same issue do not collide on + # ``egg/issue-N/slice-M`` (#2368). + issue_branch = _slice_namespace_root(pipeline_branch) # Wrap scheduler construction so the run loop doesn't crash if the # contract bypassed plan-ingestion validation and reaches the @@ -11510,28 +11627,41 @@ def _run_concurrent_phase( phase_str = phase if isinstance(phase, str) else phase.value pipeline_mode = "issue" if pipeline.issue_number is not None else "prompt" - # Slice-aware sandbox env (#2137 TASK-4-3): when running a per-slice - # team, override EGG_PIPELINE_ID with the nested ``{pipeline_id}/{slice_id}`` - # form so agent CLIs send CONSENSUS_* messages keyed on the slice's - # tracker scope. The bare pipeline_id is preserved on the caller's - # ``sandbox_env`` so we mutate a shallow copy here. + # Slice-aware sandbox env (#2137 TASK-4-3 / #2403): when running a + # per-slice team, the spawner exposes the slice id via + # ``EGG_SLICE_ID`` and leaves ``EGG_PIPELINE_ID`` as the bare + # pipeline id. An earlier shape encoded the slice into + # ``EGG_PIPELINE_ID`` itself (``{pipeline_id}/{slice_id}``) so the + # orchestrator's ``_tracker_key`` would route CONSENSUS_* to the + # slice tracker without an extra signal-level field. That broke + # every agent → orchestrator round-trip: # - # Trade-off (refine-phase decision-14 hybrid is partially honoured): - # the agent CLI uses the same env var for *every* outbound signal — - # CONSENSUS_*, HEARTBEAT, OVERSEER_ALERT — so HEARTBEAT and - # OVERSEER_ALERT also route to the slice-scoped tracker rather than - # the pipeline-scoped tracker. CONSENSUS_* isolation works as - # intended; cross-slice telemetry is per-slice today. A pipeline- - # level fan-out for OVERSEER_ALERT requires a CLI-side message- - # type-aware router and is tracked alongside the per-slice MCP - # control verbs in #2199. The orchestrator-side log line in - # ``_run_implement_phase_slices`` and the gateway broadcast on - # cascade provide the always-on fallback so a deadlocked - # downstream subtree is still surfaced to the operator. - if slice_id is not None: - sandbox_env = dict(sandbox_env) - sandbox_env["EGG_PIPELINE_ID"] = f"{pipeline_id}/{slice_id}" - sandbox_env["EGG_SLICE_ID"] = slice_id + # * the orchestrator-side ``PIPELINE_ID_PATTERN`` and the agent + # handler validator (``[a-zA-Z0-9_-]+``) both reject the slash, + # * Flask's default URL converter doesn't allow ``/``, so every + # ``POST /api/v1/pipelines/{pid}/...`` route 404s — i.e. all + # of progress, BRC, heartbeat, message, phase, decision, etc. + # + # Slice routing is plumbed explicitly instead: the BRC handlers + # pull ``EGG_SLICE_ID`` and forward it on the signal payload, and + # the orchestrator's signal handlers feed it into + # ``get_peer_consensus_tracker(pipeline_id, slice_id)``. CONSENSUS_* + # isolation is preserved; HEARTBEAT and OVERSEER_ALERT are not + # tracker-scoped at all — ``handle_heartbeat_signal`` is a no-op + # ACK with no tracker lookup, and OVERSEER_ALERT flows through the + # message bus (``MessageType.OVERSEER_ALERT``) rather than the + # consensus tracker. So per-slice scoping doesn't apply to either, + # and operator telemetry stays pipeline-wide as before. The + # pipeline-level fan-out for OVERSEER_ALERT mentioned in earlier + # comments here is tracked alongside the per-slice MCP control + # verbs in #2199. + # + # Single source of truth (#2410 v2 review): ``EGG_SLICE_ID`` is + # injected by ``KubernetesSpawner.spawn_agent_job`` from the same + # ``slice_id`` parameter that drives Job naming and worktree id, so + # there is no need to also stuff it into ``sandbox_env`` here. The + # key is in ``_PROTECTED_ENV_KEYS`` so any future caller that does + # supply a value via ``extra_env`` is logged and overridden. # Build per-role prompts for concurrent phase execution. from egg_contracts.agent_roles import get_roles_for_phase as _get_roles_for_phase @@ -11619,6 +11749,7 @@ def _run_concurrent_phase( base_branch=pipeline.base_branch, spawn_max_retries=pipeline.config.spawn_max_retries, spawn_retry_initial_backoff_seconds=pipeline.config.spawn_retry_initial_backoff_seconds, + slice_id=slice_id, ) max_concurrent = getattr(pipeline.config, "max_concurrent_agents", 6) diff --git a/orchestrator/routes/signals.py b/orchestrator/routes/signals.py index 3bf442d4f6..4ca6a85edb 100644 --- a/orchestrator/routes/signals.py +++ b/orchestrator/routes/signals.py @@ -39,6 +39,19 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] from egg_contracts.orchestrator import create_orchestrator from handoffs import AgentOutput, save_agent_output from models import AgentExecutionStatus, AgentRole, Pipeline, PipelineStatus + +# Slice-aware consensus routing (#2403): per-slice agents tag their +# signals with a ``slice_id`` field on the request body so the +# orchestrator routes each ``CONSENSUS_*`` to the per-slice tracker +# (``peer_consensus._tracker_key`` composes ``{pipeline_id}/{slice_id}``). +# The canonical extractor lives in ``slice_id_validation`` so the +# operator-triggered restart route (#2410) and the gateway-bound branch +# builders in ``concurrent_executor`` validate against the same shape. +# The alias below preserves the existing private name for the many +# handler call sites in this file. +from slice_id_validation import ( + extract_slice_id as _extract_slice_id, +) from state_store import ( InvalidPipelineIdError, PipelineNotFoundError, @@ -1013,14 +1026,20 @@ def handle_consensus_propose_signal( if summary_error: return make_error_response(summary_error, 400) + try: + slice_id = _extract_slice_id(data) + except ValueError as exc: + return make_error_response(str(exc), 400) + try: from peer_consensus import get_peer_consensus_tracker except ImportError: from ..peer_consensus import get_peer_consensus_tracker # type: ignore[no-redef] - tracker = get_peer_consensus_tracker(pipeline_id) + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if not tracker: - return make_error_response(f"No consensus tracker for pipeline {pipeline_id}", 404) + scope = f"{pipeline_id}/{slice_id}" if slice_id else pipeline_id + return make_error_response(f"No consensus tracker for pipeline {scope}", 404) # Verify commit SHA exists on the expected branch before accepting # the proposal (#1473). Reuses _verify_commit_on_branch() from the @@ -1203,14 +1222,20 @@ def handle_consensus_ack_signal( 400, ) + try: + slice_id = _extract_slice_id(data) + except ValueError as exc: + return make_error_response(str(exc), 400) + try: from peer_consensus import get_peer_consensus_tracker except ImportError: from ..peer_consensus import get_peer_consensus_tracker # type: ignore[no-redef] - tracker = get_peer_consensus_tracker(pipeline_id) + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if not tracker: - return make_error_response(f"No consensus tracker for pipeline {pipeline_id}", 404) + scope = f"{pipeline_id}/{slice_id}" if slice_id else pipeline_id + return make_error_response(f"No consensus tracker for pipeline {scope}", 404) try: try: @@ -1287,14 +1312,20 @@ def handle_consensus_nack_signal( if reason_error: return make_error_response(reason_error, 400) + try: + slice_id = _extract_slice_id(data) + except ValueError as exc: + return make_error_response(str(exc), 400) + try: from peer_consensus import get_peer_consensus_tracker except ImportError: from ..peer_consensus import get_peer_consensus_tracker # type: ignore[no-redef] - tracker = get_peer_consensus_tracker(pipeline_id) + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if not tracker: - return make_error_response(f"No consensus tracker for pipeline {pipeline_id}", 404) + scope = f"{pipeline_id}/{slice_id}" if slice_id else pipeline_id + return make_error_response(f"No consensus tracker for pipeline {scope}", 404) try: try: @@ -1353,14 +1384,20 @@ def handle_consensus_withdraw_signal( if reason_error: return make_error_response(reason_error, 400) + try: + slice_id = _extract_slice_id(data) + except ValueError as exc: + return make_error_response(str(exc), 400) + try: from peer_consensus import get_peer_consensus_tracker except ImportError: from ..peer_consensus import get_peer_consensus_tracker # type: ignore[no-redef] - tracker = get_peer_consensus_tracker(pipeline_id) + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if not tracker: - return make_error_response(f"No consensus tracker for pipeline {pipeline_id}", 404) + scope = f"{pipeline_id}/{slice_id}" if slice_id else pipeline_id + return make_error_response(f"No consensus tracker for pipeline {scope}", 404) try: result = tracker.handle_withdraw(agent_role, reason) @@ -1478,19 +1515,31 @@ def handle_consensus_confirmed_signal( if not agent_role: return make_error_response("Missing agent_role") + try: + slice_id = _extract_slice_id(data) + except ValueError as exc: + return make_error_response(str(exc), 400) + try: from peer_consensus import get_peer_consensus_tracker except ImportError: from ..peer_consensus import get_peer_consensus_tracker # type: ignore[no-redef] - tracker = get_peer_consensus_tracker(pipeline_id) + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if not tracker: # Defaults must be outside the try block so the message-bus fallback # (second try block) can reference them even if reconstruction fails. _phase = "implement" _repo = None - # Attempt reconstruction from message store before returning 404 + # Attempt reconstruction from message store before returning 404. + # Slice-scoped trackers are NOT reconstructed today: the message + # store keys messages by bare pipeline_id only, so a per-slice + # replay would mingle other slices' messages and reach false + # consensus. Tracked in #2409 (orchestrator-restart recovery for + # slice-scoped trackers; needs a slice_id field on Message and a + # filtered replay). For pipeline-level (slice_id is None) requests + # the existing replay path is unchanged. try: from peer_consensus import reconstruct_tracker_from_messages from review_graph import get_review_graph_for_phase @@ -1503,12 +1552,14 @@ def handle_consensus_confirmed_signal( except StateStoreError: pass - graph = get_review_graph_for_phase(_phase, repo=_repo) - tracker = reconstruct_tracker_from_messages(pipeline_id, graph) + if slice_id is None: + graph = get_review_graph_for_phase(_phase, repo=_repo) + tracker = reconstruct_tracker_from_messages(pipeline_id, graph) except Exception as recon_err: logger.warning( "Tracker reconstruction failed in confirmed handler", pipeline_id=pipeline_id, + slice_id=slice_id, error=str(recon_err), ) @@ -1576,7 +1627,8 @@ def handle_consensus_confirmed_signal( error=str(fallback_err), ) - return make_error_response(f"No consensus tracker for pipeline {pipeline_id}", 404) + scope = f"{pipeline_id}/{slice_id}" if slice_id else pipeline_id + return make_error_response(f"No consensus tracker for pipeline {scope}", 404) try: result = tracker.handle_confirmed(agent_role) @@ -1729,14 +1781,20 @@ def handle_consensus_excuse_producer_signal( status_code=404, ) + try: + slice_id = _extract_slice_id(data) + except ValueError as exc: + return make_error_response(str(exc), 400) + try: from peer_consensus import get_peer_consensus_tracker except ImportError: from ..peer_consensus import get_peer_consensus_tracker # type: ignore[no-redef] - tracker = get_peer_consensus_tracker(pipeline_id) + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if not tracker: - return make_error_response(f"No consensus tracker for pipeline {pipeline_id}", 404) + scope = f"{pipeline_id}/{slice_id}" if slice_id else pipeline_id + return make_error_response(f"No consensus tracker for pipeline {scope}", 404) try: result = tracker.excuse_producer(producer_role, reason) @@ -1831,14 +1889,20 @@ def handle_consensus_resolve_obligation_signal( if note_error: return make_error_response(note_error, 400) + try: + slice_id = _extract_slice_id(data) + except ValueError as exc: + return make_error_response(str(exc), 400) + try: from peer_consensus import get_peer_consensus_tracker except ImportError: from ..peer_consensus import get_peer_consensus_tracker # type: ignore[no-redef] - tracker = get_peer_consensus_tracker(pipeline_id) + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if not tracker: - return make_error_response(f"No consensus tracker for pipeline {pipeline_id}", 404) + scope = f"{pipeline_id}/{slice_id}" if slice_id else pipeline_id + return make_error_response(f"No consensus tracker for pipeline {scope}", 404) try: result = tracker.handle_resolve_obligation( @@ -1925,14 +1989,20 @@ def handle_consensus_producer_push_signal( changed_files = data.get("changed_files") + try: + slice_id = _extract_slice_id(data) + except ValueError as exc: + return make_error_response(str(exc), 400) + try: from peer_consensus import get_peer_consensus_tracker except ImportError: from ..peer_consensus import get_peer_consensus_tracker # type: ignore[no-redef] - tracker = get_peer_consensus_tracker(pipeline_id) + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if not tracker: - return make_error_response(f"No consensus tracker for pipeline {pipeline_id}", 404) + scope = f"{pipeline_id}/{slice_id}" if slice_id else pipeline_id + return make_error_response(f"No consensus tracker for pipeline {scope}", 404) try: result = tracker.handle_producer_push(agent_role, commit_sha, changed_files) diff --git a/orchestrator/slice_id_validation.py b/orchestrator/slice_id_validation.py new file mode 100644 index 0000000000..1e09c00b6d --- /dev/null +++ b/orchestrator/slice_id_validation.py @@ -0,0 +1,61 @@ +"""Canonical ``slice_id`` shape and request-payload extractor. + +The orchestrator pins slice ids to the canonical ``slice-`` shape +at every gateway-facing seam where an external caller could supply +one — signal handlers (#2403), the operator-triggered restart route +(#2410), and the gateway-bound branch builders in +``concurrent_executor``. The pattern lives here, in a small shared +module, so each seam validates against the same regex rather than +re-deriving it inline. + +The contract-side ``Slice.id`` field +(``shared/egg_contracts/models.py``) accepts the broader pattern +``^(?:slice|phase)-[0-9]+$`` for backward compatibility with +pre-#2137 contracts. The pattern below is intentionally narrower — +only the canonical ``slice-`` shape. The canonicalisation is +performed by the parent ``Contract`` model validator +``Contract._migrate_phases_to_slices`` (mode="wrap" — runs before +per-Slice field validation), which rewrites legacy ``phase-`` ids +to ``slice-`` whenever a contract is loaded via +``Contract.model_validate(json_dict)``. Slices reaching the spawn / +signal / restart path through the contract-loader path are therefore +already canonical. The guarantee does NOT extend to direct ``Slice`` +construction (e.g. ``Slice(id="phase-2", ...)``) — pydantic field +validation alone is permissive — so the regex below is doing real +work for any code path that constructs Slices outside Contract +loading (e.g. a future migration tool, a hand-built fixture, or +direct ``model_validate`` on a Slice dict). Such a caller's +``phase-`` slice will be rejected here as it should — the +registry key MUST be canonical so the per-slice tracker can be +looked up, and Job names / worktree ids that embed it MUST be +RFC-1123 safe. +""" + +from __future__ import annotations + +import re +from typing import Any + +SLICE_ID_PATTERN = re.compile(r"^slice-[0-9]+$") + + +def extract_slice_id(data: dict[str, Any]) -> str | None: + """Return ``slice_id`` from a request payload, validated. + + Returns ``None`` when no slice scope was supplied (the agent is + pipeline-level, not slice-scoped). Raises ``ValueError`` when the + caller supplied a non-empty value that does not match the + canonical ``slice-`` shape. + + Defense-in-depth: validate at every gateway-facing seam even + though the spawn-side is already canonical — a future caller + that forgets the upstream regex must not be able to smuggle path + separators or shell metacharacters into a tracker registry key, + a Job name, or a worktree id. + """ + raw = data.get("slice_id") + if raw is None or raw == "": + return None + if not isinstance(raw, str) or not SLICE_ID_PATTERN.fullmatch(raw): + raise ValueError(f"Invalid slice_id {raw!r}: must match 'slice-'") + return raw diff --git a/orchestrator/stacked_pr_reconciler.py b/orchestrator/stacked_pr_reconciler.py index c6ee7b798d..310a63e9ed 100644 --- a/orchestrator/stacked_pr_reconciler.py +++ b/orchestrator/stacked_pr_reconciler.py @@ -82,7 +82,8 @@ def _resolve_extant_new_base( slice_, slices_by_id, extant_branches: set[str], - issue_branch: str, + slice_namespace_root: str, + pipeline_branch: str, ) -> str: """Walk up the slice DAG until an extant branch is found. @@ -95,6 +96,12 @@ def _resolve_extant_new_base( whose branch is still on origin. If the entire chain has been deleted, fall back to the pipeline branch — root-targeted branches are stable. + + ``slice_namespace_root`` is the prefix slice paths are built from + (``egg/``, no ``/work`` suffix); ``pipeline_branch`` is the + actual remote ref of the umbrella pipeline tip (``egg//work``, + after #2399). They differ by exactly the ``/work`` suffix — see + :func:`routes.pipelines._ensure_pipeline_work_ref`. """ # ``dependencies[0]`` is the canonical parent under the forest # constraint enforced at plan ingestion. ``serialized_chain_order`` @@ -106,17 +113,17 @@ def _resolve_extant_new_base( parent_slice = slices_by_id.get(parent_id) if parent_slice is None: break - candidate = f"{issue_branch}/{parent_slice.id}" + candidate = f"{slice_namespace_root}/{parent_slice.id}" if candidate in extant_branches: return candidate # This ancestor's branch is also gone (cascading merge). # Walk one more level up. parent_id = parent_slice.dependencies[0] if parent_slice.dependencies else None # Either the slice has no dependencies (it's a root whose own - # PR shouldn't get here — roots target ``issue_branch`` directly, + # PR shouldn't get here — roots target ``pipeline_branch`` directly, # which is in ``extant_branches``), or every ancestor's branch # has been deleted. Either way, the pipeline branch is safe. - return issue_branch + return pipeline_branch @dataclass(frozen=True) @@ -181,20 +188,27 @@ def find_orphaned_child_prs( # (``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 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}" + # propagation), so deriving the slice namespace root from + # ``contract_key`` keeps the reconciler's lookup shape aligned + # with the producer's 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. + # + # Slice integration branches live as siblings of the pipeline tip + # under ``egg//`` (#2399); the umbrella pipeline tip is at + # ``egg//work``. ``slice_namespace_root`` is the prefix slice + # paths are built from; ``pipeline_branch`` is the actual remote + # ref of the umbrella tip — used as the cascade-fallback base. + slice_namespace_root = f"egg/{contract.contract_key}" + pipeline_branch = f"{slice_namespace_root}/work" slices_by_id = {s.id: s for s in contract.slices} for slice_ in contract.slices: parent = slice_.parent_branch_at_creation if parent is None: continue # not yet provisioned - slice_branch = f"{issue_branch}/{slice_.id}" + slice_branch = f"{slice_namespace_root}/{slice_.id}" pr = pr_by_head.get(slice_branch) if pr is None: continue # slice's PR hasn't been opened yet (or is closed) @@ -216,7 +230,13 @@ def find_orphaned_child_prs( # cascade is the primary trigger for orphan detection, and # in that case ``parent_branch_at_creation`` points at the # same just-deleted branch we're trying to escape from. - new_base = _resolve_extant_new_base(slice_, slices_by_id, extant_branches, issue_branch) + new_base = _resolve_extant_new_base( + slice_, + slices_by_id, + extant_branches, + slice_namespace_root, + pipeline_branch, + ) orphans.append( OrphanedChildPR( slice_id=slice_.id, diff --git a/orchestrator/tests/test_contracts_routes.py b/orchestrator/tests/test_contracts_routes.py index 4446b943b4..f6cc0cb87e 100644 --- a/orchestrator/tests/test_contracts_routes.py +++ b/orchestrator/tests/test_contracts_routes.py @@ -373,12 +373,13 @@ def test_get_uses_derived_branch_when_pipeline_branch_is_none( assert response.status_code == 200, response.data body = json.loads(response.data) assert body["source"] == "branch" - # With branch=None the code should derive "egg/" - # and try origin/egg/ as the preferred ref. + # With branch=None the code should derive "egg//work" + # (the /work-suffixed shape from #2399) and try + # origin/egg//work as the preferred ref. assert run_mock.call_args_list[0].args[0] == [ "git", "show", - f"origin/egg/{self.PIPELINE_ID}:.egg-state/contracts/{self.PIPELINE_ID}.json", + f"origin/egg/{self.PIPELINE_ID}/work:.egg-state/contracts/{self.PIPELINE_ID}.json", ] assert run_mock.call_args_list[0].kwargs["cwd"] == store.repo_path diff --git a/orchestrator/tests/test_kubernetes_spawner.py b/orchestrator/tests/test_kubernetes_spawner.py index f86c2f77db..b3203b8fc3 100644 --- a/orchestrator/tests/test_kubernetes_spawner.py +++ b/orchestrator/tests/test_kubernetes_spawner.py @@ -328,6 +328,79 @@ def test_spawn_extra_env_overrides(self, spawner, mock_k8s_client): assert result.environment["EGG_AGENT_ROLE"] == "custom" assert result.environment["MY_KEY"] == "val" + def test_spawn_with_slice_id_sets_egg_slice_id_env( + self, spawner, mock_k8s_client, mock_gateway + ): + """``slice_id=`` parameter propagates into ``EGG_SLICE_ID`` (#2410). + + The spawner's ``slice_id`` parameter previously only drove the Job + name and worktree id; the agent container had no slice scope in + its environment, so its BRC handlers couldn't tag CONSENSUS_* + signals with the slice (failure mode #3 from #2410). + """ + result = spawner.spawn_agent_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + slice_id="slice-2", + ) + assert result.environment.get("EGG_SLICE_ID") == "slice-2" + create_kwargs = mock_k8s_client.create_container.call_args.kwargs + assert create_kwargs["environment"].get("EGG_SLICE_ID") == "slice-2" + + def test_spawn_without_slice_id_does_not_set_egg_slice_id( + self, spawner, mock_k8s_client, mock_gateway + ): + """Pipeline-level spawns leave ``EGG_SLICE_ID`` unset.""" + result = spawner.spawn_agent_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + ) + assert "EGG_SLICE_ID" not in result.environment + + def test_extra_env_cannot_override_egg_slice_id(self, spawner, mock_k8s_client, mock_gateway): + """``extra_env`` cannot override ``EGG_SLICE_ID`` — protected key (#2410 v2 review). + + The spawner is the single source of truth: ``EGG_SLICE_ID`` is + derived from the ``slice_id`` parameter that already drives Job + naming and worktree id. A future caller that tried to ship a + different value via ``extra_env`` would silently end up with the + agent's signals tagged for one slice while its Job + worktree + belong to another. Protecting the key catches that. + """ + result = spawner.spawn_agent_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + slice_id="slice-2", + extra_env={"EGG_SLICE_ID": "slice-99"}, + ) + # Spawner's value wins, not extra_env's. + assert result.environment.get("EGG_SLICE_ID") == "slice-2" + create_kwargs = mock_k8s_client.create_container.call_args.kwargs + assert create_kwargs["environment"].get("EGG_SLICE_ID") == "slice-2" + + def test_extra_env_cannot_inject_egg_slice_id_when_pipeline_level( + self, spawner, mock_k8s_client, mock_gateway + ): + """Without ``slice_id``, ``extra_env`` cannot smuggle ``EGG_SLICE_ID`` in. + + Pipeline-level spawns must not be tagged with a slice scope — + protecting the key blocks a regression where a slice-aware + caller would forget the ``slice_id`` parameter and try to bolt + the env var on directly via ``extra_env``. + """ + result = spawner.spawn_agent_job( + pipeline_id="pipe-1", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + extra_env={"EGG_SLICE_ID": "slice-2"}, + ) + assert "EGG_SLICE_ID" not in result.environment + create_kwargs = mock_k8s_client.create_container.call_args.kwargs + assert "EGG_SLICE_ID" not in create_kwargs["environment"] + def test_spawn_labels(self, spawner, mock_k8s_client): """Spawn sets the expected labels on the Job.""" spawner.spawn_agent_job( @@ -902,7 +975,7 @@ def test_restart_limit_exceeded(self, spawner): """Restart raises when limit is exceeded.""" from kubernetes_spawner import KubernetesSpawnError - spawner._restart_counts[("pipe-1", "coder")] = 2 + spawner._restart_counts[("pipe-1", "coder", None)] = 2 with pytest.raises(KubernetesSpawnError, match="Restart limit.*exceeded"): spawner.restart_agent_job( pipeline_id="pipe-1", @@ -991,9 +1064,9 @@ def test_get_restart_count_default(self, spawner): def test_reset_restart_counts(self, spawner): """reset_restart_counts clears all counts for a pipeline.""" - spawner._restart_counts[("pipe-1", "coder")] = 3 - spawner._restart_counts[("pipe-1", "tester")] = 1 - spawner._restart_counts[("pipe-2", "coder")] = 2 + spawner._restart_counts[("pipe-1", "coder", None)] = 3 + spawner._restart_counts[("pipe-1", "tester", None)] = 1 + spawner._restart_counts[("pipe-2", "coder", None)] = 2 spawner.reset_restart_counts("pipe-1") @@ -1159,3 +1232,349 @@ def test_singleton_reuses_instance(self): assert first is second kubernetes_spawner._spawner = None + + +# --------------------------------------------------------------------------- +# Slice-scope plumbing (#2403) +# --------------------------------------------------------------------------- + + +class TestSliceScopedJobAndWorktreeIds: + """Concurrent slices in the same pipeline must spawn under distinct ids. + + Without slice scope, slice-N's coder spawn: + * builds the same Job name as slice-(N-1)'s coder, so the + pre-spawn cleanup at the top of ``spawn_agent_job`` deletes the + sibling slice's still-running Job; + * builds the same ``agent_worktree_id`` so the gateway worktree + is reused, mounting slice-(N-1)'s contents (or stepping on + them mid-flight). + Both bugs surfaced together in #2403. + """ + + def test_build_k8s_job_names_includes_slice_segment(self): + from kubernetes_spawner import KubernetesSpawner + + job_name, k8s_name = KubernetesSpawner._build_k8s_job_names( + "issue-2261-v7", AgentRole.CODER, slice_id="slice-2" + ) + assert job_name == "egg-agent-issue-2261-v7-slice-2-coder" + assert k8s_name.endswith("egg-agent-issue-2261-v7-slice-2-coder") + + def test_build_k8s_job_names_omits_slice_segment_when_unscoped(self): + from kubernetes_spawner import KubernetesSpawner + + job_name, _ = KubernetesSpawner._build_k8s_job_names("issue-2261-v7", AgentRole.CODER) + assert job_name == "egg-agent-issue-2261-v7-coder" + + def test_build_agent_worktree_id_includes_slice(self): + from kubernetes_spawner import KubernetesSpawner + + wt_id = KubernetesSpawner._build_agent_worktree_id( + "issue-2261-v7", AgentRole.CODER, slice_id="slice-2" + ) + assert wt_id == "issue-2261-v7-slice-2-coder" + + def test_build_agent_worktree_id_omits_slice_when_unscoped(self): + from kubernetes_spawner import KubernetesSpawner + + wt_id = KubernetesSpawner._build_agent_worktree_id("issue-2261-v7", AgentRole.CODER) + assert wt_id == "issue-2261-v7-coder" + + def test_concurrent_slices_get_distinct_ids(self): + """Two slice spawns for the same role must NOT collide on either id.""" + from kubernetes_spawner import KubernetesSpawner + + s1_job, _ = KubernetesSpawner._build_k8s_job_names( + "issue-2261-v7", AgentRole.CODER, slice_id="slice-1" + ) + s2_job, _ = KubernetesSpawner._build_k8s_job_names( + "issue-2261-v7", AgentRole.CODER, slice_id="slice-2" + ) + assert s1_job != s2_job + + s1_wt = KubernetesSpawner._build_agent_worktree_id( + "issue-2261-v7", AgentRole.CODER, slice_id="slice-1" + ) + s2_wt = KubernetesSpawner._build_agent_worktree_id( + "issue-2261-v7", AgentRole.CODER, slice_id="slice-2" + ) + assert s1_wt != s2_wt + + def test_underscore_role_still_hyphenated_in_job_name(self): + """``task_planner`` etc. stay hyphenated under slice scope (RFC-1123).""" + from kubernetes_spawner import KubernetesSpawner + + job_name, _ = KubernetesSpawner._build_k8s_job_names( + "issue-2261-v7", AgentRole.TASK_PLANNER, slice_id="slice-3" + ) + assert job_name == "egg-agent-issue-2261-v7-slice-3-task-planner" + + +class TestSpawnAgentJobSliceScope: + """``spawn_agent_job`` threads ``slice_id`` into the gateway worktree key.""" + + def test_create_worktrees_called_with_slice_scoped_id( + self, spawner, mock_k8s_client, mock_gateway + ): + spawner.spawn_agent_job( + pipeline_id="issue-2261-v7", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + slice_id="slice-2", + ) + # Pre-spawn worktree creation is keyed by the slice-scoped id. + cw_kwargs = mock_gateway.create_worktrees.call_args.kwargs + assert cw_kwargs["container_id"] == "issue-2261-v7-slice-2-coder" + + def test_session_register_uses_slice_scoped_worktree_container_id( + self, spawner, mock_k8s_client, mock_gateway + ): + spawner.spawn_agent_job( + pipeline_id="issue-2261-v7", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + slice_id="slice-2", + ) + # The gateway session reuses the worktree under the same key — + # without slice scope here the agent's session would dangle. + rs_kwargs = mock_gateway.register_session.call_args.kwargs + assert rs_kwargs["worktree_container_id"] == "issue-2261-v7-slice-2-coder" + + def test_concurrent_spawn_fn_forwards_slice_id(self, spawner, mock_k8s_client, mock_gateway): + spawn_fn = spawner.create_concurrent_spawn_fn( + pipeline_id="issue-2261-v7", + issue_number=2261, + repo_volumes={}, + mode="public", + repos=["owner/repo"], + phase="implement", + slice_id="slice-2", + ) + spawn_fn(role=AgentRole.CODER, branch="egg/issue-2261-v7/slice-2") + cw_kwargs = mock_gateway.create_worktrees.call_args.kwargs + assert cw_kwargs["container_id"] == "issue-2261-v7-slice-2-coder" + + +class TestRestartAgentJobSliceScope: + """``restart_agent_job`` threads ``slice_id`` into delete + respawn (#2410).""" + + def test_delete_targets_slice_scoped_job_name(self, spawner, mock_k8s_client, mock_gateway): + """A slice-scoped restart must delete the slice-scoped Job, not the pipeline-level one. + + Without the fix, ``delete_job`` was called against ``egg-sandbox-egg-agent-{pid}-{role}`` + — leaving the actual ``egg-agent-{pid}-slice-{N}-{role}`` Job running while a fresh + non-scoped Job was spawned alongside it. + """ + spawner.restart_agent_job( + pipeline_id="issue-2261-v7", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + slice_id="slice-2", + ) + delete_call = mock_k8s_client.delete_job.call_args_list[-1] + assert delete_call.args[0] == "egg-sandbox-egg-agent-issue-2261-v7-slice-2-coder" + + def test_gateway_session_cleanup_uses_slice_scoped_container_id( + self, spawner, mock_k8s_client, mock_gateway + ): + """The gateway session is keyed by the slice-scoped unprefixed name.""" + spawner.restart_agent_job( + pipeline_id="issue-2261-v7", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + slice_id="slice-2", + ) + gw_call = mock_gateway.delete_session_by_container.call_args_list[-1] + assert gw_call.args[0] == "egg-agent-issue-2261-v7-slice-2-coder" + + def test_respawn_uses_slice_scoped_worktree_id(self, spawner, mock_k8s_client, mock_gateway): + """The respawned Job mounts the slice-scoped worktree. + + Pre-spawn ``create_worktrees`` is keyed by the slice-scoped container_id + — failure mode #2 from the issue (worktree wrong / absent) is fixed by + threading slice_id into the spawn call. + """ + spawner.restart_agent_job( + pipeline_id="issue-2261-v7", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + slice_id="slice-2", + ) + cw_kwargs = mock_gateway.create_worktrees.call_args.kwargs + assert cw_kwargs["container_id"] == "issue-2261-v7-slice-2-coder" + + def test_restart_count_is_per_slice(self, spawner, mock_k8s_client, mock_gateway): + """Concurrent slices each get an independent restart budget.""" + spawner.restart_agent_job( + pipeline_id="issue-2261-v7", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + slice_id="slice-2", + ) + spawner.restart_agent_job( + pipeline_id="issue-2261-v7", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + slice_id="slice-3", + ) + # Each slice's coder has burned exactly one budget slot. + assert spawner.get_restart_count("issue-2261-v7", "coder", slice_id="slice-2") == 1 + assert spawner.get_restart_count("issue-2261-v7", "coder", slice_id="slice-3") == 1 + # The pipeline-level bucket is untouched. + assert spawner.get_restart_count("issue-2261-v7", "coder") == 0 + + def test_reset_restart_counts_clears_slice_buckets( + self, spawner, mock_k8s_client, mock_gateway + ): + """Per-pipeline reset must sweep every slice bucket too.""" + spawner._restart_counts[("issue-2261-v7", "coder", "slice-2")] = 3 + spawner._restart_counts[("issue-2261-v7", "coder", "slice-3")] = 2 + spawner._restart_counts[("issue-2261-v7", "coder", None)] = 1 + spawner._restart_counts[("issue-9999", "coder", "slice-2")] = 4 + + spawner.reset_restart_counts("issue-2261-v7") + + assert spawner.get_restart_count("issue-2261-v7", "coder", slice_id="slice-2") == 0 + assert spawner.get_restart_count("issue-2261-v7", "coder", slice_id="slice-3") == 0 + assert spawner.get_restart_count("issue-2261-v7", "coder") == 0 + # Sibling pipeline untouched. + assert spawner.get_restart_count("issue-9999", "coder", slice_id="slice-2") == 4 + + def test_restart_propagates_egg_slice_id_to_container_env( + self, spawner, mock_k8s_client, mock_gateway + ): + """The respawned Job's environment carries ``EGG_SLICE_ID``. + + Failure mode #3 from #2410: without the env var on the new Job, + the agent's BRC handlers can't tag CONSENSUS_* signals with the + slice and the orchestrator routes them to the pipeline-level + tracker. Naming + worktree id alone are insufficient — the env + is what the *agent* reads. + """ + result = spawner.restart_agent_job( + pipeline_id="issue-2261-v7", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + slice_id="slice-2", + ) + # The env on the SpawnedContainer reflects what spawn_agent_job + # assembled — and what was forwarded to ``create_container``. + assert result.environment.get("EGG_SLICE_ID") == "slice-2" + # Belt-and-braces: the env actually reached the k8s create call. + create_kwargs = mock_k8s_client.create_container.call_args.kwargs + assert create_kwargs["environment"].get("EGG_SLICE_ID") == "slice-2" + + def test_pipeline_level_restart_does_not_set_egg_slice_id( + self, spawner, mock_k8s_client, mock_gateway + ): + """Without ``slice_id``, the restarted Job's env has no slice scope.""" + result = spawner.restart_agent_job( + pipeline_id="issue-2261-v7", + agent_role=AgentRole.CODER, + repos=["owner/repo"], + ) + assert "EGG_SLICE_ID" not in result.environment + create_kwargs = mock_k8s_client.create_container.call_args.kwargs + assert "EGG_SLICE_ID" not in create_kwargs["environment"] + + +class TestDetectUncommittedChangesSliceScope: + """``detect_uncommitted_changes`` inspects the slice-scoped worktree (#2410).""" + + def test_detects_changes_in_slice_scoped_worktree(self, spawner, tmp_path): + worktree_dir = tmp_path / "issue-2261-v7-slice-2-coder" / "owner-repo" + worktree_dir.mkdir(parents=True) + + with ( + patch("kubernetes_spawner.WORKTREE_BASE_DIR", tmp_path), + patch("subprocess.run") as mock_run, + ): + mock_run.return_value = MagicMock( + returncode=0, + stdout=" M file1.py\n?? file2.py\n", + ) + result = spawner.detect_uncommitted_changes( + "issue-2261-v7", "coder", slice_id="slice-2" + ) + + assert result is not None + assert result["worktree_id"] == "issue-2261-v7-slice-2-coder" + assert result["slice_id"] == "slice-2" + assert result["file_count"] == 2 + + def test_pipeline_level_call_does_not_pick_up_slice_worktree(self, spawner, tmp_path): + """Without slice_id, only the pipeline-level worktree is inspected. + + A slice agent's uncommitted work must not surface through a + pipeline-level call — they're separate worktrees with separate + ownership semantics. + """ + # Only the slice-scoped worktree exists on disk. + slice_dir = tmp_path / "issue-2261-v7-slice-2-coder" / "owner-repo" + slice_dir.mkdir(parents=True) + + with ( + patch("kubernetes_spawner.WORKTREE_BASE_DIR", tmp_path), + patch("subprocess.run") as mock_run, + ): + mock_run.return_value = MagicMock(returncode=0, stdout=" M file1.py\n") + result = spawner.detect_uncommitted_changes("issue-2261-v7", "coder") + + # No pipeline-level worktree → returns None even though the slice worktree + # has uncommitted changes. + assert result is None + + def test_slice_call_does_not_pick_up_pipeline_worktree(self, spawner, tmp_path): + """Symmetric guard: a slice-scoped lookup must not surface pipeline-level work.""" + # Only the pipeline-level worktree exists on disk. + pipeline_dir = tmp_path / "issue-2261-v7-coder" / "owner-repo" + pipeline_dir.mkdir(parents=True) + + with ( + patch("kubernetes_spawner.WORKTREE_BASE_DIR", tmp_path), + patch("subprocess.run") as mock_run, + ): + mock_run.return_value = MagicMock(returncode=0, stdout=" M file1.py\n") + result = spawner.detect_uncommitted_changes( + "issue-2261-v7", "coder", slice_id="slice-2" + ) + + assert result is None + + +class TestCleanupPipelineSliceWorktrees: + """``cleanup_pipeline``'s filesystem scan recognises slice-scoped worktrees.""" + + def test_filesystem_scan_picks_up_slice_scoped_worktrees( + self, spawner, mock_k8s_client, mock_gateway, tmp_path, monkeypatch + ): + import kubernetes_spawner as ks_mod + + # Lay out a mix of pipeline-level, role-level, slice-scoped, and + # unrelated entries so the scan's allowlist is exercised end-to-end. + (tmp_path / "issue-2261-v7").mkdir() + (tmp_path / "issue-2261-v7-coder").mkdir() + (tmp_path / "issue-2261-v7-slice-2-coder").mkdir() + (tmp_path / "issue-2261-v7-slice-3-tester").mkdir() + # Sibling pipeline whose id starts with the same prefix — must NOT + # be swept (mirrors the #1865 regression guard). + (tmp_path / "issue-2261-v7-other-thing").mkdir() + (tmp_path / "issue-9999-coder").mkdir() + + monkeypatch.setattr(ks_mod, "WORKTREE_BASE_DIR", tmp_path) + # No Jobs returned — drive cleanup purely off the filesystem scan. + mock_k8s_client.list_containers.return_value = [] + + spawner.cleanup_pipeline("issue-2261-v7") + + cleaned = { + call.kwargs.get("container_id") for call in mock_gateway.delete_worktrees.call_args_list + } + assert "issue-2261-v7" in cleaned + assert "issue-2261-v7-coder" in cleaned + assert "issue-2261-v7-slice-2-coder" in cleaned + assert "issue-2261-v7-slice-3-tester" in cleaned + # Sibling pipelines are left alone. + assert "issue-2261-v7-other-thing" not in cleaned + assert "issue-9999-coder" not in cleaned diff --git a/orchestrator/tests/test_pipeline_branch_namespace.py b/orchestrator/tests/test_pipeline_branch_namespace.py new file mode 100644 index 0000000000..249875d01d --- /dev/null +++ b/orchestrator/tests/test_pipeline_branch_namespace.py @@ -0,0 +1,126 @@ +"""Regression tests for #2399 — pipeline-branch / slice-branch ref-namespace. + +The orchestrator pushes the pipeline tip to ``/work`` so the +``/`` namespace can hold slice integration branches as siblings +(``/slice-N``) without git's ``directory file conflict`` +rejection. A leaf ref at ```` and a child at +``/slice-N`` cannot coexist on origin. + +These tests pin the contract so a future refactor can't quietly +re-introduce the leaf-ref shape. +""" + +from routes.pipelines import _ensure_pipeline_work_ref, _slice_namespace_root + + +class TestEnsurePipelineWorkRef: + def test_appends_work_to_egg_branch(self) -> None: + assert _ensure_pipeline_work_ref("egg/issue-2261-v6") == "egg/issue-2261-v6/work" + + def test_appends_work_to_qualified_egg_branch(self) -> None: + # Qualifier suffixes (-v3, -backend) propagate; ``/work`` lives one + # level deeper. + assert _ensure_pipeline_work_ref("egg/issue-100-backend") == "egg/issue-100-backend/work" + + def test_idempotent_when_already_work_suffixed(self) -> None: + assert _ensure_pipeline_work_ref("egg/issue-2261-v6/work") == "egg/issue-2261-v6/work" + + def test_passthrough_for_none(self) -> None: + assert _ensure_pipeline_work_ref(None) is None + + def test_passthrough_for_non_egg_branch(self) -> None: + # Babysit branches are arbitrary PR head refs; the orchestrator + # does not own the namespace below them and must not rewrite them. + assert _ensure_pipeline_work_ref("feature/foo") == "feature/foo" + assert _ensure_pipeline_work_ref("main") == "main" + + def test_passthrough_for_egg_custom_branch(self) -> None: + # CUSTOM-mode auto-generates ``egg/custom-``; the same + # ``/work`` rule applies so slice integration branches can be + # added as siblings if a custom pipeline ever uses the slice DAG. + assert _ensure_pipeline_work_ref("egg/custom-deadbeef") == "egg/custom-deadbeef/work" + + def test_single_segment_egg_work_is_not_treated_as_normalised(self) -> None: + # Degenerate input: ``egg/work`` ends in ``/work`` but is a + # single-segment id, not an already-normalised pipeline branch. + # Plain ``endswith("/work")`` would return it unchanged and the + # slice DAG would push to ``egg/work/slice-N`` under the leaf + # ``egg/work`` — back to the directory/file conflict #2399 fixes. + # Structural check (≥2 slashes, last segment ``work``) treats it + # as needing normalisation. + assert _ensure_pipeline_work_ref("egg/work") == "egg/work/work" + + def test_strips_trailing_slash_before_appending_work(self) -> None: + # The branch validation regex permits trailing slashes; an + # input like ``egg/issue-1/`` must not collapse to a + # double-slash ``egg/issue-1//work``. + assert _ensure_pipeline_work_ref("egg/issue-1/") == "egg/issue-1/work" + + def test_degenerate_egg_slash_does_not_double_slash(self) -> None: + # ``egg/`` is degenerate (no id segment). Pre-fix the helper + # produced ``egg//work``; we now strip the trailing slash and + # leave the bare ``egg`` unchanged (it does not start with + # ``egg/``). The branch validation regex upstream should + # reject this shape — the helper just refuses to make it worse. + assert _ensure_pipeline_work_ref("egg/") == "egg" + + +class TestSliceNamespaceRoot: + def test_strips_work_suffix(self) -> None: + assert _slice_namespace_root("egg/issue-2261-v6/work") == "egg/issue-2261-v6" + + def test_passthrough_when_no_work_suffix(self) -> None: + # Legacy / non-normalised callers: the branch itself is the root. + assert _slice_namespace_root("egg/issue-2261-v6") == "egg/issue-2261-v6" + + def test_strips_only_trailing_work(self) -> None: + # ``/work`` mid-path is NOT a suffix and must not be stripped. + assert _slice_namespace_root("egg/work-stream/v1/work") == "egg/work-stream/v1" + + def test_qualifier_preserved(self) -> None: + assert _slice_namespace_root("egg/issue-100-backend/work") == "egg/issue-100-backend" + + def test_single_segment_egg_work_is_root_itself(self) -> None: + # Mirror of the structural check in ``_ensure_pipeline_work_ref``: + # ``egg/work`` is a single-segment id, not a normalised pipeline + # branch, so the namespace root is the branch itself rather than + # collapsing to ``egg``. + assert _slice_namespace_root("egg/work") == "egg/work" + + +class TestNamespaceCoexistence: + """Pin the design property that solves the conflict. + + The pipeline tip ``/work`` and slice integration branches + ``/slice-N`` must share a single parent path ``/`` — + that's the whole point of the #2399 fix. A regression that starts + pushing the tip to ```` (a leaf ref) would re-introduce the + ``directory file conflict`` from GitHub. + """ + + def test_pipeline_tip_and_slice_share_namespace_parent(self) -> None: + pipeline_branch = _ensure_pipeline_work_ref("egg/issue-2261-v6") + assert pipeline_branch == "egg/issue-2261-v6/work" + + namespace_root = _slice_namespace_root(pipeline_branch) + slice_branch = f"{namespace_root}/slice-1" + + # Both refs share the parent ``egg/issue-2261-v6/`` and live as + # siblings — neither is a prefix of the other. + assert pipeline_branch.rsplit("/", 1)[0] == slice_branch.rsplit("/", 1)[0] + assert not slice_branch.startswith(pipeline_branch + "/") + assert not pipeline_branch.startswith(slice_branch + "/") + + def test_pipeline_tip_is_not_a_prefix_of_slice_branch(self) -> None: + # Regression: the pre-fix shape had ``pipeline.branch == + # 'egg/issue-2261-v6'`` and slice branches at + # ``'egg/issue-2261-v6/slice-N'``, making the slice path a child + # of the pipeline ref. Git's ref storage rejects that with + # ``directory file conflict``. + pipeline_branch = _ensure_pipeline_work_ref("egg/issue-2261-v6") + namespace_root = _slice_namespace_root(pipeline_branch) + slice_branch = f"{namespace_root}/slice-1" + assert not slice_branch.startswith(pipeline_branch + "/"), ( + "Slice integration branch must not live under the pipeline tip's path " + "— that's the directory/file conflict #2399 fixes." + ) diff --git a/orchestrator/tests/test_pipelines_api.py b/orchestrator/tests/test_pipelines_api.py index f3b1fe8f65..370c2821e3 100644 --- a/orchestrator/tests/test_pipelines_api.py +++ b/orchestrator/tests/test_pipelines_api.py @@ -589,7 +589,9 @@ def test_create_pipeline_with_explicit_pipeline_id( assert response.status_code == 200 call_kwargs = mock_store.create_pipeline.call_args[1] assert call_kwargs["pipeline_id"] == "KORE-1234" - assert call_kwargs["branch"] == "egg/KORE-1234" + # #2399 — the pipeline tip is normalised to ``/work`` so slice + # integration branches at ``/slice-N`` can coexist as siblings. + assert call_kwargs["branch"] == "egg/KORE-1234/work" @patch("routes.pipelines.get_gateway_client") @patch("routes.pipelines.get_state_store") @@ -621,7 +623,8 @@ def test_create_pipeline_with_qualifier_suffix( assert response.status_code == 200 call_kwargs = mock_store.create_pipeline.call_args[1] assert call_kwargs["pipeline_id"] == "KORE-1234-backend" - assert call_kwargs["branch"] == "egg/KORE-1234-backend" + # #2399 — pipeline tip normalised to ``/work``. + assert call_kwargs["branch"] == "egg/KORE-1234-backend/work" @patch("routes.pipelines.get_gateway_client") @patch("routes.pipelines.get_state_store") @@ -661,7 +664,9 @@ def test_create_pipeline_rejects_existing_branch( assert "already exists" in body["message"] assert "qualifier" in body["message"] assert body["details"]["reason"] == "branch_exists" - assert body["details"]["branch"] == "egg/KORE-1234" + # #2399 — pipeline tip normalised to ``/work`` before the + # branch-existence check, so the error surfaces the /work shape. + assert body["details"]["branch"] == "egg/KORE-1234/work" @patch("routes.pipelines.get_gateway_client") @patch("routes.pipelines.get_state_store") @@ -710,7 +715,8 @@ def test_create_pipeline_rejects_stale_branch_with_no_active_pipeline( assert response.status_code == 409 body = response.get_json() assert body["details"]["reason"] == "stale_branch" - assert body["details"]["branch"] == "egg/issue-2137" + # #2399 — pipeline tip normalised to ``/work``. + assert body["details"]["branch"] == "egg/issue-2137/work" assert "cancel_task" in body["details"]["hint"] assert "cleanup=true" in body["details"]["hint"] @@ -822,7 +828,8 @@ def test_pipeline_id_with_issue_number_and_qualifier( assert response.status_code == 200 call_kwargs = mock_store.create_pipeline.call_args[1] assert call_kwargs["pipeline_id"] == "issue-42-frontend" - assert call_kwargs["branch"] == "egg/issue-42-frontend" + # #2399 — pipeline tip normalised to ``/work``. + assert call_kwargs["branch"] == "egg/issue-42-frontend/work" assert call_kwargs["issue_number"] == 42 @patch("routes.pipelines.get_gateway_client") diff --git a/orchestrator/tests/test_restart_agent.py b/orchestrator/tests/test_restart_agent.py index 24c6f8189c..d4915fbba3 100644 --- a/orchestrator/tests/test_restart_agent.py +++ b/orchestrator/tests/test_restart_agent.py @@ -169,7 +169,7 @@ def test_restart_tracks_count(self, spawner, mock_docker_client, mock_gateway_cl def test_restart_limit_exceeded_raises(self, spawner, mock_docker_client, mock_gateway_client): """Restart should raise ContainerSpawnError when limit is exceeded.""" # Pre-set restart count to the limit - spawner._restart_counts[("issue-100", "coder")] = 2 + spawner._restart_counts[("issue-100", "coder", None)] = 2 with pytest.raises(ContainerSpawnError, match="Restart limit"): spawner.restart_agent_container( @@ -182,7 +182,7 @@ def test_restart_limit_exceeded_raises(self, spawner, mock_docker_client, mock_g def test_restart_custom_max_restarts(self, spawner, mock_docker_client, mock_gateway_client): """Custom max_restarts should be respected.""" - spawner._restart_counts[("issue-100", "coder")] = 5 + spawner._restart_counts[("issue-100", "coder", None)] = 5 with pytest.raises(ContainerSpawnError, match="Restart limit"): spawner.restart_agent_container( @@ -303,14 +303,14 @@ def test_get_restart_count_default_zero(self, spawner): def test_get_restart_count_after_restart(self, spawner): """Count should increment after manual tracking.""" - spawner._restart_counts[("issue-100", "coder")] = 3 + spawner._restart_counts[("issue-100", "coder", None)] = 3 assert spawner.get_restart_count("issue-100", "coder") == 3 def test_reset_restart_counts_clears_pipeline(self, spawner): """reset_restart_counts should clear all counts for a pipeline.""" - spawner._restart_counts[("issue-100", "coder")] = 2 - spawner._restart_counts[("issue-100", "tester")] = 1 - spawner._restart_counts[("issue-200", "coder")] = 3 + spawner._restart_counts[("issue-100", "coder", None)] = 2 + spawner._restart_counts[("issue-100", "tester", None)] = 1 + spawner._restart_counts[("issue-200", "coder", None)] = 3 spawner.reset_restart_counts("issue-100") @@ -761,6 +761,170 @@ def test_restart_agent_uses_computed_gateway_mode( ) +# --------------------------------------------------------------------------- +# Issue #2410: slice_id forwarding through the operator restart route +# --------------------------------------------------------------------------- + + +@pytest.mark.skipif(not _HAS_FLASK, reason="Flask not available") +class TestRestartAgentEndpointSliceScope: + """``slice_id`` query / body parameter is validated and forwarded (#2410).""" + + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_slice_id_query_param_forwarded_to_spawner( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, client + ): + """``?slice_id=slice-2`` reaches ``restart_agent_container``.""" + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_resolve.return_value = (mock_store, pipeline) + + mock_spawner = MagicMock() + new_container = SpawnedContainer( + container_info=ContainerInfo( + container_id="new-container-xyz", + container_name="egg-issue-100-slice-2-coder", + status=ContainerStatus.RUNNING, + ), + session_info=None, + agent_role=AgentRole.CODER, + pipeline_id="issue-100", + environment={}, + ) + mock_spawner.restart_agent_container.return_value = new_container + mock_spawner.get_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart?slice_id=slice-2", + json={"reason": "Slice agent stalled"}, + ) + + assert response.status_code == 200 + restart_call = mock_spawner.restart_agent_container.call_args + assert restart_call.kwargs["slice_id"] == "slice-2" + # Reading the restart count after a slice-scoped restart MUST + # query the per-slice budget bucket (#2410). Without this, the + # JSON response and audit log report the pipeline-level count + # (typically 0) and operators can't trust "you've burned N of M + # restarts" telemetry. + get_count_call = mock_spawner.get_restart_count.call_args + assert get_count_call.kwargs.get("slice_id") == "slice-2" + + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_slice_id_body_field_forwarded_to_spawner( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, client + ): + """``{"slice_id": "slice-2"}`` in the body reaches the spawner.""" + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_resolve.return_value = (mock_store, pipeline) + + mock_spawner = MagicMock() + new_container = SpawnedContainer( + container_info=ContainerInfo( + container_id="new-container-xyz", + container_name="egg-issue-100-slice-2-coder", + status=ContainerStatus.RUNNING, + ), + session_info=None, + agent_role=AgentRole.CODER, + pipeline_id="issue-100", + environment={}, + ) + mock_spawner.restart_agent_container.return_value = new_container + mock_spawner.get_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={"reason": "Slice agent stalled", "slice_id": "slice-2"}, + ) + + assert response.status_code == 200 + restart_call = mock_spawner.restart_agent_container.call_args + assert restart_call.kwargs["slice_id"] == "slice-2" + get_count_call = mock_spawner.get_restart_count.call_args + assert get_count_call.kwargs.get("slice_id") == "slice-2" + + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_invalid_slice_id_returns_400(self, mock_repo, mock_resolve, client): + """Non-canonical slice_id values are rejected with 400 before spawn.""" + mock_repo.return_value = "/repo" + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_resolve.return_value = (mock_store, pipeline) + + # Path-separator and shell-metacharacter values must not reach + # the spawner — defense-in-depth against a future caller that + # forgets the upstream regex. + for bad in ("phase-2", "slice-2/etc", "../slice-2", "slice-"): + response = client.post( + f"/api/v1/pipelines/issue-100/agents/coder/restart?slice_id={bad}", + json={}, + ) + assert response.status_code == 400, f"slice_id={bad!r} should be rejected" + + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_no_slice_id_forwards_none( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, client + ): + """Pipeline-level callers (no slice_id) get ``slice_id=None`` forwarded.""" + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_running_agent() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_resolve.return_value = (mock_store, pipeline) + + mock_spawner = MagicMock() + new_container = SpawnedContainer( + container_info=ContainerInfo( + container_id="new-container-xyz", + container_name="egg-issue-100-coder", + status=ContainerStatus.RUNNING, + ), + session_info=None, + agent_role=AgentRole.CODER, + pipeline_id="issue-100", + environment={}, + ) + mock_spawner.restart_agent_container.return_value = new_container + mock_spawner.get_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart", + json={"reason": "Pipeline-level restart"}, + ) + + assert response.status_code == 200 + restart_call = mock_spawner.restart_agent_container.call_args + assert restart_call.kwargs["slice_id"] is None + get_count_call = mock_spawner.get_restart_count.call_args + assert get_count_call.kwargs.get("slice_id") is None + + # --------------------------------------------------------------------------- # Issue #1695: mode=None raises ValueError (issue 7) # --------------------------------------------------------------------------- @@ -967,7 +1131,7 @@ def test_concurrent_restarts_one_past_limit( ): """If one thread consumes the last restart, the other should get limit error.""" # Pre-set count to 1 with max_restarts=2 — only one more slot - spawner._restart_counts[("issue-100", "coder")] = 1 + spawner._restart_counts[("issue-100", "coder", None)] = 1 results = [] errors = [] @@ -1001,17 +1165,21 @@ def restart_agent(): assert "Restart limit" in str(errors[0]) def test_restart_lock_created_per_key(self, spawner): - """Each (pipeline_id, agent_role) pair should get its own lock.""" - key1 = ("issue-100", "coder") - key2 = ("issue-100", "tester") - key3 = ("issue-200", "coder") + """Each (pipeline_id, agent_role, slice_id) tuple should get its own lock.""" + key1 = ("issue-100", "coder", None) + key2 = ("issue-100", "tester", None) + key3 = ("issue-200", "coder", None) + # Slice scope splits the key further (#2410). + key4 = ("issue-100", "coder", "slice-2") lock1 = spawner._get_restart_lock(key1) lock2 = spawner._get_restart_lock(key2) lock3 = spawner._get_restart_lock(key3) + lock4 = spawner._get_restart_lock(key4) assert lock1 is not lock2, "Different agents should have different locks" assert lock1 is not lock3, "Different pipelines should have different locks" + assert lock1 is not lock4, "Different slice scopes should have different locks" # Same key should return the same lock assert spawner._get_restart_lock(key1) is lock1 @@ -1034,28 +1202,28 @@ def test_reset_clears_counts_retains_locks(self, spawner): lock for the same key — breaking mutual exclusion. """ # Create some locks by accessing them - spawner._get_restart_lock(("issue-100", "coder")) - spawner._get_restart_lock(("issue-100", "tester")) - spawner._get_restart_lock(("issue-200", "coder")) + spawner._get_restart_lock(("issue-100", "coder", None)) + spawner._get_restart_lock(("issue-100", "tester", None)) + spawner._get_restart_lock(("issue-200", "coder", None)) # Set some counts - spawner._restart_counts[("issue-100", "coder")] = 2 - spawner._restart_counts[("issue-100", "tester")] = 1 - spawner._restart_counts[("issue-200", "coder")] = 3 + spawner._restart_counts[("issue-100", "coder", None)] = 2 + spawner._restart_counts[("issue-100", "tester", None)] = 1 + spawner._restart_counts[("issue-200", "coder", None)] = 3 spawner.reset_restart_counts("issue-100") # Counts for issue-100 should be cleared - assert spawner._restart_counts.get(("issue-100", "coder"), 0) == 0 - assert spawner._restart_counts.get(("issue-100", "tester"), 0) == 0 + assert spawner._restart_counts.get(("issue-100", "coder", None), 0) == 0 + assert spawner._restart_counts.get(("issue-100", "tester", None), 0) == 0 # Locks for issue-100 should be retained (not deleted) - assert ("issue-100", "coder") in spawner._restart_locks - assert ("issue-100", "tester") in spawner._restart_locks + assert ("issue-100", "coder", None) in spawner._restart_locks + assert ("issue-100", "tester", None) in spawner._restart_locks # issue-200 should be untouched - assert spawner._restart_counts.get(("issue-200", "coder"), 0) == 3 - assert ("issue-200", "coder") in spawner._restart_locks + assert spawner._restart_counts.get(("issue-200", "coder", None), 0) == 3 + assert ("issue-200", "coder", None) in spawner._restart_locks def test_restart_lock_initialization(self): """Restart lock dicts should be initialised in constructor.""" diff --git a/orchestrator/tests/test_slice_run_loop_integration.py b/orchestrator/tests/test_slice_run_loop_integration.py index 84ae72b324..d74a21df50 100644 --- a/orchestrator/tests/test_slice_run_loop_integration.py +++ b/orchestrator/tests/test_slice_run_loop_integration.py @@ -85,10 +85,15 @@ def _make_pipeline( ) -> 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). + ``branch`` is derived from ``pipeline_id`` and carries the ``/work`` + suffix that ``create_pipeline`` applies via + :func:`routes.pipelines._ensure_pipeline_work_ref` (#2399). The + pipeline tip lives at ``egg//work`` so slice integration + branches (``egg//slice-N``) coexist as siblings under + ``egg//``. 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, @@ -99,7 +104,7 @@ def _make_pipeline( id=pipeline_id, issue_number=issue_number, repo="owner/repo", - branch=f"egg/{pipeline_id}", + branch=f"egg/{pipeline_id}/work", status=PipelineStatus.RUNNING, current_phase=PipelinePhase.IMPLEMENT, config=config, @@ -515,7 +520,9 @@ def test_single_root_slice_uses_pipeline_branch_as_parent(self) -> None: spawner.gateway.create_slice_pr.assert_called_once() pr_kwargs = spawner.gateway.create_slice_pr.call_args.kwargs assert pr_kwargs["base"] == pipeline.branch - assert pr_kwargs["head"] == f"{pipeline.branch}/slice-1" + # Slice integration branch lives as a sibling of ``/work`` under + # ``egg//`` (#2399), not as a child of ``/work``. + assert pr_kwargs["head"] == f"egg/{pipeline.id}/slice-1" assert pr_kwargs["slice_id"] == "slice-1" def test_child_slice_targets_parent_integration_branch(self) -> None: @@ -1176,7 +1183,10 @@ def test_slice_id_overrides_env_and_forwards_to_executor( worktree_repo_path=Path("/tmp/x"), slice_id="slice-3", ) - # Caller's dict is not mutated; the function takes a shallow copy. + # Caller's dict is not mutated — _run_concurrent_phase no longer + # touches sandbox_env (the EGG_SLICE_ID assignment was dropped in + # the v2 review fix; slice scope flows through the spawner via the + # slice_id kwarg instead). assert original_env == {"EGG_PIPELINE_ID": pipeline.id, "OTHER": "v"} # Executor receives slice_id="slice-3". assert MockExecutor.call_args.kwargs["slice_id"] == "slice-3" @@ -1606,7 +1616,9 @@ class TestSliceIntegrationBranchQualifierPreserved: def test_qualified_pipeline_branch_propagates_to_slice_branches(self) -> None: """``egg/issue-N-v3`` ⇒ slices stack under the qualified prefix.""" pipeline = _make_pipeline(pipeline_id="issue-2261-v3", issue_number=2261) - assert pipeline.branch == "egg/issue-2261-v3" # helper-derived; sanity-check + # Pipeline tip lives at ``/work`` (#2399); the slice + # namespace root is ``egg/issue-2261-v3``, one level up. + assert pipeline.branch == "egg/issue-2261-v3/work" # helper-derived; sanity-check contract = _make_contract( pipeline_id="issue-2261-v3", issue_number=2261, @@ -1646,9 +1658,10 @@ def _capture(*args: Any, **kwargs: Any) -> bool: 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("parent_branch") == "egg/issue-2261-v3/work", ( + "parent_branch for the root slice must be the qualified pipeline branch " + "(``/work`` per #2399), got " + f"{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 " diff --git a/orchestrator/tests/test_slice_signal_routing.py b/orchestrator/tests/test_slice_signal_routing.py new file mode 100644 index 0000000000..3e968213d5 --- /dev/null +++ b/orchestrator/tests/test_slice_signal_routing.py @@ -0,0 +1,753 @@ +"""Slice-aware spawn env + signal routing (#2403). + +Pins the wire shape that lets per-slice agents reach the orchestrator: + + * The slice-spawn path leaves ``EGG_PIPELINE_ID`` as the bare + pipeline id (passes ``state_store.PIPELINE_ID_PATTERN``) and + exposes the slice via ``EGG_SLICE_ID``. An earlier shape jammed + ``{pipeline_id}/{slice_id}`` into ``EGG_PIPELINE_ID`` itself, + which 4xx'd every agent → orchestrator round-trip (the validator + rejects ``/`` and Flask's URL converter doesn't allow it either). + * Consensus signal handlers read ``slice_id`` from the request body + and route the tracker lookup to ``get_peer_consensus_tracker( + pipeline_id, slice_id)`` so per-slice CONSENSUS_* lands on the + slice's tracker, not the pipeline-level one. +""" + +from datetime import UTC, datetime +from pathlib import Path +from unittest.mock import MagicMock, patch + +from models import ( + AgentExecution, + AgentExecutionStatus, + AgentRole, + ContainerInfo, + ContainerStatus, + PhaseExecution, + Pipeline, + PipelineConfig, + PipelinePhase, + PipelineStatus, +) +from routes.pipelines import _run_concurrent_phase +from state_store import PIPELINE_ID_PATTERN + + +def _make_pipeline() -> Pipeline: + config = PipelineConfig() + config.concurrent_execution = True + config.max_concurrent_agents = 4 + config.consensus_timeout_minutes = 30 + return Pipeline( + id="issue-2403", + issue_number=2403, + repo="owner/repo", + branch="egg/issue-2403/work", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + config=config, + ) + + +def _make_execution(role: AgentRole, container_id: str) -> AgentExecution: + return AgentExecution( + role=role, + status=AgentExecutionStatus.RUNNING, + container_id=container_id, + started_at=datetime.now(UTC), + ) + + +_CALL_ARGS = { + "repo_volumes": {}, + "gateway_mode": "public", + "repos": ["owner/repo"], + "certs_volume": None, + "worktree_repo_path": Path("/tmp/test-repo"), +} + + +def _setup_spawn(executions: list[AgentExecution]): + pipeline = _make_pipeline() + phase_exec = PhaseExecution( + phase=PipelinePhase.IMPLEMENT, + status=PipelineStatus.RUNNING, + ) + mock_store = MagicMock() + mock_pipeline_state = MagicMock() + mock_pipeline_state.get_phase_execution.return_value = phase_exec + mock_store.load_pipeline.return_value = mock_pipeline_state + + mock_docker = MagicMock() + mock_docker.get_container_info.side_effect = lambda cid: ContainerInfo( + container_id=cid, + container_name=cid, + status=ContainerStatus.EXITED, + exit_code=0, + exited_at=datetime.now(UTC), + ) + mock_spawner = MagicMock() + mock_spawner.backend = mock_docker + mock_spawner.docker = mock_docker + mock_spawner.create_concurrent_spawn_fn.return_value = MagicMock() + return pipeline, mock_store, mock_spawner + + +# --------------------------------------------------------------------------- +# call_args helpers — colocated above all consumers so a future +# ``get_peer_consensus_tracker`` signature change has one place to update. +# --------------------------------------------------------------------------- + + +def _slice_arg_from_call(call) -> str | None: + """Extract ``slice_id`` from ``get_peer_consensus_tracker``'s call_args. + + The call site uses positional args (``pipeline_id, slice_id``) but + we accept a kwarg too so future refactors don't break the test. + """ + if "slice_id" in call.kwargs: + return call.kwargs["slice_id"] + return call.args[1] if len(call.args) >= 2 else None + + +def _pipeline_arg_from_call(call) -> str | None: + """Extract ``pipeline_id`` from ``get_peer_consensus_tracker``'s call_args. + + Symmetric with ``_slice_arg_from_call`` so tests don't break if a + future refactor moves ``pipeline_id`` from a positional arg to a + kwarg. + """ + if "pipeline_id" in call.kwargs: + return call.kwargs["pipeline_id"] + return call.args[0] if call.args else None + + +class TestSliceSpawnEnvShape: + """``EGG_PIPELINE_ID`` stays canonical; slice scope rides on ``EGG_SLICE_ID``. + + Single-source-of-truth (#2410 v2 review): ``EGG_SLICE_ID`` is injected + by ``KubernetesSpawner.spawn_agent_job`` from the ``slice_id`` parameter + that already drives Job naming and worktree id, and the key is in + ``_PROTECTED_ENV_KEYS`` so ``extra_env`` cannot smuggle a mismatched + value. These tests pin the wrapper-side contract: ``slice_id`` must + flow through ``create_concurrent_spawn_fn`` as a kwarg, and + ``sandbox_env`` must not carry ``EGG_SLICE_ID`` (a duplicate setter + here would just trip the protected-key warning every spawn). + ``test_kubernetes_spawner.py`` covers the spawner-side env injection. + """ + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic", return_value=0.0) + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_slice_scope_forwards_slice_id_and_keeps_egg_pipeline_id_bare( + self, MockExecutor, mock_prompt, mock_lock, _mono, _sleep + ): + executions = [_make_execution(AgentRole.CODER, "coder-1")] + pipeline, mock_store, mock_spawner = _setup_spawn(executions) + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = { + "is_complete": True, + "has_objections": False, + "blocking_agents": [], + } + MockExecutor.return_value = mock_executor_instance + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + _run_concurrent_phase( + pipeline_id="issue-2403", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + sandbox_env={"PRESERVED": "yes"}, + slice_id="slice-2", + **_CALL_ARGS, + ) + + # ``create_concurrent_spawn_fn`` is the seam where the slice scope + # is frozen for the spawn closure. + kwargs = mock_spawner.create_concurrent_spawn_fn.call_args.kwargs + assert kwargs["pipeline_id"] == "issue-2403" + # Slice scope rides on the ``slice_id`` kwarg, not on + # ``sandbox_env``. The spawner sets ``EGG_SLICE_ID`` itself from + # this parameter (single source of truth). + assert kwargs["slice_id"] == "slice-2" + env = kwargs["sandbox_env"] + # ``sandbox_env`` must NOT carry ``EGG_SLICE_ID`` — the key is + # protected and a duplicate would log the override every spawn. + assert "EGG_SLICE_ID" not in env + # Agent CLIs read EGG_PIPELINE_ID via ``get_pipeline_id`` — only + # set if the caller seeded it. We assert here that the function + # didn't smuggle a slashed value into it. + if "EGG_PIPELINE_ID" in env: + assert PIPELINE_ID_PATTERN.match(env["EGG_PIPELINE_ID"]) is not None + # Pre-existing keys must survive the slice-aware path. + assert env["PRESERVED"] == "yes" + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic", return_value=0.0) + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_no_slice_scope_does_not_set_egg_slice_id( + self, MockExecutor, mock_prompt, mock_lock, _mono, _sleep + ): + executions = [_make_execution(AgentRole.CODER, "coder-1")] + pipeline, mock_store, mock_spawner = _setup_spawn(executions) + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = { + "is_complete": True, + "has_objections": False, + "blocking_agents": [], + } + MockExecutor.return_value = mock_executor_instance + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + _run_concurrent_phase( + pipeline_id="issue-2403", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + sandbox_env={}, + slice_id=None, + **_CALL_ARGS, + ) + + kwargs = mock_spawner.create_concurrent_spawn_fn.call_args.kwargs + assert kwargs.get("slice_id") is None + env = kwargs["sandbox_env"] + assert "EGG_SLICE_ID" not in env + + +class TestConsensusSignalSliceRouting: + """``handle_consensus_*`` look up the slice tracker when ``slice_id`` is supplied.""" + + @patch("peer_consensus.get_peer_consensus_tracker") + def test_propose_routes_to_slice_tracker(self, mock_get_tracker, app): + from routes.signals import handle_consensus_propose_signal + + mock_tracker = MagicMock() + mock_tracker.handle_propose.return_value = { + "version": 1, + "status": "proposed", + "commit_sha": "", + "reviewers": [], + "stale_reviewers": [], + } + mock_get_tracker.return_value = mock_tracker + + with app.app_context(): + handle_consensus_propose_signal( + "issue-2403", + { + "agent_role": "coder", + "slice_id": "slice-2", + "payload": { + "summary": ( + "Implemented slice-2 work with thorough commit " + "message and substantive description over fifty chars" + ), + "artifacts": ["src/a.py"], + }, + }, + Path("/tmp/repo"), + ) + + # The tracker lookup MUST forward slice_id so consensus messages + # land on the per-slice tracker (#2403). + call = mock_get_tracker.call_args + assert _pipeline_arg_from_call(call) == "issue-2403" + assert _slice_arg_from_call(call) == "slice-2" + + @patch("peer_consensus.get_peer_consensus_tracker") + def test_propose_without_slice_falls_back_to_pipeline_tracker(self, mock_get_tracker, app): + from routes.signals import handle_consensus_propose_signal + + mock_tracker = MagicMock() + mock_tracker.handle_propose.return_value = { + "version": 1, + "status": "proposed", + "commit_sha": "", + "reviewers": [], + "stale_reviewers": [], + } + mock_get_tracker.return_value = mock_tracker + + with app.app_context(): + handle_consensus_propose_signal( + "issue-2403", + { + "agent_role": "coder", + "payload": { + "summary": ( + "Implemented work with substantive description " + "over fifty chars to satisfy the validator" + ), + "artifacts": ["src/a.py"], + }, + }, + Path("/tmp/repo"), + ) + + # Pipeline-level callers (no slice_id) keep the bare-tracker + # semantics — ``get_peer_consensus_tracker(pipeline_id, None)`` + # is the same key as the legacy single-arg lookup. + call = mock_get_tracker.call_args + assert _pipeline_arg_from_call(call) == "issue-2403" + assert _slice_arg_from_call(call) is None + + def test_propose_rejects_malformed_slice_id(self, app): + from routes.signals import handle_consensus_propose_signal + + with app.app_context(): + response, status = handle_consensus_propose_signal( + "issue-2403", + { + "agent_role": "coder", + "slice_id": "../etc/passwd", + "payload": { + "summary": ( + "Implemented work with substantive description " + "over fifty chars to satisfy the validator" + ), + "artifacts": ["src/a.py"], + }, + }, + Path("/tmp/repo"), + ) + assert status == 400 + assert "slice_id" in response.get_json()["message"] + + +class TestAllConsensusHandlersRouteToSliceTracker: + """Every consensus signal handler forwards ``slice_id`` to the tracker lookup. + + The first review (#2402) noted that ``test_propose_routes_to_slice_tracker`` + only pinned the ``propose`` path even though all eight CONSENSUS_* + handlers got the same paste-and-modify treatment. These tests pin the + other seven (ack, nack, withdraw, confirmed, excuse_producer, + resolve_obligation, producer_push) so the tracker-lookup wiring + cannot regress silently. + """ + + @patch("peer_consensus.get_peer_consensus_tracker") + def test_ack_routes_to_slice_tracker(self, mock_get_tracker, app): + from routes.signals import handle_consensus_ack_signal + + mock_tracker = MagicMock() + mock_tracker.handle_ack.return_value = { + "version": 1, + "newly_ready": [], + } + mock_get_tracker.return_value = mock_tracker + + with app.app_context(): + handle_consensus_ack_signal( + "issue-2403", + { + "agent_role": "reviewer_code", + "producer_role": "coder", + "slice_id": "slice-2", + "payload": { + "reason": ( + "Reviewed slice-2 work and the diff matches the " + "proposal summary; tests cover the new path." + ), + }, + }, + Path("/tmp/repo"), + ) + + call = mock_get_tracker.call_args + assert _pipeline_arg_from_call(call) == "issue-2403" + assert _slice_arg_from_call(call) == "slice-2" + + @patch("peer_consensus.get_peer_consensus_tracker") + def test_nack_routes_to_slice_tracker(self, mock_get_tracker, app): + from routes.signals import handle_consensus_nack_signal + + mock_tracker = MagicMock() + mock_tracker.handle_nack.return_value = { + "version": 1, + "reason": "needs more tests", + "revision_count": 1, + } + mock_get_tracker.return_value = mock_tracker + + with app.app_context(): + handle_consensus_nack_signal( + "issue-2403", + { + "agent_role": "reviewer_code", + "producer_role": "coder", + "slice_id": "slice-2", + "payload": { + "reason": ( + "Slice-2 NACK: the diff is missing test coverage " + "for the new branch in the orchestrator's strip site." + ), + }, + }, + Path("/tmp/repo"), + ) + + call = mock_get_tracker.call_args + assert _pipeline_arg_from_call(call) == "issue-2403" + assert _slice_arg_from_call(call) == "slice-2" + + @patch("peer_consensus.get_peer_consensus_tracker") + def test_withdraw_routes_to_slice_tracker(self, mock_get_tracker, app): + from routes.signals import handle_consensus_withdraw_signal + + mock_tracker = MagicMock() + mock_tracker.handle_withdraw.return_value = {"version": 2} + mock_get_tracker.return_value = mock_tracker + + with app.app_context(): + handle_consensus_withdraw_signal( + "issue-2403", + { + "agent_role": "coder", + "slice_id": "slice-2", + "reason": ( + "Withdrawing slice-2 proposal: a reviewer flagged " + "an interaction with the sibling slice integration." + ), + }, + Path("/tmp/repo"), + ) + + call = mock_get_tracker.call_args + assert _pipeline_arg_from_call(call) == "issue-2403" + assert _slice_arg_from_call(call) == "slice-2" + + @patch("routes.signals._write_consensus_confirmed_marker") + @patch("routes.signals._resolve_pipeline_phase", return_value="implement") + @patch("routes.signals._existing_confirmed_for_role", return_value=(False, False)) + @patch("message_store.get_message_store") + @patch("peer_consensus.get_peer_consensus_tracker") + def test_confirmed_routes_to_slice_tracker( + self, + mock_get_tracker, + mock_get_store, + _mock_existing, + _mock_phase, + _mock_marker, + app, + ): + # Mock message_store.get_message_store and + # _existing_confirmed_for_role so the handler's "Final CONFIRMED" + # branch doesn't read or write the live in-memory message store + # (test hermeticity — repeated runs in the same process must not + # observe each other's CONSENSUS_CONFIRMED writes). + # + # The ``message_store.get_message_store`` patch reaches the + # handler because ``signals.py`` imports the symbol *inside* the + # function body (``from message_store import get_message_store`` + # at the call sites). If anyone moves that to a module-level + # ``from message_store import get_message_store`` at the top of + # ``signals.py``, this patch silently stops intercepting and the + # hermeticity guarantee breaks. Switch the patch target to + # ``routes.signals.get_message_store`` if that refactor lands. + from routes.signals import handle_consensus_confirmed_signal + + mock_tracker = MagicMock() + mock_tracker.handle_confirmed.return_value = { + "status": "confirmed", + "version": 1, + } + mock_get_tracker.return_value = mock_tracker + mock_get_store.return_value = MagicMock() + + with app.app_context(): + handle_consensus_confirmed_signal( + "issue-2403", + { + "agent_role": "coder", + "slice_id": "slice-2", + }, + Path("/tmp/repo"), + ) + + call = mock_get_tracker.call_args + assert _pipeline_arg_from_call(call) == "issue-2403" + assert _slice_arg_from_call(call) == "slice-2" + + @patch("peer_consensus.get_peer_consensus_tracker") + @patch("decision_queue.get_decision_queue") + def test_excuse_producer_routes_to_slice_tracker(self, mock_get_queue, mock_get_tracker, app): + from models import DecisionStatus + from routes.signals import handle_consensus_excuse_producer_signal + + # The excuse_producer handler is HITL-gated: short-circuit the + # decision-queue lookup with a RESOLVED decision scoped to the + # producer being excused. + mock_decision = MagicMock() + mock_decision.status = DecisionStatus.RESOLVED + mock_decision.context = "failed_role:coder" + mock_queue = MagicMock() + mock_queue.get_decision.return_value = mock_decision + mock_get_queue.return_value = mock_queue + + mock_tracker = MagicMock() + mock_tracker.excuse_producer.return_value = {"affected_reviewers": []} + mock_get_tracker.return_value = mock_tracker + + with app.app_context(): + handle_consensus_excuse_producer_signal( + "issue-2403", + { + "producer_role": "coder", + "slice_id": "slice-2", + "decision_id": "decision-1", + "reason": "Producer unresponsive after 30m heartbeat gap.", + }, + Path("/tmp/repo"), + ) + + call = mock_get_tracker.call_args + assert _pipeline_arg_from_call(call) == "issue-2403" + assert _slice_arg_from_call(call) == "slice-2" + + @patch("peer_consensus.get_peer_consensus_tracker") + def test_resolve_obligation_routes_to_slice_tracker(self, mock_get_tracker, app): + from routes.signals import handle_consensus_resolve_obligation_signal + + mock_tracker = MagicMock() + mock_tracker.handle_resolve_obligation.return_value = { + "version": 1, + "condition": "add coverage", + } + mock_get_tracker.return_value = mock_tracker + + with app.app_context(): + handle_consensus_resolve_obligation_signal( + "issue-2403", + { + "agent_role": "tester", + "reviewer_role": "reviewer_code", + "producer_role": "coder", + "slice_id": "slice-2", + "commit_sha": "deadbee", + "note": "Cherry-picked test that covers the strip site.", + }, + Path("/tmp/repo"), + ) + + call = mock_get_tracker.call_args + assert _pipeline_arg_from_call(call) == "issue-2403" + assert _slice_arg_from_call(call) == "slice-2" + + @patch("message_store.get_message_store") + @patch("peer_consensus.get_peer_consensus_tracker") + def test_producer_push_routes_to_slice_tracker(self, mock_get_tracker, mock_get_store, app): + # Mock message_store.get_message_store for hermeticity: today the + # handler's auto re-propose branch is gated by ``auto_re_propose: + # False`` so the message-bus write is skipped, but the mock pins + # the test against a future regression where the gate moves. + from routes.signals import handle_consensus_producer_push_signal + + mock_tracker = MagicMock() + mock_tracker.handle_producer_push.return_value = { + "auto_re_propose": False, + "version": 1, + } + mock_get_tracker.return_value = mock_tracker + mock_get_store.return_value = MagicMock() + + with app.app_context(): + handle_consensus_producer_push_signal( + "issue-2403", + { + "agent_role": "coder", + "slice_id": "slice-2", + "commit_sha": "deadbee", + "changed_files": ["src/a.py"], + }, + Path("/tmp/repo"), + ) + + call = mock_get_tracker.call_args + assert _pipeline_arg_from_call(call) == "issue-2403" + assert _slice_arg_from_call(call) == "slice-2" + + +class TestAllConsensusHandlersRejectMalformedSliceId: + """Defense-in-depth: every handler rejects a malformed ``slice_id`` at the boundary. + + The boundary check lives in ``_extract_slice_id`` — if any handler + forgets to call it, a path-separator-bearing ``slice_id`` would + flow through to ``get_peer_consensus_tracker``'s registry key. + These tests pin every handler against that regression. + """ + + def test_ack_rejects_malformed_slice_id(self, app): + from routes.signals import handle_consensus_ack_signal + + with app.app_context(): + response, status = handle_consensus_ack_signal( + "issue-2403", + { + "agent_role": "reviewer_code", + "producer_role": "coder", + "slice_id": "../etc/passwd", + "payload": { + "reason": ( + "Reviewed work; the diff matches the proposal " + "summary and tests cover the new path." + ), + }, + }, + Path("/tmp/repo"), + ) + assert status == 400 + assert "slice_id" in response.get_json()["message"] + + def test_nack_rejects_malformed_slice_id(self, app): + from routes.signals import handle_consensus_nack_signal + + with app.app_context(): + response, status = handle_consensus_nack_signal( + "issue-2403", + { + "agent_role": "reviewer_code", + "producer_role": "coder", + "slice_id": "phase-2", # legacy shape rejected at signal boundary + "payload": { + "reason": ( + "NACK: the diff is missing test coverage for " + "the new branch in the orchestrator strip site." + ), + }, + }, + Path("/tmp/repo"), + ) + assert status == 400 + assert "slice_id" in response.get_json()["message"] + + def test_withdraw_rejects_malformed_slice_id(self, app): + from routes.signals import handle_consensus_withdraw_signal + + with app.app_context(): + response, status = handle_consensus_withdraw_signal( + "issue-2403", + { + "agent_role": "coder", + "slice_id": "slice-2/extra", + "reason": ( + "Withdrawing proposal: a reviewer flagged an " + "interaction with the sibling slice integration." + ), + }, + Path("/tmp/repo"), + ) + assert status == 400 + assert "slice_id" in response.get_json()["message"] + + def test_confirmed_rejects_malformed_slice_id(self, app): + from routes.signals import handle_consensus_confirmed_signal + + with app.app_context(): + response, status = handle_consensus_confirmed_signal( + "issue-2403", + { + "agent_role": "coder", + "slice_id": "../etc", + }, + Path("/tmp/repo"), + ) + assert status == 400 + assert "slice_id" in response.get_json()["message"] + + @patch("decision_queue.get_decision_queue") + def test_excuse_producer_rejects_malformed_slice_id(self, mock_get_queue, app): + from models import DecisionStatus + from routes.signals import handle_consensus_excuse_producer_signal + + # The handler validates the HITL gate before slice_id, so a + # RESOLVED decision still has to short-circuit cleanly to reach + # the slice_id rejection branch. + mock_decision = MagicMock() + mock_decision.status = DecisionStatus.RESOLVED + mock_decision.context = "failed_role:coder" + mock_queue = MagicMock() + mock_queue.get_decision.return_value = mock_decision + mock_get_queue.return_value = mock_queue + + with app.app_context(): + response, status = handle_consensus_excuse_producer_signal( + "issue-2403", + { + "producer_role": "coder", + "slice_id": "../etc", + "decision_id": "decision-1", + "reason": "Producer unresponsive after heartbeat gap.", + }, + Path("/tmp/repo"), + ) + assert status == 400 + assert "slice_id" in response.get_json()["message"] + + def test_resolve_obligation_rejects_malformed_slice_id(self, app): + from routes.signals import handle_consensus_resolve_obligation_signal + + with app.app_context(): + response, status = handle_consensus_resolve_obligation_signal( + "issue-2403", + { + "agent_role": "tester", + "reviewer_role": "reviewer_code", + "producer_role": "coder", + "slice_id": "../etc", + "commit_sha": "deadbee", + "note": "Cherry-picked test that covers the strip site.", + }, + Path("/tmp/repo"), + ) + assert status == 400 + assert "slice_id" in response.get_json()["message"] + + def test_producer_push_rejects_malformed_slice_id(self, app): + from routes.signals import handle_consensus_producer_push_signal + + with app.app_context(): + response, status = handle_consensus_producer_push_signal( + "issue-2403", + { + "agent_role": "coder", + "slice_id": "slice-2/extra", + "commit_sha": "deadbee", + "changed_files": ["src/a.py"], + }, + Path("/tmp/repo"), + ) + assert status == 400 + assert "slice_id" in response.get_json()["message"] + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +import pytest # noqa: E402 + + +@pytest.fixture +def app(): + from flask import Flask + from routes.signals import signals_bp + + app = Flask(__name__) + app.register_blueprint(signals_bp) + return app diff --git a/orchestrator/tests/test_stacked_pr_reconciler.py b/orchestrator/tests/test_stacked_pr_reconciler.py index f62d22ac35..6fe64575f7 100644 --- a/orchestrator/tests/test_stacked_pr_reconciler.py +++ b/orchestrator/tests/test_stacked_pr_reconciler.py @@ -159,8 +159,10 @@ def test_child_with_deleted_base_surfaces_orphan(self) -> None: assert orphan.deleted_base == "egg/issue-2137/slice-1" # Walk-up fallback: the parent's branch is gone (and slice-1 # isn't in the contract here), so the pipeline branch is the - # last-resort target. - assert orphan.intended_new_base == "egg/issue-2137" + # last-resort target. Per #2399 the pipeline tip lives at + # ``egg//work`` so slice integration branches coexist as + # siblings under ``egg//``. + assert orphan.intended_new_base == "egg/issue-2137/work" def test_intended_new_base_walks_up_to_extant_ancestor(self) -> None: # A 3-level chain (slice-1 → slice-2 → slice-3): when @@ -223,7 +225,8 @@ def test_intended_new_base_falls_back_to_pipeline_branch(self) -> None: extant: set[str] = set() orphans = find_orphaned_child_prs(contract, prs, extant) assert len(orphans) == 1 - assert orphans[0].intended_new_base == "egg/issue-2137" + # Pipeline tip lives at ``egg//work`` — see #2399. + assert orphans[0].intended_new_base == "egg/issue-2137/work" def test_intended_new_base_ignores_pr_metadata(self) -> None: # The PR's own ``base`` may have been modified by an out-of- @@ -389,8 +392,9 @@ def test_qualified_pipeline_id_preserves_qualifier_in_issue_branch(self) -> None 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" + # is NOT the unqualified ``egg/issue-2137/work``. Pipeline tip + # lives at ``/work`` (#2399). + assert orphan.intended_new_base == "egg/issue-2137-v3/work" def test_qualified_pipeline_id_walks_up_to_qualified_ancestor(self) -> None: # The walk-up resolver must also use the qualified branch @@ -531,8 +535,9 @@ def test_one_orphan_one_rebase_called(self) -> None: called = rebaser.calls[0] assert called.branch == "egg/issue-2137/slice-2" # Walk-up fallback: slice-1 isn't in the contract here, so - # the pipeline branch is the last-resort rebase target. - assert called.intended_new_base == "egg/issue-2137" + # the pipeline branch is the last-resort rebase target. Pipeline + # tip lives at ``/work`` (#2399). + assert called.intended_new_base == "egg/issue-2137/work" assert called.deleted_base == "egg/issue-2137/slice-1" # The orphan now carries the PR number so the production # bridge can retarget the PR after the rebase. diff --git a/sandbox/egg_agent_tools/handlers/_gateway.py b/sandbox/egg_agent_tools/handlers/_gateway.py index 0f7c121cd5..30039cd1dc 100644 --- a/sandbox/egg_agent_tools/handlers/_gateway.py +++ b/sandbox/egg_agent_tools/handlers/_gateway.py @@ -85,6 +85,18 @@ def get_pipeline_id() -> str | None: return os.environ.get("EGG_PIPELINE_ID") or None +def get_slice_id() -> str | None: + """Slice ID from env (``EGG_SLICE_ID``). + + Set on agents spawned for a per-slice BRC team (#2403). When + present, BRC handlers forward it on the signal payload so the + orchestrator routes ``CONSENSUS_*`` to the slice's tracker + (see ``orchestrator.peer_consensus._tracker_key``). Pipeline-level + agents leave it unset and route to the bare pipeline tracker. + """ + return os.environ.get("EGG_SLICE_ID") or None + + def get_issue_number() -> int | None: """Issue number from env (``EGG_ISSUE_NUMBER``).""" raw = os.environ.get("EGG_ISSUE_NUMBER") diff --git a/sandbox/egg_agent_tools/handlers/brc.py b/sandbox/egg_agent_tools/handlers/brc.py index 7bda29ab29..975dde24c7 100644 --- a/sandbox/egg_agent_tools/handlers/brc.py +++ b/sandbox/egg_agent_tools/handlers/brc.py @@ -13,12 +13,32 @@ from egg_agent_tools.handlers._gateway import ( get_agent_role, get_pipeline_id, + get_slice_id, orchestrator_request, ) from egg_agent_tools.handlers.errors import GatewayError, HandlerError _COMMIT_SHA_PATTERN = re.compile(r"^[0-9a-fA-F]{7,40}$") _PIPELINE_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$") +_SLICE_ID_PATTERN = re.compile(r"^slice-[0-9]+$") + + +def _maybe_attach_slice_id(req: dict[str, Any], data: dict[str, Any]) -> None: + """Forward ``slice_id`` from the request or env onto the signal body. + + Per-slice agents set ``EGG_SLICE_ID`` so the orchestrator can route + their ``CONSENSUS_*`` to the slice tracker instead of the bare + pipeline tracker (#2403). Callers can also pass ``slice_id`` on + ``req`` to override (e.g. tests, or operator tooling acting on a + specific slice). Validation mirrors the orchestrator side so a + malformed value can't smuggle path separators into a tracker key. + """ + slice_id = req.get("slice_id") or get_slice_id() + if not slice_id: + return + if not isinstance(slice_id, str) or not _SLICE_ID_PATTERN.fullmatch(slice_id): + raise HandlerError(f"Invalid slice_id {slice_id!r}: must match 'slice-'") + data["slice_id"] = slice_id def _validate_commit_sha(sha: str) -> str: @@ -421,6 +441,7 @@ def brc_propose(req: dict[str, Any]) -> dict[str, Any]: } if req.get("changed_artifacts"): data["changed_artifacts"] = list(req["changed_artifacts"]) + _maybe_attach_slice_id(req, data) try: result = orchestrator_request(f"/api/v1/pipelines/{pid}/signal", method="POST", data=data) @@ -505,6 +526,7 @@ def brc_ack(req: dict[str, Any]) -> dict[str, Any]: "producer_role": producer_role, "payload": payload, } + _maybe_attach_slice_id(req, data) try: result = orchestrator_request(f"/api/v1/pipelines/{pid}/signal", method="POST", data=data) except GatewayError as exc: @@ -561,6 +583,7 @@ def brc_nack(req: dict[str, Any]) -> dict[str, Any]: "nack_version": nack_version, }, } + _maybe_attach_slice_id(req, data) try: result = orchestrator_request(f"/api/v1/pipelines/{pid}/signal", method="POST", data=data) except GatewayError as exc: @@ -604,10 +627,11 @@ def brc_confirm(req: dict[str, Any]) -> dict[str, Any]: pid = _require_pipeline_id(req) role = _require_role(req) - data = { + data: dict[str, Any] = { "signal_type": "consensus_confirmed", "agent_role": role, } + _maybe_attach_slice_id(req, data) result = orchestrator_request(f"/api/v1/pipelines/{pid}/signal", method="POST", data=data) if not result.get("success"): raise GatewayError(result.get("message", "confirm failed")) @@ -727,6 +751,7 @@ def brc_resolve_obligation(req: dict[str, Any]) -> dict[str, Any]: data["commit_sha"] = commit_sha if note: data["note"] = note + _maybe_attach_slice_id(req, data) result = orchestrator_request(f"/api/v1/pipelines/{pid}/signal", method="POST", data=data) if not result.get("success"): diff --git a/sandbox/egg_lib/orch_cli.py b/sandbox/egg_lib/orch_cli.py index 7573356797..fdb37968f3 100755 --- a/sandbox/egg_lib/orch_cli.py +++ b/sandbox/egg_lib/orch_cli.py @@ -146,6 +146,16 @@ def get_pipeline_id_from_env() -> str | None: return os.environ.get("EGG_PIPELINE_ID") +def get_slice_id_from_env() -> str | None: + """Get slice ID from environment (``EGG_SLICE_ID``) if set. + + Set on agents spawned for a per-slice BRC team (#2403). When + present, consensus signal commands forward it on the request body + so the orchestrator routes ``CONSENSUS_*`` to the slice's tracker. + """ + return os.environ.get("EGG_SLICE_ID") or None + + def get_agent_role_from_env() -> str | None: """Get agent role from environment if set.""" return os.environ.get("EGG_AGENT_ROLE") @@ -2533,6 +2543,9 @@ def cmd_consensus_withdraw(args: argparse.Namespace) -> int: "agent_role": role, "reason": args.reason, } + slice_id = get_slice_id_from_env() + if slice_id: + data["slice_id"] = slice_id result = orch_request(f"/api/v1/pipelines/{pid}/signal", method="POST", data=data) diff --git a/sandbox/tests/test_brc_slice_routing.py b/sandbox/tests/test_brc_slice_routing.py new file mode 100644 index 0000000000..dc060d3eca --- /dev/null +++ b/sandbox/tests/test_brc_slice_routing.py @@ -0,0 +1,210 @@ +"""BRC handlers thread ``slice_id`` from ``EGG_SLICE_ID`` onto signals (#2403). + +Per-slice agents must tag every consensus signal with their ``slice_id`` +so the orchestrator routes ``CONSENSUS_*`` to the slice's tracker. The +spawn path sets ``EGG_SLICE_ID`` (and leaves ``EGG_PIPELINE_ID`` as the +bare pipeline id); the handlers in ``egg_agent_tools.handlers.brc`` are +the agent-side end of that contract. +""" + +import sys +from pathlib import Path +from typing import Any +from unittest.mock import patch + +import pytest + +# Add sandbox to sys.path so egg_agent_tools is importable +_sandbox_path = str(Path(__file__).parent.parent) +if _sandbox_path not in sys.path: + sys.path.insert(0, _sandbox_path) + + +_PROPOSE_REQ = { + "pipeline_id": "issue-2403", + "role": "coder", + "summary": ( + "Implemented slice-2 work with substantive commit message " + "well over the fifty-character validator threshold" + ), + "artifacts": ["src/a.py"], + "tests_run": [], + "tasks": [], + "attestation": {}, +} + +_ACK_REQ = { + "pipeline_id": "issue-2403", + "role": "reviewer_code", + "producer_role": "coder", + "reason": "Reviewed src/a.py: substantive multi-file review well over fifty chars", + "files_reviewed": ["src/a.py"], + "ack_version": 1, +} + +_NACK_REQ = { + "pipeline_id": "issue-2403", + "role": "reviewer_code", + "producer_role": "coder", + "reason": "src/a.py:42 raises on empty input — substantive blocker over fifty chars", + "files_reviewed": ["src/a.py"], + "nack_version": 1, +} + +_CONFIRM_REQ = {"pipeline_id": "issue-2403", "role": "coder"} + +_RESOLVE_REQ = { + "pipeline_id": "issue-2403", + "role": "tester", + "reviewer_role": "reviewer_code", + "producer_role": "coder", + "note": "git mv old/path new/path satisfied in-cycle", +} + + +def _captured_data(mock_request: Any) -> dict[str, Any]: + assert mock_request.called, "orchestrator_request was not invoked" + return dict(mock_request.call_args.kwargs["data"]) + + +class TestSliceIdAttachedFromEnv: + """``EGG_SLICE_ID`` flows onto every CONSENSUS_* signal body.""" + + @pytest.fixture(autouse=True) + def _set_slice_env(self, monkeypatch): + monkeypatch.setenv("EGG_SLICE_ID", "slice-2") + + def test_propose_attaches_slice_id(self): + from egg_agent_tools.handlers import brc as handlers + + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value={"success": True, "data": {"consensus": {"agents": {}}}}, + ) as mock_request: + handlers.brc_propose(dict(_PROPOSE_REQ)) + assert _captured_data(mock_request)["slice_id"] == "slice-2" + + def test_ack_attaches_slice_id(self): + from egg_agent_tools.handlers import brc as handlers + + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value={"success": True, "data": {}}, + ) as mock_request: + handlers.brc_ack(dict(_ACK_REQ)) + assert _captured_data(mock_request)["slice_id"] == "slice-2" + + def test_nack_attaches_slice_id(self): + from egg_agent_tools.handlers import brc as handlers + + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value={"success": True, "data": {}}, + ) as mock_request: + handlers.brc_nack(dict(_NACK_REQ)) + assert _captured_data(mock_request)["slice_id"] == "slice-2" + + def test_confirm_attaches_slice_id(self): + from egg_agent_tools.handlers import brc as handlers + + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value={"success": True, "data": {"status": "confirmed"}}, + ) as mock_request: + handlers.brc_confirm(dict(_CONFIRM_REQ)) + assert _captured_data(mock_request)["slice_id"] == "slice-2" + + def test_resolve_obligation_attaches_slice_id(self): + from egg_agent_tools.handlers import brc as handlers + + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value={"success": True, "data": {}}, + ) as mock_request: + handlers.brc_resolve_obligation(dict(_RESOLVE_REQ)) + assert _captured_data(mock_request)["slice_id"] == "slice-2" + + +class TestSliceIdAbsentWhenEnvUnset: + """Pipeline-level agents (no ``EGG_SLICE_ID``) send no ``slice_id``.""" + + @pytest.fixture(autouse=True) + def _no_slice_env(self, monkeypatch): + monkeypatch.delenv("EGG_SLICE_ID", raising=False) + + def test_propose_omits_slice_id(self): + from egg_agent_tools.handlers import brc as handlers + + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value={"success": True, "data": {"consensus": {"agents": {}}}}, + ) as mock_request: + handlers.brc_propose(dict(_PROPOSE_REQ)) + assert "slice_id" not in _captured_data(mock_request) + + def test_ack_omits_slice_id(self): + from egg_agent_tools.handlers import brc as handlers + + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value={"success": True, "data": {}}, + ) as mock_request: + handlers.brc_ack(dict(_ACK_REQ)) + assert "slice_id" not in _captured_data(mock_request) + + +class TestSliceIdReqOverridesEnv: + """A caller-supplied ``slice_id`` on the request takes precedence.""" + + def test_req_slice_id_wins_over_env(self, monkeypatch): + from egg_agent_tools.handlers import brc as handlers + + monkeypatch.setenv("EGG_SLICE_ID", "slice-9") + + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value={"success": True, "data": {"consensus": {"agents": {}}}}, + ) as mock_request: + handlers.brc_propose({**_PROPOSE_REQ, "slice_id": "slice-3"}) + assert _captured_data(mock_request)["slice_id"] == "slice-3" + + +class TestSliceIdValidation: + """Defense-in-depth: malformed ``slice_id`` is rejected before the wire.""" + + def test_invalid_slice_id_raises(self, monkeypatch): + from egg_agent_tools.handlers import brc as handlers + from egg_agent_tools.handlers.errors import HandlerError + + # Anything other than ``slice-`` must be rejected — a trailing + # path component would corrupt the orchestrator's tracker key. + monkeypatch.setenv("EGG_SLICE_ID", "slice-2/../etc") + + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value={"success": True, "data": {"consensus": {"agents": {}}}}, + ): + with pytest.raises(HandlerError, match="slice_id"): + handlers.brc_propose(dict(_PROPOSE_REQ)) + + +class TestSliceIdHelper: + """``get_slice_id`` reads ``EGG_SLICE_ID`` and returns None when unset.""" + + def test_returns_value_when_set(self, monkeypatch): + from egg_agent_tools.handlers._gateway import get_slice_id + + monkeypatch.setenv("EGG_SLICE_ID", "slice-7") + assert get_slice_id() == "slice-7" + + def test_returns_none_when_unset(self, monkeypatch): + from egg_agent_tools.handlers._gateway import get_slice_id + + monkeypatch.delenv("EGG_SLICE_ID", raising=False) + assert get_slice_id() is None + + def test_returns_none_on_empty_string(self, monkeypatch): + from egg_agent_tools.handlers._gateway import get_slice_id + + monkeypatch.setenv("EGG_SLICE_ID", "") + assert get_slice_id() is None diff --git a/scripts/file-size-allowlist.yaml b/scripts/file-size-allowlist.yaml index e46b91af88..ab0029a198 100644 --- a/scripts/file-size-allowlist.yaml +++ b/scripts/file-size-allowlist.yaml @@ -48,3 +48,5 @@ files: issue: "2248" orchestrator/routes/deployment.py: issue: "2248" + orchestrator/kubernetes_spawner.py: + issue: "2248"