package harness import ( "bytes" "context" "database/sql" "encoding/json" "errors" "fmt" "strconv" "strings" "time" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/llm" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store" ) // RunStatus is the outcome of a run, stored in ai_runs.status. type RunStatus string // The run outcomes. const ( // StatusOK means the first reply was usable. StatusOK RunStatus = "ok" // StatusRepaired means the first reply was not usable and the reply to // the repair turn was. StatusRepaired RunStatus = "repaired" // StatusBadOutput means neither reply was usable, post-processing a // valid reply failed, or the skill's prompt could not be rendered. StatusBadOutput RunStatus = "bad_output" // StatusProviderError means the provider could not be built or failed // on either turn. StatusProviderError RunStatus = "provider_error" ) // RunRetention is how long ai_runs rows are kept. The daily purge, // PurgeSchedule, calls Purge with its fire time less RunRetention. const RunRetention = 30 * 24 * time.Hour // runSaveTimeout bounds writing the ai_runs row of a run that has failed, // timed out or been cancelled, so a blocked writer cannot hold Run up // indefinitely. A variable so tests can shorten it. var runSaveTimeout = 5 * time.Second // Purge deletes the ai_runs rows created before olderThan and returns how // many it deleted. The suggestion_batches and week_plans references to // those rows are cleared first, in the same transaction. func Purge(ctx context.Context, st *store.Store, olderThan time.Time) (int64, error) { if olderThan.IsZero() { return 0, errors.New("harness: purge ai_runs: zero cutoff") } cutoff := store.Timestamp{Time: olderThan.Truncate(time.Millisecond)}.String() var deleted int64 err := st.WithTx(ctx, func(tx *sql.Tx) error { q := db.New(tx) if _, err := q.ClearSuggestionBatchAIRunsBefore(ctx, cutoff); err != nil { return err } if _, err := q.ClearWeekPlanAIRunsBefore(ctx, cutoff); err != nil { return err } n, err := q.DeleteAIRunsBefore(ctx, cutoff) deleted = n return err }) if err != nil { return 0, fmt.Errorf("harness: purge ai_runs: %w", err) } return deleted, nil } // runLog collects one run's turns for its ai_runs row. type runLog struct { // id is the row's ID, fixed for the run so that a second save attempt // can never add a second row. id string start store.Timestamp requests []loggedRequest responses []loggedResponse tokensIn int64 tokensOut int64 // problems lists what was wrong with the first reply of a repaired // run; it is the row's error text. problems []string } // loggedRequest is one turn's llm.Request as stored in // ai_runs.request_redacted. type loggedRequest struct { System string `json:"system,omitempty"` Messages []loggedMessage `json:"messages"` Temperature *float64 `json:"temperature,omitempty"` MaxTokens int `json:"max_tokens,omitempty"` SchemaName string `json:"schema_name,omitempty"` Schema string `json:"schema,omitempty"` } type loggedMessage struct { Role llm.Role `json:"role"` Content string `json:"content"` } // loggedResponse is one turn's llm.Response, or its error, as stored in // ai_runs.response_redacted. type loggedResponse struct { Text string `json:"text,omitempty"` JSON string `json:"json,omitempty"` Model string `json:"model,omitempty"` FinishReason string `json:"finish_reason,omitempty"` InputTokens int `json:"input_tokens,omitempty"` OutputTokens int `json:"output_tokens,omitempty"` Error string `json:"error,omitempty"` } // add records one turn: the request sent and the reply or error. Every // string is redacted as it is recorded. func (l *runLog) add(apiKey string, req llm.Request, resp llm.Response, err error) { r := loggedRequest{ System: redact(req.System, apiKey), Temperature: req.Temperature, MaxTokens: req.MaxTokens, } for _, m := range req.Messages { r.Messages = append(r.Messages, loggedMessage{Role: m.Role, Content: redact(m.Content, apiKey)}) } if req.Schema != nil { r.SchemaName = redact(req.Schema.Name, apiKey) r.Schema = redact(string(req.Schema.Schema), apiKey) } l.requests = append(l.requests, r) lr := loggedResponse{ Text: redact(resp.Text, apiKey), JSON: redact(string(resp.JSON), apiKey), Model: redact(resp.Model, apiKey), FinishReason: redact(resp.FinishReason, apiKey), InputTokens: resp.Usage.InputTokens, OutputTokens: resp.Usage.OutputTokens, } if err != nil { lr.Error = redact(err.Error(), apiKey) } l.responses = append(l.responses, lr) l.tokensIn += int64(resp.Usage.InputTokens) l.tokensOut += int64(resp.Usage.OutputTokens) } // save writes the run's ai_runs row within ctx, which the caller bounds. // Every stored string is redacted. func (l *runLog) save(ctx context.Context, cfg Config, skill string, version int, status RunStatus, runErr error) error { // A run that failed before its first turn stores an empty list. request, err := marshalRedacted(nonNil(l.requests), cfg.APIKey) if err != nil { return err } response, err := marshalRedacted(nonNil(l.responses), cfg.APIKey) if err != nil { return err } var errText sql.NullString switch { case runErr != nil: errText = sql.NullString{String: redact(runErr.Error(), cfg.APIKey), Valid: true} case len(l.problems) > 0: errText = sql.NullString{String: redact("first reply: "+strings.Join(l.problems, "; "), cfg.APIKey), Valid: true} } now := store.Now() return db.New(cfg.Store.Writer).CreateAIRun(ctx, db.CreateAIRunParams{ ID: l.id, Skill: redact(skill, cfg.APIKey), SkillVersion: strconv.Itoa(version), Provider: redact(cfg.Settings.Provider, cfg.APIKey), Model: redact(cfg.Settings.Model, cfg.APIKey), Temperature: cfg.Settings.Temperature, Status: string(status), LatencyMs: sql.NullInt64{Int64: now.Sub(l.start.Time).Milliseconds(), Valid: true}, TokensIn: reported(l.tokensIn), TokensOut: reported(l.tokensOut), Error: errText, RequestRedacted: request, ResponseRedacted: sql.NullString{String: response, Valid: true}, CreatedAt: l.start.String(), UpdatedAt: now.String(), }) } // nonNil is s, or an empty slice when s is nil, so it encodes as []. func nonNil[T any](s []T) []T { if s == nil { return []T{} } return s } // reported is n as a column value, NULL when no turn reported a count. func reported(n int64) sql.NullInt64 { return sql.NullInt64{Int64: n, Valid: n > 0} } // marshalRedacted encodes v as JSON without HTML escaping, so the stored // text matches what was redacted, and redacts the result once more in case // the encoding joined text into the key. func marshalRedacted(v any, apiKey string) (string, error) { var b bytes.Buffer enc := json.NewEncoder(&b) enc.SetEscapeHTML(false) if err := enc.Encode(v); err != nil { return "", fmt.Errorf("encode ai_runs body: %w", err) } return llm.Redact(string(bytes.TrimSuffix(b.Bytes(), []byte("\n"))), apiKey), nil }