diff --git a/.egg/phase-permissions.json b/.egg/phase-permissions.json index 94f990b07b..734ea1bf64 100644 --- a/.egg/phase-permissions.json +++ b/.egg/phase-permissions.json @@ -116,32 +116,6 @@ } ], "exit_requires": "reviewer" - }, - "pr": { - "allowed_operations": [ - { - "type": "gh", - "pattern": "pr create*", - "description": "Create pull requests" - }, - { - "type": "gh", - "pattern": "pr edit *", - "description": "Edit pull requests" - }, - { - "type": "git", - "pattern": "push *", - "description": "Push code to remote" - }, - { - "type": "egg-contract", - "pattern": "show *", - "description": "View contract state" - } - ], - "blocked_operations": [], - "exit_requires": "human" } }, "phase_file_restrictions": { @@ -175,10 +149,6 @@ ".egg-state/reviews/*" ], "description": "Implement phase can push code but not .egg-state files (except checkpoints)" - }, - "pr": { - "allowed_patterns": ["*"], - "description": "PR phase can push everything" } } } diff --git a/docs/reference/agent-tools.md b/docs/reference/agent-tools.md index 0f780caa69..6410729132 100644 --- a/docs/reference/agent-tools.md +++ b/docs/reference/agent-tools.md @@ -95,7 +95,7 @@ that requires the handler docstring to explain why no CLI exists. | `mcp__sdlc__check_hitl_answers` | Return resolved decisions and feedback (submitted or pending) for the current contract. Without a `phase` arg, returns HITL across all phases; pass `phase` to narrow to a single phase. | `handlers.sdlc.check_hitl_answers` | — *(no CLI; new capability)* | | `mcp__sdlc__show_contract` | Read the current contract as a dict. Optional `fields=[…]` projection returns only the named top-level keys; an unknown field raises `HandlerError` (no silent skip). State-machine effect: **read-only**. | `handlers.sdlc.show_contract` | `egg-contract show` | | `mcp__sdlc__verify_criterion` | Mark an acceptance criterion verified on the contract. **REVIEWER role only** — the gateway rejects non-REVIEWER writers; the handler does not re-check (decision-7). State-machine effect: marks the criterion verified; no-op if already verified. | `handlers.sdlc.verify_criterion` | `egg-contract verify-criterion` | -| `mcp__sdlc__check_file_restriction` | Pure-local read against `shared/egg_restrictions/patterns.py`: returns `can_write` + `alternative_role` for a path or list of paths. Producers call this before exploring a file outside their role boundary (#2529). Read-only; no gateway round-trip. | `handlers.restrictions.check_file_restriction` | — *(no CLI; pattern matching is pure CPU and the registry ships in the sandbox image — a CLI shim would just re-import the same module)* | +| `mcp__sdlc__check_file_restriction` | Pure-local read against **both** gateway push gates: the role layer (`shared/egg_restrictions/patterns.py`) and the phase layer (`shared/egg_restrictions/phase_patterns.py`, mirror of `gateway/phase_filter.py`). `can_write` is their conjunction — it predicts push acceptance — and the split verdicts (`role_can_write`, `phase_allows`, `blocked_by`, `phase`) show which gate fires. A phase-layer block (e.g. `refiner` writing `.egg-state/drafts/*-plan.md` in the refine phase, reserved to plan) is a real gateway block, not a false claim, and carries no `alternative_role` (#2968). `role`/`phase` default to `EGG_AGENT_ROLE`/`EGG_PHASE`; an unset phase makes the phase layer a no-op (role-only, pre-#2968 behavior). **When reviewing another agent's proposal, pass `role` and `phase` explicitly** (e.g. `role="coder"`, `phase="implement"`) — the defaults give the verdict for the reviewer's *own* role/phase, not the producer's, so a reviewer's default-args check will diverge from what the gateway would have done to the producer. Producers call this before exploring a file outside their boundary (#2529). Read-only; no gateway round-trip. | `handlers.restrictions.check_file_restriction` | — *(no CLI; pattern matching is pure CPU and both pattern sets ship in the sandbox image — a CLI shim would just re-import the same modules)* | | `mcp__sdlc__report_impasse` | Persist a typed `Impasse` (category, reason, suggested_role, blocked_files, evidence, task_id) under `AgentOutput.impasse` (#2529). For `category=wrong_role`, `task_id` and `suggested_role` are **mandatory** — the handler raises `HandlerError` if either is missing, since the orchestrator's auto-delegation path needs both to rewire `task.role` unambiguously (no role-match fallback when a slice has multiple tasks per role). For other categories (`plan_bug`, `external_blocker`, `unknown`), both fields stay optional — those always escalate to HITL. The orchestrator reads the impasse post-phase and either auto-delegates to `suggested_role` (first attempt, `wrong_role` only) or escalates to HITL (second attempt or non-`wrong_role`). State-machine effect: **the agent must exit cleanly without committing after this returns**. | `handlers.restrictions.report_impasse` | — *(no CLI; structured runtime signal that lives inside agent-output JSON — a parallel CLI write path would just risk drift with the MCP one)* | ### `mcp__brc__*` — Broadcast-Review-Converge consensus diff --git a/gateway/tests/test_phase_filter_restrictions.py b/gateway/tests/test_phase_filter_restrictions.py index 02e1d0f556..a42417e534 100644 --- a/gateway/tests/test_phase_filter_restrictions.py +++ b/gateway/tests/test_phase_filter_restrictions.py @@ -216,6 +216,19 @@ def test_refine_allows_analysis_drafts(self): result = pf.check_phase_file_restrictions("refine", [".egg-state/drafts/644-analysis.md"]) assert result.allowed is True + def test_refine_blocks_plan_drafts(self): + # The inverse of the above and the crux of #2968: plan drafts are + # reserved to the plan phase, so refine rejects them even though + # every drafting role's pattern allows .egg-state/drafts/. + pf = PhaseFilter() + result = pf.check_phase_file_restrictions("refine", [".egg-state/drafts/644-plan.md"]) + assert result.allowed is False + + def test_plan_blocks_analysis_drafts(self): + pf = PhaseFilter() + result = pf.check_phase_file_restrictions("plan", [".egg-state/drafts/644-analysis.md"]) + assert result.allowed is False + def test_refine_allows_checkpoints(self): pf = PhaseFilter() result = pf.check_phase_file_restrictions("refine", [".egg-state/checkpoints/ckpt.json"]) @@ -698,3 +711,77 @@ def test_documenter_allowed_for_readme(self): # gateway, so the coarse entry never matched a real session_role in # production. The block on ``.egg-state/contracts/`` is preserved # transitively via the fine-grained role tests above. + + +class TestPhaseLayerSharedMirrorParity: + """#2968: the sandbox-side phase mirror (shared/egg_restrictions/ + phase_patterns.py) must agree with the live gateway phase gate for + every phase. This is the drift guard that lets the phase-blind + check_file_restriction MCP tool trust its own can_write.""" + + @pytest.fixture(autouse=True) + def reset_filter(self): + reset_phase_filter() + yield + reset_phase_filter() + + # Representative paths spanning every .egg-state/ subdir the configs key + # on, plus a code path and a traversal attempt. + _BATTERY = [ + ".egg-state/drafts/p-analysis.md", + ".egg-state/drafts/p-plan.md", + ".egg-state/contracts/p.json", + ".egg-state/reviews/r.json", + ".egg-state/checkpoints/c.json", + ".egg-state/agent-outputs/o.json", + ".egg-state/agent-anchors/a.json", + ".egg-state/pipelines/p.json", + "src/app.py", + "docs/guide.md", + "../escape.txt", + ] + + def test_shared_mirror_matches_gateway_for_every_phase(self): + from egg_restrictions.phase_patterns import phase_file_verdict + + pf = PhaseFilter() # loads the real .egg/phase-permissions.json + mismatches = [] + for phase in PipelinePhase: + for path in self._BATTERY: + gateway_allowed = pf.check_phase_file_restrictions(phase, [path]).allowed + mirror_allowed = phase_file_verdict(phase.value, path)[0] + if gateway_allowed != mirror_allowed: + mismatches.append( + f"{phase.value}:{path} gateway={gateway_allowed} mirror={mirror_allowed}" + ) + assert not mismatches, ( + "shared phase_patterns drifted from gateway/phase_filter.py — " + "update PHASE_FILE_PATTERNS to match the live config: " + "; ".join(mismatches) + ) + + # Off-enum strings: anything outside ``PipelinePhase``'s canonical + # lowercase set. In production the orchestrator always exports a + # canonical ``EGG_PHASE`` so this path only fires on a manual / + # test caller, but the gateway fails closed via the + # ``PipelinePhase(phase)`` coercion and the mirror has to match — + # otherwise a phase-blind caller could see ``can_write: true`` for a + # path the gateway will reject. + _OFF_ENUM_PHASES = ("IMPLEMENT", "unknown", "pr", "REFINE") + + def test_shared_mirror_fails_closed_for_off_canonical_phase(self): + from egg_restrictions.phase_patterns import phase_file_verdict + + pf = PhaseFilter() + for bad_phase in self._OFF_ENUM_PHASES: + for path in (".egg-state/drafts/p-plan.md", "src/app.py"): + gateway_allowed = pf.check_phase_file_restrictions(bad_phase, [path]).allowed + mirror_allowed = phase_file_verdict(bad_phase, path)[0] + assert gateway_allowed is False, ( + f"gateway should fail closed for off-enum phase " + f"{bad_phase!r}, got allowed={gateway_allowed}" + ) + assert mirror_allowed is False, ( + f"mirror should fail closed for off-enum phase " + f"{bad_phase!r} to match the gateway, got " + f"allowed={mirror_allowed}" + ) diff --git a/sandbox/egg_agent_tools/handlers/restrictions.py b/sandbox/egg_agent_tools/handlers/restrictions.py index 715341b747..75ecbaa00e 100644 --- a/sandbox/egg_agent_tools/handlers/restrictions.py +++ b/sandbox/egg_agent_tools/handlers/restrictions.py @@ -3,9 +3,12 @@ Two cheap handlers an agent calls when its assigned task looks structurally impossible: -- ``check_file_restriction(req)`` — pure local read against - ``shared/egg_restrictions/patterns.py``. No gateway round-trip; the - pattern registry is statically resolvable inside the sandbox image. +- ``check_file_restriction(req)`` — pure local read against both + ``shared/egg_restrictions/patterns.py`` (role layer) and + ``shared/egg_restrictions/phase_patterns.py`` (phase layer, mirroring + ``gateway/phase_filter.py``), so its ``can_write`` predicts what the + gateway will actually accept on push (#2968). No gateway round-trip; + both pattern sets are statically resolvable inside the sandbox image. - ``report_impasse(req)`` — persists a typed :class:`egg_contracts.Impasse` under ``AgentOutput.impasse`` (the same JSON file used for ``handoff_data`` today). The orchestrator @@ -29,6 +32,7 @@ from egg_agent_tools.handlers._gateway import ( get_agent_role, get_contract_identifier, + get_phase, get_repo_path, ) from egg_agent_tools.handlers.errors import HandlerError @@ -44,6 +48,14 @@ def _load_pattern_registry() -> dict[str, Any]: return AGENT_PATTERNS +def _load_phase_verdict() -> Any: + """Lazily import the phase-layer evaluator (same rationale as + :func:`_load_pattern_registry`).""" + from egg_restrictions.phase_patterns import phase_file_verdict + + return phase_file_verdict + + def _alternative_role(blocked_role: str, file_path: str) -> str | None: """Return the single producer role that *can* write ``file_path`` if one exists, else ``None``. @@ -68,43 +80,68 @@ def _alternative_role(blocked_role: str, file_path: str) -> str | None: def check_file_restriction(req: dict[str, Any]) -> dict[str, Any]: - """Check whether the named role can write the named path(s). + """Check whether the named role can write the named path(s) *in a phase*. + + The gateway gates every push on **two** independent layers, both of + which must allow a file (#2968): - Pure read against the pattern registry — does not mutate state and - does not call the gateway. Used by the agent before deciding to - explore a file or hand off the task. + - **Role layer** — ``shared/egg_restrictions/patterns.py``: can this + role ever write the path? + - **Phase layer** — ``shared/egg_restrictions/phase_patterns.py`` + (mirror of ``gateway/phase_filter.py``): is the path writable in + the current pipeline phase, regardless of role? E.g. the *refine* + phase rejects ``.egg-state/drafts/*-plan.md`` (reserved to *plan*) + even though every drafting role's pattern allows it. - No CLI counterpart: pattern matching is pure CPU and the registry - ships in the sandbox image; a CLI shim would just shell out to - re-import the same module. Decision-13 rationale. + ``can_write`` is the **conjunction**: a push of ``path`` in ``phase`` + by ``role`` will be accepted only if both layers allow it. The split + verdicts (``role_can_write`` / ``phase_allows``) and ``blocked_by`` + are surfaced so producers and reviewers can see *which* gate fires — + a phase-gate block is a true gateway block, not a false agent claim. + + Pure read — does not mutate state and does not call the gateway; both + pattern sets ship in the sandbox image. No CLI counterpart + (decision-13 rationale). Request: path (str | list[str]): a single path or a list. Required. role (str): role to check. Defaults to ``EGG_AGENT_ROLE``. + phase (str): pipeline phase to evaluate the phase layer against. + Defaults to ``EGG_PHASE``. When unset/unknown, the phase + layer is a no-op and ``can_write`` reduces to the role check + (backward-compatible with the pre-#2968 behaviour). Response (single path): { ok: True, - role: "coder", - path: "tests/test_x.py", + role: "refiner", + path: ".egg-state/drafts/p-plan.md", + phase: "refine", can_write: False, - reason: "matches blocked pattern '**/test_*.py'", - alternative_role: "tester", + role_can_write: True, + phase_allows: False, + blocked_by: "phase", + reason: "phase 'refine' blocks ... (gateway/phase_filter.py) ...", + alternative_role: None, } Response (list of paths): { ok: True, - role: "coder", + role: "refiner", + phase: "refine", results: [ - {path, can_write, reason, alternative_role}, ... + {path, can_write, role_can_write, phase_allows, + blocked_by, reason, alternative_role}, ... ], } - ``alternative_role`` is populated only when exactly one producer - role (other than the queried one) can write the path. Multi-role - or no-role coverage returns ``None`` — the agent should treat that - as "ask for HITL", not "guess a role". + ``alternative_role`` is populated only for **role**-layer blocks, and + only when exactly one producer role (other than the queried one) can + write the path. A phase-layer block has no alternative role — the + path is reserved to a different phase for everyone — so it stays + ``None``; the agent should defer the write to the owning phase rather + than hand off. """ raw_path = req.get("path") if raw_path is None: @@ -119,24 +156,58 @@ def check_file_restriction(req: dict[str, Any]) -> dict[str, Any]: if pattern is None: raise HandlerError(f"Unknown role {role!r}. Known roles: {sorted(registry.keys())}") + phase = req.get("phase") or get_phase() + if phase is not None and not isinstance(phase, str): + # Defensive: the schema declares ``phase`` a string, but the rest of + # the handler is type-checking every other input — keep the boundary + # symmetric so a malformed caller gets a structured HandlerError + # instead of an AttributeError out of ``PipelinePhase(phase)``. + raise HandlerError("'phase' must be a string when provided") + phase_file_verdict = _load_phase_verdict() + def _check_one(path: str) -> dict[str, Any]: - can_write = pattern.can_write(path) - if can_write: - return { - "path": path, - "can_write": True, - "reason": "matches an allowed pattern", - "alternative_role": None, - } - return { + role_can_write = pattern.can_write(path) + phase_allows, phase_block_reason = phase_file_verdict(phase, path) + can_write = role_can_write and phase_allows + + result: dict[str, Any] = { "path": path, - "can_write": False, - "reason": ( - f"role {role!r} is blocked from {path!r} by shared/egg_restrictions/patterns.py" - ), - "alternative_role": _alternative_role(role, path), + "can_write": can_write, + "role_can_write": role_can_write, + "phase_allows": phase_allows, + "phase": phase, + "blocked_by": None, + "alternative_role": None, + "reason": "", } + if not role_can_write: + # Role-layer block takes priority in the message: it's the + # one that may be delegable to another producer role. + result["blocked_by"] = "role" + result["alternative_role"] = _alternative_role(role, path) + result["reason"] = ( + f"role {role!r} is blocked from {path!r} by shared/egg_restrictions/patterns.py" + ) + elif not phase_allows: + result["blocked_by"] = "phase" + result["reason"] = ( + f"phase {phase!r} blocks {path!r} at the gateway phase gate " + f"(gateway/phase_filter.py): {phase_block_reason}. The role pattern " + f"alone would allow it, but this path is reserved to another phase, " + f"so a push in this phase is rejected regardless of role. Defer the " + f"write to the owning phase rather than handing off." + ) + elif phase: + result["reason"] = ( + f"matches an allowed role pattern and is permitted in phase {phase!r}" + ) + else: + # No phase context — role-only verdict (pre-#2968 wording). + result["reason"] = "matches an allowed pattern" + + return result + if isinstance(raw_path, list): if not raw_path: raise HandlerError("'path' list cannot be empty") @@ -145,7 +216,7 @@ def _check_one(path: str) -> dict[str, Any]: if not isinstance(entry, str) or not entry: raise HandlerError("'path' list entries must be non-empty strings") results.append(_check_one(entry)) - return {"ok": True, "role": role, "results": results} + return {"ok": True, "role": role, "phase": phase, "results": results} if not isinstance(raw_path, str) or not raw_path: raise HandlerError("'path' must be a non-empty string or list") diff --git a/sandbox/egg_agent_tools/tools/sdlc.py b/sandbox/egg_agent_tools/tools/sdlc.py index 837271ec62..ca71c52604 100644 --- a/sandbox/egg_agent_tools/tools/sdlc.py +++ b/sandbox/egg_agent_tools/tools/sdlc.py @@ -119,9 +119,11 @@ {"type": "array", "items": {"type": "string"}, "minItems": 1}, ], "description": ( - "Path (or list of paths) to check against the role's " - "file-write restrictions in shared/egg_restrictions/" - "patterns.py." + "Path (or list of paths) to check against BOTH gateway push " + "gates: the role layer (shared/egg_restrictions/patterns.py) " + "AND the phase layer (gateway/phase_filter.py, configured by " + ".egg/phase-permissions.json and mirrored in shared/" + "egg_restrictions/phase_patterns.py)." ), }, "role": { @@ -131,6 +133,16 @@ "left unset so the agent checks itself." ), }, + "phase": { + "type": "string", + "description": ( + "Pipeline phase to evaluate the phase-layer gate against " + "(defaults to EGG_PHASE). Leave unset to check your own " + "phase. The phase gate can block a path your role pattern " + "allows (e.g. refine cannot push *-plan.md); pass a phase " + "explicitly to ask 'would this be writable in phase X?'." + ), + }, }, "required": ["path"], } @@ -245,12 +257,21 @@ async def verify_criterion(args: dict[str, Any]) -> dict[str, Any]: @tool( "check_file_restriction", - "Check whether the named role can write the named path(s) per " - "shared/egg_restrictions/patterns.py. Read-only; no gateway round-trip. " - "Use this BEFORE exploring a file you suspect is outside your role's " - "boundary so you can hand off cleanly instead of building a workaround. " - "Returns can_write + alternative_role (the role that *can* write the " - "path, when exactly one producer role covers it).", + "Check whether the named role can write the named path(s) in the current " + "phase. Read-only; no gateway round-trip. can_write reflects BOTH gateway " + "push gates: the role layer (shared/egg_restrictions/patterns.py) AND the " + "phase layer (gateway/phase_filter.py) — so it predicts push acceptance. " + "Use this BEFORE exploring a file you suspect is outside your boundary so " + "you can hand off or defer cleanly instead of building a workaround. " + "Returns can_write plus split verdicts (role_can_write, phase_allows, " + "blocked_by) and alternative_role (set only for role-layer blocks, when " + "exactly one producer role covers the path). A phase-layer block is a real " + "gateway block, not a false claim — defer the write to the owning phase. " + "When reviewing another agent's proposal, pass `role` and `phase` " + "explicitly (e.g. role='coder', phase='implement') — the defaults read " + "EGG_AGENT_ROLE/EGG_PHASE, which give you the verdict for *your own* role/" + "phase, not the producer's. Without explicit args a reviewer's check will " + "diverge from what the gateway would have done to the producer.", _CHECK_FILE_RESTRICTION_SCHEMA, ) async def check_file_restriction(args: dict[str, Any]) -> dict[str, Any]: diff --git a/sandbox/tests/test_restrictions_handlers.py b/sandbox/tests/test_restrictions_handlers.py index 259c37df72..55f1681982 100644 --- a/sandbox/tests/test_restrictions_handlers.py +++ b/sandbox/tests/test_restrictions_handlers.py @@ -33,15 +33,22 @@ def _set_role(monkeypatch): monkeypatch.setenv("EGG_AGENT_ROLE", "coder") monkeypatch.delenv("EGG_PIPELINE_ID", raising=False) monkeypatch.delenv("EGG_ISSUE_NUMBER", raising=False) + # No phase by default: the phase layer is a no-op so these role-layer + # assertions stay deterministic regardless of the ambient EGG_PHASE. + monkeypatch.delenv("EGG_PHASE", raising=False) class TestCheckFileRestriction: def test_blocked_path_for_coder(self): - out = restrictions.check_file_restriction({"path": "tests/test_x.py"}) + # docs/ is documenter-owned: coder is blocked and documenter is the + # sole producer alternative. (Used tests/test_x.py until #2936 let + # the coder author its own tests, which made that path coder-writable + # and left this assertion stale.) + out = restrictions.check_file_restriction({"path": "docs/guide.md"}) assert out["ok"] is True assert out["role"] == "coder" assert out["can_write"] is False - assert out["alternative_role"] == "tester" + assert out["alternative_role"] == "documenter" assert "blocked" in out["reason"] def test_allowed_path_for_coder(self): @@ -58,12 +65,14 @@ def test_github_path_no_alternative(self): assert out["alternative_role"] is None def test_batch_form(self): - out = restrictions.check_file_restriction({"path": ["tests/test_x.py", "src/app.py"]}) + # docs/guide.md blocked (documenter alt), src/app.py allowed. (Was + # tests/test_x.py until #2936 — see test_blocked_path_for_coder.) + out = restrictions.check_file_restriction({"path": ["docs/guide.md", "src/app.py"]}) assert out["ok"] is True assert len(out["results"]) == 2 blocked = [r for r in out["results"] if not r["can_write"]] allowed = [r for r in out["results"] if r["can_write"]] - assert len(blocked) == 1 and blocked[0]["alternative_role"] == "tester" + assert len(blocked) == 1 and blocked[0]["alternative_role"] == "documenter" assert len(allowed) == 1 def test_explicit_role_override(self): @@ -90,6 +99,123 @@ def test_no_role_no_env_raises(self, monkeypatch): restrictions.check_file_restriction({"path": "x.py"}) +class TestCheckFileRestrictionPhase: + """The phase layer (#2968): can_write must also reflect the gateway's + phase gate, not just the role pattern.""" + + _PLAN_DRAFT = ".egg-state/drafts/pipeline-8cf1f000-plan.md" + _ANALYSIS_DRAFT = ".egg-state/drafts/pipeline-8cf1f000-analysis.md" + + def test_refiner_plan_draft_blocked_by_phase_in_refine(self, monkeypatch): + # The #2968 case: refiner's role pattern allows .egg-state/drafts/, + # but the refine phase gate reserves *-plan.md to the plan phase. + monkeypatch.setenv("EGG_AGENT_ROLE", "refiner") + monkeypatch.setenv("EGG_PHASE", "refine") + out = restrictions.check_file_restriction({"path": self._PLAN_DRAFT}) + assert out["can_write"] is False + assert out["role_can_write"] is True + assert out["phase_allows"] is False + assert out["blocked_by"] == "phase" + assert out["phase"] == "refine" + # No alternative role for a phase block — it's reserved phase-wide. + assert out["alternative_role"] is None + assert "phase_filter.py" in out["reason"] + + def test_refiner_plan_draft_writable_in_plan_phase(self, monkeypatch): + monkeypatch.setenv("EGG_AGENT_ROLE", "refiner") + monkeypatch.setenv("EGG_PHASE", "plan") + out = restrictions.check_file_restriction({"path": self._PLAN_DRAFT}) + assert out["can_write"] is True + assert out["blocked_by"] is None + + def test_refiner_analysis_draft_writable_in_refine(self, monkeypatch): + monkeypatch.setenv("EGG_AGENT_ROLE", "refiner") + monkeypatch.setenv("EGG_PHASE", "refine") + out = restrictions.check_file_restriction({"path": self._ANALYSIS_DRAFT}) + assert out["can_write"] is True + assert out["role_can_write"] is True + assert out["phase_allows"] is True + + def test_explicit_phase_arg_overrides_env(self, monkeypatch): + monkeypatch.setenv("EGG_AGENT_ROLE", "refiner") + monkeypatch.setenv("EGG_PHASE", "plan") # env says writable + out = restrictions.check_file_restriction( + {"path": self._PLAN_DRAFT, "phase": "refine"} # arg says blocked + ) + assert out["phase"] == "refine" + assert out["can_write"] is False + assert out["blocked_by"] == "phase" + + def test_role_block_takes_priority_over_phase(self, monkeypatch): + # Contracts are blocked at BOTH layers for a coder in implement + # (role: .egg-state/ minus carve-outs; phase: implement blocks + # contracts). When both fire, blocked_by reports "role" so the + # message points at the layer that might be delegable. + monkeypatch.setenv("EGG_AGENT_ROLE", "coder") + monkeypatch.setenv("EGG_PHASE", "implement") + out = restrictions.check_file_restriction({"path": ".egg-state/contracts/p.json"}) + assert out["can_write"] is False + assert out["role_can_write"] is False + assert out["phase_allows"] is False + assert out["blocked_by"] == "role" + + def test_no_phase_env_is_role_only(self, monkeypatch): + # Backward-compat: without EGG_PHASE the phase layer is a no-op, + # so the refiner's role-level allowance stands. + monkeypatch.setenv("EGG_AGENT_ROLE", "refiner") + monkeypatch.delenv("EGG_PHASE", raising=False) + out = restrictions.check_file_restriction({"path": self._PLAN_DRAFT}) + assert out["can_write"] is True + assert out["phase"] is None + assert out["phase_allows"] is True + + def test_batch_form_carries_phase(self, monkeypatch): + monkeypatch.setenv("EGG_AGENT_ROLE", "refiner") + monkeypatch.setenv("EGG_PHASE", "refine") + out = restrictions.check_file_restriction( + {"path": [self._ANALYSIS_DRAFT, self._PLAN_DRAFT]} + ) + assert out["phase"] == "refine" + by_path = {r["path"]: r for r in out["results"]} + assert by_path[self._ANALYSIS_DRAFT]["can_write"] is True + assert by_path[self._PLAN_DRAFT]["can_write"] is False + assert by_path[self._PLAN_DRAFT]["blocked_by"] == "phase" + + def test_reviewer_impersonates_producer_via_explicit_args(self, monkeypatch): + # #2968 secondary fix: a reviewer adjudicating a producer's proposal + # needs to see the producer's verdict, not its own. With explicit + # ``role`` + ``phase`` args, the reviewer's defaults + # (EGG_AGENT_ROLE=reviewer_code, EGG_PHASE=implement) are overridden + # and the tool returns what the gateway would have done to the + # producer's refine-phase push of a plan draft. + monkeypatch.setenv("EGG_AGENT_ROLE", "reviewer_code") + monkeypatch.setenv("EGG_PHASE", "implement") + out = restrictions.check_file_restriction( + { + "path": self._PLAN_DRAFT, + "role": "refiner", + "phase": "refine", + } + ) + assert out["role"] == "refiner" + assert out["phase"] == "refine" + assert out["can_write"] is False + assert out["role_can_write"] is True + assert out["phase_allows"] is False + assert out["blocked_by"] == "phase" + # No alternative role for a phase block — it's reserved phase-wide. + assert out["alternative_role"] is None + + def test_non_string_phase_rejected(self, monkeypatch): + # The schema declares ``phase`` a string; the handler boundary + # rejects a malformed caller (int, dict, etc.) with a structured + # HandlerError instead of leaking an AttributeError from + # ``PipelinePhase(phase)``. + monkeypatch.setenv("EGG_AGENT_ROLE", "refiner") + with pytest.raises(HandlerError, match="'phase' must be a string"): + restrictions.check_file_restriction({"path": self._PLAN_DRAFT, "phase": 7}) + + class TestReportImpasse: def test_persists_to_agent_output(self, tmp_path, monkeypatch): monkeypatch.setenv("EGG_REPO_PATH", str(tmp_path)) diff --git a/shared/egg_restrictions/phase_patterns.py b/shared/egg_restrictions/phase_patterns.py new file mode 100644 index 0000000000..4a346f6010 --- /dev/null +++ b/shared/egg_restrictions/phase_patterns.py @@ -0,0 +1,220 @@ +"""Phase-scoped file-write patterns — the *second* of egg's two push-gate layers. + +Every git push through the gateway is checked against two independent filters, +both of which must allow every changed file or the push is rejected: + +1. **Role layer** — :mod:`egg_restrictions.patterns` (:data:`AGENT_PATTERNS`): + *can role R ever write path P?* +2. **Phase layer** — this module (:data:`PHASE_FILE_PATTERNS`): *is path P + writable during pipeline phase φ, regardless of role?* + +The gateway's authoritative phase enforcement lives in +``gateway/phase_filter.py`` (``PhaseFileRestriction.is_file_allowed`` / +``PhaseFilter.check_phase_file_restrictions``), configured by +``.egg/phase-permissions.json``. The ``check_file_restriction`` MCP tool runs +in the sandbox — nowhere near the gateway — and historically consulted only the +role layer, so it reported ``can_write: true`` for paths the phase gate rejects +at push time (e.g. ``.egg-state/drafts/*-plan.md`` during the *refine* phase, +which is reserved to the *plan* phase). That phase-blind false positive drove a +NACK loop in #2968: a reviewer trusted the tool and NACKed a producer for a +"false gateway claim" that was in fact a true phase-gate block. + +This module mirrors the gateway's phase data and matching logic so phase-blind +callers can predict push acceptance. The mirror is kept honest by a parity test +(``gateway/tests/test_phase_filter_restrictions.py``) that compares this module +against the real :class:`PhaseFilter` for every phase. A future consolidation — +parallel to #1903, which made ``patterns.py`` the single source of truth for the +role layer — could collapse the two by having ``gateway/phase_filter.py`` derive +from here and dropping the ``phase_file_restrictions`` key from the JSON; that +touches the security-critical push path and is intentionally out of scope here. + +Phases with no configured restriction (and ``apply``, whose live gateway config +carries no row) are unrestricted at this layer. The legacy ``pr`` phase was +hard-removed in #2777 and is not a :class:`PipelinePhase`. +""" + +from __future__ import annotations + +import posixpath +from dataclasses import dataclass + +from egg_contracts.models import PipelinePhase + +from .matchers import match_pattern + +__all__ = [ + "PHASE_FILE_PATTERNS", + "PhaseFilePattern", + "phase_file_verdict", +] + + +@dataclass(frozen=True) +class PhaseFilePattern: + """File-write restriction for a single pipeline phase. + + Mirrors ``gateway/phase_filter.py``'s ``PhaseFileRestriction``: + + - ``allowed_patterns``: if non-empty, a file must match one of these (the + sentinel ``"*"`` short-circuits to allow everything). + - ``blocked_patterns``: files matching these are always rejected, and are + checked first so a block beats an allow. + """ + + allowed_patterns: tuple[str, ...] = () + blocked_patterns: tuple[str, ...] = () + description: str = "" + + def is_file_allowed(self, file_path: str) -> tuple[bool, str]: + """Return ``(allowed, reason)`` for ``file_path`` under this phase. + + Logic is a 1:1 mirror of ``PhaseFileRestriction.is_file_allowed`` in + ``gateway/phase_filter.py`` — keep them in lockstep (the parity test + enforces it). + """ + try: + normalized = _normalize_path(file_path) + except ValueError as exc: + # Paths that escape the repository are never allowed. + return False, str(exc) + + # Blocked patterns first — an explicit block beats any allow. + for pattern in self.blocked_patterns: + if match_pattern(normalized, pattern): + return False, f"File '{file_path}' matches blocked pattern '{pattern}'" + + # A non-empty allow list is a strict whitelist. + if self.allowed_patterns: + if "*" in self.allowed_patterns: + return True, "All files allowed" + for pattern in self.allowed_patterns: + if match_pattern(normalized, pattern): + return True, f"File '{file_path}' matches allowed pattern '{pattern}'" + return False, f"File '{file_path}' does not match any allowed pattern" + + # No allow list = allow by default (only blocked patterns matter). + return True, "No explicit restrictions" + + +def _normalize_path(file_path: str) -> str: + """Mirror of ``PhaseFileRestriction._normalize_path`` in the gateway. + + Resolves ``.``/``..`` and rejects paths that escape the repository. + """ + normalized = posixpath.normpath(file_path) + if normalized.startswith("./"): + normalized = normalized[2:] + if normalized.startswith("../") or normalized.startswith("/"): + raise ValueError(f"Invalid path escapes repository: {file_path}") + return normalized + + +# Single source of truth for the phase layer as consumed by phase-blind callers. +# These rows MUST stay equivalent to the live gateway config +# (``.egg/phase-permissions.json`` → ``PhaseFilter``); the parity test in +# ``gateway/tests/test_phase_filter_restrictions.py`` fails CI on drift. +# +# Only phases with real restrictions appear here. ``apply`` is intentionally +# absent: the deployed JSON carries no ``apply`` row, so the live gateway leaves +# it unrestricted at the phase layer. (The Python *fallback* in +# ``phase_filter.py`` does restrict ``apply`` per #1557, but that branch only +# runs when the JSON is absent, which is never the case in production — so the +# fallback's apply rule is currently dead. Resolving that divergence belongs to +# the consolidation follow-up, not here.) +PHASE_FILE_PATTERNS: dict[str, PhaseFilePattern] = { + "refine": PhaseFilePattern( + allowed_patterns=( + ".egg-state/contracts/*", + ".egg-state/drafts/*analysis*", + ".egg-state/checkpoints/*", + ".egg-state/agent-outputs/*", + ".egg-state/reviews/*", + ".egg-state/agent-anchors/*", + ), + description=( + "Refine phase can only push contracts, analysis drafts, " + "checkpoints, agent outputs, reviews, and agent anchors" + ), + ), + "plan": PhaseFilePattern( + allowed_patterns=( + ".egg-state/contracts/*", + ".egg-state/drafts/*plan*", + ".egg-state/checkpoints/*", + ".egg-state/agent-outputs/*", + ".egg-state/reviews/*", + ".egg-state/agent-anchors/*", + ), + description=( + "Plan phase can only push contracts, plan drafts, checkpoints, " + "agent outputs, reviews, and agent anchors" + ), + ), + "implement": PhaseFilePattern( + blocked_patterns=( + ".egg-state/contracts/*", + ".egg-state/drafts/*", + ".egg-state/pipelines/*", + ".egg-state/reviews/*", + ), + description=( + "Implement phase can push code but not .egg-state/ " + "(except checkpoints, agent-outputs, and agent-anchors)" + ), + ), +} + + +def phase_file_verdict(phase: str | None, file_path: str) -> tuple[bool, str | None]: + """Return ``(allowed, block_reason)`` for ``file_path`` under ``phase``. + + ``block_reason`` is a human-readable string only when ``allowed`` is + ``False``; it is ``None`` when the file is allowed (or when no phase-layer + restriction applies). + + Behaviour mirrors ``PhaseFilter.check_phase_file_restrictions`` in + ``gateway/phase_filter.py``: + + - ``None`` / empty string ⇒ no phase context, no-op ``(True, None)``. This + keeps a phase-less caller (no ``EGG_PHASE``) behaving exactly as the + role-only check did before #2968. Note that the gateway itself would + fail closed on an explicit ``""`` (``PipelinePhase("")`` raises), so the + empty-string branch is a small intentional divergence. It is unreachable + in practice because every live caller normalises ``""`` to ``None`` + before it reaches this function (``restrictions.py`` does + ``req.get("phase") or get_phase()``, and ``get_phase()`` returns + ``None`` for an empty ``EGG_PHASE``); if a future caller ever exposes + ``phase_file_verdict`` to a path that doesn't pre-normalise, drop the + ``or empty`` branch so ``""`` falls through to the ``ValueError`` + handler below and fails closed like the gateway. + - A string the canonical :class:`PipelinePhase` enum doesn't recognise + (e.g. ``"IMPLEMENT"``, ``"unknown"``, the dead ``"pr"`` from #2777) ⇒ + **fail closed** ``(False, reason)`` — the gateway would reject the push + with ``"Unknown phase ... blocking by default"``, and the mirror does + the same so an off-canonical caller can't slip a false ``can_write: + true`` through. In production the orchestrator always exports the + canonical lowercase ``EGG_PHASE`` (``kubernetes_spawner.py``), so this + path only fires on a manual / test caller passing a bad string. + - A canonical phase with no configured restriction (currently ``apply``, + whose deployed JSON carries no row) ⇒ ``(True, None)`` — matches the + gateway's "no phase file restrictions for phase" fall-through. + - Otherwise the mirror's :class:`PhaseFilePattern` is evaluated and its + verdict returned. + """ + if not phase: + return True, None + + try: + canonical = PipelinePhase(phase) + except ValueError: + # Match the gateway's security stance for off-canonical phase strings. + return False, ( + f"Unknown phase {phase!r}: phase-layer gate fails closed " + "(matches gateway/phase_filter.py)" + ) + + pattern = PHASE_FILE_PATTERNS.get(canonical.value) + if pattern is None: + return True, None + allowed, reason = pattern.is_file_allowed(file_path) + return allowed, (None if allowed else reason) diff --git a/shared/tests/test_phase_patterns.py b/shared/tests/test_phase_patterns.py new file mode 100644 index 0000000000..2b05d229b3 --- /dev/null +++ b/shared/tests/test_phase_patterns.py @@ -0,0 +1,87 @@ +"""Unit tests for the phase-layer file patterns (#2968). + +The companion parity test in ``gateway/tests/test_phase_filter_restrictions.py`` +asserts this module stays equivalent to the live gateway config; these tests +pin the behaviour the ``check_file_restriction`` MCP tool relies on. +""" + +from __future__ import annotations + +from egg_restrictions.phase_patterns import PHASE_FILE_PATTERNS, phase_file_verdict + + +class TestPhaseFileVerdict: + _PLAN = ".egg-state/drafts/pipeline-8cf1f000-plan.md" + _ANALYSIS = ".egg-state/drafts/pipeline-8cf1f000-analysis.md" + _CONTRACT = ".egg-state/contracts/pipeline-8cf1f000.json" + + def test_refine_blocks_plan_draft(self): + allowed, reason = phase_file_verdict("refine", self._PLAN) + assert allowed is False + assert reason and "does not match any allowed pattern" in reason + + def test_refine_allows_analysis_draft(self): + allowed, reason = phase_file_verdict("refine", self._ANALYSIS) + assert allowed is True + assert reason is None + + def test_refine_allows_contracts(self): + assert phase_file_verdict("refine", self._CONTRACT)[0] is True + + def test_plan_allows_plan_draft(self): + assert phase_file_verdict("plan", self._PLAN)[0] is True + + def test_plan_blocks_analysis_draft(self): + # The inverse of refine: a plan-phase push of an analysis draft is + # not in the plan whitelist. + assert phase_file_verdict("plan", self._ANALYSIS)[0] is False + + def test_implement_blocks_contracts_and_drafts(self): + assert phase_file_verdict("implement", self._CONTRACT)[0] is False + assert phase_file_verdict("implement", self._PLAN)[0] is False + + def test_implement_allows_code(self): + assert phase_file_verdict("implement", "src/app.py")[0] is True + + def test_off_canonical_case_fails_closed(self): + # The mirror coerces via ``PipelinePhase(phase)``, which is + # case-sensitive — same as the gateway. Off-canonical case fails + # closed instead of silently lowercasing through to a verdict. + allowed, reason = phase_file_verdict("REFINE", self._PLAN) + assert allowed is False + assert reason and "Unknown phase 'REFINE'" in reason + + def test_known_phase_with_no_restriction_is_unrestricted(self): + # ``apply`` is a valid PipelinePhase but the deployed JSON carries + # no row, so the gateway's per-phase lookup misses and the call + # falls through to allow. The mirror matches that path. + assert phase_file_verdict("apply", self._PLAN) == (True, None) + + def test_unknown_phase_fails_closed(self): + # Off-enum strings (truly unknown phases, the dead "pr" from #2777, + # garbage from a misconfigured caller) fail closed at the mirror — + # the gateway would reject the push with "Unknown phase ... + # blocking by default", and the mirror's verdict matches so a + # phase-blind caller can't slip a false can_write:true through. + allowed, reason = phase_file_verdict("not-a-phase", self._PLAN) + assert allowed is False + assert reason and "Unknown phase 'not-a-phase'" in reason + # The dead "pr" key from .egg/phase-permissions.json takes the + # same path (PipelinePhase("pr") raises after #2777 deletion). + allowed, reason = phase_file_verdict("pr", self._PLAN) + assert allowed is False + + def test_empty_phase_is_unrestricted(self): + assert phase_file_verdict(None, self._PLAN) == (True, None) + assert phase_file_verdict("", self._PLAN) == (True, None) + + def test_path_escape_is_blocked(self): + allowed, reason = phase_file_verdict("refine", "../../etc/passwd") + assert allowed is False + assert reason and "escapes repository" in reason + + +class TestPhaseFilePatternData: + def test_only_restricted_phases_present(self): + # pr (removed in #2777) and apply (unrestricted live) must not appear. + assert set(PHASE_FILE_PATTERNS) == {"refine", "plan", "implement"}