fix: rewrite prefix after carriage return (#6989)

### Description

Fixes #6969

If we encounter a `\r` in a task's outputs we make sure we rewrite our
prefix so it doesn't get overwritten by the output that follows the
`\r`.

This does result in slightly more writing to stdout than strictly
necessary on systems that use `CRLF` instead of just `LF` as we will now
print out our prefix before processing `LF`. The user still sees the
correct output and this keeps the implementation simple.

[`\r`
reference](https://en.wikipedia.org/wiki/Carriage_return#Computers)

### Testing Instructions

Unit tests for a few scenarios, but primary way to test is to run `next`
using `turbo`:
```
[0 olszewski@chriss-mbp] /tmp $ npx create-turbo@latest cr-test pnpm
[0 olszewski@chriss-mbp] /tmp $ cd cr-test 
[0 olszewski@chriss-mbp] /tmp/cr-test $ pnpm rm turbo # make sure we're using locally built turbo
```
<img width="612" alt="Screenshot 2024-01-11 at 10 11 46 AM"
src="https://github.com/vercel/turbo/assets/4131117/20ad5bfd-176c-486b-8f05-d77206168af0">



Closes TURBO-2030

crates/turborepo-ui/src/prefixed.rs

@@ -128,12 +128,24 @@ impl<W: Write> PrefixedWriter<W> {
 
 impl<W: Write> Write for PrefixedWriter<W> {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-        self.writer.write_all(self.prefix.as_bytes())?;
-
+        let mut is_first = true;
+        for chunk in buf.split_inclusive(|c| *c == b'\r') {
+            // Before we write the chunk we write the prefix as either:
+            // - this is the first iteration and we haven't written the prefix
+            // - the previous chunk ended with a \r and the cursor is currently as the start
+            //   of the line so we want to rewrite the prefix over the existing prefix in
+            //   the line
+            // or if the last chunk is just a newline we can skip rewriting the prefix
+            if is_first || chunk != b"\n" {
+                self.writer.write_all(self.prefix.as_bytes())?;
+            }
+            self.writer.write_all(chunk)?;
+            is_first = false;
+        }
         // We do end up writing more bytes than this to the underlying writer, but we
         // cannot report this to the callers as the amount of bytes we report
         // written must be less than or equal to the number of bytes in the buffer.
-        self.writer.write(buf)
+        Ok(buf.len())
     }
 
     fn flush(&mut self) -> std::io::Result<()> {
@@ -192,4 +204,22 @@ mod test {
         writer.write_all(b"cool!").unwrap();
         assert_eq!(String::from_utf8(buffer).unwrap(), expected);
     }
+
+    #[test_case("\ra whole message \n", "turbo > \rturbo > a whole message \n" ; "basic prefix cr")]
+    #[test_case("no return", "turbo > no return" ; "no return")]
+    #[test_case("foo\rbar\rbaz", "turbo > foo\rturbo > bar\rturbo > baz" ; "multiple crs")]
+    #[test_case("foo\r", "turbo > foo\r" ; "trailing cr")]
+    #[test_case("foo\r\n", "turbo > foo\r\n" ; "no double write on crlf")]
+    #[test_case("\n", "turbo > \n" ; "leading new line")]
+    fn test_prefixed_writer_cr(input: &str, expected: &str) {
+        let mut buffer = Vec::new();
+        let mut writer = PrefixedWriter::new(
+            UI::new(false),
+            Style::new().apply_to("turbo > "),
+            &mut buffer,
+        );
+
+        writer.write_all(input.as_bytes()).unwrap();
+        assert_eq!(String::from_utf8(buffer).unwrap(), expected);
+    }
 }