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.

[Volv-G]

(AI-assisted)

Approving — this is a clean extraction and the transaction-ownership story holds up. create() keeps its signature, transaction and return type, and dropping the inner session.commit() really is a no-op: SessionTransaction.__exit__ short-circuits once session._transaction is None, so the old code already committed exactly once at the explicit commit.

Approval with follow-ups, not a blocker. Four inline comments. None of them is reachable in tangle today — every current caller goes through create(), which wraps the call in with session.begin():. They become reachable the moment the oasis-backend Trigger sink owns the transaction, which is the entire point of the change, so they're worth a look before the submodule bump rather than after.

The two I'd actually act on:

• The internal session.flush() flushes rows the caller staged too, so a caller's own constraint violation surfaces from inside _create_in_transaction. That matters for the sink's UNIQUE (subscription_id, cycle) fence, where losing the race is a normal, expected IntegrityError the sink means to catch itself.
• Partially built rows survive an exception. Reproduced against this head: 1 orphan execution_node, 0 pipeline_run.

The other two are a docstring precondition that isn't enforced (and fails silently when violated), and a test gap around the specific guarantee the method exists to provide. All findings below were reproduced by executing them at 04b8c744, not inferred.

Deliberately not raised: the public-vs-private naming question, since you already flagged it in the description and oasis-backend#562's contract test fails loudly if the symbol moves.

[Volv-G]

(AI-assisted)

This flush isn't scoped to this method — it flushes whatever the caller staged, too.

session.flush() is a whole-session flush. Verified at this head: a caller that stages a row and doesn't flush finds it written to the DB by this call, and if that row violates a constraint the IntegrityError is raised from inside _create_in_transaction:

caller stages its own row, does NOT flush   -> session.new = 1
_create_in_transaction(...)                 -> caller's row is now in the DB
and when the caller's staged row collides:
                                            -> IntegrityError raised from INSIDE this method

That matters for the trigger sink in oasis-backend#562. The fence there is a UNIQUE (subscription_id, cycle) constraint and losing the race is supposed to raise IntegrityError — it's the normal, expected no-op path. If the sink stages the fence row and then calls this, the race-loss exception no longer comes out of the sink's own insert:

• a sink that wraps only its own session.add/flush in try/except IntegrityError won't catch the race loss at all;
• a sink that wraps this call can't distinguish "I lost the fence race" (expected, ignore) from "creating the pipeline run failed" (real, alert) without inspecting the constraint name.

I notice the TODO on the line above already wants this flush gone. Until then, the docstring seems like the right place to state its blast radius, since the method explicitly invites the stage-then-call pattern ("must be written atomically with the caller's own rows"):

Flushes the whole session, so rows the caller staged but has not flushed
are written too; a constraint violation on the caller's own rows will
therefore surface from this call. Flush before calling if you need to tell
your own conflicts apart from run-creation failures.

The sink-side fix is a one-liner — session.flush() right after staging the fence row — but only if this behaviour is documented, otherwise there's nothing to tell the sink author they need it.

---

[nit] While here: pipeline_run.id isn't server-generated. backend_types_sql.py:134 declares it insert_default=generate_unique_id, a Python-side callable, and IdType is a str — no autoincrement, no server_default, no sequence. The flush is still genuinely required (verified: id is None before, '01a04b0f…' after), so only the parenthetical is wrong. Worth correcting to (Python-side insert_default) precisely because this comment is the stated justification for a flush that turns out to have wider reach than advertised.

[Volv-G]

(AI-assisted)

If this raises, the caller is left holding partially-built rows.

_recursively_create_all_executions_and_artifacts does its first session.add(...) at line 1655, before all 10 of its subsequent raise ApiServiceError sites (1677, 1709, 1779, 1791, 1831, 1863, 1883, 1888, 1904, 1918). create() was protected by with session.begin():, which rolls back on exception; _create_in_transaction has no such boundary.

Reproduced at this head with a root component that has a required input and no argument supplied:

create()                 -> ApiServiceError, ExecutionNode rows = 0
_create_in_transaction() -> ApiServiceError; caller commits its own rows,
                            ExecutionNode rows = 1, PipelineRun rows = 0

An orphan execution_node with no pipeline_run referencing it. A caller that catches the error and commits its own work anyway — which is a reasonable thing for the trigger sink to want, since it may well wish to keep the fence row that records the attempt — silently persists junk.

Two ways out: (1) document that any exception from this method obliges the caller to roll back the entire transaction, or (2) wrap the body in with session.begin_nested(): so a failure unwinds only this method's writes and leaves the caller's rows intact. Option 2 looks worth the one SAVEPOINT, since batching with caller-owned rows is the whole reason this method exists — and it makes the two goals compatible instead of mutually exclusive.

[Volv-G]

(AI-assisted)

This precondition isn't enforced, and violating it fails silently.

Thanks to autobegin, calling this on a session with no open transaction appears to work: it autobegins, hands back a fully populated pipeline_run.id, and then discards everything when the session closes without a commit. Verified at this head:

in_transaction before: False
in_transaction after:  True   (autobegin)    id handed out: True
-> session closed without commit
persisted PipelineRun rows: 0

A caller gets a real-looking run ID for a run that will never exist. That's the same silent-drop failure mode this PR exists to remove, just relocated from the service into the caller — and it's the sort of mistake a new caller makes exactly once, expensively, in production.

Three lines make it loud:

if not session.in_transaction():
    raise ApiServiceError(
        "_create_in_transaction requires the caller to have an open transaction; "
        "use `create` if you just want a run created."
    )

I checked this is purely additive against every current call path: create() opens with session.begin(): before calling, all three new tests call session.begin() first, and any caller that has already staged a row has autobegun. in_transaction() is True in every legitimate case.

[Volv-G]

(AI-assisted)

These pin "the run rolls back" but not "the run and my rows roll back together" — which is the guarantee the method exists to provide.

The three new tests are good: real engine, and test_the_callers_commit_makes_the_run_durable proves the StaticPool in-memory DB is genuinely shared across sessions, so this rollback test isn't vacuous. But every test here writes only the run, so the atomicity claim in the docstring — a run written atomically with the caller's own rows — has no coverage.

def test_the_run_and_the_callers_own_rows_share_a_fate(self, session_factory, service):
    with session_factory() as session:
        session.begin()
        session.add(bts.PipelineRunAnnotation(pipeline_run_id="caller-row", key="k", value="v"))
        service._create_in_transaction(session, root_task=_make_task_spec("atomic"))
        session.rollback()
    with session_factory() as session:
        assert _count_rows(session=session, table=bts.PipelineRun) == 0
        assert _count_rows(session=session, table=bts.PipelineRunAnnotation) == 0

Two smaller gaps:

• This test counts PipelineRun and ExecutionNode but not PipelineRunAnnotation, and those rows are added by _mirror_system_annotations after the flush — exactly the writes a partial-commit regression would leak. One more assert _count_rows(..., table=bts.PipelineRunAnnotation) == 0 closes it.
• Nothing covers two calls in one transaction. I verified it works today (two distinct IDs, 2 runs and 4 annotation rows committed together), but nothing holds that behaviour still, and a trigger firing could plausibly start more than one run per cycle.