diff --git a/.env.example b/.env.example index bf4e81c..115f166 100644 --- a/.env.example +++ b/.env.example @@ -27,3 +27,9 @@ RISKKERNEL_DEFAULT_SECONDS= OTEL_EXPORTER_OTLP_ENDPOINT= OTEL_EXPORTER_OTLP_PROTOCOL=grpc OTEL_SERVICE_NAME=riskkernel + +# Human-in-the-loop approval gate. +# DEFAULT_SAFE=true requires approval for any side-effecting tool call (fail closed). +RISKKERNEL_APPROVAL_DEFAULT_SAFE=true +# Optional: POST a JSON notification here when an approval becomes pending. +RISKKERNEL_APPROVAL_WEBHOOK= diff --git a/CHANGELOG.md b/CHANGELOG.md index 55b830b..dfb36b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -52,5 +52,15 @@ surface is governed by [`COMPATIBILITY.md`](COMPATIBILITY.md). gRPC/HTTP via standard `OTEL_*` env vars. **Off unless an endpoint is configured** — spans go only to the user's backend. Example Jaeger backend + dashboard guidance in `examples/otel/`. +- **Human-in-the-loop approval gate (Surface: HITL)** — a side-effecting tool call + that policy gates pauses until a human approves or denies it. Deterministic + policy match (exact tool or side-effect glob, plus a fail-closed default-safe + mode); the gate blocks the call and is resolved via three channels: the CLI + (`riskkernel approvals list / approve / deny`), a local embedded admin page + (`/admin/approvals`), and an optional webhook (`RISKKERNEL_APPROVAL_WEBHOOK`). + `POST /v1/runs/{id}/approve`, `GET /v1/approvals`, and `GET /v1/runs/{id}` + (surfaces `pendingApproval` + `waiting_approval` status). Approvals are persisted + (migration `00003`) as an audit trail. Webhook is user-configured egress only + (see SECURITY.md). [Unreleased]: https://github.com/prashar32/riskkernel/commits/main diff --git a/SECURITY.md b/SECURITY.md index a281559..ae2ff3a 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -8,8 +8,8 @@ RiskKernel is a reliability and governance daemon for AI agents. It runs on infr This is verifiable, and we intend for you to verify it: -- **Read the code.** All network egress originates in `internal/provider/` (calls to LLM providers) and `internal/otel/` (only when *you* configure an OTLP endpoint). There is no other outbound HTTP client in the codebase. -- **Watch the wire.** Run RiskKernel under `tcpdump`/`Little Snitch`/`mitmproxy` and confirm the only destinations are your configured provider and OTLP endpoints. +- **Read the code.** All network egress originates in three places, each going only where *you* point it: `internal/provider/` (calls to the LLM providers you configure), `internal/otel/` (spans, only when you set an OTLP endpoint), and `internal/approval/` (the approval webhook, only when you set `RISKKERNEL_APPROVAL_WEBHOOK`). The webhook payload carries the pending approval's metadata (run id, tool, side effect, arguments) — never provider keys or other secrets. There is no other outbound HTTP client in the codebase. +- **Watch the wire.** Run RiskKernel under `tcpdump`/`Little Snitch`/`mitmproxy` and confirm the only destinations are your configured provider, OTLP, and (if set) approval-webhook endpoints. - **Build gates.** CI fails if a disallowed network import appears outside the provider/otel packages. Any future "you're N versions behind" hint is a local startup log line comparing against a build-time constant — it makes no network call. An optional `--share-anonymous-usage` flag may exist someday; it will be **OFF by default** and clearly documented. diff --git a/api/v1/openapi.yaml b/api/v1/openapi.yaml index f2d2f76..b7838a5 100644 --- a/api/v1/openapi.yaml +++ b/api/v1/openapi.yaml @@ -156,6 +156,33 @@ paths: '404': $ref: '#/components/responses/NotFound' + /v1/approvals: + get: + tags: [approvals] + operationId: listApprovals + summary: List approval requests + description: | + Lists human-in-the-loop approval requests, optionally filtered by status. + Used by the CLI and the local admin page to surface what is pending. + parameters: + - name: status + in: query + required: false + schema: + type: string + enum: [pending, approved, denied] + responses: + '200': + description: Matching approvals, newest first. + content: + application/json: + schema: + type: array + items: + $ref: '#/components/schemas/ApprovalRequest' + '401': + $ref: '#/components/responses/Unauthorized' + /v1/checkpoints/{run_id}: get: tags: [checkpoints] @@ -412,11 +439,14 @@ components: ApprovalRequest: type: object required: [id, tool, createdAt] - description: A side-effecting action paused awaiting human approval. + description: A side-effecting action gated for human approval (pending or resolved). properties: id: type: string format: uuid + runId: + type: string + format: uuid stepIndex: type: integer format: int32 @@ -430,9 +460,20 @@ components: type: object additionalProperties: true description: The proposed call arguments, for the human to review. + status: + type: string + enum: [pending, approved, denied] + reason: + type: string + description: Human note recorded with the decision (once resolved). + decidedBy: + type: string createdAt: type: string format: date-time + decidedAt: + type: string + format: date-time ApprovalDecision: type: object diff --git a/cmd/riskkernel/approvals.go b/cmd/riskkernel/approvals.go new file mode 100644 index 0000000..7849606 --- /dev/null +++ b/cmd/riskkernel/approvals.go @@ -0,0 +1,175 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "text/tabwriter" + "time" + + "github.com/prashar32/riskkernel/internal/config" +) + +// runApprovals implements `riskkernel approvals `. +// +// Unlike `runs`/`audit` (which read the local store), these talk to the running +// daemon's HTTP API — resolving an approval must wake the goroutine blocked inside +// the daemon, which a direct store write cannot do. +func runApprovals(args []string) error { + if len(args) == 0 { + return fmt.Errorf("usage: riskkernel approvals | deny >") + } + switch args[0] { + case "list": + return approvalsList() + case "approve", "deny": + if len(args) < 2 { + return fmt.Errorf("usage: riskkernel approvals %s [--reason ]", args[0]) + } + reason := flagValue(args[2:], "--reason") + return approvalsResolve(args[1], args[0] == "approve", reason) + default: + return fmt.Errorf("unknown approvals subcommand %q (want list|approve|deny)", args[0]) + } +} + +type approvalItem struct { + ID string `json:"id"` + RunID string `json:"runId"` + StepIndex int `json:"stepIndex"` + Tool string `json:"tool"` + SideEffect string `json:"sideEffect"` + Arguments map[string]any `json:"arguments"` + Status string `json:"status"` + CreatedAt string `json:"createdAt"` +} + +func approvalsList() error { + c, err := newDaemonClient() + if err != nil { + return err + } + var items []approvalItem + if err := c.getJSON("/v1/approvals?status=pending", &items); err != nil { + return err + } + if len(items) == 0 { + fmt.Println("no pending approvals") + return nil + } + tw := tabwriter.NewWriter(os.Stdout, 0, 2, 2, ' ', 0) + fmt.Fprintln(tw, "ID\tRUN\tSTEP\tTOOL\tSIDE EFFECT\tCREATED") + for _, a := range items { + fmt.Fprintf(tw, "%s\t%s\t%d\t%s\t%s\t%s\n", a.ID, a.RunID, a.StepIndex, a.Tool, dash(a.SideEffect), a.CreatedAt) + } + return tw.Flush() +} + +func approvalsResolve(id string, approve bool, reason string) error { + c, err := newDaemonClient() + if err != nil { + return err + } + // Discover the approval's run id from the pending list (the resolve endpoint + // is keyed by run id per the api/v1 contract). + var items []approvalItem + if err := c.getJSON("/v1/approvals?status=pending", &items); err != nil { + return err + } + runID := "" + for _, a := range items { + if a.ID == id { + runID = a.RunID + break + } + } + if runID == "" { + return fmt.Errorf("no pending approval with id %s", id) + } + + decision := "deny" + if approve { + decision = "approve" + } + body, _ := json.Marshal(map[string]string{ + "approvalId": id, "decision": decision, "reason": reason, "decidedBy": "cli", + }) + if err := c.post("/v1/runs/"+runID+"/approve", body); err != nil { + return err + } + fmt.Printf("%sd approval %s\n", decision, id) + return nil +} + +// --- minimal daemon HTTP client --- + +type daemonClient struct { + base string + token string + client *http.Client +} + +func newDaemonClient() (*daemonClient, error) { + cfg, err := config.Load() + if err != nil { + return nil, err + } + return &daemonClient{ + base: fmt.Sprintf("http://localhost:%d", cfg.Port), + token: cfg.APIToken, + client: &http.Client{Timeout: 10 * time.Second}, + }, nil +} + +func (c *daemonClient) do(method, path string, body []byte) ([]byte, error) { + var rdr io.Reader + if body != nil { + rdr = bytes.NewReader(body) + } + req, err := http.NewRequest(method, c.base+path, rdr) + if err != nil { + return nil, err + } + if body != nil { + req.Header.Set("content-type", "application/json") + } + if c.token != "" { + req.Header.Set("Authorization", "Bearer "+c.token) + } + resp, err := c.client.Do(req) + if err != nil { + return nil, fmt.Errorf("cannot reach daemon at %s (is `riskkernel serve` running?): %w", c.base, err) + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode >= 300 { + return nil, fmt.Errorf("daemon returned %d: %s", resp.StatusCode, string(raw)) + } + return raw, nil +} + +func (c *daemonClient) getJSON(path string, out any) error { + raw, err := c.do(http.MethodGet, path, nil) + if err != nil { + return err + } + return json.Unmarshal(raw, out) +} + +func (c *daemonClient) post(path string, body []byte) error { + _, err := c.do(http.MethodPost, path, body) + return err +} + +// flagValue returns the value following name in args, or "". +func flagValue(args []string, name string) string { + for i := 0; i < len(args)-1; i++ { + if args[i] == name { + return args[i+1] + } + } + return "" +} diff --git a/cmd/riskkernel/main.go b/cmd/riskkernel/main.go index c3d80ef..3a3f83e 100644 --- a/cmd/riskkernel/main.go +++ b/cmd/riskkernel/main.go @@ -44,6 +44,8 @@ func main() { err = runRuns(args) case "audit": err = runAudit(args) + case "approvals": + err = runApprovals(args) case "version", "--version", "-v": fmt.Println("riskkernel", version.String()) case "help", "--help", "-h": @@ -69,6 +71,9 @@ Usage: riskkernel runs list List persisted governed runs riskkernel runs resume Show a run's resumable state after a crash riskkernel audit export Export a run's cost ledger as JSON + riskkernel approvals list List pending human-in-the-loop approvals + riskkernel approvals approve [--reason ...] Approve a pending request + riskkernel approvals deny [--reason ...] Deny a pending request riskkernel version Print build identity riskkernel help Show this help @@ -112,7 +117,7 @@ func runServe(_ []string) error { ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() - srv := httpapi.New(cfg, deps.Gateway, deps.Runs, deps.Log) + srv := httpapi.New(cfg, deps.Gateway, deps.Runs, deps.Approvals, deps.Log) addr := fmt.Sprintf(":%d", cfg.Port) return srv.Serve(ctx, addr) } diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index c73ef85..9da7cf5 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -12,6 +12,7 @@ import ( "path/filepath" "time" + "github.com/prashar32/riskkernel/internal/approval" "github.com/prashar32/riskkernel/internal/config" "github.com/prashar32/riskkernel/internal/gateway" "github.com/prashar32/riskkernel/internal/governor" @@ -32,6 +33,7 @@ type Deps struct { Gateway *gateway.Gateway Store storage.Store Tracer *otel.Tracer + Approvals *approval.Gate } // Close releases dependencies that hold resources (the tracer's buffered spans, @@ -78,6 +80,12 @@ func Build(cfg *config.Config) (*Deps, error) { mgr := runs.NewManager(toGovernorBudget(cfg.DefaultBudget)).WithStore(store, log) gw := gateway.New(registry, mgr, prices, tracer, log) + var notifier approval.Notifier + if wh := approval.NewWebhookNotifier(cfg.Approval.WebhookURL, log); wh != nil { + notifier = wh + } + gate := approval.NewGate(store, approval.Policy{DefaultSafe: cfg.Approval.DefaultSafe}, notifier, log) + return &Deps{ Config: cfg, Log: log, @@ -87,6 +95,7 @@ func Build(cfg *config.Config) (*Deps, error) { Gateway: gw, Store: store, Tracer: tracer, + Approvals: gate, }, nil } diff --git a/internal/approval/gate.go b/internal/approval/gate.go new file mode 100644 index 0000000..fe389bc --- /dev/null +++ b/internal/approval/gate.go @@ -0,0 +1,148 @@ +package approval + +import ( + "context" + "log/slog" + "sync" + "time" + + "github.com/prashar32/riskkernel/internal/id" + "github.com/prashar32/riskkernel/internal/storage" +) + +// Decision is the outcome of an approval request. +type Decision struct { + Approved bool + Reason string + By string +} + +// Request describes a side-effecting tool call seeking approval. +type Request struct { + RunID string + StepIndex int32 + Tool string + SideEffect string + Arguments map[string]any +} + +// Notifier pushes a newly-pending approval to a channel (e.g. a webhook). The CLI +// and web page are pull channels and do not implement this. +type Notifier interface { + Notify(ctx context.Context, a storage.ApprovalRecord) +} + +// Gate is the human-in-the-loop approval queue. It persists pending approvals, +// notifies push channels, and blocks the calling (tool) goroutine until a human +// resolves the request via the API/CLI/web — or the run's context is cancelled. +type Gate struct { + store storage.Store + policy Policy + notifier Notifier + log *slog.Logger + now func() time.Time + newID func() string + + mu sync.Mutex + waiters map[string]chan Decision +} + +// NewGate constructs a Gate. notifier may be nil (no push channel). +func NewGate(store storage.Store, policy Policy, notifier Notifier, log *slog.Logger) *Gate { + return &Gate{ + store: store, + policy: policy, + notifier: notifier, + log: log, + now: time.Now, + newID: id.NewUUID, + waiters: make(map[string]chan Decision), + } +} + +// Required reports whether a call needs approval under the gate's policy. +func (g *Gate) Required(tool, sideEffect string) bool { + return g.policy.Requires(tool, sideEffect) +} + +// Request gates a side-effecting call. If approval is not required it returns an +// approved Decision immediately. Otherwise it persists a pending approval, +// notifies push channels, and BLOCKS until the request is resolved or ctx is +// done (run cancel / time budget). This is how a side-effecting call "pauses". +func (g *Gate) Request(ctx context.Context, req Request) (Decision, string, error) { + if !g.policy.Requires(req.Tool, req.SideEffect) { + return Decision{Approved: true}, "", nil + } + + rec := storage.ApprovalRecord{ + ID: g.newID(), + RunID: req.RunID, + StepIndex: req.StepIndex, + Tool: req.Tool, + SideEffect: req.SideEffect, + Arguments: req.Arguments, + Status: storage.ApprovalPending, + CreatedAt: g.now(), + } + if g.store != nil { + if err := g.store.CreateApproval(ctx, rec); err != nil { + return Decision{}, "", err + } + } + + ch := make(chan Decision, 1) + g.mu.Lock() + g.waiters[rec.ID] = ch + g.mu.Unlock() + defer func() { + g.mu.Lock() + delete(g.waiters, rec.ID) + g.mu.Unlock() + }() + + if g.notifier != nil { + g.notifier.Notify(ctx, rec) + } + g.log.Info("approval required", "id", rec.ID, "run", rec.RunID, "tool", rec.Tool, "side_effect", rec.SideEffect) + + select { + case d := <-ch: + return d, rec.ID, nil + case <-ctx.Done(): + return Decision{}, rec.ID, ctx.Err() + } +} + +// Resolve records a human decision and wakes any blocked waiter. Returns +// storage.ErrNotFound if the approval is unknown or already resolved. +func (g *Gate) Resolve(ctx context.Context, id string, approved bool, reason, by string) error { + status := storage.ApprovalApproved + if !approved { + status = storage.ApprovalDenied + } + if g.store != nil { + if err := g.store.ResolveApproval(ctx, id, status, reason, by, g.now()); err != nil { + return err + } + } + g.mu.Lock() + ch := g.waiters[id] + g.mu.Unlock() + if ch != nil { + ch <- Decision{Approved: approved, Reason: reason, By: by} + } + return nil +} + +// Pending returns the currently-pending approvals. +func (g *Gate) Pending(ctx context.Context) ([]storage.ApprovalRecord, error) { + if g.store == nil { + return nil, nil + } + return g.store.ListApprovals(ctx, storage.ApprovalPending) +} + +// Get returns a single approval by id. +func (g *Gate) Get(ctx context.Context, id string) (storage.ApprovalRecord, error) { + return g.store.GetApproval(ctx, id) +} diff --git a/internal/approval/gate_test.go b/internal/approval/gate_test.go new file mode 100644 index 0000000..45d51bc --- /dev/null +++ b/internal/approval/gate_test.go @@ -0,0 +1,144 @@ +package approval + +import ( + "context" + "errors" + "log/slog" + "path/filepath" + "testing" + "time" + + "github.com/prashar32/riskkernel/internal/storage" +) + +func newTestGate(t *testing.T, policy Policy) (*Gate, storage.Store) { + t.Helper() + store, err := storage.OpenSQLite(filepath.Join(t.TempDir(), "appr.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + // Seed a run so approval FK constraints hold. + now := time.Now().UTC() + if err := store.UpsertRun(context.Background(), storage.RunRecord{ + ID: "run-1", Status: "running", CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatal(err) + } + return NewGate(store, policy, nil, slog.New(slog.NewTextHandler(discard{}, nil))), store +} + +type discard struct{} + +func (discard) Write(p []byte) (int, error) { return len(p), nil } + +func TestGate_AutoApproveWhenNotRequired(t *testing.T) { + g, _ := newTestGate(t, Policy{DefaultSafe: true}) + // Read-only call (no side effect) → not required → immediate approve. + d, id, err := g.Request(context.Background(), Request{RunID: "run-1", Tool: "mcp://fs", SideEffect: ""}) + if err != nil || !d.Approved { + t.Fatalf("expected auto-approve, got d=%+v id=%q err=%v", d, id, err) + } + // Nothing should have been persisted. + pend, _ := g.Pending(context.Background()) + if len(pend) != 0 { + t.Errorf("auto-approve should not create a pending row: %+v", pend) + } +} + +func TestGate_RequestBlocksUntilApproved(t *testing.T) { + g, _ := newTestGate(t, Policy{DefaultSafe: true}) + + type result struct { + d Decision + err error + } + done := make(chan result, 1) + go func() { + d, _, err := g.Request(context.Background(), Request{ + RunID: "run-1", StepIndex: 2, Tool: "mcp://shell", SideEffect: "exec", + Arguments: map[string]any{"cmd": "ls"}, + }) + done <- result{d, err} + }() + + // Wait for the pending approval to appear, then approve it. + id := waitPending(t, g) + if err := g.Resolve(context.Background(), id, true, "looks fine", "tester"); err != nil { + t.Fatalf("Resolve: %v", err) + } + + select { + case r := <-done: + if r.err != nil || !r.d.Approved || r.d.By != "tester" { + t.Fatalf("expected approved decision, got %+v err=%v", r.d, r.err) + } + case <-time.After(2 * time.Second): + t.Fatal("Request did not unblock after approval") + } +} + +func TestGate_RequestBlocksUntilDenied(t *testing.T) { + g, _ := newTestGate(t, Policy{DefaultSafe: true}) + done := make(chan Decision, 1) + go func() { + d, _, _ := g.Request(context.Background(), Request{RunID: "run-1", Tool: "mcp://shell", SideEffect: "exec"}) + done <- d + }() + id := waitPending(t, g) + if err := g.Resolve(context.Background(), id, false, "nope", "tester"); err != nil { + t.Fatal(err) + } + select { + case d := <-done: + if d.Approved { + t.Fatalf("expected denied, got %+v", d) + } + case <-time.After(2 * time.Second): + t.Fatal("Request did not unblock after denial") + } +} + +func TestGate_RequestContextCancel(t *testing.T) { + g, _ := newTestGate(t, Policy{DefaultSafe: true}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + _, _, err := g.Request(ctx, Request{RunID: "run-1", Tool: "mcp://shell", SideEffect: "exec"}) + done <- err + }() + waitPending(t, g) + cancel() // e.g. run killed / time budget hit while awaiting approval + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Request did not unblock on context cancel") + } +} + +func TestGate_ResolveUnknown(t *testing.T) { + g, _ := newTestGate(t, Policy{DefaultSafe: true}) + err := g.Resolve(context.Background(), "does-not-exist", true, "", "") + if !errors.Is(err, storage.ErrNotFound) { + t.Fatalf("expected ErrNotFound, got %v", err) + } +} + +func waitPending(t *testing.T, g *Gate) string { + t.Helper() + for i := 0; i < 200; i++ { + pend, err := g.Pending(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(pend) == 1 { + return pend[0].ID + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("pending approval never appeared") + return "" +} diff --git a/internal/approval/policy.go b/internal/approval/policy.go new file mode 100644 index 0000000..adf065b --- /dev/null +++ b/internal/approval/policy.go @@ -0,0 +1,46 @@ +// Package approval implements the human-in-the-loop gate (CLAUDE.md headline +// feature). A side-effecting tool call that policy gates pauses until a human +// approves or denies it. Policy evaluation is deterministic — no LLM decides +// whether something needs approval. +package approval + +import "path" + +// Rule matches a tool call by exact tool name or by a side-effect glob (e.g. +// "*write*"). A rule with both set matches if EITHER matches. +type Rule struct { + Tool string + SideEffect string +} + +func (r Rule) matches(tool, sideEffect string) bool { + if r.Tool != "" && r.Tool == tool { + return true + } + if r.SideEffect != "" { + if ok, err := path.Match(r.SideEffect, sideEffect); err == nil && ok { + return true + } + } + return false +} + +// Policy is the deterministic rule set deciding which calls need approval. It +// mirrors the api/v1 ApprovalPolicy schema. +type Policy struct { + // RequireFor lists match rules; a call needs approval if it matches ANY rule. + RequireFor []Rule + // DefaultSafe, when true, requires approval for ANY call with a non-empty side + // effect even if no rule matched — fail closed on side effects. + DefaultSafe bool +} + +// Requires reports whether a tool call needs human approval. +func (p Policy) Requires(tool, sideEffect string) bool { + for _, r := range p.RequireFor { + if r.matches(tool, sideEffect) { + return true + } + } + return p.DefaultSafe && sideEffect != "" +} diff --git a/internal/approval/policy_test.go b/internal/approval/policy_test.go new file mode 100644 index 0000000..2d6bc92 --- /dev/null +++ b/internal/approval/policy_test.go @@ -0,0 +1,44 @@ +package approval + +import "testing" + +func TestPolicyRequires(t *testing.T) { + p := Policy{ + RequireFor: []Rule{ + {Tool: "mcp://shell"}, + {SideEffect: "*write*"}, + {Tool: "mcp://github/create_pull_request"}, + }, + } + cases := []struct { + tool, sideEffect string + want bool + }{ + {"mcp://shell", "", true}, // exact tool + {"mcp://filesystem", "file_write", true}, // side-effect glob + {"mcp://filesystem", "overwrite_x", true}, // glob substring + {"mcp://github/create_pull_request", "", true}, + {"mcp://filesystem", "read", false}, // read-only, no rule + {"mcp://search", "", false}, // unlisted, no side effect + } + for _, c := range cases { + if got := p.Requires(c.tool, c.sideEffect); got != c.want { + t.Errorf("Requires(%q,%q) = %v, want %v", c.tool, c.sideEffect, got, c.want) + } + } +} + +func TestPolicyDefaultSafe(t *testing.T) { + p := Policy{DefaultSafe: true} + if !p.Requires("mcp://anything", "write") { + t.Error("default-safe should require approval for any side effect") + } + if p.Requires("mcp://anything", "") { + t.Error("default-safe should NOT gate read-only (empty side effect) calls") + } + + open := Policy{DefaultSafe: false} + if open.Requires("mcp://anything", "write") { + t.Error("non-default-safe + no rule should not require approval") + } +} diff --git a/internal/approval/webhook.go b/internal/approval/webhook.go new file mode 100644 index 0000000..0cb6cef --- /dev/null +++ b/internal/approval/webhook.go @@ -0,0 +1,84 @@ +package approval + +import ( + "bytes" + "context" + "encoding/json" + "log/slog" + "net/http" + "time" + + "github.com/prashar32/riskkernel/internal/storage" +) + +// WebhookNotifier POSTs a JSON notification to a user-configured URL when an +// approval becomes pending. This is user-configured outbound network (like the +// OTLP endpoint) — RiskKernel calls it only because the user set the URL. It does +// NOT include any provider keys or secrets. See SECURITY.md. +type WebhookNotifier struct { + url string + client *http.Client + log *slog.Logger +} + +// NewWebhookNotifier returns a notifier, or nil if url is empty (no push channel). +func NewWebhookNotifier(url string, log *slog.Logger) *WebhookNotifier { + if url == "" { + return nil + } + return &WebhookNotifier{ + url: url, + client: &http.Client{Timeout: 10 * time.Second}, + log: log, + } +} + +type webhookPayload struct { + Event string `json:"event"` + ID string `json:"id"` + RunID string `json:"run_id"` + StepIndex int32 `json:"step_index"` + Tool string `json:"tool"` + SideEffect string `json:"side_effect"` + Arguments map[string]any `json:"arguments,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + +// Notify fires the webhook asynchronously (best-effort; failures are logged, never +// fatal — a flaky webhook must not wedge a governed run). +func (w *WebhookNotifier) Notify(_ context.Context, a storage.ApprovalRecord) { + payload := webhookPayload{ + Event: "approval.pending", + ID: a.ID, + RunID: a.RunID, + StepIndex: a.StepIndex, + Tool: a.Tool, + SideEffect: a.SideEffect, + Arguments: a.Arguments, + CreatedAt: a.CreatedAt, + } + body, err := json.Marshal(payload) + if err != nil { + w.log.Error("approval webhook marshal failed", "err", err) + return + } + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, w.url, bytes.NewReader(body)) + if err != nil { + w.log.Error("approval webhook request failed", "err", err) + return + } + req.Header.Set("content-type", "application/json") + resp, err := w.client.Do(req) + if err != nil { + w.log.Error("approval webhook delivery failed", "url", w.url, "err", err) + return + } + _ = resp.Body.Close() + if resp.StatusCode >= 300 { + w.log.Warn("approval webhook non-2xx", "url", w.url, "status", resp.StatusCode) + } + }() +} diff --git a/internal/config/config.go b/internal/config/config.go index d48fcf2..e40d4e8 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -43,6 +43,20 @@ type Config struct { // an endpoint is set — RiskKernel never emits telemetry unless the user points // it at their own OTLP backend. OTel OTelConfig + + // Approval configures the human-in-the-loop gate. + Approval ApprovalConfig +} + +// ApprovalConfig configures the human-in-the-loop approval gate. +type ApprovalConfig struct { + // DefaultSafe requires approval for any side-effecting tool call not otherwise + // allowed. Read from RISKKERNEL_APPROVAL_DEFAULT_SAFE (default true — fail + // closed on side effects). + DefaultSafe bool + // WebhookURL, if set, receives a JSON POST when an approval becomes pending. + // Read from RISKKERNEL_APPROVAL_WEBHOOK. User-configured egress only. + WebhookURL string } // OTelConfig configures OTLP trace export, using standard OpenTelemetry env vars @@ -102,10 +116,28 @@ func Load() (*Config, error) { OpenAIAPIKey: os.Getenv("OPENAI_API_KEY"), DefaultBudget: budget, OTel: loadOTel(), + Approval: ApprovalConfig{ + DefaultSafe: envBoolDefault("RISKKERNEL_APPROVAL_DEFAULT_SAFE", true), + WebhookURL: os.Getenv("RISKKERNEL_APPROVAL_WEBHOOK"), + }, } return cfg, nil } +// envBoolDefault parses a boolean env var, returning def when unset. Accepts +// "true"/"false" (case-insensitive) and "1"/"0". +func envBoolDefault(key string, def bool) bool { + v := strings.TrimSpace(os.Getenv(key)) + switch strings.ToLower(v) { + case "": + return def + case "true", "1", "yes": + return true + default: + return false + } +} + // loadOTel resolves OpenTelemetry export config from standard OTEL_* env vars. func loadOTel() OTelConfig { endpoint := os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") diff --git a/internal/httpapi/admin/approvals.html b/internal/httpapi/admin/approvals.html new file mode 100644 index 0000000..d66b21d --- /dev/null +++ b/internal/httpapi/admin/approvals.html @@ -0,0 +1,78 @@ + + + + + + RiskKernel — Approvals + + + +

