David/HERMES-GIELIE
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #2151 into stage-350

Browse source 
fix: clarify cancelled chat turn status (Jordan-SkyLF)

Conflict resolution on api/streaming.py:4549-4567 (the cancel-handler
ownership guard). Both this PR and the already-shipped PR #2136 add a
guard at the same site against stale stream writebacks, from different
angles:

  - PR #2136 (HEAD): _stream_writeback_is_current(_cs, stream_id) — strictly
    dominates by checking the active_stream_id token equality.
  - PR #2151: 'worker won the race' check via (active_stream_id != stream_id
    and not pending_user_message), with _emit_cancel_event = False to suppress
    the terminal cancel event.

Resolution merges both: keep #2136's strictly-stronger condition for skip
detection, and adopt #2151's _emit_cancel_event = False semantic so the
cancel event isn't emitted in addition to skipping the writeback (when
client may have already received the successful done payload).

55/55 tests pass across cancelled-turn-status + stale-stream-writeback +
the four cancel/data-loss sibling test files.
This commit is contained in:
Hermes Agent 2026-05-13 20:44:44 +00:00
commit 3f851051cf
8 changed files with 597 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

334
api/streaming.py
View file

@ -127,6 +127,9 @@ def _clarify_timeout_seconds(default: int = 120) -> int:
return default
_CANCEL_MARKER_PATTERNS = ('task cancelled', 'task canceled', 'response interrupted')
def _classify_provider_error(err_str: str, exc=None, *, silent_failure: bool = False) -> dict:
"""Classify provider/agent failure text for WebUI apperror UX.
@ -136,6 +139,41 @@ def _classify_provider_error(err_str: str, exc=None, *, silent_failure: bool = F
err_str = str(err_str or '')
_err_lower = err_str.lower()
_exc_name = type(exc).__name__ if exc is not None else ''
_is_cancelled = (
'cancelled by user' in _err_lower
or 'canceled by user' in _err_lower
or 'user cancelled' in _err_lower
or 'user canceled' in _err_lower
or 'task cancelled' in _err_lower
or 'task canceled' in _err_lower
or 'cancellederror' in _err_lower
or (exc is not None and _exc_name in ('CancelledError', 'CanceledError'))
)
_is_interrupted = (
not _is_cancelled
and (
'interrupted by user' in _err_lower
or 'response interrupted' in _err_lower
or 'operation interrupted' in _err_lower
or 'operation was interrupted' in _err_lower
or 'operation aborted' in _err_lower
or 'request was aborted' in _err_lower
or 'aborterror' in _err_lower
or (exc is not None and type(exc).__name__ in ('KeyboardInterrupt', 'AbortError'))
)
)
if _is_cancelled:
return {
'label': 'Task cancelled',
'type': 'cancelled',
'hint': 'The run was cancelled by the user before Skyly finished. No provider failure occurred.',
}
if _is_interrupted:
return {
'label': 'Response interrupted',
'type': 'interrupted',
'hint': 'The run stopped before a provider response completed. If you did not cancel it, try again.',
}
_is_quota = _is_quota_error_text(err_str)
_is_auth = (
not _is_quota and (
@ -213,6 +251,90 @@ def _provider_error_payload(message: str, err_type: str, hint: str = '') -> dict
return payload
def _session_has_cancel_marker(session) -> bool:
"""Return True if a visible cancel/interrupted marker is already persisted."""
for msg in reversed(getattr(session, 'messages', None) or []):
if not isinstance(msg, dict):
continue
if msg.get('role') == 'user':
return False
if msg.get('role') != 'assistant':
continue
content = msg.get('content')
text = ''
if isinstance(content, str):
text = content
elif isinstance(content, list):
parts = []
for part in content:
if isinstance(part, dict):
parts.append(str(part.get('text') or part.get('content') or ''))
text = '\n'.join(parts)
normalized = text.strip().lower()
if any(pattern in normalized for pattern in _CANCEL_MARKER_PATTERNS):
return True
return False
def _cancelled_turn_content(message: str = 'Task cancelled.') -> str:
"""Return cancelled-turn copy matching the verbose provider-error layout."""
_message = str(message or 'Task cancelled.').strip()
if not _message.endswith('.'):
_message += '.'
return (
f"**Task cancelled:** {_message}\n\n"
"*The run was cancelled by the user before Skyly finished. No provider failure occurred.*"
)
def _persist_cancelled_turn(session, *, message: str = 'Task cancelled.') -> None:
"""Persist a user-cancelled terminal state without provider-error wording.
cancel_stream() usually writes this marker first, but the streaming thread can
later unwind through the silent-failure or exception path. Those paths must
not append a misleading provider no-response error after an explicit cancel.
"""
_materialize_pending_user_turn_before_error(session)
session.active_stream_id = None
session.pending_user_message = None
session.pending_attachments = []
session.pending_started_at = None
if not _session_has_cancel_marker(session):
session.messages.append({
'role': 'assistant',
'content': _cancelled_turn_content(message),
'_error': True,
'provider_details': str(message or 'Task cancelled.').strip(),
'provider_details_label': 'Cancellation details',
'timestamp': int(time.time()),
})
def _cleanup_ephemeral_cancelled_turn(session) -> None:
"""Remove transient /btw session state after a cancel without saving it."""
session.active_stream_id = None
session.pending_user_message = None
session.pending_attachments = []
session.pending_started_at = None
try:
import pathlib
pathlib.Path(session.path).unlink(missing_ok=True)
except Exception:
logger.debug("Failed to clean up ephemeral cancelled session", exc_info=True)
def _finalize_cancelled_turn(session, *, ephemeral: bool = False, message: str = 'Task cancelled.') -> None:
"""Finalize a cancelled turn for persistent or ephemeral sessions."""
if ephemeral:
_cleanup_ephemeral_cancelled_turn(session)
return
_persist_cancelled_turn(session, message=message)
try:
session.save()
except Exception:
logger.debug("Failed to persist cancelled turn", exc_info=True)
def _aiagent_import_error_detail() -> str:
"""Return a multi-line diagnostic string for the "AIAgent not available" path.
@ -2287,6 +2409,8 @@ def _run_agent_streaming(
# TD1: set thread-local env context so concurrent sessions don't clobber globals
# Check for pre-flight cancel (user cancelled before agent even started)
if cancel_event.is_set():
with _agent_lock:
_finalize_cancelled_turn(s, ephemeral=ephemeral, message='Task cancelled before start.')
put('cancel', {'message': 'Cancelled before start'})
return
@ -3006,6 +3130,8 @@ def _run_agent_streaming(
agent.interrupt("Cancelled before start")
except Exception:
logger.debug("Failed to interrupt agent before start")
with _agent_lock:
_finalize_cancelled_turn(s, ephemeral=ephemeral, message='Task cancelled before start.')
put('cancel', {'message': 'Cancelled by user'})
return
@ -3107,6 +3233,30 @@ def _run_agent_streaming(
task_id=session_id,
persist_user_message=msg_text,
)
if cancel_event.is_set():
if _checkpoint_stop is not None:
_checkpoint_stop.set()
if _ckpt_thread is not None:
_ckpt_thread.join(timeout=15)
if ephemeral:
_cleanup_ephemeral_cancelled_turn(s)
else:
with _agent_lock:
_finalize_cancelled_turn(s, ephemeral=False)
try:
append_turn_journal_event_for_stream(
s.session_id,
stream_id,
{
"event": "interrupted",
"created_at": time.time(),
"reason": "cancelled",
},
)
except Exception:
logger.debug("Failed to append cancelled turn journal event", exc_info=True)
put('cancel', {'message': 'Cancelled by user'})
return
# ── Ephemeral mode (/btw): deliver answer, skip persistence, cleanup ──
if ephemeral:
_answer = ''
@ -3132,6 +3282,23 @@ def _run_agent_streaming(
_checkpoint_stop.set()
if _ckpt_thread is not None:
_ckpt_thread.join(timeout=15)
if cancel_event.is_set():
with _agent_lock:
_finalize_cancelled_turn(s, ephemeral=False)
try:
append_turn_journal_event_for_stream(
s.session_id,
stream_id,
{
"event": "interrupted",
"created_at": time.time(),
"reason": "cancelled",
},
)
except Exception:
logger.debug("Failed to append cancelled turn journal event", exc_info=True)
put('cancel', {'message': 'Cancelled by user'})
return
with _agent_lock:
if not ephemeral and not _stream_writeback_is_current(s, stream_id):
logger.info(
@ -3142,6 +3309,22 @@ def _run_agent_streaming(
)
return
_result_messages = result.get('messages') or _previous_context_messages
if cancel_event.is_set():
_finalize_cancelled_turn(s, ephemeral=False)
try:
append_turn_journal_event_for_stream(
s.session_id,
stream_id,
{
"event": "interrupted",
"created_at": time.time(),
"reason": "cancelled",
},
)
except Exception:
logger.debug("Failed to append cancelled turn journal event", exc_info=True)
put('cancel', {'message': 'Cancelled by user'})
return
_next_context_messages = _restore_reasoning_metadata(
_previous_context_messages,
_result_messages,
@ -3180,6 +3363,23 @@ def _run_agent_streaming(
)
# _token_sent tracks whether on_token() was called (any streamed text)
if not _assistant_added and not _token_sent:
if cancel_event.is_set():
_finalize_cancelled_turn(s, ephemeral=ephemeral)
if not ephemeral:
try:
append_turn_journal_event_for_stream(
s.session_id,
stream_id,
{
"event": "interrupted",
"created_at": time.time(),
"reason": "cancelled",
},
)
except Exception:
logger.debug("Failed to append cancelled turn journal event", exc_info=True)
put('cancel', {'message': 'Cancelled by user'})
return
_last_err = getattr(agent, '_last_error', None) or result.get('error') or ''
_err_str = str(_last_err) if _last_err else ''
_classification = _classify_provider_error(
@ -3331,6 +3531,10 @@ def _run_agent_streaming(
}
if _error_payload.get('details'):
_error_message['provider_details'] = _error_payload['details']
if _err_type == 'cancelled':
_error_message['provider_details_label'] = 'Cancellation details'
elif _err_type == 'interrupted':
_error_message['provider_details_label'] = 'Interruption details'
s.messages.append(_error_message)
try:
s.save()
@ -3620,7 +3824,39 @@ def _run_agent_streaming(
)
except Exception:
logger.debug("Failed to append assistant_started turn journal event", exc_info=True)
if cancel_event.is_set():
_finalize_cancelled_turn(s, ephemeral=False)
try:
append_turn_journal_event_for_stream(
s.session_id,
stream_id,
{
"event": "interrupted",
"created_at": time.time(),
"reason": "cancelled",
},
)
except Exception:
logger.debug("Failed to append cancelled turn journal event", exc_info=True)
put('cancel', {'message': 'Cancelled by user'})
return
s.save()
if cancel_event.is_set():
_finalize_cancelled_turn(s, ephemeral=False)
try:
append_turn_journal_event_for_stream(
s.session_id,
stream_id,
{
"event": "interrupted",
"created_at": time.time(),
"reason": "cancelled",
},
)
except Exception:
logger.debug("Failed to append cancelled turn journal event", exc_info=True)
put('cancel', {'message': 'Cancelled by user'})
return
if not ephemeral:
try:
append_turn_journal_event_for_stream(
@ -3877,12 +4113,38 @@ def _run_agent_streaming(
err_str = _stripped
_exc_lower = err_str.lower()
_classification = _classify_provider_error(err_str, e)
if cancel_event.is_set():
if s is not None:
if _checkpoint_stop is not None:
_checkpoint_stop.set()
if _ckpt_thread is not None:
_ckpt_thread.join(timeout=15)
_lock_ctx = _agent_lock if _agent_lock is not None else contextlib.nullcontext()
with _lock_ctx:
_finalize_cancelled_turn(s, ephemeral=ephemeral)
if not ephemeral:
try:
append_turn_journal_event_for_stream(
s.session_id,
stream_id,
{
"event": "interrupted",
"created_at": time.time(),
"reason": "cancelled",
},
)
except Exception:
logger.debug("Failed to append cancelled turn journal event", exc_info=True)
put('cancel', {'message': 'Cancelled by user'})
return
_exc_is_quota = _classification['type'] == 'quota_exhausted'
# Exception quota text still includes: 'more credits' in _exc_lower, 'can only afford' in _exc_lower, 'fewer max_tokens' in _exc_lower.
# Rate-limit detection remains guarded as: (not _exc_is_quota).
_exc_is_rate_limit = (_classification['type'] == 'rate_limit') and (not _exc_is_quota)
_exc_is_auth = _classification['type'] == 'auth_mismatch' # detects '401' and 'unauthorized' via _classify_provider_error.
_exc_is_not_found = _classification['type'] == 'model_not_found' # detects '404', 'not found', 'does not exist', and 'invalid model'.
_exc_is_cancelled = _classification['type'] == 'cancelled'
_exc_is_interrupted = _classification['type'] == 'interrupted'
# The user hint still points to Settings / `hermes model` from _classify_provider_error().
if _exc_is_quota:
@ -3983,6 +4245,10 @@ def _run_agent_streaming(
_exc_label, _exc_type, _exc_hint = (
_classification['label'], _classification['type'], _classification['hint'],
)
elif _exc_is_cancelled or _exc_is_interrupted:
_exc_label, _exc_type, _exc_hint = (
_classification['label'], _classification['type'], _classification['hint'],
)
else:
_exc_label, _exc_type, _exc_hint = 'Error', 'error', ''
@ -4018,6 +4284,10 @@ def _run_agent_streaming(
}
if _error_payload.get('details'):
_error_message['provider_details'] = _error_payload['details']
if _exc_type == 'cancelled':
_error_message['provider_details_label'] = 'Cancellation details'
elif _exc_type == 'interrupted':
_error_message['provider_details_label'] = 'Interruption details'
s.messages.append(_error_message)
try:
s.save()
@ -4221,13 +4491,12 @@ def cancel_stream(stream_id: str) -> bool:
except Exception:
logger.debug("Failed to clear clarify prompt during cancel")
# Put a cancel sentinel into the queue so the SSE handler wakes up
# Capture the queue while the stream still exists, but do not emit the
# terminal cancel event until the session cleanup below confirms the turn
# is still active. Otherwise a late Stop click can race with a successful
# worker save and show cancel in the client while persistence says done.
q = streams.get(stream_id)
if q:
try:
q.put_nowait(('cancel', {'message': 'Cancelled by user'}))
except Exception:
logger.debug("Failed to put cancel event to queue")
_emit_cancel_event = True
# ── Eager session lock release (fixes #653) ──────────────────────────
# Pop stream state now so the 409 guard in routes.py sees the session
@ -4277,13 +4546,21 @@ def cancel_stream(stream_id: str) -> bool:
with _get_session_agent_lock(_cancel_session_id):
try:
_cs = get_session(_cancel_session_id)
if not isinstance(getattr(_cs, 'messages', None), list):
_cs.messages = []
if not _stream_writeback_is_current(_cs, stream_id):
# The stream has rotated to a different stream id (newer
# turn started, or the worker already finalized this one).
# Skip the cancel-marker append AND suppress the terminal
# cancel event so we don't contradict a possibly-already-
# delivered done payload (#2151 + #2154 / PR #2136).
logger.info(
"Skipping stale cancel writeback for session %s stream %s; active_stream_id=%s",
_cancel_session_id,
stream_id,
getattr(_cs, 'active_stream_id', None),
)
_emit_cancel_event = False
return True
# ── Preserve the user's typed message before clearing pending state (#1298) ──
# The agent's internal messages list (where the user message was appended at
@ -4378,7 +4655,27 @@ def cancel_stream(stream_id: str) -> bool:
# reasoning-only or tool-only stream produced NO partial message).
_has_reasoning = bool(_cancel_reasoning and _cancel_reasoning.strip())
_has_tools = bool(_cancel_tool_calls)
if _stripped or _has_reasoning or _has_tools:
_cancel_marker_exists = _session_has_cancel_marker(_cs)
_cancel_marker_idx = len(_cs.messages)
if _cancel_marker_exists:
for _idx in range(len(_cs.messages) - 1, -1, -1):
_m = _cs.messages[_idx]
if not isinstance(_m, dict) or _m.get('role') != 'assistant':
continue
_content = str(_m.get('content') or '').strip().lower()
if any(pattern in _content for pattern in _CANCEL_MARKER_PATTERNS):
_cancel_marker_idx = _idx
break
_partial_already_present = False
if _stripped:
for _m in _cs.messages:
if not isinstance(_m, dict) or _m.get('role') != 'assistant' or _m.get('_error'):
continue
_existing = str(_m.get('content') or '').strip()
if _existing and (_stripped in _existing or _existing in _stripped):
_partial_already_present = True
break
if (_stripped or _has_reasoning or _has_tools) and not _partial_already_present:
_partial_msg: dict = {
'role': 'assistant',
'content': _stripped, # may be empty for reasoning/tool-only turns
@ -4405,18 +4702,27 @@ def cancel_stream(stream_id: str) -> bool:
# alongside the regular tool_calls path.
# (Opus pre-release review pass 2 of v0.50.251.)
_partial_msg['_partial_tool_calls'] = list(_cancel_tool_calls)
_cs.messages.append(_partial_msg)
_cs.messages.insert(_cancel_marker_idx, _partial_msg)
# Cancel marker — flagged _error=True so it is stripped from conversation
# history on the next turn (prevents model from seeing "Task cancelled."
# as a prior assistant reply).
_cs.messages.append({
'role': 'assistant',
'content': '*Task cancelled.*',
'_error': True,
'timestamp': int(time.time()),
})
if not _cancel_marker_exists:
_cs.messages.append({
'role': 'assistant',
'content': _cancelled_turn_content('Task cancelled.'),
'_error': True,
'provider_details': 'Task cancelled.',
'provider_details_label': 'Cancellation details',
'timestamp': int(time.time()),
})
_cs.save()
except Exception:
logger.debug("Failed to clear session state on cancel for %s", _cancel_session_id)
if _emit_cancel_event and q:
try:
q.put_nowait(('cancel', {'message': 'Cancelled by user'}))
except Exception:
logger.debug("Failed to put cancel event to queue")
return True

9
static/messages.js
View file

@ -1285,11 +1285,14 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
const isQuotaExhausted=d.type==='quota_exhausted';
const isAuthMismatch=d.type==='auth_mismatch';
const isModelNotFound=d.type==='model_not_found';
const isCancelled=d.type==='cancelled';
const isInterrupted=d.type==='interrupted';
const isNoResponse=d.type==='no_response'||d.type==='silent_failure';
const label=isQuotaExhausted?'Out of credits':isRateLimit?'Rate limit reached':isAuthMismatch?(typeof t==='function'?t('provider_mismatch_label'):'Provider mismatch'):isModelNotFound?(typeof t==='function'?t('model_not_found_label'):'Model not found'):isNoResponse?'No response received':'Error';
const label=isCancelled?'Task cancelled':isInterrupted?'Response interrupted':isQuotaExhausted?'Out of credits':isRateLimit?'Rate limit reached':isAuthMismatch?(typeof t==='function'?t('provider_mismatch_label'):'Provider mismatch'):isModelNotFound?(typeof t==='function'?t('model_not_found_label'):'Model not found'):isNoResponse?'No response from provider':'Error';
const hint=d.hint?`\n\n*${d.hint}*`:'';
const details=d.details?String(d.details).replace(/```/g,'`\u200b``'):'';
S.messages.push({role:'assistant',content:`**${label}:** ${d.message}${hint}`,provider_details:details});
const detailsLabel=isCancelled?'Cancellation details':isInterrupted?'Interruption details':undefined;
S.messages.push({role:'assistant',content:`**${label}:** ${d.message}${hint}`,provider_details:details,provider_details_label:detailsLabel});
}catch(_){
S.messages.push({role:'assistant',content:'**Error:** An error occurred. Check server logs.'});
}
@ -1380,7 +1383,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
// Fallback to local cancel message if API fails
if(S.session&&S.session.session_id===activeSid){
clearLiveToolCards();if(!assistantText)removeThinking();
S.messages.push({role:'assistant',content:'*Task cancelled.*'});renderMessages({preserveScroll:true});
S.messages.push({role:'assistant',content:'**Task cancelled:** Task cancelled.\n\n*The run was cancelled by the user before Skyly finished. No provider failure occurred.*',provider_details:'Task cancelled.',provider_details_label:'Cancellation details',_error:true});renderMessages({preserveScroll:true});
_markSessionViewed(activeSid, S.messages.length);
}
}

3
static/ui.js
View file

@ -5041,7 +5041,8 @@ function renderMessages(options){
}
let bodyHtml = isUser ? _renderUserFencedBlocks(displayContent) : renderMd(_stripXmlToolCallsDisplay(String(displayContent)));
if(!isUser&&m.provider_details){
bodyHtml += `<details class="provider-error-details"><summary>Provider details</summary><pre><code>${esc(String(m.provider_details))}</code></pre></details>`;
const summary=m.provider_details_label||'Provider details';
bodyHtml += `<details class="provider-error-details"><summary>${esc(String(summary))}</summary><pre><code>${esc(String(m.provider_details))}</code></pre></details>`;
}
const statusHtml = (!isUser&&m._statusCard) ? _statusCardHtml(m._statusCard) : '';
const isEditableUser=isUser&&rawIdx===lastUserRawIdx;

185
tests/test_cancelled_turn_status.py Normal file
View file

@ -0,0 +1,185 @@
"""Regression tests for accurate cancelled/interrupted turn status.
A user pressing Stop/Cancel must not be shown provider-empty guidance like
"No response from provider". Provider-empty remains valid only when there was
no explicit cancel/interruption signal.
"""
from __future__ import annotations
import pathlib
from api.streaming import (
_CANCEL_MARKER_PATTERNS,
_cancelled_turn_content,
_classify_provider_error,
_finalize_cancelled_turn,
)
REPO_ROOT = pathlib.Path(__file__).parent.parent.resolve()
def _read(rel_path: str) -> str:
return (REPO_ROOT / rel_path).read_text(encoding="utf-8")
class _DummySession:
def __init__(self, path: str = ''):
self.path = path
self.messages = []
self.active_stream_id = 'stream-1'
self.pending_user_message = 'hello'
self.pending_attachments = ['a.txt']
self.pending_started_at = 123
self.saved = 0
def save(self, *args, **kwargs):
self.saved += 1
class TestCancelledTurnClassification:
def test_user_cancelled_error_is_not_provider_no_response(self):
result = _classify_provider_error("Cancelled by user", Exception("Cancelled by user"))
assert result["type"] == "cancelled"
assert result["label"] == "Task cancelled"
assert "provider returned no content" not in result.get("hint", "").lower()
assert "rate limit" not in result.get("hint", "").lower()
assert "no provider failure" in result.get("hint", "").lower()
def test_string_only_cancelled_error_repr_is_cancelled(self):
result = _classify_provider_error("<CancelledError>", None, silent_failure=True)
assert result["type"] == "cancelled"
assert result["label"] == "Task cancelled"
assert "provider returned no content" not in result.get("hint", "").lower()
def test_interrupted_or_aborted_error_is_not_provider_no_response(self):
for text in (
"Interrupted by user",
"Operation aborted before provider response completed",
"AbortError: request was aborted",
):
result = _classify_provider_error(text, RuntimeError(text))
assert result["type"] == "interrupted", text
assert result["label"] == "Response interrupted", text
assert "provider returned no content" not in result.get("hint", "").lower()
def test_provider_empty_response_still_uses_no_response(self):
result = _classify_provider_error("", None, silent_failure=True)
assert result["type"] == "no_response"
assert result["label"] == "No response from provider"
assert "provider returned no content" in result.get("hint", "").lower()
class TestCancelledTurnFinalizer:
def test_persistent_cancel_finalizer_clears_pending_and_saves_cancel_marker(self):
session = _DummySession()
_finalize_cancelled_turn(session, ephemeral=False)
assert session.active_stream_id is None
assert session.pending_user_message is None
assert session.pending_attachments == []
assert session.pending_started_at is None
assert session.saved == 1
assert session.messages[-1]['content'] == _cancelled_turn_content('Task cancelled.')
assert '**Task cancelled:** Task cancelled.' in session.messages[-1]['content']
assert 'No provider failure occurred' in session.messages[-1]['content']
assert session.messages[-1]['provider_details'] == 'Task cancelled.'
assert session.messages[-1]['provider_details_label'] == 'Cancellation details'
assert session.messages[-1]['_error'] is True
def test_ephemeral_cancel_finalizer_unlinks_temp_session_without_saving_error_marker(self, tmp_path):
temp_session = tmp_path / 'btw-session.json'
temp_session.write_text('{}', encoding='utf-8')
session = _DummySession(str(temp_session))
_finalize_cancelled_turn(session, ephemeral=True)
assert session.active_stream_id is None
assert session.pending_user_message is None
assert session.pending_attachments == []
assert session.pending_started_at is None
assert session.saved == 0
assert session.messages == []
assert not temp_session.exists()
def test_message_renderer_allows_non_provider_details_label(self):
src = _read("static/ui.js")
assert "provider_details_label||'Provider details'" in src
assert "provider-error-details" in src
class TestCancelledTurnPersistenceGuards:
def test_cancel_marker_patterns_are_centralized_for_dedupe(self):
assert _CANCEL_MARKER_PATTERNS == ('task cancelled', 'task canceled', 'response interrupted')
src = _read("api/streaming.py")
assert "any(pattern in normalized for pattern in _CANCEL_MARKER_PATTERNS)" in src
assert "any(pattern in _content for pattern in _CANCEL_MARKER_PATTERNS)" in src
def test_silent_failure_path_checks_cancel_event_before_persisting_provider_error(self):
src = _read("api/streaming.py")
silent_idx = src.find("# ── Detect silent agent failure")
assert silent_idx != -1, "silent-failure block not found"
apperror_idx = src.find("put('apperror', _error_payload)", silent_idx)
assert apperror_idx != -1, "silent-failure apperror emission not found"
block = src[silent_idx:apperror_idx]
assert "cancel_event.is_set()" in block, (
"When a user cancels and the interrupted agent returns no assistant text, "
"the silent-failure path must not persist a provider no_response error."
)
assert "cancelled" in block.lower(), (
"The cancellation guard should persist/report a cancelled turn, not silently drop state."
)
def test_exception_path_classifies_after_cancel_event_before_generic_error(self):
src = _read("api/streaming.py")
except_idx = src.find("print('[webui] stream error:")
assert except_idx != -1, "stream exception handler not found"
classify_idx = src.find("_classify_provider_error", except_idx)
generic_idx = src.find("_exc_label, _exc_type, _exc_hint = 'Error', 'error', ''", except_idx)
assert classify_idx != -1 and generic_idx != -1
block = src[except_idx:generic_idx]
assert "cancel_event.is_set()" in block, (
"Exception handling must distinguish user-cancelled/aborted runs before generic errors."
)
assert "cancelled" in block.lower() or "interrupted" in block.lower()
assert "provider_details_label" in src
assert "Cancellation details" in src
assert "Interruption details" in src
def test_post_run_cancel_guard_runs_before_normal_success_merge(self):
src = _read("api/streaming.py")
run_idx = src.find("result = agent.run_conversation(")
merge_idx = src.find("_result_messages = result.get", run_idx)
assert run_idx != -1 and merge_idx != -1, "run/merge path not found"
block = src[run_idx:merge_idx]
assert "cancel_event.is_set()" in block, (
"If cancellation arrives after tokens streamed but before run_conversation returns, "
"the worker must emit/persist cancel before normal merge/save/completed handling."
)
assert "put('cancel'" in block
assert "_cleanup_ephemeral_cancelled_turn" in block or "_finalize_cancelled_turn" in block, (
"Ephemeral cancels must clean up their temporary session before returning."
)
assert "return" in block
def test_frontend_has_cancelled_and_interrupted_labels_for_apperror_fallbacks(self):
src = _read("static/messages.js")
start = src.find("source.addEventListener('apperror'")
end = src.find("source.addEventListener('warning'", start)
assert start != -1 and end != -1, "apperror handler not found"
block = src[start:end]
assert "d.type==='cancelled'" in block or 'd.type==="cancelled"' in block
assert "d.type==='interrupted'" in block or 'd.type==="interrupted"' in block
assert "Task cancelled" in block
assert "Response interrupted" in block
assert "No response from provider" in block
assert "Cancellation details" in block
assert "Interruption details" in block

72
tests/test_issue1361_cancel_data_loss.py
View file

@ -442,3 +442,75 @@ def test_materialize_helper_called_immediately_before_error_path_clears():
f"found {sites_with_helper}. PR #1760 / #1361 regression — re-wire the "
f"helper at the error-branch clear sites in api/streaming.py."
)
class TestCancelStreamIdempotentWithWorkerFinalizer:
"""The worker and explicit cancel endpoint can both finalize the same turn."""
def test_cancel_stream_does_not_duplicate_existing_worker_cancel_marker(self):
sid = "test_1361_idempotent"
stream_id = "stream_idempotent"
_make_session(
session_id=sid,
messages=[
{'role': 'user', 'content': 'Help me debug this', 'timestamp': 100},
{'role': 'assistant', 'content': '**Task cancelled:** Task cancelled.\n\n*The run was cancelled by the user before Skyly finished. No provider failure occurred.*', '_error': True, 'timestamp': 101},
],
)
_setup_cancel_state(sid, stream_id)
config.STREAM_PARTIAL_TEXT[stream_id] = "partial text before cancel"
cancel_stream(stream_id)
msgs = models.SESSIONS[sid].messages
cancel_markers = [
m for m in msgs
if isinstance(m, dict)
and m.get('role') == 'assistant'
and 'task cancelled' in str(m.get('content') or '').lower()
]
partial_idx = next(
i for i, m in enumerate(msgs)
if isinstance(m, dict) and m.get('_partial') and m.get('content') == 'partial text before cancel'
)
marker_idx = next(i for i, m in enumerate(msgs) if m in cancel_markers)
assert len(cancel_markers) == 1
assert partial_idx < marker_idx
def test_late_cancel_after_worker_finalized_does_not_add_cancel_marker(self):
sid = "test_1361_late_done"
stream_id = "stream_late_done"
s = Session(
session_id=sid,
title="Done Session",
messages=[
{'role': 'user', 'content': 'finish normally', 'timestamp': 100},
{'role': 'assistant', 'content': 'done normally', 'timestamp': 101},
],
)
s.active_stream_id = None
s.pending_user_message = None
s.pending_attachments = []
s.pending_started_at = None
s.save()
models.SESSIONS[sid] = s
q = queue.Queue()
config.STREAMS[stream_id] = q
config.CANCEL_FLAGS[stream_id] = threading.Event()
mock_agent = Mock()
mock_agent.session_id = sid
mock_agent.interrupt = Mock()
config.AGENT_INSTANCES[stream_id] = mock_agent
config.STREAM_PARTIAL_TEXT[stream_id] = 'stale partial snapshot'
assert cancel_stream(stream_id) is True
msgs = models.SESSIONS[sid].messages
assert msgs == [
{'role': 'user', 'content': 'finish normally', 'timestamp': 100},
{'role': 'assistant', 'content': 'done normally', 'timestamp': 101},
]
assert q.empty(), "late cancel must not emit a terminal cancel event after done"

9
tests/test_issue893_cancel_preserves_partial.py
View file

@ -3,7 +3,7 @@ Regression tests for #893 — cancel_stream() now preserves partial streamed
assistant content rather than discarding it.
Before this fix, clicking Stop Generation threw away all streamed text. The
session was saved with only '*Task cancelled.*' appended, so the user lost
session was saved with only a cancellation marker appended, so the user lost
whatever the agent had produced up to that point.
After this fix:
@ -118,7 +118,7 @@ class TestCancelStreamPreservesPartial:
assert any('Python is a high-level programming language' in c for c in msg_contents), (
f"Partial text not found in session messages: {msg_contents}"
)
assert any('*Task cancelled.*' in c for c in msg_contents), (
assert any('Task cancelled:' in c for c in msg_contents), (
"Cancel marker missing from session messages"
)
# Partial message should NOT have _error=True (it's real content)
@ -127,8 +127,9 @@ class TestCancelStreamPreservesPartial:
assert partial_msg.get('_partial') is True
assert not partial_msg.get('_error')
# Cancel marker should have _error=True
cancel_msg = next(m for m in saved.messages if '*Task cancelled.*' in m.get('content', ''))
cancel_msg = next(m for m in saved.messages if 'Task cancelled:' in m.get('content', ''))
assert cancel_msg.get('_error') is True
assert cancel_msg.get('provider_details_label') == 'Cancellation details'
def test_cancel_stream_with_no_partial_text_still_saves_cancel_marker(self, tmp_path, monkeypatch):
"""If no tokens were streamed before cancel, only the cancel marker is saved."""
@ -168,7 +169,7 @@ class TestCancelStreamPreservesPartial:
saved = Session.load('sess_nopartial')
msg_contents = [m.get('content', '') for m in saved.messages]
assert any('*Task cancelled.*' in c for c in msg_contents)
assert any('Task cancelled:' in c for c in msg_contents)
# No extra partial message when there was nothing streamed
assert not any(m.get('_partial') for m in saved.messages), (
"Should not add partial message when no tokens were streamed"

7
tests/test_pr1341_context_window_persistence.py
View file

@ -38,11 +38,12 @@ def test_streaming_persists_context_fields_on_session_before_save():
# Save call follows shortly after
save_call = src.find("\n s.save()", block_start)
assert save_call != -1, "s.save() not found after the post-merge marker"
# Limit bumped to 8200 by turn-journal lifecycle events: the block now also
# records `assistant_started` immediately before the durable final save.
# Limit bumped to 9000 by cancellation finalization guards: the block now also
# checks for a late user cancel immediately before the durable final save,
# preventing a race that would otherwise save/emit a completed turn after Stop.
# The context_length fallback is still a single focused resolver call with
# arg-prep scaffold and commentary explaining the failure mode it prevents.
assert save_call - block_start < 8200, (
assert save_call - block_start < 9000, (
"s.save() should be close to the post-merge marker — block expanded unexpectedly. "
"If you've added a new pre-save mutation block here, bump this limit."
)

8
tests/test_sprint36.py
View file

@ -212,16 +212,14 @@ def test_cancel_marker_flagged_as_error_to_skip_in_api_history():
_error: True so _sanitize_messages_for_api() strips it from the
conversation_history sent to the agent on the next user message.
Without this flag, the LLM sees "*Task cancelled.*" as a prior assistant
Without this flag, the LLM sees "Task cancelled" as a prior assistant
turn and may reference it in subsequent responses ("As I mentioned, I was
cancelled...") — a behavioral regression introduced when this PR started
persisting the marker to the session.
"""
src = read("api/streaming.py")
idx = src.find("'content': '*Task cancelled.*'")
if idx == -1:
idx = src.find('"content": "*Task cancelled.*"')
assert idx != -1, "cancel marker content string not found in cancel_stream()"
idx = src.find("'content': _cancelled_turn_content(message)")
assert idx != -1, "cancel marker content writer not found in cancel_stream()"
# Walk back to the start of the dict literal (opening brace)
brace_open = src.rfind("{", 0, idx)