runs

package
v1.0.0-rc.5 Latest Latest
Warning

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

 Go to latest Published: Nov 8, 2026 License: Apache-2.0 Imports: 24 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultQueryLimit = 100
	MaxQueryLimit     = 1000
	MaxLineBytes      = 16 << 10
)
View Source 
const (
	OmittedMalformed = "malformed"
	OmittedOversized = "oversized"
)

Fixed omission indicators. These are the only values written when an identifier cannot be stored; never excerpts of the rejected value.

View Source 
const (
	DispositionCompleted      = "completed"
	DispositionToolError      = "tool_error"
	DispositionDenied         = "denied"
	DispositionRoutingFailed  = "routing_failed"
	DispositionTransportError = "transport_error"
	DispositionCancelled      = "cancelled"
	DispositionTimeout        = "timeout"
	DispositionInputRequired  = "input_required"
	DispositionRetryRejected  = "retry_rejected"
)

Disposition is the final outcome of one returning dispatch attempt.

View Source 
const (
	StageGroup      = "group"
	StageScope      = "scope"
	StageGate       = "gate"
	StagePin        = "pin"
	StageRouting    = "routing"
	StageColdStart  = "cold_start"
	StageMRTR       = "mrtr"
	StageDownstream = "downstream"
	StageCodeMode   = "code_mode"
	StageDiscovery  = "discovery"
)

Stage names the decision point that produced the disposition.

View Source 
const (
	ReasonOK               = "ok"
	ReasonGroupMembership  = "group_membership"
	ReasonClientScope      = "client_scope"
	ReasonGateDenied       = "gate_denied"
	ReasonPolicyBlocked    = "policy_blocked"
	ReasonSchemaPin        = "schema_pin"
	ReasonUnknownTool      = "unknown_tool"
	ReasonNoReplica        = "no_replica"
	ReasonColdStart        = "cold_start"
	ReasonMRTRMismatch     = "mrtr_mismatch"
	ReasonTransportError   = "transport_error"
	ReasonToolError        = "tool_error"
	ReasonContextCanceled  = "context_canceled"
	ReasonDeadlineExceeded = "deadline_exceeded"
	ReasonInputRequired    = "input_required"
	ReasonCodeMode         = "code_mode"
	ReasonSearchTools      = "search_tools"
	ReasonCallTool         = "call_tool"
)

Reason codes set at the actual decision point. They are not parsed from error strings or inferred from a nil Go error.

View Source 
const (
	DropQueueFull     = "queue_full"
	DropTooLarge      = "record_too_large"
	DropClosed        = "writer_closed"
	DropGeneration    = "generation_mismatch"
	DropCapExhausted  = "cap_exhausted"
	DropWriteError    = "write_error"
	DropSyncError     = "sync_error"
	DropShutdownLimit = "shutdown_limit"
	DropOpenError     = "open_error"
	DropDiscarded     = "discarded"
)

Drop reasons for the independent recorder status. These never enter the JSONL record stream.

View Source 
const (
	HealthOK       = "ok"
	HealthDegraded = "degraded"
	HealthStopped  = "stopped"
	HealthUnknown  = "unknown"
)

Writer health values reported independently of the destination file.

View Source 
const (
	WarnPartialTail       = "partial_tail"
	WarnMalformed         = "malformed"
	WarnUnsupportedSchema = "unsupported_schema"
	WarnTruncatedQuery    = "truncated"
	WarnUnreadable        = "unreadable"
)
View Source 
const (
	DefaultQueueSize      = 1024
	DefaultMaxRecordBytes = 8192
	DefaultSyncInterval   = 2 * time.Second
	DefaultShutdownDrain  = 2 * time.Second
	DefaultMaxBytes       = 100 << 20
	DefaultMaxAge         = 7 * 24 * time.Hour
)
View Source 
const LossUnknown = "unknown"

LossUnknown is the post-restart / offline historical-loss value.

View Source 
const MaxIdentifierBytes = 128

MaxIdentifierBytes is the maximum stored length of a target or label selector. Oversized values are omitted with a fixed indicator.

View Source 
const SchemaVersion = 1

SchemaVersion is the current persisted run-record schema.

Variables

View Source 
var ErrActiveWriter = errors.New("runs wipe refused: an active recorder owns this stack; use the live daemon")

ErrActiveWriter means an offline wipe refused to unlink behind a live writer.

View Source 
var ErrPerServerRuns = errors.New("runs history is stack-wide; omit --server and wipe the stack, or use gridctl runs wipe")

