From c8d4010c9ba94a1f0bb2cd6c75cea5e704b5b9a9 Mon Sep 17 00:00:00 2001 From: Daniel Ellison Date: Wed, 12 Aug 2026 07:24:27 -0400 Subject: [PATCH] feat: atomically enqueue Workshop replies --- kai-workshop-implementation-map.md | 20 ++- src/kai/workshop/delivery_outbox.py | 165 +++++++++-------- src/kai/workshop/outbound.py | 160 +++++++++++++---- src/kai/workshop/store.py | 83 +++++---- tests/test_workshop_delivery_outbox.py | 8 + tests/test_workshop_foundation.py | 20 +++ tests/test_workshop_outbound.py | 236 ++++++++++++++++++++++++- 7 files changed, 530 insertions(+), 162 deletions(-) diff --git a/kai-workshop-implementation-map.md b/kai-workshop-implementation-map.md index df2072c9..37ce34c6 100644 --- a/kai-workshop-implementation-map.md +++ b/kai-workshop-implementation-map.md @@ -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 | @@ -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. diff --git a/src/kai/workshop/delivery_outbox.py b/src/kai/workshop/delivery_outbox.py index 84b75d48..a1dee55b 100644 --- a/src/kai/workshop/delivery_outbox.py +++ b/src/kai/workshop/delivery_outbox.py @@ -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__( @@ -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, diff --git a/src/kai/workshop/outbound.py b/src/kai/workshop/outbound.py index c940005c..37722438 100644 --- a/src/kai/workshop/outbound.py +++ b/src/kai/workshop/outbound.py @@ -1,4 +1,4 @@ -"""Assistant results and transport-delivery observations for Kai Workshop.""" +"""Assistant results, atomic delivery requests, and delivery observations.""" from __future__ import annotations @@ -6,7 +6,9 @@ from dataclasses import dataclass from datetime import datetime +from kai.workshop.delivery_outbox import DeliveryRequest, DeliveryRequestResult, WorkshopDeliveryOutbox from kai.workshop.domain import ( + ChannelBindingId, ChannelId, DeliveryId, EventEnvelope, @@ -26,6 +28,14 @@ class OutboundMessageNotFoundError(LookupError): """A referenced canonical message or its unique agent binding was not found.""" +class OutboundDeliveryBindingError(LookupError): + """The reply channel does not have exactly one canonical Telegram binding.""" + + +class OutboundDeliveryStateConflictError(RuntimeError): + """Only one half of an atomic outbound message and delivery already exists.""" + + @dataclass(frozen=True, slots=True) class OutboundMessage: in_reply_to_message_id: MessageId @@ -68,6 +78,12 @@ class _ResolvedOutbound: agent_principal_id: PrincipalId +@dataclass(frozen=True, slots=True) +class OutboundDeliveryResult: + message: AppendResult + delivery: DeliveryRequestResult + + async def _resolve_outbound(store: WorkshopEventStore, message_id: MessageId) -> _ResolvedOutbound: async with store.connection.execute( "SELECT c.workshop_id, m.channel_id, a.principal_id " @@ -93,51 +109,123 @@ def _outbound_key(message_id: MessageId) -> str: return f"workshop-outbound:v1:{message_id}" +def _outbound_payload(binding: _ResolvedOutbound, message: OutboundMessage) -> dict[str, object]: + return { + "channel_id": binding.channel_id, + "author_principal_id": binding.agent_principal_id, + "reply_to_message_id": message.in_reply_to_message_id, + "body": message.body, + } + + +def _outbound_envelope(binding: _ResolvedOutbound, message: OutboundMessage) -> EventEnvelope: + return EventEnvelope.create( + event_id=EventId.derived(binding.workshop_id, f"outbound-message-event:{message.in_reply_to_message_id}"), + event_type=WorkshopEventType.MESSAGE_CREATED, + event_version=1, + workshop_id=binding.workshop_id, + aggregate_type="message", + aggregate_id=MessageId.derived( + binding.workshop_id, + f"outbound-message:{message.in_reply_to_message_id}", + ), + actor_principal_id=binding.agent_principal_id, + occurred_at=message.occurred_at, + idempotency_key=_outbound_key(message.in_reply_to_message_id), + payload=_outbound_payload(binding, message), + metadata={"source": "agent"}, + ) + + +async def _existing_outbound( + store: WorkshopEventStore, + binding: _ResolvedOutbound, + message: OutboundMessage, +) -> AppendResult | None: + key = _outbound_key(message.in_reply_to_message_id) + existing = await store.event_by_idempotency_key(key) + if existing is None: + return None + if ( + existing.envelope.event_type != WorkshopEventType.MESSAGE_CREATED + or existing.envelope.payload != _outbound_payload(binding, message) + or existing.envelope.actor_principal_id != binding.agent_principal_id + ): + raise IdempotencyConflictError(f"Event identity {key!r} was reused with different content") + return AppendResult(event=existing, inserted=False) + + +async def _resolve_telegram_binding( + store: WorkshopEventStore, + channel_id: ChannelId, +) -> ChannelBindingId: + async with store.connection.execute( + "SELECT id FROM channel_bindings WHERE channel_id = ? AND transport = 'telegram' ORDER BY id", + (channel_id,), + ) as cursor: + rows = list(await cursor.fetchall()) + if len(rows) != 1: + raise OutboundDeliveryBindingError("Canonical reply channel must have exactly one Telegram binding") + return ChannelBindingId(str(rows[0][0])) + + async def record_outbound_message(store: WorkshopEventStore, message: OutboundMessage) -> AppendResult: """Append one canonical assistant reply to an existing inbound message.""" binding = await _resolve_outbound(store, message.in_reply_to_message_id) - key = _outbound_key(message.in_reply_to_message_id) - existing = await store.event_by_idempotency_key(key) + existing = await _existing_outbound(store, binding, message) if existing is not None: - expected_payload = { - "channel_id": binding.channel_id, - "author_principal_id": binding.agent_principal_id, - "reply_to_message_id": message.in_reply_to_message_id, - "body": message.body, - } - if ( - existing.envelope.event_type != WorkshopEventType.MESSAGE_CREATED - or existing.envelope.payload != expected_payload - or existing.envelope.actor_principal_id != binding.agent_principal_id - ): - raise IdempotencyConflictError(f"Event identity {key!r} was reused with different content") await store.project_pending(CanonicalConversationProjection()) - return AppendResult(event=existing, inserted=False) + return existing - result = await store.append( - EventEnvelope.create( - event_id=EventId.derived(binding.workshop_id, f"outbound-message-event:{message.in_reply_to_message_id}"), - event_type=WorkshopEventType.MESSAGE_CREATED, - event_version=1, - workshop_id=binding.workshop_id, - aggregate_type="message", - aggregate_id=MessageId.derived(binding.workshop_id, f"outbound-message:{message.in_reply_to_message_id}"), - actor_principal_id=binding.agent_principal_id, - occurred_at=message.occurred_at, - idempotency_key=key, - payload={ - "channel_id": binding.channel_id, - "author_principal_id": binding.agent_principal_id, - "reply_to_message_id": message.in_reply_to_message_id, - "body": message.body, - }, - metadata={"source": "agent"}, - ) - ) + result = await store.append(_outbound_envelope(binding, message)) await store.project_pending(CanonicalConversationProjection()) return result +async def record_outbound_message_with_delivery( + store: WorkshopEventStore, + message: OutboundMessage, +) -> OutboundDeliveryResult: + """Atomically create one canonical reply and its pending Telegram text delivery. + + This service is deliberately production-unused. It accepts no transport or + destination identity from its caller and does not send or register a worker. + """ + connection = store.connection + try: + await connection.execute("BEGIN IMMEDIATE") + binding = await _resolve_outbound(store, message.in_reply_to_message_id) + channel_binding_id = await _resolve_telegram_binding(store, binding.channel_id) + message_result = await _existing_outbound(store, binding, message) + if message_result is None: + message_result = await store.append_in_transaction(_outbound_envelope(binding, message)) + + projection = CanonicalConversationProjection() + await store.project_pending_in_transaction(projection) + message_id = message_result.event.envelope.aggregate_id + if not isinstance(message_id, MessageId): + raise RuntimeError("Canonical outbound event did not identify a message") + delivery_result = await WorkshopDeliveryOutbox(store).request_delivery_in_transaction( + DeliveryRequest( + message_id=message_id, + channel_binding_id=channel_binding_id, + mode="text", + occurred_at=message.occurred_at, + max_attempts=5, + ) + ) + if message_result.inserted != delivery_result.inserted: + raise OutboundDeliveryStateConflictError( + "Canonical reply and delivery request did not share one prior state" + ) + await store.project_pending_in_transaction(projection) + await connection.commit() + return OutboundDeliveryResult(message=message_result, delivery=delivery_result) + except Exception: + await connection.rollback() + raise + + async def _resolve_delivery(store: WorkshopEventStore, message_id: MessageId) -> tuple[WorkshopId, ChannelId]: async with store.connection.execute( "SELECT c.workshop_id, m.channel_id FROM messages m JOIN channels c ON c.id = m.channel_id WHERE m.id = ?", diff --git a/src/kai/workshop/store.py b/src/kai/workshop/store.py index 47c78046..b313e8bc 100644 --- a/src/kai/workshop/store.py +++ b/src/kai/workshop/store.py @@ -297,50 +297,57 @@ async def rebuild_projection(self, projection: Projection) -> ProjectionCheckpoi async def project_pending(self, projection: Projection) -> ProjectionCheckpoint: """Atomically apply events after a projection checkpoint.""" - if not _PROJECTION_NAME_PATTERN.fullmatch(projection.name): - raise ValueError("Projection name must be a lowercase identifier") - if not isinstance(projection.version, int) or isinstance(projection.version, bool) or projection.version < 1: - raise ValueError("Projection version must be a positive integer") - try: await self._connection.execute("BEGIN IMMEDIATE") - async with self._connection.execute( - "SELECT version, last_position FROM projection_checkpoints WHERE name = ?", - (projection.name,), - ) as cursor: - checkpoint_row = await cursor.fetchone() - - if checkpoint_row is None or int(checkpoint_row[0]) < projection.version: - await projection.reset(self._connection) - after_position = 0 - elif int(checkpoint_row[0]) > projection.version: - raise RuntimeError( - f"Projection {projection.name!r} is newer than this build: " - f"stored={int(checkpoint_row[0])}, supported={projection.version}" - ) - else: - after_position = int(checkpoint_row[1]) - - events = await self.read_events(after_position=after_position) - for event in events: - await projection.apply(self._connection, event) - last_position = events[-1].position if events else after_position - updated_at = _format_timestamp(datetime.now(UTC)) - await self._connection.execute( - """ - INSERT INTO projection_checkpoints (name, version, last_position, updated_at) - VALUES (?, ?, ?, ?) - ON CONFLICT(name) DO UPDATE SET - version = excluded.version, - last_position = excluded.last_position, - updated_at = excluded.updated_at - """, - (projection.name, projection.version, last_position, updated_at), - ) + checkpoint = await self.project_pending_in_transaction(projection) await self._connection.commit() + return checkpoint except Exception: await self._connection.rollback() raise + + async def project_pending_in_transaction(self, projection: Projection) -> ProjectionCheckpoint: + """Apply pending events without committing an existing transaction.""" + if not self._connection.in_transaction: + raise RuntimeError("project_pending_in_transaction requires an active transaction") + if not _PROJECTION_NAME_PATTERN.fullmatch(projection.name): + raise ValueError("Projection name must be a lowercase identifier") + if not isinstance(projection.version, int) or isinstance(projection.version, bool) or projection.version < 1: + raise ValueError("Projection version must be a positive integer") + + async with self._connection.execute( + "SELECT version, last_position FROM projection_checkpoints WHERE name = ?", + (projection.name,), + ) as cursor: + checkpoint_row = await cursor.fetchone() + + if checkpoint_row is None or int(checkpoint_row[0]) < projection.version: + await projection.reset(self._connection) + after_position = 0 + elif int(checkpoint_row[0]) > projection.version: + raise RuntimeError( + f"Projection {projection.name!r} is newer than this build: " + f"stored={int(checkpoint_row[0])}, supported={projection.version}" + ) + else: + after_position = int(checkpoint_row[1]) + + events = await self.read_events(after_position=after_position) + for event in events: + await projection.apply(self._connection, event) + last_position = events[-1].position if events else after_position + updated_at = _format_timestamp(datetime.now(UTC)) + await self._connection.execute( + """ + INSERT INTO projection_checkpoints (name, version, last_position, updated_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(name) DO UPDATE SET + version = excluded.version, + last_position = excluded.last_position, + updated_at = excluded.updated_at + """, + (projection.name, projection.version, last_position, updated_at), + ) return ProjectionCheckpoint( name=projection.name, version=projection.version, diff --git a/tests/test_workshop_delivery_outbox.py b/tests/test_workshop_delivery_outbox.py index 2b652833..cbeee17b 100644 --- a/tests/test_workshop_delivery_outbox.py +++ b/tests/test_workshop_delivery_outbox.py @@ -170,6 +170,14 @@ def _request( class TestDeliveryRequests: + async def test_transactional_request_requires_caller_transaction(self, tmp_path: Path): + store, message_id, binding_id = await _open_with_outbound(tmp_path / "kai.db") + try: + with pytest.raises(RuntimeError, match="active transaction"): + await WorkshopDeliveryOutbox(store).request_delivery_in_transaction(_request(message_id, binding_id)) + finally: + await store.close() + async def test_request_atomically_records_event_and_pending_work_without_delivering(self, tmp_path: Path): store, message_id, binding_id = await _open_with_outbound(tmp_path / "kai.db") try: diff --git a/tests/test_workshop_foundation.py b/tests/test_workshop_foundation.py index f32258d8..0d699ae9 100644 --- a/tests/test_workshop_foundation.py +++ b/tests/test_workshop_foundation.py @@ -787,6 +787,26 @@ async def apply(self, connection: aiosqlite.Connection, event) -> None: class TestProjectionReplay: + async def test_transactional_projection_requires_and_preserves_caller_transaction(self, workshop_store): + projection = MessageCountProjection() + with pytest.raises(RuntimeError, match="active transaction"): + await workshop_store.project_pending_in_transaction(projection) + + await workshop_store.append(_message_event()) + await workshop_store.connection.execute("BEGIN IMMEDIATE") + checkpoint = await workshop_store.project_pending_in_transaction(projection) + assert workshop_store.connection.in_transaction is True + assert checkpoint.last_position == 1 + async with workshop_store.connection.execute("SELECT SUM(message_count) FROM test_message_counts") as cursor: + assert (await cursor.fetchone())[0] == 1 + await workshop_store.connection.rollback() + + async with workshop_store.connection.execute( + "SELECT COUNT(*) FROM projection_checkpoints WHERE name = ?", + (projection.name,), + ) as cursor: + assert (await cursor.fetchone())[0] == 0 + async def test_rebuild_is_deterministic_and_records_checkpoint(self, workshop_store): await workshop_store.append(_message_event(idempotency_key="telegram:message:1", text="one")) await workshop_store.append(_message_event(idempotency_key="telegram:message:2", text="two")) diff --git a/tests/test_workshop_outbound.py b/tests/test_workshop_outbound.py index 57c214c1..f1ddfa26 100644 --- a/tests/test_workshop_outbound.py +++ b/tests/test_workshop_outbound.py @@ -2,21 +2,32 @@ from __future__ import annotations +import asyncio from datetime import UTC, datetime, timedelta from pathlib import Path +import aiosqlite import pytest from kai import sessions from kai.workshop.bootstrap import BootstrapHuman, bootstrap_default_workshop -from kai.workshop.domain import MessageId +from kai.workshop.domain import ( + ChannelBindingId, + EventEnvelope, + MessageId, + WorkshopEventType, + WorkshopId, +) from kai.workshop.inbound import InboundMessage, record_inbound_message from kai.workshop.outbound import ( DeliveryObservation, + OutboundDeliveryBindingError, + OutboundDeliveryStateConflictError, OutboundMessage, OutboundMessageNotFoundError, record_delivery_observation, record_outbound_message, + record_outbound_message_with_delivery, ) from kai.workshop.projection import CanonicalConversationProjection from kai.workshop.store import IdempotencyConflictError, WorkshopEventStore @@ -159,6 +170,229 @@ async def test_requires_existing_canonical_inbound_message(self, tmp_path: Path) await store.close() +class TestAtomicOutboundDelivery: + async def test_commits_message_projection_request_event_and_pending_work_together(self, tmp_path: Path): + store, inbound_id = await _open_with_inbound(tmp_path / "kai.db") + try: + result = await record_outbound_message_with_delivery(store, _outbound(inbound_id)) + + assert result.message.inserted is True + assert result.delivery.inserted is True + assert result.delivery.delivery.message_id == result.message.event.envelope.aggregate_id + assert result.delivery.delivery.transport == "telegram" + assert result.delivery.delivery.mode == "text" + assert result.delivery.delivery.status == "pending" + assert result.delivery.delivery.attempt_count == 0 + async with store.connection.execute( + "SELECT body, reply_to_message_id FROM messages WHERE id = ?", + (result.message.event.envelope.aggregate_id,), + ) as cursor: + assert tuple(await cursor.fetchone()) == ("Hello back", inbound_id) + async with store.connection.execute( + "SELECT event_type FROM event_log WHERE position IN (?, ?) ORDER BY position", + (result.message.event.position, result.delivery.delivery.requested_event_position), + ) as cursor: + assert [row[0] for row in await cursor.fetchall()] == [ + "message.created", + "delivery.requested", + ] + async with store.connection.execute( + "SELECT last_position FROM projection_checkpoints WHERE name = 'canonical_conversations'" + ) as cursor: + assert (await cursor.fetchone())[0] == result.delivery.delivery.requested_event_position + finally: + await store.close() + + async def test_retry_is_idempotent_across_restart_and_observation_time(self, tmp_path: Path): + path = tmp_path / "kai.db" + store, inbound_id = await _open_with_inbound(path) + first = await record_outbound_message_with_delivery(store, _outbound(inbound_id)) + await store.close() + + reopened = await WorkshopEventStore.open(path) + try: + retry = await record_outbound_message_with_delivery( + reopened, + OutboundMessage(inbound_id, "Hello back", _NOW + timedelta(minutes=5)), + ) + + assert retry.message.inserted is False + assert retry.delivery.inserted is False + assert retry.message.event == first.message.event + assert retry.delivery.delivery == first.delivery.delivery + async with reopened.connection.execute( + "SELECT COUNT(*) FROM event_log WHERE event_type IN ('message.created', 'delivery.requested')" + ) as cursor: + # One inbound message, one assistant message, and one request. + assert (await cursor.fetchone())[0] == 3 + async with reopened.connection.execute("SELECT COUNT(*) FROM delivery_outbox") as cursor: + assert (await cursor.fetchone())[0] == 1 + finally: + await reopened.close() + + async def test_concurrent_retry_across_connections_creates_one_message_and_delivery(self, tmp_path: Path): + path = tmp_path / "kai.db" + first_store, inbound_id = await _open_with_inbound(path) + second_store = await WorkshopEventStore.open(path) + try: + first, second = await asyncio.gather( + record_outbound_message_with_delivery(first_store, _outbound(inbound_id)), + record_outbound_message_with_delivery(second_store, _outbound(inbound_id)), + ) + + assert sorted((first.message.inserted, second.message.inserted)) == [False, True] + assert sorted((first.delivery.inserted, second.delivery.inserted)) == [False, True] + assert first.message.event == second.message.event + assert first.delivery.delivery == second.delivery.delivery + async with first_store.connection.execute( + "SELECT COUNT(*) FROM event_log WHERE event_type IN ('message.created', 'delivery.requested')" + ) as cursor: + assert (await cursor.fetchone())[0] == 3 + async with first_store.connection.execute("SELECT COUNT(*) FROM delivery_outbox") as cursor: + assert (await cursor.fetchone())[0] == 1 + finally: + await first_store.close() + await second_store.close() + + async def test_changed_body_fails_closed_without_changing_delivery(self, tmp_path: Path): + store, inbound_id = await _open_with_inbound(tmp_path / "kai.db") + try: + first = await record_outbound_message_with_delivery(store, _outbound(inbound_id, body="original")) + + with pytest.raises(IdempotencyConflictError): + await record_outbound_message_with_delivery(store, _outbound(inbound_id, body="changed")) + + async with store.connection.execute( + "SELECT body FROM messages WHERE id = ?", (first.message.event.envelope.aggregate_id,) + ) as cursor: + assert (await cursor.fetchone())[0] == "original" + async with store.connection.execute("SELECT COUNT(*) FROM delivery_outbox") as cursor: + assert (await cursor.fetchone())[0] == 1 + finally: + await store.close() + + async def test_missing_telegram_binding_rolls_back_without_creating_reply(self, tmp_path: Path): + store, inbound_id = await _open_with_inbound(tmp_path / "kai.db") + try: + await store.connection.execute("DELETE FROM channel_bindings") + await store.connection.commit() + + with pytest.raises(OutboundDeliveryBindingError): + await record_outbound_message_with_delivery(store, _outbound(inbound_id)) + + await self._assert_no_outbound_delivery_state(store, inbound_id) + finally: + await store.close() + + async def test_ambiguous_telegram_binding_rolls_back_without_guessing(self, tmp_path: Path): + store, inbound_id = await _open_with_inbound(tmp_path / "kai.db") + try: + async with store.connection.execute( + "SELECT c.workshop_id, m.channel_id FROM messages m " + "JOIN channels c ON c.id = m.channel_id WHERE m.id = ?", + (inbound_id,), + ) as cursor: + row = await cursor.fetchone() + assert row is not None + workshop_id = WorkshopId(str(row[0])) + channel_id = str(row[1]) + await store.append( + EventEnvelope.create( + event_type=WorkshopEventType.TRANSPORT_CHANNEL_BOUND, + event_version=1, + workshop_id=workshop_id, + aggregate_type="channel_binding", + aggregate_id=ChannelBindingId.new(), + occurred_at=_NOW, + idempotency_key="test:second-telegram-binding", + payload={ + "channel_id": channel_id, + "transport": "telegram", + "external_channel_id": "202", + }, + ) + ) + await store.project_pending(CanonicalConversationProjection()) + + with pytest.raises(OutboundDeliveryBindingError): + await record_outbound_message_with_delivery(store, _outbound(inbound_id)) + + await self._assert_no_outbound_delivery_state(store, inbound_id) + finally: + await store.close() + + async def test_outbox_insert_failure_rolls_back_message_event_and_projection(self, tmp_path: Path): + store, inbound_id = await _open_with_inbound(tmp_path / "kai.db") + try: + await store.connection.execute( + "CREATE TRIGGER reject_atomic_delivery BEFORE INSERT ON delivery_outbox " + "BEGIN SELECT RAISE(ABORT, 'test delivery rejection'); END" + ) + await store.connection.commit() + + with pytest.raises(aiosqlite.IntegrityError, match="test delivery rejection"): + await record_outbound_message_with_delivery(store, _outbound(inbound_id)) + + await self._assert_no_outbound_delivery_state(store, inbound_id) + finally: + await store.close() + + async def test_projection_failure_rolls_back_message_event_and_delivery(self, tmp_path: Path): + store, inbound_id = await _open_with_inbound(tmp_path / "kai.db") + try: + await store.connection.execute( + "CREATE TRIGGER reject_atomic_projection BEFORE INSERT ON messages " + "WHEN NEW.reply_to_message_id IS NOT NULL " + "BEGIN SELECT RAISE(ABORT, 'test projection rejection'); END" + ) + await store.connection.commit() + + with pytest.raises(aiosqlite.IntegrityError, match="test projection rejection"): + await record_outbound_message_with_delivery(store, _outbound(inbound_id)) + + await self._assert_no_outbound_delivery_state(store, inbound_id) + finally: + await store.close() + + async def test_preexisting_message_without_delivery_is_rejected_as_half_state(self, tmp_path: Path): + store, inbound_id = await _open_with_inbound(tmp_path / "kai.db") + try: + prior = await record_outbound_message(store, _outbound(inbound_id)) + + with pytest.raises(OutboundDeliveryStateConflictError): + await record_outbound_message_with_delivery(store, _outbound(inbound_id)) + + async with store.connection.execute("SELECT COUNT(*) FROM delivery_outbox") as cursor: + assert (await cursor.fetchone())[0] == 0 + async with store.connection.execute( + "SELECT COUNT(*) FROM event_log WHERE event_type = 'delivery.requested'" + ) as cursor: + assert (await cursor.fetchone())[0] == 0 + async with store.connection.execute( + "SELECT body FROM messages WHERE id = ?", (prior.event.envelope.aggregate_id,) + ) as cursor: + assert (await cursor.fetchone())[0] == "Hello back" + finally: + await store.close() + + @staticmethod + async def _assert_no_outbound_delivery_state(store: WorkshopEventStore, inbound_id: MessageId) -> None: + async with store.connection.execute( + "SELECT COUNT(*) FROM event_log WHERE " + "(event_type = 'message.created' AND json_extract(payload_json, '$.reply_to_message_id') = ?) " + "OR event_type = 'delivery.requested'", + (inbound_id,), + ) as cursor: + assert (await cursor.fetchone())[0] == 0 + async with store.connection.execute( + "SELECT COUNT(*) FROM messages WHERE reply_to_message_id = ?", + (inbound_id,), + ) as cursor: + assert (await cursor.fetchone())[0] == 0 + async with store.connection.execute("SELECT COUNT(*) FROM delivery_outbox") as cursor: + assert (await cursor.fetchone())[0] == 0 + + class TestDeliveryObservation: async def test_records_successful_telegram_text_delivery(self, tmp_path: Path): store, inbound_id = await _open_with_inbound(tmp_path / "kai.db")