Expose reconcile state command

docs/cli.md

@@ -21,6 +21,7 @@ distributor [--help]
 distributor help
 distributor version [--format text|json]
 distributor run [--config <path>] [--dry-run] [--force] [--format text|json]
+distributor reconcile-state --config <path> --pipeline <id> --destination <id> [--all-owners] [--dry-run] [--format text|json]
 distributor serve [--config <path>]
 distributor validate [--format text|json] <path>
 distributor validate --config <path> --pipeline <id> [--bundle <path>] [--format text|json]
@@ -33,6 +34,7 @@ distributor manifest create --id <bundle-id> [options] <bundle-path>
 
 - `version` prints the application name and version.
 - `run` executes configured pipelines against their destinations.
+- `reconcile-state` repairs destination state records for missing managed outputs.
 - `serve` starts the authenticated HTTP upload API defined by the configuration file.
 - `validate` checks a local bundle path or a configured source bundle.
 - `inspect` reports manifest and file metadata for a local bundle path or a configured source bundle.
@@ -46,7 +48,7 @@ distributor manifest create --id <bundle-id> [options] <bundle-path>
 
 ### Common Output Format
 
-`--format text|json` is supported by `version`, `run`, `validate`, `inspect`, and `manifest create`.
+`--format text|json` is supported by `version`, `run`, `reconcile-state`, `validate`, `inspect`, and `manifest create`.
 
 - `text` is the default human-readable output.
 - `json` emits one JSON document for successful command execution.
@@ -73,6 +75,21 @@ distributor run [--config <path>] [--dry-run] [--force] [--format text|json]
 
 `run` accepts no positional arguments.
 
+### `reconcile-state`
+
+```sh
+distributor reconcile-state --config <path> --pipeline <id> --destination <id> [--all-owners] [--dry-run] [--format text|json]
+```
+
+- `--config <path>` loads the pipeline configuration and is required.
+- `--pipeline <id>` selects the pipeline used to identify the destination root and is required.
+- `--destination <id>` selects the destination root and is required.
+- `--all-owners` repairs missing managed output records for every owner in a shared-root state file. Without it, shared-root repair is scoped to the selected pipeline and destination owner.
+- `--dry-run` reports repairs without rewriting `.distributor.json`.
+- `--format text|json` selects human-readable or machine-readable output.
+
+Without `--dry-run`, `reconcile-state` applies state repair by removing records for managed outputs that no longer exist in storage. It reports unmanaged entries but does not delete destination files, adopt unmanaged files, or validate output digests. The command accepts no positional arguments.
+
 ### `serve`
 
 ```sh
@@ -174,6 +191,29 @@ go run ./cmd/distributor run --config examples/local-publish.yml
 
 Use `--format json` when automation needs structured run results. Use `--force` only when the operator has reviewed the destination state conflict and intentionally wants to continue.
 
+### Repair Destination State Records
+
+Preview missing managed output records for one configured destination:
+
+```sh
+go run ./cmd/distributor reconcile-state \
+  --config examples/local-publish.yml \
+  --pipeline example-source-bundle \
+  --destination local-archive \
+  --dry-run
+```
+
+Apply the repair after reviewing the report:
+
+```sh
+go run ./cmd/distributor reconcile-state \
+  --config examples/local-publish.yml \
+  --pipeline example-source-bundle \
+  --destination local-archive
+```
+
+Use `--all-owners` only for shared-root destination state when all owners inside the selected root should be repaired.
+
 ### Run HTML And Fan-Out Examples
 
 ```sh
@@ -217,6 +257,7 @@ Text output is optimized for direct operator use. JSON output is optimized for a
 - Use `validate` before `run` when checking a bundle supplied by another process.
 - Use `inspect --format json` when automation needs manifest metadata, normalized file details, or checksum information.
 - Use `run --dry-run` before publishing to review destination actions.
