feat(ingest): shared connector builder + opt-in in-process sync scheduler (#428)

.env.example

@@ -28,6 +28,12 @@ EMBEDDING_MODEL=nomic-embed-text
 # Auto-import: re-scan ./data/docs on this interval (empty = off). Great for laptops.
 INGEST_INTERVAL=60s
 
+# Background connector sync (optional, #428): re-ingest the configured REMOTE
+# connectors (notion, slack, github, …) on this interval — the in-process form of
+# `kb sync`. Empty (default) = off, so the minimal single-user stack is unchanged;
+# markdown stays governed by INGEST_INTERVAL above. Honors INGEST_PRUNE_DELETED.
+# SYNC_INTERVAL=15m
+
 # Propagate source-side deletions on auto-import (optional, #247). OFF by default:
 # the retention default (#107) keeps content even after you delete the file. When
 # on, a doc removed from ./data/docs is removed from memory on the next sweep (only

SYSTEM.md

@@ -383,6 +383,17 @@ as the adoption signal.
 
 Newest first.
 
+- **2026-07-20** — **Shared connector builder + opt-in in-process sync scheduler (#428, ingestion P2).**
+  The ~150-line connector-construction switch was duplicated in the CLI (`buildConnector`) and the MCP server
+  (`importFunc`); it now lives once in **`internal/connectors.Build(ctx, kb, cfg, source, path, ocr)`** —
+  the CLI and MCP adapters are thin wrappers over it, so a new/changed connector is edited in one place. OAuth
+  token resolution (#246) lives inside the builder; the OCR fallback is passed in by the caller so the
+  shell-out stays in the cmd layer (the markdown connector is the only consumer). On top of it, `cmd/http`
+  gains an **opt-in background sync** (`autoSync`, mirrors `autoImport`): with `SYNC_INTERVAL` set (e.g.
+  `15m`), the HTTP server re-ingests every configured **remote** connector on a timer — the in-process form
+  of `kb sync` (#415) — incrementally, honoring `INGEST_PRUNE_DELETED`. markdown is skipped (already swept by
+  `autoImport`). **Default off** (`SYNC_INTERVAL` empty → scheduler never starts), so the minimal single-user
+  stack is unchanged. Push adapters (fsnotify for markdown, connector webhooks) remain a follow-up.
 - **2026-07-20** — **Wire async batch extraction into the ingest loop (#430, completes #420, ingestion P2).**
   With `EXTRACTION_BATCH` + the Claude extractor (`c.batchExtractor` set), ingest's extractor step now
   submits the document's hot chunks as one Message Batch (`SubmitExtractionBatch`, custom_id = content hash,

cmd/cli/commands.go

@@ -17,19 +17,10 @@ import (
 
 	"github.com/programmism/brainiac/internal/chunk"
 	"github.com/programmism/brainiac/internal/config"
+	"github.com/programmism/brainiac/internal/connectors"
 	"github.com/programmism/brainiac/internal/core"
 	"github.com/programmism/brainiac/internal/model"
 	"github.com/programmism/brainiac/internal/plugins"
-	"github.com/programmism/brainiac/internal/plugins/confluence"
-	"github.com/programmism/brainiac/internal/plugins/gdrive"
-	"github.com/programmism/brainiac/internal/plugins/github"
-	"github.com/programmism/brainiac/internal/plugins/gitlab"
-	"github.com/programmism/brainiac/internal/plugins/gmail"
-	"github.com/programmism/brainiac/internal/plugins/jira"
-	"github.com/programmism/brainiac/internal/plugins/linear"
-	"github.com/programmism/brainiac/internal/plugins/markdown"
-	"github.com/programmism/brainiac/internal/plugins/notion"
-	"github.com/programmism/brainiac/internal/plugins/slack"
 	"github.com/programmism/brainiac/internal/store"
 )
 
@@ -1133,120 +1124,11 @@ func importCmd() *cobra.Command {
 	return cmd
 }
 
-// oauthToken resolves a connector's access token (#246): a stored, auto-refreshed
-// OAuth credential if present, else the source's <TYPE>_TOKEN env value.
-func oauthToken(ctx context.Context, kb *core.Core, cfg *config.Config, source string) (string, error) {
-	env := ""
-	if sc := cfg.Source(source); sc != nil {
-		env = sc.Token
-	}
-	return kb.ResolveSourceToken(ctx, source, env)
-}
-
+// buildConnector constructs the connector for a source type via the shared
+// internal/connectors builder (#428), passing the CLI's OCR fallback. path is an
+// optional targeted import (--path).
 func buildConnector(ctx context.Context, kb *core.Core, cfg *config.Config, source, path string) (plugins.SourceConnector, error) {
-	switch source {
-	case "notion":
-		sc := cfg.Source("notion")
-		if sc == nil || sc.Token == "" {
-			return nil, fmt.Errorf("notion source not configured (set a token via NOTION_TOKEN or config.yaml)")
-		}
-		if path != "" { // --path holds a page URL/id for a targeted import
-			return notion.NewForPages(sc.Token, []string{path}), nil
-		}
-		return notion.New(sc.Token), nil
-	case "slack":
-		sc := cfg.Source("slack")
-		if sc == nil || sc.Token == "" {
-			return nil, fmt.Errorf("slack source not configured (set a token via SLACK_TOKEN or config.yaml)")
-		}
-		if path != "" { // --path holds a channel id for a targeted import
-			return slack.NewForChannels(sc.Token, []string{path}), nil
-		}
-		return slack.New(sc.Token), nil
-	case "github":
-		sc := cfg.Source("github")
-		if sc == nil || sc.Token == "" {
-			return nil, fmt.Errorf("github source not configured (set a token via GITHUB_TOKEN or config.yaml)")
-		}
-		repos := sc.Repos
-		if path != "" { // --path holds an owner/repo for a targeted import
-			repos = []string{path}
-		}
-		if len(repos) == 0 {
-			return nil, fmt.Errorf("github needs a repo: --path owner/repo, or sources[].repos / GITHUB_REPOS")
-		}
-		ghOpts := []github.Option{github.WithFiles(sc.Files)}
-		if sc.Discussions {
-			ghOpts = append(ghOpts, github.WithDiscussions())
-		}
-		return github.New(sc.Token, repos, ghOpts...), nil
-	case "gdrive":
-		tok, err := oauthToken(ctx, kb, cfg, "gdrive")
-		if err != nil {
-			return nil, err
-		}
-		if tok == "" {
-			return nil, fmt.Errorf("gdrive not configured (set GDRIVE_TOKEN or `kb oauth set --source gdrive`)")
-		}
-		return gdrive.New(tok), nil
-	case "gmail":
-		tok, err := oauthToken(ctx, kb, cfg, "gmail")
-		if err != nil {
-			return nil, err
-		}
-		if tok == "" {
-			return nil, fmt.Errorf("gmail not configured (set GMAIL_TOKEN or `kb oauth set --source gmail`)")
-		}
-		q := ""
-		if sc := cfg.Source("gmail"); sc != nil {
-			q = sc.Query
-		}
-		return gmail.New(tok, gmail.WithQuery(q)), nil
-	case "linear":
-		sc := cfg.Source("linear")
-		if sc == nil || sc.Token == "" {
-			return nil, fmt.Errorf("linear source not configured (set an API key via LINEAR_TOKEN)")
-		}
-		return linear.New(sc.Token), nil
-	case "gitlab":
-		sc := cfg.Source("gitlab")
-		if sc == nil || sc.Token == "" {
-			return nil, fmt.Errorf("gitlab source not configured (set an access token via GITLAB_TOKEN)")
-		}
-		projects := sc.Repos
-		if path != "" { // --path holds a group/project for a targeted import
-			projects = []string{path}
-		}
-		if len(projects) == 0 {
-			return nil, fmt.Errorf("gitlab needs a project: --path group/project, or sources[].repos / GITLAB_PROJECTS")
-		}
-		return gitlab.New(sc.Token, sc.BaseURL, projects), nil
-	case "jira":
-		sc := cfg.Source("jira")
-		if sc == nil || sc.BaseURL == "" || sc.Email == "" || sc.Token == "" {
-			return nil, fmt.Errorf("jira source not configured (set JIRA_BASE_URL, JIRA_EMAIL, JIRA_TOKEN)")
-		}
-		return jira.New(sc.BaseURL, sc.Email, sc.Token), nil
-	case "confluence":
-		sc := cfg.Source("confluence")
-		if sc == nil || sc.BaseURL == "" || sc.Email == "" || sc.Token == "" {
-			return nil, fmt.Errorf("confluence source not configured (set CONFLUENCE_BASE_URL, CONFLUENCE_EMAIL, CONFLUENCE_TOKEN)")
-		}
-		return confluence.New(sc.BaseURL, sc.Email, sc.Token), nil
-	case "markdown":
-		dir := path
-		if dir == "" {
-			if sc := cfg.Source("markdown"); sc != nil {
-				dir = sc.Path
-			}
-		}
-		if dir == "" {
-			return nil, fmt.Errorf("markdown source needs a directory (--path or sources[].path)")
-		}
-		return markdown.New(dir, markdown.WithOCR(ocrFunc(cfg))), nil
-	default:
-		return nil, fmt.Errorf("unknown source %q", source)
-	}
+	return connectors.Build(ctx, kb, cfg, source, path, ocrFunc(cfg))
 }
 
 func consolidateCmd() *cobra.Command {

cmd/http/main.go

@@ -23,7 +23,9 @@ import (
 	"time"
 
 	"github.com/programmism/brainiac/internal/applog"
+	"github.com/programmism/brainiac/internal/chunk"
 	"github.com/programmism/brainiac/internal/config"
+	"github.com/programmism/brainiac/internal/connectors"
 	"github.com/programmism/brainiac/internal/core"
 	"github.com/programmism/brainiac/internal/logbuf"
 	"github.com/programmism/brainiac/internal/plugins/anthropic"
@@ -157,6 +159,13 @@ func run() error {
 		go autoImport(shutdownCtx, c, cfg, d)
 	}
 
+	// Optional background connector sync (#428): opt-in via SYNC_INTERVAL, re-ingests
+	// the configured remote connectors on a timer (the in-process form of `kb sync`).
+	// Off by default, so the minimal single-user stack is unchanged.
+	if d := cfg.SyncInterval(); d > 0 {
+		go autoSync(shutdownCtx, c, cfg, d)
+	}
+
 	logStartupBanner(cfg, writable, embedderCheck)
 	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
 		return err
@@ -313,6 +322,63 @@ func autoImport(ctx context.Context, c *core.Core, cfg *config.Config, every tim
 	}
 }
 
+// autoSync re-ingests the configured remote connectors on a timer (#428) — the
+// in-process equivalent of running `kb sync` periodically. markdown is skipped
+// (autoImport already sweeps ./data/docs and markdown sources); each connector is
+// built via the shared builder and ingested incrementally so repeated runs are
+// cheap. A per-source failure is logged and the sweep continues to the next.
+func autoSync(ctx context.Context, c *core.Core, cfg *config.Config, every time.Duration) {
+	sources := make([]string, 0, len(cfg.ConnectorSources()))
+	for _, s := range cfg.ConnectorSources() {
+		if s == "markdown" {
+			continue // covered by autoImport
+		}
+		sources = append(sources, s)
+	}
+	if len(sources) == 0 {
+		log.Printf("connector-sync enabled every %s, but no remote connectors are configured — nothing to sync", every)
+		return
+	}
+	run := func() {
+		for _, src := range sources {
+			if ctx.Err() != nil {
+				return
+			}
+			conn, err := connectors.Build(ctx, c, cfg, src, "", nil)
+			if err != nil {
+				log.Printf("connector-sync: %s: %v", src, err)
+				continue
+			}
+			opts := core.IngestOptions{
+				Incremental:  true,
+				PruneMissing: cfg.Ingest.PruneDeleted,
+				Trust:        cfg.SourceTrust(src),
+				ChunkParams:  chunk.Preset(cfg.SourceChunkPreset(src)),
+			}
+			stats, err := c.Ingest(ctx, conn, opts)
+			if err != nil {
+				log.Printf("connector-sync: %s: %v", src, err)
+				continue
+			}
+			if stats.Kept+stats.Queued+stats.Deleted > 0 {
+				log.Printf("connector-sync: %s: +%d kept, %d deleted", src, stats.Kept+stats.Queued, stats.Deleted)
+			}
+		}
+	}
+	log.Printf("connector-sync enabled every %s (%s)", every, strings.Join(sources, ", "))
+	run()
+	t := time.NewTicker(every)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			run()
+		}
+	}
+}
+
 // ollamaChecker returns a readiness probe for the embedder backend. It hits the
 // Ollama tags endpoint; a failure is reported but never fatal (§11).
 // extractorOptions builds the optional local-LLM extractor wiring from config.