ErrPerServerRuns is returned when a caller asks to wipe runs for one server.

Functions

func ActivePath 

func ActivePath(stackName string) (string, error)

ActivePath returns the active JSONL path for a stack.

func BoundIdentifier 

func BoundIdentifier(value string) (stored, omitted string)

BoundIdentifier returns a stored selector and an omission indicator. Valid values are returned as-is. Malformed or oversized values yield an empty selector and a fixed indicator; the rejected text is discarded.

func Dir 

func Dir(stackName string) (string, error)

Dir returns the stack-level runs directory (~/.gridctl/runs/<stack>).

func EncodeCursor 

func EncodeCursor(c Cursor) string

EncodeCursor serializes a cursor as unpadded URL-safe base64 JSON.

func ParentAttemptID 

func ParentAttemptID(ctx context.Context) string

func RegisterActive 

func RegisterActive(stackName string, rec *Recorder)

RegisterActive records the live recorder for a stack so wipe can coordinate instead of unlinking behind an open writer.

func UnregisterActive 

func UnregisterActive(stackName string)

UnregisterActive drops the live recorder for a stack.

func ValidGeneratedID 

func ValidGeneratedID(id string) bool

ValidGeneratedID reports whether id matches the recorder's generated identifier syntax (32 lowercase hex characters).

func WipeStack 

func WipeStack(ctx context.Context, stackName string) error

WipeStack deletes owned run files for a stack. When a live recorder is registered it performs a coordinated barrier. Offline wipe refuses if a daemon is running for the stack unless coordinateOffline is true and no in-process recorder exists — then it still refuses rather than unlink behind a foreign writer.

Types

type Attempt 

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

Attempt is the per-invocation builder. Metadata is immutable once Finish materializes the record.

func (*Attempt) Context 

func (a *Attempt) Context() context.Context

Context returns the attempt context, including parent/root IDs for nested calls.

func (*Attempt) Finish 

func (a *Attempt) Finish()

Finish materializes one immutable record and enqueues it. Safe to call once.

func (*Attempt) SetDownstreamDuration 

func (a *Attempt) SetDownstreamDuration(d time.Duration)

SetDownstreamDuration labels the downstream timer separately from total duration.

func (*Attempt) SetLabels 

func (a *Attempt) SetLabels(client, access string)

SetLabels stores attribution labels. Outside OAuth mode they are caller-declared and are not authenticated principals. In OAuth mode the access label is the verified identity claim. Do not log either value at INFO or above; DEBUG structured fields only.

func (*Attempt) SetOutcome 

func (a *Attempt) SetOutcome(disposition, stage, reason string)

SetOutcome records the disposition at a real decision point.

func (*Attempt) SetPreviousAttemptID 

func (a *Attempt) SetPreviousAttemptID(id string)

SetPreviousAttemptID links a resumed round when trusted internal correlation exists.

func (*Attempt) SetResolved 

func (a *Attempt) SetResolved(server, tool string, replicaID int)

SetResolved stores bounded resolved identifiers. Unresolved targets stay absent.

func (*Attempt) SetTraceID 

func (a *Attempt) SetTraceID(id string)

SetTraceID stores a genuine trace identifier. Empty values are ignored.

type Config 

type Config struct {
	Enabled       bool
	OmitLabels    bool
	StackName     string
	MaxBytes      int64
	MaxAge        time.Duration
	QueueSize     int
	SyncInterval  time.Duration
	ShutdownDrain time.Duration
	SegmentSize   int64
	Dir           string
}

Config is the runtime recorder configuration.

type Cursor 

type Cursor struct {
	WipeEpoch   uint64    `json:"wipe_epoch"`
	SourceToken string    `json:"source_token,omitempty"`
	ReturnedAt  time.Time `json:"returned_at"`
	Sequence    uint64    `json:"sequence"`
	AttemptID   string    `json:"attempt_id"`
}

Cursor identifies a pagination position. WipeEpoch and SourceToken must match the current writer generation and on-disk file set or the cursor is invalid.

func DecodeCursor 

func DecodeCursor(s string) (Cursor, error)

DecodeCursor parses a cursor from unpadded URL-safe base64 JSON or raw JSON.

type Filter 

