diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..6b7ddcf --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,37 @@ +name: test + +# The seed of the release gate ADR-0006 requires of a consumer. Right now it runs +# unit tests; payload conformance against the pinned contracts joins it in issue #6. + +on: + push: + branches: [main] + paths: ['packages/**', 'apps/**', '.github/workflows/test.yml'] + pull_request: + paths: ['packages/**', 'apps/**', '.github/workflows/test.yml'] + workflow_dispatch: + +permissions: + contents: read + +jobs: + core: + runs-on: ubuntu-latest + timeout-minutes: 10 + strategy: + matrix: + python-version: ['3.11', '3.12', '3.13'] + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install + working-directory: packages/core + run: pip install --no-cache-dir -e '.[test]' + + - name: Test + working-directory: packages/core + run: python -m pytest --cov=devfactory_core --cov-report=term-missing diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..803929b --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +.coverage +.coverage.* +htmlcov/ +*.egg-info/ +build/ +dist/ +.venv/ +venv/ diff --git a/packages/core/README.md b/packages/core/README.md index 7c111a2..7a7d7cc 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -1 +1,87 @@ # core module + +The job state machine — the control plane's lifecycle engine. + +Spec: [`state-machine.md`](state-machine.md) — [RFC-0001](../../rfcs/0001-job-state-machine.md) +as amended by [RFC-0007](../../rfcs/0007-job-lifecycle-completeness.md), with the tenant +model from [RFC-0006](../../rfcs/0006-tenant-workspace-model.md). + +In memory only. No persistence, no policy engine, no API — those are issues #5, #6, and #7. + +## Use + +```python +from devfactory_core import Job, JobState, Principal + +alice = Principal("human", "alice") +job = Job( + job_id="job-001", + tenant_id="default", # RFC-0006: never omitted, even single-tenant + workspace_id="ws-core", + principal=alice, +) + +job.submit_for_governance(reason="ready for review") +job.approve(authority=alice, reason="scope matches milestone v0.1") +job.transition(JobState.TASK_PLANNING) +job.transition(JobState.IN_PROGRESS) + +job.pause_for_approval(reason="merge needs sign-off") +assert job.state is JobState.AWAITING_APPROVAL +assert job.awaiting_from is JobState.IN_PROGRESS +job.resume(reason="approved", principal=alice) + +job.transition(JobState.VALIDATING) +job.transition(JobState.DEPLOYABLE) +job.transition(JobState.COMPLETED) + +job.event_payloads() # audit trail in event/v1 wire shape +``` + +## What the engine refuses + +Every refusal below is deliberate. The lifecycle exists so governance can be +enforced, so the engine rejects rather than repairs. + +| call | refusal | +| --- | --- | +| an edge not in the table | `InvalidTransition`, naming what *was* allowed | +| anything out of `COMPLETED` / `FAILED` / `CANCELLED` / `TIMED_OUT` | `TerminalState` — recovery is `supersede()`, not a revival | +| `TASK_PLANNING` before `APPROVED` | `InvalidTransition` — execution is forbidden before approval | +| `FAILED` / `CANCELLED` / `TIMED_OUT` without a reason | `MissingReason` | +| `CANCELLED` without a principal | `MissingPrincipal` | +| `APPROVED` / `REJECTED` without an authority and reason | `MissingAuthority` | +| pausing outside `IN_PROGRESS` / `VALIDATING` / `DEPLOYABLE` | `MissingApprovalContext` | +| resuming into a state other than `awaiting_from` | `WrongResumeState` | +| a malformed identifier | `InvalidIdentifier` — `identity/v1` `Id` form | + +A refused call leaves the job untouched and writes nothing to the audit trail. + +## Events + +Construction emits `JOB_CREATED`; every accepted transition emits `STATE_TRANSITION`; +reaching `COMPLETED` also emits `JOB_COMPLETED`. There is no way to change state +without going through `transition()`, which is what makes *no silent state change* +hold rather than merely be documented. + +`event_payloads()` renders the trail in `event/v1` wire shape. It is **not** validated +here — owning a copy of the schema would be a parallel schema, which +[RFC-0005](../../rfcs/0005-platform-contract-authority.md) Rule 4 forbids. Validation +against the pinned contract is issue #6, and these payloads are what it will validate. + +## Tests + +```bash +cd packages/core +python -m pytest # 235 tests +python -m pytest --cov=devfactory_core # coverage gate at 90%, currently 100% +``` + +## Open question + +`state-machine.md` says `FAILED` is terminal and lists `AWAITING_APPROVAL -> FAILED`, +but never enumerates which other states may fail. This module permits `FAILED` from +`TASK_PLANNING`, `IN_PROGRESS`, `AWAITING_APPROVAL`, `VALIDATING`, and `DEPLOYABLE` — +the states where work exists to fail — and refuses it before `APPROVED`, where the +honest outcomes are `REJECTED`, `CANCELLED`, or `TIMED_OUT`. That reading needs +confirming in an RFC; see `states.FAILABLE`. diff --git a/packages/core/devfactory_core/__init__.py b/packages/core/devfactory_core/__init__.py new file mode 100644 index 0000000..dc6f651 --- /dev/null +++ b/packages/core/devfactory_core/__init__.py @@ -0,0 +1,43 @@ +"""devfactory-core — governance-first control plane. + +Phase 1: the job state machine, in memory. See ``packages/core/state-machine.md``. +""" + +from .errors import ( + ExecutionBeforeApproval, + InvalidIdentifier, + InvalidTransition, + JobStateMachineError, + MissingApprovalContext, + MissingAuthority, + MissingPrincipal, + MissingReason, + TerminalState, + WrongResumeState, +) +from .events import Event, EventType +from .identity import DEFAULT_TENANT, Principal +from .job import Job, TransitionRecord +from .states import TERMINAL, TRANSITIONS, JobState + +__all__ = [ + "DEFAULT_TENANT", + "TERMINAL", + "TRANSITIONS", + "Event", + "EventType", + "ExecutionBeforeApproval", + "InvalidIdentifier", + "InvalidTransition", + "Job", + "JobState", + "JobStateMachineError", + "MissingApprovalContext", + "MissingAuthority", + "MissingPrincipal", + "MissingReason", + "Principal", + "TerminalState", + "TransitionRecord", + "WrongResumeState", +] diff --git a/packages/core/devfactory_core/errors.py b/packages/core/devfactory_core/errors.py new file mode 100644 index 0000000..e5264bc --- /dev/null +++ b/packages/core/devfactory_core/errors.py @@ -0,0 +1,117 @@ +"""Errors raised by the job state machine. + +Every one of these is a refusal, not a fallback. The lifecycle exists so that +governance can be enforced; an engine that repairs a bad call instead of +rejecting it would defeat the reason for having it. +""" + +from __future__ import annotations + + +class JobStateMachineError(Exception): + """Base class, so a caller can catch every refusal from this module.""" + + +class InvalidTransition(JobStateMachineError): + """The requested edge is not in the transition table for the current state.""" + + def __init__(self, current: str, requested: str, allowed: list[str]) -> None: + self.current = current + self.requested = requested + self.allowed = allowed + permitted = ", ".join(allowed) if allowed else "nothing — this state is terminal" + super().__init__( + f"{current} -> {requested} is not a valid transition. Allowed: {permitted}" + ) + + +class TerminalState(JobStateMachineError): + """A transition was requested out of a terminal state. + + Recovery from FAILED is a new job carrying ``supersedes_job_id``, never a + transition out of it — RFC-0007 keeps FAILED terminal so that recovery has + to pass GOVERNANCE_ANALYSIS again rather than resume under a stale APPROVED. + """ + + def __init__(self, current: str) -> None: + self.current = current + super().__init__( + f"{current} is terminal. Recovery is a new job with supersedes_job_id, " + f"not a transition out of {current}." + ) + + +class MissingReason(JobStateMachineError): + """FAILED, CANCELLED, and TIMED_OUT each require reason metadata.""" + + def __init__(self, state: str) -> None: + self.state = state + super().__init__(f"{state} requires a reason — 'it stopped' is not an audit record") + + +class MissingPrincipal(JobStateMachineError): + """CANCELLED records who cancelled it.""" + + def __init__(self, state: str) -> None: + self.state = state + super().__init__( + f"{state} requires the principal responsible — " + f"'someone stopped this' is not an audit record" + ) + + +class MissingApprovalContext(JobStateMachineError): + """AWAITING_APPROVAL cannot be entered without knowing where to return to.""" + + def __init__(self, current: str) -> None: + self.current = current + super().__init__( + f"AWAITING_APPROVAL cannot be entered from {current} — " + f"only IN_PROGRESS, VALIDATING, or DEPLOYABLE can pause for approval" + ) + + +class WrongResumeState(JobStateMachineError): + """A paused job tried to resume somewhere other than where it paused.""" + + def __init__(self, awaiting_from: str, requested: str) -> None: + self.awaiting_from = awaiting_from + self.requested = requested + super().__init__( + f"job paused in {awaiting_from} cannot resume into {requested} — " + f"resuming elsewhere would silently lose its place in the lifecycle" + ) + + +class ExecutionBeforeApproval(JobStateMachineError): + """The direction lock: no execution before an explicit APPROVE.""" + + def __init__(self, requested: str) -> None: + self.requested = requested + super().__init__( + f"cannot reach {requested} without passing APPROVED — " + f"execution is forbidden before governance approves" + ) + + +class MissingAuthority(JobStateMachineError): + """APPROVED and REJECTED are decisions and must name who made them.""" + + def __init__(self, state: str) -> None: + self.state = state + super().__init__( + f"{state} requires an accountable authority and a reason — " + f"an approval nobody signed is not auditable" + ) + + +class InvalidIdentifier(JobStateMachineError): + """Identifiers must match the identity/v1 Id form.""" + + def __init__(self, field: str, value: str) -> None: + self.field = field + self.value = value + super().__init__( + f"{field}={value!r} is not a valid identity/v1 Id " + f"(lowercase, leading alphanumeric, [a-z0-9_-], max 63 chars)" + ) diff --git a/packages/core/devfactory_core/events.py b/packages/core/devfactory_core/events.py new file mode 100644 index 0000000..419ea73 --- /dev/null +++ b/packages/core/devfactory_core/events.py @@ -0,0 +1,97 @@ +"""Audit events emitted by the state machine. + +Shaped to ``event/v1`` from agent-platform so that the payloads this engine +produces are the ones a conformance test will validate (issue #6). This module +deliberately does **not** validate — owning a copy of the schema here would be a +parallel schema, which RFC-0005 Rule 4 forbids. It builds the payload; the +contract judges it. + +Invariants carried from RFC-0008 and enforced by construction rather than by +convention: + +* ``job_id`` is always present on events this repository emits. The field is + optional in the schema and not optional in our behaviour. +* identifiers are never fabricated — every id here comes from a real object. +* ``metadata`` carries structured facts only. Private reasoning traces are not + audit records and must never be placed here. +""" + +from __future__ import annotations + +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import Enum +from typing import Any + +from .identity import Principal + + +class EventType(str, Enum): + """The canonical vocabulary from RFC-0003. + + RFC-0009 made this a required minimum rather than a closed set at the + contract level — agent-platform may add types. These are the ones the job + state machine itself emits. + """ + + JOB_CREATED = "JOB_CREATED" + STATE_TRANSITION = "STATE_TRANSITION" + JOB_COMPLETED = "JOB_COMPLETED" + + +def new_event_id() -> str: + """A fresh event id in the identity/v1 ``Id`` form. + + ``uuid4().hex`` is 32 lowercase hex characters, which satisfies the pattern + without needing to be reshaped. + """ + return uuid.uuid4().hex + + +def utc_now() -> datetime: + return datetime.now(timezone.utc) + + +@dataclass(frozen=True, slots=True) +class Event: + """One audit record. Frozen — ``event/v1`` guarantees append-only.""" + + event_id: str + event_type: EventType + tenant_id: str + subject_type: str + subject_id: str + occurred_at: datetime + job_id: str + workspace_id: str | None = None + actor: Principal | None = None + transition: dict[str, Any] | None = None + metadata: dict[str, Any] = field(default_factory=dict) + + def as_payload(self) -> dict[str, Any]: + """Render to the ``event/v1`` wire shape. + + ``source.kind`` is ``internal`` because this engine is the origin. An + event arriving from another system keeps its own ``source`` — RFC-0008 + requires an external event to stay identifiable as external forever. + """ + payload: dict[str, Any] = { + "event_id": self.event_id, + "event_type": self.event_type.value, + "tenant_id": self.tenant_id, + "subject_type": self.subject_type, + "subject_id": self.subject_id, + "job_id": self.job_id, + "occurred_at": self.occurred_at.isoformat(), + "source": {"kind": "internal", "system": "devfactory-core"}, + } + if self.workspace_id is not None: + payload["workspace_id"] = self.workspace_id + if self.actor is not None: + payload["actor"] = self.actor.as_payload() + if self.transition is not None: + payload["transition"] = self.transition + if self.metadata: + payload["metadata"] = dict(self.metadata) + return payload diff --git a/packages/core/devfactory_core/identity.py b/packages/core/devfactory_core/identity.py new file mode 100644 index 0000000..e9fd939 --- /dev/null +++ b/packages/core/devfactory_core/identity.py @@ -0,0 +1,57 @@ +"""Identity types, mirroring ``identity/v1`` from agent-platform. + +RFC-0006 adopts ``identity/v1`` rather than defining parallel id types, so this +module validates against that contract's shape and does not invent its own. +It is not a schema — the wire format is agent-platform's. It is the minimum +needed to refuse a malformed identifier before it reaches an audit record. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Literal + +from .errors import InvalidIdentifier + +#: identity/v1 ``$defs.Id`` — lowercase, leading alphanumeric, max 63 characters. +ID_PATTERN = re.compile(r"^[a-z0-9][a-z0-9_-]{0,62}$") + +#: RFC-0006: single-tenant deployments use this literal rather than omitting the +#: field, so the payload shape is already right when a second tenant appears. +DEFAULT_TENANT = "default" + +PrincipalType = Literal["human", "agent", "service"] + + +def validate_id(field: str, value: str) -> str: + if not isinstance(value, str) or not ID_PATTERN.match(value): + raise InvalidIdentifier(field, value) + return value + + +@dataclass(frozen=True, slots=True) +class Principal: + """Who acted — the answer to "who" in every audit record. + + ``on_behalf_of`` carries a delegation chain; per identity/v1 it must never + widen scope beyond the principal it came from. + """ + + type: PrincipalType + id: str + display_name: str | None = None + on_behalf_of: "Principal | None" = None + + def __post_init__(self) -> None: + if self.type not in ("human", "agent", "service"): + raise ValueError(f"principal type must be human, agent, or service — got {self.type!r}") + validate_id("principal.id", self.id) + + def as_payload(self) -> dict: + payload: dict = {"type": self.type, "id": self.id} + if self.display_name is not None: + payload["display_name"] = self.display_name + if self.on_behalf_of is not None: + payload["on_behalf_of"] = self.on_behalf_of.as_payload() + return payload diff --git a/packages/core/devfactory_core/job.py b/packages/core/devfactory_core/job.py new file mode 100644 index 0000000..c18fd84 --- /dev/null +++ b/packages/core/devfactory_core/job.py @@ -0,0 +1,362 @@ +"""The in-memory job state machine. + +Canonical spec: ``packages/core/state-machine.md`` — RFC-0001 as amended by +RFC-0007, with the tenant model from RFC-0006. + +Scope, per issue #2: in memory, no persistence, no policy engine, no API. What +this module owns is the lifecycle and the guards on it. + +Open question surfaced by implementing this +------------------------------------------- +``state-machine.md`` says FAILED is terminal and lists ``AWAITING_APPROVAL -> +FAILED``, but never enumerates which other states may fail. This module permits +FAILED from TASK_PLANNING, IN_PROGRESS, AWAITING_APPROVAL, VALIDATING, and +DEPLOYABLE — the states where work exists to fail — and refuses it before +APPROVED, where the honest outcomes are REJECTED, CANCELLED, or TIMED_OUT. +That reading needs confirming in an RFC; see ``states.FAILABLE``. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Callable + +from .errors import ( + ExecutionBeforeApproval, + InvalidTransition, + MissingApprovalContext, + MissingAuthority, + MissingPrincipal, + MissingReason, + TerminalState, + WrongResumeState, +) +from .events import Event, EventType, new_event_id, utc_now +from .identity import Principal, validate_id +from .states import ( + APPROVAL_PAUSABLE, + POST_APPROVAL, + TERMINAL, + TRANSITIONS, + JobState, +) + +#: States whose entry requires reason metadata — RFC-0001 for FAILED, extended +#: by RFC-0007 to the two terminal states it added. +REASON_REQUIRED: frozenset[JobState] = frozenset( + {JobState.FAILED, JobState.CANCELLED, JobState.TIMED_OUT} +) + +#: States whose entry is a decision and must name an accountable authority. +AUTHORITY_REQUIRED: frozenset[JobState] = frozenset( + {JobState.APPROVED, JobState.REJECTED} +) + + +@dataclass(frozen=True, slots=True) +class TransitionRecord: + """One entry in the immutable transition history.""" + + from_state: JobState + to_state: JobState + at: datetime + reason: str | None = None + principal: Principal | None = None + event_id: str | None = None + + +class Job: + """A governed unit of work moving through the lifecycle. + + Construction emits ``JOB_CREATED``; every accepted transition emits + ``STATE_TRANSITION``; reaching COMPLETED also emits ``JOB_COMPLETED``. There + is no way to change ``state`` without going through :meth:`transition`, which + is what makes "no silent state change" hold rather than merely be documented. + """ + + def __init__( + self, + *, + job_id: str, + tenant_id: str, + workspace_id: str, + principal: Principal, + supersedes_job_id: str | None = None, + clock: Callable[[], datetime] = utc_now, + ) -> None: + # RFC-0006: tenant and workspace are required on every job. A + # single-tenant deployment passes the literal "default"; it never omits + # the field, so the payload is already correct when a second tenant appears. + self._job_id = validate_id("job_id", job_id) + self._tenant_id = validate_id("tenant_id", tenant_id) + self._workspace_id = validate_id("workspace_id", workspace_id) + if not isinstance(principal, Principal): + raise TypeError("principal must be a Principal — 'who created this' is required") + self._principal = principal + self._supersedes_job_id = ( + validate_id("supersedes_job_id", supersedes_job_id) + if supersedes_job_id is not None + else None + ) + self._clock = clock + + self._state = JobState.DRAFT + self._awaiting_from: JobState | None = None + self._approved = False + self._history: list[TransitionRecord] = [] + self._events: list[Event] = [] + + metadata: dict[str, Any] = {} + if self._supersedes_job_id is not None: + metadata["supersedes_job_id"] = self._supersedes_job_id + self._emit(EventType.JOB_CREATED, actor=principal, metadata=metadata) + + # ---- read-only surface ------------------------------------------------- + # Everything below is exposed as a copy or an immutable view. The audit trail + # is append-only, so handing out the live list would let a caller edit history. + + @property + def job_id(self) -> str: + return self._job_id + + @property + def tenant_id(self) -> str: + return self._tenant_id + + @property + def workspace_id(self) -> str: + return self._workspace_id + + @property + def principal(self) -> Principal: + return self._principal + + @property + def supersedes_job_id(self) -> str | None: + return self._supersedes_job_id + + @property + def state(self) -> JobState: + return self._state + + @property + def awaiting_from(self) -> JobState | None: + """Where a paused job returns to. Set only while in AWAITING_APPROVAL.""" + return self._awaiting_from + + @property + def is_terminal(self) -> bool: + return self._state in TERMINAL + + @property + def history(self) -> tuple[TransitionRecord, ...]: + return tuple(self._history) + + @property + def events(self) -> tuple[Event, ...]: + return tuple(self._events) + + def event_payloads(self) -> list[dict[str, Any]]: + """The audit trail in ``event/v1`` wire shape — what conformance validates.""" + return [event.as_payload() for event in self._events] + + def allowed_targets(self) -> frozenset[JobState]: + """States reachable right now, including this job's return edge. + + ``AWAITING_APPROVAL``'s way back is ``awaiting_from``, which differs per + job and so cannot live in the static table. + """ + targets = set(TRANSITIONS[self._state]) + if self._state is JobState.AWAITING_APPROVAL and self._awaiting_from is not None: + targets.add(self._awaiting_from) + return frozenset(targets) + + # ---- the engine -------------------------------------------------------- + + def transition( + self, + to: JobState, + *, + reason: str | None = None, + principal: Principal | None = None, + ) -> Event: + """Move to ``to``, or refuse and leave the job untouched. + + Returns the ``STATE_TRANSITION`` event so a caller can forward it without + reaching back into the history. + """ + to = JobState(to) + self._check_not_terminal() + self._check_edge(to) + self._check_guards(to, reason=reason, principal=principal) + + previous = self._state + # AWAITING_APPROVAL's return address is recorded on entry and cleared on + # exit, so a job paused during DEPLOYABLE cannot resume as IN_PROGRESS. + if to is JobState.AWAITING_APPROVAL: + self._awaiting_from = previous + elif previous is JobState.AWAITING_APPROVAL: + self._awaiting_from = None + + if to is JobState.APPROVED: + self._approved = True + elif to is JobState.REJECTED: + # A rejected job returns to DRAFT for revision. The approval that was + # never granted must not carry over, and an approval granted to an + # earlier revision must not authorise the revised one. + self._approved = False + + self._state = to + + event = self._emit( + EventType.STATE_TRANSITION, + actor=principal or self._principal, + transition=self._transition_payload(previous, to, reason), + ) + self._history.append( + TransitionRecord( + from_state=previous, + to_state=to, + at=event.occurred_at, + reason=reason, + principal=principal, + event_id=event.event_id, + ) + ) + if to is JobState.COMPLETED: + self._emit(EventType.JOB_COMPLETED, actor=principal or self._principal) + return event + + # ---- named transitions ------------------------------------------------- + # These exist so the arguments a guard requires are visible in the call + # signature rather than discovered at runtime. + + def submit_for_governance(self, *, reason: str | None = None) -> Event: + return self.transition(JobState.GOVERNANCE_ANALYSIS, reason=reason) + + def approve(self, *, authority: Principal, reason: str) -> Event: + return self.transition(JobState.APPROVED, reason=reason, principal=authority) + + def reject(self, *, authority: Principal, reason: str) -> Event: + return self.transition(JobState.REJECTED, reason=reason, principal=authority) + + def pause_for_approval(self, *, reason: str | None = None) -> Event: + return self.transition(JobState.AWAITING_APPROVAL, reason=reason) + + def resume(self, *, reason: str | None = None, principal: Principal | None = None) -> Event: + """Return to the state this job paused in.""" + if self._state is not JobState.AWAITING_APPROVAL or self._awaiting_from is None: + raise InvalidTransition( + self._state.value, "resume", sorted(s.value for s in self.allowed_targets()) + ) + return self.transition(self._awaiting_from, reason=reason, principal=principal) + + def fail(self, *, reason: str, principal: Principal | None = None) -> Event: + return self.transition(JobState.FAILED, reason=reason, principal=principal) + + def cancel(self, *, reason: str, principal: Principal) -> Event: + return self.transition(JobState.CANCELLED, reason=reason, principal=principal) + + def time_out(self, *, reason: str, principal: Principal | None = None) -> Event: + return self.transition(JobState.TIMED_OUT, reason=reason, principal=principal) + + def supersede(self, *, job_id: str, principal: Principal | None = None) -> "Job": + """Create the replacement job for a FAILED one. + + RFC-0007: recovery is a new job, not a revival. The replacement starts at + DRAFT and passes GOVERNANCE_ANALYSIS again, which is the guarantee — + resuming the old one would continue under an APPROVED granted to a plan + that has since failed. + """ + if self._state is not JobState.FAILED: + raise InvalidTransition( + self._state.value, "supersede", ["only a FAILED job can be superseded"] + ) + return Job( + job_id=job_id, + tenant_id=self._tenant_id, + workspace_id=self._workspace_id, + principal=principal or self._principal, + supersedes_job_id=self._job_id, + clock=self._clock, + ) + + # ---- guards ------------------------------------------------------------ + + def _check_not_terminal(self) -> None: + if self._state in TERMINAL: + raise TerminalState(self._state.value) + + def _check_edge(self, to: JobState) -> None: + allowed = self.allowed_targets() + if to in allowed: + return + # A refusal that has a specific reason says the specific reason. A generic + # table miss is the last resort, not the first answer. + if self._state is JobState.AWAITING_APPROVAL and to in APPROVAL_PAUSABLE: + raise WrongResumeState( + self._awaiting_from.value if self._awaiting_from else "unknown", to.value + ) + if to is JobState.AWAITING_APPROVAL: + raise MissingApprovalContext(self._state.value) + raise InvalidTransition( + self._state.value, to.value, sorted(s.value for s in allowed) + ) + + def _check_guards( + self, to: JobState, *, reason: str | None, principal: Principal | None + ) -> None: + if to in REASON_REQUIRED and not (reason and reason.strip()): + raise MissingReason(to.value) + if to is JobState.CANCELLED and principal is None: + raise MissingPrincipal(to.value) + if to in AUTHORITY_REQUIRED and (principal is None or not (reason and reason.strip())): + raise MissingAuthority(to.value) + # Structural backstop for the direction lock. The table already makes + # APPROVED the only way in, so this can only fire if the table is edited + # wrongly — which is exactly when it is worth having. + if to in POST_APPROVAL and not self._approved: + raise ExecutionBeforeApproval(to.value) + + # ---- emit -------------------------------------------------------------- + + def _transition_payload( + self, previous: JobState, to: JobState, reason: str | None + ) -> dict[str, Any]: + payload: dict[str, Any] = {"from": previous.value, "to": to.value} + # RFC-0008 forbids fabricating a value to fill a field. "No reason given" + # is expressed by the key being absent, never by an empty string. + if reason: + payload["reason"] = reason + return payload + + def _emit( + self, + event_type: EventType, + *, + actor: Principal | None = None, + transition: dict[str, Any] | None = None, + metadata: dict[str, Any] | None = None, + ) -> Event: + event = Event( + event_id=new_event_id(), + event_type=event_type, + tenant_id=self._tenant_id, + workspace_id=self._workspace_id, + subject_type="job", + subject_id=self._job_id, + job_id=self._job_id, + occurred_at=self._clock(), + actor=actor, + transition=transition, + metadata=metadata or {}, + ) + self._events.append(event) + return event + + def __repr__(self) -> str: + return ( + f"Job(job_id={self._job_id!r}, tenant_id={self._tenant_id!r}, " + f"state={self._state.value}, transitions={len(self._history)})" + ) diff --git a/packages/core/devfactory_core/states.py b/packages/core/devfactory_core/states.py new file mode 100644 index 0000000..771b927 --- /dev/null +++ b/packages/core/devfactory_core/states.py @@ -0,0 +1,130 @@ +"""Job lifecycle states and the transition table. + +Canonical spec: ``packages/core/state-machine.md`` — RFC-0001 as amended by RFC-0007. +This module is the single place the transition table is expressed in code; nothing +else may hard-code an edge. +""" + +from __future__ import annotations + +from enum import Enum + + +class JobState(str, Enum): + """The thirteen job states. + + ``str`` mixin so a state serialises as its own name in an event payload + without a conversion step at every emit site. + """ + + DRAFT = "DRAFT" + GOVERNANCE_ANALYSIS = "GOVERNANCE_ANALYSIS" + APPROVED = "APPROVED" + REJECTED = "REJECTED" + TASK_PLANNING = "TASK_PLANNING" + IN_PROGRESS = "IN_PROGRESS" + AWAITING_APPROVAL = "AWAITING_APPROVAL" + VALIDATING = "VALIDATING" + DEPLOYABLE = "DEPLOYABLE" + COMPLETED = "COMPLETED" + FAILED = "FAILED" + CANCELLED = "CANCELLED" + TIMED_OUT = "TIMED_OUT" + + +#: Terminal states. ``REJECTED`` is deliberately absent — it returns to ``DRAFT``. +TERMINAL: frozenset[JobState] = frozenset( + {JobState.COMPLETED, JobState.FAILED, JobState.CANCELLED, JobState.TIMED_OUT} +) + +#: States from which execution has been authorised. Reaching any of these without +#: passing APPROVED would violate "execution is forbidden before APPROVED". +POST_APPROVAL: frozenset[JobState] = frozenset( + { + JobState.TASK_PLANNING, + JobState.IN_PROGRESS, + JobState.AWAITING_APPROVAL, + JobState.VALIDATING, + JobState.DEPLOYABLE, + } +) + +#: States a job can pause in to wait for a human, and return to afterwards. +APPROVAL_PAUSABLE: frozenset[JobState] = frozenset( + {JobState.IN_PROGRESS, JobState.VALIDATING, JobState.DEPLOYABLE} +) + +#: RFC-0007: TIMED_OUT is reachable from these. AWAITING_APPROVAL is included +#: because an approval nobody answers is how a governed pipeline usually stalls. +TIMEOUTABLE: frozenset[JobState] = frozenset( + { + JobState.GOVERNANCE_ANALYSIS, + JobState.TASK_PLANNING, + JobState.IN_PROGRESS, + JobState.AWAITING_APPROVAL, + JobState.VALIDATING, + } +) + +#: A job fails only where work exists to fail. Before APPROVED nothing is +#: executing, so the honest outcomes there are REJECTED, CANCELLED, or TIMED_OUT. +#: See "Open question" in the module docstring of ``job.py``. +FAILABLE: frozenset[JobState] = frozenset( + { + JobState.TASK_PLANNING, + JobState.IN_PROGRESS, + JobState.AWAITING_APPROVAL, + JobState.VALIDATING, + JobState.DEPLOYABLE, + } +) + +# The lifecycle proper, before the cross-cutting exits are folded in. +_PROGRESSION: dict[JobState, frozenset[JobState]] = { + JobState.DRAFT: frozenset({JobState.GOVERNANCE_ANALYSIS}), + JobState.GOVERNANCE_ANALYSIS: frozenset({JobState.APPROVED, JobState.REJECTED}), + JobState.APPROVED: frozenset({JobState.TASK_PLANNING}), + JobState.REJECTED: frozenset({JobState.DRAFT}), + JobState.TASK_PLANNING: frozenset({JobState.IN_PROGRESS}), + JobState.IN_PROGRESS: frozenset({JobState.VALIDATING, JobState.AWAITING_APPROVAL}), + JobState.VALIDATING: frozenset({JobState.DEPLOYABLE, JobState.AWAITING_APPROVAL}), + JobState.DEPLOYABLE: frozenset({JobState.COMPLETED, JobState.AWAITING_APPROVAL}), + # The return edge out of AWAITING_APPROVAL is `awaiting_from` and is resolved + # per job at runtime, not from this table. + JobState.AWAITING_APPROVAL: frozenset(), + JobState.COMPLETED: frozenset(), + JobState.FAILED: frozenset(), + JobState.CANCELLED: frozenset(), + JobState.TIMED_OUT: frozenset(), +} + + +def _build() -> dict[JobState, frozenset[JobState]]: + table: dict[JobState, set[JobState]] = { + state: set(targets) for state, targets in _PROGRESSION.items() + } + for state in JobState: + if state in TERMINAL: + continue + # CANCELLED is reachable from every non-terminal state — a job can be + # stopped at any point before it settles. + table[state].add(JobState.CANCELLED) + if state in TIMEOUTABLE: + table[state].add(JobState.TIMED_OUT) + if state in FAILABLE: + table[state].add(JobState.FAILED) + return {state: frozenset(targets) for state, targets in table.items()} + + +#: Static transition table. ``AWAITING_APPROVAL`` also permits its job's +#: ``awaiting_from``, which cannot live here because it differs per job. +TRANSITIONS: dict[JobState, frozenset[JobState]] = _build() + + +def static_targets(state: JobState) -> frozenset[JobState]: + """States reachable from ``state`` without per-job context.""" + return TRANSITIONS[state] + + +def is_terminal(state: JobState) -> bool: + return state in TERMINAL diff --git a/packages/core/pyproject.toml b/packages/core/pyproject.toml new file mode 100644 index 0000000..daa1ff6 --- /dev/null +++ b/packages/core/pyproject.toml @@ -0,0 +1,32 @@ +[project] +name = "devfactory-core" +version = "0.1.0" +description = "Job state machine for the devfactory-core control plane" +requires-python = ">=3.11" +# SPDX string (PEP 639). A file= path cannot point outside this directory, +# and LICENSE lives at the repository root. +license = "MIT" +# No runtime dependencies. CORE_BOUNDARY.md forbids vendor-coupled frameworks in +# v0.x, and a lifecycle engine does not need one. +dependencies = [] + +[project.optional-dependencies] +test = ["pytest>=8", "pytest-cov>=5"] + +[build-system] +requires = ["setuptools>=77"] +build-backend = "setuptools.build_meta" + +[tool.setuptools.packages.find] +include = ["devfactory_core*"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["."] +addopts = "-q" + +[tool.coverage.report] +# Acceptance criterion of issue #2. The engine is pure logic with no I/O, so +# there is no honest excuse for an uncovered branch. +fail_under = 90 +show_missing = true diff --git a/packages/core/tests/conftest.py b/packages/core/tests/conftest.py new file mode 100644 index 0000000..9dbcd68 --- /dev/null +++ b/packages/core/tests/conftest.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from devfactory_core import Job, JobState, Principal # noqa: E402 + + +@pytest.fixture +def alice() -> Principal: + return Principal("human", "alice", display_name="Alice") + + +@pytest.fixture +def planner() -> Principal: + return Principal("agent", "planner-1") + + +@pytest.fixture +def clock(): + """Monotonic fake clock, so ordering assertions do not depend on wall time.""" + start = datetime(2026, 8, 18, 9, 0, tzinfo=timezone.utc) + state = {"n": 0} + + def tick() -> datetime: + state["n"] += 1 + return start + timedelta(seconds=state["n"]) + + return tick + + +@pytest.fixture +def job(alice, clock) -> Job: + return Job( + job_id="job-001", + tenant_id="default", + workspace_id="ws-core", + principal=alice, + clock=clock, + ) + + +def drive(job: Job, target: JobState, authority: Principal) -> Job: + """Walk a fresh job along the happy path until it reaches ``target``. + + TIMED_OUT is reached from VALIDATING rather than DEPLOYABLE — DEPLOYABLE is + deliberately not in ``TIMEOUTABLE``, since nothing there is waiting on a clock. + """ + if target is JobState.DRAFT: + return job + job.submit_for_governance() + if target is JobState.GOVERNANCE_ANALYSIS: + return job + if target is JobState.REJECTED: + job.reject(authority=authority, reason="out of scope") + return job + job.approve(authority=authority, reason="approved for milestone") + if target is JobState.APPROVED: + return job + job.transition(JobState.TASK_PLANNING) + if target is JobState.TASK_PLANNING: + return job + job.transition(JobState.IN_PROGRESS) + if target is JobState.IN_PROGRESS: + return job + if target is JobState.AWAITING_APPROVAL: + job.pause_for_approval() + return job + job.transition(JobState.VALIDATING) + if target is JobState.VALIDATING: + return job + if target is JobState.TIMED_OUT: + job.time_out(reason="sla exceeded") + return job + if target is JobState.FAILED: + job.fail(reason="orchestration exhausted retries") + return job + if target is JobState.CANCELLED: + job.cancel(reason="stopped by owner", principal=authority) + return job + job.transition(JobState.DEPLOYABLE) + if target is JobState.DEPLOYABLE: + return job + if target is JobState.COMPLETED: + job.transition(JobState.COMPLETED) + return job + raise AssertionError(f"drive() has no path to {target}") diff --git a/packages/core/tests/test_events.py b/packages/core/tests/test_events.py new file mode 100644 index 0000000..0a1a43b --- /dev/null +++ b/packages/core/tests/test_events.py @@ -0,0 +1,189 @@ +"""The audit trail — append-only, complete, and shaped to event/v1.""" + +from __future__ import annotations + +import pytest + +from conftest import drive +from devfactory_core import Event, EventType, Job, JobState +from devfactory_core.events import new_event_id +from devfactory_core.identity import ID_PATTERN + + +def _fresh(alice, clock) -> Job: + return Job( + job_id="job-001", tenant_id="acme", workspace_id="ws-core", principal=alice, clock=clock + ) + + +def test_creation_emits_job_created(alice, clock): + job = _fresh(alice, clock) + assert [e.event_type for e in job.events] == [EventType.JOB_CREATED] + + +def test_every_transition_emits_exactly_one_state_transition(alice, clock): + """RFC-0003: no silent state change.""" + job = drive(_fresh(alice, clock), JobState.DEPLOYABLE, alice) + transitions = [e for e in job.events if e.event_type is EventType.STATE_TRANSITION] + assert len(transitions) == len(job.history) + + +def test_completion_emits_job_completed(alice, clock): + job = drive(_fresh(alice, clock), JobState.COMPLETED, alice) + assert job.events[-1].event_type is EventType.JOB_COMPLETED + + +def test_history_and_events_are_read_only_copies(alice, clock): + job = drive(_fresh(alice, clock), JobState.IN_PROGRESS, alice) + before = len(job.events) + job.events # a tuple — the caller cannot append to the live log + assert isinstance(job.events, tuple) + assert isinstance(job.history, tuple) + with pytest.raises(AttributeError): + job.events.append("forged") # type: ignore[attr-defined] + assert len(job.events) == before + + +def test_events_are_frozen(alice, clock): + job = _fresh(alice, clock) + with pytest.raises(Exception): + job.events[0].tenant_id = "other" # type: ignore[misc] + + +def test_refused_transition_writes_nothing(alice, clock): + """A rejected call must not leave a trace — the log records what happened.""" + job = _fresh(alice, clock) + before = len(job.events) + with pytest.raises(Exception): + job.transition(JobState.COMPLETED) + assert len(job.events) == before + assert job.history == () + + +def test_event_ids_are_unique_and_well_formed(alice, clock): + job = drive(_fresh(alice, clock), JobState.COMPLETED, alice) + ids = [e.event_id for e in job.events] + assert len(set(ids)) == len(ids) + assert all(ID_PATTERN.match(i) for i in ids) + + +def test_new_event_id_matches_identity_v1(): + assert all(ID_PATTERN.match(new_event_id()) for _ in range(100)) + + +def test_events_are_ordered_by_occurrence(alice, clock): + job = drive(_fresh(alice, clock), JobState.COMPLETED, alice) + stamps = [e.occurred_at for e in job.events] + assert stamps == sorted(stamps) + + +# ---- event/v1 wire shape --------------------------------------------------- + +REQUIRED = {"event_id", "event_type", "tenant_id", "subject_type", "subject_id", "occurred_at", "source"} + + +def test_every_payload_carries_the_required_fields(alice, clock): + job = drive(_fresh(alice, clock), JobState.COMPLETED, alice) + for payload in job.event_payloads(): + assert REQUIRED <= set(payload), f"missing {REQUIRED - set(payload)}" + + +def test_our_events_always_carry_job_id(alice, clock): + """RFC-0008: optional in the schema, not optional in our behaviour.""" + job = drive(_fresh(alice, clock), JobState.COMPLETED, alice) + assert all(p["job_id"] == "job-001" for p in job.event_payloads()) + + +def test_subject_is_always_answerable(alice, clock): + job = drive(_fresh(alice, clock), JobState.COMPLETED, alice) + for payload in job.event_payloads(): + assert payload["subject_type"] == "job" + assert payload["subject_id"] == "job-001" + + +def test_tenant_scope_on_every_event(alice, clock): + job = drive(_fresh(alice, clock), JobState.COMPLETED, alice) + assert all(p["tenant_id"] == "acme" for p in job.event_payloads()) + assert all(p["workspace_id"] == "ws-core" for p in job.event_payloads()) + + +def test_source_marks_events_as_internal(alice, clock): + job = drive(_fresh(alice, clock), JobState.IN_PROGRESS, alice) + for payload in job.event_payloads(): + assert payload["source"] == {"kind": "internal", "system": "devfactory-core"} + + +def test_transition_payload_carries_from_and_to(alice, clock): + job = _fresh(alice, clock) + event = job.submit_for_governance(reason="ready") + assert event.transition == { + "from": "DRAFT", + "to": "GOVERNANCE_ANALYSIS", + "reason": "ready", + } + + +def test_absent_reason_is_absent_not_empty(alice, clock): + """RFC-0008: nothing is fabricated to fill a field. No reason means no key.""" + job = _fresh(alice, clock) + event = job.submit_for_governance() + assert "reason" not in event.transition + + +def test_optional_keys_are_omitted_when_unset(alice, clock): + job = _fresh(alice, clock) + payload = job.events[0].as_payload() + assert "transition" not in payload + assert "metadata" not in payload + + +def test_decision_events_name_the_authority(alice, planner, clock): + """Every APPROVE is auditable — the record says who signed it.""" + job = _fresh(alice, clock) + job.submit_for_governance() + event = job.approve(authority=planner, reason="meets milestone") + assert event.as_payload()["actor"]["id"] == "planner-1" + + +def test_transition_returns_the_emitted_event(alice, clock): + job = _fresh(alice, clock) + event = job.submit_for_governance() + assert isinstance(event, Event) + assert event is job.events[-1] + assert job.history[-1].event_id == event.event_id + + +def test_awaiting_approval_is_visible_in_the_trail(alice, clock): + """The point of RFC-0007: a job waiting on a human must be visible as one.""" + job = drive(_fresh(alice, clock), JobState.IN_PROGRESS, alice) + job.pause_for_approval(reason="merge needs sign-off") + paused = [ + p for p in job.event_payloads() + if p.get("transition", {}).get("to") == "AWAITING_APPROVAL" + ] + assert len(paused) == 1 + assert paused[0]["transition"]["reason"] == "merge needs sign-off" + + +def test_repr_is_useful(alice, clock): + job = drive(_fresh(alice, clock), JobState.IN_PROGRESS, alice) + text = repr(job) + assert "job-001" in text and "IN_PROGRESS" in text + + +def test_default_clock_is_utc(): + """Tests inject a fake clock; the real one still has to be timezone-aware.""" + from datetime import timezone + + from devfactory_core.events import utc_now + + assert utc_now().tzinfo is timezone.utc + + +def test_job_exposes_its_identity_and_settlement(alice, clock): + job = _fresh(alice, clock) + assert job.job_id == "job-001" + assert job.principal is alice + assert job.is_terminal is False + drive(job, JobState.COMPLETED, alice) + assert job.is_terminal is True diff --git a/packages/core/tests/test_guards.py b/packages/core/tests/test_guards.py new file mode 100644 index 0000000..17ecd96 --- /dev/null +++ b/packages/core/tests/test_guards.py @@ -0,0 +1,209 @@ +"""The guards — the refusals that make the lifecycle enforceable.""" + +from __future__ import annotations + +import pytest + +from conftest import drive +from devfactory_core import Job, JobState, Principal +from devfactory_core.errors import ( + ExecutionBeforeApproval, + InvalidIdentifier, + InvalidTransition, + MissingApprovalContext, + MissingAuthority, + MissingPrincipal, + MissingReason, + WrongResumeState, +) + + +def _fresh(alice, clock, **kw) -> Job: + base = dict( + job_id="job-001", tenant_id="default", workspace_id="ws-core", principal=alice, clock=clock + ) + base.update(kw) + return Job(**base) + + +# ---- tenant scope, RFC-0006 ------------------------------------------------ + + +def test_tenant_and_workspace_are_required(alice, clock): + with pytest.raises(TypeError): + Job(job_id="job-001", tenant_id="default", principal=alice, clock=clock) + + +@pytest.mark.parametrize( + "bad", ["", "Job-001", "job 001", "_job", "-job", "j" * 64, "JOB"], ids=repr +) +def test_identifiers_must_match_identity_v1(bad, alice, clock): + with pytest.raises(InvalidIdentifier): + _fresh(alice, clock, job_id=bad) + + +def test_single_tenant_uses_the_default_literal(alice, clock): + """RFC-0006: never omit the field — the payload shape is already right.""" + from devfactory_core import DEFAULT_TENANT + + job = _fresh(alice, clock, tenant_id=DEFAULT_TENANT) + assert job.tenant_id == "default" + assert all(e.tenant_id == "default" for e in job.events) + + +def test_principal_must_be_a_principal(alice, clock): + with pytest.raises(TypeError): + _fresh(alice, clock, principal="alice") + + +def test_principal_type_is_constrained(): + with pytest.raises(ValueError): + Principal("robot", "r2") + + +def test_delegation_chain_is_recorded(clock): + alice = Principal("human", "alice") + agent = Principal("agent", "planner-1", on_behalf_of=alice) + job = _fresh(alice, clock, principal=agent) + payload = job.events[0].as_payload() + assert payload["actor"]["on_behalf_of"]["id"] == "alice" + + +# ---- reason and authority -------------------------------------------------- + + +@pytest.mark.parametrize( + "target", [JobState.FAILED, JobState.CANCELLED, JobState.TIMED_OUT], ids=lambda s: s.value +) +def test_stopping_states_require_a_reason(target, alice, clock): + job = drive(_fresh(alice, clock), JobState.VALIDATING, alice) + with pytest.raises(MissingReason): + job.transition(target, principal=alice) + assert job.state is JobState.VALIDATING + + +@pytest.mark.parametrize("blank", ["", " ", "\n"], ids=repr) +def test_blank_reason_is_not_a_reason(blank, alice, clock): + job = drive(_fresh(alice, clock), JobState.IN_PROGRESS, alice) + with pytest.raises(MissingReason): + job.fail(reason=blank) + + +def test_cancel_records_who_cancelled(alice, clock): + job = drive(_fresh(alice, clock), JobState.IN_PROGRESS, alice) + with pytest.raises(MissingPrincipal): + job.transition(JobState.CANCELLED, reason="stopped") + job.cancel(reason="stopped", principal=alice) + assert job.history[-1].principal is alice + + +@pytest.mark.parametrize("target", [JobState.APPROVED, JobState.REJECTED], ids=lambda s: s.value) +def test_decisions_require_an_authority_and_a_reason(target, alice, clock): + job = drive(_fresh(alice, clock), JobState.GOVERNANCE_ANALYSIS, alice) + with pytest.raises(MissingAuthority): + job.transition(target, reason="because") + with pytest.raises(MissingAuthority): + job.transition(target, principal=alice) + assert job.state is JobState.GOVERNANCE_ANALYSIS + + +# ---- the direction lock ---------------------------------------------------- + + +def test_execution_is_forbidden_before_approval(alice, clock): + job = _fresh(alice, clock) + for target in (JobState.TASK_PLANNING, JobState.IN_PROGRESS, JobState.VALIDATING): + with pytest.raises(InvalidTransition): + job.transition(target) + + +def test_approval_flag_is_a_backstop_not_the_only_check(alice, clock): + """If the table were edited wrongly, the guard still refuses.""" + job = drive(_fresh(alice, clock), JobState.GOVERNANCE_ANALYSIS, alice) + job._state = JobState.APPROVED # bypass the engine, simulating a bad edit + with pytest.raises(ExecutionBeforeApproval): + job.transition(JobState.TASK_PLANNING) + + +# ---- mid-run approval, RFC-0007 ------------------------------------------- + + +@pytest.mark.parametrize( + "state", [JobState.IN_PROGRESS, JobState.VALIDATING, JobState.DEPLOYABLE], ids=lambda s: s.value +) +def test_pause_and_resume_returns_to_where_it_paused(state, alice, clock): + job = drive(_fresh(alice, clock), state, alice) + job.pause_for_approval(reason="needs sign-off") + assert job.state is JobState.AWAITING_APPROVAL + assert job.awaiting_from is state + job.resume(reason="approved", principal=alice) + assert job.state is state + assert job.awaiting_from is None + + +def test_cannot_pause_before_execution(alice, clock): + job = drive(_fresh(alice, clock), JobState.TASK_PLANNING, alice) + with pytest.raises(MissingApprovalContext): + job.pause_for_approval() + + +def test_resuming_into_the_wrong_state_is_refused(alice, clock): + job = drive(_fresh(alice, clock), JobState.DEPLOYABLE, alice) + job.pause_for_approval() + with pytest.raises(WrongResumeState) as excinfo: + job.transition(JobState.IN_PROGRESS) + assert excinfo.value.awaiting_from == "DEPLOYABLE" + assert job.state is JobState.AWAITING_APPROVAL + + +def test_resume_outside_awaiting_approval_is_refused(alice, clock): + job = drive(_fresh(alice, clock), JobState.IN_PROGRESS, alice) + with pytest.raises(InvalidTransition): + job.resume() + + +def test_denied_mid_run_approval_can_fail_the_job(alice, clock): + job = drive(_fresh(alice, clock), JobState.IN_PROGRESS, alice) + job.pause_for_approval() + job.fail(reason="approval denied and no path forward") + assert job.state is JobState.FAILED + + +def test_unanswered_approval_times_out(alice, clock): + """RFC-0007: an approval nobody answers is how a governed pipeline stalls.""" + job = drive(_fresh(alice, clock), JobState.IN_PROGRESS, alice) + job.pause_for_approval() + job.time_out(reason="approval request expired") + assert job.state is JobState.TIMED_OUT + + +# ---- recovery, RFC-0007 ---------------------------------------------------- + + +def test_failed_job_is_superseded_not_revived(alice, clock): + job = drive(_fresh(alice, clock), JobState.FAILED, alice) + replacement = job.supersede(job_id="job-002") + assert replacement.state is JobState.DRAFT + assert replacement.supersedes_job_id == "job-001" + assert replacement.tenant_id == job.tenant_id + assert replacement.workspace_id == job.workspace_id + + +def test_replacement_must_pass_governance_again(alice, clock): + job = drive(_fresh(alice, clock), JobState.FAILED, alice) + replacement = job.supersede(job_id="job-002") + with pytest.raises(InvalidTransition): + replacement.transition(JobState.TASK_PLANNING) + + +def test_only_a_failed_job_can_be_superseded(alice, clock): + job = drive(_fresh(alice, clock), JobState.IN_PROGRESS, alice) + with pytest.raises(InvalidTransition): + job.supersede(job_id="job-002") + + +def test_supersede_records_the_link_in_the_audit_trail(alice, clock): + job = drive(_fresh(alice, clock), JobState.FAILED, alice) + replacement = job.supersede(job_id="job-002") + created = replacement.events[0].as_payload() + assert created["metadata"]["supersedes_job_id"] == "job-001" diff --git a/packages/core/tests/test_states.py b/packages/core/tests/test_states.py new file mode 100644 index 0000000..fc22962 --- /dev/null +++ b/packages/core/tests/test_states.py @@ -0,0 +1,120 @@ +"""The transition table itself — shape, not behaviour.""" + +from __future__ import annotations + +import pytest + +from devfactory_core.states import ( + APPROVAL_PAUSABLE, + FAILABLE, + POST_APPROVAL, + TERMINAL, + TIMEOUTABLE, + TRANSITIONS, + JobState, + is_terminal, + static_targets, +) + + +def test_thirteen_states(): + """RFC-0007 amended RFC-0001's ten to thirteen. Issue #2 predates that.""" + assert len(JobState) == 13 + + +def test_every_state_has_a_table_entry(): + assert set(TRANSITIONS) == set(JobState) + + +def test_terminal_states_are_exactly_four(): + assert TERMINAL == { + JobState.COMPLETED, + JobState.FAILED, + JobState.CANCELLED, + JobState.TIMED_OUT, + } + + +def test_rejected_is_not_terminal(): + """RFC-0001: REJECTED returns to DRAFT so the job can be revised.""" + assert JobState.REJECTED not in TERMINAL + assert JobState.DRAFT in static_targets(JobState.REJECTED) + + +@pytest.mark.parametrize("state", sorted(TERMINAL, key=lambda s: s.value)) +def test_terminal_states_have_no_outgoing_edges(state): + assert static_targets(state) == frozenset() + assert is_terminal(state) + + +@pytest.mark.parametrize( + "state", [s for s in JobState if s not in TERMINAL], ids=lambda s: s.value +) +def test_cancellable_from_every_non_terminal_state(state): + """RFC-0007: a job can be stopped at any point before it settles.""" + assert JobState.CANCELLED in static_targets(state) + + +@pytest.mark.parametrize("state", sorted(TIMEOUTABLE, key=lambda s: s.value)) +def test_timeout_reachable_where_specified(state): + assert JobState.TIMED_OUT in static_targets(state) + + +def test_timeout_not_reachable_from_draft_or_approved(): + """Nothing is waiting on anything in DRAFT or APPROVED, so nothing can expire.""" + assert JobState.TIMED_OUT not in static_targets(JobState.DRAFT) + assert JobState.TIMED_OUT not in static_targets(JobState.APPROVED) + + +@pytest.mark.parametrize("state", sorted(FAILABLE, key=lambda s: s.value)) +def test_failed_reachable_only_where_work_exists(state): + assert JobState.FAILED in static_targets(state) + + +@pytest.mark.parametrize( + "state", + [JobState.DRAFT, JobState.GOVERNANCE_ANALYSIS, JobState.APPROVED, JobState.REJECTED], + ids=lambda s: s.value, +) +def test_failed_not_reachable_before_execution(state): + """Before APPROVED nothing runs, so REJECTED, CANCELLED, or TIMED_OUT apply.""" + assert JobState.FAILED not in static_targets(state) + + +def test_approved_is_the_only_gate_into_execution(): + """Direction lock: every path into POST_APPROVAL passes through APPROVED.""" + entrances = { + source + for source, targets in TRANSITIONS.items() + if targets & POST_APPROVAL and source not in POST_APPROVAL + } + assert entrances == {JobState.APPROVED} + + +def test_awaiting_approval_has_no_static_return_edge(): + """The way back is `awaiting_from`, which differs per job.""" + assert static_targets(JobState.AWAITING_APPROVAL) & APPROVAL_PAUSABLE == frozenset() + + +@pytest.mark.parametrize("state", sorted(APPROVAL_PAUSABLE, key=lambda s: s.value)) +def test_pausable_states_can_reach_awaiting_approval(state): + assert JobState.AWAITING_APPROVAL in static_targets(state) + + +def test_every_state_is_reachable_from_draft(): + """No orphans — a state nothing can reach is a state that does not exist.""" + seen = {JobState.DRAFT} + frontier = [JobState.DRAFT] + while frontier: + for target in static_targets(frontier.pop()): + if target not in seen: + seen.add(target) + frontier.append(target) + # AWAITING_APPROVAL's return edge is dynamic, so a static walk still reaches + # every state; nothing depends on it to be discovered. + assert seen == set(JobState) + + +def test_state_serialises_as_its_name(): + assert JobState.IN_PROGRESS.value == "IN_PROGRESS" + assert f"{JobState.IN_PROGRESS}" .endswith("IN_PROGRESS") diff --git a/packages/core/tests/test_transitions.py b/packages/core/tests/test_transitions.py new file mode 100644 index 0000000..680e3fb --- /dev/null +++ b/packages/core/tests/test_transitions.py @@ -0,0 +1,127 @@ +"""Every valid edge is walkable; every invalid one is refused.""" + +from __future__ import annotations + +import pytest + +from conftest import drive +from devfactory_core import Job, JobState, Principal +from devfactory_core.errors import InvalidTransition, JobStateMachineError, TerminalState +from devfactory_core.states import TERMINAL, TRANSITIONS + + +def _fresh(alice: Principal, clock, name: str = "job-001") -> Job: + return Job( + job_id=name, tenant_id="default", workspace_id="ws-core", principal=alice, clock=clock + ) + + +def _args_for(target: JobState, authority: Principal) -> dict: + """The guard-required arguments for entering ``target``.""" + if target in (JobState.APPROVED, JobState.REJECTED): + return {"reason": "decided", "principal": authority} + if target is JobState.CANCELLED: + return {"reason": "stopped", "principal": authority} + if target in (JobState.FAILED, JobState.TIMED_OUT): + return {"reason": "stopped"} + return {} + + +@pytest.mark.parametrize("state", list(JobState), ids=lambda s: s.value) +def test_drive_reaches_every_state(state, alice, clock): + job = drive(_fresh(alice, clock), state, alice) + assert job.state is state + + +@pytest.mark.parametrize( + "source,target", + [ + (source, target) + for source, targets in TRANSITIONS.items() + for target in sorted(targets, key=lambda s: s.value) + ], + ids=lambda v: v.value if isinstance(v, JobState) else str(v), +) +def test_every_static_edge_is_walkable(source, target, alice, clock): + job = drive(_fresh(alice, clock), source, alice) + assert job.state is source + job.transition(target, **_args_for(target, alice)) + assert job.state is target + + +@pytest.mark.parametrize( + "source,target", + [ + (source, target) + for source in JobState + if source not in TERMINAL + for target in JobState + if target not in TRANSITIONS[source] + # AWAITING_APPROVAL's return edge is dynamic; covered in test_guards. + and not (source is JobState.AWAITING_APPROVAL and target is JobState.IN_PROGRESS) + ], + ids=lambda v: v.value if isinstance(v, JobState) else str(v), +) +def test_every_non_edge_is_refused(source, target, alice, clock): + job = drive(_fresh(alice, clock), source, alice) + before = job.state + # Any refusal is correct here; which specific one is asserted in test_guards. + with pytest.raises(JobStateMachineError): + job.transition(target, **_args_for(target, alice)) + assert job.state is before, "a refused transition must leave the job untouched" + + +@pytest.mark.parametrize("state", sorted(TERMINAL, key=lambda s: s.value)) +def test_terminal_states_refuse_everything(state, alice, clock): + job = drive(_fresh(alice, clock), state, alice) + for target in JobState: + with pytest.raises(TerminalState): + job.transition(target, **_args_for(target, alice)) + assert job.state is state + + +def test_rejected_returns_to_draft_and_can_be_resubmitted(alice, clock): + job = drive(_fresh(alice, clock), JobState.REJECTED, alice) + job.transition(JobState.DRAFT) + assert job.state is JobState.DRAFT + job.submit_for_governance() + job.approve(authority=alice, reason="scope fixed") + assert job.state is JobState.APPROVED + + +def test_rejection_clears_a_previous_approval(alice, clock): + """An approval granted to an earlier revision must not authorise a revised one.""" + job = _fresh(alice, clock) + job.submit_for_governance() + job.approve(authority=alice, reason="first pass") + job.transition(JobState.TASK_PLANNING) + job.fail(reason="plan does not work") + + replacement = job.supersede(job_id="job-002") + replacement.submit_for_governance() + replacement.reject(authority=alice, reason="still wrong") + replacement.transition(JobState.DRAFT) + with pytest.raises(InvalidTransition): + replacement.transition(JobState.TASK_PLANNING) + + +def test_invalid_transition_names_what_was_allowed(alice, clock): + job = _fresh(alice, clock) + with pytest.raises(InvalidTransition) as excinfo: + job.transition(JobState.COMPLETED) + assert excinfo.value.current == "DRAFT" + assert excinfo.value.requested == "COMPLETED" + assert "GOVERNANCE_ANALYSIS" in excinfo.value.allowed + + +def test_terminal_error_points_at_supersede(alice, clock): + job = drive(_fresh(alice, clock), JobState.FAILED, alice) + with pytest.raises(TerminalState) as excinfo: + job.transition(JobState.IN_PROGRESS) + assert "supersedes_job_id" in str(excinfo.value) + + +def test_transition_accepts_a_plain_string(alice, clock): + job = _fresh(alice, clock) + job.transition("GOVERNANCE_ANALYSIS") + assert job.state is JobState.GOVERNANCE_ANALYSIS