Preserve messaging transcript on stop

ARCHITECTURE.md

@@ -249,10 +249,14 @@ provider, preflights the upstream request, emits trace events, counts input
 tokens, and returns an Anthropic SSE iterator.
 [api/response_streams.py](api/response_streams.py) owns public streaming egress
 commit timing. It waits for the first protocol chunk before returning a
-successful `StreamingResponse`, so provider setup failures can still become real
-non-200 JSON errors that Claude Code and Codex can retry. After the first chunk
-has escaped, HTTP status is committed; any unexpected failure must be represented
-as a protocol terminal frame where feasible.
+successful `StreamingResponse`. For streaming `/v1/messages`, once FCC has
+accepted the turn and provider execution owns the request, final provider
+failures are returned as terminal Anthropic SSE error events rather than
+retryable HTTP 429/5xx responses. HTTP error responses remain for ingress,
+auth, request validation, and preflight request-shape failures before provider
+execution. After the first chunk has escaped, HTTP status is committed; any
+unexpected failure must be represented as a protocol terminal frame where
+feasible.
 
 ```mermaid
 sequenceDiagram
@@ -448,6 +452,9 @@ Provider transports raise typed provider errors for final stream failures before
 any downstream-visible SSE chunk has escaped the recovery holdback. Once output
 has committed, transports keep ownership of midstream recovery, continuation,
 tool salvage, and protocol-specific success/error tails.
+The public streaming API boundary owns the final downstream error shape: provider
+errors may be visible to clients, but after FCC exhausts provider retry/recovery
+they must not leak as retryable HTTP statuses for accepted streaming turns.
 
 [core/openai_responses/](core/openai_responses/) owns OpenAI Responses support:
 

api/handlers/messages.py

@@ -3,7 +3,7 @@
 from collections.abc import AsyncIterator, Callable
 from dataclasses import dataclass, replace
 
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse, Response
 from loguru import logger
 
 from api.detection import is_safety_classifier_request
@@ -19,6 +19,7 @@ from api.request_errors import (
 )
 from api.response_streams import (
     EmptyStreamError,
+    anthropic_sse_error_response,
     anthropic_sse_streaming_response,
 )
 from api.web_tools.egress import WebFetchEgressPolicy, web_fetch_allowed_scheme_set
@@ -139,11 +140,18 @@ class MessagesHandler:
             pre_start_error_response=self._pre_start_error_response,
         )
 
-    def _pre_start_error_response(self, exc: BaseException) -> JSONResponse:
+    def _pre_start_error_response(self, exc: BaseException) -> Response:
         if isinstance(exc, ProviderError):
-            return JSONResponse(
+            trace_event(
+                stage="egress",
+                event="api.response.provider_error_terminalized",
+                source="api",
                 status_code=exc.status_code,
-                content=exc.to_anthropic_format(),
+                error_type=exc.error_type,
+            )
+            return anthropic_sse_error_response(
+                error_type=exc.error_type,
+                message=exc.message,
             )
         log_unexpected_api_exception(
             self._settings,
@@ -154,15 +162,16 @@ class MessagesHandler:
                 else "CREATE_MESSAGE_STREAM_START_ERROR"
             ),
         )
-        return JSONResponse(
+        trace_event(
+            stage="egress",
+            event="api.response.stream_start_error_terminalized",
+            source="api",
+            exc_type=type(exc).__name__,
             status_code=http_status_for_unexpected_api_exception(exc),
-            content={
-                "type": "error",
-                "error": {
-                    "type": "api_error",
-                    "message": _unexpected_stream_error_message(exc),
-                },
-            },
+        )
+        return anthropic_sse_error_response(
+            error_type="api_error",
+            message=_unexpected_stream_error_message(exc),
         )
 
     def _reject_unsupported_server_tools(self, routed: RoutedMessagesRequest) -> None:

api/provider_execution.py

@@ -81,13 +81,18 @@ class ProviderExecutionService:
             routed.request.system,
             routed.request.tools,
         )
-        return traced_async_stream(
-            provider.stream_response(
+
+        async def provider_body() -> AsyncIterator[str]:
+            async for chunk in provider.stream_response(
                 routed.request,
                 input_tokens=input_tokens,
                 request_id=request_id,
                 thinking_enabled=routed.resolved.thinking_enabled,
-            ),
+            ):
+                yield chunk
+
+        return traced_async_stream(
+            provider_body(),
             stage="egress",
             source="api",
             complete_event=(

api/response_streams.py

@@ -24,6 +24,21 @@ class EmptyStreamError(RuntimeError):
     """Raised when a public stream ends before emitting any protocol chunk."""
 
 
+async def _single_chunk_body(chunk: str) -> AsyncGenerator[str]:
+    yield chunk
+
+
+def anthropic_sse_error_response(*, error_type: str, message: str) -> StreamingResponse:
+    """Return a committed Anthropic SSE stream containing one terminal error."""
+    return StreamingResponse(
+        _single_chunk_body(
+            anthropic_terminal_error_frame(message, error_type=error_type)
+        ),
+        media_type="text/event-stream",
+        headers=dict(ANTHROPIC_SSE_RESPONSE_HEADERS),
+    )
+
+
 def _trace_egress_failure(exc: BaseException) -> None:
     trace_event(
         stage="egress",

core/anthropic/streaming/emitter.py

@@ -31,11 +31,13 @@ def format_sse_event(event_type: str, data: dict[str, Any]) -> str:
     return f"event: {event_type}\ndata: {json.dumps(data)}\n\n"
 
 
-def anthropic_terminal_error_frame(message: str) -> str:
+def anthropic_terminal_error_frame(
+    message: str, *, error_type: str = "api_error"
+) -> str:
     """Serialize a terminal Anthropic SSE error event for egress failures."""
     return format_sse_event(
         "error",
-        {"type": "error", "error": {"type": "api_error", "message": message}},
+        {"type": "error", "error": {"type": error_type, "message": message}},
     )
 
 

messaging/command_context.py

@@ -6,7 +6,7 @@ from .managed_protocols import ManagedClaudeSessionManagerProtocol
 from .platforms.ports import OutboundMessenger, VoiceCancellation
 from .session import SessionStore
 from .transcript import RenderCtx
-from .trees import MessageNode, MessageTree, TreeQueueManager
+from .trees import CancelledNode, MessageTree, TreeQueueManager
 
 
 class MessagingCommandContext(Protocol):
@@ -58,7 +58,7 @@ class MessagingCommandContext(Protocol):
         """Persist an outgoing platform message ID."""
         ...
 
-    def update_cancelled_nodes_ui(self, nodes: list[MessageNode]) -> None:
+    def update_cancelled_nodes_ui(self, nodes: list[CancelledNode]) -> None:
         """Render cancellation status and persist affected trees."""
         ...
 

messaging/event_parser.py

@@ -154,26 +154,29 @@ def parse_cli_event(event: Any, *, log_raw_cli: bool = False) -> list[dict]:
         if code == 0:
             logger.debug(f"CLI_PARSER: Successful exit (code={code})")
             return [{"type": "complete", "status": "success"}]
+
+        error_msg = stderr if stderr else f"Process exited with code {code}"
+        if log_raw_cli:
+            logger.warning(
+                "CLI_PARSER: Error exit (code={}): {}",
+                code,
+                error_msg,
+            )
         else:
-            # Non-zero exit is an error
-            error_msg = stderr if stderr else f"Process exited with code {code}"
-            if log_raw_cli:
-                logger.warning(
-                    "CLI_PARSER: Error exit (code={}): {}",
-                    code,
-                    error_msg,
-                )
-            else:
-                em = error_msg if isinstance(error_msg, str) else str(error_msg)
-                logger.warning(
-                    "CLI_PARSER: Error exit (code={}): message_chars={}",
-                    code,
-                    len(em),
-                )
-            return [
-                {"type": "error", "message": error_msg},
-                {"type": "complete", "status": "failed"},
-            ]
+            em = error_msg if isinstance(error_msg, str) else str(error_msg)
+            logger.warning(
+                "CLI_PARSER: Error exit (code={}): message_chars={}",
+                code,
+                len(em),
+            )
+        return [
+            {
+                "type": "error",
+                "message": error_msg,
+                "source": "exit",
+                "exit_code": code,
+            }
+        ]
 
     # Log unrecognized events for debugging
     if etype:

messaging/node_event_pipeline.py

@@ -88,6 +88,8 @@ async def process_parsed_cli_event(
     elif ptype == "block_stop":
         await update_ui(last_status, force=True)
     elif ptype == "complete":
+        if parsed.get("status") != "success":
+            return last_status, had_transcript_events
         if not had_transcript_events:
             transcript.apply({"type": "text_chunk", "text": "Done."})
         trace_event(

messaging/node_runner.py

@@ -15,7 +15,14 @@ from .platforms.ports import OutboundMessenger
 from .safe_diagnostics import format_exception_for_log
 from .session import SessionStore
 from .transcript import RenderCtx, TranscriptBuffer
-from .trees import MessageNode, MessageState, MessageTree, TreeQueueManager
+from .trees import (
+    CancellationReason,
+    MessageNode,
+    MessageState,
+    MessageTree,
+    TreeQueueManager,
+    get_cancel_reason,
+)
 from .ui_updates import ThrottledTranscriptEditor
 
 
@@ -107,6 +114,7 @@ class MessagingNodeRunner:
         transcript, render_ctx = self._create_transcript_and_render_ctx()
 
         had_transcript_events = False
+        had_non_exit_error = False
         captured_session_id = None
         temp_session_id = None
         last_status: str | None = None
@@ -238,6 +246,14 @@ class MessagingNodeRunner:
                 )
 
                 for parsed in parsed_list:
+                    ptype = parsed.get("type")
+                    if (
+                        ptype == "error"
+                        and parsed.get("source") == "exit"
+                        and had_non_exit_error
+                    ):
+                        continue
+
                     (
                         last_status,
                         had_transcript_events,
@@ -258,6 +274,8 @@ class MessagingNodeRunner:
                         propagate_error_to_children=self.propagate_error_to_children,
                         log_messaging_error_details=self._log_messaging_error_details,
                     )
+                    if ptype == "error" and parsed.get("source") != "exit":
+                        had_non_exit_error = True
 
         except asyncio.CancelledError:
             trace_event(
@@ -268,11 +286,7 @@ class MessagingNodeRunner:
                 node_id=node_id,
             )
             logger.warning(f"HANDLER: Task cancelled for node {node_id}")
-            cancel_reason = None
-            if isinstance(node.context, dict):
-                cancel_reason = node.context.get("cancel_reason")
-
-            if cancel_reason == "stop":
+            if get_cancel_reason(node) is CancellationReason.STOP:
                 await update_ui(self._format_status("⏹", "Stopped.", None), force=True)
             else:
                 transcript.apply({"type": "error", "message": "Task was cancelled"})

messaging/trees/__init__.py

@@ -1,5 +1,12 @@
 """Message tree data structures and queue management."""
 
+from .cancellation import (
+    CancellationReason,
+    CancellationUiOwner,
+    CancelledNode,
+    get_cancel_reason,
+    set_cancel_reason,
+)
 from .manager import TreeQueueManager
 from .node import MessageNode, MessageState
 from .processor import TreeQueueProcessor
@@ -8,6 +15,9 @@ from .runtime import MessageTree
 from .snapshot import ConversationSnapshot, TreeSnapshot
 
 __all__ = [
+    "CancellationReason",
+    "CancellationUiOwner",
+    "CancelledNode",
     "ConversationSnapshot",
     "MessageNode",
     "MessageState",
@@ -16,4 +26,6 @@ __all__ = [
     "TreeQueueProcessor",
     "TreeRepository",
     "TreeSnapshot",
+    "get_cancel_reason",
+    "set_cancel_reason",
 ]

messaging/trees/cancellation.py

@@ -0,0 +1,52 @@
+"""Typed cancellation facts for messaging tree UI ownership."""
+
+from dataclasses import dataclass
+from enum import Enum
+
+from .node import MessageNode
+
+_CANCEL_REASON_CONTEXT_KEY = "cancel_reason"
+
+
+class CancellationUiOwner(Enum):
+    """Who owns the final user-visible cancellation edit for a node."""
+
+    RUNNER = "runner"
+    WORKFLOW = "workflow"
+
+
+class CancellationReason(Enum):
+    """Why a node was cancelled, when the runner needs UI-specific cleanup."""
+
+    STOP = "stop"
+
+
+@dataclass(frozen=True)
+class CancelledNode:
+    """A cancelled node plus the component responsible for its final UI edit."""
+
+    node: MessageNode
+    ui_owner: CancellationUiOwner
+
+
+def set_cancel_reason(node: MessageNode, reason: CancellationReason | None) -> None:
+    """Attach or remove cancellation reason without clobbering other context."""
+    context = dict(node.context) if isinstance(node.context, dict) else {}
+    if reason is None:
+        context.pop(_CANCEL_REASON_CONTEXT_KEY, None)
+    else:
+        context[_CANCEL_REASON_CONTEXT_KEY] = reason.value
+    node.set_context(context or None)
+
+
+def get_cancel_reason(node: MessageNode) -> CancellationReason | None:
+    """Return a typed cancellation reason from node context."""
+    if not isinstance(node.context, dict):
+        return None
+    raw_reason = node.context.get(_CANCEL_REASON_CONTEXT_KEY)
+    if not isinstance(raw_reason, str):
+        return None
+    try:
+        return CancellationReason(raw_reason)
+    except ValueError:
+        return None

messaging/trees/manager.py

@@ -6,6 +6,12 @@ from collections.abc import Awaitable, Callable
 from loguru import logger
 
 from ..models import IncomingMessage
+from .cancellation import (
+    CancellationReason,
+    CancellationUiOwner,
+    CancelledNode,
+    set_cancel_reason,
+)
 from .node import MessageNode, MessageState
 from .processor import TreeQueueProcessor
 from .repository import TreeRepository
@@ -235,7 +241,9 @@ class TreeQueueManager:
 
         return affected
 
-    async def cancel_tree(self, root_id: str) -> list[MessageNode]:
+    async def cancel_tree(
+        self, root_id: str, *, reason: CancellationReason | None = None
+    ) -> list[CancelledNode]:
         """
         Cancel all queued and in-progress messages in a tree.
 
@@ -246,34 +254,46 @@ class TreeQueueManager:
         if not tree:
             return []
 
-        cancelled_nodes = []
+        cancelled_nodes: list[CancelledNode] = []
         cancelled_tasks: list[asyncio.Task] = []
 
         cleanup_count = 0
         async with tree.with_lock():
+            current_id = tree.current_node_id
+            current_node = tree.get_node(current_id) if current_id else None
+            if current_node is not None:
+                set_cancel_reason(current_node, reason)
             cancelled_task = tree.cancel_current_task()
             if cancelled_task:
                 cancelled_tasks.append(cancelled_task)
-                current_id = tree.current_node_id
-                if current_id:
-                    node = tree.get_node(current_id)
-                    if node and node.state not in (
-                        MessageState.COMPLETED,
-                        MessageState.ERROR,
-                    ):
-                        tree.set_node_error_sync(node, "Cancelled by user")
-                        cancelled_nodes.append(node)
+                if current_node and current_node.state not in (
+                    MessageState.COMPLETED,
+                    MessageState.ERROR,
+                ):
+                    tree.set_node_error_sync(current_node, "Cancelled by user")
+                    cancelled_nodes.append(
+                        CancelledNode(current_node, CancellationUiOwner.RUNNER)
+                    )
 
             queue_nodes = tree.drain_queue_and_mark_cancelled()
-            cancelled_nodes.extend(queue_nodes)
-            cancelled_ids = {n.node_id for n in cancelled_nodes}
+            for node in queue_nodes:
+                set_cancel_reason(node, reason)
+            cancelled_nodes.extend(
+                CancelledNode(node, CancellationUiOwner.WORKFLOW)
+                for node in queue_nodes
+            )
+            cancelled_ids = {entry.node.node_id for entry in cancelled_nodes}
 
             for node in tree.all_nodes():
                 if (
                     node.state in (MessageState.PENDING, MessageState.IN_PROGRESS)
                     and node.node_id not in cancelled_ids
                 ):
+                    set_cancel_reason(node, reason)
                     tree.set_node_error_sync(node, "Stale task cleaned up")
+                    cancelled_nodes.append(
+                        CancelledNode(node, CancellationUiOwner.WORKFLOW)
+                    )
                     cleanup_count += 1
 
             tree.reset_processing_state()
@@ -289,7 +309,9 @@ class TreeQueueManager:
 
         return cancelled_nodes
 
-    async def cancel_node(self, node_id: str) -> list[MessageNode]:
+    async def cancel_node(
+        self, node_id: str, *, reason: CancellationReason | None = None
+    ) -> list[CancelledNode]:
         """
         Cancel a single node (queued or in-progress) without affecting other nodes.
 
@@ -308,11 +330,14 @@ class TreeQueueManager:
             if node.state in (MessageState.COMPLETED, MessageState.ERROR):
                 return []
 
+            set_cancel_reason(node, reason)
             cancelled_tasks: list[asyncio.Task] = []
+            ui_owner = CancellationUiOwner.WORKFLOW
             if tree.is_current_node(node_id):
                 cancelled_task = self._processor.cancel_current(tree)
                 if cancelled_task:
                     cancelled_tasks.append(cancelled_task)
+                    ui_owner = CancellationUiOwner.RUNNER
 
             removed_from_queue = False
             try:
@@ -328,15 +353,17 @@ class TreeQueueManager:
             await self._processor.notify_queue_updated(tree)
         await _drain_cancelled_tasks(cancelled_tasks)
 
-        return [node]
+        return [CancelledNode(node, ui_owner)]
 
-    async def cancel_all(self) -> list[MessageNode]:
+    async def cancel_all(
+        self, *, reason: CancellationReason | None = None
+    ) -> list[CancelledNode]:
         """Cancel all messages in all trees."""
         async with self._lock:
             root_ids = list(self._repository.tree_ids())
-        all_cancelled: list[MessageNode] = []
+        all_cancelled: list[CancelledNode] = []
         for root_id in root_ids:
-            all_cancelled.extend(await self.cancel_tree(root_id))
+            all_cancelled.extend(await self.cancel_tree(root_id, reason=reason))
         return all_cancelled
 
     def cleanup_stale_nodes(self) -> int:
@@ -376,7 +403,9 @@ class TreeQueueManager:
         """Register a node ID to a tree (for external mapping)."""
         self._repository.register_node(node_id, root_id)
 
-    async def cancel_branch(self, branch_root_id: str) -> list[MessageNode]:
+    async def cancel_branch(
+        self, branch_root_id: str, *, reason: CancellationReason | None = None
+    ) -> list[CancelledNode]:
         """
         Cancel all PENDING/IN_PROGRESS nodes in the subtree (branch_root + descendants).
         """
@@ -385,7 +414,7 @@ class TreeQueueManager:
             return []
 
         branch_ids = set(tree.get_descendants(branch_root_id))
-        cancelled: list[MessageNode] = []
+        cancelled: list[CancelledNode] = []
         cancelled_tasks: list[asyncio.Task] = []
         removed_from_queue = False
 
@@ -399,17 +428,21 @@ class TreeQueueManager:
                     continue
 
                 if tree.is_current_node(nid):
+                    set_cancel_reason(node, reason)
                     cancelled_task = self._processor.cancel_current(tree)
+                    ui_owner = CancellationUiOwner.WORKFLOW
                     if cancelled_task:
                         cancelled_tasks.append(cancelled_task)
+                        ui_owner = CancellationUiOwner.RUNNER
                     tree.set_node_error_sync(node, "Cancelled by user")
-                    cancelled.append(node)
+                    cancelled.append(CancelledNode(node, ui_owner))
                 else:
+                    set_cancel_reason(node, reason)
                     removed_from_queue = (
                         tree.remove_from_queue(nid) or removed_from_queue
                     )
                     tree.set_node_error_sync(node, "Cancelled by user")
-                    cancelled.append(node)
+                    cancelled.append(CancelledNode(node, CancellationUiOwner.WORKFLOW))
 
         if cancelled:
             logger.info(f"Cancelled {len(cancelled)} nodes in branch {branch_root_id}")
@@ -440,7 +473,7 @@ class TreeQueueManager:
             removed_tree = self._repository.remove_tree(root_id)
             if removed_tree:
                 return (removed_tree.all_nodes(), root_id, True)
-            return (cancelled, root_id, True)
+            return ([entry.node for entry in cancelled], root_id, True)
 
         async with tree.with_lock():
             removed = tree.remove_branch(branch_root_id)

messaging/trees/node.py

@@ -46,8 +46,10 @@ class MessageNode:
         self.state = state
         if session_id:
             self.session_id = session_id
-        if error_message:
+        if error_message is not None:
             self.error_message = error_message
+        elif state == MessageState.COMPLETED:
+            self.error_message = None
         if state in (MessageState.COMPLETED, MessageState.ERROR):
             self.completed_at = datetime.now(UTC)
 

messaging/workflow.py

@@ -12,7 +12,13 @@ from .rendering.profiles import build_rendering_profile
 from .safe_diagnostics import format_exception_for_log
 from .session import SessionStore
 from .transcript import RenderCtx
-from .trees import MessageNode, MessageState, MessageTree, TreeQueueManager
+from .trees import (
+    CancellationReason,
+    CancellationUiOwner,
+    CancelledNode,
+    MessageTree,
+    TreeQueueManager,
+)
 from .turn_intake import MessagingTurnIntake
 
 
@@ -139,7 +145,9 @@ class MessagingWorkflow:
         Stop all pending and in-progress messaging tasks.
         """
         logger.info("Cancelling tree queue tasks...")
-        cancelled_nodes = await self.tree_queue.cancel_all()
+        cancelled_nodes = await self.tree_queue.cancel_all(
+            reason=CancellationReason.STOP
+        )
         logger.info(f"Cancelled {len(cancelled_nodes)} nodes")
 
         logger.info("Stopping all CLI sessions...")
@@ -150,13 +158,10 @@ class MessagingWorkflow:
 
     async def stop_task(self, node_id: str) -> int:
         """Stop a single queued or in-progress task node."""
-        tree = self.tree_queue.get_tree_for_node(node_id)
-        if tree:
-            node = tree.get_node(node_id)
-            if node and node.state not in (MessageState.COMPLETED, MessageState.ERROR):
-                node.set_context({"cancel_reason": "stop"})
-
-        cancelled_nodes = await self.tree_queue.cancel_node(node_id)
+        cancelled_nodes = await self.tree_queue.cancel_node(
+            node_id,
+            reason=CancellationReason.STOP,
+        )
         self.update_cancelled_nodes_ui(cancelled_nodes)
         return len(cancelled_nodes)
 
@@ -182,18 +187,20 @@ class MessagingWorkflow:
                 ),
             )
 
-    def update_cancelled_nodes_ui(self, nodes: list[MessageNode]) -> None:
+    def update_cancelled_nodes_ui(self, nodes: list[CancelledNode]) -> None:
         """Update status messages and persist tree state for cancelled nodes."""
         trees_to_save: dict[str, MessageTree] = {}
-        for node in nodes:
-            self.outbound.fire_and_forget(
-                self.outbound.queue_edit_message(
-                    node.incoming.chat_id,
-                    node.status_message_id,
-                    self.format_status("⏹", "Stopped."),
-                    parse_mode=self._parse_mode(),
+        for cancelled in nodes:
+            node = cancelled.node
+            if cancelled.ui_owner is CancellationUiOwner.WORKFLOW:
+                self.outbound.fire_and_forget(
+                    self.outbound.queue_edit_message(
+                        node.incoming.chat_id,
+                        node.status_message_id,
+                        self.format_status("⏹", "Stopped."),
+                        parse_mode=self._parse_mode(),
+                    )
                 )
-            )
             tree = self.tree_queue.get_tree_for_node(node.node_id)
             if tree:
                 trees_to_save[tree.root_id] = tree

plan.md

@@ -1,299 +1,615 @@
-# First-Frame Gated Streaming PR Plan
+# Messaging Stop And Provider Error Finalization Plan
 
 ## Summary
 
-Fix issue #1020 at the real boundary: FCC currently commits downstream
-`HTTP 200` as soon as FastAPI receives a `StreamingResponse`, before the
-provider-backed stream has proven it can emit a valid first SSE chunk. If the
-provider fails before that first chunk, Claude Code sees `200` with an empty or
-malformed stream instead of a non-200 response it can retry.
+Fix the three valid defects found while testing PR #1027:
 
-The PR should introduce an API egress first-frame gate and align provider
-transports so pre-start failures raise typed HTTP-mappable errors instead of
-being converted into synthetic successful SSE streams. Once the first chunk has
-escaped, HTTP status can no longer change, so post-start failures still need a
-protocol-correct terminal stream frame.
+1. Standalone `/stop` can preserve the transcript but still render the wrong final footer because active runner-owned nodes do not receive stop intent.
+2. `MessagingNodeRunner` suppresses successful completion after any earlier error event, instead of suppressing only duplicate process-exit errors.
+3. Exhausted provider failures, especially upstream 429s, are returned to Claude Code as retryable HTTP statuses. Claude Code then retries the whole `/v1/messages` turn outside FCC's retry policy, leaving messaging stuck on `Continuing conversation...`.
 
-Because this changes production API/provider behavior on `main`, bump the
-current patch version from `3.4.12` to `3.4.13` unless `main` advances first,
-then refresh `uv.lock`.
+Corrected error policy:
 
-## Customer-Facing Contract
+- Provider errors may be visible to Claude Code and the user.
+- Provider error diagnostics should be preserved: error type, message, upstream detail, and request id.
+- Provider execution failures must not be returned to Claude Code in a retryable HTTP shape after FCC has accepted a streaming `/v1/messages` turn.
+- FCC owns upstream retry and recovery. Claude Code should receive only the final terminal outcome.
 
-- `fcc-server` should not return `HTTP 200` for `/v1/messages` or
-  `/v1/responses` until the stream can produce its first protocol chunk.
-- Claude Code should receive a non-200 Anthropic-shaped error when upstream
-  provider setup/retry fails before any stream output is viable.
-- Codex should receive a non-200 OpenAI-shaped error when the Responses stream
-  fails before `response.created`.
-- After streaming has started, clients should receive a parseable terminal
-  protocol frame instead of a truncated connection where feasible.
-- Provider retries, midstream recovery, tool salvage, thinking/reasoning, tool
-  calls, local web tools, non-streaming `/v1/messages`, and messaging behavior
-  should remain unchanged except for the pre-start HTTP status fix.
+The customer-facing contract remains:
+
+- Telegram and Discord `/stop` preserve already-rendered transcript content and append a stopped footer.
+- Managed Claude Code turns do not remain indefinitely `in_progress` after FCC has already exhausted provider retries.
+- Claude Code still sees useful provider failures.
+- Provider retry counts, recovery prompts, transport semantics, and request bodies remain unchanged.
+- Codex `/v1/responses` is not the triggering bug for this PR, but the plan documents the same retry-ownership principle so the API boundary does not drift.
+
+Because this changes production messaging/API behavior on the PR branch, keep the existing PR version bump unless this work is split after `3.4.14` lands. If split onto a later `main`, bump patch and run `uv lock`.
+
+## Explored Code Paths
+
+### Streaming Messages Boundary
+
+`MessagesHandler.create()` resolves and preflights the request, then returns an async provider SSE body through `anthropic_sse_streaming_response()`.
+
+Important boundary:
+
+- Synchronous failures before the body iterator is returned are ingress/preflight failures.
+- Exceptions raised while pulling the first chunk from the body iterator are provider-execution or stream-conversion failures after FCC has started owning the turn.
+
+Current behavior:
+
+- `_first_chunk_streaming_response()` waits for the first chunk.
+- If the provider body raises before the first chunk, it calls `MessagesHandler._pre_start_error_response()`.
+- `_pre_start_error_response()` currently maps `ProviderError` to its HTTP status.
+- For `RateLimitError`, Claude Code receives HTTP 429 and retries the whole request.
+
+That is the live stuck-turn path seen in logs:
+
+```text
+POST /v1/messages?beta=true HTTP/1.1" 429 Too Many Requests
+POST /v1/messages?beta=true HTTP/1.1" 429 Too Many Requests
+...
+```
+
+### Provider Retry Boundary
+
+OpenAI-chat and native Anthropic transports already retry internally:
+
+- pre-response retry is owned by `providers.rate_limit.GlobalRateLimiter`;
+- early/midstream recovery is owned by `core.anthropic.streaming.RecoveryController`;
+- providers raise `map_stream_start_error(...)` only when no downstream-visible SSE has escaped.
+
+So when the API layer sees a final `ProviderError` from first-chunk probing, FCC has already decided provider retry/recovery is exhausted.
+
+Returning HTTP 429/5xx at that point leaks retry ownership back to Claude Code.
+
+### Managed Claude Boundary
+
+Managed messaging runs Claude Code as:
+
+```text
+claude --model opus -p ... --output-format stream-json --verbose
+```
+
+Messaging waits on Claude Code stdout. If Claude Code is retrying the HTTP request internally, messaging has no terminal stdout event to process, so the status remains `Continuing conversation...`.
+
+The fix should not make messaging guess about retry state. The API should return a terminal client-visible result that makes Claude Code stop the turn.
+
+### Codex Responses Boundary
+
+`/v1/responses` already converts Anthropic `event: error` into `response.failed`.
+
+However, pre-first-chunk exceptions still go through the generic `pre_start_error_response` JSON path. That is the same class of boundary issue, but it is not the observed stuck messaging bug. This PR should keep the implementation focused unless adding a shared helper makes the Responses behavior nearly free and tests prove no extra complexity.
+
+## Root Causes
+
+### 1. Global Stop Has No Runner Stop Intent
+
+`MessagingWorkflow.stop_task()` mutates node context before cancelling one node:
+
+```python
+node.set_context({"cancel_reason": "stop"})
+```
+
+`MessagingWorkflow.stop_all_tasks()` does not do the same. It calls:
+
+```python
+cancelled_nodes = await self.tree_queue.cancel_all()
+```
+
+`TreeQueueManager.cancel_all()` delegates to `cancel_tree()`, which can classify an active node as runner-owned. The workflow then correctly skips direct status-only edits for runner-owned nodes.
+
+But the runner checks `node.context.cancel_reason`; when it is missing, it renders the generic cancellation path:
+
+```python
+transcript.apply({"type": "error", "message": "Task was cancelled"})
+await update_ui(self._format_status("Cancelled", ...), force=True)
+```
+
+So reply-scoped `/stop` can be correct while standalone `/stop` can still end with `Cancelled`.
+
+### 2. Error Dedupe Is Too Broad
+
+`MessagingNodeRunner` currently tracks `had_error_events` and drops every later `complete` event:
+
+```python
+elif ptype == "complete" and had_error_events:
+    continue
+```
+
+The real duplicate is narrower:
+
+- Claude Code emits a meaningful error event.
+- The process exits non-zero.
+- `parse_cli_event()` converts the exit into a generic error: `Process exited with code 1`.
+
+The runner should suppress only that duplicate exit-sourced error. A later successful completion is the terminal fact and should win.
+
+### 3. Retryable HTTP Status Leaks Retry Ownership To Claude Code
+
+FCC already retries upstream transient errors. After those retries are exhausted, the provider transports raise a final `ProviderError`.
+
+Current streaming `/v1/messages` behavior:
+
+- final provider `RateLimitError` -> HTTP 429 JSON;
+- final provider `OverloadedError` -> HTTP 529 JSON;
+- final provider `APIError(status_code=500..599)` -> HTTP 5xx JSON.
+
+Claude Code treats those as retryable HTTP failures. That creates a second retry loop outside FCC and outside messaging's observable event stream.
+
+The root fix is not to hide provider errors. The root fix is to make final provider errors terminal in the Claude protocol instead of retryable in HTTP.
 
 ## Grill-Me Decisions
 
-### Is this caused by NIM or FCC?
+### Decision 1: Where should stop intent live?
 
-Recommended answer: FCC owns the bug. NIM or any upstream can trigger the
-failure by returning 429/5xx/504 or closing early, but FCC currently commits
-downstream `HTTP 200` before upstream viability is known. Reproducing current
-`api.response_streams.anthropic_sse_streaming_response()` shows
-`http.response.start 200` is sent immediately, before the first delayed body
-chunk and even when the body raises before its first chunk.
+Recommended answer: cancellation intent belongs at the tree cancellation boundary.
 
-### Is a terminal SSE error frame enough?
+Why:
 
-Recommended answer: no. A terminal frame fixes only post-start truncation. It
-does not restore Claude Code's HTTP retry behavior because the response is
-still `HTTP 200`. The PR needs first-frame gating for pre-start failures plus
-terminal framing for post-start failures.
+- `TreeQueueManager` knows which nodes are active, queued, stale, runner-owned, or workflow-owned.
+- Runner-owned cleanup needs the stop reason before task cancellation reaches the runner.
+- Workflow-side context mutation works only for single-node stop; global stop needs the same behavior atomically.
 
-### Should API egress own provider retry?
+Decision:
 
-Recommended answer: no. Provider transports keep upstream retries, recovery,
-tool salvage, and provider-specific fallbacks. API egress owns only the HTTP
-commit boundary: do not commit success until there is a first chunk; after
-success is committed, close the protocol cleanly if possible.
+- Add typed cancellation intent in `messaging/trees/cancellation.py`.
+- Tree cancellation APIs accept optional `reason`.
+- Stop commands pass `CancellationReason.STOP`.
+- Tree manager attaches that reason before cancelling active tasks.
 
-### Should providers keep emitting synthetic pre-start SSE errors?
+### Decision 2: Should workflow edit active runner-owned status messages as a fallback?
 
-Recommended answer: no. A provider-side final error before downstream-visible
-output should raise a mapped `ProviderError`. Synthetic Anthropic SSE error
-tails are only appropriate when the stream has already started or provider
-state has produced output that must be closed in protocol shape.
+Recommended answer: no.
 
-### Should `/v1/responses` use a fresh assembler for egress failures?
+Why:
 
-Recommended answer: no. Responses streams are stateful. A post-start
-`response.failed` must be produced by the same `ResponsesStreamAssembler` that
-emitted `response.created`, preserving `response.id`, active output flushes,
-usage, and response metadata.
+- The current PR exists because workflow-level direct edits can erase active transcript content.
+- The runner owns `TranscriptBuffer`, render context, throttling, and final transcript-preserving edits.
+- Workflow-owned nodes have no transcript buffer, so workflow may still render simple stopped status for queued/no-run nodes.
 
-## Architecture Target
+Decision:
 
-### API Egress
+- Keep runner-owned and workflow-owned cancellation UI separate.
+- Fix missing stop intent rather than adding a second UI writer.
 
-`api/response_streams.py` should own public HTTP streaming commit timing.
+### Decision 3: How should `/clear` interact with stop intent?
 
-Add an async first-frame helper that:
+Recommended answer: `/clear` may reuse stop intent for its first cancellation phase.
 
-1. pulls the first chunk from an `AsyncIterator[str]` before constructing the
-   public `StreamingResponse`;
-2. returns a protocol JSON error response if the iterator raises before the
-   first chunk;
-3. returns a `StreamingResponse` that replays the first chunk and streams the
-   rest when the first chunk exists;
-4. wraps the post-first-chunk tail with a terminal-frame guard for unexpected
-   non-cancellation failures.
+Why:
 
-The helper should be protocol-agnostic. Protocol-specific call sites provide:
+- `/clear` stops active work before deleting chat messages.
+- If a stopped message survives long enough to render, `Stopped.` is more accurate than `Cancelled`.
+- Clear/delete remains the owner of final chat cleanup.
 
-- streaming headers;
-- pre-start JSON error envelope builder;
-- post-start terminal frame behavior.
+Decision:
 
-Do not import `core/openai_responses` internals directly into API egress. API
-handlers may use their adapter/facade objects.
+- `MessagingWorkflow.stop_all_tasks()` passes stop intent.
+- Lower-level tree cancellation defaults to no intent for internal non-stop callers.
 
-### Provider Execution
+### Decision 4: Where should duplicate exit-code errors be suppressed?
 
-`api/provider_execution.py` should keep resolving providers, preflight,
-request tracing, raw-payload logging, token counting, and `traced_async_stream`.
-It should not construct `StreamingResponse` and should not own protocol error
-serialization.
+Recommended answer: in `MessagingNodeRunner`.
 
-### Provider Error Mapping
+Why:
 
-Provider transports need a single helper for pre-start final failures, for
-example under `providers/error_mapping.py`, that converts any final stream
-exception into a `ProviderError`:
+- `parse_cli_event()` should parse raw facts.
+- The runner owns turn-level event sequencing.
+- The duplicate condition depends on prior rendered events in the same turn.
 
-- preserve existing mapped provider statuses for authentication, bad request,
-  rate limit, overload, and upstream 5xx cases;
-- preserve existing user-facing error-message sanitization and request-id
-  appending;
-- wrap internal stream exceptions such as `TruncatedProviderStreamError` in an
-  upstream-style `APIError` rather than letting raw runtime exceptions escape;
-- keep verbose raw exception detail behind existing diagnostic flags.
+Decision:
 
-Do not make API egress inspect OpenAI/httpx exception classes directly. API
-egress should receive either a first chunk or a typed exception it can serialize
-for the product protocol.
+- Replace `had_error_events` with a narrower state, such as `had_non_exit_error`.
+- If parsed event is `source == "exit"` and a prior non-exit error was rendered, skip it.
+- Do not skip successful completion just because an earlier error existed.
 
-### OpenAI-Chat Transport
+### Decision 5: If an error is followed by successful completion, what wins?
 
-`providers/transports/openai_chat/stream.py` should distinguish:
+Recommended answer: successful completion wins.
 
-- **uncommitted pre-start final error**: raise mapped `ProviderError` so the API
-  first-frame gate returns non-200;
-- **committed/buffered stream failure**: preserve existing recovery/error-tail
-  behavior;
-- **early retry/recovery success**: unchanged;
-- **complete tool salvage**: unchanged.
+Why:
 
-The important classification is not merely whether `message_start` was created
-internally. It is whether anything has escaped the recovery holdback to the API
-iterator. If the holdback has not committed and no buffered events are being
-flushed as client-visible output, pre-start final errors should raise.
+- It is the actual terminal process result.
+- A prior warning/error-like event can be recoverable or superseded.
+- Keeping node state `ERROR` after successful terminal completion is less correct.
 
-### Native Anthropic Transport
+Decision:
 
-`providers/transports/anthropic_messages/stream.py` should apply the same
-boundary:
+- Let `complete(status="success")` pass through.
+- Completion updates node state to `COMPLETED`.
+- Completion path clears stale node error state if needed.
+- Do not redesign child-error propagation in this PR unless a test proves it is necessary.
 
-- no committed/buffered downstream-visible event: raise mapped `ProviderError`;
-- committed or buffered native stream: use native ledger error-tail behavior.
+### Decision 6: Should provider errors be hidden from Claude Code?
 
-This keeps local native providers and future native providers consistent with
-OpenAI-chat behavior.
+Recommended answer: no.
 
-### OpenAI Responses
+Why:
 
-`core/openai_responses/stream.py` and
-`core/openai_responses/streaming/assembler.py` should own post-start Responses
-terminal failures.
+- Users need actionable error messages.
+- Current provider mapping includes useful upstream status/body/cause/request-id detail.
+- Hiding errors would make debugging worse.
 
-Do not add a stateless `OpenAIResponsesAdapter.egress_error_frame()` that mints
-a fresh response id. Instead, the iterator returned by
-`OpenAIResponsesAdapter.iter_sse_from_anthropic()` should:
+Decision:
 
-- let pre-`response.created` failures propagate to API egress;
-- after `response.created`, catch unexpected non-cancellation failures, call
-  `ResponsesStreamAssembler.fail_response(...)` on the active assembler, yield
-  the resulting `response.failed`, then re-raise or trace according to the
-  existing egress tracing policy.
+- Preserve provider error text and error type.
+- Change only the retryability of the downstream shape.
 
-This preserves `response.failed.response.id == response.created.response.id`.
+### Decision 7: What downstream shape should streaming `/v1/messages` use for final provider errors?
 
-### Anthropic Messages
+Recommended answer: HTTP 200 with terminal Anthropic SSE `event: error`.
 
-For `/v1/messages`, post-start unexpected failures may use a stateless terminal
-Anthropic `event: error` as the final API egress fallback. Provider-owned error
-tails remain preferred when provider code can close content blocks and emit
-`message_delta`/`message_stop`.
+Why:
+
+- The request is a streaming Claude Messages turn.
+- Claude protocol already has an error event shape.
+- HTTP 200 commits the stream and prevents Claude Code from treating the request itself as retryable.
+- The error remains visible as protocol data.
+
+Decision:
+
+- For streaming `/v1/messages`, first-chunk provider-execution failures become terminal SSE error frames.
+- The terminal frame preserves `ProviderError.error_type` and `ProviderError.message`.
+- This applies to final provider auth/rate-limit/overload/API errors from execution, not just 429/5xx, because the API boundary should be uniform once provider execution owns the turn.
+- FCC auth, request validation, unsupported tools, model routing/preflight request-shape failures, and other ingress failures remain HTTP errors.
+
+### Decision 8: Should unexpected stream-body exceptions also be terminalized?
+
+Recommended answer: yes for streaming `/v1/messages`.
+
+Why:
+
+- Once the body iterator is being probed, the API has accepted a streaming turn.
+- Retrying an internal stream conversion failure externally is unlikely to help.
+- Messaging should receive a terminal result instead of waiting while Claude Code retries.
+
+Decision:
+
+- For streaming `/v1/messages`, any exception raised while pulling the first stream chunk becomes a terminal SSE error response.
+- `ProviderError` keeps its mapped type/message.
+- Unknown exceptions are logged and emitted as `api_error` with the same sanitized user-facing message policy used today.
+- `asyncio.CancelledError` and `GeneratorExit` are still re-raised by the response wrapper; cancellation is not an error result.
+
+### Decision 9: Should non-stream `/v1/messages` change now?
+
+Recommended answer: no.
+
+Why:
+
+- The observed stuck bug is the streaming Claude turn path.
+- Non-stream aggregation has no SSE protocol to commit.
+- Changing non-stream HTTP status semantics would be a separate client contract decision.
+
+Decision:
+
+- Keep `stream=false` behavior unchanged in this PR.
+- Add an explicit residual risk: if Claude Code is later shown to retry non-stream provider execution errors in a harmful loop, add a separate non-stream policy, likely a non-retryable dependency-failure status with the same Anthropic error body.
+
+### Decision 10: Should `/v1/responses` be changed in this PR?
+
+Recommended answer: only if the shared helper makes it trivial and tests remain simple; otherwise no.
+
+Why:
+
+- The current defect is managed Claude messaging.
+- Responses already converts Anthropic `event: error` into `response.failed`.
+- Pre-start Responses provider exceptions are the analogous boundary, but there is no current evidence of a Codex stuck loop.
+
+Decision:
+
+- Document the product-wide principle in `ARCHITECTURE.md`: streaming product APIs should surface final provider execution failures as terminal protocol events, not retryable HTTP statuses.
+- Implement Messages now.
+- Do not add broad Responses changes unless the implementation naturally exposes a reusable terminalization helper without additional architecture spread.
+
+## Target Architecture
+
+### Ownership Boundaries
+
+- Provider transports own upstream request construction, provider-specific retries/fallbacks, stream parsing, and recovery.
+- `providers.error_mapping` owns mapping raw SDK/HTTP errors into `ProviderError` with user-facing diagnostics.
+- `api.response_streams` owns first-chunk stream commit behavior.
+- `api.handlers.messages` owns Claude Messages product policy: what Claude Code receives for accepted streaming turns.
+- Messaging owns rendering CLI events and terminal node state, not provider retry semantics.
+
+### Streaming Error Policy
+
+Add an explicit policy at the public API boundary:
+
+```text
+Ingress/preflight failure before accepted stream
+  -> HTTP error response
+
+Accepted streaming Messages turn, provider/stream body fails before first chunk
+  -> HTTP 200 text/event-stream
+  -> terminal Anthropic event: error
+
+Accepted streaming Messages turn, provider/stream body fails after first chunk
+  -> existing terminal stream error behavior
+```
+
+This keeps the distinction precise:
+
+- HTTP status communicates whether FCC accepted the client request.
+- SSE terminal event communicates the provider execution result.
+
+### Error Payload Shape
+
+Anthropic terminal stream error:
+
+```text
+event: error
+data: {
+  "type": "error",
+  "error": {
+    "type": "<provider error_type or api_error>",
+    "message": "<sanitized provider/user-facing message>"
+  }
+}
+```
+
+The message should preserve current provider diagnostics, including copied upstream details and request id when present.
 
 ## Implementation Plan
 
-1. Add first-frame response helpers in `api/response_streams.py`.
-   - Add a small private result type for either first chunk or pre-start
-     exception.
-   - Add `async def anthropic_sse_streaming_response(...)` or a new clearly
-     named async builder, because first-frame probing must await the iterator.
-   - Add the equivalent Responses builder with injected OpenAI error-envelope
-     handling.
-   - Keep wrappers protocol-agnostic and make call sites explicit.
+### Step 1: Add Typed Cancellation Intent
 
-2. Update `MessagesHandler._to_public_response()`.
-   - Await the new Anthropic streaming response builder for `stream != false`.
-   - Convert pre-start `ProviderError` to Anthropic JSON with the provider
-     status code.
-   - Convert unexpected pre-start exceptions to a safe 500 Anthropic JSON error
-     using existing safe logging rules.
-   - Keep `stream: false` aggregation unchanged.
+Files:
 
-3. Update `ResponsesHandler.create()`.
-   - Await the new Responses streaming response builder.
-   - Convert pre-start `ProviderError` to OpenAI-shaped JSON using
-     `OpenAIResponsesAdapter.error_payload()`.
-   - Convert unexpected pre-start exceptions to safe 500 OpenAI-shaped JSON.
-   - Keep request conversion errors and `stream: false` rejection unchanged.
+- `messaging/trees/cancellation.py`
+- `messaging/trees/manager.py`
+- `messaging/workflow.py`
+- `messaging/node_runner.py`
 
-4. Add a provider-side pre-start failure exception path.
-   - Introduce a small neutral helper in provider/shared error mapping that
-     always returns a `ProviderError` for a final pre-start stream exception.
-   - In OpenAI-chat final-error handling, if the recovery holdback is not
-     committed and no event should be exposed, raise `map_error(...)` instead
-     of yielding `emit_error_tail(...)`.
-   - In native Anthropic final-error handling, raise mapped provider errors when
-     no native event has been committed/buffered to the API.
-   - Preserve existing synthetic SSE tails once events have escaped or must be
-     closed.
+Changes:
 
-5. Harden post-start terminal fallback.
-   - Add Anthropic terminal-frame serialization under
-     `core/anthropic/streaming/` only as a last-resort egress fallback.
-   - In Responses stream conversion, add same-assembler failure handling for
-     post-`response.created` exceptions.
-   - Do not mint fresh Responses IDs for a terminal failure.
+- Add `CancellationReason` enum with at least `STOP = "stop"`.
+- Add helper functions:
+  - `set_cancel_reason(node, reason)`
+  - `get_cancel_reason(node)`
+- Preserve existing node context when setting cancellation reason.
+- Update tree cancellation APIs:
+  - `cancel_tree(root_id, *, reason=None)`
+  - `cancel_node(node_id, *, reason=None)`
+  - `cancel_all(*, reason=None)`
+  - `cancel_branch(branch_root_id, *, reason=None)`
+- Apply cancellation reason before active task cancellation is delivered.
+- `MessagingWorkflow.stop_all_tasks()` and `stop_task()` pass `CancellationReason.STOP`.
+- `MessagingNodeRunner` reads cancel reason through the helper.
 
-6. Update architecture docs.
-   - Document that API egress owns first-frame HTTP commit gating.
-   - Document that providers raise pre-start final failures but own retries and
-     midstream recovery.
-   - Document that Responses terminal failures are assembler-owned because
-     Responses streams are stateful.
+Expected behavior:
 
-7. Bump version and lockfile.
-   - Update `[project].version` from `3.4.12` to `3.4.13` unless `main`
-     advances.
-   - Run `uv lock`.
+- Standalone `/stop` active task keeps transcript and final footer is `Stopped.`.
+- Reply `/stop` active task keeps existing correct behavior.
+- Queued/no-run stop remains workflow-owned and status-only.
+- Generic internal cancellation can still render generic cancellation.
+
+### Step 2: Narrow Exit Error Dedupe
+
+Files:
+
+- `messaging/node_runner.py`
+- `messaging/event_parser.py`
+- `messaging/node_event_pipeline.py`
+
+Changes:
+
+- Keep `parse_cli_event()` returning exit-sourced error facts for non-zero exits.
+- Replace `had_error_events` with narrower state:
+  - `had_non_exit_error`
+  - optionally `had_terminal_success`
+- Runner logic:
+  - non-exit error: process and set `had_non_exit_error = True`;
+  - exit-sourced error after non-exit error: skip;
+  - exit-sourced error without prior non-exit error: process;
+  - successful complete: process normally;
+  - non-success complete: keep existing ignored behavior.
+- Ensure successful completion leaves node state `COMPLETED` and does not leave a stale `error_message`.
+
+Expected behavior:
+
+- Provider error + exit code 1 renders only the useful provider error.
+- Exit code 1 alone still renders process-exit error.
+- Error-like event followed by success can complete.
+
+### Step 3: Add Anthropic Terminal Error Serialization For Provider Execution Failures
+
+Files:
+
+- `core/anthropic/streaming/emitter.py`
+- `api/response_streams.py`
+- `api/handlers/messages.py`
+
+Changes:
+
+- Add a serializer that accepts an error type and message, not only a hardcoded `api_error`.
+- Keep existing `anthropic_terminal_error_frame(message)` for generic egress interruption or replace it with a small wrapper around the typed serializer.
+- Add a helper for turning an exception into a terminal Anthropic SSE error:
+  - `ProviderError` -> use `exc.error_type` and `exc.message`;
+  - unknown `Exception` -> log through existing unexpected-error path and use sanitized `get_user_facing_error_message(exc)`;
+  - `EmptyStreamError` -> `api_error`, existing empty-stream message.
+- Do not include raw tracebacks or secrets in the SSE error.
+
+Expected behavior:
+
+- Existing post-start egress interruption still emits generic `api_error`.
+- Pre-start provider errors can preserve their specific type, e.g. `rate_limit_error`.
+
+### Step 4: Terminalize First-Chunk Streaming Messages Failures
+
+Files:
+
+- `api/handlers/messages.py`
+- `api/response_streams.py`
+- tests under `tests/api/`
+
+Changes:
+
+- Allow `pre_start_error_response` to return any `Response`, not just JSON.
+- In `MessagesHandler._pre_start_error_response()`:
+  - return `StreamingResponse` with `text/event-stream` for exceptions raised while probing the stream body;
+  - emit exactly one terminal Anthropic `event: error`;
+  - use HTTP 200;
+  - include `ANTHROPIC_SSE_RESPONSE_HEADERS`;
+  - trace `api.response.provider_error_terminalized` or `api.response.stream_start_error_terminalized`.
+- Keep synchronous `ProviderError` raised before `_to_public_response()` as HTTP. Those are preflight/ingress failures.
+- Keep `stream=false` aggregation behavior unchanged.
+
+Expected behavior:
+
+- Final provider 429 after FCC retries: Claude receives terminal SSE error, not HTTP 429.
+- Final provider 5xx/529 after FCC retries: Claude receives terminal SSE error, not retryable HTTP.
+- Final provider auth error during provider execution: Claude receives terminal SSE `authentication_error`.
+- FCC auth failure before handler: still HTTP 401.
+- Request validation/unsupported tool/preflight request-shape failure: still HTTP 400.
+
+### Step 5: Keep Provider Retry/Recovery Unchanged
+
+Files:
+
+- `providers/rate_limit.py`
+- `providers/transports/openai_chat/stream.py`
+- `providers/transports/anthropic_messages/stream.py`
+
+Changes:
+
+- No retry count changes.
+- No NIM-specific fallback.
+- No request trimming.
+- No prompt/history modification.
+- No recovery prompt changes.
+
+Expected behavior:
+
+- Provider retry ownership stays where it is.
+- API only changes the final shape returned to Claude after provider retry/recovery is exhausted.
+
+### Step 6: Document The Boundary
+
+Files:
+
+- `ARCHITECTURE.md`
+
+Changes:
+
+- Add a short note to the API/product boundary section:
+  - provider transports own upstream retry/recovery;
+  - public streaming product handlers own downstream terminal error shape;
+  - final provider execution failures must be surfaced as protocol terminal events for streaming clients, not retryable HTTP statuses.
 
 ## Test Plan
 
-### API Egress Tests
+### Messaging Stop Tests
 
-- ASGI-level test proving `http.response.start 200` is not sent until the first
-  chunk is available.
-- ASGI-level test proving a pre-first-chunk `ProviderError` returns non-200 JSON
-  and sends no SSE body.
-- ASGI-level test proving a pre-first-chunk unexpected exception returns safe
-  500 JSON and does not leak raw exception text by default.
-- ASGI-level test proving post-first-chunk exceptions yield a terminal frame
-  before the exception closes the ASGI body.
-- Cancellation and `GeneratorExit` tests proving no terminal frame is emitted
-  into a dead socket.
+Add/update:
 
-### Messages API Tests
+- `tests/messaging/test_handler.py`
+- `tests/messaging/test_tree_queue.py`
+- `tests/messaging/test_handler_markdown_and_status_edges.py`
 
-- `/v1/messages` provider pre-start `RateLimitError` returns HTTP 429
-  Anthropic-shaped JSON.
-- `/v1/messages` provider pre-start `APIError(status_code=504)` returns HTTP
-  non-200 using the existing provider mapping, not HTTP 200.
-- `/v1/messages` delayed valid provider stream still returns `text/event-stream`
-  and valid Anthropic SSE.
-- `/v1/messages stream:false` aggregation remains unchanged.
+Cases:
 
-### Responses API Tests
+- Active standalone `/stop`:
+  - node emits partial transcript;
+  - `stop_all_tasks()` is called;
+  - final message contains prior transcript and `Stopped.`;
+  - final message does not contain generic `Cancelled`.
+- Reply-scoped active `/stop` remains correct.
+- Queued standalone `/stop` still gets workflow-owned `Stopped.`.
+- `cancel_all(reason=STOP)` marks active runner-owned nodes before cancellation cleanup.
+- Cancellation context helper preserves unrelated node context keys.
 
-- `/v1/responses` provider pre-start `RateLimitError` returns HTTP 429
-  OpenAI-shaped JSON.
-- `/v1/responses` delayed valid provider stream still returns
-  `text/event-stream`.
-- Responses post-start exception emits `response.failed` with the same
-  `response.id` as `response.created`.
-- Existing provider-emitted Anthropic `event:error` still converts to
-  `response.failed` with the same active response id.
+### CLI Event Dedupe Tests
 
-### Provider Transport Tests
+Add/update:
 
-- OpenAI-chat exhausted pre-stream 429/5xx/transport failure raises mapped
-  `ProviderError` before any downstream-visible event.
-- OpenAI-chat retry success path remains unchanged.
-- OpenAI-chat midstream text failure still uses recovery/terminal tail behavior.
-- OpenAI-chat complete tool salvage remains unchanged.
-- Native Anthropic pre-send/pre-event failure raises mapped `ProviderError`.
-- Native Anthropic midstream native event failure still closes through native
-  recovery/error-tail behavior.
+- `tests/messaging/test_event_parser.py`
+- `tests/messaging/test_handler.py`
+- `tests/cli/test_cli.py`
 
-### Contract Tests
+Cases:
 
-- API response stream helper is the only owner of first-frame HTTP commit
-  gating.
-- Responses egress terminal failure is assembler-owned, not generated by a
-  stateless adapter helper.
-- `api/provider_execution.py` remains free of `StreamingResponse` ownership.
-- Architecture relative links still resolve.
+- Provider error followed by exit code 1:
+  - provider error renders once;
+  - `Process exited with code 1` is suppressed;
+  - no complete footer;
+  - node state is `ERROR`.
+- Exit code 1 without prior non-exit error:
+  - process-exit error renders.
+- Non-exit error followed by successful exit:
+  - completion is processed;
+  - final footer is `Complete`;
+  - node state is `COMPLETED`.
+- Non-success complete remains ignored.
+
+### Streaming Messages API Tests
+
+Add/update:
+
+- `tests/api/test_response_streams.py`
+- `tests/api/test_api_handlers.py`
+- `tests/api/test_api.py`
+
+Cases:
+
+- Streaming `/v1/messages`, body raises `RateLimitError` before first chunk:
+  - HTTP 200;
+  - content type `text/event-stream`;
+  - single terminal `event: error`;
+  - error type `rate_limit_error`;
+  - message preserves provider diagnostic text.
+- Streaming `/v1/messages`, body raises `OverloadedError` before first chunk:
+  - HTTP 200 terminal SSE error;
+  - no retryable HTTP 529.
+- Streaming `/v1/messages`, body raises `APIError(status_code=503)` before first chunk:
+  - HTTP 200 terminal SSE error;
+  - error type `api_error`.
+- Streaming `/v1/messages`, body raises provider `AuthenticationError` during provider execution:
+  - HTTP 200 terminal SSE error;
+  - error type `authentication_error`.
+- FCC/API ingress auth failure remains HTTP 401.
+- Invalid request/preflight `InvalidRequestError` before stream body remains HTTP 400.
+- Unexpected exception from stream body before first chunk:
+  - HTTP 200 terminal SSE `api_error`;
+  - unexpected exception is logged.
+- Empty stream before first chunk:
+  - HTTP 200 terminal SSE `api_error`.
+- Post-start exception behavior remains existing terminal stream error frame.
+- `stream=false` behavior remains unchanged.
+
+### Managed Messaging Regression Tests
+
+Add/update:
+
+- `tests/messaging/test_handler.py`
+- `tests/cli/test_cli.py`
+
+Cases:
+
+- Managed Claude emits a provider error event caused by terminal SSE, then exits non-zero:
+  - messaging renders the provider error;
+  - node leaves `IN_PROGRESS`;
+  - no duplicate process-exit error;
+  - no false `Complete`.
+- Managed Claude completes successfully after a non-exit warning/error-like event:
+  - success completion wins.
+
+### Optional Responses Guardrail
+
+Add only if implementation naturally touches shared streaming policy:
+
+- pre-start `/v1/responses` provider execution failure emits `response.created` then `response.failed`;
+- Codex does not receive retryable HTTP for final provider execution errors.
+
+If this adds meaningful complexity, leave it out and document as a follow-up.
 
 ## Verification Commands
 
-Run targeted tests first:
+Run targeted checks first:
 
 ```powershell
