fix(claude_code): do not re-send already delivered user turns to a shared session

crates/tinyagents-harness/src/providers/claude_code/README.md

@@ -18,7 +18,7 @@ surfaced](#tool-calls-are-not-surfaced)).
 | `auth_status.rs` | `probe()` — the CLI's own auth state (API key env / subscription login / signed out / unknown), via `claude auth status --json`. |
 | `driver.rs` | `run_turn`: per-turn scratch dir, permission posture, macOS Seatbelt jail, MCP config, argv, stdin/stdout piping, `DEFAULT_TURN_TIMEOUT_SECS` (900, override `OPENHUMAN_CLAUDE_CODE_TURN_TIMEOUT_SECS`), stderr diagnostics cap. |
 | `event_mapper.rs` | `ClaudeCodeEvent` → `ProviderDelta` / aggregated `ChatResponse`. `tool_use` blocks are tracked only to keep their `input_json_delta`s out of the visible text and are **not** surfaced as tool calls: the CLI is self-executing, so a tool block in the stream has already run; surfacing it would make the harness try to dispatch `Read`/`Bash`/… and loop on "unknown tool". |
-| `input_builder.rs` | `build_stdin`: JSONL user turns for `--input-format stream-json` — full history folded into a text preamble on a new session, only the pending user turn(s) on `--resume`; inline images re-hydrated from `[IMAGE:...]`/`[OH_IMAGE:...]` markers (5 MiB cap). |
+| `input_builder.rs` | `build_stdin`: JSONL user turns for `--input-format stream-json` — full history folded into a text preamble on a new session, only the pending user turn(s) on `--resume`, minus any already delivered to that session (`session_store` keeps per-thread fingerprints, so services or loop iterations sharing a thread do not re-send a turn); inline images re-hydrated from `[IMAGE:...]`/`[OH_IMAGE:...]` markers (5 MiB cap). |
 | `session_store.rs` | `SessionStore`: thread key → CC session UUID v4, persisted in `<workspace_dir>/claude-code-sessions.json`. |
 | `settings.rs` | `ClaudeCodeSettings { full_access }` persisted in `<workspace_dir>/claude_code_settings.json`. |
 | `stream_parser.rs` | Line-buffered JSONL parser for `--output-format stream-json`; permissive `serde_json::Value` payloads so a minor CLI schema bump does not break parsing. |

crates/tinyagents-harness/src/providers/claude_code/driver.rs

@@ -144,7 +144,7 @@ fn nonzero_exit_message(
 
 use super::bridge::{ChatMessage, ChatResponse, ProviderDelta};
 use super::event_mapper::EventMapper;
-use super::input_builder::build_stdin;
+use super::input_builder::{build_stdin_with_delivered, pending_fingerprints};
 use super::session_store::{SessionStore, generate_uuid_v4, is_uuid_v4};
 use super::stream_parser::{ClaudeCodeEvent, StreamJsonParser};
 
@@ -532,7 +532,17 @@ pub(crate) async fn run_turn(ctx: TurnContext<'_>) -> anyhow::Result<ChatRespons
 
     // Validate input *before* spawning so we don't launch a process we
     // can't feed (CodeRabbit: validate before spawn).
-    let stdin_bytes = build_stdin(ctx.messages, is_new);
+    // Reserve this call's pending turns atomically (per store, re-read from
+    // disk) before spawning, so two calls on one resumed thread cannot both
+    // send the same turn. The claim is released if the turn fails.
+    let pending = pending_fingerprints(ctx.messages);
+    let (delivered, claim) = if is_new || !ctx.persist_session {
+        (std::collections::HashSet::new(), None)
+    } else {
+        let (already, claim) = ctx.session_store.claim_delivered(&ctx.thread_id, &pending);
+        (already, Some(claim))
+    };
+    let stdin_bytes = build_stdin_with_delivered(ctx.messages, is_new, &delivered);
     if stdin_bytes.is_empty() {
         anyhow::bail!("[claude-code][driver] no input messages to deliver");
     }
@@ -736,6 +746,25 @@ pub(crate) async fn run_turn(ctx: TurnContext<'_>) -> anyhow::Result<ChatRespons
         }
     }
 
+    // The session now holds every pending user turn of this call. Remember
+    // them so another service or loop iteration resuming the same thread does
+    // not deliver them again (concurrent callers are covered by the claim above).
+    // One-shot (non-durable) calls never resume, so they record nothing.
+    if ctx.persist_session {
+        let accepted_id = mapper.session_id.as_deref().unwrap_or(&cc_session_id);
+        if let Err(error) = ctx.session_store.record_delivered(&ctx.thread_id, &pending) {
+            tracing::warn!(
+                "[claude-code][driver] failed to record delivered turns for thread {} session {}: {}",
+                ctx.thread_id,
+                accepted_id,
+                error
+            );
+        }
+    }
+    if let Some(claim) = claim {
+        claim.commit();
+    }
+
     Ok(mapper.into_response())
 }
 

crates/tinyagents-harness/src/providers/claude_code/driver_tests.rs

@@ -645,3 +645,86 @@ mod structured_exit {
         assert_ne!(store.get("t1").as_deref(), Some(saved));
     }
 }
