daemon/collect.bend

cswap-bend-cli@0.1.0.0 · 106 lines · 17 definitions

1import Base
2import cswap-bend-cli@0.1.0.0/json.bend as Json
3import cswap-bend-cli@0.1.0.0/act.bend as Act
4import cswap-bend-cli@0.1.0.0/store/base.bend as SBase
5import cswap-bend-cli@0.1.0.0/policy/types.bend as T
6import cswap-bend-cli@0.1.0.0/policy/usage.bend as PU
7import cswap-bend-cli@0.1.0.0/daemon/config.bend as Cfg
8import cswap-bend-cli@0.1.0.0/daemon/gather.bend as G
9import cswap-bend-cli@0.1.0.0/daemon/view.bend as V
10import cswap-bend-cli@0.1.0.0/daemon/plan.bend as Plan
11import cswap-bend-cli@0.1.0.0/daemon/fetch.bend as Fetch
12
13# The fetch half of a tick: poll what is due (the active account and one
14# candidate, then everyone when the active account is shaky) and hand back the
15# snapshot with the refreshed cache rows. Kept apart from `tick` so the policy
16# never shares a module with the usage and switch code.
17
18def ActM(-A: Type) -> Type:
19  Act.of(A)
20
21def ActM.pure(-A: Type, x: A) -> Act.of(A):
22  Act.pure(A, x)
23
24def ActM.bind(-A: Type, -B: Type, m: Act.of(A), f: A -> Act.of(B)) -> Act.of(B):
25  Act.bind(A, B, m, f)
26
27def is_token_expired(+u: T.Usage) -> Bool:
28  match u:
29    case T.Usage{s, five, fr, seven, sr, ss, fa, bo, er, fl, np, pi, ad, hu, cu, T.TokenExpired{}}:
30      True{}
31    case _:
32      False{}
33
34def escalate2(+h: Maybe<&2, Nat>, +expired: Bool, +thr: U32) -> Bool:
35  match h:
36    case None{}:
37      Bool.not(expired)
38    case Some{x}:
39      Nat.is_ge(Nat.sub(T.full(), x), Nat.sub(U32.to_nat(thr), 1500n))
40
41# Fetch every candidate too when the active account's headroom is unknown (and
42# the token is not just expired on an idle Claude Code), or it is within 15
43# points of the threshold.
44def should_escalate(+m: Maybe<&2, T.Usage>, +now: Nat, +thr: U32, +ms: List<&2, String>) -> Bool:
45  match m:
46    case None{}:
47      False{}
48    case Some{u}:
49      escalate2(PU.headroom(u, now, ms), is_token_expired(u), thr)
50
51def fetch_pass(+ctx: SBase.Ctx, +cfg: Cfg.Cfg, +targets: List<&2, G.Ent>, +active: Maybe<&2, String>, +now: U32, +rows: Json.Json) -> Act.of(Json.Json):
52  Act.cond(Json.Json, List.is_empty(&2, G.Ent, targets), Act.pure(Json.Json, rows),
53    do ActM<Json.Json>:
54      Fetch.fetch_all(ctx, cfg, targets, active, now)
55      Act.lift(Json.Json, G.read_rows(SBase.root_of(ctx))))
56
57def escalation(+ctx: SBase.Ctx, +es: List<&2, G.Ent>, +active: String, +rows1: Json.Json, +rows0: Json.Json, +now: U32, +cfg: Cfg.Cfg) -> Act.of(Json.Json):
58  fetch_pass(ctx, cfg, Plan.escalated(es, V.usages_of(rows1, es, Some{active}, now), Some{active}, Plan.first_pass(es, V.usages_of(rows0, es, Some{active}, now), Some{active}, V.nat(now), V.pol_cfg(cfg)), V.nat(now), V.pol_cfg(cfg), Cfg.cfg_models(cfg)), Some{active}, now, rows1)
59
60def shaky(+es: List<&2, G.Ent>, +active: String, +rows1: Json.Json, +now: U32, +cfg: Cfg.Cfg) -> Bool:
61  should_escalate(V.usage_for(V.usages_of(rows1, es, Some{active}, now), V.slot_nat_of(active)), V.nat(now), Cfg.cfg_threshold(cfg), Cfg.cfg_models(cfg))
62
63def second_pass(+ctx: SBase.Ctx, +es: List<&2, G.Ent>, +active: String, +rows1: Json.Json, +rows0: Json.Json, +now: U32, +cfg: Cfg.Cfg) -> Act.of(Json.Json):
64  Act.cond(Json.Json, shaky(es, active, rows1, now, cfg),
65    escalation(ctx, es, active, rows1, rows0, now, cfg),
66    Act.pure(Json.Json, rows1))
67
68# Fetch what is due, in two passes (the active account and one candidate; then
69# everyone when the active account is shaky), and answer the rows as they
70# stand.
71def collect(+ctx: SBase.Ctx, +es: List<&2, G.Ent>, +active: String, +rows0: Json.Json, +now: U32, +cfg: Cfg.Cfg) -> Act.of(Json.Json):
72  do ActM<Json.Json>:
73    rows1 : Json.Json <- fetch_pass(ctx, cfg, Plan.first_pass(es, V.usages_of(rows0, es, Some{active}, now), Some{active}, V.nat(now), V.pol_cfg(cfg)), Some{active}, now, rows0)
74    second_pass(ctx, es, active, rows1, rows0, now, cfg)
75
76def snap_with(+s: G.Snap, +rows: Json.Json) -> G.Snap:
77  match s:
78    case G.Snap{seq, live, active, st, rows0, ents}:
79      G.Snap{seq, live, active, st, rows, ents}
80
81def refresh2(+ctx: SBase.Ctx, +cfg: Cfg.Cfg, +now: U32, +s: G.Snap, +a: String) -> Act.of(G.Snap):
82  do ActM<G.Snap>:
83    rows : Json.Json <- collect(ctx, G.snap_ents(s), a, G.snap_rows(s), now, cfg)
84    return snap_with(s, rows)
85
86# The snapshot with the due accounts fetched; unchanged without an active account.
87def refresh3(+a: Maybe<&2, String>, +ctx: SBase.Ctx, +cfg: Cfg.Cfg, +now: U32, +s: G.Snap) -> Act.of(G.Snap):
88  match a:
89    case None{}:
90      Act.pure(G.Snap, s)
91    case Some{x}:
92      refresh2(ctx, cfg, now, s, x)
93
94# The snapshot with the due accounts fetched; unchanged without an active account.
95def refresh(+ctx: SBase.Ctx, +cfg: Cfg.Cfg, +now: U32, +s: G.Snap) -> Act.of(G.Snap):
96  refresh3(G.snap_active(s), ctx, cfg, now, s)
97
98def some(+s: G.Snap) -> Act.of(Maybe<&2, G.Snap>):
99  Act.pure(Maybe<&2, G.Snap>, Some{s})
100
101def refresh_opt(+ctx: SBase.Ctx, +cfg: Cfg.Cfg, +now: U32, +m: Maybe<&2, G.Snap>) -> Act.of(Maybe<&2, G.Snap>):
102  match m:
103    case None{}:
104      Act.pure(Maybe<&2, G.Snap>, None{})
105    case Some{s}:
106      Act.bind(G.Snap, Maybe<&2, G.Snap>, refresh(ctx, cfg, now, s), r => some(r))