Skip to content

feat: atomically enqueue Workshop replies #856

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dcellison merged 1 commit into from
Aug 12, 2026
Merged

feat: atomically enqueue Workshop replies #856

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 11 additions & 9 deletions kai-workshop-implementation-map.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -406,7 +406,7 @@ No entry may use an indefinite condition such as "keep for compatibility." A gen
| Telegram `chat_id` used as internal identity, namespace, and routing key | Durable principal, channel, agent, and binding IDs | Private chats, notification-only groups, duplicate updates, and restart routing all resolve correctly through bindings | Confine Telegram IDs to external identity, transport binding, and idempotency records; remove chat-shaped domain keys | Planned |
| `SubprocessPool` keyed by Telegram chat ID | Durable channel/agent session plus run and attempt orchestration | All five harnesses pass continuity, restart, cancellation, and isolation tests through durable identities | Remove chat-key compatibility lookup and move lifecycle ownership behind the orchestrator/runtime contract | Planned |
| Direct backend invocation from Telegram handlers | Transport-neutral command and run services | Telegram and the first Workshop client produce equivalent authorized runs and visible results | Remove handler-owned orchestration; leave authentication, parsing, and rendering in the Telegram adapter | Planned |
| Direct Telegram delivery from handlers, schedules, and webhooks | Durable delivery outbox and Telegram delivery adapter | Delivery outcome events preserve binding identity; retry, crash recovery, ordering, private-chat and notification-group delivery tests pass; live delivery is verified | Register the outbox worker only in an explicit cutover; remove direct Bot API sends from domain paths and delete delivery fallback flags after installed verification | Active (production-unused durable request/lease/attempt/retry/recovery, fragment progress, binding-aware terminal outcome facts, per-binding FIFO claims, and unregistered Telegram adapter/worker; installed direct-chat recovery passed, but atomic production enqueue, notification-group live evidence, startup registration, and authority change remain absent) |
| Direct Telegram delivery from handlers, schedules, and webhooks | Durable delivery outbox and Telegram delivery adapter | Delivery outcome events preserve binding identity; retry, crash recovery, ordering, private-chat and notification-group delivery tests pass; live delivery is verified | Register the outbox worker only in an explicit cutover; remove direct Bot API sends from domain paths and delete delivery fallback flags after installed verification | Active (production-unused durable request/lease/attempt/retry/recovery, fragment progress, binding-aware terminal outcome facts, per-binding FIFO claims, atomic canonical assistant-result plus delivery-request transaction, and unregistered Telegram adapter/worker; installed direct-chat recovery passed, but notification-group live evidence, startup registration, and authority change remain absent) |
| Operator-invoked Workshop delivery qualification CLI | Installed evidence followed by the production delivery worker | A configured direct-chat reply is prepared without sending, survives a service restart, recovers an intentionally abandoned lease, reaches Telegram once through the exact selected delivery, and records a terminal binding-aware outcome | Remove the qualification command and its explicit-claim-only surface after the production worker has equivalent installed restart/recovery evidence and direct delivery is retired | Active (the installed direct-chat qualification gate passed on 2026-08-12; retain until equivalent production-worker evidence exists, while the command remains unregistered and incapable of draining unrelated work) |
| Schedule firing directly into the pool or Telegram | Durable Workshop run creation | Scheduled definitions and executions survive restart and expose run/attempt state without duplicate work | Remove schedule-specific execution path; retain schedules only as authenticated run triggers | Planned |
| GitHub and generic webhook paths that route directly to Telegram or the pool | Canonical integration commands/events plus delivery/run services | Existing GitHub group notifications, generic callers, deduplication, and secret separation pass end-to-end tests | Remove direct routing while retaining verified webhook adapters and supported external contracts | Planned |
Expand Down Expand Up @@ -536,13 +536,15 @@ This decision does not reopen the mechanics already qualified. It separates a su

### 19.1 Next bounded implementation milestone

