affaan-m/ECC · error · ClaimError

a claim token is required

Error message

a claim token is required

What it means

The token parameter identifying a delivery claim must be a non-empty string; begin_dispatch, cancel, mark_unknown, and _finish all validate this before any database lookup. It prevents nonsense or None values from reaching parameterized queries and keeps the claim-identity contract explicit.

Solutions

  1. Pass the exact token string returned by claim(): token = claim(...); begin_dispatch(db, token, now=ts)
  2. Guard before calling: if not token or not isinstance(token, str): handle the missing-token case in your code
  3. Persist the token durably (database/job record) before dispatching so restarts can recover it
  4. If claim() failed, do not fabricate a token; rerun the approval/claim flow

Example fix

// before
token = None  # claim() result lost
begin_dispatch(db, token, now=ts)  # ClaimError

// after
token = claim(db, oid, did, now=ts)
assert isinstance(token, str) and token
begin_dispatch(db, token, now=ts)
Defensive patterns

Strategy: type-guard

Validate before calling

def require_token(token):
    if not isinstance(token, str) or not token:
        raise ValueError('a non-empty claim token string is required')

Type guard

def is_claim_token(v) -> bool:
    return isinstance(v, str) and len(v) > 0

Try / catch

try:
    payload = begin_dispatch(db, token, now=ts)
except ClaimError as e:
    if 'a claim token is required' in str(e):
        raise AppError('claim token missing; recover it from your job record or re-claim') from e
    raise

Prevention

When it happens

Trigger: begin_dispatch(db, None, now=ts); cancel(db, '', now=ts); passing a bytes token (b'abc'), an int ID, or a token variable that was never assigned because claim() failed earlier.

Common situations: Losing the return value of claim() and passing the obligation ID instead; deserializing tokens from JSON/config where they became None; storing tokens in a column or config that stripped/emptied them.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — 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/2e3f3e3627afecab. Report an issue: GitHub.

Appendix: source

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

        WHERE obligation_id=? AND decision_id=?''', (obligation_id, decision_id)).fetchone()
    if row is None:
        raise ClaimError('a current bound approved draft is required')
    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):

View on GitHub (pinned to 8321021c54)