analysis

package
v0.0.0-...-ec50285 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 18, 2026 License: AGPL-3.0 Imports: 35 Imported by: 0

Documentation

Overview

Package analysis runs a roster of LLM judges over finished chat sessions.

It is the generalized sibling of the skill efficacy pipeline (server/internal/skills/efficacy) and mirrors its shape exactly: a durable Postgres queue of scoring units — here (chat, judge) rather than (session, skill version) — is enqueued from the chats table, reserved against the organization's per-judge daily budget under an advisory lock, judged, and published to the chat_analysis_scores ClickHouse sink. Adding a new analysis is implementing the Judge interface and registering it in the roster; enabling it for an organization is a chat_analysis_settings row.

Index

Constants

View Source 
const (
	StatePending  = "pending"
	StateReserved = "reserved"
	StateFailed   = "failed"
)

Evaluation lifecycle states. Spending states are exactly reserved and scored: both hold a budget slot, and only StateFailed and a stale- reservation reset give one back.

View Source 
const (
	// InactivityWindow is how long a chat must have gone without a new message
	// before its session is considered finished and safe to judge.
	InactivityWindow = 30 * time.Minute

	// EnqueueLookback bounds how far back the enqueue walk reaches into a
	// project's chats. The walk restarts from the head each time it exhausts the
	// queue, so the lookback is what keeps a re-walk's cost proportional to
	// recent activity rather than to the project's full history. Sessions older
	// than this when the pipeline is enabled are never analyzed.
	EnqueueLookback = 14 * 24 * time.Hour

	// StaleReservationAfter is how long a reserved evaluation may go without an
	// updated_at bump before it is treated as crashed and returned to the queue.
	StaleReservationAfter = 24 * time.Hour

	// MaxModelAttempts is the number of model attempts an evaluation gets before
	// it terminates as failed.
	MaxModelAttempts = 3

	// ReservedClaimLease is how long a LoadReserved claim owns the rows it
	// returns. Sized exactly as the efficacy lease: one evaluation is bounded at
	// two minutes, a batch holds at most MaxReservedClaimBatch of them, and the
	// lease leaves half of itself spare over the worst sequential pass.
	ReservedClaimLease = 30 * time.Minute

	// MaxEnqueuePageSize is the widest page EnqueuePage will scan in one call.
	MaxEnqueuePageSize int32 = 100

	// PendingCandidatePage is how many pending evaluations one reservation reads
	// per keyset page.
	PendingCandidatePage int32 = 100

	// MaxPendingCandidatePages bounds the candidate walk of one reservation. The
	// whole walk runs inside the organization's advisory lock, so its cost has to
	// be a function of the bound rather than of the backlog.
	MaxPendingCandidatePages = 10

	// MaxReservedClaimBatch is the largest batch Reserve hands out or
	// LoadReserved claims — both are judged under the same lease.
	MaxReservedClaimBatch int32 = 10
)
View Source 
const DefaultJudgeDailyCap int32 = 100

DefaultJudgeDailyCap is the cap the settings surfaces suggest when switching a judge on for an organization that never stored one. The pipeline itself never falls back to it: a configured judge's row always stores its cap (the column is NOT NULL), and a judge with no row is off.

View Source 
const MaxSweepProjectPage int32 = 100

MaxSweepProjectPage is the widest page of projects one discovery call reads. A sweep reaches the rest of the estate by chaining the last project id it saw, which keeps a single call's cost independent of how many projects hold work.

View Source 
const (
	// WorkUnitsJudgeName keys the work-units judge in queue rows, settings rows
	// and score rows. The canonical constant lives in the telemetry repo so
	// score readers that cannot import this package share the same key.
	WorkUnitsJudgeName = telemetryrepo.ChatAnalysisJudgeWorkUnits
)
View Source 
const WorkUnitsScoreEventURN = "chat_analysis:work_units:score"

WorkUnitsScoreEventURN is the gram_urn stamped on work-units score events. attribute_metrics_summaries_mv admits rows by this exact value; keep the two in sync.

Variables

View Source 
var (
	// ErrModelFailure marks a failure the model owns: output that does not honour
	// the response contract, or a call that ran past its timeout. The publisher
	// charges these to the evaluation's attempt counter, and the third one is
	// terminal.
	ErrModelFailure = errors.New("chat analysis judge model failure")
	// ErrRetryable marks a failure the infrastructure owns: a throttled call or a
	// transport error. The publisher leaves the evaluation reserved and its
	// attempt counter untouched, so the same row is retried.
	ErrRetryable = errors.New("chat analysis judge failure is retryable")
)