+- Use `reconcile-state --dry-run` to inspect missing managed output records before repairing destination state.
 - Use [Configuration](config.md) for schema and default details.
 - Use [Troubleshooting](troubleshooting.md) for common errors and corrective action.
 - Use [Operations](operations.md) for HTTP upload operation, state files, and recovery workflows.

docs/integrations/destination-state.md

@@ -116,6 +116,16 @@ If a merge publication fails after writing outputs, cleanup removes only newly c
 
 Normal single-owner replacement deletes only managed output paths recorded in `outputs` plus `.distributor.json`. Shared-root replacement deletes only omitted outputs for the current owner. Merge publication retains omitted managed outputs. Forced replacement deletes the bounded destination bundle path.
 
+## State Repair Semantics
+
+`distributor reconcile-state` can remove managed output records for files that no longer exist in destination storage. It uses the configured pipeline and destination selector to open one destination root and reads that root's `.distributor.json`.
+
+For single-owner state, the state `pipeline_id` and `destination_id` must match the selected pipeline and destination. The command checks output paths recorded in `outputs`, reports missing managed outputs, reports unmanaged entries under the destination root, and removes missing output records from valid state unless `--dry-run` is set.
+
+For shared-root state, repair is scoped to the selected owner by default. With `--all-owners`, it checks and repairs missing output records for every owner in the selected shared-root state file.
+
+State repair does not validate output digests, delete destination files, adopt unmanaged entries, or rewrite invalid or mismatched state.
+
 ## Compatibility
 
 `distributor` can read schema version `1` destination state for compatibility. When v1 state is read, it is treated as single-owner state with `reconciliation.mode: replace`. Missing top-level `created_at` and `updated_at` are inferred from `published_at`, and missing per-output timestamps are also inferred from `published_at`.

docs/internal/app.md

@@ -16,7 +16,7 @@ Outputs include `RunReport`, `ReconcileStateReport`, validate/inspect/manifest r
 
 `internal/app` wires packages together but does not own manifest validation rules, destination state comparison, storage path rules, publish safety policy, transform rendering, config schema validation, or backend protocol behavior.
 
-User-facing command parsing stays in `internal/cli`. User-facing config reference stays in `docs/config.md`. External contracts live under `docs/integrations/`.
+User-facing command parsing stays in `internal/cli`, including `reconcile-state` flag validation and help text. User-facing config reference stays in `docs/config.md`. External contracts live under `docs/integrations/`.
 
 ## Config Fields Used
 
@@ -34,7 +34,7 @@ The app layer registers default transforms, including Markdown-to-HTML, and supp
 
 Run workflows discover and validate source bundles through `internal/bundle`. Destination state actions are prepared and written through `internal/publish` and `internal/state`; the app layer records report projections of those actions and results.
 
-Reconcile-state workflows load one configured pipeline/destination selector, open that destination root, parse the root `.distributor.json`, and report missing managed output records plus unmanaged storage entries. Managed output existence checks use storage `Stat`; unmanaged reporting uses bounded storage `Walk` and excludes `.distributor.json` plus all paths already recorded as managed. Apply mode removes missing managed output records from state and rewrites valid state only; dry-run reports the same repair without writing. It does not validate output digests, delete destination files, adopt unmanaged files, or rewrite invalid or mismatched state.
+Reconcile-state workflows load one configured pipeline/destination selector, open that destination root, parse the root `.distributor.json`, and report missing managed output records plus unmanaged storage entries. Managed output existence checks use storage `Stat`; unmanaged reporting uses bounded storage `Walk` and excludes `.distributor.json` plus all paths already recorded as managed. Apply mode removes missing managed output records from state and rewrites valid state only; dry-run reports the same repair without writing. Text output reports `changed`, `would_change`, or `unchanged`; JSON output uses the shared app envelope. It does not validate output digests, delete destination files, adopt unmanaged files, or rewrite invalid or mismatched state.
 
 HTTP uploads stage and validate archives before enqueueing a pipeline run with a local staged source root. Go producers can use the public `pkg/upload` package to create client-side gzip tar uploads for this server contract; `internal/app` remains the server-side orchestration boundary and does not import that producer package.
 
