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
- Pass the exact token string returned by claim(): token = claim(...); begin_dispatch(db, token, now=ts)
- Guard before calling: if not token or not isinstance(token, str): handle the missing-token case in your code
- Persist the token durably (database/job record) before dispatching so restarts can recover it
- 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
- Always capture the return value of claim() and persist it before dispatching
- Validate tokens at system boundaries (job queues, config) where they may become None
- Never substitute obligation IDs or decision IDs for tokens
- Type-annotate token parameters as str and run mypy
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
- An install component ID is required
- claim requires an owner.
- Missing issue number.
- Missing --repo <owner/repo>.
- move requires a work item id.
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)