-uv run pytest tests/api/test_response_streams.py tests/api/test_api_handlers.py tests/api/test_openai_responses.py
-uv run pytest tests/providers/test_streaming_errors.py tests/providers/test_anthropic_messages.py tests/providers/test_openai_compat_5xx_retry.py tests/providers/test_anthropic_messages_429_retry.py
-uv run pytest tests/core/openai_responses/test_sse.py tests/contracts/test_import_boundaries.py tests/contracts/test_architecture_contracts.py
+uv run pytest tests/messaging/test_handler.py tests/messaging/test_handler_markdown_and_status_edges.py tests/messaging/test_tree_queue.py tests/messaging/test_event_parser.py tests/cli/test_cli.py
+uv run pytest tests/api/test_response_streams.py tests/api/test_api_handlers.py tests/api/test_api.py
 ```
 
 Then run the final local gate:
@@ -302,23 +618,20 @@ Then run the final local gate:
 .\scripts\ci.ps1
 ```
 
-## Risks And Guardrails
-
-- Pulling the first chunk delays HTTP headers until upstream viability is known.
-  This is intentional for Claude/Codex retry correctness, but tests should prove
-  normal streams still begin promptly after the first provider chunk.
-- A provider can emit a synthetic SSE error as its first chunk. The PR should
-  avoid relying only on API first-chunk gating; providers must raise pre-start
-  final failures instead of creating synthetic success streams.
-- Responses post-start fallback must preserve assembler identity. Any test that
-  shape-asserts only `response.failed` without comparing IDs is insufficient.
-- Do not broaden API egress into retry/recovery ownership. That would erode the
-  provider transport boundaries established in `ARCHITECTURE.md`.
+If the local server is running, stop it before full CI because installer/uninstaller tests require no active `fcc-server` process.
 
 ## Out Of Scope
 
