移动 AI 请求重试如何避免重复执行与令牌重放

import hashlib
import json
import sys
from pathlib import Path

if len(sys.argv) != 3:
    raise SystemExit('usage: check_idempotency.py request.json ledger.json')
request_path = Path(sys.argv[1])
ledger_path = Path(sys.argv[2])
if not request_path.is_file() or not ledger_path.is_file():
    raise SystemExit('request or ledger file is missing')
request = json.loads(request_path.read_text(encoding='utf-8'))
ledger = json.loads(ledger_path.read_text(encoding='utf-8'))
required = {'subject', 'operation', 'operation_id', 'payload', 'idempotency_key'}
if not required <= request.keys():
    raise SystemExit('request is missing required fields')
if not isinstance(request['payload'], dict):
    raise SystemExit('payload must be an object')
if any(isinstance(value, float) for value in request['payload'].values()):
    raise SystemExit('example profile rejects floating-point payload values')
canonical_input = {
    'subject': str(request['subject']),
    'operation': str(request['operation']),
    'operation_id': str(request['operation_id']),
    'payload': request['payload'],
}
canonical = json.dumps(
    canonical_input,
    ensure_ascii=False,
    sort_keys=True,
    separators=(',', ':'),
).encode('utf-8')
request_hash = hashlib.sha256(canonical).hexdigest()
expected_key = hashlib.sha256(
    (canonical_input['subject'] + '|' + canonical_input['operation'] + '|' +
     canonical_input['operation_id']).encode('utf-8')
).hexdigest()
if request['idempotency_key'] != expected_key:
    raise SystemExit('idempotency key does not match the stable operation identity')
scope = canonical_input['subject'] + ':' + canonical_input['operation'] + ':' + expected_key
existing = ledger.get(scope)
if existing is None:
    print(json.dumps({'decision': 'accept_new', 'request_hash': request_hash}, ensure_ascii=False))
elif existing.get('request_hash') != request_hash:
    raise SystemExit('same idempotency key was reused with different parameters')
elif existing.get('status') == 'succeeded':
    print(json.dumps({'decision': 'replay_result', 'result_ref': existing.get('result_ref')}, ensure_ascii=False))
else:
    print(json.dumps({'decision': 'resume_existing', 'status': existing.get('status')}, ensure_ascii=False))