type Filter struct {
	Since             time.Time `json:"since,omitempty"`
	Until             time.Time `json:"until,omitempty"`
	RequestedName     string    `json:"requested,omitempty"`
	ResolvedServer    string    `json:"server,omitempty"`
	ResolvedTool      string    `json:"tool,omitempty"`
	Disposition       string    `json:"disposition,omitempty"`
	ClientLabel       string    `json:"client,omitempty"`
	AccessLabel       string    `json:"access,omitempty"`
	AttemptID         string    `json:"attempt,omitempty"`
	ParentAttemptID   string    `json:"parent,omitempty"`
	RootAttemptID     string    `json:"root,omitempty"`
	PreviousAttemptID string    `json:"previous,omitempty"`
	TraceID           string    `json:"trace,omitempty"`
}

Filter selects persisted records. Empty fields are unconstrained.

type Inventory 

type Inventory struct {
	Stack      string    `json:"stack"`
	Signal     string    `json:"signal"`
	Path       string    `json:"path"`
	SizeBytes  int64     `json:"sizeBytes"`
	OldestTime time.Time `json:"oldestTime"`
	NewestTime time.Time `json:"newestTime"`
	FileCount  int       `json:"fileCount"`
}

Inventory describes the on-disk footprint of one stack's run records.

func StackInventory 

func StackInventory(stackName string) (Inventory, error)

StackInventory walks owned run files for a stack. A missing directory returns a zero Inventory without error.

type QueryEnvelope 

type QueryEnvelope struct {
	SchemaVersion int         `json:"schema_version"`
	Source        QuerySource `json:"source"`
	Filters       Filter      `json:"filters"`
	Records       []Record    `json:"records"`
	Warnings      []Warning   `json:"warnings"`
	NextCursor    *Cursor     `json:"next_cursor,omitempty"`
	Partial       bool        `json:"partial"`
	WipeEpoch     uint64      `json:"wipe_epoch"`
}

QueryEnvelope is the shared live/offline/export query contract.

func EnvelopeFromResult 

func EnvelopeFromResult(res QueryResult, source QuerySource, filter Filter) QueryEnvelope

EnvelopeFromResult builds the shared query/export envelope.

type QueryResult 

type QueryResult struct {
	Records    []Record  `json:"records"`
	Warnings   []Warning `json:"warnings"`
	NextCursor *Cursor   `json:"next_cursor,omitempty"`
	Partial    bool      `json:"partial"`
	WipeEpoch  uint64    `json:"wipe_epoch"`
}

QueryResult is a bounded page of records plus explicit warnings.

func Query 

func Query(ctx context.Context, dir string, filter Filter, limit int, cursor *Cursor, wipeEpoch uint64) (QueryResult, error)

Query reads owned JSONL files newest-first and returns a bounded page. Partial tails and unknown schemas are reported as warnings without echoing their contents. ctx cancellation is honored and always marked partial.

func QueryPath 

func QueryPath(ctx context.Context, path string, filter Filter, limit int, cursor *Cursor) (QueryResult, error)

QueryPath is an explicit offline file or directory query. It never consults a live recorder.

type QuerySource 

type QuerySource struct {
	Kind  string `json:"kind"`
	Stack string `json:"stack,omitempty"`
	Path  string `json:"path,omitempty"`
}

QuerySource names the stack or path a query read.

type Record 

type Record struct {
	SchemaVersion      int       `json:"schema_version"`
	RecorderInstanceID string    `json:"recorder_instance_id"`
	Sequence           uint64    `json:"sequence"`
	AttemptID          string    `json:"attempt_id"`
	ParentAttemptID    string    `json:"parent_attempt_id,omitempty"`
	RootAttemptID      string    `json:"root_attempt_id,omitempty"`
	PreviousAttemptID  string    `json:"previous_attempt_id,omitempty"`
	StartedAt          time.Time `json:"started_at"`
	ReturnedAt         time.Time `json:"returned_at"`
	DurationMS         int64     `json:"duration_ms"`
	DownstreamMS       int64     `json:"downstream_duration_ms,omitempty"`
	RequestedName      string    `json:"requested_name,omitempty"`
	RequestedOmitted   string    `json:"requested_name_omitted,omitempty"`
	ResolvedServer     string    `json:"resolved_server,omitempty"`
	ResolvedServerOmit string    `json:"resolved_server_omitted,omitempty"`
	ResolvedTool       string    `json:"resolved_tool,omitempty"`
	ResolvedToolOmit   string    `json:"resolved_tool_omitted,omitempty"`
	Disposition        string    `json:"disposition"`
	Stage              string    `json:"stage"`
	Reason             string    `json:"reason"`
	ReplicaID          *int      `json:"replica_id,omitempty"`
	TraceID            string    `json:"trace_id,omitempty"`
	ClientLabel        string    `json:"client_label,omitempty"`
	AccessLabel        string    `json:"access_label,omitempty"`
}

