Fix DeepSWE hive action turn isolation

examples/openhuman/src/bin/deepswe_hive.rs

@@ -344,7 +344,6 @@ async fn run_seat(
     outbox: PathBuf,
     turn_timeout: Duration,
 ) -> anyhow::Result<(String, String, tinyhivemind::speech::Utterance)> {
-    let session = format!("deepswe-{}:{id}", task.instance_id);
     let mut retry_reason = None;
     mcp::clear(&outbox)?;
     for attempt in 1..=MAX_SEAT_ATTEMPTS {
@@ -354,11 +353,7 @@ async fn run_seat(
             )
         });
         let prompt = seat_prompt(&task, &id, &delta, retry_instruction.as_deref());
-        let send = tokio::time::timeout(
-            turn_timeout,
-            agent.turn(prompt).session(session.clone()).send(),
-        )
-        .await;
+        let send = tokio::time::timeout(turn_timeout, agent.turn(prompt).send()).await;
         let utterances = mcp::drain(&outbox)?;
         if utterances.len() > 1 {
             anyhow::bail!(

examples/openhuman/src/bin/deepswe_hive/README.md

@@ -54,8 +54,11 @@ accepted action and that continuation error, and that seat is not rerun. More
 than one action always fails immediately.
 
 A zero-action protocol miss or retryable provider failure may retry within the
-same budget, OpenHuman session, and frozen pre-round view. The outbox is cleared
-before the first attempt and again only after such a proven zero-action outcome.
+same budget and frozen pre-round view. Every attempt uses a fresh OpenHuman
+conversation; the TinyHiveMind desk is the only cross-turn transcript, so an
+earlier native-action acceptance cannot satisfy a later attempt or hive round.
+The outbox is cleared before the first attempt and again only after such a
+proven zero-action outcome.
 A timeout is ambiguous because a cancelled turn may still write late, so it
 fails closed without retry even when no action was observed. Provider
 retryability uses OpenHuman's structured `retryable` field when present, then a

examples/openhuman/src/bin/deepswe_hive/mcp.rs

@@ -102,18 +102,18 @@ pub(super) fn response(server: &Server, request: &Value, id: Value) -> Value {
         "tools/call" => match call(server, request) {
             Ok(text) => success(id, json!({"content": [{"type": "text", "text": text}]})),
             Err(CallError::InvalidParams(error)) => json!({
-                    "jsonrpc":"2.0", "id":id,
-                    "error":{"code":-32602,"message":error}
-                }),
+                "jsonrpc":"2.0", "id":id,
+                "error":{"code":-32602,"message":error}
+            }),
             Err(CallError::Execution(error)) => json!({
-                    "jsonrpc":"2.0", "id":id,
-                    "result":{"content":[{"type":"text","text":error}],"isError":true}
-                }),
-        },
-        other => json!({
                 "jsonrpc":"2.0", "id":id,
-                "error":{"code":-32601,"message":format!("unknown method {other}")}
+                "result":{"content":[{"type":"text","text":error}],"isError":true}
             }),
+        },
+        other => json!({
+            "jsonrpc":"2.0", "id":id,
+            "error":{"code":-32601,"message":format!("unknown method {other}")}
+        }),
     }
 }
 
