| 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/event" |
| 13 | "reasonix/internal/provider" |
| 14 | ) |
| 15 | |
| 16 | type statusFactory struct { |
| 17 | *configurableFactory |
| 18 | } |
| 19 | |
| 20 | type runtimeTrackingFactory struct { |
| 21 | *configurableFactory |
| 22 | } |
| 23 | |
| 24 | func (f *statusFactory) SessionRuntimeState(_ context.Context, p SessionRuntimeStateParams) (SessionRuntimeState, error) { |
| 25 | return SessionRuntimeState{ |
| 26 | PlannerMode: "off", |
| 27 | Sandbox: SessionSandboxState{ |
| 28 | Mode: "enforce", Engine: "bubblewrap", Available: true, WorkspaceRoot: p.Cwd, |
| 29 | WriteRoots: []string{p.Cwd}, NetworkEnabled: false, |
| 30 | }, |
| 31 | }, nil |
| 32 | } |
| 33 | |
| 34 | func (f *runtimeTrackingFactory) SessionRuntimeState(_ context.Context, p SessionRuntimeStateParams) (SessionRuntimeState, error) { |
| 35 | return SessionRuntimeState{ |
| 36 | PlannerMode: "on", |
| 37 | Sandbox: SessionSandboxState{ |
| 38 | Mode: "enforce", Engine: "bubblewrap", Available: true, WorkspaceRoot: p.Cwd, |
| 39 | WriteRoots: []string{p.Cwd}, |
| 40 | }, |
| 41 | }, nil |
| 42 | } |
| 43 | |
| 44 | func openStatusSession(t *testing.T, client *rpcClient, cwd string) string { |
| 45 | t.Helper() |
| 46 | resp := client.call(t, "session/new", SessionNewParams{Cwd: cwd}) |
| 47 | if resp.Error != nil { |
| 48 | t.Fatalf("session/new: %+v", resp.Error) |
| 49 | } |
| 50 | var opened SessionNewResult |
| 51 | if err := json.Unmarshal(resp.Result, &opened); err != nil { |
| 52 | t.Fatalf("session/new result: %v", err) |
| 53 | } |
| 54 | return opened.SessionID |
| 55 | } |
| 56 | |
| 57 | func getStatus(t *testing.T, client *rpcClient, sessionID string) ReasonixSessionStatus { |
| 58 | t.Helper() |
| 59 | resp := client.call(t, sessionStatusMethod, SessionStatusParams{SessionID: sessionID}) |
| 60 | if resp.Error != nil { |
| 61 | t.Fatalf("session/status: %+v", resp.Error) |
| 62 | } |
| 63 | var status ReasonixSessionStatus |
| 64 | if err := json.Unmarshal(resp.Result, &status); err != nil { |
| 65 | t.Fatalf("session/status result: %v", err) |
| 66 | } |
| 67 | return status |
| 68 | } |
| 69 | |
| 70 | func TestStatusExtensionTracksMultipleSessionsAndUsage(t *testing.T) { |
| 71 | factory := &statusFactory{configurableFactory: &configurableFactory{ |
| 72 | behavior: func(_ context.Context, sink event.Sink, input string, _ SessionParams) error { |
| 73 | sink.Emit(event.Event{Kind: event.Phase, Source: event.UsageSourceExecutor, Text: "executor · implementing"}) |
| 74 | sink.Emit(event.Event{Kind: event.Usage, Usage: &provider.Usage{ |
| 75 | PromptTokens: 10, CompletionTokens: 4, ReasoningTokens: 2, |
| 76 | CacheHitTokens: 7, CacheMissTokens: 3, Estimated: true, |
| 77 | }, Pricing: &provider.Pricing{CacheHit: 0.1, Input: 1, Output: 2, Currency: "USD"}, UsageSource: event.UsageSourceExecutor}) |
| 78 | sink.Emit(event.Event{Kind: event.Usage, Usage: &provider.Usage{ |
| 79 | PromptTokens: 5, CompletionTokens: 1, CacheMissTokens: 5, |
| 80 | }, Pricing: &provider.Pricing{CacheHit: 0.1, Input: 1, Output: 2, Currency: "USD"}, UsageSource: event.UsageSourceCompaction}) |
| 81 | sink.Emit(event.Event{Kind: event.Text, Text: input}) |
| 82 | return nil |
| 83 | }, |
| 84 | }} |
| 85 | client, stop := startServer(t, factory) |
| 86 | defer stop() |
| 87 | client.call(t, "initialize", InitializeParams{ProtocolVersion: 1}) |
| 88 | first := openStatusSession(t, client, t.TempDir()) |
| 89 | second := openStatusSession(t, client, t.TempDir()) |
| 90 | |
| 91 | initialSecond := getStatus(t, client, second) |
| 92 | prompt := client.callAsync("session/prompt", SessionPromptParams{SessionID: first, Prompt: []ContentBlock{{Type: "text", Text: "ship"}}}) |
| 93 | notifications, response := drainPrompt(t, client, prompt) |
| 94 | if response.Error != nil { |
| 95 | t.Fatalf("session/prompt: %+v", response.Error) |
| 96 | } |
| 97 | |
| 98 | firstStatus := getStatus(t, client, first) |
| 99 | if firstStatus.Sequence == 0 || firstStatus.State != "idle" || firstStatus.TurnOutcome.Kind != "completed" { |
| 100 | t.Fatalf("first status = %+v", firstStatus) |
| 101 | } |
| 102 | if firstStatus.PlannerMode != "off" || firstStatus.Sandbox.WorkspaceRoot == "" || len(firstStatus.Sandbox.WriteRoots) != 1 { |
| 103 | t.Fatalf("effective runtime status = %+v", firstStatus) |
| 104 | } |
| 105 | usage := firstStatus.Usage.Cumulative |
| 106 | if usage.PromptTokens != 15 || usage.CompletionTokens != 5 || usage.ReasoningTokens != 2 || usage.CacheHitTokens != 7 || usage.CacheMissTokens != 8 { |
| 107 | t.Fatalf("cumulative usage = %+v", usage) |
| 108 | } |
| 109 | if usage.UsageSource != "mixed" || usage.CacheHitRatio == nil || usage.EstimatedCost == nil || usage.Currency == nil || *usage.Currency != "USD" { |
| 110 | t.Fatalf("usage metadata = %+v", usage) |
| 111 | } |
| 112 | if !usage.Estimated { |
| 113 | t.Fatalf("cumulative usage lost estimated marker: %+v", usage) |
| 114 | } |
| 115 | secondStatus := getStatus(t, client, second) |
| 116 | if secondStatus.Sequence != initialSecond.Sequence || secondStatus.Usage.Cumulative.PromptTokens != 0 { |
| 117 | t.Fatalf("second session telemetry leaked: before=%+v after=%+v", initialSecond, secondStatus) |
| 118 | } |
| 119 | |
| 120 | var sawPhase, sawUsage, sawCompletion bool |
| 121 | for _, notification := range notifications { |
| 122 | if notification.Method != sessionStatusUpdateMethod { |
| 123 | continue |
| 124 | } |
| 125 | var update ReasonixStatusUpdate |
| 126 | if err := json.Unmarshal(notification.Params, &update); err != nil { |
| 127 | t.Fatalf("status update: %v", err) |
| 128 | } |
| 129 | if update.Sequence != update.Status.Sequence || update.SessionID != first { |
| 130 | t.Fatalf("status update correlation = %+v", update) |
| 131 | } |
| 132 | switch update.Event { |
| 133 | case "phase": |
| 134 | sawPhase = true |
| 135 | case "usage": |
| 136 | sawUsage = true |
| 137 | case "completion": |
| 138 | sawCompletion = true |
| 139 | } |
| 140 | } |
| 141 | if !sawPhase || !sawUsage || !sawCompletion { |
| 142 | t.Fatalf("status events phase=%v usage=%v completion=%v", sawPhase, sawUsage, sawCompletion) |
| 143 | } |
| 144 | } |
| 145 | |
| 146 | func TestStatusNormalizesPhaseAndRedactsPublicText(t *testing.T) { |
| 147 | telemetry := newStatusTelemetry() |
| 148 | telemetry.beginTurn() |
| 149 | telemetry.onEvent(event.Event{Kind: event.Phase, Source: event.UsageSourcePlanner, Text: "planner · private stage label"}) |
| 150 | if got := telemetry.snapshot().phase; got != "planning" { |
| 151 | t.Fatalf("planner phase = %q, want planning", got) |
| 152 | } |
| 153 | telemetry.onEvent(event.Event{Kind: event.Phase, Text: "provider-specific handoff"}) |
| 154 | if got := telemetry.snapshot().phase; got != "working" { |
| 155 | t.Fatalf("unknown phase = %q, want working", got) |
| 156 | } |
| 157 | telemetry.finishTurn(&agent.FinalReadinessError{ |
| 158 | Attempts: 1, |
| 159 | Reason: "token=secret-reason", |
| 160 | Missing: []string{"api_key=secret-risk"}, |
| 161 | }, false, "running", "authorization: bearer secret-summary") |
| 162 | snapshot := telemetry.snapshot() |
| 163 | encoded, err := json.Marshal(snapshot.finalReadiness) |
| 164 | if err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | if strings.Contains(string(encoded), "secret-") || !strings.Contains(string(encoded), "[redacted]") { |
| 168 | t.Fatalf("status text was not redacted: %s", encoded) |
| 169 | } |
| 170 | if strings.Contains(snapshot.turnOutcome.Reason, "secret-") { |
| 171 | t.Fatalf("turn outcome was not redacted: %q", snapshot.turnOutcome.Reason) |
| 172 | } |
| 173 | |
| 174 | empty, err := json.Marshal(newStatusTelemetry().snapshot().finalReadiness) |
| 175 | if err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | if !strings.Contains(string(empty), `"risks":[]`) { |
| 179 | t.Fatalf("empty risks must encode as [], got %s", empty) |
| 180 | } |
| 181 | } |
| 182 | |
| 183 | func TestRestoreStatusNormalizesLegacyPresentationPhase(t *testing.T) { |
| 184 | restored := restoreStatusTelemetry(&persistedStatusTelemetry{ |
| 185 | Phase: "executor · implementing local patch", |
| 186 | FinalReadiness: ReasonixFinalReadiness{}, |
| 187 | }) |
| 188 | if got := restored.snapshot().phase; got != "implementing" { |
| 189 | t.Fatalf("restored phase = %q, want implementing", got) |
| 190 | } |
| 191 | } |
| 192 | |
| 193 | func TestRestoreStatusMarksInterruptedTurnPaused(t *testing.T) { |
| 194 | restored := restoreStatusTelemetry(&persistedStatusTelemetry{ |
| 195 | Sequence: 7, |
| 196 | State: "running", |
| 197 | Phase: "implementing", |
| 198 | TurnOutcome: ReasonixTurnOutcome{Kind: "none"}, |
| 199 | FinalReadiness: ReasonixFinalReadiness{ |
| 200 | ReadyForReview: true, |
| 201 | Risks: []string{}, |
| 202 | }, |
| 203 | TurnUsage: persistedUsageAccumulator{PromptTokens: 3, Estimated: true, Events: 1}, |
| 204 | Cumulative: persistedUsageAccumulator{PromptTokens: 11, Estimated: true, Events: 2}, |
| 205 | }) |
| 206 | snapshot := restored.snapshot() |
| 207 | if snapshot.state != "idle" || snapshot.phase != "recovery_paused" { |
| 208 | t.Fatalf("restored interrupted state = state:%q phase:%q, want idle/recovery_paused", snapshot.state, snapshot.phase) |
| 209 | } |
| 210 | if snapshot.sequence != 8 || snapshot.turnOutcome.Kind != "paused" || snapshot.turnOutcome.Reason != "previous turn interrupted" { |
| 211 | t.Fatalf("restored interrupted outcome = sequence:%d outcome:%+v", snapshot.sequence, snapshot.turnOutcome) |
| 212 | } |
| 213 | if snapshot.finalReadiness.ReadyForReview { |
| 214 | t.Fatal("interrupted turn remained ready for review") |
| 215 | } |
| 216 | if snapshot.turnUsage.PromptTokens != 3 || snapshot.cumulative.PromptTokens != 11 { |
| 217 | t.Fatalf("interrupted usage was lost: turn=%+v cumulative=%+v", snapshot.turnUsage, snapshot.cumulative) |
| 218 | } |
| 219 | if !snapshot.turnUsage.Estimated || !snapshot.cumulative.Estimated { |
| 220 | t.Fatalf("interrupted estimated marker was lost: turn=%+v cumulative=%+v", snapshot.turnUsage, snapshot.cumulative) |
| 221 | } |
| 222 | } |
| 223 | |
| 224 | func TestStatusRecomputesPlannerModeAfterWorkModeSwitch(t *testing.T) { |
| 225 | factory := &runtimeTrackingFactory{configurableFactory: &configurableFactory{}} |
| 226 | client, stop := startServer(t, factory) |
| 227 | defer stop() |
| 228 | client.call(t, "initialize", InitializeParams{ProtocolVersion: 1}) |
| 229 | sessionID := openStatusSession(t, client, t.TempDir()) |
| 230 | if status := getStatus(t, client, sessionID); status.WorkMode != "balanced" || status.PlannerMode != "on" { |
| 231 | t.Fatalf("initial runtime status = %+v", status) |
| 232 | } |
| 233 | |
| 234 | for _, tc := range []struct { |
| 235 | profile string |
| 236 | planner string |
| 237 | }{ |
| 238 | {profile: "economy", planner: "off"}, |
| 239 | {profile: "delivery", planner: "on"}, |
| 240 | } { |
| 241 | resp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{ |
| 242 | SessionID: sessionID, |
| 243 | ConfigID: "work_mode", |
| 244 | Value: tc.profile, |
| 245 | }) |
| 246 | if resp.Error != nil { |
| 247 | t.Fatalf("set work mode %q: %+v", tc.profile, resp.Error) |
| 248 | } |
| 249 | status := getStatus(t, client, sessionID) |
| 250 | if status.WorkMode != tc.profile || status.PlannerMode != tc.planner { |
| 251 | t.Fatalf("runtime status after %q = %+v", tc.profile, status) |
| 252 | } |
| 253 | } |
| 254 | } |
| 255 | |
| 256 | func TestStatusClassifiesPauseAndError(t *testing.T) { |
| 257 | telemetry := newStatusTelemetry() |
| 258 | telemetry.beginTurn() |
| 259 | pauseEvent := telemetry.finishTurn(&agent.FinalReadinessError{Attempts: 3, Reason: "missing verification", Missing: []string{"verify"}}, false, "running", "partial") |
| 260 | paused := telemetry.snapshot() |
| 261 | if pauseEvent != "pause" || paused.turnOutcome.Kind != "paused" || len(paused.finalReadiness.Risks) != 1 { |
| 262 | t.Fatalf("pause classification = event %q snapshot %+v", pauseEvent, paused) |
| 263 | } |
| 264 | |
| 265 | telemetry.beginTurn() |
| 266 | errorEvent := telemetry.finishTurn(errors.New("provider failed"), false, "running", "") |
| 267 | failed := telemetry.snapshot() |
| 268 | if errorEvent != "error" || failed.turnOutcome.Kind != "error" || failed.goalOverride != "failed" { |
| 269 | t.Fatalf("error classification = event %q snapshot %+v", errorEvent, failed) |
| 270 | } |
| 271 | } |
| 272 | |
| 273 | func TestStatusSnapshotSurvivesSessionResume(t *testing.T) { |
| 274 | dir := t.TempDir() |
| 275 | cwd := t.TempDir() |
| 276 | sessionID := "status-reconnect" |
| 277 | telemetry := newStatusTelemetry() |
| 278 | telemetry.beginTurn() |
| 279 | telemetry.onEvent(event.Event{Kind: event.Usage, Usage: &provider.Usage{ |
| 280 | PromptTokens: 8, CompletionTokens: 2, CacheHitTokens: 6, CacheMissTokens: 2, |
| 281 | }, UsageSource: event.UsageSourceExecutor}) |
| 282 | telemetry.finishTurn(nil, false, "", "persisted summary") |
| 283 | path := filepath.Join(dir, sessionID+".jsonl") |
| 284 | if err := agent.NewSession("system").Save(path); err != nil { |
| 285 | t.Fatalf("save transcript: %v", err) |
| 286 | } |
| 287 | if err := saveACPMeta(path, acpSessionMeta{ |
| 288 | SessionID: sessionID, Cwd: cwd, Model: "fast", RuntimeProfile: "delivery", |
| 289 | Status: telemetry.persisted(), |
| 290 | }); err != nil { |
| 291 | t.Fatalf("save ACP metadata: %v", err) |
| 292 | } |
| 293 | factory := &statusFactory{configurableFactory: &configurableFactory{ |
| 294 | dir: dir, |
| 295 | }} |
| 296 | reconnected, stopReconnected := startServer(t, factory) |
| 297 | defer stopReconnected() |
| 298 | reconnected.call(t, "initialize", InitializeParams{ProtocolVersion: 1}) |
| 299 | resume := reconnected.call(t, "session/resume", SessionResumeParams{SessionID: sessionID, Cwd: cwd}) |
| 300 | if resume.Error != nil { |
| 301 | t.Fatalf("session/resume: %+v", resume.Error) |
| 302 | } |
| 303 | after := getStatus(t, reconnected, sessionID) |
| 304 | if after.Sequence != telemetry.snapshot().sequence || after.Usage.Cumulative.PromptTokens != 8 || after.State != "idle" || after.FinalReadiness.Summary != "persisted summary" { |
| 305 | t.Fatalf("recovered status = %+v", after) |
| 306 | } |
| 307 | } |
| 308 | |
| 309 | func TestStatusInterruptedSnapshotResumesPaused(t *testing.T) { |
| 310 | dir := t.TempDir() |
| 311 | cwd := t.TempDir() |
| 312 | sessionID := "status-interrupted" |
| 313 | telemetry := newStatusTelemetry() |
| 314 | telemetry.beginTurn() |
| 315 | telemetry.onEvent(event.Event{Kind: event.Usage, Usage: &provider.Usage{ |
| 316 | PromptTokens: 5, CompletionTokens: 1, |
| 317 | }, UsageSource: event.UsageSourceExecutor}) |
| 318 | path := filepath.Join(dir, sessionID+".jsonl") |
| 319 | if err := agent.NewSession("system").Save(path); err != nil { |
| 320 | t.Fatalf("save transcript: %v", err) |
| 321 | } |
| 322 | if err := saveACPMeta(path, acpSessionMeta{ |
| 323 | SessionID: sessionID, Cwd: cwd, Model: "fast", RuntimeProfile: "balanced", |
| 324 | Status: telemetry.persisted(), |
| 325 | }); err != nil { |
| 326 | t.Fatalf("save ACP metadata: %v", err) |
| 327 | } |
| 328 | |
| 329 | factory := &statusFactory{configurableFactory: &configurableFactory{dir: dir}} |
| 330 | client, stop := startServer(t, factory) |
| 331 | defer stop() |
| 332 | client.call(t, "initialize", InitializeParams{ProtocolVersion: 1}) |
| 333 | resume := client.call(t, "session/resume", SessionResumeParams{SessionID: sessionID, Cwd: cwd}) |
| 334 | if resume.Error != nil { |
| 335 | t.Fatalf("session/resume: %+v", resume.Error) |
| 336 | } |
| 337 | after := getStatus(t, client, sessionID) |
| 338 | if after.State != "idle" || after.Phase != "recovery_paused" || after.TurnOutcome.Kind != "paused" { |
| 339 | t.Fatalf("resumed interrupted status = %+v", after) |
| 340 | } |
| 341 | if after.Sequence != telemetry.snapshot().sequence+1 || after.Usage.Turn.PromptTokens != 5 || after.Usage.Cumulative.PromptTokens != 5 { |
| 342 | t.Fatalf("resumed interrupted sequence/usage = %+v", after) |
| 343 | } |
| 344 | } |
| 345 |