trajectory

package
v0.0.4 Latest Latest
Warning

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

 Go to latest Published: Aug 29, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Overview

Package trajectory implements the append-only session ledger (M9–M12).

Owns: event catalog, in-memory store, async JSONL persist, live Hub fan-out, export/list/search, import append, policy replay, log fork. Must not: hook wire decode (hookedge), config compile (config), or own route matching. ReplayPolicy accepts an injected dispatch.Invoker for offline policy dry-run only (does not compile routes or match on the live hot path).

Invariants:

  • No disk I/O on the sync Invoke path; enqueue only.
  • Contiguous seq per session; events immutable after append.
  • Opt-in via config.Trajectory.Enabled (default off).
  • schema_version frozen at SchemaVersion for v0.0.2 contract.

Entry: Recorder.Record, Hub.Publish, ListSessions, Export, ExportToFile, Search, AppendImported, ReplayPolicy, ReplayPolicyFromConfig, ForkSession, ResolveSessionKey, ResolveSessionKeyID, EventFromSessionEvent, EventToSessionEvent. Import orchestration and L2 importer status: importer.ImportSession, importer.ProviderImporterStatus. See DESIGN.md §1.5 (async_side), §14.

Index

Constants

View Source 
const (
	TypeSessionOpen        = "session/open"
	TypeHookInvoked        = "hook/invoked"
	TypeHookDecided        = "hook/decided"
	TypeAsyncDispatched    = "async/dispatched"
	TypeAsyncDropped       = "async/dropped"
	TypeTranscriptMessage  = "transcript/message"
	TypeTranscriptThinking = "transcript/thinking"
	TypeSessionFork        = "session/fork"
	TypeSessionEndSeed     = "session/end-seed"
)
View Source 
const (
	SourceSystem     = "system"
	SourceHook       = "hook"
	SourceDecision   = "decision"
	SourceTranscript = "transcript"
)
View Source 
const SchemaVersion uint32 = 1

SchemaVersion is the frozen trajectory event contract version (v0.0.2).

Variables

View Source 
var (
	// ErrSessionsDirUnavailable means the default sessions state directory is unavailable.
	ErrSessionsDirUnavailable = errors.New("sessions dir unavailable")
	// ErrSessionNotFound means no session JSONL exists for the provider and session id.
	ErrSessionNotFound = errors.New("session not found")
	// ErrNewSessionIDRequired means fork requires a non-empty destination session id.
	ErrNewSessionIDRequired = errors.New("new session id required")
	// ErrSourceSessionEmpty means the fork source ledger has no events.
	ErrSourceSessionEmpty = errors.New("source session is empty")
	// ErrSessionAlreadyExists means the fork destination session already exists.
	ErrSessionAlreadyExists = errors.New("session already exists")
	// ErrReplayNoRaw means no hook/invoked events have stored Raw payloads.
	ErrReplayNoRaw = errors.New("policy replay requires stored raw payloads at record time")
	// ErrReplayNoEvents means the session has no hook/invoked events to replay.
	ErrReplayNoEvents = errors.New("no hook/invoked events to replay")
	// ErrReplaySeqNotFound means no hook/invoked event with Raw exists for the requested seq.
	ErrReplaySeqNotFound = errors.New("no hook/invoked event with raw payload for seq")
	// ErrNilConfigSnap means ReplayPolicy was called without a config snapshot.
	ErrNilConfigSnap = errors.New("nil config snapshot")
	// ErrNilEngine means ReplayPolicy was called without a dispatch engine.
	ErrNilEngine = errors.New("nil dispatch engine")
)

Functions

func AppendEvents 

func AppendEvents(root string, key SessionKey, events []Event) error

AppendEvents writes events to the session JSONL file immediately (offline import/fork).

func AppendImported 

func AppendImported(root string, key SessionKey, events []Event) error

AppendImported assigns contiguous seq after existing ledger events and persists.

func CanonicalProvider 

func CanonicalProvider(name string) string

CanonicalProvider normalizes CLI/provider ids for ledger keys and path filters.

func DefaultImportConfig 

func DefaultImportConfig() config.TrajectoryConfig

DefaultImportConfig returns conservative defaults for offline import.

func DefaultSessionsDir 

func DefaultSessionsDir() string

DefaultSessionsDir returns the ledger root ($XDG_STATE_HOME/agentd/sessions, else ~/.local/state/agentd/sessions when XDG_STATE_HOME is unset).

func EventToSessionEvent 

func EventToSessionEvent(ev Event) *agentdv1.SessionEvent

EventToSessionEvent maps a ledger Event to agentd.v1.SessionEvent.