@@ -146,7 +146,11 @@ fn call(server: &Server, request: &Value) -> Result<String, CallError> {
     let required = match server {
         Server::Hive { .. } => match name {
             "broadcast" | "complete_episode" => ["message"].as_slice(),
-            _ => return Err(CallError::InvalidParams(format!("unknown hive tool {name}"))),
+            _ => {
+                return Err(CallError::InvalidParams(format!(
+                    "unknown hive tool {name}"
+                )));
+            }
         },
         Server::Workspace { .. } => match name {
             "file_read" => ["path"].as_slice(),
@@ -317,7 +321,9 @@ fn take(
 }
 
 pub(super) fn clear(path: &Path) -> anyhow::Result<()> {
-    let parent = path.parent().filter(|parent| !parent.as_os_str().is_empty())
+    let parent = path
+        .parent()
+        .filter(|parent| !parent.as_os_str().is_empty())
         .unwrap_or_else(|| Path::new("."));
     let parent_metadata = std::fs::symlink_metadata(parent)?;
     if !parent_metadata.is_dir() || parent_metadata.file_type().is_symlink() {

examples/openhuman/src/bin/deepswe_hive/sandbox.rs

@@ -429,7 +429,9 @@ pub(super) fn mount(source: &Path, destination: &str, writable: bool) -> anyhow:
         anyhow::bail!("Docker bind mount paths must not contain commas")
     }
     let readonly = if writable { "" } else { ",readonly" };
-    Ok(format!("type=bind,src={source},dst={destination}{readonly}"))
+    Ok(format!(
+        "type=bind,src={source},dst={destination}{readonly}"
+    ))
 }
 
 fn run_container_output(

examples/openhuman/src/bin/deepswe_hive/task.rs

@@ -150,7 +150,10 @@ impl Task {
             .args(args)
             .current_dir(&self.repo_path)
             .env_clear()
-            .env("PATH", std::env::var_os("PATH").unwrap_or_else(|| "/usr/bin:/bin:/usr/local/bin".into()))
+            .env(
+                "PATH",
+                std::env::var_os("PATH").unwrap_or_else(|| "/usr/bin:/bin:/usr/local/bin".into()),
+            )
             .output()?;
         if !output.status.success() {
             anyhow::bail!(
@@ -250,7 +253,10 @@ fn git<const N: usize>(repo_path: &Path, args: [&str; N], action: &str) -> anyho
         .args(args)
         .current_dir(repo_path)
         .env_clear()
-        .env("PATH", std::env::var_os("PATH").unwrap_or_else(|| "/usr/bin:/bin:/usr/local/bin".into()))
+        .env(
+            "PATH",
+            std::env::var_os("PATH").unwrap_or_else(|| "/usr/bin:/bin:/usr/local/bin".into()),
+        )
         .output()?;
     if !output.status.success() {
         anyhow::bail!(

examples/openhuman/src/bin/deepswe_hive/test.rs

@@ -379,7 +379,10 @@ fn refuses_to_schedule_a_round_past_the_exact_turn_cap() {
 
 #[test]
 fn docker_mounts_reject_delimiter_paths_before_argument_rendering() {
-    for (source, destination) in [("/tmp/source,comma", "/workspace"), ("/tmp/source", "/workspace,comma")] {
+    for (source, destination) in [
+        ("/tmp/source,comma", "/workspace"),
+        ("/tmp/source", "/workspace,comma"),
+    ] {
         let error = super::sandbox::mount(std::path::Path::new(source), destination, false)
             .expect_err("comma cannot enter Docker mount syntax");
         assert!(error.to_string().contains("commas"));
@@ -414,7 +417,11 @@ fn hive_mcp_writes_only_native_tool_calls_and_separates_protocol_errors() {
         serde_json::json!({"method": "tools/call", "params": {"name": "broadcast", "arguments": {}}}),
     ] {
         let response = super::mcp::response(&server, &request, serde_json::json!(2));
-        let expected = if request["method"] == "unknown" { -32601 } else { -32602 };
+        let expected = if request["method"] == "unknown" {
+            -32601
+        } else {
+            -32602
+        };
         assert_eq!(response["error"]["code"], expected);
     }
     std::fs::remove_file(&outbox).expect("remove prepared outbox");
@@ -428,10 +435,12 @@ fn hive_mcp_writes_only_native_tool_calls_and_separates_protocol_errors() {
     );
     assert_eq!(response["result"]["isError"], true);
     std::fs::write(&outbox, vec![b'x'; 64 * 1024 + 1]).expect("oversized outbox");
-    assert!(super::mcp::drain(&outbox)
-        .expect_err("oversized outbox is rejected")
-        .to_string()
-        .contains("exceeds"));
+    assert!(
+        super::mcp::drain(&outbox)
+            .expect_err("oversized outbox is rejected")
+            .to_string()
+            .contains("exceeds")
+    );
 }
 
 #[test]

examples/openhuman/src/bin/deepswe_hive/test/retry.rs

@@ -22,7 +22,8 @@ use crate::sandbox::{DockerSandbox, SandboxConfig};
 mod response;
 
 use response::{
-    completion_response, empty_completion_response, provider_error_response, tool_call_response,
+    completion_response, empty_completion_response, hive_action_response, provider_error_response,
+    tool_call_response,
 };
 
 const PRINTED_JSON: &str =
@@ -41,6 +42,8 @@ struct ScriptedHive {
     lead_misses: u32,
     lead_failure: LeadFailure,
     lead_continuation_failure: Option<ContinuationFailure>,
+    broadcast_first_turn: bool,
+    refuse_after_stale_acceptance: bool,
 }
 
 #[derive(Clone, Copy)]
@@ -89,6 +92,15 @@ impl Respond for ScriptedHive {
         }
 
         assert_eq!(last_role, "user", "unexpected provider turn boundary");
+        if self.refuse_after_stale_acceptance
+            && messages
+                .iter()
+                .filter(|message| message["role"] == "user")
+                .count()
+                > 1
+        {
+            return completion_response("stale action already accepted");
+        }
         let attempt = {
             let entry = state.turn_starts.entry(seat.clone()).or_default();
             *entry = entry.saturating_add(1);
@@ -103,6 +115,9 @@ impl Respond for ScriptedHive {
                 LeadFailure::Delay(duration) => completion_response("too late").set_delay(duration),
             };
         }
+        if self.broadcast_first_turn && attempt == 1 {
+            return hive_action_response(&seat, state.calls, "broadcast");
+        }
         tool_call_response(&seat, state.calls)
     }
 }
@@ -146,11 +161,14 @@ while IFS= read -r line; do
   id=$(printf '%s' "$line" | sed -E 's/.*"id"[ ]*:[ ]*([^,}]+).*/\1/')
   case "$line" in
     *initialize*) result='{"protocolVersion":"2024-11-05","capabilities":{"tools":{}},"serverInfo":{"name":"fixture","version":"1"}}' ;;
-    *tools/list*) result='{"tools":[{"name":"complete_episode","description":"complete","inputSchema":{"type":"object","properties":{"message":{"type":"string"}},"required":["message"]}}]}' ;;
+    *tools/list*) result='{"tools":[{"name":"broadcast","description":"broadcast","inputSchema":{"type":"object","properties":{"message":{"type":"string"}},"required":["message"]}},{"name":"complete_episode","description":"complete","inputSchema":{"type":"object","properties":{"message":{"type":"string"}},"required":["message"]}}]}' ;;
     *tools/call*)
+      kind=complete_episode
+      message=complete
+      case "$line" in *'"name":"broadcast"'*) kind=broadcast; message=broadcast ;; esac
       count=0
       while test "$count" -lt ACTION_COPIES; do
-        printf '{"kind":"complete_episode","message":"%s complete"}\n' "$agent" >> "$outbox"
+        printf '{"kind":"%s","message":"%s %s"}\n' "$kind" "$agent" "$message" >> "$outbox"
         count=$((count + 1))
       done
       result=$(printf '{"content":[{"type":"text","text":"accepted from @%s"}]}' "$agent")
@@ -193,12 +211,92 @@ async fn provider_with_continuation(
             lead_misses,
             lead_failure,
             lead_continuation_failure,
+            broadcast_first_turn: false,
+            refuse_after_stale_acceptance: false,
+        })
+        .mount(&provider)
+        .await;
+    (provider, state)
+}
+
+async fn two_round_provider() -> (MockServer, Arc<Mutex<ScriptState>>) {
+    let provider = MockServer::start().await;
+    let state = Arc::new(Mutex::new(ScriptState::default()));
+    Mock::given(method("POST"))
+        .and(path("/v1/chat/completions"))
+        .respond_with(ScriptedHive {
+            state: Arc::clone(&state),
+            lead_misses: 0,
+            lead_failure: LeadFailure::Protocol,
+            lead_continuation_failure: None,
+            broadcast_first_turn: true,
+            refuse_after_stale_acceptance: true,
         })
         .mount(&provider)
         .await;
     (provider, state)
 }
 
+#[test]
+fn a_new_hive_turn_does_not_inherit_a_prior_action_acceptance() {
+    let _guard = retry_test_guard();
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .thread_stack_size(16 * 1024 * 1024)
+        .build()
+        .expect("test runtime");
+    runtime.block_on(async {
+        let (directory, task) = fixture();
+        let sandbox = fake_sandbox(&directory, &task);
+        let mcp = fake_mcp(&directory, 1);
+        let (provider, state) = two_round_provider().await;
+        let output_directory = TempDir::new().expect("output directory");
+        let output = output_directory.path().join("output.json");
+        let cli = Cli {
+            task: directory.path().join("unused.json"),
+            api_base: format!("{}/v1", provider.uri()),
+            model: DEFAULT_MODEL.into(),
+            output: output.clone(),
+        };
+
+        let error = tokio::spawn(async move {
+            run_with_mcp_executable(cli, task, sandbox, "loopback-only".into(), &mcp).await
+        })
+        .await
+        .expect("adapter task")
+        .expect_err("empty patch makes the completed episode fail");
+        assert!(
+            error.to_string().contains("episode result is failed"),
+            "a stale acceptance poisoned the next hive turn: {error:#}"
+        );
+        let result: Value = serde_json::from_slice(&std::fs::read(&output).expect("result file"))
+            .expect("result JSON");
+        assert_eq!(result["turns"], 8, "two four-seat rounds commit");
+
+        let state = state.lock().expect("script state");
+        for seat in SEATS {
+            assert_eq!(state.turn_starts.get(seat), Some(&2), "@{seat} turn count");
+        }
+        let second_lead = state
+            .start_requests
+            .iter()
+            .filter(|(seat, _)| seat == "lead")
+            .nth(1)
+            .map(|(_, request)| request)
+            .expect("second lead turn");
+        assert!(
+            second_lead["messages"]
+                .as_array()
+                .expect("messages")
+                .iter()
+                .filter(|message| message["role"] == "user")
+                .count()
+                == 1,
+            "the next hive turn retained an old prompt: {second_lead:#}"
+        );
+    });
+}
+
 #[test]
 fn accepted_action_survives_post_tool_provider_failure_without_retry() {
     let _guard = retry_test_guard();
@@ -356,7 +454,7 @@ fn retry_test_guard() -> std::sync::MutexGuard<'static, ()> {
 }
 
 #[test]
-fn one_missing_seat_retries_in_session_and_round_commits_once() {
+fn one_missing_seat_retries_in_a_fresh_session_and_round_commits_once() {
     let _guard = retry_test_guard();
     let runtime = tokio::runtime::Builder::new_multi_thread()
         .enable_all()
@@ -416,7 +514,8 @@ fn one_missing_seat_retries_in_session_and_round_commits_once() {
         assert!(
             messages
                 .iter()
-                .any(|message| message["content"] == PRINTED_JSON)
+                .all(|message| message["content"] != PRINTED_JSON),
+            "the retry retained the invalid provider response"
         );
         let retry_prompt = messages
             .last()
@@ -555,7 +654,11 @@ fn empty_provider_response_retries_only_that_seat_and_commits_once() {
         assert_eq!(result["turns"], 4, "provider failure is not committed");
 
         let state = state.lock().expect("script state");
-        assert_eq!(state.turn_starts.get("lead"), Some(&2));
+        assert_eq!(
+            state.turn_starts.get("lead"),
+            Some(&3),
+            "one OpenHuman empty-response fallback and one fresh hive retry"
+        );
         for seat in ["implementer", "tester", "reviewer"] {
             assert_eq!(state.turn_starts.get(seat), Some(&1), "@{seat} reran");
         }

examples/openhuman/src/bin/deepswe_hive/test/retry/response.rs

@@ -53,10 +53,18 @@ pub(super) fn completion_response(content: &str) -> ResponseTemplate {
 }
 
 pub(super) fn tool_call_response(seat: &str, serial: u32) -> ResponseTemplate {
+    hive_action_response(seat, serial, "complete_episode")
+}
+
+pub(super) fn hive_action_response(seat: &str, serial: u32, tool: &str) -> ResponseTemplate {
+    let message = match tool {
+        "complete_episode" => format!("{seat} complete"),
+        _ => format!("{seat} {tool}"),
+    };
     let arguments = json!({
         "server":"tinyhive",
-        "tool":"complete_episode",
-        "arguments":{"message":format!("{seat} complete")}
+        "tool":tool,
+        "arguments":{"message":message}
     })
     .to_string();
     ResponseTemplate::new(200).set_body_json(json!({

examples/openhuman/src/bin/deepswe_hive/test/security.rs

@@ -130,10 +130,13 @@ fn outbox_initialization_truncates_regular_files_and_rejects_hostile_children()
     let child = directory.path().join("lead.jsonl");
     std::fs::write(&child, "stale action\n").expect("regular child");
     super::super::mcp::clear(&child).expect("regular child truncates");
-    assert!(std::fs::read_to_string(&child).expect("read child").is_empty());
+    assert!(
+        std::fs::read_to_string(&child)
+            .expect("read child")
+            .is_empty()
+    );
     std::fs::remove_file(&child).expect("remove child");
-    std::os::unix::fs::symlink(directory.path().join("missing"), &child)
-        .expect("outbox symlink");
+    std::os::unix::fs::symlink(directory.path().join("missing"), &child).expect("outbox symlink");
     let error = super::super::mcp::clear(&child).expect_err("hostile child rejected");
     assert!(error.to_string().contains("regular file"), "{error:#}");
 }