Offsets for a batch the sink refused are no longer committed on shutdown

[turbolytics]

A stateful pipeline loses rows on main whenever its sink refuses a batch. The process exits with the sink's error, but the offsets for that batch become durable, and a restart resumes past rows nothing wrote.

The bug

• Positions advance when a message reaches the handler, before its batch is flushed.
• A failed flush rolls back the batch's state writes, but leaves those positions in memory.
• run syncs state twice after the loop returns, before and after the table managers' final poll. Those syncs save the positions held in memory.
• The DuckDB driver ignores the context on commit, rollback and statements, so nothing about a failed run stops the commit.

A second path loses rows the same way. A cancel that lands exactly as a batch fills returned a clean stop without writing the batch, exited 0, and the shutdown committed its positions.

Found by an adversarial review of #273. It predates #273.

The fix

• The turbine keeps a copy of the positions its last successful commit made durable. That is the state transaction's offsets, or the positions the source accepted for a pipeline with no state database.
• Any error out of ConsumeLoop resets the in-memory positions to that copy, so a later commit saves nothing new.
• A cancel at a batch boundary drains the batch, the same as the cancel branch of the loop's select.
• Marks.Reset is the one operation allowed to move a position backwards. It reuses its maps, so recording the copy on every commit allocates nothing new per commit.

The harness

The new invariant pipeline.shutdown.commits_only_delivered is enforced. Every earlier pipeline check stopped when the loop returned, so none of them could see this bug.

• The recorder now tracks positions: what the sink acknowledged, what a state commit made durable, and what the source was told to commit.
• The check runs every trigger, clean and with a failed flush, then run's two shutdown syncs, and fails if any durable position covers a message the sink did not acknowledge.
• A fifth trigger, boundary-cancel, lands the cancel as the batch fills. Every pipeline invariant now runs on it.
• A self-test proves the comparison fails on a position past delivery, at the state database or the source.

Against main's turbine, the new harness fails pipeline.shutdown.commits_only_delivered on a failed flush, and fails after_flush, nothing_on_failure and state.with_offsets on the boundary-cancel trigger.

Verified

• go test -short -race ./..., go vet, gofmt and pytest tests/tooling pass.
• Four core tests with a real file-backed DuckDB, reopening the file to read the durable offsets. All four fail on main.
• End to end, same binary build from each branch: a stateful Kafka to ClickHouse pipeline delivers batch 1, the ClickHouse table is renamed so batch 2 is refused, the process exits, the table is restored and the pipeline restarts.

	Durable offset after the refusal	Rows after the restart
main	9	5, users u6 to u10 lost
this branch	4	10, batch 2 replayed

Coverage status files follow from this PR's CI report artifacts.