store

package
v0.116.3 Latest Latest
Warning

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

 Go to latest Published: Jul 9, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Overview

Package store owns the SQLite database that holds kojo v1's structured state. It is the single primary store for agents, messages, memory entries, tasks, kv settings, blob refs, peer registry, agent locks and migration status. Blob bodies (avatars, books, MEMORY.md, RAG indices) are stored on the filesystem under internal/blob and referenced by URI from blob_refs.

Layout: 

~/.config/kojo-v1/
  kojo.db                          # this database (WAL + SHM)
  kojo.db-wal
  kojo.db-shm
  global/  local/  machine/        # blob trees (managed by internal/blob)

The database is opened with WAL journaling, foreign_keys ON, and busy_timeout=5000ms. Migrations are numbered SQL files embedded into the binary; on Open, all pending files are applied inside an explicit transaction in lexical order. The applied schema version is stored in the `schema_migrations` table and is exposed via Store.SchemaVersion().

Index

Constants

View Source 
const (
	PeerStatusOnline  = "online"
	PeerStatusOffline = "offline"
)

PeerStatus values accepted by the schema's CHECK constraint.

View Source 
const (
	// DBFileName is the SQLite file name under the config dir.
	DBFileName = "kojo.db"
)
View Source 
const IfMatchAny = "*"

IfMatchAny is the sentinel ifMatch value that asserts "row must not already exist". Mirrors the HTTP `If-None-Match: *` semantic used by the future PUT handler (§4.1).

View Source 
const MaxHandoffQueuedPerAgent = 100

MaxHandoffQueuedPerAgent caps the hub-side queue-and-forward buffer per agent. Enforced inside the enqueue transaction so concurrent enqueues cannot overshoot. 100 is comfortably above any realistic "typed while the laptop was closed" burst while still bounding the DB row count and the reconnect drain burst on the holder.

View Source 
const UserSenderID = "user"