@@ -57,6 +57,7 @@ HTTP upload startup fails if upload tokens are missing, empty, or duplicated. Up
 ## Tests To Inspect
 
 - `internal/app/*_test.go`
+- `internal/cli/reconcile_state_test.go`
 - `internal/cli/root_test.go`
 - `internal/config/*_test.go`
 - `internal/ingest/*_test.go`

docs/internal/state.md

@@ -48,7 +48,7 @@ Comparison is pure. It returns outcomes for absent state, unmanaged content, inv
 
 Shared-root comparison is owner-scoped. It compares only the owner keyed by the current pipeline id and destination id, treats an absent owner as absent destination state for that owner, and can compare compatible single-owner state for the current owner without converting unrelated single-owner state.
 
-Missing-output removal helpers are pure state transformations. They remove matching output records only and leave storage inspection, timestamp updates, validation, and state rewrites to callers.
+Missing-output removal helpers are pure state transformations used by app-level state repair. They remove matching output records only and leave storage inspection, timestamp updates, validation, and state rewrites to callers.
 
 ## Failure Behavior
 
@@ -59,6 +59,8 @@ Parsing rejects invalid JSON, trailing data, missing required fields, invalid ti
 - `internal/state/distributor_test.go`
 - `internal/state/shared_root_test.go`
 - `internal/state/compare_test.go`
+- `internal/app/reconcile_state_test.go`
+- `internal/cli/reconcile_state_test.go`
 - `internal/publish/*_test.go`
 
 ## Architectural Invariants

docs/internal/storage.md

