journal

package
v0.11.0-beta.1 Latest Latest
Warning

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

 Go to latest Published: Oct 23, 2026 License: Apache-2.0, MIT Imports: 21 Imported by: 0

Documentation

Overview

Package journal — CLAIM-CONFIRM async compression queue (§4.19, inspired by claude-mem).

The problem: an observation arrives. Persisting it is fast (one JSONL append). Compressing it — LLM rewrite into structured facts/concepts, embedding for vector search, deduping against neighbors — is slow and unreliable (model timeouts, bridge down, disk hiccups). If we do the slow part on the hot path, every captured observation pays the latency. If we just kick off a goroutine and forget, a crashed worker means the observation stays half-processed forever.

CLAIM-CONFIRM splits the lifecycle:

  1. CAPTURE: durable write to disk (raw observation JSONL). Never blocks; agent moves on.
  2. CLAIM: a worker pulls a pending observation and marks it `claimed` with its PID + a deadline. Other workers skip it while the deadline is in the future.
  3. CONFIRM: the worker finishes compression and marks it `confirmed`. The job is done.
  4. RECOVERY: if a worker dies mid-CLAIM, its deadline expires and another worker re-claims. No "stuck forever" state.

Storage shape: one tiny JSON file per observation under <dir>/queue/ with the state machine. Atomic-rename writes for the state transitions so a crash mid-write doesn't corrupt the queue file (the previous state survives intact).

