Run executes fn and persists one ledger step when Store is non-nil. When ledger is disabled (Store nil), fn still runs; OnStep is invoked if set.
(ctx context.Context, stepType, backend string, input map[string]any, fn func() (output map[string]any, err error))
| 34 | // Run executes fn and persists one ledger step when Store is non-nil. |
| 35 | // When ledger is disabled (Store nil), fn still runs; OnStep is invoked if set. |
| 36 | func (w *StepWriter) Run(ctx context.Context, stepType, backend string, input map[string]any, fn func() (output map[string]any, err error)) error { |
| 37 | if w == nil { |
| 38 | _, err := fn() |
| 39 | return err |
| 40 | } |
| 41 | t0 := time.Now() |
| 42 | inBytes, _ := json.Marshal(input) |
| 43 | out, err := fn() |
| 44 | lat := time.Since(t0).Milliseconds() |
| 45 | status := "success" |
| 46 | errMsg := "" |
| 47 | if err != nil { |
| 48 | status = "failed" |
| 49 | errMsg = err.Error() |
| 50 | } |
| 51 | if w.OnStep != nil { |
| 52 | w.OnStep(ctx, stepType, backend, status, t0, lat) |
| 53 | } |
| 54 | if w.Store == nil { |
| 55 | return err |
| 56 | } |
| 57 | if out == nil { |
| 58 | out = map[string]any{} |
| 59 | } |
| 60 | outBytes, _ := json.Marshal(out) |
| 61 | meta := map[string]any{"step": stepType} |
| 62 | if err != nil && stepType == StepPolicyCheck { |
| 63 | meta["policy_rejected"] = true |
| 64 | } |
| 65 | metaBytes, _ := json.Marshal(meta) |
| 66 | _ = w.Store.AppendStep(ctx, requests.StepRow{ |
| 67 | RequestID: w.RequestID, |
| 68 | StepIndex: w.next, |
| 69 | StepType: stepType, |
| 70 | InputJSON: inBytes, |
| 71 | OutputJSON: outBytes, |
| 72 | Backend: backend, |
| 73 | Status: status, |
| 74 | Error: errMsg, |
| 75 | LatencyMs: lat, |
| 76 | MetadataJSON: metaBytes, |
| 77 | }) |
| 78 | w.next++ |
| 79 | return err |
| 80 | } |
| 81 | |
| 82 | // Index returns the next step index without incrementing (for branching). |
| 83 | func (w *StepWriter) Index() int { |