diff --git a/DEPLOYMENT.md b/DEPLOYMENT.md index 9f7fd79..bf7c5f4 100644 --- a/DEPLOYMENT.md +++ b/DEPLOYMENT.md @@ -223,7 +223,7 @@ display_name = "Local Chat" # For a hosted API instead, e.g.: # provider = "openai" # valid: llama-server | openai | ollama | vllm | llama-completion # endpoint = "https://api.openai.com/v1" # any OpenAI-compatible base URL -# api_key = "sk-..." # inline (test only) — or use vault_key for the encrypted vault +# api_key = "${OPENAI_API_KEY}" # expanded from the environment; there is no vault # (There is no dedicated "anthropic" provider — reach Anthropic via an # OpenAI-compatible endpoint under provider = "openai".) @@ -463,8 +463,9 @@ familiar-gateway` → health check: cd ~/repos/familiar-engine && ./familiar-deploy.sh ``` -(It assumes the repo is at `~/repos/familiar-engine` and a `familiar-gateway` -systemd unit exists.) +(It deploys the checkout it is run from — set `FAMILIAR_REPO` to deploy another +— and assumes a `familiar-gateway` systemd unit whose paths point at that same +checkout; the unit templates in `systemd/` use `~/repos/familiar-engine`.) Workspace **binary** change (Go code under `familiar-workspace/`, not static): `cd familiar-workspace && make build && sudo systemctl restart familiar-workspace`. diff --git a/README.md b/README.md index 0958220..7ca51e9 100644 --- a/README.md +++ b/README.md @@ -76,8 +76,9 @@ See `tests/e2e/MAKE_TESTS.md` for the E2E harness details. ## Configuration `config.example.toml` documents every gateway block. System prompts live in -`prompts/` as a tiered set (`base.md`, `tier_*.md`, `tool_policy.md`); point -`[system_prompt].dir` at it. The single config knobs you must set for a usable +`prompts/tiers/` as a tiered set (`base.md`, `tier_*.md`, `tool_policy.md`); +point `[system_prompt].prompt_dir` at that directory (the key is `prompt_dir`, +not `dir`; an unknown key is reported at boot and otherwise ignored). The single config knobs you must set for a usable instance: `[adapter.http].listen_addr`, `[memory].local_dsn`, at least one `[[models]]` chat model (with `"tools"` in its capabilities), `[system_prompt]`, and the `[admin]` WebAuthn relying-party (`rp_id` / `rp_origins`). diff --git a/config.example.toml b/config.example.toml index 35f37fa..a0d769c 100644 --- a/config.example.toml +++ b/config.example.toml @@ -3,9 +3,21 @@ # Copy to ~/.familiar/gateway.toml and edit for your deployment. # The gateway auto-discovers this path; override with --config . # -# Environment variable substitution is NOT supported in TOML values. -# Secrets (API keys, bot tokens) go inline; move to the vault once -# you have one set up. +# Secrets: this file holds the database password, API keys, bot tokens +# and the VAPID signing key, so keep it private (chmod 600; the gateway +# warns when others can read it). There is no vault. +# +# Environment variables: $VAR and ${VAR} are expanded in memory.local_dsn, +# model api_key and endpoint, [embedder].endpoint, the Slack tokens, the +# Brave and PirateWeather keys, the [push] VAPID keys and subject, and +# path fields (which also expand a leading ~/). Write $$ for a literal $ +# (a password containing $, say); a $ that names no set variable is +# dropped, with a warning naming the field. A model that sets +# vault_key (any name) and no api_key reads its key from the +# environment variable FAMILIAR__KEY (its id uppercased, other +# characters as _). +# +# Unknown keys are reported at boot ("... isn't a config key"). # # ─── Minimum viable config ────────────────────────────────────────── # To get a working single-user install, you need: @@ -77,12 +89,9 @@ help_notes = "" # freeform guidance injected into the tool # Keep this list small — it's meant for the owner and any accounts # that must survive a database reset, not for day-to-day users. -# OWNER-MIGRATION: the legacy "owner" canonical_id was retired in -# favour of an explicit per-user identifier (e.g. "operator", "ali"). -# Pre-migration deploys can rename their data with -# `internal/db/migrate.go`'s "owner_to_operator_rename" step; new -# deploys should seed themselves with the canonical_id they actually -# want. Examples below use "operator" — substitute your own. +# Seed yourself with the canonical_id you want (a per-user identifier +# such as "operator" or "ali"). Examples below use "operator" — +# substitute your own. # # [[identity.seed]] # platform = "cli" @@ -92,7 +101,7 @@ help_notes = "" # freeform guidance injected into the tool # # [[identity.seed]] # platform = "openai" -# platform_id = "operator@example.com" # email used in X-User-Email header +# platform_id = "operator@example.com" # canonical_id = "operator" # display_name = "Operator" # @@ -124,8 +133,8 @@ history_file = "~/.familiar/cli_history" # listener, and the gateway serves plain HTTP. Bind wider only if the # workspace runs on another host, and firewall the port to it. listen_addr = "127.0.0.1:8000" -# OWNER-MIGRATION removed the default_sender_id field. Requests with no -# X-User-Email header and no admin session cookie now return 401. +# Requests with no admin session cookie (or shard token, on the shard +# API) get 401. [adapter.slack] # Slack Socket Mode adapter. Requires app_mentions:read, chat:write, @@ -363,7 +372,6 @@ pool_size = 50 # hybrid candidates fed to the reranker window_size = 32768 output_reservation = 4096 system_prompt_ratio = 0.10 # ~10% system prompt + tool schemas -working_context_ratio = 0.05 # ~5% user profile (Layer 1) memory_ratio = 0.12 # ~12% retrieved memories tool_result_ratio = 0.12 # ~12% skill output # Per-result cap: a single tool result is head+tail truncated to this @@ -563,12 +571,11 @@ first_user_id = "operator" # Legacy single-RP form. Used when no [[admin.relying_party]] blocks # above are configured. rp_id must be the hostname the browser sees # (no scheme, no port); rp_origins must list every scheme+host+port -# the UI is loaded from. +# the UI is loaded from — the workspace's, not this gateway's (the +# workspace serves the UI; :3000 without TLS, :8443 with it). rp_id = "localhost" rp_origins = [ - "http://localhost:8000", - # "http://100.64.0.1:8000", # Tailscale - # "http://10.0.0.104:8000", # LAN + "http://localhost:3000", # "https://your-host.ts.net:8443", ] diff --git a/familiar-deploy.sh b/familiar-deploy.sh index 8cf321c..ee576b3 100755 --- a/familiar-deploy.sh +++ b/familiar-deploy.sh @@ -1,7 +1,11 @@ #!/usr/bin/env bash set -euo pipefail -REPO="${FAMILIAR_REPO:-$HOME/repos/familiar}" +# The checkout this script is in (FAMILIAR_REPO overrides). It was a +# hardcoded ~/repos/familiar while the docs and units used +# ~/repos/familiar-engine, so on a box with both it could build one tree +# and restart the other. +REPO="${FAMILIAR_REPO:-$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)}" GATEWAY="$REPO/familiar-gateway" WORKSPACE="$REPO/familiar-workspace" diff --git a/familiar-gateway/cmd/gateway/config.go b/familiar-gateway/cmd/gateway/config.go index a0c5077..24f8e87 100644 --- a/familiar-gateway/cmd/gateway/config.go +++ b/familiar-gateway/cmd/gateway/config.go @@ -1,26 +1,24 @@ package main import ( - "context" "log" "os" "path/filepath" "strings" - "time" "github.com/familiar/gateway/internal/config" "github.com/familiar/gateway/internal/engine" ) -// loadConfig finds and loads the gateway config, falling back to defaults. +// loadConfig finds and loads the gateway config, falling back to defaults: +// $FAMILIAR_HOME/gateway.toml (~/.familiar without it), then ./gateway.toml. func loadConfig(explicit string) (*config.Config, error) { if explicit != "" { return config.Load(explicit) } - home, _ := os.UserHomeDir() candidates := []string{ - filepath.Join(home, ".familiar", "gateway.toml"), + filepath.Join(config.FamiliarHome(), "gateway.toml"), "./gateway.toml", } @@ -35,9 +33,13 @@ func loadConfig(explicit string) (*config.Config, error) { return config.DefaultConfig(), nil } -// makeAPIKeyFn returns a function that resolves vault keys to API key strings. -// Resolution order: engine vault → model's api_key field → env var FAMILIAR__KEY. -func makeAPIKeyFn(eng engine.Service, models []config.ModelConfig) func(string) string { +// makeAPIKeyFn returns a function that resolves a model's vault_key to its +// API key (the router asks only for models that set one): the model's +// api_key field, else the env var FAMILIAR__KEY (the ID uppercased, +// anything but letters, digits and _ as _). There is no vault: the +// engine's VaultGet is unsupported, and asking it first cost a 3s +// timeout per resolution for nothing. +func makeAPIKeyFn(_ engine.Service, models []config.ModelConfig) func(string) string { // Build a map from model ID to ModelConfig for env-var fallback. modelsByVaultKey := make(map[string]config.ModelConfig) modelsByID := make(map[string]config.ModelConfig) @@ -49,21 +51,12 @@ func makeAPIKeyFn(eng engine.Service, models []config.ModelConfig) func(string) } return func(vaultKey string) string { - // 1. Try engine vault. - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) - defer cancel() - - val, found, err := eng.VaultGet(ctx, vaultKey) - if err == nil && found && val != "" { - return val - } - - // 2. Try direct api_key from config. + // 1. The model's api_key from config. if m, ok := modelsByVaultKey[vaultKey]; ok { if m.APIKey != "" { return m.APIKey } - // 3. Try env var FAMILIAR__KEY. + // 2. The env var FAMILIAR__KEY. envKey := "FAMILIAR_" + strings.ToUpper(strings.ReplaceAll(m.ID, "/", "_")) + "_KEY" // Replace non-alphanumeric characters with underscores. envKey = sanitizeEnvKey(envKey) diff --git a/familiar-gateway/cmd/gateway/config_test.go b/familiar-gateway/cmd/gateway/config_test.go new file mode 100644 index 0000000..e21f6ca --- /dev/null +++ b/familiar-gateway/cmd/gateway/config_test.go @@ -0,0 +1,43 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + + "github.com/familiar/gateway/internal/config" +) + +// A model's key comes from its api_key or FAMILIAR__KEY. The +// resolver asked an engine vault first, which is unsupported (a 3s +// timeout per resolution); it no longer touches the engine at all. +func TestAPIKeyFn_NoVault(t *testing.T) { + t.Setenv("FAMILIAR_OPENAI_GPT_KEY", "from-env") + fn := makeAPIKeyFn(nil, []config.ModelConfig{ + {ID: "openai/gpt", VaultKey: "gpt"}, + {ID: "inline", VaultKey: "inline", APIKey: "from-config"}, + }) + if got := fn("gpt"); got != "from-env" { + t.Errorf("env key = %q", got) + } + if got := fn("inline"); got != "from-config" { + t.Errorf("inline key = %q", got) + } +} + +// The config is looked for in $FAMILIAR_HOME (the units set it). +func TestLoadConfig_FamiliarHome(t *testing.T) { + dir := t.TempDir() + t.Setenv("FAMILIAR_HOME", dir) + body := "[[models]]\nid = \"from-home\"\nprovider = \"llama-server\"\nendpoint = \"http://m\"\nchat = true\n" + if err := os.WriteFile(filepath.Join(dir, "gateway.toml"), []byte(body), 0o600); err != nil { + t.Fatal(err) + } + cfg, err := loadConfig("") + if err != nil { + t.Fatal(err) + } + if len(cfg.Models) == 0 || cfg.Models[0].ID != "from-home" { + t.Errorf("loaded %+v, want the config in FAMILIAR_HOME", cfg.Models) + } +} diff --git a/familiar-gateway/cmd/gateway/main.go b/familiar-gateway/cmd/gateway/main.go index 018ec59..e62be6e 100644 --- a/familiar-gateway/cmd/gateway/main.go +++ b/familiar-gateway/cmd/gateway/main.go @@ -877,6 +877,30 @@ func main() { } else { log.Printf("[gateway] shutdown signal received — draining") } + // Cancel the root context on every exit path, before anything is + // closed: on an adapter error it was cancelled only by the deferred + // stop, which runs after the deferred pool.Close, so background + // loops ran against a closed pool. + stop() + // The other adapters unwind on the cancel, each draining its own + // in-flight requests; wait for them rather than closing the pool + // under them. + for i := 1; i < adapterCount; i++ { + select { + case err := <-adapterErrs: + if err != nil && !errors.Is(err, context.Canceled) { + log.Printf("[gateway] adapter error while stopping: %v", err) + } + case <-time.After(shutdownWait): + log.Printf("[gateway] %d adapter(s) still stopping after %v — closing anyway", adapterCount-i, shutdownWait) + i = adapterCount + } + } + // Post-turn work (fact extraction, summaries) runs detached from + // any request; let it finish before the pool closes. + if pl != nil && !pl.Drain(shutdownWait) { + log.Printf("[gateway] post-turn work still running after %v — closing anyway", shutdownWait) + } // Explicitly stop the consolidation cycle here, before the deferred // pool.Close runs, so a mid-pass sleep query can't race the pool // closing. Idempotent: the deferred eng.Close() calls Stop() again @@ -884,6 +908,10 @@ func main() { inProcMem.Close() } +// shutdownWait bounds each wait of a shutdown: for the remaining +// adapters to stop, and for post-turn work to finish. +const shutdownWait = 15 * time.Second + // openDBWithRetry opens the pool, retrying with backoff until it // connects or wait elapses. On a reboot the database often isn't up yet // when the gateway starts (the Postgres container comes up after it), diff --git a/familiar-gateway/cmd/gateway/main_test.go b/familiar-gateway/cmd/gateway/main_test.go index 714426a..135401e 100644 --- a/familiar-gateway/cmd/gateway/main_test.go +++ b/familiar-gateway/cmd/gateway/main_test.go @@ -57,6 +57,7 @@ enabled = false [[models]] id = "test/dummy" +provider = "llama-server" endpoint = "http://127.0.0.1:1" role = "small" `, filepath.Join(dir, "skills")) diff --git a/familiar-gateway/internal/adapter/native/native.go b/familiar-gateway/internal/adapter/native/native.go index f059c58..5210661 100644 --- a/familiar-gateway/internal/adapter/native/native.go +++ b/familiar-gateway/internal/adapter/native/native.go @@ -351,9 +351,12 @@ func listenAddr(configured string) string { return configured } +// shutdownGrace is how long a shutdown waits for in-flight requests. +const shutdownGrace = 5 * time.Second + // Run starts the HTTP server. Blocks until ctx is cancelled or the -// server fails. Graceful shutdown waits up to 5s for in-flight -// requests. +// server fails; on cancellation it returns once in-flight requests have +// finished, or after shutdownGrace. func (a *Adapter) Run(ctx context.Context) error { addr := listenAddr(a.cfg.ListenAddr) @@ -371,19 +374,34 @@ func (a *Adapter) Run(ctx context.Context) error { IdleTimeout: 60 * time.Second, } + ln, err := net.Listen("tcp", addr) + if err != nil { + return fmt.Errorf("HTTP server error: %w", err) + } + log.Printf("[http] Listening on %s", addr) + log.Printf("[http] Native chat at POST /api/chat; memlog at /events/{session_id}") + return serve(ctx, server, ln, shutdownGrace) +} + +// serve runs server on ln until ctx is cancelled, then shuts it down and +// returns once the in-flight requests finish (or grace runs out). +// Serve returns as soon as Shutdown begins, not when it's done, so +// returning then let the process exit mid-request (a wiki save cut off +// in its transaction) while the drain was still going. +func serve(ctx context.Context, server *http.Server, ln net.Listener, grace time.Duration) error { + drained := make(chan error, 1) go func() { <-ctx.Done() - shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + shutdownCtx, cancel := context.WithTimeout(context.Background(), grace) defer cancel() - server.Shutdown(shutdownCtx) + drained <- server.Shutdown(shutdownCtx) }() - - log.Printf("[http] Listening on %s", addr) - log.Printf("[http] Native chat at POST /api/chat; memlog at /events/{session_id}") - - if err := server.ListenAndServe(); err != http.ErrServerClosed { + if err := server.Serve(ln); err != http.ErrServerClosed { return fmt.Errorf("HTTP server error: %w", err) } + if err := <-drained; err != nil { + log.Printf("[http] shutdown: in-flight requests still running after %v were cut (%v)", grace, err) + } return nil } diff --git a/familiar-gateway/internal/adapter/native/shutdown_test.go b/familiar-gateway/internal/adapter/native/shutdown_test.go new file mode 100644 index 0000000..7827dde --- /dev/null +++ b/familiar-gateway/internal/adapter/native/shutdown_test.go @@ -0,0 +1,46 @@ +package native + +import ( + "context" + "net" + "net/http" + "sync/atomic" + "testing" + "time" +) + +// A shutdown returns once the in-flight request has finished. Serve +// returns as soon as Shutdown begins, and Run returned then, so the +// process exited mid-request while the drain was still going. +func TestServe_WaitsForInFlightRequests(t *testing.T) { + var finished atomic.Bool + started := make(chan struct{}) + srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(started) + time.Sleep(300 * time.Millisecond) + finished.Store(true) + w.WriteHeader(http.StatusNoContent) + })} + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- serve(ctx, srv, ln, 5*time.Second) }() + + go func() { _, _ = http.Get("http://" + ln.Addr().String() + "/") }() + <-started + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("serve: %v", err) + } + if !finished.Load() { + t.Error("serve returned before the in-flight request finished") + } + case <-time.After(5 * time.Second): + t.Fatal("serve never returned") + } +} diff --git a/familiar-gateway/internal/config/config.go b/familiar-gateway/internal/config/config.go index e1ba463..9307121 100644 --- a/familiar-gateway/internal/config/config.go +++ b/familiar-gateway/internal/config/config.go @@ -62,6 +62,9 @@ type Config struct { // ignored names the keys in the file that nothing reads (see // ignoredKeys); Validate warns about each. ignored []string `toml:"-"` + // unknown names the keys in the file that match no field (typos, + // keys that were renamed or removed); Validate warns about each. + unknown []string `toml:"-"` } // PushConfig holds the VAPID keypair the Web Push subsystem signs with @@ -1165,7 +1168,7 @@ type SlackConfig struct { // rejected with 401; the native adapter requires a workspace // session cookie (the X-User-Email path was removed). type HTTPConfig struct { - ListenAddr string `toml:"listen_addr"` // e.g., "0.0.0.0:8000" + ListenAddr string `toml:"listen_addr"` // default 127.0.0.1:8000 (native.DefaultListenAddr) // Identity / auth on /api/chat is the admin session cookie only — // a server-verified canonical identity. Two prior knobs were @@ -1193,8 +1196,7 @@ type EmbedderConfig struct { // DefaultConfig returns a config with sensible defaults. func DefaultConfig() *Config { - home, _ := os.UserHomeDir() - _ = home // home was used by the legacy engine socket path + home := FamiliarHome() return &Config{ Sleep: DefaultSleepConfig(), Node: NodeConfig{ @@ -1254,7 +1256,7 @@ func DefaultConfig() *Config { }, Adapter: AdapterConfig{ CLI: CLIConfig{ - HistoryFile: filepath.Join(home, ".familiar", "cli_history"), + HistoryFile: filepath.Join(home, "cli_history"), Prompt: "> ", }, }, @@ -1263,12 +1265,24 @@ func DefaultConfig() *Config { Dimension: 768, }, SystemPrompt: SystemPromptConfig{ - Dir: "~/.familiar/prompts", - File: "~/.familiar/system_prompt.md", + Dir: filepath.Join(home, "prompts"), + File: filepath.Join(home, "system_prompt.md"), }, } } +// FamiliarHome is the gateway's home directory: $FAMILIAR_HOME when +// set, else ~/.familiar. The config file is looked for there and the +// default skills, media, prompt and history paths live under it. The +// systemd units and scripts set FAMILIAR_HOME, and nothing read it. +func FamiliarHome() string { + if h := os.Getenv("FAMILIAR_HOME"); h != "" { + return h + } + home, _ := os.UserHomeDir() + return filepath.Join(home, ".familiar") +} + // Load reads and parses a TOML config file, applying env var expansion and // tilde expansion to string fields. Falls back to defaults for unset values. func Load(path string) (*Config, error) { @@ -1284,6 +1298,13 @@ func Load(path string) (*Config, error) { return nil, fmt.Errorf("parsing config %s: %w", path, err) } cfg.ignored = ignoredKnobs(md) + cfg.unknown = UnknownKeys(md) + + // The file holds the DSN password, API keys, bot tokens and the + // VAPID signing key. + if fi, err := os.Stat(path); err == nil && fi.Mode().Perm()&0o077 != 0 { + log.Printf("[config] warning: %s is readable by other users (mode %04o); it holds secrets — chmod 600 it", path, fi.Mode().Perm()) + } // Apply expansions. expandConfig(cfg) @@ -1550,7 +1571,20 @@ const legacyEmbedderModelID = "embedder/legacy" // participates in health checks and can be given a backup. No-op when // [roles.embedder] is set explicitly or no [embedder].endpoint exists. func (c *Config) normalizeEmbedderRole() { - if !c.Roles.Embedder.IsEmpty() || c.Embedder.Endpoint == "" { + if !c.Roles.Embedder.IsEmpty() { + return + } + // A model tagged role = "embedder" is the embedder. The tag was + // accepted and documented but read by nothing, so facts were stored + // without vectors. + for _, m := range c.Models { + if m.Role == ModelRoleEmbedder { + c.Roles.Embedder.Primary = m.ID + log.Printf("[config] embedder: model %q (role=embedder) is the [roles.embedder] primary", m.ID) + return + } + } + if c.Embedder.Endpoint == "" { return } exists := false @@ -1575,55 +1609,55 @@ func (c *Config) normalizeEmbedderRole() { c.Embedder.Endpoint, legacyEmbedderModelID) } -// expandConfig applies tilde and env var expansion to path-like fields. +// expandConfig applies tilde and env var expansion to path-like fields +// and to the fields that carry secrets or endpoints (see expandEnv). func expandConfig(cfg *Config) { - cfg.Adapter.CLI.HistoryFile = expandPath(cfg.Adapter.CLI.HistoryFile) - cfg.Adapter.CLI.Prompt = expandEnv(cfg.Adapter.CLI.Prompt) - cfg.Embedder.Endpoint = expandEnv(cfg.Embedder.Endpoint) + cfg.Adapter.CLI.HistoryFile = expandPath("adapter.cli.history_file", cfg.Adapter.CLI.HistoryFile) + cfg.Adapter.CLI.Prompt = expandEnv("adapter.cli.prompt", cfg.Adapter.CLI.Prompt) + cfg.Embedder.Endpoint = expandEnv("embedder.endpoint", cfg.Embedder.Endpoint) for i := range cfg.Models { - cfg.Models[i].APIKey = expandEnv(cfg.Models[i].APIKey) - cfg.Models[i].Endpoint = expandEnv(cfg.Models[i].Endpoint) + field := fmt.Sprintf("models[%d] (%s)", i, cfg.Models[i].ID) + cfg.Models[i].APIKey = expandEnv(field+" api_key", cfg.Models[i].APIKey) + cfg.Models[i].Endpoint = expandEnv(field+" endpoint", cfg.Models[i].Endpoint) } - cfg.Sidecar.SocketPath = expandPath(cfg.Sidecar.SocketPath) - cfg.Memory.LocalDSN = expandEnv(cfg.Memory.LocalDSN) - - cfg.Adapter.Slack.BotToken = expandEnv(cfg.Adapter.Slack.BotToken) - cfg.Adapter.Slack.AppToken = expandEnv(cfg.Adapter.Slack.AppToken) + cfg.Sidecar.SocketPath = expandPath("sidecar.socket_path", cfg.Sidecar.SocketPath) + cfg.Memory.LocalDSN = expandEnv("memory.local_dsn", cfg.Memory.LocalDSN) - cfg.Push.VAPIDPrivateKey = expandEnv(cfg.Push.VAPIDPrivateKey) + cfg.Adapter.Slack.BotToken = expandEnv("adapter.slack.bot_token", cfg.Adapter.Slack.BotToken) + cfg.Adapter.Slack.AppToken = expandEnv("adapter.slack.app_token", cfg.Adapter.Slack.AppToken) - cfg.SystemPrompt.Dir = expandPath(cfg.SystemPrompt.Dir) - cfg.SystemPrompt.File = expandPath(cfg.SystemPrompt.File) + cfg.SystemPrompt.Dir = expandPath("system_prompt.prompt_dir", cfg.SystemPrompt.Dir) + cfg.SystemPrompt.File = expandPath("system_prompt.file", cfg.SystemPrompt.File) if cfg.Skills.Dir == "" { - cfg.Skills.Dir = "~/.familiar/skills" + cfg.Skills.Dir = filepath.Join(FamiliarHome(), "skills") } - cfg.Skills.Dir = expandPath(cfg.Skills.Dir) + cfg.Skills.Dir = expandPath("skills.dir", cfg.Skills.Dir) if cfg.Media.Dir == "" { - cfg.Media.Dir = "~/.familiar/media" + cfg.Media.Dir = filepath.Join(FamiliarHome(), "media") } - cfg.Media.Dir = expandPath(cfg.Media.Dir) + cfg.Media.Dir = expandPath("media.dir", cfg.Media.Dir) if cfg.Media.MaxUploadMB <= 0 { cfg.Media.MaxUploadMB = 10 } - cfg.Tools.Brave.APIKey = expandEnv(cfg.Tools.Brave.APIKey) - cfg.Tools.PirateWeather.APIKey = expandEnv(cfg.Tools.PirateWeather.APIKey) + cfg.Tools.Brave.APIKey = expandEnv("tools.brave.api_key", cfg.Tools.Brave.APIKey) + cfg.Tools.PirateWeather.APIKey = expandEnv("tools.pirateweather.api_key", cfg.Tools.PirateWeather.APIKey) // Web Push VAPID keys — the private key is a secret, expand it from // the environment. Default the contact subject when push is on. - cfg.Push.VAPIDPublicKey = expandEnv(cfg.Push.VAPIDPublicKey) - cfg.Push.VAPIDPrivateKey = expandEnv(cfg.Push.VAPIDPrivateKey) - cfg.Push.Subject = expandEnv(cfg.Push.Subject) + cfg.Push.VAPIDPublicKey = expandEnv("push.vapid_public_key", cfg.Push.VAPIDPublicKey) + cfg.Push.VAPIDPrivateKey = expandEnv("push.vapid_private_key", cfg.Push.VAPIDPrivateKey) + cfg.Push.Subject = expandEnv("push.subject", cfg.Push.Subject) if cfg.Push.Enabled() && cfg.Push.Subject == "" { cfg.Push.Subject = "mailto:admin@example.com" } } // expandPath expands ~/ prefix and env vars in a path string. -func expandPath(s string) string { - s = expandEnv(s) +func expandPath(field, s string) string { + s = expandEnv(field, s) if strings.HasPrefix(s, "~/") { home, err := os.UserHomeDir() if err == nil { @@ -1633,7 +1667,27 @@ func expandPath(s string) string { return s } -// expandEnv expands $VAR and ${VAR} patterns. -func expandEnv(s string) string { - return os.ExpandEnv(s) +// expandEnv expands $VAR and ${VAR} in a config value; $$ is a literal +// $. The fields expanded are listed in expandConfig. A $ that doesn't +// name a set variable expands to nothing, as before, but is now +// reported (by field, never by value): a password or key containing $ +// lost characters silently, and the example said nothing was expanded. +func expandEnv(field, s string) string { + if !strings.Contains(s, "$") { + return s + } + const literal = "\x00literal-dollar\x00" + s = strings.ReplaceAll(s, "$$", literal) + unset := false + s = os.Expand(s, func(name string) string { + v, ok := os.LookupEnv(name) + if !ok { + unset = true + } + return v + }) + if unset { + log.Printf("[config] warning: %s has a $ that doesn't name a set environment variable; that part was dropped (write $$ for a literal $)", field) + } + return strings.ReplaceAll(s, literal, "$") } diff --git a/familiar-gateway/internal/config/config_test.go b/familiar-gateway/internal/config/config_test.go index d01b519..3b76499 100644 --- a/familiar-gateway/internal/config/config_test.go +++ b/familiar-gateway/internal/config/config_test.go @@ -26,7 +26,7 @@ func TestExpandPath(t *testing.T) { t.Skip("cannot determine home dir") } - got := expandPath("~/some/path") + got := expandPath("test", "~/some/path") expected := filepath.Join(home, "some/path") if got != expected { t.Fatalf("expected %q, got %q", expected, got) @@ -34,7 +34,7 @@ func TestExpandPath(t *testing.T) { } func TestExpandPathNoTilde(t *testing.T) { - got := expandPath("/absolute/path") + got := expandPath("test", "/absolute/path") if got != "/absolute/path" { t.Fatalf("expected /absolute/path, got %q", got) } @@ -72,7 +72,7 @@ fallback_model = "test-model" func TestExpandEnv(t *testing.T) { t.Setenv("FAMILIAR_TEST_VAR", "hello") - got := expandEnv("prefix-$FAMILIAR_TEST_VAR-suffix") + got := expandEnv("test", "prefix-$FAMILIAR_TEST_VAR-suffix") if !strings.Contains(got, "hello") { t.Fatalf("expected env expansion, got %q", got) } diff --git a/familiar-gateway/internal/config/example_check_test.go b/familiar-gateway/internal/config/example_check_test.go index a986e1a..c2499c2 100644 --- a/familiar-gateway/internal/config/example_check_test.go +++ b/familiar-gateway/internal/config/example_check_test.go @@ -21,6 +21,12 @@ func TestConfigExampleParsesAndValidates(t *testing.T) { if err := cfg.Validate(); err != nil { t.Fatalf("config.example.toml does not validate: %v", err) } + // Every key it shows is one the gateway reads: an unknown key is + // ignored at boot, and operators copy them (working_context_ratio + // sat here long after nothing read it). + if len(cfg.unknown) > 0 { + t.Errorf("config.example.toml sets keys the gateway doesn't know: %v", cfg.unknown) + } // The example's chat role should resolve to the heavy backend it // documents, via the chat=true/lex-order derivation. if cfg.Roles.Chat.Primary == "" { diff --git a/familiar-gateway/internal/config/platform_test.go b/familiar-gateway/internal/config/platform_test.go new file mode 100644 index 0000000..de3aa4e --- /dev/null +++ b/familiar-gateway/internal/config/platform_test.go @@ -0,0 +1,152 @@ +package config + +import ( + "bytes" + "log" + "os" + "path/filepath" + "strings" + "testing" +) + +// captureLog collects what the package logs while f runs. +func captureLog(f func()) string { + var buf bytes.Buffer + old := log.Writer() + log.SetOutput(&buf) + defer log.SetOutput(old) + f() + return buf.String() +} + +// $$ is a literal $, and a $ naming no set variable is reported by +// field, not value. A password containing $ lost characters silently, +// while the example said nothing was expanded. +func TestExpandEnv_EscapeAndWarning(t *testing.T) { + t.Setenv("FAMILIAR_TEST_PART", "v") + if got := expandEnv("f", "a$$b${FAMILIAR_TEST_PART}"); got != "a$bv" { + t.Errorf("expandEnv = %q, want a$bv", got) + } + var got string + out := captureLog(func() { got = expandEnv("memory.local_dsn", "postgresql://u:pa$$w0rd$ecret@h/db") }) + if got != "postgresql://u:pa$w0rd@h/db" { + t.Errorf("expandEnv = %q", got) + } + if !strings.Contains(out, "memory.local_dsn") || strings.Contains(out, "ecret") || strings.Contains(out, "w0rd") { + t.Errorf("warning should name the field and nothing of the value: %q", out) + } + if out := captureLog(func() { expandEnv("f", "plain") }); out != "" { + t.Errorf("a value without $ warned: %q", out) + } +} + +// The threshold ordering is checked on the values that run: an unset +// supersede_threshold runs at its default and went unchecked. +func TestValidate_ThresholdsUseDefaults(t *testing.T) { + base := func() *Config { + c := DefaultConfig() + c.Models = []ModelConfig{{ID: "m", Endpoint: "http://m", Provider: "llama-server", Chat: true}} + c.normalizeRoles() + return c + } + c := base() + c.Memory.RelevanceThreshold = 0.8 // above the 0.75 supersede default + if err := c.Validate(); err == nil || !strings.Contains(err.Error(), "supersede_threshold") { + t.Errorf("relevance above the default supersede threshold: err = %v", err) + } + c = base() + c.Memory.RelevanceThreshold = 0.6 + c.Memory.DedupThreshold = 0.7 // below the 0.75 supersede default + if err := c.Validate(); err == nil || !strings.Contains(err.Error(), "supersede_threshold") { + t.Errorf("dedup below the default supersede threshold: err = %v", err) + } + if err := base().Validate(); err != nil { + t.Errorf("defaults don't validate: %v", err) + } +} + +// A provider or formatter the router can't build fails at boot, not on +// every call (the removed "anthropic" provider booted fine). +func TestValidate_UnknownProviderAndFormatter(t *testing.T) { + for _, m := range []ModelConfig{ + {ID: "a", Endpoint: "e", Provider: "anthropic", Chat: true}, + {ID: "b", Endpoint: "e", Provider: "llama_server", Chat: true}, + {ID: "c", Endpoint: "e", Provider: "", Chat: true}, + {ID: "d", Endpoint: "e", Provider: "llama-completion", Formatter: "gemma", Chat: true}, + } { + c := DefaultConfig() + c.Models = []ModelConfig{m} + c.normalizeRoles() + if err := c.Validate(); err == nil || !strings.Contains(err.Error(), "unknown") { + t.Errorf("model %+v: err = %v, want unknown provider/formatter", m, err) + } + } + c := DefaultConfig() + c.Models = []ModelConfig{{ID: "ok", Endpoint: "e", Provider: "llama-completion", Formatter: "cohere2", Chat: true}} + c.normalizeRoles() + if err := c.Validate(); err != nil { + t.Errorf("a known formatter failed: %v", err) + } +} + +// role = "embedder" was accepted and documented but read by nothing. +func TestNormalizeRoles_EmbedderRoleTag(t *testing.T) { + c := DefaultConfig() + c.Embedder = EmbedderConfig{} + c.Models = []ModelConfig{ + {ID: "chat", Endpoint: "e", Provider: "llama-server", Chat: true}, + {ID: "embed/x", Endpoint: "e", Provider: "embeddings", Role: ModelRoleEmbedder, Dimension: 768}, + } + c.normalizeRoles() + if c.Roles.Embedder.Primary != "embed/x" { + t.Errorf("embedder primary = %q, want the model tagged role=embedder", c.Roles.Embedder.Primary) + } +} + +// Unknown keys are reported (a README told operators to set +// [system_prompt].dir, which nothing reads), and so is a config file +// others can read. +func TestLoad_WarnsOnUnknownKeysAndLoosePermissions(t *testing.T) { + path := filepath.Join(t.TempDir(), "gateway.toml") + body := "[system_prompt]\ndir = \"/x\"\n\n[[models]]\nid = \"m\"\nprovider = \"llama-server\"\nendpoint = \"http://m\"\nchat = true\n" + if err := os.WriteFile(path, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + var cfg *Config + out := captureLog(func() { + var err error + if cfg, err = Load(path); err != nil { + t.Fatal(err) + } + _ = cfg.Validate() + }) + if !strings.Contains(out, "system_prompt.dir isn't a config key") { + t.Errorf("unknown key not reported: %q", out) + } + if !strings.Contains(out, "readable by other users") { + t.Errorf("0644 config not reported: %q", out) + } + if err := os.Chmod(path, 0o600); err != nil { + t.Fatal(err) + } + if out := captureLog(func() { _, _ = Load(path) }); strings.Contains(out, "readable by other users") { + t.Errorf("0600 config reported: %q", out) + } +} + +// FAMILIAR_HOME moves the default paths (the units and scripts set it, +// and nothing read it). +func TestFamiliarHome(t *testing.T) { + dir := t.TempDir() + t.Setenv("FAMILIAR_HOME", dir) + c := DefaultConfig() + expandConfig(c) + for name, got := range map[string]string{ + "prompt file": c.SystemPrompt.File, "prompt dir": c.SystemPrompt.Dir, + "skills": c.Skills.Dir, "media": c.Media.Dir, "history": c.Adapter.CLI.HistoryFile, + } { + if !strings.HasPrefix(got, dir) { + t.Errorf("%s = %q, want under FAMILIAR_HOME", name, got) + } + } +} diff --git a/familiar-gateway/internal/config/roles_test.go b/familiar-gateway/internal/config/roles_test.go index db89c95..e816395 100644 --- a/familiar-gateway/internal/config/roles_test.go +++ b/familiar-gateway/internal/config/roles_test.go @@ -226,7 +226,7 @@ func TestValidateAcceptsMatchedEmbedderChain(t *testing.T) { c.Models = []ModelConfig{ {ID: "ea", Endpoint: "e", Provider: "embeddings", Model: "nomic", Dimension: 768}, {ID: "eb", Endpoint: "e", Provider: "embeddings", Model: "nomic", Dimension: 768}, - {ID: "chat", Endpoint: "e", Chat: true}, + {ID: "chat", Endpoint: "e", Provider: "llama-server", Chat: true}, } c.Roles.Embedder = RoleChain{Primary: "ea", Backup: "eb"} c.normalizeRoles() diff --git a/familiar-gateway/internal/config/validate.go b/familiar-gateway/internal/config/validate.go index ba0f798..57adcda 100644 --- a/familiar-gateway/internal/config/validate.go +++ b/familiar-gateway/internal/config/validate.go @@ -3,6 +3,7 @@ package config import ( "fmt" "log" + "sort" "strings" "github.com/BurntSushi/toml" @@ -59,25 +60,30 @@ func (c *Config) Validate() error { if c.Memory.DedupThreshold < 0 || c.Memory.DedupThreshold > 1 { return fmt.Errorf("memory.dedup_threshold must be in [0,1] (got %v)", c.Memory.DedupThreshold) } - if c.Memory.DedupThreshold > 0 && c.Memory.DedupThreshold < c.Memory.RelevanceThreshold { - return fmt.Errorf("memory.dedup_threshold (%v) must be >= memory.relevance_threshold (%v)", - c.Memory.DedupThreshold, c.Memory.RelevanceThreshold) - } if c.Memory.SupersedeThreshold < 0 || c.Memory.SupersedeThreshold > 1 { return fmt.Errorf("memory.supersede_threshold must be in [0,1] (got %v)", c.Memory.SupersedeThreshold) } + // The ordering relevance <= supersede <= dedup is checked on the + // values that run, defaults included: an unset supersede_threshold + // runs at its default, which went unchecked against a raised + // relevance or lowered dedup threshold. + relevance := c.Memory.RelevanceThreshold + supersede := c.Memory.SupersedeThresholdOrDefault() + dedup := c.Memory.DedupThresholdOrDefault() + if dedup < relevance { + return fmt.Errorf("memory.dedup_threshold (%v) must be >= memory.relevance_threshold (%v)", dedup, relevance) + } // Superseding hides a row, so it must demand at least as much // confidence as merely retrieving one. A floor below the retrieval // threshold would let an extraction bury facts it would not even have // been shown. - if c.Memory.SupersedeThreshold > 0 && c.Memory.SupersedeThreshold < c.Memory.RelevanceThreshold { - return fmt.Errorf("memory.supersede_threshold (%v) must be >= memory.relevance_threshold (%v) — superseding hides a fact, so it needs more confidence than retrieving one", - c.Memory.SupersedeThreshold, c.Memory.RelevanceThreshold) + if supersede < relevance { + return fmt.Errorf("memory.supersede_threshold (%v%s) must be >= memory.relevance_threshold (%v) — superseding hides a fact, so it needs more confidence than retrieving one", + supersede, defaultNote(c.Memory.SupersedeThreshold), relevance) } - if c.Memory.SupersedeThreshold > 0 && c.Memory.DedupThreshold > 0 && - c.Memory.SupersedeThreshold > c.Memory.DedupThreshold { - return fmt.Errorf("memory.supersede_threshold (%v) must be <= memory.dedup_threshold (%v) — a fact similar enough to be a duplicate is skipped before it can supersede anything", - c.Memory.SupersedeThreshold, c.Memory.DedupThreshold) + if supersede > dedup { + return fmt.Errorf("memory.supersede_threshold (%v%s) must be <= memory.dedup_threshold (%v) — a fact similar enough to be a duplicate is skipped before it can supersede anything", + supersede, defaultNote(c.Memory.SupersedeThreshold), dedup) } // At least one model must be configured and each entry must have an @@ -95,6 +101,14 @@ func (c *Config) Validate() error { if m.Endpoint == "" { return fmt.Errorf("models[%d] (%s): endpoint is required", i, m.ID) } + // Checked here, not on the first call: a typo or a removed + // provider ("anthropic") booted fine and failed every request. + if !KnownProviders[m.Provider] { + return fmt.Errorf("models[%d] (%s): unknown provider %q (want one of %s)", i, m.ID, m.Provider, knownList(KnownProviders)) + } + if m.Provider == "llama-completion" && !KnownFormatters[m.Formatter] { + return fmt.Errorf("models[%d] (%s): unknown formatter %q (want one of %s)", i, m.ID, m.Formatter, knownList(KnownFormatters)) + } modelIDs[m.ID] = struct{}{} // Role validation. Reject unknown role strings so a typo @@ -227,6 +241,9 @@ func (c *Config) Validate() error { for _, knob := range c.ignored { log.Printf("[config] warning: %s is set but does nothing; remove it", knob) } + for _, key := range c.unknown { + log.Printf("[config] warning: %s isn't a config key (a typo, or renamed or removed) and is ignored", key) + } // Heartbeat tuning — coerce non-positive values to defaults with a // warning rather than erroring; a bad interval shouldn't block boot. @@ -378,6 +395,44 @@ var ignoredKeys = [][]string{ {"roles", "condense"}, } +// KnownProviders are the [[models]] provider values the router builds +// (router.buildProvider; a router test keeps the two in step). +var KnownProviders = map[string]bool{ + "llama-server": true, "openai": true, "ollama": true, "vllm": true, + "embeddings": true, "llama-completion": true, +} + +// KnownFormatters are the llama-completion formatters (router.pickFormatter). +var KnownFormatters = map[string]bool{"": true, "qwen35": true, "cohere2": true} + +func knownList(m map[string]bool) string { + var out []string + for k := range m { + if k != "" { + out = append(out, fmt.Sprintf("%q", k)) + } + } + sort.Strings(out) + return strings.Join(out, ", ") +} + +// defaultNote marks a threshold that wasn't set. +func defaultNote(set float64) string { + if set > 0 { + return "" + } + return ", the default" +} + +// UnknownKeys lists the keys in the file that match no config field. +func UnknownKeys(md toml.MetaData) []string { + var out []string + for _, k := range md.Undecoded() { + out = append(out, k.String()) + } + return out +} + // ignoredKnobs names the ignoredKeys the file sets. func ignoredKnobs(md toml.MetaData) []string { var out []string diff --git a/familiar-gateway/internal/db/db.go b/familiar-gateway/internal/db/db.go index 897925b..8415a7e 100644 --- a/familiar-gateway/internal/db/db.go +++ b/familiar-gateway/internal/db/db.go @@ -30,7 +30,7 @@ type Pool struct { func Open(dsn string) (*Pool, error) { sqlDB, err := sql.Open("postgres", dsn) if err != nil { - return nil, fmt.Errorf("db: open postgres: %w", err) + return nil, fmt.Errorf("db: open postgres: %s", SanitizeDSN(err.Error())) } sqlDB.SetMaxOpenConns(5) @@ -41,7 +41,10 @@ func Open(dsn string) (*Pool, error) { defer cancel() if err := sqlDB.PingContext(ctx); err != nil { sqlDB.Close() - return nil, fmt.Errorf("db: ping postgres: %w", err) + // The driver's errors can quote the DSN (a malformed URL's parse + // error carries it whole), password included, and this ends up + // in the log. + return nil, fmt.Errorf("db: ping postgres: %s", SanitizeDSN(err.Error())) } log.Printf("[memory] connected to pgvector at %s", SanitizeDSN(dsn)) @@ -56,14 +59,18 @@ func (p *Pool) Close() error { return p.DB.Close() } -// SanitizeDSN strips the password from a Postgres DSN for logging. -// URL form (postgres://user:pass@host/db) is rewritten to mask the -// password segment; DSNs without a `user:pass@` block pass through -// unchanged, and key-value form `password=...` is also masked. +// SanitizeDSN strips the password from a Postgres DSN, or from text +// that quotes one, for logging. URL form (postgres://user:pass@host/db) +// is rewritten to mask the password segment, up to the last @ before +// whitespace (a password can contain one); DSNs without a `user:pass@` +// block pass through unchanged, and key-value form `password=...` is +// also masked, quoted values with spaces whole. func SanitizeDSN(dsn string) string { - urlRe := regexp.MustCompile(`(postgres(?:ql)?://[^:@/]+):[^@]*@`) - out := urlRe.ReplaceAllString(dsn, "${1}:***@") - kvRe := regexp.MustCompile(`password=[^\s]+`) - out = kvRe.ReplaceAllString(out, "password=***") - return out + out := dsnURLPassword.ReplaceAllString(dsn, "${1}:***@") + return dsnKVPassword.ReplaceAllString(out, "password=***") } + +var ( + dsnURLPassword = regexp.MustCompile(`(postgres(?:ql)?://[^:@/\s]+):\S*@`) + dsnKVPassword = regexp.MustCompile(`password\s*=\s*(?:'(?:[^'\\]|\\.)*'|[^\s]+)`) +) diff --git a/familiar-gateway/internal/db/migrate.go b/familiar-gateway/internal/db/migrate.go index 08d2a6f..f95814a 100644 --- a/familiar-gateway/internal/db/migrate.go +++ b/familiar-gateway/internal/db/migrate.go @@ -56,14 +56,18 @@ CREATE TABLE IF NOT EXISTS user_profiles ( // those are facts and re-accrete as memory rows going forward. // Wrapped in a DO block so it's idempotent across boots: the // second run finds working_context already gone and just - // ensures the typed column exists. + // ensures the typed column exists. The probe is confined to the + // schema being migrated: another schema's user_profiles (public, + // on a database migrated into a separate schema) made it try to + // drop a column this one doesn't have. name: "user_profiles_scrub_working_context", ddl: ` DO $$ BEGIN IF EXISTS ( SELECT 1 FROM information_schema.columns - WHERE table_name = 'user_profiles' AND column_name = 'working_context' + WHERE table_schema = current_schema() + AND table_name = 'user_profiles' AND column_name = 'working_context' ) THEN ALTER TABLE user_profiles ADD COLUMN IF NOT EXISTS user_prompt TEXT NOT NULL DEFAULT ''; UPDATE user_profiles @@ -284,7 +288,7 @@ CREATE INDEX IF NOT EXISTS idx_admin_sessions_expires_at // // user_id follows the same visibility rule as memories: NULL is // global, non-NULL is owned by one canonical user. The unique - // index on (subject, predicate, user_id_key) lets UpsertRelationships + // keys (relationships_links_multi, below) let UpsertRelationships // do ON CONFLICT DO UPDATE when the object changes (IP moved, // version bumped, etc.); user_id_key collapses NULL to '' so // Postgres treats global rows as a single conflict class. @@ -302,8 +306,6 @@ CREATE TABLE IF NOT EXISTS relationships ( created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); -CREATE UNIQUE INDEX IF NOT EXISTS idx_rel_subject_pred_user - ON relationships (subject, predicate, user_id_key); CREATE INDEX IF NOT EXISTS idx_rel_subject ON relationships (subject); CREATE INDEX IF NOT EXISTS idx_rel_object ON relationships (object); CREATE INDEX IF NOT EXISTS idx_rel_user ON relationships (user_id) WHERE user_id IS NOT NULL;`, @@ -1341,9 +1343,11 @@ CREATE INDEX IF NOT EXISTS idx_push_subscriptions_user // (tool_call_id referenced before tool_ids_seen is built). // // A BIGSERIAL seq column gives a globally monotonic insertion - // order independent of timestamp collisions. Backfill existing - // rows by created_at then id so historical conversations keep - // a stable (if approximate) order. + // order independent of timestamp collisions. Rows that existed + // when the column was added got their seq in physical (heap) + // order, not by created_at: on a database where deleted rows' + // space was reused, some older conversations can hydrate out of + // order. Rows written since are in insertion order. name: "messages_seq_ordering", ddl: ` ALTER TABLE messages ADD COLUMN IF NOT EXISTS seq BIGSERIAL; @@ -1692,11 +1696,32 @@ ALTER TABLE admin_sessions ADD COLUMN IF NOT EXISTS absolute_expires_at TIMESTAM // relationships_object_lowercase: entity names are lowercase // everywhere they're looked up, and subjects always were; objects // kept the extractor's case, so "Acme" was a second entity no - // lookup, merge or delete could reach. The unique index is - // (subject, predicate, owner), so this can't collide. + // lookup, merge or delete could reach. The key for these rows is + // (subject, predicate, owner), so this can't collide. links_to + // rows are keyed by their object (relationships_links_multi) and + // left alone: their objects are page keys built from slugs, + // lowercase already, and lowercasing one that differed only in + // case from another would fail the boot on every restart. name: "relationships_object_lowercase", ddl: ` -UPDATE relationships SET object = lower(object) WHERE object <> lower(object);`, +UPDATE relationships SET object = lower(object) + WHERE object <> lower(object) AND predicate <> 'links_to';`, + }, + { + // relationships_links_multi: a triple was unique on (subject, + // predicate, owner), so a page's links_to edges overwrote each + // other and only its last link survived. links_to is keyed by its + // object too; every other predicate keeps one object per subject + // (an IP moved, a deadline changed). The old index is dropped + // once, after the two that replace it exist; existing rows + // already satisfy both. + name: "relationships_links_multi", + ddl: ` +CREATE UNIQUE INDEX IF NOT EXISTS idx_rel_links_to + ON relationships (subject, object, user_id_key) WHERE predicate = 'links_to'; +CREATE UNIQUE INDEX IF NOT EXISTS idx_rel_subject_pred_user_single + ON relationships (subject, predicate, user_id_key) WHERE predicate <> 'links_to'; +DROP INDEX IF EXISTS idx_rel_subject_pred_user;`, }, } diff --git a/familiar-gateway/internal/db/migrate_test.go b/familiar-gateway/internal/db/migrate_test.go index 100ce90..078e3ee 100644 --- a/familiar-gateway/internal/db/migrate_test.go +++ b/familiar-gateway/internal/db/migrate_test.go @@ -5,6 +5,7 @@ import ( "database/sql" "errors" "os" + "strings" "testing" "time" @@ -247,6 +248,63 @@ func TestMigrateNilPool(t *testing.T) { } } +// The scrub migration's probe looked for working_context in every +// schema: another schema's user_profiles carrying it (public, on a +// database whose own schema is migrated separately) made the migration +// update and drop a column the migrated schema doesn't have, and +// Migrate aborted. +func TestMigrate_ScrubProbeStaysInItsSchema(t *testing.T) { + dsn := os.Getenv("FAMILIAR_TEST_DSN") + if dsn == "" { + t.Skip("skipping: FAMILIAR_TEST_DSN not set") + } + ctx := context.Background() + admin, err := Open(dsn) + if err != nil { + t.Fatalf("db.Open (admin): %v", err) + } + t.Cleanup(func() { _ = admin.Close() }) + const other, fresh = "migrate_probe_other", "migrate_probe_fresh" + for _, sch := range []string{other, fresh} { + if _, err := admin.ExecContext(ctx, "DROP SCHEMA IF EXISTS "+sch+" CASCADE; CREATE SCHEMA "+sch); err != nil { + t.Fatalf("schema %s: %v", sch, err) + } + sch := sch + t.Cleanup(func() { _, _ = admin.ExecContext(context.Background(), "DROP SCHEMA IF EXISTS "+sch+" CASCADE") }) + } + if _, err := admin.ExecContext(ctx, "CREATE TABLE "+other+".user_profiles (user_id TEXT PRIMARY KEY, working_context JSONB)"); err != nil { + t.Fatal(err) + } + pool, err := Open(testdsn.Scoped(t, dsn, fresh)) + if err != nil { + t.Fatalf("db.Open (scoped): %v", err) + } + t.Cleanup(func() { _ = pool.Close() }) + if err := Migrate(ctx, pool); err != nil { + t.Fatalf("Migrate with another schema's working_context present: %v", err) + } + var n int + if err := admin.QueryRowContext(ctx, `SELECT count(*) FROM information_schema.columns + WHERE table_schema = $1 AND table_name = 'user_profiles' AND column_name = 'working_context'`, other).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Error("the migration reached into another schema's table") + } +} + +// A malformed DSN's error quoted it whole, password included, and the +// error is logged. +func TestOpen_ErrorDoesNotQuoteThePassword(t *testing.T) { + _, err := Open("postgresql://familiar:s3cr%zzet@localhost:5432/familiar") + if err == nil { + t.Fatal("a malformed DSN opened") + } + if strings.Contains(err.Error(), "s3cr") { + t.Errorf("error leaks the password: %v", err) + } +} + // scratchPools returns a pool on dsn and one on a freshly migrated // scratch schema of the same database, which no other package's tests // touch. The schema is dropped when the test ends. diff --git a/familiar-gateway/internal/memory/pgvector_test.go b/familiar-gateway/internal/memory/pgvector_test.go index d43f112..49303f3 100644 --- a/familiar-gateway/internal/memory/pgvector_test.go +++ b/familiar-gateway/internal/memory/pgvector_test.go @@ -334,6 +334,13 @@ func TestSanitizeDSN(t *testing.T) { {"postgres://user:secret@host/db", "postgres://user:***@host/db"}, {"postgres://user@host/db", "postgres://user@host/db"}, {"host/db", "host/db"}, + // A password containing @, and one inside an error message. + {"postgresql://u:p@ss@host/db", "postgresql://u:***@host/db"}, + {`parse "postgresql://familiar:s3cr%zzet@localhost:5432/familiar": invalid URL escape "%zz"`, + `parse "postgresql://familiar:***@localhost:5432/familiar": invalid URL escape "%zz"`}, + // Key-value form, quoted with spaces. + {"host=h password='a b c' dbname=d", "host=h password=*** dbname=d"}, + {"host=h password=plain dbname=d", "host=h password=*** dbname=d"}, } for _, tc := range cases { if got := db.SanitizeDSN(tc.in); got != tc.want { diff --git a/familiar-gateway/internal/memory/relationships.go b/familiar-gateway/internal/memory/relationships.go index 3d493ae..f427b42 100644 --- a/familiar-gateway/internal/memory/relationships.go +++ b/familiar-gateway/internal/memory/relationships.go @@ -14,6 +14,7 @@ import ( "github.com/familiar/gateway/internal/db" "github.com/familiar/gateway/internal/safego" + "github.com/lib/pq" ) // Relationship is one (subject, predicate, object) triple extracted @@ -38,7 +39,9 @@ type Relationship struct { // lookups keyed by retrieved memory contents, and the write path // needs an upsert that collapses (subject, predicate) pairs to one // row so IP changes / version bumps replace the previous value -// instead of accumulating contradictory triples. +// instead of accumulating contradictory triples — except links_to, +// where a page has one edge per link (ReplacePageLinks keeps the set +// current). type RelationshipStore interface { UpsertRelationships(ctx context.Context, rels []Relationship) error RelatedForContents(ctx context.Context, contents []string, userID string, limit int) ([]Relationship, error) @@ -108,7 +111,7 @@ func (s *PgRelationshipStore) UpsertRelationships(ctx context.Context, rels []Re _, err := s.db.ExecContext(ctx, ` INSERT INTO relationships (subject, predicate, object, user_id, source_fact, confidence, scope_tag, created_at, updated_at) VALUES ($1, $2, $3, $4, $5::uuid, $6, $7, $8, $8) - ON CONFLICT (subject, predicate, user_id_key) DO UPDATE + ON CONFLICT `+conflictTarget(r.Predicate)+` DO UPDATE SET object = EXCLUDED.object, source_fact = EXCLUDED.source_fact, confidence = EXCLUDED.confidence, @@ -125,6 +128,51 @@ func (s *PgRelationshipStore) UpsertRelationships(ctx context.Context, rels []Re return nil } +// linksTo is the one multi-valued predicate: a page links to many +// pages, one edge each. +const linksTo = "links_to" + +// conflictTarget is the ON CONFLICT target matching the unique index +// that covers predicate (relationships_links_multi). +func conflictTarget(predicate string) string { + if strings.EqualFold(predicate, linksTo) { + return `(subject, object, user_id_key) WHERE predicate = 'links_to'` + } + return `(subject, predicate, user_id_key) WHERE predicate <> 'links_to'` +} + +// ReplacePageLinks makes subject's links_to edges in scopeTag exactly +// objects, owned by userID: edges to pages no longer linked are +// deleted (whoever saved them; a page's links are the page's, not a +// member's), and the rest upserted. A page saved with no links keeps +// none. +func (s *PgRelationshipStore) ReplacePageLinks(ctx context.Context, subject, userID, scopeTag string, objects []string) error { + subject = strings.ToLower(subject) + lower := make([]string, 0, len(objects)) + for _, o := range objects { + if o != "" { + lower = append(lower, strings.ToLower(o)) + } + } + var scopeArg any + if scopeTag != "" { + scopeArg = scopeTag + } + if _, err := s.db.ExecContext(ctx, ` + DELETE FROM relationships + WHERE subject = $1 AND predicate = 'links_to' + AND scope_tag IS NOT DISTINCT FROM $2 + AND NOT (object = ANY($3::text[]) AND user_id_key = $4)`, + subject, scopeArg, pq.Array(lower), userID); err != nil { + return fmt.Errorf("replace links of %s: %w", subject, err) + } + rels := make([]Relationship, 0, len(lower)) + for _, o := range lower { + rels = append(rels, Relationship{Subject: subject, Predicate: linksTo, Object: o, UserID: userID, ScopeTag: scopeTag, Confidence: 1.0}) + } + return s.UpsertRelationships(ctx, rels) +} + // InsertRelationshipsIfAbsent adds the triples that aren't stored yet // (by subject, predicate and owner, lowercased like // UpsertRelationships) and leaves existing ones alone, reporting how @@ -154,7 +202,7 @@ func (s *PgRelationshipStore) InsertRelationshipsIfAbsent(ctx context.Context, r res, err := s.db.ExecContext(ctx, ` INSERT INTO relationships (subject, predicate, object, user_id, source_fact, confidence, scope_tag, created_at, updated_at) VALUES ($1, $2, $3, $4, $5::uuid, $6, $7, NOW(), NOW()) - ON CONFLICT (subject, predicate, user_id_key) DO NOTHING`, + ON CONFLICT `+conflictTarget(r.Predicate)+` DO NOTHING`, strings.ToLower(r.Subject), strings.ToLower(r.Predicate), strings.ToLower(r.Object), userArg, sourceArg, conf, scopeArg) if err != nil { diff --git a/familiar-gateway/internal/memory/relationships_links_test.go b/familiar-gateway/internal/memory/relationships_links_test.go new file mode 100644 index 0000000..71f6993 --- /dev/null +++ b/familiar-gateway/internal/memory/relationships_links_test.go @@ -0,0 +1,128 @@ +package memory + +import ( + "context" + "sort" + "strings" + "testing" + + "github.com/familiar/gateway/internal/db" + "github.com/familiar/gateway/internal/testutil" +) + +// A page's links_to edges are one per link. The unique key was +// (subject, predicate, owner), so saving a page that links to a, b and +// c left only the edge to c. Other predicates keep one object per +// subject (a changed value replaces the old one). +func TestRelationships_PageLinksAreMultiValued(t *testing.T) { + pool := testutil.PgScopedPool(t, "memory_rel_test") + testutil.TruncateTables(t, pool, "relationships") + ctx := context.Background() + // An upgraded database still has the old key; the migration replaces + // it (the table is empty, so recreating it here can't fail). + if _, err := pool.ExecContext(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_rel_subject_pred_user + ON relationships (subject, predicate, user_id_key)`); err != nil { + t.Fatal(err) + } + if err := db.Migrate(ctx, pool); err != nil { + t.Fatalf("Migrate: %v", err) + } + s, err := NewPgRelationshipStore(pool) + if err != nil { + t.Fatalf("store: %v", err) + } + const page, scope = "page:eng/deploy", "book:b1" + links := func() string { + rows, err := pool.QueryContext(ctx, `SELECT object || '@' || user_id_key FROM relationships + WHERE subject = $1 AND predicate = 'links_to' ORDER BY 1`, page) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []string + for rows.Next() { + var o string + _ = rows.Scan(&o) + out = append(out, o) + } + sort.Strings(out) + return strings.Join(out, " ") + } + + if err := s.ReplacePageLinks(ctx, page, "alice", scope, []string{"page:eng/a", "page:eng/b", "page:eng/c"}); err != nil { + t.Fatal(err) + } + if got := links(); got != "page:eng/a@alice page:eng/b@alice page:eng/c@alice" { + t.Fatalf("after first save: %q, want all three links", got) + } + + // A link removed from the page loses its edge; another member's save + // replaces the set rather than adding a second one. + if err := s.ReplacePageLinks(ctx, page, "bob", scope, []string{"page:eng/a", "page:eng/d"}); err != nil { + t.Fatal(err) + } + if got := links(); got != "page:eng/a@bob page:eng/d@bob" { + t.Fatalf("after bob's save: %q, want a and d, bob's", got) + } + // Saving the same links again updates them in place. + if err := s.ReplacePageLinks(ctx, page, "bob", scope, []string{"page:eng/a", "Page:Eng/D"}); err != nil { + t.Fatalf("re-save: %v", err) + } + if got := links(); got != "page:eng/a@bob page:eng/d@bob" { + t.Fatalf("after re-save: %q", got) + } + + // No links left: no edges. + if err := s.ReplacePageLinks(ctx, page, "bob", scope, nil); err != nil { + t.Fatal(err) + } + if got := links(); got != "" { + t.Fatalf("after removing every link: %q", got) + } + + // A single-valued predicate still replaces its object. + for _, obj := range []string{"march", "april"} { + if err := s.UpsertRelationships(ctx, []Relationship{{Subject: "acme", Predicate: "deadline", Object: obj, UserID: "alice"}}); err != nil { + t.Fatal(err) + } + } + var n int + var obj string + if err := pool.QueryRowContext(ctx, `SELECT count(*), max(object) FROM relationships WHERE subject = 'acme' AND predicate = 'deadline'`).Scan(&n, &obj); err != nil { + t.Fatal(err) + } + if n != 1 || obj != "april" { + t.Errorf("deadline rows = %d (%s), want one, april", n, obj) + } + + // The insert-if-absent path (relationship backfill) follows the + // same keys. + added, err := s.InsertRelationshipsIfAbsent(ctx, []Relationship{ + {Subject: page, Predicate: "links_to", Object: "page:eng/x", UserID: "bob", ScopeTag: scope}, + {Subject: page, Predicate: "links_to", Object: "page:eng/y", UserID: "bob", ScopeTag: scope}, + {Subject: "acme", Predicate: "deadline", Object: "may", UserID: "alice"}, + }) + if err != nil { + t.Fatal(err) + } + if added != 2 { + t.Errorf("added = %d, want both links and not the existing deadline", added) + } + + // The old index is gone after migration; its replacements exist. + var idx []string + rows, err := pool.QueryContext(ctx, `SELECT indexname FROM pg_indexes WHERE tablename = 'relationships' AND schemaname = current_schema() ORDER BY 1`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + for rows.Next() { + var name string + _ = rows.Scan(&name) + idx = append(idx, name) + } + joined := strings.Join(idx, ",") + if strings.Contains(joined, "idx_rel_subject_pred_user,") || !strings.Contains(joined, "idx_rel_links_to") || !strings.Contains(joined, "idx_rel_subject_pred_user_single") { + t.Errorf("relationships indexes = %s", joined) + } +} diff --git a/familiar-gateway/internal/pipeline/pipeline.go b/familiar-gateway/internal/pipeline/pipeline.go index 687e019..3284b31 100644 --- a/familiar-gateway/internal/pipeline/pipeline.go +++ b/familiar-gateway/internal/pipeline/pipeline.go @@ -40,6 +40,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" "github.com/familiar/gateway/internal/classifier" @@ -185,6 +186,9 @@ func researchNoteFrom(toolName string, data json.RawMessage) (ResearchNoteRef, b // Pipeline wires together the engine, router, and session manager. type Pipeline struct { + // background counts post-turn goroutines still running (Drain). + background atomic.Int64 + engine engine.Service router *router.Router sessions *session.Manager diff --git a/familiar-gateway/internal/pipeline/summarize.go b/familiar-gateway/internal/pipeline/summarize.go index bb7ab45..eac59c7 100644 --- a/familiar-gateway/internal/pipeline/summarize.go +++ b/familiar-gateway/internal/pipeline/summarize.go @@ -88,7 +88,7 @@ func (p *Pipeline) maybeSummarize(sess *session.Session, overrides *ShardOverrid if !sess.TryBeginSummarize() { return // another goroutine is already summarizing this session } - go p.runSummarize(sess, overrides) + p.goBackground(func() { p.runSummarize(sess, overrides) }) } // runSummarize does the async summarization + fact extraction for a session. @@ -264,7 +264,33 @@ func (p *Pipeline) kickoffPostTurnExtract(sess *session.Session, userMsg, respon if strings.TrimSpace(userMsg) == "" && strings.TrimSpace(responseText) == "" { return } - go p.runPostTurnExtract(sess, userMsg, responseText, prior, retrievedRels, overrides) + p.goBackground(func() { p.runPostTurnExtract(sess, userMsg, responseText, prior, retrievedRels, overrides) }) +} + +// goBackground runs post-turn work (summaries, fact extraction) on its +// own goroutine, counted so a shutdown can wait for it (Drain). +func (p *Pipeline) goBackground(fn func()) { + p.background.Add(1) + go func() { + defer p.background.Add(-1) + fn() + }() +} + +// Drain waits up to timeout for post-turn work still running and +// reports whether it all finished. Shutdown calls it before closing the +// database: that work runs detached from any request, and the process +// used to exit under it, so a turn completed just before a restart had +// its facts never extracted. +func (p *Pipeline) Drain(timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for p.background.Load() > 0 { + if time.Now().After(deadline) { + return false + } + time.Sleep(25 * time.Millisecond) + } + return true } // runPostTurnExtract executes the per-turn memory write pipeline: diff --git a/familiar-gateway/internal/pipeline/turncontext_test.go b/familiar-gateway/internal/pipeline/turncontext_test.go index 23a44ed..7e01f13 100644 --- a/familiar-gateway/internal/pipeline/turncontext_test.go +++ b/familiar-gateway/internal/pipeline/turncontext_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/http/httptest" "sync" + "sync/atomic" "testing" "time" @@ -315,3 +316,22 @@ func TestTurnContext_BoundToCallerEndsWithCaller(t *testing.T) { t.Error("CallerBound doesn't reflect the mark") } } + +// Drain waits for post-turn work and gives up at its timeout. +func TestPipeline_DrainWaitsForBackgroundWork(t *testing.T) { + p := &Pipeline{} + var done atomic.Bool + p.goBackground(func() { time.Sleep(100 * time.Millisecond); done.Store(true) }) + if !p.Drain(2*time.Second) || !done.Load() { + t.Error("Drain returned before the post-turn work finished") + } + block := make(chan struct{}) + p.goBackground(func() { <-block }) + if p.Drain(50 * time.Millisecond) { + t.Error("Drain reported done with work still running") + } + close(block) + if !p.Drain(2 * time.Second) { + t.Error("Drain didn't see the work finish") + } +} diff --git a/familiar-gateway/internal/router/registry_test.go b/familiar-gateway/internal/router/registry_test.go index 5c06a56..a77d507 100644 --- a/familiar-gateway/internal/router/registry_test.go +++ b/familiar-gateway/internal/router/registry_test.go @@ -198,3 +198,22 @@ func TestGetEmbeddingsProvider(t *testing.T) { t.Errorf("embeddings model should build as a Provider for health checks: %v", err) } } + +// config.Validate rejects providers and formatters outside +// config.KnownProviders / KnownFormatters; those sets must be exactly +// what buildProvider and pickFormatter build. +func TestBuildProvider_MatchesConfigKnownProviders(t *testing.T) { + for p := range config.KnownProviders { + if _, err := buildProvider(config.ModelConfig{ID: "m", Provider: p, Endpoint: "http://m"}, ""); err != nil { + t.Errorf("known provider %q doesn't build: %v", p, err) + } + } + for f := range config.KnownFormatters { + if _, err := pickFormatter(f); err != nil { + t.Errorf("known formatter %q doesn't build: %v", f, err) + } + } + if _, err := buildProvider(config.ModelConfig{ID: "m", Provider: "anthropic", Endpoint: "e"}, ""); err == nil { + t.Error("an unknown provider built") + } +} diff --git a/familiar-gateway/internal/wikiknowledge/pipeline.go b/familiar-gateway/internal/wikiknowledge/pipeline.go index 1d0e0a7..0f0d1d6 100644 --- a/familiar-gateway/internal/wikiknowledge/pipeline.go +++ b/familiar-gateway/internal/wikiknowledge/pipeline.go @@ -79,6 +79,7 @@ type MemoryStore interface { // depend on for triple persistence. type RelationshipStore interface { UpsertRelationships(ctx context.Context, rels []memory.Relationship) error + ReplacePageLinks(ctx context.Context, subject, userID, scopeTag string, objects []string) error } // EmbedFunc computes an embedding for a fact's content. Pipeline @@ -524,16 +525,18 @@ func (p *Pipeline) upsertExtractedRelationships(ctx context.Context, evt SaveEve } } -// upsertLinkTriples emits one "links_to" triple per resolved -// outbound link. Subjects + objects use the canonical +// upsertLinkTriples sets the page's "links_to" triples to its resolved +// outbound links, one per link. Subjects + objects use the canonical // "page:{book_slug}/{page_slug}" form so they sit alongside -// content-extracted entities in the same relationship graph. +// content-extracted entities in the same relationship graph. The set +// is replaced, not added to: a link removed from the page loses its +// edge, and a page saved with no links has none (it used to keep +// whatever it had). func (p *Pipeline) upsertLinkTriples(ctx context.Context, evt SaveEvent, scopeTag string) { - if p.deps.RelStore == nil || len(evt.Links) == 0 { + if p.deps.RelStore == nil { return } - out := make([]memory.Relationship, 0, len(evt.Links)) - subject := pageEntityKey(evt.BookSlug, evt.PageSlug) + var objects []string for _, l := range evt.Links { if l.TargetPageID == nil { continue // broken link @@ -542,20 +545,11 @@ func (p *Pipeline) upsertLinkTriples(ctx context.Context, evt SaveEvent, scopeTa if targetBook == "" { targetBook = evt.BookSlug } - out = append(out, memory.Relationship{ - Subject: subject, - Predicate: "links_to", - Object: pageEntityKey(targetBook, l.TargetPageSlug), - UserID: evt.UserID, - ScopeTag: scopeTag, - Confidence: 1.0, - }) + objects = append(objects, pageEntityKey(targetBook, l.TargetPageSlug)) } - if len(out) == 0 { - return - } - if err := p.deps.RelStore.UpsertRelationships(ctx, out); err != nil { - log.Printf("[wikiknowledge] upsert link triples failed for %s/%s: %v", + subject := pageEntityKey(evt.BookSlug, evt.PageSlug) + if err := p.deps.RelStore.ReplacePageLinks(ctx, subject, evt.UserID, scopeTag, objects); err != nil { + log.Printf("[wikiknowledge] replace link triples failed for %s/%s: %v", evt.BookSlug, evt.PageSlug, err) } } diff --git a/familiar-gateway/internal/wikiknowledge/pipeline_test.go b/familiar-gateway/internal/wikiknowledge/pipeline_test.go index 2ae4a78..fe19c4a 100644 --- a/familiar-gateway/internal/wikiknowledge/pipeline_test.go +++ b/familiar-gateway/internal/wikiknowledge/pipeline_test.go @@ -140,6 +140,28 @@ func (f *fakeMem) DeleteMemoriesBySource(_ context.Context, st, sr, sg string) ( type fakeRel struct { upserts [][]memory.Relationship + // replaces records ReplacePageLinks: the page's whole link set. + replaces []linkReplace +} + +type linkReplace struct { + subject, userID, scopeTag string + objects []string +} + +// ReplacePageLinks records the call and, for the tests that look at +// links_to rows, the triples it stands for. +func (f *fakeRel) ReplacePageLinks(_ context.Context, subject, userID, scopeTag string, objects []string) error { + f.replaces = append(f.replaces, linkReplace{subject, userID, scopeTag, append([]string(nil), objects...)}) + if len(objects) == 0 { + return nil + } + batch := make([]memory.Relationship, 0, len(objects)) + for _, o := range objects { + batch = append(batch, memory.Relationship{Subject: subject, Predicate: "links_to", Object: o, UserID: userID, ScopeTag: scopeTag, Confidence: 1.0}) + } + f.upserts = append(f.upserts, batch) + return nil } func (f *fakeRel) UpsertRelationships(_ context.Context, rels []memory.Relationship) error { @@ -413,6 +435,24 @@ func TestOnPageSaved_BrokenLinkSkipped(t *testing.T) { } } } + // The page's link set is still replaced — with nothing — so edges + // from links it used to have go (it kept them before). + if len(rel.replaces) != 1 || len(rel.replaces[0].objects) != 0 { + t.Errorf("replaces = %+v, want one call with no links", rel.replaces) + } +} + +// A page saved with no links at all clears its link set: the edges from +// links it used to have don't outlive them. +func TestOnPageSaved_NoLinksClearsTheSet(t *testing.T) { + rel := &fakeRel{} + p := newPipelineWith(nil, nil, nil, rel) + evt := sampleEvent() + evt.Links = nil + p.OnPageSaved(context.Background(), evt) + if len(rel.replaces) != 1 || len(rel.replaces[0].objects) != 0 || rel.replaces[0].subject != "page:engineering/deploy-process" { + t.Errorf("replaces = %+v, want the page's set replaced with nothing", rel.replaces) + } } func TestOnPageSaved_CrossBookLinkUsesTargetBook(t *testing.T) { diff --git a/init-db.sql b/init-db.sql index 4ac0cc3..f93d097 100644 --- a/init-db.sql +++ b/init-db.sql @@ -1,79 +1,10 @@ --- Familiar — initial database schema +-- Familiar — database bootstrap for the Docker image. -- Runs automatically on first `docker compose up`. +-- +-- The gateway creates and migrates its whole schema itself at startup +-- (familiar-gateway/internal/db/migrate.go). This file only enables +-- pgvector, which needs a superuser the gateway's role may not be on a +-- native cluster. It used to create a legacy schema (facts, +-- conversation_turns, entity_edges) that nothing reads. CREATE EXTENSION IF NOT EXISTS vector; -CREATE EXTENSION IF NOT EXISTS "uuid-ossp"; - --- Semantic facts / memories -CREATE TABLE IF NOT EXISTS facts ( - id UUID PRIMARY KEY DEFAULT uuid_generate_v4(), - content TEXT NOT NULL, - embedding vector(768), -- matches nomic-embed-text default dim - source_type TEXT NOT NULL DEFAULT 'conversation', - source_ref TEXT, - source_desc TEXT, - confidence REAL NOT NULL DEFAULT 0.5, - conf_basis TEXT, - scope TEXT NOT NULL DEFAULT 'user', - tags TEXT[] DEFAULT '{}', - domains TEXT[] DEFAULT '{}', - allowed_channels TEXT[] DEFAULT '{}', - allowed_agents TEXT[] DEFAULT '{}', - agent_private BOOLEAN NOT NULL DEFAULT FALSE, - internal_only BOOLEAN NOT NULL DEFAULT FALSE, - supersedes UUID REFERENCES facts(id), - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), - last_accessed TIMESTAMPTZ NOT NULL DEFAULT now(), - last_verified TIMESTAMPTZ, - access_count BIGINT NOT NULL DEFAULT 0 -); - --- HNSW index for fast ANN queries on embeddings -CREATE INDEX IF NOT EXISTS facts_embedding_idx - ON facts USING hnsw (embedding vector_cosine_ops); - -CREATE INDEX IF NOT EXISTS facts_scope_idx ON facts (scope); -CREATE INDEX IF NOT EXISTS facts_created_idx ON facts (created_at); - --- Conversation history -CREATE TABLE IF NOT EXISTS conversation_turns ( - id BIGSERIAL PRIMARY KEY, - session_id TEXT NOT NULL, - role TEXT NOT NULL, - content TEXT NOT NULL, - created_at TIMESTAMPTZ NOT NULL DEFAULT now() -); - -CREATE INDEX IF NOT EXISTS conv_session_idx ON conversation_turns (session_id, created_at); - --- Entity-relationship edges (persistent knowledge graph tier) -CREATE TABLE IF NOT EXISTS entity_edges ( - id BIGSERIAL PRIMARY KEY, - from_entity TEXT NOT NULL, - to_entity TEXT NOT NULL, - rel_type TEXT NOT NULL, - weight REAL NOT NULL DEFAULT 1.0, - metadata JSONB DEFAULT '{}', - created_at TIMESTAMPTZ NOT NULL DEFAULT now() -); - -CREATE INDEX IF NOT EXISTS edges_from_idx ON entity_edges (from_entity); -CREATE INDEX IF NOT EXISTS edges_to_idx ON entity_edges (to_entity); -CREATE INDEX IF NOT EXISTS edges_rel_idx ON entity_edges (rel_type); - --- Layer 1 working context: small per-user profile always injected into the --- system prompt by the ctxbuild assembler. Phase 2 seeds this manually; a --- later phase lets the LLM edit it via a core_memory_update tool. -CREATE TABLE IF NOT EXISTS user_profiles ( - user_id TEXT PRIMARY KEY, - working_context JSONB NOT NULL DEFAULT '{}'::jsonb, - updated_at TIMESTAMPTZ NOT NULL DEFAULT now() -); - --- Example seed (commented — edit and run manually): --- INSERT INTO user_profiles (user_id, working_context) VALUES ( --- 'U0123EXAMPLE', --- '{"name":"Ada Lovelace","location":"Boring, OR","role":"Software Engineer", --- "preferences":{"communication_style":"direct, technical, no fluff","timezone":"America/Los_Angeles"}, --- "active_goals":[],"recent_context":""}'::jsonb --- ) ON CONFLICT (user_id) DO UPDATE SET working_context = EXCLUDED.working_context, updated_at = now(); diff --git a/run.sh b/run.sh index 11222d9..0c3714a 100755 --- a/run.sh +++ b/run.sh @@ -18,6 +18,7 @@ ROOT_DIR="$(cd "$(dirname "$0")" && pwd)" GATEWAY_DIR="$ROOT_DIR/familiar-gateway" FAMILIAR_HOME="${FAMILIAR_HOME:-$HOME/.familiar}" +export FAMILIAR_HOME # the gateway reads its config and default paths from it BUILD=true RUN_SIDECAR=true RUN_DB=true @@ -80,15 +81,17 @@ if $RUN_DB; then docker compose -f "$ROOT_DIR/docker-compose.yml" up -d echo " Waiting for Postgres to be ready..." TRIES=0 + PG_READY=true until docker compose -f "$ROOT_DIR/docker-compose.yml" exec -T postgres pg_isready -U familiar >/dev/null 2>&1; do sleep 1 TRIES=$((TRIES + 1)) if [ $TRIES -ge 30 ]; then echo " WARNING: Postgres not ready after 30s — continuing without DB." + PG_READY=false break fi done - echo " Postgres ready." + $PG_READY && echo " Postgres ready." else echo " WARNING: No PostgreSQL available. Memory store will not persist." fi @@ -175,6 +178,20 @@ echo "==> Starting gateway..." GATEWAY_BIN="$GATEWAY_DIR/familiar-gateway" [ -x "$GATEWAY_BIN" ] || GATEWAY_BIN="$GATEWAY_DIR/gateway" +# Without --http or --slack the gateway runs the interactive CLI, which +# reads this terminal: run it in the foreground. Backgrounded, its stdin +# was /dev/null, it read EOF and exited at once, and the exit trap tore +# down everything just started. +case " $GATEWAY_FLAGS " in + *" --http "*|*" --slack "*) ;; + *) + echo "==> Familiar CLI (Ctrl-D to quit)" + # shellcheck disable=SC2086 + "$GATEWAY_BIN" $GATEWAY_FLAGS + exit $? + ;; +esac + # shellcheck disable=SC2086 "$GATEWAY_BIN" $GATEWAY_FLAGS & GATEWAY_PID=$! diff --git a/setup.sh b/setup.sh index d46f624..0250421 100755 --- a/setup.sh +++ b/setup.sh @@ -144,7 +144,7 @@ if ! $SKIP_DB; then fi psql -d familiar -f "$ROOT_DIR/init-db.sql" 2>/dev/null || true fi - ok "Database schema applied (facts, conversation_turns, entity_edges)" + ok "pgvector enabled (the gateway creates its schema on first start)" fi fi @@ -162,9 +162,6 @@ if [ ! -f "$GATEWAY_CONFIG" ]; then echo "" echo "==> Generating default config at $GATEWAY_CONFIG ..." - # Detect sidecar socket path to use. - SIDECAR_SOCK="$SIDECAR_SOCK_DIR/sidecar.sock" - # Docker Postgres needs the generated password; a native cluster # authenticates the local user and needs none. LOCAL_DSN="postgresql://familiar@localhost:5432/familiar" @@ -172,9 +169,16 @@ if [ ! -f "$GATEWAY_CONFIG" ]; then PGPW=$(sed -n 's/^POSTGRES_PASSWORD=//p' "$ROOT_DIR/.env" | tail -1) LOCAL_DSN="postgresql://familiar:${PGPW}@localhost:5432/familiar" fi - # gateway.toml carries credentials (the DSN, tokens): owner-only. - (umask 077; : > "$GATEWAY_CONFIG") - cat > "$GATEWAY_CONFIG" < "$GATEWAY_CONFIG" <