Documentation
¶
Overview ¶
Package ops supplies operational primitives for caller-owned factory loops: retry decisions, accounting and budget signals, capacity and degraded episodes, bounded events, fair slots and level-triggered wakes. It assigns no meaning to work keys, tags, roles or outcomes. Callers own workflow, logging, persistence migration and the consequences of operational signals.
Index ¶
- func InfraReason(res *agent.Result) string
- func OverBudget(cost, limit float64) bool
- func PauseUntil(now, resets time.Time, backoff, maxPause time.Duration) time.Time
- func RateLimitedText(msg string) bool
- func SelectProfile(p agent.Profile, n int) (agent.Profile, bool)
- func Sleep(ctx context.Context, delay time.Duration) error
- func Spend(entries []LedgerEntry, key work.Key, since time.Time) (cost float64, sessions, unknown int)
- type BudgetSignal
- type Bus
- type CapacityPause
- type Degraded
- type Event
- type FailureKind
- type Ledger
- type LedgerEntry
- type LedgerLineError
- type Member
- type OpFailure
- type RetryDecision
- type RetryPolicy
- type SharedPool
- type Streaks
- type Wake
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func InfraReason ¶
InfraReason names an infrastructure failure for logs and escalations.
func OverBudget ¶
OverBudget reports the strict crossing used for completed sessions and work. Nonpositive limits are unlimited; a cost exactly at the limit is allowed.
func PauseUntil ¶
PauseUntil uses backoff for a missing or past reset and caps a future reset. Backoff is caller policy and is preserved even if larger than maxPause.
func RateLimitedText ¶
RateLimitedText recognizes capacity and transient service-limit messages.
func SelectProfile ¶
SelectProfile walks n steps down a profile's fallback chain, the way a caller retrying a session that had no capacity moves on: the first retry runs the profile's fallback, the second that one's own, and a chain that ends sooner stays on its last profile. It reports whether the profile returned is a fallback; n <= 0 is the profile itself.
func Spend ¶
func Spend(entries []LedgerEntry, key work.Key, since time.Time) (cost float64, sessions, unknown int)
Spend totals sessions at or after since. An empty key selects all work. Every session counts as one, and unknown is how many of them reported no cost (LedgerEntry.CostUnknown): those add nothing to cost, which is only what the sessions that reported one cost, so a caller with unknown > 0 has a total that is short by an amount nobody knows.
Types ¶
type BudgetSignal ¶
BudgetSignal is a rolling-budget evaluation, independent of dispatch policy. Unknown is how many sessions in the window reported no cost, which Spent leaves out.
func EvaluateWindow ¶
func EvaluateWindow(entries []LedgerEntry, now time.Time, window time.Duration, limit, resumePercent float64, previouslyReached bool) BudgetSignal
EvaluateWindow preserves hysteresis: reach at limit, release strictly below the resume threshold. previouslyReached is the caller's prior evaluation.
type Bus ¶
type Bus struct {
// contains filtered or unexported fields
}
Bus publishes bounded, best-effort events. Slow subscribers lose new events without blocking producers. Channels live for the lifetime of the bus and are never closed. The supplied clock must be safe for concurrent calls.
type CapacityPause ¶
type CapacityPause struct {
// contains filtered or unexported fields
}
CapacityPause tracks an account-wide pause episode. Its zero value is ready.
func (*CapacityPause) Check ¶
func (p *CapacityPause) Check(now time.Time) (paused, released bool)
Check reports whether paused and whether an expired episode was released. Only the first check of an expired episode signals release.
func (*CapacityPause) Extend ¶
Extend never shortens an episode and reports whether this starts a new one.
func (*CapacityPause) Until ¶
func (p *CapacityPause) Until() time.Time
type Degraded ¶
type Degraded struct {
// contains filtered or unexported fields
}
Degraded tracks operation failures. A zero value is ready for use.
func (*Degraded) Record ¶
func (d *Degraded) Record(name string, failed bool, message string, now time.Time, threshold int) (OpFailure, bool)
Record clears a streak on success; otherwise it returns the new snapshot and a threshold-crossing signal once per streak. The caller supplies any error-text normalization and decides how to surface the crossing.
type Event ¶
type Event struct {
Work work.Ref `json:"work"`
Kind string
Activity string
Started time.Time
Phase string
Completed int
Total int
Success bool
Time time.Time
Role string
Session string
Dir string
Stage string
Round int
Model string
Fallback bool
Sandbox string
Outcome string
Note string
Turns int
CostUSD float64
CostKnown bool
Duration time.Duration
Err string
}
Event describes caller-defined activity. Kind, role, stage, phase and outcome are opaque strings with no workflow interpretation. Publishers own their input; each subscriber receives its own copy of Work.Tags.
type FailureKind ¶
type FailureKind int
FailureKind classifies why a session did not produce a usable result.
const ( // FailureNone is reserved for callers that have not classified a session; // ClassifyFailure returns FailureInfra or FailureBehavioural. FailureNone FailureKind = iota // FailureInfra is a failure of the machinery around the model — a // timeout, an API error, exhausted turns, a crashed agent process. // Retrying it later is likely to work. FailureInfra // FailureBehavioural is the session itself: it ran and reported any // outcome, or chose not to report. Retrying repeats the same decision. FailureBehavioural )
func ClassifyFailure ¶
func ClassifyFailure(res *agent.Result) FailureKind
ClassifyFailure decides whether a finished session is worth retrying.
func (FailureKind) String ¶
func (k FailureKind) String() string
type Ledger ¶
Ledger serializes append and trim operations. Share one instance per file. Dir and Now must not change while the ledger is in use. A nil Now uses time.Now.
func (*Ledger) AppendLedger ¶
func (s *Ledger) AppendLedger(e LedgerEntry) error
AppendLedger appends one entry to the ledger, creating it if needed. The line is written with a single Write to an O_APPEND file so concurrent workers never interleave. When the file does not end in a newline (a write torn by a crash), the entry is preceded by one, so it starts a line of its own: the torn tail is left as it is, and once it is a middle line ReadLedger fails closed naming it until a person removes it.
func (*Ledger) LedgerPath ¶
LedgerPath returns the ledger file.
func (*Ledger) ReadLedger ¶
func (s *Ledger) ReadLedger(since time.Time) ([]LedgerEntry, error)
ReadLedger returns the entries recorded at or after since (a zero since returns everything). The ledger is read fail-closed: a line that does not parse is a LedgerLineError naming the file and the line, and no entries are returned, so no caller can report or budget against a total that is quietly short of a session. The one exception is the final line, which is ignored when it does not parse: a session killed mid-write leaves a truncated tail, and that must never break accounting. Blank lines are ignored. Read and scan failures return an error without entries too.
func (*Ledger) TrimLedger ¶
TrimLedger removes the entries recorded before before and returns how many it removed. The ledger is rewritten through a temporary file renamed over it, so a reader sees either the old ledger or the trimmed one, and only when there is something to remove. A line that does not parse is kept where it is: its age is unknown, and ReadLedger is what reports it.
type LedgerEntry ¶
type LedgerEntry struct {
Work work.Ref `json:"work"`
Time time.Time `json:"time"`
Role string `json:"role"`
Session string `json:"session"`
Turns int `json:"turns"`
// CostUSD is what the session cost, when the agent reported a cost.
// CostUnknown says it reported none: the session ran and is counted,
// but its cost is not zero, it is not known, and no total it is part
// of is complete. An entry written before the field was recorded reads
// as known.
CostUSD float64 `json:"cost_usd"`
CostUnknown bool `json:"cost_unknown,omitempty"`
DurationMS int64 `json:"duration_ms"`
Outcome string `json:"outcome"`
ErrorSubtype string `json:"error_subtype"`
TimedOut bool `json:"timed_out"`
}
LedgerEntry records the accounting and caller-defined outcome of one session.
type LedgerLineError ¶
LedgerLineError is a ledger line that does not parse, other than the last one: which file, which line, and what was wrong with it.
func (*LedgerLineError) Error ¶
func (e *LedgerLineError) Error() string
func (*LedgerLineError) Unwrap ¶
func (e *LedgerLineError) Unwrap() error
type Member ¶
type Member struct {
// contains filtered or unexported fields
}
Member is one scheduler's membership of a SharedPool.
func (*Member) Acquire ¶
Acquire takes n slots without waiting, all of them or none. Refused, the member is queued, if it is not already, and woken when its turn comes.
func (*Member) Leave ¶
func (m *Member) Leave()
Leave drops a stopped scheduler's pending claim without releasing the slots its in-flight workers still own. Their normal releases handle those.
func (*Member) Pass ¶
func (m *Member) Pass()
Pass tells the pool the scheduler's dispatch pass ended. One in which nothing was refused leaves the queue: the scheduler has nothing waiting for a slot, and the turn passes on.
func (*Member) Pool ¶
func (m *Member) Pool() *SharedPool
Pool returns the pool this member belongs to.
type OpFailure ¶
type OpFailure struct {
Op string `json:"op"`
Count int `json:"count"`
// First and Last are the ends of the streak: the failure that started
// it and the most recent one.
First time.Time `json:"first"`
Last time.Time `json:"last"`
// LastError is the most recent caller-supplied error text.
LastError string `json:"last_error,omitempty"`
// Escalated records that this streak has already crossed its threshold.
// The caller decides whether that signal warrants escalation.
Escalated bool `json:"escalated,omitempty"`
}
OpFailure is a stable snapshot of one operation's current failure streak.
type RetryDecision ¶
RetryDecision reports whether and when the next attempt should run.
type RetryPolicy ¶
RetryPolicy counts retries after the initial attempt.
func (RetryPolicy) Decide ¶
func (p RetryPolicy) Decide(attempt int, retryable bool) RetryDecision
Decide evaluates a completed, one-based attempt. Callers may also pass retryable=true for a budget crossing eligible for another attempt.
type SharedPool ¶
type SharedPool struct {
// contains filtered or unexported fields
}
SharedPool bounds concurrent work across caller-owned loops. Refused members queue FIFO, with all-or-none claims. A served member rejoins at the tail if it needs more, so repeated passes share slots round-robin. End each dispatch pass with Pass and call Leave when a loop stops; in-flight claims are still released normally. Claims must be between one and Size slots.
func NewSharedPool ¶
func NewSharedPool(size int) *SharedPool
NewSharedPool makes a pool of size slots. A size below 1 is one slot.
func (*SharedPool) InUse ¶
func (p *SharedPool) InUse() int
InUse is how many of them are taken right now.
func (*SharedPool) Join ¶
func (p *SharedPool) Join(name string, wake func()) *Member
Join makes a member called name. wake must be non-blocking and must not call back into the pool; it runs under the pool lock.
func (*SharedPool) Waiting ¶
func (p *SharedPool) Waiting() []string
Waiting names the schedulers queued for a slot, in the order they are served.
type Streaks ¶
type Streaks[K comparable] struct { // contains filtered or unexported fields }
Streaks counts consecutive crossings per caller-owned subject. Its zero value is ready. A false crossing clears only that subject's streak.
type Wake ¶
type Wake chan struct{}
Wake is a level-triggered notification for one caller-owned reconcile loop. Create one per loop with NewWake. Signals coalesce until consumed; never close it.
func (Wake) Drain ¶
func (w Wake) Drain()
Drain consumes a pending wake before a full reconciliation, which supersedes it.
func (Wake) Signal ¶
func (w Wake) Signal()
Signal requests a local reconciliation without blocking the producer.
func (Wake) Wait ¶
Wait runs local reconciliations until the scheduled tick or cancellation. The caller owns and stops the tick source. A wake never resets its cadence. A ready tick supersedes a pending wake. Signals during local reconciliation remain pending, since they may describe work arriving after its snapshot.
Source Files