-- Changing retry counts, backoff, or adding `Retry-After` support.
-- Redesigning OpenAI-chat tool-call buffering.
-- Changing model routing, thinking/reasoning policy, or provider request bodies.
-- Changing messaging queue/cancellation behavior.
-- Restoring non-streaming `/v1/responses`.
+- Changing provider retry counts, backoff, or rate-limit blocking.
+- Adding NIM-specific request fallbacks.
+- Trimming prompts, tools, or conversation history.
+- Changing `stream=false` response policy.
+- Redesigning `/v1/responses` unless the shared helper makes the fix trivial.
+- Changing Claude Code invocation flags.
+- Changing Telegram/Discord platform adapters beyond tests expecting final UI text.
+
+## Residual Risks
+
+- Claude Code could theoretically retry terminal SSE `event: error`. If observed, the next escalation is a managed-Claude-specific request marker plus a known non-retryable terminal shape. That should not be implemented until evidence proves SSE terminal errors are retried.
+- `stream=false` provider errors may still use retryable HTTP statuses. This is acceptable for this PR because the live stuck path is streaming; if a non-stream retry loop is observed, add a separate non-stream provider-error policy.
+- Allowing success completion after a prior non-exit error may expose a stale propagated child cancellation if a child was already marked failed. The first implementation should test the known event order; if a real recoverable-error-success stream exists, delay child propagation until terminal process state in a later focused design.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "free-claude-code"
-version = "3.4.13"
+version = "3.4.14"
 description = "Middleware between Claude Code CLI (Anthropic API) and NVIDIA NIM"
 readme = "README.md"
 requires-python = ">=3.14.0"

