tegwick
ff59199c3b
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029c-c003-7283-acc9-930b434e6cae
68 lines
3.3 KiB
Python
68 lines
3.3 KiB
Python
"""Leased local integration outbox. Owner composition supplies bounded transports.
|
|
|
|
Unknown adapters leave rows pending; hub resolutions remain pending while the
|
|
write-back flag is off. Dispatch failure never rolls back a human disposition.
|
|
Audit/notification receivers must deduplicate using the stable delivery id.
|
|
"""
|
|
import json
|
|
import time
|
|
import uuid
|
|
|
|
from .hub_writeback import Answer, enabled, write_back
|
|
from .audit import OutboxWorker
|
|
|
|
|
|
class IntakeDeliveryWorker:
|
|
def __init__(self, store, *, audit=None, notify=None, hub_transport=None,
|
|
hub_base_url='', env=None, clock=time.time):
|
|
self.store, self.audit, self.notify = store, audit, notify
|
|
self.hub_transport, self.hub_base_url = hub_transport, hub_base_url
|
|
self.env, self.clock = env, clock
|
|
|
|
def run_once(self, limit=10):
|
|
if type(limit) is not int or not 1 <= limit <= 100:
|
|
raise ValueError('delivery limit must be 1..100')
|
|
delivered = 0
|
|
if self.audit:
|
|
self.store.stage_intake_audit()
|
|
OutboxWorker(self.store, self.audit).run_once(limit=limit)
|
|
delivered += self.store.settle_intake_audit()
|
|
kinds = []
|
|
if self.notify:
|
|
kinds.append('notification')
|
|
if self.hub_transport and enabled(self.env):
|
|
kinds.append('hub-resolution')
|
|
if not kinds:
|
|
return delivered
|
|
for _ in range(limit):
|
|
now, lease = self.clock(), str(uuid.uuid4())
|
|
with self.store._transaction() as db:
|
|
placeholders = ','.join('?' for kind in kinds)
|
|
row = db.execute(f"SELECT * FROM intake_deliveries WHERE kind IN ({placeholders}) "
|
|
"AND state IN ('pending','in_flight') AND next_attempt<=? "
|
|
"AND (lease_until IS NULL OR lease_until<=?) ORDER BY rowid LIMIT 1", (*kinds, now, now)).fetchone()
|
|
if row is None:
|
|
break
|
|
db.execute("UPDATE intake_deliveries SET state='in_flight',lease=?,lease_until=?,attempts=attempts+1 WHERE id=?",
|
|
(lease, now + 30, row['id']))
|
|
state = 'pending'
|
|
try:
|
|
payload = json.loads(row['payload'])
|
|
if row['kind'] == 'notification':
|
|
# notify(id, metadata) returns only after a deduplicated receiver receipt.
|
|
if self.notify(row['id'], payload) is True:
|
|
state = 'delivered'
|
|
else:
|
|
result = write_back(Answer(**payload['answer']), payload['serves'],
|
|
base_url=self.hub_base_url, transport=self.hub_transport, env=self.env)
|
|
if result.status in ('resolved', 'already_resolved'):
|
|
state = 'delivered'
|
|
elif result.status == 'conflict':
|
|
state = 'blocked'
|
|
except Exception:
|
|
pass # Never persist upstream bodies, tokens, or exception text.
|
|
with self.store._transaction() as db:
|
|
db.execute("UPDATE intake_deliveries SET state=?,lease=NULL,lease_until=NULL,next_attempt=? "
|
|
"WHERE id=? AND lease=?", (state, now + min(3600, 2 ** min(row['attempts'] + 1, 11)), row['id'], lease))
|
|
delivered += state == 'delivered'
|
|
return delivered
|