From 068f53df0b07fd02427589a639dbf263fcf3f13d Mon Sep 17 00:00:00 2001 From: monthop-gmail Date: Wed, 19 Aug 2026 20:31:46 +0700 Subject: [PATCH] End-to-end flow simulation, and a replay that proves the trail is complete (#7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Milestone v0.1's closing gate: a governed job driven end to end through the real state machine, everything it emits filed in the real audit log, and the trail then read back to check it accounts for what happened. Delivered as both forms issue #7 allows. `simulation/e2e_flow.py` is what a person runs to watch a job move; `simulation/tests/test_e2e_flow.py` is what fails a pull request when a guarantee regresses. Both run in CI. The forward path is read out of `states.py`, not retyped -------------------------------------------------------- `flows.main_line()` derives DRAFT → … → COMPLETED from the transition table by discarding the exits available from nearly everywhere, and raises rather than choosing if the table ever stops naming one way forward. The flow #7 spells out appears exactly once, as the thing that derivation is compared *against* — so a disagreement between the issue and the table fails the check instead of the simulation quietly following the issue and reporting success. `states.reachable_from()` folds AWAITING_APPROVAL's per-job return edge into the table read, so the engine and anything reading the trail back ask the same question of the same source. Replay is a completeness proof, not a convenience ------------------------------------------------- `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. It re-checks the guarantees against what was actually written: every edge against `states.reachable_from`, every APPROVED against a GOVERNANCE_DECISION that really produced it, and the direction lock against the record rather than the engine's memory of it. `conformance/payload_check.py` now drives the same flows from `flows.py`. It asks a different question — do the payloads conform — but it should not be asking it about a different journey. Two limits recorded rather than smoothed over --------------------------------------------- A trail truncated at the end is detectable only for a job that completed: JOB_COMPLETED is the one record that says a transition should have followed. Nothing says so for FAILED, CANCELLED, TIMED_OUT, or a job still in flight, and closing that needs a per-job sequence number in event/v1 — a contract change. `UnauditedExecution` cannot fire on a table-consistent trail, since TASK_PLANNING is reachable only from APPROVED and APPROVED is refused without a decision. It is a structural backstop, kept for the reason job.py keeps ExecutionBeforeApproval. Nothing here weakens an existing guarantee, and no transition is declared outside `states.py`. 350 tests → 441; payload conformance unchanged at 13 passed, 0 fail. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/test.yml | 40 +- conformance/payload_check.py | 76 ++- packages/core/README.md | 4 +- packages/core/devfactory_core/job.py | 10 +- packages/core/devfactory_core/states.py | 18 + packages/core/state-machine.md | 8 + packages/observability/README.md | 52 +- .../devfactory_observability/__init__.py | 23 +- .../devfactory_observability/errors.py | 155 +++++ .../devfactory_observability/replay.py | 273 +++++++++ packages/observability/tests/test_replay.py | 377 ++++++++++++ simulation/README.md | 77 +++ simulation/__init__.py | 37 ++ simulation/e2e_flow.py | 578 ++++++++++++++++++ simulation/flows.py | 229 +++++++ simulation/tests/conftest.py | 56 ++ simulation/tests/test_e2e_flow.py | 554 +++++++++++++++++ 17 files changed, 2511 insertions(+), 56 deletions(-) create mode 100644 packages/observability/devfactory_observability/replay.py create mode 100644 packages/observability/tests/test_replay.py create mode 100644 simulation/README.md create mode 100644 simulation/__init__.py create mode 100644 simulation/e2e_flow.py create mode 100644 simulation/flows.py create mode 100644 simulation/tests/conftest.py create mode 100644 simulation/tests/test_e2e_flow.py 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