Package journal — typed failed-hypothesis tracker (paper #48 design input #5).

The shape we care about: the agent forms a hypothesis ("I think X is the bug"), tries it, and reports back that it didn't work. Today that signal lives only in the prose of agent_reply entries — it is not searchable, not surfaceable in future sessions, and tends to get retried verbatim a week later.

FailedHypothesis is a structured record carved out of those replies. We extract them with conservative regexes (it's better to miss than to populate the store with noise) and persist them as their own JSONL stream alongside the flight recorder.

Future readers — what the next session can do with this:

  • Before exploring a fix, query "have we already tried this on this codebase?" — saves a full debug loop on repeats.
  • Group by Subject (the thing we hypothesised about) to surface "we tried 4 things on auth, none worked" patterns that warrant a different angle.

Package journal — 3-layer query protocol over the FlightRecorder raw JSONL store (master plan §4.19).

Separate from the ObservationStore-backed JournalQuery in query.go: that path uses structured observation records via the journal sub- agent. THIS path is the raw flight-recorder readback — every turn, every tool call, every error logged as-it-happened.

The agent calls these mid-session via tools, NOT from boot. The design follows claude-mem's progressive disclosure: cheap index first, drill in only when interesting.

  • SearchFlight → compact index: ID, timestamp, type, title
  • TimelineFlight → chronological context around an anchor entry
  • GetFlight → full Entry on demand

Implementation: streaming scan over raw/ JSONL files. Entries are never fully materialized — we stream with bufio.Scanner and early-terminate when enough results are found.

Package journal — session replay (§8.3 Phase 5 #5).

Reads a journaled session's flight recorder entries and yields them in chronological order with optional filtering. Useful for:

  • Debugging "why did the agent decide X?" — step through the exact sequence of inputs, tool calls, results, errors.
  • Post-mortem rendering — the SSE dashboard subscribes to the replay stream to redraw a past session.
  • Test fixtures — feed a deterministic replay into a stub agent to verify deterministic-by-design behaviour.

What this is NOT: re-executing tools. Replay never invokes side effects; it just yields recorded entries. The tools' real results are in the journal — we surface them, we don't reproduce them.

Package journal — SSE memory dashboard endpoint (§4.19 last open item). Streams observation + alert events to subscribers over Server-Sent Events so users can build their own dashboard without us baking in a UI choice.

Why SSE not WebSocket: one-way is the right shape (the dashboard reads, doesn't write), browsers handle reconnect for free, and curl can subscribe with no extra tooling.

What gets streamed:

  • `observation` — every observation stored via Broadcast
  • `alert` — every alert created via BroadcastAlert
  • `heartbeat` — empty event every 30s so reverse proxies don't time out idle connections

Auth: loopback-only by default. The caller (`overkill dashboard` CLI) sets a bearer Token; clients pass `Authorization: Bearer X`. Empty token disables auth — fine for local-only deployments.

Package journal — vector similarity for observations (§4.19 hybrid search). Wires the bridge package's embedding service to the ObservationStore so journal_search can blend FTS hits with semantic neighbors.

Design:

  • VectorIndex is a tiny interface: Embed(text) → vector, Store(id, vec, content, meta), Search(vec, topK, threshold). The wiring layer (cmd/overkill) plugs internal/memory's BridgeAdapter in — internal/journal doesn't import the Python bridge directly.
  • Indexing is best-effort and asynchronous-ish: we kick off Store under a short timeout. A bridge that's down or slow never blocks observation capture. The observation is still persisted to disk via the existing ObservationStore path; vector search just won't find it until the bridge catches up.
  • Search is synchronous (the caller is asking RIGHT NOW), but bounded by a 5s timeout and returns ("", nil) when the bridge is unavailable — caller falls back to FTS-only.

This closes the "vector path stubbed" item in §4.19. The shape here is intentionally narrow: the journal owns the observation lifecycle, the bridge owns the math.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func FormatReplayEvent 

func FormatReplayEvent(ev ReplayEvent) string

FormatReplayEvent renders one event as a single-line summary suitable for CLI streaming. Types get a single-character glyph; content is truncated to keep terminal output readable.

Types

type Alert 

type Alert struct {
	ID           string    `json:"id"`
	Type         AlertType `json:"type"`
	Message      string    `json:"message"`
	SessionID    string    `json:"session_id"`
	Timestamp    time.Time `json:"timestamp"`
	Acknowledged bool      `json:"acknowledged"`
}

type AlertStore 

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

func NewAlertStore 

func NewAlertStore(dir string) *AlertStore

func (*AlertStore) Acknowledge 

func (s *AlertStore) Acknowledge(id string) error

func (*AlertStore) Close 

func (s *AlertStore) Close()

Close cancels the background prune goroutine.

func (*AlertStore) Create 

func (s *AlertStore) Create(alertType AlertType, message string, sessionID string) error

func (*AlertStore) DismissAll 

func (s *AlertStore) DismissAll() error

func (*AlertStore) Load 

func (s *AlertStore) Load() error

func (*AlertStore) Pending 

func (s *AlertStore) Pending() []Alert

func (*AlertStore) Save 

func (s *AlertStore) Save() error

func (*AlertStore) SetMaxAlerts 

func (s *AlertStore) SetMaxAlerts(n int)

SetMaxAlerts configures the cap on total in-memory alerts (default 1000).

func (*AlertStore) SetRetention 

func (s *AlertStore) SetRetention(d time.Duration)

SetRetention configures how long acknowledged alerts are kept before pruning. Set <= 0 to disable time-based pruning (still capped by max).

type AlertType 

type AlertType string
const (
	AlertCompactionSkip   AlertType = "compaction_skip"
	AlertTaskDeferred     AlertType = "task_deferred"
	AlertPatternDetected  AlertType = "pattern_detected"
	AlertFrustration      AlertType = "frustration_signal"
	AlertDelegationFailed AlertType = "delegation_failure"
	// AlertMemoryCorruption fires when a DB open or integrity
	// check fails (§4.20). The TUI surfaces this with a restore
	// prompt: "Memory corrupted. I knew I knew you. I don't know what
	// I knew. Last export was 3 days ago — want me to restore from
	// that?"
	AlertMemoryCorruption AlertType = "memory_corruption"

	// AlertTaskCompleted fires from the ledger when a long-running
	// background task transitions to a terminal state (Completed /
	// Failed / Cancelled / Lost / TimedOut). §7.1 Layer 6: push
	// notification on completion. The gateway hub reads these out
	// of the alert store and delivers to the user's active channels.
	AlertTaskCompleted AlertType = "task_completed"
)

type CompressionQueue 

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

CompressionQueue is the on-disk JSONL-per-job store. Concurrent access from multiple workers (in-process or cross-process) is arbitrated by the file-rename atomicity: a Claim that loses the rename race wins nothing, the other worker has the job.

func NewCompressionQueue 

func NewCompressionQueue(dir string) *CompressionQueue

NewCompressionQueue wires the queue to a directory. Files land at <dir>/<job-id>.json. Lazy-created on first Enqueue.

func (*CompressionQueue) Claim 

func (q *CompressionQueue) Claim(workerPID int, lease time.Duration) (*QueueJob, error)

Claim atomically takes the oldest pending job (or one whose claim has expired) and marks it Claimed by the calling process for the given lease duration. Returns (nil, nil) when nothing is available — caller should sleep + retry.

Lease semantics: claim expires after `lease`. A worker that can't finish in time can call Renew to extend; a dead worker's claim simply expires and another worker picks the job up. No distributed lock service required.

func (*CompressionQueue) Confirm 

func (q *CompressionQueue) Confirm(jobID string, workerPID int) error

Confirm marks a job done. Caller must hold a valid claim.

func (*CompressionQueue) Enqueue 

func (q *CompressionQueue) Enqueue(observationID string) (*QueueJob, error)

Enqueue creates a new pending job for the observation. Idempotent by observation ID: re-enqueuing the same observation returns the existing job (caller can read its state instead of creating a duplicate).

func (*CompressionQueue) Fail 

func (q *CompressionQueue) Fail(jobID string, workerPID int, err error, maxAttempts int) error

Fail records a compression error. After maxAttempts the job transitions to QueueFailed and stops being re-claimed. Below the threshold it goes back to QueuePending for retry.

func (*CompressionQueue) Get 

func (q *CompressionQueue) Get(id string) (*QueueJob, error)

Get returns one job by ID.

func (*CompressionQueue) List 

func (q *CompressionQueue) List() ([]QueueJob, error)

List returns every job currently on disk. Useful for operators inspecting the queue ("which observations are stuck failed?").

func (*CompressionQueue) PendingCount 

func (q *CompressionQueue) PendingCount() (int, error)

PendingCount returns the number of jobs that could be claimed right now — pending plus jobs whose claim has expired.

func (*CompressionQueue) Renew 

func (q *CompressionQueue) Renew(jobID string, workerPID int, lease time.Duration) error

Renew extends the claim deadline. Used by long-running workers (LLM compression on a slow connection) to avoid expiration. Errors when the job no longer belongs to this worker — the lease expired and another worker took over.

type DashboardServer 

type DashboardServer struct {
	Listen string // default 127.0.0.1:7802
	Token  string // empty disables auth (loopback deployments)
	// contains filtered or unexported fields
}

DashboardServer is the SSE endpoint hub. Subscribers register via Subscribe and receive events through their channel until they close it. Best-effort delivery: a slow subscriber gets its event dropped rather than backing up the broadcast path.

func NewDashboardServer 

func NewDashboardServer() *DashboardServer

NewDashboardServer wires the server. Caller controls Listen + Token; defaults are loopback-only with no auth.

func (*DashboardServer) BroadcastAlert 

func (d *DashboardServer) BroadcastAlert(alert *Alert)

BroadcastAlert mirrors BroadcastObservation for alert events.

func (*DashboardServer) BroadcastObservation 

func (d *DashboardServer) BroadcastObservation(obs *Observation)

BroadcastObservation fans an observation event to every live subscriber. Non-blocking: slow subscribers drop the event rather than gating the publisher.

func (*DashboardServer) Run 

func (d *DashboardServer) Run(ctx context.Context) error

Run starts the HTTP server and blocks until ctx cancels.

func (*DashboardServer) SubscriberCount 

func (d *DashboardServer) SubscriberCount() int

SubscriberCount returns the number of live SSE clients. Exposed for tests and operator visibility (`dashboard status` could read this).

type Entry 

type Entry struct {
	ID        string          `json:"id"`
	Type      EntryType       `json:"type"`
	SessionID string          `json:"session_id"`
	Timestamp time.Time       `json:"timestamp"`
	Content   string          `json:"content"`
	Metadata  json.RawMessage `json:"metadata,omitempty"`
}

type EntryType 

type EntryType string
const (
	EntryUserInput  EntryType = "user_input"
	EntryAgentReply EntryType = "agent_reply"
	EntryToolCall   EntryType = "tool_call"
	EntryToolResult EntryType = "tool_result"
	EntryError      EntryType = "error"
	EntrySystem     EntryType = "system"
)

type FailedHypothesis 

type FailedHypothesis struct {
	ID         string    `json:"id"`
	SessionID  string    `json:"session_id"`
	EntryID    string    `json:"entry_id"`
	Timestamp  time.Time `json:"timestamp"`
	Subject    string    `json:"subject,omitempty"`
	Hypothesis string    `json:"hypothesis"`
	Reason     string    `json:"reason"`
	// ModelID tags the record with the model that produced the
	// failure. Optional (older records lack it). Used at read time
	// to filter "I've hit this wall twice" down to the CURRENT
	// model — a stale record from a previous model swap is noise.
	// See §4.16 model fingerprinting.
	ModelID string `json:"model_id,omitempty"`
}

FailedHypothesis is one "tried X, failed because Y" record. Subject is the noun-phrase the hypothesis was about (the file, the function, the system); Hypothesis is what was tried; Reason is the failure mode reported by the agent. EntryID points back to the agent_reply this was extracted from so callers can read the full turn for context.

func ExtractFailedHypotheses 

func ExtractFailedHypotheses(e Entry) []FailedHypothesis

ExtractFailedHypotheses pulls structured records out of one agent_reply entry. Returns at most one finding per pattern match — duplicates within a single reply are deduped by hypothesis text.

type FailedHypothesisStore 

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

FailedHypothesisStore is the durable, append-only stream of extracted records. One file per day (failed_hypotheses/YYYY-MM-DD.jsonl) mirroring the flight-recorder layout — same disk-budget story, same roll-up cadence.

func NewFailedHypothesisStore 

func NewFailedHypothesisStore(dir string) *FailedHypothesisStore

func (*FailedHypothesisStore) All 

func (s *FailedHypothesisStore) All() ([]FailedHypothesis, error)

All loads every persisted record across every day file. Suitable for small histories; if the store grows past tens of thousands we add a date-range API.

func (*FailedHypothesisStore) Append 

func (s *FailedHypothesisStore) Append(h FailedHypothesis) error

func (*FailedHypothesisStore) Search 

func (s *FailedHypothesisStore) Search(query string) ([]FailedHypothesis, error)

Search returns records whose Subject, Hypothesis, or Reason contains the given query (case-insensitive). The intent is "have we tried this?" not "exact phrase match" — substring is good enough.

func (*FailedHypothesisStore) SearchForModel 

func (s *FailedHypothesisStore) SearchForModel(query, modelID string) ([]FailedHypothesis, error)

SearchForModel is Search restricted to records produced by the given model ID. Empty modelID returns no filtering (same as Search). Records without a ModelID (older entries that predate fingerprinting) are INCLUDED — we'd rather show stale-but-marked records than silently drop them. The caller can dedupe by Subject if needed.

type FlightIndexHit 

type FlightIndexHit struct {
	ID        string    `json:"id"`
	Type      EntryType `json:"type"`
	Timestamp string    `json:"timestamp"` // RFC3339 for readability
	Title     string    `json:"title"`
	SessionID string    `json:"session_id,omitempty"`
}

FlightIndexHit is one row in the compact search index. Title is the first ~80 chars of content (newlines collapsed). Tiny on purpose so the model can fit many results in one tool call.

type FlightRecorder 

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

func NewFlightRecorder 

func NewFlightRecorder(dir string, sessionID string) *FlightRecorder

func (*FlightRecorder) GetFlight 

func (r *FlightRecorder) GetFlight(id string) (*Entry, error)

GetFlight returns the raw-journal entry with the matching ID, or (nil, nil) when not found. Streams files until the ID is found.

func (*FlightRecorder) ReadDay 

func (r *FlightRecorder) ReadDay(date time.Time) ([]Entry, error)

func (*FlightRecorder) ReadSession 

func (r *FlightRecorder) ReadSession(sessionID string) ([]Entry, error)

func (*FlightRecorder) Record 

func (r *FlightRecorder) Record(entryType EntryType, content string, metadata json.RawMessage) (recordErr error)

func (*FlightRecorder) RecordError 

func (r *FlightRecorder) RecordError(err error) error

func (*FlightRecorder) RecordInput 

func (r *FlightRecorder) RecordInput(content string) error

func (*FlightRecorder) RecordReply 

func (r *FlightRecorder) RecordReply(content string) error

func (*FlightRecorder) RecordToolCall 

func (r *FlightRecorder) RecordToolCall(toolName string, input json.RawMessage) error

func (*FlightRecorder) RecordToolResult 

func (r *FlightRecorder) RecordToolResult(toolName string, output json.RawMessage) error

func (*FlightRecorder) Replay 

func (r *FlightRecorder) Replay(ctx context.Context, sessionID string, opts ReplayOptions) (<-chan ReplayEvent, <-chan error)

Replay walks a session in order and emits ReplayEvents on the returned channel. The channel closes when:

  • All entries have been yielded.
  • ctx is cancelled.
  • An unrecoverable read error occurs (sent on errCh too).

errCh is buffered so a caller that ignores it doesn't leak the goroutine. Both channels are 1-shot from the producer side.

func (*FlightRecorder) SearchFlight 

func (r *FlightRecorder) SearchFlight(opts FlightSearchOptions) ([]FlightIndexHit, error)

SearchFlight returns matching raw-journal entries as a compact index, newest first. Streams JSONL files in reverse-chronological order and early-terminates after hitting the limit — entries are never fully materialized.

func (*FlightRecorder) TimelineFlight 

func (r *FlightRecorder) TimelineFlight(anchorID string, depth int) ([]Entry, error)

TimelineFlight returns `depth` entries before AND after the anchor, in chronological order. The anchor is included as the middle entry. Uses binary-search over date-named files: first finds the anchor timestamp, then only reads files in the surrounding time window.

type FlightSearchOptions 

type FlightSearchOptions struct {
	Query   string    // case-insensitive substring; empty = "all"
	Type    EntryType // empty = any type
	Session string    // empty = any session
	Limit   int       // 0 → 20
}

FlightSearchOptions controls SearchFlight behaviour.

type JournalQuery 

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

func NewJournalQuery 

func NewJournalQuery(store *ObservationStore) *JournalQuery

func (*JournalQuery) Get 

func (jq *JournalQuery) Get(id string) (*Observation, error)

func (*JournalQuery) Search 

func (jq *JournalQuery) Search(query string, obsType ObservationType, limit int) []ObservationIndex

func (*JournalQuery) Timeline 

func (jq *JournalQuery) Timeline(anchorID string, depth int) []Observation

type Observation 

type Observation struct {
	ID            string          `json:"id"`
	Type          ObservationType `json:"type"`
	Title         string          `json:"title"`
	Narrative     string          `json:"narrative"`
	Facts         []string        `json:"facts"`
	Concepts      []string        `json:"concepts"`
	FilesRead     []string        `json:"files_read"`
	FilesModified []string        `json:"files_modified"`
	SessionID     string          `json:"session_id"`
	Timestamp     time.Time       `json:"timestamp"`
	ContentHash   string          `json:"content_hash"`
}

func NewObservation 

func NewObservation(obsType ObservationType, title, narrative, sessionID string) *Observation

func (*Observation) ComputeHash 

func (o *Observation) ComputeHash() string

func (*Observation) Index 

func (o *Observation) Index() ObservationIndex

type ObservationIndex 

type ObservationIndex struct {
	ID        string          `json:"id"`
	Type      ObservationType `json:"type"`
	Title     string          `json:"title"`
	Timestamp time.Time       `json:"timestamp"`
}

type ObservationStore 

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

func NewObservationStore 

func NewObservationStore(dir string) *ObservationStore

func (*ObservationStore) Get 

func (s *ObservationStore) Get(id string) (*Observation, error)

func (*ObservationStore) List 

func (s *ObservationStore) List(obsType ObservationType, limit int) []ObservationIndex

func (*ObservationStore) Store 

func (s *ObservationStore) Store(obs *Observation) error

func (*ObservationStore) Timeline 

func (s *ObservationStore) Timeline(anchorID string, depth int) []Observation

type ObservationType 

type ObservationType string
const (
	ObsBugfix    ObservationType = "bugfix"
	ObsFeature   ObservationType = "feature"
	ObsDecision  ObservationType = "decision"
	ObsDiscovery ObservationType = "discovery"
	ObsChange    ObservationType = "change"
	ObsRefactor  ObservationType = "refactor"
)

type QueueJob 

type QueueJob struct {
	ID            string     `json:"id"`
	ObservationID string     `json:"observation_id"`
	State         QueueState `json:"state"`
	EnqueuedAt    time.Time  `json:"enqueued_at"`
	ClaimedAt     time.Time  `json:"claimed_at,omitempty"`
	ClaimedBy     int        `json:"claimed_by,omitempty"`
	ClaimedUntil  time.Time  `json:"claimed_until,omitempty"`
	ConfirmedAt   time.Time  `json:"confirmed_at,omitempty"`
	Attempts      int        `json:"attempts"`
	LastError     string     `json:"last_error,omitempty"`
}

QueueJob is one item in the CLAIM-CONFIRM queue. ObservationID points at the raw observation; the actual narrative + embedding live in the ObservationStore, not duplicated here.

type QueueState 

type QueueState string

QueueState is the CLAIM-CONFIRM lifecycle position. 

const (
	// QueuePending: enqueued, waiting for a worker.
	QueuePending QueueState = "pending"
	// QueueClaimed: a worker took it; check ClaimedUntil to know
	// whether the claim is still valid.
	QueueClaimed QueueState = "claimed"
	// QueueConfirmed: compression done; safe to delete or archive.
	QueueConfirmed QueueState = "confirmed"
	// QueueFailed: compression hit a hard error too many times.
	// Surfaced for operator review; not retried.
	QueueFailed QueueState = "failed"
)

type ReplayEvent 

type ReplayEvent struct {
	Entry  Entry
	Offset time.Duration // since first entry in the replay
	Index  int           // 0-based position within the filtered stream
}

ReplayEvent is one yielded entry plus the elapsed offset from session start. Consumers can use Offset for timeline rendering without re-deriving it.

type ReplayOptions 

type ReplayOptions struct {
	// Types filters to specific entry types. Empty means "all".
	Types []EntryType
	// From skips entries before this timestamp. Zero means "no
	// lower bound".
	From time.Time
	// Until stops at this timestamp. Zero means "no upper bound".
	Until time.Time
	// Speed >0 enables real-time playback: sleep between entries
	// so they emit at recorded pace, divided by speed (1.0 = real
	// time, 2.0 = 2× faster). 0 yields as fast as the consumer can
	// read.
	Speed float64
}

ReplayOptions tunes what to yield. Zero-value yields every entry in the session.

type Replayer 

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

Replayer is a stateful wrapper around Replay for clients that want push-style consumption with backpressure. The simpler Replay channel API is usually enough — Replayer is a convenience for SSE / TUI consumers that need to control stepping themselves.

func NewReplayer 

func NewReplayer(rec *FlightRecorder, sessionID string, opts ReplayOptions) *Replayer

func (*Replayer) Snapshot 

func (r *Replayer) Snapshot() ([]ReplayEvent, error)

Snapshot returns the full filtered+sorted entry list synchronously. Useful for "give me the whole session as one blob" callers (e.g. the CLI's --no-stream mode).

type Summarizer 

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

func NewSummarizer 

func NewSummarizer(recorder *FlightRecorder, provider providers.Provider, model string) *Summarizer

func (*Summarizer) NarrateSession 

func (s *Summarizer) NarrateSession(ctx context.Context, dir, sessionID string) (string, string, error)

NarrateSession runs the full diary-renderer pipeline for one session: read entries → produce narrative → persist to the day file. Returns the path written and the narrative text. Empty sessions (no journal entries) are a no-op — returns ("", "", nil).

Intended call sites: TUI session-end defer (best-effort, errors logged), `overkill journal narrate` CLI (errors surfaced).

func (*Summarizer) Summarize 

func (s *Summarizer) Summarize(ctx context.Context, sessionID string) (string, error)

func (*Summarizer) SummarizeDay 

func (s *Summarizer) SummarizeDay(ctx context.Context, date times.Time) (string, error)

func (*Summarizer) WriteDayNarrative 

func (s *Summarizer) WriteDayNarrative(dir string, sessionID string, narrative string, when times.Time) error

WriteDayNarrative writes the day's diary entry to <dir>/entries/<YYYY-MM-DD>.md. When the file already exists (multiple sessions in one day), the new narrative is appended under a `## session <id>` sub-section so the day stays one file per the §4.19 spec while still preserving per-session attribution.

func (*Summarizer) WriteSummary 

func (s *Summarizer) WriteSummary(dir string, sessionID string, summary string) error

type VectorEnabledStore 

type VectorEnabledStore struct {
	*ObservationStore
	// contains filtered or unexported fields
}

VectorEnabledStore wraps an ObservationStore with an optional vector index. When the index is nil (or wiring failed at boot), the wrapper falls through to the bare store — vector calls are no-ops and SearchSimilar returns (nil, nil).

func NewVectorEnabledStore 

func NewVectorEnabledStore(store *ObservationStore, index VectorIndex, opts VectorOptions) *VectorEnabledStore

NewVectorEnabledStore wraps store with an index. Passing a nil index is legal and gives FTS-only behaviour — useful for tests and for users running without the Python bridge.

func (*VectorEnabledStore) SearchSimilar 

func (s *VectorEnabledStore) SearchSimilar(ctx context.Context, query string, topK int, threshold float64) ([]VectorSearchHit, error)

SearchSimilar embeds the query and asks the index for top-K neighbors. Returns (nil, nil) when no index is wired or when the bridge is unreachable — caller falls back to FTS results.

Bounded by VectorOptions.searchTimeout(); a slow bridge degrades to "no semantic hits" rather than stalling the agent.

func (*VectorEnabledStore) StoreWithVector 

func (s *VectorEnabledStore) StoreWithVector(ctx context.Context, obs *Observation) error

StoreWithVector persists the observation AND embeds it into the vector index. The observation lands on disk first (durability is non-negotiable); the embedding is best-effort.

type VectorIndex 

type VectorIndex interface {
	Embed(ctx context.Context, text, model string) ([]float32, int32, error)
	Store(ctx context.Context, id string, embedding []float32, content string, metadata map[string]string) (string, error)
	Search(ctx context.Context, query []float32, topK int, threshold float64) ([]VectorSearchHit, error)
}

VectorIndex is the minimal surface the journal needs from a vector store. Pass nil in the constructor to disable the vector path (FTS-only mode).

type VectorOptions 

type VectorOptions struct {
	// EmbedModel passed to Embed. Empty → bridge's default.
	EmbedModel string
	// IndexTimeout caps the per-observation Store call. Default 3s.
	IndexTimeout time.Duration
	// SearchTimeout caps SearchSimilar. Default 5s.
	SearchTimeout time.Duration
	// DefaultTopK when caller passes 0. Default 10.
	DefaultTopK int
	// DefaultThreshold when caller passes 0. Default 0.5 cosine.
	DefaultThreshold float64
}

VectorOptions tunes embed model + search defaults. Zero-value defaults are reasonable for most callers.

type VectorSearchHit 

type VectorSearchHit struct {
	ID       string
	Score    float64
	Content  string
	Metadata map[string]string
}

VectorSearchHit is one neighbor returned by the vector index. Score is provider-defined (cosine similarity for the default embedding service); higher is more similar.

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.