feat(cli): add local tenant store subscribe

Author: altaywtf

README.md

@@ -79,6 +79,15 @@ FINCRAWL_AGE_IDENTITY=<age-identity> \
   fincrawl import --in snapshots/local.jsonl.zst.age --dry-run
 ```
 
+For a local tenant-controlled store with a `manifest.json`, verify and import
+the listed encrypted JSONL snapshots in one step:
+
+```bash
+fincrawl store verify <tenant-store-root>
+FINCRAWL_AGE_IDENTITY=<age-identity> \
+  fincrawl subscribe <tenant-store-root> --dry-run
+```
+
 Plaintext archive output is local scratch data only. Tenant encrypted snapshots
 still belong in tenant-controlled private storage, not in this repository.
 
@@ -115,6 +124,7 @@ Common flows:
 | Export encrypted snapshot | `fincrawl publish --recipient <recipient> --out snapshots/local.jsonl.zst.age` |
 | Import encrypted snapshot | `fincrawl import --identity <identity> --in snapshots/local.jsonl.zst.age` |
 | Verify an encrypted tenant store | `fincrawl store verify <path>` |
+| Import a local tenant store | `fincrawl subscribe <path> --identity <identity>` |
 | Check repo guardrails | `fincrawl guard --json` |
 
 ## Docs

docs/architecture.md

@@ -31,6 +31,7 @@ fincrawl archive --fixture testdata/synthetic --recipient <age-recipient> --out
 fincrawl publish --recipient <age-recipient> --out snapshots/local.jsonl.zst.age
 fincrawl import --identity <age-identity> --in snapshots/local.jsonl.zst.age
 fincrawl store verify <tenant-store-root>
+fincrawl subscribe <tenant-store-root> --dry-run
 fincrawl guard --json
 ```
 
@@ -116,6 +117,11 @@ verifier reads `manifest.json`, requires manifest snapshots to point at existing
 compressed age-encrypted artifacts, and rejects plaintext archives, SQLite
 stores, runtime state, logs, reports, screenshots, and transcripts.
 
+`fincrawl subscribe <path>` is a local one-shot subscriber flow. It verifies a
+tenant-controlled store, reads the manifest, and imports listed
+`.jsonl.zst.age` snapshots into local SQLite. It does not clone, pull, push,
+schedule jobs, or persist tenant-specific subscription config.
+
 ## Agent Use
 
 Agents should load [the repo skill](../skills/fincrawl/SKILL.md) before using a

docs/roadmap.md

@@ -25,19 +25,21 @@ this repo.
 - zstd + age encrypted `archive`, `publish`, and `import` flows.
 - Generic `store verify` checks for tenant-controlled encrypted snapshot
   manifests.
+- Local one-shot `subscribe` imports `.jsonl.zst.age` snapshots from a verified
+  tenant store path.
 - Repository guardrails for plaintext archives, generated artifacts, secret
   patterns, provider URLs, and transcript-like files.
 - Agent-facing skill guidance under `skills/fincrawl/`.
 - CI/release automation for `0.0.x` bootstrap releases with protected `main`.
 
 ## Next
 
-- Add a read-only subscriber flow that can pull/import a local tenant store
-  path without adding remote push or schedule mechanics.
 - Broaden exact hydration/search ergonomics around known provider URLs while
   keeping provider IDs path-safe.
 - Add more sync torture coverage around interrupted multi-page windows and
   repeated transient failures.
+- Add persisted local subscription metadata only if repeated local store imports
+  need stale checks or operator reminders.
 
 ## Later
 

docs/tenant-data-boundary.md

@@ -87,10 +87,17 @@ existing compressed age-encrypted snapshots with relative paths and rejects
 plaintext archives, local databases, runtime state, logs, reports, screenshots,
 and transcripts.
 