func Export 

func Export(w io.Writer, root, providerFilter, sessionID string) error

Export writes session JSONL to w. When sessionID is empty, concatenates all sessions for providerFilter (or all providers when filter is empty).

func ExportToFile 

func ExportToFile(outPath, root, providerFilter, sessionID string) error

ExportToFile writes export output to outPath.

func FindSessionPath 

func FindSessionPath(root, provider, sessionID string) (string, error)

FindSessionPath resolves provider + session id to a JSONL path under root.

func ImportSidecarPath 

func ImportSidecarPath(sessionsRoot, provider, sessionID string) string

ImportSidecarPath returns the import checkpoint path beside a session ledger.

func InvocationModeString 

func InvocationModeString(m agentdv1.InvocationMode) string

InvocationModeString maps proto enum to ledger strings.

func PrepareRaw 

func PrepareRaw(raw []byte, cfg config.TrajectoryConfig) json.RawMessage

PrepareRaw returns raw payload bytes for storage per trajectory config.

func PrepareTranscriptText 

func PrepareTranscriptText(text string, cfg config.TrajectoryConfig) string

PrepareTranscriptText applies redaction and max size to imported transcript text.

func SaveImportCheckpoint 

func SaveImportCheckpoint(sidecarPath string, cp ImportCheckpoint) error

SaveImportCheckpoint writes the sidecar atomically.

func SessionFileName 

func SessionFileName(sessionID string) string

SessionFileName returns the on-disk basename for a session id.

func SessionFilePath 

func SessionFilePath(root string, key SessionKey) string

SessionFilePath returns the JSONL path for one session key.

Types

type AsyncDispatchedData 

type AsyncDispatchedData struct {
	Count uint32 `json:"count"`
}

AsyncDispatchedData is the payload for async/dispatched.

type AsyncDroppedData 

type AsyncDroppedData struct {
	Reason string `json:"reason"`
}

AsyncDroppedData is the payload for async/dropped (trajectory queue overflow).

type Event 

type Event struct {
	SchemaVersion  uint32          `json:"schema_version,omitempty"`
	Seq            uint64          `json:"seq"`
	Type           string          `json:"type"`
	Source         string          `json:"source"`
	TS             time.Time       `json:"ts"`
	Provider       string          `json:"provider"`
	InvocationMode string          `json:"invocation_mode,omitempty"`
	SessionID      string          `json:"session_id"`
	ProjectRoot    string          `json:"project_root,omitempty"`
	CWD            string          `json:"cwd,omitempty"`
	Data           json.RawMessage `json:"data,omitempty"`
	Raw            json.RawMessage `json:"raw,omitempty"`
	Ignorable      bool            `json:"ignorable,omitempty"`
}

Event is one append-only ledger record (DESIGN §14.2).

func EventFromSessionEvent 

func EventFromSessionEvent(ev *agentdv1.SessionEvent) Event

EventFromSessionEvent maps agentd.v1.SessionEvent to a ledger Event.

func ReadEvents 

func ReadEvents(path string) ([]Event, error)

ReadEvents loads all events from a session JSONL file.

type ForkResult 

type ForkResult struct {
	Provider      string `json:"provider"`
	ParentSession string `json:"parent_session"`
	NewSessionID  string `json:"new_session_id"`
	BoundarySeq   uint64 `json:"boundary_seq"`
	Copied        int    `json:"copied"`
	Path          string `json:"path"`
}

ForkResult summarizes a successful log fork.

func ForkSession 

func ForkSession(root string, src SessionKey, newSessionID string, atSeq uint64) (ForkResult, error)

ForkSession copies events with seq <= atSeq (or all if atSeq == 0) into a new session ledger and appends session/fork + session/end-seed metadata. The source JSONL is never modified.

type HookDecidedData 

type HookDecidedData struct {
	Kind                 string `json:"kind"`
	Decision             string `json:"decision"`
	Reason               string `json:"reason,omitempty"`
	ConfigGeneration     uint64 `json:"config_generation"`
	ConfigFingerprint    string `json:"config_fingerprint"`
	AsyncDispatchedCount uint32 `json:"async_dispatched_count,omitempty"`
}

HookDecidedData is the payload for hook/decided.

type HookInvokedData 

type HookInvokedData struct {
	Kind      string `json:"kind"`
	ToolName  string `json:"tool_name,omitempty"`
	ToolUseID string `json:"tool_use_id,omitempty"`
	HasRoute  bool   `json:"has_route"`
}

HookInvokedData is the payload for hook/invoked.

type Hub 

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

Hub fans out post-commit ledger events to live subscribers.

