Merge pull request #322 from Gentleman-Programming/fix/cloud-chunks-observation-materialization

fix(sync): materialize mutation chunks for recovery

Author: Alan-TheGentleman

DOCS.md

@@ -393,6 +393,7 @@ Inspect or replay the `sync_apply_deferred` queue.
 - `engram cloud upgrade bootstrap --project <project> [--resume]` — resumable checkpointed enroll/push/verify flow
 - `engram cloud upgrade status --project <project>` — show upgrade stage/class/reason
 - `engram cloud upgrade rollback --project <project>` — restore pre-upgrade local snapshot before `bootstrap_verified`; blocked afterwards
+- `engram cloud repair materialize-mutations --project <project> (--dry-run|--apply)` — explicit server-side Postgres repair that backfills existing `cloud_mutations` into compatible `cloud_chunks` without deleting remote data
 
 Cloud auth token is provided at runtime via `ENGRAM_CLOUD_TOKEN` (not by a dedicated CLI subcommand).
 Cloud server startup fails closed when the token is missing unless `ENGRAM_CLOUD_INSECURE_NO_AUTH=1` is explicitly set for local insecure development.
@@ -443,6 +444,17 @@ engram sync --cloud --project <project>
 
 Sync/autosync never auto-applies repairs; only the explicit `repair --apply` command mutates local repairable upgrade state.
 
