Skip to content

Commit bb1d3c1

Browse files
author
gluck
committed
feat(doctor): figaro doctor provider - the round-trip ledger
`figaro doctor` had six subcommands and not one of them could answer "why is this aria not responding". During a real incident the surfaces said: figaro status 94f0752b -> idle figaro queue ls -> empty logs.jsonl -> nothing traces.jsonl -> the answer, to anyone willing to write a JSON scanner, and only for turns that had already ENDED The request that was still hanging - the only one that mattered - appeared nowhere at all, because spans export on end. figaro doctor provider --id 94f0752b time aria status duration req retry endpoint 17:09:17.298 94f0752b 200 11.1s 712.4 KiB - /v1/messages 17:09:36.101 94f0752b 429 1.1s 712.4 KiB 1h0m0s /v1/messages 18:56:02.550 94f0752b … in flight 712.4 KiB - /v1/messages 1 of 3 round-trips were refused for quota. the provider asked for up to 1h0m0s; a wait that long is a usage window, not a throttle, and no amount of retrying will shorten it. limit resets 2026-08-14T20:00:00Z 1 request(s) still in flight. The table is evidence; the paragraph under it is the verdict, and the verdict is the product. It prints nothing when the ledger is healthy - a diagnostic that always says something trains people to stop reading it. Reads the transport's ring over a new angelus.provider_ledger RPC rather than adding hundreds of rows to angelus.status, which is polled. Against a daemon that predates the method the CLI says so and names the fix.
1 parent 67397cd commit bb1d3c1

5 files changed

Lines changed: 400 additions & 6 deletions

File tree

