Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 0 additions & 30 deletions .egg/phase-permissions.json
Original file line number Diff line number Diff line change
Expand Up @@ -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": {
Expand Down Expand Up @@ -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"
}
}
}
2 changes: 1 addition & 1 deletion docs/reference/agent-tools.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
87 changes: 87 additions & 0 deletions gateway/tests/test_phase_filter_restrictions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"])
Expand Down Expand Up @@ -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}"
)
141 changes: 106 additions & 35 deletions sandbox/egg_agent_tools/handlers/restrictions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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``.
Expand All @@ -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:
Expand All @@ -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")
Expand All @@ -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")
Expand Down
Loading
Loading