fix(office): stop over-budget policies from spamming duplicate alerts (#3287)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-12 00:43:56 +08:00
committed by GitHub
parent fd9b78e58d
commit d93b86a338
15 changed files with 2229 additions and 89 deletions
@@ -0,0 +1,17 @@
package costs
import "expvar"
// budgetClaimFailuresTotal counts claim-store failures encountered while
// evaluating a budget policy, exported through the stdlib expvar surface at
// /debug/vars in dev mode. Shaped as an expvar.Map, matching
// cost_events_written_total / cost_events_dropped_total in
// internal/office/service/cost_metrics.go, rather than a plain Int, so
// /debug/vars stays consistent. The "op" label has exactly one value,
// "claim": it covers both a policy's own level claim and its companion
// alert claim, and does not grow — a failed claim discard during a policy
// update is a different, already non-silent failure and is deliberately
// not counted here.
var budgetClaimFailuresTotal = expvar.NewMap("budget_claim_failures_total")
const budgetClaimFailureLabel = "op=claim"
@@ -0,0 +1,605 @@
package costs_test
import (
"context"
"errors"
"expvar"
"sync"
"testing"
"time"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/office/costs"
"github.com/kandev/kandev/internal/office/models"
)
// claimFailureCount reads the current value of budget_claim_failures_total's
// "op=claim" entry. Tests must compare before/after deltas, never the
// absolute value: the expvar.Map is process-global and accumulates across
// every test in this package.
func claimFailureCount(t *testing.T) int64 {
t.Helper()
v := expvar.Get("budget_claim_failures_total")
if v == nil {
return 0
}
m, ok := v.(*expvar.Map)
if !ok {
t.Fatalf("budget_claim_failures_total is not an expvar.Map, got %T", v)
}
kv := m.Get("op=claim")
if kv == nil {
return 0
}
iv, ok := kv.(*expvar.Int)
if !ok {
t.Fatalf("op=claim counter is not an expvar.Int, got %T", kv)
}
return iv.Value()
}
// REQ-OFFICE-COSTS-002: budget notifications fire on crossings, not on
// every evaluation. See docs/specs/office/requirements/costs.md and
// docs/specs/office/system-design/costs-03.md / costs-04.md.
func newIdempotencyTestPolicy(scopeID string, limit int64) *models.BudgetPolicy {
return &models.BudgetPolicy{
WorkspaceID: "ws-1",
ScopeType: "agent",
ScopeID: scopeID,
LimitSubcents: limit,
Period: "monthly",
AlertThresholdPct: 80,
ActionOnExceed: "notify_only",
}
}
// monthlyPeriodKey mirrors the RFC3339-UTC month-start rendering the
// service computes internally, for tests that assert on stored claim rows.
func monthlyPeriodKey() string {
now := time.Now().UTC()
return time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, time.UTC).Format(time.RFC3339)
}
func TestCheckBudget_EvaluateTwice_AlertEmitsOnce(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-alert-twice")
agents := &repoAgents{repo: repo}
spy := &budgetActivitySpy{}
svc := costs.NewCostService(repo, logger.Default(), spy, agents, agents)
policy := newIdempotencyTestPolicy("agent-alert-twice", 1000)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-alert-twice", "task-1", 850)
first, err := svc.CheckBudget(ctx, "ws-1", "agent-alert-twice", "proj-1")
if err != nil {
t.Fatalf("first CheckBudget: %v", err)
}
if !first[0].AlertFired || !first[0].AlertSubmitted {
t.Fatalf("first evaluation: AlertFired=%v AlertSubmitted=%v, want true/true", first[0].AlertFired, first[0].AlertSubmitted)
}
second, err := svc.CheckBudget(ctx, "ws-1", "agent-alert-twice", "proj-1")
if err != nil {
t.Fatalf("second CheckBudget: %v", err)
}
if !second[0].AlertFired {
t.Error("second evaluation should still report the level reached")
}
if second[0].AlertSubmitted {
t.Error("second evaluation must not resubmit an already-claimed level")
}
if got := spy.count("budget.alert"); got != 1 {
t.Fatalf("budget.alert submissions = %d, want 1", got)
}
}
func TestCheckBudget_EvaluateTwice_ExceededEmitsOnce(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-exceed-twice")
agents := &repoAgents{repo: repo}
spy := &budgetActivitySpy{}
svc := costs.NewCostService(repo, logger.Default(), spy, agents, agents)
policy := newIdempotencyTestPolicy("agent-exceed-twice", 500)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-exceed-twice", "task-1", 600)
for i := 0; i < 2; i++ {
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-exceed-twice", "proj-1"); err != nil {
t.Fatalf("CheckBudget[%d]: %v", i, err)
}
}
if got := spy.count("budget.exceeded"); got != 1 {
t.Fatalf("budget.exceeded submissions = %d, want 1", got)
}
}
// TestCheckBudget_ExceededClaimsCompanionAlert covers AC-OFFICE-COSTS-002.5
// on a healthy store: spend jumping straight past the limit still holds
// the alert-level claim, so a later evaluation landing back in the alert
// band does not emit a budget.alert describing a spend reduction.
func TestCheckBudget_ExceededClaimsCompanionAlert(t *testing.T) {
repo, queryRow, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-companion")
agents := &repoAgents{repo: repo}
spy := &budgetActivitySpy{}
svc := costs.NewCostService(repo, logger.Default(), spy, agents, agents)
policy := newIdempotencyTestPolicy("agent-companion", 500)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
// Jumps straight past the limit; the alert band is never reached on its
// own by a rollup query, only inferred via the companion claim.
insertBudgetTestCostEvent(t, execSQL, "agent-companion", "task-1", 600)
results, err := svc.CheckBudget(ctx, "ws-1", "agent-companion", "proj-1")
if err != nil {
t.Fatalf("CheckBudget: %v", err)
}
if !results[0].ExceededSubmitted {
t.Fatal("expected the exceeded row to be submitted")
}
if results[0].AlertSubmitted {
t.Fatal("the alert-level field must be false: only budget.exceeded was emitted (AC-OFFICE-COSTS-002.13)")
}
var count int
if err := queryRow(
`SELECT COUNT(*) FROM office_budget_claims WHERE policy_id = ? AND period_key = ? AND level = 'alert'`,
policy.ID, monthlyPeriodKey(),
).Scan(&count); err != nil {
t.Fatalf("count companion claim: %v", err)
}
if count != 1 {
t.Fatalf("companion alert-level claim rows = %d, want 1", count)
}
}
// TestCheckBudget_CompanionAlertClaimFault_StillSubmitsExceeded covers the
// AC-OFFICE-COSTS-002.5 fault-injection carve-out: when the companion
// alert-level claim errors, the evaluation still submits budget.exceeded
// (already earned by the exceeded-level claim) and the failure is logged
// and counted like any other claim-store error. The test deliberately does
// NOT assert the companion claim was recorded, per the spec's own
// instruction — that is the failure this test injects.
func TestCheckBudget_CompanionAlertClaimFault_StillSubmitsExceeded(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-companion-fault")
faulty := &claimFaultRepo{Repository: repo, failLevels: map[string]error{
"alert": errors.New("injected companion claim failure"),
}}
agents := &repoAgents{repo: repo}
spy := &budgetActivitySpy{}
svc := costs.NewCostService(faulty, logger.Default(), spy, agents, agents)
policy := newIdempotencyTestPolicy("agent-companion-fault", 500)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-companion-fault", "task-1", 600)
results, err := svc.CheckBudget(ctx, "ws-1", "agent-companion-fault", "proj-1")
if err != nil {
t.Fatalf("CheckBudget: %v", err)
}
if !results[0].ExceededSubmitted {
t.Fatal("budget.exceeded must still be submitted despite the companion claim failure")
}
if got := spy.count("budget.exceeded"); got != 1 {
t.Fatalf("budget.exceeded submissions = %d, want 1", got)
}
}
// TestCheckBudget_ClaimStoreFault_EmitsAndReportsSubmittedTrue covers
// AC-OFFICE-COSTS-002.14: a claim-store error degrades to emit-anyway, and
// the result's submitted field is true even though no claim was held — the
// case a "held the claim and submitted" reading gets backwards.
func TestCheckBudget_ClaimStoreFault_EmitsAndReportsSubmittedTrue(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-fault")
injected := errors.New("injected claim store failure")
faulty := &claimFaultRepo{Repository: repo, failLevels: map[string]error{
"alert": injected,
}}
agents := &repoAgents{repo: repo}
spy := &budgetActivitySpy{}
svc := costs.NewCostService(faulty, logger.Default(), spy, agents, agents)
policy := newIdempotencyTestPolicy("agent-fault", 1000)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-fault", "task-1", 850)
before := claimFailureCount(t)
// Two evaluations, both against the always-failing claim store: a
// broken store degrades to duplicate notifications, never to silence.
for i := 0; i < 2; i++ {
results, err := svc.CheckBudget(ctx, "ws-1", "agent-fault", "proj-1")
if err != nil {
t.Fatalf("CheckBudget[%d]: %v", i, err)
}
if !results[0].AlertSubmitted {
t.Fatalf("evaluation %d: AlertSubmitted = false, want true (fail-open emission)", i)
}
}
if got := spy.count("budget.alert"); got != 2 {
t.Fatalf("budget.alert submissions under a faulted store = %d, want 2 (duplicates permitted)", got)
}
if delta := claimFailureCount(t) - before; delta != 2 {
t.Fatalf("budget_claim_failures_total delta = %d, want 2 (AC-OFFICE-COSTS-002.14: one per faulted evaluation)", delta)
}
}
// TestCheckBudget_ClaimStoreFault_ExceededPath_EmitsAndCountsTwicePerEval
// extends TestCheckBudget_ClaimStoreFault_EmitsAndReportsSubmittedTrue to the
// exceeded level: claimAndEmit's afterClaim hook claims the alert-level
// companion right after the exceeded-level claim, so a broken store fails
// both claim attempts once per evaluation (AC-OFFICE-COSTS-002.14).
func TestCheckBudget_ClaimStoreFault_ExceededPath_EmitsAndCountsTwicePerEval(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-fault-exc")
injected := errors.New("injected claim store failure")
faulty := &claimFaultRepo{Repository: repo, failLevels: map[string]error{
"alert": injected,
"exceeded": injected,
}}
agents := &repoAgents{repo: repo}
spy := &budgetActivitySpy{}
svc := costs.NewCostService(faulty, logger.Default(), spy, agents, agents)
policy := newIdempotencyTestPolicy("agent-fault-exc", 500)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-fault-exc", "task-1", 600) // >= limit
before := claimFailureCount(t)
for i := 0; i < 2; i++ {
results, err := svc.CheckBudget(ctx, "ws-1", "agent-fault-exc", "proj-1")
if err != nil {
t.Fatalf("CheckBudget[%d]: %v", i, err)
}
if !results[0].ExceededSubmitted {
t.Fatalf("eval %d: ExceededSubmitted=false, want true (fail-open)", i)
}
}
if got := spy.count("budget.exceeded"); got != 2 {
t.Fatalf("budget.exceeded = %d, want 2 (duplicates permitted under broken store)", got)
}
// exceeded claim + companion alert claim both fail => 2 increments per evaluation.
if delta := claimFailureCount(t) - before; delta != 4 {
t.Fatalf("budget_claim_failures_total delta = %d, want 4 (2 per exceeded eval)", delta)
}
}
// TestCheckBudget_AlreadyClaimedMiss_NoFailureCounterDelta covers the
// AC-OFFICE-COSTS-002.14a half of the failure counter: a Claim call
// returning (claimed=false, err=nil) — whether because an earlier
// evaluation already holds it or because the referenced policy was deleted
// mid-evaluation (a foreign-key violation, classified identically) — is an
// ordinary miss, not a claim-store failure, and must not move
// budget_claim_failures_total.
func TestCheckBudget_AlreadyClaimedMiss_NoFailureCounterDelta(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-miss")
missy := &claimFaultRepo{Repository: repo, missLevels: map[string]bool{"alert": true}}
agents := &repoAgents{repo: repo}
spy := &budgetActivitySpy{}
svc := costs.NewCostService(missy, logger.Default(), spy, agents, agents)
policy := newIdempotencyTestPolicy("agent-miss", 1000)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-miss", "task-1", 850)
before := claimFailureCount(t)
results, err := svc.CheckBudget(ctx, "ws-1", "agent-miss", "proj-1")
if err != nil {
t.Fatalf("CheckBudget: %v", err)
}
if results[0].AlertSubmitted {
t.Fatal("a (claimed=false, err=nil) miss must not submit")
}
if got := spy.count("budget.alert"); got != 0 {
t.Fatalf("budget.alert submissions = %d, want 0", got)
}
if delta := claimFailureCount(t) - before; delta != 0 {
t.Fatalf("budget_claim_failures_total delta = %d, want 0 (AC-OFFICE-COSTS-002.14a: a miss is not a store failure)", delta)
}
}
// TestCheckBudget_ExceededClaimsCompanionBeforeEmit covers AC-OFFICE-COSTS-002.5's
// claim-then-emit ordering: costs-03.md's Claim-then-emit step 2 records both
// the exceeded claim and its alert-level companion before the emission
// decision. Asserting call order (not just eventual outcome) is what proves
// the ordering — outcome alone cannot distinguish this from a companion claim
// recorded after the emit.
func TestCheckBudget_ExceededClaimsCompanionBeforeEmit(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-order")
order := &callOrderRecorder{}
orderedRepo := &orderRecordingRepo{Repository: repo, order: order}
agents := &repoAgents{repo: repo}
svc := costs.NewCostService(orderedRepo, logger.Default(), &orderRecordingActivity{order: order}, agents, agents)
policy := newIdempotencyTestPolicy("agent-order", 500)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-order", "task-1", 600)
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-order", "proj-1"); err != nil {
t.Fatalf("CheckBudget: %v", err)
}
got := order.snapshot()
want := []string{"claim:exceeded", "claim:alert", "emit:budget.exceeded"}
if len(got) != len(want) {
t.Fatalf("call order = %v, want %v", got, want)
}
for i := range want {
if got[i] != want[i] {
t.Fatalf("call order = %v, want %v (both claims must precede the emission)", got, want)
}
}
}
// TestCheckBudget_PauseAgent_ReArmsDespiteExistingClaim covers
// AC-OFFICE-COSTS-002.12: suppressing a notification must not suppress
// enforcement. An evaluation landing on an already-claimed level must still
// pause an agent the user has since unpaused.
func TestCheckBudget_PauseAgent_ReArmsDespiteExistingClaim(t *testing.T) {
svc, repo, execSQL := newBudgetTestService(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-rearm")
policy := newIdempotencyTestPolicy("agent-rearm", 500)
policy.ActionOnExceed = "pause_agent"
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-rearm", "task-1", 600)
first, err := svc.CheckBudget(ctx, "ws-1", "agent-rearm", "proj-1")
if err != nil {
t.Fatalf("first CheckBudget: %v", err)
}
if !first[0].AgentPaused {
t.Fatal("first evaluation should pause the agent")
}
if err := repo.UpdateAgentStatusFields(ctx, "agent-rearm", "idle", ""); err != nil {
t.Fatalf("unpause agent: %v", err)
}
second, err := svc.CheckBudget(ctx, "ws-1", "agent-rearm", "proj-1")
if err != nil {
t.Fatalf("second CheckBudget: %v", err)
}
if !second[0].AgentPaused {
t.Fatal("second evaluation must still pause the agent despite the level already being claimed")
}
agent, err := repo.GetAgentInstance(ctx, "agent-rearm")
if err != nil {
t.Fatalf("get agent: %v", err)
}
if agent.Status != "paused" {
t.Fatalf("agent status = %q, want paused", agent.Status)
}
}
// TestCheckBudget_ConcurrentEvaluation_EmitsOnce covers
// AC-OFFICE-COSTS-002.10 on a healthy store: the claim table's primary key
// resolves the race, so concurrent evaluations of the same policy, period
// and level submit exactly one activity row.
func TestCheckBudget_ConcurrentEvaluation_EmitsOnce(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-concurrent")
agents := &repoAgents{repo: repo}
spy := &budgetActivitySpy{}
svc := costs.NewCostService(repo, logger.Default(), spy, agents, agents)
policy := newIdempotencyTestPolicy("agent-concurrent", 1000)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-concurrent", "task-1", 850)
const n = 8
var wg sync.WaitGroup
wg.Add(n)
for i := 0; i < n; i++ {
go func() {
defer wg.Done()
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-concurrent", "proj-1"); err != nil {
t.Errorf("CheckBudget: %v", err)
}
}()
}
wg.Wait()
if got := spy.count("budget.alert"); got != 1 {
t.Fatalf("budget.alert submissions under concurrent evaluation = %d, want 1", got)
}
}
// TestCheckBudgetAndEvaluateProjectBudget_ShareAlertClaim covers the two
// callers that can evaluate a project policy. They must use the same durable
// claim so a cost event followed by task reassignment emits one alert.
func TestCheckBudgetAndEvaluateProjectBudget_ShareAlertClaim(t *testing.T) {
spy := &budgetActivitySpy{}
svc, _, execSQL := newBudgetTestServiceWithActivity(t, spy)
ctx := context.Background()
policy := newIdempotencyTestPolicy("proj-shared-claim", 1000)
policy.ScopeType = "project"
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestTask(t, execSQL, "task-shared-claim", "ws-1", "proj-shared-claim")
insertBudgetTestCostEvent(t, execSQL, "agent-shared-claim", "task-shared-claim", 850)
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-shared-claim", "proj-shared-claim"); err != nil {
t.Fatalf("CheckBudget: %v", err)
}
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-shared-claim"); err != nil {
t.Fatalf("EvaluateProjectBudget: %v", err)
}
if got := spy.count("budget.alert"); got != 1 {
t.Fatalf("budget.alert submissions across callers = %d, want 1", got)
}
}
// TestCheckBudget_PeriodKeyMatchesSpendBoundary covers AC-OFFICE-COSTS-002.6a:
// the claim recorded for a monthly policy is keyed to the same window
// boundary the spend rollup used, rendered as RFC3339 UTC.
func TestCheckBudget_PeriodKeyMatchesSpendBoundary(t *testing.T) {
repo, queryRow, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-period")
agents := &repoAgents{repo: repo}
svc := costs.NewCostService(repo, logger.Default(), &noopActivity{}, agents, agents)
policy := newIdempotencyTestPolicy("agent-period", 1000)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-period", "task-1", 850)
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-period", "proj-1"); err != nil {
t.Fatalf("CheckBudget: %v", err)
}
var gotKey string
if err := queryRow(
`SELECT period_key FROM office_budget_claims WHERE policy_id = ? AND level = 'alert'`, policy.ID,
).Scan(&gotKey); err != nil {
t.Fatalf("query claim period_key: %v", err)
}
if gotKey != monthlyPeriodKey() {
t.Fatalf("claim period_key = %q, want %q (must match the spend rollup's boundary)", gotKey, monthlyPeriodKey())
}
}
// TestCheckBudget_LifetimePolicyClaimsLifetimeKey covers the "lifetime"
// literal branch of AC-OFFICE-COSTS-002.6: a policy whose spend window
// never resets (period "total") gets a claim that never resets either.
func TestCheckBudget_LifetimePolicyClaimsLifetimeKey(t *testing.T) {
repo, queryRow, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-lifetime")
agents := &repoAgents{repo: repo}
svc := costs.NewCostService(repo, logger.Default(), &noopActivity{}, agents, agents)
policy := newIdempotencyTestPolicy("agent-lifetime", 1000)
policy.Period = "total"
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-lifetime", "task-1", 850)
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-lifetime", "proj-1"); err != nil {
t.Fatalf("CheckBudget: %v", err)
}
var gotKey string
if err := queryRow(
`SELECT period_key FROM office_budget_claims WHERE policy_id = ? AND level = 'alert'`, policy.ID,
).Scan(&gotKey); err != nil {
t.Fatalf("query claim period_key: %v", err)
}
if gotKey != "lifetime" {
t.Fatalf("claim period_key = %q, want %q", gotKey, "lifetime")
}
}
// TestCheckBudget_SpendBelowLevelKeepsExistingClaim covers
// AC-OFFICE-COSTS-002.15: nothing in the evaluation flow ever releases a
// claim, even when a later evaluation finds spend below the level it
// claimed.
func TestCheckBudget_SpendBelowLevelKeepsExistingClaim(t *testing.T) {
repo, _, _ := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-keep-claim")
agents := &repoAgents{repo: repo}
svc := costs.NewCostService(repo, logger.Default(), &noopActivity{}, agents, agents)
policy := newIdempotencyTestPolicy("agent-keep-claim", 1000)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
periodKey := monthlyPeriodKey()
if claimed, err := repo.Claim(ctx, policy.ID, periodKey, "alert"); err != nil || !claimed {
t.Fatalf("seed claim: claimed=%v err=%v", claimed, err)
}
// No cost events: spend is 0, well below the alert level, so this
// evaluation takes the "no claim, no emission" branch entirely.
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-keep-claim", "proj-1"); err != nil {
t.Fatalf("CheckBudget: %v", err)
}
claimed, err := repo.Claim(ctx, policy.ID, periodKey, "alert")
if err != nil {
t.Fatalf("re-claim: %v", err)
}
if claimed {
t.Fatal("existing claim must survive an evaluation where spend does not reach the level")
}
}
// TestCheckPreExecutionBudget_StillDeniesAfterAlreadyClaimed covers
// AC-OFFICE-COSTS-002.12: suppressing a notification must not suppress
// enforcement. block_new_tasks keeps denying on every subsequent check
// regardless of whether a claim already exists.
func TestCheckPreExecutionBudget_StillDeniesAfterAlreadyClaimed(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-enforce")
agents := &repoAgents{repo: repo}
svc := costs.NewCostService(repo, logger.Default(), &noopActivity{}, agents, agents)
policy := newIdempotencyTestPolicy("agent-enforce", 500)
policy.ActionOnExceed = "block_new_tasks"
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-enforce", "task-1", 600)
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-enforce", "proj-1"); err != nil {
t.Fatalf("CheckBudget: %v", err)
}
allowed, reason, err := svc.CheckPreExecutionBudget(ctx, "agent-enforce", "", "ws-1")
if err != nil {
t.Fatalf("CheckPreExecutionBudget: %v", err)
}
if allowed {
t.Errorf("block_new_tasks must still deny once the notification is already claimed; reason=%q", reason)
}
}
+122 -16
View File
@@ -26,16 +26,36 @@ const (
budgetActionBlockNewTasks = "block_new_tasks"
)
// Claim levels. See docs/specs/office/requirements/costs.md#terminology.
const (
claimLevelAlert = "alert"
claimLevelExceeded = "exceeded"
)
// claimPeriodLifetime is the period_key for a policy whose spend window
// never resets (periodCutoff's zero-time "no filter" answer). It is a
// literal rather than the RFC3339 rendering of the zero time so a future
// change to periodCutoff's zero value can never collide with it.
const claimPeriodLifetime = "lifetime"
// BudgetCheckResult describes the outcome of a budget check.
// SpentSubcents / LimitSubcents are hundredths of a cent.
// SpentSubcents / LimitSubcents are hundredths of a cent. AlertFired /
// LimitExceed report whether spend reaches that level *now* (used for
// gating); AlertSubmitted / ExceededSubmitted report whether *this*
// evaluation handed that level's row to the activity logger (used for
// assertions and for callers that want to react only to new crossings).
// The two pairs are independent: holding a claim does not imply a
// submission, and a submission does not imply a claim was held.
type BudgetCheckResult struct {
PolicyID string
ActionOnExceed string
SpentSubcents int64
LimitSubcents int64
AlertFired bool
LimitExceed bool
AgentPaused bool
PolicyID string
ActionOnExceed string
SpentSubcents int64
LimitSubcents int64
AlertFired bool
LimitExceed bool
AlertSubmitted bool
ExceededSubmitted bool
AgentPaused bool
}
// CheckBudget evaluates all budget policies for the given agent and project.
@@ -95,7 +115,8 @@ func (s *CostService) evaluatePolicy(
workspaceID string,
policy *BudgetPolicy,
) (BudgetCheckResult, error) {
spent, err := s.getSpendForPolicy(ctx, workspaceID, policy)
boundary := periodCutoff(string(policy.Period), time.Now())
spent, err := s.getSpendForPolicy(ctx, workspaceID, policy, boundary)
if err != nil {
return BudgetCheckResult{}, err
}
@@ -108,23 +129,96 @@ func (s *CostService) evaluatePolicy(
LimitSubcents: limit,
}
periodKey := periodKeyFor(boundary)
threshold := limit * int64(policy.AlertThresholdPct) / 100
if spent >= threshold && spent < limit {
result.AlertFired = true
s.logBudgetAlert(ctx, workspaceID, policy, spent)
}
if spent >= limit {
switch {
case spent >= limit:
result.LimitExceed = true
if policy.ActionOnExceed == budgetActionPauseAgent && policy.ScopeType == scopeAgent {
result.AgentPaused = s.pauseAgentForBudget(ctx, policy.ScopeID)
}
s.logBudgetExceeded(ctx, workspaceID, policy, spent)
result.ExceededSubmitted = s.claimAndEmit(ctx, workspaceID, policy, spent, periodKey, claimLevelExceeded,
// A spend that jumps straight from below the alert level to
// above the limit must also hold the alert-level claim before
// budget.exceeded is emitted, so a concurrent evaluation
// landing in the alert band cannot win that claim in the gap
// and emit a budget.alert describing a reduction in spend. The
// outcome is discarded (claimed or already-claimed are equally
// fine here); a claim-store error is still logged and counted
// like any other.
func() { s.claimCompanionAlert(ctx, policy, periodKey) })
case spent >= threshold:
result.AlertFired = true
result.AlertSubmitted = s.claimAndEmit(ctx, workspaceID, policy, spent, periodKey, claimLevelAlert, nil)
}
return result, nil
}
// claimAndEmit resolves the three outcomes of claiming (policy, periodKey,
// level) and emits that level's activity row when the claim allows it.
// Reports whether this evaluation submitted the row. afterClaim, when
// non-nil, runs immediately after the claim and before the emission
// decision, regardless of the claim's outcome — the seam the exceeded
// level's alert-level companion claim uses so both claims land before
// either emission is decided.
func (s *CostService) claimAndEmit(
ctx context.Context,
workspaceID string,
policy *BudgetPolicy,
spent int64,
periodKey, level string,
afterClaim func(),
) bool {
claimed, err := s.repo.Claim(ctx, policy.ID, periodKey, level)
if afterClaim != nil {
afterClaim()
}
if err != nil {
s.recordClaimFailure(policy.ID, level, err)
s.emitLevel(ctx, workspaceID, policy, spent, level)
return true
}
if !claimed {
return false
}
s.emitLevel(ctx, workspaceID, policy, spent, level)
return true
}
// claimCompanionAlert records the alert-level claim alongside an
// exceeded-level emission, so a spend that jumps straight past the limit
// still de-escalates correctly. Its outcome never changes the
// budget.exceeded emission; only a claim-store error is reported, exactly
// like any other claim attempt.
func (s *CostService) claimCompanionAlert(ctx context.Context, policy *BudgetPolicy, periodKey string) {
if _, err := s.repo.Claim(ctx, policy.ID, periodKey, claimLevelAlert); err != nil {
s.recordClaimFailure(policy.ID, claimLevelAlert, err)
}
}
func (s *CostService) emitLevel(
ctx context.Context, workspaceID string, policy *BudgetPolicy, spent int64, level string,
) {
if level == claimLevelExceeded {
s.logBudgetExceeded(ctx, workspaceID, policy, spent)
return
}
s.logBudgetAlert(ctx, workspaceID, policy, spent)
}
// recordClaimFailure logs and counts a claim-store failure encountered
// while evaluating a policy. Never called for a foreign-key violation (the
// referenced policy no longer exists, not a store failure) or for a failed
// claim discard during a policy update, both of which are reported through
// different channels.
func (s *CostService) recordClaimFailure(policyID, level string, err error) {
budgetClaimFailuresTotal.Add(budgetClaimFailureLabel, 1)
s.logger.Error("budget claim store failure",
zap.String("policy_id", policyID), zap.String("level", level), zap.Error(err))
}
// periodCutoff returns the time.Time at which the policy's spend window
// starts. A zero time means "no filter" (lifetime / total) or an unknown
// period retained for compatibility with stored policies.
@@ -136,12 +230,24 @@ func periodCutoff(period string, now time.Time) time.Time {
return start
}
// periodKeyFor renders a period boundary as the claim's stored identity.
// The layout is contract, not local style: a non-zero boundary is RFC3339 in
// UTC, and periodCutoff's zero-time "lifetime" answer renders as the
// literal "lifetime" rather than the RFC3339 zero time, so it can never be
// mistaken for a real instant.
func periodKeyFor(boundary time.Time) string {
if boundary.IsZero() {
return claimPeriodLifetime
}
return boundary.UTC().Format(time.RFC3339)
}
func (s *CostService) getSpendForPolicy(
ctx context.Context,
workspaceID string,
policy *BudgetPolicy,
since time.Time,
) (int64, error) {
since := periodCutoff(string(policy.Period), time.Now())
switch policy.ScopeType {
case scopeAgent:
return s.repo.GetCostForAgentSince(ctx, policy.ScopeID, since)
@@ -2,6 +2,8 @@ package costs_test
import (
"context"
"database/sql"
"sync"
"testing"
"time"
@@ -17,6 +19,123 @@ import (
"github.com/kandev/kandev/internal/office/shared"
)
// budgetActivitySpy implements shared.ActivityLogger and records every
// call so tests can assert on *submission* (AC-OFFICE-COSTS-002.2a), never
// with a SELECT against office_activity_log: LogActivity returns nothing,
// so the durability of any one row is not something the contract promises.
// The mutex makes it safe for TestCheckBudget_ConcurrentEvaluation_EmitsOnce,
// which drives LogActivity from multiple goroutines.
type budgetActivitySpy struct {
mu sync.Mutex
calls []budgetActivityCall
}
// budgetActivityCall records one LogActivity invocation observed by
// budgetActivitySpy.
type budgetActivityCall struct {
action, targetType, targetID, details string
}
func (s *budgetActivitySpy) LogActivity(_ context.Context, _, _, _, action, targetType, targetID, details string) {
s.mu.Lock()
defer s.mu.Unlock()
s.calls = append(s.calls, budgetActivityCall{action, targetType, targetID, details})
}
func (s *budgetActivitySpy) LogActivityWithRun(
ctx context.Context, wsID, actorType, actorID, action, targetType, targetID, details, _, _ string,
) {
s.LogActivity(ctx, wsID, actorType, actorID, action, targetType, targetID, details)
}
func (s *budgetActivitySpy) count(action string) int {
s.mu.Lock()
defer s.mu.Unlock()
n := 0
for _, c := range s.calls {
if c.action == action {
n++
}
}
return n
}
// claimFaultRepo wraps a real *sqlite.Repository and overrides Claim to
// fail (failLevels) or miss without error (missLevels) for the named
// levels, so tests can drive AC-OFFICE-COSTS-002.5's, .14's and .14a's
// claim-store outcomes deterministically without a fault-injecting SQL
// driver. missLevels stands in for either an already-held claim or a
// foreign-key-violation-as-miss (AC-OFFICE-COSTS-002.14a): both return
// (false, nil), and the counter under test cannot and should not
// distinguish them.
type claimFaultRepo struct {
*sqlite.Repository
failLevels map[string]error
missLevels map[string]bool
}
func (r *claimFaultRepo) Claim(ctx context.Context, policyID, periodKey, level string) (bool, error) {
if err, ok := r.failLevels[level]; ok {
return false, err
}
if r.missLevels[level] {
return false, nil
}
return r.Repository.Claim(ctx, policyID, periodKey, level)
}
// callOrderRecorder records events in the order they occur, for tests that
// must prove sequencing rather than just eventual outcome (AC-OFFICE-COSTS-002.5's
// claim-then-emit ordering).
type callOrderRecorder struct {
mu sync.Mutex
events []string
}
func (r *callOrderRecorder) record(event string) {
r.mu.Lock()
defer r.mu.Unlock()
r.events = append(r.events, event)
}
func (r *callOrderRecorder) snapshot() []string {
r.mu.Lock()
defer r.mu.Unlock()
out := make([]string, len(r.events))
copy(out, r.events)
return out
}
// orderRecordingRepo wraps a real *sqlite.Repository and records each
// Claim call's level into a shared callOrderRecorder.
type orderRecordingRepo struct {
*sqlite.Repository
order *callOrderRecorder
}
func (r *orderRecordingRepo) Claim(ctx context.Context, policyID, periodKey, level string) (bool, error) {
claimed, err := r.Repository.Claim(ctx, policyID, periodKey, level)
r.order.record("claim:" + level)
return claimed, err
}
// orderRecordingActivity implements shared.ActivityLogger and records each
// submitted action into the same callOrderRecorder as orderRecordingRepo,
// so a test can assert the interleaving of claims and emissions.
type orderRecordingActivity struct {
order *callOrderRecorder
}
func (a *orderRecordingActivity) LogActivity(_ context.Context, _, _, _, action, _, _, _ string) {
a.order.record("emit:" + action)
}
func (a *orderRecordingActivity) LogActivityWithRun(
ctx context.Context, wsID, actorType, actorID, action, targetType, targetID, details, _, _ string,
) {
a.LogActivity(ctx, wsID, actorType, actorID, action, targetType, targetID, details)
}
// repoAgents adapts the office repository to shared.AgentReader +
// shared.AgentWriter so budget tests can exercise the real pause-agent
// path end-to-end against an in-memory DB.
@@ -40,17 +159,13 @@ func (a *repoAgents) UpdateAgentStatusFields(ctx context.Context, agentID, statu
return a.repo.UpdateAgentStatusFields(ctx, agentID, status, pauseReason)
}
func newBudgetTestService(t *testing.T) (*costs.CostService, *sqlite.Repository, func(string, ...interface{})) {
t.Helper()
return newBudgetTestServiceWithActivity(t, &noopActivity{})
}
// newBudgetTestServiceWithActivity is the same setup as newBudgetTestService
// but lets project-scope tests supply a spy activity logger to observe which
// budget.alert / budget.exceeded rows fire.
func newBudgetTestServiceWithActivity(
t *testing.T, activity shared.ActivityLogger,
) (*costs.CostService, *sqlite.Repository, func(string, ...interface{})) {
// newBudgetTestRepo builds the in-memory office repo shared by every budget
// test, without wiring a CostService, so fault-injection tests can wrap the
// repo (claimFaultRepo) or swap the activity logger (budgetActivitySpy)
// before constructing the service. queryRow exposes read-only access to the
// underlying connection for tests that assert directly on
// office_budget_claims rather than through the CostService API.
func newBudgetTestRepo(t *testing.T) (*sqlite.Repository, func(string, ...interface{}) *sql.Row, func(string, ...interface{})) {
t.Helper()
db, err := sqlx.Open("sqlite3", ":memory:")
if err != nil {
@@ -80,16 +195,35 @@ func newBudgetTestServiceWithActivity(
if err != nil {
t.Fatalf("new repo: %v", err)
}
log := logger.Default()
agents := &repoAgents{repo: repo}
svc := costs.NewCostService(repo, log, activity, agents, agents)
execSQL := func(query string, args ...interface{}) {
t.Helper()
if _, err := db.Exec(query, args...); err != nil {
t.Fatalf("exec sql: %v", err)
}
}
queryRow := func(query string, args ...interface{}) *sql.Row {
t.Helper()
return db.QueryRow(query, args...)
}
return repo, queryRow, execSQL
}
func newBudgetTestService(t *testing.T) (*costs.CostService, *sqlite.Repository, func(string, ...interface{})) {
t.Helper()
return newBudgetTestServiceWithActivity(t, &noopActivity{})
}
// newBudgetTestServiceWithActivity is the same setup as newBudgetTestService
// but lets project-scope tests supply a spy activity logger to observe which
// budget.alert / budget.exceeded rows fire.
func newBudgetTestServiceWithActivity(
t *testing.T, activity shared.ActivityLogger,
) (*costs.CostService, *sqlite.Repository, func(string, ...interface{})) {
t.Helper()
repo, _, execSQL := newBudgetTestRepo(t)
log := logger.Default()
agents := &repoAgents{repo: repo}
svc := costs.NewCostService(repo, log, activity, agents, agents)
return svc, repo, execSQL
}
@@ -105,29 +239,6 @@ func insertBudgetTestTask(t *testing.T, execSQL func(string, ...interface{}), ta
)
}
// budgetActivityCall records one LogActivity invocation observed by
// budgetActivitySpy.
type budgetActivityCall struct {
action string
targetType string
targetID string
}
// budgetActivitySpy implements shared.ActivityLogger and records every call,
// so project-budget tests can assert exactly which alerts fired without
// depending on evaluatePolicy's return value (EvaluateProjectBudget only
// returns an error).
type budgetActivitySpy struct {
calls []budgetActivityCall
}
func (s *budgetActivitySpy) LogActivity(_ context.Context, _, _, _, action, targetType, targetID, _ string) {
s.calls = append(s.calls, budgetActivityCall{action: action, targetType: targetType, targetID: targetID})
}
func (s *budgetActivitySpy) LogActivityWithRun(_ context.Context, _, _, _, _, _, _, _, _, _ string) {
}
func createBudgetTestAgent(t *testing.T, repo *sqlite.Repository, wsID, agentID string) {
t.Helper()
agent := &models.AgentInstance{
@@ -468,13 +579,18 @@ func TestEvaluateProjectBudget_AlertAtThreshold(t *testing.T) {
insertBudgetTestTask(t, execSQL, "task-1", "ws-1", "proj-1")
insertBudgetTestCostEvent(t, execSQL, "agent-1", "task-1", int64(850))
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-1"); err != nil {
t.Fatalf("EvaluateProjectBudget: %v", err)
for i := 0; i < 2; i++ {
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-1"); err != nil {
t.Fatalf("EvaluateProjectBudget[%d]: %v", i, err)
}
}
if !hasBudgetActivity(spy.calls, "budget.alert", "proj-1") {
t.Errorf("expected budget.alert for proj-1, got calls=%+v", spy.calls)
}
if got := spy.count("budget.alert"); got != 1 {
t.Fatalf("budget.alert submissions = %d, want 1", got)
}
}
// TestEvaluateProjectBudget_ExceededAtLimit covers the limit-exceeded branch.
@@ -498,13 +614,18 @@ func TestEvaluateProjectBudget_ExceededAtLimit(t *testing.T) {
insertBudgetTestTask(t, execSQL, "task-1", "ws-1", "proj-1")
insertBudgetTestCostEvent(t, execSQL, "agent-1", "task-1", int64(600))
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-1"); err != nil {
t.Fatalf("EvaluateProjectBudget: %v", err)
for i := 0; i < 2; i++ {
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-1"); err != nil {
t.Fatalf("EvaluateProjectBudget[%d]: %v", i, err)
}
}
if !hasBudgetActivity(spy.calls, "budget.exceeded", "proj-1") {
t.Errorf("expected budget.exceeded for proj-1, got calls=%+v", spy.calls)
}
if got := spy.count("budget.exceeded"); got != 1 {
t.Fatalf("budget.exceeded submissions = %d, want 1", got)
}
}
// TestEvaluateProjectBudget_PauseAgentPolicyDoesNotPause locks in the design
@@ -43,6 +43,9 @@ type Repository interface {
) (models.SpendWindow, error)
GetWorkspaceBudgetDefault(ctx context.Context, workspaceID string) (limitSubcents int64, found bool, err error)
SetWorkspaceBudgetDefault(ctx context.Context, workspaceID string, limitSubcents int64) error
// Claim atomically records that (policyID, periodKey, level) may emit its
// budget notification. See docs/specs/office/requirements/costs.md.
Claim(ctx context.Context, policyID, periodKey, level string) (bool, error)
}
// CostService handles cost recording, summaries, and budget evaluation.
@@ -64,6 +64,12 @@ type Repository struct {
ro *sqlx.DB // reader
log *logger.Logger
migrate *db.MigrateLogger
// failBudgetPolicyUpdateErr is a test-only failpoint: when set,
// updateBudgetPolicyTx returns it after the claim discard has run but
// before the policy row update, so a test can prove the two are
// transactional without a fault-injecting driver.
failBudgetPolicyUpdateErr error
}
// NewWithDB creates a new office repository with existing database connections.
@@ -333,6 +339,23 @@ func (r *Repository) createCostTables() error {
limit_subcents INTEGER NOT NULL,
updated_at TIMESTAMP NOT NULL
);
-- office_budget_claims must be created after office_budget_policies:
-- its foreign key references that table, and PostgreSQL rejects a
-- forward reference at CREATE TABLE time even though SQLite tolerates
-- it. One row per (policy, evaluation period, level) that has already
-- produced its budget.alert / budget.exceeded notification; the
-- primary key is the concurrency control for a claim attempt.
-- ON DELETE CASCADE removes a policy's claims on every deletion path
-- without a matching code change at any of them.
CREATE TABLE IF NOT EXISTS office_budget_claims (
policy_id TEXT NOT NULL,
period_key TEXT NOT NULL,
level TEXT NOT NULL,
claimed_at TIMESTAMP NOT NULL,
PRIMARY KEY (policy_id, period_key, level),
FOREIGN KEY (policy_id) REFERENCES office_budget_policies(id) ON DELETE CASCADE
);
`)
return err
}
@@ -53,6 +53,7 @@ func TestInitSchema_AllTablesExist(t *testing.T) {
"office_agent_runtime",
"office_cost_events",
"office_budget_policies",
"office_budget_claims",
"runs",
"office_routines",
"office_routine_triggers",
@@ -0,0 +1,83 @@
package sqlite
import (
"context"
"errors"
"testing"
"github.com/jmoiron/sqlx"
_ "github.com/mattn/go-sqlite3"
settingsstore "github.com/kandev/kandev/internal/agent/settings/store"
"github.com/kandev/kandev/internal/office/models"
)
// TestUpdateBudgetPolicy_FailedUpdateRollsBackDiscard proves the claim
// discard and the policy row update are one transaction
// (AC-OFFICE-COSTS-002.8): when the row update fails after the discard has
// already run, the whole transaction rolls back, so the claim survives
// alongside the un-updated policy. failBudgetPolicyUpdateErr is a
// test-only failpoint (see base.go) standing in for a fault-injecting
// driver.
func TestUpdateBudgetPolicy_FailedUpdateRollsBackDiscard(t *testing.T) {
db, err := sqlx.Open("sqlite3", ":memory:?_foreign_keys=on")
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
// Pin all work to one connection: each SQLite :memory: connection is a
// separate empty database, so a pooled second connection would miss the
// tables/rows the first one created.
db.SetMaxOpenConns(1)
t.Cleanup(func() { _ = db.Close() })
if _, err := db.Exec(`PRAGMA foreign_keys = ON`); err != nil {
t.Fatalf("pragma fk: %v", err)
}
if _, _, err := settingsstore.Provide(db, db, nil); err != nil {
t.Fatalf("settings store init: %v", err)
}
repo, err := NewWithDB(db, db, nil)
if err != nil {
t.Fatalf("new repo: %v", err)
}
ctx := context.Background()
policy := &models.BudgetPolicy{
WorkspaceID: "ws-rollback",
ScopeType: "workspace",
ScopeID: "ws-rollback",
LimitSubcents: 1000,
Period: "monthly",
AlertThresholdPct: 80,
ActionOnExceed: "notify_only",
}
if err := repo.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
if _, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert"); err != nil {
t.Fatalf("claim: %v", err)
}
repo.failBudgetPolicyUpdateErr = errors.New("injected update failure")
policy.LimitSubcents = 5000
if err := repo.UpdateBudgetPolicy(ctx, policy); err == nil {
t.Fatal("expected the injected update failure to surface")
}
var claimCount int
if err := db.QueryRow(
`SELECT COUNT(*) FROM office_budget_claims WHERE policy_id = ?`, policy.ID,
).Scan(&claimCount); err != nil {
t.Fatalf("count claims: %v", err)
}
if claimCount != 1 {
t.Fatalf("claim count after rolled-back update = %d, want 1 (discard must roll back with the failed update)", claimCount)
}
stored, err := repo.GetBudgetPolicy(ctx, policy.ID)
if err != nil {
t.Fatalf("get policy: %v", err)
}
if stored.LimitSubcents != 1000 {
t.Fatalf("limit_subcents after rolled-back update = %d, want unchanged 1000", stored.LimitSubcents)
}
}
@@ -0,0 +1,97 @@
package sqlite_test
import (
"context"
"testing"
"github.com/google/uuid"
"github.com/kandev/kandev/internal/office/models"
"github.com/kandev/kandev/internal/office/repository/sqlite"
taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite"
"github.com/kandev/kandev/internal/testutil"
)
// TestPostgresBudgetClaims is the PostgreSQL twin of the office_budget_claims
// coverage in budgets_test.go (REQ-OFFICE-COSTS-002, ADR 0027): office_budget_claims
// carries a forward foreign key to office_budget_policies, which SQLite tolerates at
// CREATE TABLE time but PostgreSQL rejects, so the two tables' creation order is a
// dialect-sensitive contract; Claim's INSERT ... ON CONFLICT DO NOTHING and the
// db.IsForeignKeyViolation classifier it relies on are dialect-sensitive too. This
// proves fresh-boot, boot replay, claim win/lose, the foreign-key-violation miss, and
// cascade delete all hold on a real PostgreSQL backend. Skips unless
// KANDEV_TEST_POSTGRES_DSN is set.
func TestPostgresBudgetClaims(t *testing.T) {
db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t))
ctx := context.Background()
// office_budget_policies (and thus office_budget_claims' FK target) has no
// dependency of its own on the task repository, but office repo init as a
// whole does, mirroring production boot order (see cost_event_contract_postgres_test.go).
if _, err := taskrepo.NewWithDB(db, db, nil); err != nil {
t.Fatalf("init task repo: %v", err)
}
// First boot: office_budget_claims must be created after office_budget_policies
// so its forward FK reference resolves on Postgres.
repo, err := sqlite.NewWithDB(db, db, nil)
if err != nil {
t.Fatalf("first boot: %v", err)
}
// Second boot: the schema must replay without error.
if _, err := sqlite.NewWithDB(db, db, nil); err != nil {
t.Fatalf("second boot (replay): %v", err)
}
policy := &models.BudgetPolicy{
WorkspaceID: "pg-budget-ws",
ScopeType: "workspace",
ScopeID: "pg-budget-ws",
LimitSubcents: 1000,
Period: "monthly",
AlertThresholdPct: 80,
ActionOnExceed: "notify_only",
}
if err := repo.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create budget policy: %v", err)
}
claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert")
if err != nil {
t.Fatalf("first claim: %v", err)
}
if !claimed {
t.Fatal("first claim should win")
}
claimed, err = repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert")
if err != nil {
t.Fatalf("second claim: %v", err)
}
if claimed {
t.Fatal("second claim on the same (policy, period, level) must not win")
}
// A foreign-key violation (no such policy) must classify as an ordinary
// miss, not a store error — db.IsForeignKeyViolation's PostgreSQL branch.
claimed, err = repo.Claim(ctx, uuid.NewString(), "lifetime", "alert")
if err != nil {
t.Fatalf("claim against a nonexistent policy must not be a store error, got: %v", err)
}
if claimed {
t.Fatal("claim against a nonexistent policy must not win")
}
if err := repo.DeleteBudgetPolicy(ctx, policy.ID); err != nil {
t.Fatalf("delete budget policy: %v", err)
}
var claimCount int
if err := db.GetContext(ctx, &claimCount,
`SELECT COUNT(*) FROM office_budget_claims WHERE policy_id = $1`, policy.ID,
); err != nil {
t.Fatalf("count claims after delete: %v", err)
}
if claimCount != 0 {
t.Fatalf("claim count after policy delete = %d, want 0 (FK ON DELETE CASCADE)", claimCount)
}
}
@@ -7,7 +7,9 @@ import (
"time"
"github.com/google/uuid"
"github.com/jmoiron/sqlx"
"github.com/kandev/kandev/internal/db"
"github.com/kandev/kandev/internal/office/models"
)
@@ -42,11 +44,13 @@ func (r *Repository) GetBudgetPolicy(ctx context.Context, id string) (*models.Bu
return &policy, err
}
// ListBudgetPolicies returns all budget policies for a workspace.
// ListBudgetPolicies returns all budget policies for a workspace, ordered by
// created_at with id as a tiebreak so evaluation order is total and
// reproducible.
func (r *Repository) ListBudgetPolicies(ctx context.Context, workspaceID string) ([]*models.BudgetPolicy, error) {
var policies []*models.BudgetPolicy
err := r.ro.SelectContext(ctx, &policies, r.ro.Rebind(
`SELECT * FROM office_budget_policies WHERE workspace_id = ? ORDER BY created_at`), workspaceID)
`SELECT * FROM office_budget_policies WHERE workspace_id = ? ORDER BY created_at ASC, id ASC`), workspaceID)
if err != nil {
return nil, err
}
@@ -56,22 +60,74 @@ func (r *Repository) ListBudgetPolicies(ctx context.Context, workspaceID string)
return policies, nil
}
// UpdateBudgetPolicy updates an existing budget policy.
// UpdateBudgetPolicy updates an existing budget policy and discards that
// policy's claims in the same transaction, so either both apply or neither
// does.
func (r *Repository) UpdateBudgetPolicy(ctx context.Context, policy *models.BudgetPolicy) error {
policy.UpdatedAt = time.Now().UTC()
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
tx, err := r.db.BeginTxx(ctx, nil)
if err != nil {
return err
}
if err := r.updateBudgetPolicyTx(ctx, tx, policy); err != nil {
if rollbackErr := tx.Rollback(); rollbackErr != nil {
return fmt.Errorf("%w; rollback: %v", err, rollbackErr)
}
return err
}
return tx.Commit()
}
func (r *Repository) updateBudgetPolicyTx(ctx context.Context, tx *sqlx.Tx, policy *models.BudgetPolicy) error {
if _, err := tx.ExecContext(ctx, tx.Rebind(
`DELETE FROM office_budget_claims WHERE policy_id = ?`), policy.ID); err != nil {
return fmt.Errorf("discard claims: %w", err)
}
if r.failBudgetPolicyUpdateErr != nil {
return r.failBudgetPolicyUpdateErr
}
_, err := tx.ExecContext(ctx, tx.Rebind(`
UPDATE office_budget_policies SET
scope_type = ?, scope_id = ?, limit_subcents = ?, period = ?,
alert_threshold_pct = ?, action_on_exceed = ?, updated_at = ?
WHERE id = ?
`), policy.ScopeType, policy.ScopeID, policy.LimitSubcents, policy.Period,
policy.AlertThresholdPct, policy.ActionOnExceed, policy.UpdatedAt, policy.ID)
return err
if err != nil {
return fmt.Errorf("update policy: %w", err)
}
return nil
}
// DeleteBudgetPolicy deletes a budget policy by ID.
// DeleteBudgetPolicy deletes a budget policy by ID. Its claims are removed by
// the office_budget_claims foreign key's ON DELETE CASCADE, not here.
func (r *Repository) DeleteBudgetPolicy(ctx context.Context, id string) error {
_, err := r.db.ExecContext(ctx, r.db.Rebind(
`DELETE FROM office_budget_policies WHERE id = ?`), id)
return err
}
// Claim atomically records that (policyID, periodKey, level) may emit its
// budget notification. claimed=true means this call won and the caller
// should emit; claimed=false with a nil error means an earlier evaluation
// already holds the claim, or the referenced policy no longer exists — a
// foreign-key violation is deliberately reported as an ordinary miss, not a
// store failure.
func (r *Repository) Claim(ctx context.Context, policyID, periodKey, level string) (bool, error) {
res, err := r.db.ExecContext(ctx, r.db.Rebind(`
INSERT INTO office_budget_claims (policy_id, period_key, level, claimed_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(policy_id, period_key, level) DO NOTHING
`), policyID, periodKey, level, time.Now().UTC())
if err != nil {
if db.IsForeignKeyViolation(err) {
return false, nil
}
return false, err
}
affected, err := res.RowsAffected()
if err != nil {
return false, err
}
return affected == 1, nil
}
@@ -3,53 +3,247 @@ package sqlite_test
import (
"context"
"testing"
"time"
"github.com/google/uuid"
"github.com/jmoiron/sqlx"
_ "github.com/mattn/go-sqlite3"
settingsstore "github.com/kandev/kandev/internal/agent/settings/store"
"github.com/kandev/kandev/internal/office/models"
"github.com/kandev/kandev/internal/office/repository/sqlite"
)
func TestBudgetPolicy_CRUD(t *testing.T) {
repo := newTestRepo(t)
ctx := context.Background()
// newBudgetClaimsRepoWithFK builds an in-memory repo with
// PRAGMA foreign_keys=ON enabled, so the office_budget_claims cascade is
// actually enforced (production always runs with _foreign_keys=on; the
// default newTestRepo does not). Mirrors newRouteAttemptsRepoWithFK.
func newBudgetClaimsRepoWithFK(t *testing.T) (*sqlite.Repository, *sqlx.DB) {
t.Helper()
db, err := sqlx.Open("sqlite3", ":memory:?_foreign_keys=on")
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
// Pin all work to one connection: each SQLite :memory: connection is a
// separate empty database, so a pooled second connection would miss the
// tables/rows the first one created.
db.SetMaxOpenConns(1)
t.Cleanup(func() { _ = db.Close() })
if _, err := db.Exec(`PRAGMA foreign_keys = ON`); err != nil {
t.Fatalf("pragma fk: %v", err)
}
if _, _, err := settingsstore.Provide(db, db, nil); err != nil {
t.Fatalf("settings store init: %v", err)
}
// office_task_tree_holds and friends carry an FK to the task-package
// owned "tasks" table; with foreign_keys=ON that table must exist (even
// empty) for a workspace-wide DELETE to validate the constraint.
if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS tasks (
id TEXT PRIMARY KEY,
workspace_id TEXT NOT NULL DEFAULT ''
)`); err != nil {
t.Fatalf("create tasks: %v", err)
}
repo, err := sqlite.NewWithDB(db, db, nil)
if err != nil {
t.Fatalf("new repo: %v", err)
}
return repo, db
}
func createTestBudgetPolicy(t *testing.T, repo *sqlite.Repository, workspaceID string) *models.BudgetPolicy {
t.Helper()
policy := &models.BudgetPolicy{
WorkspaceID: "ws-1",
ScopeType: "agent",
ScopeID: "agent-1",
LimitSubcents: 50000,
WorkspaceID: workspaceID,
ScopeType: "workspace",
ScopeID: workspaceID,
LimitSubcents: 1000,
Period: "monthly",
AlertThresholdPct: 80,
ActionOnExceed: "notify_only",
}
if err := repo.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create: %v", err)
if err := repo.CreateBudgetPolicy(context.Background(), policy); err != nil {
t.Fatalf("create budget policy: %v", err)
}
return policy
}
got, err := repo.GetBudgetPolicy(ctx, policy.ID)
func countBudgetClaims(t *testing.T, db *sqlx.DB, policyID string) int {
t.Helper()
var count int
if err := db.QueryRow(
`SELECT COUNT(*) FROM office_budget_claims WHERE policy_id = ?`, policyID,
).Scan(&count); err != nil {
t.Fatalf("count claims: %v", err)
}
return count
}
func TestClaim_FirstCallWinsSecondCallDoesNotReclaim(t *testing.T) {
repo, _ := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-claim")
claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert")
if err != nil {
t.Fatalf("get: %v", err)
t.Fatalf("first claim: %v", err)
}
if got.LimitSubcents != 50000 {
t.Errorf("limit_subcents = %d, want 50000", got.LimitSubcents)
if !claimed {
t.Fatal("first claim should win")
}
policies, err := repo.ListBudgetPolicies(ctx, "ws-1")
claimed, err = repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert")
if err != nil {
t.Fatalf("list: %v", err)
t.Fatalf("second claim: %v", err)
}
if len(policies) != 1 {
t.Fatalf("count = %d, want 1", len(policies))
if claimed {
t.Fatal("second claim on the same (policy, period, level) must not win")
}
}
policy.LimitSubcents = 100000
if err := repo.UpdateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("update: %v", err)
func TestClaim_DifferentPeriodOrLevelIsIndependent(t *testing.T) {
repo, _ := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-claim-2")
if claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert"); err != nil || !claimed {
t.Fatalf("claim september alert: claimed=%v err=%v", claimed, err)
}
got, _ = repo.GetBudgetPolicy(ctx, policy.ID)
if got.LimitSubcents != 100000 {
t.Errorf("updated limit_subcents = %d, want 100000", got.LimitSubcents)
// A different period (new window) is unclaimed (AC-OFFICE-COSTS-002.7).
if claimed, err := repo.Claim(ctx, policy.ID, "2026-10-01T00:00:00Z", "alert"); err != nil || !claimed {
t.Fatalf("claim october alert: claimed=%v err=%v", claimed, err)
}
// A different level in the same period is unclaimed.
if claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "exceeded"); err != nil || !claimed {
t.Fatalf("claim september exceeded: claimed=%v err=%v", claimed, err)
}
}
func TestClaim_ForeignKeyViolationReturnsFalseWithNoError(t *testing.T) {
repo, _ := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
claimed, err := repo.Claim(ctx, uuid.NewString(), "lifetime", "alert")
if err != nil {
t.Fatalf("claim against a nonexistent policy must not be a store error, got: %v", err)
}
if claimed {
t.Fatal("claim against a nonexistent policy must not win")
}
}
func TestDeleteBudgetPolicy_CascadesClaims(t *testing.T) {
repo, db := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-cascade-1")
if _, err := repo.Claim(ctx, policy.ID, "lifetime", "alert"); err != nil {
t.Fatalf("claim: %v", err)
}
if got := countBudgetClaims(t, db, policy.ID); got != 1 {
t.Fatalf("claim count before delete = %d, want 1", got)
}
if err := repo.DeleteBudgetPolicy(ctx, policy.ID); err != nil {
t.Fatalf("delete: %v", err)
t.Fatalf("delete budget policy: %v", err)
}
if got := countBudgetClaims(t, db, policy.ID); got != 0 {
t.Fatalf("claim count after single-policy delete = %d, want 0", got)
}
}
func TestDeleteWorkspaceData_CascadesBudgetClaims(t *testing.T) {
repo, db := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-cascade-2")
if _, err := repo.Claim(ctx, policy.ID, "lifetime", "alert"); err != nil {
t.Fatalf("claim: %v", err)
}
if err := repo.DeleteWorkspaceData(ctx, "ws-cascade-2"); err != nil {
t.Fatalf("delete workspace data: %v", err)
}
if got := countBudgetClaims(t, db, policy.ID); got != 0 {
t.Fatalf("claim count after workspace delete = %d, want 0", got)
}
}
func TestDeleteBudgetPoliciesForRemovedScopes_CascadesClaims(t *testing.T) {
repo, db := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-cascade-3")
if _, err := repo.Claim(ctx, policy.ID, "lifetime", "alert"); err != nil {
t.Fatalf("claim: %v", err)
}
// No valid scope IDs remain, so the reconciliation sweep removes every
// workspace-scoped policy (mirrors infra.Reconciler.reconcileBudgetPolicies).
if _, err := repo.DeleteBudgetPoliciesForRemovedScopes(ctx, "workspace", nil); err != nil {
t.Fatalf("delete for removed scopes: %v", err)
}
if got := countBudgetClaims(t, db, policy.ID); got != 0 {
t.Fatalf("claim count after reconcile delete = %d, want 0", got)
}
}
func TestUpdateBudgetPolicy_DiscardsClaims(t *testing.T) {
repo, db := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-update-1")
if _, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert"); err != nil {
t.Fatalf("claim: %v", err)
}
if got := countBudgetClaims(t, db, policy.ID); got != 1 {
t.Fatalf("claim count before update = %d, want 1", got)
}
policy.LimitSubcents = 2000
if err := repo.UpdateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("update budget policy: %v", err)
}
if got := countBudgetClaims(t, db, policy.ID); got != 0 {
t.Fatalf("claim count after update = %d, want 0 (AC-OFFICE-COSTS-002.8)", got)
}
updated, err := repo.GetBudgetPolicy(ctx, policy.ID)
if err != nil {
t.Fatalf("get updated policy: %v", err)
}
if updated.LimitSubcents != 2000 {
t.Fatalf("limit_subcents = %d, want 2000", updated.LimitSubcents)
}
}
func TestListBudgetPolicies_OrdersByCreatedAtThenID(t *testing.T) {
repo, db := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
now := time.Now().UTC().Truncate(time.Second)
// Two policies sharing the same created_at have no defined order from
// created_at alone; id is the deterministic tiebreak (AC-OFFICE-COSTS-002.16).
for _, id := range []string{"policy-b", "policy-a"} {
if _, err := db.Exec(
`INSERT INTO office_budget_policies (
id, workspace_id, scope_type, scope_id, limit_subcents, period,
alert_threshold_pct, action_on_exceed, created_at, updated_at
) VALUES (?, 'ws-order', 'workspace', 'ws-order', 100, 'monthly', 80, 'notify_only', ?, ?)`,
id, now, now,
); err != nil {
t.Fatalf("insert %s: %v", id, err)
}
}
policies, err := repo.ListBudgetPolicies(ctx, "ws-order")
if err != nil {
t.Fatalf("list budget policies: %v", err)
}
if len(policies) != 2 {
t.Fatalf("policies = %d, want 2", len(policies))
}
if policies[0].ID != "policy-a" || policies[1].ID != "policy-b" {
t.Fatalf("order = [%s, %s], want [policy-a, policy-b]", policies[0].ID, policies[1].ID)
}
}
+2
View File
@@ -100,6 +100,8 @@ dashboard projections, and Office testing contracts.
- [Office Config Sync Reconciliation System Design](system-design/config-sync-reconciliation.md)
- [Office: Cost Tracking & Budget Management System Design Part 1](system-design/costs-01.md)
- [Office: Cost Tracking & Budget Management System Design Part 2](system-design/costs-02.md)
- [Office: Budget Notification Idempotency System Design (Part 3)](system-design/costs-03.md)
- [Office: Budget Notification Idempotency System Design (Part 4)](system-design/costs-04.md)
- [Office Live Updates System Design Part 1](system-design/live-updates-01.md)
- [Office Live Updates System Design Part 2](system-design/live-updates-02.md)
- [Office per-agent and per-role tier selection System Design Part 1](system-design/office-agent-tier-routing-01.md)
+252 -1
View File
@@ -2,7 +2,7 @@
status: active
system: office
created: 2026-04-25
updated: 2026-08-18
updated: 2026-09-02
owners:
- cfl
---
@@ -12,6 +12,31 @@ owners:
Autonomous agents consume tokens on every turn, and when multiple agents run unattended across many tasks, spending can escalate without visibility. Kandev tracks analytics (task counts, turn counts, agent usage) but has no concept of monetary cost. Users have no way to know how much a task cost, which agent is most expensive, or to set guardrails that prevent runaway spending. Without cost tracking and budgets, autonomous operation is a financial black box.
## Terminology
- **Evaluation period:** The window over which a policy's spend is summed,
identified by the window's start instant. A policy whose window is the whole
history of the workspace has a single, non-resetting period.
- **Alert level:** `limit_subcents * alert_threshold_pct / 100`, using the
existing integer arithmetic.
- **Limit level:** `limit_subcents`.
- **Reaches a level:** Spend is at or above the alert level and strictly below
the limit (alert level), or spend is at or above the limit (limit level).
These are mutually exclusive within one evaluation; this requirement does not
change that.
- **Claim:** A durable record that a given (policy, evaluation period, level) has
already produced its notification.
- **Emit:** Write the corresponding `budget.alert` or `budget.exceeded` row to
the Office activity log.
- **Submit:** Hand that row to the Office activity logger. Submission is
observable to the evaluation that performs it; the durability of the resulting
write is not, because `ActivityLogger.LogActivity` returns nothing. Criteria
describing what an evaluation *reports about itself* therefore say "submit";
criteria describing the user-visible outcome say "emit". Submitting is not the
same as holding a claim, and neither implies the other: an evaluation can hold a
claim without submitting (AC-OFFICE-COSTS-002.5), and submit without holding one
(AC-OFFICE-COSTS-002.14).
## Requirements
### REQ-OFFICE-COSTS-001: Office: Cost Tracking & Budget Management
@@ -29,6 +54,232 @@ Autonomous agents consume tokens on every turn, and when multiple agents run una
- **AC-OFFICE-COSTS-001.7:** For providers without usable wire telemetry (codex per-turn split, amp), a disk-runner subsystem reads on-disk session files via pinned `@ccusage/*` packages and feeds normalized cost events into the same pipeline.
- **AC-OFFICE-COSTS-001.8:** The cost explorer UI surfaces aggregations and lets the user manage budget policies.
### REQ-OFFICE-COSTS-002: Budget notifications fire on crossings, not on every evaluation
**Intent:** `budget.alert` and `budget.exceeded` are notifications, and the
[state machine](../system-design/costs-01.md#state-machine) defines them as
crossing events ("spend crosses `alert_threshold_pct`"). Budget evaluation is
implemented as a level check that runs on unrelated triggers: after every cost
event, through the pre-execution budget API, and after every task reassignment
into a project. Once a policy sits above its threshold each of those writes
another activity row, so one over-budget policy floods the Office inbox (which
lists both budget notification rows and counts them toward the badge) with
duplicates carrying no new information. A reassignment adding no spend at all
still produces one. This requirement makes the notification idempotent per
policy, per period, per level, while leaving enforcement (pausing an agent,
blocking a run) the level check it must remain. The current scheduler admission
path uses `admitRun` and the pure-read `EvaluatePreLaunch` evaluator; that path
does not emit these notification rows.
**User story:** As a workspace owner with an over-budget project, I want one
budget notification per policy per period, so that my inbox stays usable and a
new notification still means something changed.
#### Acceptance criteria
Terminology used below is defined in [Terminology](#terminology).
- **AC-OFFICE-COSTS-002.1:** When a budget policy is evaluated and its spend is
at or above its alert level but below its limit, and no alert-level claim
exists for that policy and evaluation period, the system shall record the
claim and submit a `budget.alert` row to the Office activity log.
- **AC-OFFICE-COSTS-002.2:** When a budget policy is evaluated and its spend is
at or above its limit, and no exceeded-level claim exists for that policy and
evaluation period, the system shall record the claim and submit a
`budget.exceeded` row to the Office activity log.
- **AC-OFFICE-COSTS-002.2a:** Every criterion asserting an emission outcome
(AC-OFFICE-COSTS-002.1, .2, .7, .8, .10, .14, .17, and any added later) requires the
row to be *submitted*, not durably observed: `LogActivity` returns no error, so a
failed activity-log write is invisible to the evaluation that claimed.
At most one notification per (policy, evaluation period, level) may be lost this
way, and the system shall not retry it, release the claim, or report that loss.
This is the accepted direction of claim-then-emit ordering; the opposite ordering
would re-introduce the flood this requirement exists to remove.
- **AC-OFFICE-COSTS-002.3:** When a budget policy is evaluated and a claim
already exists for the level that the current spend reaches, for that policy
and evaluation period, the system shall record no activity row for that level.
- **AC-OFFICE-COSTS-002.4:** AC-OFFICE-COSTS-002.1 through AC-OFFICE-COSTS-002.3 shall hold identically
for every caller that evaluates a policy through `evaluatePolicy`: the
post-cost-event hook, the `CheckPreExecutionBudget` API, and the
task-reassignment project evaluation. The behavior shall be implemented once,
at the single evaluation function all these callers converge on, so that no
caller carries its own copy of it. The current scheduler admission path uses
`admitRun` and the pure-read `EvaluatePreLaunch` evaluator; it does not emit
these notification rows.
- **AC-OFFICE-COSTS-002.4a:** The task-reassignment caller
`EvaluateProjectBudget` SHALL use the same `evaluatePolicy` path. Tests SHALL
cover alert, exceeded, and repeated-evaluation behavior for this caller, plus
claim sharing with `CheckBudget`.
- **AC-OFFICE-COSTS-002.5:** When an evaluation records an exceeded-level claim,
the system shall also record the alert-level claim for the same policy and
period, so that a spend that jumps from below the alert level to above the
limit never emits a later `budget.alert` for the same period. This obligation is
conditional on the claim store accepting that companion write. When the
companion alert-level claim fails with a claim-store error, the system shall not
retry it, shall not fail the evaluation, and shall not withhold the
`budget.exceeded` row that the exceeded-level claim has already earned; it shall
record the failure exactly as AC-OFFICE-COSTS-002.14 requires.
AC-OFFICE-COSTS-002.14 takes precedence over this criterion, for the same reason
it takes precedence over AC-OFFICE-COSTS-002.10: a degraded claim store must
degrade to an extra notification, never to a failed evaluation or a withheld
`budget.exceeded`. The cost is bounded and accepted — the de-escalation
protection lapses for that one period, so a later evaluation landing in the alert
band may emit one `budget.alert`. A test of this criterion shall therefore assert
the companion claim on a healthy store, and a fault-injection test shall not
assert that the companion claim was recorded.
- **AC-OFFICE-COSTS-002.6:** A claim shall be scoped to the policy's evaluation
period, and the evaluation period shall be the same window used to compute
that policy's spend, so a policy whose spend window never resets shall have a
claim that never resets.
- **AC-OFFICE-COSTS-002.6a:** A single evaluation shall determine its period
boundary exactly once, and shall use that one value both to compute the policy's
spend and to identify the claim. An evaluation shall not derive the boundary a
second time from a second reading of the clock. The unit here is the evaluation
of one policy: when one trigger evaluates several policies, each policy's spend
and claim shall agree with each other, but the policies need not share a boundary
value, since no criterion compares one policy's period to another's. An evaluation spanning a period boundary would otherwise sum spend
against one window and claim against the next, re-arming the notification once
per boundary for every policy whose window resets — invisible in steady state,
so it shall be closed structurally rather than left to the implementation.
- **AC-OFFICE-COSTS-002.7:** When the evaluation period advances to a new
window, the system shall treat the new period as unclaimed and emit again on
the first evaluation that reaches a level.
- **AC-OFFICE-COSTS-002.8:** When a budget policy is updated, the system shall
discard that policy's claims for every period and level, so the next
evaluation that reaches a level under the updated policy emits. The row update
and the claim discard shall be atomic: either both apply or neither does. When
the discard fails, the update shall fail with it and return an error to the
caller, leaving the stored policy and its claims both unchanged. Partially
applying the pair is prohibited: an update that succeeds while its discard fails
leaves claims that silently suppress the notification this criterion mandates.
The discard's guarantee is scoped to evaluations that begin
after the update commits. An evaluation already in flight when the discard
commits has read its policy, spend and levels beforehand, and may therefore
insert a claim keyed to the pre-update period and level, suppressing the first
post-update notification; that outcome is permitted rather than prevented. The
window is bounded by one evaluation's duration, costs at most one suppressed
notification, self-corrects at the next period, and is side-stepped entirely by a
`period` change, which moves `period_key`. The alternative, a lock spanning both
the evaluation and the update, is rejected: it would put a user-facing write
behind a background evaluation.
- **AC-OFFICE-COSTS-002.9:** When a budget policy row is deleted by any path, the
system shall delete that policy's claims. This shall hold for all three
deletion paths that exist: single-policy deletion, workspace deletion, and the
startup reconciliation that removes policies whose agent or project scope no
longer exists. Claim deletion shall be a property of the policy row's removal,
not of any one calling method, so a future fourth deletion path inherits it
without change.
- **AC-OFFICE-COSTS-002.10:** When two evaluations of the same policy, period and
level run concurrently **and the claim store returns no error to either
evaluation**, the system shall submit exactly one activity row for that level.
When the claim store returns an error to either evaluation,
AC-OFFICE-COSTS-002.14 takes precedence over this criterion and duplicate rows
are permitted: the healthy caller emits because it won the claim, the faulted
caller because it fails open. Ordered this way deliberately — a broken claim store
must degrade to duplicate notifications, never to silence, and a duplicate under
an already-degraded store is the pre-existing behavior this requirement replaces,
so it is no worse than today.
- **AC-OFFICE-COSTS-002.11:** Claim state shall survive a backend restart: a
restart within an unchanged evaluation period shall not cause a policy that
has already emitted at a level to emit again at that level.
- **AC-OFFICE-COSTS-002.12:** Suppressing a notification shall not suppress
enforcement: when spend is at or above the limit, `pause_agent` shall still
pause an unpaused in-scope agent, and `pause_agent` / `block_new_tasks` shall
still deny a pre-execution budget check, regardless of whether a claim already
exists.
- **AC-OFFICE-COSTS-002.13:** An evaluation result shall report, per level, both
whether the policy is at or above that level and whether this evaluation
submitted the notification, so a caller can distinguish "over budget" from
"newly over budget". "Submitted" carries exactly the meaning fixed in
[Terminology](#terminology) — this evaluation handed that level's row to the
activity logger — and nothing beyond it. The claim is the mechanism that decides
whether to submit; it is not part of what the field reports. The field shall
therefore be true in every case where this evaluation put a row into the
activity log and false in every case where it did not, which settles every case,
including these:
- reaches the level and wins the claim: row submitted, field true;
- reaches the level but an earlier evaluation already holds the claim: no row,
field false;
- records the companion alert-level claim under AC-OFFICE-COSTS-002.5 while
emitting `budget.exceeded`: no `budget.alert` row, so the alert-level field is
false even though this evaluation held that claim;
- the claim store errors and the evaluation emits anyway under
AC-OFFICE-COSTS-002.14: a row went out, so the field is **true** even though no
claim was held.
The last case is the one a "held the claim *and* submitted" reading gets wrong:
it would report false while duplicating a user-visible notification, hiding every
fail-open emission from the callers this field exists to serve. Narrowed by
AC-OFFICE-COSTS-002.2a in the other direction: `LogActivity` returns nothing, so
a caller shall not read the field as a guarantee that the row is readable.
- **AC-OFFICE-COSTS-002.14:** When the claim store cannot be read or written, the
system shall emit the notification and record the failure as an error log and
a counter increment, so a broken claim store degrades to the previous
duplicate-notification behavior rather than to silence. An evaluation that emits
under this criterion reports `submitted` true, per AC-OFFICE-COSTS-002.13. This
criterion is scoped to claim reads and writes performed **during an evaluation**,
the only place a claim-store fault is otherwise silent: the evaluation continues,
the caller sees no error, and without the counter nothing distinguishes a broken
store from a quiet one. A failed claim *discard* during a policy update is
deliberately outside it and shall not be counted or logged under it:
AC-OFFICE-COSTS-002.8 governs it, rolling the update back and returning the error,
so it is already visible to whoever made the edit, and a second weaker signal
would imply the discard can fail silently, which AC-OFFICE-COSTS-002.8 forbids.
- **AC-OFFICE-COSTS-002.14a:** A claim rejected because the policy it references
no longer exists is not a claim-store failure and shall be excluded from
AC-OFFICE-COSTS-002.14: the system shall record no activity row, no error log and
no counter increment. A policy deleted mid-evaluation warrants no notification,
and counting its absence as a storage fault would both emit a notification naming
a policy the user removed and report a healthy store as broken.
- **AC-OFFICE-COSTS-002.15:** When a policy's spend falls back below a level it
has already claimed within the same period, the system shall keep the claim, so
spend that re-crosses the same level in the same period does not emit again.
- **AC-OFFICE-COSTS-002.16:** Policies applicable to one evaluation shall be
evaluated in ascending `created_at` order with ascending `id` as the tiebreak.
- **AC-OFFICE-COSTS-002.17:** On the first evaluation after this behavior is
deployed, a policy already at or above a level shall have no claim for the
current period and shall therefore emit once at that level, and then be quiet
for the remainder of the period.
## System design
The migrated technical source is split into [part 1](../system-design/costs-01.md), [part 2](../system-design/costs-02.md).
`REQ-OFFICE-COSTS-002` is designed in [part 3](../system-design/costs-03.md) (the
claim store and control flow) and [part 4](../system-design/costs-04.md)
(suppression boundary, failure and recovery, persistence, security,
observability). Parts 3 and 4 are one design; part 3 carries the requirement
mapping for both.
## Out of scope
These exclusions scope `REQ-OFFICE-COSTS-002`.
- **Changing when a policy reaches a level.** The arithmetic named under
[Terminology](#terminology) is preserved verbatim, including its existing edge
behavior: `alert_threshold_pct = 100` never reaches the alert level (the
alert level equals the limit and the alert branch requires spend strictly
below the limit); `limit_subcents <= 0` is permanently at the limit level; and
an evaluation never reaches both levels at once. This requirement changes how
often a level emits, never whether it is reached. Correcting any of that is a
separate contract change.
- **Unifying the three spend-window implementations.** `costs.periodCutoff`,
`backendapp.budgetEvaluator.spendForPolicy` and `cron.periodStartFor` each
compute a period boundary for a different caller. The cost evaluator uses
calendar starts for `daily`, `monthly`, and `yearly` policies, and `lifetime`
for `total`. The backend cron evaluator computes a monthly start and uses a
lifetime window for other periods. The scheduler cron helper also supports
weekly periods. This requirement does not unify these implementations or
change the separate cron notification channel.
- **The cron budget-alert trigger.** `internal/scheduler/cron.BudgetHandler` fires
`engine.TriggerOnBudgetAlert` on its own thresholds (50/80/90/100) with its own
in-process dedup, and is inert in production because
`budgetTaskScope.ResolveAlertTaskID` returns an empty task id. Prior art for the
claim key shape (see the system design), but a different notification channel and
unchanged here.
- **Resolving or expiring inbox items.** `budget.alert` rows are surfaced as inbox
items with a hardcoded `active` status and no resolve action. This requirement
reduces how many are created; it adds no lifecycle to the ones that exist.
- **`budget.exceeded` visibility.** The Office inbox list and badge count read both
`budget.alert` and `budget.exceeded` activity rows. This requirement makes both
idempotent without changing which surfaces read them.
- **A user-facing setting for notification frequency.** Idempotency is
unconditional; no per-policy "re-notify every N" control.
+368
View File
@@ -0,0 +1,368 @@
---
status: draft
system: office
requirements:
- REQ-OFFICE-COSTS-002
created: 2026-09-02
updated: 2026-09-02
owners:
- cfl
---
# Office: Budget Notification Idempotency System Design (Part 3)
Part 3 holds the contract: what the claim store is, and the control flow that uses
it. Its operational half — suppression boundary, failure and recovery, persistence,
security and observability — is [part 4](costs-04.md).
## Purpose and boundaries
Office owns budget policies, their evaluation, and the `budget.alert` /
`budget.exceeded` rows they produce. This part designs the durable claim state
that turns those two rows from a level check into a crossing notification, per
`REQ-OFFICE-COSTS-002`.
Adjacent contracts this design uses but does not own:
- **Office activity log** (`office_activity_log`, `shared.ActivityLogger`) is the
emission sink. It is fire-and-forget: `LogActivity` returns nothing, so a
failed write is not observable to the caller. That constraint drives the
claim-then-emit ordering below.
- **Office inbox** (`dashboard.DashboardService.inboxBudgetAlertItems`,
`GetInboxCount`) is the user-visible surface that reads both `budget.alert` and
`budget.exceeded` rows. It is unchanged; it simply stops receiving duplicates.
- **Spend rollups** (`Repository.GetCostForAgentSince` /
`GetCostForProjectSince` / `SumCostsSince`) define the spend window. This
design reads the window boundary they are given; it does not redefine it.
- **Agent pause** (`shared.AgentWriter.UpdateAgentStatusFields`) and the
pre-execution gate (`shared.BudgetChecker`) are enforcement, not notification,
and are deliberately outside the suppression boundary.
## Prior art
### Other prior reasoning — unavailable
The repository's own prior reasoning and external product references were
unavailable in this session. The in-repo prior art below provides the basis for
the claim key and persistence choices.
### In-repo prior art — found, and it decides the key shape
`internal/scheduler/cron/budget.go` already solved this exact problem for a
different notification channel. `BudgetHandler.tryFire` dedupes on
```text
workspace_id | scope_type | scope_id | threshold | period_start
```
and `periodStartFor` anchors that key to the policy's period start so it resets at
a boundary. Its own comment names the gap this design closes: *"Phase 5 keeps
state in-memory for simplicity; a process restart re-fires once per active
crossing [...] A future iteration may persist this."*
We take the key shape and reject two details. The scope tuple becomes
`policy_id`, because two policies can cover one scope (the schema and
`costs-01.md` both allow "monthly + total") and a scope-keyed claim would let one
policy's emission silence the other's. And the state is persisted rather than
in-process, because `costs-01.md` already promises "Monthly reset is idempotent:
backend restart mid-month does not refire", which an in-memory map cannot
deliver (AC-002.11).
We depart once more: the cron handler dedupes on a fixed ladder (50/80/90/100)
ignoring the policy's `alert_threshold_pct`, while this design keys on the two
levels the policy actually defines. The mechanisms stay independent; see the
requirement's `## Out of scope`.
## Requirement mapping
| Requirement | Design section |
| --- | --- |
| `REQ-OFFICE-COSTS-002` (all ACs) | [Control flow](#control-flow) |
| AC-002.1–.3, .2a, .5, .10 | [Claim-then-emit](#claim-then-emit) |
| AC-002.4, .4a | [Call sites](#call-sites) |
| AC-002.6, .6a, .7, .15 | [Period identity](#period-identity) |
| AC-002.8, .9, .11, .17 | [Persistence](costs-04.md#persistence) (part 4) |
| AC-002.8, .9, .14a | [Repository surface](#repository-surface) |
| AC-002.14 (counter, log), .8 (why the discard is excluded) | [Observability](costs-04.md#observability) (part 4) |
| AC-002.13 | [Evaluation result](#evaluation-result) |
| AC-002.12 | [Suppression boundary](costs-04.md#suppression-boundary) (part 4) |
| AC-002.10, .14, .14a | [Failure and recovery](costs-04.md#failure-and-recovery) (part 4) |
| AC-002.16 | [Evaluation order](#evaluation-order) |
## Components and responsibilities
- **`costs.CostService.evaluatePolicy`** — where all three call sites converge.
It computes spend and levels exactly as today and gains one responsibility: ask
the claim store whether this (policy, period, level) may emit, and emit only if
so.
- **Claim store** — a new Office-owned table plus repository methods on
`costs.Repository`: atomically claim a (policy, period, level), and discard a
policy's claims as part of that policy's update. No business logic, no clock,
and no deletion method — deletion is a schema property, see below.
- **`costs.BudgetCheckResult`** — extended so a caller can distinguish "at or
above this level" from "this evaluation emitted" (AC-002.13). The existing
`AlertFired` / `LimitExceed` fields keep their present meaning (level reached),
because `CheckPreExecutionBudget` gates on `LimitExceed` and must not start
allowing runs the moment a claim exists.
- **`office/repository/sqlite`** — owns the table DDL in the Office schema
initializer, the foreign key that deletes claims, and the transaction that
discards them on update. **The claim lifecycle lives at the persistence layer,
not at a service method.** This is the load-bearing decision of this design.
Anchoring it to a service method would be wrong here, because two service
facades already expose those exact method names —
`costs.CostService` (`internal/office/costs/budgets.go`) and
`office/service.Service` (`internal/office/service/service.go`) — and the second
calls `s.repo` directly, inheriting nothing. Three deletion paths already
exist, and only one of them goes through `DeleteBudgetPolicy` at all (see
[Persistence](costs-04.md#persistence)). Anchoring the rule to the row rather than to a
method is what makes AC-002.8 and AC-002.9 hold for every path, including the
ones nobody remembers to update.
Nothing changes in the frontend: the inbox and badge count read both budget
notification rows, and the costs page reads the same policy and activity data.
## Data and contracts
### Claim record
One row per (policy, evaluation period, level).
| Field | Type | Notes |
| --- | --- | --- |
| `policy_id` | string | The budget policy this claim belongs to. Not the scope tuple — see [Prior art](#prior-art). |
| `period_key` | string | Canonical identity of the evaluation period. See [Period identity](#period-identity). |
| `level` | string | Exactly `alert` or `exceeded`. Any other value is a bug, not an extension point. |
| `claimed_at` | timestamp | When recorded. Diagnostic only; never read for a decision. |
`(policy_id, period_key, level)` is the primary key. That uniqueness constraint
is the concurrency control (AC-002.10) — not an advisory lock, not a
read-then-write guard.
`policy_id` carries
`FOREIGN KEY (policy_id) REFERENCES office_budget_policies(id) ON DELETE CASCADE`.
That constraint — not a method — is how AC-002.9 is satisfied, for every deletion
path at once. The pattern is already established in this schema: nine `ON DELETE CASCADE`
foreign keys across seven tables, `office_routine_triggers` and `office_routine_runs` hang off
`office_routines` exactly this way, and the SQLite DSN sets `_foreign_keys=on` for
both the read-write and read-only pools, so the cascade is live rather than
declarative. `labels.go` already relies on the same mechanism in production.
`workspace_id` is intentionally absent: a claim is reachable only through its
policy, which already carries the workspace, and duplicating it would create a
second answer that could drift. The foreign key makes that structural.
### Repository surface
**One** new operation on `costs.Repository`, plus a change to an existing one.
Both are implemented in `office/repository/sqlite`:
- **Claim** `(policy_id, period_key, level) -> (claimed bool, err error)`.
Inserts the row and reports whether *this* call inserted it. Must be a single
statement so two concurrent callers cannot both observe "not claimed". Returns
`claimed=false` with no error when the row already existed.
**The conflict-resolution form is contract, not local style: use
`INSERT ... ON CONFLICT(policy_id, period_key, level) DO NOTHING`, and
specifically NOT `INSERT OR IGNORE`.** The targeted form resolves only the named
primary-key conflict and lets every other constraint violation surface as an
ordinary Go `error`, which is what makes the foreign-key rule below implementable
at all. `claimed` is the statement's **affected-row count**: 1 inserted, 0 already
existed. That reading holds on both dialects (SQLite's `sqlite3_changes()`
excludes a row skipped by conflict resolution; PostgreSQL's command tag counts
only inserted rows), so no `RETURNING` and no follow-up `SELECT` is needed — and a
`SELECT` would reintroduce the read-then-write race the primary key prevents.
`INSERT OR IGNORE` instead applies IGNORE resolution to *every* constraint
violation, foreign keys included: an insert failing only its FK check returns zero
rows and **no error**, indistinguishable from "already claimed", leaving nothing
for `db.IsForeignKeyViolation` and making the FK rule below dead code. Worth
naming because `INSERT OR IGNORE` is the established idiom *in this package* for a
table of this exact shape (`labels.go` writes `office_task_labels`, which carries
two `ON DELETE CASCADE` foreign keys, that way), and AC-002.14a's black-box
outcome converges under both forms, so no test catches the substitution.
**Foreign-key violation is not a store failure.** An evaluation that listed a
policy, then raced its deletion, fails the insert on the foreign key because the
parent row is gone. Treat that specific error as `claimed=false, err=nil` —
suppress the emission, do not log it, do not increment
`budget_claim_failures_total`. It is not a degraded store but a policy that no
longer exists, and none warrants a notification; routing it through AC-002.14's
fail-open path would emit a `budget.alert` naming a deleted policy and report a
healthy store as broken. Classify with the shared `internal/db` helpers, never a
local `strings.Contains` on the message.
- **`Repository.UpdateBudgetPolicy`** gains the discard, *inside* the same
transaction as the row update, and not as a separate exported method — that is
the point: both service facades already call this one repository method, so
putting the discard here is what makes them agree (AC-002.8, closing the second
half of the two-facade problem in
[Components](#components-and-responsibilities)). Order within the transaction:
delete the policy's claims, then update the row. On any error roll back and return
it; the caller sees a failed update with policy and claims both unchanged.
`UpdateBudgetPolicy` is today a single untransacted `ExecContext`, so this
introduces a `BeginTxx` / `Commit` / `Rollback` wrapper matching
`workspace_deletion.go`'s. The discard is idempotent: deleting zero claim rows is
success, not an error — a policy that never reached a level has no claims, and
that is the common case, not the edge case.
There is **no** "discard by policy" and **no** "discard by workspace" method: the
foreign key does both deletions, so an explicit method would be a second, divergent
answer to the same question — the one call sites forget. There is no read-only "is
claimed?" method either; it would invite a check-then-emit call site that
reintroduces the race the primary key exists to prevent.
### Evaluation result
`BudgetCheckResult` gains one boolean per level recording whether this
evaluation **submitted** that level's notification. Existing fields are
unchanged:
- `AlertFired` / `LimitExceed` — the policy reaches that level *now*. Level
check. Used for gating.
- the two new fields — this evaluation handed that level's row to the activity
logger. Crossing. Used for assertions and for any future caller that wants to
react only to new crossings.
A field reports **submission only**; holding the claim is the mechanism that
decides whether to submit and is not part of what the field says (AC-002.13
enumerates all four reachable cases). Two of them are counter-intuitive and are
why this is spelled out: the AC-002.5 companion claim is *held* while the
alert-level field reads **false** (no `budget.alert` row went out), and a
fail-open emission under AC-002.14 holds *no* claim while the field reads **true**
(a row did). Reading the field as "held the claim and submitted" gets that second
case backwards and hides every fail-open emission from the callers the field
exists to serve.
The word is **submitted**, not *emitted* (AC-002.13, and the `Emit` / `Submit`
split in the requirement's Terminology). `LogActivity` returns nothing, so the
evaluation cannot observe whether its row became durable. A field documented as
"emitted" would promise a durability guarantee AC-002.2a says is unavailable, and
a caller would be entitled to believe it.
## Control flow
### Period identity
`period_key` is derived from the same window boundary the policy's spend is
computed against, not from an independently-chosen calendar rule. The evaluation
reads the clock **once** and derives that boundary **once**: `evaluatePolicy`
calls `periodCutoff` with a single evaluation timestamp, hands the resulting
instant to the spend rollup, and formats that same instant as `period_key`
(AC-002.6a).
**This requires a contract change at an internal seam.** Today
`getSpendForPolicy` calls `periodCutoff(string(policy.Period), time.Now())`
itself and returns only `(int64, error)`, so the instant is discarded before
`evaluatePolicy` regains control and is unreachable from the claim site. The
boundary must move up to the caller — either `evaluatePolicy` computes it and
passes it in, or `getSpendForPolicy` returns it alongside the spend. Either
satisfies AC-002.6a; leaving the seam alone does not.
Recomputing the boundary at claim time is **prohibited**, and is named because it
is the path of least resistance: a second `periodCutoff(period, time.Now())`
compiles, reads naturally, and agrees with the first except exactly at a boundary
— spend summed against September, claim written against October — re-arming the
policy once per boundary forever, invisibly in steady state. That is why AC-002.6
is phrased as an equality rather than as a calendar:
- A non-zero window start renders as that instant formatted RFC3339 in UTC
(`2026-09-01T00:00:00Z`). The layout is fixed and is part of the contract: it
is a stored key, so changing the layout later orphans every existing claim and
re-fires every policy once.
- A zero window start — `periodCutoff`'s "no filter / lifetime" answer — renders
as the literal `lifetime`. It is not the RFC3339 rendering of the zero time,
because that reads as a real instant and would be silently reset by any future
change that starts passing a real epoch-anchored floor.
Daily and yearly policies use their calendar window starts, so their claims reset
at the same boundaries as their spend rollups. Total policies use `lifetime`, so
their claims do not reset. The cost evaluator and the separate cron handlers
still have different period implementations; this design does not unify them.
Because the key is the window start, a new window is automatically unclaimed
(AC-002.7), and no scheduled job, cleanup pass, or reset trigger exists or is
needed.
### Claim-then-emit
Within `evaluatePolicy`, for whichever level the spend reaches:
1. Compute spend and levels exactly as today.
2. If the spend reaches the **limit** level: claim `exceeded`; then also claim
`alert` (AC-002.5). Run the `pause_agent` side effect regardless — see
[Suppression boundary](costs-04.md#suppression-boundary). `Claim` returns
`(claimed bool, err error)`, so the `exceeded` claim has **three** outcomes and
each decides the emission differently. Do not collapse them to two:
- `claimed=true, err=nil` — this evaluation won the claim: **emit**.
- `claimed=false, err=nil` — an earlier evaluation already holds it, or the
policy was deleted mid-evaluation (AC-002.14a): **do not emit**.
- `err != nil` — the claim store is degraded: **emit anyway**, and log and count
it (AC-002.14, [Observability](costs-04.md#observability)). Suppressing here is
the one outcome AC-002.14 exists to prevent, and it is what any two-state
summary of this step ("emit only if the claim succeeded") silently produces,
which is why all three outcomes are written out instead.
The companion `alert` claim's **outcome** is discarded — claimed and
already-claimed are equally fine, and neither changes whether
`budget.exceeded` is emitted. Its **error** is not discarded: a claim-store
error here is logged and counted exactly like any other (see
[Observability](costs-04.md#observability)), because it is the same degraded
store AC-002.14 exists to make visible. It still does not change the
`budget.exceeded` emission, and it is not retried. Naming this matters
because the consequence is specific and otherwise silent: if the companion
claim errors, the de-escalation hole below reopens **for that period only** —
a later evaluation landing in the alert band emits one `budget.alert` — and
the counter is the only signal that this is what happened. **AC-002.5 carries
the matching carve-out**: its "shall also record" is conditional on the store
accepting the companion write, with AC-002.14 taking precedence when it does
not. So a test of AC-002.5 asserts the companion claim on a **healthy** store,
and a fault-injection test must not assert it was recorded — the same split
AC-002.10 and AC-002.14 already require.
3. Else if the spend reaches the **alert** level: claim `alert`, and decide the
`budget.alert` emission from the same three outcomes as step 2.
4. Else: no claim, no emission. Note that a spend that has fallen back below a
claimed level does **not** release the claim (AC-002.15); nothing in this
flow ever deletes a claim.
The order is claim first, emit second, and that ordering is forced rather than
preferred. `ActivityLogger.LogActivity` returns nothing, so emit-then-claim
cannot know whether it has anything to claim, and a claim written after a failed
emit would silence a notification that never happened. Claim-then-emit has the
opposite failure: a successful claim whose emission then fails loses exactly one
notification for that period. That is the accepted trade — one lost notification
under an already-degraded activity log, versus a flood under normal operation,
which is the defect being fixed.
Step 2's unconditional alert-claim closes the de-escalation hole: without it, a
policy whose spend jumps straight past the limit never claims `alert`, so a
later reassignment that drags spend back into the band between the alert level
and the limit would emit a `budget.alert` — a *new* notification describing a
*reduction* in spend.
### Call sites
The post-event hook, the pre-execution budget API, and task reassignment converge
on `evaluatePolicy`, so they inherit the behavior with no per-caller logic
(AC-002.4):
| Caller | Trigger | Path |
| --- | --- | --- |
| `Service.CheckBudget` (post-cost-event subscriber, `service/event_subscribers.go`) | every recorded cost event | `EvaluateBudget` → `CheckBudget` → `evaluatePolicy` |
| `CostService.CheckPreExecutionBudget` (pre-execution budget API) | every direct pre-execution check | `CheckPreExecutionBudget` → `CheckBudget` → `evaluatePolicy` |
| `DashboardService` task reassignment (`dashboard/service_tasks.go`) | every reassignment into a project | `EvaluateProjectBudget` → `evaluatePolicy` |
The current scheduler admission path does not use `CheckPreExecutionBudget`.
`SchedulerIntegration.admitRun` calls `EvaluatePreLaunch`, which is a pure read
and does not emit `budget.alert` or `budget.exceeded`; its admission activity
entries are owned by `admitRun` instead.
### Evaluation order
`Repository.ListBudgetPolicies` currently orders by `created_at` alone, which is
not unique: two policies created in the same second have no defined relative
order, so which emits first can differ between runs. The ordering becomes
`created_at ASC, id ASC` (AC-002.16); `id` is the primary key, so the ordering is
total. This matters because each policy's emission is now one-shot per period, so
a non-deterministic order makes the activity-log sequence non-reproducible and
any test asserting on it flaky rather than wrong.
---
Enforcement boundaries, failure handling, persistence, security and observability
are designed in [part 4](costs-04.md). They are split out because this design
reached the specification linter's 32,768-byte ceiling for a system-design file;
the two parts are one design and part 4 has no requirement mapping of its own.
+213
View File
@@ -0,0 +1,213 @@
---
status: draft
system: office
requirements:
- REQ-OFFICE-COSTS-002
created: 2026-09-02
updated: 2026-09-02
owners:
- cfl
---
# Office: Budget Notification Idempotency System Design (Part 4)
The operational half of [part 3](costs-03.md), split out when that file reached the
specification linter's 32,768-byte ceiling for a system-design file. Part 3 defines
the claim store and the control flow that uses it; part 4 defines what the claim
must **not** gate, how the design behaves when a component fails, where the state
lives, and what it exposes. The two parts are one design against
`REQ-OFFICE-COSTS-002` and share part 3's
[requirement mapping](costs-03.md#requirement-mapping).
## Suppression boundary
The most dangerous way to implement this requirement is to gate too much on the
claim. Enforcement is a level check and must stay one (AC-002.12):
- **`pause_agent`** runs on every evaluation where spend is at or above the limit,
claim or no claim: a user who un-pauses an agent that is still over budget must
see it re-paused next evaluation. `pauseAgentForBudget` already no-ops on an
already-paused agent, so repeating it is cheap and idempotent.
- **`CheckPreExecutionBudget`** keeps denying on `LimitExceed` for `pause_agent`
and `block_new_tasks` policies, reading the level and never the claim. A policy
that already emitted must still block every subsequent run.
- **`BudgetCheckResult.AlertFired` / `LimitExceed`** keep meaning "reaches this
level". Only the two new fields report submission.
One rule: the claim gates writes to `office_activity_log` and nothing else.
## Failure and recovery
- **Claim store read/write error** (AC-002.14): treat the claim as not held, emit,
log at error level, increment the counter, and report `submitted` true because a
row went out (AC-002.13). Rationale: this requirement fixes a flood produced by
*normal* operation, whereas a broken store is exceptional, and degrading it to
"duplicates plus an error signal" beats "silently stop warning the user about
money". A store broken for a whole period reproduces today's behavior exactly,
and the counter makes that visible. The error must not fail the evaluation:
`evaluatePolicy` still returns its result and the cost event is not rolled back.
- **Activity write fails after a successful claim**: one notification is lost for
that (policy, period, level). Not detectable from the caller; see
[Claim-then-emit](costs-03.md#claim-then-emit) for why this is the accepted direction.
- **Concurrent evaluation of the same policy**: resolved by the primary key.
Exactly one caller inserts; the other observes `claimed=false` and suppresses.
AC-002.10's "exactly one" is a count of **submissions**, not of durable rows:
assert it on the activity logger, never with a `SELECT` against
`office_activity_log`. AC-002.2a makes the durability of any one row unobservable,
so a durable-row assertion is testing something the contract does not promise.
No transaction spanning claim and emit is required, and none should be added —
the activity write is not transactional with anything.
- **Concurrent evaluation where the claim store also faults**: the two rules above
compose into a duplicate, and AC-002.10 says so explicitly. The healthy caller
wins the insert and emits; the faulted caller fails open per AC-002.14 and also
emits. Two rows for one crossing. **AC-002.14 takes precedence over AC-002.10** —
a documented tie-break, not an oversight: a duplicate under a broken store is
today's behavior, whereas suppressing on claim-store error converts a storage
fault into silent under-notification about money. Do not "fix" it with a retry
loop inside the claim; a retry outliving the evaluation moves the fault into the
cost-event path, which AC-002.14 forbids. Unlikely on SQLite, where
`internal/db/pool.go` uses a single writer connection to serialize writes; a real
path on Postgres. Test consequence, which is why this is written down: the
AC-002.10 concurrency test must run against a **healthy** store, and the
AC-002.14 fault-injection test must not assert single emission. Written the other
way round they contradict each other and one is flaky rather than wrong.
- **Claim discard fails during a policy update** (AC-002.8): the whole update
fails and rolls back. See [Repository surface](costs-03.md#repository-surface). The
caller gets an error, the policy keeps its old values, and the claims survive
intact — consistent with the un-updated policy they belong to.
- **An in-flight evaluation claims just after an update's discard commits**: it
read its spend and levels before the update, so it may insert a claim keyed to
the pre-update period and level, suppressing one notification under the new
policy. Real rather than theoretical: `CheckBudget` calls `ListBudgetPolicies`
once, then passes each already-read `*BudgetPolicy` into `evaluatePolicy`, so a
policy updated mid-loop is evaluated from a stale read. **AC-002.8 carries the
matching carve-out**, scoping the discard's guarantee to evaluations beginning
after the update commits. Bounded by one evaluation's duration, and accepted
rather than locked against: the alternative is a lock spanning the evaluation and
the update, putting a user-facing write behind a background evaluation. A
`period` change moves `period_key` and side-steps it entirely; a limit or
threshold change costs at most one suppressed notification, self-correcting at
the next period.
- **Backend restart mid-period**: claims are on disk, so a restart changes nothing
(AC-002.11) — the promise `costs-01.md` already makes and the cron handler's
in-memory map cannot keep.
- **Policy re-created with the same scope**: a new policy row means a new
`policy_id` means no claims, so it emits once. Correct — it is a new policy.
## Persistence
- **Table**: one new Office-owned table created in the Office SQLite schema
initializer, using `CREATE TABLE IF NOT EXISTS` so startup replay is safe. It
must be created **after** `office_budget_policies` in the initializer, because
its foreign key references that table: SQLite tolerates a forward reference at
`CREATE TABLE` and only fails later at DML time, but PostgreSQL rejects it
outright, so the ordering is a correctness requirement on one dialect and a
latent trap on the other. Per ADR 0027, the change needs fresh-plus-replay
coverage on the same connection at the schema owner's boundary, and no new
local `strings.Contains(err.Error(), ...)` replay classifier —
`db.IsDuplicateColumnError` / `db.IsAlreadyExistsError` are the only
classifiers.
- **Deployment / first run** (AC-002.17): the table starts empty, so the first
evaluation after deployment emits once per policy per reached level, then goes
quiet for the rest of the period. No backfill runs, and none should be written:
pre-claiming from existing activity rows would suppress the one notification
telling the user the new behavior is live, and the activity log carries the
policy id only inside a JSON `details` string with no index to match on.
- **Policy update**: discard all of the policy's claims on *any* update
(AC-002.8), unconditionally, without comparing old and new field values.
`limit_subcents` and `alert_threshold_pct` matter most since they redefine what
a level means, but `UpdateBudgetPolicy` receives only the new state, so a
narrower rule would need a read-compare-write against the stored row — both a
race and a second place for "changed" to drift. The cost is that editing an
unrelated field (say `action_on_exceed`) re-arms the notification once, which
is acceptable: a human just edited the policy. A `period` change moves
`period_key` anyway, so it would re-arm regardless.
- **Policy deletion, all three paths** (AC-002.9): handled by the foreign key's
`ON DELETE CASCADE`, so **no code change is required at any of them**. All three
are enumerated because only one goes through `DeleteBudgetPolicy`:
1. `Repository.DeleteBudgetPolicy` — the single-policy delete behind the HTTP
route.
2. The workspace-deletion sweep in `repository/sqlite/workspace_deletion.go`,
whose existing `DELETE FROM office_budget_policies WHERE workspace_id = ?`
now cascades. It needs **no** new statement, and must not gain one ordered
before the policies delete — the cascade already covers it.
3. `Repository.DeleteBudgetPoliciesForRemovedScopes`, called from
`infra.Reconciler.reconcileBudgetPolicies` at startup to drop policies
whose agent or project scope no longer exists. This path bypasses
`DeleteBudgetPolicy` entirely and would have orphaned claims permanently,
since [Retention](#persistence) is deliberately none.
A fourth deletion path added later inherits the cascade for free — the whole
argument for putting this in the schema rather than a service method, and why
AC-002.9 is phrased as a property of the row's removal.
- **Retention**: none. Growth is bounded by `policies × periods × 2` (two rows
per policy per period) and needs no pruning job. Row count is not surfaced.
- **Backups**: covered by the standard Office SQLite snapshot.
## Security
No new trust boundary. Claims are reachable only through a policy, and policies are
already workspace-scoped by `officeWorkspaceScopeMiddleware`. No new HTTP route is
added, so `officeParamScopeResolvers`, `officeScopedSubResourceParams` and
`officeWorkspacelessRoutes` need no entry and `TestOfficeRouteScopeCompleteness`
stays green unmodified. Claims carry no user content, spend figures or PII: a
policy id, a period key, a level, a timestamp.
## Observability
- **Counter** for claim-store failures (AC-002.14), exported through the
existing `expvar` surface at `/debug/vars` in dev mode. Concretely:
| | |
| --- | --- |
| Name | `budget_claim_failures_total` |
| Shape | `expvar.NewMap`, matching its siblings — **not** a plain `Int` |
| Label key | `op`, with exactly one value: `claim` |
| Home | a new file in **`internal/office/costs`**, which owns `evaluatePolicy` and every increment site |
The home needs stating because the obvious neighbours sit in a *different*
package: `cost_events_written_total` / `cost_events_dropped_total` live in
`internal/office/service/cost_metrics.go`, and their increment helpers are
unexported methods on `*service.Service`, unreachable from `costs`. So this
counter sits beside the code that increments it, while following their *shape*
(label-keyed `expvar.Map`, not Int) because `/debug/vars` is a contract surface
and a lone Int among Maps would be the odd one out.
`claim` covers both the level claim and the AC-002.5 companion claim, and the
label stays at that one value: the log below carries the policy id and level, so
the counter does not need that cardinality and must not grow it.
**The in-transaction discard is deliberately NOT counted, and a later reader must
not "fix" the omission.** Three reasons, the first decisive. (1) A discard failure
is not silent: AC-002.8 rolls the update back and returns the error, so the person
who edited the policy sees it fail. This counter is for the failure that *is*
silent — an evaluation that fails open, continues, and tells nobody (AC-002.14 is
now scoped to evaluation-time claim reads and writes for exactly this reason). A
second, weaker signal for a failure that already surfaces would imply the discard
can fail quietly, which AC-002.8 forbids. (2) A discard spans every period and
level of one policy, so it has no `level` for the log line below; a sentinel would
make that field mean two things. (3) It is not reachable from this package: the
discard runs inside `Repository.UpdateBudgetPolicy` in `office/repository/sqlite`,
which has no production import of `internal/office/costs`; **both** facades reach
it and `office/service.Service.UpdateBudgetPolicy` calls `s.repo` directly without
entering `costs`; and the repository returns one `error` for both halves of the
transaction, so even from inside `costs` a discard failure is indistinguishable
from a row-update failure without a typed error this design does not define.
Counting it would mean an unnamed `sqlite`→`costs` import edge or over-counting
every failed update — the same two-facade trap
[Components](costs-03.md#components-and-responsibilities) puts the discard in the repository
to avoid.
A non-zero value means notifications are being duplicated and the claim store
needs attention — it is the only signal that distinguishes "quiet because
nothing crossed" from "loud because the store is broken".
- **Structured log** at error level on a claim-store failure during an
evaluation, carrying the policy id and the level being claimed. Both fields are
always available here, because every increment site is a claim attempt for a
known policy at a known level.
- **No log line on ordinary suppression.** Suppression is the steady state; a
log per suppressed evaluation would move the flood from the inbox to the log
file. The activity log itself is the record of what fired.
## Related decisions
- [ADR 0027: Replayable schema migrations across SQLite and Postgres](../../../decisions/0027-replayable-schema-migrations.md)