Functions

func CallStructured 

func CallStructured(ctx context.Context, logger *slog.Logger, client openrouter.CompletionClient, limiter *ratelimit.Limiter, in JudgeInput, call StructuredCall) (string, string, error)

CallStructured performs one structured-output completion on the platform's internal key, drawing on the shared key-scoped judge rate limiter, and returns the raw response text plus the model that produced it. Errors wrap ErrModelFailure or ErrRetryable exactly as the publisher expects.

func NewObserver 

func NewObserver(logger *slog.Logger, signaler Signaler) chat.MessageObserver

NewObserver builds the chat.MessageObserver that turns durable chat-message persistence into a chat analysis wake. Register it on the chat message writer.

func Reserve 

func Reserve(ctx context.Context, db *pgxpool.Pool, judges *Judges, projectID uuid.UUID, cursor PendingCursor, batchSize int32) ([]Evaluation, PendingCursor, error)

Reserve moves pending evaluations to reserved, spending the organization's per-judge budgets for the current UTC day.

A project that no longer resolves reserves nothing and reports no error: a coordinator can be holding an id that was deleted between passes.

The whole pass is one transaction whose first statement is an advisory lock keyed on the project's organization. The lock is held to commit, so counting and reserving are serialised per organization even across its projects: a concurrent reserver waits and then reads the committed spend of the batch before it. That, not the row locks, is what makes double spending impossible.

Candidates are walked recent-first and every judge's counter is decremented in memory as units are admitted, so one batch can never grant more slots than the caps leave. The walk pages through the queue rather than reading a fixed head: a judge whose cap is spent would otherwise fill that head every pass and starve the judges behind it. The returned cursor is where the walk stopped; the zero cursor — returned when the walk reached the end of the queue — starts the next one at the head.

func ResetStaleReservations 

func ResetStaleReservations(ctx context.Context, db *pgxpool.Pool, projectID uuid.UUID, staleAfter time.Duration) (int64, error)

ResetStaleReservations returns evaluations whose owner is gone to the queue. The reset deliberately re-opens the budget slot; attempts is preserved, so a unit that poisons the judge still terminates at MaxModelAttempts.

Types

type EnqueueCursor 

type EnqueueCursor struct {
	CreatedAt time.Time
	ID        uuid.UUID
}

EnqueueCursor is a position in a project's chats, which the enqueue walks oldest-first on the immutable (created_at, id) key. The zero value starts at the head. A holder stores the cursor EnqueuePage returned and hands it back on the next call, which is how a walk that spans more pages than one call may scan is resumed across process restarts and a coordinator's own retries.

type EnqueuePageResult 

type EnqueuePageResult struct {
	// Scanned is the number of candidate chats the page read. It never exceeds
	// the page size.
	Scanned int
	// NextCursor is where the next page starts. It is strictly past the cursor
	// the page was given whenever the page read anything, and equal to it when
	// the queue was already empty. Chaining it is what carries a walk across
	// calls.
	NextCursor EnqueueCursor
	// Exhausted reports that the page reached the end of the candidate set — it
	// read fewer chats than it asked for. A caller resumes from NextCursor when
	// it is false and stops when it is true.
	Exhausted bool
}

EnqueuePageResult reports what one page did for a project.

func EnqueuePage 

func EnqueuePage(ctx context.Context, db *pgxpool.Pool, judges *Judges, projectID uuid.UUID, cursor EnqueueCursor, pageSize int32) (EnqueuePageResult, error)

EnqueuePage turns one bounded page of a project's chats into pending evaluations, one per (chat, judge) unit for every judge the organization has enabled.

This is the durable primitive the pipeline is built on: a call reads at most pageSize chats in one short pass and returns the cursor it stopped at, so a coordinator — a Temporal workflow persisting NextCursor between activities — decides how far a walk goes. Pass a zero EnqueueCursor to start at the head. The walk is bounded to EnqueueLookback, and the cursor key (created_at, id) is immutable, so a restarted walk covers exactly the chats a continuous one would have.