‎internal/angelus/authz_wiring_test.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@ func TestEveryServedMethodIsGuarded(t *testing.T) {
119119
rpc.MethodCreate, rpc.MethodFork, rpc.MethodPromote, rpc.MethodKill,
120120
rpc.MethodList, rpc.MethodAttach, rpc.MethodBind, rpc.MethodResolve,
121121
rpc.MethodUnbind, rpc.MethodStatus, rpc.MethodSaveBindings, rpc.MethodAriaRead,
122+
rpc.MethodProviderLedger,
122123
} {
123124
if hs.Map[m] == nil {
124125
t.Fatalf("method %s missing from the guarded map", m)
Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
package angelus
2+
3+
import (
4+
"context"
5+
"encoding/json"
6+
"net/http"
7+
"net/http/httptest"
8+
"strings"
9+
"testing"
10+
11+
"github.com/jack-work/figaro/internal/rpc"
12+
"github.com/jack-work/figaro/internal/wirelog"
13+
)
14+
15+
// The ledger is only worth having if it survives the trip to the CLI. This
16+
// drives a real refused round-trip through the real transport and reads it
17+
// back through the handler, because that is the path an operator uses at
18+
// three in the morning.
19+
func TestProviderLedgerHandlerReportsARefusal(t *testing.T) {
20+
wirelog.ResizeLedger(64)
21+
t.Cleanup(func() { wirelog.ResizeLedger(256) })
22+
23+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
24+
w.Header().Set("Retry-After", "3600")
25+
w.Header().Set("request-id", "req_quota")
26+
w.Header().Set("anthropic-ratelimit-input-tokens-reset", "2026-08-14T20:00:00Z")
27+
w.WriteHeader(http.StatusTooManyRequests)
28+
}))
29+
defer srv.Close()
30+
31+
client := &http.Client{Transport: &wirelog.Transport{Inner: http.DefaultTransport}}
32+
req, err := http.NewRequestWithContext(
33+
wirelog.WithAria(context.Background(), "94f0752b"),
34+
http.MethodPost, srv.URL, strings.NewReader("{}"))
35+
if err != nil {
36+
t.Fatal(err)
37+
}
38+
resp, err := client.Do(req)
39+
if err != nil {
40+
t.Fatal(err)
41+
}
42+
resp.Body.Close()
43+
44+
h := &handlers{}
45+
params, _ := json.Marshal(rpc.ProviderLedgerRequest{Aria: "94f0752b"})
46+
out, err := h.providerLedger(context.Background(), params)
47+
if err != nil {
48+
t.Fatal(err)
49+
}
50+
got, ok := out.(rpc.ProviderLedgerResponse)
51+
if !ok {
52+
t.Fatalf("unexpected response type %T", out)
53+
}
54+
if len(got.Rounds) != 1 {
55+
t.Fatalf("want 1 round for the aria, got %d", len(got.Rounds))
56+
}
57+
r := got.Rounds[0]
58+
if r.Status != http.StatusTooManyRequests {
59+
t.Errorf("status = %d", r.Status)
60+
}
61+
if r.RetryAfterS != 3600 {
62+
t.Errorf("retry_after_s = %d; this is the field the whole feature exists for", r.RetryAfterS)
63+
}
64+
if r.RequestID != "req_quota" {
65+
t.Errorf("request_id = %q", r.RequestID)
66+
}
67+
if r.StartedAtMS == 0 {
68+
t.Error("started_at_ms not carried to the wire")
69+
}
70+
if r.RateLimit["input-tokens-reset"] != "2026-08-14T20:00:00Z" {
71+
t.Errorf("ratelimit headers lost: %v", r.RateLimit)
72+
}
73+
}
74+
75+
// The aria filter has to actually filter, or "why is 94f0752b stuck" comes
76+
// back as the whole daemon's traffic.
77+
func TestProviderLedgerFiltersByAriaAndLimit(t *testing.T) {
78+
wirelog.ResizeLedger(64)
79+
t.Cleanup(func() { wirelog.ResizeLedger(256) })
80+
81+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
82+
w.WriteHeader(http.StatusOK)
83+
}))
84+
defer srv.Close()
85+
86+
client := &http.Client{Transport: &wirelog.Transport{Inner: http.DefaultTransport}}
87+
for _, aria := range []string{"aaaa1111", "bbbb2222", "aaaa1111", "aaaa1111"} {
88+
req, _ := http.NewRequest(http.MethodGet, srv.URL, nil)
89+
req = req.WithContext(wirelog.WithAria(req.Context(), aria))
90+
resp, err := client.Do(req)
91+
if err != nil {
92+
t.Fatal(err)
93+
}
94+
resp.Body.Close()
95+
}
96+
97+
h := &handlers{}
98+
params, _ := json.Marshal(rpc.ProviderLedgerRequest{Aria: "aaaa1111"})
99+
out, _ := h.providerLedger(context.Background(), params)
100+
got := out.(rpc.ProviderLedgerResponse)
101+
if len(got.Rounds) != 3 {
102+
t.Fatalf("aria filter returned %d rounds, want 3", len(got.Rounds))
103+
}
104+
if got.Retained != 4 {
105+
t.Errorf("retained = %d, want 4 - the caller must be able to tell a filter from an empty ring", got.Retained)
106+
}
107+
108+
params, _ = json.Marshal(rpc.ProviderLedgerRequest{Aria: "aaaa1111", Limit: 2})
109+
out, _ = h.providerLedger(context.Background(), params)
110+
got = out.(rpc.ProviderLedgerResponse)
111+
if len(got.Rounds) != 2 {
112+
t.Fatalf("limit ignored: %d rounds", len(got.Rounds))
113+
}
114+
// Newest last: a limit takes the tail, not the head.
115+
if got.Rounds[0].Seq >= got.Rounds[1].Seq {
116+
t.Error("rounds must come back oldest-first")
117+
}
118+
}

