monthop-gmail · monthop-gmail · Aug 19, 2026 · Aug 19, 2026
@@ -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.

@@ -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:

@@ -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
 

@@ -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 --------------------------------------------------------
 

@@ -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
@@ -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
 

@@ -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.
@@ -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",
 ]