internal/formation/formation.go
internal/formation · 377 lines · 12 declarations · source
This file carries the package documentation, rendered on the package page.
Declarations
type Former
type Former struct {
observations *pg.ObservationStore
facts *pg.FactStore
extractor *extract.Extractor
}
Former runs extraction over one turn and writes what it produced.
func NewFormer
func NewFormer(observations *pg.ObservationStore, facts *pg.FactStore, extractor *extract.Extractor) *Former
NewFormer builds a former over the stores and the extractor it drives.
type Report
type Report struct {
ObservationID string
MessagesRead int
FactsAsserted int
// ClaimsRejected is proposals refused by the vocabulary or by the span rule. Recorded, because
// their rate is the only evidence that either is wrong.
ClaimsRejected int
// ClaimsRetracted counts source-backed human withdrawals enforced before projection writes.
ClaimsRetracted int
// ClaimsRefusedByRole is claims that would have become facts about the principal, from a message
// the principal did not speak. Counted and not recorded: unlike the other two this is not a
// quality signal — it is the policy working exactly as designed, and its rate is already
// knowable from the roles of the messages in the turn.
ClaimsRefusedByRole int
}
Report is what one turn produced. Counted rather than returned in full, because the rows themselves are already in the database and a caller that wants them can read them there.
method Former.Form
func (f *Former) Form(ctx context.Context, schema pg.Schema, observation domain.Observation) (Report, error)
Form extracts from every message in a turn and writes the result.
The role is the MESSAGE's, never the turn's
A turn opened by the user does not make an assistant message three positions later speak for the principal. The policy is per message because that is the only granularity at which it is true, and passing the turn's role would quietly promote every model sentence in any turn a person started.
Every message is extracted, including ones whose claims will be refused
A message that cannot speak for the principal costs a model call whose every claim is then refused. Skipping it would save that call and produce the same rows — but only because formation would be applying the role policy itself, in a second place, using reasoning that has to stay in step with the first. The policy lives on the road to the fact table so that a path which forgets it does not exist, and that is worth more than the call.
Not marked formed here
Marking is the caller's, because "extraction ran" and "the watermark may advance" are separable: a caller re-running a turn to compare extractor versions wants the first without the second.
type Policy
33 lines of declaration
type Policy struct {
// MaxAttempts is how many times a turn is tried before it is parked. Generous rather than tight,
// because the failure that matters most is the one that is not the turn's fault.
MaxAttempts int
// RetryAfter is the wait before a failed turn is offered again. It doubles per attempt, so this
// is the first interval rather than the only one.
RetryAfter time.Duration
// TurnBudget bounds one attempt. Extraction is a model call per message, so a long turn is a
// request that never ends.
TurnBudget time.Duration
// Interval is how long the driver waits after a pass that found nothing.
Interval time.Duration
// RetentionInterval is how often expiry is swept. Separate from Interval because expiry is a
// daily-scale concern and running it every few seconds would scan an index for nothing.
RetentionInterval time.Duration
// ReportsPerPass bounds how many community reports one pass writes per scope, and
// SegmentsPerPass how many compaction segments: each is a model call, so the bound is what
// makes a pass's cost predictable.
ReportsPerPass int
SegmentsPerPass int
// DerivedScopesPerPass bounds how many projects with no backlog are given the derived passes in
// one tick. Each one is a partition rebuild and a model call, so it is bounded like the two
// above rather than by however many projects happen to owe a report.
DerivedScopesPerPass int
// NotificationsPerPass bounds how many deliveries one pass attempts. Unlike the two above this
// is not a model call but an outbound request to somebody else's server, and the bound exists
// for the same reason: a queue of failing destinations must not be able to stop memory forming.
NotificationsPerPass int
// SealInterval is how often the audit ledger is sealed. It is the exposure window: an entry
// written after the last seal is covered by nothing, so this number is the size of the gap
// somebody could write into unnoticed.
SealInterval time.Duration
}
Policy bounds what forming will try before it gives up.
Every field here exists because its absence is a way for the backlog to stop making progress without anybody being told.
func DefaultPolicy
func DefaultPolicy() Policy
DefaultPolicy is deliberately patient.
Six attempts with a doubling wait from ten seconds means a turn is not parked until roughly ten minutes of failing, which is long enough that a provider blip does not park a backlog and short enough that a genuinely poisonous turn does not hold a scope for a working day. No latency budget has been agreed anywhere in this system, so these are chosen for the shape of the failure rather than derived from a target — and that is worth saying out loud rather than presenting them as measured.
type Worker
type Worker struct {
pool *pgxpool.Pool
observations *pg.ObservationStore
former *Former
policy Policy
health func(context.Context, pg.HealthExecutor, int64, int64)
}
Worker forms a scope's backlog.
func NewWorker
func NewWorker(pool *pgxpool.Pool, observations *pg.ObservationStore, former *Former) *Worker
NewWorker builds a worker over a pool, a store and a former.
func NewWorkerWithPolicy
func NewWorkerWithPolicy(pool *pgxpool.Pool, observations *pg.ObservationStore, former *Former,
policy Policy) *Worker
NewWorkerWithPolicy is the same worker with the bounds stated, which is what a test needs in order to drive a turn to exhaustion without waiting ten minutes.
var ErrScopeBusy
var ErrScopeBusy = errors.New("another worker holds this scope")
ErrScopeBusy is returned when another worker is already forming this scope's backlog.
method Worker.Drain
func (w *Worker) Drain(ctx context.Context, schema pg.Schema, scope string) ([]Report, error)
Drain forms every unformed turn in a scope, oldest first, and returns what each produced.
One worker per scope, held by an advisory lock
Two workers taking the same turn would extract it twice and assert every fact twice, and the duplicates would be indistinguishable from a person saying the same thing in two conversations. The turns are picked one at a time by offset, so nothing weaker than a lock held across the whole drain prevents it: a lock inside the transaction that picks a turn is released before the model call it was meant to protect.
A scope, not the whole instance. Formation for one project has no reason to wait on another's, and an instance-wide lock would make a busy project starve every quiet one beside it.
Oldest first
The formed watermark is the offset below the lowest unformed turn. Forming newest-first would leave the oldest unformed and pin that number at the bottom while everything above it completed — a scope reported as entirely behind when it is nearly current.
method Worker.recordFailure
func (w *Worker) recordFailure(ctx context.Context, schema pg.Schema,
observation domain.Observation, cause error) (parked bool, err error)
recordFailure counts the attempt and parks the turn if it has run out of them.
Parking is a decision to stop trying, recorded and reversible: the observation and its messages are untouched, so unparking is one update and re-forming is a rebuild. It advances the watermark past the turn, which is what stops one turn nobody can form from freezing a scope — and the count of parked turns travels with freshness so that exception is visible rather than hidden.