Extract _create_in_transaction so a caller can own the pipeline-run transaction

[yuechao-qin]

What this changes

PipelineRunsApiService_Sql.create() is split in two. A new private
_create_in_transaction() does the work — build the execution-node tree, insert the
PipelineRun, flush, mirror the system annotations — and returns the bts.PipelineRun.
create() keeps its signature, its return type and its transaction, and is now a thin
wrapper around it.

before                              after
──────                              ─────
create()                            create()
  with session.begin():               with session.begin():
    ...build and insert...              _create_in_transaction()  ← the work, no commit
    session.commit()   ← redundant    session.refresh(); return
  session.refresh(); return

Purely additive for existing callers: same signature, same transaction ownership, same
response object. The one behavioural tidy-up is the session.commit() that sat inside
with session.begin(): — the block already commits on exit, so it is gone.

Why

Starting a pipeline run is a same-database insert, not a remote call, so it can share a
caller's transaction. Today it cannot: create() calls session.begin(), and SQLAlchemy
refuses that on a session that already has a transaction open.

That blocks any caller that has to write a run atomically with its own rows. Ours is an
event-driven trigger: it claims a "this cycle has fired" fence row and starts a run, and the
two must commit or roll back together — otherwise a crash between them either fires the same
trigger twice or drops the run silently. With _create_in_transaction() the caller keeps one
transaction around both.

Private on purpose

The underscore is deliberate: this is an internal seam, not a new public API, so nothing here
is promised to stay. Happy to make it public if other consumers want the same guarantee —
that is a question for reviewers, and it changes nothing about the code.

Tests

TestCreateInTransaction pins the contract, since a private method has nothing else
protecting it:

• the run is flushed, so the caller can use its ID inside the transaction;
• an outer rollback leaves zero pipeline_run and zero execution_node rows;
• a caller holding with session.begin(): — the same shape create() uses — gets a durable
  run once the block exits.

The last two are mutation-checked: putting session.commit() back inside
_create_in_transaction turns both red, so the "never commits" rule has two independent
guards. Existing coverage of create() is unchanged — 50 tests across 8 classes reach it, all
against a real SQLite engine with no mocks, and they too go red if create() stops committing.
Full suite: 471 passed.

Assisted-By: devx/11fb0c55-5ff5-402f-abc7-1a722b0b09cb

[yuechao-qin]

• Extract _create_in_transaction so a caller can own the pipeline-run transaction #344 👈 (View in Graphite)
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.