tests/api/test_api.py

@@ -4,6 +4,7 @@ import pytest
 from fastapi.testclient import TestClient
 
 from api.app import create_app
+from core.anthropic.stream_contracts import parse_sse_text
 from providers.exceptions import RateLimitError
 from providers.nvidia_nim import NvidiaNimProvider
 
@@ -30,6 +31,14 @@ async def _mock_pre_start_rate_limit(*args, **kwargs):
     yield "unreachable"
 
 
+def _stream_error(response):
+    assert response.status_code == 200
+    assert "text/event-stream" in response.headers.get("content-type", "")
+    events = parse_sse_text(response.text)
+    assert [event.event for event in events] == ["error"]
+    return events[0].data["error"]
+
+
 mock_provider.stream_response = _mock_stream_response
 
 
@@ -103,10 +112,10 @@ def test_create_message_stream(client: TestClient):
     assert b"message_start" in content or b"event:" in content
 
 
-def test_create_message_pre_start_provider_error_returns_non_200_json(
+def test_create_message_pre_start_provider_error_returns_terminal_sse(
     client: TestClient,
 ):
-    """Pre-first-chunk provider errors should not commit HTTP 200."""
+    """Provider execution failures should not leak retryable HTTP status."""
     mock_provider.stream_response = _mock_pre_start_rate_limit
     payload = {
         "model": "claude-3-sonnet",
@@ -117,12 +126,8 @@ def test_create_message_pre_start_provider_error_returns_non_200_json(
 
     response = client.post("/v1/messages", json=payload)
 
-    assert response.status_code == 429
-    assert response.headers["content-type"].startswith("application/json")
-    assert response.json() == {
-        "type": "error",
-        "error": {"type": "rate_limit_error", "message": "upstream is busy"},
-    }
+    error = _stream_error(response)
+    assert error == {"type": "rate_limit_error", "message": "upstream is busy"}
     mock_provider.stream_response = _mock_stream_response
 
 
@@ -189,30 +194,27 @@ def test_error_fallbacks(client: TestClient):
     def _raise_overloaded(*args, **kwargs):
         raise OverloadedError("Server Overloaded")
 
-    # 1. Authentication Error (401)
+    # 1. Provider authentication during execution is terminal SSE, not retryable HTTP.
     mock_provider.stream_response = _raise_auth
     response = client.post("/v1/messages", json=base_payload)
-    assert response.status_code == 401
-    assert response.json()["error"]["type"] == "authentication_error"
+    assert _stream_error(response)["type"] == "authentication_error"
 
-    # 2. Rate Limit (429)
+    # 2. Provider rate limit during execution is terminal SSE, not retryable HTTP.
     mock_provider.stream_response = _raise_rate_limit
     response = client.post("/v1/messages", json=base_payload)
-    assert response.status_code == 429
-    assert response.json()["error"]["type"] == "rate_limit_error"
+    assert _stream_error(response)["type"] == "rate_limit_error"
 
-    # 3. Overloaded (529)
+    # 3. Provider overload during execution is terminal SSE, not retryable HTTP.
     mock_provider.stream_response = _raise_overloaded
     response = client.post("/v1/messages", json=base_payload)
-    assert response.status_code == 529
-    assert response.json()["error"]["type"] == "overloaded_error"
+    assert _stream_error(response)["type"] == "overloaded_error"
 
     # Reset for subsequent tests
     mock_provider.stream_response = _mock_stream_response
 
 
-def test_generic_exception_returns_500(client: TestClient):
-    """Non-ProviderError exceptions are caught and returned as HTTPException(500)."""
+def test_generic_stream_exception_returns_terminal_sse(client: TestClient):
+    """Unexpected provider execution failures also terminalize the accepted stream."""
 
     def _raise_runtime(*args, **kwargs):
         raise RuntimeError("unexpected crash")
@@ -227,12 +229,16 @@ def test_generic_exception_returns_500(client: TestClient):
             "stream": True,
         },
     )
-    assert response.status_code == 500
+    error = _stream_error(response)
+    assert error["type"] == "api_error"
+    assert error["message"] == "unexpected crash"
     mock_provider.stream_response = _mock_stream_response
 
 
-def test_generic_exception_with_status_code(client: TestClient):
-    """Unexpected errors always map to HTTP 500 (ignore ad-hoc status_code attrs)."""
+def test_generic_stream_exception_with_status_code_returns_terminal_sse(
+    client: TestClient,
+):
+    """Ad-hoc status_code attrs do not become retryable HTTP responses."""
 
     class ExceptionWithStatus(RuntimeError):
         def __init__(self, msg: str, status_code: int = 500):
@@ -252,11 +258,15 @@ def test_generic_exception_with_status_code(client: TestClient):
             "stream": True,
         },
     )
