feat: add durable conversations and sanitized audit storage

.env.example

@@ -18,6 +18,17 @@ NIVORA_SSE_HEARTBEAT=15s
 NIVORA_MAX_HISTORY_TURNS=12
 NIVORA_MAX_QUESTION_BYTES=16384
 
+# Durable conversation and audit storage.
+# Production should set NIVORA_STORAGE_REQUIRED=true and use PostgreSQL.
+NIVORA_STORAGE_REQUIRED=false
+NIVORA_STORAGE_DRIVER=sqlite
+NIVORA_STORAGE_DSN=file:/var/lib/nivora/nivora.db?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)
+# PostgreSQL example:
+# NIVORA_STORAGE_DRIVER=pgx
+# NIVORA_STORAGE_DSN=postgres://nivora:password@127.0.0.1:5432/nivora?sslmode=require
+NIVORA_STORAGE_RETENTION=720h
+NIVORA_STORAGE_CLEANUP_INTERVAL=1h
+
 # Volcengine Ark through the official Eino extension.
 ARK_API_KEY=ark-xxxxxxxx
 # Ordered endpoint IDs. Nivora falls back only before streaming begins.

README.md

@@ -17,11 +17,13 @@ Lumio is the first planned provider integration, but Nivora itself does not know
 - provider-neutral Tools for knowledge, customer context, resources, diagnosis, transactions, and human-support cases
 - Provider-side approved-knowledge reference service using the official Eino VikingDB retriever
 - tenant, approval, freshness, provenance, and score validation after semantic retrieval
+- SQLite development and PostgreSQL production storage for public transcripts, run metadata, sanitized Tool audits, and support-case references
+- deterministic replay protection and tenant-scoped transcript access
 - black-box customer-support and knowledge-retrieval JSONL evaluation tools
 - bounded Provider retries for idempotent reads and idempotent support-case creation
 - stable Server-Sent Events protocol with heartbeat comments
 - private service authentication between the product BFF and Nivora
-- real Provider readiness checks with short caching
+- real Provider and storage readiness checks with short caching
 - global concurrency and queue protection
 - Prometheus-compatible runtime metrics
 - loopback-first production deployment examples
@@ -32,13 +34,14 @@ Lumio is the first planned provider integration, but Nivora itself does not know
 Browser
   -> Product BFF (session, tenant, brand, scopes, rate limit)
      -> Nivora :3100 (Eino runtime, private)
+        -> Nivora conversation/audit database
         -> Product Provider API (authorization and business truth)
            -> Product services and database
            -> approved knowledge service :3110
               -> VikingDB
 ```
 
-Nivora does not accept a Provider URL from chat requests and does not connect to a product database or VikingDB. The configured Provider remains the source of truth.
+Nivora does not accept a Provider URL from chat requests and does not connect to a product business database or VikingDB. The configured Provider remains the source of truth.
 
 ## Run locally
 
@@ -54,6 +57,8 @@ Useful endpoints:
 curl http://127.0.0.1:3100/healthz
 curl -i http://127.0.0.1:3100/readyz
 curl http://127.0.0.1:3100/metrics
+curl -H 'X-Nivora-Key: replace-with-a-long-random-secret' \
+  http://127.0.0.1:3100/v1/conversations/conv-id/transcript
 ```
 
 Chat requests must come from a trusted BFF. The BFF must replace browser-supplied tenant and principal data with trusted server-side values.
@@ -103,6 +108,7 @@ Tool results are not forwarded to the browser. They remain inside the Agent run.
 - Use separate secrets for product-to-Nivora, Nivora-to-Provider, and Provider-to-knowledge authentication.
 - The Provider API must enforce customer ownership and redact internal fields.
 - Anonymous requests can receive only explicitly granted knowledge and case scopes.
+- Durable storage contains public messages and sanitized audit metadata only; it never stores chain of thought, bearer contexts, Tool payloads, or product recipes.
 - Nivora currently performs read operations plus idempotent `case.create` only.
 
 ## Documentation
@@ -111,6 +117,7 @@ Tool results are not forwarded to the browser. They remain inside the Agent run.
 - [Provider API v1](docs/provider-api.md)
 - [CozeLoop integration](docs/cozeloop.md)
 - [Approved VikingDB knowledge](docs/approved-knowledge.md)
