| 1 | package cli |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "strings" |
| 9 | "sync" |
| 10 | "time" |
| 11 | |
| 12 | "reasonix/internal/billing" |
| 13 | "reasonix/internal/control" |
| 14 | "reasonix/internal/event" |
| 15 | "reasonix/internal/eventwire" |
| 16 | ) |
| 17 | |
| 18 | type runOutputFormat string |
| 19 | |
| 20 | const ( |
| 21 | runOutputText runOutputFormat = "text" |
| 22 | runOutputJSON runOutputFormat = "json" |
| 23 | runOutputStreamJSON runOutputFormat = "stream-json" |
| 24 | runOutputEventsJSONL runOutputFormat = "events-jsonl" |
| 25 | ) |
| 26 | |
| 27 | func parseRunOutputFormat(value string) (runOutputFormat, error) { |
| 28 | switch runOutputFormat(strings.ToLower(strings.TrimSpace(value))) { |
| 29 | case runOutputText: |
| 30 | return runOutputText, nil |
| 31 | case runOutputJSON: |
| 32 | return runOutputJSON, nil |
| 33 | case runOutputStreamJSON: |
| 34 | return runOutputStreamJSON, nil |
| 35 | default: |
| 36 | return "", fmt.Errorf("unknown output format %q (want text, json, or stream-json)", value) |
| 37 | } |
| 38 | } |
| 39 | |
| 40 | // runOutputSessionID preserves the established json/stream-json contract while |
| 41 | // keeping the redacted events-jsonl surface independent from transcript names. |
| 42 | func runOutputSessionID(format runOutputFormat, rawSessionID string, identityKey []byte) string { |
| 43 | if format == runOutputEventsJSONL { |
| 44 | return machineSessionIDWithKey(rawSessionID, identityKey) |
| 45 | } |
| 46 | return rawSessionID |
| 47 | } |
| 48 | |
| 49 | type runResultUsage struct { |
| 50 | InputTokens int `json:"input_tokens"` |
| 51 | OutputTokens int `json:"output_tokens"` |
| 52 | CacheReadInputTokens int `json:"cache_read_input_tokens"` |
| 53 | CacheCreationInputTokens int `json:"cache_creation_input_tokens"` |
| 54 | Estimated bool `json:"estimated,omitempty"` |
| 55 | } |
| 56 | |
| 57 | type runResult struct { |
| 58 | Type string `json:"type"` |
| 59 | Subtype string `json:"subtype"` |
| 60 | IsError bool `json:"is_error"` |
| 61 | DurationMS int64 `json:"duration_ms"` |
| 62 | NumTurns int `json:"num_turns"` |
| 63 | Result string `json:"result"` |
| 64 | SessionID string `json:"session_id,omitempty"` |
| 65 | TotalCost float64 `json:"total_cost,omitempty"` |
| 66 | Currency string `json:"currency,omitempty"` |
| 67 | // TotalCostUSD is the released compatibility alias. It mirrors TotalCost; |
| 68 | // new consumers must pair TotalCost with Currency instead of assuming USD. |
| 69 | TotalCostUSD float64 `json:"total_cost_usd,omitempty"` |
| 70 | // CostComplete is false when mixed originals lack a shared display valuation. |
| 71 | CostComplete bool `json:"cost_complete"` |
| 72 | DisplayComplete bool `json:"display_complete"` |
| 73 | DisplayStatus string `json:"display_status,omitempty"` |
| 74 | AggregateMode string `json:"aggregate_mode,omitempty"` |
| 75 | // OriginalCosts lists per-ISO original totals (never cross-added). |
| 76 | OriginalCosts map[string]float64 `json:"original_costs,omitempty"` |
| 77 | OriginalTotals []billing.Money `json:"original_totals,omitempty"` |
| 78 | CostQuote *billing.CostQuote `json:"cost_quote,omitempty"` |
| 79 | Usage runResultUsage `json:"usage"` |
| 80 | ErrorCode string `json:"error_code,omitempty"` |
| 81 | Authentication string `json:"authentication_status,omitempty"` |
| 82 | Recovery []string `json:"recovery_actions,omitempty"` |
| 83 | } |
| 84 | |
| 85 | type machineEventUsage struct { |
| 86 | InputTokens int `json:"input_tokens"` |
| 87 | OutputTokens int `json:"output_tokens"` |
| 88 | CacheHitTokens int `json:"cache_hit_tokens"` |
| 89 | CacheMissTokens int `json:"cache_miss_tokens"` |
| 90 | Estimated bool `json:"estimated,omitempty"` |
| 91 | } |
| 92 | |
| 93 | // machineEventRecord is deliberately content-free. The existing stream-json |
| 94 | // format is a rich UI transport and includes prompts, tool arguments, results, |
| 95 | // and reasoning; this contract is for automation that must not receive them. |
| 96 | type machineEventRecord struct { |
| 97 | SchemaVersion int `json:"schema_version"` |
| 98 | Sequence uint64 `json:"sequence"` |
| 99 | Kind string `json:"kind"` |
| 100 | Code string `json:"code,omitempty"` |
| 101 | Level string `json:"level,omitempty"` |
| 102 | ToolID string `json:"tool_id,omitempty"` |
| 103 | ToolName string `json:"tool_name,omitempty"` |
| 104 | ToolReadOnly bool `json:"tool_read_only,omitempty"` |
| 105 | ToolError bool `json:"tool_error,omitempty"` |
| 106 | ToolTruncated bool `json:"tool_truncated,omitempty"` |
| 107 | ToolDurationMS int64 `json:"tool_duration_ms,omitempty"` |
| 108 | Usage *machineEventUsage `json:"usage,omitempty"` |
| 109 | ApprovalID string `json:"approval_id,omitempty"` |
| 110 | ApprovalKind string `json:"approval_kind,omitempty"` |
| 111 | AskID string `json:"ask_id,omitempty"` |
| 112 | Outcome string `json:"outcome,omitempty"` |
| 113 | Cancelled bool `json:"cancelled,omitempty"` |
| 114 | Error bool `json:"error,omitempty"` |
| 115 | Recovery *event.RecoveryStatus `json:"recovery,omitempty"` |
| 116 | RetryAttempt int `json:"retry_attempt,omitempty"` |
| 117 | RetryMax int `json:"retry_max,omitempty"` |
| 118 | CompactionType string `json:"compaction_type,omitempty"` |
| 119 | CompactionMsgs int `json:"compaction_messages,omitempty"` |
| 120 | GuardianResult string `json:"guardian_result,omitempty"` |
| 121 | GuardianRisk string `json:"guardian_risk,omitempty"` |
| 122 | } |
| 123 | |
| 124 | type machineRunDone struct { |
| 125 | SchemaVersion int `json:"schema_version"` |
| 126 | Sequence uint64 `json:"sequence"` |
| 127 | Kind string `json:"kind"` |
| 128 | SessionID string `json:"session_id,omitempty"` |
| 129 | OK bool `json:"ok"` |
| 130 | DurationMS int64 `json:"duration_ms"` |
| 131 | NumTurns int `json:"num_turns"` |
| 132 | Usage machineEventUsage `json:"usage"` |
| 133 | ErrorCode string `json:"error_code,omitempty"` |
| 134 | Authentication string `json:"authentication_status,omitempty"` |
| 135 | Recovery []string `json:"recovery_actions,omitempty"` |
| 136 | } |
| 137 | |
| 138 | func runAuthenticationMetadata(err error) (code, status string, actions []string) { |
| 139 | var authErr *control.AuthenticationError |
| 140 | if !errors.As(err, &authErr) || authErr == nil { |
| 141 | return "", "", nil |
| 142 | } |
| 143 | state := authErr.State |
| 144 | code = state.Code |
| 145 | if code == "" { |
| 146 | code = string(state.Status) |
| 147 | } |
| 148 | status = string(state.Status) |
| 149 | switch state.Status { |
| 150 | case control.AuthenticationMissingCredential: |
| 151 | actions = []string{"configure_credentials", "select_model", "diagnose_credentials"} |
| 152 | case control.AuthenticationRejected: |
| 153 | actions = []string{"update_credentials", "select_model", "test_connection", "retry_authentication"} |
| 154 | case control.AuthenticationCredentialStoreUnavailable: |
| 155 | actions = []string{"diagnose_credentials", "select_model"} |
| 156 | } |
| 157 | return code, status, actions |
| 158 | } |
| 159 | |
| 160 | type runOutputSink struct { |
| 161 | mu sync.Mutex |
| 162 | format runOutputFormat |
| 163 | out io.Writer |
| 164 | encoder *json.Encoder |
| 165 | final string |
| 166 | usage runResultUsage |
| 167 | cost float64 |
| 168 | currency string |
| 169 | costComplete bool |
| 170 | displayComplete bool |
| 171 | displayStatus string |
| 172 | aggregateMode string |
| 173 | originalTotals []billing.Money |
| 174 | sawQuote bool |
| 175 | originalCosts map[string]float64 |
| 176 | quoteLedger *billing.Ledger |
| 177 | turns int |
| 178 | sequence uint64 |
| 179 | machineToolIDs map[string]string |
| 180 | machineToolNames map[string]string |
| 181 | nextMachineToolID uint64 |
| 182 | nextMachineToolName uint64 |
| 183 | err error |
| 184 | } |
| 185 | |
| 186 | func newRunOutputSink(out io.Writer, format runOutputFormat) *runOutputSink { |
| 187 | return &runOutputSink{ |
| 188 | format: format, |
| 189 | out: out, |
| 190 | encoder: json.NewEncoder(out), |
| 191 | machineToolIDs: make(map[string]string), |
| 192 | machineToolNames: make(map[string]string), |
| 193 | } |
| 194 | } |
| 195 | |
| 196 | func (s *runOutputSink) Emit(e event.Event) { |
| 197 | s.mu.Lock() |
| 198 | defer s.mu.Unlock() |
| 199 | if e.Kind == event.Message { |
| 200 | s.final = e.Text |
| 201 | } |
| 202 | if e.Kind == event.Usage && e.Usage != nil { |
| 203 | s.usage.InputTokens += e.Usage.PromptTokens |
| 204 | s.usage.OutputTokens += e.Usage.CompletionTokens |
| 205 | s.usage.CacheReadInputTokens += e.Usage.CacheHitTokens |
| 206 | s.usage.CacheCreationInputTokens += e.Usage.CacheMissTokens |
| 207 | s.usage.Estimated = s.usage.Estimated || e.Usage.Estimated |
| 208 | q := e.CostQuote |
| 209 | if q == nil && e.Pricing != nil { |
| 210 | q = event.EnsureCostQuote(e, nil) |
| 211 | } |
| 212 | if q != nil { |
| 213 | s.sawQuote = true |
| 214 | if !q.CostComplete { |
| 215 | s.costComplete = false |
| 216 | } |
| 217 | // First complete quote establishes complete=true. |
| 218 | if q.Complete && s.quoteLedger == nil { |
| 219 | s.costComplete = true |
| 220 | } |
| 221 | if s.originalCosts == nil { |
| 222 | s.originalCosts = map[string]float64{} |
| 223 | } |
| 224 | if cur := billing.NormalizeCurrency(q.Original.Currency); cur != "" { |
| 225 | s.originalCosts[cur] += q.Original.Float64() |
| 226 | } |
| 227 | if q.Selected != nil && (s.currency == "" || s.currency == q.LegacyCurrencyCode()) { |
| 228 | s.cost += q.Selected.Float64() |
| 229 | s.currency = q.LegacyCurrencyCode() |
| 230 | } else if q.Selected != nil { |
| 231 | s.currency = "" |
| 232 | s.cost = 0 |
| 233 | } |
| 234 | if s.quoteLedger == nil { |
| 235 | s.quoteLedger = billing.NewLedger() |
| 236 | } |
| 237 | s.quoteLedger.Add(*q, billing.UsageTokens{ |
| 238 | PromptTokens: e.Usage.PromptTokens, |
| 239 | CompletionTokens: e.Usage.CompletionTokens, |
| 240 | CacheHitTokens: e.Usage.CacheHitTokens, |
| 241 | CacheMissTokens: e.Usage.CacheMissTokens, |
| 242 | CacheWriteTokens: e.Usage.CacheWriteTokens, |
| 243 | CacheWriteBilledTokens: e.Usage.CacheWriteBilledTokens, |
| 244 | Estimated: e.Usage.Estimated, |
| 245 | }, time.Now().UTC()) |
| 246 | } |
| 247 | } |
| 248 | if e.Kind == event.TurnDone { |
| 249 | s.turns++ |
| 250 | } |
| 251 | if s.format == runOutputStreamJSON && s.err == nil { |
| 252 | s.err = s.encoder.Encode(eventwire.ToWire(e)) |
| 253 | } else if s.format == runOutputEventsJSONL && s.err == nil { |
| 254 | s.sequence++ |
| 255 | s.err = s.encoder.Encode(s.machineEventRecordFor(e, s.sequence)) |
| 256 | } |
| 257 | } |
| 258 | |
| 259 | func (s *runOutputSink) Finalize(sessionID string, started time.Time, runErr error) error { |
| 260 | s.mu.Lock() |
| 261 | defer s.mu.Unlock() |
| 262 | if s.err != nil { |
| 263 | return s.err |
| 264 | } |
| 265 | // Mixed original currencies no longer error: totals use shared display |
| 266 | // valuations when complete, otherwise cost_complete=false with original_costs. |
| 267 | if s.format == runOutputText { |
| 268 | if s.final != "" { |
| 269 | _, s.err = fmt.Fprintln(s.out, s.final) |
| 270 | } |
| 271 | return s.err |
| 272 | } |
| 273 | completion := classifyRunCompletion(runErr) |
| 274 | errorCode, authentication, recovery := runAuthenticationMetadata(runErr) |
| 275 | if s.format == runOutputEventsJSONL { |
| 276 | s.sequence++ |
| 277 | turns := s.turns |
| 278 | if turns == 0 && !completion.isError { |
| 279 | turns = 1 |
| 280 | } |
| 281 | return s.encoder.Encode(machineRunDone{ |
| 282 | SchemaVersion: machineSchemaVersion, |
| 283 | Sequence: s.sequence, |
| 284 | Kind: "run_done", |
| 285 | SessionID: sessionID, |
| 286 | OK: !completion.isError, |
| 287 | DurationMS: time.Since(started).Milliseconds(), |
| 288 | NumTurns: turns, |
| 289 | Usage: machineEventUsage{InputTokens: s.usage.InputTokens, OutputTokens: s.usage.OutputTokens, CacheHitTokens: s.usage.CacheReadInputTokens, CacheMissTokens: s.usage.CacheCreationInputTokens}, |
| 290 | ErrorCode: errorCode, |
| 291 | Authentication: authentication, |
| 292 | Recovery: recovery, |
| 293 | }) |
| 294 | } |
| 295 | resultText := s.final |
| 296 | if runErr != nil { |
| 297 | if resultText == "" { |
| 298 | resultText = runErr.Error() |
| 299 | } |
| 300 | } |
| 301 | turns := s.turns |
| 302 | if turns == 0 && !completion.isError { |
| 303 | turns = 1 |
| 304 | } |
| 305 | var aggQuote *billing.CostQuote |
| 306 | if s.quoteLedger != nil && len(s.quoteLedger.Entries) > 0 { |
| 307 | agg := s.quoteLedger.Total("") |
| 308 | aggQuote = &agg |
| 309 | if agg.Selected != nil { |
| 310 | s.cost = agg.Selected.Float64() |
| 311 | s.currency = agg.LegacyCurrencyCode() |
| 312 | } |
| 313 | if agg.Selected == nil { |
| 314 | s.cost = 0 |
| 315 | s.currency = "" |
| 316 | } |
| 317 | s.costComplete = agg.CostComplete |
| 318 | s.displayComplete = agg.DisplayComplete |
| 319 | s.displayStatus = agg.DisplayStatus |
| 320 | s.aggregateMode = agg.AggregateMode |
| 321 | if agg.OriginalTotals != nil { |
| 322 | s.originalTotals = append([]billing.Money(nil), agg.OriginalTotals...) |
| 323 | } |
| 324 | } |
| 325 | return s.encoder.Encode(runResult{ |
| 326 | Type: "result", |
| 327 | Subtype: completion.subtype, |
| 328 | IsError: completion.isError, |
| 329 | DurationMS: time.Since(started).Milliseconds(), |
| 330 | NumTurns: turns, |
| 331 | Result: resultText, |
| 332 | SessionID: sessionID, |
| 333 | TotalCost: s.cost, |
| 334 | Currency: s.currency, |
| 335 | TotalCostUSD: s.cost, |
| 336 | CostComplete: s.costComplete || (!s.sawQuote && s.currency != ""), |
| 337 | DisplayComplete: s.displayComplete, |
| 338 | DisplayStatus: s.displayStatus, |
| 339 | AggregateMode: s.aggregateMode, |
| 340 | OriginalCosts: s.originalCosts, |
| 341 | OriginalTotals: s.originalTotals, |
| 342 | CostQuote: aggQuote, |
| 343 | Usage: s.usage, |
| 344 | ErrorCode: errorCode, |
| 345 | Authentication: authentication, |
| 346 | Recovery: recovery, |
| 347 | }) |
| 348 | } |
| 349 | |
| 350 | func (s *runOutputSink) machineEventRecordFor(e event.Event, sequence uint64) machineEventRecord { |
| 351 | record := machineEventRecord{SchemaVersion: machineSchemaVersion, Sequence: sequence, Kind: machineEventKind(e.Kind)} |
| 352 | switch e.Kind { |
| 353 | case event.Notice: |
| 354 | record.Code = e.Code |
| 355 | if e.Level == event.LevelWarn { |
| 356 | record.Level = "warn" |
| 357 | } else { |
| 358 | record.Level = "info" |
| 359 | } |
| 360 | case event.ToolDispatch, event.ToolResult, event.ToolProgress: |
| 361 | // Tool-call IDs and names originate in the provider stream. Treat both as |
| 362 | // untrusted content: an OpenAI-compatible endpoint may put arbitrary prompt |
| 363 | // or argument text in either field. Per-run opaque aliases preserve event |
| 364 | // correlation without exposing the provider-controlled values. |
| 365 | record.ToolID = machineOpaqueValue(s.machineToolIDs, &s.nextMachineToolID, "tool", e.Tool.ID) |
| 366 | record.ToolName = machineOpaqueValue(s.machineToolNames, &s.nextMachineToolName, "tool_name", e.Tool.Name) |
| 367 | record.ToolReadOnly = e.Tool.ReadOnly |
| 368 | record.ToolError = e.Tool.Err != "" |
| 369 | record.ToolTruncated = e.Tool.Truncated |
| 370 | record.ToolDurationMS = e.Tool.DurationMs |
| 371 | case event.Usage: |
| 372 | if e.Usage != nil { |
| 373 | record.Usage = &machineEventUsage{InputTokens: e.Usage.PromptTokens, OutputTokens: e.Usage.CompletionTokens, CacheHitTokens: e.Usage.CacheHitTokens, CacheMissTokens: e.Usage.CacheMissTokens, Estimated: e.Usage.Estimated} |
| 374 | } |
| 375 | case event.ApprovalRequest: |
| 376 | record.ApprovalID = e.Approval.ID |
| 377 | record.ApprovalKind = e.Approval.Kind |
| 378 | case event.AskRequest: |
| 379 | record.AskID = e.Ask.ID |
| 380 | case event.TurnDone: |
| 381 | record.Outcome = e.Outcome |
| 382 | record.Cancelled = e.Cancelled |
| 383 | record.Error = e.Err != nil |
| 384 | case event.CompactionStarted, event.CompactionDone: |
| 385 | record.CompactionType = e.Compaction.Trigger |
| 386 | record.CompactionMsgs = e.Compaction.Messages |
| 387 | case event.GuardianAssessment: |
| 388 | record.GuardianResult = e.Guardian.Outcome |
| 389 | record.GuardianRisk = e.Guardian.RiskLevel |
| 390 | case event.Retrying: |
| 391 | record.Recovery = e.Recovery |
| 392 | record.RetryAttempt = e.RetryAttempt |
| 393 | record.RetryMax = e.RetryMax |
| 394 | } |
| 395 | return record |
| 396 | } |
| 397 | |
| 398 | func machineOpaqueValue(values map[string]string, next *uint64, prefix, value string) string { |
| 399 | if value == "" { |
| 400 | return "" |
| 401 | } |
| 402 | if opaque := values[value]; opaque != "" { |
| 403 | return opaque |
| 404 | } |
| 405 | (*next)++ |
| 406 | opaque := fmt.Sprintf("%s_%d", prefix, *next) |
| 407 | values[value] = opaque |
| 408 | return opaque |
| 409 | } |
| 410 | |
| 411 | func machineEventKind(kind event.Kind) string { |
| 412 | names := eventwire.KindNames() |
| 413 | if int(kind) >= 0 && int(kind) < len(names) { |
| 414 | return names[kind] |
| 415 | } |
| 416 | return "unknown" |
| 417 | } |
| 418 |