fix(agent): pi agent final output excludes intermediate steps (#4894) (MUL-4030)

* fix(agent): pi agent final output excludes intermediate steps

Updated PI agent to only retain the final result in JSON output.

Previously, `text_delta` included both intermediate steps and final
content.

Now, output is reset on each `text_start` to concatenate only the
final text.

* fix(agent) Replace `message_update.text_start` with `turn_start` event. add test

`turn_start` begins a new turn, Reset output on it to exclude
intermediate texts.

a1b336d73e/packages/coding-agent/docs/rpc.md (events)

server/pkg/agent/pi.go

@@ -282,6 +282,10 @@ func (b *piBackend) Execute(ctx context.Context, prompt string, opts ExecOptions
 			case "agent_start":
 				trySend(msgCh, Message{Type: MessageStatus, Status: "running"})
 
+			case "turn_start":
+				output.Reset()
+				textBuffer.Reset()
+
 			case "message_update":
 				if evt.AssistantMessageEvent == nil {
 					continue

server/pkg/agent/pi_test.go

@@ -119,6 +119,72 @@ func TestPiExecuteAttachesStdinPipe(t *testing.T) {
 	}
 }
 
+// piEventStreamScript builds a sh script that prints each JSON event on
+// its own stdout line. Fixtures must not contain single quotes.
+func piEventStreamScript(events []string) string {
+	var b strings.Builder
+	b.WriteString("#!/bin/sh\n")
+	for _, e := range events {
+		b.WriteString("printf '%s\\n' '")
+		b.WriteString(e)
+		b.WriteString("'\n")
+	}
+	return b.String()
+}
+
+// TestPiExecuteRetainsOnlyLastTurnOutput verifies turn_start resets the
+// output buffer so Result.Output keeps only the final turn's text.
+func TestPiExecuteRetainsOnlyLastTurnOutput(t *testing.T) {
+	t.Parallel()
+	if runtime.GOOS == "windows" {
+		t.Skip("shell-script fixture is POSIX-only")
+	}
+
+	events := []string{
+		`{"type":"agent_start"}`,
+		`{"type":"turn_start"}`,
+		`{"type":"message_update","assistantMessageEvent":{"type":"text_delta","delta":"intermediate"}}`,
+		`{"type":"tool_execution_start","toolCallId":"call_1","toolName":"bash","args":{"command":"echo hi"}}`,
+		`{"type":"tool_execution_end","toolCallId":"call_1","toolName":"bash","result":{"content":[{"type":"text","text":"hi"}]},"isError":false}`,
+		`{"type":"turn_end","message":{"role":"assistant","model":"test","usage":{"input":1,"output":1}}}`,
+		`{"type":"turn_start"}`,
+		`{"type":"message_update","assistantMessageEvent":{"type":"text_delta","delta":"final"}}`,
+		`{"type":"message_update","assistantMessageEvent":{"type":"text_delta","delta":" "}}`,
+		`{"type":"message_update","assistantMessageEvent":{"type":"text_delta","delta":"answer"}}`,
+		`{"type":"turn_end","message":{"role":"assistant","model":"test","usage":{"input":2,"output":2}}}`,
+	}
+	fakePath := filepath.Join(t.TempDir(), "pi")
+	writeTestExecutable(t, fakePath, []byte(piEventStreamScript(events)))
+
+	backend, err := New("pi", Config{ExecutablePath: fakePath, Logger: slog.Default()})
+	if err != nil {
+		t.Fatalf("new pi backend: %v", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	session, err := backend.Execute(ctx, "prompt-ignored", ExecOptions{Timeout: 5 * time.Second})
+	if err != nil {
+		t.Fatalf("execute: %v", err)
+	}
+	go func() {
+		for range session.Messages {
+		}
+	}()
+
+	select {
+	case result := <-session.Result:
+		if result.Status != "completed" {
+			t.Fatalf("expected status=completed, got %q (error=%q)", result.Status, result.Error)
+		}
+		if result.Output != "final answer" {
+			t.Fatalf("Output: got %q, want %q", result.Output, "final answer")
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("timeout waiting for result")
+	}
+}
+
 func TestStripPiToolCallMarkup(t *testing.T) {
 	tests := map[string]string{
 		`before call:bash{command:<|"|>cd repo/path && ls -F<|"|>}<tool_call|> after`:                           "before  after",