Documentation
¶
Overview ¶
Package jsonlstore implements the current versioned snapshot port.SessionStore, plus append-only port.ToolCallRecorder and port.EventLog sidecars. Save atomically replaces one bounded current snapshot; older namespaces are ignored and left untouched. ToolCall and Append retain their cumulative JSONL audit semantics.
SESSION-FAMILY NAMING. Each session's files share a bounded, injective family stem under the owner-only `sid-v1` subdirectory:
<dir>/sid-v1/sid-v1-<token>.session.json <dir>/sid-v1/sid-v1-<token>.tools.jsonl <dir>/sid-v1/sid-v1-<token>.events.jsonl
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 Reader
- func (r *Reader) List(ctx context.Context) ([]port.StoredSession, error)
- func (r *Reader) Load(ctx context.Context, id session.SessionID) (*session.Session, error)
- func (r *Reader) MetaList(ctx context.Context) ([]port.SessionMeta, error)
- func (r *Reader) PageSessionMetadata(ctx context.Context, req port.SessionMetadataPageRequest) (port.SessionMetadataPage, error)
- func (r *Reader) Read(ctx context.Context, id session.SessionID) iter.Seq2[session.Event, error]
- func (r *Reader) ReadAfter(ctx context.Context, id session.SessionID, after port.Cursor, ...) iter.Seq2[port.LogRecord, error]
- func (r *Reader) ReadSessionLineage(ctx context.Context, query port.SessionLineageQuery) (port.SessionLineageResult, error)
- func (*Reader) SupportsSessionActivityProjection() bool
- type SnapshotDurabilityCapability
- type Store
- func (st *Store) Append(ctx context.Context, id session.SessionID, ev session.Event) error
- func (st *Store) AppendEvent(ctx context.Context, id session.SessionID, ev session.Event) (port.Cursor, error)
- func (st *Store) AppendGap(ctx context.Context, id session.SessionID, reason string) (port.Cursor, 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) List(_ context.Context) ([]port.StoredSession, error)
- func (st *Store) Load(ctx context.Context, id session.SessionID) (*session.Session, error)
- func (st *Store) MetaList(ctx context.Context) ([]port.SessionMeta, 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) ReadAfter(ctx context.Context, id session.SessionID, after port.Cursor, ...) iter.Seq2[port.LogRecord, error]
- func (st *Store) ReadSessionLineage(ctx context.Context, query port.SessionLineageQuery) (port.SessionLineageResult, error)
- func (st *Store) Save(ctx context.Context, s *session.Session) error
- func (st *Store) ScheduleStore() port.ScheduleStore
- func (st *Store) SessionStorageHealth(ctx context.Context) (port.SessionStorageHealth, error)
- func (st *Store) SnapshotDurability() SnapshotDurabilityCapability
- func (*Store) SupportsSessionActivityProjection() bool
- 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 (`adapters/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 Reader ¶ added in v0.2.0
type Reader struct {
// contains filtered or unexported fields
}
Reader opens an existing JSONL store without initializing, repairing, or mutating it. It intentionally does not implement any writing port.
func OpenReader ¶ added in v0.2.0
OpenReader requires an existing current store namespace. Use New to prepare a store and its derivative indexes before opening it read-only.
func (*Reader) MetaList ¶ added in v0.2.0
MetaList reads metadata from the current prepared catalog.
func (*Reader) PageSessionMetadata ¶ added in v0.2.0
func (r *Reader) PageSessionMetadata(ctx context.Context, req port.SessionMetadataPageRequest) (port.SessionMetadataPage, error)
PageSessionMetadata reads a bounded page from the current prepared catalog.
func (*Reader) ReadAfter ¶ added in v0.2.0
func (r *Reader) ReadAfter(ctx context.Context, id session.SessionID, after port.Cursor, opts port.ReadOptions) iter.Seq2[port.LogRecord, error]
ReadAfter reads after a cursor and optionally follows new events.
func (*Reader) ReadSessionLineage ¶ added in v0.2.0
func (r *Reader) ReadSessionLineage(ctx context.Context, query port.SessionLineageQuery) (port.SessionLineageResult, error)
ReadSessionLineage reads selected existing partitions under shared locks.
func (*Reader) SupportsSessionActivityProjection ¶ added in v0.2.0
SupportsSessionActivityProjection reports support for saved activity metadata.
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 issue all primitives required to make both snapshot contents and the replacement directory entry durable before returning nil. The underlying filesystem and storage stack must honor successful sync/rename calls.
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) 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. A nil return means the newline-committed record has been file-synced and its containing directory has been synced on every append.
It is AppendEvent with the cursor discarded — one append path, so a caller that does not want a position cannot end up on a different code path with different locking or a missing generation header.
func (*Store) AppendEvent ¶
func (st *Store) AppendEvent(ctx context.Context, id session.SessionID, ev session.Event) (port.Cursor, error)
AppendEvent durably records ev and returns the cursor positioned after it.
func (*Store) AppendGap ¶
func (st *Store) AppendGap(ctx context.Context, id session.SessionID, reason string) (port.Cursor, error)
AppendGap durably records a gap marker and returns the cursor positioned after it.
func (*Store) Create ¶
Create atomically publishes the adapter-private current snapshot only when no current snapshot exists.
func (*Store) Delete ¶
Delete removes current sidecars before the current snapshot. Old snapshot families are unsupported and are left untouched.
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) Load ¶
Load reads only the authoritative current snapshot without modifying storage. Old snapshot families are rejected with a data-preserving error.
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) 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 newline-terminated record, an unknown format tag, or an I/O error — is yielded as the error on a zero-value event and the consumer stops. Only an unterminated final record is ignored: newline is the event-record commit marker.
func (*Store) ReadAfter ¶
func (st *Store) ReadAfter(ctx context.Context, id session.SessionID, after port.Cursor, opts port.ReadOptions) iter.Seq2[port.LogRecord, error]
ReadAfter yields the log's records strictly after the cursor, optionally following the tail until ctx is done.
func (*Store) ReadSessionLineage ¶
func (st *Store) ReadSessionLineage(ctx context.Context, query port.SessionLineageQuery) (port.SessionLineageResult, error)
ReadSessionLineage reads only the selected root record and exact-incarnation direct-edge partitions. It never locks, reconciles, or writes.
func (*Store) Save ¶
Save atomically replaces the adapter-private current snapshot. Old snapshot families are rejected and left untouched.
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) SupportsSessionActivityProjection ¶
SupportsSessionActivityProjection reports that the JSONL metadata catalog atomically reflects the latest snapshot's activity.
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