feat(orchestrator): a seam for taking queued executions off the launch path

[yuechao-qin]

What

Adds a generic seam to the orchestrator so a downstream implementation can take a queued
execution off the launch path, and makes UNINITIALIZED mean parked rather than legacy.

Nothing here mentions quotas: Shopify's quota groups are the first user of the seam, and they
live entirely in oasis-backend.

How it works

  sweep picks a QUEUED node
        │
        ├─ inputs ready? cache hit? cancelled?      (unchanged)
        │
        ├─ interceptor.intercept(session, execution)
        │        │
        │        ├─ True  ──► it committed its own status. We return. No container.
        │        └─ False ──► fall through
        │
        └─ create container execution               (unchanged)

Change	File	Why
Sweep selects QUEUED only, not UNINITIALIZED too	orchestrator_sql.py:126	Parking must actually hide a node. Otherwise it is re-selected next tick, redoes the work above the gate and re-parks — and with no ORDER BY the same low-id node is picked every time, spending the whole 2–3/sec sweep budget on one parked execution
QueuedExecutionInterceptor Protocol	orchestrator_sql.py:42	The seam. Returning True means "I own this execution": the implementation sets whatever status it wants and commits, and the orchestrator makes no assumption about which
One keyword-only queued_execution_interceptor=None on OrchestratorService_Sql	orchestrator_sql.py:60	Every existing caller is unaffected
The call site, after the cancel check and before container creation	orchestrator_sql.py:626	The latest point at which the execution is known to be launchable
ContainerExecutionStatus docstring; # Remove → # Parked by an interceptor; not swept	backend_types_sql.py:14	The status is no longer a leftover

Tests

tests/test_orchestrator_sql.py, 9 passing in the file and 473 in the suite.

• a parked (UNINITIALIZED) node is not selected, and is left exactly as found
• a QUEUED node still is — narrowing the selector did not break the sweep
• interceptor returns True: no launch, no container, its status survives
• interceptor returns False, and no interceptor at all: launches as today

Known gap, deliberately left to the caller

An execution parked at UNINITIALIZED no longer sees the run-level TERMINATED flag, because
the sweep no longer looks at it. Cancelling a run whose parked nodes have no live sibling in the
same group therefore leaves those nodes non-terminal until something requeues them.

This is a property of parking, not of this diff — nothing upstream parks today — so the duty
sits with whoever installs an interceptor: they must un-park on the cancel path. Shopify's
implementation does so, and the four cancellation scenarios are covered by its own tests. Say
the word if you would rather the orchestrator kept selecting parked rows that belong to a
terminated run, and I will add it here instead.

[Volv-G]

(AI-assisted)

The Protocol licenses commit() on the orchestrator's own session, but doesn't state the invariant that makes that safe — or what happens if the implementation raises.

The commit() is only safe because of something invisible from here: at the call site (line 626) the session has no pending orchestrator writes. session.rollback() at line 597 clears everything, and the only work between that rollback and intercept is the two extra_data reads for the cancellation check. Nothing pins that invariant. Add a last_processed_at bump or a status-history append between 597 and 626 later, and an implementation's commit() silently flushes orchestrator state that was never meant to be durable — presenting as half-applied execution rows far from this diff.

Separately, raising out of intercept is a much bigger deal than the docstring implies: it propagates to the handler in internal_process_queued_executions_queue (lines 165-185), which marks the execution SYSTEM_ERROR and _mark_all_downstream_executions_as_skipped. So a transient error in downstream code — which by design talks to its own tables — permanently kills the run.

Both belong in this docstring:

    The session is the orchestrator's own and has no uncommitted orchestrator
    changes at call time, so an implementation may commit freely; it must not
    assume that remains true if it holds the session past `intercept`.

    Raising propagates: the orchestrator marks the execution SYSTEM_ERROR and
    skips everything downstream. Swallow transient errors and return False.

Worth a one-line comment above session.rollback() at line 597 too, noting the clean-session invariant is load-bearing for the seam.

[Volv-G]

(AI-assisted)

Overloading UNINITIALIZED as the parked state is the part of this design I'd push back on — I think this wants a dedicated status, which is where the design discussion landed too ("look into creating a new state", left unresolved).

Three reasons the reuse bothers me:

