diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index ac0dc20..e1096ac 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -1,7 +1,8 @@ name: test # The release gate ADR-0006 requires of a consumer: unit tests for both packages -# plus payload conformance against the pinned contracts. +# plus payload conformance against the pinned contracts. Issue #7 adds the +# end-to-end flow simulation, which is the closing gate on milestone v0.1. # # A failing job here has to block the merge for this to be a gate rather than a # report. That needs branch protection with these checks marked required, which is @@ -10,9 +11,19 @@ name: test on: push: branches: [main] - paths: ['packages/**', 'apps/**', 'conformance/**', '.github/workflows/test.yml'] + paths: + - 'packages/**' + - 'apps/**' + - 'conformance/**' + - 'simulation/**' + - '.github/workflows/test.yml' pull_request: - paths: ['packages/**', 'apps/**', 'conformance/**', '.github/workflows/test.yml'] + paths: + - 'packages/**' + - 'apps/**' + - 'conformance/**' + - 'simulation/**' + - '.github/workflows/test.yml' workflow_dispatch: permissions: @@ -47,6 +58,29 @@ jobs: working-directory: packages/observability run: python -m pytest --cov=devfactory_observability --cov-report=term-missing + simulation: + # Issue #7 — the end-to-end flow simulation. Runs both forms deliberately: + # the test suite is what fails a pull request, and the script is the + # deliverable the issue asks for, so a script that stopped running would be a + # regression the suite alone might not notice. + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + + - name: Install + run: pip install --no-cache-dir 'pytest>=8' + + - name: End-to-end flow suite + run: python3 -m pytest simulation/tests + + - name: End-to-end flow simulation + run: python3 simulation/e2e_flow.py + conformance: # ADR-0006 requirement 2 — validate payloads the engine actually produces # against the contracts pinned in conformance/pinned.yaml. diff --git a/conformance/payload_check.py b/conformance/payload_check.py index f144115..e04a479 100644 --- a/conformance/payload_check.py +++ b/conformance/payload_check.py @@ -40,7 +40,11 @@ import urllib.request ROOT = pathlib.Path(__file__).resolve().parents[1] -sys.path[:0] = [str(ROOT / "packages" / "core"), str(ROOT / "packages" / "observability")] +sys.path[:0] = [ + str(ROOT), + str(ROOT / "packages" / "core"), + str(ROOT / "packages" / "observability"), +] PINNED = ROOT / "conformance" / "pinned.yaml" CACHE = ROOT / "conformance" / ".schema_cache" @@ -136,63 +140,51 @@ def run_scenario(): Six jobs across two tenants, covering every terminal state, the mid-run approval pause, rejection and resubmission, and recovery by supersession — plus two inbound external events that no job caused. + + The journeys themselves are ``simulation/flows.py``, which issue #7 made the + one place they are written down. Two files describing the same lifecycle + differently is how they end up disagreeing about it, and this check exists to + catch disagreement rather than to add one. """ - from devfactory_core import Job, JobState, Principal + from devfactory_core import JobState, Principal from devfactory_observability import EventLog, accept_external + from simulation.flows import ( + cancelled_by_a_person, + failed_at, + happy_path, + job_factory, + rejected_then_resubmitted, + stalled_awaiting_approval, + ) log = EventLog() owner = Principal("human", "alice", display_name="Alice") reviewer = Principal("human", "bob", display_name="Bob") - agent = Principal("agent", "planner-1", on_behalf_of=owner) - - def new(job_id: str, tenant: str = "acme", workspace: str = "ws-core") -> Job: - return Job( - job_id=job_id, tenant_id=tenant, workspace_id=workspace, principal=owner - ) + acme = job_factory(tenant_id="acme", workspace_id="ws-core", principal=owner) + globex = job_factory( + tenant_id="globex", workspace_id="ws-platform", principal=owner + ) # 1. the happy path, with a mid-run approval pause before deploy - happy = new("job-001") - happy.submit_for_governance(reason="ready for review") - happy.approve(authority=reviewer, reason="scope matches milestone v0.1") - happy.transition(JobState.TASK_PLANNING) - happy.transition(JobState.IN_PROGRESS) - happy.transition(JobState.VALIDATING) - happy.transition(JobState.DEPLOYABLE) - happy.pause_for_approval(reason="deploy needs sign-off") - happy.resume(reason="signed off", principal=reviewer) - happy.transition(JobState.COMPLETED) + happy = happy_path( + acme, job_id="job-001", authority=reviewer, pause_in=JobState.DEPLOYABLE + ) # 2. rejected, revised, resubmitted, approved - revised = new("job-002") - revised.submit_for_governance() - revised.reject(authority=reviewer, reason="missing risk analysis") - revised.transition(JobState.DRAFT) - revised.submit_for_governance(reason="risk analysis added") - revised.approve(authority=reviewer, reason="addressed") + revised = rejected_then_resubmitted(acme, job_id="job-002", authority=reviewer) # 3. failed, then superseded by a fresh job that re-enters governance - failed = new("job-003") - failed.submit_for_governance() - failed.approve(authority=reviewer, reason="approved") - failed.transition(JobState.TASK_PLANNING) - failed.transition(JobState.IN_PROGRESS) - failed.fail(reason="orchestration exhausted execution retries") + failed = failed_at( + acme, job_id="job-003", authority=reviewer, state=JobState.IN_PROGRESS + ) replacement = failed.supersede(job_id="job-004", principal=owner) replacement.submit_for_governance(reason="retry with a different plan") # 4. cancelled by a person - cancelled = new("job-005") - cancelled.submit_for_governance() - cancelled.cancel(reason="superseded by an urgent request", principal=owner) - - # 5. an approval nobody answered - stalled = new("job-006", tenant="globex", workspace="ws-platform") - stalled.submit_for_governance() - stalled.approve(authority=reviewer, reason="approved") - stalled.transition(JobState.TASK_PLANNING) - stalled.transition(JobState.IN_PROGRESS) - stalled.pause_for_approval(reason="needs a human before merge") - stalled.time_out(reason="approval request expired after 72h") + cancelled = cancelled_by_a_person(acme, job_id="job-005", owner=owner) + + # 5. an approval nobody answered, in a second tenant + stalled = stalled_awaiting_approval(globex, job_id="job-006", authority=reviewer) jobs = [happy, revised, failed, replacement, cancelled, stalled] for job in jobs: diff --git a/packages/core/README.md b/packages/core/README.md index 759f335..c7a1d54 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -84,7 +84,9 @@ is what makes *every APPROVE is auditable* hold. `Decision.as_payload()` renders a decision in `approval/v1` shape. Neither is 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. -`conformance/payload_check.py` validates both against the pinned contracts. +`conformance/payload_check.py` validates both against the pinned contracts, and +`devfactory_observability.replay` reads the trail back to check it can account for how +the job got where it is — driven end to end by [`simulation/`](../../simulation/). ## Decisions diff --git a/packages/core/devfactory_core/job.py b/packages/core/devfactory_core/job.py index 0dd09d3..0ba09a8 100644 --- a/packages/core/devfactory_core/job.py +++ b/packages/core/devfactory_core/job.py @@ -46,8 +46,8 @@ DECISION_TARGET, POST_APPROVAL, TERMINAL, - TRANSITIONS, JobState, + reachable_from, ) #: States whose entry requires reason metadata — RFC-0001 for FAILED, extended @@ -199,12 +199,10 @@ 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. + job and so cannot live in the static table; ``states.reachable_from`` + folds it in, and is the same call anything else reading the table makes. """ - 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) + return reachable_from(self._state, awaiting_from=self._awaiting_from) # ---- the engine -------------------------------------------------------- diff --git a/packages/core/devfactory_core/states.py b/packages/core/devfactory_core/states.py index 23d7875..3af5642 100644 --- a/packages/core/devfactory_core/states.py +++ b/packages/core/devfactory_core/states.py @@ -179,5 +179,23 @@ def static_targets(state: JobState) -> frozenset[JobState]: return TRANSITIONS[state] +def reachable_from( + state: JobState, *, awaiting_from: JobState | None = None +) -> frozenset[JobState]: + """States reachable from ``state``, including the one dynamic edge. + + ``AWAITING_APPROVAL``'s way back differs per job and so cannot be a row in + ``TRANSITIONS``. Folding it in here rather than at each call site means + "reading the table correctly" has one implementation: the engine asks this + before it moves a job, and a replay reading the trail back asks the same + thing before it believes a recorded edge. Neither gets to hold an opinion + about the lifecycle that this module does not already state. + """ + targets = TRANSITIONS[state] + if state is JobState.AWAITING_APPROVAL and awaiting_from is not None: + return targets | {awaiting_from} + return targets + + def is_terminal(state: JobState) -> bool: return state in TERMINAL diff --git a/packages/core/state-machine.md b/packages/core/state-machine.md index 6124ed4..8c8d612 100644 --- a/packages/core/state-machine.md +++ b/packages/core/state-machine.md @@ -108,6 +108,14 @@ definitions: - `APPROVED` requires an explicit governance decision, recorded and emitted. - `FAILED`, `CANCELLED`, and `TIMED_OUT` all require reason metadata. - `CANCELLED` records the cancelling principal. +- The trail is complete enough to reconstruct the job from it — + `devfactory_observability.replay` rebuilds state, `awaiting_from`, the approval in + force, and the whole history from the events alone, and refuses a trail that has a + gap in it. This is what RFC-0010 anticipated when it noted that consumers can + validate transitions "against a declared table instead of inferring one": replay + checks every recorded edge against `states.reachable_from`, so nothing outside + `states.py` holds an opinion about the lifecycle. Driven end to end by + [`simulation/`](../../simulation/) — issue #7. ## Open questions diff --git a/packages/observability/README.md b/packages/observability/README.md index 9b0e612..1c61ea1 100644 --- a/packages/observability/README.md +++ b/packages/observability/README.md @@ -10,7 +10,7 @@ In memory for v0.1. No metrics backend, no dashboard — out of scope per the is and per [`CORE_BOUNDARY.md`](../../docs/governance/CORE_BOUNDARY.md). The `Event` type itself lives in `devfactory_core.events`, because the state machine -emits events. This package owns **storage** and **intake**. +emits events. This package owns **storage**, **intake**, and **replay**. ## Use @@ -79,13 +79,59 @@ external by the fact of arriving here, whatever it claims about itself. Guessing a tenant is treated as worse than losing the event. A lost event is a visible gap; one tenant's activity written into another tenant's immutable trail is not. +## Replay — RFC-0003 + +`replay_job(events)` rebuilds a job from its trail alone; `replay_tenant(log, tenant)` +does it for every job in one partition. + +```python +from devfactory_observability import replay_job, replay_tenant + +seen = replay_job(log.read("acme", job_id="job-001")) +seen.state # JobState.COMPLETED +seen.awaiting_from # where a paused job would return to +seen.approval_decision_id # the APPROVE it was executing under +seen.history # every transition, reconstructed +``` + +RFC-0003 lists *"enable replayable job history"* as a goal. This makes it checkable, +and it is a **completeness proof** rather than a convenience: every `STATE_TRANSITION` +names the state it left, so a replay holding a running state notices a record that is +missing or out of order — which the engine cannot, having written them. + +It also re-checks the guarantees against what was actually written, by something that +was not there when it happened. Every edge is validated against +`devfactory_core.states.reachable_from`, the same call the engine makes, so there is +no second transition table here to drift. + +| the trail | outcome | +| --- | --- | +| nothing to replay | `EmptyTrail` | +| does not start at `JOB_CREATED` | `UnstartedTrail` | +| a transition leaves a state the replay is not in | `BrokenTrail` | +| records an edge `states.py` does not declare | `UndeclaredTransition` | +| enters `APPROVED`/`REJECTED` with no matching decision | `UnauditedDecision` | +| begins execution with no `APPROVE` before it | `UnauditedExecution` | +| `COMPLETED` and `JOB_COMPLETED` disagree | `IncompleteSettlement` | +| an unrecognised event type | **skipped** — `event/v1`: keep it, do not interpret it | +| an external event | **skipped** — RFC-0008: another system is not an authority on our lifecycle | + +A trail truncated at the end is noticeable only for a job that completed, because +`JOB_COMPLETED` is the one record that says a transition should have followed. Nothing +says so for `FAILED`, `CANCELLED`, or `TIMED_OUT`, or for a job still in flight — +closing that needs a per-job sequence number in `event/v1`, which is a contract change +and not something replay can infer. + ## Tests ```bash cd packages/observability -python -m pytest # 59 tests -python -m pytest --cov=devfactory_observability # gate at 90%, currently 100% +python -m pytest # 89 tests +python -m pytest --cov=devfactory_observability # gate at 90% ``` +The end-to-end flows that exercise replay over whole journeys are +[`simulation/`](../../simulation/) at the repository root — issue #7. + Payload conformance against the pinned contracts is [`conformance/`](../../conformance/) at the repository root. diff --git a/packages/observability/devfactory_observability/__init__.py b/packages/observability/devfactory_observability/__init__.py index 795c28c..8eb5a5f 100644 --- a/packages/observability/devfactory_observability/__init__.py +++ b/packages/observability/devfactory_observability/__init__.py @@ -2,28 +2,49 @@ Issue #6. Spec: RFC-0003 as amended by RFC-0008, with the tenant model from RFC-0006. The ``Event`` type itself lives in ``devfactory_core.events``; this -package owns storage and intake. +package owns storage, intake, and reading a trail back (``replay``). """ from .errors import ( AuditLogError, + BrokenTrail, DuplicateEvent, + EmptyTrail, ExternalSourceRequired, FabricatedIdentifier, + IncompleteSettlement, MissingSubject, MissingTenant, + ReplayError, + UnauditedDecision, + UnauditedExecution, + UndeclaredTransition, + UnstartedTrail, ) from .intake import PLACEHOLDERS, accept_external +from .replay import ReplayedJob, ReplayedTransition, replay_job, replay_tenant from .store import EventLog __all__ = [ "PLACEHOLDERS", "AuditLogError", + "BrokenTrail", "DuplicateEvent", + "EmptyTrail", "EventLog", "ExternalSourceRequired", "FabricatedIdentifier", + "IncompleteSettlement", "MissingSubject", "MissingTenant", + "ReplayError", + "ReplayedJob", + "ReplayedTransition", + "UnauditedDecision", + "UnauditedExecution", + "UndeclaredTransition", + "UnstartedTrail", "accept_external", + "replay_job", + "replay_tenant", ] diff --git a/packages/observability/devfactory_observability/errors.py b/packages/observability/devfactory_observability/errors.py index b51b653..7a10475 100644 --- a/packages/observability/devfactory_observability/errors.py +++ b/packages/observability/devfactory_observability/errors.py @@ -71,3 +71,158 @@ def __init__(self) -> None: "an external event must name its source system so it stays " "identifiable as external forever" ) + + +# ---- replay ---------------------------------------------------------------- +# RFC-0003 asks the log to "enable replayable job history". These are what a +# replay refuses. Each one is a *finding about the log*, not about the caller: +# reaching any of them means the trail cannot account for how the job got where +# it is, which is the failure mode an audit trail exists to make impossible. + + +class ReplayError(AuditLogError): + """Base class for a trail that cannot be replayed.""" + + +class EmptyTrail(ReplayError): + """There is nothing to replay.""" + + def __init__(self, job_id: str | None = None) -> None: + self.job_id = job_id + subject = f" for job {job_id}" if job_id else "" + super().__init__(f"no events{subject} — a job with no trail cannot be reconstructed") + + +class UnstartedTrail(ReplayError): + """The trail does not begin at ``JOB_CREATED``. + + Every job starts in DRAFT and says so in its first event. A trail starting + anywhere else is missing its beginning, and replaying it would silently + assume the beginning it cannot see. + """ + + def __init__(self, job_id: str | None, first_event_type: str) -> None: + self.job_id = job_id + self.first_event_type = first_event_type + super().__init__( + f"trail for job {job_id} starts at {first_event_type}, not JOB_CREATED — " + f"its beginning is missing and replay will not assume one" + ) + + +class BrokenTrail(ReplayError): + """The trail contradicts itself: a record is missing, or they are out of order. + + A ``STATE_TRANSITION`` names the state it left. If that is not the state the + replay is standing in, then either a transition between the two was never + written or the records arrived in the wrong order. Both mean the same thing + for an audit trail — it can no longer account for the job — so replay stops + rather than papering over the gap. + """ + + def __init__(self, job_id: str, expected: str, recorded: str, event_id: str) -> None: + self.job_id = job_id + self.expected = expected + self.recorded = recorded + self.event_id = event_id + super().__init__( + f"job {job_id}: event {event_id} records a transition out of {recorded}, " + f"but the trail so far leaves the job in {expected} — a record is missing " + f"or the trail is out of order" + ) + + +class IncompleteSettlement(ReplayError): + """``COMPLETED`` and ``JOB_COMPLETED`` do not agree in the trail. + + Reaching ``COMPLETED`` emits ``JOB_COMPLETED``, so a trail carrying one and + not the other is missing a record. + + This is also the only way a trail truncated at the *end* is noticeable. + Every other record is checked by the one after it naming the state it left; + nothing follows the last one. ``JOB_COMPLETED`` closes that gap for a job + that finished — and only for that job. A trail cut short mid-flight, or at + ``FAILED``, ``CANCELLED``, or ``TIMED_OUT``, still replays cleanly into the + state it was cut at, because nothing in ``event/v1`` says how many records a + job should have. Making that detectable needs a per-job sequence number, + which is a contract change, not something replay can infer. + """ + + def __init__(self, job_id: str, state: str, *, announced: bool) -> None: + self.job_id = job_id + self.state = state + self.announced = announced + detail = ( + f"the trail carries JOB_COMPLETED but its transitions leave the job in " + f"{state} — the transition into COMPLETED was never written" + if announced + else f"the job reached {state} with no JOB_COMPLETED to announce it" + ) + super().__init__(f"job {job_id}: {detail}") + + +class UndeclaredTransition(ReplayError): + """The trail records an edge the lifecycle does not declare. + + Checked against ``devfactory_core.states``, which is the only place the + transition table exists. An edge that is not in it was not made by an engine + following the lifecycle, whatever the record says. + """ + + def __init__(self, job_id: str, from_state: str, to_state: str, event_id: str) -> None: + self.job_id = job_id + self.from_state = from_state + self.to_state = to_state + self.event_id = event_id + super().__init__( + f"job {job_id}: event {event_id} records {from_state} -> {to_state}, which " + f"the transition table does not declare — the lifecycle is defined in " + f"devfactory_core.states and nowhere else" + ) + + +class UnauditedDecision(ReplayError): + """A job entered a decision state with no decision behind it in the trail. + + Either nothing was recorded, or what was recorded does not produce this + transition. RFC-0002: an approval that leaves no record is not auditable, and + a record whose meaning is not the meaning that was decided is worse than + none. Both are guarantees about the log, so they have to hold when the log is + read back and not only when it is written. + """ + + def __init__( + self, job_id: str, state: str, event_id: str, recorded: str | None = None + ) -> None: + self.job_id = job_id + self.state = state + self.event_id = event_id + self.recorded = recorded + detail = ( + f"the decision before it is {recorded}, which does not send a job there" + if recorded is not None + else "no GOVERNANCE_DECISION appears before it" + ) + super().__init__( + f"job {job_id}: event {event_id} enters {state} and {detail} — the trail " + f"cannot say who decided, or what they decided" + ) + + +class UnauditedExecution(ReplayError): + """The trail shows execution beginning with no APPROVE behind it. + + The direction lock read back off the log: *execution is forbidden before + APPROVED*. The engine enforces it as it writes; this is the same claim + checked against what was actually written, by something that was not there + when it happened. + """ + + def __init__(self, job_id: str, state: str, event_id: str) -> None: + self.job_id = job_id + self.state = state + self.event_id = event_id + super().__init__( + f"job {job_id}: event {event_id} enters {state}, but no APPROVE decision " + f"appears in the trail before it — execution without a recorded approval" + ) diff --git a/packages/observability/devfactory_observability/replay.py b/packages/observability/devfactory_observability/replay.py new file mode 100644 index 0000000..b3add7a --- /dev/null +++ b/packages/observability/devfactory_observability/replay.py @@ -0,0 +1,273 @@ +"""Rebuild a job's state from its audit trail, and refuse a trail that cannot. + +[RFC-0003](../../../rfcs/0003-audit-event-log-schema.md) lists *"enable replayable +job history"* as a goal of the event log. This module is that goal made +checkable: hand it the events for one job and it returns where the job ended up, +having derived it from nothing but what was written down. + +Two things make that worth having, and neither is "a second state machine". + +**It is a completeness proof, not a convenience.** Every ``STATE_TRANSITION`` +names the state it left as well as the one it entered. Replay walks the trail +holding a running state and compares: if a transition claims to leave a state the +replay is not standing in, a record between the two was never written, or the +records are out of order. The engine cannot detect that — it wrote them — so the +check has to live here, on the reading side. A trail that replays cleanly is a +trail with no gaps in it. + +**It re-checks the guarantees against what was actually written.** The engine +enforces "execution is forbidden before ``APPROVED``" as it moves a job. That is +a claim about the log too, and this replays it: reaching a post-approval state +with no ``APPROVE`` in the trail before it raises +:class:`~devfactory_observability.errors.UnauditedExecution`, whatever the engine +believed at the time. + +What this module does **not** own is the lifecycle. Every edge it accepts is +checked against ``devfactory_core.states.reachable_from`` — the same call the +engine makes — so there is no second copy of the transition table here and no way +for one to drift in. An edge the table does not declare is refused rather than +followed, even though following it would be easier. + +Events this repository did not emit are skipped rather than interpreted: RFC-0008 +keeps an external event identifiable as external forever, and an outside system +is not an authority on our lifecycle. Unrecognised event types are skipped for +the reason ``event/v1`` gives — keep it, do not interpret it. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import TYPE_CHECKING, Any, Iterable + +from devfactory_core.decision import DecisionType +from devfactory_core.events import INTERNAL_SOURCE, Event, EventType +from devfactory_core.states import ( + DECISION_BY_TARGET, + POST_APPROVAL, + TERMINAL, + JobState, + reachable_from, +) + +from .errors import ( + BrokenTrail, + EmptyTrail, + IncompleteSettlement, + UnauditedDecision, + UnauditedExecution, + UndeclaredTransition, + UnstartedTrail, +) + +if TYPE_CHECKING: # pragma: no cover + from .store import EventLog + + +@dataclass(frozen=True, slots=True) +class ReplayedTransition: + """One transition as the trail recorded it. + + Deliberately shaped like ``devfactory_core.job.TransitionRecord`` without + being it: this one is reconstructed rather than remembered, and the two being + comparable is the whole point of the exercise. + """ + + from_state: JobState + to_state: JobState + at: datetime + reason: str | None + event_id: str + decision_id: str | None + + +@dataclass(frozen=True, slots=True) +class ReplayedJob: + """A job as its audit trail alone describes it.""" + + job_id: str + tenant_id: str + workspace_id: str | None + supersedes_job_id: str | None + state: JobState + awaiting_from: JobState | None + #: The APPROVE this job was last executing under, by id. Cleared by a REJECT, + #: exactly as the engine clears it — an approval granted to an earlier + #: revision must not appear to authorise the revised one when read back. + approval_decision_id: str | None + decision_ids: tuple[str, ...] + history: tuple[ReplayedTransition, ...] + #: Whether the trail carries the ``JOB_COMPLETED`` that COMPLETED implies. + #: Always agrees with ``state`` — a trail where it did not raises + #: :class:`IncompleteSettlement` rather than returning the disagreement. + completed: bool + + @property + def is_terminal(self) -> bool: + return self.state in TERMINAL + + @property + def states_visited(self) -> tuple[JobState, ...]: + """DRAFT and then every state entered, in order.""" + return (JobState.DRAFT,) + tuple(t.to_state for t in self.history) + + +def is_ours(event: Event) -> bool: + """Whether this repository emitted the event. + + ``source`` is unset on events the engine emits and filled in on the wire; an + event that arrived through intake always carries ``kind: external``. + """ + return (event.source or INTERNAL_SOURCE).get("kind") == "internal" + + +def _approval_of(event: Event) -> dict[str, Any] | None: + """The approval payload a ``GOVERNANCE_DECISION`` carries, if it carries one.""" + approval = (event.metadata or {}).get("approval") + if not isinstance(approval, dict): + return None + if not approval.get("approval_id") or not approval.get("decision"): + return None + return approval + + +def replay_job(events: Iterable[Event]) -> ReplayedJob: + """Reconstruct one job from its events, in the order they were logged. + + The order is taken as given rather than sorted. The log is append-only and + keeps write order, and re-sorting by ``occurred_at`` would quietly repair a + trail that arrived scrambled — which is a thing worth reporting, not fixing. + An out-of-order trail surfaces as :class:`BrokenTrail`. + """ + trail = [event for event in events if is_ours(event)] + if not trail: + raise EmptyTrail() + + creation = trail[0] + if creation.type_value != EventType.JOB_CREATED.value: + raise UnstartedTrail(creation.job_id, creation.type_value) + if creation.job_id is None: + raise EmptyTrail() + + job_id = creation.job_id + state = JobState.DRAFT + awaiting_from: JobState | None = None + approval_decision_id: str | None = None + pending: dict[str, Any] | None = None + decision_ids: list[str] = [] + history: list[ReplayedTransition] = [] + completed = False + + for event in trail[1:]: + if event.job_id != job_id: + raise BrokenTrail(job_id, job_id, str(event.job_id), event.event_id) + kind = event.type_value + + if kind == EventType.GOVERNANCE_DECISION.value: + approval = _approval_of(event) + if approval is not None: + pending = approval + decision_ids.append(approval["approval_id"]) + continue + + if kind == EventType.JOB_COMPLETED.value: + completed = True + continue + + if kind != EventType.STATE_TRANSITION.value: + # Everything else is something to keep, not something to interpret. + continue + + payload = event.transition or {} + recorded_from, recorded_to = payload.get("from"), payload.get("to") + if not recorded_from or not recorded_to: + raise BrokenTrail(job_id, state.value, str(recorded_from), event.event_id) + + from_state, to_state = JobState(recorded_from), JobState(recorded_to) + if from_state is not state: + raise BrokenTrail(job_id, state.value, from_state.value, event.event_id) + if to_state not in reachable_from(state, awaiting_from=awaiting_from): + raise UndeclaredTransition( + job_id, from_state.value, to_state.value, event.event_id + ) + + decision_id = _settle_decision(job_id, to_state, pending, event.event_id) + if decision_id is not None: + pending = None + if to_state is JobState.APPROVED: + approval_decision_id = decision_id + elif to_state is JobState.REJECTED: + approval_decision_id = None + + # The direction lock, checked against the record rather than the engine. + if to_state in POST_APPROVAL and approval_decision_id is None: + raise UnauditedExecution(job_id, to_state.value, event.event_id) + + if to_state is JobState.AWAITING_APPROVAL: + awaiting_from = state + elif state is JobState.AWAITING_APPROVAL: + awaiting_from = None + + history.append( + ReplayedTransition( + from_state=from_state, + to_state=to_state, + at=event.occurred_at, + reason=payload.get("reason"), + event_id=event.event_id, + decision_id=decision_id, + ) + ) + state = to_state + + if completed is not (state is JobState.COMPLETED): + raise IncompleteSettlement(job_id, state.value, announced=completed) + + return ReplayedJob( + job_id=job_id, + tenant_id=creation.tenant_id, + workspace_id=creation.workspace_id, + supersedes_job_id=(creation.metadata or {}).get("supersedes_job_id"), + state=state, + awaiting_from=awaiting_from, + approval_decision_id=approval_decision_id, + decision_ids=tuple(decision_ids), + history=tuple(history), + completed=completed, + ) + + +def _settle_decision( + job_id: str, to_state: JobState, pending: dict[str, Any] | None, event_id: str +) -> str | None: + """The decision id behind this transition, or None if it needed no decision. + + Which states are decisions is ``DECISION_BY_TARGET``, which lives in + ``devfactory_core.states`` beside the table it has to agree with. Asking it + rather than listing the decision states again is what stops this module from + acquiring an opinion of its own about what counts as a decision. + """ + expected = DECISION_BY_TARGET.get(to_state) + if expected is None: + return None + if pending is None: + raise UnauditedDecision(job_id, to_state.value, event_id) + recorded = pending["decision"] + if DecisionType(recorded) is not expected: + raise UnauditedDecision(job_id, to_state.value, event_id, recorded=str(recorded)) + return str(pending["approval_id"]) + + +def replay_tenant(log: "EventLog", tenant_id: str) -> dict[str, ReplayedJob]: + """Replay every job the log holds for one tenant. + + Events that no job caused are skipped rather than filed under a job — + RFC-0008 rule 2, read back: absent means absent, so there is nothing here to + attribute them to. + """ + trails: dict[str, list[Event]] = {} + for event in log.read(tenant_id): + if event.job_id is None: + continue + trails.setdefault(event.job_id, []).append(event) + return {job_id: replay_job(events) for job_id, events in trails.items()} diff --git a/packages/observability/tests/test_replay.py b/packages/observability/tests/test_replay.py new file mode 100644 index 0000000..f70d312 --- /dev/null +++ b/packages/observability/tests/test_replay.py @@ -0,0 +1,377 @@ +"""Reading a trail back — RFC-0003's *"enable replayable job history"*. + +The end-to-end flows that exercise replay over whole journeys live in +``simulation/tests/test_e2e_flow.py`` (issue #7). What is here is what this +package owns: the boundary between a trail and everything else in the log. +""" + +from __future__ import annotations + +import dataclasses + +import pytest +from devfactory_core import Event, EventType, JobState, Principal +from devfactory_core.events import new_event_id, utc_now + +from devfactory_observability import ( + BrokenTrail, + EmptyTrail, + EventLog, + IncompleteSettlement, + ReplayError, + UnauditedDecision, + UnauditedExecution, + UndeclaredTransition, + UnstartedTrail, + accept_external, + replay_job, + replay_tenant, +) +from devfactory_observability.replay import is_ours + + +def approved(job, reviewer): + job.submit_for_governance(reason="ready") + job.approve(authority=reviewer, reason="approved") + return job + + +@pytest.fixture +def reviewer() -> Principal: + return Principal("human", "bob") + + +# ---- what replay reconstructs ---------------------------------------------- + + +def test_a_fresh_job_replays_as_draft(make_job): + seen = replay_job(make_job().events) + assert seen.state is JobState.DRAFT + assert seen.history == () + assert seen.awaiting_from is None + assert seen.approval_decision_id is None + assert seen.is_terminal is False + + +def test_identity_comes_from_the_creation_event(make_job): + job = make_job(job_id="job-009", tenant_id="globex", workspace_id="ws-platform") + seen = replay_job(job.events) + assert (seen.job_id, seen.tenant_id, seen.workspace_id) == ( + "job-009", + "globex", + "ws-platform", + ) + assert seen.supersedes_job_id is None + + +def test_the_supersession_link_survives_the_round_trip(make_job, reviewer): + job = approved(make_job(), reviewer) + job.transition(JobState.TASK_PLANNING) + job.fail(reason="the plan does not work") + replacement = job.supersede(job_id="job-002") + assert replay_job(replacement.events).supersedes_job_id == "job-001" + + +def test_decisions_are_recovered_in_order(make_job, reviewer): + job = make_job() + job.submit_for_governance() + job.reject(authority=reviewer, reason="no") + job.transition(JobState.DRAFT) + job.submit_for_governance() + job.approve(authority=reviewer, reason="yes") + seen = replay_job(job.events) + assert list(seen.decision_ids) == [d.decision_id for d in job.decisions] + assert seen.approval_decision_id == job.approval.decision_id + + +def test_a_rejection_clears_the_approval_on_replay_too(make_job, reviewer): + """The engine drops a stale approval; so must anything reading the trail.""" + job = approved(make_job(), reviewer) + job.transition(JobState.TASK_PLANNING) + job.fail(reason="failed") + replacement = job.supersede(job_id="job-002") + replacement.submit_for_governance() + replacement.reject(authority=reviewer, reason="still wrong") + seen = replay_job(replacement.events) + assert seen.state is JobState.REJECTED + assert seen.approval_decision_id is None + + +def test_states_visited_starts_at_draft(make_job, reviewer): + job = approved(make_job(), reviewer) + assert replay_job(job.events).states_visited == ( + JobState.DRAFT, + JobState.GOVERNANCE_ANALYSIS, + JobState.APPROVED, + ) + + +def test_a_pause_and_its_return_address_are_recovered(make_job, reviewer): + job = approved(make_job(), reviewer) + job.transition(JobState.TASK_PLANNING) + job.transition(JobState.IN_PROGRESS) + job.pause_for_approval(reason="needs a human") + paused = replay_job(job.events) + assert paused.state is JobState.AWAITING_APPROVAL + assert paused.awaiting_from is JobState.IN_PROGRESS + + job.resume(reason="signed off", principal=reviewer) + resumed = replay_job(job.events) + assert resumed.state is JobState.IN_PROGRESS + assert resumed.awaiting_from is None + + +def test_a_terminal_job_replays_as_terminal(make_job): + job = make_job() + job.submit_for_governance() + job.cancel(reason="stopped", principal=Principal("human", "alice")) + seen = replay_job(job.events) + assert seen.state is JobState.CANCELLED + assert seen.is_terminal + assert seen.history[-1].reason == "stopped" + + +# ---- what replay refuses --------------------------------------------------- + + +def test_an_empty_trail_is_refused(): + with pytest.raises(EmptyTrail): + replay_job([]) + + +def test_a_trail_of_nothing_but_external_events_is_refused(external): + with pytest.raises(EmptyTrail): + replay_job([accept_external(external())]) + + +def test_a_trail_that_does_not_start_at_creation_is_refused(make_job): + job = make_job() + job.submit_for_governance() + with pytest.raises(UnstartedTrail) as excinfo: + replay_job(job.events[1:]) + assert excinfo.value.first_event_type == "STATE_TRANSITION" + + +def test_a_gap_in_the_trail_is_refused(make_job, reviewer): + job = approved(make_job(), reviewer) + without_submission = [job.events[0], *job.events[2:]] + with pytest.raises(BrokenTrail) as excinfo: + replay_job(without_submission) + assert (excinfo.value.expected, excinfo.value.recorded) == ( + "DRAFT", + "GOVERNANCE_ANALYSIS", + ) + + +def test_a_transition_event_with_no_transition_payload_is_refused(make_job): + job = make_job() + job.submit_for_governance() + stripped = [ + dataclasses.replace(e, transition=None) + if e.type_value == "STATE_TRANSITION" + else e + for e in job.events + ] + with pytest.raises(BrokenTrail): + replay_job(stripped) + + +def test_an_edge_the_table_does_not_declare_is_refused(make_job): + job = make_job() + job.submit_for_governance() + forged = [ + dataclasses.replace(e, transition={"from": "DRAFT", "to": "DEPLOYABLE"}) + if e.type_value == "STATE_TRANSITION" + else e + for e in job.events + ] + with pytest.raises(UndeclaredTransition) as excinfo: + replay_job(forged) + assert excinfo.value.to_state == "DEPLOYABLE" + + +def test_another_jobs_event_in_the_trail_is_refused(make_job): + mine, theirs = make_job(job_id="job-001"), make_job(job_id="job-002") + theirs.submit_for_governance() + with pytest.raises(BrokenTrail): + replay_job([*mine.events, theirs.events[-1]]) + + +def test_entering_approved_with_no_decision_is_refused(make_job, reviewer): + job = approved(make_job(), reviewer) + with pytest.raises(UnauditedDecision) as excinfo: + replay_job([e for e in job.events if e.type_value != "GOVERNANCE_DECISION"]) + assert excinfo.value.recorded is None + + +def test_a_decision_event_carrying_no_approval_counts_as_no_decision( + make_job, reviewer +): + job = approved(make_job(), reviewer) + hollow = [ + dataclasses.replace(e, metadata={}) + if e.type_value == "GOVERNANCE_DECISION" + else e + for e in job.events + ] + with pytest.raises(UnauditedDecision): + replay_job(hollow) + + +def test_a_decision_that_does_not_produce_the_transition_is_refused(make_job, reviewer): + job = approved(make_job(), reviewer) + forged = [ + dataclasses.replace( + e, + metadata={ + "approval": {**e.metadata["approval"], "decision": "REQUIRE_CHANGES"} + }, + ) + if e.type_value == "GOVERNANCE_DECISION" + else e + for e in job.events + ] + with pytest.raises(UnauditedDecision) as excinfo: + replay_job(forged) + assert excinfo.value.recorded == "REQUIRE_CHANGES" + + +def test_execution_with_no_approve_behind_it_is_refused(make_job, reviewer, monkeypatch): + """The direction lock as a backstop, the way ``job.py`` keeps its own. + + A table-consistent trail cannot reach this: ``TASK_PLANNING`` is only + reachable from ``APPROVED``, and ``APPROVED`` is refused without a decision. + Emptying the decision map is how a wrongly edited table would look from here, + and the point is that the lock still holds when it happens. + """ + from devfactory_observability import replay as replay_module + + job = approved(make_job(), reviewer) + job.transition(JobState.TASK_PLANNING) + monkeypatch.setattr(replay_module, "DECISION_BY_TARGET", {}) + with pytest.raises(UnauditedExecution) as excinfo: + replay_job(job.events) + assert excinfo.value.state == "TASK_PLANNING" + + +def test_a_completion_the_transitions_do_not_reach_is_refused(make_job): + """The one truncation replay can notice — see ``IncompleteSettlement``.""" + job = make_job() + announcement = Event( + event_id=new_event_id(), + event_type=EventType.JOB_COMPLETED, + tenant_id=job.tenant_id, + subject_type="job", + subject_id=job.job_id, + job_id=job.job_id, + occurred_at=utc_now(), + ) + with pytest.raises(IncompleteSettlement) as excinfo: + replay_job([*job.events, announcement]) + assert excinfo.value.announced is True + + +def test_a_completion_with_nothing_announcing_it_is_refused(make_job, reviewer): + job = approved(make_job(), reviewer) + for target in ( + JobState.TASK_PLANNING, + JobState.IN_PROGRESS, + JobState.VALIDATING, + JobState.DEPLOYABLE, + JobState.COMPLETED, + ): + job.transition(target) + silent = [e for e in job.events if e.type_value != "JOB_COMPLETED"] + with pytest.raises(IncompleteSettlement) as excinfo: + replay_job(silent) + assert excinfo.value.announced is False + + +def test_a_creation_event_with_no_job_is_not_a_job_trail(make_job): + """``job_id`` is never fabricated, so an absent one has nothing to replay.""" + job = make_job() + with pytest.raises(EmptyTrail): + replay_job([dataclasses.replace(job.events[0], job_id=None)]) + + +def test_every_refusal_is_a_replay_error(make_job): + with pytest.raises(ReplayError): + replay_job(make_job().events[1:]) + + +# ---- what replay leaves alone ---------------------------------------------- + + +def test_an_event_type_replay_does_not_know_is_kept_not_interpreted( + make_job, external +): + """``event/v1``: keep it, skip interpreting it. Skipping is not failing.""" + job = make_job() + job.submit_for_governance() + noise = dataclasses.replace( + job.events[-1], + event_id=new_event_id(), + event_type="TASK_ASSIGNED", + transition=None, + ) + assert replay_job([*job.events, noise]).state is JobState.GOVERNANCE_ANALYSIS + + +def test_an_external_event_is_not_an_authority_on_our_lifecycle(make_job, external): + """RFC-0008 keeps an external event identifiable as external forever. + + A forged ``STATE_TRANSITION`` from another system is kept in the log and has + no effect on what this repository says its own job did. + """ + job = make_job() + job.submit_for_governance() + intruder = accept_external( + external( + event_type="STATE_TRANSITION", + subject_type="job", + subject_id=job.job_id, + job_id=job.job_id, + ) + ) + assert is_ours(intruder) is False + assert replay_job([*job.events, intruder]).state is JobState.GOVERNANCE_ANALYSIS + + +def test_our_own_events_are_ours(make_job): + assert all(is_ours(e) for e in make_job().events) + + +# ---- reading a whole partition --------------------------------------------- + + +def test_replay_tenant_returns_one_entry_per_job(make_job, reviewer): + log = EventLog() + first = approved(make_job(job_id="job-001"), reviewer) + second = make_job(job_id="job-002") + log.extend(first.events) + log.extend(second.events) + + replayed = replay_tenant(log, "acme") + assert set(replayed) == {"job-001", "job-002"} + assert replayed["job-001"].state is JobState.APPROVED + assert replayed["job-002"].state is JobState.DRAFT + + +def test_replay_tenant_skips_events_no_job_caused(make_job, external): + log = EventLog() + job = make_job() + log.extend(job.events) + log.append(accept_external(external())) + assert set(replay_tenant(log, "acme")) == {job.job_id} + + +def test_replay_tenant_reads_only_the_tenant_it_names(make_job, reviewer): + log = EventLog() + log.extend(make_job(job_id="job-001", tenant_id="acme").events) + log.extend(make_job(job_id="job-002", tenant_id="globex").events) + assert set(replay_tenant(log, "acme")) == {"job-001"} + assert set(replay_tenant(log, "globex")) == {"job-002"} + + +def test_an_unknown_tenant_replays_as_nothing(make_job): + assert replay_tenant(EventLog(), "nobody") == {} diff --git a/simulation/README.md b/simulation/README.md new file mode 100644 index 0000000..ad70bff --- /dev/null +++ b/simulation/README.md @@ -0,0 +1,77 @@ +# End-to-end flow simulation + +Issue [#7](https://github.com/monthop-gmail/devfactory-core/issues/7) — the closing +gate on [milestone v0.1](../docs/governance/MILESTONE_v0.1.md). + +A governed job is driven end to end through the real state machine +([`packages/core`](../packages/core/)), everything it emits is filed in the real +audit log ([`packages/observability`](../packages/observability/)), and the trail +is then read back to check it accounts for what happened. + +## Run it + +```bash +python3 simulation/e2e_flow.py # the simulation, with findings +python3 simulation/e2e_flow.py --trail # …and the audit trail it produced +python3 simulation/e2e_flow.py --json # machine-readable result +python3 -m pytest simulation/tests # the same guarantees, as a suite +``` + +Both are run in CI. The script is what the issue asks for and what a person runs +to watch a job move; the suite is what stops the guarantees regressing when +nobody is watching. Exit code is non-zero if any check fails. + +## What it checks + +| # | issue #7 asks | where | +| --- | --- | --- | +| 1 | the full flow, `DRAFT → … → COMPLETED` | `check_full_flow` | +| 2 | `REJECTED → DRAFT`, then resubmitted | `check_rejection_flow` | +| 3 | `FAILED` with a reason, from a state `FAILABLE` allows | `check_failure_flow` | +| 4 | the governance gate blocks execution without an `APPROVE` | `check_governance_gate` | +| 5 | every transition emits an audit event | `check_every_transition_is_audited` | +| 6 | the log is complete and replays to the same state | `check_replay` | +| 7 | a runnable script or a test suite | this directory — both | + +## The forward path is read, not retyped + +`packages/core/devfactory_core/states.py` is the only place the transition table +is expressed. A simulation that wrote `APPROVED → TASK_PLANNING` into itself in +order to walk it would be a second declaration with the first one's authority, so +`flows.main_line()` *derives* the path instead: at each state, discard the exits +available from nearly everywhere — `CANCELLED`, `TIMED_OUT`, `FAILED`, `REJECTED`, +and the `AWAITING_APPROVAL` pause — and one successor is left. + +The flow issue #7 spells out appears exactly once, as the thing that derivation is +compared *against*. If the table and the issue ever disagree, the comparison fails +and says so, instead of the simulation quietly following the issue and reporting +success. + +## Replay is a completeness proof + +`devfactory_observability.replay` rebuilds a job from its events alone. Every +`STATE_TRANSITION` names the state it left, so a replay holding a running state +notices a record that is missing or out of order — which the engine cannot, +having written them. A trail that replays cleanly is a trail with no gaps in it, +and that is what makes "the audit log is complete" a checked claim rather than a +stated one. + +Two limits, recorded rather than smoothed over: + +- **A truncated tail is only detectable for a job that completed.** `JOB_COMPLETED` + is what says a transition should have followed; nothing says so for `FAILED`, + `CANCELLED`, or `TIMED_OUT`, or for a job still in flight. Closing that needs a + per-job sequence number in `event/v1`, which is a contract change. +- **`UnauditedExecution` cannot fire on a table-consistent trail**, because + `TASK_PLANNING` is reachable only from `APPROVED` and `APPROVED` is refused + without a decision. It is a structural backstop, kept for the same reason + `job.py` keeps `ExecutionBeforeApproval`: it fires when the table itself is + wrong, which is exactly when it is worth having. + +## Shared with conformance + +`conformance/payload_check.py` drives the same flows from `flows.py`. It asks a +different question — do the payloads conform to the pinned `event/v1` and +`approval/v1` — but it should not be asking it about a *different* journey. Two +files describing the same lifecycle differently is how they end up disagreeing +about it. diff --git a/simulation/__init__.py b/simulation/__init__.py new file mode 100644 index 0000000..68ef1ba --- /dev/null +++ b/simulation/__init__.py @@ -0,0 +1,37 @@ +"""End-to-end flow simulation — issue #7, the closing gate on milestone v0.1. + +``flows`` holds one definition of each flow the control plane is asked to +demonstrate. ``e2e_flow`` is the runnable simulation over them. + +Nothing here is a second state machine: the flows drive the real ``Job`` engine, +the audit trail is the real ``EventLog``, and the forward path they walk is read +out of ``devfactory_core.states`` rather than retyped. +""" + +from .flows import ( + MAIN_LINE, + JobFactory, + advance_to, + cancelled_by_a_person, + failed_at, + happy_path, + job_factory, + main_line, + never_approved, + rejected_then_resubmitted, + stalled_awaiting_approval, +) + +__all__ = [ + "MAIN_LINE", + "JobFactory", + "advance_to", + "cancelled_by_a_person", + "failed_at", + "happy_path", + "job_factory", + "main_line", + "never_approved", + "rejected_then_resubmitted", + "stalled_awaiting_approval", +] diff --git a/simulation/e2e_flow.py b/simulation/e2e_flow.py new file mode 100644 index 0000000..225fc08 --- /dev/null +++ b/simulation/e2e_flow.py @@ -0,0 +1,578 @@ +#!/usr/bin/env python3 +"""End-to-end flow simulation — issue #7, the closing gate on milestone v0.1. + +Drives real jobs through the real state machine, collects what they emit into the +real audit log, and then checks the six things issue #7 asks for. Nothing is +hand-written to make a check pass: every assertion below is made against records +the engine actually wrote. + +The six checks map one-to-one onto the issue's list:: + + [1] the full flow, DRAFT → … → COMPLETED + [2] rejection, revision, resubmission + [3] failure, from every state RFC-0010 allows and from nowhere else + [4] the governance gate — no execution without an APPROVE *record* + [5] every transition leaves an audit event + [6] the trail is complete and replays to the state the job is really in + +The seventh item is this file: a runnable script, per the issue. The same checks +also run under pytest — see ``simulation/tests/test_e2e_flow.py`` — so CI fails on +them without anyone having to remember to run this by hand. + +Usage:: + + python3 simulation/e2e_flow.py # run the simulation + python3 simulation/e2e_flow.py --json # machine-readable result + python3 simulation/e2e_flow.py --trail # also print the audit trail it produced + +Style follows ``conformance/payload_check.py``, which got here first: findings +printed as they are found, a non-zero exit if any of them failed. +""" + +from __future__ import annotations + +import argparse +import dataclasses +import json +import pathlib +import sys +from datetime import datetime, timedelta, timezone + +ROOT = pathlib.Path(__file__).resolve().parents[1] +sys.path[:0] = [ + str(ROOT), + str(ROOT / "packages" / "core"), + str(ROOT / "packages" / "observability"), +] + +from devfactory_core import Job, JobState, Principal # noqa: E402 +from devfactory_core.errors import ( # noqa: E402 + ExecutionBeforeApproval, + InvalidTransition, + JobStateMachineError, +) +from devfactory_core.states import FAILABLE, POST_APPROVAL, TERMINAL # noqa: E402 +from devfactory_observability import ( # noqa: E402 + BrokenTrail, + EventLog, + UnauditedDecision, + UndeclaredTransition, + replay_job, + replay_tenant, +) +from simulation.flows import ( # noqa: E402 + MAIN_LINE, + cancelled_by_a_person, + failed_at, + happy_path, + job_factory, + never_approved, + rejected_then_resubmitted, + stalled_awaiting_approval, +) + +TENANT = "acme" +WORKSPACE = "ws-core" + +#: The flow issue #7 spells out, kept here as the thing to compare *against* the +#: table. It is written down once, in the assertion, and never used to drive +#: anything — the simulation walks ``MAIN_LINE``, which is read out of +#: ``states.py``. If the two ever disagree, check [1] says so instead of the +#: simulation quietly following the issue and reporting success. +ISSUE_7_FLOW = ( + "DRAFT", + "GOVERNANCE_ANALYSIS", + "APPROVED", + "TASK_PLANNING", + "IN_PROGRESS", + "VALIDATING", + "DEPLOYABLE", + "COMPLETED", +) + +findings: list[tuple[str, str, str]] = [] +passed = 0 + + +def ok(area: str, message: str) -> None: + global passed + passed += 1 + findings.append(("ok", area, message)) + print(f" ok {area}: {message}") + + +def fail(area: str, message: str) -> None: + findings.append(("FAIL", area, message)) + print(f" FAIL {area}: {message}") + + +def check(area: str, condition: bool, when_ok: str, when_bad: str) -> bool: + (ok if condition else fail)(area, when_ok if condition else when_bad) + return condition + + +def monotonic_clock(): + """A fake clock, so the trail's ordering does not depend on wall time.""" + start = datetime(2026, 8, 19, 9, 0, tzinfo=timezone.utc) + state = {"n": 0} + + def tick() -> datetime: + state["n"] += 1 + return start + timedelta(seconds=state["n"]) + + return tick + + +def visited(job: Job) -> tuple[str, ...]: + """The states the job was actually in, read from its own history.""" + return (JobState.DRAFT.value,) + tuple(h.to_state.value for h in job.history) + + +# ---- [1] the full flow ------------------------------------------------------ + + +def check_full_flow(new, reviewer) -> Job: + if [s.value for s in MAIN_LINE] != list(ISSUE_7_FLOW): + fail( + "flow", + f"เส้นทางที่ตารางประกาศ {[s.value for s in MAIN_LINE]} ไม่ตรงกับที่ #7 ระบุ " + f"{list(ISSUE_7_FLOW)} — ถ้าตารางถูก ต้องแก้ issue ไม่ใช่แก้ simulation", + ) + else: + ok("flow", f"เส้นทางเดินหน้าอ่านจาก states.py ได้ตรงกับ #7 ทั้ง {len(MAIN_LINE)} state") + + job = happy_path(new, job_id="job-001", authority=reviewer) + check( + "flow", + visited(job) == ISSUE_7_FLOW, + f"job-001 เดินครบเส้นทาง {' → '.join(ISSUE_7_FLOW)}", + f"job-001 เดินได้ {visited(job)}", + ) + check( + "flow", + job.state is JobState.COMPLETED and job.is_terminal, + "จบที่ COMPLETED และเป็น terminal", + f"จบที่ {job.state.value}", + ) + check( + "flow", + any(e.type_value == "JOB_COMPLETED" for e in job.events), + "COMPLETED มี JOB_COMPLETED กำกับ", + "ถึง COMPLETED แต่ไม่มี JOB_COMPLETED", + ) + return job + + +# ---- [2] rejection and resubmission ----------------------------------------- + + +def check_rejection_flow(new, reviewer) -> Job: + job = rejected_then_resubmitted(new, job_id="job-002", authority=reviewer) + expected = ( + "DRAFT", + "GOVERNANCE_ANALYSIS", + "REJECTED", + "DRAFT", + "GOVERNANCE_ANALYSIS", + "APPROVED", + "TASK_PLANNING", + ) + check( + "reject", + visited(job) == expected, + f"job-002 เดิน {' → '.join(expected)}", + f"job-002 เดินได้ {visited(job)}", + ) + + rejection, approval = job.decisions[0], job.decisions[-1] + check( + "reject", + rejection.decision.value == "REJECT" and approval.decision.value == "APPROVE", + "ยื่นใหม่แล้วได้ decision ใบที่สอง ไม่ใช่การแก้ใบเดิม", + f"decision ที่บันทึกคือ {[d.decision.value for d in job.decisions]}", + ) + check( + "reject", + approval.supersedes_decision_id == rejection.decision_id, + "APPROVE ใบใหม่อ้างถึง REJECT ใบเดิม (approval/v1)", + "APPROVE ใบใหม่ไม่ได้อ้างใบเดิม", + ) + check( + "reject", + job.approval is not None and job.approval.decision_id == approval.decision_id, + "งานกำลังทำงานภายใต้ APPROVE ใบล่าสุด", + "งานเดินต่อได้โดยไม่ได้ถือ APPROVE ใบล่าสุด", + ) + + # A rejection must not leave an approval behind — proved on a job that had one. + approved = new("job-002b") + approved.submit_for_governance() + approved.approve(authority=reviewer, reason="first pass") + approved.transition(JobState.TASK_PLANNING) + approved.fail(reason="the approved plan does not work") + replacement = approved.supersede(job_id="job-002c") + replacement.submit_for_governance(reason="revised plan") + replacement.reject(authority=reviewer, reason="still wrong") + replacement.transition(JobState.DRAFT) + try: + replacement.transition(JobState.TASK_PLANNING) + fail("reject", "งานที่ถูก REJECT ยังเดินเข้า TASK_PLANNING ได้") + except InvalidTransition: + ok("reject", "REJECT ล้าง approval ทิ้ง — งานที่ยื่นใหม่ยังทำงานไม่ได้จนกว่าจะอนุมัติอีกครั้ง") + return job + + +# ---- [3] failure ------------------------------------------------------------ + + +def check_failure_flow(new, reviewer) -> list[Job]: + jobs: list[Job] = [] + for index, state in enumerate(sorted(FAILABLE, key=lambda s: s.value)): + reason = f"orchestration exhausted execution retries in {state.value}" + job = failed_at( + new, + job_id=f"job-003-{index}", + authority=reviewer, + state=state, + reason=reason, + ) + jobs.append(job) + last = job.history[-1] + if not ( + job.state is JobState.FAILED + and last.from_state is state + and last.reason == reason + ): + fail("fail", f"ล้มเหลวจาก {state.value} ไม่ได้บันทึกตามที่เกิดจริง") + continue + stored = [ + e + for e in job.events + if e.type_value == "STATE_TRANSITION" and e.transition["to"] == "FAILED" + ] + check( + "fail", + len(stored) == 1 and stored[0].transition.get("reason") == reason, + f"FAILED จาก {state.value} พร้อมเหตุผลใน audit trail", + f"FAILED จาก {state.value} แต่เหตุผลไม่ได้ลงใน event", + ) + + refused = sorted(set(JobState) - FAILABLE - TERMINAL, key=lambda s: s.value) + for state in refused: + job = new(f"job-003-x-{state.value.lower().replace('_', '-')}") + _park_in(job, state, reviewer) + try: + job.fail(reason="pretending there is work to fail") + fail("fail", f"{state.value} เข้า FAILED ได้ ทั้งที่ RFC-0010 ไม่อนุญาต") + except JobStateMachineError: + pass + ok( + "fail", + f"FAILED เข้าได้เฉพาะ {len(FAILABLE)} state ที่ RFC-0010 อนุญาต " + f"และถูกปฏิเสธจากอีก {len(refused)} state ({', '.join(s.value for s in refused)})", + ) + + # RFC-0007: recovery is a new job that passes governance again, not a revival. + origin = jobs[0] + replacement = origin.supersede(job_id="job-003-next") + check( + "fail", + replacement.state is JobState.DRAFT + and replacement.supersedes_job_id == origin.job_id, + "งานที่ล้มเหลวถูกแทนที่ด้วย job ใหม่ที่ชี้กลับไปหาใบเดิม ไม่ใช่การปลุกใบเดิม", + "supersede ไม่ได้สร้าง job ใหม่ที่ชี้กลับใบเดิม", + ) + try: + replacement.transition(JobState.TASK_PLANNING) + fail("fail", "job ที่มาแทนเดินเข้า TASK_PLANNING ได้โดยไม่ผ่าน governance") + except InvalidTransition: + ok("fail", "job ที่มาแทนต้องผ่าน GOVERNANCE_ANALYSIS ใหม่") + jobs.append(replacement) + return jobs + + +def _park_in(job: Job, state: JobState, authority: Principal) -> None: + """Put a fresh job into ``state``, for the states that cannot fail. + + Only ever called with DRAFT, GOVERNANCE_ANALYSIS, APPROVED, or REJECTED — + everything RFC-0010 refuses ``FAILED`` from. + """ + if state is JobState.DRAFT: + return + job.submit_for_governance() + if state is JobState.GOVERNANCE_ANALYSIS: + return + if state is JobState.REJECTED: + job.reject(authority=authority, reason="out of scope") + return + job.approve(authority=authority, reason="approved") + + +# ---- [4] the governance gate ------------------------------------------------ + + +def check_governance_gate(new, reviewer) -> list[Job]: + job = never_approved(new, job_id="job-004") + check( + "gate", + job.approval is None, + "งานที่ยังไม่ได้ตัดสิน ไม่ถือ Decision ใด ๆ", + "งานที่ยังไม่ได้ตัดสินกลับถือ approval อยู่", + ) + blocked = [] + for target in sorted(POST_APPROVAL, key=lambda s: s.value): + try: + job.transition(target) + fail("gate", f"เข้า {target.value} ได้ทั้งที่ยังไม่มี APPROVE") + except JobStateMachineError: + blocked.append(target.value) + check( + "gate", + len(blocked) == len(POST_APPROVAL), + f"ไม่มี APPROVE แล้วเข้า execution ไม่ได้ทั้ง {len(blocked)} state " + f"({', '.join(blocked)})", + "มี state ฝั่ง execution ที่เข้าได้โดยไม่มี APPROVE", + ) + + # The gate is a record, not a flag: forcing the *state* is not enough. + forced = new("job-004b") + forced.submit_for_governance() + forced._state = JobState.APPROVED # simulating a wrongly edited table + try: + forced.transition(JobState.TASK_PLANNING) + fail("gate", "อยู่ใน APPROVED โดยไม่มี Decision record แล้วยังเดินต่อได้") + except ExecutionBeforeApproval: + ok("gate", "gate ผูกกับ Decision record — สถานะ APPROVED เปล่า ๆ เดินต่อไม่ได้") + + # And what authorises execution is a record with an accountable authority. + approved = new("job-004c") + approved.submit_for_governance() + record = approved.approve(authority=reviewer, reason="scope matches milestone v0.1") + approved.transition(JobState.TASK_PLANNING) + check( + "gate", + approved.approval is record + and record.authority.id == reviewer.id + and bool(record.reason) + and record.decided_at is not None, + "สิ่งที่อนุญาตให้ execute คือ record ที่ตอบได้ว่าใครตัดสิน ด้วยเหตุผลอะไร เมื่อไร", + "approval ที่ถืออยู่ไม่ได้ตอบว่าใครตัดสินด้วยเหตุผลอะไร", + ) + + # Read back off the log: strip the decision and the trail no longer holds up. + tampered = [e for e in approved.events if e.type_value != "GOVERNANCE_DECISION"] + try: + replay_job(tampered) + fail("gate", "trail ที่ไม่มี GOVERNANCE_DECISION ยัง replay ผ่าน") + except UnauditedDecision: + ok("gate", "replay ปฏิเสธ trail ที่เข้า APPROVED โดยไม่มี decision บันทึกไว้") + + # ``forced`` is deliberately corrupted and is not filed: an audit log is not + # the place to keep a job whose state was set behind the engine's back. + return [job, approved] + + +# ---- [5] every transition emits an audit event ------------------------------ + + +def check_every_transition_is_audited(jobs, log) -> None: + silent = [ + job.job_id + for job in jobs + if len([e for e in job.events if e.type_value == "STATE_TRANSITION"]) + != len(job.history) + ] + check( + "audit", + not silent, + f"ทุก transition ของ {len(jobs)} job มี STATE_TRANSITION ตรงจำนวน " + f"(รวม {sum(len(j.history) for j in jobs)} transition)", + f"มี transition ที่ไม่ได้ emit event: {silent}", + ) + + check( + "audit", + len(log) == sum(len(job.events) for job in jobs) and log.tenants() == (TENANT,), + f"ล็อกเก็บครบทุก event ที่ {len(jobs)} job ผลิต ในพาร์ทิชันของ {TENANT} พาร์ทิชันเดียว", + f"ล็อกมี {len(log)} event แต่ job ผลิตรวม {sum(len(j.events) for j in jobs)}", + ) + + by_id = {e.event_id: e for e in log.read(TENANT)} + unlinked = [ + (job.job_id, record.from_state.value, record.to_state.value) + for job in jobs + for record in job.history + if record.event_id not in by_id + or by_id[record.event_id].transition + != { + k: v + for k, v in ( + ("from", record.from_state.value), + ("to", record.to_state.value), + ("reason", record.reason), + ) + if v is not None + } + ] + check( + "audit", + not unlinked, + "ทุก transition ชี้ไปที่ event ที่มีอยู่จริงในล็อก และเนื้อหาตรงกัน", + f"transition ที่ event หายไปหรือเนื้อหาไม่ตรง: {unlinked}", + ) + + undecided = [ + job.job_id + for job in jobs + if len([h for h in job.history if h.to_state in (JobState.APPROVED, JobState.REJECTED)]) + != len([e for e in job.events if e.type_value == "GOVERNANCE_DECISION"]) + ] + check( + "audit", + not undecided, + "ทุกครั้งที่เข้า APPROVED หรือ REJECTED มี GOVERNANCE_DECISION คู่กับ STATE_TRANSITION", + f"job ที่เข้า state ตัดสินใจโดยไม่มี GOVERNANCE_DECISION: {undecided}", + ) + + +# ---- [6] the trail is complete and replays ---------------------------------- + + +def check_replay(jobs, log) -> None: + replayed = replay_tenant(log, TENANT) + live = {job.job_id: job for job in jobs} + + missing = sorted(set(live) - set(replayed)) + if missing: + fail("replay", f"job ที่ไม่มีร่องรอยในล็อก: {missing}") + return + + wrong = [] + for job_id, job in sorted(live.items()): + seen = replayed[job_id] + expected_approval = job.approval.decision_id if job.approval else None + if ( + seen.state is not job.state + or seen.awaiting_from is not job.awaiting_from + or seen.approval_decision_id != expected_approval + or seen.tenant_id != job.tenant_id + or seen.workspace_id != job.workspace_id + or seen.supersedes_job_id != job.supersedes_job_id + or [(t.from_state, t.to_state, t.reason) for t in seen.history] + != [(h.from_state, h.to_state, h.reason) for h in job.history] + or [t.decision_id for t in seen.history if t.decision_id] + != [h.decision_id for h in job.history if h.decision_id] + or seen.completed is not (job.state is JobState.COMPLETED) + ): + wrong.append(f"{job_id} (live={job.state.value} replay={seen.state.value})") + check( + "replay", + not wrong, + f"replay จาก audit log อย่างเดียว ได้สถานะเดิมครบทั้ง {len(live)} job " + f"— state, awaiting_from, approval, และประวัติทุกก้าว", + f"replay แล้วไม่ตรงกับของจริง: {wrong}", + ) + + # A trail is complete only if losing a record is detectable. Drop one and see. + donor = live["job-001"] + transitions = [e for e in donor.events if e.type_value == "STATE_TRANSITION"] + dropped = [e for e in donor.events if e is not transitions[len(transitions) // 2]] + try: + replay_job(dropped) + fail("replay", "ลบ STATE_TRANSITION ออกหนึ่งใบแล้ว replay ยังผ่าน — ล็อกไม่ครบก็ไม่รู้") + except BrokenTrail: + ok("replay", "ลบ event ออกหนึ่งใบแล้ว replay จับได้ — ล็อกครบจริงจึง replay ผ่าน") + + # And only if a forged edge is refused rather than followed. + target = next( + e + for e in donor.events + if e.type_value == "STATE_TRANSITION" and e.transition["from"] == "IN_PROGRESS" + ) + forged = [ + dataclasses.replace(e, transition={**e.transition, "to": "COMPLETED"}) + if e is target + else e + for e in donor.events + ] + try: + replay_job(forged) + fail("replay", "trail ที่บันทึก IN_PROGRESS → COMPLETED ถูก replay ตามไปด้วย") + except UndeclaredTransition: + ok("replay", "replay ตรวจทุก edge กับ states.py — edge ที่ตารางไม่ประกาศถูกปฏิเสธ") + + +# ---- run -------------------------------------------------------------------- + + +def simulate(log: EventLog) -> list[Job]: + """Run every flow, file everything in the log, and return the jobs.""" + clock = monotonic_clock() + owner = Principal("human", "alice", display_name="Alice") + reviewer = Principal("human", "bob", display_name="Bob") + new = job_factory( + tenant_id=TENANT, workspace_id=WORKSPACE, principal=owner, clock=clock + ) + + print(f"\n[1] flow เต็ม — {' → '.join(ISSUE_7_FLOW)}") + happy = check_full_flow(new, reviewer) + + print("\n[2] flow ปฏิเสธ — REJECTED กลับ DRAFT แล้วยื่นใหม่") + rejected = check_rejection_flow(new, reviewer) + + print("\n[3] flow ล้มเหลว — FAILED จาก state ที่ RFC-0010 อนุญาตเท่านั้น") + failures = check_failure_flow(new, reviewer) + + print("\n[4] governance gate — ไม่มี APPROVE record ก็ไม่มี execution") + gated = check_governance_gate(new, reviewer) + + # Two flows this repository already exercised in conformance, kept in the run + # so the trail the replay checks work on covers every terminal state. + cancelled = cancelled_by_a_person(new, job_id="job-005", owner=owner) + stalled = stalled_awaiting_approval(new, job_id="job-006", authority=reviewer) + + jobs = [happy, rejected, *failures, *gated, cancelled, stalled] + for job in jobs: + log.extend(job.events) + return jobs + + +def main() -> int: + parser = argparse.ArgumentParser(description="end-to-end flow simulation (issue #7)") + parser.add_argument("--json", action="store_true", help="พิมพ์ผลเป็น JSON") + parser.add_argument("--trail", action="store_true", help="พิมพ์ audit trail ที่ผลิตได้") + args = parser.parse_args() + + print("=" * 70) + print("END-TO-END FLOW SIMULATION — issue #7 (v0.1.0)") + print("=" * 70) + + log = EventLog() + jobs = simulate(log) + + print(f"\n[5] ทุก transition ต้องมี audit event — {len(log)} event จาก {len(jobs)} job") + check_every_transition_is_audited(jobs, log) + + print("\n[6] audit log ครบและ replay ได้") + check_replay(jobs, log) + + if args.trail: + print("\naudit trail") + for payload in log.payloads(TENANT): + move = payload.get("transition") + arrow = f" {move['from']} → {move['to']}" if move else "" + print(f" {payload['occurred_at']} {payload['event_type']:<20}" + f" {payload.get('job_id', '-'):<16}{arrow}") + + fails = [f for f in findings if f[0] == "FAIL"] + print("\n" + "=" * 70) + print(f" passed={passed} FAIL={len(fails)}") + print("=" * 70) + + if args.json: + print(json.dumps({"passed": passed, "findings": findings}, ensure_ascii=False, indent=2)) + return 1 if fails else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/simulation/flows.py b/simulation/flows.py new file mode 100644 index 0000000..e66b6b5 --- /dev/null +++ b/simulation/flows.py @@ -0,0 +1,229 @@ +"""The flows a job can take, defined once and driven through the real engine. + +Issue #7 names four of them — the full path, rejection and resubmission, failure +from a state that may fail, and the governance gate refusing execution. Two more +live here because ``conformance/payload_check.py`` was already driving them and +there is no reason for two files to describe the same journey differently. + +Why the forward path is not written out here +-------------------------------------------- +``packages/core/devfactory_core/states.py`` is the only place the transition +table is expressed, and a simulation that retyped ``APPROVED -> TASK_PLANNING`` +in order to walk it would be a second declaration with the first one's authority. +So :func:`main_line` *reads* the path out of the table instead: at every state, +discard the exits that are available everywhere and are not "forward", and what +remains is a single successor. If the lifecycle ever gains a fork, this raises +rather than picking one — which is the right failure, because a fork is a +lifecycle change and belongs in an RFC before it belongs in a simulation. + +The only per-state knowledge kept here is *which method to call*, because +entering ``GOVERNANCE_ANALYSIS`` and entering ``APPROVED`` need arguments the +guards require. That is a fact about the call signature, not about the table. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Callable, Protocol + +from devfactory_core import Job, JobState, Principal +from devfactory_core.states import FAILABLE, TERMINAL, reachable_from + + +class JobFactory(Protocol): + """How a flow gets a fresh job — supplied by the caller, so the flows stay + free of any opinion about tenancy, workspace, or clock.""" + + def __call__(self, job_id: str) -> Job: ... + + +#: Exits available from nearly everywhere, none of which is progress. Removing +#: them from a state's targets is what leaves the forward edge exposed. +#: +#: * the three unhappy terminals — every non-terminal state offers ``CANCELLED``, +#: and most offer ``TIMED_OUT`` or ``FAILED`` +#: * ``REJECTED`` — a verdict, and the flow that follows it is its own +#: * ``AWAITING_APPROVAL`` — a pause inside the path, not a step along it +_NOT_FORWARD: frozenset[JobState] = (TERMINAL - {JobState.COMPLETED}) | { + JobState.REJECTED, + JobState.AWAITING_APPROVAL, +} + + +def main_line() -> tuple[JobState, ...]: + """The forward path from ``DRAFT`` to ``COMPLETED``, read out of the table. + + Raises if the table stops naming exactly one way forward, rather than + choosing between them. + """ + path: list[JobState] = [JobState.DRAFT] + while path[-1] is not JobState.COMPLETED: + forward = sorted(reachable_from(path[-1]) - _NOT_FORWARD, key=lambda s: s.value) + if len(forward) != 1: + raise ValueError( + f"{path[-1].value} has {len(forward)} ways forward " + f"({[s.value for s in forward] or 'none'}) — the forward path is no " + f"longer a line, and which branch a simulation walks is an RFC " + f"question, not this module's to answer" + ) + if forward[0] in path: + raise ValueError(f"the forward path loops back to {forward[0].value}") + path.append(forward[0]) + return tuple(path) + + +#: ``DRAFT → GOVERNANCE_ANALYSIS → APPROVED → TASK_PLANNING → IN_PROGRESS → +#: VALIDATING → DEPLOYABLE → COMPLETED``, as the table declares it. +MAIN_LINE: tuple[JobState, ...] = main_line() + + +def advance_to(job: Job, target: JobState, *, authority: Principal) -> Job: + """Walk ``job`` forward along :data:`MAIN_LINE` until it is in ``target``.""" + if target not in MAIN_LINE: + raise ValueError(f"{target.value} is not on the forward path") + if job.state not in MAIN_LINE: + raise ValueError(f"a job in {job.state.value} is not on the forward path") + for state in MAIN_LINE[MAIN_LINE.index(job.state) + 1 : MAIN_LINE.index(target) + 1]: + _enter(job, state, authority=authority) + return job + + +def _enter(job: Job, state: JobState, *, authority: Principal) -> None: + """Take the one step into ``state``. + + Two states are entered through a named method because the guards require + arguments that the generic call would have to be told about anyway: + ``GOVERNANCE_ANALYSIS`` is a submission, and ``APPROVED`` is a decision and + must name the authority accountable for it. + """ + if state is JobState.GOVERNANCE_ANALYSIS: + job.submit_for_governance(reason="ready for governance review") + elif state is JobState.APPROVED: + job.approve(authority=authority, reason="scope matches milestone v0.1") + else: + job.transition(state) + + +# ---- the flows ------------------------------------------------------------- + + +def happy_path( + new: JobFactory, + *, + job_id: str, + authority: Principal, + pause_in: JobState | None = None, +) -> Job: + """The whole lifecycle, ``DRAFT`` through ``COMPLETED``. + + ``pause_in`` optionally takes the mid-run approval detour on the way out of + that state — the same journey, with a human interrupting it once. + """ + job = new(job_id) + for state in MAIN_LINE[1:]: + if pause_in is not None and job.state is pause_in: + job.pause_for_approval(reason="needs a human before the next step") + job.resume(reason="signed off", principal=authority) + _enter(job, state, authority=authority) + return job + + +def rejected_then_resubmitted( + new: JobFactory, *, job_id: str, authority: Principal +) -> Job: + """Rejected, revised, resubmitted, approved — and then actually executing. + + It carries on into ``TASK_PLANNING`` on purpose. Reaching ``APPROVED`` a + second time proves the job may be resubmitted; taking the next step proves + the *second* decision restored the authority the first one withheld, which is + the part that would be silently broken if a rejection left an approval behind. + """ + job = new(job_id) + job.submit_for_governance(reason="first submission") + job.reject(authority=authority, reason="missing risk analysis") + job.transition(JobState.DRAFT) + job.submit_for_governance(reason="risk analysis added") + job.approve(authority=authority, reason="the gap raised in review is addressed") + job.transition(JobState.TASK_PLANNING) + return job + + +def failed_at( + new: JobFactory, + *, + job_id: str, + authority: Principal, + state: JobState, + reason: str = "orchestration exhausted execution retries", +) -> Job: + """Drive to ``state`` and fail there. + + ``state`` must be in ``states.FAILABLE`` — RFC-0010. Asking for anything else + is refused here rather than left to the engine, so that a flow which cannot + happen is not written down as though it could. + """ + if state not in FAILABLE: + raise ValueError( + f"{state.value} is not in states.FAILABLE — RFC-0010 says a job fails " + f"only where work exists to fail" + ) + job = new(job_id) + if state is JobState.AWAITING_APPROVAL: + advance_to(job, JobState.IN_PROGRESS, authority=authority) + job.pause_for_approval(reason="needs a human before merge") + else: + advance_to(job, state, authority=authority) + job.fail(reason=reason) + return job + + +def never_approved(new: JobFactory, *, job_id: str) -> Job: + """A job sitting in ``GOVERNANCE_ANALYSIS`` with no decision made about it. + + The subject of the governance-gate checks: it holds no ``Decision``, so + nothing about it may execute. + """ + job = new(job_id) + job.submit_for_governance(reason="awaiting a verdict") + return job + + +def cancelled_by_a_person(new: JobFactory, *, job_id: str, owner: Principal) -> Job: + job = new(job_id) + job.submit_for_governance() + job.cancel(reason="superseded by an urgent request", principal=owner) + return job + + +def stalled_awaiting_approval( + new: JobFactory, *, job_id: str, authority: Principal +) -> Job: + """An approval nobody answered — RFC-0007's characteristic stall.""" + job = new(job_id) + advance_to(job, JobState.IN_PROGRESS, authority=authority) + job.pause_for_approval(reason="needs a human before merge") + job.time_out(reason="sla_exceeded — approval request expired after 72h") + return job + + +def job_factory( + *, + tenant_id: str = "acme", + workspace_id: str = "ws-core", + principal: Principal, + clock: Callable[[], datetime] | None = None, +) -> JobFactory: + """A :class:`JobFactory` bound to one tenant, workspace, and clock.""" + + def new(job_id: str) -> Job: + kwargs = { + "job_id": job_id, + "tenant_id": tenant_id, + "workspace_id": workspace_id, + "principal": principal, + } + if clock is not None: + kwargs["clock"] = clock + return Job(**kwargs) + + return new diff --git a/simulation/tests/conftest.py b/simulation/tests/conftest.py new file mode 100644 index 0000000..df84b30 --- /dev/null +++ b/simulation/tests/conftest.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +sys.path[:0] = [ + str(ROOT), + str(ROOT / "packages" / "core"), + str(ROOT / "packages" / "observability"), +] + +from devfactory_core import Principal # noqa: E402 +from devfactory_observability import EventLog # noqa: E402 +from simulation.flows import job_factory # noqa: E402 + +TENANT = "acme" +WORKSPACE = "ws-core" + + +@pytest.fixture +def owner() -> Principal: + return Principal("human", "alice", display_name="Alice") + + +@pytest.fixture +def reviewer() -> Principal: + return Principal("human", "bob", display_name="Bob") + + +@pytest.fixture +def clock(): + """Monotonic fake clock, so trail ordering does not depend on wall time.""" + start = datetime(2026, 8, 19, 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 new(owner, clock): + return job_factory( + tenant_id=TENANT, workspace_id=WORKSPACE, principal=owner, clock=clock + ) + + +@pytest.fixture +def log() -> EventLog: + return EventLog() diff --git a/simulation/tests/test_e2e_flow.py b/simulation/tests/test_e2e_flow.py new file mode 100644 index 0000000..8ea98b1 --- /dev/null +++ b/simulation/tests/test_e2e_flow.py @@ -0,0 +1,554 @@ +"""Issue #7 — the end-to-end flow simulation, as a test suite. + +The issue asks for "a runnable script or a test suite". This repository gets +both, because they answer different questions: ``simulation/e2e_flow.py`` is what +a person runs to watch a governed job move end to end, and this is what stops the +same guarantees from regressing without anyone watching. + +Each section below is one item from the issue's list. Nothing here re-implements +the lifecycle — every flow drives the real engine, and every expectation about +which states exist or connect is read out of ``devfactory_core.states``. +""" + +from __future__ import annotations + +import dataclasses +import subprocess +import sys +from pathlib import Path + +import pytest +from conftest import TENANT, WORKSPACE + +from devfactory_core import Job, JobState +from devfactory_core.errors import ( + ExecutionBeforeApproval, + InvalidTransition, + JobStateMachineError, +) +from devfactory_core.states import FAILABLE, POST_APPROVAL, TERMINAL +from devfactory_observability import ( + BrokenTrail, + EmptyTrail, + IncompleteSettlement, + ReplayError, + UnauditedDecision, + UnauditedExecution, + UndeclaredTransition, + UnstartedTrail, + replay_job, + replay_tenant, +) +from simulation.flows import ( + MAIN_LINE, + advance_to, + cancelled_by_a_person, + failed_at, + happy_path, + main_line, + never_approved, + rejected_then_resubmitted, + stalled_awaiting_approval, +) + +ROOT = Path(__file__).resolve().parents[2] + +#: The flow issue #7 writes out. Compared against the table rather than used to +#: drive anything — see ``test_the_forward_path_is_the_one_the_issue_asks_for``. +ISSUE_7_FLOW = ( + "DRAFT", + "GOVERNANCE_ANALYSIS", + "APPROVED", + "TASK_PLANNING", + "IN_PROGRESS", + "VALIDATING", + "DEPLOYABLE", + "COMPLETED", +) + + +def visited(job: Job) -> tuple[str, ...]: + return (JobState.DRAFT.value,) + tuple(h.to_state.value for h in job.history) + + +# ---- [1] the full flow ------------------------------------------------------ + + +def test_the_forward_path_is_the_one_the_issue_asks_for(): + """The table and the issue agree, and the table is what gets walked. + + ``MAIN_LINE`` is derived from ``states.TRANSITIONS``. Asserting it equals the + flow #7 spells out is the only place the issue's wording is treated as data — + everywhere else the simulation follows the table, so this test is what would + fail if the two drifted, rather than the simulation quietly following the + issue and reporting success. + """ + assert tuple(s.value for s in MAIN_LINE) == ISSUE_7_FLOW + + +def test_the_forward_path_is_recomputed_not_cached(): + assert main_line() == MAIN_LINE + + +def test_full_flow_reaches_completed(new, reviewer): + job = happy_path(new, job_id="job-001", authority=reviewer) + assert visited(job) == ISSUE_7_FLOW + assert job.state is JobState.COMPLETED + assert job.is_terminal + + +def test_completion_is_announced_in_the_trail(new, reviewer): + job = happy_path(new, job_id="job-001", authority=reviewer) + assert [e.type_value for e in job.events].count("JOB_COMPLETED") == 1 + assert job.events[-1].type_value == "JOB_COMPLETED" + + +def test_the_full_flow_survives_a_mid_run_approval_pause(new, reviewer): + """RFC-0007's pause is a detour inside the path, not a different path.""" + job = happy_path( + new, job_id="job-001", authority=reviewer, pause_in=JobState.DEPLOYABLE + ) + assert job.state is JobState.COMPLETED + assert job.awaiting_from is None + assert "AWAITING_APPROVAL" in visited(job) + # Drop the pause and the re-entry it returns to, and the same path is left. + forward = [s for s in visited(job) if s != "AWAITING_APPROVAL"] + resumed = [s for i, s in enumerate(forward) if i == 0 or s != forward[i - 1]] + assert tuple(resumed) == ISSUE_7_FLOW + + +# ---- [2] rejection and resubmission ----------------------------------------- + + +def test_rejected_returns_to_draft_and_can_be_resubmitted(new, reviewer): + job = rejected_then_resubmitted(new, job_id="job-002", authority=reviewer) + assert visited(job) == ( + "DRAFT", + "GOVERNANCE_ANALYSIS", + "REJECTED", + "DRAFT", + "GOVERNANCE_ANALYSIS", + "APPROVED", + "TASK_PLANNING", + ) + + +def test_resubmission_produces_a_second_decision_not_an_edited_first(new, reviewer): + job = rejected_then_resubmitted(new, job_id="job-002", authority=reviewer) + assert [d.decision.value for d in job.decisions] == ["REJECT", "APPROVE"] + assert job.decisions[1].supersedes_decision_id == job.decisions[0].decision_id + + +def test_the_second_approval_is_what_authorises_the_work(new, reviewer): + job = rejected_then_resubmitted(new, job_id="job-002", authority=reviewer) + assert job.approval is not None + assert job.approval.decision_id == job.decisions[-1].decision_id + assert job.state is JobState.TASK_PLANNING + + +def test_a_rejection_does_not_leave_an_earlier_approval_behind(new, reviewer): + """The revised job must be re-approved; the old APPROVE does not carry over.""" + approved = new("job-002b") + approved.submit_for_governance() + approved.approve(authority=reviewer, reason="first pass") + approved.transition(JobState.TASK_PLANNING) + approved.fail(reason="the approved plan does not work") + + replacement = approved.supersede(job_id="job-002c") + replacement.submit_for_governance(reason="revised plan") + replacement.reject(authority=reviewer, reason="still wrong") + replacement.transition(JobState.DRAFT) + assert replacement.approval is None + with pytest.raises(InvalidTransition): + replacement.transition(JobState.TASK_PLANNING) + + +# ---- [3] failure ------------------------------------------------------------ + + +@pytest.mark.parametrize( + "state", sorted(FAILABLE, key=lambda s: s.value), ids=lambda s: s.value +) +def test_failure_from_every_state_rfc_0010_allows(state, new, reviewer): + reason = f"orchestration exhausted execution retries in {state.value}" + job = failed_at( + new, job_id="job-003", authority=reviewer, state=state, reason=reason + ) + assert job.state is JobState.FAILED + assert job.history[-1].from_state is state + assert job.history[-1].reason == reason + + +@pytest.mark.parametrize( + "state", sorted(FAILABLE, key=lambda s: s.value), ids=lambda s: s.value +) +def test_the_failure_reason_reaches_the_audit_trail(state, new, reviewer): + reason = f"orchestration exhausted execution retries in {state.value}" + job = failed_at( + new, job_id="job-003", authority=reviewer, state=state, reason=reason + ) + entering = [ + e + for e in job.events + if e.type_value == "STATE_TRANSITION" and e.transition["to"] == "FAILED" + ] + assert len(entering) == 1 + assert entering[0].transition["reason"] == reason + + +@pytest.mark.parametrize( + "state", + sorted(set(JobState) - FAILABLE - TERMINAL, key=lambda s: s.value), + ids=lambda s: s.value, +) +def test_failure_is_refused_everywhere_else(state, new, reviewer): + """RFC-0010: a job fails only where work exists to fail.""" + job = new("job-003-x") + _park_in(job, state, reviewer) + assert job.state is state + with pytest.raises(JobStateMachineError): + job.fail(reason="pretending there is work to fail") + assert job.state is state + + +def test_a_flow_that_cannot_happen_is_refused_before_it_is_written_down(new, reviewer): + with pytest.raises(ValueError, match="FAILABLE"): + failed_at( + new, job_id="job-003", authority=reviewer, state=JobState.GOVERNANCE_ANALYSIS + ) + + +def test_recovery_is_a_new_job_that_passes_governance_again(new, reviewer): + origin = failed_at( + new, job_id="job-003", authority=reviewer, state=JobState.IN_PROGRESS + ) + replacement = origin.supersede(job_id="job-003-next") + assert replacement.state is JobState.DRAFT + assert replacement.supersedes_job_id == "job-003" + with pytest.raises(InvalidTransition): + replacement.transition(JobState.TASK_PLANNING) + + +def _park_in(job: Job, state: JobState, authority) -> None: + """Put a fresh job into one of the states RFC-0010 refuses ``FAILED`` from.""" + if state is JobState.DRAFT: + return + job.submit_for_governance() + if state is JobState.GOVERNANCE_ANALYSIS: + return + if state is JobState.REJECTED: + job.reject(authority=authority, reason="out of scope") + return + job.approve(authority=authority, reason="approved") + + +# ---- [4] the governance gate ------------------------------------------------ + + +@pytest.mark.parametrize( + "target", sorted(POST_APPROVAL, key=lambda s: s.value), ids=lambda s: s.value +) +def test_no_execution_state_is_reachable_without_a_decision(target, new): + job = never_approved(new, job_id="job-004") + assert job.approval is None + with pytest.raises(JobStateMachineError): + job.transition(target) + assert job.state is JobState.GOVERNANCE_ANALYSIS + + +def test_the_gate_is_the_record_not_the_state(new): + """Forcing APPROVED without a ``Decision`` still does not authorise anything. + + The state alone is not the authorisation — the record is. Poking ``_state`` + simulates a wrongly edited transition table, which is the only way this guard + can fire and the reason it exists. + """ + job = never_approved(new, job_id="job-004b") + job._state = JobState.APPROVED + with pytest.raises(ExecutionBeforeApproval): + job.transition(JobState.TASK_PLANNING) + + +def test_what_authorises_execution_answers_who_decided_and_why(new, reviewer): + job = never_approved(new, job_id="job-004c") + record = job.approve(authority=reviewer, reason="scope matches milestone v0.1") + job.transition(JobState.TASK_PLANNING) + assert job.approval is record + assert record.authority.id == reviewer.id + assert record.reason + assert record.decided_at is not None + assert record.subject.id == job.job_id + + +def test_the_gate_holds_when_the_trail_is_read_back(new, reviewer): + """Strip the decision out of the trail and the replay refuses it.""" + job = never_approved(new, job_id="job-004c") + job.approve(authority=reviewer, reason="approved") + job.transition(JobState.TASK_PLANNING) + + tampered = [e for e in job.events if e.type_value != "GOVERNANCE_DECISION"] + with pytest.raises(UnauditedDecision): + replay_job(tampered) + + +def test_a_decision_that_does_not_produce_the_transition_is_refused_on_replay( + new, reviewer +): + job = never_approved(new, job_id="job-004d") + job.approve(authority=reviewer, reason="approved") + forged = [ + dataclasses.replace( + e, + metadata={**e.metadata, "approval": {**e.metadata["approval"], "decision": "REJECT"}}, + ) + if e.type_value == "GOVERNANCE_DECISION" + else e + for e in job.events + ] + with pytest.raises(UnauditedDecision) as excinfo: + replay_job(forged) + assert excinfo.value.recorded == "REJECT" + + +def test_the_direction_lock_is_a_backstop_on_replay_too(new, reviewer, monkeypatch): + """If replay forgot APPROVED were a decision, execution would still be refused. + + Mirrors ``test_the_gate_is_the_record_not_the_state``: the check can only fire + when something upstream is already wrong, and that is exactly when it earns + its place. + """ + from devfactory_observability import replay as replay_module + + job = never_approved(new, job_id="job-004e") + job.approve(authority=reviewer, reason="approved") + job.transition(JobState.TASK_PLANNING) + + monkeypatch.setattr(replay_module, "DECISION_BY_TARGET", {}) + with pytest.raises(UnauditedExecution) as excinfo: + replay_job(job.events) + assert excinfo.value.state == "TASK_PLANNING" + + +# ---- [5] every transition emits an audit event ------------------------------ + + +@pytest.fixture +def run(new, owner, reviewer, log): + """Every flow, driven once, with everything they emitted filed in the log.""" + jobs = [ + happy_path(new, job_id="job-001", authority=reviewer), + happy_path( + new, job_id="job-001b", authority=reviewer, pause_in=JobState.DEPLOYABLE + ), + rejected_then_resubmitted(new, job_id="job-002", authority=reviewer), + failed_at(new, job_id="job-003", authority=reviewer, state=JobState.IN_PROGRESS), + never_approved(new, job_id="job-004"), + cancelled_by_a_person(new, job_id="job-005", owner=owner), + stalled_awaiting_approval(new, job_id="job-006", authority=reviewer), + ] + jobs.append(jobs[3].supersede(job_id="job-003-next")) + for job in jobs: + log.extend(job.events) + return jobs, log + + +def test_every_transition_emits_exactly_one_state_transition(run): + jobs, _ = run + for job in jobs: + emitted = [e for e in job.events if e.type_value == "STATE_TRANSITION"] + assert len(emitted) == len(job.history), job.job_id + + +def test_every_transition_points_at_an_event_that_is_really_in_the_log(run): + jobs, log = run + by_id = {e.event_id: e for e in log.read(TENANT)} + for job in jobs: + for record in job.history: + event = by_id[record.event_id] + assert event.transition["from"] == record.from_state.value + assert event.transition["to"] == record.to_state.value + assert event.transition.get("reason") == record.reason + + +def test_every_decision_state_entry_is_accompanied_by_a_decision_event(run): + jobs, _ = run + for job in jobs: + entries = [ + h + for h in job.history + if h.to_state in (JobState.APPROVED, JobState.REJECTED) + ] + emitted = [e for e in job.events if e.type_value == "GOVERNANCE_DECISION"] + assert len(entries) == len(emitted) == len(job.decisions), job.job_id + + +def test_the_whole_run_reaches_one_tenant_partition(run): + jobs, log = run + assert log.tenants() == (TENANT,) + assert len(log) == sum(len(job.events) for job in jobs) + assert all(p["workspace_id"] == WORKSPACE for p in log.payloads(TENANT)) + + +# ---- [6] the trail is complete and replays ---------------------------------- + + +def test_replay_returns_every_job_the_run_produced(run): + jobs, log = run + assert set(replay_tenant(log, TENANT)) == {job.job_id for job in jobs} + + +def test_replay_recovers_the_state_each_job_is_really_in(run): + jobs, log = run + replayed = replay_tenant(log, TENANT) + for job in jobs: + assert replayed[job.job_id].state is job.state, job.job_id + + +def test_replay_recovers_the_whole_history_not_just_the_last_state(run): + jobs, log = run + replayed = replay_tenant(log, TENANT) + for job in jobs: + seen = replayed[job.job_id] + assert [(t.from_state, t.to_state, t.reason) for t in seen.history] == [ + (h.from_state, h.to_state, h.reason) for h in job.history + ], job.job_id + assert [t.event_id for t in seen.history] == [ + h.event_id for h in job.history + ], job.job_id + + +def test_replay_recovers_the_governance_context(run): + jobs, log = run + replayed = replay_tenant(log, TENANT) + for job in jobs: + seen = replayed[job.job_id] + expected = job.approval.decision_id if job.approval else None + assert seen.approval_decision_id == expected, job.job_id + assert list(seen.decision_ids) == [d.decision_id for d in job.decisions] + + +def test_replay_recovers_the_pause_a_job_is_sitting_in(new, reviewer): + job = advance_to(new("job-006"), JobState.IN_PROGRESS, authority=reviewer) + job.pause_for_approval(reason="needs a human before merge") + seen = replay_job(job.events) + assert seen.state is JobState.AWAITING_APPROVAL + assert seen.awaiting_from is JobState.IN_PROGRESS + + +def test_replay_recovers_the_identity_and_the_supersession_link(run): + jobs, log = run + replayed = replay_tenant(log, TENANT) + for job in jobs: + seen = replayed[job.job_id] + assert seen.tenant_id == job.tenant_id + assert seen.workspace_id == job.workspace_id + assert seen.supersedes_job_id == job.supersedes_job_id + + +def test_replay_agrees_about_completion(run): + jobs, log = run + replayed = replay_tenant(log, TENANT) + for job in jobs: + assert replayed[job.job_id].completed is (job.state is JobState.COMPLETED) + + +def test_states_visited_is_the_path_the_job_took(new, reviewer): + job = happy_path(new, job_id="job-001", authority=reviewer) + assert tuple(s.value for s in replay_job(job.events).states_visited) == ISSUE_7_FLOW + + +# ---- [6b] the trail is complete because losing a record is detectable ------- + + +@pytest.mark.parametrize("index", range(1, 8), ids=lambda i: f"drop-{i}") +def test_dropping_any_transition_breaks_the_replay(index, new, reviewer): + """Completeness is only a claim if a gap would be noticed. Here it is. + + Every transition but the last is caught by the one after it naming the state + it left. The last one is caught by ``JOB_COMPLETED`` having nothing to + announce — which is why this covers a completed job, and why the same is not + true of a trail cut short at ``FAILED``; see :class:`IncompleteSettlement`. + """ + job = happy_path(new, job_id="job-001", authority=reviewer) + transitions = [e for e in job.events if e.type_value == "STATE_TRANSITION"] + assert len(transitions) == 7 + dropped = [e for e in job.events if e is not transitions[index - 1]] + with pytest.raises(ReplayError): + replay_job(dropped) + + +def test_a_missing_completion_announcement_is_noticed(new, reviewer): + job = happy_path(new, job_id="job-001", authority=reviewer) + without = [e for e in job.events if e.type_value != "JOB_COMPLETED"] + with pytest.raises(IncompleteSettlement) as excinfo: + replay_job(without) + assert excinfo.value.announced is False + + +def test_a_trail_cut_short_at_a_terminal_other_than_completed_still_replays( + new, reviewer +): + """Recorded because it is a real limit, not because it is desirable. + + Truncation is detectable only where something in the trail says more should + follow. Nothing does for ``FAILED``, so a trail that lost its last record + replays cleanly into the state before it. Closing that needs a per-job + sequence number in ``event/v1`` — a contract change, tracked in the report + for issue #7 rather than papered over here. + """ + job = failed_at( + new, job_id="job-003", authority=reviewer, state=JobState.IN_PROGRESS + ) + truncated = [e for e in job.events if e.transition is None or e.transition["to"] != "FAILED"] + assert replay_job(truncated).state is JobState.IN_PROGRESS + + +def test_reordering_the_trail_breaks_the_replay(new, reviewer): + job = happy_path(new, job_id="job-001", authority=reviewer) + events = list(job.events) + events[-2], events[-3] = events[-3], events[-2] + with pytest.raises(BrokenTrail): + replay_job(events) + + +def test_a_forged_edge_is_refused_rather_than_followed(new, reviewer): + """``states.py`` is the only transition table, on the reading side as well.""" + job = happy_path(new, job_id="job-001", authority=reviewer) + forged = [ + dataclasses.replace(e, transition={**e.transition, "to": "COMPLETED"}) + if e.type_value == "STATE_TRANSITION" and e.transition["from"] == "IN_PROGRESS" + else e + for e in job.events + ] + with pytest.raises(UndeclaredTransition) as excinfo: + replay_job(forged) + assert (excinfo.value.from_state, excinfo.value.to_state) == ( + "IN_PROGRESS", + "COMPLETED", + ) + + +def test_a_trail_with_no_beginning_is_refused(new, reviewer): + job = happy_path(new, job_id="job-001", authority=reviewer) + with pytest.raises(UnstartedTrail): + replay_job(job.events[1:]) + + +def test_an_empty_trail_is_refused(): + with pytest.raises(EmptyTrail): + replay_job([]) + + +# ---- [7] the deliverable is runnable ---------------------------------------- + + +def test_the_simulation_script_runs_and_reports_no_failures(): + """Issue #7 asks for a runnable script. This is that claim, checked.""" + proc = subprocess.run( + [sys.executable, str(ROOT / "simulation" / "e2e_flow.py"), "--json"], + cwd=ROOT, + capture_output=True, + text=True, + ) + assert proc.returncode == 0, proc.stdout + proc.stderr + assert "FAIL=0" in proc.stdout