UserSenderID is the reserved agent_id for messages posted by the human user (the same constant exists in v0's internal/agent package). Posts under this id skip the member-of-group check and don't FK against the agents table — the repo recognizes the sentinel explicitly.

Variables

View Source 
var ErrETagMismatch = errors.New("store: etag mismatch")

ErrETagMismatch is returned by Update helpers when an If-Match check fails. Callers should treat this as a 412 Precondition Failed at the API layer.

View Source 
var ErrFencingMismatch = errors.New("store: fencing token mismatch")

ErrFencingMismatch signals that the supplied fencing_token does not match the row currently in agent_locks. Used by every agent-runtime write path so a peer that lost its lock cannot keep appending after the lease expired and another peer took over (split-brain prevention).

View Source 
var ErrHandoffPending = errors.New("store: blob_refs row is mid-handoff (handoff_pending=1)")

ErrHandoffPending is returned by InsertOrReplaceBlobRef when the existing row's handoff_pending flag is set. A device-switch (§3.7) has marked this URI as mid-handoff, and accepting a new body would silently overwrite the digest the orchestrator committed the target peer to pull. Callers above the store layer (blob.Store.Put → agent-runtime writes) should surface this as 409 to whatever client tried the write; the operator can finish or abort the handoff and retry.

View Source 
var ErrHandoffQueueFull = errors.New("store: handoff queue full for agent")

ErrHandoffQueueFull is returned by EnqueueHandoffQueuedMessage when the per-agent cap is reached. The HTTP layer maps it to 429.

View Source 
var ErrIdempotencyConflict = errors.New("store: idempotency-key reused for a different request")

ErrIdempotencyConflict is returned when the supplied key exists but the request_hash does not match — i.e. the caller reused the same Idempotency-Key for a *different* request. Per RFC standards for idempotency the correct response is 409 Conflict; surfacing a distinct sentinel lets the HTTP layer differentiate this from a generic "key exists" case.

View Source 
var ErrIdempotencyInFlight = errors.New("store: idempotency-key request still in flight")

ErrIdempotencyInFlight is returned when the supplied key matches an existing row whose ResponseStatus is still 0 — i.e. another worker is currently executing this request. Callers should respond 409 rather than re-execute (which would defeat the dedup).

View Source 
var ErrLockHeld = errors.New("store: agent lock held by another peer")

ErrLockHeld signals that AcquireAgentLock found a live lock held by a different peer. Callers (Hub) handle this as "the agent is busy on peer X" and surface it to the UI; the standard remediation is to wait for lease expiry then retry.

View Source 
var ErrNotFound = errors.New("store: not found")

ErrNotFound is returned by repo Get/Update helpers when no row matches.

View Source 
var ErrOplogOpIDReused = errors.New("store: op_id reused with different fingerprint or agent")

ErrOplogOpIDReused is returned when an idempotency probe finds a ledger row whose fingerprint or agent_id doesn't match the caller's claim. Indicates a peer-side bug; surfaces as a per-entry error in the receive handler.

View Source 
var ErrRestoreSuperseded = errors.New("store: RestoreBlobRef: row was modified concurrently; leaving current state")

ErrRestoreSuperseded is returned by RestoreBlobRef when the optimistic-concurrency check (rec.ExpectedCurrentSHA256) does NOT match the row's current sha256. A concurrent abort / complete / scrub repaired or advanced the row while the blob layer was inside its commit window; restoring our stale snapshot would undo their work, so the helper refuses. The blob layer treats this as "leave the row alone" and the next scrub pass picks up any residual inconsistency.

View Source 
var ErrStaleHead = errors.New("store: stale head (group head advanced)")

ErrStaleHead is returned by AppendGroupDMMessage when the caller's ExpectedLatestSeq does not match the current head — used by the v0 PostMessage CAS path to surface concurrent-reply conflicts to the UI.

Functions

func CanonicalETag 

func CanonicalETag(version int, canonicalRecord any) (string, error)

CanonicalETag computes the etag string used by every domain table.

Format: "<version>-<sha256(canonical_record)[:8]>"

canonicalRecord must be a value safely JSON-encodable. Callers are responsible for assembling the table-specific subset of fields that the design doc lists for each table (3.3, "etag canonical_record" table).

This helper guarantees:

  • Map keys are sorted (canonical JSON), so re-encoding the same logical record always yields the same digest regardless of struct field order.
  • Floating-point and unsupported types fail loudly via the underlying encoder rather than silently producing a divergent digest.

func IsValidWorkspaceFileKind 

func IsValidWorkspaceFileKind(kind WorkspaceFileKind) bool

IsValidWorkspaceFileKind reports whether kind is one of the workspace-file kinds the table accepts. Mirrors validMemoryKinds in memory.go — adding a new kind requires both a schema migration and an update here so the contract stays visible in code review.

func MaxSupportedSchemaVersion 

func MaxSupportedSchemaVersion() (int, error)

MaxSupportedSchemaVersion returns the highest migration version this build can apply. Callers (the startup gate, snapshot tools) compare it against an on-disk schema_version to detect "DB is from a newer build than this binary" — in which case starting up could silently mix incompatible state. Reads the embedded migration list at call time so the answer is always in sync with the binary's own migrations.

func NewHandoffQueuedMessageID added in v0.111.0 

func NewHandoffQueuedMessageID() (string, error)

NewHandoffQueuedMessageID mints a queue row id. Exported so the HTTP layer can pre-generate the id BEFORE the synchronous forward attempt: the same id (via its derived idempotency key) then covers both the initial forward and any later drain redelivery, so a "processed but connection dropped before the response" forward followed by enqueue cannot double-inject.

func NextGlobalSeq 

func NextGlobalSeq() int64

NextGlobalSeq returns a strictly increasing 64-bit value, suitable as the `seq` column for tables that partition seq globally. Resolution is 1ms but every call advances by at least 1, so within a tight loop seq still increments.

func NowMillis 

func NowMillis() int64

NowMillis returns the current UTC time in epoch milliseconds. All timestamps in kojo's structured store use this representation (3.2 in the design doc). Callers that need to inject a fake clock should do so at the boundary of their handler rather than monkey-patching this function.

func RecordEvent 

func RecordEvent(ctx context.Context, tx *sql.Tx, table, id, etag string, op EventOp, ts int64) (int64, error)

RecordEvent appends one row to the events table inside the caller's tx. Seq is allocated from NextGlobalSeq() — the SAME source domain rows draw from, so a peer can use any seq it has seen as a cursor here without per-table bookkeeping. ts == 0 picks NowMillis().

Callers should invoke this AFTER the domain mutation succeeds in the tx but BEFORE commit, so a transient DB error rolls both back. For inserts/updates pass the post-write etag; for deletes pass "".

Returns the allocated seq so the caller can echo it back into a process-local broadcast (eventbus.Event.Seq) — the broadcast then matches exactly what /changes?since=<seq> will return for the same row, eliminating any "did the peer already see this?" ambiguity.

func SHA256Hex 

func SHA256Hex(body []byte) string

SHA256Hex returns the lowercase hex sha256 digest of body. Used by agent_memory.body_sha256 and blob_refs.sha256.

Types

type AgentInsertOptions 

type AgentInsertOptions struct {
	Now       int64 // 0 = NowMillis()
	Seq       int64 // 0 = NextGlobalSeq()
	CreatedAt int64 // 0 = Now
	UpdatedAt int64 // 0 = Now
	PeerID    string
	// AllowOverwrite lets UpsertAgentPersona blindly replace an existing
	// live persona without an If-Match etag. Reserved for the v0→v1
	// importer and daemon-internal callers that already serialize against
	// per-agent state. API handlers MUST leave this false and supply
	// ifMatchETag instead.
	AllowOverwrite bool
}

AgentInsertOptions lets callers override timestamps/seq for tests and the v0→v1 importer (which must preserve the original CreatedAt to keep the change feed's "happened-before" relation intact).

type AgentLockRecord 

type AgentLockRecord struct {
	AgentID        string
	HolderPeer     string
	FencingToken   int64
	LeaseExpiresAt int64
	AcquiredAt     int64
	// AllowedProxyPeer names the signer that may drive proxied
	// requests on /api/v1/agents/{id}/* against this host while
	// the lock is live. Mirrors the §3.7 device-switch model: the
	// host that handed the runtime here is the only legitimate
	// orchestrator. Holder == self acquires set this to holder
	// itself; device-switch transfer sets it to the source peer;
	// force-reclaim sets it to whoever is reclaiming (self).
	AllowedProxyPeer string
}

AgentLockRecord mirrors one row of `agent_locks`. Per agent_id at most one row exists (PK is agent_id); the row records which peer currently holds the lock plus the fencing_token paired with the holder. Every agent-runtime write must present this token alongside its op so a delayed write from an old holder is rejected after lease expiry — see docs §3.7 fencing.

type AgentMemoryInsertOptions 

type AgentMemoryInsertOptions struct {
	Now            int64
	Seq            int64
	CreatedAt      int64
	UpdatedAt      int64
	PeerID         string
	LastTxID       *string
	AllowOverwrite bool
	Fencing        *FencingPredicate
	Idempotency    *IdempotencyTag
}

AgentMemoryInsertOptions follows the same shape as AgentInsertOptions. Fencing makes the upsert atomic with an agent_locks holder check; Idempotency makes a replayed op_id short-circuit to the prior etag.

type AgentMemoryRecord 

type AgentMemoryRecord struct {
	AgentID    string
	Body       string
	BodySHA256 string
	LastTxID   *string

	Seq       int64
	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

AgentMemoryRecord mirrors the `agent_memory` table. The body is the denormalized copy of the global blob `<v1>/global/agents/<id>/MEMORY.md`; the daemon syncs the file into this row using the intent-file protocol (§2.5). LastTxID identifies the originating write transaction; NULL means the row was loaded from a CLI-direct write before the daemon caught up.

type AgentPersonaRecord 

type AgentPersonaRecord struct {
	AgentID    string
	Body       string
	BodySHA256 string

	Seq       int64
	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

AgentPersonaRecord mirrors the `agent_persona` table. Body is the canonical persona text; body_sha256 is recomputed on every update so the change feed can de-dupe no-op writes upstream.

type AgentRecord 

type AgentRecord struct {
	ID          string
	Name        string
	PersonaRef  string
	Settings    map[string]any
	WorkspaceID string

	// common columns
	Seq       int64
	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

AgentRecord mirrors the `agents` table. Settings carries every backend/UI preference flag that doesn't deserve its own column; the DB stores the raw JSON so older binaries can preserve forward-introduced keys via round-trip (the "soft" forward-compat the design doc calls for in §3.3).

type AgentSyncPayload 

type AgentSyncPayload struct {
	Agent          *AgentRecord
	Persona        *AgentPersonaRecord         // nil = no persona on source
	Memory         *AgentMemoryRecord          // nil = no MEMORY.md on source
	Messages       []*MessageRecord            // empty = clear (full mode) OR no new rows (incremental mode)
	MemoryEntries  []*MemoryEntryRecord        // empty = clear (full mode) OR no new rows (incremental mode)
	Tasks          []*AgentTaskRecord          // empty = clear target's tasks
	WorkspaceFiles []*AgentWorkspaceFileRecord // empty = clear (full mode) OR no new rows (incremental mode)

	// IncrementalMessages, when true, switches syncMessagesTx
	// from "delete-then-insert" (source-wins full replace) to
	// "upsert" (UPSERT supplied rows, leave the rest alone).
	// The §3.7 incremental device-switch orchestrator sets this
	// after fetching target's max(agent_messages.seq) and
	// shipping only rows newer than that — the existing rows on
	// target are bit-identical to source's by virtue of seq +
	// etag, so DELETE would just waste I/O. ON CONFLICT(id)
	// keeps the path idempotent across retries.
	IncrementalMessages bool

	// IncrementalMemoryEntries is the analogue for memory_entries.
	IncrementalMemoryEntries bool
}

AgentSyncPayload is the bundle of rows a §3.7 device-switch pushes from source to target so the agent's runtime can continue on the new home_peer. The orchestrator builds this on the source side from its own kojo.db; the target applies it via SyncAgentFromPeer inside one transaction.

Empty slices / nil pointers are honoured: a target that receives no Persona / Memory / Messages / MemoryEntries / WorkspaceFiles clears its corresponding rows (the agent existed on source without that surface, so target should mirror).

type AgentSyncState 

type AgentSyncState struct {
	Known bool `json:"known"`
	// MaxMessageSeq tracks max(agent_messages.seq) for the agent
	// on target. Used as the seq cursor for the §3.7 incremental
	// device-switch protocol — source ships rows with seq > this.
	// Append-only transcripts (no edits / no truncates) are the
	// common case and this cursor is sufficient; mixed-mode
	// transcripts trigger a full-replace downgrade on source via
	// HasNonAppendOnlyMessages.
	MaxMessageSeq int64 `json:"max_message_seq"`
	// MaxMemoryEntrySeq is reported for diagnostics but NOT used
	// as a delta cursor — memory_entries allow body updates +
	// soft-deletes + recreations on the same seq, so the
	// orchestrator keys off MaxMemoryEntryUpdatedAt instead.
	MaxMemoryEntrySeq int64 `json:"max_memory_entry_seq"`
	// MaxMemoryEntryUpdatedAt is the cursor the orchestrator
	// uses for memory_entries delta. Every InsertMemoryEntry /
	// UpdateMemoryEntry / SoftDeleteMemoryEntry bumps updated_at,
	// so source's `updated_at >= this` filter catches every
	// mutation target hasn't observed — including tombstones and
	// recreations. (`>=`, not `>`, defends against same-
	// millisecond mutations colliding on this cursor; every row
	// sharing the cursor timestamp — one or more — gets
	// idempotently re-shipped via the receiver's
	// ON CONFLICT(id) DO UPDATE.) The handler runs
	// IncrementalMemoryEntries when the wire
	// SinceMemoryEntryUpdatedAt > 0.
	//
	// Includes tombstoned rows (deleted_at != NULL) — their
	// updated_at IS the soft-delete instant, and the delta MUST
	// see them so target can mirror the deletion. A
	// COALESCE-style "live max only" would silently drop
	// tombstones whose updated_at exceeds the latest live row.
	MaxMemoryEntryUpdatedAt int64 `json:"max_memory_entry_updated_at"`
	// NOTE: workspace files (agent_workspace_files) intentionally do
	// NOT have a max-updated-at cursor here. They are tiny per-agent
	// singletons (≤ 2 rows: user.md, checkin.md), so the bytes saved
	// by an incremental delta are negligible — and incremental mode
	// risks SILENT data loss across peer clock skew (source's
	// updated_at is behind target's cursor → the delta skips the
	// edit). The orchestrator always full-ships workspace files; the
	// DELETE-then-INSERT in syncWorkspaceFilesTx covers the
	// tombstone-bump case without needing a cursor.
	AgentETag   string `json:"agent_etag,omitempty"`
	PersonaETag string `json:"persona_etag,omitempty"`
	MemoryETag  string `json:"memory_etag,omitempty"`
}

AgentSyncState is the snapshot a target peer reports back to a device-switch orchestrator so the source can send only the delta the target doesn't already have. Returned by GetAgentSyncState and surfaced over POST /api/v1/peers/agent-sync/state.

Known == false signals the target has never seen this agent — the orchestrator must do a full sync (every field below is 0/"").

Max*Seq are the largest seq on each table for the agent; the orchestrator filters source rows with `seq > Max*Seq` and ships only those, halving payload size on every switch after the first. *ETag fields let the orchestrator skip retransmitting agents / agent_persona / agent_memory bodies that already match; the source still ships task rows in full because they're small and routinely mutated.

type AgentTaskInsertOptions 

type AgentTaskInsertOptions struct {
	Now       int64
	CreatedAt int64
	UpdatedAt int64
	PeerID    string
}

AgentTaskInsertOptions lets callers (notably the v0→v1 importer) preserve original timestamps and override the clock for tests.

Seq is intentionally not exposed here: both CreateAgentTask and BulkInsertAgentTasks allocate seq from MAX(seq)+1 inside the transaction. Caller-supplied seq would race against parallel writers for the same agent; serialization on the SQLite write lock + the unique (agent_id, seq) constraint is the canonical guard. The v0 → v1 importer doesn't need to preserve original task ordering either: tasks have no inherent timeline (unlike messages) and seq is a stable-sort tiebreaker, not part of the row's identity.

type AgentTaskListOptions 

type AgentTaskListOptions struct {
	Status string // "" = all statuses
	Limit  int    // 0 = unbounded
	Cursor int64  // seq strictly greater (keyset paging); 0 = from start
}

AgentTaskListOptions configures ListAgentTasks.

type AgentTaskPatch 

type AgentTaskPatch struct {
	Title      *string
	Body       *string
	Status     *string
	DueAt      *int64
	ClearDueAt bool
}

AgentTaskPatch supports partial updates. nil = leave unchanged. To clear DueAt, set ClearDueAt=true (a nil DueAt is otherwise read as "leave alone" — Go has no tri-state pointer-of-pointer convention here).

type AgentTaskRecord 

type AgentTaskRecord struct {
	ID      string
	AgentID string
	Seq     int64 // per-agent monotonic
	Title   string
	Body    string // optional, may be empty
	Status  string // 'pending'|'in_progress'|'done'|'cancelled'
	DueAt   *int64

	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

AgentTaskRecord mirrors the `agent_tasks` table.

type AgentWorkspaceFileInsertOptions 

type AgentWorkspaceFileInsertOptions struct {
	Now            int64
	Seq            int64
	CreatedAt      int64
	UpdatedAt      int64
	PeerID         string
	AllowOverwrite bool
	Fencing        *FencingPredicate
	Idempotency    *IdempotencyTag
}

AgentWorkspaceFileInsertOptions mirrors AgentMemoryInsertOptions minus LastTxID. Fencing makes the upsert atomic with an agent_locks holder check; Idempotency makes a replayed op_id short-circuit to the prior etag.

type AgentWorkspaceFileRecord 

type AgentWorkspaceFileRecord struct {
	AgentID    string
	Kind       WorkspaceFileKind
	Body       string
	BodySHA256 string

	Seq       int64
	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

AgentWorkspaceFileRecord mirrors the `agent_workspace_files` table. The body is the denormalized copy of the on-disk file under `<v1>/global/agents/<id>/<kind>.md`; the daemon syncs the file into this row (and back out again) using the workspace-sync reconcile path.

Unlike AgentMemoryRecord this struct intentionally has no LastTxID column: workspace files are user-driven content (the user edits user.md or checkin.md directly), not CLI-tx outputs that need to be tied back to a specific oplog write transaction. The §2.5 intent-file protocol does not apply.

type BlobRefInsertOptions 

type BlobRefInsertOptions struct {
	CreatedAt int64
	UpdatedAt int64
	// AllowHandoffPending lets the §3.7 target-side pull
	// (peer.PullClient via blob.Store.Put with
	// BypassHandoffPending=true) update a row whose existing
	// handoff_pending flag would otherwise refuse the write.
	// Reserved for the orchestrator's pull path; agent-runtime
	// writes MUST leave this false so the §3.7 invariant holds.
	//
	// v1 trust model: the orchestrator (Hub) is the only
	// authority that runs begin/complete/abort, so concurrent
	// handoffs on the same URI can't happen and an unconditional
	// UPDATE is safe. A future multi-orchestrator slice would
	// need to thread an expected-pre-state token through the
	// bypass to refuse "I expected source X but the row says Y".
	AllowHandoffPending bool
}

BlobRefInsertOptions tunes timestamps for InsertOrReplaceBlobRef. Mirrors the pattern used by other Insert* helpers so v0→v1 importers can preserve original CreatedAt values.

type BlobRefRecord 

type BlobRefRecord struct {
	URI            string
	Scope          string
	HomePeer       string
	Size           int64
	SHA256         string // hex
	Refcount       int64
	PinPolicy      string // raw JSON, empty = NULL
	LastSeenOK     int64  // unix millis, 0 = NULL
	MarkedForGCAt  int64  // unix millis, 0 = NULL
	HandoffPending bool
	CreatedAt      int64
	UpdatedAt      int64

	// ExpectedCurrentSHA256 / ExpectedCurrentUpdatedAt are
	// consumed ONLY by RestoreBlobRef to enforce optimistic-
	// concurrency: the restore refuses if either disagrees with
	// the row's current state. (sha256, updated_at) together
	// catch every meaningful concurrent mutation — sha256 moves
	// on body replace, updated_at moves on every write including
	// handoff_pending flips and scrub timestamp bumps that keep
	// sha256 stable. Both fields must be set to enable the OCC
	// gate; either empty disables it.
	ExpectedCurrentSHA256    string
	ExpectedCurrentUpdatedAt int64
}

BlobRefRecord mirrors one row of the `blob_refs` table. The blob URI is `kojo://<scope>/<path>` and is the primary key. Bodies live on the filesystem under internal/blob; this row is the canonical metadata cache so Head / IfMatch / List avoid reading the full body to recompute sha256.

Refcount, PinPolicy, LastSeenOK, MarkedForGCAt and HandoffPending are reserved for later phases (pin / dedup, scrub repair, device handoff). Phase 3 slice 2 keeps refcount at 1 for every row and leaves the rest at their default zero values; slice 3+ will use them. They are surfaced on the record so the schema and the API type stay aligned across slices.

type CompleteHandoffResult 

type CompleteHandoffResult struct {
	// Lock is the agent_locks row state after the tx. nil when no
	// lock row exists for this agent (transient pre-acquire state).
	Lock *AgentLockRecord
	// LockTransferred reports whether holder_peer + fencing_token
	// moved as part of this call. False on the idempotent re-call
	// (holder already at target) — the row state in Lock still
	// reflects the converged target.
	LockTransferred bool
	// SwitchedURIs is the set of blob_refs URIs whose home_peer
	// just flipped to target (and whose handoff_pending cleared).
	SwitchedURIs []string
	// AlreadyAtTargetURIs is the set of URIs in the agent's prefix
	// range that were already at target before this call —
	// idempotent re-call detection.
	AlreadyAtTargetURIs []string
	// LeftoverURIs is the set of URIs in the agent's prefix range
	// whose state is neither "switched by us" nor "already at
	// target" (e.g. handoff_pending=0 but home_peer is still the
	// source). The orchestrator surfaces these so an operator
	// knows complete did not converge every row.
	LeftoverURIs []string
}

CompleteHandoffResult bundles the outcome of one atomic device- switch complete. agent_locks transfer + every blob_refs.home_peer flip run in ONE tx; SwitchedURIs / AlreadyAtTargetURIs let the orchestrator render the per-row report without a follow-up read.

type EventListener 

type EventListener func(EventRecord)

EventListener is invoked AFTER a tx that recorded one or more events table rows commits successfully. The listener fires once per recorded row, with the same seq the row holds in the events table — so a peer receiving a live broadcast can match it 1:1 against a `/api/v1/changes?since=<seq>` poll.

Listeners run synchronously on the writer's goroutine; they MUST be fast and non-blocking. A typical implementation forwards to an in-memory broadcast bus and returns.

type EventOp 

type EventOp string

EventOp is the verb persisted in events.op. Restricted to the three values the schema CHECK constraint accepts. 

const (
	EventOpInsert EventOp = "insert"
	EventOpUpdate EventOp = "update"
	EventOpDelete EventOp = "delete"

	ChangesKVNamespace     = "changes"
	ChangesKVPrunedThrough = "events_pruned_through"
)

type EventRecord 

type EventRecord struct {
	Seq   int64
	Table string
	ID    string
	ETag  string // empty for delete
	Op    EventOp
	TS    int64 // unix millis
}

EventRecord is one durable row from the events table. It mirrors the WebSocket broadcast payload shape (eventbus.Event) so a peer that caught up via /changes?since=<seq> can replay the same logic it uses for live frames.

type ExternalChatCursorInsertOptions 

type ExternalChatCursorInsertOptions struct {
	Now       int64
	CreatedAt int64
	UpdatedAt int64
	PeerID    string
}

ExternalChatCursorInsertOptions lets the v0→v1 importer preserve original timestamps and override the clock for tests. PeerID is recorded on every imported row: cursors are global-scoped (other peers must see the same cursor to avoid re-fetching the same external history on device switch) but the row remembers which peer last advanced it.

type ExternalChatCursorRecord 

type ExternalChatCursorRecord struct {
	ID        string
	Source    string  // 'slack' | 'discord' | ...
	AgentID   *string // nullable
	ChannelID *string // nullable
	Cursor    string  // opaque to kojo

	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

ExternalChatCursorRecord mirrors the `external_chat_cursors` table.

id is a composite identifier — by convention the v0→v1 importer composes it as "<agent_id>:<source>:<channel_id>:<thread_id>" for per-thread cursors (the only shape v0 produces today; channel-level rollups are not cursor-driven and are not imported). The schema carries channel_id as a regular column (no dedicated index — agent_id is the only secondary index, see migration 0001); thread_id lives only inside the composite id. Runtime callers that resume polling reuse the same composition so a re-imported row matches an in-flight live row by primary key.

agent_id is nullable to accommodate a future deployment-scoped integration (e.g. a system-wide bot account) that wouldn't be tied to one agent. Every v0 cursor today is per-agent, so all imported rows have agent_id populated.

channel_id is nullable for the same reason — a future cursor for a non- channel-shaped source (DM list, PM stream) may not have a channel id. Every v0 cursor today corresponds to a Slack channel and populates this column.

cursor is opaque to kojo — the polling plugin sets it (Slack: the latest numeric ts; Discord: a snowflake; etc.) and the runtime hands it back unchanged on the next poll.

external_chat_cursors has no `seq` column: cursor freshness ordering is implicit in updated_at and the WS invalidator uses the global event log.

type FencingPredicate 

type FencingPredicate struct {
	AgentID      string
	Peer         string
	FencingToken int64
}

FencingPredicate makes a write atomic with the agent_lock holder check (docs/multi-device-storage.md §3.5 — "agent-runtime write requires lock holder peer + matching fencing token"). When passed through *InsertOptions to a write helper that supports it, the helper opens BEGIN IMMEDIATE, runs CheckFencingTx as the first statement, and gates the write on it; concurrent peers that stole the lock between the caller's read and the write see ErrFencingMismatch instead of a silently-committed stale write.

Zero value (or nil pointer) skips the check — single-Hub callers that already hold their own lock guarantee can opt out.

type GroupDMDeadLetter added in v0.111.0 

type GroupDMDeadLetter struct {
	ID        int64  `json:"id"`
	GroupDMID string `json:"groupdmId"`
	AgentID   string `json:"agentId"`
	Reason    string `json:"reason"`
	Payload   string `json:"payload,omitempty"`
	Attempts  int    `json:"attempts"`
	CreatedAt int64  `json:"createdAt"`
}

GroupDMDeadLetter is a permanently failed notification delivery. Kept as an audit record instead of silently dropping the batch. Payload is the rendered notification text (may be truncated by the caller).

type GroupDMInsertOptions 

type GroupDMInsertOptions struct {
	Now       int64
	Seq       int64
	CreatedAt int64
	UpdatedAt int64
	PeerID    string
	// AllowDeadMembers skips the "every member must be a live agent"
	// check. Reserved for the v0→v1 importer where a group's
	// groups.json may reference an agent_id that has since vanished
	// from agents.json — without this escape hatch, the entire group
	// (plus its transcript) would have to be dropped to import the
	// rest. API handlers MUST leave this false.
	AllowDeadMembers bool
}

GroupDMInsertOptions follows the AgentInsertOptions shape.

type GroupDMMember 

type GroupDMMember struct {
	AgentID      string `json:"agent_id"`
	NotifyMode   string `json:"notify_mode,omitempty"`   // ""|"realtime"|"digest"
	DigestWindow int    `json:"digest_window,omitempty"` // seconds, only meaningful with digest mode
}

GroupDMMember is one entry in members_json. Per-member notify/digest settings travel here so the v0→v1 importer doesn't lose calibration data before Phase 4 normalizes membership into a separate table.

type GroupDMMessageInsertOptions 

type GroupDMMessageInsertOptions struct {
	Now       int64
	CreatedAt int64
	UpdatedAt int64
	Seq       int64
	PeerID    string
	// ExpectedLatestSeq, when non-zero, requires the current per-group
	// head to equal this value — otherwise ErrStaleHead is returned. The
	// API layer uses it to map v0's expectedLatestMessageId conflict
	// flow. ExpectedLatestSeq=0 skips the check (importer/admin path).
	//
	// v0 keys CAS by the head *message id* not seq — the API layer
	// resolves the id via LatestGroupDMMessageID before calling this.
	// We deliberately keep the repo on seq because seq is monotonic and
	// indexed by the UNIQUE(groupdm_id,seq) constraint; the id mapping
	// is a single SELECT on the way in and stays in the API layer.
	ExpectedLatestSeq int64
	// RequireExpectedLatestSeq enforces the head comparison even when
	// ExpectedLatestSeq is 0, making "I expect the room to be EMPTY"
	// expressible: two concurrent first posts to an empty room then
	// collide (the loser gets ErrStaleHead) instead of both passing.
	RequireExpectedLatestSeq bool
	// AllowNonMember lets the caller post under an agent_id that is not
	// in members_json. Reserved for the v0→v1 importer where a member
	// has since been removed but historical messages must round-trip,
	// and for tests. The UserSenderID sentinel is allowed regardless.
	AllowNonMember bool
	// AllowMissingAuthor relaxes the live-author check for the importer:
	// a v0 group transcript may reference an agent_id that has since been
	// hard-deleted, but the message body still needs to round-trip. With
	// this flag set, AppendGroupDMMessage will not query agents and will
	// not enforce membership. Combine with AllowNonMember when both
	// constraints would otherwise block the import. API handlers must
	// leave both flags false.
	AllowMissingAuthor bool
}

GroupDMMessageInsertOptions matches MessageInsertOptions, plus ExpectedLatestSeq for compare-and-set head protection.

type GroupDMMessageListOptions 

type GroupDMMessageListOptions struct {
	Limit     int
	BeforeSeq int64
	SinceSeq  int64
	Order     string // "asc"|"desc"
}

GroupDMMessageListOptions matches MessageListOptions.

type GroupDMMessageRecord 

type GroupDMMessageRecord struct {
	ID          string
	GroupDMID   string
	Seq         int64  // per-group
	AgentID     string // "" for system messages (NULL in DB)
	Content     string
	Attachments json.RawMessage
	// Hop is the agent-relay depth: 0 for user/system/fresh agent posts,
	// trigger-hop + 1 for posts made from a notification-triggered turn.
	Hop int
	// Mentions is a JSON array of mentioned member ids ("user" for the
	// human operator). nil when the message mentions nobody.
	Mentions json.RawMessage
	// Usage is a JSON object of token usage for an agent thread reply
	// (mirrors agent.Usage). nil for user/system posts and agent posts
	// made outside a thread turn.
	Usage json.RawMessage
	// Thinking is the extended-thinking text for an agent thread reply.
	// "" for user/system posts and agent posts made outside a thread turn.
	Thinking string
	// ToolUses is a JSON array of tool calls (mirrors []agent.ToolUse) for
	// an agent thread reply. nil for user/system posts and agent posts
	// made outside a thread turn.
	ToolUses json.RawMessage
	// Model/Effort record the agent's configured model and effort level at
	// the time of an agent thread reply. "" for user/system posts and agent
	// posts made outside a thread turn.
	Model  string
	Effort string
	// Interrupted marks a thread reply whose turn ended before completion
	// (operator stop, error, or timeout) — the content is partial output.
	// false for user/system posts and agent posts made outside a thread turn.
	Interrupted bool

	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

GroupDMMessageRecord mirrors the `groupdm_messages` table.

type GroupDMRecord 

type GroupDMRecord struct {
	ID       string
	Name     string
	Members  []GroupDMMember
	Style    string
	Cooldown int
	Venue    string
	// MaxHops caps agent-to-agent notification relay depth for this room.
	// 0 means "use the built-in default" (agent.defaultMaxHops).
	MaxHops int
	// Kind is "group" (default) or "dm" (first-class 1:1 room).
	Kind string

	Seq       int64
	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

GroupDMRecord mirrors the `groupdms` table.

Caveat: members_json may contain agent_ids that have since been soft-deleted. The repo layer does not filter them out at read time because the etag canonical record covers the unfiltered list — masking dead members here would let a returned record's etag drift from the stored row. Phase 4 normalizes this into a groupdm_members FK table with a cascade-on-tombstone trigger; until then, callers that surface the member list to the UI should overlay a live-agents lookup.

type HandoffQueuedMessage added in v0.111.0 

type HandoffQueuedMessage struct {
	ID         string
	AgentID    string
	HolderPeer string
	Content    string
	CreatedAt  int64
	Status     string
}

HandoffQueuedMessage mirrors one row of handoff_queued_messages (migration 0023). HolderPeer is the holder at enqueue time — informational only; the drain re-resolves the current holder from agent_locks.

type IdempotencyEntry 

type IdempotencyEntry struct {
	Key            string
	OpID           string
	RequestHash    string
	ResponseStatus int
	ResponseEtag   string
	ResponseBody   string
	ExpiresAt      int64
}

IdempotencyEntry mirrors one row of `idempotency_keys`. The table is the 24-hour dedup window for write-API retries (3.5) and op-log replay (3.13.1): a client that retries the same write under the same Idempotency-Key gets back the prior response instead of risking a double-execute, and an op-log replay during Hub failover (3.6 / 3.13.1) uses the same gate to suppress duplicate apply.

`RequestHash` captures method + canonical path + body sha256 so a client that reuses an Idempotency-Key for a *different* request is caught (409 Conflict) rather than served a stale cached response.

`ResponseStatus == 0` is a sentinel for "claim row inserted, handler is still running". A second concurrent request with the same key must NOT execute the handler again — it returns 409 ("in flight"). When the handler finishes the row is rewritten with the real status / etag / body.

type IdempotencyTag 

type IdempotencyTag struct {
	OpID        string
	Fingerprint string
}

IdempotencyTag is the op-log applied-ledger key + fingerprint the store records alongside a write. When non-nil on a write helper's Idempotency field, the function:

  1. Probes oplog_applied(OpID) inside its tx. - hit + fingerprint match: returns the saved etag without re-running the write. - hit + fingerprint MISMATCH: returns ErrOplogOpIDReused (a peer-side bug — the same op_id was used for two different (table, op, body) tuples). - hit + agent_id mismatch: returns ErrOplogOpIDReused for the same reason — the ledger is keyed on op_id, and the claimed agent must match the ledger row.
  2. Runs the write.
  3. Records oplog_applied(OpID, agent_id, fingerprint, etag) in the same tx so a crash between dispatch commit and ledger write is impossible.

Fingerprint is opaque to the store — the caller (oplog_handler) computes sha256 over (table || sep || op || sep || body) so a replay with the same op_id but a different body is detected.

type KVPutOptions 

type KVPutOptions struct {
	// IfMatchETag, when non-empty, requires the existing row's etag to
	// match. Empty matches any (idempotent overwrite). The literal
	// string "*" is reserved for "row must not exist" checks.
	IfMatchETag string
	// Now overrides the wall clock for tests. 0 = NowMillis().
	Now int64
}

KVPutOptions narrows the Put path.

type KVRecord 

type KVRecord struct {
	Namespace      string
	Key            string
	Value          string // plaintext; empty when Secret is true
	ValueEncrypted []byte // envelope ciphertext; nil when Secret is false
	Type           KVType
	Secret         bool
	Scope          KVScope
	Version        int64
	ETag           string
	CreatedAt      int64
	UpdatedAt      int64
}

KVRecord is one kv row. Exactly one of Value (plaintext) and ValueEncrypted (envelope-protected) is set per row, gated by Secret.

type KVScope 

type KVScope string

KVScope mirrors the kv.scope CHECK constraint. Validated at Put/Get boundaries so a typo can't slip a misclassified row past the schema check on insert (the DB would reject it, but a clean Go-level error is friendlier). 

const (
	KVScopeGlobal  KVScope = "global"
	KVScopeLocal   KVScope = "local"
	KVScopeMachine KVScope = "machine"
)

type KVType 

type KVType string

KVType mirrors the kv.type CHECK constraint. 

const (
	KVTypeString KVType = "string"
	KVTypeJSON   KVType = "json"
	KVTypeBinary KVType = "binary"
)

type ListBlobRefsOptions 

type ListBlobRefsOptions struct {
	// Scope filters to one scope; "" returns every scope.
	Scope string
	// URIPrefix filters to URIs that HasPrefix(uri, URIPrefix). The
	// caller is expected to build the full `kojo://<scope>/<path>`
	// form; this helper does no scope-prefixing of its own so a
	// caller can list across scopes when Scope == "".
	URIPrefix string
	// IncludeMarkedForGC: false hides rows whose marked_for_gc_at is
	// non-null. Slice 2 always passes false; the scrub job (slice 3+)
	// passes true to enumerate sweep candidates.
	IncludeMarkedForGC bool
	// Limit caps the row count. 0 = unlimited.
	Limit int
}

ListBlobRefsOptions tunes ListBlobRefs.

type ListEventsResult 

type ListEventsResult struct {
	Events []EventRecord
	// NextSince is the seq value the caller should pass as ?since on
	// the next poll. Equal to the largest Seq in Events, or the
	// caller-supplied since when Events is empty (so a poll loop never
	// rewinds).
	NextSince int64
	// Watermark is the smallest seq currently present in the events
	// table.
	Watermark int64
	// PrunedThrough is the largest seq that retention has deleted from
	// events. A caller whose since cursor is below this floor missed rows
	// and must full-resync.
	PrunedThrough int64
}

ListEventsResult is the body of a /changes?since=<seq> response.

type ListEventsSinceOptions 

type ListEventsSinceOptions struct {
	// Table, when non-empty, filters to a single domain. Defaults to
	// all tables.
	Table string
	// Limit caps the number of rows returned. <= 0 picks 500. The
	// caller is expected to page by passing the largest seq from the
	// previous batch as the next ?since.
	Limit int
}

ListEventsSinceOptions narrows the cursor read.

type ListPeersOptions 

type ListPeersOptions struct {
	// Status filters to a single status value. Empty returns every row.
	Status string
	// Limit caps the row count. 0 = unlimited.
	Limit int
}

ListPeersOptions tunes ListPeers.

type MemoryEntryInsertOptions 

type MemoryEntryInsertOptions struct {
	Now         int64
	CreatedAt   int64
	UpdatedAt   int64
	Seq         int64 // 0 = allocate next per-agent
	PeerID      string
	Fencing     *FencingPredicate
	Idempotency *IdempotencyTag
}

MemoryEntryInsertOptions lets the importer override id/seq/ timestamps. Fencing, when non-nil, makes the insert atomic with an agent_locks holder check (see FencingPredicate).

type MemoryEntryListOptions 

type MemoryEntryListOptions struct {
	Kind   string // "" = all kinds
	Limit  int    // 0 = unbounded
	Cursor int64  // seq strictly greater (keyset paging); 0 = from start

	// IncludeDeleted: include soft-deleted rows in the result.
	// Default false (callers see only live entries). The §3.7
	// incremental device-switch path sets true so source can
	// ship tombstones to target; target's syncMemoryEntriesTx
	// upserts them and the row becomes deleted on target too.
	IncludeDeleted bool

	// UpdatedAtSince: return only rows with updated_at >= this.
	// 0 = no filter. `>=` is inclusive — defends against
	// same-millisecond mutations colliding on the boundary,
	// trading idempotent resend of every row sharing the cursor
	// timestamp (typically 1, but a burst of mutations could
	// share a single millisecond) for full coverage.
	// Used by the §3.7 device-switch delta (memory_entries
	// cursor) because seq alone misses in-place body updates
	// and soft-deletes. Order flips to updated_at ASC when this
	// is > 0 so tombstones precede recreations on the same
	// (agent_id, kind, name) — the alive UNIQUE index requires
	// the old row to be tombstoned BEFORE the new row's INSERT
	// lands on target.
	UpdatedAtSince int64
}

MemoryEntryListOptions configures ListMemoryEntries.

type MemoryEntryPatch 

type MemoryEntryPatch struct {
	Kind *string
	Name *string
	Body *string
	// Fencing, when non-nil, makes the patch atomic with an
	// agent_locks holder check (see FencingPredicate). The
	// predicate's agent_id must match the entry's agent_id; the
	// store re-reads the row inside the tx and verifies the
	// match before applying the patch.
	Fencing *FencingPredicate
	// Idempotency makes the patch crash-safe across the dispatch
	// commit boundary — a prior matching ledger row short-
	// circuits the update.
	Idempotency *IdempotencyTag
}

MemoryEntryPatch supports partial updates. Pass nil to leave a field unchanged.

type MemoryEntryRecord 

type MemoryEntryRecord struct {
	ID         string
	AgentID    string
	Seq        int64 // per-agent
	Kind       string
	Name       string
	Body       string
	BodySHA256 string

	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

MemoryEntryRecord mirrors the `memory_entries` table. Body lives both here (canonical) and on the filesystem under `<v1>/global/agents/<id>/memory/<kind>/<name>.md` — the blob mirror is wired up in Phase 3.

type MessageInsertOptions 

type MessageInsertOptions struct {
	Now       int64
	CreatedAt int64
	UpdatedAt int64
	Seq       int64 // 0 = allocate next per-agent
	PeerID    string
	// Fencing, when non-nil, makes the insert atomic with an
	// agent_locks holder check. The store runs CheckFencingTx as
	// the first statement of the same BEGIN IMMEDIATE tx the
	// INSERT lives in, so a peer that stole the lock between the
	// caller's check and the write surfaces as ErrFencingMismatch
	// instead of a committed stale-token write. The op-log
	// replay path threads its (peer, token) here for each entry.
	Fencing *FencingPredicate
	// Idempotency, when non-nil, gates the write on the
	// oplog_applied ledger inside the same tx — a matching row
	// short-circuits the write and returns the saved etag, a
	// mismatched row surfaces as ErrOplogOpIDReused, and a
	// missing row leads to a fresh write + ledger insert at the
	// end of the tx. Used by the op-log receive handler so
	// replays are crash-safe across the dispatch commit boundary.
	Idempotency *IdempotencyTag
}

MessageInsertOptions lets the v0→v1 importer preserve original timestamps and seq, and lets tests inject fixed clocks.

type MessageListOptions 

type MessageListOptions struct {
	// Limit caps the number of returned rows (after pagination). 0 = no cap.
	Limit int
	// BeforeSeq returns only rows with seq < BeforeSeq. 0 = no upper bound.
	// Used for keyset pagination ("older messages") matching the v0 jsonl
	// "before <id>" semantics — but we use seq, not id, because seq is
	// monotonic and survives ID re-issuance during the importer.
	BeforeSeq int64
	// SinceSeq returns only rows with seq > SinceSeq. 0 = no lower bound.
	// Used by the WebSocket invalidation feed (Phase 4) to backfill clients.
	SinceSeq int64
	// Order: "asc" returns oldest-first, "desc" newest-first. Empty = "asc".
	Order string
	// IncludeDeleted: include soft-deleted rows. Default false.
	IncludeDeleted bool
}

MessageListOptions configures ListMessages.

type MessagePatch 

type MessagePatch struct {
	Content     *string
	Thinking    *string
	ToolUses    json.RawMessage // nil = unchanged; empty []byte("null") to clear
	Attachments json.RawMessage
	Usage       json.RawMessage
}

UpdateMessageContent rewrites the content/thinking/tool_uses of a message in place, bumping version+etag. Only fields the caller specifies via non-nil pointers are touched; passing all nils is a no-op (returns the existing record).

type MessageRecord 

type MessageRecord struct {
	ID          string
	AgentID     string
	Seq         int64 // per-agent seq, monotonic
	Role        string
	Content     string
	Thinking    string
	ToolUses    json.RawMessage // nil-safe: empty == NULL
	Attachments json.RawMessage
	Usage       json.RawMessage

	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

MessageRecord mirrors the `agent_messages` table. Tool/attachment/usage payloads are kept as opaque JSON so message-shape upgrades don't require a schema migration; downstream consumers are expected to validate their own shape on read.

type OplogAppliedRecord 

type OplogAppliedRecord struct {
	OpID        string
	AgentID     string
	Fingerprint string
	ResultETag  string
	AppliedAt   int64
}

OplogAppliedRecord mirrors one row of the oplog_applied ledger. docs/multi-device-storage.md §3.13.1 — recorded inside the dispatch tx so a peer's retry returns the saved etag without re-running the write.

type Options 

type Options struct {
	// ConfigDir is the directory that will hold kojo.db. If empty, callers
	// should pass configdir.Path() explicitly — store has no implicit
	// dependency on configdir to keep this package testable.
	ConfigDir string

	// Path overrides ConfigDir/DBFileName. Used by tests and snapshots.
	Path string

	// ReadOnly opens the database read-only. Migrations are NOT applied.
	ReadOnly bool
}

Options configures Open. All fields are optional.

type PeerPendingRecord 

type PeerPendingRecord struct {
	DeviceID  string
	Name      string
	URL       string
	NodeKey   string
	FirstSeen int64
	LastSeen  int64
}

PeerPendingRecord mirrors one row of `peer_pending` — a peer that has called Hub's POST /api/v1/peers/join-request but is still awaiting Owner approval. Approve moves the row into `peer_registry` and drops the pending row. Reject drops it only (peer is free to re-request).

first_seen / last_seen are unix millis. NodeKey is the Tailscale stable NodeKey of the requesting peer, observed by the Hub via LocalClient.WhoIs on the inbound HTTP request — the peer never sends it. The handler refuses a repeat /join-request whose device_id binds to a different NodeKey than the one originally recorded (the operator must explicitly delete the stale row to rotate identity).

type PeerRecord 

type PeerRecord struct {
	DeviceID string
	// Name is the human-readable device label (OS hostname by default).
	// Operator-overridable from the UI; agents address peers by Name
	// rather than URL because the dial address (`<host>:<port>` or
	// `http://...`) is meaningless to a human and changes when the
	// network topology shifts.
	Name string
	// URL is the dial address other peers reach this row on. The
	// registrar stamps it from tsnet's FQDN (Hub) or the Tailscale
	// IPv4 (--peer). Empty until the daemon has been started at least
	// once so the listener bound its port.
	URL string
	// NodeKey is the Tailscale stable NodeKey of this peer
	// (`nodekey:...`). Empty when the row was imported before
	// migration 0013 or when the registering peer hadn't observed its
	// own NodeKey yet. The tsnet identity middleware refuses to admit
	// requests whose WhoIs-resolved NodeKey doesn't match a row here,
	// so a NULL/empty column effectively quarantines the row until
	// the next join-request re-stamps it.
	NodeKey  string
	LastSeen int64 // unix millis, 0 = NULL
	Status   string
}

PeerRecord mirrors one row of the `peer_registry` table. device_id is a stable GUID minted by the peer the first time it joins the cluster. Per docs/peer-tsnet-identity.md, identity is anchored on the Tailscale stable NodeKey (column `node_key`), looked up via tsnet.LocalClient.WhoIs on every incoming inter-peer request.

`Status` is one of the schema's CHECK values: 'online' | 'offline'. The Hub flips it to 'offline' after a heartbeat-miss threshold (3.7).

type PushSubscriptionInsertOptions 

type PushSubscriptionInsertOptions struct {
	Now       int64
	CreatedAt int64
	UpdatedAt int64
}

PushSubscriptionInsertOptions narrows the bulk-insert path. There is intentionally no PeerID field: the schema has no peer_id column for this table (see §3.3 exception above). Now / CreatedAt / UpdatedAt follow the same precedence pattern as the other importers — caller record value wins, then opts, then the clock.

type PushSubscriptionRecord 

type PushSubscriptionRecord struct {
	Endpoint       string
	DeviceID       *string // nullable; v0 never set it
	UserAgent      *string // nullable; v0 never set it
	VAPIDPublicKey string  // copied from vapid.json at import time
	P256dh         string
	Auth           string
	ExpiredAt      *int64 // tombstone marker; nil = active
	CreatedAt      int64
	UpdatedAt      int64
}

PushSubscriptionRecord mirrors the `push_subscriptions` table.

§3.3 exception (documented in 0001_initial.sql line 367-378): this table does NOT carry version/etag/seq/deleted_at/peer_id. The endpoint URL is the entire identity (the user agent produces it; nothing kojo writes would optimistic-lock against), liveness is signalled by ExpiredAt (set on 401/410 from the push provider — a tombstone-without-soft- delete), and audit-by-device is supplanted by DeviceID copied straight from peer_registry. A row therefore has no etag to recompute on import, no seq to allocate against the global counter, and no peer_id to stamp.

DeviceID and UserAgent are nullable: v0's push_subscriptions.json never stored either column (webpush.Subscription only carries endpoint+keys), so every imported row leaves both NULL. Live runtime that subscribes a new browser may populate them once the future API surfaces them.

type RemoteMirrorMessage 

type RemoteMirrorMessage struct {
	ID          string
	Role        string
	Content     string
	Thinking    string
	ToolUses    []byte // JSON, opaque (nil 可)
	Attachments []byte // JSON, opaque (nil 可)
	Usage       []byte // JSON, opaque (nil 可)
	Timestamp   string // RFC3339
}

RemoteMirrorMessage は mirror テーブル 1 行の wire 表現。 agent.Message と field 名・型を揃えており、handlers 層で agent.Message にそのままコピーできる。

type SessionInsertOptions 

type SessionInsertOptions struct {
	Now       int64
	CreatedAt int64
	UpdatedAt int64
	PeerID    string
}

SessionInsertOptions lets the v0→v1 importer preserve original timestamps and override the clock for tests.

Seq is intentionally not exposed: BulkInsertSessions allocates seq via NextGlobalSeq(), the same atomic CAS counter that backs every global-partition table (agents / agent_persona / agent_memory / groupdms / sessions / events). Caller-supplied seq would diverge from the global counter and let two writers across different tables produce the same seq value.

type SessionRecord 

type SessionRecord struct {
	ID        string
	AgentID   *string // nullable — sessions can be detached / pre-agent
	Status    string  // 'running' | 'stopped' | 'archived'
	PID       *int64
	Cmd       string
	WorkDir   string
	StartedAt *int64
	StoppedAt *int64
	ExitCode  *int64

	Seq       int64
	Version   int
	ETag      string
	CreatedAt int64
	UpdatedAt int64
	DeletedAt *int64
	PeerID    string
}

SessionRecord mirrors the `sessions` table. The session row is local- scoped (per-peer PTY state) — peer_id records which physical device launched / archived this session so cross-device snapshot inspection can attribute the row to its origin. The schema enforces a global `id TEXT PRIMARY KEY`, so two peers cannot legally allocate the same id; if a snapshot merge surfaces a collision the merger must reject one side rather than silently overwrite. peer_id is disambiguating metadata, not a tiebreaker.

type Store 

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

Store wraps the SQLite database handle and exposes high-level helpers used by feature packages. The zero value is not usable; obtain one via Open.

func Open 

func Open(ctx context.Context, opts Options) (*Store, error)

Open creates the config directory if needed, opens kojo.db with WAL and pragmas tuned for kojo's workload, and applies any pending migrations.

On any error after the underlying *sql.DB has been opened, the handle is closed before returning so callers do not leak descriptors.

func (*Store) AbandonIdempotencyKey 

func (s *Store) AbandonIdempotencyKey(ctx context.Context, key, claimToken string) error

AbandonIdempotencyKey deletes the claim row when the handler panicked / context-cancelled before producing a saveable response. Scoped to (key, claim_token) so a stale handler can't drop a fresh claim's pending row. Idempotent — a missing or already-completed row returns nil.

func (*Store) AcquireAgentLock 

func (s *Store) AcquireAgentLock(ctx context.Context, agentID, peer string, now, leaseDuration int64) (*AgentLockRecord, error)

AcquireAgentLock takes the lock for (agentID, peer) under a fresh fencing_token. The semantics:

  • If no row exists yet, INSERT with a freshly minted token from agent_fencing_counters (1 for the very first acquisition, monotonically advancing for every subsequent post-Release re-acquisition).
  • If a row exists held by `peer` (re-acquire from same holder), refresh lease_expires_at and KEEP the existing fencing_token — the holder's prior writes stay valid.
  • If a row exists held by another peer:
  • lease still alive → return ErrLockHeld with the current row so the caller can show "busy on peer X" without a second SELECT.
  • lease expired → reassign to `peer` and INCREMENT the counter so any in-flight write from the old holder gets rejected.

All branches run inside one BEGIN IMMEDIATE tx to serialize concurrent acquisitions; the writer-lock-up-front recipe is the same one AppendMessage / etc. use to dodge SQLITE_BUSY_SNAPSHOT under WAL.

`now` is injected so callers (the Hub's lease scheduler) can drive deterministic clock for tests; pass NowMillis() in production. `leaseDuration` is the milliseconds the lease is granted for; the schema is clock-agnostic so the resulting lease_expires_at is `now + leaseDuration`.

func (*Store) AppendGroupDMMessage 

func (s *Store) AppendGroupDMMessage(ctx context.Context, rec *GroupDMMessageRecord, opts GroupDMMessageInsertOptions) (*GroupDMMessageRecord, error)

AppendGroupDMMessage inserts a new message at the next per-group seq.

Sender semantics:

  • rec.AgentID == "" → system message (NULL in DB)
  • rec.AgentID == UserSenderID → human-user post; bypasses member-of-group check, no FK against agents
  • otherwise → must be a live, member agent unless opts.AllowNonMember is set

func (*Store) AppendMessage 

func (s *Store) AppendMessage(ctx context.Context, rec *MessageRecord, opts MessageInsertOptions) (*MessageRecord, error)

AppendMessage inserts a new message at the next per-agent seq. The seq allocation runs inside the same transaction as the insert so concurrent appenders for the same agent serialize on the table's UNIQUE(agent_id,seq) constraint rather than racing.

Returns the inserted record with seq/etag/timestamps filled in.

func (*Store) ApprovePeerPending 

func (s *Store) ApprovePeerPending(ctx context.Context, deviceID string) (*PeerRecord, error)

ApprovePeerPending promotes the pending row keyed by device_id into peer_registry (carrying NodeKey across) and removes the pending row in one transaction. Returns ErrNotFound when no pending row matches.

func (*Store) BulkAppendMessages 

func (s *Store) BulkAppendMessages(ctx context.Context, agentID string, recs []*MessageRecord, opts MessageInsertOptions) (int, error)

BulkAppendMessages inserts a batch of messages for one agent inside a single transaction. Used by the v0→v1 importer for transcripts large enough that the per-row AppendMessage transaction overhead dominates — every AppendMessage opens its own BEGIN/COMMIT pair, which on SQLite means one fsync per message. A 100k-row transcript is roughly 100k fsyncs through AppendMessage; through BulkAppendMessages with a 5k-row chunk it is ~20.

Contract:

  • Every record's AgentID must equal agentID. A mismatch fails the batch before any insert so a misrouted slice can't poison the transcript.
  • Roles are validated against validRoles up front; an invalid role on any record fails the whole batch.
  • Seqs are allocated as MAX(seq)+1, +2, ... at the start of the transaction. A non-zero rec.Seq overrides allocation for that row; callers using explicit seqs are responsible for keeping the UNIQUE(agent_id, seq) constraint satisfied.
  • Per-row CreatedAt/UpdatedAt: when non-zero on the record they are honored verbatim, otherwise opts.CreatedAt/UpdatedAt are used, otherwise opts.Now, otherwise NowMillis(). This lets the importer preserve original v0 timestamps without filling MessageInsertOptions row-by-row.
  • All-or-nothing: any error mid-batch rolls the transaction back. A partial-success contract would leave the caller without a clean way to recover seqs after a crash.
  • On success the input records are mutated in place — Seq, Version, ETag, CreatedAt, UpdatedAt, PeerID, and the normalized JSON columns are filled in so callers can read the assigned values without an extra ListMessages round-trip.

Cross-agent id collisions still surface as a SQLite UNIQUE error on the PRIMARY KEY — the bulk path does not weaken that guard.

func (*Store) BulkInsertAgentTasks 

func (s *Store) BulkInsertAgentTasks(ctx context.Context, agentID string, recs []*AgentTaskRecord, opts AgentTaskInsertOptions) (int, error)

BulkInsertAgentTasks inserts many tasks for one agent in a single transaction. Used by the v0→v1 importer; live UI flows should use CreateAgentTask one at a time so per-task etags / events are emitted.

Idempotency: rows whose id already exists for *this same agent* are skipped silently. A row whose id is already present under a *different* agent surfaces as a hard error rather than a silent skip — that situation is a v0 data-integrity violation (task_<random> ids should not collide across agents) and corrupting the v1 store by attaching a different agent's row to this batch would be silent data loss.

Seq allocation: the per-row Seq is allocated inside the transaction from MAX(seq)+1 and advances only when an INSERT actually lands. Skipped duplicates (same-agent ids hit by the preload) do not burn seq values, so re-runs over the same input don't widen seq gaps. Caller-supplied r.Seq is ignored. This sidesteps the race where a parallel CreateAgentTask for the same agent could claim a seq the bulk caller wanted.

Returned count is the number of rows actually inserted (new for this run). After commit, each successfully-inserted record in `recs` is updated in-place with its canonical Seq / Version / ETag / timestamps; records that were skipped (duplicate id under same agent) are left untouched.

CreatedAt/UpdatedAt fallback: r.CreatedAt → opts.CreatedAt → now, then r.UpdatedAt → opts.UpdatedAt → CreatedAt. This matches the "explicit per-row > batch-level > clock" precedence the rest of the store layer follows.

func (*Store) BulkInsertExternalChatCursors 

func (s *Store) BulkInsertExternalChatCursors(ctx context.Context, recs []*ExternalChatCursorRecord, opts ExternalChatCursorInsertOptions) (int, error)

BulkInsertExternalChatCursors inserts many external_chat_cursor rows in a single transaction. It is used by the v0→v1 importer; runtime cursor writers still own their source-specific storage rather than this table.

Idempotency contract: rows whose id already exists are skipped via ON CONFLICT DO NOTHING + a preload-set so a re-run leaves the existing row untouched. Caller records are mutated in place AFTER commit with assigned etag/timestamps; skipped rows are left untouched so callers can distinguish "imported now" from "already there".

AgentID=&"" and ChannelID=&"" are normalized to nil on the staged copy so the canonical ETag (computed from the staged record) agrees with the round-tripped record read back from SQL (NULL → nil pointer).

func (*Store) BulkInsertPushSubscriptions 

func (s *Store) BulkInsertPushSubscriptions(ctx context.Context, recs []*PushSubscriptionRecord, opts PushSubscriptionInsertOptions) (int, error)

BulkInsertPushSubscriptions inserts many push_subscriptions rows in a single transaction. Used by the v0→v1 importer; live subscribe/ unsubscribe goes through dedicated single-row APIs.

Idempotency contract matches BulkInsertSessions: rows whose endpoint already exists are skipped via ON CONFLICT DO NOTHING + a preload-set so a re-run leaves the existing row untouched. Caller records are mutated in place AFTER commit with assigned timestamps; skipped rows are left untouched so callers can distinguish "imported now" (CreatedAt populated) from "already there" (CreatedAt zero) without re-querying.

DeviceID / UserAgent are normalized: a non-nil pointer to "" is rewritten to nil on the staged copy so the persisted NULL round-trips to a nil pointer on read-back. There is no etag to keep symmetric (no etag column), but consistent NULL handling avoids surprising callers that compare records pre- and post-insert.

func (*Store) BulkInsertSessions 

func (s *Store) BulkInsertSessions(ctx context.Context, recs []*SessionRecord, opts SessionInsertOptions) (int, error)

BulkInsertSessions inserts many session rows in a single transaction. Used by the v0→v1 importer; live runtime callers (PTY launch, status transitions) go through dedicated single-row APIs that emit per-row events. Same idempotency contract as BulkInsertAgentTasks: rows whose id already exists are skipped via ON CONFLICT DO NOTHING + a preload-set. Seq is allocated via NextGlobalSeq() (the global CAS counter shared across global- partition tables; see SessionInsertOptions.Seq comment).

Caller records are mutated in place AFTER commit with their assigned seq/etag/timestamps. Records skipped by the preload are left untouched.

func (*Store) CheckFencing 

func (s *Store) CheckFencing(ctx context.Context, agentID, peer string, fencingToken int64) error

CheckFencing returns nil iff agent_locks contains a row for agent_id with the given (peer, fencing_token). Convenience wrapper around CheckFencingTx using a short-lived read tx.

IMPORTANT: this performs a single SELECT, so the result is only valid until something else writes — there is a TOCTOU window between CheckFencing and the caller's subsequent write where another peer could steal the lock and bump the token. Production write paths MUST use CheckFencingTx inside the same BEGIN IMMEDIATE transaction that performs the write. CheckFencing exists for read-only / non- critical callers (audit endpoints, tests) where the window is tolerable.

func (*Store) CheckFencingTx 

func (s *Store) CheckFencingTx(ctx context.Context, tx *sql.Tx, agentID, peer string, fencingToken int64) error

CheckFencingTx is the tx-scoped variant write paths must use. The caller opens BEGIN IMMEDIATE (or s.db.BeginTx(ctx, nil) — kojo's DSN pins txlock=immediate so the writer lock is held up front), calls CheckFencingTx as the FIRST statement, then issues its INSERT / UPDATE / DELETE inside the same tx. Because BEGIN IMMEDIATE has already taken the WAL writer lock, no other writer can steal the agent_locks row between the check and the gated write.

Returns ErrNotFound if no row exists, ErrFencingMismatch if the row exists but the predicate doesn't match.

func (*Store) ClaimIdempotencyKey 

func (s *Store) ClaimIdempotencyKey(ctx context.Context, key, opID, requestHash string, expiresAt int64) (*IdempotencyEntry, error)

ClaimIdempotencyKey atomically reserves the key for the caller. It inserts a "pending" row (response_status=0, response_etag/body NULL) when no live row exists for `key`, and returns (nil, nil) — the caller should run the handler and then call FinalizeIdempotencyKey.

When a live row is already present:

  • request_hash mismatch → ErrIdempotencyConflict
  • request_hash match, response_status == 0 → ErrIdempotencyInFlight
  • request_hash match, response_status != 0 → returns the saved entry; the caller short-circuits the handler and replays the stored response verbatim.

"Live" means expires_at > now (we evaluate it inside the SQL so the behaviour matches the index-driven sweep). Expired rows are transparently overwritten so a delayed retry beyond the dedup window correctly re-executes (per RFC the only client guarantee is "within the dedup window").

The atomicity rests on a single statement (INSERT … ON CONFLICT DO UPDATE WHERE expires_at <= now), which SQLite serializes at the database level — two concurrent claim attempts cannot both win.

func (*Store) ClearPeerNodeKey 

func (s *Store) ClearPeerNodeKey(ctx context.Context, deviceID string) error

ClearPeerNodeKey sets the node_key column to NULL for the given device_id. Used on Hub-mode startup to drop a stale value left by a previous binary that wrote tsnet.Server's NodeKey into the self-row; the post-clear value is then refilled with the host tailscaled NodeKey via the OS LocalAPI capture goroutine. UpsertPeer treats empty NodeKey as "no change", so this dedicated path is the only way to actively wipe the column without touching the other columns.

Returns ErrNotFound when no row matches.

func (*Store) Close 

func (s *Store) Close() error

Close flushes WAL and closes the database. Idempotent.

func (*Store) CompleteHandoff 

func (s *Store) CompleteHandoff(ctx context.Context, agentID, targetPeer, blobURIPrefix string, leaseDurationMs int64) (*CompleteHandoffResult, error)

CompleteHandoff atomically transfers the agent_lock from its current holder to targetPeer AND switches every blob_refs row in the agent's prefix (kojo://global/agents/<id>/) to home_peer= target AND clears handoff_pending. All three mutations run in ONE transaction so a crash between them rolls back to the pre- call state — no half-completed handoff can survive a daemon restart.

Idempotency: when the lock is ALREADY at targetPeer the fencing_token is NOT re-bumped (that would invalidate the target's current writes); the handler echoes the existing row and only the blob_refs loop runs.

Returns ErrFencingMismatch when the lock exists but its (holder, token) tuple doesn't match a freshly-read current state — a concurrent abort / steal raced us. Returns ErrNotFound when the agent has no agent_locks row at all AND no blob_refs rows in the prefix; callers treat that as "no state to migrate" and surface it as a 404. A no-lock-but-blobs case proceeds normally (blobs switch, LockTransferred=false).

func (*Store) CompleteHandoffSelectedBlobs added in v0.101.7 

func (s *Store) CompleteHandoffSelectedBlobs(ctx context.Context, agentID, targetPeer string, blobURIs []string, leaseDurationMs int64) (*CompleteHandoffResult, error)

CompleteHandoffSelectedBlobs atomically transfers the agent_lock and switches only the supplied blob_refs URIs. It is used by the device-switch orchestrator for the small set of portable binary state that must follow the agent (currently avatar blobs) while leaving historical delivery artifacts such as attach/ on their original home peer.

func (*Store) CountHandoffQueuedMessages added in v0.111.0 

func (s *Store) CountHandoffQueuedMessages(ctx context.Context, agentID string) (int, error)

CountHandoffQueuedMessages returns the queue depth for one agent.

func (*Store) CountMessages 

func (s *Store) CountMessages(ctx context.Context, agentID string) (int64, error)

CountMessages returns the number of live messages for agentID. Returns 0 for soft-deleted agents (cascade-on-tombstone via the JOIN).

func (*Store) CreateAgentTask 

func (s *Store) CreateAgentTask(ctx context.Context, rec *AgentTaskRecord, opts AgentTaskInsertOptions) (*AgentTaskRecord, error)

CreateAgentTask inserts a new task at the next per-agent seq. Seq allocation runs inside the same transaction as the insert so concurrent creators serialize on the table's UNIQUE(agent_id, seq) constraint rather than racing.

func (*Store) DB 

func (s *Store) DB() *sql.DB

DB returns the underlying *sql.DB. Feature packages should keep usage scoped to short transactions and prefer the helpers in this package over raw SQL where possible so cross-cutting concerns (etag, seq) stay consistent.

func (*Store) DeleteAllAgentTasks 

func (s *Store) DeleteAllAgentTasks(ctx context.Context, agentID string) (int64, error)

DeleteAllAgentTasks hard-deletes every task row for agentID. Used by the data-reset flow that nukes an agent's persistent state. Tombstones are removed too — the operation is "vacate this agent's task slate" regardless of liveness. Bypasses etag because the data-reset flow owns the agent for the duration of the call.

func (*Store) DeleteBlobRef 

func (s *Store) DeleteBlobRef(ctx context.Context, uri string) error

DeleteBlobRef removes the row keyed by uri. Returns nil even if the row was already absent so callers can drive idempotent cleanup loops without sniffing ErrNotFound. The on-disk body is the caller's responsibility (internal/blob.Delete handles both halves under one pathLock).

func (*Store) DeleteBlobRefIfMatches 

func (s *Store) DeleteBlobRefIfMatches(ctx context.Context, uri, sha256 string, updatedAt int64) (bool, error)

DeleteBlobRefIfMatches removes the row keyed by uri only when its current (sha256, updated_at) match the supplied tuple. blob.Store uses this on the rollback path when its prior Snapshot saw NO row: a 3rd-party writer that recreated the URI between Snapshot and rollback would otherwise be silently deleted. Returns (deleted bool, err error): deleted=false with err=nil means the row state moved on (or the row was already absent) and the caller should leave it alone.

func (*Store) DeleteHandoffQueuedMessage added in v0.111.0 

func (s *Store) DeleteHandoffQueuedMessage(ctx context.Context, agentID, id string) error

DeleteHandoffQueuedMessage removes one queued row (delivered or operator-cancelled). agentID is part of the key so an id guessed across agents cannot cancel another agent's message. Returns ErrNotFound when no row matches.

func (*Store) DeleteKV 

func (s *Store) DeleteKV(ctx context.Context, namespace, key, ifMatchETag string) error

DeleteKV removes the row. Empty ifMatchETag → idempotent (missing row is not an error). Non-empty ifMatchETag → conditional: missing row or etag mismatch surfaces as ErrETagMismatch so the caller can refetch + retry. Mirrors the SoftDeleteAgentMemory contract for tombstone-style endpoints.

func (*Store) DeletePeer 

func (s *Store) DeletePeer(ctx context.Context, deviceID string) error

DeletePeer removes the peer_registry row and the matching pending row in a single transaction. With the Bearer-issuance flow retired (docs/peer-tsnet-identity.md) there is no longer any token or kv stash to clean up — identity is anchored on the NodeKey column and disappears with the row.

Idempotent: missing rows return nil.

Callers driving a "decommission" flow should also audit any agent_locks rows whose holder_peer == deviceID; releasing those is a separate operation (see ReleaseAgentLockByPeer).

func (*Store) DeletePeerPending 

func (s *Store) DeletePeerPending(ctx context.Context, deviceID string) error

DeletePeerPending removes a pending row by device_id. Idempotent; returns nil even when no row matches (Reject must be a no-op when a concurrent Approve already promoted the row).

func (*Store) DeleteRemoteMirrorForAgent 

func (s *Store) DeleteRemoteMirrorForAgent(ctx context.Context, agentID string) (int64, error)

DeleteRemoteMirrorForAgent は agent の mirror 行を全て削除する。 帰還時 (holder==self になった瞬間) と ReleaseAgentLocally で呼ぶ。 戻り値は削除された行数。

func (*Store) EnqueueHandoffQueuedMessage added in v0.111.0 

func (s *Store) EnqueueHandoffQueuedMessage(ctx context.Context, agentID, holderPeer, content string) (*HandoffQueuedMessage, error)

EnqueueHandoffQueuedMessage appends one message to the agent's queue, enforcing MaxHandoffQueuedPerAgent inside a transaction. Returns ErrHandoffQueueFull when the cap is reached.

func (*Store) EnqueueHandoffQueuedMessageWithID added in v0.111.0 

func (s *Store) EnqueueHandoffQueuedMessageWithID(ctx context.Context, id, agentID, holderPeer, content string) (*HandoffQueuedMessage, error)

EnqueueHandoffQueuedMessageWithID is EnqueueHandoffQueuedMessage with a caller-supplied id (empty → minted here); see NewHandoffQueuedMessageID for why callers pre-generate.

func (*Store) ExpireIdempotencyKeys 

func (s *Store) ExpireIdempotencyKeys(ctx context.Context, cutoff int64) (int, error)

ExpireIdempotencyKeys deletes rows whose expires_at <= cutoff. Returns the count deleted. Run periodically (1× / hour is fine — the dedup window is 24 h) so the table doesn't grow unbounded.

func (*Store) FinalizeIdempotencyKey 

func (s *Store) FinalizeIdempotencyKey(ctx context.Context, key, claimToken string, status int, etag, body string) error

FinalizeIdempotencyKey rewrites the pending row with the handler's response. Called once the wrapped handler has produced its full status/body. The expires_at is left intact (set at claim time).

`claimToken` MUST be the value the caller passed as `opID` to ClaimIdempotencyKey — the WHERE clause scopes the UPDATE to the exact claim row. Without that scope, an old handler whose claim expired and was overwritten by a fresh claim could finalize over the new claim's pending row, replaying its response under the new requester's expectations.

Returns ErrNotFound when the row is missing or no longer matches the claim token — surfaces an actionable failure rather than a silent no-op.

func (*Store) FindMemoryEntryByName 

func (s *Store) FindMemoryEntryByName(ctx context.Context, agentID, kind, name string) (*MemoryEntryRecord, error)

FindMemoryEntryByName returns the live entry for (agent_id, kind, name). Used by the v0→v1 importer (which keys entries by filesystem path) and by the merge-queue resolver.

func (*Store) ForceReclaimAgentToLocal 

func (s *Store) ForceReclaimAgentToLocal(ctx context.Context, agentID, localPeerID string, now, leaseDurationMs int64) (*AgentLockRecord, error)

ForceReclaimAgentToLocal restores ONE agent's distributed state to a "this host owns the runtime" baseline atomically. Every row the §3.7 device-switch ever flipped — agent_locks, blob_refs, handoff kv markers — converges to local-self in a single transaction so a partial failure either updates every table or none. Designed to be the SOLE recovery path that operators (or the startup reconciler) drive when a switch left state in an unrecoverable shape:

  • agent_locks: row upserted with holder_peer = localPeerID, allowed_proxy_peer = localPeerID, fencing_token bumped via agent_fencing_counters (so any in-flight write from the former holder is invalidated), lease refreshed.
  • blob_refs `kojo://global/agents/<id>/%`: home_peer rewritten to localPeerID, handoff_pending cleared. Without this the next device-switch attempt would 409 wrong_source because blob_refs still claims the agent lives elsewhere.
  • kv namespace="handoff":
  • `released/<id>` marker deleted (otherwise startup eviction throws the just-reclaimed runtime away).
  • `arrived/<id>` marker deleted (stale arrival from prior half-finished switch would mask the next legitimate one).
  • `pending/<id>/*` sealed agent-sync tokens deleted (a stranded sync entry would let a retry of the prior orchestrator step double-commit).

Caller MUST refresh the in-memory cache + side channels after this call returns; the store has no view into the runtime layer.

`now` is injected for tests; pass NowMillis() in production. `leaseDurationMs` follows AgentLockLeaseDuration to match what AgentLockGuard's refresh loop expects.

func (*Store) GetAgent 

func (s *Store) GetAgent(ctx context.Context, id string) (*AgentRecord, error)

GetAgent returns the agent by id. Returns ErrNotFound if missing or soft-deleted (callers wanting to inspect tombstones must use a dedicated helper — there is none yet because no caller needs that).

func (*Store) GetAgentLock 

func (s *Store) GetAgentLock(ctx context.Context, agentID string) (*AgentLockRecord, error)

GetAgentLock returns the row for agent_id or ErrNotFound.

func (*Store) GetAgentMemory 

func (s *Store) GetAgentMemory(ctx context.Context, agentID string) (*AgentMemoryRecord, error)

GetAgentMemory returns the MEMORY.md row for agentID, or ErrNotFound. Filters out soft-deleted parents the same way persona/messages reads do.

func (*Store) GetAgentPersona 

func (s *Store) GetAgentPersona(ctx context.Context, agentID string) (*AgentPersonaRecord, error)

GetAgentPersona returns the persona for agentID, or ErrNotFound. The join against agents enforces the parent-alive invariant on read so callers can't resurrect a soft-deleted agent's persona via a stale id reference.

func (*Store) GetAgentSyncState 

func (s *Store) GetAgentSyncState(ctx context.Context, agentID string) (*AgentSyncState, error)

GetAgentSyncState reads the per-agent high-water marks the §3.7 incremental device-switch protocol needs.

The function is read-only and safe to call from any peer's kojo.db. Returns Known=false (with zero everything else) when the agents row is missing or deleted — that's the "first-time sync" signal for the orchestrator.

All four queries run in a single DB connection (no transaction — they are read-only and SQLite's snapshot isolation under journal_mode=WAL gives consistent reads across them without holding a write lock).

func (*Store) GetAgentTask 

func (s *Store) GetAgentTask(ctx context.Context, id string) (*AgentTaskRecord, error)

GetAgentTask returns a single live task by id. ErrNotFound on miss, tombstone, or tombstoned parent agent.

func (*Store) GetAgentWorkspaceFile 

func (s *Store) GetAgentWorkspaceFile(ctx context.Context, agentID string, kind WorkspaceFileKind) (*AgentWorkspaceFileRecord, error)

GetAgentWorkspaceFile returns the workspace file row for (agentID, kind), or ErrNotFound when the row is missing OR tombstoned. Filters out soft-deleted parents the same way persona/memory reads do — a tombstoned agent has no readable workspace files.

func (*Store) GetBlobRef 

func (s *Store) GetBlobRef(ctx context.Context, uri string) (*BlobRefRecord, error)

GetBlobRef returns the row keyed by uri or ErrNotFound.

func (*Store) GetExternalChatCursor 

func (s *Store) GetExternalChatCursor(ctx context.Context, id string) (*ExternalChatCursorRecord, error)

GetExternalChatCursor returns the row by id. ErrNotFound on miss.

func (*Store) GetGroupDM 

func (s *Store) GetGroupDM(ctx context.Context, id string) (*GroupDMRecord, error)

GetGroupDM returns the live group DM by id.

Caveat: members_json may contain agent_ids that have since been soft-deleted. The repo layer does not filter them out at read time because the etag canonical record covers the unfiltered list — masking dead members here would let a returned record's etag drift from the stored row. The proper fix is Phase 4's normalized groupdm_members table + cascade-on-tombstone trigger; until then, callers that surface the member list to the UI should overlay a live-agents lookup themselves.

func (*Store) GetGroupDMReadCursor added in v0.111.0 

func (s *Store) GetGroupDMReadCursor(ctx context.Context, groupID string) (seq int64, ok bool, err error)

GetGroupDMReadCursor returns the persisted read-cursor seq for a room. ok is false when no cursor has been recorded yet (the room has never been marked read on any device).

func (*Store) GetHandoffQueuedMessage added in v0.111.0 

func (s *Store) GetHandoffQueuedMessage(ctx context.Context, agentID, id string) (*HandoffQueuedMessage, error)

GetHandoffQueuedMessage returns one queued row or ErrNotFound. The drain re-checks row existence right before delivery so an owner cancel that landed after the drain's list snapshot wins.

func (*Store) GetKV 

func (s *Store) GetKV(ctx context.Context, namespace, key string) (*KVRecord, error)

GetKV returns the row by composite key. Returns ErrNotFound when no row matches.

func (*Store) GetMemoryEntry 

func (s *Store) GetMemoryEntry(ctx context.Context, id string) (*MemoryEntryRecord, error)

GetMemoryEntry returns a single live memory entry by id. ErrNotFound on miss, on tombstone, or on tombstoned parent agent.

func (*Store) GetMessage 

func (s *Store) GetMessage(ctx context.Context, id string) (*MessageRecord, error)

GetMessage returns a single message by id. Returns ErrNotFound on miss OR on a soft-deleted parent agent. Cascade-on-tombstone is enforced via the JOIN rather than by mass UPDATE on SoftDeleteAgent — a 100k-message agent would otherwise pay a heavy delete cost.

func (*Store) GetOplogApplied 

func (s *Store) GetOplogApplied(ctx context.Context, opID string) (*OplogAppliedRecord, error)

GetOplogApplied returns the ledger row for opID. Returns ErrNotFound when no row exists.

func (*Store) GetPeer 

func (s *Store) GetPeer(ctx context.Context, deviceID string) (*PeerRecord, error)

GetPeer returns the row keyed by device_id or ErrNotFound.

func (*Store) GetPeerByNodeKey 

func (s *Store) GetPeerByNodeKey(ctx context.Context, nodeKey string) (*PeerRecord, error)

GetPeerByNodeKey returns the registry row whose node_key column matches `nodeKey`. Used by the tsnet identity middleware to translate the WhoIs-resolved NodeKey into a Principal.PeerID. Empty nodeKey is rejected so a row with a NULL/empty column can never match a caller that failed WhoIs resolution.

func (*Store) GetPeerPending 

func (s *Store) GetPeerPending(ctx context.Context, deviceID string) (*PeerPendingRecord, error)

GetPeerPending returns the row keyed by device_id or ErrNotFound.

func (*Store) GetPeerPendingByNodeKey 

func (s *Store) GetPeerPendingByNodeKey(ctx context.Context, nodeKey string) (*PeerPendingRecord, error)

GetPeerPendingByNodeKey returns the pending row whose node_key column matches `nodeKey`. Used by the join-request handler to detect a repeat POST from the same Tailscale node arriving against a different device_id (= NodeKey collision; handler rejects with 409 so the operator can clean up the old row).

func (*Store) GetPushSubscription 

func (s *Store) GetPushSubscription(ctx context.Context, endpoint string) (*PushSubscriptionRecord, error)

GetPushSubscription returns the row by endpoint. ErrNotFound on miss.

func (*Store) GetSession 

func (s *Store) GetSession(ctx context.Context, id string) (*SessionRecord, error)

GetSession returns the row by id. ErrNotFound on miss. Used by tests and audit tools; the live PTY runtime keeps its own in-memory state.

func (*Store) HasNonAppendOnlyMessages 

func (s *Store) HasNonAppendOnlyMessages(ctx context.Context, agentID string) (bool, error)

HasNonAppendOnlyMessages returns true when any row in agent_messages for the agent shows a sign of in-place mutation — either soft-deleted (deleted_at IS NOT NULL) or edited (version > 1 / regenerate / transcript edit). The §3.7 incremental device-switch orchestrator uses this to downgrade to full-replace mode: a seq-cursor delta would skip both kinds of mutation (the seq is preserved across tombstone and edit) and let target keep showing stale transcript that source has updated. The bool is conservative — false ONLY when every row is in its original append state.

LIMIT 1 + EXISTS-shape semantics keep this O(1) against the agent_id index even for agents with tens of thousands of messages.

func (*Store) InsertAgent 

func (s *Store) InsertAgent(ctx context.Context, rec *AgentRecord, opts AgentInsertOptions) (*AgentRecord, error)

InsertAgent writes a new agents row. Seq, ETag, CreatedAt, UpdatedAt are filled in from opts (or defaulted) and reflected back into the returned record so callers don't have to re-read.

func (*Store) InsertGroupDM 

func (s *Store) InsertGroupDM(ctx context.Context, rec *GroupDMRecord, opts GroupDMInsertOptions) (*GroupDMRecord, error)

InsertGroupDM creates a new group DM. Member agent_ids are validated against the live agents table — silently accepting a bogus member would surface as an empty avatar/profile in the UI later, hard to trace.

func (*Store) InsertGroupDMDeadLetter added in v0.111.0 

func (s *Store) InsertGroupDMDeadLetter(ctx context.Context, dl *GroupDMDeadLetter) error

InsertGroupDMDeadLetter records a permanently failed delivery.

func (*Store) InsertMemoryEntry 

func (s *Store) InsertMemoryEntry(ctx context.Context, rec *MemoryEntryRecord, opts MemoryEntryInsertOptions) (*MemoryEntryRecord, error)

InsertMemoryEntry creates a new memory entry under (agent_id, kind, name). The schema's partial unique index (idx_memory_entries_alive_natkey) blocks duplicates among live rows; resurrecting a previously soft-deleted entry with the same natural key requires the caller to hard-delete the tombstone first or to use UpsertMemoryEntry which handles the resurrection path.

func (*Store) InsertOrReplaceBlobRef 

func (s *Store) InsertOrReplaceBlobRef(ctx context.Context, rec *BlobRefRecord, opts BlobRefInsertOptions) (*BlobRefRecord, error)

InsertOrReplaceBlobRef writes rec into blob_refs. On URI conflict the existing row is overwritten in place (Put-style semantics — blob bodies are replace-on-write at the URI level; refcount stays at the table default until pin / dedup phases use it). The CreatedAt of the existing row is preserved so audit history holds.

Returns ErrHandoffPending when the existing row has handoff_pending=1. The §3.7 device-switch state machine requires both source and target to refuse agent-runtime writes against rows whose flag is set; this guard is the canonical enforcement point. SwitchBlobRefHome / SetBlobRefHandoffPending bypass it (they are the state-machine drivers themselves) and edit the row through dedicated targeted UPDATEs.

func (*Store) InsertPeerPendingIfAbsent 

func (s *Store) InsertPeerPendingIfAbsent(ctx context.Context, rec *PeerPendingRecord) (*PeerPendingRecord, bool, error)

InsertPeerPendingIfAbsent tries to insert a fresh pending row. Returns (inserted=true, …) when the row landed; (inserted=false, …) when a row already exists for that device_id and the existing state is returned untouched (no metadata overwrite, no node_key overwrite).

The handler is expected to use this on first contact. A concurrent race is serialised by SQLite at the row level; one caller wins as the inserter, the other receives inserted=false and is told to take the repeat path (which the handler gates on NodeKey equality before calling UpsertPeerPending).

func (*Store) LatestGroupDMMessageID 

func (s *Store) LatestGroupDMMessageID(ctx context.Context, groupID string) (id string, seq int64, err error)

LatestGroupDMMessageID returns the id of the most recent live message in groupID, or "" if the group has no messages. Used by the API layer to map the client's expectedLatestMessageId into the seq used by AppendGroupDMMessage's ExpectedLatestSeq.

func (*Store) LatestMessage 

func (s *Store) LatestMessage(ctx context.Context, agentID string) (*MessageRecord, error)

LatestMessage returns the highest-seq live message for agentID, or ErrNotFound if the transcript is empty / the agent is tombstoned.

func (*Store) ListActivePushSubscriptions 

func (s *Store) ListActivePushSubscriptions(ctx context.Context) ([]*PushSubscriptionRecord, error)

ListActivePushSubscriptions returns every row whose expired_at IS NULL ordered by (created_at, endpoint) ASC. Expired rows are kept in the table as a debug breadcrumb but excluded here so the live notify path doesn't waste a webpush.Send round-trip on a known-dead endpoint.

Tiebreak by endpoint: the v0→v1 importer stamps every row in a single batch with the same fileMTimeMillis(), so created_at alone would let SQLite return them in implementation-defined order — tests asserting an ordered slice would flake under `go test -count=N` and audit-log diffing across hosts would surface phantom changes.

func (*Store) ListAgentLocksByHolder 

func (s *Store) ListAgentLocksByHolder(ctx context.Context, peer string) ([]AgentLockRecord, error)

ListAgentLocksByHolder returns every agent_locks row whose holder_peer matches the given peer. Returns an empty slice when no rows match — never ErrNotFound, because the absence of rows is itself the intended answer (this peer doesn't hold anything).

Used by --peer startup to seed AgentLockGuard's desired set from durable state: in peer mode the in-memory list is empty until a live finalize fires AddAgent, so a daemon restart between handoff and any subsequent activity would otherwise let the lease expire and ownership become recoverable by another peer. Lease expiry is NOT consulted here — even an expired row is meaningful (the guard's refresh loop will re-Acquire on the next tick).

func (*Store) ListAgentLocksNotHeldBy added in v0.111.0 

func (s *Store) ListAgentLocksNotHeldBy(ctx context.Context, selfPeer string) ([]AgentLockRecord, error)

ListAgentLocksNotHeldBy returns every agent_locks row whose holder_peer differs from selfPeer — i.e. the set of agents that are currently transferred to some other device (§3.7 device-switch). Returns an empty slice when nothing is remote. Lease expiry is not consulted: even an expired remote row means the agent's canonical transcript still lives elsewhere, so the mirror refresher should keep trying while the holder is online.

Used by the Hub-side remote_message_mirror refresher to enumerate its work set without scanning the agents table.

func (*Store) ListAgentTasks 

func (s *Store) ListAgentTasks(ctx context.Context, agentID string, opts AgentTaskListOptions) ([]*AgentTaskRecord, error)

ListAgentTasks returns the live tasks for agentID ordered by seq ASC.

func (*Store) ListAgentWorkspaceFiles 

func (s *Store) ListAgentWorkspaceFiles(ctx context.Context, agentID string, opts WorkspaceFileListOptions) ([]*AgentWorkspaceFileRecord, error)

ListAgentWorkspaceFiles returns workspace-file rows for agentID, honoring opts. Drives the peer-sync delta and the workspace-sync reconcile loop's "what does the DB say is here?" query.

Always joins agents to filter out tombstoned parents — even with IncludeDeleted, a workspace file under a dead agent is uninteresting: the cascade delete will fire eventually and the row is on its way out. The DB-side ON DELETE CASCADE handles the actual cleanup; this filter keeps live readers from observing the doomed rows in the interim.

func (*Store) ListAgents 

func (s *Store) ListAgents(ctx context.Context) ([]*AgentRecord, error)

ListAgents returns all live agents, ordered by seq ASC (stable for UI lists and deterministic for snapshot diffs in tests).

func (*Store) ListBlobRefs 

func (s *Store) ListBlobRefs(ctx context.Context, opts ListBlobRefsOptions) ([]*BlobRefRecord, error)

ListBlobRefs returns rows matching opts, ordered by uri ASC for deterministic output. The query uses the (scope, uri) prefix on the PK so single-scope listings stay index-friendly.

func (*Store) ListEventsSince 

func (s *Store) ListEventsSince(ctx context.Context, since int64, opts ListEventsSinceOptions) (*ListEventsResult, error)

ListEventsSince returns rows with seq > since, ordered by seq. The store retention may have trimmed earlier rows; the result includes a Watermark so the caller can detect that case.

func (*Store) ListExternalChatCursorsByAgent 

func (s *Store) ListExternalChatCursorsByAgent(ctx context.Context, agentID string) ([]*ExternalChatCursorRecord, error)

ListExternalChatCursorsByAgent returns cursors owned by agentID ordered by id. agentID="" returns rows with NULL agent_id (deployment-scoped cursors; not used in v0 today but the schema allows them).

func (*Store) ListGroupDMDeadLetters added in v0.111.0 

func (s *Store) ListGroupDMDeadLetters(ctx context.Context, groupID string, limit int) ([]*GroupDMDeadLetter, error)

ListGroupDMDeadLetters returns dead letters for a group, newest first, capped at limit (0 = 50).

func (*Store) ListGroupDMMessages 

func (s *Store) ListGroupDMMessages(ctx context.Context, groupID string, opts GroupDMMessageListOptions) ([]*GroupDMMessageRecord, error)

ListGroupDMMessages returns messages for groupID. Hides messages of tombstoned parent groups (cascade-on-tombstone via JOIN).

func (*Store) ListGroupDMs 

func (s *Store) ListGroupDMs(ctx context.Context) ([]*GroupDMRecord, error)

ListGroupDMs returns all live group DMs ordered by seq.

func (*Store) ListHandoffQueuedAgentIDs added in v0.111.0 

func (s *Store) ListHandoffQueuedAgentIDs(ctx context.Context) ([]string, error)

ListHandoffQueuedAgentIDs returns the distinct agent_ids that have at least one queued message. The drain iterates this set and re-resolves each agent's current holder.

func (*Store) ListHandoffQueuedMessages added in v0.111.0 

func (s *Store) ListHandoffQueuedMessages(ctx context.Context, agentID string) ([]*HandoffQueuedMessage, error)

ListHandoffQueuedMessages returns the agent's queued messages in enqueue order (created_at, then rowid for same-millisecond ties — rowid is monotonic per insert so ties keep arrival order).

func (*Store) ListKV 

func (s *Store) ListKV(ctx context.Context, namespace string) ([]*KVRecord, error)

ListKV returns all rows in the namespace ordered by key. For now no pagination — kv is intended for small config / secret blobs.

func (*Store) ListMemoryEntries 

func (s *Store) ListMemoryEntries(ctx context.Context, agentID string, opts MemoryEntryListOptions) ([]*MemoryEntryRecord, error)

ListMemoryEntries returns the live entries for agentID. Ordered by seq ASC — this matches the "most recent at end" semantics the v0 list helpers use and keeps cursor pagination monotonic.

func (*Store) ListMessages 

func (s *Store) ListMessages(ctx context.Context, agentID string, opts MessageListOptions) ([]*MessageRecord, error)

ListMessages returns the messages for agentID matching opts. Always orders by seq (the natural conversation order); ID/timestamp order can diverge if the importer rewrites IDs but the design doc treats seq as authoritative.

func (*Store) ListPeerPending 

func (s *Store) ListPeerPending(ctx context.Context) ([]*PeerPendingRecord, error)

ListPeerPending returns every pending row ordered by last_seen DESC then device_id ASC.

func (*Store) ListPeers 

func (s *Store) ListPeers(ctx context.Context, opts ListPeersOptions) ([]*PeerRecord, error)

ListPeers returns rows matching opts ordered by last_seen DESC then device_id ASC for deterministic output. last_seen=NULL sorts last so recently-active peers float to the top.

func (*Store) ListRemoteMirrorMessages 

func (s *Store) ListRemoteMirrorMessages(ctx context.Context, agentID string, limit int, before string) ([]RemoteMirrorMessage, bool, error)

ListRemoteMirrorMessages は agent の mirror から timestamp DESC 順で 最新 limit 件 (oldest-first で返却) を取得する。before に message_id を 渡すと、その id より古い (timestamp が小さい、同 ts なら message_id が小さい) 行を limit 件返す。MessagesPaginated と同じ envelope (oldest-first slice + hasMore) を提供する。limit <= 0 は全件返却 + hasMore=false。

func (*Store) ListSessionsByAgent 

func (s *Store) ListSessionsByAgent(ctx context.Context, agentID string) ([]*SessionRecord, error)

ListSessionsByAgent returns sessions owned by agentID ordered by seq. Includes archived rows so the importer can verify a round-trip. agentID="" returns rows with NULL agent_id only — matching the "detached / pre-agent" sessions semantic.

func (*Store) MarkBlobRefForGC 

func (s *Store) MarkBlobRefForGC(ctx context.Context, uri string) error

MarkBlobRefForGC stamps marked_for_gc_at = now on the row. Used by internal/blob.Delete when slice 3+ adds the 24h grace window; slice 2 calls DeleteBlobRef directly. Idempotent.

func (*Store) MarkStalePeersOffline 

func (s *Store) MarkStalePeersOffline(ctx context.Context, before int64, excludeDeviceID string) (int, error)

MarkStalePeersOffline flips every row whose `last_seen` is older than `before` (unix millis) AND whose `status` is currently 'online' to 'offline'. The optional `excludeDeviceID` is left untouched — pass the local peer's id so the sweeper can never race against its own registrar's heartbeat (the registrar is the authoritative writer for the self-row's liveness).

Returns the number of rows flipped. The single UPDATE statement is race-safe against concurrent TouchPeer / RegisterPeerMetadata because SQLite serializes writers and the WHERE clause re-checks `status = 'online' AND last_seen < ?` at apply time — a heartbeat landing between the sweeper's intended-set scan and the UPDATE would refresh last_seen above `before`, fall outside the WHERE, and the row would correctly stay online.

Rows already 'offline' are not touched: the peer has been gone long enough that we already gave up.

`last_seen IS NULL` is treated as "never heartbeated" and *is* swept (NULL < `before` is false in SQL, so we OR-include it explicitly via COALESCE so a freshly-inserted row that the peer never actually checked in to gets cleared up by the next sweep).

func (*Store) MarkStalePeersOfflineDetail 

func (s *Store) MarkStalePeersOfflineDetail(ctx context.Context, before int64, excludeDeviceID string) ([]string, error)

MarkStalePeersOfflineDetail flips stale online peers offline AND returns the device_ids it changed, so a caller (the cross-peer status push) can publish per-row events without an extra ListPeers round trip. SELECT-then-UPDATE inside a tx so a concurrent TouchPeer can't slip in between the candidate list and the commit.

Same predicate as MarkStalePeersOffline; the splitting exists so callers that don't need the row list keep the cheaper count-only path.

func (*Store) Path 

func (s *Store) Path() string

Path returns the absolute path to kojo.db.

func (*Store) PurgeAgentRuntimeStateForRetry 

func (s *Store) PurgeAgentRuntimeStateForRetry(ctx context.Context, agentID string) error

PurgeAgentRuntimeStateForRetry drops every per-agent device-switch artefact this host carries so a fresh agent-sync from a trusted source can land on a clean slate. Unlike ForceReclaimAgentToLocal (which leaves the local host as holder), this helper does NOT re-insert agent_locks — the orchestrator's upcoming agent-sync + AgentLockGuard.AddAgent re-creates the row in the right shape. Used by state-probe self-heal: detecting a stale row that points at the wrong holder means the orchestrator wants to retry the switch, and keeping any prior row alive would just trip the next sync's existingLock.HolderPeer check with another 409.

Scope (single tx, rolled back on error):

  • agent_locks: DELETE the row outright. The fencing_token counter (agent_fencing_counters) is left in place so the next acquire mints a NEW token — any in-flight write from the previous holder still fails fencing.
  • blob_refs `kojo://global/agents/<id>/%`: clear handoff_pending so a half-finished switch flag doesn't block the upcoming SetBlobRefHandoffPending in begin. home_peer is intentionally NOT changed here — the row records where the blob body lives, and rewriting it blind would mask a legitimate "blob lives on the source" state. CompleteHandoff drives home_peer to target inside its own tx after the pull succeeds.
  • kv handoff markers (released/<id>, arrived/<id>, pending/<id>/*): DELETE so the prior switch's audit trail doesn't replay-evict the agent on restart.

Caller is responsible for refusing untrusted invocations — this method is the engine, not the gate.

func (*Store) PutKV 

func (s *Store) PutKV(ctx context.Context, rec *KVRecord, opts KVPutOptions) (*KVRecord, error)

PutKV inserts or updates a kv row. The (Value, ValueEncrypted) pair MUST agree with Secret: a secret row stores ValueEncrypted and an empty Value, a non-secret row stores Value and a nil ValueEncrypted. Mixing the two is a programming error and returned as a Go-level error rather than left to surface as a constraint violation later.

func (*Store) RefreshAgentLock 

func (s *Store) RefreshAgentLock(ctx context.Context, agentID, peer string, fencingToken, now, leaseDuration int64) (*AgentLockRecord, error)

RefreshAgentLock extends the lease on an already-held lock. Returns ErrFencingMismatch if (peer, fencingToken) doesn't match the row; ErrNotFound if no row exists for agent_id. Both diagnoses are run inside the same tx as the conditional UPDATE so a concurrent steal can't cause the diagnose-SELECT to disagree with the UPDATE about which case actually fired (BEGIN IMMEDIATE serializes writers).

func (*Store) RegisterPeerMetadata 

func (s *Store) RegisterPeerMetadata(ctx context.Context, rec *PeerRecord) (*PeerRecord, error)

RegisterPeerMetadata is the operator-driven peer registration path: it inserts a new row (last_seen=0, status='offline') or updates only the metadata columns (name, capabilities) of an existing row, preserving last_seen / status / public_key.

This is the atomic alternative to GetPeer→UpsertPeer in the HTTP handler. The race that read-modify-write opens is concrete: the registrar's heartbeat loop ticks every few seconds and fires TouchPeer(status=online, last_seen=now); a UI rename that happened to land in that window would, with the read-then-upsert path, write the now-stale offline/0 row back over the heartbeat update. The single-statement INSERT ... ON CONFLICT below closes that window because SQLite serializes statements at the row level.

public_key is preserved on conflict for the same reason as UpsertPeer — re-registration cannot rotate the long-lived identity key without going through an explicit RotatePeerKey path that audits the swap.

Empty Capabilities is stored as NULL.

func (*Store) ReleaseAgentLock 

func (s *Store) ReleaseAgentLock(ctx context.Context, agentID, peer string, fencingToken int64) error

ReleaseAgentLock removes the row iff (peer, fencingToken) matches. The fencing-token guard prevents a peer that lost the lock from "releasing" it on behalf of the new holder. Idempotent in the sense that calling Release again after success returns ErrNotFound, not ErrFencingMismatch.

The agent_fencing_counters row is intentionally NOT cleared so a subsequent Acquire on the same agent_id keeps advancing past every prior token; a delayed write from this caller after Release will fail CheckFencing on the new token.

As with Refresh, the conditional DELETE and the diagnostic SELECT run in the same tx so the answer cannot flip under concurrent writers.

func (*Store) ReleaseAgentLockByPeer 

func (s *Store) ReleaseAgentLockByPeer(ctx context.Context, peer string) (int64, error)

ReleaseAgentLockByPeer drops every lock row currently held by peer. Used during peer decommission and on Hub-driven failover (an admin has marked the peer offline; its locks must clear so other peers can pick them up). Returns the number of rows actually removed.

The agent_fencing_counters rows are intentionally retained so the next Acquire on each affected agent advances past the prior token — without that, a delayed write from the just-released peer would be silently accepted by a new holder that picked up token=1.

func (*Store) RemoteMirrorMessageExists 

func (s *Store) RemoteMirrorMessageExists(ctx context.Context, agentID, messageID string) (bool, error)

RemoteMirrorMessageExists は (agent_id, message_id) の行が存在するか だけを判定する。serveRemoteMessagesMerged が cursor の出所 source を 判別するために使う。エラーは false + err で返す。

func (*Store) ReplaceRemoteMirrorWindowIfHolder added in v0.111.0 

func (s *Store) ReplaceRemoteMirrorWindowIfHolder(ctx context.Context, agentID, expectedHolder string, msgs []RemoteMirrorMessage) error

ReplaceRemoteMirrorWindowIfHolder replaces the mirrored window for one agent with the holder's authoritative latest-N snapshot, inside the same holder-checked BEGIN IMMEDIATE transaction shape as UpsertRemoteMirrorMessagesIfHolder.

Semantics: msgs is the holder's newest window (a GET ?limit=N with no `before` cursor). Any mirror row that sorts inside that window — (timestamp, message_id) >= the oldest fetched message — but was NOT returned by the holder has been deleted there, so it is pruned here. Rows older than the window are preserved (they came from deeper paginated proxy reads and the holder said nothing about them). msgs == empty means the holder's transcript is empty: the whole mirror for the agent is cleared. Callers must only pass empty msgs when the holder reported hasMore=false.

The holder check (agent_locks.holder_peer == expectedHolder) skips the entire operation — including the prune — when the agent moved or came home, so a stale in-flight response can't clear a mirror it no longer owns.

func (*Store) RestoreBlobRef 

func (s *Store) RestoreBlobRef(ctx context.Context, rec *BlobRefRecord) error

RestoreBlobRef writes rec back to blob_refs verbatim — every column (including managed fields refcount, pin_policy, last_seen_ok, marked_for_gc_at, handoff_pending, created_at, updated_at) is set from the record. Used by the blob layer to roll back a failed Put after the prior row was snapshotted via StoreRefs.Snapshot — InsertOrReplaceBlobRef's ON-CONFLICT path would silently re-derive last_seen_ok / clear marked_for_gc_at, which is wrong on a restore.

Optimistic concurrency: when rec.ExpectedCurrentSHA256 is non-empty, the UPDATE branch only fires if the row's current sha256 matches that value. A mismatch (or absent row when ExpectedCurrentSHA256 is set to the snapshot's digest) returns ErrRestoreSuperseded so a concurrent state-machine driver isn't clobbered. ExpectedCurrentSHA256=="" disables the check — used by importers that rebuild the row from scratch.

If no row exists for rec.URI and ExpectedCurrentSHA256 is "", RestoreBlobRef inserts one instead of failing — keeps the importer path idempotent.

func (*Store) SchemaVersion 

func (s *Store) SchemaVersion(ctx context.Context) (int, error)

SchemaVersion returns the highest migration version applied. Returns 0 on a freshly-initialized database with no migrations.

func (*Store) SetBlobRefHandoffPending 

func (s *Store) SetBlobRefHandoffPending(ctx context.Context, uri string, pending bool) error

SetBlobRefHandoffPending toggles the handoff_pending flag for the row, returning ErrNotFound if no row matches. Used by the device-switch state machine (docs §3.7):

  • true at step 3 (source peer marks the row mid-switch so concurrent writes on either side surface as 409),
  • false at step 5 success or rollback (post-pull commit or pull failure → revert).

Idempotent on the boolean (re-setting to the same value is a no-op). updated_at is bumped unconditionally so a subscriber of the changes cursor sees the activity.

func (*Store) SetEventListener 

func (s *Store) SetEventListener(l EventListener)

SetEventListener installs (or replaces) the post-commit listener. Pass nil to disable.

func (*Store) SetGroupDMReadCursor added in v0.111.0 

func (s *Store) SetGroupDMReadCursor(ctx context.Context, groupID string, seq int64) error

SetGroupDMReadCursor persists the operator's read cursor (highest read seq) for a room. Idempotent upsert; a lower seq never overwrites a higher one so an out-of-order/stale mark-read can't resurrect unread badges.

func (*Store) SoftDeleteAgent 

func (s *Store) SoftDeleteAgent(ctx context.Context, id string) error

SoftDeleteAgent tombstones the agent. Idempotent: deleting an already- deleted (or absent) row returns nil so callers can use it in at-least-once delivery paths without bookkeeping. The etag is recomputed because agentETagInput captures DeletedAt — leaving the old etag would surface a stale "alive" cache hit for any reader that resolves the row by id.

func (*Store) SoftDeleteAgentMemory 

func (s *Store) SoftDeleteAgentMemory(ctx context.Context, agentID, ifMatchETag string) error

SoftDeleteAgentMemory tombstones the agent_memory row for agentID. Idempotent (already-tombstoned, missing-row, or missing-agent all return nil) so the daemon-side file→DB sync can call it on every "file gone, row exists" detection without conditionals on the caller side.

ifMatchETag is an optional optimistic-concurrency precondition checked inside the TX. Empty means "tombstone unconditionally" — matches the daemon-sync path. Non-empty + mismatch returns ErrETagMismatch (NOT idempotent: a stale precondition refuses rather than silently no-oping). Non-empty + missing-row also surfaces ErrETagMismatch — a caller asserting a specific etag against a vanished row needs to refetch, not silently succeed.

Recomputes etag on the way out so cross-device readers observe a fresh strong-etag and can distinguish "tombstoned" from the prior live state via 304 vs new ETag — the body-sha256 stays put because the canonical etag input includes deleted_at, so the tombstone itself is what shifts the hash.

func (*Store) SoftDeleteAgentTask 

func (s *Store) SoftDeleteAgentTask(ctx context.Context, id, ifMatchETag string) error

SoftDeleteAgentTask tombstones a task.

Empty ifMatchETag → unconditional, idempotent (a missing or already tombstoned row returns nil). Non-empty ifMatchETag → conditional: missing/tombstoned/mismatched surface as ErrETagMismatch.

func (*Store) SoftDeleteAgentWorkspaceFile 

func (s *Store) SoftDeleteAgentWorkspaceFile(ctx context.Context, agentID string, kind WorkspaceFileKind, ifMatchETag string) error

SoftDeleteAgentWorkspaceFile tombstones the agent_workspace_files row for (agentID, kind). Idempotent (already-tombstoned, missing-row, or missing-agent all return nil) so the daemon-side file→DB sync can call it on every "file gone, row exists" detection without conditionals on the caller side. Mirrors SoftDeleteAgentMemory.

ifMatchETag is an optional optimistic-concurrency precondition checked inside the TX. Empty means "tombstone unconditionally" — matches the daemon-sync path. Non-empty + mismatch returns ErrETagMismatch (NOT idempotent: a stale precondition refuses rather than silently no-oping). Non-empty + missing-row also surfaces ErrETagMismatch — a caller asserting a specific etag against a vanished row needs to refetch, not silently succeed.

Recomputes etag on the way out so cross-device readers observe a fresh strong-etag and can distinguish "tombstoned" from the prior live state via 304 vs new ETag — the body-sha256 stays put because the canonical etag input includes deleted_at, so the tombstone itself is what shifts the hash.

func (*Store) SoftDeleteAllMemoryEntries added in v0.101.4 

func (s *Store) SoftDeleteAllMemoryEntries(ctx context.Context, agentID string) (int64, error)

SoftDeleteAllMemoryEntries tombstones every live memory_entries row for agentID in a single UPDATE statement. Used by ResetData to wipe memory entries from the DB before the post-reset syncMemoryEntriesToDB runs — without this, the sync sees an empty memory/ directory, enters hydrate mode, and writes the pre-reset DB rows back to disk, effectively undoing the reset.

Idempotent: calling on an agent with no live entries is a no-op (0 rows affected). Does NOT recompute per-row etags — the rows are dead and the next sync (if any) will either ignore them or insert fresh rows.

func (*Store) SoftDeleteGroupDM 

func (s *Store) SoftDeleteGroupDM(ctx context.Context, id string) error

SoftDeleteGroupDM tombstones the group DM. Idempotent. Recomputes etag.

func (*Store) SoftDeleteGroupDMMessages added in v0.103.0 

func (s *Store) SoftDeleteGroupDMMessages(ctx context.Context, groupID string) (int64, error)

SoftDeleteGroupDMMessages tombstones every live message in groupID and returns the number of rows newly marked deleted. The parent group must be live; deleting a transcript for an unknown or tombstoned group is treated as ErrNotFound so callers don't silently clear stale views.

func (*Store) SoftDeleteMemoryEntry 

func (s *Store) SoftDeleteMemoryEntry(ctx context.Context, id, ifMatchETag string) error

SoftDeleteMemoryEntry tombstones an entry.

Empty ifMatchETag → unconditional, idempotent: a missing or already- tombstoned row returns nil so daemon-side reconciliation (sync) can blindly call this without racing with a parallel CLI delete.

Non-empty ifMatchETag → conditional: missing row, tombstoned row, or etag mismatch all surface as ErrETagMismatch. The HTTP handler maps that to 412 so a Web client asserting a specific live etag never silently succeeds against a row that drifted out from under it.

func (*Store) SoftDeleteMessage 

func (s *Store) SoftDeleteMessage(ctx context.Context, id, ifMatchETag string) error

SoftDeleteMessage tombstones a message. Idempotent on a missing/dead row when ifMatchETag is empty. Recomputes etag so the tombstone is visible in the change feed (messageETagInput captures DeletedAt — a stale "alive" etag would bypass invalidation).

ifMatchETag, when non-empty, enforces optimistic locking against the pre-delete row's etag: a missing/dead row returns ErrNotFound (so the caller can distinguish "already gone" from "your view is stale"), and a live-but-different etag returns ErrETagMismatch. Empty ifMatchETag preserves the legacy idempotent behaviour for daemon-internal callers.

func (*Store) SwitchBlobRefHome 

func (s *Store) SwitchBlobRefHome(ctx context.Context, uri, newHomePeer string) error

SwitchBlobRefHome flips home_peer to the new owner AND clears handoff_pending in a single tx. Used at §3.7 step 5 when the target peer's pull succeeded and the row's authoritative home moves to the target. Returns ErrNotFound on missing row.

The single-statement update is atomic at the SQLite level so a concurrent reader either sees the old (source, pending=true) pair or the new (target, pending=false) pair, never a half- committed mix.

func (*Store) SyncAgentFromPeer 

func (s *Store) SyncAgentFromPeer(ctx context.Context, payload AgentSyncPayload) error

SyncAgentFromPeer overwrites the target's local copy of one agent's metadata + transcript + memory with the source-supplied payload. Atomic across all five tables: a crash mid-call rolls back so the target never observes a half-synced state.

Semantics are "source wins": every field in payload is written verbatim; existing rows on target are replaced (messages / memory_entries are DELETE-then-INSERT, agents / persona / memory are UPSERT). Version + etag are taken from the source record so reads after sync return the same canonical state the source saw — UI cross-peer reads stay consistent.

Caller MUST validate the payload's agent_id is the one it expects (the orchestrator does this against the agent it's currently switching). This function does not enforce authorization; the HTTP layer (handlePeerAgentSync) gates RolePeer + caller-source identity matching.

func (*Store) TouchBlobRefLastSeenOK 

func (s *Store) TouchBlobRefLastSeenOK(ctx context.Context, uri, expectedSHA256 string, ts int64) error

TouchBlobRefLastSeenOK stamps last_seen_ok on the row. Called by the scrub job (§3.15-bis) after a successful sha256 re-hash verifies the on-disk body matches the canonical hash recorded in the row. The timestamp lets the scrub job skip rows it has recently checked and gives operators a way to see when a blob was last known-good.

`expectedSHA256` scopes the UPDATE to the row whose sha256 is still the value the scrubber hashed against: a concurrent Put that landed between the scrubber's Open and this Touch would have advanced the row's sha256, and stamping last_seen_ok on a row whose body we did NOT actually verify would lie to operators + later scrubs. Returns ErrNotFound when no row matches (skipped in that case; the next pass will re-hash the fresh body).

`ts` is taken from the caller so tests can supply a deterministic clock; production callers pass NowMillis().

func (*Store) TouchGroupDM added in v0.111.0 

func (s *Store) TouchGroupDM(ctx context.Context, id string, ts int64) error

TouchGroupDM advances a group's updated_at (and version/etag) to reflect new activity — e.g. a freshly posted message — without changing any settings field. It exists because UpdateGroupDM short-circuits on a settings no-op (a message post changes nothing UpdateGroupDM tracks), so it would never bump updated_at; the room-list "last active" time reads updated_at, so without this the row would freeze at creation time across restarts. No-op when the group is missing or tombstoned.

ts is the desired updated_at in epoch millis; a value not strictly greater than the current updated_at falls back to now (then to current+1) so the timestamp advances monotonically even under coarse-resolution (RFC3339 seconds) callers.

func (*Store) TouchPeer 

func (s *Store) TouchPeer(ctx context.Context, deviceID, status string, lastSeen int64) error

TouchPeer stamps last_seen = now and (optionally) updates status. If status is "" the existing status column is preserved. Used by the heartbeat handler so a noisy peer doesn't pay an UPSERT cycle just to update its last_seen.

Returns ErrNotFound when device_id is not registered — callers (handler / Hub) must Upsert first; we don't auto-create rows here because that would let an unauthenticated peer materialize itself just by calling /heartbeat.

func (*Store) TranscriptRevision 

func (s *Store) TranscriptRevision(ctx context.Context, agentID string) (count int64, maxUpdatedAt int64, err error)

TranscriptRevision returns (count, maxUpdatedAt) for the agent's live messages. Callers (FTS index, snapshot diff) use the pair as a cheap "did anything change?" cursor — count alone misses edit-in-place, max(updated_at) alone misses pure deletions that bring it back to a prior value. Together they detect every observable change to the transcript without scanning rows.

Returns (0, 0, nil) for a tombstoned agent or empty transcript so the caller can treat the zero pair as the canonical "no transcript" state.

func (*Store) TransferAgentLock 

func (s *Store) TransferAgentLock(ctx context.Context, agentID, currentPeer string, currentToken int64, newPeer string, leaseDurationMs, now int64) (*AgentLockRecord, error)

TransferAgentLock atomically moves the lock from currentPeer to newPeer, bumping the fencing_token via the per-agent counter so a delayed write from currentPeer with the prior token fails CheckFencing on the new value.

docs §3.7 step 6 — the device-switch handoff moves the lock to the target peer AFTER its blob pull succeeded (step 5). Without the token bump, a write from the source peer that was queued in flight could land on the target's row.

Returns ErrNotFound when no row matches (agent_id has no lock or current holder differs); ErrFencingMismatch when the lock row exists but the (currentPeer, currentToken) tuple doesn't match the row.

func (*Store) TruncateForRegenerate 

func (s *Store) TruncateForRegenerate(ctx context.Context, agentID, pivotID, pivotETag, sourceID, sourceETag string, killPivot bool) error

TruncateForRegenerate atomically validates the pivot row's etag and tombstones the suffix derived from the pivot's *immutable* seq — all inside a single transaction. This is the regenerate-flow counterpart to TruncateMessagesAfterSeq: callers don't compute afterSeq themselves (avoiding the TOCTOU window where a cross-device prefix delete shifts the boundary), they just describe the click.

killPivot:

  • true: tombstone seq >= pivot.Seq (assistant-mode regenerate — the user clicked an assistant response, wants it and everything after gone, then re-runs from the preceding user message).
  • false: tombstone seq > pivot.Seq (user-mode regenerate — the user clicked their own message, wants to keep it but redo everything after).

pivotID must be non-empty (the boundary is pivot-relative — empty pivot is rejected rather than silently accepting "kill everything" since the legacy kill-all path goes through TruncateMessagesAfterSeq instead). pivotETag is optional: when non-empty, mismatch returns ErrETagMismatch; when empty, the etag check is skipped but the pivot's seq is still used to derive the boundary atomically — so callers without optimistic-locking still get correct boundary semantics under cross-device prefix mutations.

sourceID + sourceETag close a second TOCTOU window specific to assistant-mode regenerate: the row whose content drives the chat (the user message preceding the clicked assistant pivot) is read outside this transaction by Manager.Regenerate to feed prepareChat. If a cross-device edit / tombstone lands between that read and this truncate, the chat would re-run against stale content. By passing the snapshot's etag here, the same TX that validates the pivot can also re-validate the source — mismatch returns ErrETagMismatch so the caller can surface 412 instead of committing a regen against disappeared/edited input.

When sourceID == "" the source-side precondition is dropped entirely (CLI / internal callers without optimistic locking). When sourceID == pivotID (user-mode regenerate, where the source IS the pivot) the pivot SELECT's curETag is reused to validate sourceETag — no extra round trip, but the precondition is still honoured even when the caller left pivotETag empty. When sourceID != pivotID (assistant-mode), a second SELECT inside this TX reads the source row and validates identity + etag (etag check only when sourceETag != "").

func (*Store) TruncateMessagesAfterSeq 

func (s *Store) TruncateMessagesAfterSeq(ctx context.Context, agentID string, afterSeq int64, pivotID, pivotETag string) (int64, error)

TruncateMessagesAfterSeq tombstones every live message for agentID whose seq > afterSeq. Used by lifecycle paths (Reset). The regenerate flow uses TruncateForRegenerate instead, which derives afterSeq from a pivot inside the transaction so the boundary cannot drift on a cross-device prefix mutation.

pivotID + pivotETag are accepted as a soft precondition for callers that want optimistic locking against an arbitrary row (e.g. future "rollback to this checkpoint" features); both empty disables the check. Returns ErrNotFound if the pivot row is gone/tombstoned and ErrETagMismatch on a live-but-different etag.

Iterates row-by-row inside a single transaction so each affected row gets a freshly recomputed etag. A bulk UPDATE would leave stale etags and silently break If-Match for downstream readers.

func (*Store) TruncateMessagesFromCreatedAt 

func (s *Store) TruncateMessagesFromCreatedAt(ctx context.Context, agentID string, sinceMillis int64) (int64, error)

TruncateMessagesFromCreatedAt tombstones every live message for agentID whose created_at >= sinceMillis. Sister method to TruncateMessagesAfterSeq for callers that need a wall-clock boundary instead of a seq pivot — the memory-truncate flow uses it to drop everything at or after a user-picked datetime in one shot.

Same row-by-row tombstone-with-recomputed-etag pattern as TruncateMessagesAfterSeq, for the same reason: a bulk UPDATE would leave stale etags and silently break If-Match for downstream readers.

func (*Store) UpdateAgent 

func (s *Store) UpdateAgent(ctx context.Context, id string, ifMatchETag string, mutate func(*AgentRecord) error) (*AgentRecord, error)

UpdateAgent applies mutate inside a transaction with optimistic locking on etag. ifMatchETag may be empty to skip the check (used by daemon-internal callers that already serialize against per-agent state); API handlers MUST pass the client's If-Match value.

mutate receives a pointer to a copy of the current record. The function must not mutate read-only fields (ID, CreatedAt, Seq); UpdatedAt and ETag are recomputed by UpdateAgent regardless of what mutate sets.

func (*Store) UpdateAgentLockAllowedProxy 

func (s *Store) UpdateAgentLockAllowedProxy(ctx context.Context, agentID, expectedHolder, allowedProxyPeer string) error

UpdateAgentLockAllowedProxy stamps the allowed_proxy_peer column on an existing lock row without disturbing holder / fencing_token / lease. Used by the §3.7 device-switch finalize path: target's AgentLockGuard.AddAgent (called inside the finalize hook) ran AcquireAgentLock which inserted with allowed_proxy_peer = self. That default would refuse the post-switch Hub→target proxy whose signer is the source. This call rewrites allowed_proxy_peer to the source so admit-by-orchestrator works without re-issuing the lock row (which would bump the fencing_token and invalidate the agent's just-adopted runtime tokens).

expectedHolder is a precondition: the UPDATE only fires when agent_locks.holder_peer matches it. A concurrent steal that flipped the holder between the caller's verify and this call will leave the row's allowed_proxy_peer untouched and return ErrFencingMismatch — the finalize handler treats that the same as lock_not_self.

Returns ErrNotFound when no row matches the agent_id; ErrFencingMismatch when the row exists but holder_peer no longer matches expectedHolder.

func (*Store) UpdateAgentTask 

func (s *Store) UpdateAgentTask(ctx context.Context, id, ifMatchETag string, patch AgentTaskPatch) (*AgentTaskRecord, error)

UpdateAgentTask applies patch to the task identified by id with optional If-Match etag check. Returns the updated record.

func (*Store) UpdateGroupDM 

func (s *Store) UpdateGroupDM(ctx context.Context, id, ifMatchETag string, mutate func(*GroupDMRecord) error) (*GroupDMRecord, error)

UpdateGroupDM applies mutate inside a tx with optimistic locking. Members added by mutate are re-validated against live agents.

func (*Store) UpdateMemoryEntry 

func (s *Store) UpdateMemoryEntry(ctx context.Context, id, ifMatchETag string, patch MemoryEntryPatch) (*MemoryEntryRecord, error)

UpdateMemoryEntry applies patch to the entry identified by id with optional If-Match. Recomputes BodySHA256 + ETag.

func (*Store) UpdateMessage 

func (s *Store) UpdateMessage(ctx context.Context, id, ifMatchETag string, patch MessagePatch) (*MessageRecord, error)

UpdateMessage applies patch to the message identified by id, with optional If-Match etag check. Returns the new record.

func (*Store) UpdatePeerMetadata 

func (s *Store) UpdatePeerMetadata(ctx context.Context, deviceID, name, url string) error

UpdatePeerMetadata mutates the human-editable name + url of an existing peer row without touching public_key, trusted, capabilities, last_seen, or status. Operator-driven path for the GUI's inline edit form.

Why capabilities is out: the UI never edits it, so sending the current value alongside an edit would race a concurrent peer- reported capabilities refresh and silently roll it back. The other preserved columns are owned by separate code paths (identity rotation, trust flip, heartbeat) and likewise must not be overwritten by a metadata edit. `name` and `url` empty values are rejected because the operator-facing UI has no legitimate path to wipe them.

Returns ErrNotFound when no row matches the device_id.

func (*Store) UpsertAgentMemory 

func (s *Store) UpsertAgentMemory(ctx context.Context, agentID, body, ifMatchETag string, opts AgentMemoryInsertOptions) (*AgentMemoryRecord, error)

UpsertAgentMemory writes (or replaces) the singleton MEMORY.md row for agentID. Mirrors the persona upsert semantics — see UpsertAgentPersona for the rationale on AllowOverwrite, ifMatchETag, and the live-vs-tombstone branching.

func (*Store) UpsertAgentPersona 

func (s *Store) UpsertAgentPersona(ctx context.Context, agentID, body, ifMatchETag string, opts AgentInsertOptions) (*AgentPersonaRecord, error)

UpsertAgentPersona writes (or replaces) the canonical persona row for agentID. The CLI-facing persona.md file is a hydrated per-agent mirror maintained by the agent manager sync paths; this store helper only touches the DB row.

ifMatchETag enforces optimistic locking against the prior row's etag — pass "" to skip (used by the v0→v1 importer and by daemon-internal callers that already serialize against per-agent state). API handlers MUST pass the client's If-Match value or "" only when no prior row exists.

func (*Store) UpsertAgentWorkspaceFile 

func (s *Store) UpsertAgentWorkspaceFile(ctx context.Context, agentID string, kind WorkspaceFileKind, body, ifMatchETag string, opts AgentWorkspaceFileInsertOptions) (*AgentWorkspaceFileRecord, error)

UpsertAgentWorkspaceFile writes (or replaces) the singleton workspace file row for (agentID, kind). Mirrors UpsertAgentMemory exactly — see that function for the rationale on AllowOverwrite, ifMatchETag, and the live-vs-tombstone branching.

Three branches: (1) live row → version-bumping UPDATE in place; (2) tombstoned row → DELETE + fresh INSERT to revive (preserves no history — the previous tombstone is gone); (3) no row → plain INSERT. Branch (2) keeps the primary key intact: SQLite's ON CONFLICT clauses would also work but require schema-level UPSERT support that varies between dialects, and the explicit branching makes the resurrection path easier to reason about in review.

func (*Store) UpsertPeer 

func (s *Store) UpsertPeer(ctx context.Context, rec *PeerRecord) (*PeerRecord, error)

UpsertPeer inserts or updates a peer_registry row keyed by device_id. Identity-stable columns (public_key) are preserved on conflict so a peer that re-registers cannot silently rotate its key without going through an explicit RotatePeerKey path (Phase 4 slice 2+). Mutable columns (name, capabilities, last_seen, status) overwrite.

Empty Capabilities is stored as NULL (matches the schema's nullable column) so JSON parsers downstream can branch on the difference between "no caps reported" and "empty caps object".

func (*Store) UpsertPeerPending 

func (s *Store) UpsertPeerPending(ctx context.Context, rec *PeerPendingRecord) (*PeerPendingRecord, error)

UpsertPeerPending refreshes name/url/last_seen on an existing pending row. node_key is PRESERVED on conflict — only the original NodeKey observed at insert time stays authoritative. A repeat /join-request whose WhoIs-resolved NodeKey differs from the stored value should be rejected at the handler before reaching here.

func (*Store) UpsertRemoteMirrorMessages 

func (s *Store) UpsertRemoteMirrorMessages(ctx context.Context, agentID, holderPeer string, msgs []RemoteMirrorMessage) error

UpsertRemoteMirrorMessages は msgs を agent_id 単位に upsert する。 message_id PK で dedup される。空 slice は no-op。1 トランザクション。

func (*Store) UpsertRemoteMirrorMessagesIfHolder 

func (s *Store) UpsertRemoteMirrorMessagesIfHolder(ctx context.Context, agentID, expectedHolder string, msgs []RemoteMirrorMessage) error

UpsertRemoteMirrorMessagesIfHolder は agent_locks.holder_peer が expectedHolder と一致するときだけ upsert する。in-flight proxy 応答が finalize/force-reclaim 直後に到着して stale 行を再挿入するレースを 単一 BEGIN IMMEDIATE トランザクション内で閉じる。holder が一致しない (または lock 行が消えている) ときは upsert を skip し、エラーは返さない。

type WorkspaceFileKind 

type WorkspaceFileKind string

WorkspaceFileKind is the typed alias for the `kind` column on agent_workspace_files. The CHECK constraint at the SQL layer also enforces this; mirroring the allowed values here turns a buggy caller into a typed error rather than a sqlite "CHECK constraint failed" surfacing through three layers of UI. Keep in sync with 0013_agent_workspace_files.sql. 

const (
	// WorkspaceFileKindUser is the user-authored workspace markdown
	// (user.md). Free-form prose the user maintains for the agent; the
	// CLI process consumes it via the on-disk mirror.
	WorkspaceFileKindUser WorkspaceFileKind = "user"

	// WorkspaceFileKindCheckin is the check-in markdown (checkin.md)
	// used as the prompt body for cron-scheduled check-ins. Previously
	// stored as settings_json.cronMessage; see the migration header
	// for the data-migration story.
	WorkspaceFileKindCheckin WorkspaceFileKind = "checkin"

	// WorkspaceFileKindStatus is the agent's self-maintained state file
	// (status.json on disk — JSON, not markdown): freeform key-value
	// pairs (mood, energy, sleepiness, affection, ...) injected into the
	// system prompt tail and edited by the agent itself as its state
	// drifts. Added by migration 0021.
	WorkspaceFileKindStatus WorkspaceFileKind = "status"

	// WorkspaceFileKindAnchor is the agent's optional persona anchor file
	// (anchor.md on disk — markdown): a 2-3 line first-person distillation
	// of the agent's persona (pronoun, tone, attitude) appended to the tail
	// of every turn's volatile context so the persona survives long-context
	// drift. Optional — nothing is injected when absent/empty — and edited
	// by the agent itself.
	WorkspaceFileKindAnchor WorkspaceFileKind = "anchor"
)

type WorkspaceFileListOptions 

type WorkspaceFileListOptions struct {
	Kind           WorkspaceFileKind
	IncludeDeleted bool
	UpdatedAtSince int64
}

WorkspaceFileListOptions filters ListAgentWorkspaceFiles. All zero- value fields are no-ops; the default list is "live rows for the agent, ordered by seq ASC".

  • Kind: when non-empty, restricts to one workspace-file kind. Unknown kinds return an error rather than an empty list so a buggy caller surfaces immediately.
  • IncludeDeleted: when true, tombstoned rows are returned. The §3.7 device-switch delta path MUST set this — otherwise a workspace file the user deleted on device A will silently stick around on device B because the tombstone never crosses the wire.
  • UpdatedAtSince: when > 0, returns rows whose updated_at >= UpdatedAtSince (inclusive — same-millisecond writes are NOT skipped). Ordering switches to updated_at ASC, kind ASC so ties at the same wall-clock millisecond are stable, and the caller resumes by replaying the last-seen updated_at and deduping on (agent_id, kind) — never by adding +1ms, which would lose any concurrent same-ms write.

Source Files

View all Source files

Directories

Path Synopsis
Package secretcrypto implements the envelope encryption used by the kv table's secret rows and (future) by the snapshot path described in docs/multi-device-storage.md §3.5 / §3.6.
 Package secretcrypto implements the envelope encryption used by the kv table's secret rows and (future) by the snapshot path described in docs/multi-device-storage.md §3.5 / §3.6.

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.