fix(execution): gate retry admission by error category

src/apps/cli/tests/cli_command_contracts/exec_cli_contracts.rs

@@ -368,44 +368,52 @@ fn stream_json_patch_success_emits_one_success_terminal() {
 }
 
 #[test]
-fn stream_json_malformed_sse_retries_then_completes() {
+fn stream_json_malformed_sse_is_a_deterministic_terminal_error() {
     let server = MockOpenAiServer::malformed_sse_then_immediate();
     let environment = CliTestEnvironment::new();
     environment.configure_mock_model(server.base_url());
     let mut command = environment.std_command();
     command.args([
         "exec",
-        "exercise malformed provider stream retry",
+        "exercise malformed provider stream",
         "--output-format",
         "stream-json",
     ]);
     let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30));
-    server.assert_chat_completion_requests(2);
+    // A malformed SSE frame is a deterministic provider protocol error (not a
+    // transient network/rate/timeout failure), so upfix-4 admission leaves no
+    // retry budget: the turn terminates at attempt 1 after a single request.
+    server.assert_chat_completion_requests(1);
 
     let stdout = stdout(&output);
-    assert!(output.status.success(), "{}\n{stdout}", stderr(&output));
+    assert!(!output.status.success(), "{stdout}");
+    assert_eq!(output.status.code(), Some(1), "{}", stderr(&output));
     let events = jsonl_events(&stdout);
-    assert!(
-        events.iter().any(|value| {
-            value["event"]["type"] == "TextChunk"
-                && value["event"]["text"]
-                    .as_str()
-                    .is_some_and(|text| text.contains(STREAM_COMPLETED_MARKER))
-        }),
-        "retried model stream did not complete: {stdout}"
-    );
     assert_eq!(
         events
             .iter()
             .filter(|value| is_terminal_event(value))
             .count(),
         1,
-        "retried stream must emit exactly one terminal envelope: {stdout}"
+        "malformed stream must emit exactly one terminal envelope: {stdout}"
     );
     assert_eq!(
-        events.last().expect("retried stream terminal event")["event"]["type"],
-        "DialogTurnCompleted",
-        "retried stream terminal must be last: {stdout}"
+        events.last().expect("malformed stream terminal event")["event"]["type"],
+        "DialogTurnFailed",
+        "malformed stream terminal must be last: {stdout}"
+    );
+    let terminal_error = events.last().expect("malformed stream terminal event")["event"]["error"]
+        .as_str()
+        .expect("malformed stream error text");
+    assert!(
+        terminal_error.contains("SSE parsing error"),
+        "malformed stream reason was lost: {stdout}"
+    );
+    assert!(
+        events.iter().all(|value| {
+            !(value["event"]["type"] == "DialogTurnCompleted" && value["event"]["success"] == true)
+        }),
+        "malformed stream emitted a successful completion: {stdout}"
     );
 }
 
@@ -422,7 +430,10 @@ fn stream_json_provider_http_403_emits_one_error_terminal() {
         "stream-json",
     ]);
     let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30));
-    server.assert_chat_completion_requests(10);
+    // An authorization rejection (403) is a deterministic error: upfix-4 terminates
+    // the turn at attempt 1 after a single provider request instead of exhausting
+    // the retry ladder.
+    server.assert_chat_completion_requests(1);
 
     let stdout = stdout(&output);
     assert!(!output.status.success(), "{stdout}");
@@ -485,7 +496,9 @@ fn stream_json_provider_and_patch_failures_publish_one_final_classification() {
         &output_target,
     ]);
     let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30));
-    server.assert_chat_completion_requests(10);
+    // The 403 is deterministic, so only a single attempt reaches the provider: the
+    // patch delivery (not the provider) becomes the final classifier.
+    server.assert_chat_completion_requests(1);
 
     let stdout = stdout(&output);
     let stderr = stderr(&output);
@@ -535,7 +548,10 @@ fn stream_json_disconnect_then_exhausted_retry_failure_emits_one_error_terminal(
         "stream-json",
     ]);
     let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30));
-    server.assert_chat_completion_requests(10);
+    // The first attempt's mid-stream disconnect is transient (Network), so upfix-4
+    // still retries once; the retried request is a deterministic 403, which then
+    // terminates the turn at a terminal error (two provider requests in total).
+    server.assert_chat_completion_requests(2);
 
     let stdout = stdout(&output);
     assert!(!output.status.success(), "{stdout}");