Add a production-unused application service that atomically creates one canonical assistant reply and its Telegram text delivery request in the same SQLite transaction. The service must:
The production-unused application service atomically creates one canonical assistant reply and its Telegram text delivery request in the same SQLite transaction. It:

- resolve the existing inbound message, agent principal, channel, and canonical Telegram binding without accepting a transport identity from its caller;
- append `message.created` and `delivery.requested`, project both facts, and insert the pending outbox row under one transaction;
- use deterministic identities and idempotency keys so replay returns the same message and delivery without duplicate work;
- roll back both the canonical reply and delivery request when binding resolution, event append, projection, or outbox insertion fails;
- reject ambiguous or missing bindings rather than guessing a destination;
- remain unused by production handlers and leave the worker unregistered.
- resolves the existing inbound message, agent principal, channel, and canonical Telegram binding without accepting a transport identity from its caller;
- appends `message.created` and `delivery.requested`, projects the canonical message, advances the projection checkpoint across both facts, and inserts the pending outbox row under one transaction;
- uses deterministic identities and idempotency keys so replay returns the same message and delivery without duplicate work;
- rolls back both the canonical reply and delivery request when binding resolution, event append, projection, or outbox insertion fails;
- rejects ambiguous or missing bindings rather than guessing a destination;
- remains unused by production handlers and leaves the worker unregistered.

After that contract passes rollback and idempotency tests, separately qualify installed notification-group delivery, define lifecycle ownership, and conduct another explicit cutover review. No temporary delivery feature flag is introduced by this sequence.
Rollback tests cover message projection failure and outbox insertion failure; both leave no assistant event, projected reply, delivery-request event, or pending work. Restart and concurrent-connection retries produce one deterministic message and one delivery, changed content fails closed, and a missing or ambiguous canonical Telegram binding is rejected without accepting a destination from the caller. A pre-existing message-only half-state is not silently repaired.

The service remains production-unused. Next, separately qualify installed notification-group delivery, define lifecycle ownership, and conduct another explicit cutover review. No temporary delivery feature flag is introduced by this sequence.
165 changes: 87 additions & 78 deletions src/kai/workshop/delivery_outbox.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,8 +156,9 @@ def _retry_delay(attempt_number: int) -> timedelta:
class WorkshopDeliveryOutbox:
"""Durable delivery state with lease-based, at-least-once work claims.

No production worker or transport adapter uses this class yet. Existing
direct Telegram delivery remains authoritative until a later cutover.
Only explicit qualification and production-unused application services use
this class. Existing direct Telegram delivery remains authoritative until
a later cutover registers a production worker.
"""