func NewHub 

func NewHub(log *slog.Logger) *Hub

NewHub returns an empty subscriber registry.

func (*Hub) Close 

func (h *Hub) Close()

Close ends all subscriptions.

func (*Hub) Publish 

func (h *Hub) Publish(events []Event)

Publish delivers events to matching subscribers without blocking callers.

func (*Hub) Register 

func (h *Hub) Register(filter SubscribeFilter) (<-chan Event, func())

Register adds a filtered subscriber. The returned unregister func removes it.

type ImportCheckpoint 

type ImportCheckpoint struct {
	LastLineIndex int       `json:"last_line_index"`
	SourcePath    string    `json:"source_path"`
	SourceModTime time.Time `json:"source_mod_time"`
}

ImportCheckpoint tracks incremental import progress for one transcript source.

func LoadImportCheckpoint 

func LoadImportCheckpoint(sidecarPath string) (ImportCheckpoint, error)

LoadImportCheckpoint reads the sidecar for a session, or zero value if missing.

type Persister 

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

Persister writes JSONL lines asynchronously with debounced flush.

func NewPersister 

func NewPersister(root string, log *slog.Logger) *Persister

NewPersister returns a persister rooted at dir (typically DefaultSessionsDir()).

func (*Persister) Flush 

func (p *Persister) Flush(_ context.Context) error

Flush writes all pending events to disk.

func (*Persister) Schedule 

func (p *Persister) Schedule(key SessionKey, events []Event)

Schedule queues events for debounced disk append.

type Queue 

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

Queue is a bounded worker pool for trajectory append + persist.

func NewQueue 

func NewQueue(capacity int, store *Store, persist *Persister, hub *Hub, log *slog.Logger) *Queue

NewQueue starts workers for trajectory side effects.

func (*Queue) Close 

func (q *Queue) Close(timeout time.Duration)

Close drains pending jobs up to timeout then stops workers.

func (*Queue) Dropped 

func (q *Queue) Dropped() uint64

Dropped returns overflow drop count.

func (*Queue) Enqueue 

func (q *Queue) Enqueue(key SessionKey, events []Event) bool

Enqueue submits events for async append. Never blocks the caller.

type RecordInput 

type RecordInput struct {
	Provider       agentdv1.Provider
	InvocationMode agentdv1.InvocationMode
	CWD            string
	ProjectRoot    string
	RawPayload     []byte
	Result         dispatch.InvokeResult
	Snap           *config.Snapshot
}

RecordInput is one Invoke worth of trajectory data.

type Recorder 

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

Recorder enqueues trajectory events from HookService.Invoke (async side).

func NewRecorder 

func NewRecorder(sessionsDir string, capacity int, log *slog.Logger) *Recorder

NewRecorder wires store + persist + hub into a bounded queue.

func (*Recorder) Close 

func (r *Recorder) Close(timeout time.Duration)

Close drains workers, closes hub subscriptions, and flushes pending JSONL.

func (*Recorder) Hub 

func (r *Recorder) Hub() *Hub

Hub returns the live event fan-out registry.

func (*Recorder) Queue 

func (r *Recorder) Queue() *Queue

Queue returns the underlying queue (Status drop counter).

func (*Recorder) Record 

func (r *Recorder) Record(in RecordInput)

Record enqueues ledger events when trajectory is enabled in snap.

type ReplayHit 

type ReplayHit struct {
	Seq            uint64 `json:"seq"`
	Kind           string `json:"kind,omitempty"`
	StoredDecision string `json:"stored_decision,omitempty"`
	ReplayDecision string `json:"replay_decision,omitempty"`
	Match          bool   `json:"match"`
	Error          string `json:"error,omitempty"`
}

ReplayHit is one replayed hook/invoked event.

type ReplayOptions 

type ReplayOptions struct {
	SessionsRoot string
	Provider     string
	SessionID    string
	Seq          uint64 // 0 = all hook/invoked with Raw
	Snap         *config.Snapshot
	Engine       dispatch.Invoker
}

ReplayOptions configures an offline policy dry-run against stored Raw payloads.

type ReplayPolicyConfigOptions 

type ReplayPolicyConfigOptions struct {
	ConfigPath   string
	SessionsRoot string
	Provider     string
	SessionID    string
	Seq          uint64
}

ReplayPolicyConfigOptions configures offline policy replay with config load and engine setup.

type ReplayResult 

type ReplayResult struct {
	Provider  string      `json:"provider"`
	SessionID string      `json:"session_id"`
	Hits      []ReplayHit `json:"hits"`
}

ReplayResult is the full policy replay output.

func ReplayPolicy 

