From e5b7a8d8c765dfaa71994bdef2b1d4c156b4ca34 Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Fri, 2 Oct 2026 03:11:17 -0400 Subject: [PATCH 1/5] fix(pi): honor settled runs and aggregate turn usage [#638] Pi 1.0 RPC can continue after agent_end with automatic retries, compaction recovery, or queued work; agent_settled is the only point where no further automatic work follows. The adapter stopped at agent_end and closed stdin, so a recoverable provider failure returned no answer, and usage was overwritten per event, keeping only the last turn. - Read events through agent_settled and close stdin only after settlement or a terminal failure, draining stdout during shutdown. - Take the answer only from the final completed assistant message; a final error or abort fails with Pi's diagnostic, and earlier turns or failed attempts never stand in for the answer. - Fail truncated or unsettled streams, prompts handled without a run, and runtimes that go quiet after a run boundary without agent_settled (incompatible runtime after a bounded grace period). - Sum usage once per completed assistant message and compaction summary, ignoring cumulative message_update usage and agent_end snapshots. - Document the Pi 1.0 completion contract in the README. --- README.md | 24 +- internal/llmadapters/pi_rpc.go | 242 ++++++++++---- internal/llmadapters/pi_rpc_test.go | 468 ++++++++++++++++++++++++++-- 3 files changed, 648 insertions(+), 86 deletions(-) diff --git a/README.md b/README.md index 007e8518..839c7f22 100644 --- a/README.md +++ b/README.md @@ -115,14 +115,22 @@ cr init --non-interactive \ ``` Setup with Pi's local RPC runtime. Install Pi's coding agent and make sure the -`pi` binary is available on `PATH` before running `cr review`. New installs -should use the current npm package (`@earendil-works/pi-coding-agent`); existing -installs from the previous npm scope can also work when CR's compatibility -preflight confirms the reviewer controls it requires: RPC/system-prompt mode; -`--no-builtin-tools` with an exact `--tools` allowlist; explicit `--extension` loading while -`--no-extensions` disables discovery; and `--no-context-files`, `--no-approve`, -`--no-skills`, `--no-prompt-templates`, `--no-themes`, and `--no-session`. -CR preflights these capabilities before starting a Pi reviewer and returns an +`pi` binary is available on `PATH` before running `cr review`. Use Pi 1.0 or +newer from the current npm package (`@earendil-works/pi-coding-agent`). CR +relies on Pi 1.0's RPC completion contract: it keeps reading events through +automatic retries and compaction recovery until `agent_settled`, takes the +answer only from the final completed assistant message, and sums usage from +each completed assistant message and compaction summary. A final provider +error or abort fails the task with Pi's diagnostic. Older runtimes, including +installs from the previous npm scope, that end a run without `agent_settled` +fail with an incompatible-runtime error after a short grace period instead of +returning a result. + +CR also preflights the reviewer controls it requires before starting a Pi +reviewer: RPC/system-prompt mode; `--no-builtin-tools` with an exact `--tools` +allowlist; explicit `--extension` loading while `--no-extensions` disables +discovery; and `--no-context-files`, `--no-approve`, `--no-skills`, +`--no-prompt-templates`, `--no-themes`, and `--no-session`. It returns an incompatible-runtime error when any control is unavailable. ```bash diff --git a/internal/llmadapters/pi_rpc.go b/internal/llmadapters/pi_rpc.go index a98dee19..15c7e4ce 100644 --- a/internal/llmadapters/pi_rpc.go +++ b/internal/llmadapters/pi_rpc.go @@ -14,6 +14,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/open-cli-collective/codereview-cli/internal/llm" @@ -27,9 +28,14 @@ const ( piRPCReviewerToolNames = "cr_read,cr_search,cr_list,cr_diff" piRPCToolEvidenceReserve = 256 piRPCToolDiagnosticMaxRunes = 128 + piRPCErrorMaxRunes = 512 piRPCPreflightTimeout = 5 * time.Second piRPCPreflightOutputBytes = 64 * 1024 piRPCPreflightRegistration = "codereview-pi-reviewer-tools-registered cr_read,cr_search,cr_list,cr_diff" + // piRPCSettlementGrace bounds the quiet period after a run boundary that + // does not announce more work. Pi 1.0 follows it with agent_settled; older + // runtimes never do and would otherwise idle until the task deadline. + piRPCSettlementGrace = 30 * time.Second ) // ErrPiRPCIncompatible reports that the installed Pi runtime cannot enforce @@ -54,6 +60,7 @@ type PiRPCAdapter struct { timeout time.Duration scratchDirFactory ScratchDirFactory fastModeModels []string + settlementGrace time.Duration preflightMu sync.Mutex preflightReady bool } @@ -84,6 +91,7 @@ func NewPiRPCAdapter(opts PiRPCOptions) *PiRPCAdapter { timeout: timeout, scratchDirFactory: factory, fastModeModels: append([]string(nil), opts.FastModeModels...), + settlementGrace: piRPCSettlementGrace, } } @@ -165,6 +173,7 @@ func (a *PiRPCAdapter) Start(ctx context.Context, req Request) (Stream, error) { stdin: process.Stdin(), allowReviewerTools: req.ReviewerWorkspace != nil, logBytesLeft: -1, + settlementGrace: a.settlementGrace, } if req.ReviewerWorkspace != nil { stream.toolEvidenceBytesLeft = min(piRPCToolEvidenceReserve, req.ReviewerWorkspace.MaxToolOutputBytes) @@ -489,6 +498,7 @@ type piRPCStream struct { baseStream stdin io.Closer allowReviewerTools bool + settlementGrace time.Duration logLimitMu sync.Mutex logBytesLeft int logCapped bool @@ -512,10 +522,19 @@ func (s *piRPCStream) run(ctx context.Context, cmd *exec.Cmd, stdout io.Reader, }() scanResult := s.scanStdout(stdout) + // Closing stdin asks Pi to shut down, so it happens only after settlement + // or a terminal failure. Keep reading stdout so shutdown never blocks on a + // full pipe. if s.stdin != nil { _ = s.stdin.Close() } + stdoutDone := make(chan struct{}) + go func() { + defer close(stdoutDone) + _, _ = io.Copy(piRPCLogWriter{stream: s}, stdout) + }() waitErr := cmd.Wait() + <-stdoutDone <-stderrDone evidence := s.reviewerToolEvidence() s.writeReviewerToolEvidence(evidence) @@ -527,18 +546,18 @@ func (s *piRPCStream) run(ctx context.Context, cmd *exec.Cmd, stdout io.Reader, result.err = scanResult.err case errors.Is(ctx.Err(), context.DeadlineExceeded): result.err = ctx.Err() - case scanResult.agentEnd && waitErr != nil: + case scanResult.settled && waitErr != nil: result.err = waitErr - case scanResult.agentEnd && len(scanResult.response.StructuredOutput) == 0: + case scanResult.settled && len(scanResult.response.StructuredOutput) == 0: result.err = errors.New("llm pi rpc: no structured output") - case scanResult.agentEnd: + case scanResult.settled: result.err = nil case ctx.Err() != nil: result.err = ctx.Err() case waitErr != nil: result.err = waitErr - case !scanResult.agentEnd: - result.err = errors.New("llm pi rpc: missing agent_end") + default: + result.err = errors.New("llm pi rpc: stream ended before agent_settled") } recordRequestDuration(&result.response, start, result.err) s.Cancel() @@ -551,14 +570,26 @@ func (s *piRPCStream) run(ctx context.Context, cmd *exec.Cmd, stdout io.Reader, type piRPCScanResult struct { response Response err error - agentEnd bool + settled bool } +// scanStdout consumes RPC records until agent_settled, the only boundary after +// which Pi does no further automatic work. agent_end can be followed by +// retries, compaction recovery, or queued work, and a successful prompt +// response only acknowledges acceptance. func (s *piRPCStream) scanStdout(stdout io.Reader) piRPCScanResult { scanner := bufio.NewScanner(stdout) scanner.Buffer(make([]byte, 0, 64*1024), 4*1024*1024) var result piRPCScanResult + var final *piRPCAssistantMessage + var retryFailure string + settlement := piRPCSettlementWatch{grace: s.settlementGrace, expire: s.Cancel} + defer settlement.stop() for scanner.Scan() { + if settlement.expired.Load() { + break + } + settlement.stop() line := append([]byte(nil), scanner.Bytes()...) s.writeLog(normalizePiRPCLogLine(line)) event, err := parsePiRPCEvent(line) @@ -581,14 +612,30 @@ func (s *piRPCStream) scanStdout(stdout io.Reader) piRPCScanResult { result.err = fmt.Errorf("llm pi rpc: prompt failed: %s", event.responseFailure) return result } - if len(event.structuredOutput) > 0 { - result.response.StructuredOutput = event.structuredOutput + if event.promptHandled { + s.Cancel() + result.err = errors.New("llm pi rpc: prompt was handled without starting a run") + return result + } + result.response.Usage = addPiRPCUsage(result.response.Usage, event.usage) + if event.assistant != nil { + final = event.assistant } - result.response.Usage = mergeUsage(result.response.Usage, event.usage) - if event.agentEnd { - result.agentEnd = true + if event.retryFailure != "" { + retryFailure = event.retryFailure + } + if event.settled { + result.settled = true + result.response.StructuredOutput, result.err = piRPCFinalAnswer(final, retryFailure) return result } + if event.awaitsSettlement { + settlement.arm() + } + } + if settlement.expired.Load() { + result.err = fmt.Errorf("%w: no agent_settled within %s after the run ended; CR requires the Pi 1.0 RPC completion contract", ErrPiRPCIncompatible, s.settlementGrace) + return result } if err := scanner.Err(); err != nil && result.err == nil { result.err = err @@ -596,6 +643,80 @@ func (s *piRPCStream) scanStdout(stdout io.Reader) piRPCScanResult { return result } +// piRPCFinalAnswer accepts only the last completed assistant message, so text +// from an earlier turn or failed attempt never stands in for the answer. +func piRPCFinalAnswer(final *piRPCAssistantMessage, retryFailure string) ([]byte, error) { + switch { + case final == nil: + return nil, nil + case final.stopReason == "error" || final.stopReason == "aborted": + diagnostic := final.errorMessage + if diagnostic == "" { + diagnostic = retryFailure + } + if diagnostic == "" { + diagnostic = "no error message reported" + } + return nil, fmt.Errorf("llm pi rpc: final assistant message %s: %s", final.stopReason, boundPiRPCDiagnostic(diagnostic, piRPCErrorMaxRunes)) + case final.stopReason == "toolUse": + return nil, errors.New("llm pi rpc: run settled without a final answer after tool use") + default: + return final.output, nil + } +} + +// piRPCSettlementWatch fails a run that goes quiet after a boundary which +// announced no further work but never reports agent_settled. +type piRPCSettlementWatch struct { + grace time.Duration + expire func() + timer *time.Timer + expired atomic.Bool +} + +func (w *piRPCSettlementWatch) arm() { + if w.grace <= 0 { + return + } + w.timer = time.AfterFunc(w.grace, func() { + w.expired.Store(true) + w.expire() + }) +} + +func (w *piRPCSettlementWatch) stop() { + if w.timer != nil { + w.timer.Stop() + w.timer = nil + } +} + +// addPiRPCUsage adds one completed usage report to the run total. Fields that +// no report has provided stay nil (unknown) rather than becoming zero. +func addPiRPCUsage(total Usage, next Usage) Usage { + total.TokensIn = addPiRPCUsageField(total.TokensIn, next.TokensIn) + total.TokensOut = addPiRPCUsageField(total.TokensOut, next.TokensOut) + total.CacheRead = addPiRPCUsageField(total.CacheRead, next.CacheRead) + total.CacheCreate = addPiRPCUsageField(total.CacheCreate, next.CacheCreate) + total.CacheCreate5m = addPiRPCUsageField(total.CacheCreate5m, next.CacheCreate5m) + total.CacheCreate1h = addPiRPCUsageField(total.CacheCreate1h, next.CacheCreate1h) + total.CostUSD = addPiRPCUsageField(total.CostUSD, next.CostUSD) + return total +} + +func addPiRPCUsageField[T int | float64](total *T, next *T) *T { + switch { + case next == nil: + return total + case total == nil: + value := *next + return &value + default: + value := *total + *next + return &value + } +} + type piRPCLogWriter struct{ stream *piRPCStream } func (w piRPCLogWriter) Write(p []byte) (int, error) { @@ -670,7 +791,7 @@ func (s *piRPCStream) reviewerToolEvidence() *llm.ReviewerToolEvidence { } evidence := &llm.ReviewerToolEvidence{DiffStatus: status} if status == llm.DiffToolStatusFailed { - evidence.DiffDiagnostic = boundPiRPCToolError(s.diffToolError) + evidence.DiffDiagnostic = boundPiRPCDiagnostic(s.diffToolError, piRPCToolDiagnosticMaxRunes) } return evidence } @@ -693,13 +814,13 @@ func (s *piRPCStream) writeReviewerToolEvidence(evidence *llm.ReviewerToolEviden } } -func boundPiRPCToolError(value string) string { +func boundPiRPCDiagnostic(value string, maxRunes int) string { value = strings.Join(strings.Fields(strings.TrimSpace(value)), " ") runes := []rune(value) - if len(runes) <= piRPCToolDiagnosticMaxRunes { + if len(runes) <= maxRunes { return value } - return string(runes[:piRPCToolDiagnosticMaxRunes-3]) + "..." + return string(runes[:maxRunes-3]) + "..." } func piRPCToolError(raw json.RawMessage) string { @@ -872,7 +993,7 @@ func compactPiRPCPartialForLog(partialRaw json.RawMessage) json.RawMessage { type piRPCEvent struct { sessionID string - structuredOutput []byte + assistant *piRPCAssistantMessage usage Usage toolUse bool toolName string @@ -881,7 +1002,18 @@ type piRPCEvent struct { toolFailed bool toolError string responseFailure string - agentEnd bool + promptHandled bool + retryFailure string + awaitsSettlement bool + settled bool +} + +// piRPCAssistantMessage is the authoritative message_end form of one +// completed assistant message. +type piRPCAssistantMessage struct { + output []byte + stopReason string + errorMessage string } func parsePiRPCEvent(line []byte) (piRPCEvent, error) { @@ -896,7 +1028,6 @@ func parsePiRPCEvent(line []byte) (piRPCEvent, error) { eventType := rawString(raw, "type") event := piRPCEvent{ toolUse: piRPCEventIndicatesToolUse(eventType) || valueIndicatesToolUse(decoded), - usage: parsePiRPCUsage(raw), } if event.toolUse { event.toolName = firstRawString(raw, "toolName", "tool_name", "name") @@ -921,26 +1052,41 @@ func parsePiRPCEvent(line []byte) (piRPCEvent, error) { if id := firstRawString(raw, "sessionId", "session_id"); id != "" { event.sessionID = id } - if eventType == "response" && rawString(raw, "command") == "prompt" && !rawBool(raw, "success") { - event.responseFailure = rawString(raw, "error") - if event.responseFailure == "" { - event.responseFailure = "unknown error" + if eventType == "response" && rawString(raw, "command") == "prompt" { + if !rawBool(raw, "success") { + event.responseFailure = rawString(raw, "error") + if event.responseFailure == "" { + event.responseFailure = "unknown error" + } + return event, nil + } + var data map[string]json.RawMessage + if err := json.Unmarshal(raw["data"], &data); err == nil && rawString(data, "disposition") == "handled" { + event.promptHandled = true } return event, nil } - if eventType == "message_end" { - if output, usage := parsePiRPCMessageEnd(raw["message"]); len(output) > 0 { - event.structuredOutput = output - event.usage = mergeUsage(event.usage, usage) - } else { - event.usage = mergeUsage(event.usage, usage) + // Usage is counted only from completed reports: message_end for each + // assistant message and compaction_end for summary generation. + // message_update usage is cumulative, and agent_end.messages repeats + // messages already reported by message_end. + switch eventType { + case "message_end": + event.assistant, event.usage = parsePiRPCMessageEnd(raw["message"]) + case "compaction_end": + var result map[string]json.RawMessage + if err := json.Unmarshal(raw["result"], &result); err == nil { + event.usage = parsePiRPCUsage(result) } - } - if eventType == "agent_end" { - event.agentEnd = true - if output := parsePiRPCAgentEnd(raw["messages"]); len(output) > 0 { - event.structuredOutput = output + event.awaitsSettlement = !rawBool(raw, "willRetry") + case "auto_retry_end": + if !rawBool(raw, "success") { + event.retryFailure = rawString(raw, "finalError") } + case "agent_end": + event.awaitsSettlement = !rawBool(raw, "willRetry") + case "agent_settled": + event.settled = true } return event, nil } @@ -1026,7 +1172,7 @@ func piRPCEventIndicatesToolUse(value string) bool { normalized == "bash" } -func parsePiRPCMessageEnd(value json.RawMessage) ([]byte, Usage) { +func parsePiRPCMessageEnd(value json.RawMessage) (*piRPCAssistantMessage, Usage) { var raw map[string]json.RawMessage if len(value) == 0 { return nil, Usage{} @@ -1034,30 +1180,14 @@ func parsePiRPCMessageEnd(value json.RawMessage) ([]byte, Usage) { if err := json.Unmarshal(value, &raw); err != nil { return nil, Usage{} } - usage := parsePiRPCUsage(raw) if rawString(raw, "role") != "assistant" { - return nil, usage - } - return extractPiRPCText(raw["content"]), usage -} - -func parsePiRPCAgentEnd(value json.RawMessage) []byte { - var messages []map[string]json.RawMessage - if len(value) == 0 { - return nil - } - if err := json.Unmarshal(value, &messages); err != nil { - return nil - } - for i := len(messages) - 1; i >= 0; i-- { - if rawString(messages[i], "role") != "assistant" { - continue - } - if output := extractPiRPCText(messages[i]["content"]); len(output) > 0 { - return output - } + return nil, Usage{} } - return nil + return &piRPCAssistantMessage{ + output: extractPiRPCText(raw["content"]), + stopReason: rawString(raw, "stopReason"), + errorMessage: rawString(raw, "errorMessage"), + }, parsePiRPCUsage(raw) } func extractPiRPCText(value json.RawMessage) []byte { diff --git a/internal/llmadapters/pi_rpc_test.go b/internal/llmadapters/pi_rpc_test.go index d4cd494a..b20364e6 100644 --- a/internal/llmadapters/pi_rpc_test.go +++ b/internal/llmadapters/pi_rpc_test.go @@ -1137,7 +1137,7 @@ func TestPiRPCProtocolFailures(t *testing.T) { } }) - t.Run("caller cancellation propagates before agent_end", func(t *testing.T) { + t.Run("caller cancellation propagates before any event", func(t *testing.T) { recordPath := filepath.Join(t.TempDir(), "record.json") ctx, cancel := context.WithCancel(context.Background()) adapter := NewPiRPCAdapter(PiRPCOptions{ @@ -1178,7 +1178,7 @@ func TestPiRPCProtocolFailures(t *testing.T) { } }) - t.Run("missing agent_end fails stream", func(t *testing.T) { + t.Run("stream ending before agent_settled fails", func(t *testing.T) { recordPath := filepath.Join(t.TempDir(), "record.json") adapter := NewPiRPCAdapter(PiRPCOptions{ Command: os.Args[0], @@ -1191,17 +1191,17 @@ func TestPiRPCProtocolFailures(t *testing.T) { t.Fatalf("Start: %v", err) } _, err = stream.Wait(context.Background()) - if err == nil || !strings.Contains(err.Error(), "agent_end") { - t.Fatalf("Wait error = %v, want missing agent_end", err) + if err == nil || !strings.Contains(err.Error(), "agent_settled") { + t.Fatalf("Wait error = %v, want missing agent_settled", err) } }) - t.Run("non-zero exit after agent_end fails stream", func(t *testing.T) { + t.Run("non-zero exit after agent_settled fails stream", func(t *testing.T) { recordPath := filepath.Join(t.TempDir(), "record.json") adapter := NewPiRPCAdapter(PiRPCOptions{ Command: os.Args[0], commandArgsPrefix: piRPCHelperPrefix(), - Env: piRPCHelperEnv("agent-end-exit-failure", recordPath), + Env: piRPCHelperEnv("settled-exit-failure", recordPath), Timeout: 5 * time.Second, }) stream, err := adapter.Start(context.Background(), Request{Model: "opencode-go/kimi-k2.6", Prompt: "prompt"}) @@ -1214,7 +1214,7 @@ func TestPiRPCProtocolFailures(t *testing.T) { } }) - t.Run("non-zero exit before agent_end fails stream", func(t *testing.T) { + t.Run("non-zero exit before agent_settled fails stream", func(t *testing.T) { recordPath := filepath.Join(t.TempDir(), "record.json") adapter := NewPiRPCAdapter(PiRPCOptions{ Command: os.Args[0], @@ -1233,6 +1233,389 @@ func TestPiRPCProtocolFailures(t *testing.T) { }) } +const piRPCTestPromptStarted = `{"id":"prompt-1","type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}` + +func TestPiRPCSettlementContract(t *testing.T) { + answer := `{"ok":true}` + overloaded := piRPCTestTurn{stopReason: "error", errorMessage: "503 overloaded", usage: piRPCTestUsage(7, 0, nil, 0.25)} + final := piRPCTestTurn{text: answer, stopReason: "stop", usage: piRPCTestUsage(100, 10, map[string]any{"cacheRead": 3, "cacheWrite": 4}, 0.5)} + finalUsage := Usage{TokensIn: piRPCTestPtr(100), TokensOut: piRPCTestPtr(10), CacheRead: piRPCTestPtr(3), CacheCreate: piRPCTestPtr(4), CostUSD: piRPCTestPtr(0.5)} + for _, tt := range []struct { + name string + script []string + wantOutput string + wantErr string + wantUsage Usage + }{ + { + name: "recoverable agent_end continues through retry to settlement", + script: []string{ + piRPCTestPromptStarted, + `{"type":"agent_start"}`, + overloaded.end(), + piRPCTestAgentEnd(true, overloaded), + `#pause`, + `{"type":"auto_retry_start","attempt":1,"maxAttempts":3,"delayMs":100,"errorMessage":"503 overloaded"}`, + `{"type":"agent_start"}`, + `{"type":"message_update","usage":{"input":90,"output":5,"cost":{"total":9}},"assistantMessageEvent":{"type":"text_delta","contentIndex":0,"delta":"{"}}`, + final.end(), + `{"type":"auto_retry_end","success":true,"attempt":1}`, + piRPCTestAgentEnd(false, final), + `{"type":"agent_settled"}`, + }, + wantOutput: answer, + wantUsage: Usage{TokensIn: piRPCTestPtr(107), TokensOut: piRPCTestPtr(10), CacheRead: piRPCTestPtr(3), CacheCreate: piRPCTestPtr(4), CostUSD: piRPCTestPtr(0.75)}, + }, + { + name: "overflow compaction recovery counts summary usage once", + script: []string{ + piRPCTestPromptStarted, + piRPCTestTurn{stopReason: "error", errorMessage: "context overflow", usage: piRPCTestUsage(0, 0, nil, 0)}.end(), + `{"type":"agent_end","messages":[],"willRetry":false}`, + `{"type":"compaction_start","reason":"overflow"}`, + `{"type":"compaction_end","reason":"overflow","result":{"summary":"s","tokensBefore":9000,"usage":{"input":50,"output":5,"cost":{"total":0.25}}},"aborted":false,"willRetry":true}`, + `{"type":"agent_start"}`, + final.end(), + piRPCTestAgentEnd(false, final), + `{"type":"agent_settled"}`, + }, + wantOutput: answer, + wantUsage: Usage{TokensIn: piRPCTestPtr(150), TokensOut: piRPCTestPtr(15), CacheRead: piRPCTestPtr(3), CacheCreate: piRPCTestPtr(4), CostUSD: piRPCTestPtr(0.75)}, + }, + { + name: "completed assistant turns sum usage once and keep only the final answer", + script: []string{ + piRPCTestPromptStarted, + piRPCTestTurn{text: "Checking the diff.", toolCall: true, stopReason: "toolUse", usage: piRPCTestUsage(100, 10, map[string]any{"cacheRead": 1}, 0.25)}.end(), + `{"type":"message_end","message":{"role":"toolResult","toolCallId":"call-1","toolName":"cr_diff","content":[{"type":"text","text":"diff"}],"isError":false}}`, + `{"type":"message_update","usage":{"input":200,"output":20,"cost":{"total":0.5}},"assistantMessageEvent":{"type":"text_delta","contentIndex":0,"delta":"{"}}`, + `{"type":"message_update","usage":{"input":200,"output":20,"cost":{"total":0.5}},"assistantMessageEvent":{"type":"text_end","contentIndex":0,"content":"{\"ok\":true}"}}`, + piRPCTestTurn{text: answer, stopReason: "stop", usage: piRPCTestUsage(200, 20, map[string]any{"cacheRead": 2}, 0.5)}.end(), + `{"type":"agent_end","messages":[{"role":"assistant","content":[],"usage":{"input":100,"output":10},"stopReason":"toolUse"},{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"usage":{"input":200,"output":20},"stopReason":"stop"}],"willRetry":false}`, + `{"type":"agent_settled"}`, + }, + wantOutput: answer, + wantUsage: Usage{TokensIn: piRPCTestPtr(300), TokensOut: piRPCTestPtr(30), CacheRead: piRPCTestPtr(3), CostUSD: piRPCTestPtr(0.75)}, + }, + { + name: "provider error after exhausted retries fails with diagnostic", + script: []string{ + piRPCTestPromptStarted, + piRPCTestTurn{text: answer, stopReason: "error", errorMessage: "503 overloaded", usage: piRPCTestUsage(7, 0, nil, 0.25)}.end(), + `{"type":"agent_end","messages":[],"willRetry":true}`, + `{"type":"auto_retry_start","attempt":1,"maxAttempts":1,"delayMs":1,"errorMessage":"503 overloaded"}`, + overloaded.end(), + `{"type":"auto_retry_end","success":false,"attempt":1,"finalError":"503 overloaded"}`, + `{"type":"agent_end","messages":[],"willRetry":false}`, + `{"type":"agent_settled"}`, + }, + wantErr: "503 overloaded", + wantUsage: Usage{TokensIn: piRPCTestPtr(14), TokensOut: piRPCTestPtr(0), CostUSD: piRPCTestPtr(0.5)}, + }, + { + name: "aborted final assistant fails", + script: []string{ + piRPCTestPromptStarted, + piRPCTestTurn{text: `{"ok":`, stopReason: "aborted", errorMessage: "Request was aborted", usage: piRPCTestUsage(5, 1, nil, 0)}.end(), + `{"type":"agent_end","messages":[],"willRetry":false}`, + `{"type":"agent_settled"}`, + }, + wantErr: "Request was aborted", + wantUsage: Usage{TokensIn: piRPCTestPtr(5), TokensOut: piRPCTestPtr(1), CostUSD: piRPCTestPtr(0.0)}, + }, + { + name: "earlier valid answer then final provider error fails", + script: []string{ + piRPCTestPromptStarted, + piRPCTestTurn{text: answer, toolCall: true, stopReason: "toolUse", usage: piRPCTestUsage(10, 2, nil, 0)}.end(), + piRPCTestTurn{stopReason: "error", errorMessage: "terminated", usage: piRPCTestUsage(0, 0, nil, 0)}.end(), + `{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"stopReason":"toolUse"}],"willRetry":false}`, + `{"type":"agent_settled"}`, + }, + wantErr: "terminated", + wantUsage: Usage{TokensIn: piRPCTestPtr(10), TokensOut: piRPCTestPtr(2), CostUSD: piRPCTestPtr(0.0)}, + }, + { + name: "settled run without a final answer fails", + script: []string{ + piRPCTestPromptStarted, + piRPCTestTurn{text: "Checking the diff.", toolCall: true, stopReason: "toolUse", usage: piRPCTestUsage(10, 2, nil, 0)}.end(), + `{"type":"agent_end","messages":[],"willRetry":false}`, + `{"type":"agent_settled"}`, + }, + wantErr: "final answer", + wantUsage: Usage{TokensIn: piRPCTestPtr(10), TokensOut: piRPCTestPtr(2), CostUSD: piRPCTestPtr(0.0)}, + }, + { + name: "stream truncated after agent_end fails", + script: []string{ + piRPCTestPromptStarted, + final.end(), + piRPCTestAgentEnd(false, final), + `#exit`, + }, + wantErr: "agent_settled", + wantUsage: finalUsage, + }, + { + name: "malformed record after valid answer fails without output", + script: []string{ + piRPCTestPromptStarted, + final.end(), + `{"type":`, + }, + wantErr: "malformed JSONL", + wantUsage: finalUsage, + }, + { + name: "prompt handled without a run fails", + script: []string{ + `{"id":"prompt-1","type":"response","command":"prompt","success":true,"data":{"disposition":"handled"}}`, + }, + wantErr: "handled", + }, + { + name: "prompt rejection fails", + script: []string{ + `{"id":"prompt-1","type":"response","command":"prompt","success":false,"error":"No API key found for opencode-go"}`, + }, + wantErr: "No API key found", + }, + } { + t.Run(tt.name, func(t *testing.T) { + adapter := piRPCScriptAdapter(t, 5*time.Second, tt.script...) + stream, err := adapter.Start(context.Background(), Request{Model: "opencode-go/kimi-k2.6", Prompt: "prompt"}) + if err != nil { + t.Fatalf("Start: %v", err) + } + start := time.Now() + response, err := stream.Wait(context.Background()) + if elapsed := time.Since(start); elapsed > 3*time.Second { + t.Fatalf("Wait took %s, want completion without waiting for the task deadline", elapsed) + } + if tt.wantErr == "" && err != nil { + t.Fatalf("Wait: %v", err) + } + if tt.wantErr != "" && (err == nil || !strings.Contains(err.Error(), tt.wantErr)) { + t.Fatalf("Wait error = %v, want %q", err, tt.wantErr) + } + if string(response.StructuredOutput) != tt.wantOutput { + t.Fatalf("StructuredOutput = %q, want %q", response.StructuredOutput, tt.wantOutput) + } + if !reflect.DeepEqual(response.Usage, tt.wantUsage) { + t.Fatalf("Usage = %s, want %s", formatPiRPCTestUsage(response.Usage), formatPiRPCTestUsage(tt.wantUsage)) + } + }) + } +} + +func TestPiRPCReviewerSettlementPreservesToolEvidence(t *testing.T) { + tempDir := t.TempDir() + repoDir := filepath.Join(tempDir, "repo") + scratchDir := filepath.Join(tempDir, "scratch") + for _, dir := range []string{repoDir, scratchDir} { + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatalf("MkdirAll(%s): %v", dir, err) + } + } + diffPath := filepath.Join(tempDir, "diff.patch") + if err := os.WriteFile(diffPath, []byte("fixed diff\n"), 0o600); err != nil { + t.Fatalf("WriteFile(diff): %v", err) + } + commentary := piRPCTestTurn{text: "Checking the diff.", toolCall: true, stopReason: "toolUse", usage: piRPCTestUsage(100, 10, nil, 0.25)} + overloaded := piRPCTestTurn{stopReason: "error", errorMessage: "503 overloaded", usage: piRPCTestUsage(0, 0, nil, 0)} + final := piRPCTestTurn{text: `{"ok":true}`, stopReason: "stop", usage: piRPCTestUsage(200, 20, nil, 0.5)} + adapter := piRPCScriptAdapter(t, 5*time.Second, + piRPCTestPromptStarted, + commentary.end(), + `{"type":"tool_execution_start","toolCallId":"diff-1","toolName":"cr_diff","args":{}}`, + `{"type":"tool_execution_end","toolCallId":"diff-1","toolName":"cr_diff","result":{"content":[{"type":"text","text":"diff"}]},"isError":false}`, + overloaded.end(), + piRPCTestAgentEnd(true, commentary, overloaded), + `#pause`, + `{"type":"auto_retry_start","attempt":1,"maxAttempts":3,"delayMs":100,"errorMessage":"503 overloaded"}`, + final.end(), + `{"type":"auto_retry_end","success":true,"attempt":1}`, + piRPCTestAgentEnd(false, final), + `{"type":"agent_settled"}`, + ) + stream, err := adapter.Start(context.Background(), Request{ + Prompt: "review", + ReviewerWorkspace: &ReviewerWorkspaceRequest{ + RepoDir: repoDir, ScratchDir: scratchDir, DiffPath: diffPath, MaxToolOutputBytes: 2048, + }, + }) + if err != nil { + t.Fatalf("Start: %v", err) + } + response, err := stream.Wait(context.Background()) + if err != nil { + t.Fatalf("Wait: %v", err) + } + if string(response.StructuredOutput) != `{"ok":true}` { + t.Fatalf("StructuredOutput = %q, want recovered final answer", response.StructuredOutput) + } + if response.ReviewerToolEvidence == nil || response.ReviewerToolEvidence.DiffStatus != llm.DiffToolStatusSucceeded { + t.Fatalf("reviewer tool evidence = %#v, want succeeded cr_diff", response.ReviewerToolEvidence) + } + want := Usage{TokensIn: piRPCTestPtr(300), TokensOut: piRPCTestPtr(30), CostUSD: piRPCTestPtr(0.75)} + if !reflect.DeepEqual(response.Usage, want) { + t.Fatalf("Usage = %s, want %s", formatPiRPCTestUsage(response.Usage), formatPiRPCTestUsage(want)) + } +} + +func TestPiRPCPendingRetryHonorsDeadlineAndCancellation(t *testing.T) { + pendingRetry := []string{ + piRPCTestPromptStarted, + piRPCTestTurn{stopReason: "error", errorMessage: "503 overloaded", usage: piRPCTestUsage(0, 0, nil, 0)}.end(), + `{"type":"agent_end","messages":[],"willRetry":true}`, + `{"type":"auto_retry_start","attempt":1,"maxAttempts":3,"delayMs":60000,"errorMessage":"503 overloaded"}`, + } + t.Run("deadline", func(t *testing.T) { + adapter := piRPCScriptAdapter(t, 300*time.Millisecond, pendingRetry...) + stream, err := adapter.Start(context.Background(), Request{Model: "opencode-go/kimi-k2.6", Prompt: "prompt"}) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err := stream.Wait(context.Background()); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Wait error = %v, want deadline exceeded while retry is pending", err) + } + }) + t.Run("cancellation", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + adapter := piRPCScriptAdapter(t, 5*time.Second, pendingRetry...) + stream, err := adapter.Start(ctx, Request{Model: "opencode-go/kimi-k2.6", Prompt: "prompt"}) + if err != nil { + t.Fatalf("Start: %v", err) + } + time.Sleep(300 * time.Millisecond) + cancel() + if _, err := stream.Wait(context.Background()); !errors.Is(err, context.Canceled) { + t.Fatalf("Wait error = %v, want context canceled while retry is pending", err) + } + }) +} + +func TestPiRPCMissingSettlementFailsAsIncompatibleRuntime(t *testing.T) { + final := piRPCTestTurn{text: `{"ok":true}`, stopReason: "stop", usage: piRPCTestUsage(100, 10, nil, 0.5)} + adapter := piRPCScriptAdapter(t, 5*time.Second, + `{"id":"prompt-1","type":"response","command":"prompt","success":true}`, + final.end(), + `{"type":"agent_end","messages":[]}`, + ) + adapter.settlementGrace = 100 * time.Millisecond + stream, err := adapter.Start(context.Background(), Request{Model: "opencode-go/kimi-k2.6", Prompt: "prompt"}) + if err != nil { + t.Fatalf("Start: %v", err) + } + start := time.Now() + response, err := stream.Wait(context.Background()) + if !errors.Is(err, ErrPiRPCIncompatible) || !strings.Contains(err.Error(), "agent_settled") { + t.Fatalf("Wait error = %v, want incompatible runtime without agent_settled", err) + } + if elapsed := time.Since(start); elapsed > 3*time.Second { + t.Fatalf("Wait took %s, want bounded settlement grace", elapsed) + } + if len(response.StructuredOutput) != 0 { + t.Fatalf("StructuredOutput = %q, want none without settlement", response.StructuredOutput) + } +} + +func TestPiRPCSettlementGraceRestartsOnFollowUpWork(t *testing.T) { + final := piRPCTestTurn{text: `{"ok":true}`, stopReason: "stop", usage: piRPCTestUsage(100, 10, nil, 0.5)} + adapter := piRPCScriptAdapter(t, 5*time.Second, + piRPCTestPromptStarted, + final.end(), + piRPCTestAgentEnd(false, final), + `#pause`, + `{"type":"compaction_start","reason":"threshold"}`, + `#pause`, `#pause`, `#pause`, `#pause`, `#pause`, `#pause`, `#pause`, `#pause`, + `{"type":"compaction_end","reason":"threshold","result":{"summary":"s","tokensBefore":9000},"aborted":false,"willRetry":false}`, + `{"type":"agent_settled"}`, + ) + adapter.settlementGrace = 500 * time.Millisecond + stream, err := adapter.Start(context.Background(), Request{Model: "opencode-go/kimi-k2.6", Prompt: "prompt"}) + if err != nil { + t.Fatalf("Start: %v", err) + } + response, err := stream.Wait(context.Background()) + if err != nil { + t.Fatalf("Wait: %v", err) + } + if string(response.StructuredOutput) != `{"ok":true}` { + t.Fatalf("StructuredOutput = %q, want answer after threshold compaction settles", response.StructuredOutput) + } +} + +type piRPCTestTurn struct { + text string + toolCall bool + stopReason string + errorMessage string + usage map[string]any +} + +func (turn piRPCTestTurn) message() map[string]any { + content := []map[string]any{} + if turn.text != "" { + content = append(content, map[string]any{"type": "text", "text": turn.text}) + } + if turn.toolCall { + content = append(content, map[string]any{"type": "toolCall", "id": "call-1", "name": "cr_diff", "arguments": map[string]any{}}) + } + message := map[string]any{"role": "assistant", "content": content, "usage": turn.usage, "stopReason": turn.stopReason} + if turn.errorMessage != "" { + message["errorMessage"] = turn.errorMessage + } + return message +} + +func (turn piRPCTestTurn) end() string { + return piRPCTestLine(map[string]any{"type": "message_end", "message": turn.message()}) +} + +func piRPCTestAgentEnd(willRetry bool, turns ...piRPCTestTurn) string { + messages := make([]map[string]any, 0, len(turns)) + for _, turn := range turns { + messages = append(messages, turn.message()) + } + return piRPCTestLine(map[string]any{"type": "agent_end", "messages": messages, "willRetry": willRetry}) +} + +func piRPCTestUsage(input, output int, cache map[string]any, cost float64) map[string]any { + usage := map[string]any{"input": input, "output": output, "cost": map[string]any{"total": cost}} + for key, value := range cache { + usage[key] = value + } + return usage +} + +func piRPCTestLine(event map[string]any) string { + data, err := json.Marshal(event) + if err != nil { + panic(err) + } + return string(data) +} + +func piRPCTestPtr[T any](value T) *T { return &value } + +func formatPiRPCTestUsage(usage Usage) string { + data, _ := json.Marshal(usage) + return string(data) +} + +func piRPCScriptAdapter(t *testing.T, timeout time.Duration, script ...string) *PiRPCAdapter { + t.Helper() + return NewPiRPCAdapter(PiRPCOptions{ + Command: os.Args[0], + commandArgsPrefix: piRPCHelperPrefix(), + Env: append(piRPCHelperEnv("script", filepath.Join(t.TempDir(), "record.json")), "LLM_PI_RPC_SCRIPT="+strings.Join(script, "\n")), + Timeout: timeout, + }) +} + func TestPiRPCRejectsUnsafeSpecs(t *testing.T) { adapter := NewPiRPCAdapter(PiRPCOptions{}) req := Request{Model: "opencode-go/kimi-k2.6", Prompt: "prompt"} @@ -1362,17 +1745,54 @@ func TestPiRPCHelperProcess(_ *testing.T) { } os.Exit(0) } + // Pi RPC stays alive after agent_settled and shuts down when stdin closes. + stdinClosed := make(chan struct{}) + go func() { + for scanner.Scan() { + } + close(stdinClosed) + }() + awaitStdinClose := func() { + select { + case <-stdinClosed: + case <-time.After(10 * time.Second): + } + } + settle := func() { + fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"stopReason":"stop"}],"willRetry":false}`) + fmt.Println(`{"type":"agent_settled"}`) + awaitStdinClose() + } switch os.Getenv("LLM_HELPER_MODE") { case "success", "slow-success": if os.Getenv("LLM_HELPER_MODE") == "slow-success" { time.Sleep(slowSuccessSleep) } - fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":true}`) + fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":true,"data":{"disposition":"started"}}`) fmt.Println(`{"type":"agent_start","sessionId":"session-1"}`) fmt.Println(`{"type":"message_end","message":{"role":"user","content":"review this diff"}}`) - fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"thinking","text":"ignored"},{"type":"text","text":"{\"ok\":true}"}],"usage":{"tokensIn":3915,"tokensOut":50,"cacheRead":5,"cacheWrite":7,"cost":{"total":0.00392005}}}}`) - fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}]}`) + fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"thinking","text":"ignored"},{"type":"text","text":"{\"ok\":true}"}],"usage":{"tokensIn":3915,"tokensOut":50,"cacheRead":5,"cacheWrite":7,"cost":{"total":0.00392005}},"stopReason":"stop"}}`) + fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"usage":{"tokensIn":3915,"tokensOut":50,"cacheRead":5,"cacheWrite":7,"cost":{"total":0.00392005}},"stopReason":"stop"}],"willRetry":false}`) + fmt.Println(`{"type":"agent_settled"}`) + awaitStdinClose() + case "script": + for _, line := range strings.Split(os.Getenv("LLM_PI_RPC_SCRIPT"), "\n") { + switch line { + case "#pause": + // Emulate retry delay; Pi shuts down instead of retrying once stdin closes. + select { + case <-stdinClosed: + os.Exit(0) + case <-time.After(100 * time.Millisecond): + } + case "#exit": + os.Exit(0) + default: + fmt.Println(line) + } + } + awaitStdinClose() case "streaming-partials": fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":true}`) fmt.Println(`{"type":"agent_start","sessionId":"session-1"}`) @@ -1417,7 +1837,9 @@ func TestPiRPCHelperProcess(_ *testing.T) { 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}},"stopReason":"stop","timestamp":1700000000000}}`) - fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}]}`) + fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}],"willRetry":false}`) + fmt.Println(`{"type":"agent_settled"}`) + awaitStdinClose() case "prompt-failure": fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":false,"error":"No API key found for opencode-go"}`) case "tool": @@ -1430,28 +1852,28 @@ func TestPiRPCHelperProcess(_ *testing.T) { fmt.Printf("{\"type\":\"tool_execution_start\",\"toolCallId\":\"%s\",\"toolName\":%q,\"args\":{}}\n", tool, tool) fmt.Printf("{\"type\":\"tool_execution_end\",\"toolCallId\":\"%s\",\"toolName\":%q,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"ok\"}]}}\n", tool, tool) } - fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}}`) - fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}]}`) + fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"stopReason":"stop"}}`) + settle() case "reviewer-log-flood": fmt.Fprintln(os.Stderr, strings.Repeat("stderr flood\n", 1000)) fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":true}`) for i := 0; i < 20; i++ { fmt.Printf("{\"type\":\"tool_execution_end\",\"toolCallId\":\"tool-%d\",\"toolName\":\"cr_read\",\"result\":{\"content\":[{\"type\":\"text\",\"text\":%q}]}}\n", i, strings.Repeat("tool output ", 500)) } - fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}}`) - fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}]}`) + fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"stopReason":"stop"}}`) + settle() case "reviewer-diff-failure-log-flood": fmt.Fprintln(os.Stderr, strings.Repeat("stderr flood\n", 1000)) fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":true}`) fmt.Println(`{"type":"tool_execution_start","toolCallId":"diff-1","toolName":"cr_diff","args":{}}`) fmt.Println(`{"type":"tool_execution_end","toolCallId":"diff-1","toolName":"cr_diff","result":{"content":[{"type":"text","text":"fixed diff unavailable"}],"isError":true}}`) - fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}}`) - fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}]}`) + fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"stopReason":"stop"}}`) + settle() case "reviewer-diff-incomplete": fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":true}`) fmt.Println(`{"type":"tool_execution_start","toolCallId":"diff-1","toolName":"cr_diff","args":{}}`) - fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}}`) - fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}]}`) + fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"stopReason":"stop"}}`) + settle() case "sleep": time.Sleep(10 * time.Second) case "malformed": @@ -1459,10 +1881,12 @@ func TestPiRPCHelperProcess(_ *testing.T) { case "no-agent-end": fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":true}`) fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}}`) - case "agent-end-exit-failure": + case "settled-exit-failure": fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":true}`) - fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}}`) - fmt.Println(`{"type":"agent_end","messages":[{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}]}]}`) + fmt.Println(`{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"stopReason":"stop"}}`) + fmt.Println(`{"type":"agent_end","messages":[],"willRetry":false}`) + fmt.Println(`{"type":"agent_settled"}`) + awaitStdinClose() os.Exit(42) case "exit-before-agent-end": fmt.Println(`{"id":"prompt-1","type":"response","command":"prompt","success":true}`) From 8797d43863628badb8ba1293ddacf77c9ee3ad23 Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Fri, 2 Oct 2026 03:20:42 -0400 Subject: [PATCH 2/5] fix(pi): retain one-hour cache write usage [#638] Pi reports usage.cacheWrite1h as the one-hour-retention subset of the cacheWrite aggregate, but the adapter never mapped it, so per-task usage lost the cache duration split that Anthropic-backed models report. Map cacheWrite1h to the one-hour cache-create field and sum it once per completed assistant message alongside the aggregate. Messages that omit it leave it unknown, and no five-minute value is derived because Pi does not label the remainder with a retention. --- internal/llmadapters/pi_rpc.go | 6 +++- internal/llmadapters/pi_rpc_test.go | 56 +++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/internal/llmadapters/pi_rpc.go b/internal/llmadapters/pi_rpc.go index 15c7e4ce..41e3e309 100644 --- a/internal/llmadapters/pi_rpc.go +++ b/internal/llmadapters/pi_rpc.go @@ -1223,7 +1223,11 @@ func parsePiRPCUsage(raw map[string]json.RawMessage) Usage { TokensOut: firstRawIntPtr(usageRaw, "tokens_out", "tokensOut", "output", "outputTokens", "completionTokens"), CacheRead: firstRawIntPtr(usageRaw, "cache_read", "cacheRead"), CacheCreate: firstRawIntPtr(usageRaw, "cache_create", "cacheCreate", "cache_write", "cacheWrite"), - CostUSD: firstRawFloatPtr(usageRaw, "cost_usd", "costUSD", "totalCost", "totalCostUSD"), + // Pi reports cacheWrite1h as the one-hour subset of cacheWrite. The + // remainder is not labeled with a retention, so no five-minute value + // is derived from it. + CacheCreate1h: rawIntPtr(usageRaw, "cacheWrite1h"), + CostUSD: firstRawFloatPtr(usageRaw, "cost_usd", "costUSD", "totalCost", "totalCostUSD"), } if usage.CostUSD == nil { usage.CostUSD = nestedRawFloatPtr(usageRaw, "cost", "total") diff --git a/internal/llmadapters/pi_rpc_test.go b/internal/llmadapters/pi_rpc_test.go index b20364e6..986d5f2b 100644 --- a/internal/llmadapters/pi_rpc_test.go +++ b/internal/llmadapters/pi_rpc_test.go @@ -1297,6 +1297,31 @@ func TestPiRPCSettlementContract(t *testing.T) { wantOutput: answer, wantUsage: Usage{TokensIn: piRPCTestPtr(300), TokensOut: piRPCTestPtr(30), CacheRead: piRPCTestPtr(3), CostUSD: piRPCTestPtr(0.75)}, }, + { + name: "one-hour cache writes sum once per completed assistant message", + script: []string{ + piRPCTestPromptStarted, + piRPCTestTurn{text: "Checking the diff.", toolCall: true, stopReason: "toolUse", usage: piRPCTestUsage(100, 10, map[string]any{"cacheWrite": 5, "cacheWrite1h": 1}, 0.25)}.end(), + `{"type":"message_update","usage":{"input":200,"output":20,"cacheWrite":7,"cacheWrite1h":2,"cost":{"total":0.5}},"assistantMessageEvent":{"type":"text_delta","contentIndex":0,"delta":"{"}}`, + piRPCTestTurn{text: answer, stopReason: "stop", usage: piRPCTestUsage(200, 20, map[string]any{"cacheWrite": 7, "cacheWrite1h": 2}, 0.5)}.end(), + `{"type":"agent_end","messages":[{"role":"assistant","content":[],"usage":{"input":100,"output":10,"cacheWrite":5,"cacheWrite1h":1},"stopReason":"toolUse"},{"role":"assistant","content":[{"type":"text","text":"{\"ok\":true}"}],"usage":{"input":200,"output":20,"cacheWrite":7,"cacheWrite1h":2},"stopReason":"stop"}],"willRetry":false}`, + `{"type":"agent_settled"}`, + }, + wantOutput: answer, + wantUsage: Usage{TokensIn: piRPCTestPtr(300), TokensOut: piRPCTestPtr(30), CacheCreate: piRPCTestPtr(12), CacheCreate1h: piRPCTestPtr(3), CostUSD: piRPCTestPtr(0.75)}, + }, + { + name: "one-hour cache split reported by only some messages sums the reported values", + script: []string{ + piRPCTestPromptStarted, + piRPCTestTurn{text: "Checking the diff.", toolCall: true, stopReason: "toolUse", usage: piRPCTestUsage(100, 10, map[string]any{"cacheWrite": 5}, 0.25)}.end(), + piRPCTestTurn{text: answer, stopReason: "stop", usage: piRPCTestUsage(200, 20, map[string]any{"cacheWrite": 7, "cacheWrite1h": 2}, 0.5)}.end(), + piRPCTestAgentEnd(false), + `{"type":"agent_settled"}`, + }, + wantOutput: answer, + wantUsage: Usage{TokensIn: piRPCTestPtr(300), TokensOut: piRPCTestPtr(30), CacheCreate: piRPCTestPtr(12), CacheCreate1h: piRPCTestPtr(2), CostUSD: piRPCTestPtr(0.75)}, + }, { name: "provider error after exhausted retries fails with diagnostic", script: []string{ @@ -1409,6 +1434,37 @@ func TestPiRPCSettlementContract(t *testing.T) { } } +func TestParsePiRPCUsageCacheWriteRetention(t *testing.T) { + for _, tt := range []struct { + name string + usage string + want Usage + }{ + { + name: "one-hour subset is kept beside the aggregate", + usage: `{"input":10,"output":2,"cacheRead":3,"cacheWrite":7,"cacheWrite1h":4,"cost":{"total":0.5}}`, + want: Usage{TokensIn: piRPCTestPtr(10), TokensOut: piRPCTestPtr(2), CacheRead: piRPCTestPtr(3), CacheCreate: piRPCTestPtr(7), CacheCreate1h: piRPCTestPtr(4), CostUSD: piRPCTestPtr(0.5)}, + }, + { + name: "zero one-hour subset is a reported zero", + usage: `{"input":10,"output":2,"cacheRead":0,"cacheWrite":7,"cacheWrite1h":0,"cost":{"total":0.5}}`, + want: Usage{TokensIn: piRPCTestPtr(10), TokensOut: piRPCTestPtr(2), CacheRead: piRPCTestPtr(0), CacheCreate: piRPCTestPtr(7), CacheCreate1h: piRPCTestPtr(0), CostUSD: piRPCTestPtr(0.5)}, + }, + { + name: "absent one-hour subset stays unknown", + usage: `{"input":10,"output":2,"cacheRead":0,"cacheWrite":7,"cost":{"total":0.5}}`, + want: Usage{TokensIn: piRPCTestPtr(10), TokensOut: piRPCTestPtr(2), CacheRead: piRPCTestPtr(0), CacheCreate: piRPCTestPtr(7), CostUSD: piRPCTestPtr(0.5)}, + }, + } { + t.Run(tt.name, func(t *testing.T) { + got := parsePiRPCUsage(map[string]json.RawMessage{"usage": json.RawMessage(tt.usage)}) + if !reflect.DeepEqual(got, tt.want) { + t.Fatalf("parsePiRPCUsage = %s, want %s", formatPiRPCTestUsage(got), formatPiRPCTestUsage(tt.want)) + } + }) + } +} + func TestPiRPCReviewerSettlementPreservesToolEvidence(t *testing.T) { tempDir := t.TempDir() repoDir := filepath.Join(tempDir, "repo") From 6ebf041eb42e7fb3d5a84d29e12a4bf0eb206da6 Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Fri, 2 Oct 2026 03:53:36 -0400 Subject: [PATCH 3/5] test(pi): gate the installed Pi 1.0.0 RPC runtime [#639] Most pi_rpc tests use a fake Pi, so they stay green when Pi's real completion and event contracts drift. Add tests that run the adapter against the installed Pi with an isolated agent directory and home, offline mode, and a scripted loopback OpenAI-compatible provider with a dummy key. Reviewer tool calls go through the real __pi-review-tool helper, which TestMain dispatches the way the cr binary does. The tests cover structured JSON without tools, a 503 followed by a successful retry, permanent provider failures, a multi-turn reviewer run with the diff-first gate, a path escape, and exact usage, cache, and cost totals, a denied shell tool, and caller cancellation and task deadlines with process-group cleanup. Hostile global and project instructions, extensions, MCP servers, and skills must not load. Ordinary go test skips them. make test-pi-runtime sets CR_PI_RUNTIME_VERSION from the Makefile pin, so a missing or different Pi fails. A new non-required pi-runtime CI job installs that version with npm --ignore-scripts on pinned Node 22.22.3 and runs the target. --- .github/workflows/ci.yml | 14 + Makefile | 13 +- docs/development.md | 32 + internal/llmadapters/pi_rpc_runtime_test.go | 840 ++++++++++++++++++ .../llmadapters/pi_rpc_runtime_unix_test.go | 105 +++ 5 files changed, 1003 insertions(+), 1 deletion(-) create mode 100644 internal/llmadapters/pi_rpc_runtime_test.go create mode 100644 internal/llmadapters/pi_rpc_runtime_unix_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d10ae486..b7cdb8b7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -70,6 +70,20 @@ jobs: go-version-file: go.mod target: test-static-smoke + pi-runtime: # installed Pi RPC runtime gate — not a required check + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5.1.0 + - uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 + with: + node-version: 22.22.3 # Pi 1.0 requires Node >= 22.19.0 + # The Makefile owns the Pi version; test-pi-runtime fails if it differs. + - run: npm install -g --ignore-scripts "@earendil-works/pi-coding-agent@$(make -s pi-runtime-version)" + - uses: open-cli-collective/.github/actions/go-test@b6382514b809d960c6fbdc85745a60c14cde04f1 # shared .github + with: + go-version-file: go.mod + target: test-pi-runtime + lint: runs-on: ubuntu-latest steps: diff --git a/Makefile b/Makefile index b572a772..7864e5fc 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: all build test test-cover test-static-smoke lint fmt tidy deps check install snapshot package-render-check release clean +.PHONY: all build test test-cover test-static-smoke test-pi-runtime pi-runtime-version lint fmt tidy deps check install snapshot package-render-check release clean # Standard keyring tags: enable 1Password support, keep passage disabled. GOFLAGS ?= -tags=keyring_nopassage @@ -18,6 +18,17 @@ test-cover: test-static-smoke: go test -v ./internal/... ./cmd/... -count=1 +# Exact Pi version the installed-runtime gate requires. CI installs this +# version; test-pi-runtime fails, rather than skips, when `pi --version` on PATH +# reports anything else or Pi is missing. +PI_RUNTIME_VERSION ?= 1.0.0 + +test-pi-runtime: + CR_PI_RUNTIME_VERSION=$(PI_RUNTIME_VERSION) go test -v -race -count=1 -run 'TestPiRPCRuntime|TestPiRPCReviewerExtensionLoadsInInstalledPi' ./internal/llmadapters + +pi-runtime-version: + @echo $(PI_RUNTIME_VERSION) + lint: golangci-lint run diff --git a/docs/development.md b/docs/development.md index 84a4814c..82cff5fb 100644 --- a/docs/development.md +++ b/docs/development.md @@ -95,10 +95,42 @@ make tidy # go mod tidy and verify go.mod is unchanged make deps # download and verify Go modules make check # tidy + fmt + lint + test + build make clean # remove build artifacts +make test-pi-runtime # installed Pi runtime gate (see Pi Runtime Gate) ``` `make snapshot` runs a local GoReleaser snapshot build without publishing. +## Pi Runtime Gate + +`make test-pi-runtime` runs the `pi_rpc` adapter against the installed Pi on +`PATH` instead of the fake Pi the ordinary adapter tests use. It sets +`CR_PI_RUNTIME_VERSION` to the Makefile's `PI_RUNTIME_VERSION` (currently +`1.0.0`, printed by `make -s pi-runtime-version`). When that variable is set, +the runtime tests fail if Pi is missing or `pi --version` reports any other +version. Without it, as in `make test`, they skip. To install the pinned +runtime: + +```bash +npm install -g --ignore-scripts "@earendil-works/pi-coding-agent@$(make -s pi-runtime-version)" +make test-pi-runtime +``` + +The tests make no paid or external calls: + +- Each test gives Pi a fresh `PI_CODING_AGENT_DIR` and `HOME` with + `PI_OFFLINE=1`. +- The only model is a scripted OpenAI-compatible endpoint on loopback, with a + dummy key. +- Reviewer tool calls run the real `__pi-review-tool` helper, dispatched to the + test binary by `TestMain`. +- Hostile instructions, extensions, MCP servers, and skills are planted at Pi's + default global location and in the reviewer checkout. The tests fail if any + of them reaches a provider request or runs. + +The CI `pi-runtime` job installs the Makefile version on the Node release it +pins and runs the same target. It is not a required check. To move to a new Pi +release, change `PI_RUNTIME_VERSION` and run the target locally first. + ## Repo-Local Shape - Module: `github.com/open-cli-collective/codereview-cli` diff --git a/internal/llmadapters/pi_rpc_runtime_test.go b/internal/llmadapters/pi_rpc_runtime_test.go new file mode 100644 index 00000000..4aabe103 --- /dev/null +++ b/internal/llmadapters/pi_rpc_runtime_test.go @@ -0,0 +1,840 @@ +package llmadapters + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "reflect" + "runtime" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/open-cli-collective/codereview-cli/internal/llm" + "github.com/open-cli-collective/codereview-cli/internal/pireviewtool" +) + +// piRuntimeVersionEnv opts this package into tests that drive the installed Pi +// runtime against a local mock provider. Its value is the exact version that +// `pi --version` must report: once it is set, a missing or different Pi fails +// instead of skipping. `make test-pi-runtime` sets it from the Makefile pin. +const piRuntimeVersionEnv = "CR_PI_RUNTIME_VERSION" + +const ( + piRuntimeProviderName = "crruntime" + piRuntimeModelID = "local" + piRuntimeModel = piRuntimeProviderName + "/" + piRuntimeModelID + piRuntimeAPIKey = "cr-runtime-fixture-key" // #nosec G101 -- dummy key accepted only by the loopback mock provider. + // piRuntimeHostileMarker is planted in global and project Pi resources the + // fixture must never load. It must not reach the provider. + piRuntimeHostileMarker = "CR_PI_RUNTIME_HOSTILE_RESOURCE" + piRuntimeTestTimeout = 60 * time.Second + piRuntimeTaskTimeout = 45 * time.Second +) + +// TestMain lets the generated reviewer extension dispatch its tool helper to +// this test binary the way it dispatches to the cr binary in production, so +// runtime tests exercise the real pireviewtool.Run. +func TestMain(m *testing.M) { + if len(os.Args) > 1 && os.Args[1] == "__pi-review-tool" { + os.Exit(pireviewtool.Run(context.Background(), os.Args[2:], os.Stdin, os.Stdout, os.Stderr)) + } + os.Exit(m.Run()) +} + +func TestPiRPCRuntimeStructuredJSONWithoutTools(t *testing.T) { + fixture := newPiRuntimeFixture(t, + piRuntimeAnswer(`{"ok":true}`, piRuntimeUsage{prompt: 120, completion: 9, cached: 40, cacheWrite: 10}), + ) + + response, err := fixture.run(fixture.adapter(piRuntimeTaskTimeout), Request{Model: piRuntimeModel, Prompt: `Return {"ok":true}.`}) + if err != nil { + t.Fatalf("run: %v", err) + } + if string(response.StructuredOutput) != `{"ok":true}` { + t.Fatalf("StructuredOutput = %q, want the provider answer", response.StructuredOutput) + } + // input = 120 - 40 cached - 10 written; cost = 70*1 + 9*2 + 40*0.5 + 10*2.5. + assertPiRuntimeUsage(t, response.Usage, 70, 9, 40, 10, 133) + if response.ReviewerToolEvidence != nil { + t.Fatalf("ReviewerToolEvidence = %#v, want none for a no-tools call", response.ReviewerToolEvidence) + } + requests := fixture.provider.requests() + if len(requests) != 1 { + t.Fatalf("provider requests = %d, want 1", len(requests)) + } + if names := requests[0].toolNames(); len(names) != 0 { + t.Fatalf("provider request tools = %v, want none", names) + } + if system := requests[0].systemText(); !strings.Contains(system, piRPCSystemPrompt) { + t.Fatalf("provider system prompt = %q, want the CR structured-output prompt", system) + } +} + +func TestPiRPCRuntimeRecoversTransientProviderFailure(t *testing.T) { + fixture := newPiRuntimeFixture(t, + piRuntimeFailure(http.StatusServiceUnavailable, "fixture transient overload"), + piRuntimeAnswer(`{"ok":true}`, piRuntimeUsage{prompt: 100, completion: 10}), + ) + + response, err := fixture.run(fixture.adapter(piRuntimeTaskTimeout), Request{Model: piRuntimeModel, Prompt: `Return {"ok":true}.`}) + if err != nil { + t.Fatalf("run: %v", err) + } + if string(response.StructuredOutput) != `{"ok":true}` { + t.Fatalf("StructuredOutput = %q, want the answer from the retried attempt", response.StructuredOutput) + } + // The failed attempt reports zero usage, so the total is the retry alone. + assertPiRuntimeUsage(t, response.Usage, 100, 10, 0, 0, 120) + if requests := fixture.provider.requests(); len(requests) != 2 { + t.Fatalf("provider requests = %d, want the failed attempt and one retry", len(requests)) + } +} + +func TestPiRPCRuntimePreservesPermanentProviderFailure(t *testing.T) { + for _, tc := range []struct { + name string + steps []piRuntimeStep + requests int + want string + }{ + { + name: "non-retryable rejection", + steps: []piRuntimeStep{piRuntimeFailure(http.StatusBadRequest, "fixture permanent rejection")}, + requests: 1, + want: "fixture permanent rejection", + }, + { + name: "retries exhausted", + steps: []piRuntimeStep{ + piRuntimeFailure(http.StatusServiceUnavailable, "fixture overload one"), + piRuntimeFailure(http.StatusServiceUnavailable, "fixture overload two"), + }, + requests: 2, + want: "fixture overload two", + }, + } { + t.Run(tc.name, func(t *testing.T) { + fixture := newPiRuntimeFixture(t, tc.steps...) + + response, err := fixture.run(fixture.adapter(piRuntimeTaskTimeout), Request{Model: piRuntimeModel, Prompt: `Return {"ok":true}.`}) + if err == nil { + t.Fatalf("run succeeded with %q, want the provider failure", response.StructuredOutput) + } + if !strings.Contains(err.Error(), "final assistant message error") || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("run error = %v, want final provider diagnostic %q", err, tc.want) + } + if len(response.StructuredOutput) != 0 { + t.Fatalf("StructuredOutput = %q, want none on failure", response.StructuredOutput) + } + if requests := fixture.provider.requests(); len(requests) != tc.requests { + t.Fatalf("provider requests = %d, want %d", len(requests), tc.requests) + } + }) + } +} + +func TestPiRPCRuntimeReviewerToolsAndUsage(t *testing.T) { + fixture := newPiRuntimeFixture(t, + piRuntimeToolCalls(piRuntimeUsage{prompt: 100, completion: 5}, + piRuntimeToolCall{id: "read-early", name: "cr_read", args: `{"path":"main.go"}`}, + ), + piRuntimeToolCalls(piRuntimeUsage{prompt: 150, completion: 7, cached: 60, cacheWrite: 20}, + piRuntimeToolCall{id: "diff", name: "cr_diff", args: `{}`}, + ), + piRuntimeToolCalls(piRuntimeUsage{prompt: 200, completion: 11, cached: 120}, + piRuntimeToolCall{id: "read", name: "cr_read", args: `{"path":"main.go"}`}, + piRuntimeToolCall{id: "search", name: "cr_search", args: `{"query":"func Hello"}`}, + piRuntimeToolCall{id: "list", name: "cr_list", args: `{}`}, + piRuntimeToolCall{id: "escape", name: "cr_read", args: `{"path":"../outside.txt"}`}, + ), + piRuntimeAnswer(`{"findings":[]}`, piRuntimeUsage{prompt: 260, completion: 13, cached: 180, cacheWrite: 30}), + ) + workspace := fixture.reviewerWorkspace() + + response, err := fixture.run(fixture.adapter(piRuntimeTaskTimeout), Request{ + Model: piRuntimeModel, + Prompt: "Review the pinned change.", + ReviewerWorkspace: workspace, + }) + if err != nil { + t.Fatalf("run: %v", err) + } + if string(response.StructuredOutput) != `{"findings":[]}` { + t.Fatalf("StructuredOutput = %q, want the final answer after every tool turn", response.StructuredOutput) + } + if response.ReviewerToolEvidence == nil || response.ReviewerToolEvidence.DiffStatus != llm.DiffToolStatusSucceeded { + t.Fatalf("ReviewerToolEvidence = %#v, want a succeeded cr_diff", response.ReviewerToolEvidence) + } + // Every assistant turn counts once: input 100+70+80+50, output 5+7+11+13, + // cache read 0+60+120+180, cache write 0+20+0+30, cost 300 + 36*2 + + // 360*0.5 + 50*2.5. + assertPiRuntimeUsage(t, response.Usage, 300, 36, 360, 50, 677) + + requests := fixture.provider.requests() + if len(requests) != 4 { + t.Fatalf("provider requests = %d, want one per assistant turn", len(requests)) + } + wantTools := []string{"cr_diff", "cr_list", "cr_read", "cr_search"} + for i, request := range requests { + if got := request.toolNames(); !reflect.DeepEqual(got, wantTools) { + t.Fatalf("request %d tools = %v, want only %v", i, got, wantTools) + } + if system := request.systemText(); !strings.Contains(system, piRPCReviewerSystemPrompt) { + t.Fatalf("request %d system prompt = %q, want the CR reviewer prompt", i, system) + } + } + assertPiRuntimeToolResult(t, requests[1], "read-early", "cr_diff must be invoked before inspecting repository files") + assertPiRuntimeToolResult(t, requests[2], "diff", "+func Hello() string") + assertPiRuntimeToolResult(t, requests[3], "read", `return "hello from the head revision"`) + assertPiRuntimeToolResult(t, requests[3], "search", "main.go") + assertPiRuntimeToolResult(t, requests[3], "list", "main.go\n") + assertPiRuntimeToolResult(t, requests[3], "escape", "path traversal") + if strings.Contains(string(requests[3].raw), "outside the repository") { + t.Fatal("cr_read returned content from outside the repository root") + } + assertPiRuntimeDirEmpty(t, workspace.ScratchDir) +} + +func TestPiRPCRuntimeReviewerRejectsUnlistedTool(t *testing.T) { + shellMarker := filepath.Join(t.TempDir(), "shell-ran") + fixture := newPiRuntimeFixture(t, + piRuntimeToolCalls(piRuntimeUsage{prompt: 100, completion: 5}, + piRuntimeToolCall{id: "diff", name: "cr_diff", args: `{}`}, + ), + piRuntimeToolCalls(piRuntimeUsage{prompt: 150, completion: 5}, + piRuntimeToolCall{id: "shell", name: "bash", args: mustPiRuntimeJSON(t, map[string]string{"command": "touch " + shellMarker})}, + ), + piRuntimeAnswer(`{"findings":[]}`, piRuntimeUsage{prompt: 200, completion: 5}), + ) + workspace := fixture.reviewerWorkspace() + + response, err := fixture.run(fixture.adapter(piRuntimeTaskTimeout), Request{ + Model: piRuntimeModel, + Prompt: "Review the pinned change.", + ReviewerWorkspace: workspace, + }) + if !errors.Is(err, ErrToolUse) { + t.Fatalf("run error = %v (output %q), want ErrToolUse for an unlisted tool", err, response.StructuredOutput) + } + if len(response.StructuredOutput) != 0 { + t.Fatalf("StructuredOutput = %q, want none after a denied tool", response.StructuredOutput) + } + if _, statErr := os.Stat(shellMarker); !errors.Is(statErr, os.ErrNotExist) { + t.Fatalf("shell marker stat = %v, want the bash request never executed", statErr) + } + if requests := fixture.provider.requests(); len(requests) != 2 { + t.Fatalf("provider requests = %d, want the run stopped at the denied tool call", len(requests)) + } + assertPiRuntimeDirEmpty(t, workspace.ScratchDir) +} + +// requirePiRuntime returns the installed Pi path when runtime tests are +// enabled. It skips only when the opt-in is unset. +func requirePiRuntime(t *testing.T) string { + t.Helper() + want := os.Getenv(piRuntimeVersionEnv) + if want == "" { + t.Skipf("set %s to the required Pi version (make test-pi-runtime) to run installed Pi runtime tests", piRuntimeVersionEnv) + } + install := installedPiRuntime() + if install.err != nil { + t.Fatalf("%s=%s requires the installed Pi runtime: %v", piRuntimeVersionEnv, want, install.err) + } + if install.version != want { + t.Fatalf("installed Pi %s reports version %q, want exactly %q", install.path, install.version, want) + } + return install.path +} + +type piRuntimeInstall struct { + path string + version string + err error +} + +var installedPiRuntime = sync.OnceValue(func() piRuntimeInstall { + path, err := exec.LookPath("pi") + if err != nil { + return piRuntimeInstall{err: err} + } + agentDir, err := os.MkdirTemp("", "cr-pi-runtime-version-*") + if err != nil { + return piRuntimeInstall{path: path, err: err} + } + defer func() { _ = os.RemoveAll(agentDir) }() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, path, "--version") // #nosec G204 -- test runs the discovered Pi executable with a fixed argument. + cmd.Env = append(os.Environ(), piRuntimeIsolationEnv(agentDir, agentDir)...) + output, err := cmd.Output() + if err != nil { + return piRuntimeInstall{path: path, err: fmt.Errorf("pi --version: %w", err)} + } + return piRuntimeInstall{path: path, version: strings.TrimSpace(string(output))} +}) + +// piRuntimeGoroutines counts live goroutines, leaving out LaunchProcess's +// pipe-close grace, which outlives a finished run by a bounded delay by design. +func piRuntimeGoroutines() (int, string) { + buf := make([]byte, 1<<20) + for { + n := runtime.Stack(buf, true) + if n < len(buf) { + buf = buf[:n] + break + } + buf = make([]byte, 2*len(buf)) + } + var kept []string + for _, stack := range strings.Split(string(buf), "\n\n") { + if strings.Contains(stack, "internal/llm.LaunchProcess.func") && strings.Contains(stack, "time.Sleep") { + continue + } + kept = append(kept, stack) + } + return len(kept), strings.Join(kept, "\n\n") +} + +// piRuntimeIsolationEnv points Pi at a fixture-owned agent directory and home +// and turns off automatic network activity other than the model request. +func piRuntimeIsolationEnv(agentDir, home string) []string { + return []string{ + "PI_CODING_AGENT_DIR=" + agentDir, + "PI_CODING_AGENT_SESSION_DIR=" + filepath.Join(agentDir, "sessions"), + "PI_OFFLINE=1", + "PI_SKIP_VERSION_CHECK=1", + "PI_TELEMETRY=0", + "HOME=" + home, + "USERPROFILE=" + home, + } +} + +type piRuntimeFixture struct { + t *testing.T + piPath string + // command launches Pi; tests may point it at a wrapper of piPath. + command string + root string + agentDir string + env []string + provider *piRuntimeProvider + // sideEffects are files that hostile global or project resources would + // create if Pi loaded them. + sideEffects []string +} + +func newPiRuntimeFixture(t *testing.T, steps ...piRuntimeStep) *piRuntimeFixture { + t.Helper() + piPath := requirePiRuntime(t) + goroutines, _ := piRuntimeGoroutines() + t.Cleanup(func() { + deadline := time.Now().Add(5 * time.Second) + for { + got, stacks := piRuntimeGoroutines() + if got <= goroutines { + return + } + if time.Now().After(deadline) { + t.Errorf("goroutines = %d after the runtime test, want at most %d:\n%s", got, goroutines, stacks) + return + } + time.Sleep(20 * time.Millisecond) + } + }) + + root := t.TempDir() + fixture := &piRuntimeFixture{ + t: t, + piPath: piPath, + command: piPath, + root: root, + agentDir: filepath.Join(root, "agent"), + provider: newPiRuntimeProvider(t, steps), + } + home := filepath.Join(root, "home") + fixture.env = piRuntimeIsolationEnv(fixture.agentDir, home) + fixture.writeJSON(filepath.Join(fixture.agentDir, "models.json"), map[string]any{ + "providers": map[string]any{ + piRuntimeProviderName: map[string]any{ + "baseUrl": fixture.provider.server.URL + "/v1", + "api": "openai-completions", + "apiKey": piRuntimeAPIKey, + "models": []any{map[string]any{ + "id": piRuntimeModelID, + "name": "CR runtime fixture", + "reasoning": false, + "input": []string{"text"}, + "contextWindow": 100000, + "maxTokens": 1000, + // Rates per million tokens that make each token cost an + // exactly representable amount: 1, 2, 0.5, and 2.5. + "cost": map[string]any{"input": 1000000, "output": 2000000, "cacheRead": 500000, "cacheWrite": 2500000}, + }}, + }, + }, + }) + fixture.writeJSON(filepath.Join(fixture.agentDir, "settings.json"), map[string]any{ + "retry": map[string]any{"enabled": true, "maxRetries": 1, "baseDelayMs": 10}, + "enableInstallTelemetry": false, + }) + // Resources at Pi's default global location. The agent directory override + // must keep every one of them out of the run. + fixture.plantHostileResources(filepath.Join(home, ".pi", "agent"), "global") + fixture.writeFile(filepath.Join(home, ".pi", "agent", "settings.json"), `{"retry":{"enabled":false},"defaultProjectTrust":"always"}`) + fixture.writeFile(filepath.Join(home, ".agents", "skills", "hostile", "SKILL.md"), "---\nname: hostile\ndescription: "+piRuntimeHostileMarker+"\n---\n"+piRuntimeHostileMarker+"\n") + + t.Cleanup(fixture.assertIsolated) + return fixture +} + +// plantHostileResources writes instructions, an extension, and an MCP server +// under dir. Loading any of them leaves the marker in a provider request or +// creates a side-effect file. +func (f *piRuntimeFixture) plantHostileResources(dir, label string) { + f.t.Helper() + extensionMarker := filepath.Join(f.root, label+"-extension-loaded") + mcpMarker := filepath.Join(f.root, label+"-mcp-started") + f.sideEffects = append(f.sideEffects, extensionMarker, mcpMarker) + for _, name := range []string{"AGENTS.md", "CLAUDE.md", "SYSTEM.md", "APPEND_SYSTEM.md"} { + f.writeFile(filepath.Join(dir, name), piRuntimeHostileMarker+" "+label+" "+name+"\n") + } + f.writeFile(filepath.Join(dir, "extensions", "hostile.js"), + "import fs from \"node:fs\";\nexport default function () { fs.writeFileSync("+mustPiRuntimeJSON(f.t, extensionMarker)+", \"loaded\"); }\n") + f.writeJSON(filepath.Join(dir, "mcp.json"), map[string]any{ + "mcpServers": map[string]any{ + "hostile": map[string]any{ + "command": "node", + "args": []string{"-e", "require('node:fs').writeFileSync(" + mustPiRuntimeJSON(f.t, mcpMarker) + ", 'started')"}, + "description": piRuntimeHostileMarker, + }, + }, + }) +} + +// reviewerWorkspace builds a disposable checkout that also carries hostile +// project-level Pi resources, as an untrusted pull request could. +func (f *piRuntimeFixture) reviewerWorkspace() *ReviewerWorkspaceRequest { + f.t.Helper() + workspaceRoot := filepath.Join(f.root, "workspace") + repoDir := filepath.Join(workspaceRoot, "repo") + scratchDir := filepath.Join(workspaceRoot, "scratch") + f.writeFile(filepath.Join(repoDir, "main.go"), "package main\n\nfunc Hello() string {\n\treturn \"hello from the head revision\"\n}\n") + f.writeFile(filepath.Join(workspaceRoot, "outside.txt"), "outside the repository\n") + f.plantHostileResources(filepath.Join(repoDir, ".pi"), "project") + for _, name := range []string{"AGENTS.md", "CLAUDE.md"} { + f.writeFile(filepath.Join(repoDir, name), piRuntimeHostileMarker+" project root "+name+"\n") + } + f.writeFile(filepath.Join(repoDir, ".agents", "skills", "hostile", "SKILL.md"), "---\nname: hostile\ndescription: "+piRuntimeHostileMarker+"\n---\n"+piRuntimeHostileMarker+"\n") + if err := os.MkdirAll(scratchDir, 0o700); err != nil { + f.t.Fatalf("MkdirAll(scratch): %v", err) + } + diffPath := filepath.Join(workspaceRoot, "diff.patch") + f.writeFile(diffPath, "diff --git a/main.go b/main.go\n--- a/main.go\n+++ b/main.go\n@@ -1,2 +1,5 @@\n package main\n+\n+func Hello() string {\n+\treturn \"hello from the head revision\"\n+}\n") + return &ReviewerWorkspaceRequest{ + RepoDir: repoDir, + ScratchDir: scratchDir, + DiffPath: diffPath, + MaxToolOutputBytes: 64 * 1024, + } +} + +func (f *piRuntimeFixture) adapter(timeout time.Duration) *PiRPCAdapter { + return NewPiRPCAdapter(PiRPCOptions{ + Command: f.command, + Env: f.env, + Timeout: timeout, + ScratchDirFactory: func() (string, func() error, error) { + dir, err := os.MkdirTemp(f.root, "scratch-*") + if err != nil { + return "", nil, err + } + return dir, func() error { return os.RemoveAll(dir) }, nil + }, + }) +} + +func (f *piRuntimeFixture) run(adapter *PiRPCAdapter, req Request) (Response, error) { + f.t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), piRuntimeTestTimeout) + defer cancel() + stream, err := adapter.Start(ctx, req) + if err != nil { + return Response{}, err + } + return stream.Wait(ctx) +} + +func (f *piRuntimeFixture) assertIsolated() { + t := f.t + t.Helper() + for i, request := range f.provider.requests() { + if request.authorization != "Bearer "+piRuntimeAPIKey { + t.Errorf("request %d authorization is not the fixture key", i) + } + if bytes.Contains(request.raw, []byte(piRuntimeHostileMarker)) { + t.Errorf("request %d carries a global or project resource the fixture must not load", i) + } + } + for _, unexpected := range f.provider.unexpectedRequests() { + t.Errorf("unexpected provider request: %s", unexpected) + } + for _, path := range f.sideEffects { + if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) { + t.Errorf("hostile resource side effect %s exists (stat err %v)", filepath.Base(path), err) + } + } + entries, err := os.ReadDir(f.root) + if err != nil { + t.Errorf("ReadDir(fixture root): %v", err) + return + } + for _, entry := range entries { + if strings.HasPrefix(entry.Name(), "scratch-") { + t.Errorf("adapter scratch %s was not removed", entry.Name()) + } + } +} + +func (f *piRuntimeFixture) writeJSON(path string, value any) { + f.t.Helper() + data, err := json.MarshalIndent(value, "", " ") + if err != nil { + f.t.Fatalf("Marshal(%s): %v", filepath.Base(path), err) + } + f.writeFile(path, string(data)+"\n") +} + +func (f *piRuntimeFixture) writeFile(path, content string) { + f.t.Helper() + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + f.t.Fatalf("MkdirAll(%s): %v", filepath.Dir(path), err) + } + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { // #nosec G703 -- path is rooted in the fixture's t.TempDir. + f.t.Fatalf("WriteFile(%s): %v", path, err) + } +} + +// piRuntimeStep answers one provider request. +type piRuntimeStep func(w http.ResponseWriter, r *http.Request, stop <-chan struct{}) + +type piRuntimeUsage struct { + prompt int + completion int + cached int + cacheWrite int +} + +func (u piRuntimeUsage) payload() map[string]any { + return map[string]any{ + "prompt_tokens": u.prompt, + "completion_tokens": u.completion, + "total_tokens": u.prompt + u.completion, + "prompt_tokens_details": map[string]any{ + "cached_tokens": u.cached, + "cache_write_tokens": u.cacheWrite, + }, + } +} + +type piRuntimeToolCall struct { + id string + name string + args string +} + +func piRuntimeAnswer(text string, usage piRuntimeUsage) piRuntimeStep { + return func(w http.ResponseWriter, _ *http.Request, _ <-chan struct{}) { + writePiRuntimeSSE(w, + piRuntimeChunk(map[string]any{"role": "assistant", "content": text}, nil, nil), + piRuntimeChunk(map[string]any{}, "stop", &usage), + ) + } +} + +func piRuntimeToolCalls(usage piRuntimeUsage, calls ...piRuntimeToolCall) piRuntimeStep { + toolCalls := make([]any, 0, len(calls)) + for i, call := range calls { + toolCalls = append(toolCalls, map[string]any{ + "index": i, + "id": call.id, + "type": "function", + "function": map[string]any{"name": call.name, "arguments": call.args}, + }) + } + return func(w http.ResponseWriter, _ *http.Request, _ <-chan struct{}) { + writePiRuntimeSSE(w, + piRuntimeChunk(map[string]any{"role": "assistant", "tool_calls": toolCalls}, nil, nil), + piRuntimeChunk(map[string]any{}, "tool_calls", &usage), + ) + } +} + +// piRuntimeFailure returns an OpenAI-style error. Pi retries by matching the +// error text, so only 5xx responses carry the retryable server_error type. +func piRuntimeFailure(status int, message string) piRuntimeStep { + errorType := "invalid_request_error" + if status >= http.StatusInternalServerError { + errorType = "server_error" + } + return func(w http.ResponseWriter, _ *http.Request, _ <-chan struct{}) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(map[string]any{"error": map[string]any{"message": message, "type": errorType}}) + } +} + +// piRuntimeStall starts a stream and then holds the request open until the +// client disconnects or the fixture shuts down. started is closed once the +// stream is open; disconnected is closed if the client goes away first. +func piRuntimeStall(started, disconnected chan<- struct{}) piRuntimeStep { + return func(w http.ResponseWriter, r *http.Request, stop <-chan struct{}) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + data, _ := json.Marshal(piRuntimeChunk(map[string]any{"role": "assistant", "content": ""}, nil, nil)) + _, _ = fmt.Fprintf(w, "data: %s\n\n", data) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + close(started) + select { + case <-r.Context().Done(): + close(disconnected) + case <-stop: + } + } +} + +func piRuntimeChunk(delta map[string]any, finishReason any, usage *piRuntimeUsage) map[string]any { + chunk := map[string]any{ + "id": "cr-runtime-completion", + "object": "chat.completion.chunk", + "created": 0, + "model": piRuntimeModelID, + "choices": []any{map[string]any{"index": 0, "delta": delta, "finish_reason": finishReason}}, + } + if usage != nil { + chunk["usage"] = usage.payload() + } + return chunk +} + +func writePiRuntimeSSE(w http.ResponseWriter, chunks ...map[string]any) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + for _, chunk := range chunks { + data, _ := json.Marshal(chunk) + _, _ = fmt.Fprintf(w, "data: %s\n\n", data) + } + _, _ = io.WriteString(w, "data: [DONE]\n\n") +} + +// piRuntimeProvider is a scripted OpenAI-compatible chat completions endpoint +// bound to loopback. Request n gets step n; any extra request is rejected with +// a non-retryable error and reported by the fixture. +type piRuntimeProvider struct { + server *httptest.Server + stop chan struct{} + mu sync.Mutex + steps []piRuntimeStep + recorded []piRuntimeRequest + unexpected []string +} + +func newPiRuntimeProvider(t *testing.T, steps []piRuntimeStep) *piRuntimeProvider { + t.Helper() + provider := &piRuntimeProvider{stop: make(chan struct{}), steps: steps} + provider.server = httptest.NewServer(provider) + t.Cleanup(provider.server.Close) + // Release stalled handlers before Close waits for them. + t.Cleanup(func() { close(provider.stop) }) + host, _, err := net.SplitHostPort(provider.server.Listener.Addr().String()) + if err != nil || !net.ParseIP(host).IsLoopback() { + t.Fatalf("mock provider address %s is not loopback (%v)", provider.server.Listener.Addr(), err) + } + return provider +} + +func (p *piRuntimeProvider) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/v1/chat/completions" { + p.recordUnexpected(r.Method + " " + r.URL.Path) + http.Error(w, "unexpected endpoint", http.StatusNotFound) + return + } + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 8<<20)) + if err != nil { + p.recordUnexpected("unreadable request body: " + err.Error()) + http.Error(w, "unreadable body", http.StatusBadRequest) + return + } + request := piRuntimeRequest{authorization: r.Header.Get("Authorization"), raw: body} + if err := json.Unmarshal(body, &request.body); err != nil { + p.recordUnexpected("malformed request body: " + err.Error()) + } + p.mu.Lock() + index := len(p.recorded) + p.recorded = append(p.recorded, request) + var step piRuntimeStep + if index < len(p.steps) { + step = p.steps[index] + } + p.mu.Unlock() + if step == nil { + p.recordUnexpected(fmt.Sprintf("request %d beyond the %d scripted steps", index, len(p.steps))) + piRuntimeFailure(http.StatusBadRequest, "unexpected extra provider request")(w, r, p.stop) + return + } + step(w, r, p.stop) +} + +func (p *piRuntimeProvider) recordUnexpected(message string) { + p.mu.Lock() + defer p.mu.Unlock() + p.unexpected = append(p.unexpected, message) +} + +func (p *piRuntimeProvider) requests() []piRuntimeRequest { + p.mu.Lock() + defer p.mu.Unlock() + return append([]piRuntimeRequest(nil), p.recorded...) +} + +func (p *piRuntimeProvider) unexpectedRequests() []string { + p.mu.Lock() + defer p.mu.Unlock() + return append([]string(nil), p.unexpected...) +} + +type piRuntimeRequest struct { + authorization string + raw []byte + body struct { + Messages []struct { + Role string `json:"role"` + Content json.RawMessage `json:"content"` + ToolCallID string `json:"tool_call_id"` + } `json:"messages"` + Tools []struct { + Function struct { + Name string `json:"name"` + } `json:"function"` + } `json:"tools"` + } +} + +func (r piRuntimeRequest) toolNames() []string { + names := make([]string, 0, len(r.body.Tools)) + for _, tool := range r.body.Tools { + names = append(names, tool.Function.Name) + } + slices.Sort(names) + return names +} + +func (r piRuntimeRequest) systemText() string { + var out strings.Builder + for _, message := range r.body.Messages { + if message.Role == "system" || message.Role == "developer" { + out.WriteString(piRuntimeContentText(message.Content)) + } + } + return out.String() +} + +func (r piRuntimeRequest) toolResult(id string) (string, bool) { + for _, message := range r.body.Messages { + if message.Role == "tool" && message.ToolCallID == id { + return piRuntimeContentText(message.Content), true + } + } + return "", false +} + +func piRuntimeContentText(content json.RawMessage) string { + var text string + if err := json.Unmarshal(content, &text); err == nil { + return text + } + var parts []struct { + Type string `json:"type"` + Text string `json:"text"` + } + if err := json.Unmarshal(content, &parts); err != nil { + return "" + } + var out strings.Builder + for _, part := range parts { + if part.Type == "text" { + out.WriteString(part.Text) + } + } + return out.String() +} + +func assertPiRuntimeToolResult(t *testing.T, request piRuntimeRequest, id, want string) { + t.Helper() + got, ok := request.toolResult(id) + if !ok { + t.Fatalf("provider request has no tool result for %s", id) + } + if !strings.Contains(got, want) { + t.Fatalf("tool result %s = %q, want it to contain %q", id, got, want) + } +} + +func assertPiRuntimeUsage(t *testing.T, usage Usage, tokensIn, tokensOut, cacheRead, cacheCreate int, costUSD float64) { + t.Helper() + got := fmt.Sprintf("in=%s out=%s cacheRead=%s cacheCreate=%s cacheCreate1h=%s cost=%s", + piRuntimeIntField(usage.TokensIn), piRuntimeIntField(usage.TokensOut), piRuntimeIntField(usage.CacheRead), + piRuntimeIntField(usage.CacheCreate), piRuntimeIntField(usage.CacheCreate1h), piRuntimeFloatField(usage.CostUSD)) + want := fmt.Sprintf("in=%d out=%d cacheRead=%d cacheCreate=%d cacheCreate1h=nil cost=%g", + tokensIn, tokensOut, cacheRead, cacheCreate, costUSD) + if got != want { + t.Fatalf("usage = %s, want %s", got, want) + } +} + +func piRuntimeIntField(value *int) string { + if value == nil { + return "nil" + } + return fmt.Sprint(*value) +} + +func piRuntimeFloatField(value *float64) string { + if value == nil { + return "nil" + } + return fmt.Sprintf("%g", *value) +} + +func assertPiRuntimeDirEmpty(t *testing.T, dir string) { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("ReadDir(%s): %v", dir, err) + } + if len(entries) != 0 { + t.Fatalf("%s still holds %d entries after the run, want adapter cleanup", dir, len(entries)) + } +} + +func mustPiRuntimeJSON(t *testing.T, value any) string { + t.Helper() + data, err := json.Marshal(value) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + return string(data) +} diff --git a/internal/llmadapters/pi_rpc_runtime_unix_test.go b/internal/llmadapters/pi_rpc_runtime_unix_test.go new file mode 100644 index 00000000..c592f8fa --- /dev/null +++ b/internal/llmadapters/pi_rpc_runtime_unix_test.go @@ -0,0 +1,105 @@ +//go:build unix + +package llmadapters + +import ( + "context" + "errors" + "os" + "path/filepath" + "strconv" + "strings" + "syscall" + "testing" + "time" +) + +// Process-group assertions are POSIX-specific; the CI runtime gate runs on +// Linux. +func TestPiRPCRuntimeCancellationStopsPi(t *testing.T) { + for _, tc := range []struct { + name string + timeout time.Duration + cancel bool + want error + }{ + {name: "caller cancellation", timeout: piRuntimeTaskTimeout, cancel: true, want: context.Canceled}, + {name: "task deadline", timeout: 3 * time.Second, want: context.DeadlineExceeded}, + } { + t.Run(tc.name, func(t *testing.T) { + started := make(chan struct{}) + disconnected := make(chan struct{}) + fixture := newPiRuntimeFixture(t, piRuntimeStall(started, disconnected)) + // exec keeps the wrapper's PID, which is also the process group the + // adapter starts Pi in. + pidPath := filepath.Join(fixture.root, "pi.pid") + wrapperPath := filepath.Join(fixture.root, "pi-wrapper.sh") + fixture.writeFile(wrapperPath, "#!/bin/sh\necho $$ > "+strconv.Quote(pidPath)+"\nexec "+strconv.Quote(fixture.piPath)+" \"$@\"\n") + if err := os.Chmod(wrapperPath, 0o700); err != nil { // #nosec G302 -- test wrapper must be executable and is rooted in t.TempDir. + t.Fatalf("Chmod(wrapper): %v", err) + } + fixture.command = wrapperPath + + ctx, cancel := context.WithTimeout(context.Background(), piRuntimeTestTimeout) + defer cancel() + stream, err := fixture.adapter(tc.timeout).Start(ctx, Request{Model: piRuntimeModel, Prompt: `Return {"ok":true}.`}) + if err != nil { + t.Fatalf("Start: %v", err) + } + select { + case <-started: + case <-time.After(30 * time.Second): + cancel() + _, _ = stream.Wait(context.Background()) + t.Fatal("Pi never reached the mock provider") + } + pid := readPiRuntimePID(t, pidPath) + + began := time.Now() + if tc.cancel { + cancel() + } + response, err := stream.Wait(ctx) + elapsed := time.Since(began) + if !errors.Is(err, tc.want) { + t.Fatalf("Wait error = %v, want %v", err, tc.want) + } + if len(response.StructuredOutput) != 0 { + t.Fatalf("StructuredOutput = %q, want none after %s", response.StructuredOutput, tc.name) + } + limit := 10 * time.Second + if !tc.cancel { + limit += tc.timeout + } + if elapsed > limit { + t.Fatalf("Wait returned after %s, want termination within %s", elapsed, limit) + } + select { + case <-disconnected: + case <-time.After(5 * time.Second): + t.Fatal("Pi kept the provider request open after the run ended") + } + eventually(t, 5*time.Second, func() bool { + return errors.Is(syscall.Kill(-pid, 0), syscall.ESRCH) + }) + if requests := fixture.provider.requests(); len(requests) != 1 { + t.Fatalf("provider requests = %d, want the stalled request only", len(requests)) + } + }) + } +} + +func readPiRuntimePID(t *testing.T, path string) int { + t.Helper() + var data []byte + eventually(t, 5*time.Second, func() bool { + var err error + data, err = os.ReadFile(path) // #nosec G304 -- path is rooted in the fixture's t.TempDir. + return err == nil && strings.TrimSpace(string(data)) != "" + }) + pid, err := strconv.Atoi(strings.TrimSpace(string(data))) + if err != nil || pid <= 1 { + t.Fatalf("Pi PID file = %q: %v", data, err) + } + return pid +} From e9d463ee16231f1a7d9759bf6ef1ddf3de87c844 Mon Sep 17 00:00:00 2001 From: Aaron Wong <6979793+zzwong@users.noreply.github.com> Date: Fri, 2 Oct 2026 04:24:11 -0400 Subject: [PATCH 4/5] test(pi): reject a blank Pi runtime pin [#639] A blank PI_RUNTIME_VERSION made make test-pi-runtime export an empty CR_PI_RUNTIME_VERSION, so every runtime test skipped and the target reported ok. The same value made pi-runtime-version print nothing, which the CI install step would turn into an unpinned, latest Pi install. Both Make targets now require exactly one version word before doing anything. The runtime tests also fail, rather than skip, when the opt-in variable is set but blank. The CI step reads the version in its own assignment so a failed lookup stops the step before npm runs. A focused test runs both targets against a fake go and checks that blank, whitespace, and multi-word pins fail without dispatching go test. The development guide now states that the runtime fixture's isolation does not extend to production agent directories (#642). --- .github/workflows/ci.yml | 6 +- Makefile | 18 ++++-- docs/development.md | 24 ++++--- internal/llmadapters/pi_rpc_runtime_test.go | 10 ++- .../llmadapters/pi_rpc_runtime_unix_test.go | 63 +++++++++++++++++++ 5 files changed, 104 insertions(+), 17 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b7cdb8b7..11c36518 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -78,7 +78,11 @@ jobs: with: node-version: 22.22.3 # Pi 1.0 requires Node >= 22.19.0 # The Makefile owns the Pi version; test-pi-runtime fails if it differs. - - run: npm install -g --ignore-scripts "@earendil-works/pi-coding-agent@$(make -s pi-runtime-version)" + # Read it first so a failed lookup stops the step instead of installing + # an unpinned (latest) Pi. + - run: | + pi_version="$(make -s pi-runtime-version)" + npm install -g --ignore-scripts "@earendil-works/pi-coding-agent@${pi_version}" - uses: open-cli-collective/.github/actions/go-test@b6382514b809d960c6fbdc85745a60c14cde04f1 # shared .github with: go-version-file: go.mod diff --git a/Makefile b/Makefile index 7864e5fc..97f52efa 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: all build test test-cover test-static-smoke test-pi-runtime pi-runtime-version lint fmt tidy deps check install snapshot package-render-check release clean +.PHONY: all build test test-cover test-static-smoke test-pi-runtime pi-runtime-version check-pi-runtime-version lint fmt tidy deps check install snapshot package-render-check release clean # Standard keyring tags: enable 1Password support, keep passage disabled. GOFLAGS ?= -tags=keyring_nopassage @@ -23,11 +23,19 @@ test-static-smoke: # reports anything else or Pi is missing. PI_RUNTIME_VERSION ?= 1.0.0 -test-pi-runtime: - CR_PI_RUNTIME_VERSION=$(PI_RUNTIME_VERSION) go test -v -race -count=1 -run 'TestPiRPCRuntime|TestPiRPCReviewerExtensionLoadsInInstalledPi' ./internal/llmadapters +# A blank pin would make the runtime tests skip and let npm install the latest +# Pi, so both targets below require exactly one version. +check-pi-runtime-version: + @if [ "$(words $(PI_RUNTIME_VERSION))" != 1 ]; then \ + echo "PI_RUNTIME_VERSION must be exactly one Pi version" >&2; \ + exit 1; \ + fi + +test-pi-runtime: check-pi-runtime-version + CR_PI_RUNTIME_VERSION=$(strip $(PI_RUNTIME_VERSION)) go test -v -race -count=1 -run 'TestPiRPCRuntime|TestPiRPCReviewerExtensionLoadsInInstalledPi' ./internal/llmadapters -pi-runtime-version: - @echo $(PI_RUNTIME_VERSION) +pi-runtime-version: check-pi-runtime-version + @echo $(strip $(PI_RUNTIME_VERSION)) lint: golangci-lint run diff --git a/docs/development.md b/docs/development.md index 82cff5fb..8581b7f7 100644 --- a/docs/development.md +++ b/docs/development.md @@ -105,13 +105,15 @@ make test-pi-runtime # installed Pi runtime gate (see Pi Runtime Gate) `make test-pi-runtime` runs the `pi_rpc` adapter against the installed Pi on `PATH` instead of the fake Pi the ordinary adapter tests use. It sets `CR_PI_RUNTIME_VERSION` to the Makefile's `PI_RUNTIME_VERSION` (currently -`1.0.0`, printed by `make -s pi-runtime-version`). When that variable is set, -the runtime tests fail if Pi is missing or `pi --version` reports any other -version. Without it, as in `make test`, they skip. To install the pinned -runtime: +`1.0.0`, printed by `make -s pi-runtime-version`). Both targets fail when +`PI_RUNTIME_VERSION` is blank or holds more than one version. When +`CR_PI_RUNTIME_VERSION` is set, the runtime tests fail if it is blank, if Pi is +missing, or if `pi --version` reports any other version. When it is unset, as in +`make test`, they skip. To install the pinned runtime: ```bash -npm install -g --ignore-scripts "@earendil-works/pi-coding-agent@$(make -s pi-runtime-version)" +pi_version="$(make -s pi-runtime-version)" +npm install -g --ignore-scripts "@earendil-works/pi-coding-agent@${pi_version}" make test-pi-runtime ``` @@ -123,9 +125,15 @@ The tests make no paid or external calls: dummy key. - Reviewer tool calls run the real `__pi-review-tool` helper, dispatched to the test binary by `TestMain`. -- Hostile instructions, extensions, MCP servers, and skills are planted at Pi's - default global location and in the reviewer checkout. The tests fail if any - of them reaches a provider request or runs. +- Hostile instructions, extensions, MCP servers, and skills are planted under + the fixture `HOME`'s default `.pi/agent` location and in the reviewer + checkout, while Pi reads the fresh agent directory. The tests fail if any of + them reaches a provider request or runs. + +This shows that the fixture is isolated, not that production runs are. CR does +not replace the developer's own Pi agent directory, so its global instructions +can still reach prompts +([#642](https://github.com/open-cli-collective/codereview-cli/issues/642)). The CI `pi-runtime` job installs the Makefile version on the Node release it pins and runs the same target. It is not a required check. To move to a new Pi diff --git a/internal/llmadapters/pi_rpc_runtime_test.go b/internal/llmadapters/pi_rpc_runtime_test.go index 4aabe103..e9045418 100644 --- a/internal/llmadapters/pi_rpc_runtime_test.go +++ b/internal/llmadapters/pi_rpc_runtime_test.go @@ -241,13 +241,17 @@ func TestPiRPCRuntimeReviewerRejectsUnlistedTool(t *testing.T) { } // requirePiRuntime returns the installed Pi path when runtime tests are -// enabled. It skips only when the opt-in is unset. +// enabled. It skips only when the opt-in is unset; a set but blank version +// fails so a broken pin cannot pass by skipping. func requirePiRuntime(t *testing.T) string { t.Helper() - want := os.Getenv(piRuntimeVersionEnv) - if want == "" { + want, set := os.LookupEnv(piRuntimeVersionEnv) + if !set { t.Skipf("set %s to the required Pi version (make test-pi-runtime) to run installed Pi runtime tests", piRuntimeVersionEnv) } + if strings.TrimSpace(want) == "" { + t.Fatalf("%s is set but blank; it must name the exact required Pi version", piRuntimeVersionEnv) + } install := installedPiRuntime() if install.err != nil { t.Fatalf("%s=%s requires the installed Pi runtime: %v", piRuntimeVersionEnv, want, install.err) diff --git a/internal/llmadapters/pi_rpc_runtime_unix_test.go b/internal/llmadapters/pi_rpc_runtime_unix_test.go index c592f8fa..d1c3cb69 100644 --- a/internal/llmadapters/pi_rpc_runtime_unix_test.go +++ b/internal/llmadapters/pi_rpc_runtime_unix_test.go @@ -6,6 +6,7 @@ import ( "context" "errors" "os" + "os/exec" "path/filepath" "strconv" "strings" @@ -103,3 +104,65 @@ func readPiRuntimePID(t *testing.T, path string) int { } return pid } + +// A blank Pi pin must stop both Make targets: it would make the runtime tests +// skip and report ok, and it would let npm install the latest Pi. A fake go on +// PATH records any dispatch instead of running the suite. +func TestPiRuntimeMakeTargetsRejectBlankPin(t *testing.T) { + makePath, err := exec.LookPath("make") + if err != nil { + t.Skip("make is not installed") + } + repoRoot, err := filepath.Abs(filepath.Join("..", "..")) + if err != nil { + t.Fatalf("Abs(repo root): %v", err) + } + binDir := t.TempDir() + recordPath := filepath.Join(t.TempDir(), "go-dispatch") + fakeGo := "#!/bin/sh\nprintf '%s' \"$CR_PI_RUNTIME_VERSION\" > " + strconv.Quote(recordPath) + "\n" + if err := os.WriteFile(filepath.Join(binDir, "go"), []byte(fakeGo), 0o700); err != nil { // #nosec G306 -- fake go must be executable and is rooted in t.TempDir. + t.Fatalf("WriteFile(fake go): %v", err) + } + env := []string{"PATH=" + binDir + string(os.PathListSeparator) + os.Getenv("PATH")} + for _, entry := range os.Environ() { + switch key, _, _ := strings.Cut(entry, "="); key { + case "PATH", "MAKEFLAGS", "MFLAGS", "MAKELEVEL", "PI_RUNTIME_VERSION", piRuntimeVersionEnv: + default: + env = append(env, entry) + } + } + runMake := func(args ...string) (string, bool, error) { + t.Helper() + _ = os.Remove(recordPath) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, makePath, append([]string{"-s", "--no-print-directory", "-C", repoRoot}, args...)...) // #nosec G204 -- test runs make with fixed targets. + cmd.Env = env + output, err := cmd.Output() + recorded, readErr := os.ReadFile(recordPath) // #nosec G304 -- recordPath is rooted in t.TempDir. + if readErr != nil && !errors.Is(readErr, os.ErrNotExist) { + t.Fatalf("ReadFile(go dispatch record): %v", readErr) + } + if readErr == nil { + return string(recorded), true, err + } + return string(output), false, err + } + + pin, dispatched, err := runMake("pi-runtime-version") + pin = strings.TrimSpace(pin) + if err != nil || dispatched || pin == "" || strings.ContainsAny(pin, " \t") { + t.Fatalf("make pi-runtime-version = %q (err %v), want one default version", pin, err) + } + if got, dispatched, err := runMake("test-pi-runtime"); err != nil || !dispatched || got != pin { + t.Fatalf("make test-pi-runtime dispatched=%v with %s=%q (err %v), want go test with %q", dispatched, piRuntimeVersionEnv, got, err, pin) + } + for _, bad := range []string{"", " ", "\t", pin + " " + pin} { + if got, dispatched, err := runMake("pi-runtime-version", "PI_RUNTIME_VERSION="+bad); err == nil || dispatched || strings.TrimSpace(got) != "" { + t.Errorf("make pi-runtime-version PI_RUNTIME_VERSION=%q printed %q (err %v), want failure without output", bad, got, err) + } + if got, dispatched, err := runMake("test-pi-runtime", "PI_RUNTIME_VERSION="+bad); err == nil || dispatched { + t.Errorf("make test-pi-runtime PI_RUNTIME_VERSION=%q dispatched=%v with %q (err %v), want failure before go test", bad, dispatched, got, err) + } + } +} From 73bf803c51fdb5ee75bf3b0add13ed96d6410818 Mon Sep 17 00:00:00 2001 From: zzwong <6979793+zzwong@users.noreply.github.com> Date: Thu, 8 Oct 2026 00:27:57 -0400 Subject: [PATCH 5/5] fix(pi): isolate RPC prompts from ambient instructions [#642] --- README.md | 9 ++ docs/development.md | 34 ++++-- internal/llmadapters/pi_rpc.go | 46 ++++++--- internal/llmadapters/pi_rpc_runtime_test.go | 48 ++++++++- internal/llmadapters/pi_rpc_test.go | 109 +++++++++++++++++++- 5 files changed, 217 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index 839c7f22..c1b00a53 100644 --- a/README.md +++ b/README.md @@ -133,6 +133,15 @@ discovery; and `--no-context-files`, `--no-approve`, `--no-skills`, `--no-prompt-templates`, `--no-themes`, and `--no-session`. It returns an incompatible-runtime error when any control is unavailable. +Every CR-owned Pi request disables `AGENTS.md`/`CLAUDE.md` context discovery, +supplies its own system prompt, and passes an explicit empty +`--append-system-prompt` source. This also suppresses ambient `SYSTEM.md` and +`APPEND_SYSTEM.md` inputs, including those in the selected agent directory. +CR leaves Pi's selected agent directory and authentication storage in place; +credentials are never copied into disposable scratch state, so Pi can persist +OAuth refresh updates to its original store. Resource discovery remains +disabled, and only reviewers load the CR-owned read-only tool extension. + ```bash cr init --non-interactive \ --llm-provider pi \ diff --git a/docs/development.md b/docs/development.md index 8581b7f7..cdba393f 100644 --- a/docs/development.md +++ b/docs/development.md @@ -122,18 +122,32 @@ The tests make no paid or external calls: - Each test gives Pi a fresh `PI_CODING_AGENT_DIR` and `HOME` with `PI_OFFLINE=1`. - The only model is a scripted OpenAI-compatible endpoint on loopback, with a - dummy key. + dummy key stored in the selected directory's `auth.json`. - Reviewer tool calls run the real `__pi-review-tool` helper, dispatched to the test binary by `TestMain`. -- Hostile instructions, extensions, MCP servers, and skills are planted under - the fixture `HOME`'s default `.pi/agent` location and in the reviewer - checkout, while Pi reads the fresh agent directory. The tests fail if any of - them reaches a provider request or runs. - -This shows that the fixture is isolated, not that production runs are. CR does -not replace the developer's own Pi agent directory, so its global instructions -can still reach prompts -([#642](https://github.com/open-cli-collective/codereview-cli/issues/642)). +- Hostile instructions, extensions, and MCP servers are planted in the + actually selected agent directory as well as under the fixture `HOME`'s + default `.pi/agent` location and in the reviewer checkout. Hostile skills + are also planted under the fixture home and checkout. The tests fail if any + resource reaches a provider request or runs. +- Both structured and reviewer cases plant each selected `AGENTS.md`, + `CLAUDE.md`, `AGENTS.override.md`, `SYSTEM.md`, and `APPEND_SYSTEM.md` + independently, so context-file precedence cannot hide an uncovered input. + +The Pi 1.0 prompt-resource contract is explicit: `--no-context-files` disables +context discovery, CR's `--system-prompt` suppresses `SYSTEM.md` discovery, +and `--append-system-prompt` with an empty argument supplies an explicit CLI +source list that suppresses `APPEND_SYSTEM.md` discovery while contributing no +text. See Pi's pinned [argument parser](https://github.com/earendil-works/pi/blob/v1.0.0/packages/coding-agent/src/cli/args.ts) +and [resource loader](https://github.com/earendil-works/pi/blob/v1.0.0/packages/coding-agent/src/core/resource-loader.ts). + +CR does not redirect `PI_CODING_AGENT_DIR`, `HOME`, or Pi's credential store to +achieve isolation. Authentication, model configuration, and OAuth refresh +persistence continue to use the selected directory. The ordinary subprocess +test verifies that a synthetic auth-state update remains in the selected store +after scratch cleanup; it does not exercise a real OAuth provider or refresh. +All installed-runtime evidence is local/mock-provider evidence, not deployed +validation or an operating-system sandbox claim. The CI `pi-runtime` job installs the Makefile version on the Node release it pins and runs the same target. It is not a required check. To move to a new Pi diff --git a/internal/llmadapters/pi_rpc.go b/internal/llmadapters/pi_rpc.go index 41e3e309..e6e9e5fe 100644 --- a/internal/llmadapters/pi_rpc.go +++ b/internal/llmadapters/pi_rpc.go @@ -241,6 +241,7 @@ func (a *PiRPCAdapter) preflightReviewerRuntime(parent context.Context) error { args := append(append([]string(nil), a.commandArgsPrefix...), "--mode", "rpc", "--system-prompt", piRPCReviewerSystemPrompt, + "--append-system-prompt", "", "--no-builtin-tools", "--tools", piRPCReviewerToolNames, "--extension", extensionPath, @@ -395,6 +396,13 @@ func (a *PiRPCAdapter) buildArgs(req Request, extensionPath string) ([]string, e args := []string{ "--mode", "rpc", "--system-prompt", systemPrompt, + // An explicit empty CLI append source suppresses APPEND_SYSTEM.md + // discovery in Pi 1.0. --no-context-files separately suppresses + // AGENTS.md/CLAUDE.md; --system-prompt overrides SYSTEM.md discovery. + // Keep the selected agent directory intact so authentication refreshes + // continue to persist in the user's original credential store. + "--append-system-prompt", "", + "--no-context-files", "--no-extensions", "--no-skills", "--no-prompt-templates", @@ -405,7 +413,7 @@ func (a *PiRPCAdapter) buildArgs(req Request, extensionPath string) ([]string, e args = append(args, "--no-tools") } else { args[3] = piRPCReviewerSystemPrompt - args = append(args, "--no-builtin-tools", "--no-context-files", "--no-approve", "--tools", piRPCReviewerToolNames, "--extension", extensionPath) + args = append(args, "--no-builtin-tools", "--no-approve", "--tools", piRPCReviewerToolNames, "--extension", extensionPath) } if req.Model != "" { args = append(args, "--model", req.Model) @@ -418,21 +426,22 @@ func (a *PiRPCAdapter) buildArgs(req Request, extensionPath string) ([]string, e func (a *PiRPCAdapter) validateArgs(args []string, req Request, extensionPath string) error { allowedFlags := map[string]bool{ - "--mode": true, - "--system-prompt": true, - "--no-tools": false, - "--no-builtin-tools": false, - "--no-context-files": false, - "--no-approve": false, - "--tools": true, - "--no-extensions": false, - "--extension": true, - "--no-skills": false, - "--no-prompt-templates": false, - "--no-themes": false, - "--no-session": false, - "--model": true, - "--thinking": true, + "--mode": true, + "--system-prompt": true, + "--append-system-prompt": true, + "--no-tools": false, + "--no-builtin-tools": false, + "--no-context-files": false, + "--no-approve": false, + "--tools": true, + "--no-extensions": false, + "--extension": true, + "--no-skills": false, + "--no-prompt-templates": false, + "--no-themes": false, + "--no-session": false, + "--model": true, + "--thinking": true, } if err := validateAllowedFlags("pi_rpc", args, allowedFlags); err != nil { return err @@ -445,11 +454,14 @@ func (a *PiRPCAdapter) validateArgs(args []string, req Request, extensionPath st if flagValue(args, "--mode") != "rpc" { return fmt.Errorf("%w: pi_rpc must use rpc mode", ErrUnsafeSubprocessConfig) } - for _, flag := range []string{"--system-prompt", "--no-extensions", "--no-skills", "--no-prompt-templates", "--no-themes", "--no-session"} { + for _, flag := range []string{"--system-prompt", "--append-system-prompt", "--no-context-files", "--no-extensions", "--no-skills", "--no-prompt-templates", "--no-themes", "--no-session"} { if !containsFlag(args, flag) { return fmt.Errorf("%w: missing %s", ErrUnsafeSubprocessConfig, flag) } } + if appendPrompt, ok := flagValueOK(args, "--append-system-prompt"); !ok || appendPrompt != "" { + return fmt.Errorf("%w: pi_rpc must use an explicit empty append system prompt", ErrUnsafeSubprocessConfig) + } wantSystemPrompt := piRPCSystemPrompt if req.ReviewerWorkspace == nil { if !containsFlag(args, "--no-tools") || containsFlag(args, "--no-builtin-tools") || containsFlag(args, "--tools") || containsFlag(args, "-t") || containsFlag(args, "--extension") { diff --git a/internal/llmadapters/pi_rpc_runtime_test.go b/internal/llmadapters/pi_rpc_runtime_test.go index e9045418..bd6c5106 100644 --- a/internal/llmadapters/pi_rpc_runtime_test.go +++ b/internal/llmadapters/pi_rpc_runtime_test.go @@ -82,6 +82,47 @@ func TestPiRPCRuntimeStructuredJSONWithoutTools(t *testing.T) { } } +// Plant each instruction independently so Pi's AGENTS/CLAUDE precedence cannot +// hide an uncovered source. The selected directory also contains the fixture's +// real models.json and dummy auth.json, which must continue to be used. +func TestPiRPCRuntimeSelectedAgentInstructionIsolation(t *testing.T) { + for _, reviewer := range []bool{false, true} { + mode := "structured" + if reviewer { + mode = "reviewer" + } + t.Run(mode, func(t *testing.T) { + for _, filename := range []string{"AGENTS.md", "CLAUDE.md", "AGENTS.override.md", "SYSTEM.md", "APPEND_SYSTEM.md"} { + t.Run(filename, func(t *testing.T) { + fixture := newPiRuntimeFixture(t, piRuntimeAnswer(`{"ok":true}`, piRuntimeUsage{prompt: 10, completion: 1})) + for _, candidate := range []string{"AGENTS.md", "CLAUDE.md", "SYSTEM.md", "APPEND_SYSTEM.md"} { + if err := os.Remove(filepath.Join(fixture.agentDir, candidate)); err != nil { // #nosec G703 -- synthetic instruction fixtures only. + t.Fatalf("Remove(selected instruction): %v", err) + } + } + fixture.writeFile(filepath.Join(fixture.agentDir, filename), piRuntimeHostileMarker+" selected "+filename+"\n") + req := Request{Model: piRuntimeModel, Prompt: `Return {"ok":true}.`} + wantPrompt := piRPCSystemPrompt + if reviewer { + req.ReviewerWorkspace = fixture.reviewerWorkspace() + wantPrompt = piRPCReviewerSystemPrompt + } + response, err := fixture.run(fixture.adapter(piRuntimeTaskTimeout), req) + if err != nil || string(response.StructuredOutput) != `{"ok":true}` { + t.Fatalf("run output = %q (err %v), want CR response", response.StructuredOutput, err) + } + requests := fixture.provider.requests() + if len(requests) != 1 || !strings.Contains(requests[0].systemText(), wantPrompt) { + t.Fatalf("provider requests = %d, want one with the CR prompt", len(requests)) + } + // assertIsolated checks every request for the marker and the + // dummy selected-store authorization, including tool turns. + }) + } + }) + } +} + func TestPiRPCRuntimeRecoversTransientProviderFailure(t *testing.T) { fixture := newPiRuntimeFixture(t, piRuntimeFailure(http.StatusServiceUnavailable, "fixture transient overload"), @@ -374,7 +415,6 @@ func newPiRuntimeFixture(t *testing.T, steps ...piRuntimeStep) *piRuntimeFixture piRuntimeProviderName: map[string]any{ "baseUrl": fixture.provider.server.URL + "/v1", "api": "openai-completions", - "apiKey": piRuntimeAPIKey, "models": []any{map[string]any{ "id": piRuntimeModelID, "name": "CR runtime fixture", @@ -389,10 +429,16 @@ func newPiRuntimeFixture(t *testing.T, steps ...piRuntimeStep) *piRuntimeFixture }, }, }) + fixture.writeJSON(filepath.Join(fixture.agentDir, "auth.json"), map[string]any{ + piRuntimeProviderName: map[string]any{"type": "api_key", "key": piRuntimeAPIKey}, + }) fixture.writeJSON(filepath.Join(fixture.agentDir, "settings.json"), map[string]any{ "retry": map[string]any{"enabled": true, "maxRetries": 1, "baseDelayMs": 10}, "enableInstallTelemetry": false, }) + // The actual selected directory must reject instruction resources without + // hiding its model configuration, credentials, or refresh destination. + fixture.plantHostileResources(fixture.agentDir, "selected-agent") // Resources at Pi's default global location. The agent directory override // must keep every one of them out of the run. fixture.plantHostileResources(filepath.Join(home, ".pi", "agent"), "global") diff --git a/internal/llmadapters/pi_rpc_test.go b/internal/llmadapters/pi_rpc_test.go index 986d5f2b..ef26be61 100644 --- a/internal/llmadapters/pi_rpc_test.go +++ b/internal/llmadapters/pi_rpc_test.go @@ -73,7 +73,8 @@ func TestPiRPCLaunchSafetyAndSuccess(t *testing.T) { assertFlagValue(t, record.AdapterArgs, "--model", "opencode-go/kimi-k2.6") assertFlagValue(t, record.AdapterArgs, "--thinking", "max") assertFlagValue(t, record.AdapterArgs, "--system-prompt", piRPCSystemPrompt) - for _, flag := range []string{"--no-tools", "--no-extensions", "--no-skills", "--no-prompt-templates", "--no-themes", "--no-session"} { + assertPiRPCEmptyAppend(t, record.AdapterArgs) + for _, flag := range []string{"--no-tools", "--no-context-files", "--no-extensions", "--no-skills", "--no-prompt-templates", "--no-themes", "--no-session"} { if !containsFlag(record.AdapterArgs, flag) { t.Fatalf("args = %#v, want %s", record.AdapterArgs, flag) } @@ -120,6 +121,86 @@ func TestPiRPCStreamRecordsDuration(t *testing.T) { assertSlowSuccessDuration(t, response) } +// This subprocess double models an auth.json refresh without using credentials +// or a provider. Prompt isolation must leave the selected store in place, so +// its updated synthetic state survives invocation scratch cleanup. +func TestPiRPCPreservesSelectedAgentAuthenticationState(t *testing.T) { + for _, reviewer := range []bool{false, true} { + name := "structured" + if reviewer { + name = "reviewer" + } + t.Run(name, func(t *testing.T) { + root := t.TempDir() + agentDir := filepath.Join(root, "selected-agent") + home := filepath.Join(root, "home") + repoDir := filepath.Join(root, "repo") + scratchDir := filepath.Join(root, "scratch") + for _, dir := range []string{agentDir, home, repoDir, scratchDir} { + if err := os.Mkdir(dir, 0o700); err != nil { + t.Fatalf("Mkdir: %v", err) + } + } + authPath := filepath.Join(agentDir, "auth.json") + if err := os.WriteFile(authPath, []byte(`{"fixture":"initial synthetic auth state"}`), 0o600); err != nil { + t.Fatalf("WriteFile(synthetic auth): %v", err) + } + recordPath := filepath.Join(root, "record.json") + adapter := NewPiRPCAdapter(PiRPCOptions{ + Command: os.Args[0], + commandArgsPrefix: piRPCHelperPrefix(), + Env: append(piRPCHelperEnv("success", recordPath), + "PI_CODING_AGENT_DIR="+agentDir, + "HOME="+home, + "LLM_PI_RPC_SYNTHETIC_AUTH_REFRESH=1", + ), + Timeout: 5 * time.Second, + ScratchDirFactory: func() (string, func() error, error) { + dir, err := os.MkdirTemp(scratchDir, "invocation-*") + return dir, func() error { return os.RemoveAll(dir) }, err + }, + }) + req := Request{Model: "fixture/model", Prompt: "fixture prompt"} + wantPrompt := piRPCSystemPrompt + if reviewer { + wantPrompt = piRPCReviewerSystemPrompt + diffPath := filepath.Join(root, "diff.patch") + if err := os.WriteFile(diffPath, []byte("fixture diff\n"), 0o600); err != nil { + t.Fatalf("WriteFile(diff): %v", err) + } + req.ReviewerWorkspace = &ReviewerWorkspaceRequest{ + RepoDir: repoDir, ScratchDir: scratchDir, DiffPath: diffPath, MaxToolOutputBytes: 2048, + } + } + stream, err := adapter.Start(context.Background(), req) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err := stream.Wait(context.Background()); err != nil { + t.Fatalf("Wait: %v", err) + } + record := readPiRPCRecord(t, recordPath) + if record.Env["PI_CODING_AGENT_DIR"] != agentDir || record.Env["HOME"] != home { + t.Fatalf("selected agent/home = %q/%q, want %q/%q", record.Env["PI_CODING_AGENT_DIR"], record.Env["HOME"], agentDir, home) + } + assertFlagValue(t, record.AdapterArgs, "--system-prompt", wantPrompt) + if appendPrompt, present := flagValueOK(record.AdapterArgs, "--append-system-prompt"); !present || appendPrompt != "" { + t.Fatalf("append system prompt = %q (present %v), want explicit empty input", appendPrompt, present) + } + if !containsFlag(record.AdapterArgs, "--no-context-files") { + t.Fatalf("args = %#v, want context discovery disabled", record.AdapterArgs) + } + data, err := os.ReadFile(authPath) // #nosec G304 -- synthetic test-owned auth state only. + if err != nil || string(data) != `{"fixture":"refreshed synthetic auth state"}` { + t.Fatalf("synthetic auth state = %q (err %v), want refresh in original selected store", data, err) + } + if entries, err := os.ReadDir(scratchDir); err != nil || len(entries) != 0 { + t.Fatalf("scratch entries = %v (err %v), want cleanup without auth state loss", entries, err) + } + }) + } +} + func TestPiRPCFailureLeavesDurationZero(t *testing.T) { recordPath := filepath.Join(t.TempDir(), "record.json") adapter := NewPiRPCAdapter(PiRPCOptions{ @@ -198,6 +279,7 @@ func TestPiRPCReviewerWorkspaceLaunchUsesOnlyCROwnedTools(t *testing.T) { t.Fatalf("args = %#v, reviewer extension tools must remain enabled", record.AdapterArgs) } assertFlagValue(t, record.AdapterArgs, "--tools", piRPCReviewerToolNames) + assertPiRPCEmptyAppend(t, record.AdapterArgs) reviewerPrompt := flagValue(record.AdapterArgs, "--system-prompt") for _, instruction := range []string{"Invoke cr_diff before cr_read, cr_search, or cr_list", "If cr_diff fails"} { if !strings.Contains(reviewerPrompt, instruction) { @@ -548,6 +630,7 @@ func TestPiRPCReviewerPreflightUsesEmptyDiscoveryDisabledDirectory(t *testing.T) } assertFlagValue(t, record.AdapterArgs, "--mode", "rpc") assertFlagValue(t, record.AdapterArgs, "--tools", piRPCReviewerToolNames) + assertPiRPCEmptyAppend(t, record.AdapterArgs) for _, flag := range []string{"--no-builtin-tools", "--no-extensions", "--no-skills", "--no-prompt-templates", "--no-themes", "--no-context-files", "--no-approve", "--no-session"} { if !containsFlag(record.AdapterArgs, flag) { t.Fatalf("preflight args = %#v, want %s", record.AdapterArgs, flag) @@ -1686,6 +1769,10 @@ func TestPiRPCRejectsUnsafeSpecs(t *testing.T) { {name: "tools allowlist", args: append([]string{"--tools", "read"}, args...)}, {name: "missing no tools", args: removeFlag(args, "--no-tools")}, {name: "missing no extensions", args: removeFlag(args, "--no-extensions")}, + {name: "missing context disable", args: removeFlag(args, "--no-context-files")}, + {name: "missing append override", args: removeFlagWithValue(args, "--append-system-prompt")}, + {name: "nonempty append override", args: replaceFlagValue(args, "--append-system-prompt", "ambient instructions")}, + {name: "duplicate append override", args: append(append([]string(nil), args...), "--append-system-prompt", "")}, {name: "unexpected flag", args: append([]string{"--unexpected"}, args...)}, {name: "text mode", args: replaceFlagValue(args, "--mode", "text")}, {name: "wrong system prompt", args: replaceFlagValue(args, "--system-prompt", "be loose")}, @@ -1712,6 +1799,9 @@ func TestPiRPCRejectsUnsafeReviewerSpecs(t *testing.T) { }{ {name: "missing builtin disable", args: removeFlag(args, "--no-builtin-tools")}, {name: "missing context disable", args: removeFlag(args, "--no-context-files")}, + {name: "missing append override", args: removeFlagWithValue(args, "--append-system-prompt")}, + {name: "nonempty append override", args: replaceFlagValue(args, "--append-system-prompt", "ambient instructions")}, + {name: "duplicate append override", args: append(append([]string(nil), args...), "--append-system-prompt", "")}, {name: "missing project approval disable", args: removeFlag(args, "--no-approve")}, {name: "all tools disabled", args: append(removeFlagWithValue(args, "--tools"), "--no-tools")}, {name: "native bash added", args: replaceFlagValue(args, "--tools", piRPCReviewerToolNames+",bash")}, @@ -1801,6 +1891,14 @@ func TestPiRPCHelperProcess(_ *testing.T) { } os.Exit(0) } + if os.Getenv("LLM_PI_RPC_SYNTHETIC_AUTH_REFRESH") == "1" { + // Only this opt-in fixture passes a synthetic selected store and home. + record.Env["PI_CODING_AGENT_DIR"] = os.Getenv("PI_CODING_AGENT_DIR") + record.Env["HOME"] = os.Getenv("HOME") + data, _ := json.Marshal(record) + _ = os.WriteFile(recordPath, data, 0o600) // #nosec G703 -- fixture-owned record path. + _ = os.WriteFile(filepath.Join(record.Env["PI_CODING_AGENT_DIR"], "auth.json"), []byte(`{"fixture":"refreshed synthetic auth state"}`), 0o600) // #nosec G703 -- opt-in synthetic fixture store, never real credentials. + } // Pi RPC stays alive after agent_settled and shuts down when stdin closes. stdinClosed := make(chan struct{}) go func() { @@ -2084,3 +2182,12 @@ func TestPiRPCToolUseKillsProcessGroup(t *testing.T) { return !processExists(pid) }) } + +// Absence is distinct from an explicitly empty append source in Pi. +func assertPiRPCEmptyAppend(t *testing.T, args []string) { + t.Helper() + value, present := flagValueOK(args, "--append-system-prompt") + if !present || value != "" || countPiRPCFlag(args, "--append-system-prompt") != 1 { + t.Fatalf("append system prompt = %q (present %v) in %#v, want exactly one empty source", value, present, args) + } +}