From 3c70fe3357919d84a383e3797a0719f746b23da4 Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Mon, 31 Aug 2026 13:59:34 -0400 Subject: [PATCH 1/7] fix(llm): reduce Pi RPC message update logs --- internal/llmadapters/pi_rpc.go | 1 + internal/llmadapters/pi_rpc_test.go | 75 +++++++++++++++++++++++++---- 2 files changed, 67 insertions(+), 9 deletions(-) diff --git a/internal/llmadapters/pi_rpc.go b/internal/llmadapters/pi_rpc.go index 36e1a32..1760e9a 100644 --- a/internal/llmadapters/pi_rpc.go +++ b/internal/llmadapters/pi_rpc.go @@ -753,6 +753,7 @@ func normalizePiRPCLogLine(line []byte) []byte { return append(logLine, '\n') } event["assistantMessageEvent"] = normalizedAssistantEvent + delete(event, "message") normalized, err := json.Marshal(event) if err != nil { return append(logLine, '\n') diff --git a/internal/llmadapters/pi_rpc_test.go b/internal/llmadapters/pi_rpc_test.go index b5ebf2d..2865950 100644 --- a/internal/llmadapters/pi_rpc_test.go +++ b/internal/llmadapters/pi_rpc_test.go @@ -588,6 +588,9 @@ func TestPiRPCLogStripsCumulativeStreamingPartials(t *testing.T) { if string(response.StructuredOutput) != `{"ok":true}` { t.Fatalf("StructuredOutput = %s, want final assistant text", response.StructuredOutput) } + if stream.SessionID() != "session-1" { + t.Fatalf("SessionID = %q, want session-1", stream.SessionID()) + } logged, err := os.ReadFile(logPath) // #nosec G304 -- test reads the log path it created with t.TempDir. if err != nil { @@ -604,6 +607,7 @@ func TestPiRPCLogStripsCumulativeStreamingPartials(t *testing.T) { t.Fatalf("log = %s, want final RPC events preserved", logText) } assertPiRPCMessageUpdatesCompacted(t, logged) + assertPiRPCFinalEventsPreserved(t, logged) if len(logged) > 10_000 { t.Fatalf("log size = %d bytes, want normalized bounded log", len(logged)) } @@ -620,6 +624,9 @@ func assertPiRPCMessageUpdatesCompacted(t *testing.T, logged []byte) { if event["type"] != "message_update" { continue } + if _, ok := event["message"]; ok { + t.Fatalf("message_update = %#v, want redundant root message stripped", event) + } assistantEvent, ok := event["assistantMessageEvent"].(map[string]any) if !ok { t.Fatalf("message_update = %#v, want assistantMessageEvent", event) @@ -640,6 +647,53 @@ func assertPiRPCMessageUpdatesCompacted(t *testing.T, logged []byte) { } } +func assertPiRPCFinalEventsPreserved(t *testing.T, logged []byte) { + t.Helper() + scanner := bufio.NewScanner(strings.NewReader(string(logged))) + var messageEndFound, agentEndFound bool + for scanner.Scan() { + var event map[string]any + if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { + t.Fatalf("Unmarshal(log line): %v", err) + } + switch event["type"] { + case "message_end": + message, ok := event["message"].(map[string]any) + if !ok { + t.Fatalf("message_end = %#v, want message", event) + } + content, ok := message["content"].([]any) + if !ok || len(content) == 0 { + t.Fatalf("message_end.message = %#v, want final content", message) + } + contentBlock, ok := content[0].(map[string]any) + if !ok || contentBlock["text"] != `{"ok":true}` { + t.Fatalf("message_end.message = %#v, want final structured content", message) + } + usage, ok := message["usage"].(map[string]any) + if !ok || usage["tokensIn"] != float64(100) || usage["tokensOut"] != float64(50) { + t.Fatalf("message_end.message = %#v, want final usage", message) + } + messageEndFound = true + case "agent_end": + messages, ok := event["messages"].([]any) + if !ok || len(messages) == 0 { + t.Fatalf("agent_end = %#v, want messages", event) + } + agentEndFound = true + } + } + if err := scanner.Err(); err != nil { + t.Fatalf("Scan(log): %v", err) + } + if !messageEndFound { + t.Fatal("log missing message_end with final message content and usage") + } + if !agentEndFound { + t.Fatal("log missing agent_end messages") + } +} + func TestPiRPCProtocolFailures(t *testing.T) { t.Run("prompt response failure", func(t *testing.T) { recordPath := filepath.Join(t.TempDir(), "record.json") @@ -938,20 +992,23 @@ func TestPiRPCHelperProcess(_ *testing.T) { thinking := "" for i := 0; i < 25; i++ { thinking += strings.Repeat("x", 200) + partial := map[string]any{ + "role": "assistant", + "provider": "opencode-go", + "model": "deepseek-v4-pro", + "stopReason": "stop", + "content": []map[string]any{{"type": "thinking", "thinking": thinking}}, + "usage": map[string]any{"tokensIn": 100, "tokensOut": 50, "totalTokens": 150}, + } event := map[string]any{ - "type": "message_update", + "type": "message_update", + "message": partial, + "sessionId": "session-1", "assistantMessageEvent": map[string]any{ "type": "thinking_delta", "contentIndex": 0, "delta": "x", - "partial": map[string]any{ - "role": "assistant", - "provider": "opencode-go", - "model": "deepseek-v4-pro", - "stopReason": "stop", - "content": []map[string]any{{"type": "thinking", "thinking": thinking}}, - "usage": map[string]any{"tokensIn": 100, "tokensOut": 50, "totalTokens": 150}, - }, + "partial": partial, }, } data, _ := json.Marshal(event) From 4bd85666e53377b2f8f2ab4c770b6314daa9cf3b Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Mon, 31 Aug 2026 14:08:16 -0400 Subject: [PATCH 2/7] fix(llm): preserve Pi RPC fallback messages --- internal/llmadapters/pi_rpc.go | 7 +++++-- internal/llmadapters/pi_rpc_test.go | 17 +++++++++++++++++ 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/internal/llmadapters/pi_rpc.go b/internal/llmadapters/pi_rpc.go index 1760e9a..12fe329 100644 --- a/internal/llmadapters/pi_rpc.go +++ b/internal/llmadapters/pi_rpc.go @@ -743,7 +743,8 @@ func normalizePiRPCLogLine(line []byte) []byte { if !ok { return append(logLine, '\n') } - if compactPartial := compactPiRPCPartialForLog(partialRaw); len(compactPartial) > 0 { + compactPartial := compactPiRPCPartialForLog(partialRaw) + if len(compactPartial) > 0 { assistantEvent["partial"] = compactPartial } else { delete(assistantEvent, "partial") @@ -753,7 +754,9 @@ func normalizePiRPCLogLine(line []byte) []byte { return append(logLine, '\n') } event["assistantMessageEvent"] = normalizedAssistantEvent - delete(event, "message") + if len(compactPartial) > 0 { + delete(event, "message") + } normalized, err := json.Marshal(event) if err != nil { return append(logLine, '\n') diff --git a/internal/llmadapters/pi_rpc_test.go b/internal/llmadapters/pi_rpc_test.go index 2865950..86f0f46 100644 --- a/internal/llmadapters/pi_rpc_test.go +++ b/internal/llmadapters/pi_rpc_test.go @@ -694,6 +694,23 @@ func assertPiRPCFinalEventsPreserved(t *testing.T, logged []byte) { } } +func TestNormalizePiRPCLogLinePreservesRootMessageForUnrecognizedPartial(t *testing.T) { + line := []byte(`{"type":"message_update","message":{"role":"assistant","content":[{"type":"thinking","thinking":"cumulative"}]},"assistantMessageEvent":{"type":"thinking_delta","contentIndex":0,"delta":"x","partial":{"content":[{"type":"unknown","value":"unrecognized"}],"usage":{"tokensIn":100,"tokensOut":50}}}}`) + + normalized := normalizePiRPCLogLine(line) + var event map[string]any + if err := json.Unmarshal(normalized, &event); err != nil { + t.Fatalf("Unmarshal(normalized): %v", err) + } + message, ok := event["message"].(map[string]any) + if !ok { + t.Fatalf("normalized message_update = %#v, want root message preserved", event) + } + if message["role"] != "assistant" { + t.Fatalf("root message = %#v, want assistant message preserved", message) + } +} + func TestPiRPCProtocolFailures(t *testing.T) { t.Run("prompt response failure", func(t *testing.T) { recordPath := filepath.Join(t.TempDir(), "record.json") From 4353d9f85477fac42e359802c6273c164d50baac Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Mon, 31 Aug 2026 14:30:58 -0400 Subject: [PATCH 3/7] fix(llm): validate Pi RPC message update roots --- internal/llmadapters/pi_rpc.go | 24 +++++- internal/llmadapters/pi_rpc_test.go | 127 ++++++++++++++++++++++++++++ 2 files changed, 150 insertions(+), 1 deletion(-) diff --git a/internal/llmadapters/pi_rpc.go b/internal/llmadapters/pi_rpc.go index 12fe329..162fa89 100644 --- a/internal/llmadapters/pi_rpc.go +++ b/internal/llmadapters/pi_rpc.go @@ -2,6 +2,7 @@ package llmadapters import ( "bufio" + "bytes" "context" "encoding/json" "errors" @@ -754,7 +755,7 @@ func normalizePiRPCLogLine(line []byte) []byte { return append(logLine, '\n') } event["assistantMessageEvent"] = normalizedAssistantEvent - if len(compactPartial) > 0 { + if len(compactPartial) > 0 && piRPCMessageUpdateRootIsRedundant(event, assistantEvent, partialRaw, compactPartial) { delete(event, "message") } normalized, err := json.Marshal(event) @@ -764,6 +765,27 @@ func normalizePiRPCLogLine(line []byte) []byte { return append(normalized, '\n') } +func piRPCMessageUpdateRootIsRedundant(event, assistantEvent map[string]json.RawMessage, partialRaw, compactPartial json.RawMessage) bool { + if !isKnownPiRPCAssistantMessageEventType(rawString(assistantEvent, "type")) { + return false + } + var compact map[string]json.RawMessage + if err := json.Unmarshal(compactPartial, &compact); err != nil || rawString(compact, "role") != "assistant" { + return false + } + rootRaw, ok := event["message"] + return ok && bytes.Equal(rootRaw, partialRaw) +} + +func isKnownPiRPCAssistantMessageEventType(eventType string) bool { + switch eventType { + case "text_start", "text_delta", "text_end", "thinking_start", "thinking_delta", "thinking_end", "toolcall_start", "toolcall_delta", "toolcall_end": + return true + default: + return false + } +} + func compactPiRPCPartialForLog(partialRaw json.RawMessage) json.RawMessage { var partial map[string]json.RawMessage if err := json.Unmarshal(partialRaw, &partial); err != nil { diff --git a/internal/llmadapters/pi_rpc_test.go b/internal/llmadapters/pi_rpc_test.go index 86f0f46..3774d70 100644 --- a/internal/llmadapters/pi_rpc_test.go +++ b/internal/llmadapters/pi_rpc_test.go @@ -9,6 +9,7 @@ import ( "os" "os/exec" "path/filepath" + "reflect" "strconv" "strings" "testing" @@ -627,14 +628,36 @@ func assertPiRPCMessageUpdatesCompacted(t *testing.T, logged []byte) { if _, ok := event["message"]; ok { t.Fatalf("message_update = %#v, want redundant root message stripped", event) } + if event["sessionId"] != "session-1" { + t.Fatalf("message_update sessionId = %#v, want session-1", event["sessionId"]) + } assistantEvent, ok := event["assistantMessageEvent"].(map[string]any) if !ok { t.Fatalf("message_update = %#v, want assistantMessageEvent", event) } + if assistantEvent["type"] != "thinking_delta" { + t.Fatalf("assistantMessageEvent.type = %#v, want thinking_delta", assistantEvent["type"]) + } + if assistantEvent["contentIndex"] != float64(0) { + t.Fatalf("assistantMessageEvent.contentIndex = %#v, want 0", assistantEvent["contentIndex"]) + } + if assistantEvent["delta"] != "x" { + t.Fatalf("assistantMessageEvent.delta = %#v, want x", assistantEvent["delta"]) + } partial, ok := assistantEvent["partial"].(map[string]any) if !ok { t.Fatalf("message_update = %#v, want compact partial metadata", event) } + wantPartial := map[string]any{ + "role": "assistant", + "provider": "opencode-go", + "model": "deepseek-v4-pro", + "api": "openai-completions", + "stopReason": "stop", + } + if !reflect.DeepEqual(partial, wantPartial) { + t.Fatalf("partial = %#v, want exact compact metadata %#v", partial, wantPartial) + } if _, ok := partial["content"]; ok { t.Fatalf("partial = %#v, want content stripped", partial) } @@ -709,6 +732,109 @@ func TestNormalizePiRPCLogLinePreservesRootMessageForUnrecognizedPartial(t *test if message["role"] != "assistant" { t.Fatalf("root message = %#v, want assistant message preserved", message) } + assistantEvent, ok := event["assistantMessageEvent"].(map[string]any) + if !ok { + t.Fatalf("normalized message_update = %#v, want assistantMessageEvent", event) + } + if _, ok := assistantEvent["partial"]; ok { + t.Fatalf("assistantMessageEvent = %#v, want unrecognized partial removed", assistantEvent) + } +} + +func TestNormalizePiRPCLogLinePreservesRootMessageForInvalidAssistantEvent(t *testing.T) { + for _, tt := range []struct { + name string + eventType string + role string + }{ + {name: "unknown event type", eventType: "future_or_bad_event", role: "assistant"}, + {name: "invalid partial role", eventType: "thinking_delta", role: "user"}, + } { + t.Run(tt.name, func(t *testing.T) { + partial := map[string]any{ + "role": tt.role, + "provider": "opencode-go", + "model": "deepseek-v4-pro", + "api": "openai-completions", + "stopReason": "stop", + "content": []map[string]any{{"type": "thinking", "thinking": "cumulative"}}, + "usage": map[string]any{"tokensIn": 100, "tokensOut": 50, "totalTokens": 150}, + } + event := map[string]any{ + "type": "message_update", + "message": partial, + "assistantMessageEvent": map[string]any{ + "type": tt.eventType, + "contentIndex": 0, + "delta": "x", + "partial": partial, + }, + } + line, err := json.Marshal(event) + if err != nil { + t.Fatalf("Marshal(event): %v", err) + } + normalized := normalizePiRPCLogLine(line) + var decoded map[string]any + if err := json.Unmarshal(normalized, &decoded); err != nil { + t.Fatalf("Unmarshal(normalized): %v", err) + } + if _, ok := decoded["message"]; !ok { + t.Fatalf("normalized message_update = %#v, want root message preserved", decoded) + } + }) + } +} + +func TestNormalizePiRPCLogLinePreservesNonidenticalRootMessage(t *testing.T) { + partial := map[string]any{ + "role": "assistant", + "provider": "opencode-go", + "model": "deepseek-v4-pro", + "api": "openai-completions", + "stopReason": "stop", + "content": []map[string]any{{"type": "thinking", "thinking": "partial"}}, + "usage": map[string]any{"tokensIn": 100, "tokensOut": 50, "totalTokens": 150}, + } + event := map[string]any{ + "type": "message_update", + "message": map[string]any{ + "role": "assistant", + "provider": "opencode-go", + "model": "deepseek-v4-pro", + "api": "openai-completions", + "stopReason": "stop", + "content": []map[string]any{{"type": "thinking", "thinking": "root"}}, + "usage": map[string]any{"tokensIn": 100, "tokensOut": 50, "totalTokens": 150}, + }, + "assistantMessageEvent": map[string]any{ + "type": "thinking_delta", + "contentIndex": 0, + "delta": "x", + "partial": partial, + }, + } + line, err := json.Marshal(event) + if err != nil { + t.Fatalf("Marshal(event): %v", err) + } + normalized := normalizePiRPCLogLine(line) + var decoded map[string]any + if err := json.Unmarshal(normalized, &decoded); err != nil { + t.Fatalf("Unmarshal(normalized): %v", err) + } + message, ok := decoded["message"].(map[string]any) + if !ok { + t.Fatalf("normalized message_update = %#v, want nonidentical root message preserved", decoded) + } + content, ok := message["content"].([]any) + if !ok || len(content) == 0 { + t.Fatalf("root message = %#v, want root content preserved", message) + } + contentBlock, ok := content[0].(map[string]any) + if !ok || contentBlock["thinking"] != "root" { + t.Fatalf("root message = %#v, want root snapshot preserved", message) + } } func TestPiRPCProtocolFailures(t *testing.T) { @@ -1013,6 +1139,7 @@ func TestPiRPCHelperProcess(_ *testing.T) { "role": "assistant", "provider": "opencode-go", "model": "deepseek-v4-pro", + "api": "openai-completions", "stopReason": "stop", "content": []map[string]any{{"type": "thinking", "thinking": thinking}}, "usage": map[string]any{"tokensIn": 100, "tokensOut": 50, "totalTokens": 150}, From 28d5246055a248b82be7b957f6e089bb64c597b7 Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Mon, 31 Aug 2026 14:52:05 -0400 Subject: [PATCH 4/7] fix(llm): validate Pi RPC update payloads --- internal/llmadapters/pi_rpc.go | 183 +++++++++++++++++++++++++++- internal/llmadapters/pi_rpc_test.go | 163 ++++++++++++++++++++++++- 2 files changed, 335 insertions(+), 11 deletions(-) diff --git a/internal/llmadapters/pi_rpc.go b/internal/llmadapters/pi_rpc.go index 162fa89..eba3320 100644 --- a/internal/llmadapters/pi_rpc.go +++ b/internal/llmadapters/pi_rpc.go @@ -745,8 +745,13 @@ func normalizePiRPCLogLine(line []byte) []byte { return append(logLine, '\n') } compactPartial := compactPiRPCPartialForLog(partialRaw) + rootMessageRedundant := false if len(compactPartial) > 0 { + if !piRPCMessageUpdateRootIsRedundant(event, assistantEvent, partialRaw) { + return append(logLine, '\n') + } assistantEvent["partial"] = compactPartial + rootMessageRedundant = true } else { delete(assistantEvent, "partial") } @@ -755,7 +760,7 @@ func normalizePiRPCLogLine(line []byte) []byte { return append(logLine, '\n') } event["assistantMessageEvent"] = normalizedAssistantEvent - if len(compactPartial) > 0 && piRPCMessageUpdateRootIsRedundant(event, assistantEvent, partialRaw, compactPartial) { + if rootMessageRedundant { delete(event, "message") } normalized, err := json.Marshal(event) @@ -765,18 +770,186 @@ func normalizePiRPCLogLine(line []byte) []byte { return append(normalized, '\n') } -func piRPCMessageUpdateRootIsRedundant(event, assistantEvent map[string]json.RawMessage, partialRaw, compactPartial json.RawMessage) bool { - if !isKnownPiRPCAssistantMessageEventType(rawString(assistantEvent, "type")) { +func piRPCMessageUpdateRootIsRedundant(event, assistantEvent map[string]json.RawMessage, partialRaw json.RawMessage) bool { + eventType := rawString(assistantEvent, "type") + if !isKnownPiRPCAssistantMessageEventType(eventType) { return false } - var compact map[string]json.RawMessage - if err := json.Unmarshal(compactPartial, &compact); err != nil || rawString(compact, "role") != "assistant" { + if !validPiRPCAssistantMessageUpdate(assistantEvent, partialRaw, eventType) { return false } rootRaw, ok := event["message"] return ok && bytes.Equal(rootRaw, partialRaw) } +func validPiRPCAssistantMessageUpdate(assistantEvent map[string]json.RawMessage, partialRaw json.RawMessage, eventType string) bool { + partial, ok := piRPCJSONObject(partialRaw) + if !ok || rawString(partial, "role") != "assistant" || !validPiRPCAssistantMessage(partial) { + return false + } + if !piRPCJSONNumber(assistantEvent["contentIndex"]) { + return false + } + switch eventType { + case "text_delta", "thinking_delta", "toolcall_delta": + return piRPCJSONString(assistantEvent["delta"]) + case "text_end", "thinking_end": + return piRPCJSONString(assistantEvent["content"]) + case "toolcall_end": + return validPiRPCToolCall(assistantEvent["toolCall"]) + default: + return true + } +} + +func validPiRPCAssistantMessage(partial map[string]json.RawMessage) bool { + for _, key := range []string{"api", "provider", "model"} { + if !piRPCNonEmptyJSONString(partial[key]) { + return false + } + } + if rawString(partial, "stopReason") == "" || !isKnownPiRPCStopReason(rawString(partial, "stopReason")) { + return false + } + if !piRPCJSONNumber(partial["timestamp"]) || !validPiRPCContent(partial["content"]) { + return false + } + return validPiRPCUsage(partial["usage"]) +} + +func validPiRPCContent(value json.RawMessage) bool { + blocks, ok := piRPCJSONArray(value) + if !ok { + return false + } + for _, rawBlock := range blocks { + block, ok := piRPCJSONObject(rawBlock) + if !ok { + return false + } + switch rawString(block, "type") { + case "text": + if !piRPCJSONString(block["text"]) { + return false + } + case "thinking": + if !piRPCJSONString(block["thinking"]) { + return false + } + case "toolCall": + if !piRPCJSONString(block["id"]) || !piRPCJSONString(block["name"]) { + return false + } + if _, ok := piRPCJSONObject(block["arguments"]); !ok { + return false + } + default: + return false + } + } + return true +} + +func validPiRPCUsage(value json.RawMessage) bool { + usage, ok := piRPCJSONObject(value) + if !ok { + return false + } + for _, key := range []string{"input", "output", "cacheRead", "cacheWrite", "totalTokens", "timestamp"} { + if !piRPCJSONNumber(usage[key]) { + return false + } + } + cost, ok := piRPCJSONObject(usage["cost"]) + if !ok { + return false + } + for _, key := range []string{"input", "output", "cacheRead", "cacheWrite", "total"} { + if !piRPCJSONNumber(cost[key]) { + return false + } + } + for _, key := range []string{"cacheWrite1h", "reasoning"} { + if value, ok := usage[key]; ok && !piRPCJSONNumber(value) { + return false + } + } + if value, ok := usage["rawStopReason"]; ok && !piRPCJSONString(value) { + return false + } + return true +} + +func validPiRPCToolCall(value json.RawMessage) bool { + toolCall, ok := piRPCJSONObject(value) + if !ok || rawString(toolCall, "type") != "toolCall" { + return false + } + if !piRPCJSONString(toolCall["id"]) || !piRPCJSONString(toolCall["name"]) { + return false + } + _, ok = piRPCJSONObject(toolCall["arguments"]) + return ok +} + +func isKnownPiRPCStopReason(value string) bool { + switch value { + case "pending", "stop", "length", "toolUse", "error", "aborted": + return true + default: + return false + } +} + +func piRPCJSONObject(value json.RawMessage) (map[string]json.RawMessage, bool) { + if len(value) == 0 || bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return nil, false + } + var object map[string]json.RawMessage + if err := json.Unmarshal(value, &object); err != nil || object == nil { + return nil, false + } + return object, true +} + +func piRPCJSONArray(value json.RawMessage) ([]json.RawMessage, bool) { + if len(value) == 0 || bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return nil, false + } + var array []json.RawMessage + if err := json.Unmarshal(value, &array); err != nil || array == nil { + return nil, false + } + return array, true +} + +func piRPCJSONString(value json.RawMessage) bool { + if len(value) == 0 || bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return false + } + var text string + return json.Unmarshal(value, &text) == nil +} + +func piRPCNonEmptyJSONString(value json.RawMessage) bool { + if !piRPCJSONString(value) { + return false + } + var text string + if err := json.Unmarshal(value, &text); err != nil { + return false + } + return text != "" +} + +func piRPCJSONNumber(value json.RawMessage) bool { + if len(value) == 0 || bytes.Equal(bytes.TrimSpace(value), []byte("null")) { + return false + } + var number float64 + return json.Unmarshal(value, &number) == nil +} + func isKnownPiRPCAssistantMessageEventType(eventType string) bool { switch eventType { case "text_start", "text_delta", "text_end", "thinking_start", "thinking_delta", "thinking_end", "toolcall_start", "toolcall_delta", "toolcall_end": diff --git a/internal/llmadapters/pi_rpc_test.go b/internal/llmadapters/pi_rpc_test.go index 3774d70..819c331 100644 --- a/internal/llmadapters/pi_rpc_test.go +++ b/internal/llmadapters/pi_rpc_test.go @@ -2,6 +2,7 @@ package llmadapters import ( "bufio" + "bytes" "context" "encoding/json" "errors" @@ -694,7 +695,7 @@ func assertPiRPCFinalEventsPreserved(t *testing.T, logged []byte) { t.Fatalf("message_end.message = %#v, want final structured content", message) } usage, ok := message["usage"].(map[string]any) - if !ok || usage["tokensIn"] != float64(100) || usage["tokensOut"] != float64(50) { + if !ok || usage["input"] != float64(100) || usage["output"] != float64(50) { t.Fatalf("message_end.message = %#v, want final usage", message) } messageEndFound = true @@ -757,8 +758,23 @@ func TestNormalizePiRPCLogLinePreservesRootMessageForInvalidAssistantEvent(t *te "model": "deepseek-v4-pro", "api": "openai-completions", "stopReason": "stop", + "timestamp": 1700000000000, "content": []map[string]any{{"type": "thinking", "thinking": "cumulative"}}, - "usage": map[string]any{"tokensIn": 100, "tokensOut": 50, "totalTokens": 150}, + "usage": map[string]any{ + "input": 100, + "output": 50, + "cacheRead": 0, + "cacheWrite": 0, + "totalTokens": 150, + "cost": map[string]any{ + "input": 0, + "output": 0, + "cacheRead": 0, + "cacheWrite": 0, + "total": 0, + }, + "timestamp": 1700000000000, + }, } event := map[string]any{ "type": "message_update", @@ -793,8 +809,23 @@ func TestNormalizePiRPCLogLinePreservesNonidenticalRootMessage(t *testing.T) { "model": "deepseek-v4-pro", "api": "openai-completions", "stopReason": "stop", + "timestamp": 1700000000000, "content": []map[string]any{{"type": "thinking", "thinking": "partial"}}, - "usage": map[string]any{"tokensIn": 100, "tokensOut": 50, "totalTokens": 150}, + "usage": map[string]any{ + "input": 100, + "output": 50, + "cacheRead": 0, + "cacheWrite": 0, + "totalTokens": 150, + "cost": map[string]any{ + "input": 0, + "output": 0, + "cacheRead": 0, + "cacheWrite": 0, + "total": 0, + }, + "timestamp": 1700000000000, + }, } event := map[string]any{ "type": "message_update", @@ -804,8 +835,23 @@ func TestNormalizePiRPCLogLinePreservesNonidenticalRootMessage(t *testing.T) { "model": "deepseek-v4-pro", "api": "openai-completions", "stopReason": "stop", + "timestamp": 1700000000000, "content": []map[string]any{{"type": "thinking", "thinking": "root"}}, - "usage": map[string]any{"tokensIn": 100, "tokensOut": 50, "totalTokens": 150}, + "usage": map[string]any{ + "input": 100, + "output": 50, + "cacheRead": 0, + "cacheWrite": 0, + "totalTokens": 150, + "cost": map[string]any{ + "input": 0, + "output": 0, + "cacheRead": 0, + "cacheWrite": 0, + "total": 0, + }, + "timestamp": 1700000000000, + }, }, "assistantMessageEvent": map[string]any{ "type": "thinking_delta", @@ -835,6 +881,96 @@ func TestNormalizePiRPCLogLinePreservesNonidenticalRootMessage(t *testing.T) { if !ok || contentBlock["thinking"] != "root" { t.Fatalf("root message = %#v, want root snapshot preserved", message) } + want := append(append([]byte(nil), line...), '\n') + if !bytes.Equal(normalized, want) { + t.Fatalf("normalized = %s, want original line preserved", normalized) + } +} + +func TestNormalizePiRPCLogLinePreservesOriginalForInvalidPiUpdateShape(t *testing.T) { + for _, tt := range []struct { + name string + mutate func(partial, assistantEvent map[string]any) + }{ + { + name: "missing assistant timestamp", + mutate: func(partial, _ map[string]any) { + delete(partial, "timestamp") + }, + }, + { + name: "missing assistant usage", + mutate: func(partial, _ map[string]any) { + delete(partial, "usage") + }, + }, + { + name: "missing event content index", + mutate: func(_, assistantEvent map[string]any) { + delete(assistantEvent, "contentIndex") + }, + }, + { + name: "wrong event delta type", + mutate: func(_, assistantEvent map[string]any) { + assistantEvent["delta"] = 1 + }, + }, + { + name: "unknown content block", + mutate: func(partial, _ map[string]any) { + partial["content"] = []map[string]any{{"type": "unknown", "value": "unrecognized"}} + }, + }, + } { + t.Run(tt.name, func(t *testing.T) { + partial := map[string]any{ + "role": "assistant", + "provider": "opencode-go", + "model": "deepseek-v4-pro", + "api": "openai-completions", + "stopReason": "stop", + "timestamp": float64(1700000000000), + "content": []map[string]any{{"type": "thinking", "thinking": "cumulative"}}, + "usage": map[string]any{ + "input": 100, + "output": 50, + "cacheRead": 0, + "cacheWrite": 0, + "totalTokens": 150, + "cost": map[string]any{ + "input": 0, + "output": 0, + "cacheRead": 0, + "cacheWrite": 0, + "total": 0, + }, + "timestamp": float64(1700000000000), + }, + } + assistantEvent := map[string]any{ + "type": "thinking_delta", + "contentIndex": 0, + "delta": "x", + "partial": partial, + } + event := map[string]any{ + "type": "message_update", + "message": partial, + "assistantMessageEvent": assistantEvent, + } + tt.mutate(partial, assistantEvent) + line, err := json.Marshal(event) + if err != nil { + t.Fatalf("Marshal(event): %v", err) + } + normalized := normalizePiRPCLogLine(line) + want := append(append([]byte(nil), line...), '\n') + if !bytes.Equal(normalized, want) { + t.Fatalf("normalized = %s, want original line preserved", normalized) + } + }) + } } func TestPiRPCProtocolFailures(t *testing.T) { @@ -1141,8 +1277,23 @@ func TestPiRPCHelperProcess(_ *testing.T) { "model": "deepseek-v4-pro", "api": "openai-completions", "stopReason": "stop", + "timestamp": 1700000000000, "content": []map[string]any{{"type": "thinking", "thinking": thinking}}, - "usage": map[string]any{"tokensIn": 100, "tokensOut": 50, "totalTokens": 150}, + "usage": map[string]any{ + "input": 100, + "output": 50, + "cacheRead": 0, + "cacheWrite": 0, + "totalTokens": 150, + "cost": map[string]any{ + "input": 0, + "output": 0, + "cacheRead": 0, + "cacheWrite": 0, + "total": 0, + }, + "timestamp": 1700000000000, + }, } event := map[string]any{ "type": "message_update", @@ -1158,7 +1309,7 @@ func TestPiRPCHelperProcess(_ *testing.T) { data, _ := json.Marshal(event) fmt.Println(string(data)) } - fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"usage":{"tokensIn":100,"tokensOut":50}}}`) + fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"api":"openai-completions","provider":"opencode-go","model":"deepseek-v4-pro","usage":{"input":100,"output":50,"cacheRead":0,"cacheWrite":0,"totalTokens":150,"cost":{"input":0,"output":0,"cacheRead":0,"cacheWrite":0,"total":0},"timestamp":1700000000000},"stopReason":"stop","timestamp":1700000000000}}`) fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}]}`) case "prompt-failure": fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":false,"error":"No API key found for opencode-go"}`) From d274cf22635ee5891a681b4c4316586af2844063 Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Mon, 31 Aug 2026 14:56:51 -0400 Subject: [PATCH 5/7] fix(llm): align Pi RPC usage validation --- internal/llmadapters/pi_rpc.go | 5 +---- internal/llmadapters/pi_rpc_test.go | 7 +------ 2 files changed, 2 insertions(+), 10 deletions(-) diff --git a/internal/llmadapters/pi_rpc.go b/internal/llmadapters/pi_rpc.go index eba3320..03bf266 100644 --- a/internal/llmadapters/pi_rpc.go +++ b/internal/llmadapters/pi_rpc.go @@ -855,7 +855,7 @@ func validPiRPCUsage(value json.RawMessage) bool { if !ok { return false } - for _, key := range []string{"input", "output", "cacheRead", "cacheWrite", "totalTokens", "timestamp"} { + for _, key := range []string{"input", "output", "cacheRead", "cacheWrite", "totalTokens"} { if !piRPCJSONNumber(usage[key]) { return false } @@ -874,9 +874,6 @@ func validPiRPCUsage(value json.RawMessage) bool { return false } } - if value, ok := usage["rawStopReason"]; ok && !piRPCJSONString(value) { - return false - } return true } diff --git a/internal/llmadapters/pi_rpc_test.go b/internal/llmadapters/pi_rpc_test.go index 819c331..33c950c 100644 --- a/internal/llmadapters/pi_rpc_test.go +++ b/internal/llmadapters/pi_rpc_test.go @@ -773,7 +773,6 @@ func TestNormalizePiRPCLogLinePreservesRootMessageForInvalidAssistantEvent(t *te "cacheWrite": 0, "total": 0, }, - "timestamp": 1700000000000, }, } event := map[string]any{ @@ -824,7 +823,6 @@ func TestNormalizePiRPCLogLinePreservesNonidenticalRootMessage(t *testing.T) { "cacheWrite": 0, "total": 0, }, - "timestamp": 1700000000000, }, } event := map[string]any{ @@ -850,7 +848,6 @@ func TestNormalizePiRPCLogLinePreservesNonidenticalRootMessage(t *testing.T) { "cacheWrite": 0, "total": 0, }, - "timestamp": 1700000000000, }, }, "assistantMessageEvent": map[string]any{ @@ -945,7 +942,6 @@ func TestNormalizePiRPCLogLinePreservesOriginalForInvalidPiUpdateShape(t *testin "cacheWrite": 0, "total": 0, }, - "timestamp": float64(1700000000000), }, } assistantEvent := map[string]any{ @@ -1292,7 +1288,6 @@ func TestPiRPCHelperProcess(_ *testing.T) { "cacheWrite": 0, "total": 0, }, - "timestamp": 1700000000000, }, } event := map[string]any{ @@ -1309,7 +1304,7 @@ func TestPiRPCHelperProcess(_ *testing.T) { data, _ := json.Marshal(event) fmt.Println(string(data)) } - fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"api":"openai-completions","provider":"opencode-go","model":"deepseek-v4-pro","usage":{"input":100,"output":50,"cacheRead":0,"cacheWrite":0,"totalTokens":150,"cost":{"input":0,"output":0,"cacheRead":0,"cacheWrite":0,"total":0},"timestamp":1700000000000},"stopReason":"stop","timestamp":1700000000000}}`) + fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"api":"openai-completions","provider":"opencode-go","model":"deepseek-v4-pro","usage":{"input":100,"output":50,"cacheRead":0,"cacheWrite":0,"totalTokens":150,"cost":{"input":0,"output":0,"cacheRead":0,"cacheWrite":0,"total":0}},"stopReason":"stop","timestamp":1700000000000}}`) fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}]}`) case "prompt-failure": fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":false,"error":"No API key found for opencode-go"}`) From aeb8f637e7e305a00dd1a57447a6c4f9bbd3095f Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Mon, 31 Aug 2026 15:33:11 -0400 Subject: [PATCH 6/7] fix(llm): narrow Pi RPC log projection --- internal/llmadapters/pi_rpc.go | 145 ++------- internal/llmadapters/pi_rpc_test.go | 442 +++++++++++++++------------- 2 files changed, 258 insertions(+), 329 deletions(-) diff --git a/internal/llmadapters/pi_rpc.go b/internal/llmadapters/pi_rpc.go index 03bf266..75c36c2 100644 --- a/internal/llmadapters/pi_rpc.go +++ b/internal/llmadapters/pi_rpc.go @@ -744,25 +744,20 @@ func normalizePiRPCLogLine(line []byte) []byte { if !ok { return append(logLine, '\n') } + if !piRPCMessageUpdateRootIsRedundant(event, assistantEvent, partialRaw) { + return append(logLine, '\n') + } compactPartial := compactPiRPCPartialForLog(partialRaw) - rootMessageRedundant := false - if len(compactPartial) > 0 { - if !piRPCMessageUpdateRootIsRedundant(event, assistantEvent, partialRaw) { - return append(logLine, '\n') - } - assistantEvent["partial"] = compactPartial - rootMessageRedundant = true - } else { - delete(assistantEvent, "partial") + if len(compactPartial) == 0 { + return append(logLine, '\n') } + assistantEvent["partial"] = compactPartial normalizedAssistantEvent, err := json.Marshal(assistantEvent) if err != nil { return append(logLine, '\n') } event["assistantMessageEvent"] = normalizedAssistantEvent - if rootMessageRedundant { - delete(event, "message") - } + delete(event, "message") normalized, err := json.Marshal(event) if err != nil { return append(logLine, '\n') @@ -771,25 +766,27 @@ func normalizePiRPCLogLine(line []byte) []byte { } func piRPCMessageUpdateRootIsRedundant(event, assistantEvent map[string]json.RawMessage, partialRaw json.RawMessage) bool { - eventType := rawString(assistantEvent, "type") - if !isKnownPiRPCAssistantMessageEventType(eventType) { - return false - } - if !validPiRPCAssistantMessageUpdate(assistantEvent, partialRaw, eventType) { + if !validPiRPCProjection(assistantEvent, partialRaw) { return false } rootRaw, ok := event["message"] return ok && bytes.Equal(rootRaw, partialRaw) } -func validPiRPCAssistantMessageUpdate(assistantEvent map[string]json.RawMessage, partialRaw json.RawMessage, eventType string) bool { - partial, ok := piRPCJSONObject(partialRaw) - if !ok || rawString(partial, "role") != "assistant" || !validPiRPCAssistantMessage(partial) { +func validPiRPCProjection(assistantEvent map[string]json.RawMessage, partialRaw json.RawMessage) bool { + eventType := rawString(assistantEvent, "type") + if !isKnownPiRPCAssistantMessageEventType(eventType) || !piRPCJSONNumber(assistantEvent["contentIndex"]) { return false } - if !piRPCJSONNumber(assistantEvent["contentIndex"]) { + partial, ok := piRPCJSONObject(partialRaw) + if !ok || rawString(partial, "role") != "assistant" { return false } + for _, key := range []string{"role", "provider", "model", "api", "stopReason"} { + if !piRPCJSONString(partial[key]) { + return false + } + } switch eventType { case "text_delta", "thinking_delta", "toolcall_delta": return piRPCJSONString(assistantEvent["delta"]) @@ -802,81 +799,6 @@ func validPiRPCAssistantMessageUpdate(assistantEvent map[string]json.RawMessage, } } -func validPiRPCAssistantMessage(partial map[string]json.RawMessage) bool { - for _, key := range []string{"api", "provider", "model"} { - if !piRPCNonEmptyJSONString(partial[key]) { - return false - } - } - if rawString(partial, "stopReason") == "" || !isKnownPiRPCStopReason(rawString(partial, "stopReason")) { - return false - } - if !piRPCJSONNumber(partial["timestamp"]) || !validPiRPCContent(partial["content"]) { - return false - } - return validPiRPCUsage(partial["usage"]) -} - -func validPiRPCContent(value json.RawMessage) bool { - blocks, ok := piRPCJSONArray(value) - if !ok { - return false - } - for _, rawBlock := range blocks { - block, ok := piRPCJSONObject(rawBlock) - if !ok { - return false - } - switch rawString(block, "type") { - case "text": - if !piRPCJSONString(block["text"]) { - return false - } - case "thinking": - if !piRPCJSONString(block["thinking"]) { - return false - } - case "toolCall": - if !piRPCJSONString(block["id"]) || !piRPCJSONString(block["name"]) { - return false - } - if _, ok := piRPCJSONObject(block["arguments"]); !ok { - return false - } - default: - return false - } - } - return true -} - -func validPiRPCUsage(value json.RawMessage) bool { - usage, ok := piRPCJSONObject(value) - if !ok { - return false - } - for _, key := range []string{"input", "output", "cacheRead", "cacheWrite", "totalTokens"} { - if !piRPCJSONNumber(usage[key]) { - return false - } - } - cost, ok := piRPCJSONObject(usage["cost"]) - if !ok { - return false - } - for _, key := range []string{"input", "output", "cacheRead", "cacheWrite", "total"} { - if !piRPCJSONNumber(cost[key]) { - return false - } - } - for _, key := range []string{"cacheWrite1h", "reasoning"} { - if value, ok := usage[key]; ok && !piRPCJSONNumber(value) { - return false - } - } - return true -} - func validPiRPCToolCall(value json.RawMessage) bool { toolCall, ok := piRPCJSONObject(value) if !ok || rawString(toolCall, "type") != "toolCall" { @@ -889,15 +811,6 @@ func validPiRPCToolCall(value json.RawMessage) bool { return ok } -func isKnownPiRPCStopReason(value string) bool { - switch value { - case "pending", "stop", "length", "toolUse", "error", "aborted": - return true - default: - return false - } -} - func piRPCJSONObject(value json.RawMessage) (map[string]json.RawMessage, bool) { if len(value) == 0 || bytes.Equal(bytes.TrimSpace(value), []byte("null")) { return nil, false @@ -909,17 +822,6 @@ func piRPCJSONObject(value json.RawMessage) (map[string]json.RawMessage, bool) { return object, true } -func piRPCJSONArray(value json.RawMessage) ([]json.RawMessage, bool) { - if len(value) == 0 || bytes.Equal(bytes.TrimSpace(value), []byte("null")) { - return nil, false - } - var array []json.RawMessage - if err := json.Unmarshal(value, &array); err != nil || array == nil { - return nil, false - } - return array, true -} - func piRPCJSONString(value json.RawMessage) bool { if len(value) == 0 || bytes.Equal(bytes.TrimSpace(value), []byte("null")) { return false @@ -928,17 +830,6 @@ func piRPCJSONString(value json.RawMessage) bool { return json.Unmarshal(value, &text) == nil } -func piRPCNonEmptyJSONString(value json.RawMessage) bool { - if !piRPCJSONString(value) { - return false - } - var text string - if err := json.Unmarshal(value, &text); err != nil { - return false - } - return text != "" -} - func piRPCJSONNumber(value json.RawMessage) bool { if len(value) == 0 || bytes.Equal(bytes.TrimSpace(value), []byte("null")) { return false diff --git a/internal/llmadapters/pi_rpc_test.go b/internal/llmadapters/pi_rpc_test.go index 33c950c..ae739b3 100644 --- a/internal/llmadapters/pi_rpc_test.go +++ b/internal/llmadapters/pi_rpc_test.go @@ -718,244 +718,158 @@ func assertPiRPCFinalEventsPreserved(t *testing.T, logged []byte) { } } -func TestNormalizePiRPCLogLinePreservesRootMessageForUnrecognizedPartial(t *testing.T) { - line := []byte(`{"type":"message_update","message":{"role":"assistant","content":[{"type":"thinking","thinking":"cumulative"}]},"assistantMessageEvent":{"type":"thinking_delta","contentIndex":0,"delta":"x","partial":{"content":[{"type":"unknown","value":"unrecognized"}],"usage":{"tokensIn":100,"tokensOut":50}}}}`) - - normalized := normalizePiRPCLogLine(line) - var event map[string]any - if err := json.Unmarshal(normalized, &event); err != nil { - t.Fatalf("Unmarshal(normalized): %v", err) - } - message, ok := event["message"].(map[string]any) - if !ok { - t.Fatalf("normalized message_update = %#v, want root message preserved", event) - } - if message["role"] != "assistant" { - t.Fatalf("root message = %#v, want assistant message preserved", message) - } - assistantEvent, ok := event["assistantMessageEvent"].(map[string]any) - if !ok { - t.Fatalf("normalized message_update = %#v, want assistantMessageEvent", event) - } - if _, ok := assistantEvent["partial"]; ok { - t.Fatalf("assistantMessageEvent = %#v, want unrecognized partial removed", assistantEvent) - } -} - -func TestNormalizePiRPCLogLinePreservesRootMessageForInvalidAssistantEvent(t *testing.T) { - for _, tt := range []struct { +func TestNormalizePiRPCLogLineCompactsAllKnownPiAssistantEvents(t *testing.T) { + tests := []struct { name string eventType string - role string + fields map[string]any }{ - {name: "unknown event type", eventType: "future_or_bad_event", role: "assistant"}, - {name: "invalid partial role", eventType: "thinking_delta", role: "user"}, - } { + {name: "text start", eventType: "text_start"}, + {name: "text delta", eventType: "text_delta", fields: map[string]any{"delta": "text chunk"}}, + {name: "text end", eventType: "text_end", fields: map[string]any{"content": "complete text"}}, + {name: "thinking start", eventType: "thinking_start"}, + {name: "thinking delta", eventType: "thinking_delta", fields: map[string]any{"delta": "thinking chunk"}}, + {name: "thinking end", eventType: "thinking_end", fields: map[string]any{"content": "complete thinking"}}, + {name: "tool call start", eventType: "toolcall_start"}, + {name: "tool call delta", eventType: "toolcall_delta", fields: map[string]any{"delta": "{\"path\":\"x\"}"}}, + { + name: "tool call end", + eventType: "toolcall_end", + fields: map[string]any{"toolCall": map[string]any{ + "type": "toolCall", "id": "call-1", "name": "Read", "arguments": map[string]any{"path": "x"}, + }}, + }, + } + for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - partial := map[string]any{ - "role": tt.role, - "provider": "opencode-go", - "model": "deepseek-v4-pro", - "api": "openai-completions", - "stopReason": "stop", - "timestamp": 1700000000000, - "content": []map[string]any{{"type": "thinking", "thinking": "cumulative"}}, - "usage": map[string]any{ - "input": 100, - "output": 50, - "cacheRead": 0, - "cacheWrite": 0, - "totalTokens": 150, - "cost": map[string]any{ - "input": 0, - "output": 0, - "cacheRead": 0, - "cacheWrite": 0, - "total": 0, - }, - }, - } - event := map[string]any{ - "type": "message_update", - "message": partial, - "assistantMessageEvent": map[string]any{ - "type": tt.eventType, - "contentIndex": 0, - "delta": "x", - "partial": partial, - }, + event, _, assistantEvent := piRPCProjectionEventFixture(tt.eventType) + for key, value := range tt.fields { + assistantEvent[key] = value } line, err := json.Marshal(event) if err != nil { t.Fatalf("Marshal(event): %v", err) } normalized := normalizePiRPCLogLine(line) - var decoded map[string]any - if err := json.Unmarshal(normalized, &decoded); err != nil { - t.Fatalf("Unmarshal(normalized): %v", err) - } - if _, ok := decoded["message"]; !ok { - t.Fatalf("normalized message_update = %#v, want root message preserved", decoded) + gotAssistantEvent := assertPiRPCProjectionCompacted(t, normalized, tt.eventType, piRPCProjectionPartialFixture()) + for key, want := range tt.fields { + if !reflect.DeepEqual(gotAssistantEvent[key], want) { + t.Fatalf("assistantMessageEvent[%q] = %#v, want %#v", key, gotAssistantEvent[key], want) + } } }) } } -func TestNormalizePiRPCLogLinePreservesNonidenticalRootMessage(t *testing.T) { - partial := map[string]any{ - "role": "assistant", - "provider": "opencode-go", - "model": "deepseek-v4-pro", - "api": "openai-completions", - "stopReason": "stop", - "timestamp": 1700000000000, - "content": []map[string]any{{"type": "thinking", "thinking": "partial"}}, - "usage": map[string]any{ - "input": 100, - "output": 50, - "cacheRead": 0, - "cacheWrite": 0, - "totalTokens": 150, - "cost": map[string]any{ - "input": 0, - "output": 0, - "cacheRead": 0, - "cacheWrite": 0, - "total": 0, +func TestNormalizePiRPCLogLinePreservesOriginalForProjectionFailures(t *testing.T) { + tests := []struct { + name string + mutate func(event, partial, assistantEvent map[string]any) + }{ + {name: "missing role", mutate: piRPCDeleteProjectionField("role")}, + {name: "missing provider", mutate: piRPCDeleteProjectionField("provider")}, + {name: "missing model", mutate: piRPCDeleteProjectionField("model")}, + {name: "missing api", mutate: piRPCDeleteProjectionField("api")}, + {name: "missing stop reason", mutate: piRPCDeleteProjectionField("stopReason")}, + {name: "non-string role", mutate: piRPCSetProjectionField("role", 1)}, + {name: "non-string provider", mutate: piRPCSetProjectionField("provider", 1)}, + {name: "non-string model", mutate: piRPCSetProjectionField("model", 1)}, + {name: "non-string api", mutate: piRPCSetProjectionField("api", 1)}, + {name: "non-string stop reason", mutate: piRPCSetProjectionField("stopReason", 1)}, + { + name: "invalid role", + mutate: func(_, partial, _ map[string]any) { partial["role"] = "user" }, + }, + { + name: "unknown event type", + mutate: func(_, _, assistantEvent map[string]any) { assistantEvent["type"] = "future_event" }, + }, + { + name: "missing content index", + mutate: func(_, _, assistantEvent map[string]any) { delete(assistantEvent, "contentIndex") }, + }, + { + name: "non-numeric content index", + mutate: func(_, _, assistantEvent map[string]any) { assistantEvent["contentIndex"] = "0" }, + }, + { + name: "missing delta", + mutate: func(_, _, assistantEvent map[string]any) { + assistantEvent["type"] = "text_delta" + delete(assistantEvent, "delta") }, }, - } - event := map[string]any{ - "type": "message_update", - "message": map[string]any{ - "role": "assistant", - "provider": "opencode-go", - "model": "deepseek-v4-pro", - "api": "openai-completions", - "stopReason": "stop", - "timestamp": 1700000000000, - "content": []map[string]any{{"type": "thinking", "thinking": "root"}}, - "usage": map[string]any{ - "input": 100, - "output": 50, - "cacheRead": 0, - "cacheWrite": 0, - "totalTokens": 150, - "cost": map[string]any{ - "input": 0, - "output": 0, - "cacheRead": 0, - "cacheWrite": 0, - "total": 0, - }, + { + name: "non-string delta", + mutate: func(_, _, assistantEvent map[string]any) { + assistantEvent["type"] = "text_delta" + assistantEvent["delta"] = 1 }, }, - "assistantMessageEvent": map[string]any{ - "type": "thinking_delta", - "contentIndex": 0, - "delta": "x", - "partial": partial, + { + name: "missing end content", + mutate: func(_, _, assistantEvent map[string]any) { + assistantEvent["type"] = "text_end" + delete(assistantEvent, "content") + }, }, - } - line, err := json.Marshal(event) - if err != nil { - t.Fatalf("Marshal(event): %v", err) - } - normalized := normalizePiRPCLogLine(line) - var decoded map[string]any - if err := json.Unmarshal(normalized, &decoded); err != nil { - t.Fatalf("Unmarshal(normalized): %v", err) - } - message, ok := decoded["message"].(map[string]any) - if !ok { - t.Fatalf("normalized message_update = %#v, want nonidentical root message preserved", decoded) - } - content, ok := message["content"].([]any) - if !ok || len(content) == 0 { - t.Fatalf("root message = %#v, want root content preserved", message) - } - contentBlock, ok := content[0].(map[string]any) - if !ok || contentBlock["thinking"] != "root" { - t.Fatalf("root message = %#v, want root snapshot preserved", message) - } - want := append(append([]byte(nil), line...), '\n') - if !bytes.Equal(normalized, want) { - t.Fatalf("normalized = %s, want original line preserved", normalized) - } -} - -func TestNormalizePiRPCLogLinePreservesOriginalForInvalidPiUpdateShape(t *testing.T) { - for _, tt := range []struct { - name string - mutate func(partial, assistantEvent map[string]any) - }{ { - name: "missing assistant timestamp", - mutate: func(partial, _ map[string]any) { - delete(partial, "timestamp") + name: "non-string end content", + mutate: func(_, _, assistantEvent map[string]any) { + assistantEvent["type"] = "thinking_end" + assistantEvent["content"] = 1 }, }, { - name: "missing assistant usage", - mutate: func(partial, _ map[string]any) { - delete(partial, "usage") + name: "missing tool call", + mutate: func(_, _, assistantEvent map[string]any) { + assistantEvent["type"] = "toolcall_end" + delete(assistantEvent, "toolCall") }, }, { - name: "missing event content index", - mutate: func(_, assistantEvent map[string]any) { - delete(assistantEvent, "contentIndex") + name: "invalid tool call", + mutate: func(_, _, assistantEvent map[string]any) { + assistantEvent["type"] = "toolcall_end" + assistantEvent["toolCall"] = map[string]any{"type": "toolCall", "id": "call-1"} }, }, { - name: "wrong event delta type", - mutate: func(_, assistantEvent map[string]any) { - assistantEvent["delta"] = 1 + name: "missing partial", + mutate: func(_, _, assistantEvent map[string]any) { delete(assistantEvent, "partial") }, + }, + { + name: "non-object partial", + mutate: func(_, _, assistantEvent map[string]any) { assistantEvent["partial"] = []any{} }, + }, + { + name: "partial without projection keys", + mutate: func(_, partial, _ map[string]any) { + for _, key := range []string{"role", "provider", "model", "api", "stopReason"} { + delete(partial, key) + } }, }, { - name: "unknown content block", - mutate: func(partial, _ map[string]any) { - partial["content"] = []map[string]any{{"type": "unknown", "value": "unrecognized"}} + name: "missing root message", + mutate: func(event, _, _ map[string]any) { delete(event, "message") }, + }, + { + name: "nonidentical root message", + mutate: func(event, partial, _ map[string]any) { + root := make(map[string]any, len(partial)) + for key, value := range partial { + root[key] = value + } + root["stopReason"] = "different" + event["message"] = root }, }, - } { + } + for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - partial := map[string]any{ - "role": "assistant", - "provider": "opencode-go", - "model": "deepseek-v4-pro", - "api": "openai-completions", - "stopReason": "stop", - "timestamp": float64(1700000000000), - "content": []map[string]any{{"type": "thinking", "thinking": "cumulative"}}, - "usage": map[string]any{ - "input": 100, - "output": 50, - "cacheRead": 0, - "cacheWrite": 0, - "totalTokens": 150, - "cost": map[string]any{ - "input": 0, - "output": 0, - "cacheRead": 0, - "cacheWrite": 0, - "total": 0, - }, - }, - } - assistantEvent := map[string]any{ - "type": "thinking_delta", - "contentIndex": 0, - "delta": "x", - "partial": partial, - } - event := map[string]any{ - "type": "message_update", - "message": partial, - "assistantMessageEvent": assistantEvent, - } - tt.mutate(partial, assistantEvent) + event, partial, assistantEvent := piRPCProjectionEventFixture("thinking_delta") + tt.mutate(event, partial, assistantEvent) line, err := json.Marshal(event) if err != nil { t.Fatalf("Marshal(event): %v", err) @@ -969,6 +883,130 @@ func TestNormalizePiRPCLogLinePreservesOriginalForInvalidPiUpdateShape(t *testin } } +func TestNormalizePiRPCLogLineCompactsProjectionWithMalformedCumulativeFields(t *testing.T) { + event, partial, _ := piRPCProjectionEventFixture("thinking_delta") + partial["content"] = map[string]any{"unexpected": true} + partial["usage"] = []any{"unexpected"} + partial["timestamp"] = "not-a-number" + line, err := json.Marshal(event) + if err != nil { + t.Fatalf("Marshal(event): %v", err) + } + assertPiRPCProjectionCompacted(t, normalizePiRPCLogLine(line), "thinking_delta", piRPCProjectionPartialFixture()) +} + +func TestNormalizePiRPCLogLineCompactsStringProjectionFieldsWithoutPolicyChecks(t *testing.T) { + event, partial, _ := piRPCProjectionEventFixture("thinking_delta") + partial["provider"] = "" + partial["model"] = "" + partial["api"] = "" + partial["stopReason"] = "future-stop-reason" + line, err := json.Marshal(event) + if err != nil { + t.Fatalf("Marshal(event): %v", err) + } + assertPiRPCProjectionCompacted(t, normalizePiRPCLogLine(line), "thinking_delta", map[string]any{ + "role": "assistant", "provider": "", "model": "", "api": "", "stopReason": "future-stop-reason", + }) +} + +func assertPiRPCProjectionCompacted(t *testing.T, normalized []byte, eventType string, wantPartial map[string]any) map[string]any { + t.Helper() + var event map[string]any + if err := json.Unmarshal(normalized, &event); err != nil { + t.Fatalf("Unmarshal(normalized): %v", err) + } + if _, ok := event["message"]; ok { + t.Fatalf("normalized message_update = %#v, want redundant root message stripped", event) + } + if event["sessionId"] != "session-1" || event["preservedSibling"] != "keep" { + t.Fatalf("normalized event siblings = %#v, want session and sibling preserved", event) + } + assistantEvent, ok := event["assistantMessageEvent"].(map[string]any) + if !ok { + t.Fatalf("normalized message_update = %#v, want assistantMessageEvent", event) + } + if assistantEvent["type"] != eventType || assistantEvent["contentIndex"] != float64(0) || assistantEvent["preservedSibling"] != "keep" { + t.Fatalf("assistantMessageEvent = %#v, want retained event fields", assistantEvent) + } + partial, ok := assistantEvent["partial"].(map[string]any) + if !ok { + t.Fatalf("assistantMessageEvent = %#v, want compact partial", assistantEvent) + } + if !reflect.DeepEqual(partial, wantPartial) { + t.Fatalf("partial = %#v, want exact compact projection %#v", partial, wantPartial) + } + return assistantEvent +} + +func piRPCProjectionPartialFixture() map[string]any { + return map[string]any{ + "role": "assistant", + "provider": "opencode-go", + "model": "deepseek-v4-pro", + "api": "openai-completions", + "stopReason": "stop", + } +} + +func piRPCProjectionEventFixture(eventType string) (map[string]any, map[string]any, map[string]any) { + partial := map[string]any{ + "role": "assistant", + "provider": "opencode-go", + "model": "deepseek-v4-pro", + "api": "openai-completions", + "stopReason": "stop", + "timestamp": 1700000000000, + "content": []map[string]any{{"type": "thinking", "thinking": "cumulative"}}, + "usage": map[string]any{ + "input": 100, + "output": 50, + "cacheRead": 0, + "cacheWrite": 0, + "totalTokens": 150, + "cost": map[string]any{ + "input": 0, + "output": 0, + "cacheRead": 0, + "cacheWrite": 0, + "total": 0, + }, + }, + } + assistantEvent := map[string]any{ + "type": eventType, + "contentIndex": 0, + "partial": partial, + "preservedSibling": "keep", + } + switch eventType { + case "text_delta", "thinking_delta", "toolcall_delta": + assistantEvent["delta"] = "x" + case "text_end", "thinking_end": + assistantEvent["content"] = "complete" + case "toolcall_end": + assistantEvent["toolCall"] = map[string]any{ + "type": "toolCall", "id": "call-1", "name": "Read", "arguments": map[string]any{"path": "x"}, + } + } + event := map[string]any{ + "type": "message_update", + "message": partial, + "sessionId": "session-1", + "preservedSibling": "keep", + "assistantMessageEvent": assistantEvent, + } + return event, partial, assistantEvent +} + +func piRPCDeleteProjectionField(key string) func(event, partial, assistantEvent map[string]any) { + return func(_, partial, _ map[string]any) { delete(partial, key) } +} + +func piRPCSetProjectionField(key string, value any) func(event, partial, assistantEvent map[string]any) { + return func(_, partial, _ map[string]any) { partial[key] = value } +} + func TestPiRPCProtocolFailures(t *testing.T) { t.Run("prompt response failure", func(t *testing.T) { recordPath := filepath.Join(t.TempDir(), "record.json") From a77505a8c93e4a7e01bb5fa3c25a647715ab57b6 Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Mon, 31 Aug 2026 16:36:00 -0400 Subject: [PATCH 7/7] test(llm): preserve differently encoded Pi roots --- internal/llmadapters/pi_rpc_test.go | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/internal/llmadapters/pi_rpc_test.go b/internal/llmadapters/pi_rpc_test.go index ae739b3..d8c6551 100644 --- a/internal/llmadapters/pi_rpc_test.go +++ b/internal/llmadapters/pi_rpc_test.go @@ -883,6 +883,35 @@ func TestNormalizePiRPCLogLinePreservesOriginalForProjectionFailures(t *testing. } } +func TestNormalizePiRPCLogLinePreservesOriginalForDifferentlyEncodedEqualRoot(t *testing.T) { + line := []byte(`{"type":"message_update","message":{"role": "assistant", "provider": "opencode-go", "model": "deepseek-v4-pro", "api": "openai-completions", "stopReason": "stop", "content": [], "usage": {"input": 100}, "timestamp": 1700000000000},"assistantMessageEvent":{"type":"thinking_delta","contentIndex":0,"delta":"x","partial":{"timestamp":1700000000000,"usage":{"input":100},"content":[],"stopReason":"stop","api":"openai-completions","model":"deepseek-v4-pro","provider":"opencode-go","role":"assistant"}}}`) + + var event map[string]json.RawMessage + if err := json.Unmarshal(line, &event); err != nil { + t.Fatalf("Unmarshal(line): %v", err) + } + var root, partial any + if err := json.Unmarshal(event["message"], &root); err != nil { + t.Fatalf("Unmarshal(root): %v", err) + } + var assistantEvent map[string]json.RawMessage + if err := json.Unmarshal(event["assistantMessageEvent"], &assistantEvent); err != nil { + t.Fatalf("Unmarshal(assistantMessageEvent): %v", err) + } + if err := json.Unmarshal(assistantEvent["partial"], &partial); err != nil { + t.Fatalf("Unmarshal(partial): %v", err) + } + if !reflect.DeepEqual(root, partial) { + t.Fatalf("root and partial differ semantically: root=%#v partial=%#v", root, partial) + } + + normalized := normalizePiRPCLogLine(line) + want := append(append([]byte(nil), line...), '\n') + if !bytes.Equal(normalized, want) { + t.Fatalf("normalized = %s, want exact original line preserved", normalized) + } +} + func TestNormalizePiRPCLogLineCompactsProjectionWithMalformedCumulativeFields(t *testing.T) { event, partial, _ := piRPCProjectionEventFixture("thinking_delta") partial["content"] = map[string]any{"unexpected": true}