src/crates/assembly/core/src/agentic/execution/round_executor.rs

@@ -433,7 +433,17 @@ impl RoundExecutor {
                     error!("AI request failed: {}", e);
                     let provider_error = e.downcast_ref::<AiProviderError>().cloned();
                     let err_msg = e.to_string();
-                    if local_attempt_index < max_attempts - 1 {
+                    // Classify upfront so the retry admission gate can reject deterministic
+                    // request errors (4xx/context_length_exceeded) without replaying the request.
+                    let category = provider_error
+                        .as_ref()
+                        .map(|error| error.category.clone())
+                        .unwrap_or_else(|| {
+                            bitfun_core_types::errors::classify_ai_error_message(&err_msg)
+                        });
+                    if local_attempt_index < max_attempts - 1
+                        && bitfun_core_types::errors::is_retryable_category(&category)
+                    {
                         self.record_retry_diagnostic(
                             &context,
                             &round_id,
@@ -463,12 +473,6 @@ impl RoundExecutor {
                         local_attempt_index += 1;
                         continue;
                     }
-                    let category = provider_error
-                        .as_ref()
-                        .map(|error| error.category.clone())
-                        .unwrap_or_else(|| {
-                            bitfun_core_types::errors::classify_ai_error_message(&err_msg)
-                        });
                     let error = if category == ErrorCategory::ContextOverflow {
                         BitFunError::RecoverableContextOverflow(provider_error.unwrap_or_else(
                             || AiProviderError::classified(err_msg, ErrorCategory::ContextOverflow),
@@ -813,7 +817,9 @@ impl RoundExecutor {
                         Self::error_trace_response("error", err_msg.clone()),
                     )
                     .await;
-                    if local_attempt_index < max_attempts - 1 {
+                    if local_attempt_index < max_attempts - 1
+                        && bitfun_core_types::errors::is_retryable_category(&stream_error_category)
+                    {
                         self.record_retry_diagnostic(
                             &context,
                             &round_id,

src/crates/contracts/core-types/src/errors.rs

@@ -422,7 +422,7 @@ fn is_context_overflow_message(message: &str) -> bool {
         || (message.contains("prompt has") && message.contains("configured context size"))
 }
 
-fn is_retryable_category(category: &ErrorCategory) -> bool {
+pub fn is_retryable_category(category: &ErrorCategory) -> bool {
     matches!(
         category,
         ErrorCategory::Network
@@ -512,7 +512,7 @@ fn extract_http_status(message: &str) -> Option<u16> {
 mod tests {
     use super::{
         ai_error_detail_from_message, classify_ai_error_message, classify_ai_error_parts,
-        AiProviderError, ErrorCategory,
+        is_retryable_category, AiProviderError, ErrorCategory,
     };
 
     #[test]
@@ -535,6 +535,28 @@ mod tests {
         );
     }
 
+    #[test]
+    fn is_retryable_category_matches_transient_vs_terminal_semantic() {
+        // Transient categories must be retryable so the round_executor retry gate keeps
+        // replaying them (preserved correct behavior).
+        assert!(is_retryable_category(&ErrorCategory::Network));
+        assert!(is_retryable_category(&ErrorCategory::RateLimit));
+        assert!(is_retryable_category(&ErrorCategory::Timeout));
+        assert!(is_retryable_category(&ErrorCategory::ProviderUnavailable));
+
+        // Deterministic request/terminal categories must NOT be retryable so the retry gate
+        // turns them into an immediate terminal error instead of replaying the request.
+        assert!(!is_retryable_category(&ErrorCategory::ContextOverflow));
+        assert!(!is_retryable_category(&ErrorCategory::InvalidRequest));
+        assert!(!is_retryable_category(&ErrorCategory::Auth));
+        assert!(!is_retryable_category(&ErrorCategory::Permission));
+        assert!(!is_retryable_category(&ErrorCategory::ProviderQuota));
+        assert!(!is_retryable_category(&ErrorCategory::ProviderBilling));
+        assert!(!is_retryable_category(&ErrorCategory::ModelError));
+        assert!(!is_retryable_category(&ErrorCategory::ContentPolicy));
+        assert!(!is_retryable_category(&ErrorCategory::Unknown));
+    }
+
     #[test]
     fn builds_ai_error_detail_from_provider_metadata() {
         let detail = ai_error_detail_from_message(