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
11 changes: 10 additions & 1 deletion src/adm_cli/gates.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ def can_advance(
if missing:
return False, f"Missing required artefacts in {phase_dir_name}/: {', '.join(missing)}"

# Ratchet gate: zero PENDING/OPEN resolutions
# Ratchet gate: zero PENDING/OPEN resolutions + stability window
if target_phase == 6:
open_clrs = [
k for k, v in domain_state.resolutions.items()
Expand All @@ -90,6 +90,15 @@ def can_advance(
if open_clrs:
return False, f"Cannot ratchet: {len(open_clrs)} open clarifications ({', '.join(open_clrs)})"

if domain_state.stability_window_days > 0 and domain_state.last_thesis_change:
from datetime import date

last_change = date.fromisoformat(domain_state.last_thesis_change)
days_elapsed = (date.today() - last_change).days
if days_elapsed < domain_state.stability_window_days:
remaining = domain_state.stability_window_days - days_elapsed
return False, f"Stability window: {remaining} days remaining (need {domain_state.stability_window_days}, elapsed {days_elapsed})"

return True, "Gate passed"


Expand Down
2 changes: 2 additions & 0 deletions src/adm_cli/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ class DomainState(BaseModel):
iteration: Annotated[int, Field(ge=1)] = 1
contract_version: str | None = None
source_path: str | None = None
stability_window_days: Annotated[int, Field(ge=0)] = 0
last_thesis_change: str | None = None
resolutions: dict[str, ResolutionStatus] = Field(default_factory=dict)

@field_validator("contract_version")
Expand Down
41 changes: 41 additions & 0 deletions tests/test_gates.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,47 @@ def test_ratchet_gate_passes_when_all_resolved(self, tmp_path: Path):
allowed, reason = can_advance("holdings", 6, state, tmp_path)
assert allowed is True

def test_stability_window_blocks_when_too_recent(self, tmp_path: Path):
from datetime import date

state = _setup_project(
tmp_path,
phase=5,
stability_window_days=7,
last_thesis_change=date.today().isoformat(),
)
_create_phase_artefacts(tmp_path, "holdings", 5)
allowed, reason = can_advance("holdings", 6, state, tmp_path)
assert allowed is False
assert "stability window" in reason.lower()

def test_stability_window_passes_when_elapsed(self, tmp_path: Path):
from datetime import date, timedelta

past = (date.today() - timedelta(days=10)).isoformat()
state = _setup_project(
tmp_path,
phase=5,
stability_window_days=7,
last_thesis_change=past,
)
_create_phase_artefacts(tmp_path, "holdings", 5)
allowed, reason = can_advance("holdings", 6, state, tmp_path)
assert allowed is True

def test_stability_window_zero_allows_immediate(self, tmp_path: Path):
from datetime import date

state = _setup_project(
tmp_path,
phase=5,
stability_window_days=0,
last_thesis_change=date.today().isoformat(),
)
_create_phase_artefacts(tmp_path, "holdings", 5)
allowed, reason = can_advance("holdings", 6, state, tmp_path)
assert allowed is True


class TestAdvancePhase:
def test_increments_phase(self, tmp_path: Path):
Expand Down
Loading