Quiet is not checked here: the insert stores the chat's latest message time as observed_at, and the reservation's candidate read applies the inactivity window live. An organization with no enabled judge gets no queue built for it at all, rather than a backlog of pending rows nothing will ever reserve.

type Evaluation 

type Evaluation struct {
	ID             uuid.UUID
	OrganizationID string
	ProjectID      uuid.UUID
	ChatID         uuid.UUID
	Judge          string
	ObservedAt     time.Time
	State          string
	// ReservedOn is the UTC day the evaluation spent its budget slot on, zero
	// while the row has never been reserved.
	ReservedOn time.Time
	Attempts   int32
}

Evaluation is a queued scoring unit. It carries no verdict: score and detail live only in ClickHouse, PostgreSQL holds pipeline state.

func LoadReserved 

func LoadReserved(ctx context.Context, db *pgxpool.Pool, projectID uuid.UUID, batchSize int32) ([]Evaluation, error)

LoadReserved claims reserved evaluations for processing, recent-first.

The claim is the crash-recovery path: a batch that Reserve has just handed out is processed from its own return value, and this only picks up rows whose previous owner is gone. Ownership is soft and leased — the updated_at bump IS the claim — so a concurrent or immediately repeated claim selects nothing and the model call that follows never has to hold a transaction open.

func NewEvaluation 

func NewEvaluation(row repo.ChatAnalysisEvaluation) Evaluation

NewEvaluation projects a stored row onto the domain type.

type Judge 

type Judge interface {
	// Name is the judge's stable identifier. It keys queue rows, settings rows
	// and score rows, so changing it orphans all three.
	Name() string
	Judge(ctx context.Context, in JudgeInput) (JudgeResult, error)
}

Judge scores one finished chat session. Implementations must return errors wrapping ErrModelFailure or ErrRetryable so the publisher can tell an answer it should charge the model for from one it should simply retry; the CallStructured helper classifies completion-call failures that way already.

type JudgeInput 

type JudgeInput struct {
	EvaluationID uuid.UUID
	OrgID        string
	ProjectID    string
	ChatID       uuid.UUID
	AuthorID     string
	Transcript   efficacy.Transcript
}

JudgeInput is one scoring unit: a finished session's rendered transcript and the identifiers a verdict row needs. The transcript rendering is shared with the skill efficacy judge, so every session judge sees the same prompt-injection-hardened shape.

type JudgeResult 

type JudgeResult struct {
	Verdict       Verdict
	Model         string
	PromptVersion string
}

JudgeResult carries the verdict plus the attribution the score row needs. Cost and token counts are deliberately absent: the completion client already bills and records them against the chat-analysis usage source.

type Judges 

type Judges struct {
	// contains filtered or unexported fields
}

Judges is the immutable judge roster the pipeline runs. Order is the registration order and has no behavioural weight; identity is the name.

func NewJudges 

func NewJudges(judges ...Judge) (*Judges, error)

NewJudges validates and assembles the roster. Every judge needs a unique, non-empty, lowercase name because the name is a persistence key.

func (*Judges) Get 

func (j *Judges) Get(name string) (Judge, bool)

Get resolves a judge by name.

func (*Judges) Names 

func (j *Judges) Names() []string

Names lists the roster's judge names in registration order.

type PendingCursor 

type PendingCursor struct {
	ObservedAt time.Time
	ID         uuid.UUID
}

PendingCursor is a position in a project's pending evaluations, which are walked recent-first on the unique (observed_at, id) key. The zero value starts at the head of the queue, and a reservation returns it again once its walk has reached the end.

type PendingWorkProject 

type PendingWorkProject struct {
	ProjectID uuid.UUID `json:"project_id"`
	HasStale  bool      `json:"has_stale"`
}

PendingWorkProject is a project the sweep has to visit, and whether it holds a reservation to recover.

func PendingWorkProjects 

func PendingWorkProjects(ctx context.Context, db *pgxpool.Pool, after uuid.UUID, staleAfter time.Duration, pageLimit int32) ([]PendingWorkProject, error)

PendingWorkProjects returns the next page of projects that hold analysis work the pipeline has not finished, ordered by project id and starting strictly after the given one. The zero uuid starts at the head of the estate.

Two things count as unfinished work: a live pending evaluation, and a reservation whose owner has been gone for staleAfter. The second source is what makes this a recovery pass rather than a discovery one — a project whose only work is a crashed reservation would otherwise never be visited again. Sessions no signal ever enqueued are not discovered here: enqueue coverage comes from the chat-writer observer, which wakes the coordinator on every durable message write.