+For cloud servers that already accepted mutation pushes before mutation payloads were materialized into chunk history, run the server-side backfill against the Postgres DSN used by `engram cloud serve`:
+
+```bash
+ENGRAM_DATABASE_URL='postgres://...' engram cloud repair materialize-mutations --project <project> --dry-run
+ENGRAM_DATABASE_URL='postgres://...' engram cloud repair materialize-mutations --project <project> --apply
+```
+
+The backfill is project-scoped, non-destructive, and idempotent: it inserts missing compatible chunks and leaves existing `cloud_mutations` and chunks in place.
+
+`engram cloud serve` also runs this materialization repair automatically for every configured `ENGRAM_CLOUD_ALLOWED_PROJECTS` entry at startup. The explicit repair command remains available for operator verification, dry-runs, and re-running a project after an upgrade.
+
 ### Local Cloud Bring-Up (Docker + Postgres)
 
 ```bash

cmd/engram/cloud.go

@@ -84,6 +84,10 @@ var newCloudRuntime = func(cfg cloud.Config) (cloudServerRuntime, error) {
 		return nil, err
 	}
 	allowedProjects := normalizeAllowedProjects(cfg.AllowedProjects)
+	if err := backfillAllowedProjectMutationChunks(context.Background(), cs, allowedProjects); err != nil {
+		_ = cs.Close()
+		return nil, err
+	}
 	projectAuth := auth.NewProjectScopeAuthorizer(allowedProjects)
 	token := strings.TrimSpace(os.Getenv("ENGRAM_CLOUD_TOKEN"))
 	cs.SetDashboardAllowedProjects(allowedProjects)
@@ -114,6 +118,22 @@ var newCloudRuntime = func(cfg cloud.Config) (cloudServerRuntime, error) {
 	}, nil
 }
 
+func backfillAllowedProjectMutationChunks(ctx context.Context, cs *cloudstore.CloudStore, projects []string) error {
+	for _, project := range projects {
+		report, err := cs.BackfillMutationChunks(ctx, project, true)
+		if err != nil {
+			return fmt.Errorf("cloud repair materialize-mutations for project %q: %w", project, err)
+		}
+		if report.CandidateMutations > 0 || report.ChunksInserted > 0 {
+			fmt.Fprintf(os.Stderr,
+				"engram cloud repair materialize-mutations: project=%s candidates=%d already_materialized=%d chunks_planned=%d chunks_inserted=%d\n",
+				report.Project, report.CandidateMutations, report.AlreadyMaterialized, report.ChunksPlanned, report.ChunksInserted,
+			)
+		}
+	}
+	return nil
+}
+
 var runUpgradeBootstrap = func(s *store.Store, project string, cc *cloudConfig) (*engramsync.UpgradeBootstrapResult, error) {
 	transport, err := remote.NewRemoteTransport(cc.ServerURL, cc.Token, project)
 	if err != nil {
@@ -130,12 +150,12 @@ type cloudConfig struct {
 func cmdCloud(cfg store.Config) {
 	if len(os.Args) < 3 {
 		fmt.Fprintln(os.Stderr, "usage: engram cloud <subcommand> [options]")
-		fmt.Fprintln(os.Stderr, "supported subcommands: status, enroll, config, serve, upgrade")
+		fmt.Fprintln(os.Stderr, "supported subcommands: status, enroll, config, serve, upgrade, repair")
 		exitFunc(1)
 	}
 	if os.Args[2] == "--help" || os.Args[2] == "-h" || os.Args[2] == "help" {
 		fmt.Println("usage: engram cloud <subcommand> [options]")
-		fmt.Println("supported subcommands: status, enroll, config, serve, upgrade")
+		fmt.Println("supported subcommands: status, enroll, config, serve, upgrade, repair")
 		return
 	}
 
@@ -150,11 +170,61 @@ func cmdCloud(cfg store.Config) {
 		cmdCloudServe()
 	case "upgrade":
 		cmdCloudUpgrade(cfg)
+	case "repair":
+		cmdCloudRepair()
 	default:
 		fmt.Fprintf(os.Stderr, "unknown cloud command: %s\n", os.Args[2])
-		fmt.Fprintln(os.Stderr, "supported subcommands: status, enroll, config, serve, upgrade")
+		fmt.Fprintln(os.Stderr, "supported subcommands: status, enroll, config, serve, upgrade, repair")
+		exitFunc(1)
+	}
+}
+
+func cmdCloudRepair() {
+	if len(os.Args) < 4 || os.Args[3] == "--help" || os.Args[3] == "-h" || os.Args[3] == "help" {
+		fmt.Println("usage: engram cloud repair materialize-mutations --project <name> (--dry-run|--apply)")
+		fmt.Println("repairs existing cloud_mutations into compatible cloud_chunks without deleting remote data")
+		return
+	}
+	command := strings.TrimSpace(strings.ToLower(os.Args[3]))
+	if command != "materialize-mutations" {
+		fmt.Fprintf(os.Stderr, "unknown cloud repair command: %s\n", command)
+		fmt.Fprintln(os.Stderr, "supported cloud repair commands: materialize-mutations")
+		exitFunc(1)
+		return
+	}
+	project := parseCloudUpgradeProjectArg(os.Args[4:])
+	if project == "" {
+		fmt.Fprintln(os.Stderr, "usage: engram cloud repair materialize-mutations --project <name> (--dry-run|--apply)")
+		fmt.Fprintln(os.Stderr, "error: --project is required")
 		exitFunc(1)
+		return
+	}
+	dryRun := hasCloudUpgradeFlag(os.Args[4:], "--dry-run")
+	apply := hasCloudUpgradeFlag(os.Args[4:], "--apply")
+	if dryRun == apply {
+		fmt.Fprintln(os.Stderr, "usage: engram cloud repair materialize-mutations --project <name> (--dry-run|--apply)")
+		fmt.Fprintln(os.Stderr, "error: exactly one of --dry-run or --apply is required")
+		exitFunc(1)
+		return
+	}
+
+	cs, err := cloudstore.New(cloud.ConfigFromEnv())
+	if err != nil {
+		fatal(err)
+		return
+	}
+	defer cs.Close()
+	report, err := cs.BackfillMutationChunks(context.Background(), project, apply)
+	if err != nil {
+		fatal(err)
+		return
+	}
+	encoded, err := jsonMarshalIndent(report, "", "  ")
+	if err != nil {
+		fatal(err)
+		return
 	}
+	fmt.Println(string(encoded))
 }
 
 func cmdCloudUpgrade(cfg store.Config) {