Record is the schema-versioned metadata-only final-disposition record. It is the allowlist: generated IDs, times/duration, bounded target identifiers, disposition/stage/reason, optional replica and genuine trace correlation, and optional caller-declared or OAuth principal labels. Argument and result values, hashes, code, raw errors, tokens, headers, URLs, host paths, and approval claims are excluded by construction.

type Recorder 

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

Recorder is one stack-owned, single-writer run recorder.

func NewRecorder 

func NewRecorder(cfg Config, logger *slog.Logger) (*Recorder, error)

NewRecorder starts a single writer goroutine. Enablement is requested independently of destination health: an unopenable destination yields a degraded recorder rather than a nil one.

func (*Recorder) Apply 

func (r *Recorder) Apply(cfg Config) error

Apply updates enablement, labels, and retention without racing in-flight attempts. In-flight attempts stay bound to the generation they began with.

func (*Recorder) Begin 

func (r *Recorder) Begin(ctx context.Context, requestedName string) *Attempt

Begin starts one attempt bound to the recorder generation active at entry.

func (*Recorder) Close 

func (r *Recorder) Close()

Close drains the queue up to the shutdown bound and closes the writer.

func (*Recorder) DirPath 

func (r *Recorder) DirPath() string

DirPath returns the configured storage directory.

func (*Recorder) Disable 

func (r *Recorder) Disable()

Disable recording without deleting retained history.

func (*Recorder) Status 

func (r *Recorder) Status() Status

Status returns independent recorder health.

func (*Recorder) Wipe 

func (r *Recorder) Wipe(ctx context.Context) error

Wipe is a writer-coordinated barrier. Pre-wipe queued events are dropped by epoch mismatch. Partial deletion is returned as an error.

func (*Recorder) WipeEpoch 

func (r *Recorder) WipeEpoch() uint64

WipeEpoch returns the current wipe generation.

type Status 

type Status struct {
	Enabled              bool              `json:"enabled"`
	Effective            bool              `json:"effective"`
	WriterHealth         string            `json:"writer_health"`
	QueueDepth           int               `json:"queue_depth"`
	QueueCapacity        int               `json:"queue_capacity"`
	Drops                map[string]uint64 `json:"drops"`
	Failures             map[string]uint64 `json:"failures"`
	LastSuccessfulAppend *time.Time        `json:"last_successful_append,omitempty"`
	LastSuccessfulSync   *time.Time        `json:"last_successful_sync,omitempty"`
	ProcessStartedAt     time.Time         `json:"process_started_at"`
	HistoricalLoss       string            `json:"historical_loss"`
	WipeEpoch            uint64            `json:"wipe_epoch"`
	Generation           uint64            `json:"generation"`
	RecorderInstanceID   string            `json:"recorder_instance_id"`
	LogicalBytes         int64             `json:"logical_bytes"`
	Synced               bool              `json:"synced"`
	Known                bool              `json:"known"`
	OmitLabels           bool              `json:"omit_labels"`
	RetentionMaxBytes    int64             `json:"retention_max_bytes"`
	RetentionMaxAgeDays  int               `json:"retention_max_age_days"`
}

Status is value-free recorder health. It is independent of the JSONL destination so a broken writer can still report failure.

type Warning 

type Warning struct {
	Code    string `json:"code"`
	Count   int    `json:"count"`
	Message string `json:"message"`
}

Warning is a value-free reader diagnostic. Malformed contents are never included.

type Writer 

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

Writer appends schema-versioned JSONL records with rotation, a total logical-byte budget, periodic sync, and coordinated reopen for wipe.

func (*Writer) Append 

func (w *Writer) Append(ctx context.Context, payload []byte) error

func (*Writer) Close 

func (w *Writer) Close() error

func (*Writer) Prune 

func (w *Writer) Prune(now time.Time) error

func (*Writer) Sync 

func (w *Writer) Sync(ctx context.Context) error

type WriterConfig 

type WriterConfig struct {
	Dir         string
	MaxBytes    int64
	MaxAge      time.Duration
	SegmentSize int64
}

WriterConfig bounds on-disk retention. MaxBytes is the total logical record-byte budget across the active file and closed segments.

Source Files

View all Source files

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.