def __init__(
Expand All @@ -173,87 +174,95 @@ async def request_delivery(self, request: DeliveryRequest) -> DeliveryRequestRes
connection = self._store.connection
try:
await connection.execute("BEGIN IMMEDIATE")
resolved = await self._resolve_target(request)
workshop_id, channel_id, author_principal_id, transport = resolved
delivery_id = DeliveryId.derived(
workshop_id,
f"delivery:{request.message_id}:{request.channel_binding_id}:{request.mode}",
)

existing = await self._state_for_identity(
request.message_id,
request.channel_binding_id,
request.mode,
)
if existing is not None:
if (
existing.delivery_id != delivery_id
or existing.channel_binding_id != request.channel_binding_id
or existing.transport != transport
or existing.max_attempts != request.max_attempts
):
raise DeliveryRequestConflictError("Delivery request identity has different semantics")
await connection.commit()
return DeliveryRequestResult(delivery=existing, inserted=False)

occurred_at = request.occurred_at.astimezone(UTC)
event = EventEnvelope.create(
event_id=EventId.derived(
workshop_id,
f"delivery-request-event:{request.message_id}:{request.channel_binding_id}:{request.mode}",
),
event_type=WorkshopEventType.DELIVERY_REQUESTED,
event_version=1,
workshop_id=workshop_id,
aggregate_type="delivery",
aggregate_id=delivery_id,
actor_principal_id=author_principal_id,
occurred_at=occurred_at,
idempotency_key=(
f"workshop-delivery-request:v1:{request.message_id}:{request.channel_binding_id}:{request.mode}"
),
payload={
"message_id": request.message_id,
"channel_id": channel_id,
"channel_binding_id": request.channel_binding_id,
"transport": transport,
"mode": request.mode,
"max_attempts": request.max_attempts,
},
metadata={"source": "delivery_outbox"},
)
appended = await self._store.append_in_transaction(event)
if not appended.inserted:
raise DeliveryRequestConflictError("Delivery request event exists without outbox state")

timestamp = _format_timestamp(occurred_at)
await connection.execute(
"INSERT INTO delivery_outbox "
"(id, workshop_id, channel_id, channel_binding_id, message_id, transport, mode, "
"status, max_attempts, attempt_count, available_at, requested_event_position, "
"created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, 0, ?, ?, ?, ?)",
(
delivery_id,
workshop_id,
channel_id,
request.channel_binding_id,
request.message_id,
transport,
request.mode,
request.max_attempts,
timestamp,
appended.event.position,
timestamp,
timestamp,
),
)
state = await self._state_by_id(delivery_id)
result = await self.request_delivery_in_transaction(request)
await connection.commit()
return DeliveryRequestResult(delivery=state, inserted=True)
return result
except Exception:
await connection.rollback()
raise

async def request_delivery_in_transaction(self, request: DeliveryRequest) -> DeliveryRequestResult:
"""Persist one delivery request without committing an existing transaction."""
connection = self._store.connection
if not connection.in_transaction:
raise RuntimeError("request_delivery_in_transaction requires an active transaction")

resolved = await self._resolve_target(request)
workshop_id, channel_id, author_principal_id, transport = resolved
delivery_id = DeliveryId.derived(
workshop_id,
f"delivery:{request.message_id}:{request.channel_binding_id}:{request.mode}",
)

existing = await self._state_for_identity(
request.message_id,
request.channel_binding_id,
request.mode,
)
if existing is not None:
if (
existing.delivery_id != delivery_id
or existing.channel_binding_id != request.channel_binding_id
or existing.transport != transport
or existing.max_attempts != request.max_attempts
):
raise DeliveryRequestConflictError("Delivery request identity has different semantics")
return DeliveryRequestResult(delivery=existing, inserted=False)

occurred_at = request.occurred_at.astimezone(UTC)
event = EventEnvelope.create(
event_id=EventId.derived(
workshop_id,
f"delivery-request-event:{request.message_id}:{request.channel_binding_id}:{request.mode}",
),
event_type=WorkshopEventType.DELIVERY_REQUESTED,
event_version=1,
workshop_id=workshop_id,
aggregate_type="delivery",
aggregate_id=delivery_id,
actor_principal_id=author_principal_id,
occurred_at=occurred_at,
idempotency_key=(
f"workshop-delivery-request:v1:{request.message_id}:{request.channel_binding_id}:{request.mode}"
),
payload={
"message_id": request.message_id,
"channel_id": channel_id,
"channel_binding_id": request.channel_binding_id,
"transport": transport,
"mode": request.mode,
"max_attempts": request.max_attempts,
},
metadata={"source": "delivery_outbox"},
)
appended = await self._store.append_in_transaction(event)
if not appended.inserted:
raise DeliveryRequestConflictError("Delivery request event exists without outbox state")

timestamp = _format_timestamp(occurred_at)
await connection.execute(
"INSERT INTO delivery_outbox "
"(id, workshop_id, channel_id, channel_binding_id, message_id, transport, mode, "
"status, max_attempts, attempt_count, available_at, requested_event_position, "
"created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, 0, ?, ?, ?, ?)",
(
delivery_id,
workshop_id,
channel_id,
request.channel_binding_id,
request.message_id,
transport,
request.mode,
request.max_attempts,
timestamp,
appended.event.position,
timestamp,
timestamp,
),
)
state = await self._state_by_id(delivery_id)
return DeliveryRequestResult(delivery=state, inserted=True)

async def claim_next(
self,
worker_id: str,
Expand Down
Loading