[ULTRA EXPERIMENTAL — DO NOT REVIEW] Durable workflows exploration - #6912
Draft
Alek99 wants to merge 145 commits into
Draft
[ULTRA EXPERIMENTAL — DO NOT REVIEW] Durable workflows exploration#6912Alek99 wants to merge 145 commits into
Alek99 wants to merge 145 commits into
Conversation
Implements the first slice of the Reflex Workflows design: durable automation on the existing Reflex programming model. Authoring contract: - rx.WorkflowConfig on a workflow-focused rx.State class (reserved __workflow__ attribute, excluded from state schemas) - @rx.event(durable=True, effect=...) with id/trigger/retry/timeout/ queue/on_failure/on_timeout options, validated at decoration time - rx.Retry with per-effect-class defaults materialized at compile (TransientWorkflowError is the explicit retryable signal) - declarative triggers rx.manual() / rx.webhook() / rx.schedule() - control returns rx.complete/fail/needs_attention and rx.after delays Runtime: - compile_workflow() validates the class contract (durable-only public handlers, no substates/backend vars/mixed scopes, resolvable hooks, stable unique ids) and produces an immutable digest-pinned definition - app.add_workflow() registers the class and detaches it from the session state tree: no per-session instances, no browser setters, no frontend event dispatch to durable handlers - WorkflowKernel executes runs against a RunStore: single-writer ordered mailbox with preallocated ordinals, fenced claims, atomic commit of state snapshot + successor slots + history, retry with exponential backoff and jitter as persisted timers, per-attempt execution timeouts, failure/timeout hooks after tombstoning, NEEDS_ATTENTION suspension for uncertain non-idempotent effects, drain-based cancellation, run deadlines, max_steps bounds, and orphan recovery with a separate infrastructure recovery budget - MemoryRunStore for tests; SqliteRunStore (stdlib, WAL) for crash-safe local persistence, including admission dedupe by request_key surviving restarts - rx.workflows.start/cancel/get_run namespace served by the app lifespan; WorkflowTestHarness runs definitions deterministically on virtual time Deliberately out of scope for this slice (per the design's MVP/Beta ledger): webhook/schedule ingress execution, the connector broker and effect evidence records, mixed-scope classes, UI run projections, operator commands beyond cancel, and multi-worker kernels.
A second worker starting while the first was mid-attempt reclaimed the live claim and executed the same step concurrently: recovery treated every CLAIMED row as an orphan left by a dead process. Proven with two OS processes against one SQLite file, where the peer ran the handler a second time and the original worker's commit was then discarded as stale. Claims now carry a lease. claim_next stamps lease_expires_at, the executing kernel renews it in the background while the attempt runs, and recovery reclaims only claims whose lease has lapsed, so a peer mid-attempt is never disturbed. Lease loss consumes the infrastructure recovery budget, never a business attempt, and is recorded as attempt_abandoned evidence. Also: - Recovery is now periodic rather than startup-only, so a peer that dies long after this worker booted is still reclaimed. - Renewal cadence runs on real time while expiry is measured on the injected clock, so virtual-time tests stay deterministic and a jumping clock cannot expire a live attempt (recover() renews own claims before sweeping). - Cancellation is now disambiguated three ways: a lost lease abandons the attempt, an operator cancel releases the step as CANCELLED, and any other cancellation (worker shutdown) re-raises and leaves the step claimed for lease recovery. Previously every cancellation terminally cancelled the step and wedged the run at RUNNING with no path forward. - SqliteRunStore migrates databases written before this change; a step left claimed by the previous binary has no lease and is a genuine orphan. - The worker loop no longer dies on a transient store error. Crash recovery is no longer instant: it is bounded below by the lease duration (default 30s, sweeping every 15s). That is the price of not double-executing. Multi-worker SQLite remains unsupported and is now documented as one worker process per database file: the store's calls are synchronous, so cross-process write contention blocks the caller's event loop including its own renewals.
rx.Retry(max_attempts=5) on a handler raising an ordinary exception executed the handler exactly once. Both the effect-class defaults and any explicit policy that did not name retry_on were resolved to retry only on TransientWorkflowError, so a flaky HTTP call or a dropped connection failed the run immediately while the declared policy promised five attempts. Failures now retry by default: none, read, and idempotent_write resolve to three attempts with exponential backoff on any Exception, and an explicit policy without retry_on retries on Exception too. Narrow it with do_not_retry_on to fail fast. non_idempotent_write still gets exactly one attempt and routes to NEEDS_ATTENTION, since the runtime cannot prove the external effect did not already land. This is a deliberate divergence from the design doc's 'unknown code defects do not retry by default'. That rule is incoherent with an idempotent_write declaring three attempts that can never fire, and it makes the common case -- surviving a flaky dependency, which is the whole point of a durable step -- require boilerplate. Every comparable engine retries by default. TransientWorkflowError remains as an explicit marker for intent and for staying retryable under a narrowed policy.
Runs were pinned to a hash of the whole compiled definition, so adding a state field, retuning a retry policy, or changing a timeout parked every live run of that workflow in NEEDS_ATTENTION -- with no API to get it back. That fires on the second deploy of any real app, not on an exotic one. Runs are now gated on what can actually strand a step: the handler it names is gone, or its persisted payload no longer fits that handler's parameters. Both suspend with a precise, actionable reason instead of a bare digest mismatch. Everything else -- new fields, retuned retries and timeouts, changed hooks, effects, and triggers -- deploys without disturbing work in flight. The definition digest is still recorded on each run as provenance. Adds rx.workflows.resume(run_id) so suspension is a door rather than a wall: it clears the error, grants the frontier step a fresh attempt budget, and makes it claimable. A handler that returns rx.needs_attention() now leaves its step NEEDS_ATTENTION rather than SUCCEEDED, so resuming re-runs the handler that suspended and lets it take a different branch once a human has acted.
Calling another durable handler directly (self.charge()) ran it inline inside the caller's attempt: no retry policy of its own, no effect tracking, no step in the mailbox, and a silent re-execution of its side effect whenever the caller retried. The run still reported success, which is what makes it dangerous -- and it is exactly the shape a code generator reaches for. The compiler now parses each handler body and rejects two traps with an actionable message: an inline call to a sibling durable handler (pointing at 'return MyFlow.charge' instead), and a return of a plain literal (listing the transitions a durable handler may return). Both fail at compile time, before a run exists. Handlers whose source is unavailable, as in a REPL, are skipped rather than guessed at.
Runs could only be fetched one at a time by id, so labels were recorded but never readable and nothing could build an operator view, a CLI listing, or a customer-facing 'your jobs' page. Adds RunQuery and RunStore.list_runs on both stores, surfaced as rx.workflows.list_runs(workflow_id=..., statuses=..., labels=..., limit=...), newest first with a created_before cursor for pagination. SQLite filters labels through json_extract so it stays a single indexed scan rather than loading every run.
rx.webhook(...) compiled but could never fire: only manual roots were
reachable, so the whole provider-driven half of the product was declarative
decoration.
Adds the ingress endpoint at POST /_workflow/webhook/{topic}, registered only
when a workflow actually declares a webhook root. It preserves the raw request
body, verifies the provider signature over those exact bytes, validates the
payload against the declared model, and durably admits the run before
acknowledging -- so a provider that never sees a 202 can safely redeliver.
Redelivery reaches the same run through dedupe_by rather than starting a
second one.
Authentication is not optional by default: a webhook trigger without a
verifier is a compile error naming the fix, and an endpoint that really is
public must say so with allow_unverified plus a reason. rx.hmac_signature()
covers the Stripe/GitHub/Shopify shape, reading the secret from the
environment at request time so it never enters workflow state, history, or a
browser bundle.
Trigger kind is now part of admission: a webhook root cannot be started by
application code, and a manual root is not reachable over HTTP.
rx.schedule(...) validated its five fields and then did nothing; scheduled workflows never ran. Adds a dependency-free UTC cron evaluator supporting the standard five fields with ranges, steps and lists, including the day-of-month OR day-of-week rule that every other cron implementation follows. Expressions are validated when the workflow compiles, so a bad expression fails at add_workflow() rather than silently never firing. The kernel admits one run per occurrence under a request key derived from the occurrence time, which reuses the existing dedupe path: a restart, a second process, or an overlapping sweep all converge on exactly one run per occurrence. Cursors are seeded when the kernel is constructed, so deploying a schedule never backfills history, and catch-up after an outage is capped at ten occurrences so a restart cannot stampede. The worker wakes for the next occurrence rather than polling for it.
A run could only move forward on its own timers, so human approvals, event-driven waits, and anything needing an answer from outside were simply inexpressible -- the gap that most separated this from Temporal and Inngest. A wait is now a BLOCKED slot at the run's frontier, which keeps the mailbox strictly serial: no second open slot, no change to the per-run fence, no concurrent commits. The slot carries the address a delivery must match, and its due_at doubles as the deadline, so a wait timeout fires through the same timer path that rx.after() already used and the virtual clock drives it with no new machinery. The race is settled on one row. A delivery compare-and-swaps the blocked slot to ready and hands the payload to the resume handler; a deadline instead makes the slot claimable, and claiming it *is* the timeout branch. Whichever lands first erases the other's trigger, so a late signal to a finished run is refused rather than silently dropped. A signal that arrives before the run reaches its wait is buffered and consumed by the arming commit itself, so a sender faster than the workflow cannot block it forever. Deliveries never write run state or the state version, so a delivery can never fence a live attempt. BLOCKED is deliberately not a claimable status: a wait with no deadline would otherwise report itself due at time zero and spin the worker against the database, starving lease renewal. Claimability is now one predicate both stores share, with a test asserting an unbounded wait leaves nothing claimable and nothing scheduled.
One dunning workflow written the way a user would: a manual root, a flaky charge with retries and a failure hook, a durable delay, a human decision with a deadline, and completion, failure, and suspension outcomes -- plus a second class reached by a verified webhook and a cron schedule. It caught a real authoring subtlety worth keeping in front of us: run state cannot count attempts, because a failed attempt's patch is discarded by design. The example now models gateway flakiness outside the run, which is where it actually lives.
Nothing about workflows was documented, which also matters because this page is the surface a text-to-workflow generator will learn from. Covers the whole shipped surface: durable steps and effect classes, retries and timeouts, the transition table, waits and typed signals with the approval example, manual/webhook/schedule triggers, inspecting and steering runs, the virtual-clock harness, and what a redeploy does to runs in flight. Every example was run against the real engine rather than written from memory, which caught one error in the draft: rx.Base no longer exists on main, so payload models use pydantic BaseModel.
Runs could only do one thing at a time: the mailbox is strictly serial, so there was no way to enrich and score a lead concurrently, or to do anything Temporal and Inngest express with child workflows. Concurrency now lives in the run graph rather than the mailbox. rx.parallel() commits a BLOCKED join slot in the parent and admits one child run per branch, each with its own state, retries, timers, and history. A child that finishes reports its outcome to the parent's join slot through a compare-and-swap on an arrival counter, so a redelivered result cannot be counted twice, and the slot becomes claimable exactly when the last expected branch lands. The join handler receives one entry per branch carrying run_id, status, result, and error. Keeping each run's mailbox serial is what makes this safe: the parent never has two open slots, so the per-run fence and the one-claim-per-run invariant are untouched, and a failing branch fails its own run rather than the parent's. Children are admitted after the parent's commit lands, so a crash in between leaves a join with no children -- which recovery re-runs -- rather than orphans with no parent.
Nothing bounded how often a root could start, so a chatty webhook produced one
run per delivery and two clicks produced two concurrent syncs of the same
customer. Flow control is the main thing Inngest sells and we had none of it.
A root now declares one start policy, applied at admission and grouped by a
payload field:
singleton one active run per key; a second start either returns the first
(skip) or replaces it (cancel)
debounce a burst collapses into one run, each start pushing it out until
things go quiet
rate_limit starts beyond the cap are refused with retry_after, which is what
you want when a provider can flood you
throttle the excess is delayed rather than dropped, for when every start
matters but the downstream is slow
The grouping key must name a real parameter of the root, checked at compile
time, and a policy without a trigger is rejected since it governs starting.
Only one policy per root, so its behavior stays predictable.
throttle= and debounce= are overloaded by type rather than given second names:
an int is still the browser event action on a session handler, while the
policy object is the durable start policy. Same word, same meaning, and a
session handler passing an int is untouched.
RunStore is a public extension point -- a deployment can back workflows with Postgres or a hosted kernel -- but the protocol's signatures say nothing about the invariants that make durable execution correct. Two implementations were already drifting apart with nothing but shared test files to hold them together, across 22 methods. reflex.workflow.CONFORMANCE_CHECKS is now the specification: 22 checks, each taking a fresh store and asserting one property. Frontier ordering, atomic commit, fenced claims discarding their work, failed attempts discarding state, lease renewal sparing a live claim, recovery reclaiming only lapsed ones, next_due never promising work that is not claimable, a deadline-less wait never becoming due, delivery never touching run state, joins counting each arrival once, finalize refusing while a step is claimed, and the queries start policies depend on. Both shipped stores pass all 22. Anyone adding a store runs the same suite; it is exported and documented for that purpose. Postgres is deliberately not in this commit: no server was available to test against, and an unverified store implementation is worse than none.
Runs could only be reached from inside the app, so diagnosing one meant writing a script against the store. That is the wrong tool at 2am. reflex workflows list filters by workflow, status, and label; show renders a run's state, steps with their attempt and recovery counts, and optionally its history; cancel and resume steer a run without opening the app. Both list and show take --json so the output is scriptable. The commands read the same SQLite database the app writes, so they work against a running deployment or a stopped one, and resume refuses a run that is not actually suspended rather than pretending to act.
Runs recorded a full history but nothing could watch them happen: diagnosing a production workflow meant querying the database after the fact, and there was no way to get workflow activity into metrics or tracing. WorkflowObserver receives every transition the kernel records -- admission, each attempt and its outcome, retries, waits, joins, and terminal dispositions -- with the correlation a durable system needs: run, workflow, step, and attempt. Install one with rx.App(workflow_observer=...). LoggingObserver is bundled for the common case. Instrumentation is deliberately not allowed to break execution: an observer that raises is reported and ignored, which a test asserts by running a workflow to completion under an observer that always throws.
An audit of the whole feature reproduced four real problems, each of which could stop a run permanently. A handler that raised CancelledError itself -- as any handler wrapping its own asyncio work might -- propagated past the cancellation branch and killed the worker task, so every later run in the process silently never executed. The kernel now distinguishes a task that was cancelled from a coroutine that raised: the first is a control signal, the second is an ordinary handler failure that retries. A crash between a fan-out's commit and the creation of its children left a join blocked on children that did not exist, with nothing to recover it; the original comment claiming recovery handled this was simply wrong. Children are now inserted in the same transaction as the join slot, so the window is gone rather than merely narrowed. A child that was cancelled or blew its run deadline never reported to its parent's join, because reporting only happened on commit and those paths finalize without one. Both now report, so a join can no longer wait forever on a child that already stopped. Singleton with mode='cancel' left the superseded run in CANCELLING, so a burst of starts produced several simultaneously active runs under one key. The replacement now waits for the old run to reach a terminal state first. Also fixes a store divergence the conformance suite missed: the memory store kept only the most recent buffered signal per wait key while SQLite queued them, so a second early signal was silently dropped. Two conformance checks now cover buffered-delivery ordering and children being created with their join.
A second, adversarial pass over the whole feature reproduced four problems, one of which was a fix from the previous commit that did not actually work. The guard added for 'a handler that raises CancelledError kills the worker' was a branch that could never be true: asyncio marks a task cancelled whether the kernel cancelled it or the coroutine let CancelledError escape, so the task's own flag cannot tell them apart. Verified directly -- both cases report cancelled() is True -- so the worker still died and every later run in the process silently never ran. The kernel now discriminates on its own control signals plus whether cancellation was requested on the executing task: a handler that raises is an ordinary failure, while a real shutdown still leaves the step claimed for lease recovery. The worker loop no longer dies on any exception, and a dead worker can be replaced. Admission dedupe ran after start policies, so a provider redelivering an event was judged as a new start: with singleton(mode='cancel') the redelivery cancelled the very run it deduplicated to, and answered the provider 202. With debounce, a provider retrying one event faster than the window starved it forever. request_key is now resolved before any policy. rx.fail(details=...) and rx.needs_attention(details=...) passed user values straight to the store, so a datetime raised inside the transaction recording the failure and left the run stuck RUNNING. Details are normalized first, and an unserializable value is recorded as its repr rather than losing the failure. timeout= on a synchronous handler was a lie: asyncio.wait_for cancels the wrapper while the thread runs on, so a timed-out step kept executing and its retries ran concurrently with it. It is now a compile error naming the fix.
Every kernel-level test used the in-memory store, so the harness was certifying semantics production does not necessarily have: any divergence between the two stores could ship green, and one already had. Harness-based tests are now parametrised over both stores, doubling that coverage to 478 workflow tests. Turning it on immediately caught a real divergence: the memory store handed callers live references to the values it was storing, so mutating a returned run's state silently changed committed data -- something a database-backed store cannot do. Reads now detach their mutable payloads, and a conformance check pins it. It also exposed an order-dependent test of its own, which assumed the first child listed was the branch that had already reported to the join. This is the gap that let earlier store divergences through, so it goes in before any further features.
Working through the confirmed findings, highest severity first. A class with __workflow__ that was never passed to app.add_workflow() stayed a session substate, so its durable handlers -- including non_idempotent_write -- remained dispatchable from a browser, and the only feedback was a later error from rx.workflows.start(). Tying detachment to registration meant the omitted line, exactly the one a generator drops, left money-moving handlers exposed. Detachment now happens when the class is created, so registration only adds the definition to the kernel. A wait whose deadline had already fallen due still accepted its signal, because delivery matched on BLOCKED alone and never compared the deadline. A seven-day approval that expired three weeks ago would be approved, and a sender that lost the race was told 'buffered' and then refused as 'duplicate'. Delivery now refuses with a distinct 'expired' disposition, which also stops a stale signal resolving a later wait on the same channel. A child failed by exhausting its recovery budget never reported to its parent's join: that path fails the run inside the store, and recover() only returned a count. The parent, and every ancestor, waited forever. recover_orphans now returns the runs it failed so the kernel can report them. Start policies of the wrong type vanished instead of raising: debounce='30s' -- a very plausible generation given every other duration in the API is a string -- silently disabled debouncing, and singleton='cid' failed only in production once two runs overlapped. All four are now type-checked at decoration, discriminating the browser event action on int. Run pagination used created_at alone, so runs sharing a timestamp -- the shape every fan-out produces -- were silently skipped; the cursor is now (created_at, run_id). The SQLite label filter interpolated user-supplied keys into a JSON path expression and now matches them as values.
Two more confirmed findings. Lease renewal failures were swallowed at debug level, so a store that kept failing -- which a contended SQLite file does, raising after a five second block -- left the attempt running while its lease quietly lapsed. Recovery then handed the step to someone else and the external effect ran twice, with a log line that is off by default as the only evidence. The kernel now tracks when the lease actually expires and abandons the attempt once too little of it remains to survive another failed round-trip: better to stop work you can no longer prove you own than to race the worker that is about to take it. rx.parallel resolved its branches through a path that skipped the trigger check start() enforces, so a webhook-only root -- one that exists precisely because only a verified provider may start it -- could be started from application code by naming it as a branch. Branches are now held to the same manual-root rule as a direct start. Fanning out to a handler of your own class is now a compile error. Each branch becomes a child run with fresh state, so a same-class branch cannot see anything the parent did; it is the most natural spelling and it silently did the wrong thing.
Every store call is synchronous on the caller's event loop, which also serves HTTP, websockets, and session state. SQLite's default busy timeout is multiple seconds, so a second process writing the same file could stall the whole app before raising -- and the error then landed in lease renewal, where a silent failure used to mean the kernel re-executed its own attempt. The busy timeout is now short: contention surfaces quickly as a transient error the kernel retries, rather than freezing everything first. Combined with the previous commit, a store that cannot be reached now degrades to abandoning the attempt instead of duplicating its effect. This bounds the symptom rather than removing the cause. Offloading the store's calls to a thread is the real fix, and one worker process per database file remains the supported deployment -- now stated plainly in the docs alongside why horizontal scale wants a different store.
The kernel claimed one step and awaited it before claiming again, so a single process executed one step at a time no matter how many runs were waiting. One ten-minute step stalled every other run in the deployment -- including their timers and deadlines -- which is not a throughput story any durable engine can be compared on. The scheduler now fills up to max_concurrency slots, eight by default. This needs no new locking: each run has exactly one claimable frontier step, so two concurrent claims are necessarily different runs and a run's own mailbox stays strictly serial. A test asserts both halves -- that attempts really do overlap, and that one run's steps still complete in order. Three details that had to be right: finished attempts are pruned synchronously rather than by a done-callback, or the scheduler spins on work it already finished; a round waits only on the attempts it started, so a second caller pumping the same kernel is not blocked by an attempt it does not own; and cancelling the scheduler cancels the attempts it started, since asyncio.wait does not cancel what it waits on. The test harness stays single-slot so virtual-clock tests remain deterministic; production defaults to eight.
rx.parallel always waited for every branch, so the one shape everybody reaches for -- ask two vendors, take whoever answers first -- could not be expressed. Worse, the losing branch kept running and kept calling out to a vendor after the order was already booked. mode="first" sets the join's expected count to one. The arrival that resolves the join now says so, and the kernel cancels the siblings still in flight, found through a new list_children(parent_run_id, parent_ordinal) store query rather than a scan of the run table -- indexed in sqlite, and covered by a conformance check so any future store has to answer it too. Two compile-time diagnostics came out of writing the tests, both failure modes a code generator will hit: self.book passed as then= is a bound method, not a routable transition, so it failed at runtime -- as a retrying durable step, which is the worst place to learn about a typo. The handler-body guard already rejected self.step() calls; it now rejects bare self.step references too and names the class form. Passing a list of workflow classes where varargs were expected died with "cannot use 'list' as a dict key" from inside the registry. register() now says what it wanted. The docs example was executed before committing.
Throttle deferred every excess start by exactly one window, so a burst of a hundred with limit=10 admitted ten now and scheduled ninety for the same instant one window later. That is not throttling, it is a delay line: the downstream the throttle exists to protect sees the same spike, just later, and the next window then holds ninety starts against a limit of ten. Each start is now placed at least a window after the limit-th most recent scheduled start under its key, which spaces the backlog at exactly the configured rate and holds the sliding-window bound rather than a per-window one: any window of length period contains at most limit starts, because a start is always a full window after its limit-th predecessor. This needs a new store query, nth_recent_start, since counting admissions in a window cannot see where the deferred ones are already scheduled. A run's scheduled start is when its root slot comes due, so a debounced or throttled run counts at the time it will run, not the time it was admitted. Covered by a conformance check, so a future store has to answer it the same way. The regression test asserts the schedule directly (0, 0, 10, 10, 20, 20 for six starts at limit=2 over ten seconds) and then advances the clock to confirm the runs execute on it.
SQLite takes one writer, so a deployment was capped at one worker process per database file. That is the ceiling that keeps this from being comparable to Temporal or Inngest at all, and it is not something the kernel can fix -- it is the store. PostgresRunStore claims a run's frontier step with FOR UPDATE ... SKIP LOCKED, so workers never queue behind each other and never take the same step. Adding a process adds throughput. A worker that dies mid-step holds a lease that another worker may only reclaim once it lapses, so a slow step is never duplicated -- there is a test for exactly that, and one asserting twenty non_idempotent_write runs across two kernels execute exactly once each, with both kernels provably taking a share of the work. The store is not trusted on assertion: the 29 conformance checks run against it, and the whole workflow suite -- 260 harness tests -- now runs a third time against a real server whenever REFLEX_TEST_POSTGRES names one. That third parameter immediately paid for itself twice. It caught a test of mine that asserted a race always starts every branch, which is not an invariant: a loser cancelled before its first step is the better outcome, and only Postgres's ordering exposed the assumption. And driving the real CLI against a Postgres URL surfaced that each command ran its own asyncio.run, which is fine for a file and fails for a pool, whose connections belong to the loop that opened them. Commands now run in one loop. Two things came out of the port. Pyright rejected the schema name spliced into DDL, since psycopg types raw SQL as LiteralString; the name is now composed as an identifier, which is a better guarantee than the check it replaces. And the observer turned out to see only admissions and commits -- not attempt starts, which are the spans a tracer actually wants -- so every recording site now reports, correlated to its workflow. A throwaway schema per test isolates them. Dropping one first evicts its own backends, because a test loop that dies mid-transaction leaves a pooled connection idle holding locks, and the DROP would otherwise wait on it forever. That was a real hang, reproducible only under random test ordering. Postgres is optional: pip install 'psycopg[binary,pool]'.
A run waiting on a decision is the most common human-in-the-loop shape, and the naive implementation -- a URL carrying a run id -- is an open door. An approval link here is an HMAC token over run, channel, payload, delivery key, and expiry, so an edited link is refused rather than believed; it is spent once; and it never decides on GET, because mail clients and scanners fetch URLs before a person reads the message, so a link that approved on GET would approve itself in transit. The secret comes from REFLEX_WORKFLOW_APPROVAL_SECRET with deliberately no default, and a server missing it says 'not configured' instead of masquerading as an expired link. Executing the docs example before committing caught a real defect: a channel declared with a pydantic model failed to serialize into the token -- and every realistic channel is typed -- so payloads now go through the same reduction the signal path uses. Links need to know which run built them, which is a capability handlers were missing generally. rx.current_run() now exposes the attempt's identity -- run, workflow, slot, attempt, epoch -- bound per attempt via a ContextVar that to_thread carries into sync handlers, plus idempotency_key(): stable across retries of a step, distinct across steps, which is exactly the contract a payment API's idempotency header wants. Chasing an intermittent test hang also root-caused a real defect: the worker loop's treated CancelledError as a retryable error, so anything that cancelled the worker task without calling aclose() -- a task group, a supervisor, an event loop tearing down -- waited forever on a task that had gone back to polling. The hang reproduced with the fix removed and disappears with it. The kernel also cancels its in-flight attempts on close instead of leaving them running against a store nobody reads, and the test harness closes stores it created, which previously leaked a Postgres pool into every later test in the process.
A handler is the unit of retry, so a handler that makes three calls and fails
after the second repeats the first two on retry -- three charges for one
order. rx.step(name, fn, ...) runs a callable once, records its result
durably at the moment it returns, and replays it to every later attempt of
the same handler, including one recovered from a crashed worker. The journal
is epoch-fenced: an attempt whose lease was reclaimed cannot record, so a
zombie stops instead of duplicating a side effect. Results round-trip through
serialization before the handler ever sees them, so the first execution and a
replay produce identical shapes -- a difference there would only surface
during retries, the worst place to find it. Async handlers await the call;
sync handlers call it bare and block, bounded so a stalled loop fails the
step rather than pinning the worker thread forever. A name reused in a loop
is numbered per occurrence. Recorded keys appear in run history.
queue= on a durable handler has been accepted since the first commit and
silently ignored, which is the worst state a parameter can be in. It now
routes: every step is stamped with its handler's queue ('default' when none),
a worker claims only from queues it serves (rx.App(workflow_queues=...)), and
per-run order holds across queues -- a run whose frontier sits on an unserved
queue waits for the right worker rather than running the step somewhere it
was configured not to. Wait and join slots take the queue of the handler that
resumes them; children take their root's.
Both are covered by conformance checks, so all three stores answer the same
way, and the whole workflow suite passes against memory, SQLite, and
Postgres. The teardown watchdog added to the workflow conftest (REFLEX_DEBUG_HANG=1)
names any task that outlives its cancellation instead of hanging the run
silently -- built while chasing an intermittent suite hang that so far only
reproduces when several suites contend for one database.
Contributor
Greptile SummaryThe PR introduces an experimental durable-workflow engine for Reflex, including persistent stores, leased execution, triggers, flow control, composition, operator APIs, testing utilities, and documentation.
Confidence Score: 3/5The PR is not safe to merge because two outstanding same-run lifecycle races can leave attempts untracked during shutdown or remove a live replacement attempt’s lease renewal.
Files Needing Attention: reflex/workflow/kernel.py and focused lease/shutdown regression tests in tests/units/workflow/test_lease.py
|
| Filename | Overview |
|---|---|
| reflex/workflow/kernel.py | Implements concurrent leased workflow execution, recovery, retries, transitions, shutdown, and worker scheduling. |
| reflex/workflow/store.py | Defines the workflow store contract and the in-memory and SQLite persistence implementations. |
| reflex/workflow/postgres.py | Adds the multi-worker Postgres implementation with transactional claiming and schema namespacing. |
| packages/reflex-base/src/reflex_base/workflow.py | Adds public authoring-time workflow configuration, trigger, policy, and transition value types. |
| reflex/workflow/definition.py | Compiles workflow state classes and durable handlers into validated runtime definitions. |
| reflex/workflow/ingress.py | Adds durable webhook admission with payload validation, signature verification, and deduplication. |
| tests/units/workflow/test_lease.py | Covers lease renewal, loss, recovery, and shutdown behavior, but does not exercise overlapping attempts for one run. |
Reviews (16): Last reviewed commit: "Pin two more contract clauses with confo..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Phase 2's foundation, on all three stores: a webhook event addressed to a signal channel by business key is durable from the moment it is acknowledged, delivered exactly once whenever its run exists, and visible when it cannot be. ingest_channel_delivery makes one transaction of the whole decision. The row keyed by (workflow, channel, correlation_key, event id) is written first, so the ack and the record are one fact: a provider redelivery and a crash-after-ack replay both collapse into "duplicate". If the correlation key already admitted a run, the payload is delivered in the same transaction -- through the same per-run inbox and dedupe as every other signal. If not, the row waits PENDING. Admission flushes parked mail inside the admitting transaction, on both doors -- plain admit and policy admit_flow -- so a crash cannot separate "the run exists" from "its early mail reached it". On Postgres the channel rows are locked before the run row on every path (ingest, admit, admit_flow, replay), one order, no cycle. A delivery nothing can take is a dead letter, never a silent drop: a terminal or past-deadline run at ingest, or a PENDING row unclaimed past its TTL (sweep_parked). Dead letters carry their reason, list via list_parked, and replay via replay_parked with the same idempotency -- replaying a delivered row is "duplicate", never a second signal. Store surface: ingest_channel_delivery, list_parked, replay_parked, sweep_parked; new dispositions "parked" and "dead_letter" (documented in the contract vocabulary, which this repo's guard test enforces); SQLite schema v4 adds workflow_channel_inbox; Postgres adds the same table to its advisory-locked DDL. deliver() on SQLite and Postgres is factored into an in-transaction form both admission and ingest share, with refusal branches that write nothing so the caller's transaction stays committable. Six conformance checks pin it everywhere, including the acceptance flow: park before the run exists, redeliver three times, admit the run -- the wait resolves with exactly one payload, and the late redelivery is still "duplicate". Ingress wiring (rx.Signal(trigger=rx.webhook(..., correlate_by=...)) routing) and the SIGKILL crash test ride the next commit; the base API for it (correlate_by on webhook(), Signal.trigger validation) is already in.
Phase 2 complete on top of the channel inbox. A provider event now
reaches a waiting run with no glue code:
class Order(rx.State):
shipped = rx.Signal(
Shipment,
trigger=rx.webhook(
"shippo.shipped",
verify=shippo_verifier,
dedupe_by="event_id",
correlate_by="order_id",
),
)
Root webhooks start runs; channel webhooks locate them. correlate_by
names the payload field carrying the business key, matched against
request keys; dedupe_by names the provider's event identity. A channel
trigger requires both -- without them a delivery cannot be routed
exactly once -- and one topic identifies exactly one target, root or
channel, enforced at route collection.
The webhook endpoint verifies the signature, canonicalizes against the
channel's model, extracts both identities (a missing one is a 400
before anything exists -- accepting it would mint a dead letter for a
sender error a 400 would have fixed), and hands the payload to
kernel.ingest_channel, which checks the channel is declared and wakes
the worker on a resolved delivery. Every durable outcome acknowledges
202: once the row is committed the provider must stop retrying,
whether the payload landed, parked, deduplicated, or died visibly.
Recovery sweeps PENDING deliveries unclaimed past 30 days into
"unclaimed" dead letters and says so. Operators get the loop on both
surfaces: reflex workflows deadletters [--status ...|--all|--replay ID]
and GET /deadletters + POST /deadletters/{id}/replay on the standalone
service (read/operate scopes).
The plan's acceptance scenario now runs with a real SIGKILL: the
shipment webhook arrives before the order workflow exists, the process
is killed immediately after acknowledging it, the provider redelivers
twice from fresh processes, the order workflow starts later -- exactly
one signal arrives, held by the fsynced effect ledger, by the run's
inbox read straight from the database, and by the single DELIVERED
channel row. The same flow minus the kill runs over real HTTP in the
ingress suite, and the dead-letter loop runs over the standalone
service's own webhook route.
Ten canonical patterns from five ecosystems, recreated as executable Reflex workflows from the projects' own official examples and run as tests on every store: Temporal's money transfer, Restate's travel-saga compensation, Inngest's onboarding wait-for-event / fan-out / checkpointing, Prefect's API-sourced ETL, Celery's chain and chord, Airflow's TaskFlow ETL, and DBOS's transactional outbox. 48 test rows across Memory, SQLite, and Postgres, all green. Beyond regression cover, this is the parity ledger: when a competitor's pattern needs contortions here, the contortion is written down where CI runs it. Gaps the exercise surfaced, tracked for the roadmap: per-step retry / timeout / queue overrides on rx.step; pub/sub fan-out atop the new correlated-event inbox; a first-class saga/compensation stack; result piping and lighter parallel branches; and transaction-coupled database steps, where DBOS's same-transaction outbox guarantee has no current equivalent.
The engine half of workflow-only deploys, per the roadmap's Phase 3 release-safety spec and the standing design decision: runs pin to the release that admitted them, and a deploy is a set of reads, not a ceremony. Two identities per run now, doing different jobs. The definition digest (structural, existing) decides whether code CAN run a payload -- mismatches suspend at dispatch. The new release id -- REFLEX_RELEASE_ID or WorkflowRuntime(release=...) -- decides whether code MAY: every run and every fan-out child stamps the release that admitted it, and claim_next on all three stores skips runs pinned elsewhere, so a run drains on the code that recorded its payloads and never silently mixes two releases. Pinning binds only when both sides declare: unpinned runs are anyone's, and a worker with no release (dev, tests) serves everything -- which is also why the entire existing suite runs unchanged. Workers register a durable identity -- id, release, queues, capacity -- at startup (after the first recovery, so the timestamp is store-synced), heartbeat on the lease-renewal cadence rather than a second timer, and deregister on clean shutdown; a crashed worker stays listed with a stale heartbeat, which is exactly what a fleet page should show. reflex workflows fleet renders the registry and per-release active-run counts, and --can-retire RELEASE is the deploy gate: nonzero while any active run is pinned to that release, because stopping its workers early strands those runs until leases lapse. RunQuery gains a release_id filter (all three stores + the memory matcher), RunSnapshot and the HTTP read surfaces report each run's release, and SQLite reaches schema v5 (release_id column + the workflow_workers table; Postgres adds both additively in its advisory-locked DDL). Pinned by three conformance checks on every store -- routing, registry round-trip, retirement counts -- and a rolling N/N-1 kernel test: a run sleeps for a day on v1, v2 deploys and takes new admissions, v2 never claims v1's sleeping run, v1 drains it, and the retirement count reaches zero. One test lesson recorded in the diff: worker heartbeats are store-clock timestamps, and two clock syncs can differ by milliseconds either way, so freshness assertions carry the sync's own jitter tolerance.
The audit half of the operator console's acceptance criterion ("every
mutation shows who performed it and why"), landed ahead of the UI so
the console renders a record that already exists.
Attribution rides the history. The operator store operations --
request_cancel, retry_run, skip_step, resume_run, finalize_run -- take
an optional attribution mapping and merge it into the operator-facing
history event, inside the same transaction as the mutation itself. An
audit that lives outside the history would drift from it; this one
cannot, because the record that answers "what happened" is the record
that answers "who did this".
The kernel's operator methods (cancel, retry, skip, resume,
force_finalize) take actor and reason and build the payload. The CLI
stamps the invoking user (REFLEX_ACTOR overrides, getpass fallback)
and grows --reason on cancel, retry, skip, resume, and complete; fail
already demanded a reason and now records it as attribution too. The
HTTP API records the caller's X-Actor claim -- tokens are anonymous,
so the header is a claim, recorded as given, and an authenticating
proxy can stamp it; "api" beats naming nobody -- plus the body's
reason on cancel, retry, and resume.
Pinned by a conformance check exercising all five operations on all
three stores, a serve test proving X-Actor and the body reason land in
history, and a CLI subprocess test proving --reason and REFLEX_ACTOR
do. Contract 9 states the rule.
Phase 4's UI over the data layer the last commits finished: `reflex workflows console` serves a Reflex app with four pages -- runs, one run's story, the worker fleet, and channel deliveries -- so an operator finds and repairs a stuck run without SQL or the CLI. Runs: filter by workflow and status, status badges, release, ages. Run detail: state, steps with attempts and errors, result or error, children by join slot, and the full history with actor and reason per event; cancel, retry, skip, and resume buttons that take a reason and go through the same kernel operations the CLI uses, so the mutation lands attributed in the run's own history. Fleet: registered workers with release, queues, capacity, heartbeat, and per-release active counts. Events: parked, delivered, and dead-letter deliveries with replay. The console is a read-and-repair surface, never a worker: it opens the store through a worker-less client runtime, held once per process under a lock and closed by the app's lifespan. It has no login of its own, so the CLI binds loopback by default and warns when asked to do otherwise. The command materializes the minimal Reflex project the app needs and hands off to `reflex run`; the scaffold is rewritten each launch. Auditing the draft before this commit found four defects, all fixed: `async for` over coroutine event handlers (a TypeError on the first click); a lifespan hook written as a bare async generator, which the framework refuses at registration (it must be an asynccontextmanager); a declared `run_id` state var shadowing the dynamic route argument, refused at add_page; and an unlocked lazy runtime init that two concurrent page loads would race into two pools. Separately, the validation adapter cache now survives an unhashable type hint instead of failing every boundary that meets it. Tested by building every page's component tree and the app, and by driving each state's handlers directly against a seeded SQLite store -- listing and filtering runs, loading a run and retrying it with a reason that shows up in history under the operator's name, listing and replaying a parked delivery, reading an empty fleet -- plus a CLI test that the materialized project is the console.
The audit's first architecture gap was the console having no login of its own. It now signs operators in with the service's scoped API tokens -- read to view, operate to repair -- and the token model gained the piece that makes attribution trustworthy on both surfaces. Tokens can be bound to principals: REFLEX_WORKFLOW_API_TOKEN_PRINCIPALS maps names to tokens (alek=tok1;deploy-bot=tok2). A bound token signs its actions as that principal on the console and on the HTTP API alike, where the credential now outranks the X-Actor header -- the caller's claim is recorded only when the credential says nothing, and "api" only when neither does. This is the model Temporal Cloud uses (API keys bound to users or service accounts) rather than the free-form identity string alone, and it is what lets an audit answer "who" from something the caller could not have written themselves. The console's LoginState checks a typed token against the configured scopes, records the principal or the typed name, drops the secret from state on success, and redirects to the runs page. Every page's on_load sends an unauthenticated visitor to /login; every mutation needs the operate scope and reports a notice rather than acting when it is missing. With no token configured at all the console stays open -- the loopback default -- because a console nobody can log in to protects nothing; the CLI still warns when asked to bind elsewhere. get_state has no event context on a directly constructed state, so every handler that resolves the login has a seam (act_as, replay_as, load_runs, load_fleet, load_deliveries) that the tests drive with an explicit LoginState; ScopedTokens accepts explicit grants and principals so tests stop poking __new__. Tests cover a rejected token, a bound token naming its principal over a typed name, an unbound token recording the typed name, scopes enforced only once a token exists, a read-only login refused on retry and replay with a notice, and on the API a bound token beating a spoofed X-Actor while an unbound one still honors the header.
Each console page keeps itself current while it is mounted: a background
event re-reads the store every three seconds and stops when the browser
leaves the page or the login no longer admits reading. Polling the store
rather than streaming from a worker is the point -- the store is the one
thing every worker shares, so the view survives any worker restart
without depending on one being alive, which is the acceptance bullet
this closes ("live updates work across worker restarts").
Registered alongside each page's first load (on_load=[refresh, watch]),
gated on the page route so a watcher for the runs table dies when the
operator navigates to a run, and bounded by the same read-scope rule as
the page itself. The stop predicate is a pure function with its own
test; the app still registers every page with its watcher.
Run-level operator actions ride each run's own history, attributed. Two mutations had no run to carry them -- replaying a dead letter and purging finished runs -- and so were the only operator decisions nobody could account for. They now land in an append-only audit log, written inside the operation's own transaction on every store, with who asked, what was done, to what, the outcome, and why. Only attributed actions are recorded. The TTL sweep and recovery replay nothing on anyone's behalf and leave no entry, so the log reads as operator decisions rather than as noise. This is the split Temporal makes between workflow history and its control-plane audit stream, kept deliberately narrow: a run-level action never appears here twice. Surfaces: `reflex workflows audit` lists entries; `deadletters --replay` and `purge` take --reason and stamp the invoking user; the standalone service's replay endpoint records the actor from a principal-bound token (else the X-Actor claim) plus the body's reason, and GET /audit reads the log under the read scope; the console gains an Audit page with a live watcher, and its replays are signed with the login's name. SQLite schema v6 adds workflow_audit; Postgres adds the same table to its advisory-locked DDL. A conformance check pins the semantics on all three stores -- newest first, no entry without an actor, purge and replay both recorded -- and surface tests cover the CLI, the service (bound principal beating no header), and the console. Contract 9 states the rule and the boundary between history and audit.
What starts each workflow -- and where each schedule stands -- is now one
summary (reflex.workflow.triggers.describe_triggers) behind three
surfaces: reflex workflows triggers, GET /triggers on the standalone
service (read scope), and a Triggers page in the console. Three views
that computed this separately could disagree about which URL a provider
posts to or when a cron next fires; one function cannot.
Each webhook row names the topic, the path a provider posts to, whether
it is signature-verified, and -- the part an operator cannot see from
the source -- whether the verifier's secret is actually present in the
environment. Channel webhooks appear beside root webhooks with their
correlation field. Each schedule row carries its cron, the next
occurrence, and how far its durable cursor lags, read from the store.
Manual roots are listed so "how does this one start" always has an
answer.
The console learns definitions by being started with the workflow
module (reflex workflows console workflows.py, exported to the app as
REFLEX_WORKFLOW_CONSOLE_TARGET); it registers them read-only through the
same worker-less runtime and executes nothing. Without a module the
page says how to get one instead of showing an empty table.
Also: discover_workflows(module) replaces three copies of the
class-discovery expression in the CLI; WEBHOOK_ROUTE carries a Starlette
path converter ({topic:path}), which a plain "{topic}" substitution
silently left in place -- the summary substitutes the converter form.
Tests cover every trigger kind including the secret-present tri-state,
cursor lookup by the kernel's key, module discovery ignoring inherited
declarations, the console reading a real module file, and the HTTP
route under auth.
…onsole The things whose absence is silent in production -- a webhook verifier's secret that is unset, an approval key missing, the API token that decides whether the HTTP surface exists at all -- were checked only by `reflex workflows doctor`, on a laptop, before deploying. They are now one summary (reflex.workflow.health.describe_connections) that doctor, GET /connections on the standalone service, and a Connections page in the console all report from, so the answer to "why did nothing happen" is on screen wherever an operator is looking. Each row is a dependency by name and presence, never by value: every verifier secret variable with who depends on it (a missing one is a problem -- those webhooks will refuse every delivery), unverified webhooks with their declared reason, the approval-link key, the API token or any scoped variant, and schedules that need a serving process. The console badges the problem count; the service never echoes a secret; doctor keeps its store-reachability check and its exit code. doctor is rebuilt on the summary and on discover_workflows, so its checks and the console's cannot drift. Tests cover the summary against a half-configured deployment (present and missing secrets, an unverified root, a channel webhook, a schedule), a scoped token counting as the API being configured, the HTTP route under auth with a secret value proven absent from the response, and the console page counting a missing secret from a real module file.
The first of Phase 4's "management as mutation" items: a durable per-
schedule pause flag the sweep honors, reachable from the CLI, the
standalone service, and the console, and audited like every other
run-less operator action.
A paused schedule skips its occurrences and keeps its cursor moving, so
resuming never backfills the pause -- an operator who paused a nightly
job for a week wants one run when they resume, not seven. That is the
semantics Temporal's schedule pause and Inngest's function pause both
settled on, and the harness test pins it: three paused hours of an
hourly schedule admit nothing while the cursor advances, and resuming
admits exactly the next occurrence. Skipped occurrences are said in the
worker log but feed no lost-work counter; they were asked for.
The flag lives in a new workflow_schedule_state table on all three
stores (SQLite schema v7, Postgres additive DDL) behind
set_schedule_paused/paused_schedules; pausing and resuming write
pause_schedule/resume_schedule audit entries in the same transaction
when attributed. Surfaces: reflex workflows schedules pause|resume KEY
--reason; POST /schedules/{key}/pause|resume under the operate scope
with the actor from the credential; a paused badge and Pause/Resume
buttons on the console's Triggers page, gated on operate, via the same
testable seam the other console mutations use. /triggers rows and the
trigger summary carry the schedule key and its paused flag.
Tests: the kernel semantics above; the service end to end (403 for a
read token, /triggers showing paused, audit naming the bound
principal); the console refusing a reader and showing paused after an
operator's click; the CLI writing an attributed audit entry; the
conformance suite pinning durability and audit on every store.
The last "management" item from the audit that did not need a new API: rotating a webhook secret. A verifier's secret variable may now hold several comma-separated secrets, and a delivery verifies if it was signed with any of them -- so rotation is list the new secret beside the old, cut the provider over, drop the old, with no moment at which one side has rotated and the other refuses everything. Both rx.hmac_signature and rx.stripe_signature read through one helper and try every configured secret with a constant-time compare each, accumulating rather than returning early, so the comparison count does not tell an attacker which secret was close. An unrelated secret never verifies; dropping the old one ends its window immediately. doctor, GET /connections, and the console keep reporting the variable's presence and never its contents. Tests sign the same body with the outgoing and incoming secret and require both to verify while both are listed, a stranger to fail, and the outgoing one to fail the moment it is removed -- for the plain HMAC verifier and the Stripe scheme alike. The contract states the rotation rule.
… failure matrix covers deliveries, releases, alerts, paused schedules; docs state the non-idempotent-write recovery exception
…st-commit successor and child arrival, N claims, killed sweep, dead release, hour-long outage, pre-ack admission
… console runs page filters by labels; docs start-here
…a token, not the database
…onterminal state it was in at the kill
…ly-once ledgers and terminal invariants after; SQLite one worker, Postgres several
…s; child-arrival crash test recovers in two phases
Resolves: the durable-event decorator keeps both its config block and main's supersedes marker; reflex deploy moved to the Cloud CLI, so the workflow-only deploy guard now wraps the imported command instead of living in the framework's own deploy; pyi hashes regenerated.
…ff; chaos soak models the idempotent provider for signal handlers and asserts one arrival per wait The Postgres soak, severing connections while respawning workers, caught a fresh worker taking AdminShutdown straight out of startup and exiting. The first recovery and registration now retry five times with backoff before failing with the real error. The soak's two remaining false alarms were its own: an unguarded signal handler may re-execute after a kill (contract §2), and a signal that arrives before its wait is armed is recorded as buffered, not resolved.
…t to the PR; fix fragments that had no type; add the reflex-base fragment
Merging this PR will not alter performance
Comparing 
Footnotes
|
…ng reflex pydantic is an optional extra of Reflex and CI runs the suite once without it; the kernel and ingress imported it at module level and the CLI imports the workflow package eagerly, so every reflex command failed without it. Those imports are now lazy, a reflex[workflows] extra names the dependency, and a runtime built without pydantic fails once with the install line rather than degrading validation or raising from the first admission. The workflow test directory skips as a unit when pydantic is absent. The package root no longer re-exports the conformance suite, which imports pytest and was being pulled into every app. Two tests that deliberately signalled a misspelled channel now use an unknown name codespell accepts.
main deprecated console.error/warn/debug in favour of logging; the workflow package and CLI had fifty-six such sites, each now a deprecation warning per call. They use logging.getLogger(__name__) like the rest of the CLI; console.print stays for plain command output, as main still uses it. CLI tests that asserted on error text now read caplog, the convention main's CLI tests already follow.
…ws; the soak only counts kills that land on a held claim datetime.UTC is 3.11-only and the cron module used it, so schedules would have crashed on 3.10; every site now uses timezone.utc. The real-kill, chaos, and SIGTERM-drain tests need SIGKILL semantics and skip on Windows, where the contract's crash evidence is not claimed. On shared runners a respawned worker can take longer to boot than the kill interval, so the soak now waits for a worker to hold a claim before killing it and keeps going until at least one kill landed. The serve lifecycle test accepts that its background worker may finish the run before the duplicate post arrives; either disposition refuses a second handling. The catch-up warning test reads caplog now that the warning is logged.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Warning
ULTRA EXPERIMENTAL — DO NOT REVIEW, DO NOT MERGE.
This is a messing-around branch for collaborating on ideas between agents. We are not at all sure any of this comes in. No review wanted; nothing here is final.
What this is
An exploration of durable workflows embedded in Reflex — a Temporal / Inngest / Trigger.dev / Zapier-class engine where a workflow is an
rx.Statesubclass, steps are@rx.event(durable=True)handlers, and control flow is the handler's return value.Core design bet: no replay — state is snapshotted per step, never rebuilt by re-executing user code, so handlers are ordinary Python with no determinism constraints.
What's in the branch (37 commits)
FOR UPDATE ... SKIP LOCKED, multi-worker, schema-namespaced) + a runnable conformance suite (32 checks) any store must passREFLEX_WORKFLOW_DATABASEresolves the store for the app, its workers, and the CLI alikerx.parallel, incl.mode="first"racing), signed single-use approval links for email (GET never decides)queue=+rx.App(workflow_queues=...))rx.steprecorded substeps (epoch-fenced journal replayed to retries and crash recoveries);rx.current_run()+ retry-stable idempotency keys; policy keys reach into model payloadsrx.workflows.*,reflex workflows list/show/cancel/resume/check(SQLite path or Postgres URL;check --jsonis the generation-loop validator)WorkflowTestHarnesswith a virtual clock; every workflow test runs against all three stores (~930 tests); Playwright integration tests prove the engine inside real dev and prod servers (browser → durable timer → completion; signed webhook over HTTP)Notable war story: an intermittent teardown hang was root-caused to a swallowed task cancellation during lease release (
task.cancelled()is never a valid "who cancelled" discriminator — third instance of that bug class here), fixed with a deterministic repro and verified over 30 clean suite runs.🤖 Generated with Claude Code