-    assert response.status_code == 500
+    error = _stream_error(response)
+    assert error["type"] == "api_error"
+    assert error["message"] == "bad gateway"
     mock_provider.stream_response = _mock_stream_response
 
 
-def test_generic_exception_empty_message_returns_non_empty_detail(client: TestClient):
+def test_generic_stream_exception_empty_message_returns_non_empty_error(
+    client: TestClient,
+):
     """Exceptions with empty __str__ still return a readable HTTP detail."""
 
     class SilentError(RuntimeError):
@@ -276,8 +286,9 @@ def test_generic_exception_empty_message_returns_non_empty_detail(client: TestCl
             "stream": True,
         },
     )
-    assert response.status_code == 500
-    assert response.json()["detail"] != ""
+    error = _stream_error(response)
+    assert error["type"] == "api_error"
+    assert error["message"] != ""
     mock_provider.stream_response = _mock_stream_response
 
 

tests/api/test_api_handlers.py

@@ -12,6 +12,7 @@ from api.models.openai_responses import OpenAIResponsesRequest
 from config.settings import Settings
 from core.anthropic.streaming import format_sse_event
 from providers.base import BaseProvider, ProviderConfig
+from providers.exceptions import InvalidRequestError
 
 _CLASSIFIER_SYSTEM = (
     "You are a security monitor. Respond with <block>yes</block> or <block>no</block>."
@@ -110,6 +111,26 @@ async def test_messages_handler_passes_routed_request_and_stream_metadata() -> N
     assert len(provider.preflight_calls) == 1
 
 
+@pytest.mark.asyncio
+async def test_messages_handler_preflight_invalid_request_stays_http_error() -> None:
+    class RejectPreflightProvider(FakeProvider):
+        def preflight_stream(
+            self, request: Any, *, thinking_enabled: bool | None = None
+        ) -> None:
+            raise InvalidRequestError("bad tool shape")
+
+    provider = RejectPreflightProvider()
+    handler = MessagesHandler(Settings(), provider_getter=lambda _: provider)
+    request = MessagesRequest(
+        model="nvidia_nim/test-model",
+        max_tokens=100,
+        messages=[Message(role="user", content="hi")],
+    )
+
+    with pytest.raises(InvalidRequestError):
+        await handler.create(request)
+
+
 @pytest.mark.asyncio
 async def test_messages_handler_aggregates_provider_stream_when_stream_false() -> None:
     provider = FakeProvider(

tests/api/test_response_streams.py

@@ -9,6 +9,7 @@ from fastapi.responses import JSONResponse, StreamingResponse
 
 from api.response_streams import (
     EGRESS_STREAM_INTERRUPTED_MESSAGE,
+    anthropic_sse_error_response,
     anthropic_sse_streaming_response,
 )
 from core.anthropic.stream_contracts import parse_sse_text
@@ -94,6 +95,24 @@ async def test_anthropic_pre_start_provider_error_returns_non_200_json() -> None
     assert body["error"]["message"] == "provider says slow down"
 
 
+@pytest.mark.asyncio
+async def test_anthropic_sse_error_response_preserves_error_type() -> None:
+    response = anthropic_sse_error_response(
+        error_type="rate_limit_error",
+        message="provider says slow down",
+    )
+
+    assert isinstance(response, StreamingResponse)
+    assert response.status_code == 200
+    text = await _drain(response)
+    events = parse_sse_text(text)
+    assert [event.event for event in events] == ["error"]
+    assert events[0].data["error"] == {
+        "type": "rate_limit_error",
+        "message": "provider says slow down",
+    }
+
+
 @pytest.mark.asyncio
 async def test_anthropic_post_start_exception_emits_terminal_error_frame() -> None:
     response = await anthropic_sse_streaming_response(

tests/api/test_safe_logging.py

@@ -4,12 +4,14 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 from fastapi import HTTPException
+from fastapi.responses import StreamingResponse
 
 from api import provider_execution, request_errors
 from api.handlers import MessagesHandler, TokenCountHandler
 from api.models.anthropic import Message, MessagesRequest
 from config.settings import Settings
 from core.anthropic import AnthropicStreamLedger
+from core.anthropic.stream_contracts import parse_sse_text
 
 
 @pytest.mark.asyncio
@@ -92,6 +94,16 @@ def _flatten_log_calls(mock_log) -> str:
     return " ".join(parts)
 
 
+async def _streaming_body_text(response: StreamingResponse) -> str:
+    parts: list[str] = []
+    async for chunk in response.body_iterator:
+        if isinstance(chunk, bytes):
+            parts.append(chunk.decode("utf-8"))
+        else:
+            parts.append(str(chunk))
+    return "".join(parts)
+
+
 @pytest.mark.asyncio
 async def test_create_message_unexpected_error_default_logs_exclude_exception_text():
     settings = Settings()
@@ -111,20 +123,21 @@ async def test_create_message_unexpected_error_default_logs_exclude_exception_te
         messages=[Message(role="user", content="hi")],
     )
 
-    with (
-        patch.object(request_errors.logger, "error") as log_err,
-        pytest.raises(HTTPException),
-    ):
-        await service.create(request)
+    with patch.object(request_errors.logger, "error") as log_err:
+        response = await service.create(request)
 
     blob = _flatten_log_calls(log_err)
     assert secret not in blob
     assert "RuntimeError" in blob
+    assert isinstance(response, StreamingResponse)
+    events = parse_sse_text(await _streaming_body_text(response))
+    assert [event.event for event in events] == ["error"]
+    assert events[0].data["error"]["type"] == "api_error"
 
 
 @pytest.mark.asyncio
-async def test_create_message_unexpected_error_always_returns_500():
-    """Non-provider failures must not leak arbitrary status_code attributes."""
+async def test_create_message_unexpected_error_terminal_sse_ignores_status_code():
+    """Non-provider stream failures must not leak arbitrary HTTP status attributes."""
 
     class WeirdError(Exception):
         status_code = 418
@@ -143,10 +156,13 @@ async def test_create_message_unexpected_error_always_returns_500():
         messages=[Message(role="user", content="hi")],
     )
 
-    with pytest.raises(HTTPException) as excinfo:
-        await service.create(request)
+    response = await service.create(request)
 
-    assert excinfo.value.status_code == 500
+    assert isinstance(response, StreamingResponse)
+    assert response.status_code == 200
+    events = parse_sse_text(await _streaming_body_text(response))
+    assert [event.event for event in events] == ["error"]
+    assert events[0].data["error"] == {"type": "api_error", "message": "no"}
 
 
 def test_parse_cli_event_error_logs_metadata_by_default():

tests/cli/test_cli.py

@@ -119,18 +119,20 @@ class TestCLIParser:
         assert result[0]["status"] == "success"
 
     def test_parse_exit_failure(self):
-        """Test parsing exit event with failure returns error then complete."""
+        """Test parsing exit event with failure returns an error only."""
         event = {"type": "exit", "code": 1}
         result = parse_cli_event(event)
-        # Non-zero exit now returns error first, then complete
-        assert len(result) == 2
-        assert result[0]["type"] == "error"
+        assert len(result) == 1
+        assert result[0] == {
+            "type": "error",
+            "message": "Process exited with code 1",
+            "source": "exit",
+            "exit_code": 1,
+        }
         assert (
             "exit" in result[0]["message"].lower()
             or "code" in result[0]["message"].lower()
         )
-        assert result[1]["type"] == "complete"
-        assert result[1]["status"] == "failed"
 
     def test_parse_invalid_event(self):
         """Test parsing returns empty list for unrecognized event."""

tests/messaging/test_event_parser.py

@@ -135,9 +135,14 @@ def test_parse_cli_event_exit_success():
 def test_parse_cli_event_exit_failure():
     event = {"type": "exit", "code": 1, "stderr": "fatal error"}
     results = parse_cli_event(event)
-    assert len(results) == 2
-    assert results[0] == {"type": "error", "message": "fatal error"}
-    assert results[1] == {"type": "complete", "status": "failed"}
+    assert results == [
+        {
+            "type": "error",
+            "message": "fatal error",
+            "source": "exit",
+            "exit_code": 1,
+        }
+    ]
 
 
 def test_parse_cli_event_invalid_input():

tests/messaging/test_handler.py

@@ -4,7 +4,15 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 from messaging.models import IncomingMessage
-from messaging.trees import MessageNode, MessageState, MessageTree, TreeQueueManager
+from messaging.trees import (
+    CancellationReason,
+    CancellationUiOwner,
+    CancelledNode,
+    MessageNode,
+    MessageState,
+    MessageTree,
+    TreeQueueManager,
+)
 from messaging.workflow import MessagingWorkflow
 
 
@@ -357,15 +365,87 @@ async def test_stop_all_tasks(handler, mock_cli_manager, mock_platform):
     mock_node.status_message_id = "status_1"
 
     with patch.object(
-        handler.tree_queue, "cancel_all", AsyncMock(return_value=[mock_node])
-    ):
+        handler.tree_queue,
+        "cancel_all",
+        AsyncMock(
+            return_value=[CancelledNode(mock_node, CancellationUiOwner.WORKFLOW)]
+        ),
+    ) as cancel_all:
         count = await handler.stop_all_tasks()
 
         assert count == 1