1. The two meanings are opposites. "Never initialized" and "deliberately taken off the launch path by a gate that intends to put it back" are different facts about a node, and after this PR nothing distinguishes them. A reader of a row — or of a dashboard — cannot tell which one they're looking at.

2. Legacy rows change meaning retroactively. 3a2173b API server - Changed the initial status from UNINITIALIZED to QUEUED means this was the initial status for new nodes. Any surviving row at UNINITIALIZED is drained by the sweep today and becomes permanently invisible after this change. Probably zero rows in practice — SELECT COUNT(*) FROM execution_node WHERE container_execution_status = 'UNINITIALIZED' settles it — but with a distinct state the question wouldn't arise at all.

3. It's user-visible. A quota-parked node will render as UNINITIALIZED in the UI, which is meaningless to the person whose pipeline it is. That label mapping does not live in this repo, so nothing here can soften it.

The honest counter-argument, which I don't think is fatal: container_execution_status compiles to a MySQL ENUM(...), so a new member is ALTER TABLE execution_node MODIFY COLUMN … on a very hot table — and at least one downstream consumer builds its schema with metadata.create_all() and no migration framework, so it would need a hand-written migration there too. That is real cost. But it is a one-time cost paid at the bottom of a six-PR stack, and it only gets more expensive once parked rows exist in production and the ambiguity is load-bearing.

So: either add the state now, or — if the cost wins — please record the decision in the enum comment (backend_types_sql.py:17-23) as an explicit, priced trade-off rather than a reuse of a spare member, so the next person to touch this knows it was chosen and not inherited.

[Volv-G]

(AI-assisted)

The seam's contract is positional, and no test pins the position — which matters more than usual because a consumer vendoring this repo at a pinned SHA cannot detect it moving.

The docstring's promise is where this runs: "once the execution is known to be launchable — inputs present, not conditionally skipped, no cache hit, not cancelled." TestQueuedExecutionInterceptor covers True / False / no-interceptor, all on a plain launchable single-task pipeline. Every one of those would still pass if a later refactor moved this block above the cache lookup or the cancel check.

The failure that allows is quiet. Move it above the cache lookup and a cache hit now consults the interceptor — an implementation that gates on capacity would charge a slot for work that is about to be satisfied from cache, and could park a node waiting for a slot it never needed. Move it above the cancel check and cancelled executions do the same. Neither errors; both surface much later as "the gate is mysteriously saturated."

Worth pinning here specifically because this is a library seam. A downstream implementation typically vendors this repo at a fixed revision, so its own CI compiles against a frozen copy — a move here cannot fail any test it owns until someone advances the pin, at which point the behaviour change looks like it came from the bump rather than the refactor. These tests are the only place the guarantee can be stated where it runs on every push.

Cheap, given _make_orchestrator and _StubInterceptor already exist — each case is one assertion that intercept was not called:

# interceptor.calls == [] for each of:
#  1. missing input artifact      -> WAITING_FOR_UPSTREAM
#  2. cache hit                   -> reuses cached execution
#  3. desired_state = TERMINATED  -> CANCELLED   (node-level and run-level)
#  4. is_enabled: false           -> SKIPPED

If only one is worth it, take the cache-hit case: it is the one whose failure mode is silent. The cancel case at least ends in a terminal status somebody notices.

[Volv-G]

(AI-assisted)

An implementation that returns True but leaves the row QUEUED silently re-creates the single-row sweep starvation this PR narrows the query to remove — and nothing errors or logs when it happens.

The contract is prose only: "it sets whatever status it wants and commits." If an implementation returns True but leaves the status as QUEUED, or sets a status and forgets to commit, then:

• this method returns without committing;
• process_each_queue_once exits its with self._session_factory() as session: block, which rolls back;
• the row is still QUEUED;
• the next sweep runs SELECT ... WHERE status = QUEUED LIMIT 1 with no ORDER BY (lines 141-150 — .order_by(...) is still commented out), so it deterministically re-selects the same low-id row;
• the orchestrator spends its whole sweep budget re-parsing the task spec, re-deriving the cache key and re-querying candidates for one execution, indefinitely.

That is the exact failure your own description gives as the reason for narrowing the selector, reintroduced through the seam — but silent this time, because a buggy implementation looks identical to a working one from here.

Cheap fix — make the violation loud rather than fatal:

