fix: reconcile uncertain voice admission with bounded cancellation
This commit is contained in:
1 parent
0cb00e6095
commit
0c57217ad9
3 files changed
+40
-4
No files matched your search
@@ -24,7 +24,7 @@ jobs:
|
|||||||
tests/unit/test_accounts.py tests/unit/test_provisioner.py \
|
tests/unit/test_accounts.py tests/unit/test_provisioner.py \
|
||||||
tests/unit/test_token_bootstrap.py tests/unit/test_deploy_rollback.py \
|
tests/unit/test_token_bootstrap.py tests/unit/test_deploy_rollback.py \
|
||||||
tests/unit/test_model_preferences_store.py tests/unit/test_model_settings_api.py \
|
tests/unit/test_model_preferences_store.py tests/unit/test_model_settings_api.py \
|
||||||
tests/unit/test_chat_store.py tests/unit/test_chat_runs.py \
|
tests/unit/test_chat_store.py tests/unit/test_chat_runs.py tests/unit/test_chat_runs_limits.py \
|
||||||
tests/unit/test_chat_api.py tests/unit/test_chat_exports.py \
|
tests/unit/test_chat_api.py tests/unit/test_chat_exports.py \
|
||||||
tests/unit/test_chat_features.py tests/unit/test_chat_security_review.py \
|
tests/unit/test_chat_features.py tests/unit/test_chat_security_review.py \
|
||||||
tests/unit/test_chat_cross_client.py tests/unit/test_chat_packaging.py -q
|
tests/unit/test_chat_cross_client.py tests/unit/test_chat_packaging.py -q
|
||||||
|
|||||||
@@ -65,6 +65,7 @@ class ConversationStream(llm.LLMStream):
|
|||||||
if not text.strip(): raise HermesLLMError('empty_voice_turn')
|
if not text.strip(): raise HermesLLMError('empty_voice_turn')
|
||||||
timer = None
|
timer = None
|
||||||
admission = None
|
admission = None
|
||||||
|
admission_rejected = False
|
||||||
try:
|
try:
|
||||||
# Do not abandon uncertain admission on barge-in. A successfully
|
# Do not abandon uncertain admission on barge-in. A successfully
|
||||||
# admitted remote action must be identified and cancelled, not left
|
# admitted remote action must be identified and cancelled, not left
|
||||||
@@ -72,7 +73,9 @@ class ConversationStream(llm.LLMStream):
|
|||||||
admission = asyncio.create_task(self.bridge.client.post(self.bridge.root + '/messages', headers=self.bridge.headers,
|
admission = asyncio.create_task(self.bridge.client.post(self.bridge.root + '/messages', headers=self.bridge.headers,
|
||||||
json={'content': text, 'idempotency_key': self.operation_id, 'modality': 'voice'}, timeout=httpx.Timeout(20, connect=5)))
|
json={'content': text, 'idempotency_key': self.operation_id, 'modality': 'voice'}, timeout=httpx.Timeout(20, connect=5)))
|
||||||
response = await asyncio.shield(admission)
|
response = await asyncio.shield(admission)
|
||||||
if response.status_code != 202: raise HermesLLMError('voice_admission_failed')
|
if response.status_code != 202:
|
||||||
|
admission_rejected = 400 <= response.status_code < 500
|
||||||
|
raise HermesLLMError('voice_admission_failed')
|
||||||
self.run_id = response.json()['run']['id']
|
self.run_id = response.json()['run']['id']
|
||||||
await self.progress('working', 'Working on your request.')
|
await self.progress('working', 'Working on your request.')
|
||||||
timer = asyncio.create_task(self.notice())
|
timer = asyncio.create_task(self.notice())
|
||||||
@@ -117,17 +120,25 @@ class ConversationStream(llm.LLMStream):
|
|||||||
# admission, then return the run identity for cancellation.
|
# admission, then return the run identity for cancellation.
|
||||||
try:
|
try:
|
||||||
admitted = await self.bridge.client.post(self.bridge.root + '/messages', headers=self.bridge.headers,
|
admitted = await self.bridge.client.post(self.bridge.root + '/messages', headers=self.bridge.headers,
|
||||||
json={'content': text, 'idempotency_key': self.operation_id, 'modality': 'voice'})
|
json={'content': text, 'idempotency_key': self.operation_id, 'modality': 'voice'}, timeout=httpx.Timeout(20, connect=5))
|
||||||
if admitted.status_code == 202:
|
if admitted.status_code == 202:
|
||||||
self.run_id = admitted.json()['run']['id']
|
self.run_id = admitted.json()['run']['id']
|
||||||
except Exception: pass
|
except Exception: pass
|
||||||
if self.run_id:
|
if self.run_id:
|
||||||
try:
|
try:
|
||||||
await asyncio.shield(self.bridge.client.post(self.bridge.root + '/runs/' + self.run_id + '/cancel', headers=self.bridge.headers))
|
await asyncio.shield(self.bridge.client.post(self.bridge.root + '/runs/' + self.run_id + '/cancel', headers=self.bridge.headers, timeout=httpx.Timeout(10, connect=5)))
|
||||||
except Exception: pass
|
except Exception: pass
|
||||||
await self.progress('failed', 'Request cancelled.')
|
await self.progress('failed', 'Request cancelled.')
|
||||||
raise
|
raise
|
||||||
except Exception:
|
except Exception:
|
||||||
|
# A failed HTTP receipt is not proof admission failed. Fail safely
|
||||||
|
# by tombstoning the stable operation key, including when JSON/SSE
|
||||||
|
# decoding failed after the coordinator already accepted the turn.
|
||||||
|
if not admission_rejected:
|
||||||
|
try:
|
||||||
|
await self.bridge.client.post(self.bridge.root + '/operations/' + self.operation_id + '/cancel',
|
||||||
|
headers=self.bridge.headers, timeout=httpx.Timeout(10, connect=5))
|
||||||
|
except Exception: pass
|
||||||
await self.progress('failed', 'Hope could not complete this response.')
|
await self.progress('failed', 'Hope could not complete this response.')
|
||||||
raise HermesLLMError('voice_coordinator_failed') from None
|
raise HermesLLMError('voice_coordinator_failed') from None
|
||||||
finally:
|
finally:
|
||||||
|
|||||||
@@ -174,6 +174,31 @@ class ConversationLLMTests(unittest.IsolatedAsyncioTestCase):
|
|||||||
await asyncio.gather(consumer, return_exceptions=True)
|
await asyncio.gather(consumer, return_exceptions=True)
|
||||||
await bridge.aclose()
|
await bridge.aclose()
|
||||||
|
|
||||||
|
async def test_normal_lost_admission_receipt_persists_key_cancellation(self):
|
||||||
|
requests, progress_events = [], []
|
||||||
|
async def handler(request):
|
||||||
|
requests.append(request.url.path)
|
||||||
|
if request.url.path.endswith('/messages'):
|
||||||
|
raise httpx.ReadError('Offline accepted admission receipt lost', request=request)
|
||||||
|
if '/operations/' in request.url.path and request.url.path.endswith('/cancel'):
|
||||||
|
return httpx.Response(200, json={'ok': True})
|
||||||
|
self.fail('Unknown admission identity must not stream or invent a run stop')
|
||||||
|
async def progress(value): progress_events.append(value)
|
||||||
|
bridge = ConversationLLM(control_url='https://offline.test', secret='offline', room_id='bound-room', on_progress=progress)
|
||||||
|
await bridge.client.aclose()
|
||||||
|
bridge.client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
||||||
|
context = llm.ChatContext([llm.ChatMessage(role='user', content=['Lost receipt without barge-in'])])
|
||||||
|
try:
|
||||||
|
with self.assertRaises(Exception):
|
||||||
|
async with bridge.chat(chat_ctx=context) as stream:
|
||||||
|
async for _ in stream:
|
||||||
|
self.fail('No answer is safe after uncertain admission failure')
|
||||||
|
self.assertEqual(requests, ['/internal/chat/rooms/bound-room/messages',
|
||||||
|
'/internal/chat/rooms/bound-room/operations/' + context.items[-1].id + '/cancel'])
|
||||||
|
self.assertEqual(progress_events[-1]['status'], 'failed')
|
||||||
|
finally:
|
||||||
|
await bridge.aclose()
|
||||||
|
|
||||||
async def test_key_cancel_receipt_prevents_late_rejected_admission_action(self):
|
async def test_key_cancel_receipt_prevents_late_rejected_admission_action(self):
|
||||||
entered, release, tombstoned = asyncio.Event(), asyncio.Event(), asyncio.Event()
|
entered, release, tombstoned = asyncio.Event(), asyncio.Event(), asyncio.Event()
|
||||||
requests, progress_events = [], []
|
requests, progress_events = [], []
|
||||||
|
|||||||
Reference in new issue
Block a user