+        cancel_all.assert_awaited_once_with(reason=CancellationReason.STOP)
         mock_cli_manager.stop_all.assert_called_once()
         mock_platform.fire_and_forget.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_stop_all_tasks_skips_direct_edit_for_active_runner_owned_node(
+    handler, mock_cli_manager, mock_platform
+):
+    """Active node cleanup owns transcript-preserving stopped rendering."""
+    incoming = IncomingMessage(
+        text="work",
+        chat_id="chat_1",
+        user_id="user_1",
+        message_id="node_1",
+        platform="telegram",
+    )
+    node = MessageNode(
+        node_id="node_1", incoming=incoming, status_message_id="status_1"
+    )
+    tree = MagicMock()
+    tree.root_id = "root_1"
+    tree.snapshot.return_value = {"root": "snapshot"}
+
+    with (
+        patch.object(
+            handler.tree_queue,
+            "cancel_all",
+            AsyncMock(return_value=[CancelledNode(node, CancellationUiOwner.RUNNER)]),
+        ),
+        patch.object(
+            handler.tree_queue, "get_tree_for_node", MagicMock(return_value=tree)
+        ),
+    ):
+        count = await handler.stop_all_tasks()
+
+    assert count == 1
+    mock_cli_manager.stop_all.assert_called_once()
+    mock_platform.fire_and_forget.assert_not_called()
+    mock_platform.queue_edit_message.assert_not_called()
+    handler.session_store.save_tree_snapshot.assert_called_once_with(
+        {"root": "snapshot"}
+    )
+
+
+@pytest.mark.asyncio
+async def test_stop_all_tasks_edits_queued_workflow_owned_node(
+    handler, mock_cli_manager, mock_platform
+):
+    """Queued nodes have no transcript owner, so workflow renders simple stop."""
+    incoming = IncomingMessage(
+        text="queued",
+        chat_id="chat_1",
+        user_id="user_1",
+        message_id="node_1",
+        platform="telegram",
+    )
+    node = MessageNode(
+        node_id="node_1", incoming=incoming, status_message_id="status_1"
+    )
+
+    with patch.object(
+        handler.tree_queue,
+        "cancel_all",
+        AsyncMock(return_value=[CancelledNode(node, CancellationUiOwner.WORKFLOW)]),
+    ):
+        count = await handler.stop_all_tasks()
+
+    assert count == 1
+    mock_platform.fire_and_forget.assert_called_once()
+
+
 async def mock_async_gen(events):
     for e in events:
         yield e