A page shorter than pageLimit is the end of the estate.

type PublishResult 

type PublishResult struct {
	// Loaded is the number of still-reserved evaluations the batch resolved.
	Loaded int
	// AlreadyPublished is how many of those the existence guard found in
	// ClickHouse, so they were marked scored without being judged again.
	AlreadyPublished int
	// Scored is how many evaluations ended the pass in state scored.
	Scored int
	// ModelFailures is how many took a non-terminal model failure and stayed
	// reserved with an incremented attempt count.
	ModelFailures int
	// Failed is how many terminated, either after exhausting MaxModelAttempts or
	// immediately because row validation proved a retry cannot succeed.
	Failed int
	// Retryable is how many hit an infrastructure failure that still needs
	// another pass.
	Retryable int
}

PublishResult reports what one publication pass did with the reserved evaluations it was handed.

type Publisher 

type Publisher struct {
	// contains filtered or unexported fields
}

Publisher judges reserved evaluations and publishes their verdicts.

func NewPublisher 

func NewPublisher(logger *slog.Logger, tracerProvider trace.TracerProvider, db *pgxpool.Pool, scores ScoreSink, events ScoreEventSink, judges *Judges) *Publisher

NewPublisher constructs a Publisher over the given judge roster. events may be nil, which disables work-units score event emission (scores still publish).

func (*Publisher) Publish 

func (p *Publisher) Publish(ctx context.Context, projectID uuid.UUID, ids []uuid.UUID, heartbeat func()) (PublishResult, error)

Publish judges the given reserved evaluations and writes their verdicts.

Publication order per evaluation is existence guard → judge → synchronous insert → mark scored, and the guard runs for the WHOLE batch before any judge call: a retry that follows a crash between insert and mark must not pay for inference a second time. The score id is the evaluation id, so every physical retry has the same logical event identity and analytical reads collapse it.

A model failure charges the evaluation an attempt and the batch continues; the third one terminates that evaluation as failed and never writes a score. A deterministic row-validation failure — including a judge name the roster no longer runs — terminates immediately. An infrastructure failure changes no state and charges no attempt, with the one exception of a sink failure after the judge has answered, which is charged as well so a broken sink cannot buy the same inference forever.

heartbeat, when given, is called once before each evaluation. It is what lets the caller's own lease on the batch stay live across a long pass and, on the Temporal path, what delivers a cancellation.

type ScoreEventSink 

type ScoreEventSink interface {
	LogBulk(ctx context.Context, params []telemetry.LogParams) error
}

ScoreEventSink emits the synthetic per-session telemetry events derived from published work-units verdicts — the rows attribute_metrics_summaries_mv folds into the work-units efficiency measures. Satisfied by *telemetry.Logger; nil disables emission.

type ScoreSink 

type ScoreSink interface {
	ListExistingChatAnalysisScoreIDs(ctx context.Context, arg telemetryrepo.ListExistingChatAnalysisScoreIDsParams) ([]string, error)
	InsertChatAnalysisScores(ctx context.Context, rows []telemetryrepo.ChatAnalysisScore) error
	GetChatSessionFactsByChatIDs(ctx context.Context, arg telemetryrepo.GetChatSessionFactsByChatIDsParams) (map[string]telemetryrepo.ChatSessionFacts, error)
}

ScoreSink is the ClickHouse side of publication: the existence guard, the synchronous insert, and the session-facts read that decorates score events. Satisfied by *telemetryrepo.Queries.

type ScorelessJudge 

type ScorelessJudge interface {
	SkipScoreSink() bool
}

ScorelessJudge performs its durable work and marks its evaluation scored atomically inside Judge, so it does not publish a verdict to the shared ClickHouse score sink.

type Settings 

type Settings struct {
	OrganizationID string
	// JudgeDailyCaps holds one entry per enabled judge. A cap of 0 disables the
	// judge as surely as enabled=false does — the reservation can never admit
	// its units.
	JudgeDailyCaps map[string]int32
}

Settings are the effective per-organization budgets: the daily cap for each enabled judge. A judge absent from the map is off for the organization.

type Signaler 

type Signaler interface {
	Signal(ctx context.Context, projectID uuid.UUID) error
}

