What causes "unknown claim token" in affaan-m/ECC?

Verify the token exists: SELECT * FROM obligation_delivery_claims WHERE token=? before calling

affaan-m/ECC · error · ClaimError

unknown claim token

Error message

unknown claim token

What it means

After validating the token format, the library looks it up in obligation_delivery_claims. If no row matches, the token does not correspond to any claim this database issued, so the operation is refused. This keeps dispatch/cancel/unknown-marking strictly tied to claims created by claim() in the same database.

Solutions

Verify the token exists: SELECT * FROM obligation_delivery_claims WHERE token=? before calling
Confirm you are connected to the same database that issued the token
Use the full, exact token string returned by claim() (64 hex chars from secrets.token_hex(32)) — copy it, don't retype it
If the claim row is gone, re-create it via the approval + claim flow; there is no way to resurrect a deleted token

Example fix

// before
begin_dispatch(db, token_from_other_env, now=ts)  # unknown claim token

// after
exists = db.execute('SELECT 1 FROM obligation_delivery_claims WHERE token=?', (token,)).fetchone()
if not exists:
    raise AppError('token not found in this database; check environment')
begin_dispatch(db, token, now=ts)

Defensive patterns

Strategy: try-catch

Validate before calling

def token_known(db, token) -> bool:
    return db.execute('SELECT 1 FROM obligation_delivery_claims WHERE token=?',
                      (token,)).fetchone() is not None

Try / catch

try:
    cancel(db, token, now=ts)
except ClaimError as e:
    if 'unknown claim token' in str(e):
        log.warning('token %s... not found; wrong DB or expired claim', token[:8])
        # safe to treat as already-resolved for cancel; escalate for dispatch
    else:
        raise

Prevention

Store tokens only in the DB that issued them; tag records with the DB/environment
Copy the full 64-char token verbatim; never retype or truncate
Avoid cleanup jobs that delete obligation_delivery_claims rows still referenced by workers
In multi-DB setups, look up the issuing database before dispatch/cancel operations

When it happens

Trigger: Passing a token from a different database/environment; a typo'd or truncated token; the claim row was deleted; calling cancel/mark_unknown twice after the row was purged by a cleanup job.

Common situations: Replaying tokens recorded in logs from another environment (staging vs prod); multi-worker setups where each worker uses its own DB file; retyping a token by hand from a console message; tests reusing fixtures across DBs.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

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

Appendix: source

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

    try:
        digest = hashlib.sha256(row['draft_text'].encode('utf-8')).hexdigest()
    except (AttributeError, UnicodeError) as error:
        raise ClaimError('approved text must be valid UTF-8 text') from error
    stored_digest = row['draft_sha256']
    if (not isinstance(stored_digest, str) or len(stored_digest) != 64
            or any(character not in '0123456789abcdef' for character in stored_digest)):
        raise ClaimError('approved hash must be lowercase SHA-256 hexadecimal')
    if not secrets.compare_digest(digest, stored_digest):
        raise ClaimError('approved text hash does not match')
    return dict(row)


def _claim_row(db, token):
    if not isinstance(token, str) or not token:
        raise ClaimError('a claim token is required')
    row = db.execute('SELECT * FROM obligation_delivery_claims WHERE token=?', (token,)).fetchone()
    if row is None:
        raise ClaimError('unknown claim token')
    return row


def claim(db, obligation_id, decision_id, *, now):
    """Reserve one already-authorized decision; return only a random claim token."""
    with _transaction(db, now):
        _snapshot(db, obligation_id, decision_id)
        token = secrets.token_hex(32)
        db.execute('''INSERT INTO obligation_delivery_claims
            (obligation_id,decision_id,token,state,created_ts,updated_ts)
            VALUES (?,?,?,'claimed',?,?)''', (obligation_id, decision_id, token, now, now))
    return token


def begin_dispatch(db, token, *, now):
    """Return bound payload once, only after dispatching state has committed.

    A crash after this boundary is uncertain even if transport has not started.

View on GitHub (pinned to 8321021c54)