返回 DeepSeek-Reasonix
status_test.go
根目录 / internal / acp / status_test.go
1 package acp
2
3 import (
4 "context"
5 "encoding/json"
6 "errors"
7 "path/filepath"
8 "strings"
9 "testing"
10
11 "reasonix/internal/agent"
12 "reasonix/internal/billing"
13 "reasonix/internal/event"
14 "reasonix/internal/provider"
15 )
16
17 type statusFactory struct {
18 *configurableFactory
19 }
20
21 func TestUsageAccumulatorTotalsMoreThanAuditLimit(t *testing.T) {
22 var accumulator usageAccumulator
23 usage := &provider.Usage{PromptTokens: 1_000_000}
24 pricing := &provider.Pricing{Input: 1, Currency: "USD"}
25 for range 65 {
26 quote := billing.BuildQuote(billing.QuoteInput{
27 Usage: billing.UsageTokens{PromptTokens: usage.PromptTokens},
28 Rates: billing.RateCard{Input: pricing.Input, Currency: pricing.Currency},
29 DisplayCurrency: "USD",
30 })
31 accumulator.addQuoted(usage, pricing, &quote, event.UsageSourceExecutor)
32 }
33 wire := accumulator.wire()
34 if wire.EstimatedCost == nil || *wire.EstimatedCost != 65 || wire.Currency == nil || *wire.Currency != "USD" {
35 t.Fatalf("65-event ACP total was truncated: %+v", wire)
36 }
37 if wire.CostQuote == nil || wire.CostQuote.Selected == nil || wire.CostQuote.Selected.Amount != "65" {
38 t.Fatalf("65-event ACP aggregate quote = %+v", wire.CostQuote)
39 }
40 if wire.CostComplete == nil || !*wire.CostComplete {
41 t.Fatalf("65-event ACP quote incomplete: %+v", wire)
42 }
43 }
44
45 func TestUsageAccumulatorExposesAuthoritativeTotalWithoutCacheDoubleCount(t *testing.T) {
46 var accumulator usageAccumulator
47 accumulator.addQuoted(&provider.Usage{
48 PromptTokens: 1_000, CompletionTokens: 500, ReasoningTokens: 300,
49 CacheHitTokens: 800, CacheMissTokens: 200,
50 }, nil, nil, event.UsageSourceExecutor)
51
52 wire := accumulator.wire()
53 if wire.TotalTokens != 1_500 {
54 t.Fatalf("total tokens = %d, want 1500: %+v", wire.TotalTokens, wire)
55 }
56 if wire.PromptTokens != wire.CacheHitTokens+wire.CacheMissTokens {
57 t.Fatalf("cache split no longer partitions prompt tokens: %+v", wire)
58 }
59 }
60
61 func TestRestoreUsageReconstructsTotalTokensFromLegacySnapshot(t *testing.T) {
62 wire := restoreUsage(persistedUsageAccumulator{
63 PromptTokens: 1_000, CompletionTokens: 500,
64 CacheHitTokens: 800, CacheMissTokens: 200,
65 }).wire()
66
67 if wire.TotalTokens != 1_500 {
68 t.Fatalf("restored total tokens = %d, want 1500: %+v", wire.TotalTokens, wire)
69 }
70 }
71
72 func TestRestoredUsageKeepsFullScalarTotalAfterNewQuote(t *testing.T) {
73 complete := true
74 accumulator := restoreUsage(persistedUsageAccumulator{
75 PromptTokens: 1_000_000, Events: 1, PricedEvents: 1,
76 EstimatedCost: 2, Currency: "USD", CostComplete: &complete,
77 })
78 usage := &provider.Usage{PromptTokens: 1_000_000}
79 pricing := &provider.Pricing{Input: 1, Currency: "USD"}
80 quote := billing.BuildQuote(billing.QuoteInput{
81 Usage: billing.UsageTokens{PromptTokens: usage.PromptTokens},
82 Rates: billing.RateCard{Input: pricing.Input, Currency: pricing.Currency},
83 DisplayCurrency: "USD",
84 })
85 accumulator.addQuoted(usage, pricing, &quote, event.UsageSourceExecutor)
86 wire := accumulator.wire()
87 if wire.EstimatedCost == nil || *wire.EstimatedCost != 3 {
88 t.Fatalf("restored scalar history was replaced by the new ledger fragment: %+v", wire)
89 }
90 if wire.CostComplete == nil || !*wire.CostComplete {
91 t.Fatalf("restored complete state was lost: %+v", wire)
92 }
93 }
94
95 type runtimeTrackingFactory struct {
96 *configurableFactory
97 }
98
99 func (f *statusFactory) SessionRuntimeState(_ context.Context, p SessionRuntimeStateParams) (SessionRuntimeState, error) {
100 return SessionRuntimeState{
101 PlannerMode: "off",
102 Sandbox: SessionSandboxState{
103 Mode: "enforce", Engine: "bubblewrap", Available: true, WorkspaceRoot: p.Cwd,
104 WriteRoots: []string{p.Cwd}, NetworkEnabled: false,
105 },
106 }, nil
107 }
108
109 func (f *runtimeTrackingFactory) SessionRuntimeState(_ context.Context, p SessionRuntimeStateParams) (SessionRuntimeState, error) {
110 return SessionRuntimeState{
111 PlannerMode: "on",
112 Sandbox: SessionSandboxState{
113 Mode: "enforce", Engine: "bubblewrap", Available: true, WorkspaceRoot: p.Cwd,
114 WriteRoots: []string{p.Cwd},
115 },
116 }, nil
117 }
118
119 func openStatusSession(t *testing.T, client *rpcClient, cwd string) string {
120 t.Helper()
121 resp := client.call(t, "session/new", SessionNewParams{Cwd: cwd})
122 if resp.Error != nil {
123 t.Fatalf("session/new: %+v", resp.Error)
124 }
125 var opened SessionNewResult
126 if err := json.Unmarshal(resp.Result, &opened); err != nil {
127 t.Fatalf("session/new result: %v", err)
128 }
129 return opened.SessionID
130 }
131
132 func getStatus(t *testing.T, client *rpcClient, sessionID string) ReasonixSessionStatus {
133 t.Helper()
134 resp := client.call(t, sessionStatusMethod, SessionStatusParams{SessionID: sessionID})
135 if resp.Error != nil {
136 t.Fatalf("session/status: %+v", resp.Error)
137 }
138 var status ReasonixSessionStatus
139 if err := json.Unmarshal(resp.Result, &status); err != nil {
140 t.Fatalf("session/status result: %v", err)
141 }
142 return status
143 }
144
145 func TestStatusExtensionTracksMultipleSessionsAndUsage(t *testing.T) {
146 factory := &statusFactory{configurableFactory: &configurableFactory{
147 behavior: func(_ context.Context, sink event.Sink, input string, _ SessionParams) error {
148 sink.Emit(event.Event{Kind: event.Phase, Source: event.UsageSourceExecutor, Text: "executor · implementing"})
149 sink.Emit(event.Event{Kind: event.Usage, Usage: &provider.Usage{
150 PromptTokens: 10, CompletionTokens: 4, ReasoningTokens: 2,
151 CacheHitTokens: 7, CacheMissTokens: 3, Estimated: true,
152 }, Pricing: &provider.Pricing{CacheHit: 0.1, Input: 1, Output: 2, Currency: "USD"}, UsageSource: event.UsageSourceExecutor})
153 sink.Emit(event.Event{Kind: event.Usage, Usage: &provider.Usage{
154 PromptTokens: 5, CompletionTokens: 1, CacheMissTokens: 5,
155 }, Pricing: &provider.Pricing{CacheHit: 0.1, Input: 1, Output: 2, Currency: "USD"}, UsageSource: event.UsageSourceCompaction})
156 sink.Emit(event.Event{Kind: event.Text, Text: input})
157 return nil
158 },
159 }}
160 client, stop := startServer(t, factory)
161 defer stop()
162 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
163 first := openStatusSession(t, client, t.TempDir())
164 second := openStatusSession(t, client, t.TempDir())
165
166 initialSecond := getStatus(t, client, second)
167 prompt := client.callAsync("session/prompt", SessionPromptParams{SessionID: first, Prompt: []ContentBlock{{Type: "text", Text: "ship"}}})
168 notifications, response := drainPrompt(t, client, prompt)
169 if response.Error != nil {
170 t.Fatalf("session/prompt: %+v", response.Error)
171 }
172
173 firstStatus := getStatus(t, client, first)
174 if firstStatus.Sequence == 0 || firstStatus.State != "idle" || firstStatus.TurnOutcome.Kind != "completed" {
175 t.Fatalf("first status = %+v", firstStatus)
176 }
177 if firstStatus.PlannerMode != "off" || firstStatus.Sandbox.WorkspaceRoot == "" || len(firstStatus.Sandbox.WriteRoots) != 1 {
178 t.Fatalf("effective runtime status = %+v", firstStatus)
179 }
180 usage := firstStatus.Usage.Cumulative
181 if usage.TotalTokens != 20 || usage.PromptTokens != 15 || usage.CompletionTokens != 5 || usage.ReasoningTokens != 2 || usage.CacheHitTokens != 7 || usage.CacheMissTokens != 8 {
182 t.Fatalf("cumulative usage = %+v", usage)
183 }
184 if usage.UsageSource != "mixed" || usage.CacheHitRatio == nil || usage.EstimatedCost == nil || usage.Currency == nil || *usage.Currency != "USD" {
185 t.Fatalf("usage metadata = %+v", usage)
186 }
187 if !usage.Estimated {
188 t.Fatalf("cumulative usage lost estimated marker: %+v", usage)
189 }
190 secondStatus := getStatus(t, client, second)
191 if secondStatus.Sequence != initialSecond.Sequence || secondStatus.Usage.Cumulative.PromptTokens != 0 {
192 t.Fatalf("second session telemetry leaked: before=%+v after=%+v", initialSecond, secondStatus)
193 }
194
195 var sawPhase, sawUsage, sawCompletion bool
196 for _, notification := range notifications {
197 if notification.Method != sessionStatusUpdateMethod {
198 continue
199 }
200 var update ReasonixStatusUpdate
201 if err := json.Unmarshal(notification.Params, &update); err != nil {
202 t.Fatalf("status update: %v", err)
203 }
204 if update.Sequence != update.Status.Sequence || update.SessionID != first {
205 t.Fatalf("status update correlation = %+v", update)
206 }
207 switch update.Event {
208 case "phase":
209 sawPhase = true
210 case "usage":
211 sawUsage = true
212 case "completion":
213 sawCompletion = true
214 }
215 }
216 if !sawPhase || !sawUsage || !sawCompletion {
217 t.Fatalf("status events phase=%v usage=%v completion=%v", sawPhase, sawUsage, sawCompletion)
218 }
219 }
220
221 func TestStatusNormalizesPhaseAndRedactsPublicText(t *testing.T) {
222 const opaqueSecret = "readinessSecretAbc123"
223 telemetry := newStatusTelemetry()
224 telemetry.beginTurn()
225 telemetry.onEvent(event.Event{Kind: event.Phase, Source: event.UsageSourcePlanner, Text: "planner · private stage label"})
226 if got := telemetry.snapshot().phase; got != "planning" {
227 t.Fatalf("planner phase = %q, want planning", got)
228 }
229 telemetry.onEvent(event.Event{Kind: event.Phase, Text: "provider-specific handoff"})
230 if got := telemetry.snapshot().phase; got != "working" {
231 t.Fatalf("unknown phase = %q, want working", got)
232 }
233 telemetry.finishTurn(&agent.FinalReadinessError{
234 Attempts: 1,
235 Reason: "token=secret-reason credential " + opaqueSecret,
236 Missing: []string{"api_key=secret-risk"},
237 }, false, "running", "authorization: bearer secret-summary")
238 snapshot := telemetry.snapshot()
239 encoded, err := json.Marshal(snapshot.finalReadiness)
240 if err != nil {
241 t.Fatal(err)
242 }
243 if strings.Contains(string(encoded), "secret-") || !strings.Contains(string(encoded), "[redacted]") {
244 t.Fatalf("status text was not redacted: %s", encoded)
245 }
246 if strings.Contains(snapshot.turnOutcome.Reason, "secret-") || strings.Contains(snapshot.turnOutcome.Reason, opaqueSecret) {
247 t.Fatalf("turn outcome was not redacted: %q", snapshot.turnOutcome.Reason)
248 }
249
250 empty, err := json.Marshal(newStatusTelemetry().snapshot().finalReadiness)
251 if err != nil {
252 t.Fatal(err)
253 }
254 if !strings.Contains(string(empty), `"risks":[]`) {
255 t.Fatalf("empty risks must encode as [], got %s", empty)
256 }
257 }
258
259 func TestRestoreStatusNormalizesLegacyPresentationPhase(t *testing.T) {
260 restored := restoreStatusTelemetry(&persistedStatusTelemetry{
261 Phase: "executor · implementing local patch",
262 FinalReadiness: ReasonixFinalReadiness{},
263 })
264 if got := restored.snapshot().phase; got != "implementing" {
265 t.Fatalf("restored phase = %q, want implementing", got)
266 }
267 }
268
269 func TestRestoreStatusStronglyRedactsLegacyTurnOutcome(t *testing.T) {
270 const opaqueSecret = "readinessSecretAbc123"
271 const bearerSecret = "bearerSecretAbc123"
272 restored := restoreStatusTelemetry(&persistedStatusTelemetry{
273 TurnOutcome: ReasonixTurnOutcome{
274 Kind: "error",
275 Reason: "credential " + opaqueSecret + " Authorization: Bearer " + bearerSecret,
276 },
277 })
278
279 snapshot := restored.snapshot()
280 persisted := restored.persisted()
281 for name, reason := range map[string]string{
282 "public snapshot": snapshot.turnOutcome.Reason,
283 "repersisted data": persisted.TurnOutcome.Reason,
284 } {
285 if strings.Contains(reason, opaqueSecret) || strings.Contains(reason, bearerSecret) {
286 t.Errorf("%s leaked a legacy credential: %q", name, reason)
287 }
288 }
289 }
290
291 func TestRestoreStatusMarksInterruptedTurnPaused(t *testing.T) {
292 restored := restoreStatusTelemetry(&persistedStatusTelemetry{
293 Sequence: 7,
294 State: "running",
295 Phase: "implementing",
296 TurnOutcome: ReasonixTurnOutcome{Kind: "none"},
297 FinalReadiness: ReasonixFinalReadiness{
298 ReadyForReview: true,
299 Risks: []string{},
300 },
301 TurnUsage: persistedUsageAccumulator{PromptTokens: 3, Estimated: true, Events: 1},
302 Cumulative: persistedUsageAccumulator{PromptTokens: 11, Estimated: true, Events: 2},
303 })
304 snapshot := restored.snapshot()
305 if snapshot.state != "idle" || snapshot.phase != "recovery_paused" {
306 t.Fatalf("restored interrupted state = state:%q phase:%q, want idle/recovery_paused", snapshot.state, snapshot.phase)
307 }
308 if snapshot.sequence != 8 || snapshot.turnOutcome.Kind != "paused" || snapshot.turnOutcome.Reason != "previous turn interrupted" {
309 t.Fatalf("restored interrupted outcome = sequence:%d outcome:%+v", snapshot.sequence, snapshot.turnOutcome)
310 }
311 if snapshot.finalReadiness.ReadyForReview {
312 t.Fatal("interrupted turn remained ready for review")
313 }
314 if snapshot.turnUsage.PromptTokens != 3 || snapshot.cumulative.PromptTokens != 11 {
315 t.Fatalf("interrupted usage was lost: turn=%+v cumulative=%+v", snapshot.turnUsage, snapshot.cumulative)
316 }
317 if !snapshot.turnUsage.Estimated || !snapshot.cumulative.Estimated {
318 t.Fatalf("interrupted estimated marker was lost: turn=%+v cumulative=%+v", snapshot.turnUsage, snapshot.cumulative)
319 }
320 }
321
322 func TestStatusWorkModeSetConfigOptionIsHiddenCompatibilityNoOp(t *testing.T) {
323 factory := &runtimeTrackingFactory{configurableFactory: &configurableFactory{}}
324 client, stop := startServer(t, factory)
325 defer stop()
326 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
327 sessionID := openStatusSession(t, client, t.TempDir())
328 if status := getStatus(t, client, sessionID); status.WorkMode != "standard" || status.PlannerMode != "on" {
329 t.Fatalf("initial runtime status = %+v", status)
330 }
331 buildsBefore := factory.buildCount()
332
333 for _, value := range []string{"economy", "delivery", "light"} {
334 resp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
335 SessionID: sessionID,
336 ConfigID: "work_mode",
337 Value: value,
338 })
339 if resp.Error != nil {
340 t.Fatalf("set work mode %q: %+v", value, resp.Error)
341 }
342 var set SetSessionConfigOptionResult
343 if err := json.Unmarshal(resp.Result, &set); err != nil {
344 t.Fatalf("set work mode %q result: %v", value, err)
345 }
346 var floorOpt *SessionConfigOption
347 for i := range set.ConfigOptions {
348 if set.ConfigOptions[i].ID == "quality_floor" {
349 floorOpt = &set.ConfigOptions[i]
350 }
351 }
352 if floorOpt != nil {
353 t.Fatalf("retired quality floor option still advertised after work_mode %q: %+v", value, floorOpt)
354 }
355 if set.DeprecatedNotice == "" {
356 t.Fatalf("work_mode %q missing retirement notice", value)
357 }
358 status := getStatus(t, client, sessionID)
359 if status.WorkMode != "standard" || status.PlannerMode != "on" {
360 t.Fatalf("runtime status after deprecated work_mode %q = %+v", value, status)
361 }
362 }
363 if got := factory.buildCount(); got != buildsBefore {
364 t.Fatalf("work_mode rebuilt controller: builds=%d, want %d", got, buildsBefore)
365 }
366 }
367
368 func TestStatusClassifiesPauseAndError(t *testing.T) {
369 telemetry := newStatusTelemetry()
370 telemetry.beginTurn()
371 pauseEvent := telemetry.finishTurn(&agent.FinalReadinessError{Attempts: 3, Reason: "missing verification", Missing: []string{"verify"}}, false, "running", "partial")
372 paused := telemetry.snapshot()
373 if pauseEvent != "pause" || paused.turnOutcome.Kind != "paused" || len(paused.finalReadiness.Risks) != 1 {
374 t.Fatalf("pause classification = event %q snapshot %+v", pauseEvent, paused)
375 }
376
377 telemetry.beginTurn()
378 errorEvent := telemetry.finishTurn(errors.New("provider failed"), false, "running", "")
379 failed := telemetry.snapshot()
380 if errorEvent != "error" || failed.turnOutcome.Kind != "error" || failed.goalOverride != "failed" {
381 t.Fatalf("error classification = event %q snapshot %+v", errorEvent, failed)
382 }
383 }
384
385 func TestStatusSnapshotSurvivesSessionResume(t *testing.T) {
386 dir := t.TempDir()
387 cwd := t.TempDir()
388 sessionID := "status-reconnect"
389 telemetry := newStatusTelemetry()
390 telemetry.beginTurn()
391 telemetry.onEvent(event.Event{Kind: event.Usage, Usage: &provider.Usage{
392 PromptTokens: 8, CompletionTokens: 2, CacheHitTokens: 6, CacheMissTokens: 2,
393 }, UsageSource: event.UsageSourceExecutor})
394 telemetry.finishTurn(nil, false, "", "persisted summary")
395 path := filepath.Join(dir, sessionID+".jsonl")
396 if err := agent.NewSession("system").Save(path); err != nil {
397 t.Fatalf("save transcript: %v", err)
398 }
399 if err := saveACPMeta(path, acpSessionMeta{
400 SessionID: sessionID, Cwd: cwd, Model: "fast", RuntimeProfile: "delivery",
401 Status: telemetry.persisted(),
402 }); err != nil {
403 t.Fatalf("save ACP metadata: %v", err)
404 }
405 factory := &statusFactory{configurableFactory: &configurableFactory{
406 dir: dir,
407 }}
408 reconnected, stopReconnected := startServer(t, factory)
409 defer stopReconnected()
410 reconnected.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
411 resume := reconnected.call(t, "session/resume", SessionResumeParams{SessionID: sessionID, Cwd: cwd})
412 if resume.Error != nil {
413 t.Fatalf("session/resume: %+v", resume.Error)
414 }
415 after := getStatus(t, reconnected, sessionID)
416 if after.Sequence != telemetry.snapshot().sequence || after.Usage.Cumulative.PromptTokens != 8 || after.State != "idle" || after.FinalReadiness.Summary != "persisted summary" {
417 t.Fatalf("recovered status = %+v", after)
418 }
419 }
420
421 func TestStatusInterruptedSnapshotResumesPaused(t *testing.T) {
422 dir := t.TempDir()
423 cwd := t.TempDir()
424 sessionID := "status-interrupted"
425 telemetry := newStatusTelemetry()
426 telemetry.beginTurn()
427 telemetry.onEvent(event.Event{Kind: event.Usage, Usage: &provider.Usage{
428 PromptTokens: 5, CompletionTokens: 1,
429 }, UsageSource: event.UsageSourceExecutor})
430 path := filepath.Join(dir, sessionID+".jsonl")
431 if err := agent.NewSession("system").Save(path); err != nil {
432 t.Fatalf("save transcript: %v", err)
433 }
434 if err := saveACPMeta(path, acpSessionMeta{
435 SessionID: sessionID, Cwd: cwd, Model: "fast", RuntimeProfile: "balanced",
436 Status: telemetry.persisted(),
437 }); err != nil {
438 t.Fatalf("save ACP metadata: %v", err)
439 }
440
441 factory := &statusFactory{configurableFactory: &configurableFactory{dir: dir}}
442 client, stop := startServer(t, factory)
443 defer stop()
444 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
445 resume := client.call(t, "session/resume", SessionResumeParams{SessionID: sessionID, Cwd: cwd})
446 if resume.Error != nil {
447 t.Fatalf("session/resume: %+v", resume.Error)
448 }
449 after := getStatus(t, client, sessionID)
450 if after.State != "idle" || after.Phase != "recovery_paused" || after.TurnOutcome.Kind != "paused" {
451 t.Fatalf("resumed interrupted status = %+v", after)
452 }
453 if after.Sequence != telemetry.snapshot().sequence+1 || after.Usage.Turn.PromptTokens != 5 || after.Usage.Cumulative.PromptTokens != 5 {
454 t.Fatalf("resumed interrupted sequence/usage = %+v", after)
455 }
456 }
457
457 lines GO