Documentation
¶
Overview ¶
Package jsonlstore implements a versioned current-snapshot port.SessionStore, plus append-only port.ToolCallRecorder and port.EventLog sidecars. Save atomically replaces one bounded v2 current snapshot; Load also accepts historical v1 JSONL snapshots and a successful save lazily promotes that session. ToolCall and Append retain their cumulative JSONL audit semantics.
SESSION-FAMILY NAMING. Each session's files share a family stem under the owner-only `sid-v1` subdirectory. The reversible token is `sid-v1-` plus Raw URL-base64 of the complete opaque valid-UTF-8 session id:
<dir>/sid-v1/sid-v1-<token>.session.json (v2 current) <dir>/sid-v1/sid-v1-<token>.session.jsonl (readable v1 history) <dir>/sid-v1/sid-v1-<token>.tools.jsonl <dir>/sid-v1/sid-v1-<token>.events.jsonl
A pre-rewrite family may still use the lossy legacySafeName stem. The sessionResolver in resolve.go is the single authority for canonical/legacy paths, ownership checks, and write-time migration. Reads never migrate.
The .events.jsonl log is PARALLEL to (not a superset of) .tools.jsonl: the tool log is the structured per-tool AUDIT seam (args, queue/exec timing), the event log is the relayed STREAM (reasoning, ask/verdict pairs, delegation lifecycle) the server otherwise discards. Neither subsumes the other.
Physical names are confined to filename-safe tokens under the store dir.
Index ¶
- Constants
- Variables
- type SnapshotDurabilityCapability
- type Store
- func (st *Store) AcquireSessionMigrationJob(ctx context.Context, id string) (context.Context, func() error, error)
- func (st *Store) Append(ctx context.Context, id session.SessionID, ev session.Event) error
- func (st *Store) CheckSessionMigrationJobOwnership(ctx context.Context) error
- func (st *Store) Create(ctx context.Context, s *session.Session) error
- func (st *Store) Delete(ctx context.Context, id session.SessionID) error
- func (st *Store) DeleteSessionIfUnchanged(ctx context.Context, expected port.SessionDiscoveryMeta) (bool, error)
- func (st *Store) InspectSessionMigration(ctx context.Context) (port.SessionMigrationInspection, error)
- func (st *Store) List(_ context.Context) ([]port.StoredSession, error)
- func (st *Store) Load(ctx context.Context, id session.SessionID) (*session.Session, error)
- func (st *Store) LoadSessionMigrationJob(_ context.Context, id string) (port.SessionMigrationJob, error)
- func (st *Store) MetaList(ctx context.Context) ([]port.SessionMeta, error)
- func (st *Store) MigrateSessionFamily(ctx context.Context, expected port.SessionMigrationFamily) (string, error)
- func (st *Store) PageSessionMetadata(ctx context.Context, request port.SessionMetadataPageRequest) (port.SessionMetadataPage, error)
- func (st *Store) Read(ctx context.Context, id session.SessionID) iter.Seq2[session.Event, error]
- func (st *Store) Save(ctx context.Context, s *session.Session) error
- func (st *Store) SaveSessionMigrationJob(ctx context.Context, job port.SessionMigrationJob) error
- func (st *Store) ScheduleStore() port.ScheduleStore
- func (st *Store) SessionStorageHealth(ctx context.Context) (port.SessionStorageHealth, error)
- func (st *Store) SnapshotDurability() SnapshotDurabilityCapability
- func (st *Store) ToolCall(id session.SessionID, call session.ToolCall, result session.ToolResult, ...)
Constants ¶
const EventLogFormat = "eventlog-json/1"
EventLogFormat is the per-record format tag written on every event-log line. It versions the on-disk encoding so the language-neutral driver wire (3c) and any future format change can be distinguished; Read rejects an unknown tag as an infra error (a forward-incompatible log must fail loud, not silently skip).
It is EXPORTED so the gRPC driver wire (`internal/adapter/grpcdriver.EventLogFormat`) can be pinned EQUAL to it by a test: the wire payload is exactly this record's "ev" bytes (json.Marshal of a session.Event), so the two tags MUST agree or a log written by one path is unreadable by the other (the one-codec claim). The unexported alias keeps the in-file call sites terse.
Variables ¶
var ErrNotFound = fmt.Errorf("jsonlstore: session not found: %w", port.ErrSessionNotFound)
ErrNotFound is returned by Load when no snapshot file exists for the id. It wraps port.ErrSessionNotFound so a consumer that may not import this adapter can distinguish not-found from an infra failure via errors.Is.
var ErrScheduleNotFound = fmt.Errorf("jsonlstore: schedule not found: %w", port.ErrScheduleNotFound)
ErrScheduleNotFound is returned by Load/Delete/Claim/RecordFire/LoadFire when no schedule (or fire) exists under the requested name/id. It wraps port.ErrScheduleNotFound so a consumer that may not import this adapter can distinguish not-found from an infra failure via errors.Is, the same discipline the session store applies for ErrNotFound.
Functions ¶
This section is empty.
Types ¶
type SnapshotDurabilityCapability ¶
SnapshotDurabilityCapability reports which crash-durability primitives the current jsonlstore filesystem supports. Atomic replacement without all sync steps prevents torn snapshots but does not claim survival across a host crash.
func (SnapshotDurabilityCapability) HostCrashSafe ¶
func (c SnapshotDurabilityCapability) HostCrashSafe() bool
HostCrashSafe reports whether Save can make both snapshot contents and the replacement directory entry durable before returning nil.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is a current-snapshot SessionStore with append-only tool/event sidecars.
func New ¶
New constructs a Store writing under dir, creating dir if needed. The dir is created at mode 0700: the store holds raw conversation transcripts (session snapshots, tool-call args/results, and the relayed event stream) in plaintext, so it is owner-only by construction.
func (*Store) AcquireSessionMigrationJob ¶
func (st *Store) AcquireSessionMigrationJob(ctx context.Context, id string) (context.Context, func() error, error)
AcquireSessionMigrationJob holds one stable cross-process job exclusion until the returned release function is called and binds that acquisition to the returned context.
func (*Store) Append ¶
Append records one relayed event under id as a format-tagged JSON line on the per-session event log. It satisfies port.EventLog. The event is marshalled to its session.Event JSON verbatim (already redacted at the relay) and wrapped in the {"v":"eventlog-json/1","ev":...} envelope so Read can validate the format. It uses the same stable per-family cross-process mutation identity as Save/Delete/ToolCall, and is best-effort durable: the relay logs a WARN on a returned error and never aborts the run.
func (*Store) CheckSessionMigrationJobOwnership ¶
CheckSessionMigrationJobOwnership rejects contexts without the active exact acquisition used by this store.
func (*Store) Create ¶
Create atomically publishes the adapter-private v2 snapshot only when no authoritative current, canonical-v1, or matching legacy snapshot exists. Every supported writer takes the same stable family lock, so the presence check and final rename form one cross-process create-once operation.
func (*Store) Delete ¶
Delete removes canonical sidecars before the canonical snapshot, and a legacy family in the same order — REMOVAL ORDER is load-bearing, see familyOrder's doc comment (resolve.go): the snapshot file is what List enumerates, so removing it last means a partial failure leaves the family still VISIBLE (the next retention sweep retries it), where the reverse order would leave an invisible orphaned sidecar no sweep could ever find. With two families (canonical + legacy) this now has to hold TWICE per call: canonical sidecars before the canonical snapshot, AND — only when the legacy snapshot's embedded id proves it belongs to this session — legacy sidecars before the legacy snapshot. A legacy family that fails ownership (mismatch or absent) is left untouched; that mismatch and absence are both idempotent success.
The canonical family is removed on PRESENCE alone, never on the snapshot parsing: the token is injective, so the file is ours whatever it contains, and gating removal on validity made a torn snapshot line permanently unprunable — every retention sweep re-failed on it while List, which skips undecodable files, never surfaced it. port.PrunableStore requires that a Delete either remove or be idempotent success, so an unreadable snapshot must not be a third outcome.
func (*Store) DeleteSessionIfUnchanged ¶
func (st *Store) DeleteSessionIfUnchanged(ctx context.Context, expected port.SessionDiscoveryMeta) (bool, error)
DeleteSessionIfUnchanged holds the family lock across durable metadata revalidation and sidecar-first/snapshot-last deletion.
func (*Store) InspectSessionMigration ¶
func (st *Store) InspectSessionMigration(ctx context.Context) (port.SessionMigrationInspection, error)
InspectSessionMigration performs a read-only physical inventory. It never creates a catalog, job record, lock, quarantine, or replacement file.
func (*Store) List ¶
List returns one row per logical session id. IDs come from latest snapshots, never filenames; canonical files win when canonical and legacy coexist.
COST: ids are decoded from each session file's latest snapshot line rather than from filenames (a filename is not invertible back to the id, and now there are two directories — canonical and legacy — to reconcile), so List costs one directory read per dir plus one reverse TAIL read per snapshot file. The reader grows its EOF window only to the latest record, never scanning older snapshot history. Fine for a retention sweep on a startup/hourly cadence; indexed inventory is a separate concern.
List enumerates v2 current snapshots and historical *.session.jsonl files; sidecars are never inventory authority. That keeps sidecars-before-snapshot removal load-bearing (see familyOrder in resolve.go): a sidecar without any session snapshot is invisible here and can never be swept.
func (*Store) Load ¶
Load reads the authoritative snapshot without modifying storage. Canonical presence prevents fallback; legacy is accepted only when its latest embedded id exactly matches the requested id.
func (*Store) LoadSessionMigrationJob ¶
func (st *Store) LoadSessionMigrationJob(_ context.Context, id string) (port.SessionMigrationJob, error)
LoadSessionMigrationJob reloads one validated durable job record by opaque handle.
func (*Store) MetaList ¶
MetaList returns every stored session's picker metadata by reading ONLY the last snapshot line of each *.session.jsonl file and decoding into a small struct that skips the messages array. It satisfies port.MetaLister.
It is the CHEAP-listing path Service.ListSessions prefers (via type assertion) over the Load-per-row fallback: listing N sessions is O(N × last-line-read) instead of O(N × filesize), because each file is tail-read (seek near the end, find the last newline) rather than fully scanned, and the unmarshal skips the conversation entirely. A file smaller than the seek window is read whole (small file = fast). Catalog rebuilds use a dedicated process mutex plus cross-process catalog flock; they never take a store-wide session-operation lock.
The last line is the LATEST snapshot (append-only, latest-line-wins), so the metadata reflects the session's CURRENT state/turns/model/title, exactly as a full Load would.
CORRUPT-ROW CONTRACT (mirrors the Load-per-row path): a row whose last line decodes a valid id but CANNOT be fully restored (an unknown state RestoreState would reject, a missing id, or undecodable JSON) still surfaces with its id + modified_at but ZEROED snapshot-derived fields — the same behaviour ListSessions had when Load failed per row. So a corrupt snapshot file is visible in the picker with its id/mtime even if it can't be opened, exactly as before.
func (*Store) MigrateSessionFamily ¶
func (st *Store) MigrateSessionFamily(ctx context.Context, expected port.SessionMigrationFamily) (string, error)
MigrateSessionFamily holds the stable family flock across revalidation, promotion, v2 verification, and v1 removal. Item failures are returned as a closed reason code; raw backend errors never cross the maintenance boundary.
func (*Store) PageSessionMetadata ¶
func (st *Store) PageSessionMetadata(ctx context.Context, request port.SessionMetadataPageRequest) (port.SessionMetadataPage, error)
PageSessionMetadata reads at most Limit+1 rows from the owner-specific, pre-ordered derivative catalog. The cursor's byte position seeks directly to page two; no prior catalog row or snapshot payload is traversed.
func (*Store) Read ¶
Read scans the per-session event log and yields every recorded event in append order (cumulative — NOT latest-line-wins like the snapshot read). It satisfies port.EventLog. A MISS (no event file) yields an EMPTY sequence: absence is data. A genuine fault — an undecodable record, an unknown format tag, or an I/O error — is yielded as the error on a zero-value event and the consumer stops (the iterator returns after the consumer's range body returns false on the error item, the standard iter.Seq2 error idiom).
func (*Store) Save ¶
Save atomically replaces the adapter-private v2 current snapshot. Existing v1 JSONL snapshots remain readable and are promoted lazily on the next save; their file modification time becomes the v2 logical modification time.
func (*Store) SaveSessionMigrationJob ¶
SaveSessionMigrationJob atomically checkpoints one sanitized durable job record.
func (*Store) ScheduleStore ¶
func (st *Store) ScheduleStore() port.ScheduleStore
ScheduleStore returns a port.ScheduleStore backed by the SAME directory as the session store (a sibling struct sharing the dir + the single-process mutex). Composition discovers it via type-assertion on this accessor — NOT by asserting the *Store itself implements port.ScheduleStore (the schedule store is a separate concern; the accessor keeps session-store and schedule-store methods from bloating one struct, the way PrunableStore is discovered on the store itself but here the schedule store is a sibling struct, not the session store). A caller that does not need schedules never calls this; the byte-identical default is no schedules.
func (*Store) SessionStorageHealth ¶
SessionStorageHealth reports content-free aggregate health from the current derivative inventory catalog and cheap filesystem metadata. A missing or stale catalog is reported unavailable; health inspection never rebuilds it.
func (*Store) SnapshotDurability ¶
func (st *Store) SnapshotDurability() SnapshotDurabilityCapability
SnapshotDurability reports the filesystem primitives this Store verified at construction. A false field is an explicit weaker guarantee, not a Save error.
func (*Store) ToolCall ¶
func (st *Store) ToolCall(id session.SessionID, call session.ToolCall, result session.ToolResult, queued, took time.Duration)
ToolCall appends a structured tool-call record to the per-session tool log, including both the dispatch queue time (queued) and the execution wall time (took) in microseconds. It satisfies port.ToolCallRecorder. Errors are intentionally swallowed (the port has no error return) but the record is best-effort durable. Because the port carries no caller context, lock acquisition is capped at five seconds; on timeout the best-effort record is dropped rather than blocking a run.
Source Files