diff --git a/roboco/enforcement/task_lifecycle.py b/roboco/enforcement/task_lifecycle.py index d08652a1..06511db8 100644 --- a/roboco/enforcement/task_lifecycle.py +++ b/roboco/enforcement/task_lifecycle.py @@ -84,8 +84,10 @@ _LEGACY_OPERATIONAL_EDGES: dict[Status, frozenset[Status]] = { } ), # Self-fail out of verifying (QA / PM only — role gate enforced - # in ROLE_RESTRICTED_TRANSITIONS below). - Status.VERIFYING: frozenset({Status.NEEDS_REVISION, Status.AWAITING_DOCUMENTATION}), + # in ROLE_RESTRICTED_TRANSITIONS below). The canonical exit is submit_qa + # -> awaiting_qa -> (qa_pass) -> awaiting_documentation; a direct + # verifying->awaiting_documentation edge would bypass the QA review hop. + Status.VERIFYING: frozenset({Status.NEEDS_REVISION}), # QA can park a task as blocked while waiting on dev clarification. Status.AWAITING_QA: frozenset({Status.BLOCKED}), # PM claim + PM reject path on review queue. @@ -251,6 +253,7 @@ def is_waiting_state(status: str) -> bool: "paused", "awaiting_qa", "awaiting_documentation", + "awaiting_pr_review", "awaiting_pm_review", "awaiting_ceo_approval", ) diff --git a/roboco/foundation/_validate_lifecycle.py b/roboco/foundation/_validate_lifecycle.py index 84cd1af2..2c801884 100644 --- a/roboco/foundation/_validate_lifecycle.py +++ b/roboco/foundation/_validate_lifecycle.py @@ -53,14 +53,63 @@ def reachable_from(start: Status) -> set[Status]: def _check_status_enum_coverage() -> None: - """Every Status appears in STATUS_GRAPH (as key).""" - from roboco.foundation.policy.lifecycle import STATUS_GRAPH, Status + """Every Status is reachable in the transition set, bidirectionally. - missing = set(Status) - set(STATUS_GRAPH) - if missing: - missing_values = sorted(s.value for s in missing) + The old check (`set(Status) - set(STATUS_GRAPH)`) was a tautology: + ``_build_status_graph`` seeds ``{s: set() for s in Status}``, so every + Status is a key by construction and the check could never raise. It gave + false assurance while an orphan state (no transition references it) or a + stray-string target slipped past. This checks the real coverage: + + (a) every non-terminal Status is the SOURCE of at least one transition — + an orphan state with no outgoing edge is caught; + (b) every source/target referenced in _STATUS_TRANSITIONS is a real Status + member — a stray-string target (e.g. a typo or a removed status) is + caught. + """ + from roboco.foundation.policy.lifecycle import _STATUS_TRANSITIONS, Status + + all_statuses = set(Status) + terminals = {Status.COMPLETED, Status.CANCELLED} + sources = {t.source for t in _STATUS_TRANSITIONS} + targets = {t.target for t in _STATUS_TRANSITIONS} + referenced = sources | targets + + orphan_sources = {s for s in all_statuses - terminals if s not in sources} + stray = referenced - all_statuses + + problems: list[str] = [] + if orphan_sources: + problems.append( + "non-terminal statuses with no outgoing transition: " + + sorted(s.value for s in orphan_sources).__repr__() + ) + if stray: + stray_repr = sorted(repr(s) for s in stray) + problems.append(f"transitions reference non-Status values: {stray_repr}") + if problems: + raise LifecycleSpecError("; ".join(problems)) + + +def _check_status_enum_parity() -> None: + """The spec's Status enum must match models.base.TaskStatus exactly. + + TaskType has long had this guard (test_task_type_enum_matches_models); Status + did not. The two StrEnums are textually identical today, but the seam was + unguarded: a status added to the ORM column type (TaskStatus) but not the + lifecycle map (Status) drifts silently — TaskService writes it to the DB, + get_valid_transitions returns [] for the orphan, and the task looks terminal + with no valid exits. Fail the build at import before any test runs. + """ + from roboco.foundation.policy.lifecycle import Status + from roboco.models.base import TaskStatus + + spec_values = {s.value for s in Status} + model_values = {s.value for s in TaskStatus} + if spec_values != model_values: raise LifecycleSpecError( - f"Statuses missing from STATUS_GRAPH keys: {missing_values}" + f"Status enum drift between lifecycle.spec and models.base: " + f"{spec_values ^ model_values}" ) @@ -79,17 +128,25 @@ def _check_status_reachability() -> None: def _check_terminal_exits() -> None: - """Every non-terminal status exits to either COMPLETED or CANCELLED.""" + """Every non-terminal status reaches COMPLETED and can be CANCELLED. + + The cancel fan-out generates a ``cancel`` edge from every non-terminal + status to CANCELLED, so the old ``reachable & {COMPLETED, CANCELLED}`` + check was structurally trivial — a status whose sole exit was cancel passed + the guard with no real forward completion path. Split the check so a + state-machine hole that traps work behind a cancel-only exit is caught at + import. + """ from roboco.foundation.policy.lifecycle import Status terminals = {Status.COMPLETED, Status.CANCELLED} non_terminal = set(Status) - terminals for s in non_terminal: reachable = reachable_from(s) - if not (reachable & terminals): - raise LifecycleSpecError( - f"Status '{s.value}' has no path to COMPLETED or CANCELLED" - ) + if Status.COMPLETED not in reachable: + raise LifecycleSpecError(f"Status '{s.value}' has no path to COMPLETED") + if Status.CANCELLED not in reachable: + raise LifecycleSpecError(f"Status '{s.value}' has no cancel exit") def _check_intent_compositions() -> None: @@ -273,6 +330,7 @@ def _check_unmigrated_is_subset() -> None: _LIFECYCLE_VALIDATORS = ( _check_status_enum_coverage, + _check_status_enum_parity, _check_status_reachability, _check_terminal_exits, _check_intent_compositions, diff --git a/tests/foundation/test_lifecycle_spec.py b/tests/foundation/test_lifecycle_spec.py index 3de10029..499cea5a 100644 --- a/tests/foundation/test_lifecycle_spec.py +++ b/tests/foundation/test_lifecycle_spec.py @@ -11,6 +11,7 @@ from roboco.foundation import _validate_lifecycle as _validate from roboco.foundation._validate_lifecycle import reachable_from from roboco.foundation.policy import lifecycle as spec from roboco.foundation.policy.lifecycle import _INTENT_VERBS, IntentSpec +from roboco.models.base import TaskStatus as ModelTaskStatus from roboco.models.base import TaskType as ModelTaskType @@ -79,6 +80,73 @@ def test_task_type_enum_matches_models() -> None: ) +def test_status_enum_matches_models() -> None: + """The spec's Status must match models.base.TaskStatus — the ORM column + type and the lifecycle map must not drift. TaskType has this guard; Status + did not, so adding/renaming a status in one enum only wedged silently.""" + spec_values = {s.value for s in spec.Status} + model_values = {s.value for s in ModelTaskStatus} + assert spec_values == model_values, ( + f"Status drift between lifecycle.spec and models.base: " + f"{spec_values ^ model_values}" + ) + + +def test_status_enum_parity_validator_passes_on_real_spec() -> None: + """The import-time parity validator must agree with the real enums.""" + _validate._check_status_enum_parity() # no raise + + +def test_status_coverage_rejects_stray_string_target() -> None: + """A transition referencing a non-Status target string must fail the + coverage validator — the old check was a tautology (STATUS_GRAPH keys every + Status by construction) and let stray-string targets through.""" + fake = ( + SimpleNamespace( + source=spec.Status.PENDING, target="bogus_state", triggered_by_action="x" + ), + ) + original = spec._STATUS_TRANSITIONS + spec._STATUS_TRANSITIONS = fake # type: ignore[attr-defined] + try: + with pytest.raises(_validate.LifecycleSpecError, match="non-Status"): + _validate._check_status_enum_coverage() + finally: + spec._STATUS_TRANSITIONS = original # type: ignore[attr-defined] + + +def test_status_coverage_rejects_orphan_non_terminal_source() -> None: + """A non-terminal status that is the source of no transition (an orphan + state) must fail the coverage validator — the cancel fan-out made the old + 'is a key in STATUS_GRAPH' check structurally always-true.""" + original = spec._STATUS_TRANSITIONS + spec._STATUS_TRANSITIONS = tuple( # type: ignore[attr-defined] + t for t in original if t.source is not spec.Status.PAUSED + ) + try: + with pytest.raises( + _validate.LifecycleSpecError, match="no outgoing transition" + ): + _validate._check_status_enum_coverage() + finally: + spec._STATUS_TRANSITIONS = original # type: ignore[attr-defined] + + +def test_terminal_exit_requires_a_completed_path() -> None: + """Every non-terminal status must reach COMPLETED specifically — the cancel + fan-out made the old {COMPLETED, CANCELLED} check trivial, so a status whose + sole exit was cancel passed the guard with no real forward completion path.""" + original = spec.STATUS_GRAPH + fake = dict(original) + fake[spec.Status.PAUSED] = frozenset({spec.Status.CANCELLED}) + spec.STATUS_GRAPH = fake # type: ignore[attr-defined] + try: + with pytest.raises(_validate.LifecycleSpecError, match="no path to COMPLETED"): + _validate._check_terminal_exits() + finally: + spec.STATUS_GRAPH = original # type: ignore[attr-defined] + + def test_decision_allow_has_no_rejection_kind() -> None: d = spec.Decision.allow() assert d.allowed is True diff --git a/tests/unit/enforcement/test_task_lifecycle.py b/tests/unit/enforcement/test_task_lifecycle.py index 50215487..911c447b 100644 --- a/tests/unit/enforcement/test_task_lifecycle.py +++ b/tests/unit/enforcement/test_task_lifecycle.py @@ -17,6 +17,7 @@ from roboco.enforcement.task_lifecycle import ( validate_task_transition, ) from roboco.exceptions import TaskLifecycleError +from roboco.foundation.policy.lifecycle import Status # --------------------------------------------------------------------------- # validate_task_transition @@ -231,3 +232,46 @@ def test_validate_cancel_from_awaiting_ceo_raises_for_pm() -> None: validate_task_transition("awaiting_ceo_approval", "cancelled", "cell_pm") # CEO is allowed — no raise. assert validate_task_transition("awaiting_ceo_approval", "cancelled", "ceo") is True + + +# --------------------------------------------------------------------------- +# VERIFYING must go through the QA hop (awaiting_qa), not straight to docs +# --------------------------------------------------------------------------- + + +def test_verifying_to_awaiting_documentation_is_rejected() -> None: + """VERIFYING is the dev self-verification state; the canonical exit is + submit_qa -> awaiting_qa -> (qa_pass) -> awaiting_documentation. A spurious + verifying->awaiting_documentation edge bypassed the entire QA review hop.""" + with pytest.raises(TaskLifecycleError): + validate_task_transition("verifying", "awaiting_documentation", "qa") + + +def test_verifying_self_fail_to_needs_revision_still_allowed() -> None: + """The legitimate self-fail out of verifying (QA/PM only) is preserved.""" + assert validate_task_transition("verifying", "needs_revision", "qa") is True + + +# --------------------------------------------------------------------------- +# is_waiting_state must cover the in-path PR-review gate +# --------------------------------------------------------------------------- + + +def test_is_waiting_state_includes_awaiting_pr_review() -> None: + """The PR-review gate parks the PM on the reviewer; it is a waiting state, + not an active one (the predicates were never updated when AWAITING_PR_REVIEW + was added to the enum).""" + assert is_waiting_state("awaiting_pr_review") is True + assert is_active_state("awaiting_pr_review") is False + + +def test_status_classification_is_mutually_disjoint() -> None: + """No status may be classified as both active and waiting — the structural + invariant that catches miscategorization (the awaiting_pr_review leak was + an instance of a status falling into no category).""" + active = {s.value for s in Status if is_active_state(s.value)} + waiting = {s.value for s in Status if is_waiting_state(s.value)} + terminal = {s.value for s in Status if is_terminal_state(s.value)} + assert active & waiting == set() + assert active & terminal == set() + assert waiting & terminal == set()