+- [Durable conversation storage](docs/durable-storage.md)
 - [Customer-support evaluation](docs/evaluation.md)
 - [Volcengine production stack](docs/volcengine-production-stack.md)
 
@@ -128,6 +135,5 @@ make knowledge-eval
 
 ## Roadmap
 
-1. Add durable conversations, audit logs, and support cases in Nivora's own storage.
-2. Add the production security, load, and shadow-traffic acceptance suite.
-3. Add Eino interrupt/resume for human approval of future high-risk actions.
+1. Add the production security, load, and shadow-traffic acceptance suite.
+2. Add Eino interrupt/resume for human approval of future high-risk actions.

cmd/nivora/main.go

@@ -15,8 +15,13 @@ import (
 
 	"github.com/Nesoriel/nivora/internal/agent"
 	"github.com/Nesoriel/nivora/internal/config"
+	"github.com/Nesoriel/nivora/internal/conversation"
+	conversationhttp "github.com/Nesoriel/nivora/internal/conversation/httpapi"
+	"github.com/Nesoriel/nivora/internal/conversation/sqlstore"
+	"github.com/Nesoriel/nivora/internal/dependency"
 	"github.com/Nesoriel/nivora/internal/model/failover"
 	"github.com/Nesoriel/nivora/internal/promptpolicy"
+	"github.com/Nesoriel/nivora/internal/provider"
 	providerhttp "github.com/Nesoriel/nivora/internal/provider/httpclient"
 	"github.com/Nesoriel/nivora/internal/requestctx"
 	looptrace "github.com/Nesoriel/nivora/internal/runtrace/cozeloop"
@@ -31,6 +36,8 @@ func main() {
 		logger.Error("load configuration", "error", err)
 		os.Exit(1)
 	}
+	serviceCtx, stopBackground := context.WithCancel(context.Background())
+	defer stopBackground()
 
 	loopRuntime, loopErr := looptrace.New(cfg.CozeLoopEnabled, logger)
 	if loopErr != nil {
@@ -40,7 +47,7 @@ func main() {
 
 	var policy promptpolicy.Source = promptpolicy.Static("", "bundled-v1", "bundled")
 	if loopRuntime.Client() != nil && cfg.CozeLoopPromptKey != "" {
-		remotePolicy, promptErr := promptpolicy.Remote(context.Background(), loopRuntime.Client(), promptpolicy.RemoteConfig{
+		remotePolicy, promptErr := promptpolicy.Remote(serviceCtx, loopRuntime.Client(), promptpolicy.RemoteConfig{
 			Key:             cfg.CozeLoopPromptKey,
 			Version:         cfg.CozeLoopPromptVersion,
 			RefreshInterval: cfg.CozeLoopPromptRefresh,
@@ -55,6 +62,26 @@ func main() {
 		}
 	}
 
+	conversationStore := conversation.Nop()
+	storageEnabled := cfg.StorageDriver != "" && cfg.StorageDSN != ""
+	if storageEnabled {
+		durableStore, storageErr := sqlstore.Open(serviceCtx, cfg.StorageDriver, cfg.StorageDSN)
+		if storageErr != nil {
+			logger.Error("open durable conversation store", "error", storageErr)
+			os.Exit(1)
+		}
+		conversationStore = durableStore
+		if cfg.StorageCleanupInterval > 0 {
+			go retentionLoop(serviceCtx, conversationStore, cfg.StorageRetention, cfg.StorageCleanupInterval, logger)
+		}
+		logger.Info("durable conversation storage configured", "driver", cfg.StorageDriver)
+	} else if cfg.StorageRequired {
+		logger.Error("durable conversation storage is required but not configured")
+		os.Exit(1)
+	} else {
+		logger.Warn("durable conversation storage is disabled; production acceptance requires enabling it")
+	}
+
 	providerClient, err := providerhttp.New(
 		cfg.ProviderBaseURL,
 		cfg.ProviderSharedSecret,
@@ -65,13 +92,22 @@ func main() {
 		logger.Error("create provider client", "error", err)
 		os.Exit(1)
 	}
+	var runtimeProvider provider.Provider = providerClient
+	if storageEnabled {
+		recordedProvider, recordErr := conversation.NewProviderRecorder(runtimeProvider, conversationStore, cfg.TenantID)
+		if recordErr != nil {
+			logger.Error("create Provider audit recorder", "error", recordErr)
+			os.Exit(1)
+		}
+		runtimeProvider = recordedProvider
+	}
 
-	var runtime *agent.Service
+	var streamer conversation.Streamer
 	if cfg.ArkAPIKey != "" && len(cfg.ArkModels) > 0 {
 		models := make([]einomodel.ToolCallingChatModel, 0, len(cfg.ArkModels))
 		for _, modelID := range cfg.ArkModels {
 			modelTimeout := cfg.RequestTimeout
-			chatModel, modelErr := ark.NewChatModel(context.Background(), &ark.ChatModelConfig{
+			chatModel, modelErr := ark.NewChatModel(serviceCtx, &ark.ChatModelConfig{
 				APIKey:  cfg.ArkAPIKey,
 				Model:   modelID,
 				BaseURL: cfg.ArkBaseURL,
@@ -92,24 +128,40 @@ func main() {
 				os.Exit(1)
 			}
 		}
-		runtime, err = agent.New(
+		runtime, runtimeErr := agent.New(
 			runtimeModel,
-			providerClient,
+			runtimeProvider,
 			agent.WithPolicySource(policy),
 			agent.WithTracer(loopRuntime.Tracer()),
 			agent.WithBuildInfo(cfg.Version, cfg.Commit),
 		)
-		if err != nil {
-			logger.Error("create agent runtime", "error", err)
+		if runtimeErr != nil {
+			logger.Error("create agent runtime", "error", runtimeErr)
 			os.Exit(1)
 		}
+		streamer = runtime
+		if storageEnabled {
+			recorder, recordErr := conversation.NewRecorder(runtime, conversationStore, cfg.Version, cfg.Commit, func() (string, string) {
+				snapshot := policy.Current()
+				return snapshot.Version, snapshot.Source
+			})
+			if recordErr != nil {
+				logger.Error("create durable run recorder", "error", recordErr)
+				os.Exit(1)
+			}
+			streamer = recorder
+		}
 		logger.Info("Ark runtime configured", "model_endpoints", len(models))
 	} else {
 		logger.Warn("Ark model is not configured; health endpoints are available but readiness will fail")
 	}
 
+	checkers := []dependency.Checker{providerClient}
+	if storageEnabled {
+		checkers = append(checkers, conversationStore)
+	}
 	metrics := telemetry.New()
-	transport := httpserver.New(cfg, runtime, providerClient, metrics, logger)
+	transport := httpserver.New(cfg, streamer, dependency.New(checkers...), metrics, logger)
 	root := http.NewServeMux()
 	root.HandleFunc("GET /version", func(w http.ResponseWriter, _ *http.Request) {
 		snapshot := policy.Current()
@@ -121,8 +173,18 @@ func main() {
 			"prompt_version":   snapshot.Version,
 			"prompt_source":    snapshot.Source,
 			"cozeloop_enabled": cfg.CozeLoopEnabled && loopRuntime.Client() != nil,
+			"storage_enabled":  storageEnabled,
+			"storage_driver":   cfg.StorageDriver,
 		})
 	})
+	if storageEnabled {
+		transcriptAPI, transcriptErr := conversationhttp.New(conversationStore, cfg.TenantID, cfg.SharedSecret)
+		if transcriptErr != nil {
+			logger.Error("create transcript API", "error", transcriptErr)
+			os.Exit(1)
+		}
+		transcriptAPI.Register(root)
+	}
 	root.Handle("/", requestctx.Middleware(transport.Handler()))
 
 	server := &http.Server{
@@ -144,6 +206,7 @@ func main() {
 	}()
 
 	<-shutdown
+	stopBackground()
 	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
 	defer cancel()
 	if err := server.Shutdown(ctx); err != nil {
@@ -152,5 +215,39 @@ func main() {
 	}
 	policy.Close()
 	loopRuntime.Close(ctx)
+	if err := conversationStore.Close(); err != nil {
+		logger.Error("close conversation store", "error", err)
+		os.Exit(1)
+	}
 	logger.Info("Nivora stopped")
 }
+
+func retentionLoop(ctx context.Context, store conversation.Store, retention, interval time.Duration, logger *slog.Logger) {
+	run := func() {
+		result, err := store.DeleteBefore(ctx, time.Now().UTC().Add(-retention))
+		if err != nil {
+			if ctx.Err() == nil {
+				logger.Error("conversation retention cleanup failed", "error", err)
+			}
+			return
+		}
+		logger.Info("conversation retention cleanup completed",
+			"runs", result.Runs,
+			"messages", result.Messages,
+			"tool_audits", result.ToolAudits,
+			"support_cases", result.SupportCases,
+			"conversations", result.Conversations,
+		)
+	}
+	run()
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			run()
+		}
+	}
+}

docs/durable-storage.md

@@ -0,0 +1,86 @@
+# Durable conversation and audit storage
+
+Nivora owns customer-support conversation state, but never reads or copies a product's business database. Product facts remain behind the Provider API.
+
+## Stored data
+
+- conversation IDs and tenant IDs;
+- public user questions and final assistant answers;
+- Agent run status, Nivora version/commit, and active Prompt version/source;
+- sanitized Tool lifecycle records containing only Tool name, Tool Call ID, status, and timestamps;
+- Provider support-case ID and status for human handoff.
+
+Nivora never persists:
+
+- model chain of thought or hidden reasoning;
+- bearer contexts, API keys, or service secrets;
+- Tool arguments or unrestricted Tool results;
+- raw Provider payloads;
+- product-internal prompts, generation recipes, or other customers' data.
+
+## Drivers
+
+Development and single-node testing may use SQLite:
+
+```env
+NIVORA_STORAGE_DRIVER=sqlite
+NIVORA_STORAGE_DSN=file:/var/lib/nivora/nivora.db?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)
+```
+
+Production should use PostgreSQL:
+
+```env
+NIVORA_STORAGE_DRIVER=pgx
+NIVORA_STORAGE_DSN=postgres://nivora:password@db.internal:5432/nivora?sslmode=require
+NIVORA_STORAGE_REQUIRED=true
+```
+
+When storage is enabled it participates in `/readyz`. When `NIVORA_STORAGE_REQUIRED=true`, missing storage configuration prevents startup readiness.
+
+## Idempotency
+
+- `request_id` is the primary key for an Agent run.
+- user and assistant message IDs are deterministic: `<request_id>:user` and `<request_id>:assistant`.
+- Tool audits use `(request_id, tool_call_id)`.
+- Provider case references use `(tenant_id, provider_case_id)`.
+- replaying the same identity is a no-op;
+- reusing an identity for another tenant or conversation returns an idempotency conflict.
+
+Provider `case.create` remains protected by its separate stable idempotency key. If the Provider creates a case but the local audit write fails, the customer run fails closed; a retry must return the same Provider case before the local reference is recorded.
+
+## Tenant isolation
+
+Every transcript query requires both the configured tenant ID and conversation ID. The private API does not accept a browser-selected tenant:
+
+```http
+GET /v1/conversations/{conversation_id}/transcript
+X-Nivora-Key: <internal service key>
+```
+
+Only public user/assistant messages are returned. Tool audit records and Provider payloads are never part of the transcript.
+
+## Retention
+
+```env
+NIVORA_STORAGE_RETENTION=720h
+NIVORA_STORAGE_CLEANUP_INTERVAL=1h
+```
+
+Cleanup is transactional and restart-safe. It removes completed runs, public messages, Tool audits, old support-case references, and finally conversations that no longer have retained records.
+
+For regulated deployments, set retention according to the product's privacy policy and legal requirements. Database backups must use the same encryption, access control, deletion, and retention policy as the live database.
+
+## Migration and rollback
+
+Schema migrations run at startup inside a transaction and are recorded in `schema_migrations`.
+
+Before deployment:
+
+1. back up the database;
+2. test the migration against a production-sized copy;
+3. deploy Nivora with traffic disabled;
+4. verify `/readyz` and transcript reads;
+5. enable shadow traffic;
+6. keep the previous binary available for rollback.
+
+The initial schema is additive. Rolling back the binary does not require immediately deleting the new tables.