+Use `fincrawl subscribe <tenant-store-root>` only for local one-shot imports from
+a tenant-controlled store. The command verifies the store before import and
+hydrates local SQLite from encrypted JSONL snapshots; it does not make the
+generic repo a home for tenant store state, schedules, or remote credentials.
+
 Encryption recipients may be native `age1...` recipients or SSH public keys
 accepted by age. Private age identities and private SSH keys must stay outside
-the repo. Import dry-runs still need a private decrypt identity because they
-validate encrypted record contents and counts.
+the repo. `import --dry-run` still needs a private decrypt identity because it
+validates encrypted record contents and counts. `subscribe --dry-run` verifies
+the store manifest and planned snapshots without decrypting or mutating local
+SQLite.
 
 ## Fixture Rule
 

internal/cli/agent_dx.go

@@ -238,6 +238,25 @@ func describeCommands(command string) (cliSchema, error) {
 					"Input paths must be relative, stay under the current working directory, and end in .jsonl.zst.age.",
 				},
 			},
+			"subscribe": {
+				Name:    "subscribe",
+				Summary: "Import snapshots from a local encrypted tenant store.",
+				Mutates: true,
+				JSON:    true,
+				Args: []paramSchema{
+					{Name: "store", Type: "path", Required: true, Help: "Local tenant store root containing manifest.json."},
+				},
+				Flags: []paramSchema{
+					{Name: "identity", Type: "age-identity", Help: "Age identity. Defaults to FINCRAWL_AGE_IDENTITY."},
+					{Name: "dry-run", Type: "bool", Help: "Verify and list planned imports without writing local state."},
+					{Name: "json", Type: "bool", Default: "true", Help: "Print JSON output."},
+				},
+				Examples: []string{"fincrawl subscribe ../tenant-store --dry-run", "fincrawl subscribe ../tenant-store --identity AGE-SECRET-KEY-..."},
+				Notes: []string{
+					"Local filesystem stores only; remote clone, pull, and scheduling are out of scope.",
+					"The tenant store is verified before import, and this command currently imports .jsonl.zst.age snapshots.",
+				},
+			},
 			"store verify": {
 				Name:    "store verify",
 				Summary: "Verify a generic encrypted tenant-store manifest and artifact boundary.",

internal/cli/cli.go

@@ -26,19 +26,20 @@ import (
 )
 
 type app struct {
-	Doctor   doctorCmd   `cmd:"" help:"Check local configuration."`
-	Metadata metadataCmd `cmd:"" help:"Print machine-readable metadata."`
-	Describe describeCmd `cmd:"" help:"Print machine-readable command schemas."`
-	Status   statusCmd   `cmd:"" help:"Print local archive status."`
-	Sync     syncCmd     `cmd:"" help:"Sync conversations from fixtures or provider APIs."`
-	Search   searchCmd   `cmd:"" help:"Search the local archive."`
-	Show     showCmd     `cmd:"" help:"Show one local conversation."`
-	Archive  archiveCmd  `cmd:"" help:"Write compressed age-encrypted archive output."`
-	Publish  publishCmd  `cmd:"" help:"Publish local SQLite state as an encrypted snapshot."`
-	Import   importCmd   `cmd:"" help:"Import an encrypted snapshot into local SQLite."`
-	Store    storeCmd    `cmd:"" help:"Inspect generic tenant-store contracts."`
-	Guard    guardCmd    `cmd:"" help:"Check commit guardrails."`
-	Version  versionCmd  `cmd:"" help:"Print version information."`
+	Doctor    doctorCmd    `cmd:"" help:"Check local configuration."`
+	Metadata  metadataCmd  `cmd:"" help:"Print machine-readable metadata."`
+	Describe  describeCmd  `cmd:"" help:"Print machine-readable command schemas."`
+	Status    statusCmd    `cmd:"" help:"Print local archive status."`
+	Sync      syncCmd      `cmd:"" help:"Sync conversations from fixtures or provider APIs."`
+	Search    searchCmd    `cmd:"" help:"Search the local archive."`
+	Show      showCmd      `cmd:"" help:"Show one local conversation."`
+	Archive   archiveCmd   `cmd:"" help:"Write compressed age-encrypted archive output."`
+	Publish   publishCmd   `cmd:"" help:"Publish local SQLite state as an encrypted snapshot."`
+	Import    importCmd    `cmd:"" help:"Import an encrypted snapshot into local SQLite."`
+	Subscribe subscribeCmd `cmd:"" help:"Import snapshots from a local encrypted tenant store."`
+	Store     storeCmd     `cmd:"" help:"Inspect generic tenant-store contracts."`
+	Guard     guardCmd     `cmd:"" help:"Check commit guardrails."`
+	Version   versionCmd   `cmd:"" help:"Print version information."`
 }
 
 type commandContext struct {
@@ -633,6 +634,117 @@ func (cmd importCmd) Run(ctx commandContext) error {
 	return writeMaybeJSON(ctx.stdout, cmd.JSON, importResult{Input: cmd.In, Records: len(records), Sync: result})
 }
 
+type subscribeCmd struct {
+	Store    string `arg:"" help:"Local tenant store root containing manifest.json."`
+	Identity string `help:"Age identity. Defaults to FINCRAWL_AGE_IDENTITY."`
+	DryRun   bool   `name:"dry-run" help:"Verify and list planned imports without writing local state."`
+	JSON     bool   `help:"Print JSON output." default:"true"`
+}
+
+type subscribeResult struct {
+	Store             string              `json:"store"`
+	DryRun            bool                `json:"dry_run,omitempty"`
+	Snapshots         []subscribeSnapshot `json:"snapshots"`
+	ImportedSnapshots int                 `json:"imported_snapshots"`
+	Records           int                 `json:"records"`
+	Sync              *store.SyncResult   `json:"sync,omitempty"`
+}
+
+type subscribeSnapshot struct {
+	Path     string `json:"path"`
+	Records  int    `json:"records,omitempty"`
+	Imported bool   `json:"imported,omitempty"`
+}
+
+type pendingSnapshotImport struct {
+	snapshot tenantstore.SnapshotFile
+	records  []archive.Record
+	fixture  store.Fixture
+}
+
+func (cmd subscribeCmd) Run(ctx commandContext) error {
+	rt, err := config.LoadRuntime()
+	if err != nil {
+		return err
+	}
+	report, snapshots, err := tenantstore.VerifiedSnapshots(ctx, cmd.Store)
+	if err != nil {
+		return err
+	}
+	if !report.OK {
+		if writeErr := writeMaybeJSON(ctx.stdout, cmd.JSON, report); writeErr != nil {
+			return writeErr
+		}
+		return fmt.Errorf("tenant store verification failed with %d finding(s)", len(report.Findings))
+	}
+	result := subscribeResult{Store: report.Root, DryRun: cmd.DryRun, Snapshots: make([]subscribeSnapshot, 0, len(snapshots))}
+	for _, snapshot := range snapshots {
+		if !strings.HasSuffix(snapshot.Path, ".jsonl.zst.age") {
+			return fmt.Errorf("subscribe currently imports only .jsonl.zst.age snapshots: %s", snapshot.Path)
+		}
+		result.Snapshots = append(result.Snapshots, subscribeSnapshot{Path: snapshot.Path, Records: snapshot.Records})
+		result.Records += snapshot.Records
+	}
+	if cmd.DryRun {
+		return writeMaybeJSON(ctx.stdout, cmd.JSON, result)
+	}
+	identity := strings.TrimSpace(cmd.Identity)
+	if identity == "" {
+		identity = config.AgeIdentity()
+	}
+	pending := make([]pendingSnapshotImport, 0, len(snapshots))
+	totalRecords := 0
+	for _, snapshot := range snapshots {
+		records, err := archive.ReadEncryptedJSONL(snapshot.FullPath, identity)
+		if err != nil {
+			return fmt.Errorf("read tenant store snapshot %s: %w", snapshot.Path, err)
+		}
+		fixture, err := archive.RecordsFixture(records)
+		if err != nil {
+			return fmt.Errorf("decode tenant store snapshot %s: %w", snapshot.Path, err)
+		}
+		pending = append(pending, pendingSnapshotImport{snapshot: snapshot, records: records, fixture: fixture})
+		totalRecords += len(records)
+	}
+	if err := config.EnsureDirs(rt); err != nil {
+		return err
+	}
+	lck, err := lock.Acquire(rt.Config.DBPath)
+	if err != nil {
+		return err
+	}
+	defer lck.Release()
+	aggregate := store.SyncResult{}
+	for index, item := range pending {
+		syncResult, err := store.SyncFixture(ctx, rt.Config.DBPath, item.fixture)
+		if err != nil {
+			return fmt.Errorf("import tenant store snapshot %s: %w", item.snapshot.Path, err)
+		}
+		mergeSyncResult(&aggregate, syncResult)
+		result.Snapshots[index].Records = len(item.records)
+		result.Snapshots[index].Imported = true
+		result.Snapshots[index].Path = item.snapshot.Path
+		result.ImportedSnapshots++
+	}
+	result.Records = totalRecords
+	result.Sync = &aggregate
+	return writeMaybeJSON(ctx.stdout, cmd.JSON, result)
+}
+
+func mergeSyncResult(dst *store.SyncResult, src store.SyncResult) {
+	if dst.WorkspaceID == "" {
+		dst.WorkspaceID = src.WorkspaceID
+	}
+	dst.Conversations += src.Conversations
+	dst.ConversationParts += src.ConversationParts
+	dst.Admins += src.Admins
+	dst.Teams += src.Teams
+	dst.Tags += src.Tags
+	dst.Contacts += src.Contacts
+	dst.RawBlobs += src.RawBlobs
+	dst.Warnings = append(dst.Warnings, src.Warnings...)
+}
+
 type storeCmd struct {
 	Verify storeVerifyCmd `cmd:"" help:"Verify an encrypted tenant-store manifest."`
 }

internal/cli/cli_test.go

@@ -707,6 +707,96 @@ func TestPublishImportEncryptedSnapshotRoundTrip(t *testing.T) {
 	}
 }
 
+func TestSubscribeDryRunVerifiesLocalStoreWithoutIdentity(t *testing.T) {
+	storeRoot := filepath.Join(t.TempDir(), "store")
+	if err := os.MkdirAll(filepath.Join(storeRoot, "snapshots"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(storeRoot, "snapshots", "synthetic.jsonl.zst.age"), []byte("encrypted placeholder"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(storeRoot, "manifest.json"), []byte(`{"snapshots":[{"path":" snapshots/synthetic.jsonl.zst.age ","records":13}]}`), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("FINCRAWL_HOME", t.TempDir())
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	if err := Run(context.Background(), []string{"subscribe", storeRoot, "--dry-run", "--json"}, &stdout, &stderr); err != nil {
+		t.Fatal(err)
+	}
+	var result struct {
+		DryRun            bool `json:"dry_run"`
+		ImportedSnapshots int  `json:"imported_snapshots"`
+		Records           int  `json:"records"`
+		Snapshots         []struct {
+			Path    string `json:"path"`
+			Records int    `json:"records"`
+		} `json:"snapshots"`
+	}
+	if err := json.Unmarshal(stdout.Bytes(), &result); err != nil {
+		t.Fatal(err)
+	}
+	if !result.DryRun || result.ImportedSnapshots != 0 || result.Records != 13 || len(result.Snapshots) != 1 {
+		t.Fatalf("result = %#v", result)
+	}
+	if result.Snapshots[0].Path != "snapshots/synthetic.jsonl.zst.age" {
+		t.Fatalf("snapshot path = %q", result.Snapshots[0].Path)
+	}
+}
+
+func TestSubscribeImportsLocalTenantStore(t *testing.T) {
+	sourceHome := t.TempDir()
+	targetHome := t.TempDir()
+	t.Setenv("FINCRAWL_HOME", sourceHome)
+	identity, err := age.GenerateX25519Identity()
+	if err != nil {
+		t.Fatal(err)
+	}
+	storeRoot := filepath.Join("tmp", fmt.Sprintf("subscribe-store-%d", time.Now().UnixNano()))
+	out := filepath.Join(storeRoot, "snapshots", "synthetic.jsonl.zst.age")
+	t.Cleanup(func() {
+		_ = os.RemoveAll(storeRoot)
+	})
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	if err := Run(context.Background(), []string{"sync", "--fixture", filepath.Join("..", "..", "testdata", "synthetic")}, &stdout, &stderr); err != nil {
+		t.Fatal(err)
+	}
+	stdout.Reset()
+	if err := Run(context.Background(), []string{"publish", "--recipient", identity.Recipient().String(), "--out", out}, &stdout, &stderr); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(storeRoot, "manifest.json"), []byte(`{"snapshots":[{"path":"snapshots/synthetic.jsonl.zst.age"}]}`), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	stdout.Reset()
+	if err := os.Setenv("FINCRAWL_HOME", targetHome); err != nil {
+		t.Fatal(err)
+	}
+	if err := Run(context.Background(), []string{"subscribe", storeRoot, "--identity", identity.String(), "--json"}, &stdout, &stderr); err != nil {
+		t.Fatal(err)
+	}
+	var result struct {
+		ImportedSnapshots int `json:"imported_snapshots"`
+		Records           int `json:"records"`
+	}
+	if err := json.Unmarshal(stdout.Bytes(), &result); err != nil {
+		t.Fatal(err)
+	}
+	if result.ImportedSnapshots != 1 || result.Records == 0 {
+		t.Fatalf("result = %#v", result)
+	}
+	stdout.Reset()
+	if err := Run(context.Background(), []string{"search", "Morgan", "--fields", "provider_id,subject", "--json"}, &stdout, &stderr); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Contains(stdout.Bytes(), []byte(`"provider_id": "ic_syn_002"`)) {
+		t.Fatalf("stdout = %q", stdout.String())
+	}
+}
+
 func TestParseSinceAcceptsDayDurations(t *testing.T) {
 	now := time.Date(2026, 5, 16, 12, 0, 0, 0, time.UTC)
 	got, err := parseSince("2d", now, "updated-since")

internal/control/control.go

@@ -43,20 +43,21 @@ func Manifest(rt config.Runtime) ckcontrol.Manifest {
 		DefaultShare:    rt.Config.ShareDir,
 	}
 	manifest.Commands = map[string]ckcontrol.Command{
-		"doctor":   {Title: "Check local configuration", Argv: []string{"fincrawl", "doctor", "--offline"}, JSON: true},
-		"metadata": {Title: "Print machine-readable metadata", Argv: []string{"fincrawl", "metadata"}, JSON: true},
-		"describe": {Title: "Print command schemas", Argv: []string{"fincrawl", "describe"}, JSON: true},
-		"status":   {Title: "Print archive status", Argv: []string{"fincrawl", "status"}, JSON: true},
-		"sync":     {Title: "Sync conversations", Argv: []string{"fincrawl", "sync", "--fixture", "testdata/synthetic", "--dry-run"}, JSON: true, Mutates: true},
-		"search":   {Title: "Search local archive", Argv: []string{"fincrawl", "search", "query", "--fields", "provider_id,subject,score,updated_at"}, JSON: true},
-		"show":     {Title: "Show one local conversation", Argv: []string{"fincrawl", "show", "provider-conversation-id", "--fields", "provider_id,subject,tags,snippet"}, JSON: true},
-		"archive":  {Title: "Write encrypted archive", Argv: []string{"fincrawl", "archive", "--fixture", "testdata/synthetic", "--recipient", syntheticArchiveRecipient, "--out", "tmp/fincrawl-smoke.jsonl.zst.age", "--dry-run"}, JSON: true, Mutates: true},
-		"publish":  {Title: "Publish encrypted snapshot", Argv: []string{"fincrawl", "publish", "--recipient", syntheticArchiveRecipient, "--out", "tmp/fincrawl-publish.jsonl.zst.age", "--dry-run"}, JSON: true, Mutates: true},
-		"import":   {Title: "Import encrypted snapshot", Argv: []string{"fincrawl", "import", "--in", "tmp/fincrawl-publish.jsonl.zst.age", "--dry-run"}, JSON: true, Mutates: true},
-		"store":    {Title: "Verify encrypted tenant store", Argv: []string{"fincrawl", "store", "verify", "."}, JSON: true},
-		"guard":    {Title: "Check commit guardrails", Argv: []string{"fincrawl", "guard"}, JSON: true},
+		"doctor":    {Title: "Check local configuration", Argv: []string{"fincrawl", "doctor", "--offline"}, JSON: true},
+		"metadata":  {Title: "Print machine-readable metadata", Argv: []string{"fincrawl", "metadata"}, JSON: true},
+		"describe":  {Title: "Print command schemas", Argv: []string{"fincrawl", "describe"}, JSON: true},
+		"status":    {Title: "Print archive status", Argv: []string{"fincrawl", "status"}, JSON: true},
+		"sync":      {Title: "Sync conversations", Argv: []string{"fincrawl", "sync", "--fixture", "testdata/synthetic", "--dry-run"}, JSON: true, Mutates: true},
+		"search":    {Title: "Search local archive", Argv: []string{"fincrawl", "search", "query", "--fields", "provider_id,subject,score,updated_at"}, JSON: true},
+		"show":      {Title: "Show one local conversation", Argv: []string{"fincrawl", "show", "provider-conversation-id", "--fields", "provider_id,subject,tags,snippet"}, JSON: true},
+		"archive":   {Title: "Write encrypted archive", Argv: []string{"fincrawl", "archive", "--fixture", "testdata/synthetic", "--recipient", syntheticArchiveRecipient, "--out", "tmp/fincrawl-smoke.jsonl.zst.age", "--dry-run"}, JSON: true, Mutates: true},
+		"publish":   {Title: "Publish encrypted snapshot", Argv: []string{"fincrawl", "publish", "--recipient", syntheticArchiveRecipient, "--out", "tmp/fincrawl-publish.jsonl.zst.age", "--dry-run"}, JSON: true, Mutates: true},
+		"import":    {Title: "Import encrypted snapshot", Argv: []string{"fincrawl", "import", "--in", "tmp/fincrawl-publish.jsonl.zst.age", "--dry-run"}, JSON: true, Mutates: true},
+		"subscribe": {Title: "Import local tenant store snapshots", Argv: []string{"fincrawl", "subscribe", "tenant-store", "--dry-run"}, JSON: true, Mutates: true},
+		"store":     {Title: "Verify encrypted tenant store", Argv: []string{"fincrawl", "store", "verify", "."}, JSON: true},
+		"guard":     {Title: "Check commit guardrails", Argv: []string{"fincrawl", "guard"}, JSON: true},
 	}
-	manifest.Capabilities = []string{"intercom-api-sync", "sqlite-fts-search", "conversation-show", "zstd-age-archive", "encrypted-snapshot-import", "tenant-store-verify", "synthetic-fixtures"}
+	manifest.Capabilities = []string{"intercom-api-sync", "sqlite-fts-search", "conversation-show", "zstd-age-archive", "encrypted-snapshot-import", "local-tenant-store-subscribe", "tenant-store-verify", "synthetic-fixtures"}
 	manifest.Privacy = ckcontrol.Privacy{
 		ContainsPrivateMessages: true,
 		ExportsSecrets:          false,