‎internal/cli/cli.go‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -876,7 +876,7 @@ slot, so a lone positional is then the aria. With no figaro available
876876
first: if the casting then fails, that partial is spelled out.
877877
878878
ATTENDING A FORM, with no arguments at all, mints a figaro for it and
879-
casts it: the other entrance to `+"`figaro new -C`"+`, the same operation from
879+
casts it: the other entrance to ` + "`figaro new -C`" + `, the same operation from
880880
the other end. There, -O/-S dress the ARIA, because the form already
881881
exists; attending an ARIA they mint the ROLE, for the same reason.
882882
@@ -1471,15 +1471,18 @@ flow writes both through the daemon, so a client never has to know the path.`,
14711471
r.Register(&cmdkit.Command{
14721472
Name: "doctor",
14731473
Group: "System",
1474-
Short: "Store maintenance: gc removes dead channels; schema reports channel versions; mem reports the daemon's footprint; librettos recounts derived-form observers; skills lists what the binary ships",
1475-
Usage: "doctor <gc [--dry-run] | schema | term | mem [-j] | librettos [--dry-run] | skills [-j]>",
1474+
Short: "Store and provider health: gc removes dead channels; schema reports channel versions; mem reports the daemon's footprint; provider shows recent provider round-trips (including in-flight); librettos recounts derived-form observers; skills lists what the binary ships",
1475+
Usage: "doctor <gc [--dry-run] | schema | term | mem [-j] | provider [--id <aria>] [-c N] [-j] | librettos [--dry-run] | skills [-j]>",
14761476
Flags: []cmdkit.FlagDef{
14771477
{Long: "dry-run", Short: "n", IsBool: true, Description: "Report what would be removed without touching the store"},
1478-
{Long: "json", Short: "j", IsBool: true, Description: "Machine-readable output (mem)"},
1478+
{Long: "json", Short: "j", IsBool: true, Description: "Machine-readable output (mem, provider)"},
1479+
{Long: "id", Description: "Restrict to one aria (provider)"},
1480+
// -n is already dry-run on this command, so count takes -c.
1481+
{Long: "count", Short: "c", Description: "How many rows to show (provider); default 20"},
14791482
},
14801483
Run: func(ctx *cmdkit.RunContext) error {
14811484
if len(ctx.Args) != 1 {
1482-
return fmt.Errorf("usage: doctor <gc [--dry-run] | schema | term | mem [-j] | librettos [--dry-run] | skills [-j]>")
1485+
return fmt.Errorf("usage: doctor <gc [--dry-run] | schema | term | mem [-j] | provider [--id <aria>] [-c N] [-j] | librettos [--dry-run] | skills [-j]>")
14831486
}
14841487
switch ctx.Args[0] {
14851488
case "gc":
@@ -1490,12 +1493,14 @@ flow writes both through the daemon, so a client never has to know the path.`,
14901493
return runDoctorTerm()
14911494
case "mem":
14921495
return runDoctorMem(ctx.BoolFlag("json"))
1496+
case "provider":
1497+
return runDoctorProvider(ctx.Flag("id"), ctx.Flag("count"), ctx.BoolFlag("json"))
14931498
case "librettos":
14941499
return runDoctorLibrettos(ctx.BoolFlag("dry-run"))
14951500
case "skills":
14961501
return runDoctorSkills(ctx.BoolFlag("json"))
14971502
}
1498-
return fmt.Errorf("usage: doctor <gc [--dry-run] | schema | term | mem [-j] | librettos [--dry-run] | skills [-j]>")
1503+
return fmt.Errorf("usage: doctor <gc [--dry-run] | schema | term | mem [-j] | provider [--id <aria>] [-c N] [-j] | librettos [--dry-run] | skills [-j]>")
14991504
},
15001505
})
15011506

‎internal/cli/doctor_provider.go‎

