[Refactor/Chore] Define transaction ownership for core workflow persistence layers #41560

AI disclosure: This issue was drafted with Codex. I have reviewed the report, and I am responsible for the content.

Self Checks

• I have read the Contributing Guide and Language Policy.
• This is only for refactors or chores; if you would like to ask a question, please head to Discussions.
• I have searched for existing issues search for existing issues, including closed ones.
• I confirm that I am using English to submit this report, otherwise it will be closed.
• 【中文用户 & Non English User】请使用英语提交，否则会被关闭 ：）
• Please do not modify this template :) and fill in all the required fields.

Description

Core workflow persistence currently has no single owner for the transaction that makes a workflow pause durable.

When Graphon emits GraphRunPausedEvent, related state is persisted by independent layers:

• WorkflowPersistenceLayer persists WorkflowRun.status = PAUSED, outputs, and execution statistics through WorkflowExecutionRepository.save().
• PauseStatePersistenceLayer separately serializes the resumption context and calls create_workflow_pause() to persist the snapshot reference, WorkflowPause, and WorkflowPauseReason records.

These layers use separate repository calls and separate database transactions. Their registration order is therefore part of the correctness contract, even though that dependency is not represented by either layer's interface.

The generic Graphon layer dispatcher also catches and logs every on_event() exception. This is appropriate for observational extensions, but pause persistence is a correctness-critical side effect. If pause persistence fails after the workflow run update commits, execution can expose a durable partial state:

• WorkflowRun.status is PAUSED;
• there is no active, readable pause state for the current pause;
• the paused event can still be published to application consumers;
• the workflow cannot be resumed consistently.

The invariant should instead be owned and committed by one persistence boundary:

A workflow run is observably PAUSED only when its active pause record, pause reasons, and readable resumption snapshot have been committed successfully.

Proposed direction

1. Consolidate persistence responsibilities that own the workflow execution aggregate. WorkflowPersistenceLayer and PauseStatePersistenceLayer should not independently write different parts of the same pause transition. This can be achieved by integrating pause handling into the core workflow persistence layer, while keeping tracing, inspector publication, notifications, and other observational side effects in separate best-effort layers.
2. Add a precise repository aggregate operation for pausing a workflow. It should hide the database transaction that updates the workflow run and replaces the pause and pause-reason records, rather than requiring multiple self-committing repository calls.
3. Treat the core persistence path as fail-fast. A persistence failure must prevent publication of a committed paused outcome and allow the run to converge to an explicit failure state; it must not be swallowed by the generic best-effort layer dispatcher.
4. Publish pause notifications and application events only after the pause transaction commits.

For snapshot storage, a pragmatic consistency model is sufficient: write the new immutable snapshot object before committing its database reference, and clean up the previous object on a best-effort basis after it is no longer needed. A failed cleanup may leave an orphan object, but must not invalidate workflow state.

This proposal is about transaction ownership rather than introducing a generic coordinator. The core persistence layer should map engine events to durable workflow state transitions, while the repository should hide storage and database mechanics.

Motivation

The current design is temporally decomposed: correctness depends on which layer happens to run first and which independent commit succeeds. This creates hidden coupling and makes failures in any critical layer indistinguishable from failures in optional observers.

Establishing one transaction owner would provide the following guarantees:

• observers see either the previous valid state or a complete new paused state, never a partially committed pause;
• a successful paused outcome implies that resumption state is available;
• changes to layer registration order cannot alter transaction correctness;
• failures have explicit control-flow semantics instead of being reduced to logs;
• storage cleanup remains operationally independent from workflow correctness.

Suggested regression coverage should inject failures before snapshot creation, before database commit, and after commit during cleanup, and assert the workflow-run/pause invariant and emitted terminal event in each case.

Additional Context

• Internal tracking issue: ENG-815
• Related upstream issue: Race condition between Celery workflow persistence and HITL pause creation #40445 documents one concrete manifestation caused by asynchronous Celery workflow persistence. This issue addresses the broader transaction-ownership problem that also exists with synchronous repositories when one core layer commits and another fails.
• Relevant components:
  • api/core/app/workflow/layers/persistence.py
  • api/core/app/layers/pause_state_persist_layer.py
  • api/repositories/sqlalchemy_api_workflow_run_repository.py
  • Graphon's layer event dispatcher