Pending approvals

+

Side-effecting tool calls paused for human review. The LLM proposes; you dispose.

+ +
+ + +
+ +

Loading…

+ + + + diff --git a/internal/httpapi/approvals.go b/internal/httpapi/approvals.go new file mode 100644 index 0000000..42aad2d --- /dev/null +++ b/internal/httpapi/approvals.go @@ -0,0 +1,220 @@ +package httpapi + +import ( + _ "embed" + "encoding/json" + "errors" + "net/http" + "time" + + "github.com/prashar32/riskkernel/internal/httpx" + "github.com/prashar32/riskkernel/internal/storage" +) + +//go:embed admin/approvals.html +var approvalsPage []byte + +// approvalDecision is the POST /v1/runs/{id}/approve body (api/v1 ApprovalDecision). +type approvalDecision struct { + ApprovalID string `json:"approvalId"` + Decision string `json:"decision"` // approve | deny + Reason string `json:"reason"` + DecidedBy string `json:"decidedBy"` +} + +// handleApprove resolves a human-in-the-loop gate (POST /v1/runs/{id}/approve). +func (s *Server) handleApprove(w http.ResponseWriter, r *http.Request) { + runID := r.PathValue("id") + var dec approvalDecision + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&dec); err != nil { + httpx.WriteError(w, http.StatusBadRequest, "bad_request", "invalid JSON: "+err.Error()) + return + } + approve, ok := decisionToBool(dec.Decision) + if !ok { + httpx.WriteError(w, http.StatusBadRequest, "bad_request", `decision must be "approve" or "deny"`) + return + } + + // Resolve the named approval, or the run's single pending one. + approvalID := dec.ApprovalID + if approvalID == "" { + pending, err := s.pendingForRun(r, runID) + if err != nil { + httpx.WriteError(w, http.StatusInternalServerError, "internal_error", err.Error()) + return + } + switch len(pending) { + case 0: + httpx.WriteError(w, http.StatusConflict, "no_pending_approval", "no approval is pending for this run") + return + case 1: + approvalID = pending[0].ID + default: + httpx.WriteError(w, http.StatusConflict, "ambiguous_approval", "run has multiple pending approvals; specify approvalId") + return + } + } else { + // Validate the approval belongs to this run. + a, err := s.approvals.Get(r.Context(), approvalID) + if err != nil { + httpx.WriteError(w, http.StatusNotFound, "not_found", "approval not found") + return + } + if a.RunID != runID { + httpx.WriteError(w, http.StatusBadRequest, "bad_request", "approval does not belong to this run") + return + } + } + + if err := s.approvals.Resolve(r.Context(), approvalID, approve, dec.Reason, dec.DecidedBy); err != nil { + if errors.Is(err, storage.ErrNotFound) { + httpx.WriteError(w, http.StatusConflict, "already_resolved", "approval is unknown or already resolved") + return + } + httpx.WriteError(w, http.StatusInternalServerError, "internal_error", err.Error()) + return + } + + // Per the api/v1 contract, /approve returns the run's current state. + body, err := s.runViewBody(r, runID) + if err != nil { + // Resolution succeeded even if we can't read the run back; report success. + a, _ := s.approvals.Get(r.Context(), approvalID) + httpx.WriteJSON(w, http.StatusOK, map[string]any{"resolved": approvalView(a)}) + return + } + httpx.WriteJSON(w, http.StatusOK, body) +} + +// handleListApprovals lists approvals (GET /v1/approvals?status=pending). +func (s *Server) handleListApprovals(w http.ResponseWriter, r *http.Request) { + status := r.URL.Query().Get("status") + store := s.runs.Store() + if store == nil { + httpx.WriteJSON(w, http.StatusOK, []any{}) + return + } + list, err := store.ListApprovals(r.Context(), status) + if err != nil { + httpx.WriteError(w, http.StatusInternalServerError, "internal_error", err.Error()) + return + } + out := make([]map[string]any, 0, len(list)) + for _, a := range list { + out = append(out, approvalView(a)) + } + httpx.WriteJSON(w, http.StatusOK, out) +} + +// handleGetRun returns a run's state including any pending approval (GET /v1/runs/{id}). +func (s *Server) handleGetRun(w http.ResponseWriter, r *http.Request) { + body, err := s.runViewBody(r, r.PathValue("id")) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + httpx.WriteError(w, http.StatusNotFound, "not_found", "run not found") + return + } + if errors.Is(err, errNoStore) { + httpx.WriteError(w, http.StatusServiceUnavailable, "no_store", "no durable store configured") + return + } + httpx.WriteError(w, http.StatusInternalServerError, "internal_error", err.Error()) + return + } + httpx.WriteJSON(w, http.StatusOK, body) +} + +var errNoStore = errors.New("no durable store configured") + +// runViewBody builds the api/v1 Run view from the store, overlaying a pending +// approval (and the waiting_approval status) when one exists. +func (s *Server) runViewBody(r *http.Request, runID string) (map[string]any, error) { + store := s.runs.Store() + if store == nil { + return nil, errNoStore + } + rec, err := store.GetRun(r.Context(), runID) + if err != nil { + return nil, err + } + body := map[string]any{ + "id": rec.ID, + "name": rec.Name, + "status": rec.Status, + "haltReason": rec.HaltReason, + "usage": map[string]any{ + "tokens": rec.UsagePromptTokens + rec.UsageCompletionTokens, + "promptTokens": rec.UsagePromptTokens, + "completionTokens": rec.UsageCompletionTokens, + "dollars": rec.UsageDollars, + "loops": rec.UsageLoops, + }, + "createdAt": rec.CreatedAt, + "updatedAt": rec.UpdatedAt, + } + if pending, err := s.pendingForRun(r, runID); err == nil && len(pending) > 0 { + body["pendingApproval"] = approvalView(pending[0]) + body["status"] = "waiting_approval" + } + return body, nil +} + +// handleAdminApprovalsPage serves the embedded local approvals page. +func (s *Server) handleAdminApprovalsPage(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/html; charset=utf-8") + _, _ = w.Write(approvalsPage) +} + +// pendingForRun returns the run's pending approvals. +func (s *Server) pendingForRun(r *http.Request, runID string) ([]storage.ApprovalRecord, error) { + store := s.runs.Store() + if store == nil { + return nil, nil + } + all, err := store.ListApprovals(r.Context(), storage.ApprovalPending) + if err != nil { + return nil, err + } + var out []storage.ApprovalRecord + for _, a := range all { + if a.RunID == runID { + out = append(out, a) + } + } + return out, nil +} + +func decisionToBool(d string) (approve, ok bool) { + switch d { + case "approve": + return true, true + case "deny": + return false, true + default: + return false, false + } +} + +func approvalView(a storage.ApprovalRecord) map[string]any { + v := map[string]any{ + "id": a.ID, + "runId": a.RunID, + "stepIndex": a.StepIndex, + "tool": a.Tool, + "sideEffect": a.SideEffect, + "arguments": a.Arguments, + "status": a.Status, + "createdAt": a.CreatedAt, + } + if a.Reason != "" { + v["reason"] = a.Reason + } + if a.DecidedBy != "" { + v["decidedBy"] = a.DecidedBy + } + if a.DecidedAt != nil { + v["decidedAt"] = a.DecidedAt.Format(time.RFC3339) + } + return v +} diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 4ffef03..19645d7 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -13,6 +13,7 @@ import ( "net/http" "time" + "github.com/prashar32/riskkernel/internal/approval" "github.com/prashar32/riskkernel/internal/config" "github.com/prashar32/riskkernel/internal/gateway" "github.com/prashar32/riskkernel/internal/httpx" @@ -23,15 +24,16 @@ import ( // Server wires dependencies into an http.Handler. It holds no per-request state. type Server struct { - cfg *config.Config - gateway *gateway.Gateway - runs *runs.Manager - log *slog.Logger + cfg *config.Config + gateway *gateway.Gateway + runs *runs.Manager + approvals *approval.Gate + log *slog.Logger } // New constructs a Server. -func New(cfg *config.Config, gw *gateway.Gateway, mgr *runs.Manager, log *slog.Logger) *Server { - return &Server{cfg: cfg, gateway: gw, runs: mgr, log: log} +func New(cfg *config.Config, gw *gateway.Gateway, mgr *runs.Manager, gate *approval.Gate, log *slog.Logger) *Server { + return &Server{cfg: cfg, gateway: gw, runs: mgr, approvals: gate, log: log} } // Handler returns the root HTTP handler with all routes mounted. @@ -53,6 +55,13 @@ func (s *Server) Handler() http.Handler { // Public /v1 contract routes (authenticated). if s.runs != nil { mux.HandleFunc("GET /v1/checkpoints/{run_id}", s.requireAuth(s.handleGetCheckpoint)) + mux.HandleFunc("GET /v1/runs/{id}", s.requireAuth(s.handleGetRun)) + } + if s.approvals != nil { + mux.HandleFunc("POST /v1/runs/{id}/approve", s.requireAuth(s.handleApprove)) + mux.HandleFunc("GET /v1/approvals", s.requireAuth(s.handleListApprovals)) + // Local admin web page (Surface: human-in-the-loop, pull channel). + mux.HandleFunc("GET /admin/approvals", s.requireAuth(s.handleAdminApprovalsPage)) } // Further /v1 routes (full runs API) land in later build steps. diff --git a/internal/httpapi/server_test.go b/internal/httpapi/server_test.go index 0c134b5..2a2a5ac 100644 --- a/internal/httpapi/server_test.go +++ b/internal/httpapi/server_test.go @@ -1,21 +1,25 @@ package httpapi import ( + "context" "encoding/json" "io" "log/slog" "net/http" "net/http/httptest" "path/filepath" + "strings" "testing" + "time" + "github.com/prashar32/riskkernel/internal/approval" "github.com/prashar32/riskkernel/internal/config" "github.com/prashar32/riskkernel/internal/governor" "github.com/prashar32/riskkernel/internal/runs" "github.com/prashar32/riskkernel/internal/storage" ) -func newTestServer(t *testing.T, token string) (*Server, *runs.Manager) { +func newTestServer(t *testing.T, token string) (*Server, *runs.Manager, *approval.Gate) { t.Helper() store, err := storage.OpenSQLite(filepath.Join(t.TempDir(), "srv.db")) if err != nil { @@ -24,12 +28,13 @@ func newTestServer(t *testing.T, token string) (*Server, *runs.Manager) { t.Cleanup(func() { _ = store.Close() }) log := slog.New(slog.NewTextHandler(io.Discard, nil)) mgr := runs.NewManager(governor.Budget{Tokens: 100000}).WithStore(store, log) - srv := New(&config.Config{APIToken: token}, nil, mgr, log) - return srv, mgr + gate := approval.NewGate(store, approval.Policy{DefaultSafe: true}, nil, log) + srv := New(&config.Config{APIToken: token}, nil, mgr, gate, log) + return srv, mgr, gate } func TestHealthAndVersion(t *testing.T) { - srv, _ := newTestServer(t, "") + srv, _, _ := newTestServer(t, "") h := srv.Handler() for _, path := range []string{"/healthz", "/version"} { @@ -42,7 +47,7 @@ func TestHealthAndVersion(t *testing.T) { } func TestGetCheckpoint(t *testing.T) { - srv, mgr := newTestServer(t, "") + srv, mgr, _ := newTestServer(t, "") h := srv.Handler() // Seed a run with one recorded call → a checkpoint is written per step. @@ -80,8 +85,96 @@ func TestGetCheckpoint(t *testing.T) { } } +func TestApproveFlow(t *testing.T) { + srv, mgr, gate := newTestServer(t, "") + h := srv.Handler() + + mgr.Create(runs.CreateOptions{ID: "run-x"}) // persist the run row (FK) + + // A side-effecting tool call blocks awaiting approval. + type res struct { + d approval.Decision + err error + } + done := make(chan res, 1) + go func() { + d, _, err := gate.Request(context.Background(), approval.Request{ + RunID: "run-x", StepIndex: 1, Tool: "mcp://shell", SideEffect: "exec", + Arguments: map[string]any{"cmd": "rm -rf /tmp/x"}, + }) + done <- res{d, err} + }() + + // Wait until it's pending, then confirm the API surfaces it. + id := waitPendingAPI(t, h) + + // GET /v1/runs/run-x shows status waiting_approval + the pending approval. + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v1/runs/run-x", nil)) + var run map[string]any + _ = json.Unmarshal(w.Body.Bytes(), &run) + if run["status"] != "waiting_approval" || run["pendingApproval"] == nil { + t.Fatalf("run should be waiting_approval with a pending approval: %v", run) + } + + // Approve via the API (no approvalId → resolves the single pending one). + body := `{"decision":"approve","decidedBy":"tester"}` + w = httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/v1/runs/run-x/approve", strings.NewReader(body))) + if w.Code != http.StatusOK { + t.Fatalf("approve status = %d, body=%s", w.Code, w.Body.String()) + } + + select { + case r := <-done: + if r.err != nil || !r.d.Approved { + t.Fatalf("blocked call should be approved: %+v err=%v", r.d, r.err) + } + case <-time.After(2 * time.Second): + t.Fatal("approval via API did not unblock the call") + } + _ = id + + // Pending list is now empty. + w = httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v1/approvals?status=pending", nil)) + var list []map[string]any + _ = json.Unmarshal(w.Body.Bytes(), &list) + if len(list) != 0 { + t.Errorf("expected no pending approvals after approve, got %d", len(list)) + } +} + +func TestApprove_NoPending(t *testing.T) { + srv, mgr, _ := newTestServer(t, "") + h := srv.Handler() + mgr.Create(runs.CreateOptions{ID: "run-y"}) + + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/v1/runs/run-y/approve", strings.NewReader(`{"decision":"approve"}`))) + if w.Code != http.StatusConflict { + t.Fatalf("expected 409 when nothing pending, got %d", w.Code) + } +} + +func waitPendingAPI(t *testing.T, h http.Handler) string { + t.Helper() + for i := 0; i < 200; i++ { + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/v1/approvals?status=pending", nil)) + var list []map[string]any + _ = json.Unmarshal(w.Body.Bytes(), &list) + if len(list) == 1 { + return list[0]["id"].(string) + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("pending approval never appeared via API") + return "" +} + func TestGetCheckpoint_AuthRequired(t *testing.T) { - srv, _ := newTestServer(t, "sekret") + srv, _, _ := newTestServer(t, "sekret") h := srv.Handler() w := httptest.NewRecorder() diff --git a/internal/storage/approvals.go b/internal/storage/approvals.go new file mode 100644 index 0000000..8aa82e5 --- /dev/null +++ b/internal/storage/approvals.go @@ -0,0 +1,99 @@ +package storage + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "time" +) + +// CreateApproval persists a new (pending) approval request. +func (s *SQLite) CreateApproval(ctx context.Context, a ApprovalRecord) error { + args, err := json.Marshal(orEmptyMapAny(a.Arguments)) + if err != nil { + return fmt.Errorf("storage: marshal approval args: %w", err) + } + _, err = s.db.ExecContext(ctx, ` + INSERT INTO approvals (id, run_id, step_idx, tool, side_effect, arguments, status, reason, decided_by, created_at, decided_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?)`, + a.ID, a.RunID, a.StepIndex, a.Tool, a.SideEffect, string(args), + a.Status, a.Reason, a.DecidedBy, fmtTime(a.CreatedAt), fmtTimePtr(a.DecidedAt)) + if err != nil { + return fmt.Errorf("storage: create approval: %w", err) + } + return nil +} + +// GetApproval returns an approval by id, or ErrNotFound. +func (s *SQLite) GetApproval(ctx context.Context, id string) (ApprovalRecord, error) { + row := s.db.QueryRowContext(ctx, ` + SELECT id, run_id, step_idx, tool, side_effect, arguments, status, reason, decided_by, created_at, decided_at + FROM approvals WHERE id = ?`, id) + a, err := scanApproval(row) + if err == sql.ErrNoRows { + return ApprovalRecord{}, ErrNotFound + } + return a, err +} + +// ResolveApproval records a decision on a pending approval. It is a no-op if the +// approval is not currently pending (returns ErrNotFound so callers can detect a +// double-resolve or unknown id). +func (s *SQLite) ResolveApproval(ctx context.Context, id, status, reason, decidedBy string, decidedAt time.Time) error { + res, err := s.db.ExecContext(ctx, ` + UPDATE approvals SET status = ?, reason = ?, decided_by = ?, decided_at = ? + WHERE id = ? AND status = ?`, + status, reason, decidedBy, fmtTime(decidedAt), id, ApprovalPending) + if err != nil { + return fmt.Errorf("storage: resolve approval: %w", err) + } + n, _ := res.RowsAffected() + if n == 0 { + return ErrNotFound + } + return nil +} + +// ListApprovals returns approvals filtered by status ("" = all), newest first. +func (s *SQLite) ListApprovals(ctx context.Context, status string) ([]ApprovalRecord, error) { + q := `SELECT id, run_id, step_idx, tool, side_effect, arguments, status, reason, decided_by, created_at, decided_at + FROM approvals` + var rows *sql.Rows + var err error + if status == "" { + rows, err = s.db.QueryContext(ctx, q+` ORDER BY created_at DESC`) + } else { + rows, err = s.db.QueryContext(ctx, q+` WHERE status = ? ORDER BY created_at DESC`, status) + } + if err != nil { + return nil, fmt.Errorf("storage: list approvals: %w", err) + } + defer rows.Close() + var out []ApprovalRecord + for rows.Next() { + a, err := scanApproval(rows) + if err != nil { + return nil, err + } + out = append(out, a) + } + return out, rows.Err() +} + +func scanApproval(row rowScanner) (ApprovalRecord, error) { + var a ApprovalRecord + var args, created string + var decided sql.NullString + if err := row.Scan(&a.ID, &a.RunID, &a.StepIndex, &a.Tool, &a.SideEffect, &args, + &a.Status, &a.Reason, &a.DecidedBy, &created, &decided); err != nil { + return ApprovalRecord{}, err + } + a.Arguments = unmarshalMapAny(args) + a.CreatedAt = parseTime(created) + if decided.Valid && decided.String != "" { + t := parseTime(decided.String) + a.DecidedAt = &t + } + return a, nil +} diff --git a/internal/storage/migrations/00003_approvals.sql b/internal/storage/migrations/00003_approvals.sql new file mode 100644 index 0000000..385a60e --- /dev/null +++ b/internal/storage/migrations/00003_approvals.sql @@ -0,0 +1,24 @@ +-- +goose Up +-- Human-in-the-loop approvals. A side-effecting tool call that policy gates pauses +-- here as a pending row until a human approves or denies it. Resolved rows are an +-- audit trail of who allowed which side effect, when, and why. + +CREATE TABLE approvals ( + id TEXT PRIMARY KEY, + run_id TEXT NOT NULL REFERENCES runs(id), + step_idx INTEGER NOT NULL, + tool TEXT NOT NULL, + side_effect TEXT NOT NULL DEFAULT '', + arguments TEXT NOT NULL DEFAULT '{}', + status TEXT NOT NULL, -- pending | approved | denied + reason TEXT NOT NULL DEFAULT '', + decided_by TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL, + decided_at TEXT +); + +CREATE INDEX idx_approvals_run ON approvals(run_id); +CREATE INDEX idx_approvals_status ON approvals(status); + +-- +goose Down +-- Forward-only migrations (COMPATIBILITY.md). No down migration is provided. diff --git a/internal/storage/sqlite_test.go b/internal/storage/sqlite_test.go index 8c8dfed..5d595ab 100644 --- a/internal/storage/sqlite_test.go +++ b/internal/storage/sqlite_test.go @@ -253,6 +253,46 @@ func TestListRunsByStatus(t *testing.T) { } } +func TestApprovals(t *testing.T) { + s := openTemp(t) + ctx := context.Background() + now := time.Now().UTC() + mustRun(t, s, "run-ap", now) + + a := ApprovalRecord{ + ID: "ap-1", RunID: "run-ap", StepIndex: 3, Tool: "mcp://shell", SideEffect: "exec", + Arguments: map[string]any{"cmd": "ls"}, Status: ApprovalPending, CreatedAt: now, + } + if err := s.CreateApproval(ctx, a); err != nil { + t.Fatalf("CreateApproval: %v", err) + } + + pending, err := s.ListApprovals(ctx, ApprovalPending) + if err != nil || len(pending) != 1 { + t.Fatalf("ListApprovals pending = %v, %v", pending, err) + } + if pending[0].Arguments["cmd"] != "ls" { + t.Errorf("args not round-tripped: %+v", pending[0].Arguments) + } + + // Resolve it. + if err := s.ResolveApproval(ctx, "ap-1", ApprovalApproved, "ok", "alice", now.Add(time.Minute)); err != nil { + t.Fatalf("ResolveApproval: %v", err) + } + got, _ := s.GetApproval(ctx, "ap-1") + if got.Status != ApprovalApproved || got.DecidedBy != "alice" || got.DecidedAt == nil { + t.Fatalf("resolved approval = %+v", got) + } + + // Pending list now empty; double-resolve fails. + if pend, _ := s.ListApprovals(ctx, ApprovalPending); len(pend) != 0 { + t.Errorf("expected no pending after resolve, got %d", len(pend)) + } + if err := s.ResolveApproval(ctx, "ap-1", ApprovalDenied, "", "", now); !errors.Is(err, ErrNotFound) { + t.Errorf("double-resolve should return ErrNotFound, got %v", err) + } +} + func mustRun(t *testing.T, s *SQLite, id string, now time.Time) { t.Helper() if err := s.UpsertRun(context.Background(), RunRecord{ diff --git a/internal/storage/store.go b/internal/storage/store.go index 290097b..45724ba 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -89,6 +89,28 @@ type CheckpointRecord struct { CreatedAt time.Time } +// Approval status values. +const ( + ApprovalPending = "pending" + ApprovalApproved = "approved" + ApprovalDenied = "denied" +) + +// ApprovalRecord is a human-in-the-loop gate on a side-effecting tool call. +type ApprovalRecord struct { + ID string + RunID string + StepIndex int32 + Tool string + SideEffect string + Arguments map[string]any + Status string // pending | approved | denied + Reason string + DecidedBy string + CreatedAt time.Time + DecidedAt *time.Time +} + // LedgerTotals aggregates spend for audit/reporting. type LedgerTotals struct { RunID string @@ -125,6 +147,15 @@ type Store interface { // AppendToolCall records a tool invocation. AppendToolCall(ctx context.Context, t ToolCallRecord) error + // CreateApproval persists a new (pending) approval request. + CreateApproval(ctx context.Context, a ApprovalRecord) error + // GetApproval returns an approval by id, or ErrNotFound. + GetApproval(ctx context.Context, id string) (ApprovalRecord, error) + // ResolveApproval records a decision (approved/denied) on a pending approval. + ResolveApproval(ctx context.Context, id, status, reason, decidedBy string, decidedAt time.Time) error + // ListApprovals returns approvals filtered by status ("" = all), newest first. + ListApprovals(ctx context.Context, status string) ([]ApprovalRecord, error) + // SaveCheckpoint appends a crash-resumable checkpoint. SaveCheckpoint(ctx context.Context, c CheckpointRecord) error // LatestCheckpoint returns a run's most recent checkpoint, or ErrNotFound.