+
+// ---- one session shared by several callers (openhuman#5877) ----
+
+/// Single-quote `s` for embedding in a `/bin/sh` script.
+#[cfg(unix)]
+fn sh_quote(s: &str) -> String {
+    format!("'{}'", s.replace('\'', "'\\''"))
+}
+
+/// Text of every user message the stub CLI received (one JSON object per line;
+/// the `---` separators the stub writes are skipped).
+#[cfg(unix)]
+fn received_user_texts(log: &str) -> Vec<String> {
+    log.lines()
+        .filter(|l| l.starts_with('{'))
+        .map(|l| {
+            let v: serde_json::Value = serde_json::from_str(l).expect("stdin line is json");
+            v["message"]["content"]
+                .as_array()
+                .expect("content blocks")
+                .iter()
+                .filter_map(|b| b["text"].as_str())
+                .collect::<Vec<_>>()
+                .join("")
+        })
+        .collect()
+}
+
+/// Repeated calls on one thread (sequential services or loop iterations) carry
+/// the same pending user turn. The CLI must receive its text once. Concurrent
+/// callers are covered at the store level (`claim_delivered`).
+#[cfg(unix)]
+#[tokio::test]
+async fn shared_session_receives_a_pending_user_turn_once() {
+    use std::os::unix::fs::PermissionsExt;
+    let dir = tempfile::tempdir().expect("tempdir");
+    let log = dir.path().join("stdin.log");
+    let bin = dir.path().join("claude");
+    std::fs::write(
+        &bin,
+        format!(
+            "#!/bin/sh\ncat >> {log}\necho '---' >> {log}\necho '{{\"type\":\"result\",\"subtype\":\"success\"}}'\n",
+            log = sh_quote(&log.display().to_string())
+        ),
+    )
+    .expect("script");
+    std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).expect("chmod");
+    let store = Arc::new(SessionStore::open(&dir.path().join("ws")));
+    let messages = [ChatMessage::user("UNIQUE-PENDING-TURN")];
+
+    for _ in 0..3 {
+        run_turn(TurnContext {
+            bin_path: bin.clone(),
+            workspace_dir: dir.path().join("ws"),
+            project_dir: dir.path().join("project"),
+            thread_id: "t-5877".into(),
+            persist_session: true,
+            model: "sonnet".into(),
+            append_system_prompt: None,
+            messages: &messages,
+            session_store: store.clone(),
+            stream: None,
+            anthropic_api_key: None,
+            mcp_provider: None,
+        })
+        .await
+        .expect("turn");
+    }
+
+    let logged = std::fs::read_to_string(&log).expect("log");
+    let texts = received_user_texts(&logged);
+    assert_eq!(
+        texts
+            .iter()
+            .filter(|t| t.as_str() == "UNIQUE-PENDING-TURN")
+            .count(),
+        1,
+        "pending user turn was re-sent to the shared session:\n{logged}"
+    );
+    assert_eq!(texts.len(), 3, "every call still ran the CLI");
+    assert!(texts[1].contains("already delivered"), "{texts:?}");
+    assert_eq!(logged.matches("---").count(), 3);
+}

crates/tinyagents-harness/src/providers/claude_code/input_builder.rs

@@ -15,6 +15,13 @@
 use super::bridge::ChatMessage;
 use base64::Engine as _;
 use serde_json::{Value, json};
+use sha2::{Digest, Sha256};
+use std::collections::HashSet;
+
+/// Sent on resume when every pending user turn is already in the Claude
+/// session (another service or loop iteration on the same thread delivered it).
+/// Re-sending the text would make the model see the same message twice.
+const ALREADY_DELIVERED_NOTICE: &str = "[The user's latest message was already delivered earlier in this session. Respond to it now; do not treat it as a new, repeated message.]";
 
 /// Upper bound on a single decoded inline image's byte size; larger images
 /// are dropped rather than inlined (see [`image_block`]).
