diff --git a/CHANGELOG.md b/CHANGELOG.md index 3634a57..7904b26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,14 @@ surface is governed by [`COMPATIBILITY.md`](COMPATIBILITY.md). ## [Unreleased] ### Added +- **Spend rollup across runs (`riskkernel audit summary`).** Per-run `audit export` + has a cross-run companion: `riskkernel audit summary --by >` + rolls cost-ledger spend up by a dimension — so spend **by team/user/feature** comes + straight from the tags you put on runs (`--by metadata.team`), with `--since`/`--until` + windows and `--json` output. Deterministic SQL over the ledger you own; no LLM in the + path. The OTel model-call span now also carries `riskkernel.run.name` and one + `riskkernel.run.meta.` per run tag, so the same grouping works in Datadog/Grafana/etc. + without a separate run→team map. - **Importable SigNoz dashboard.** [`examples/otel/signoz`](examples/otel/signoz) ships a ready-made SigNoz dashboard (spend per run, budget halts by reason, tool-call outcomes, latency and token burn by model) built from the same diff --git a/api/v1/otel-genai.md b/api/v1/otel-genai.md index 94ad9b9..fa8fc1c 100644 --- a/api/v1/otel-genai.md +++ b/api/v1/otel-genai.md @@ -57,6 +57,8 @@ conventions don't model. Names are stable per COMPATIBILITY.md. | Attribute | Type | Example | Notes | |---|---|---|---| | `riskkernel.run.id` | string | `9f1c…` | Correlates every span to a governed run. | +| `riskkernel.run.name` | string | `nightly-report` | The run's name, for grouping spend by run (set when non-empty). | +| `riskkernel.run.meta.` | string | `riskkernel.run.meta.team` = `payments` | One attribute per user-supplied run metadata tag, so spend can be grouped by team/user/feature in the backend without a separate run→tag map. Cardinality is the user's own. | | `riskkernel.step.index` | int | `3` | Loop iteration. | | `riskkernel.cost.usd` | double | `0.0042` | Cost charged to the ledger for this call. | | `riskkernel.budget.tokens.limit` | int | `200000` | The run's token budget, if set. | diff --git a/cmd/riskkernel/admin.go b/cmd/riskkernel/admin.go index 67dadbd..a2dc375 100644 --- a/cmd/riskkernel/admin.go +++ b/cmd/riskkernel/admin.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "os" + "strings" "text/tabwriter" "time" @@ -156,19 +157,135 @@ func maxFloat(a, b float64) float64 { // runAudit implements the local audit read-back commands. func runAudit(args []string) error { - if len(args) < 2 { - return fmt.Errorf("usage: riskkernel audit ") + if len(args) < 1 { + return fmt.Errorf("usage: riskkernel audit ...") } switch args[0] { case "export": + if len(args) < 2 { + return errAuditRunID("export") + } return auditExport(args[1]) case "tools": + if len(args) < 2 { + return errAuditRunID("tools") + } return auditTools(args[1]) case "compliance": + if len(args) < 2 { + return errAuditRunID("compliance") + } return auditCompliance(args[1]) + case "summary": + return auditSummary(args[1:]) default: - return fmt.Errorf("unknown audit subcommand %q (want export|tools|compliance)", args[0]) + return fmt.Errorf("unknown audit subcommand %q (want export|tools|compliance|summary)", args[0]) + } +} + +func errAuditRunID(sub string) error { + return fmt.Errorf("usage: riskkernel audit %s ", sub) +} + +// auditSummary rolls cost-ledger spend up across runs, grouped by a dimension — +// provider, model, day, run name, or a run metadata key (metadata., e.g. +// metadata.team). This is the cross-run rollup the per-run `audit export` doesn't +// give: spend by team/user/feature via the tags you put on runs. Deterministic SQL +// over the ledger the user owns; no LLM in the path. +func auditSummary(args []string) error { + var by, sinceStr, untilStr string + asJSON := false + for i := 0; i < len(args); i++ { + switch args[i] { + case "--by": + if i+1 >= len(args) { + return fmt.Errorf("--by requires a value (provider|model|day|name|metadata.)") + } + i++ + by = args[i] + case "--since": + if i+1 >= len(args) { + return fmt.Errorf("--since requires a value (RFC3339 or YYYY-MM-DD)") + } + i++ + sinceStr = args[i] + case "--until": + if i+1 >= len(args) { + return fmt.Errorf("--until requires a value (RFC3339 or YYYY-MM-DD)") + } + i++ + untilStr = args[i] + case "--json": + asJSON = true + default: + return fmt.Errorf("unknown flag %q for audit summary", args[i]) + } + } + if by == "" { + return fmt.Errorf("usage: riskkernel audit summary --by > [--since T] [--until T] [--json]") + } + + opts := storage.SummarizeOptions{By: by} + if sinceStr != "" { + t, err := parseTimeFlag(sinceStr) + if err != nil { + return fmt.Errorf("--since: %w", err) + } + opts.Since = &t + } + if untilStr != "" { + t, err := parseTimeFlag(untilStr) + if err != nil { + return fmt.Errorf("--until: %w", err) + } + opts.Until = &t + } + + store, err := openStoreForCLI() + if err != nil { + return err + } + defer store.Close() + + sum, err := store.SummarizeLedger(context.Background(), opts) + if err != nil { + return err + } + + if asJSON { + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + return enc.Encode(sum) + } + + tw := tabwriter.NewWriter(os.Stdout, 0, 2, 2, ' ', 0) + fmt.Fprintf(tw, "%s\tCALLS\tPROMPT\tCOMPLETION\tDOLLARS\n", summaryHeader(by)) + for _, g := range sum.Groups { + fmt.Fprintf(tw, "%s\t%d\t%d\t%d\t%.6f\n", + g.Key, g.Calls, g.PromptTokens, g.CompletionTokens, g.Dollars) + } + fmt.Fprintf(tw, "TOTAL\t%d\t%d\t%d\t%.6f\n", + sum.Total.Calls, sum.Total.PromptTokens, sum.Total.CompletionTokens, sum.Total.Dollars) + return tw.Flush() +} + +// summaryHeader is the column label for the grouping dimension (metadata.team → TEAM). +func summaryHeader(by string) string { + if k, ok := strings.CutPrefix(by, "metadata."); ok { + return strings.ToUpper(k) + } + return strings.ToUpper(by) +} + +// parseTimeFlag accepts an RFC3339 timestamp or a plain YYYY-MM-DD date (UTC). +func parseTimeFlag(s string) (time.Time, error) { + if t, err := time.Parse(time.RFC3339, s); err == nil { + return t, nil + } + if t, err := time.Parse("2006-01-02", s); err == nil { + return t, nil } + return time.Time{}, fmt.Errorf("invalid time %q (want RFC3339 or YYYY-MM-DD)", s) } // auditCompliance prints an auditor-ready compliance export for a run: its diff --git a/cmd/riskkernel/admin_test.go b/cmd/riskkernel/admin_test.go index f31450b..3256000 100644 --- a/cmd/riskkernel/admin_test.go +++ b/cmd/riskkernel/admin_test.go @@ -69,6 +69,104 @@ func TestAuditExportIncludesToolCalls(t *testing.T) { } } +func TestAuditSummaryByMetadata(t *testing.T) { + dir := seedSummaryStore(t) + t.Setenv("RISKKERNEL_DATA_DIR", dir) + + out := captureStdout(t, func() error { + return runAudit([]string{"summary", "--by", "metadata.team", "--json"}) + }) + + var sum struct { + By string `json:"by"` + Groups []struct { + Key string `json:"key"` + Calls int64 `json:"calls"` + PromptTokens int64 `json:"promptTokens"` + CompletionTokens int64 `json:"completionTokens"` + Dollars float64 `json:"dollars"` + } `json:"groups"` + Total struct { + Calls int64 `json:"calls"` + Dollars float64 `json:"dollars"` + } `json:"total"` + } + if err := json.Unmarshal(out, &sum); err != nil { + t.Fatalf("decode: %v (out=%s)", err, out) + } + if sum.By != "metadata.team" || len(sum.Groups) != 2 { + t.Fatalf("summary = %+v", sum) + } + byTeam := map[string]float64{} + for _, g := range sum.Groups { + byTeam[g.Key] = g.Dollars + } + if byTeam["alpha"] != 0.01 || byTeam["beta"] != 0.03 { + t.Fatalf("per-team dollars = %v, want alpha=0.01 beta=0.03", byTeam) + } + if sum.Total.Calls != 2 || sum.Total.Dollars != 0.04 { + t.Fatalf("total = %+v, want calls=2 dollars=0.04", sum.Total) + } +} + +func TestAuditSummaryRequiresBy(t *testing.T) { + if err := runAudit([]string{"summary"}); err == nil { + t.Fatal("audit summary with no --by should error") + } +} + +func TestParseTimeFlag(t *testing.T) { + if _, err := parseTimeFlag("2026-06-15"); err != nil { + t.Errorf("date: %v", err) + } + if _, err := parseTimeFlag("2026-06-15T10:00:00Z"); err != nil { + t.Errorf("rfc3339: %v", err) + } + if _, err := parseTimeFlag("nope"); err == nil { + t.Error("invalid time should error") + } +} + +func TestSummaryHeader(t *testing.T) { + if got := summaryHeader("metadata.team"); got != "TEAM" { + t.Errorf("metadata.team header = %q, want TEAM", got) + } + if got := summaryHeader("provider"); got != "PROVIDER" { + t.Errorf("provider header = %q, want PROVIDER", got) + } +} + +func seedSummaryStore(t *testing.T) string { + t.Helper() + dir := t.TempDir() + s, err := storage.OpenSQLite(filepath.Join(dir, "riskkernel.db")) + if err != nil { + t.Fatal(err) + } + defer s.Close() + ctx := context.Background() + now := time.Date(2026, 6, 15, 12, 0, 0, 0, time.UTC) + + for _, r := range []struct { + id, team string + dollars float64 + }{{"run-a", "alpha", 0.01}, {"run-b", "beta", 0.03}} { + if err := s.UpsertRun(ctx, storage.RunRecord{ + ID: r.id, Status: "running", Metadata: map[string]string{"team": r.team}, + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatal(err) + } + if err := s.AppendLedger(ctx, storage.LedgerEntry{ + RunID: r.id, StepIndex: 1, Provider: "anthropic", Model: "claude", + PromptTokens: 100, CompletionTokens: 50, Dollars: r.dollars, Priced: true, CreatedAt: now, + }); err != nil { + t.Fatal(err) + } + } + return dir +} + func seedAuditStore(t *testing.T) string { t.Helper() dir := t.TempDir() diff --git a/cmd/riskkernel/completion.go b/cmd/riskkernel/completion.go index 5e3eedf..c00999d 100644 --- a/cmd/riskkernel/completion.go +++ b/cmd/riskkernel/completion.go @@ -20,7 +20,7 @@ var topLevelCommands = []string{ // approvals.go, memory.go). var subCommands = map[string][]string{ "runs": {"list", "resume"}, - "audit": {"export", "tools", "compliance"}, + "audit": {"export", "tools", "compliance", "summary"}, "policy": {"validate", "dry-run"}, "approvals": {"list", "approve", "deny"}, "memory": {"list", "show"}, diff --git a/cmd/riskkernel/main.go b/cmd/riskkernel/main.go index 5bc3ce3..3598ac4 100644 --- a/cmd/riskkernel/main.go +++ b/cmd/riskkernel/main.go @@ -87,6 +87,7 @@ Usage: riskkernel audit export Export a run's cost ledger as JSON riskkernel audit tools Export a run's governed tool calls as JSON riskkernel audit compliance Auditor-ready OWASP / EU AI Act evidence export + riskkernel audit summary --by Roll spend up across runs (provider|model|day|name|metadata.) riskkernel policy validate Validate a riskkernel.yaml policy file riskkernel policy dry-run Show what a policy would gate/halt on a run riskkernel approvals list List pending human-in-the-loop approvals diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index 73e9f49..8b04ff7 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -167,7 +167,8 @@ func (g *Gateway) emitCallSpan(run *runs.Run, step int32, providerName string, p } v := run.View() call := otel.Call{ - RunID: run.ID, StepIndex: step, Provider: providerName, Operation: "chat", + RunID: run.ID, RunName: v.Name, Metadata: v.Metadata, + StepIndex: step, Provider: providerName, Operation: "chat", RequestModel: preq.Model, ResponseModel: resp.Model, MaxTokens: preq.MaxTokens, Temperature: preq.Temperature, PromptTokens: resp.Usage.PromptTokens, OutputTokens: resp.Usage.CompletionTokens, diff --git a/internal/otel/otel.go b/internal/otel/otel.go index 60aea80..98a7cad 100644 --- a/internal/otel/otel.go +++ b/internal/otel/otel.go @@ -41,8 +41,12 @@ const ( attrGenAIResponseID = "gen_ai.response.id" attrErrorType = "error.type" - attrRunID = "riskkernel.run.id" - attrStepIndex = "riskkernel.step.index" + attrRunID = "riskkernel.run.id" + attrRunName = "riskkernel.run.name" + attrStepIndex = "riskkernel.step.index" + // attrRunMetaPrefix is prepended to each user-supplied run metadata key so a + // backend can group cost by team/user/feature without a separate run→tag map. + attrRunMetaPrefix = "riskkernel.run.meta." attrCostUSD = "riskkernel.cost.usd" attrBudgetTokLimit = "riskkernel.budget.tokens.limit" attrBudgetTokRemain = "riskkernel.budget.tokens.remaining" @@ -146,6 +150,8 @@ func (t *Tracer) Shutdown(ctx context.Context) error { // Call is the data for one governed model-call span. type Call struct { RunID string + RunName string // the run's name, emitted for spend attribution + Metadata map[string]string // user-supplied run tags (team/user/feature/…) StepIndex int32 Provider string Operation string // e.g. "chat" @@ -191,6 +197,14 @@ func (t *Tracer) RecordCall(ctx context.Context, c Call) { attribute.String(attrRunID, c.RunID), attribute.Int(attrStepIndex, int(c.StepIndex)), } + if c.RunName != "" { + attrs = append(attrs, attribute.String(attrRunName, c.RunName)) + } + // Emit each run tag as riskkernel.run.meta. so spend can be grouped by + // team/user/feature in the backend. Cardinality is the user's own (their tags). + for k, v := range c.Metadata { + attrs = append(attrs, attribute.String(attrRunMetaPrefix+k, v)) + } if c.ResponseModel != "" { attrs = append(attrs, attribute.String(attrGenAIResponseModel, c.ResponseModel)) } diff --git a/internal/otel/otel_test.go b/internal/otel/otel_test.go index 8fc5b4c..624b6b6 100644 --- a/internal/otel/otel_test.go +++ b/internal/otel/otel_test.go @@ -70,6 +70,39 @@ func TestRecordCall_Attributes(t *testing.T) { } } +func TestRecordCall_RunNameAndMetadata(t *testing.T) { + tr, sr := newRecording() + tr.RecordCall(context.Background(), Call{ + RunID: "r1", RunName: "nightly-report", StepIndex: 1, + Provider: "openai", Operation: "chat", RequestModel: "gpt-4o", + Metadata: map[string]string{"team": "payments", "env": "prod"}, + }) + a := attrMap(sr.Ended()[0].Attributes()) + if a[attrRunName].AsString() != "nightly-report" { + t.Errorf("run.name = %v, want nightly-report", a[attrRunName]) + } + if a["riskkernel.run.meta.team"].AsString() != "payments" { + t.Errorf("meta.team = %v, want payments", a["riskkernel.run.meta.team"]) + } + if a["riskkernel.run.meta.env"].AsString() != "prod" { + t.Errorf("meta.env = %v, want prod", a["riskkernel.run.meta.env"]) + } +} + +func TestRecordCall_NoNameNoMetadata(t *testing.T) { + // An unnamed, untagged run emits neither the name nor any meta.* attribute. + tr, sr := newRecording() + tr.RecordCall(context.Background(), Call{ + RunID: "r1", StepIndex: 1, Provider: "openai", Operation: "chat", RequestModel: "gpt-4o", + }) + for _, kv := range sr.Ended()[0].Attributes() { + k := string(kv.Key) + if k == attrRunName || len(k) > len(attrRunMetaPrefix) && k[:len(attrRunMetaPrefix)] == attrRunMetaPrefix { + t.Errorf("unexpected attribute %q on an unnamed/untagged run", k) + } + } +} + func TestRecordCall_Error(t *testing.T) { tr, sr := newRecording() tr.RecordCall(context.Background(), Call{