hive-runtime: report an empty ACP end_turn as a turn error (#4819)

opencode answers `end_turn` even when its provider rejected the request
with a non-retryable error (401, 400), and forwards nothing over ACP, so
the turn looked like an empty success. A turn that ends with `end_turn`,
no event, no `usage_update` and no `usage` in the prompt response now
fails with `AcpError::EmptyEndTurn`.

A turn that really produced nothing and reported no usage is reported
the same way; that false positive is accepted.

The test agent's plain `end_turn` replies now carry a response `usage`,
so its ordinary turns stay successes; a response `usage` feeds only cost
telemetry, not the compaction watermark. A new `blank` mode keeps the
empty reply for the error case.

hive-runtime/src/acp/mod.rs

@@ -107,6 +107,8 @@ pub enum AcpError {
          provider error the agent retries without reporting it, such as HTTP 429, looks like this"
     )]
     IdleTimeout { silent_secs: u64 },
+    #[error("agent ended the turn with no output; likely a provider error")]
+    EmptyEndTurn,
     #[error(
         "ACP agent sent nothing for {silent_secs}s and did not stop within {grace_secs}s of \
          `session/cancel`, so it was killed"
@@ -413,7 +415,12 @@ impl<P: CompactionPolicy> AcpRuntime<P> {
             Some((Stop::Operator, _)) => Some("cancelled"),
             None => response["stopReason"].as_str(),
         };
+        let empty = stop_reason == Some("end_turn")
+            && !mapper.has_content()
+            && !mapper.has_usage_update()
+            && response.get("usage").is_none();
         match stop_reason {
+            Some("end_turn") if empty => return Err(AcpError::EmptyEndTurn.into()),
             Some("end_turn") => {}
             reason => {
                 let reason = reason.unwrap_or("none");
@@ -736,16 +743,19 @@ mod tests {
     /// sessions `s1`, `s2`, …, appends every method it is sent to
     /// `<log>.methods` (and `start` when it starts), and every
     /// `session/prompt` line to `<log>`. It answers every prompt with
-    /// `end_turn`, except the very first one it is sent across restarts,
-    /// which it handles per its mode:
+    /// `end_turn` and a token `usage`, except the very first one it is sent
+    /// across restarts, which it handles per its mode:
     ///
     /// - `fail`: an error;
     /// - `silent`: nothing until `session/cancel`, then `end_turn` — the
     ///   reply an agent gives a cancelled prompt it was retrying against a
     ///   rate-limited (HTTP 429) provider without reporting it;
     /// - `deaf`: nothing, and `session/cancel` is ignored;
-    /// - `trickle`: five text chunks 100ms apart, then `end_turn`;
-    /// - `ok`: `end_turn`, like the rest.
+    /// - `trickle`: five text chunks 100ms apart, then `end_turn` with no
+    ///   `usage`;
+    /// - `blank`: `end_turn` with no `usage` and nothing before it — how
+    ///   opencode ends a turn its provider rejected;
+    /// - `ok`: `end_turn` and `usage`, like the rest.
     ///
     /// With `COMMANDS` set, it advertises that one command on each session it
     /// creates or loads. With `USED` set, it reports that many of 1000 context
@@ -760,6 +770,9 @@ update() {
 advertise() {
   [ -n "$COMMANDS" ] && update "$1" "{\"sessionUpdate\":\"available_commands_update\",\"availableCommands\":[{\"name\":\"$COMMANDS\",\"description\":\"\"}]}"
 }
+ended() {
+  printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn","usage":{"inputTokens":1,"outputTokens":1}}}\n' "$1"
+}
 while IFS= read -r line; do
   id=${line#*\"id\":}; id=${id%%[,\}]*}
   m=${line#*\"method\":\"}; m=${m%%\"*}
@@ -782,13 +795,14 @@ while IFS= read -r line; do
         *'"text":"/compact"'*) [ -n "$STALL_COMPACT" ] && prompt=$id && continue ;;
       esac
       if [ -e "$1.once" ]; then
-        printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id"
+        ended "$id"
       else
         : > "$1.once"
         case $2 in
           fail)
             printf '{"jsonrpc":"2.0","id":%s,"error":{"code":-32603,"message":"provider down"}}\n' "$id" ;;
-          ok)
+          ok) ended "$id" ;;
+          blank)
             printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id" ;;
           silent) prompt=$id ;;
           trickle)
@@ -1018,6 +1032,27 @@ done
         assert_eq!(count(dir.path(), "session/cancel"), 0);
     }
 
+    #[tokio::test]
+    async fn an_end_turn_with_no_output_and_no_usage_fails() {
+        let dir = tempfile::tempdir().unwrap();
+        let (runtime, config) = (runtime(dir.path(), "blank"), config(dir.path()));
+
+        let empty = runtime.run(&config, "one", &NoopSink).await;
+        assert!(
+            matches!(empty, Err(Error::Acp(AcpError::EmptyEndTurn))),
+            "{empty:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn an_end_turn_with_output_but_no_usage_succeeds() {
+        let dir = tempfile::tempdir().unwrap();
+        let (runtime, config) = (runtime(dir.path(), "trickle"), config(dir.path()));
+
+        let done = runtime.run(&config, "one", &NoopSink).await;
+        assert!(done.is_ok(), "{done:?}");
+    }
+
     #[tokio::test]
     async fn a_cancelled_turn_stops_and_ends_normally() {
         let dir = tempfile::tempdir().unwrap();

hive-runtime/src/acp/stream.rs

@@ -21,6 +21,7 @@ pub(super) struct StreamMapper {
     thought: String,
     tools: HashMap<String, ToolCall>,
     usage: Option<(u64, u64)>,
+    emitted: bool,
 }
 
 struct ToolCall {
@@ -39,6 +40,7 @@ impl StreamMapper {
             thought: String::new(),
             tools: HashMap::new(),
             usage: None,
+            emitted: false,
         }
     }
 
@@ -69,6 +71,7 @@ impl StreamMapper {
             }
             _ => {}
         }
+        self.emitted |= !out.is_empty();
         out
     }
 
@@ -82,9 +85,20 @@ impl StreamMapper {
         for (id, tool) in pending {
             out.push(tool_use(&id, &tool.name, &tool.input));
         }
+        self.emitted |= !out.is_empty();
         out
     }
 
+    /// Whether the turn has emitted any event.
+    pub(super) fn has_content(&self) -> bool {
+        self.emitted
+    }
+
+    /// Whether the turn has had a `usage_update`.
+    pub(super) fn has_usage_update(&self) -> bool {
+        self.usage.is_some()
+    }
+
     /// The turn's telemetry: context from the last `usage_update`, cost from
     /// the `session/prompt` response's `usage` when the agent sends one.
     pub(super) fn telemetry(&self, response: &Value, model: Option<&str>) -> Telemetry {