@@ -23,6 +30,17 @@ const MAX_IMAGE_BYTES: usize = 5 * 1024 * 1024;
 /// Build the bytes to write to claude's stdin. Returns an empty `Vec`
 /// when there is nothing to send (caller should abort).
 pub fn build_stdin(messages: &[ChatMessage], is_new_session: bool) -> Vec<u8> {
+    build_stdin_with_delivered(messages, is_new_session, &HashSet::new())
+}
+
+/// Like [`build_stdin`], but on a resumed session skips pending user turns whose
+/// fingerprint (see [`pending_fingerprints`]) is in `delivered`: the session
+/// already received them. Ignored for a new session, which replays context.
+pub fn build_stdin_with_delivered(
+    messages: &[ChatMessage],
+    is_new_session: bool,
+    delivered: &HashSet<String>,
+) -> Vec<u8> {
     // Resolve any `[Image: … #att:<id>]` placeholders to on-disk `[IMAGE:<path>]`
     // markers so pasted images can be inlined below. No-op for messages that
     // carry no image placeholder, so plain-text turns are unaffected.
@@ -38,7 +56,30 @@ pub fn build_stdin(messages: &[ChatMessage], is_new_session: bool) -> Vec<u8> {
     // context on a new session; never resubmit an earlier answered prompt.
     let last_user_pos = non_system.iter().rposition(|m| m.role == "user");
     let active_user_pos = last_user_pos.filter(|&pos| pos == non_system.len() - 1);
-    let active_user_content = active_user_pos.and_then(|_| pending_user_content(&non_system));
+    let active_user_content = active_user_pos.and_then(|_| {
+        let skip = if is_new_session {
+            &HashSet::new()
+        } else {
+            delivered
+        };
+        let turns = pending_user_turns(&non_system);
+        let fresh: Vec<&str> = turns
+            .iter()
+            .filter(|(fingerprint, _)| !skip.contains(fingerprint))
+            .map(|(_, content)| *content)
+            .collect();
+        if turns.is_empty() {
+            None
+        } else if fresh.is_empty() {
+            tracing::debug!(
+                "[claude-code][input] all {} pending user turn(s) already delivered to session",
+                turns.len()
+            );
+            Some(ALREADY_DELIVERED_NOTICE.to_string())
+        } else {
+            Some(fresh.join("\n\n"))
+        }
+    });
 
     let mut content: Vec<Value> = Vec::new();
     if is_new_session {
@@ -86,21 +127,66 @@ pub fn build_stdin(messages: &[ChatMessage], is_new_session: bool) -> Vec<u8> {
     out.into_bytes()
 }
 
-/// Join user turns that arrived after the most recent assistant response. A
-/// single Claude input message is required, but queued steering/user messages
-/// must retain their order instead of silently dropping every turn except the
-/// last one.
-fn pending_user_content(non_system: &[&ChatMessage]) -> Option<String> {
-    let after_assistant = non_system
+/// User turns that arrived after the most recent assistant response, each with
+/// its delivery fingerprint. A single Claude input message is required, but
+/// queued steering/user messages must retain their order instead of silently
+/// dropping every turn except the last one.
+fn pending_user_turns<'a>(non_system: &[&'a ChatMessage]) -> Vec<(String, &'a str)> {
+    let last_assistant = non_system
         .iter()
-        .rposition(|message| message.role == "assistant")
-        .map_or(0, |position| position + 1);
-    let pending: Vec<&str> = non_system[after_assistant..]
+        .rposition(|message| message.role == "assistant");
+    let anchor = last_assistant.map_or("", |position| non_system[position].content.as_str());
+    // Replies are not unique ("Done" twice), so the reply's ordinal among the
+    // assistant turns marks which exchange the pending turns belong to.
+    let reply_ordinal = non_system
+        .iter()
+        .filter(|message| message.role == "assistant")
+        .count();
+    let after_assistant = last_assistant.map_or(0, |position| position + 1);
+    let mut seen: std::collections::HashMap<&str, usize> = std::collections::HashMap::new();
+    non_system[after_assistant..]
         .iter()
         .filter(|message| message.role == "user" && !message.content.is_empty())
-        .map(|message| message.content.as_str())
-        .collect();
-    (!pending.is_empty()).then(|| pending.join("\n\n"))
+        .map(|message| {
+            // Occurrence of this exact text among the pending turns: distinct
+            // for a repeated message, yet unchanged when an earlier, different
+            // pending turn is absent from another service's slice.
+            let occurrence = seen.entry(message.content.as_str()).or_insert(0);
+            let fp = fingerprint(anchor, reply_ordinal, *occurrence, &message.content);
+            *occurrence += 1;
+            (fp, message.content.as_str())
+        })
+        .collect()
+}
+
+/// Identity of one pending user turn: the assistant reply it follows (text and
+/// ordinal, so two identical replies are still different boundaries), how many
+/// times the same text already appeared among the pending turns, and the text.
+/// User-turn history position is deliberately not part of it, so services that
+/// hold different slices of the same thread still agree on it, while a user
+/// repeating the same words after a new reply gets a new identity.
+fn fingerprint(anchor: &str, reply_ordinal: usize, occurrence: usize, content: &str) -> String {
+    let mut hasher = Sha256::new();
+    hasher.update(anchor.as_bytes());
+    hasher.update([0u8]);
+    hasher.update(reply_ordinal.to_le_bytes());
+    hasher.update(occurrence.to_le_bytes());
+    hasher.update(content.as_bytes());
+    let digest = hasher.finalize();
+    digest[..12].iter().map(|b| format!("{b:02x}")).collect()
+}
+
+/// Fingerprints of the user turns a resumed call would deliver for `messages`
+/// (empty unless the final non-system turn is from the user).
+pub fn pending_fingerprints(messages: &[ChatMessage]) -> Vec<String> {
+    let non_system: Vec<&ChatMessage> = messages.iter().filter(|m| m.role != "system").collect();
+    if non_system.last().is_none_or(|m| m.role != "user") {
+        return Vec::new();
+    }
+    pending_user_turns(&non_system)
+        .into_iter()
+        .map(|(fingerprint, _)| fingerprint)
+        .collect()
 }
 
 /// Render the turns before `end` (the latest user turn) as a plain-text preamble