Lines changed: 171 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,171 @@
1+
package cli
2+
3+
import (
4+
"context"
5+
"encoding/json"
6+
"fmt"
7+
"os"
8+
"sort"
9+
"strconv"
10+
"strings"
11+
"time"
12+
13+
"github.com/jack-work/figaro/internal/angelus"
14+
"github.com/jack-work/figaro/internal/rpc"
15+
"github.com/jack-work/figaro/internal/transport"
16+
)
17+
18+
// runDoctorProvider answers the question no figaro surface could answer:
19+
// "this aria is not responding - what is the provider actually doing?"
20+
//
21+
// `status` said idle. `queue` was empty. The log said nothing. The trace file
22+
// had the answer - a 429 with an hour-long Retry-After - but only for turns
23+
// that had already ended, because spans export on end, and only to someone
24+
// willing to write a JSON scanner. Meanwhile the request that was STILL
25+
// hanging appeared nowhere at all.
26+
//
27+
// This reads the transport's own ring: rows are written at departure and
28+
// completed in place, so an in-flight request is visible while it is the
29+
// problem rather than after it is history.
30+
func runDoctorProvider(ariaID, count string, asJSON bool) error {
31+
limit := 20
32+
if count != "" {
33+
n, err := strconv.Atoi(count)
34+
if err != nil || n < 0 {
35+
return fmt.Errorf("-n takes a count, got %q", count)
36+
}
37+
limit = n
38+
}
39+
40+
cli, err := angelus.DialClient(transport.UnixEndpoint(angelusSocketPath()))
41+
if err != nil {
42+
return fmt.Errorf("no angelus running: %w", err)
43+
}
44+
defer cli.Close()
45+
46+
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
47+
defer cancel()
48+
resp, err := cli.ProviderLedger(ctx, ariaID, limit)
49+
if err != nil {
50+
return fmt.Errorf("the running angelus cannot report provider round-trips; `figaro stop` and retry: %w", err)
51+
}
52+
53+
if asJSON {
54+
enc := json.NewEncoder(os.Stdout)
55+
enc.SetIndent("", " ")
56+
return enc.Encode(resp)
57+
}
58+
59+
if len(resp.Rounds) == 0 {
60+
if resp.Retained == 0 {
61+
fmt.Println("no provider round-trips recorded yet")
62+
} else {
63+
fmt.Printf("no round-trips for that filter (%d retained overall)\n", resp.Retained)
64+
}
65+
return nil
66+
}
67+
68+
fmt.Printf("%-12s %-9s %-6s %-9s %-9s %-8s %s\n",
69+
"time", "aria", "status", "duration", "req", "retry", "endpoint")
70+
for _, r := range resp.Rounds {
71+
when := time.UnixMilli(r.StartedAtMS).Format("15:04:05.000")
72+
aria := r.Aria
73+
if aria == "" {
74+
aria = "-"
75+
}
76+
77+
status, dur := "…", "in flight"
78+
if !r.InFlight {
79+
status = strconv.Itoa(r.Status)
80+
if r.Status == 0 {
81+
status = "err"
82+
}
83+
dur = humanMillis(r.DurationMS)
84+
} else {
85+
// An in-flight row's age is the number that matters: a request
86+
// out for four minutes is the story.
87+
dur = "in flight " + humanMillis(time.Since(time.UnixMilli(r.StartedAtMS)).Milliseconds())
88+
}
89+
90+
retry := "-"
91+
if r.RetryAfterS > 0 {
92+
retry = (time.Duration(r.RetryAfterS) * time.Second).String()
93+
}
94+
95+
fmt.Printf("%-12s %-9s %-6s %-9s %-9s %-8s %s\n",
96+
when, aria, status, dur, humanBytes(r.ReqBytes), retry, shortEndpoint(r.URL))
97+
}
98+
99+
summarizeProviderTrouble(resp.Rounds)
100+
return nil
101+
}
102+
103+
// summarizeProviderTrouble names the diagnosis rather than leaving it in the
104+
// table. The table is evidence; an operator wants the verdict.
105+
func summarizeProviderTrouble(rounds []rpc.ProviderRound) {
106+
var refused, inFlight int
107+
var worstRetry int64
108+
var reset string
109+
for _, r := range rounds {
110+
if r.InFlight {
111+
inFlight++
112+
}
113+
if r.Status == 429 || r.Status == 529 {
114+
refused++
115+
if r.RetryAfterS > worstRetry {
116+
worstRetry = r.RetryAfterS
117+
}
118+
for _, k := range sortedRateLimitKeys(r.RateLimit) {
119+
if strings.Contains(k, "reset") && reset == "" {
120+
reset = r.RateLimit[k]
121+
}
122+
}
123+
}
124+
}
125+
if refused == 0 && inFlight == 0 {
126+
return
127+
}
128+
fmt.Println()
129+
if refused > 0 {
130+
fmt.Printf("%d of %d round-trips were refused for quota.\n", refused, len(rounds))
131+
if worstRetry > 0 {
132+
fmt.Printf("the provider asked for up to %s; a wait that long is a usage window,\n",
133+
(time.Duration(worstRetry) * time.Second).String())
134+
fmt.Println("not a throttle, and no amount of retrying will shorten it.")
135+
}
136+
if reset != "" {
137+
fmt.Printf("limit resets %s\n", reset)
138+
}
139+
}
140+
if inFlight > 0 {
141+
fmt.Printf("%d request(s) still in flight.\n", inFlight)
142+
}
143+
}
144+
145+
func sortedRateLimitKeys(m map[string]string) []string {
146+
out := make([]string, 0, len(m))
147+
for k := range m {
148+
out = append(out, k)
149+
}
150+
sort.Strings(out)
151+
return out
152+
}
153+
154+
func humanMillis(ms int64) string {
155+
if ms < 1000 {
156+
return fmt.Sprintf("%dms", ms)
157+
}
158+
return (time.Duration(ms) * time.Millisecond).Round(100 * time.Millisecond).String()
159+
}
160+
161+
// shortEndpoint keeps the part of a URL that distinguishes one call from
162+
// another; the host repeats on every row and earns no column.
163+
func shortEndpoint(raw string) string {
164+
if i := strings.Index(raw, "://"); i >= 0 {
165+
raw = raw[i+3:]
166+
}
167+
if i := strings.Index(raw, "/"); i >= 0 {
168+
return raw[i:]
169+
}
170+
return raw
171+
}

0 commit comments

Comments
 (0)