Signaler wakes a project's chat analysis coordinator. Implementations are expected to be idempotent: every producer signals on each durable write, so the same project is woken many times over one session and a wake carries no payload beyond the project it names.

Declared here rather than imported from the workflow layer so the producers depend on the analysis domain and never on the background package that runs the coordinator.

type StructuredCall 

type StructuredCall struct {
	Model        string
	SystemPrompt string
	Prompt       string
	// SchemaName names the response schema for the provider.
	SchemaName string
	// Schema is the structured-output JSON schema. Do not use minimum/maximum or
	// maxLength keywords: Anthropic routes (via Amazon Bedrock) reject those with
	// a 400. Enforce bounds in the judge's normalization instead.
	Schema map[string]any
	// Timeout bounds the call; judges reading whole transcripts should allow the
	// same 60 seconds the efficacy judge does.
	Timeout time.Duration
}

StructuredCall is one structured-output judge completion: the model, the framing, and the response contract. Judges describe their call with this and hand it to CallStructured, so every judge shares the same rate limiting, timeout handling and failure classification.

type Verdict 

type Verdict struct {
	Score  float64
	Detail json.RawMessage
}

Verdict is a judge's normalized answer. Score is the headline metric whose meaning the judge defines (work units delivered, resolution likelihood, …) and Detail is the full structured verdict as JSON in the judge's own shape. Both land verbatim in the chat_analysis_scores sink, keyed by the judge's name.

type WorkUnitsJudge 

type WorkUnitsJudge struct {
	// contains filtered or unexported fields
}

WorkUnitsJudge estimates how many meaningful work units a session delivered.

func NewWorkUnitsJudge 

func NewWorkUnitsJudge(logger *slog.Logger, tracerProvider trace.TracerProvider, client openrouter.CompletionClient, limiter *ratelimit.Limiter) *WorkUnitsJudge

NewWorkUnitsJudge constructs the judge. Pass the limiter from openrouter.NewJudgeRateLimiter so its calls draw from the same bucket as every other judge spending the org's key on the same model.

func (*WorkUnitsJudge) Judge 

func (j *WorkUnitsJudge) Judge(ctx context.Context, in JudgeInput) (JudgeResult, error)

Judge scores one session. Errors wrap ErrModelFailure or ErrRetryable so the publisher can tell an answer it should charge the model for from one it should simply retry.

func (*WorkUnitsJudge) Name 

func (j *WorkUnitsJudge) Name() string

type WorkUnitsTask 

type WorkUnitsTask struct {
	ID              int     `json:"id"`
	Request         string  `json:"request"`
	Band            string  `json:"band"`
	BaseUnits       float64 `json:"base_units"`
	Modifier        float64 `json:"modifier"`
	Completion      float64 `json:"completion"`
	Units           float64 `json:"units"`
	NearestExemplar string  `json:"nearest_exemplar"`
	Rationale       string  `json:"rationale"`
}

WorkUnitsTask is one user-requested task the judge identified and scored.

type WorkUnitsVerdict 

type WorkUnitsVerdict struct {
	Tasks        []WorkUnitsTask `json:"tasks"`
	SessionUnits float64         `json:"session_units"`
	Flags        []string        `json:"flags"`
}

WorkUnitsVerdict is the judge's structured answer: per-task work units plus the session total. SessionUnits is the verdict's headline score.

func ParseWorkUnitsVerdict 

func ParseWorkUnitsVerdict(raw string) (WorkUnitsVerdict, error)

ParseWorkUnitsVerdict decodes the judge's raw structured output and normalizes it. Unparseable output is a model failure: the model returned something outside the contract it was given, and a retry can produce a different answer.

func (WorkUnitsVerdict) Normalize 

func (v WorkUnitsVerdict) Normalize() (WorkUnitsVerdict, error)

Normalize forces the verdict inside the prompt's own contract: per-task units recomputed from the factors the judge reported — round(base_units × modifier × completion), clamped to [-30, 100] — so the stored score can never disagree with the arithmetic behind it, the session total recomputed as the tasks' sum, flags restricted to the allowed set, and free text capped. The structured-output schema already requires every field, so the shape checks here are defense in depth against a model that returned empty or null anyway. A non-finite number, a modifier or completion outside its snap set, or a base outside the task's band is unfixable — repairing any of them would invent a score the judge never gave — so each is reported as a model failure.

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.