feat(chat): human-language agent processing — server tool labels + persisted thoughts (#4054)

tests/agent_harness_e2e.rs

@@ -756,9 +756,11 @@ async fn subagent_delegation_happy_path_inner() {
     //   request[1] = researcher subagent inner LLM call: canary text returned
     //   request[2] = orchestrator synthesis: canary forwarded in final reply
     //
-    // NOTE: threads_turn_state_get returns None after a successful turn completion —
-    // TurnStateMirror deletes the snapshot on TurnCompleted (mirror.rs:338-341).
-    // Therefore we verify delegation via captured upstream requests, not turn state.
+    // NOTE: a completed turn's snapshot is now RETAINED (lifecycle `Completed`)
+    // so "View processing" can replay a finished turn; the snapshot is overwritten
+    // by the next turn, not deleted on completion. We verify delegation via the
+    // captured upstream requests rather than turn state, which keeps this test
+    // independent of the snapshot's retention/lifecycle details.
     let requests = with_captured(|c| c.clone());
     assert!(
         requests.len() >= 3,

tests/memory_core_threads_raw_coverage_e2e.rs

@@ -550,6 +550,8 @@ async fn thread_ops_welcome_migration_and_turn_state_cover_error_and_cleanup_pat
         tool_name: "memory.search".into(),
         arguments: json!({"q": "phoenix"}),
         iteration: 1,
+        display_label: None,
+        display_detail: None,
     }));
     assert!(mirror.observe(&AgentProgress::SubagentSpawned {
         agent_id: "researcher".into(),

tests/memory_raw_coverage_e2e.rs

@@ -611,7 +611,10 @@ fn threads_turn_state_store_skips_corrupt_entries_and_marks_interrupted() {
                 iteration: Some(1),
                 elapsed_ms: Some(20),
                 output_chars: Some(64),
+                display_name: None,
+                detail: None,
             }],
+            transcript: vec![],
         }),
     });
     let second = TurnState::started("thread-b", "req-b", 2, "2026-05-29T12:01:00Z");

tests/memory_threads_raw_coverage_e2e.rs

@@ -3042,6 +3042,14 @@ impl ComposioProvider for EmptySlugProvider {
 #[tokio::test]
 async fn memory_sync_provider_trait_defaults_and_connection_hook_are_deterministic() {
     let tmp = TempDir::new().expect("tempdir");
+    // Determinism (as this test's name promises): `identity_set` →
+    // `persist_provider_profile` writes through the PROCESS-GLOBAL memory client
+    // and returns 0 when it isn't ready. Other tests in this binary rebind that
+    // global, so under parallel execution this test could otherwise observe an
+    // unready client and see 0 instead of 1. Bind the global to this test's
+    // workspace up front so the assertion is independent of execution order.
+    openhuman_core::openhuman::memory::global::init(tmp.path().to_path_buf())
+        .expect("init global memory client");
     let ctx = ProviderContext {
         config: Arc::new(config_in(&tmp)),
         toolkit: "raw_coverage".into(),
@@ -3140,6 +3148,8 @@ fn turn_state_mirror_persists_progress_edges_from_public_events() {
         tool_name: "memory.search".into(),
         arguments: json!({ "q": "coverage" }),
         iteration: 2,
+        display_label: None,
+        display_detail: None,
     }));
     assert!(mirror.observe(&AgentProgress::ToolCallCompleted {
         call_id: "call-1".into(),
@@ -3174,15 +3184,22 @@ fn turn_state_mirror_persists_progress_edges_from_public_events() {
         max_iterations: 3,
         extended_policy: false,
     }));
-    assert!(!mirror.observe(&AgentProgress::SubagentToolCallStarted {
+    // Sub-agent tool boundaries now FLUSH (return `true`): while a sub-agent
+    // runs, the parent is blocked on its long-lived `spawn_subagent` tool, so
+    // the parent's own iteration/tool flushes don't fire. Flushing on the
+    // child's tool start/complete is what lets the sub-agent's streamed thoughts
+    // reach disk mid-run (survives tab-switch / reload).
+    assert!(mirror.observe(&AgentProgress::SubagentToolCallStarted {
         agent_id: "researcher".into(),
         task_id: "task-1".into(),
         call_id: "child-call".into(),
         tool_name: "memory.read".into(),
         arguments: serde_json::Value::Null,
         iteration: 1,
+        display_label: None,
+        display_detail: None,
     }));
-    assert!(!mirror.observe(&AgentProgress::SubagentToolCallCompleted {
+    assert!(mirror.observe(&AgentProgress::SubagentToolCallCompleted {
         agent_id: "researcher".into(),
         task_id: "task-1".into(),
         call_id: "child-call".into(),
@@ -3259,10 +3276,14 @@ fn turn_state_mirror_persists_progress_edges_from_public_events() {
     let mut complete = TurnStateMirror::new(store.clone(), "thread/completed", "request-complete");
     assert!(complete.observe(&AgentProgress::TurnCompleted { iterations: 2 }));
     complete.finish();
-    assert!(store
+    // A completed turn's snapshot is now RETAINED (lifecycle `Completed`) rather
+    // than deleted, so the "View processing" panel can replay a finished turn
+    // after reload. `finish()`/mark-all-interrupted skips `Completed` snapshots.
+    let completed = store
         .get("thread/completed")
         .expect("completed snapshot lookup")
-        .is_none());
+        .expect("completed snapshot retained for replay");
+    assert_eq!(completed.lifecycle, TurnLifecycle::Completed);
 }
 
 #[test]
@@ -3451,7 +3472,10 @@ fn turn_state_store_persists_lists_marks_and_clears_snapshots() {
                 iteration: Some(1),
                 elapsed_ms: Some(100),
                 output_chars: Some(10),
+                display_name: None,
+                detail: None,
             }],
+            transcript: vec![],
         }),
     });
     let second = TurnState::started("thread/b", "request-2", 2, "2026-05-29T12:01:00Z");