func ReplayPolicy(ctx context.Context, opts ReplayOptions) (ReplayResult, error)

ReplayPolicy re-Invokes stored Raw through Engine (offline; no live agent).

func ReplayPolicyFromConfig 

func ReplayPolicyFromConfig(ctx context.Context, opts ReplayPolicyConfigOptions) (ReplayResult, error)

ReplayPolicyFromConfig loads config, builds a dispatch engine, and runs ReplayPolicy.

type SearchHit 

type SearchHit struct {
	Provider  string `json:"provider"`
	SessionID string `json:"session_id"`
	Seq       uint64 `json:"seq"`
	Type      string `json:"type"`
	Source    string `json:"source"`
	Snippet   string `json:"snippet,omitempty"`
	Path      string `json:"path"`
}

SearchHit is one matching ledger event. 

func Search(opts SearchOptions) ([]SearchHit, error)

Search scans session JSONL files under root and returns matching events.

type SearchOptions 

type SearchOptions struct {
	Root      string
	Provider  string
	SessionID string
	Types     []string
	Source    string
	Query     string
	Limit     int
}

SearchOptions filters ledger JSONL scans. Search walks every matching session file line-by-line — O(total bytes) with no index (by design; see docs/en/trajectory.md).

type SessionEndSeedData 

type SessionEndSeedData struct {
	ParentProvider string `json:"parent_provider"`
	ParentSession  string `json:"parent_session"`
	BoundarySeq    uint64 `json:"boundary_seq"`
}

SessionEndSeedData marks the lineage boundary after a fork seed copy.

type SessionForkData 

type SessionForkData struct {
	ParentProvider string `json:"parent_provider"`
	ParentSession  string `json:"parent_session"`
	BoundarySeq    uint64 `json:"boundary_seq"`
}

SessionForkData is the payload for session/fork (audit lineage).

type SessionKey 

type SessionKey struct {
	Provider    provider.ID
	SessionID   string
	ProjectRoot string
}

SessionKey identifies one ledger stream.

func ResolveSessionKey 

func ResolveSessionKey(providerName, sessionID, projectRoot, cwd string) SessionKey

ResolveSessionKey builds a stable ledger key; empty session_id gets a weak synthetic id.

func ResolveSessionKeyID 

func ResolveSessionKeyID(id provider.ID, sessionID, projectRoot, cwd string) SessionKey

ResolveSessionKeyID is ResolveSessionKey with a validated provider.ID.

func (SessionKey) String 

func (k SessionKey) String() string

type SessionOpenData 

type SessionOpenData struct {
	Provider    string `json:"provider"`
	CWD         string `json:"cwd,omitempty"`
	ProjectRoot string `json:"project_root,omitempty"`
}

SessionOpenData is the payload for session/open.

type SessionSummary 

type SessionSummary struct {
	Provider       string `json:"provider"`
	SessionID      string `json:"session_id"`
	Path           string `json:"path"`
	Lines          int    `json:"lines,omitempty"`
	ImporterStatus string `json:"importer_status,omitempty"`
}

SessionSummary is one on-disk session ledger.

func ListSessions 

func ListSessions(root, providerFilter string) ([]SessionSummary, error)

ListSessions scans root for session JSONL files, optionally filtered by provider.

type Store 

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

Store holds in-memory append-only session logs.

func NewStore 

func NewStore() *Store

NewStore returns an empty trajectory store.

func (*Store) Append 

func (s *Store) Append(key SessionKey, events []Event) []Event

Append adds events to the session log and returns copies of appended records.

func (*Store) Events 

func (s *Store) Events(key SessionKey) []Event

Events returns a copy of all events for a session key.

type SubscribeFilter 

type SubscribeFilter struct {
	Provider  string
	SessionID string
	Source    string
}

SubscribeFilter selects events for one subscriber.

type TranscriptMessageData 

type TranscriptMessageData struct {
	Role                string `json:"role,omitempty"`
	Text                string `json:"text,omitempty"`
	ToolUseID           string `json:"tool_use_id,omitempty"`
	TranscriptLineIndex int    `json:"transcript_line_index"`
}

TranscriptMessageData is the payload for transcript/message.

type TranscriptThinkingData 

type TranscriptThinkingData struct {
	Text                string `json:"text,omitempty"`
	TranscriptLineIndex int    `json:"transcript_line_index"`
}

TranscriptThinkingData is the payload for transcript/thinking.

Source Files

View all Source files

Directories

Path Synopsis
Package importer maps provider on-disk transcripts into trajectory events.
 Package importer maps provider on-disk transcripts into trajectory events.

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.