if self._queued_execution_interceptor.intercept(session=session, execution=execution):
    session.refresh(execution)
    if execution.container_execution_status == bts.ContainerExecutionStatus.QUEUED:
        _logger.error(
            "Interceptor claimed execution %s but left it QUEUED; it will be re-swept.",
            execution.id,
        )
    return

Stronger alternative: make the contract enforceable by construction — have intercept return ContainerExecutionStatus | None, where None means "not mine, carry on" and a status means "park it here", and let the orchestrator do the write and the commit. That removes an implementation's ability to leave the row unchanged or the session half-committed, and it also settles the question of what session state the implementation is allowed to assume. If the boolean form was a deliberate choice — to let an implementation pick a status the orchestrator has no opinion about — a sentence saying so would help; it is currently the load-bearing assumption of the whole seam.

[Volv-G]

(AI-assisted)

Building on the docstring comment above (#discussion_r3919337398) rather than repeating it — that one asks you to document that raising here is fatal. This is the argument that the call site should instead make it survivable.

To restate only the conclusion: an exception out of intercept propagates to the handler in internal_process_queued_executions_queue (lines 165-185), which commits SYSTEM_ERROR on the node and then skips every downstream execution. Unrecoverable.

The reason that is worth more than a docstring: this is an optional seam whose implementation is, by construction, foreign code talking to storage the orchestrator knows nothing about. As written, installing an admission gate silently couples every pipeline's survival to that gate's availability — a lock-wait timeout or a deploy-time blip during a sweep is enough to kill a run outright. That is a much stronger commitment than an opt-in hook implies, and an implementer reading the signature has no way to infer it.

Fail-open at the call site:

if self._queued_execution_interceptor is not None:
    try:
        if self._queued_execution_interceptor.intercept(session=session, execution=execution):
            return
    except Exception as exc:
        _logger.exception("Queued execution interceptor raised; launching anyway.")
        bugsnag_instrumentation.notify(exception=exc)
        session.rollback()

A broken gate then degrades to "no gating" rather than "no pipelines". The cost is a bounded, self-correcting overshoot while the implementation is down; the benefit is that an optional component cannot take the orchestrator's core job with it.

If you would rather fail closed, one caveat worth stating explicitly: swallowing the exception and falling through without launching leaves the row QUEUED, which re-enters the sweep next tick and re-creates the single-row starvation described in the adjacent comment. Failing closed safely would mean parking the execution — and the orchestrator cannot do that, because the whole premise of this seam is that it has no opinion about which status a claimed execution should hold. That asymmetry is itself an argument for fail-open.

[Volv-G]

(AI-assisted)

Requesting changes — with a caveat about what that does and does not mean here.

I did not find a functional defect in this diff. The seam is placed correctly: it is on the single path before container creation, both routes into QUEUED funnel through it, UNINITIALIZED really is excluded from the sweep now, and the default is a genuine no-op for every existing caller. The tests that exist are good ones. If this were internal code I would approve it.

What holds it up is that it is not internal code: this PR defines a public extension point, and the contract around it is currently prose in a docstring rather than something the code states or enforces. Once implementations exist against it, that contract is expensive to change. The threads worth resolving before then:

Contract, unstated or unenforced

• The Protocol licenses commit() on the orchestrator's session but never states the invariant that makes it safe, nor that raising is fatal to the run.
• return True without actually changing and committing the status silently re-creates the single-row sweep starvation this PR narrows the selector to remove. Nothing errors, nothing logs.
• An exception out of intercept ends the run and skips the whole downstream subgraph — a very strong coupling for an optional gate.

A design question rather than a defect

• Overloading UNINITIALIZED as the parked state. Reasonable given the cost of a new enum member, but I think it deserves an explicit decision recorded in the enum comment rather than being inherited as reuse of a spare value.

Testing

• Nothing pins the seam's position, which is the whole of its contract. A later refactor can move it above the cache or cancel check with every test still green.

None of these requires much code — most are a docstring paragraph, a five-line guard, and a few assertions. Several could reasonably be resolved by you replying "deliberate, here's why", and I would clear the block on that basis. I have also answered your closing question about whether the orchestrator should keep selecting parked rows belonging to a terminated run, and my answer is no — but that thread is worth having before this merges, because it is the one place where I think the "known gap" is better closed here than delegated to implementations.

Happy to re-review quickly on any of these.