@@ -30,9 +30,11 @@ Storage owns `.distributor.json` path helpers through `StateFileName`, `StatePat
 
 Logical paths are slash-separated and relative to a backend root. Prefix validation allows an empty prefix to mean the backend root; file path validation requires a non-empty path.
 
+Reconcile-state callers use `Stat` to check whether managed output paths still exist and bounded recursive `Walk` to report unmanaged entries under a selected destination root. Storage does not decide whether entries are managed; callers compare entries against destination state.
+
 ## Skip And Resume Behavior
 
-Storage has no publication skip policy. It supplies `HasAny` for unmanaged-content checks, `DeleteManagedBundle` target construction for normal replacement cleanup, and `DeletePrefix` semantics for explicit forced replacement.
+Storage has no publication skip policy. It supplies `HasAny` for unmanaged-content checks, `Stat` and `Walk` for state repair inspection, `DeleteManagedBundle` target construction for normal replacement cleanup, and `DeletePrefix` semantics for explicit forced replacement.
 
 ## Failure Behavior
 
@@ -42,6 +44,7 @@ Storage errors use typed categories: not found, already exists, not empty, inval
 
 - `internal/storage/*_test.go`
 - `internal/storage/fake/*_test.go`
+- `internal/app/reconcile_state_test.go`
 - `internal/adapters/local/*_test.go`
 - `internal/adapters/ssh/*_test.go`
 - `internal/adapters/s3/*_test.go`

docs/operations.md

@@ -87,6 +87,35 @@ If a write fails after some outputs were written, `distributor` attempts cleanup
 
 Fan-out destinations are independent. If one destination fails after planning or execution begins, later destinations are still attempted. The command exits non-zero if any destination failed.
 
+## Destination State Repair
+
+Use `reconcile-state` when `.distributor.json` still records managed outputs that no longer exist in destination storage. This repairs the state record only; it does not restore missing files.
+
+Preview the repair first:
+
+```sh
+go run ./cmd/distributor reconcile-state \
+  --config <config-path> \
+  --pipeline <pipeline-id> \
+  --destination <destination-id> \
+  --dry-run
+```
+
+Apply after reviewing the report:
+
+```sh
+go run ./cmd/distributor reconcile-state \
+  --config <config-path> \
+  --pipeline <pipeline-id> \
+  --destination <destination-id>
+```
+
+The command opens the configured destination root selected by `--pipeline` and `--destination`, reads the root `.distributor.json`, checks each managed output path with storage metadata, reports missing managed outputs, and reports unmanaged entries under that root. It excludes `.distributor.json` and already managed paths from unmanaged reporting.
+
+Without `--dry-run`, it removes missing managed output records from valid state and rewrites `.distributor.json`. It never deletes destination files, adopts unmanaged files, validates output digests, or rewrites invalid or mismatched state.
+
+For single-owner state, the state owner must match the selected pipeline and destination. For shared-root state, repair is scoped to the selected owner by default. Add `--all-owners` only when every owner in the selected shared-root state should have missing managed output records removed.
+
 ## Dry Runs And Output Review
 
 `run --dry-run` loads config, resolves credentials, discovers source bundles, opens destinations, inspects destination state, builds publish plans, and prints actions. It does not write outputs, `.distributor.json`, or SSH `known_hosts` entries. For reconciliation, dry runs report the same high-level action labels as execution; inspect the configured destination's `reconciliation.mode` to determine whether `replace_older` will replace the managed set or merge into it.
@@ -224,6 +253,7 @@ Use these recovery boundaries:
 - For an empty or missing destination, rerun after fixing config or storage access.
 - For unmanaged destination content, move unrelated files aside or use a different destination path before publishing.
 - For shared-root ownership conflicts, change one owner so it writes a different destination path, or use a separate destination root.
+- For missing managed output files recorded in state, run `reconcile-state --dry-run`, then apply `reconcile-state` if the missing files should no longer be considered managed.
 - For failed writes, inspect the destination bundle path, remove only confirmed partial outputs if needed, then rerun `--dry-run`. In merge mode, retained outputs may be intentional managed outputs from the prior state.
 - For state conflicts, verify the source, pipeline, destination, and existing `.distributor.json` before considering `--force`.
 - For HTTP upload failures, inspect `/runs/<run-id>` while retained; after expiry or restart, rely on destination state and logs/output from the publishing run.

docs/troubleshooting.md

@@ -69,6 +69,24 @@ Safe fix: use either `distributor validate <path>` / `distributor inspect <path>
 
 Reference: [CLI](cli.md#validate).
 
+## Reconcile-State Selector Is Missing Or Wrong
+
+Symptom: `reconcile-state requires --config`, `requires --pipeline`, `requires --destination`, `pipeline "<id>" not found`, `destination <id> not found`, or `state owner is ... not ...`.
+
+Likely cause: the command did not identify one configured destination root, or the selected root contains state for a different single-owner pipeline/destination.
+
+Diagnostic:
+
+```sh
+go run ./cmd/distributor reconcile-state --help
+rg -n 'pipelines:|destinations:|id:' <config-path>
+cat <destination-path>/.distributor.json
+```
+
+Safe fix: pass the configured `--config`, `--pipeline`, and `--destination` values that identify the destination root containing the state file. For unrelated single-owner state, use the correct config selector or a separate destination root; `reconcile-state` will not take over mismatched state.
+
+Reference: [CLI](cli.md#reconcile-state).
+
 ## Output Format Is Invalid
 
 Symptom: `format must be text or json`.
@@ -182,6 +200,26 @@ Safe fix: verify the source and destination are intended to match. Use a separat
 
 Reference: [Operations](operations.md#destination-state-and-retry-behavior).
 
+## Destination State References Missing Managed Outputs
+
+Symptom: `reconcile-state --dry-run` reports `status=would_change` or JSON `missing_managed_outputs` entries.
+
+Likely cause: files that were recorded as managed in `.distributor.json` were removed outside `distributor`, or a previous external cleanup removed destination files without updating state.
+
+Diagnostic:
+
+```sh
+go run ./cmd/distributor reconcile-state \
+  --config <config-path> \
+  --pipeline <pipeline-id> \
+  --destination <destination-id> \
+  --dry-run
+```
+
+Safe fix: if the missing files should no longer be managed, rerun the same command without `--dry-run` to remove only the missing managed output records from `.distributor.json`. The command does not delete destination files or adopt unmanaged entries. If the files should exist, restore them from backup or republish instead.
+
+Reference: [Operations](operations.md#destination-state-repair).
+
 ## Destination Is Newer Than Source
 
 Symptom: `skip_destination_newer` or `destination is newer and replacement requires --force`.

internal/cli/reconcile_state.go

@@ -0,0 +1,84 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/app"
+)
+
+func reconcileStateCommand(ctx context.Context, args []string, stdout, stderr io.Writer) int {
+	if hasHelp(args) {
+		printReconcileStateHelp(stdout)
+		return exitOK
+	}
+
+	flags := newFlagSet("reconcile-state", stderr)
+	configPath := flags.String("config", "", "path to config file")
+	pipelineID := flags.String("pipeline", "", "pipeline id")
+	destinationID := flags.String("destination", "", "destination id")
+	allOwners := flags.Bool("all-owners", false, "repair all shared-root owners in the selected destination root")
+	dryRun := flags.Bool("dry-run", false, "report repairs without rewriting state")
+	formatFlag := addFormatFlag(flags)
+	if err := flags.Parse(args); err != nil {
+		return exitUsage
+	}
+	if rejectPositionalArgs(stderr, "reconcile-state", flags.Args()) {
+		return exitUsage
+	}
+	format, ok := parseOutputFormat(stderr, "reconcile-state", *formatFlag)
+	if !ok {
+		return exitUsage
+	}
+	if !validateReconcileStateFlags(stderr, *configPath, *pipelineID, *destinationID) {
+		return exitUsage
+	}
+
+	if _, err := app.ReconcileState(ctx, app.ReconcileStateOptions{
+		ConfigPath:    *configPath,
+		PipelineID:    *pipelineID,
+		DestinationID: *destinationID,
+		AllOwners:     *allOwners,
+		DryRun:        *dryRun,
+		Stdout:        stdout,
+		OutputFormat:  format,
+	}); err != nil {
+		return fail(stderr, err)
+	}
+	return exitOK
+}
+
+func validateReconcileStateFlags(stderr io.Writer, configPath, pipelineID, destinationID string) bool {
+	if configPath == "" {
+		fmt.Fprintf(stderr, "%s: reconcile-state requires --config\n", app.Name)
+		return false
+	}
+	if pipelineID == "" {
+		fmt.Fprintf(stderr, "%s: reconcile-state requires --pipeline\n", app.Name)
+		return false
+	}
+	if destinationID == "" {
+		fmt.Fprintf(stderr, "%s: reconcile-state requires --destination\n", app.Name)
+		return false
+	}
+	return true
+}
+
+func printReconcileStateHelp(w io.Writer) {
+	fmt.Fprint(w, `Usage:
+  distributor reconcile-state --config <path> --pipeline <id> --destination <id> [--all-owners] [--dry-run] [--format text|json]
+
+Options:
+  --config <path>       Path to config file
+  --pipeline <id>       Pipeline id that selects the destination root
+  --destination <id>    Destination id that selects the destination root
+  --all-owners          Repair all shared-root owners in the selected destination root
+  --dry-run             Report repairs without rewriting state
+  --format text|json    Output format
+
+Reconcile-state checks managed output records against destination storage and
+removes records for missing managed outputs unless --dry-run is set. It reports
+unmanaged entries but does not delete or adopt destination files.
+`)
+}

internal/cli/reconcile_state_test.go

@@ -0,0 +1,190 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/state"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
+)
+
+func TestExecuteReconcileStateAppliesByDefault(t *testing.T) {
+	_, destinationRoot, configPath := writeReconcileStateLocalFixture(t)
+	if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("# Report\nSunny.\n"), 0o600); err != nil {
+		t.Fatalf("write managed output: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(destinationRoot, "extra.txt"), []byte("unmanaged"), 0o600); err != nil {
+		t.Fatalf("write unmanaged output: %v", err)
+	}
+	var stdout, stderr bytes.Buffer
+
+	code := Execute(context.Background(), []string{
+		"reconcile-state",
+		"--config", configPath,
+		"--pipeline", "reports",
+		"--destination", "archive",
+	}, &stdout, &stderr)
+
+	if code != exitOK {
+		t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
+	}
+	if !strings.Contains(stdout.String(), "status=changed") {
+		t.Fatalf("stdout = %q, want changed status", stdout.String())
+	}
+	destinationState := testutil.ReadDestinationState(t, filepath.Join(destinationRoot, storage.StateFileName))
+	if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "report.md" {
+		t.Fatalf("state outputs = %q, want report.md", got)
+	}
+	assertLocalFile(t, filepath.Join(destinationRoot, "extra.txt"), "unmanaged")
+	if stderr.Len() != 0 {
+		t.Fatalf("stderr = %q, want empty", stderr.String())
+	}
+}
+
+func TestExecuteReconcileStateDryRunReportsWithoutWriting(t *testing.T) {
+	_, destinationRoot, configPath := writeReconcileStateLocalFixture(t)
+	if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("# Report\nSunny.\n"), 0o600); err != nil {
+		t.Fatalf("write managed output: %v", err)
+	}
+	var stdout, stderr bytes.Buffer
+
+	code := Execute(context.Background(), []string{
+		"reconcile-state",
+		"--config", configPath,
+		"--pipeline", "reports",
+		"--destination", "archive",
+		"--dry-run",
+	}, &stdout, &stderr)
+
+	if code != exitOK {
+		t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
+	}
+	if !strings.Contains(stdout.String(), "status=would_change") {
+		t.Fatalf("stdout = %q, want would_change status", stdout.String())
+	}
+	destinationState := testutil.ReadDestinationState(t, filepath.Join(destinationRoot, storage.StateFileName))
+	if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "report.md,summary.txt" {
+		t.Fatalf("state outputs = %q, want original outputs", got)
+	}
+	if stderr.Len() != 0 {
+		t.Fatalf("stderr = %q, want empty", stderr.String())
+	}
+}
+
+func TestExecuteReconcileStateJSONReport(t *testing.T) {
+	_, destinationRoot, configPath := writeReconcileStateLocalFixture(t)
+	if err := os.WriteFile(filepath.Join(destinationRoot, "report.md"), []byte("# Report\nSunny.\n"), 0o600); err != nil {
+		t.Fatalf("write managed output: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(destinationRoot, "extra.txt"), []byte("unmanaged"), 0o600); err != nil {
+		t.Fatalf("write unmanaged output: %v", err)
+	}
+	var stdout, stderr bytes.Buffer
+
+	code := Execute(context.Background(), []string{
+		"reconcile-state",
+		"--config", configPath,
+		"--pipeline", "reports",
+		"--destination", "archive",
+		"--dry-run",
+		"--format", "json",
+	}, &stdout, &stderr)
+
+	if code != exitOK {
+		t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
+	}
+	envelope := decodeEnvelope(t, &stdout)
+	if envelope["command"] != "reconcile-state" || envelope["ok"] != true {
+		t.Fatalf("envelope = %#v, want reconcile-state ok", envelope)
+	}
+	result := envelopeResult(t, envelope)
+	if result["would_change"] != true || result["changed"] != false || result["dry_run"] != true {
+		t.Fatalf("result = %#v, want dry-run pending change", result)
+	}
+	missing, ok := result["missing_managed_outputs"].([]any)
+	if !ok || len(missing) != 1 {
+		t.Fatalf("missing outputs = %#v, want one", result["missing_managed_outputs"])
+	}
+	unmanaged, ok := result["unmanaged_entries"].([]any)
+	if !ok || len(unmanaged) != 1 {
+		t.Fatalf("unmanaged entries = %#v, want one", result["unmanaged_entries"])
+	}
+	if stderr.Len() != 0 {
+		t.Fatalf("stderr = %q, want empty", stderr.String())
+	}
+}
+
+func TestExecuteReconcileStateRejectsInvalidFlags(t *testing.T) {
+	tests := []struct {
+		name       string
+		args       []string
+		wantStderr string
+	}{
+		{
+			name:       "missing config",
+			args:       []string{"reconcile-state", "--pipeline", "reports", "--destination", "archive"},
+			wantStderr: "requires --config",
+		},
+		{
+			name:       "missing pipeline",
+			args:       []string{"reconcile-state", "--config", "config.yml", "--destination", "archive"},
+			wantStderr: "requires --pipeline",
+		},
+		{
+			name:       "missing destination",
+			args:       []string{"reconcile-state", "--config", "config.yml", "--pipeline", "reports"},
+			wantStderr: "requires --destination",
+		},
+		{
+			name:       "invalid format",
+			args:       []string{"reconcile-state", "--config", "config.yml", "--pipeline", "reports", "--destination", "archive", "--format", "xml"},
+			wantStderr: "format must be text or json",
+		},
+		{
+			name:       "positional",
+			args:       []string{"reconcile-state", "--config", "config.yml", "--pipeline", "reports", "--destination", "archive", "extra"},
+			wantStderr: "does not accept positional arguments",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var stdout, stderr bytes.Buffer
+			code := Execute(context.Background(), tt.args, &stdout, &stderr)
+			if code != exitUsage {
+				t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitUsage, stderr.String())
+			}
+			if stdout.Len() != 0 {
+				t.Fatalf("stdout = %q, want empty", stdout.String())
+			}
+			if !strings.Contains(stderr.String(), tt.wantStderr) {
+				t.Fatalf("stderr = %q, want substring %q", stderr.String(), tt.wantStderr)
+			}
+		})
+	}
+}
+
+func writeReconcileStateLocalFixture(t *testing.T) (string, string, string) {
+	t.Helper()
+	sourceRoot := t.TempDir()
+	destinationRoot := t.TempDir()
+	manifest := testutil.WriteSourceBundle(t, sourceRoot, "", testutil.BundleOptions{})
+	testutil.WriteDestinationState(t, destinationRoot, "", manifest, testutil.DestinationStateOptions{})
+	configPath := testutil.WriteMinimalLocalConfig(t, sourceRoot, destinationRoot)
+	return sourceRoot, destinationRoot, configPath
+}
+
+func assertLocalFile(t *testing.T, path, want string) {
+	t.Helper()
+	data, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read file %s: %v", path, err)
+	}
+	if string(data) != want {
+		t.Fatalf("file %s = %q, want %q", path, data, want)
+	}
+}

internal/cli/root.go

@@ -29,6 +29,8 @@ func Execute(ctx context.Context, args []string, stdout, stderr io.Writer) int {
 		return versionCommand(ctx, args[1:], stdout, stderr)
 	case "run":
 		return runCommand(ctx, args[1:], stdout, stderr)
+	case "reconcile-state":
+		return reconcileStateCommand(ctx, args[1:], stdout, stderr)
 	case "serve":
 		return serveCommand(ctx, args[1:], stdout, stderr)
 	case "validate":
@@ -53,6 +55,8 @@ Usage:
 Commands:
   version    Print version information
   run        Run configured distribution pipelines
+  reconcile-state
+             Repair missing managed-output records in destination state
   serve      Run the HTTP upload server
   validate   Validate a source bundle or bundle tree
   inspect    Inspect bundles or distributor state