From 0a827551c09bf74b705257f2dff631a4ebd77f69 Mon Sep 17 00:00:00 2001 From: Wojtek Date: Thu, 21 May 2026 20:22:12 -0400 Subject: [PATCH] Add channel-memory SQLite adapter example --- examples/channel-memory/.claw-describe.json | 36 + examples/channel-memory/Dockerfile | 15 + examples/channel-memory/README.md | 59 + examples/channel-memory/main.go | 1361 +++++++++++++++++++ examples/channel-memory/main_test.go | 313 +++++ go.mod | 16 + go.sum | 45 + 7 files changed, 1845 insertions(+) create mode 100644 examples/channel-memory/.claw-describe.json create mode 100644 examples/channel-memory/Dockerfile create mode 100644 examples/channel-memory/README.md create mode 100644 examples/channel-memory/main.go create mode 100644 examples/channel-memory/main_test.go diff --git a/examples/channel-memory/.claw-describe.json b/examples/channel-memory/.claw-describe.json new file mode 100644 index 00000000..c3349629 --- /dev/null +++ b/examples/channel-memory/.claw-describe.json @@ -0,0 +1,36 @@ +{ + "version": 2, + "description": "Channel-memory adapter for source-backed Discord channel awareness digests", + "endpoints": [ + { + "method": "POST", + "path": "/ingest", + "description": "Retain one source channel message with stable message identity and content hash" + }, + { + "method": "POST", + "path": "/source-messages", + "description": "Fetch exact retained source messages by channel and message id" + }, + { + "method": "POST", + "path": "/digest", + "description": "Return deterministic digest blocks over retained source messages" + }, + { + "method": "POST", + "path": "/coverage-gaps", + "description": "Record an explicit source coverage gap for a channel and time range" + }, + { + "method": "POST", + "path": "/forget", + "description": "Suppress retained source messages and mark derived blocks dirty" + }, + { + "method": "GET", + "path": "/health", + "description": "Health check" + } + ] +} diff --git a/examples/channel-memory/Dockerfile b/examples/channel-memory/Dockerfile new file mode 100644 index 00000000..b69947c4 --- /dev/null +++ b/examples/channel-memory/Dockerfile @@ -0,0 +1,15 @@ +FROM golang:1.23-alpine AS builder +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY examples/channel-memory ./examples/channel-memory +RUN CGO_ENABLED=0 go build -o /channel-memory ./examples/channel-memory + +FROM alpine:3.20 +RUN apk add --no-cache wget +COPY --from=builder /channel-memory /channel-memory +COPY examples/channel-memory/.claw-describe.json /.claw-describe.json +EXPOSE 8080 +HEALTHCHECK --interval=5s --timeout=3s --retries=6 \ + CMD wget -qO- http://localhost:8080/health > /dev/null || exit 1 +CMD ["/channel-memory"] diff --git a/examples/channel-memory/README.md b/examples/channel-memory/README.md new file mode 100644 index 00000000..ddd6e8e2 --- /dev/null +++ b/examples/channel-memory/README.md @@ -0,0 +1,59 @@ +# Channel Memory Adapter + +This is the first executable slice of the digest-backed channel-awareness design. +It is a standalone HTTP service; `claw-wall` does not call it yet. + +The adapter stores Discord-style channel messages in SQLite and keeps exact +source provenance for later digest-backed `channel-awareness` work. + +## Endpoints + +- `POST /ingest` stores one source message. Idempotency is keyed by + `(source_kind, channel_id, message_id, content_hash)`. +- `POST /source-messages` fetches exact retained source messages by channel and + message id. By default it returns the current non-deleted version; set + `include_history: true` to inspect older content-hash versions. +- `POST /digest` returns deterministic digest blocks over already-retained + messages. It does not call an LLM. +- `POST /coverage-gaps` records an explicit missing source range. +- `POST /forget` suppresses source messages and marks derived blocks dirty. +- `GET /health` reports process liveness. + +## Deterministic Processor + +The deterministic path is intentionally conservative: + +- preserves obvious hard events as faithful `hard_event` blocks +- keeps ordinary retained content as `raw_excerpt` blocks +- collapses runtime/status noise into sparse `telemetry_count` blocks +- emits coverage-gap metadata from stored gap records +- creates tombstone blocks for deleted messages without carrying deleted content + +Higher-quality `topic_rollup` and `sequence_rollup` blocks belong to the async +LLM worker tracked separately. + +## Storage + +State is stored under `CHANNEL_MEMORY_DIR` and defaults to +`/data/channel-memory`. Set `CHANNEL_MEMORY_DB` to choose an exact SQLite file. + +The schema includes: + +- `source_messages` +- `derived_blocks` +- `derived_block_sources` +- `coverage_gaps` +- `processing_queue` + +`source_messages` uses explicit `observed_seq`, `observed_at`, and `is_current` +fields so edited messages create new rows while exact retrieval can still select +the current version deterministically. + +## Build + +This Dockerfile expects the repository root as build context because the service +uses the repository Go module: + +```sh +docker build -f examples/channel-memory/Dockerfile -t channel-memory:latest . +``` diff --git a/examples/channel-memory/main.go b/examples/channel-memory/main.go new file mode 100644 index 00000000..e093457a --- /dev/null +++ b/examples/channel-memory/main.go @@ -0,0 +1,1361 @@ +package main + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "log" + "net/http" + "os" + "path/filepath" + "slices" + "strings" + "time" + + _ "modernc.org/sqlite" +) + +const ( + defaultDataDir = "/data/channel-memory" + defaultDBFile = "channel-memory.sqlite" + defaultListenAddress = ":8080" + defaultSourceKind = "discord" + defaultSourceService = "claw-wall" +) + +type ingestRequest struct { + ChannelID string `json:"channel_id"` + Message ingestMessage `json:"message"` + Source ingestSource `json:"source,omitempty"` + Metadata map[string]string `json:"metadata,omitempty"` + Scope string `json:"scope,omitempty"` +} + +type ingestMessage struct { + ID string `json:"id"` + AuthorID string `json:"author_id,omitempty"` + AuthorName string `json:"author_name,omitempty"` + CreatedAt string `json:"created_at,omitempty"` + EditedAt string `json:"edited_at,omitempty"` + Deleted bool `json:"deleted,omitempty"` + Content string `json:"content,omitempty"` + ContentHash string `json:"content_hash,omitempty"` +} + +type ingestSource struct { + Kind string `json:"kind,omitempty"` + Service string `json:"service,omitempty"` + Surface string `json:"surface,omitempty"` + GuildID string `json:"guild_id,omitempty"` +} + +type ingestResponse struct { + Inserted bool `json:"inserted"` + SourceKind string `json:"source_kind"` + ChannelID string `json:"channel_id"` + MessageID string `json:"message_id"` + SourceHandle string `json:"source_handle"` + ContentHash string `json:"content_hash"` + ObservedSeq int64 `json:"observed_seq"` + Current bool `json:"current"` +} + +type sourceMessagesRequest struct { + SourceKind string `json:"source_kind,omitempty"` + ChannelID string `json:"channel_id"` + MessageIDs []string `json:"message_ids"` + IncludeHistory bool `json:"include_history,omitempty"` +} + +type sourceMessagesResponse struct { + Messages []sourceMessageRecord `json:"messages"` + NotFound []sourceRef `json:"not_found,omitempty"` +} + +type sourceRef struct { + SourceKind string `json:"source_kind"` + ChannelID string `json:"channel_id"` + MessageID string `json:"message_id"` +} + +type sourceMessageRecord struct { + SourceKind string `json:"source_kind"` + ChannelID string `json:"channel_id"` + MessageID string `json:"message_id"` + SourceHandle string `json:"source_handle"` + ContentHash string `json:"content_hash"` + AuthorID string `json:"author_id,omitempty"` + AuthorName string `json:"author_name,omitempty"` + CreatedAt string `json:"created_at"` + EditedAt string `json:"edited_at,omitempty"` + Deleted bool `json:"deleted,omitempty"` + Content string `json:"content,omitempty"` + Service string `json:"service,omitempty"` + Surface string `json:"surface,omitempty"` + GuildID string `json:"guild_id,omitempty"` + VisibilityScope string `json:"visibility_scope,omitempty"` + ObservedSeq int64 `json:"observed_seq"` + ObservedAt string `json:"observed_at"` + IsCurrent bool `json:"is_current"` + SupersededBy *int64 `json:"superseded_by,omitempty"` + ForgottenAt string `json:"forgotten_at,omitempty"` + ForgetReason string `json:"forget_reason,omitempty"` +} + +type digestRequest struct { + SourceKind string `json:"source_kind,omitempty"` + ChannelIDs []string `json:"channel_ids,omitempty"` + Since string `json:"since,omitempty"` + Budget digestBudget `json:"budget,omitempty"` +} + +type digestBudget struct { + MaxBlocks int `json:"max_blocks,omitempty"` +} + +type digestResponse struct { + Status string `json:"status"` + GeneratedAt string `json:"generated_at"` + Coverage digestCoverage `json:"coverage"` + Blocks []digestBlock `json:"blocks"` + Cost digestCost `json:"cost"` +} + +type digestCoverage struct { + From string `json:"from,omitempty"` + To string `json:"to,omitempty"` + SourceMessages int `json:"source_messages"` + DigestMessages int `json:"digest_messages"` + RawRecentMessages int `json:"raw_recent_messages"` + Gaps []coverageGap `json:"gaps,omitempty"` +} + +type digestCost struct { + DeterministicOnly bool `json:"deterministic_only"` + LLMCallsToday int `json:"llm_calls_today"` +} + +type digestBlock struct { + ID int64 `json:"id,omitempty"` + Kind string `json:"kind"` + EventType string `json:"event_type,omitempty"` + Text string `json:"text"` + SourceChannel string `json:"source_channel"` + SourceMessages []string `json:"source_messages"` + CoveredContentHashes []string `json:"covered_content_hashes,omitempty"` + SourceWindow sourceWindow `json:"source_window"` + Sparse bool `json:"sparse"` + Score float64 `json:"score"` + GeneratedAt string `json:"generated_at"` + Stale bool `json:"stale,omitempty"` + Dirty bool `json:"dirty,omitempty"` + Processor string `json:"processor"` +} + +type sourceWindow struct { + From string `json:"from"` + To string `json:"to"` +} + +type coverageGapRequest struct { + ChannelID string `json:"channel_id"` + From string `json:"from"` + To string `json:"to"` + Reason string `json:"reason,omitempty"` +} + +type coverageGap struct { + ID int64 `json:"id,omitempty"` + ChannelID string `json:"channel_id"` + From string `json:"from"` + To string `json:"to"` + Reason string `json:"reason,omitempty"` + CreatedAt string `json:"created_at,omitempty"` +} + +type forgetRequest struct { + SourceKind string `json:"source_kind,omitempty"` + ChannelID string `json:"channel_id"` + MessageIDs []string `json:"message_ids"` + Reason string `json:"reason,omitempty"` +} + +type forgetResponse struct { + Forgotten int `json:"forgotten"` +} + +type channelMemoryStore struct { + db *sql.DB + now func() time.Time +} + +type storedSourceMessage struct { + ID int64 + SourceKind string + ChannelID string + MessageID string + ContentHash string + AuthorID string + AuthorName string + CreatedAt string + EditedAt string + Deleted bool + Content string + Service string + Surface string + GuildID string + VisibilityScope string + ObservedSeq int64 + ObservedAt string + IsCurrent bool +} + +func main() { + store, err := openStoreFromEnv() + if err != nil { + log.Fatalf("open channel-memory store: %v", err) + } + defer store.Close() + + addr := strings.TrimSpace(os.Getenv("PORT")) + if addr == "" { + addr = defaultListenAddress + } else if !strings.Contains(addr, ":") { + addr = ":" + addr + } + + log.Printf("channel-memory listening on %s", addr) + if err := http.ListenAndServe(addr, newHandler(store)); err != nil { + log.Fatalf("listen: %v", err) + } +} + +func openStoreFromEnv() (*channelMemoryStore, error) { + dbPath := strings.TrimSpace(os.Getenv("CHANNEL_MEMORY_DB")) + if dbPath == "" { + dataDir := strings.TrimSpace(os.Getenv("CHANNEL_MEMORY_DIR")) + if dataDir == "" { + dataDir = defaultDataDir + } + dbPath = filepath.Join(dataDir, defaultDBFile) + } + return openStore(dbPath) +} + +func openStore(dbPath string) (*channelMemoryStore, error) { + if strings.TrimSpace(dbPath) == "" { + return nil, errors.New("db path is required") + } + if err := os.MkdirAll(filepath.Dir(dbPath), 0o755); err != nil { + return nil, err + } + db, err := sql.Open("sqlite", dbPath) + if err != nil { + return nil, err + } + db.SetMaxOpenConns(1) + + store := &channelMemoryStore{ + db: db, + now: func() time.Time { return time.Now().UTC() }, + } + if err := store.initSchema(context.Background()); err != nil { + _ = db.Close() + return nil, err + } + return store, nil +} + +func (s *channelMemoryStore) Close() error { + return s.db.Close() +} + +func (s *channelMemoryStore) initSchema(ctx context.Context) error { + statements := []string{ + `PRAGMA foreign_keys = ON`, + `PRAGMA journal_mode = WAL`, + `CREATE TABLE IF NOT EXISTS source_messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_kind TEXT NOT NULL, + channel_id TEXT NOT NULL, + message_id TEXT NOT NULL, + content_hash TEXT NOT NULL, + author_id TEXT NOT NULL DEFAULT '', + author_name TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL, + edited_at TEXT NOT NULL DEFAULT '', + deleted INTEGER NOT NULL DEFAULT 0, + content TEXT NOT NULL DEFAULT '', + service TEXT NOT NULL DEFAULT '', + surface TEXT NOT NULL DEFAULT '', + guild_id TEXT NOT NULL DEFAULT '', + visibility_scope TEXT NOT NULL DEFAULT '', + observed_seq INTEGER NOT NULL, + observed_at TEXT NOT NULL, + is_current INTEGER NOT NULL DEFAULT 1, + superseded_by INTEGER, + forgotten_at TEXT NOT NULL DEFAULT '', + forget_reason TEXT NOT NULL DEFAULT '', + UNIQUE(source_kind, channel_id, message_id, content_hash) + )`, + `CREATE INDEX IF NOT EXISTS idx_source_messages_current ON source_messages(source_kind, channel_id, message_id, is_current, deleted, forgotten_at)`, + `CREATE INDEX IF NOT EXISTS idx_source_messages_channel_time ON source_messages(source_kind, channel_id, created_at)`, + `CREATE TABLE IF NOT EXISTS derived_blocks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + block_key TEXT NOT NULL UNIQUE, + kind TEXT NOT NULL, + event_type TEXT NOT NULL DEFAULT '', + text TEXT NOT NULL, + source_channel TEXT NOT NULL, + source_window_from TEXT NOT NULL, + source_window_to TEXT NOT NULL, + sparse INTEGER NOT NULL, + score REAL NOT NULL, + generated_at TEXT NOT NULL, + stale INTEGER NOT NULL DEFAULT 0, + dirty INTEGER NOT NULL DEFAULT 0, + processor TEXT NOT NULL, + metadata_json TEXT NOT NULL DEFAULT '{}' + )`, + `CREATE INDEX IF NOT EXISTS idx_derived_blocks_channel_window ON derived_blocks(source_channel, source_window_to, dirty, stale)`, + `CREATE TABLE IF NOT EXISTS derived_block_sources ( + block_id INTEGER NOT NULL REFERENCES derived_blocks(id) ON DELETE CASCADE, + source_kind TEXT NOT NULL, + channel_id TEXT NOT NULL, + message_id TEXT NOT NULL, + content_hash TEXT NOT NULL, + PRIMARY KEY(block_id, source_kind, channel_id, message_id, content_hash) + )`, + `CREATE INDEX IF NOT EXISTS idx_derived_block_sources_identity ON derived_block_sources(source_kind, channel_id, message_id)`, + `CREATE TABLE IF NOT EXISTS coverage_gaps ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + channel_id TEXT NOT NULL, + from_ts TEXT NOT NULL, + to_ts TEXT NOT NULL, + reason TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS idx_coverage_gaps_channel_time ON coverage_gaps(channel_id, from_ts, to_ts)`, + `CREATE TABLE IF NOT EXISTS processing_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_kind TEXT NOT NULL, + channel_id TEXT NOT NULL, + message_id TEXT NOT NULL, + content_hash TEXT NOT NULL, + status TEXT NOT NULL, + kind TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + last_error TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS idx_processing_queue_status ON processing_queue(status, updated_at)`, + } + for _, stmt := range statements { + if _, err := s.db.ExecContext(ctx, stmt); err != nil { + return err + } + } + return nil +} + +func newHandler(store *channelMemoryStore) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/ingest", func(w http.ResponseWriter, r *http.Request) { + if !requireMethod(w, r, http.MethodPost) { + return + } + var req ingestRequest + if !decodeJSON(w, r, &req) { + return + } + resp, err := store.Ingest(r.Context(), req) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + writeJSON(w, http.StatusAccepted, resp) + }) + + mux.HandleFunc("/source-messages", func(w http.ResponseWriter, r *http.Request) { + if !requireMethod(w, r, http.MethodPost) { + return + } + var req sourceMessagesRequest + if !decodeJSON(w, r, &req) { + return + } + resp, err := store.SourceMessages(r.Context(), req) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + writeJSON(w, http.StatusOK, resp) + }) + + mux.HandleFunc("/digest", func(w http.ResponseWriter, r *http.Request) { + if !requireMethod(w, r, http.MethodPost) { + return + } + var req digestRequest + if !decodeJSON(w, r, &req) { + return + } + resp, err := store.Digest(r.Context(), req) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + writeJSON(w, http.StatusOK, resp) + }) + + mux.HandleFunc("/coverage-gaps", func(w http.ResponseWriter, r *http.Request) { + if !requireMethod(w, r, http.MethodPost) { + return + } + var req coverageGapRequest + if !decodeJSON(w, r, &req) { + return + } + gap, err := store.AddCoverageGap(r.Context(), req) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + writeJSON(w, http.StatusAccepted, gap) + }) + + mux.HandleFunc("/forget", func(w http.ResponseWriter, r *http.Request) { + if !requireMethod(w, r, http.MethodPost) { + return + } + var req forgetRequest + if !decodeJSON(w, r, &req) { + return + } + resp, err := store.Forget(r.Context(), req) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + writeJSON(w, http.StatusOK, resp) + }) + + mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + w.WriteHeader(http.StatusOK) + }) + + return mux +} + +func (s *channelMemoryStore) Ingest(ctx context.Context, req ingestRequest) (ingestResponse, error) { + normalized, err := normalizeIngestRequest(req, s.now()) + if err != nil { + return ingestResponse{}, err + } + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return ingestResponse{}, err + } + defer rollbackUnlessCommitted(tx) + + var nextObservedSeq int64 + if err := tx.QueryRowContext(ctx, ` + SELECT COALESCE(MAX(observed_seq), 0) + 1 + FROM source_messages + WHERE source_kind = ? AND channel_id = ? AND message_id = ?`, + normalized.SourceKind, normalized.ChannelID, normalized.MessageID, + ).Scan(&nextObservedSeq); err != nil { + return ingestResponse{}, err + } + normalized.ObservedSeq = nextObservedSeq + + result, err := tx.ExecContext(ctx, ` + INSERT OR IGNORE INTO source_messages ( + source_kind, channel_id, message_id, content_hash, + author_id, author_name, created_at, edited_at, deleted, content, + service, surface, guild_id, visibility_scope, + observed_seq, observed_at, is_current + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1)`, + normalized.SourceKind, normalized.ChannelID, normalized.MessageID, normalized.ContentHash, + normalized.AuthorID, normalized.AuthorName, normalized.CreatedAt, normalized.EditedAt, boolInt(normalized.Deleted), normalized.Content, + normalized.Service, normalized.Surface, normalized.GuildID, normalized.VisibilityScope, + normalized.ObservedSeq, normalized.ObservedAt, + ) + if err != nil { + return ingestResponse{}, err + } + + rows, err := result.RowsAffected() + if err != nil { + return ingestResponse{}, err + } + + inserted := rows > 0 + if inserted { + sourceID, err := result.LastInsertId() + if err != nil { + return ingestResponse{}, err + } + normalized.ID = sourceID + if _, err := tx.ExecContext(ctx, ` + UPDATE source_messages + SET is_current = 0, superseded_by = ? + WHERE source_kind = ? AND channel_id = ? AND message_id = ? AND id <> ? AND is_current = 1`, + sourceID, normalized.SourceKind, normalized.ChannelID, normalized.MessageID, sourceID, + ); err != nil { + return ingestResponse{}, err + } + if err := markIdentityBlocksDirtyTx(ctx, tx, normalized.SourceKind, normalized.ChannelID, normalized.MessageID, normalized.ContentHash); err != nil { + return ingestResponse{}, err + } + if err := enqueueProcessedTx(ctx, tx, normalized); err != nil { + return ingestResponse{}, err + } + if err := upsertDeterministicBlocksTx(ctx, tx, normalized, s.now()); err != nil { + return ingestResponse{}, err + } + } else { + loaded, err := loadSourceByIdentityAndHashTx(ctx, tx, normalized.SourceKind, normalized.ChannelID, normalized.MessageID, normalized.ContentHash) + if err != nil { + return ingestResponse{}, err + } + normalized = loaded + } + + if err := tx.Commit(); err != nil { + return ingestResponse{}, err + } + + return ingestResponse{ + Inserted: inserted, + SourceKind: normalized.SourceKind, + ChannelID: normalized.ChannelID, + MessageID: normalized.MessageID, + SourceHandle: sourceHandle(normalized.ChannelID, normalized.MessageID), + ContentHash: normalized.ContentHash, + ObservedSeq: normalized.ObservedSeq, + Current: normalized.IsCurrent, + }, nil +} + +func (s *channelMemoryStore) SourceMessages(ctx context.Context, req sourceMessagesRequest) (sourceMessagesResponse, error) { + sourceKind := normalizeSourceKind(req.SourceKind, "") + channelID := strings.TrimSpace(req.ChannelID) + if channelID == "" { + return sourceMessagesResponse{}, errors.New("channel_id is required") + } + if len(req.MessageIDs) == 0 { + return sourceMessagesResponse{}, errors.New("message_ids is required") + } + + resp := sourceMessagesResponse{} + for _, rawMessageID := range req.MessageIDs { + messageID := strings.TrimSpace(rawMessageID) + if messageID == "" { + continue + } + records, err := s.querySourceMessages(ctx, sourceKind, channelID, messageID, req.IncludeHistory) + if err != nil { + return sourceMessagesResponse{}, err + } + if len(records) == 0 { + resp.NotFound = append(resp.NotFound, sourceRef{SourceKind: sourceKind, ChannelID: channelID, MessageID: messageID}) + continue + } + resp.Messages = append(resp.Messages, records...) + } + return resp, nil +} + +func (s *channelMemoryStore) Digest(ctx context.Context, req digestRequest) (digestResponse, error) { + sourceKind := normalizeSourceKind(req.SourceKind, "") + cutoff, cutoffText, err := parseSince(req.Since, s.now()) + if err != nil { + return digestResponse{}, err + } + channels := trimNonEmpty(req.ChannelIDs) + if len(channels) == 0 { + var err error + channels, err = s.allChannels(ctx, sourceKind) + if err != nil { + return digestResponse{}, err + } + } + maxBlocks := req.Budget.MaxBlocks + if maxBlocks <= 0 { + maxBlocks = 32 + } + + blocks := make([]digestBlock, 0) + for _, channelID := range channels { + channelBlocks, err := s.queryDigestBlocks(ctx, channelID, cutoff.Format(time.RFC3339), maxBlocks-len(blocks)) + if err != nil { + return digestResponse{}, err + } + blocks = append(blocks, channelBlocks...) + if len(blocks) >= maxBlocks { + break + } + } + + gaps, err := s.queryCoverageGaps(ctx, channels, cutoff.Format(time.RFC3339)) + if err != nil { + return digestResponse{}, err + } + sourceCount, err := s.countCurrentSources(ctx, sourceKind, channels, cutoff.Format(time.RFC3339)) + if err != nil { + return digestResponse{}, err + } + + status := "ok" + if len(gaps) > 0 { + status = "coverage_gap" + } else if len(blocks) == 0 { + status = "unavailable" + } + + rawRecent := 0 + for _, block := range blocks { + if block.Kind == "raw_excerpt" || block.Kind == "hard_event" || block.Kind == "tombstone" { + rawRecent++ + } + } + + return digestResponse{ + Status: status, + GeneratedAt: s.now().Format(time.RFC3339), + Coverage: digestCoverage{ + From: cutoffText, + To: s.now().Format(time.RFC3339), + SourceMessages: sourceCount, + DigestMessages: len(blocks), + RawRecentMessages: rawRecent, + Gaps: gaps, + }, + Blocks: blocks, + Cost: digestCost{ + DeterministicOnly: true, + LLMCallsToday: 0, + }, + }, nil +} + +func (s *channelMemoryStore) AddCoverageGap(ctx context.Context, req coverageGapRequest) (coverageGap, error) { + channelID := strings.TrimSpace(req.ChannelID) + if channelID == "" { + return coverageGap{}, errors.New("channel_id is required") + } + from, err := normalizeTimestamp(req.From, s.now()) + if err != nil { + return coverageGap{}, fmt.Errorf("from: %w", err) + } + to, err := normalizeTimestamp(req.To, s.now()) + if err != nil { + return coverageGap{}, fmt.Errorf("to: %w", err) + } + createdAt := s.now().Format(time.RFC3339) + result, err := s.db.ExecContext(ctx, ` + INSERT INTO coverage_gaps(channel_id, from_ts, to_ts, reason, created_at) + VALUES (?, ?, ?, ?, ?)`, + channelID, from, to, strings.TrimSpace(req.Reason), createdAt, + ) + if err != nil { + return coverageGap{}, err + } + id, err := result.LastInsertId() + if err != nil { + return coverageGap{}, err + } + return coverageGap{ID: id, ChannelID: channelID, From: from, To: to, Reason: strings.TrimSpace(req.Reason), CreatedAt: createdAt}, nil +} + +func (s *channelMemoryStore) Forget(ctx context.Context, req forgetRequest) (forgetResponse, error) { + sourceKind := normalizeSourceKind(req.SourceKind, "") + channelID := strings.TrimSpace(req.ChannelID) + if channelID == "" { + return forgetResponse{}, errors.New("channel_id is required") + } + if len(req.MessageIDs) == 0 { + return forgetResponse{}, errors.New("message_ids is required") + } + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return forgetResponse{}, err + } + defer rollbackUnlessCommitted(tx) + + forgottenAt := s.now().Format(time.RFC3339) + total := 0 + for _, rawMessageID := range req.MessageIDs { + messageID := strings.TrimSpace(rawMessageID) + if messageID == "" { + continue + } + result, err := tx.ExecContext(ctx, ` + UPDATE source_messages + SET forgotten_at = ?, forget_reason = ?, is_current = 0 + WHERE source_kind = ? AND channel_id = ? AND message_id = ? AND forgotten_at = ''`, + forgottenAt, strings.TrimSpace(req.Reason), sourceKind, channelID, messageID, + ) + if err != nil { + return forgetResponse{}, err + } + rows, err := result.RowsAffected() + if err != nil { + return forgetResponse{}, err + } + total += int(rows) + if err := markIdentityBlocksDirtyTx(ctx, tx, sourceKind, channelID, messageID, ""); err != nil { + return forgetResponse{}, err + } + } + if err := tx.Commit(); err != nil { + return forgetResponse{}, err + } + return forgetResponse{Forgotten: total}, nil +} + +func normalizeIngestRequest(req ingestRequest, now time.Time) (storedSourceMessage, error) { + sourceKind := normalizeSourceKind(req.Source.Kind, req.Source.Surface) + channelID := strings.TrimSpace(req.ChannelID) + messageID := strings.TrimSpace(req.Message.ID) + if channelID == "" { + return storedSourceMessage{}, errors.New("channel_id is required") + } + if messageID == "" { + return storedSourceMessage{}, errors.New("message.id is required") + } + createdAt, err := normalizeTimestamp(req.Message.CreatedAt, now) + if err != nil { + return storedSourceMessage{}, fmt.Errorf("message.created_at: %w", err) + } + editedAt := "" + if strings.TrimSpace(req.Message.EditedAt) != "" { + editedAt, err = normalizeTimestamp(req.Message.EditedAt, now) + if err != nil { + return storedSourceMessage{}, fmt.Errorf("message.edited_at: %w", err) + } + } + observedAt := now.UTC().Format(time.RFC3339) + content := req.Message.Content + contentHash := normalizeContentHash(req.Message.ContentHash, sourceKind, channelID, messageID, content, req.Message.Deleted) + service := firstNonEmpty(req.Source.Service, defaultSourceService) + surface := firstNonEmpty(req.Source.Surface, sourceKind) + + return storedSourceMessage{ + SourceKind: sourceKind, + ChannelID: channelID, + MessageID: messageID, + ContentHash: contentHash, + AuthorID: strings.TrimSpace(req.Message.AuthorID), + AuthorName: strings.TrimSpace(req.Message.AuthorName), + CreatedAt: createdAt, + EditedAt: editedAt, + Deleted: req.Message.Deleted, + Content: content, + Service: strings.TrimSpace(service), + Surface: strings.TrimSpace(surface), + GuildID: strings.TrimSpace(req.Source.GuildID), + VisibilityScope: strings.TrimSpace(firstNonEmpty(req.Scope, req.Metadata["visibility_scope"])), + ObservedAt: observedAt, + IsCurrent: true, + }, nil +} + +func loadSourceByIdentityAndHashTx(ctx context.Context, tx *sql.Tx, sourceKind, channelID, messageID, contentHash string) (storedSourceMessage, error) { + row := tx.QueryRowContext(ctx, ` + SELECT id, source_kind, channel_id, message_id, content_hash, author_id, author_name, + created_at, edited_at, deleted, content, service, surface, guild_id, visibility_scope, + observed_seq, observed_at, is_current + FROM source_messages + WHERE source_kind = ? AND channel_id = ? AND message_id = ? AND content_hash = ?`, + sourceKind, channelID, messageID, contentHash, + ) + return scanStoredSource(row) +} + +type rowScanner interface { + Scan(dest ...any) error +} + +func scanStoredSource(row rowScanner) (storedSourceMessage, error) { + var record storedSourceMessage + var deleted, current int + if err := row.Scan( + &record.ID, &record.SourceKind, &record.ChannelID, &record.MessageID, &record.ContentHash, + &record.AuthorID, &record.AuthorName, &record.CreatedAt, &record.EditedAt, &deleted, &record.Content, + &record.Service, &record.Surface, &record.GuildID, &record.VisibilityScope, + &record.ObservedSeq, &record.ObservedAt, ¤t, + ); err != nil { + return storedSourceMessage{}, err + } + record.Deleted = deleted != 0 + record.IsCurrent = current != 0 + return record, nil +} + +func enqueueProcessedTx(ctx context.Context, tx *sql.Tx, source storedSourceMessage) error { + now := time.Now().UTC().Format(time.RFC3339) + _, err := tx.ExecContext(ctx, ` + INSERT INTO processing_queue(source_kind, channel_id, message_id, content_hash, status, kind, created_at, updated_at) + VALUES (?, ?, ?, ?, 'processed', 'deterministic', ?, ?)`, + source.SourceKind, source.ChannelID, source.MessageID, source.ContentHash, now, now, + ) + return err +} + +func markIdentityBlocksDirtyTx(ctx context.Context, tx *sql.Tx, sourceKind, channelID, messageID, exceptContentHash string) error { + query := ` + UPDATE derived_blocks + SET dirty = 1 + WHERE id IN ( + SELECT block_id FROM derived_block_sources + WHERE source_kind = ? AND channel_id = ? AND message_id = ?` + args := []any{sourceKind, channelID, messageID} + if exceptContentHash != "" { + query += ` AND content_hash <> ?` + args = append(args, exceptContentHash) + } + query += `)` + _, err := tx.ExecContext(ctx, query, args...) + return err +} + +func upsertDeterministicBlocksTx(ctx context.Context, tx *sql.Tx, source storedSourceMessage, now time.Time) error { + if source.Deleted { + return upsertSourceBlockTx(ctx, tx, source, deterministicBlock{ + Key: sourceBlockKey("tombstone", source), + Kind: "tombstone", + Text: formatTombstone(source), + Sparse: true, + Score: 1, + Processor: "deterministic", + }, now) + } + if isTelemetryNoise(source.Content) { + return rebuildTelemetryBlockTx(ctx, tx, source, now) + } + kind, eventType, score := classifySourceContent(source.Content) + return upsertSourceBlockTx(ctx, tx, source, deterministicBlock{ + Key: sourceBlockKey(kind, source), + Kind: kind, + EventType: eventType, + Text: formatSourceLine(source), + Sparse: false, + Score: score, + Processor: "deterministic", + }, now) +} + +type deterministicBlock struct { + Key string + Kind string + EventType string + Text string + Sparse bool + Score float64 + Processor string +} + +func upsertSourceBlockTx(ctx context.Context, tx *sql.Tx, source storedSourceMessage, block deterministicBlock, now time.Time) error { + if strings.TrimSpace(block.Text) == "" { + return nil + } + generatedAt := now.UTC().Format(time.RFC3339) + if _, err := tx.ExecContext(ctx, ` + INSERT INTO derived_blocks( + block_key, kind, event_type, text, source_channel, + source_window_from, source_window_to, sparse, score, generated_at, stale, dirty, processor + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0, ?) + ON CONFLICT(block_key) DO UPDATE SET + kind = excluded.kind, + event_type = excluded.event_type, + text = excluded.text, + source_channel = excluded.source_channel, + source_window_from = excluded.source_window_from, + source_window_to = excluded.source_window_to, + sparse = excluded.sparse, + score = excluded.score, + generated_at = excluded.generated_at, + stale = 0, + dirty = 0, + processor = excluded.processor`, + block.Key, block.Kind, block.EventType, block.Text, source.ChannelID, + source.CreatedAt, firstNonEmpty(source.EditedAt, source.CreatedAt), boolInt(block.Sparse), block.Score, generatedAt, block.Processor, + ); err != nil { + return err + } + var blockID int64 + if err := tx.QueryRowContext(ctx, `SELECT id FROM derived_blocks WHERE block_key = ?`, block.Key).Scan(&blockID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM derived_block_sources WHERE block_id = ?`, blockID); err != nil { + return err + } + _, err := tx.ExecContext(ctx, ` + INSERT OR IGNORE INTO derived_block_sources(block_id, source_kind, channel_id, message_id, content_hash) + VALUES (?, ?, ?, ?, ?)`, + blockID, source.SourceKind, source.ChannelID, source.MessageID, source.ContentHash, + ) + return err +} + +func rebuildTelemetryBlockTx(ctx context.Context, tx *sql.Tx, source storedSourceMessage, now time.Time) error { + from, to := telemetryBucket(source.CreatedAt) + rows, err := tx.QueryContext(ctx, ` + SELECT id, source_kind, channel_id, message_id, content_hash, author_id, author_name, + created_at, edited_at, deleted, content, service, surface, guild_id, visibility_scope, + observed_seq, observed_at, is_current + FROM source_messages + WHERE source_kind = ? AND channel_id = ? AND is_current = 1 AND deleted = 0 AND forgotten_at = '' + AND created_at >= ? AND created_at < ? + ORDER BY created_at, observed_seq`, + source.SourceKind, source.ChannelID, from, to, + ) + if err != nil { + return err + } + defer rows.Close() + + sources := make([]storedSourceMessage, 0) + for rows.Next() { + record, err := scanStoredSource(rows) + if err != nil { + return err + } + if isTelemetryNoise(record.Content) { + sources = append(sources, record) + } + } + if err := rows.Err(); err != nil { + return err + } + if len(sources) == 0 { + return nil + } + + blockKey := strings.Join([]string{"telemetry", source.SourceKind, source.ChannelID, from}, ":") + generatedAt := now.UTC().Format(time.RFC3339) + text := fmt.Sprintf("[%s-%s] runtime/status noise elided: %d messages.", clockText(from), clockText(to), len(sources)) + if _, err := tx.ExecContext(ctx, ` + INSERT INTO derived_blocks( + block_key, kind, text, source_channel, source_window_from, source_window_to, + sparse, score, generated_at, stale, dirty, processor + ) VALUES (?, 'telemetry_count', ?, ?, ?, ?, 1, 0.25, ?, 0, 0, 'deterministic') + ON CONFLICT(block_key) DO UPDATE SET + text = excluded.text, + source_window_from = excluded.source_window_from, + source_window_to = excluded.source_window_to, + generated_at = excluded.generated_at, + stale = 0, + dirty = 0`, + blockKey, text, source.ChannelID, sources[0].CreatedAt, sources[len(sources)-1].CreatedAt, generatedAt, + ); err != nil { + return err + } + var blockID int64 + if err := tx.QueryRowContext(ctx, `SELECT id FROM derived_blocks WHERE block_key = ?`, blockKey).Scan(&blockID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM derived_block_sources WHERE block_id = ?`, blockID); err != nil { + return err + } + for _, record := range sources { + if _, err := tx.ExecContext(ctx, ` + INSERT OR IGNORE INTO derived_block_sources(block_id, source_kind, channel_id, message_id, content_hash) + VALUES (?, ?, ?, ?, ?)`, + blockID, record.SourceKind, record.ChannelID, record.MessageID, record.ContentHash, + ); err != nil { + return err + } + } + return nil +} + +func (s *channelMemoryStore) querySourceMessages(ctx context.Context, sourceKind, channelID, messageID string, includeHistory bool) ([]sourceMessageRecord, error) { + query := ` + SELECT id, source_kind, channel_id, message_id, content_hash, author_id, author_name, + created_at, edited_at, deleted, content, service, surface, guild_id, visibility_scope, + observed_seq, observed_at, is_current, superseded_by, forgotten_at, forget_reason + FROM source_messages + WHERE source_kind = ? AND channel_id = ? AND message_id = ?` + args := []any{sourceKind, channelID, messageID} + if !includeHistory { + query += ` AND is_current = 1 AND deleted = 0 AND forgotten_at = ''` + } + query += ` ORDER BY observed_seq` + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + records := make([]sourceMessageRecord, 0) + for rows.Next() { + var record sourceMessageRecord + var id int64 + var deleted, current int + var supersededBy sql.NullInt64 + if err := rows.Scan( + &id, &record.SourceKind, &record.ChannelID, &record.MessageID, &record.ContentHash, + &record.AuthorID, &record.AuthorName, &record.CreatedAt, &record.EditedAt, &deleted, &record.Content, + &record.Service, &record.Surface, &record.GuildID, &record.VisibilityScope, + &record.ObservedSeq, &record.ObservedAt, ¤t, &supersededBy, &record.ForgottenAt, &record.ForgetReason, + ); err != nil { + return nil, err + } + if supersededBy.Valid { + value := supersededBy.Int64 + record.SupersededBy = &value + } + record.Deleted = deleted != 0 + record.IsCurrent = current != 0 + record.SourceHandle = sourceHandle(record.ChannelID, record.MessageID) + records = append(records, record) + } + return records, rows.Err() +} + +func (s *channelMemoryStore) queryDigestBlocks(ctx context.Context, channelID, cutoff string, limit int) ([]digestBlock, error) { + if limit <= 0 { + return nil, nil + } + rows, err := s.db.QueryContext(ctx, ` + SELECT id, kind, event_type, text, source_channel, source_window_from, source_window_to, + sparse, score, generated_at, stale, dirty, processor + FROM derived_blocks + WHERE source_channel = ? AND source_window_to >= ? AND stale = 0 AND dirty = 0 + ORDER BY source_window_from ASC, id ASC + LIMIT ?`, + channelID, cutoff, limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + + blocks := make([]digestBlock, 0) + for rows.Next() { + var block digestBlock + var sparse, stale, dirty int + var from, to string + if err := rows.Scan( + &block.ID, &block.Kind, &block.EventType, &block.Text, &block.SourceChannel, &from, &to, + &sparse, &block.Score, &block.GeneratedAt, &stale, &dirty, &block.Processor, + ); err != nil { + return nil, err + } + block.SourceWindow = sourceWindow{From: from, To: to} + block.Sparse = sparse != 0 + block.Stale = stale != 0 + block.Dirty = dirty != 0 + blocks = append(blocks, block) + } + if err := rows.Err(); err != nil { + return nil, err + } + if err := rows.Close(); err != nil { + return nil, err + } + + for i := range blocks { + if err := s.loadBlockSources(ctx, &blocks[i]); err != nil { + return nil, err + } + } + return blocks, nil +} + +func (s *channelMemoryStore) loadBlockSources(ctx context.Context, block *digestBlock) error { + rows, err := s.db.QueryContext(ctx, ` + SELECT message_id, content_hash + FROM derived_block_sources + WHERE block_id = ? + ORDER BY message_id, content_hash`, + block.ID, + ) + if err != nil { + return err + } + defer rows.Close() + + seenMessages := make(map[string]struct{}) + for rows.Next() { + var messageID, contentHash string + if err := rows.Scan(&messageID, &contentHash); err != nil { + return err + } + if _, ok := seenMessages[messageID]; !ok { + block.SourceMessages = append(block.SourceMessages, messageID) + seenMessages[messageID] = struct{}{} + } + block.CoveredContentHashes = append(block.CoveredContentHashes, contentHash) + } + return rows.Err() +} + +func (s *channelMemoryStore) queryCoverageGaps(ctx context.Context, channels []string, cutoff string) ([]coverageGap, error) { + gaps := make([]coverageGap, 0) + for _, channelID := range channels { + rows, err := s.db.QueryContext(ctx, ` + SELECT id, channel_id, from_ts, to_ts, reason, created_at + FROM coverage_gaps + WHERE channel_id = ? AND to_ts >= ? + ORDER BY from_ts, id`, + channelID, cutoff, + ) + if err != nil { + return nil, err + } + for rows.Next() { + var gap coverageGap + if err := rows.Scan(&gap.ID, &gap.ChannelID, &gap.From, &gap.To, &gap.Reason, &gap.CreatedAt); err != nil { + _ = rows.Close() + return nil, err + } + gaps = append(gaps, gap) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, err + } + if err := rows.Close(); err != nil { + return nil, err + } + } + return gaps, nil +} + +func (s *channelMemoryStore) countCurrentSources(ctx context.Context, sourceKind string, channels []string, cutoff string) (int, error) { + total := 0 + for _, channelID := range channels { + var count int + if err := s.db.QueryRowContext(ctx, ` + SELECT COUNT(*) + FROM source_messages + WHERE source_kind = ? AND channel_id = ? AND is_current = 1 AND deleted = 0 AND forgotten_at = '' AND created_at >= ?`, + sourceKind, channelID, cutoff, + ).Scan(&count); err != nil { + return 0, err + } + total += count + } + return total, nil +} + +func (s *channelMemoryStore) allChannels(ctx context.Context, sourceKind string) ([]string, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT DISTINCT channel_id + FROM source_messages + WHERE source_kind = ? + ORDER BY channel_id`, + sourceKind, + ) + if err != nil { + return nil, err + } + defer rows.Close() + channels := make([]string, 0) + for rows.Next() { + var channelID string + if err := rows.Scan(&channelID); err != nil { + return nil, err + } + channels = append(channels, channelID) + } + return channels, rows.Err() +} + +func normalizeSourceKind(kind, surface string) string { + for _, candidate := range []string{kind, surface, defaultSourceKind} { + candidate = strings.ToLower(strings.TrimSpace(candidate)) + if candidate != "" { + return candidate + } + } + return defaultSourceKind +} + +func normalizeTimestamp(value string, fallback time.Time) (string, error) { + value = strings.TrimSpace(value) + if value == "" { + return fallback.UTC().Format(time.RFC3339), nil + } + parsed, err := time.Parse(time.RFC3339, value) + if err != nil { + return "", err + } + return parsed.UTC().Format(time.RFC3339), nil +} + +func normalizeContentHash(value, sourceKind, channelID, messageID, content string, deleted bool) string { + value = strings.TrimSpace(value) + if value != "" { + return value + } + seed := strings.Join([]string{sourceKind, channelID, messageID, content, fmt.Sprintf("deleted=%t", deleted)}, "\x00") + sum := sha256.Sum256([]byte(seed)) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func parseSince(value string, now time.Time) (time.Time, string, error) { + value = strings.TrimSpace(value) + if value == "" { + value = "24h" + } + if duration, err := time.ParseDuration(value); err == nil { + cutoff := now.UTC().Add(-duration) + return cutoff, cutoff.Format(time.RFC3339), nil + } + parsed, err := time.Parse(time.RFC3339, value) + if err != nil { + return time.Time{}, "", fmt.Errorf("since must be a duration or RFC3339 timestamp") + } + return parsed.UTC(), parsed.UTC().Format(time.RFC3339), nil +} + +func sourceHandle(channelID, messageID string) string { + return channelID + "/" + messageID +} + +func sourceBlockKey(kind string, source storedSourceMessage) string { + return strings.Join([]string{kind, source.SourceKind, source.ChannelID, source.MessageID, source.ContentHash}, ":") +} + +func classifySourceContent(content string) (kind, eventType string, score float64) { + lower := strings.ToLower(content) + switch { + case containsAny(lower, "proposed", "proposal", "[proposed]"): + return "hard_event", "trade_proposal", 1 + case containsAny(lower, "approved", "approval", "[approved]"): + return "hard_event", "trade_approval", 1 + case containsAny(lower, "confirmed", "confirmation", "[confirmed]"): + return "hard_event", "trade_confirmation", 1 + case containsAny(lower, "filled", "fill "): + return "hard_event", "trade_fill", 1 + case containsAny(lower, "stop", "target"): + return "hard_event", "stop_or_target_change", 0.95 + case containsAny(lower, "route", "no route"): + return "hard_event", "route_decision", 0.95 + case containsAny(lower, "watchlist", "held position", "thesis", "risk limit"): + return "hard_event", "thesis_update", 0.9 + default: + return "raw_excerpt", "", 0.65 + } +} + +func isTelemetryNoise(content string) bool { + lower := strings.ToLower(content) + return containsAny(lower, + "heartbeat_ok", + "heartbeat ok", + "runtime status", + "provider retry", + "gateway reconnect", + "gateway shutdown", + "cron status", + "upstream request failed", + "context deadline exceeded", + ) +} + +func formatSourceLine(source storedSourceMessage) string { + author := firstNonEmpty(source.AuthorName, source.AuthorID, "unknown") + return fmt.Sprintf("[%s] %s: %s", clockText(source.CreatedAt), author, strings.TrimSpace(source.Content)) +} + +func formatTombstone(source storedSourceMessage) string { + author := firstNonEmpty(source.AuthorName, source.AuthorID, "unknown") + ts := firstNonEmpty(source.EditedAt, source.CreatedAt) + return fmt.Sprintf("[%s] %s: [message removed]; source=%s", clockText(ts), author, sourceHandle(source.ChannelID, source.MessageID)) +} + +func telemetryBucket(ts string) (from, to string) { + parsed, err := time.Parse(time.RFC3339, ts) + if err != nil { + parsed = time.Now().UTC() + } + start := parsed.UTC().Truncate(time.Hour) + return start.Format(time.RFC3339), start.Add(time.Hour).Format(time.RFC3339) +} + +func clockText(ts string) string { + parsed, err := time.Parse(time.RFC3339, ts) + if err != nil { + return ts + } + return parsed.UTC().Format("15:04") +} + +func containsAny(text string, needles ...string) bool { + for _, needle := range needles { + if strings.Contains(text, needle) { + return true + } + } + return false +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { + return value + } + } + return "" +} + +func trimNonEmpty(values []string) []string { + out := make([]string, 0, len(values)) + for _, value := range values { + value = strings.TrimSpace(value) + if value != "" && !slices.Contains(out, value) { + out = append(out, value) + } + } + return out +} + +func boolInt(value bool) int { + if value { + return 1 + } + return 0 +} + +func rollbackUnlessCommitted(tx *sql.Tx) { + _ = tx.Rollback() +} + +func requireMethod(w http.ResponseWriter, r *http.Request, method string) bool { + if r.Method != method { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return false + } + return true +} + +func decodeJSON(w http.ResponseWriter, r *http.Request, value any) bool { + if err := json.NewDecoder(r.Body).Decode(value); err != nil { + http.Error(w, fmt.Sprintf("decode JSON: %v", err), http.StatusBadRequest) + return false + } + return true +} + +func writeJSON(w http.ResponseWriter, status int, value any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(value) +} diff --git a/examples/channel-memory/main_test.go b/examples/channel-memory/main_test.go new file mode 100644 index 00000000..e08ec3ee --- /dev/null +++ b/examples/channel-memory/main_test.go @@ -0,0 +1,313 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestChannelMemoryIngestIdempotentAndEditedVersionIsCurrent(t *testing.T) { + store := newTestStore(t) + defer store.Close() + + base := ingestRequest{ + ChannelID: "chan-1", + Message: ingestMessage{ + ID: "101", + AuthorName: "analyst-a", + CreatedAt: "2026-05-21T16:00:00Z", + Content: "[PROPOSED] signal-101 SELL ACME", + ContentHash: "sha256:first", + }, + } + + first, err := store.Ingest(context.Background(), base) + if err != nil { + t.Fatalf("ingest first: %v", err) + } + if !first.Inserted || first.ObservedSeq != 1 || !first.Current { + t.Fatalf("unexpected first ingest response: %+v", first) + } + + duplicate, err := store.Ingest(context.Background(), base) + if err != nil { + t.Fatalf("ingest duplicate: %v", err) + } + if duplicate.Inserted || duplicate.ObservedSeq != 1 { + t.Fatalf("duplicate should be idempotent, got %+v", duplicate) + } + + edited := base + edited.Message.EditedAt = "2026-05-21T16:02:00Z" + edited.Message.Content = "[PROPOSED] signal-101 SELL ACME after revised momentum check" + edited.Message.ContentHash = "sha256:second" + editResp, err := store.Ingest(context.Background(), edited) + if err != nil { + t.Fatalf("ingest edit: %v", err) + } + if !editResp.Inserted || editResp.ObservedSeq != 2 { + t.Fatalf("edit should be a new observed source row, got %+v", editResp) + } + + current, err := store.SourceMessages(context.Background(), sourceMessagesRequest{ + ChannelID: "chan-1", + MessageIDs: []string{"101"}, + }) + if err != nil { + t.Fatalf("current source query: %v", err) + } + if len(current.Messages) != 1 { + t.Fatalf("expected one current source message, got %+v", current) + } + if got := current.Messages[0].ContentHash; got != "sha256:second" { + t.Fatalf("current query returned wrong content hash %q", got) + } + if !strings.Contains(current.Messages[0].Content, "revised momentum") { + t.Fatalf("current query returned old content: %+v", current.Messages[0]) + } + + history, err := store.SourceMessages(context.Background(), sourceMessagesRequest{ + ChannelID: "chan-1", + MessageIDs: []string{"101"}, + IncludeHistory: true, + }) + if err != nil { + t.Fatalf("history source query: %v", err) + } + if len(history.Messages) != 2 { + t.Fatalf("expected two content-hash versions, got %+v", history.Messages) + } + if history.Messages[0].IsCurrent || history.Messages[0].SupersededBy == nil { + t.Fatalf("old source row should be superseded, got %+v", history.Messages[0]) + } + if !history.Messages[1].IsCurrent { + t.Fatalf("edited source row should be current, got %+v", history.Messages[1]) + } +} + +func TestChannelMemoryDeleteTombstoneAndForgetSuppressDerivedBlocks(t *testing.T) { + store := newTestStore(t) + defer store.Close() + + msg := ingestRequest{ + ChannelID: "chan-1", + Message: ingestMessage{ + ID: "201", + AuthorName: "analyst-a", + CreatedAt: "2026-05-21T17:00:00Z", + Content: "[APPROVED] signal-201 BUY ACME", + ContentHash: "sha256:approved", + }, + } + if _, err := store.Ingest(context.Background(), msg); err != nil { + t.Fatalf("ingest approved message: %v", err) + } + + beforeDelete, err := store.Digest(context.Background(), digestRequest{ChannelIDs: []string{"chan-1"}, Since: "24h"}) + if err != nil { + t.Fatalf("digest before delete: %v", err) + } + if len(beforeDelete.Blocks) != 1 || beforeDelete.Blocks[0].Kind != "hard_event" || beforeDelete.Blocks[0].Sparse { + t.Fatalf("expected one faithful hard_event before delete, got %+v", beforeDelete.Blocks) + } + + deleted := msg + deleted.Message.Deleted = true + deleted.Message.EditedAt = "2026-05-21T17:03:00Z" + deleted.Message.Content = "" + deleted.Message.ContentHash = "sha256:deleted" + if _, err := store.Ingest(context.Background(), deleted); err != nil { + t.Fatalf("ingest delete tombstone: %v", err) + } + + current, err := store.SourceMessages(context.Background(), sourceMessagesRequest{ChannelID: "chan-1", MessageIDs: []string{"201"}}) + if err != nil { + t.Fatalf("current after delete: %v", err) + } + if len(current.Messages) != 0 || len(current.NotFound) != 1 { + t.Fatalf("deleted current source should not be served by default, got %+v", current) + } + + afterDelete, err := store.Digest(context.Background(), digestRequest{ChannelIDs: []string{"chan-1"}, Since: "24h"}) + if err != nil { + t.Fatalf("digest after delete: %v", err) + } + if len(afterDelete.Blocks) != 1 || afterDelete.Blocks[0].Kind != "tombstone" || !afterDelete.Blocks[0].Sparse { + t.Fatalf("expected sparse tombstone after delete, got %+v", afterDelete.Blocks) + } + if strings.Contains(afterDelete.Blocks[0].Text, "BUY ACME") { + t.Fatalf("tombstone leaked deleted source content: %+v", afterDelete.Blocks[0]) + } + + forget, err := store.Forget(context.Background(), forgetRequest{ + ChannelID: "chan-1", + MessageIDs: []string{"201"}, + Reason: "operator requested removal", + }) + if err != nil { + t.Fatalf("forget: %v", err) + } + if forget.Forgotten != 2 { + t.Fatalf("expected both content versions forgotten, got %+v", forget) + } + + afterForget, err := store.Digest(context.Background(), digestRequest{ChannelIDs: []string{"chan-1"}, Since: "24h"}) + if err != nil { + t.Fatalf("digest after forget: %v", err) + } + if len(afterForget.Blocks) != 0 || afterForget.Status != "unavailable" { + t.Fatalf("forget should suppress derived blocks, got %+v", afterForget) + } +} + +func TestChannelMemoryDeterministicTelemetryAndCoverageGap(t *testing.T) { + store := newTestStore(t) + defer store.Close() + + for i := 0; i < 3; i++ { + req := ingestRequest{ + ChannelID: "chan-ops", + Message: ingestMessage{ + ID: string(rune('a' + i)), + AuthorName: "agent-status", + CreatedAt: "2026-05-21T18:0" + string(rune('0'+i)) + ":00Z", + Content: "HEARTBEAT_OK runtime status nominal", + ContentHash: "sha256:heartbeat-" + string(rune('0'+i)), + }, + } + if _, err := store.Ingest(context.Background(), req); err != nil { + t.Fatalf("ingest heartbeat %d: %v", i, err) + } + } + + digest, err := store.Digest(context.Background(), digestRequest{ChannelIDs: []string{"chan-ops"}, Since: "24h"}) + if err != nil { + t.Fatalf("digest telemetry: %v", err) + } + if len(digest.Blocks) != 1 { + t.Fatalf("expected one collapsed telemetry block, got %+v", digest.Blocks) + } + block := digest.Blocks[0] + if block.Kind != "telemetry_count" || !block.Sparse { + t.Fatalf("expected sparse telemetry_count, got %+v", block) + } + if len(block.SourceMessages) != 3 || !strings.Contains(block.Text, "3 messages") { + t.Fatalf("telemetry block should cover all source messages, got %+v", block) + } + + gap, err := store.AddCoverageGap(context.Background(), coverageGapRequest{ + ChannelID: "chan-ops", + From: "2026-05-21T17:00:00Z", + To: "2026-05-21T17:30:00Z", + Reason: "backfill rate limited", + }) + if err != nil { + t.Fatalf("add coverage gap: %v", err) + } + if gap.ID == 0 { + t.Fatalf("expected persisted gap id, got %+v", gap) + } + + withGap, err := store.Digest(context.Background(), digestRequest{ChannelIDs: []string{"chan-ops"}, Since: "24h"}) + if err != nil { + t.Fatalf("digest coverage gap: %v", err) + } + if withGap.Status != "coverage_gap" || len(withGap.Coverage.Gaps) != 1 { + t.Fatalf("expected coverage_gap status, got %+v", withGap) + } +} + +func TestChannelMemoryHTTPAPI(t *testing.T) { + store := newTestStore(t) + defer store.Close() + + server := httptest.NewServer(newHandler(store)) + defer server.Close() + + postJSONExpect(t, server.URL+"/ingest", ingestRequest{ + ChannelID: "chan-http", + Message: ingestMessage{ + ID: "301", + AuthorName: "analyst-b", + CreatedAt: "2026-05-21T19:00:00Z", + Content: "ACME note remains relevant", + ContentHash: "sha256:http", + }, + }, http.StatusAccepted) + + sources := postJSONDecode[sourceMessagesResponse](t, server.URL+"/source-messages", sourceMessagesRequest{ + ChannelID: "chan-http", + MessageIDs: []string{"301"}, + }, http.StatusOK) + if len(sources.Messages) != 1 || sources.Messages[0].SourceHandle != "chan-http/301" { + t.Fatalf("unexpected source query response: %+v", sources) + } + + digest := postJSONDecode[digestResponse](t, server.URL+"/digest", digestRequest{ + ChannelIDs: []string{"chan-http"}, + Since: "24h", + }, http.StatusOK) + if len(digest.Blocks) != 1 || digest.Blocks[0].Kind != "raw_excerpt" { + t.Fatalf("unexpected digest response: %+v", digest) + } +} + +func newTestStore(t *testing.T) *channelMemoryStore { + t.Helper() + store, err := openStore(filepath.Join(t.TempDir(), "channel-memory.sqlite")) + if err != nil { + t.Fatalf("open test store: %v", err) + } + fixedNow := time.Date(2026, 5, 21, 20, 0, 0, 0, time.UTC) + store.now = func() time.Time { return fixedNow } + return store +} + +func postJSONExpect(t *testing.T, url string, body any, wantStatus int) { + t.Helper() + reqBody, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal request: %v", err) + } + resp, err := http.Post(url, "application/json", bytes.NewReader(reqBody)) + if err != nil { + t.Fatalf("post %s: %v", url, err) + } + defer resp.Body.Close() + if resp.StatusCode != wantStatus { + var got bytes.Buffer + _, _ = got.ReadFrom(resp.Body) + t.Fatalf("expected status %d from %s, got %d body=%s", wantStatus, url, resp.StatusCode, got.String()) + } +} + +func postJSONDecode[T any](t *testing.T, url string, body any, wantStatus int) T { + t.Helper() + reqBody, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal request: %v", err) + } + resp, err := http.Post(url, "application/json", bytes.NewReader(reqBody)) + if err != nil { + t.Fatalf("post %s: %v", url, err) + } + defer resp.Body.Close() + + if resp.StatusCode != wantStatus { + var got bytes.Buffer + _, _ = got.ReadFrom(resp.Body) + t.Fatalf("expected status %d from %s, got %d body=%s", wantStatus, url, resp.StatusCode, got.String()) + } + + var decoded T + if err := json.NewDecoder(resp.Body).Decode(&decoded); err != nil { + t.Fatalf("decode response from %s: %v", url, err) + } + return decoded +} diff --git a/go.mod b/go.mod index 52ec9d9c..a27084f4 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,7 @@ require ( github.com/spf13/cobra v1.8.1 golang.org/x/mod v0.13.0 gopkg.in/yaml.v3 v3.0.1 + modernc.org/sqlite v1.28.0 oras.land/oras-go/v2 v2.6.0 ) @@ -19,15 +20,21 @@ require ( github.com/distribution/reference v0.5.0 // indirect github.com/docker/go-connections v0.5.0 // indirect github.com/docker/go-units v0.5.0 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/go-logr/logr v1.3.0 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/gogo/protobuf v1.3.2 // indirect + github.com/google/uuid v1.5.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect + github.com/ncruces/go-strftime v1.0.0 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pkg/errors v0.9.1 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/spf13/pflag v1.0.5 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect go.opentelemetry.io/otel v1.21.0 // indirect @@ -38,4 +45,13 @@ require ( golang.org/x/tools v0.14.0 // indirect google.golang.org/protobuf v1.31.0 // indirect gotest.tools/v3 v3.5.2 // indirect + lukechampine.com/uint128 v1.2.0 // indirect + modernc.org/cc/v3 v3.41.0 // indirect + modernc.org/ccgo/v3 v3.16.15 // indirect + modernc.org/libc v1.41.0 // indirect + modernc.org/mathutil v1.7.1 // indirect + modernc.org/memory v1.8.0 // indirect + modernc.org/opt v0.1.3 // indirect + modernc.org/strutil v1.1.3 // indirect + modernc.org/token v1.1.0 // indirect ) diff --git a/go.sum b/go.sum index bcb18c2e..f8e19852 100644 --- a/go.sum +++ b/go.sum @@ -19,6 +19,8 @@ github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc= github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= @@ -34,12 +36,22 @@ github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiu github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/pprof v0.0.0-20230323073829-e72429f035bd h1:r8yyd+DJDmsUhGrRBxH5Pj7KeFK5l+Y3FsgT8keqKtk= +github.com/google/pprof v0.0.0-20230323073829-e72429f035bd/go.mod h1:79YE0hCXdHag9sBkw2o+N/YnZtTkXi0UT9Nnixa5eYk= +github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU= +github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 h1:YBftPWNWd4WwGqtY2yeZL2ef8rHAxPBD8KFhJpmcqms= github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0/go.mod h1:YN5jB8ie0yfIUg6VvR9Kz84aCaG7AsGZnLjhHbUqwPg= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs= +github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mattn/go-sqlite3 v1.14.16 h1:yOQRA0RpS5PFz/oikGwBEqvAWhWg5ufRz4ETLjwpU1Y= +github.com/mattn/go-sqlite3 v1.14.16/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg= github.com/moby/buildkit v0.13.2 h1:nXNszM4qD9E7QtG7bFWPnDI1teUQFQglBzon/IU3SzI= github.com/moby/buildkit v0.13.2/go.mod h1:2cyVOv9NoHM7arphK9ZfHIWKn9YVZRFd1wXB8kKmEzY= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= @@ -48,6 +60,8 @@ github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0= github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y= github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= +github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= +github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= @@ -56,6 +70,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= @@ -107,6 +123,7 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4= golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= @@ -142,5 +159,33 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= +lukechampine.com/uint128 v1.2.0 h1:mBi/5l91vocEN8otkC5bDLhi2KdCticRiwbdB0O+rjI= +lukechampine.com/uint128 v1.2.0/go.mod h1:c4eWIwlEGaxC/+H1VguhU4PHXNWDCDMUlWdIWl2j1gk= +modernc.org/cc/v3 v3.41.0 h1:QoR1Sn3YWlmA1T4vLaKZfawdVtSiGx8H+cEojbC7v1Q= +modernc.org/cc/v3 v3.41.0/go.mod h1:Ni4zjJYJ04CDOhG7dn640WGfwBzfE0ecX8TyMB0Fv0Y= +modernc.org/ccgo/v3 v3.16.15 h1:KbDR3ZAVU+wiLyMESPtbtE/Add4elztFyfsWoNTgxS0= +modernc.org/ccgo/v3 v3.16.15/go.mod h1:yT7B+/E2m43tmMOT51GMoM98/MtHIcQQSleGnddkUNI= +modernc.org/ccorpus v1.11.6 h1:J16RXiiqiCgua6+ZvQot4yUuUy8zxgqbqEEUuGPlISk= +modernc.org/ccorpus v1.11.6/go.mod h1:2gEUTrWqdpH2pXsmTM1ZkjeSrUWDpjMu2T6m29L/ErQ= +modernc.org/httpfs v1.0.6 h1:AAgIpFZRXuYnkjftxTAZwMIiwEqAfk8aVB2/oA6nAeM= +modernc.org/httpfs v1.0.6/go.mod h1:7dosgurJGp0sPaRanU53W4xZYKh14wfzX420oZADeHM= +modernc.org/libc v1.41.0 h1:g9YAc6BkKlgORsUWj+JwqoB1wU3o4DE3bM3yvA3k+Gk= +modernc.org/libc v1.41.0/go.mod h1:w0eszPsiXoOnoMJgrXjglgLuDy/bt5RR4y3QzUUeodY= +modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= +modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= +modernc.org/memory v1.8.0 h1:IqGTL6eFMaDZZhEWwcREgeMXYwmW83LYW8cROZYkg+E= +modernc.org/memory v1.8.0/go.mod h1:XPZ936zp5OMKGWPqbD3JShgd/ZoQ7899TUuQqxY+peU= +modernc.org/opt v0.1.3 h1:3XOZf2yznlhC+ibLltsDGzABUGVx8J6pnFMS3E4dcq4= +modernc.org/opt v0.1.3/go.mod h1:WdSiB5evDcignE70guQKxYUl14mgWtbClRi5wmkkTX0= +modernc.org/sqlite v1.28.0 h1:Zx+LyDDmXczNnEQdvPuEfcFVA2ZPyaD7UCZDjef3BHQ= +modernc.org/sqlite v1.28.0/go.mod h1:Qxpazz0zH8Z1xCFyi5GSL3FzbtZ3fvbjmywNogldEW0= +modernc.org/strutil v1.1.3 h1:fNMm+oJklMGYfU9Ylcywl0CO5O6nTfaowNsh2wpPjzY= +modernc.org/strutil v1.1.3/go.mod h1:MEHNA7PdEnEwLvspRMtWTNnp2nnyvMfkimT1NKNAGbw= +modernc.org/tcl v1.15.2 h1:C4ybAYCGJw968e+Me18oW55kD/FexcHbqH2xak1ROSY= +modernc.org/tcl v1.15.2/go.mod h1:3+k/ZaEbKrC8ePv8zJWPtBSW0V7Gg9g8rkmhI1Kfs3c= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= +modernc.org/z v1.7.3 h1:zDJf6iHjrnB+WRD88stbXokugjyc0/pB91ri1gO6LZY= +modernc.org/z v1.7.3/go.mod h1:Ipv4tsdxZRbQyLq9Q1M6gdbkxYzdlrciF2Hi/lS7nWE= oras.land/oras-go/v2 v2.6.0 h1:X4ELRsiGkrbeox69+9tzTu492FMUu7zJQW6eJU+I2oc= oras.land/oras-go/v2 v2.6.0/go.mod h1:magiQDfG6H1O9APp+rOsvCPcW1GD2MM7vgnKY0Y+u1o=