affaan-m/ECC · error · ClaimError

a top-level committed transaction is required

Error message

a top-level committed transaction is required

What it means

This library refuses to run a claim operation when the SQLite connection already has an open transaction (db.in_transaction is true). It exists so that dispatch permission is never granted inside a transaction whose commit belongs to an outer caller: a nested or caller-controlled commit could leave the claim rolled back after the caller already acted on it. Every public API (claim, begin_dispatch, cancel, mark_unknown, _finish) requires a clean, top-level transaction it owns.

Solutions

1. Commit or rollback the open transaction on the connection before calling any function in this module
2. Create the connection with isolation_level=None (autocommit) so this library's BEGIN IMMEDIATE is top-level, as its connect() helper does
3. Do not wrap claim()/begin_dispatch() calls inside your own transaction; perform outer work in a separate transaction or separate connection
4. Check db.in_transaction yourself before calling and fail fast with a clear application error

Example fix

// before
db = sqlite3.connect('app.db')
db.execute('BEGIN')
claim(db, obligation_id, decision_id, now=ts)  # ClaimError

// after
db = sqlite3.connect('app.db', isolation_level=None)
db.commit()  # or rollback any pending work first
claim(db, obligation_id, decision_id, now=ts)

def ensure_top_level(db):
    if db.in_transaction:
        raise RuntimeError('commit or rollback the open transaction before calling the claims API')

def is_clean_connection(db) -> bool:
    return not getattr(db, 'in_transaction', False)

try:
    token = claim(db, oid, did, now=ts)
except ClaimError as e:
    if 'top-level committed transaction' in str(e):
        db.rollback()  # or commit, per your semantics
        token = claim(db, oid, did, now=ts)
    else:
        raise

When it happens

Trigger: Calling claim/begin_dispatch/cancel/mark_unknown/_finish while the same connection has a pending BEGIN started by your own code, by a with db: block, or by a sqlite3 implicit transaction (isolation_level not None and uncommitted DML).

Common situations: Wrapping a claim() call inside application-level DB transaction code; reusing a connection shared with an ORM that keeps transactions open; forgetting to commit/rollback a prior statement because isolation_level was left at the default (empty-string autocommit deferral).

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• claim transaction failed; no permission granted
• a current bound approved draft is required
• approved hash must be lowercase SHA-256 hexadecimal
• approved text must be valid UTF-8 text
• candidate alias collision with different immutable target

AI-assisted analysis of affaan-m/ECC@8321021c54 (2026-09-16). Data as JSON: /api/errors/51098f63546e44c4. Report an issue: GitHub.

Appendix: source

Thrown at skills/operator-approval-loop/references/approval_claims.py:34

class ClaimError(Exception):
    """No dispatch permission or state transition was granted."""


def connect(path):
    """Open an existing caller-selected database; never apply schema/migrations."""
    uri = Path(path).resolve().as_uri() + '?mode=rw'
    db = sqlite3.connect(uri, uri=True, isolation_level=None, timeout=5)
    db.row_factory = sqlite3.Row
    db.execute('PRAGMA foreign_keys=ON')
    db.execute('PRAGMA recursive_triggers=ON')
    return db


@contextmanager
def _transaction(db, now):
    # Never return permission whose commit belongs to an outer caller transaction.
    if db.in_transaction:
        raise ClaimError('a top-level committed transaction is required')
    if type(now) is not int or now < 0:
        raise ClaimError('now must be a nonnegative integer')
    if any(db.execute(f'PRAGMA {name}').fetchone()[0] != 1
           for name in ('foreign_keys', 'recursive_triggers')):
        raise ClaimError('required SQLite guards are disabled')
    try:
        db.execute('BEGIN IMMEDIATE')
        yield
        db.commit()
    except BaseException as error:
        db.rollback()
        if isinstance(error, sqlite3.Error):
            raise ClaimError('claim transaction failed; no permission granted') from error
        raise


def _snapshot(db, obligation_id, decision_id):
    row = db.execute('''SELECT * FROM approval_bound_drafts

View on GitHub (pinned to 8321021c54)