@@ -522,6 +602,174 @@ async def test_node_runner_process_node_error_flow(
         assert "CLI crashed" in last_call[0][2]
 
 
+@pytest.mark.asyncio
+async def test_node_runner_process_node_provider_error_exit_does_not_complete(
+    handler, mock_cli_manager, mock_platform
+):
+    node_id = "node_1"
+    mock_node = MagicMock()
+    mock_node.incoming.chat_id = "chat_1"
+    mock_node.incoming.text = "hello"
+    mock_node.status_message_id = "status_1"
+    mock_node.parent_id = None
+
+    mock_session = MagicMock()
+    events = [
+        {
+            "type": "error",
+            "error": {
+                "message": "API Error: Request rejected (429)\nProvider rate limit reached."
+            },
+        },
+        {"type": "exit", "code": 1},
+    ]
+    mock_session.start_task.return_value = mock_async_gen(events)
+    mock_cli_manager.get_or_create_session.return_value = (
+        mock_session,
+        "session_1",
+        False,
+    )
+
+    mock_tree = MagicMock()
+    mock_tree.root_id = "root_1"
+    mock_tree.snapshot.return_value = {"data": "tree"}
+    mock_tree.update_state = AsyncMock()
+
+    with (
+        patch.object(
+            handler.tree_queue, "get_tree_for_node", MagicMock(return_value=mock_tree)
+        ),
+        patch.object(
+            handler.tree_queue, "mark_node_error", AsyncMock(return_value=[mock_node])
+        ) as mark_node_error,
+    ):
+        await handler.node_runner.process_node(node_id, mock_node)
+
+    mark_node_error.assert_called_once_with(
+        node_id,
+        "API Error: Request rejected (429)\nProvider rate limit reached.",
+        propagate_to_children=True,
+    )
+    mock_tree.update_state.assert_any_call(node_id, MessageState.IN_PROGRESS)
+    assert not any(
+        call.args == (node_id, MessageState.COMPLETED)
+        or (
+            len(call.args) >= 2
+            and call.args[0] == node_id
+            and call.args[1] is MessageState.COMPLETED
+        )
+        for call in mock_tree.update_state.call_args_list
+    )
+
+    rendered = mock_platform.queue_edit_message.call_args_list[-1].args[2]
+    assert "❌ *Error*" in rendered
+    assert "API Error: Request rejected" in rendered
+    assert "Process exited with code" not in rendered
+    assert "✅ *Complete*" not in rendered
+
+
+@pytest.mark.asyncio
+async def test_node_runner_process_node_success_complete_wins_after_non_exit_error(
+    handler, mock_cli_manager, mock_platform
+):
+    node_id = "node_1"
+    mock_node = MagicMock()
+    mock_node.incoming.chat_id = "chat_1"
+    mock_node.incoming.text = "hello"
+    mock_node.status_message_id = "status_1"
+    mock_node.parent_id = None
+
+    mock_session = MagicMock()
+    events = [
+        {"type": "error", "error": {"message": "recoverable warning"}},
+        {"type": "exit", "code": 0},
+    ]
+    mock_session.start_task.return_value = mock_async_gen(events)
+    mock_cli_manager.get_or_create_session.return_value = (
+        mock_session,
+        "session_1",
+        False,
+    )
+
+    mock_tree = MagicMock()
+    mock_tree.root_id = "root_1"
+    mock_tree.snapshot.return_value = {"data": "tree"}
+    mock_tree.update_state = AsyncMock()
+
+    with (
+        patch.object(
+            handler.tree_queue, "get_tree_for_node", MagicMock(return_value=mock_tree)
+        ),
+        patch.object(
+            handler.tree_queue, "mark_node_error", AsyncMock(return_value=[mock_node])
+        ) as mark_node_error,
+    ):
+        await handler.node_runner.process_node(node_id, mock_node)
+
+    mark_node_error.assert_called_once_with(
+        node_id,
+        "recoverable warning",
+        propagate_to_children=True,
+    )
+    mock_tree.update_state.assert_any_call(
+        node_id, MessageState.COMPLETED, session_id="session_1"
+    )
+    rendered = mock_platform.queue_edit_message.call_args_list[-1].args[2]
+    assert "✅ *Complete*" in rendered
+
+
+@pytest.mark.asyncio
+async def test_node_runner_stop_cancellation_preserves_transcript(
+    handler, mock_cli_manager, mock_platform
+):
+    started = asyncio.Event()
+
+    async def start_task(*args, **kwargs):
+        yield {
+            "type": "assistant",
+            "message": {"content": [{"type": "text", "text": "partial answer"}]},
+        }
+        started.set()
+        try:
+            await asyncio.sleep(60)
+        except asyncio.CancelledError:
+            raise
+
+    mock_session = MagicMock()
+    mock_session.start_task = start_task
+    mock_cli_manager.get_or_create_session.return_value = (
+        mock_session,
+        "session_1",
+        False,
+    )
+
+    incoming = IncomingMessage(
+        text="work",
+        chat_id="chat_1",
+        user_id="user_1",
+        message_id="node_1",
+        platform="telegram",
+    )
+    node = MessageNode(
+        node_id="node_1", incoming=incoming, status_message_id="status_1"
+    )
+    node.set_context({"cancel_reason": "stop"})
+
+    tree = await handler.tree_queue.create_tree("node_1", incoming, "status_1")
+    task = asyncio.create_task(handler.node_runner.process_node("node_1", node))
+    await started.wait()
+
+    task.cancel()
+    await task
+
+    last_call = mock_platform.queue_edit_message.call_args_list[-1]
+    rendered = last_call.args[2]
+    assert "partial answer" in rendered
+    assert "⏹ *Stopped\\.*" in rendered
+    assert rendered.index("partial answer") < rendered.index("⏹ *Stopped\\.*")
+    assert tree.get_node("node_1").state == MessageState.ERROR
+
+
 @pytest.mark.asyncio
 async def test_handle_message_clear_command_stops_deletes_and_wipes_state(
     handler, mock_platform, mock_session_store, incoming_message_factory

tests/messaging/test_handler_markdown_and_status_edges.py

@@ -6,7 +6,12 @@ import pytest
 from messaging.models import IncomingMessage
 from messaging.node_event_pipeline import process_parsed_cli_event
 from messaging.rendering.telegram_markdown import render_markdown_to_mdv2
-from messaging.trees import MessageNode, MessageState
+from messaging.trees import (
+    CancellationUiOwner,
+    CancelledNode,
+    MessageNode,
+    MessageState,
+)
 from messaging.workflow import MessagingWorkflow
 
 
@@ -249,7 +254,11 @@ async def test_stop_all_tasks_saves_tree_for_cancelled_nodes():
     tree.root_id = "root"
     tree.snapshot = MagicMock(return_value={"root": "ok"})
     with (
-        patch.object(handler.tree_queue, "cancel_all", AsyncMock(return_value=[node])),
+        patch.object(
+            handler.tree_queue,
+            "cancel_all",
+            AsyncMock(return_value=[CancelledNode(node, CancellationUiOwner.RUNNER)]),
+        ),
         patch.object(
             handler.tree_queue, "get_tree_for_node", MagicMock(return_value=tree)
         ),
@@ -420,6 +429,41 @@ async def test_process_parsed_event_malformed_content_continues():
     assert had is False
 
 
+@pytest.mark.asyncio
+async def test_process_parsed_event_failed_complete_does_not_mark_success():
+    """Failed terminal events are not rendered as successful completion."""
+    platform = MagicMock()
+    platform.queue_edit_message = AsyncMock()
+
+    cli_manager = MagicMock()
+    session_store = MagicMock()
+    handler = MessagingWorkflow(platform, cli_manager, session_store)
+
+    transcript = MagicMock()
+    update_ui = AsyncMock()
+    tree = MagicMock()
+    tree.update_state = AsyncMock()
+
+    last_status, had = await process_parsed_cli_event(
+        parsed={"type": "complete", "status": "failed"},
+        transcript=transcript,
+        update_ui=update_ui,
+        last_status="❌ Error",
+        had_transcript_events=True,
+        tree=tree,
+        node_id="n1",
+        captured_session_id="session_1",
+        session_store=session_store,
+        format_status=handler.format_status,
+        propagate_error_to_children=AsyncMock(),
+    )
+
+    assert last_status == "❌ Error"
+    assert had is True
+    update_ui.assert_not_awaited()
+    tree.update_state.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_handler_update_ui_edit_failure_does_not_crash():
     """When queue_edit_message raises during streaming, node_runner.process_node continues and completes."""

tests/messaging/test_tree_queue.py

@@ -8,11 +8,15 @@ import pytest
 
 from messaging.models import IncomingMessage
 from messaging.trees import (
+    CancellationReason,
+    CancellationUiOwner,
     MessageNode,
     MessageState,
     MessageTree,
     TreeQueueManager,
     TreeSnapshot,
+    get_cancel_reason,
+    set_cancel_reason,
 )
 from messaging.trees import manager as tree_manager_module
 from messaging.trees.graph import MessageTreeGraph
@@ -101,6 +105,46 @@ class TestMessageNode:
         assert node.parent_id == "parent_1"
         assert "child_1" in node.children_ids
 
+    def test_completed_state_clears_stale_error_message(self):
+        """Successful completion is terminal and should clear prior node errors."""
+        incoming = IncomingMessage(
+            text="Test",
+            chat_id="1",
+            user_id="2",
+            message_id="3",
+            platform="test",
+        )
+        node = MessageNode(node_id="3", incoming=incoming, status_message_id="s1")
+        node.mark_error("temporary failure")
+
+        node.update_state(MessageState.COMPLETED, session_id="session_1")
+
+        assert node.state is MessageState.COMPLETED
+        assert node.session_id == "session_1"
+        assert node.error_message is None
+
+    def test_cancel_reason_helpers_preserve_existing_context(self):
+        """Cancellation reason is typed and does not clobber other context."""
+        incoming = IncomingMessage(
+            text="Test",
+            chat_id="1",
+            user_id="2",
+            message_id="3",
+            platform="test",
+        )
+        node = MessageNode(node_id="3", incoming=incoming, status_message_id="s1")
+        node.set_context({"other": "value"})
+
+        set_cancel_reason(node, CancellationReason.STOP)
+
+        assert get_cancel_reason(node) is CancellationReason.STOP
+        assert node.context == {"other": "value", "cancel_reason": "stop"}
+
+        set_cancel_reason(node, None)
+
+        assert get_cancel_reason(node) is None
+        assert node.context == {"other": "value"}
+
 
 class TestMessageTree:
     """Test MessageTree class."""
@@ -581,9 +625,11 @@ class TestTreeQueueManager:
         await manager.enqueue("m1", slow_processor)
         await started.wait()
 
-        cancelled = await manager.cancel_tree("m1")
+        cancelled = await manager.cancel_tree("m1", reason=CancellationReason.STOP)
 
-        assert [node.node_id for node in cancelled] == ["m1"]
+        assert [entry.node.node_id for entry in cancelled] == ["m1"]
+        assert [entry.ui_owner for entry in cancelled] == [CancellationUiOwner.RUNNER]
+        assert get_cancel_reason(cancelled[0].node) is CancellationReason.STOP
         assert cleanup_done.is_set()
 
     @pytest.mark.asyncio
@@ -612,9 +658,11 @@ class TestTreeQueueManager:
         await manager.enqueue("m1", slow_processor)
         await started.wait()
 
-        cancelled = await manager.cancel_node("m1")
+        cancelled = await manager.cancel_node("m1", reason=CancellationReason.STOP)
 
-        assert [node.node_id for node in cancelled] == ["m1"]
+        assert [entry.node.node_id for entry in cancelled] == ["m1"]
+        assert [entry.ui_owner for entry in cancelled] == [CancellationUiOwner.RUNNER]
+        assert get_cancel_reason(cancelled[0].node) is CancellationReason.STOP
         assert cleanup_done.is_set()
 
     @pytest.mark.asyncio
@@ -649,7 +697,8 @@ class TestTreeQueueManager:
 
         cancelled = await manager.cancel_branch("child")
         assert len(cancelled) == 1
-        assert cancelled[0].node_id == "child"
+        assert cancelled[0].node.node_id == "child"
+        assert cancelled[0].ui_owner is CancellationUiOwner.WORKFLOW
 
         child_node = tree.get_node("child")
         assert child_node is not None
@@ -692,7 +741,8 @@ class TestTreeQueueManager:
 
         cancelled = await manager.cancel_branch("child")
 
-        assert [node.node_id for node in cancelled] == ["child"]
+        assert [entry.node.node_id for entry in cancelled] == ["child"]
+        assert [entry.ui_owner for entry in cancelled] == [CancellationUiOwner.RUNNER]
         assert cleanup_done.is_set()
 
     @pytest.mark.asyncio
@@ -725,9 +775,13 @@ class TestTreeQueueManager:
             await manager.enqueue(node_id, slow_processor)
         await all_started.wait()
 
-        cancelled = await manager.cancel_all()
+        cancelled = await manager.cancel_all(reason=CancellationReason.STOP)
 
-        assert {node.node_id for node in cancelled} == {"a", "b"}
+        assert {entry.node.node_id for entry in cancelled} == {"a", "b"}
+        assert {entry.ui_owner for entry in cancelled} == {CancellationUiOwner.RUNNER}
+        assert {get_cancel_reason(entry.node) for entry in cancelled} == {
+            CancellationReason.STOP
+        }
         assert cleanup_done == {"a", "b"}
 
     @pytest.mark.asyncio
@@ -803,7 +857,8 @@ class TestTreeQueueManager:
 
         cancelled = await manager.cancel_node("queued_first")
 
-        assert [node.node_id for node in cancelled] == ["queued_first"]
+        assert [entry.node.node_id for entry in cancelled] == ["queued_first"]
+        assert [entry.ui_owner for entry in cancelled] == [CancellationUiOwner.WORKFLOW]
         assert await tree.get_queue_snapshot() == ["queued_second"]
         queue_updated.assert_awaited_once_with(tree)
 
@@ -845,7 +900,8 @@ class TestTreeQueueManager:
 
         cancelled = await manager.cancel_branch("queued_first")
 
-        assert [node.node_id for node in cancelled] == ["queued_first"]
+        assert [entry.node.node_id for entry in cancelled] == ["queued_first"]
+        assert [entry.ui_owner for entry in cancelled] == [CancellationUiOwner.WORKFLOW]
         assert await tree.get_queue_snapshot() == ["queued_second"]
         queue_updated.assert_awaited_once_with(tree)
 

uv.lock

@@ -561,7 +561,7 @@ wheels = [
 
 [[package]]
 name = "free-claude-code"
-version = "3.4.13"
+version = "3.4.14"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },