From d6deccf3e8b2888255145b3838fced1769fc61ea Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 19:47:12 +0000 Subject: [PATCH] Add immutable remote commit reader --- docs/internal/command-restore.md | 5 +- docs/internal/manifest.md | 24 ++ docs/operations.md | 14 ++ docs/roadmap/implementation.md | 2 +- internal/app/restore_discovery.go | 6 +- internal/artifacts/current_state.go | 98 +------- internal/artifacts/current_state_commit.go | 187 +++++++++++++++ internal/artifacts/current_state_legacy.go | 98 ++++++++ internal/artifacts/current_state_test.go | 27 +++ internal/artifacts/remote_commit.go | 264 +++++++++++++++++++++ internal/artifacts/remote_commit_test.go | 253 ++++++++++++++++++++ internal/artifacts/s3_keys.go | 12 + internal/artifacts/s3_keys_test.go | 8 + internal/config/defaults.go | 14 +- 14 files changed, 914 insertions(+), 98 deletions(-) create mode 100644 internal/artifacts/current_state_commit.go create mode 100644 internal/artifacts/current_state_legacy.go create mode 100644 internal/artifacts/remote_commit.go create mode 100644 internal/artifacts/remote_commit_test.go diff --git a/docs/internal/command-restore.md b/docs/internal/command-restore.md index 2befcb4..a7d5078 100644 --- a/docs/internal/command-restore.md +++ b/docs/internal/command-restore.md @@ -65,7 +65,10 @@ Audio restore path: ## Invariants - restore uses committed remote current state as authority; -- `current/run_id.txt` is the remote publish commit marker; +- a verified `current/commit-pointer.json` and its selected immutable commit + establish new-protocol remote commitment; coherent legacy + `current/run_id.txt` plus `current/manifest.json` remains read-only migration + support; - restore does not execute pipeline stages. ## Implementation And Tests diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index c9f1239..0c48593 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -42,6 +42,25 @@ The model admits these stage states: - per-stage status - overall run status (`running`, `succeeded`, `failed`) +## Remote Commit Manifest + +`artifacts.RemoteCommitManifest` is a separate, versioned remote snapshot +contract. It is not a serialized session manifest and contains no local +post-publication assertion such as `current_pointer_written`. A remote commit +identifies one campaign, session, and run and declares its immutable artifact +set. Each artifact has a typed source, immutable destination key, SHA-256 +checksum, size, and storage generation. + +`current/commit-pointer.json` is the sole mutable selector for the new +contract. It identifies exactly one run-scoped `runs/{run_id}/commit.json` and +binds that object by checksum, size, and generation. Readers strictly reject +unknown fields, version mismatches, pointer/commit identity mismatches, and +objects that do not match their declaration. + +The reader retains a temporary, clearly isolated compatibility path for a +coherent legacy `current/manifest.json` plus `current/run_id.txt` pair. That +path is removable after migration and is never used to write new state. + ## Persistence Semantics `manifest.LocalStore`: @@ -112,11 +131,16 @@ where a durable running record can require operator interpretation. - stale stages retain prior details until replacement execution starts. - force reruns stale downstream succeeded stages. - run manifest does not replace session manifest as progress authority. +- remote commitment is established by a verified current pointer and remote + commit relationship, never by a mutable session-manifest boolean. ## Implementation And Tests - Models and transitions: `internal/manifest/manifest.go`, `internal/manifest/run_manifest.go` +- Remote commit model and readers: `internal/artifacts/remote_commit.go`, + `internal/artifacts/current_state_commit.go`, + `internal/artifacts/current_state_legacy.go` - Persistence and validation: `internal/manifest/store.go` - Package tests: `internal/manifest/*_test.go` - Assembled execution behavior: `internal/app/runner_test.go`, diff --git a/docs/operations.md b/docs/operations.md index 1d5be24..839a1d9 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -208,6 +208,20 @@ Publish commit model: `current/run_id.txt` is the remote current-state commit marker. +## Remote Commit Migration + +The immutable remote commit contract uses +`runs/{run_id}/commit.json` to declare a run's complete object set and a small +`current/commit-pointer.json` to select it. The pointer binds the selected +commit by version, checksum, size, and storage generation; committed artifacts +are also checksum- and generation-bound. Readers accept this contract now and +strictly reject mismatched or unknown data. + +The publish workflow above remains the legacy writer until its planned cutover. +Legacy reads are limited to a coherent `current/manifest.json` and +`current/run_id.txt` pair; a torn pair is rejected. New remote commit state does +not carry local `current_pointer_written` metadata. + ## Publish Locks Lock sources: diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index e087c70..8d15533 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -28,7 +28,7 @@ All stages are pending when this plan is created. | 10 | Confine publish archive reads | COR-005 | Completed | | 11 | Make manifest and run identity singular | COR-001, TST-006 | Completed | | 12 | Centralize handled terminal-failure persistence | RSK-001, TST-002, SIM-001, COM-001 | Completed | -| 13 | Introduce the immutable remote-commit model and legacy boundary | ARC-003 | Pending | +| 13 | Introduce the immutable remote-commit model and legacy boundary | ARC-003 | Completed | | 14 | Publish through immutable commits and canonical mappings | COR-004, COR-011, DUP-002, TST-004 | Pending | | 15 | Make remote locks generation-safe and harden pagination | RSK-005, RSK-014 | Pending | | 16 | Persist retryable post-commit cleanup state | COR-006, COR-007 | Pending | diff --git a/internal/app/restore_discovery.go b/internal/app/restore_discovery.go index b08d080..a45774d 100644 --- a/internal/app/restore_discovery.go +++ b/internal/app/restore_discovery.go @@ -39,8 +39,6 @@ func discoverRemoteCurrentState(ctx context.Context, cfg *config.Config, store s if err != nil { return nil, fmt.Errorf("resolve publish session prefix: %w", err) } - currentManifestKey, currentRunIDKey := artifacts.ResolveCurrentStateKeys(sessionPrefix) - requestedSession := strings.TrimSpace(cfg.Session.SessionID) requestedCampaign := strings.TrimSpace(cfg.Session.Campaign) current, err := artifacts.LoadCurrentState(ctx, store, sessionPrefix, artifacts.CurrentStateValidation{ @@ -54,8 +52,8 @@ func discoverRemoteCurrentState(ctx context.Context, cfg *config.Config, store s return &RemoteCurrentState{ Bucket: bucket, SessionPrefix: sessionPrefix, - CurrentRunIDKey: currentRunIDKey, - CurrentManifestKey: currentManifestKey, + CurrentRunIDKey: current.CurrentRunIDKey, + CurrentManifestKey: current.CurrentManifestKey, RunID: current.RunID, SessionID: strings.TrimSpace(current.Manifest.SessionID), Campaign: strings.TrimSpace(current.Manifest.Campaign), diff --git a/internal/artifacts/current_state.go b/internal/artifacts/current_state.go index 0fe0966..238112c 100644 --- a/internal/artifacts/current_state.go +++ b/internal/artifacts/current_state.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "os" "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" @@ -44,8 +43,11 @@ type CurrentState struct { SessionPrefix string CurrentRunIDKey string CurrentManifestKey string + CurrentPointerKey string RunID string Manifest *manifest.Manifest + Commit *RemoteCommitManifest + Pointer *CurrentCommitPointer } type CurrentStateValidation struct { @@ -55,74 +57,6 @@ type CurrentStateValidation struct { ValidateRunID bool } -func LoadCurrentRunPointer(ctx context.Context, store storage.ObjectStore, currentRunIDKey string) (string, error) { - if store == nil { - return "", fmt.Errorf("object store is required") - } - key := strings.TrimSpace(currentRunIDKey) - if key == "" { - return "", fmt.Errorf("current run pointer key is required") - } - - exists, err := store.Exists(ctx, key) - if err != nil { - return "", fmt.Errorf("check current run pointer %q: %w", key, err) - } - if !exists { - return "", &CurrentRunPointerMissingError{Key: key} - } - - localPath, err := storage.DownloadObjectToTemp(ctx, store, key, "narratio-current-run-id-*.txt") - if err != nil { - return "", fmt.Errorf("download current run pointer %q: %w", key, err) - } - defer func() { _ = os.Remove(localPath) }() - - data, err := os.ReadFile(localPath) - if err != nil { - return "", fmt.Errorf("read downloaded current run pointer %q: %w", key, err) - } - runID := strings.TrimSpace(string(data)) - if runID == "" { - return "", fmt.Errorf("current run pointer %q is empty", key) - } - if err := ValidateRunIdentity(runID); err != nil { - return "", fmt.Errorf("current run pointer %q contains an unsafe legacy run id; migrate remote state before use: %w", key, err) - } - return runID, nil -} - -func LoadCurrentManifest(ctx context.Context, store storage.ObjectStore, currentManifestKey string) (*manifest.Manifest, error) { - if store == nil { - return nil, fmt.Errorf("object store is required") - } - key := strings.TrimSpace(currentManifestKey) - if key == "" { - return nil, fmt.Errorf("current manifest key is required") - } - - exists, err := store.Exists(ctx, key) - if err != nil { - return nil, fmt.Errorf("check current manifest %q: %w", key, err) - } - if !exists { - return nil, &CurrentManifestMissingError{Key: key} - } - - localPath, err := storage.DownloadObjectToTemp(ctx, store, key, "narratio-current-manifest-*.json") - if err != nil { - return nil, fmt.Errorf("download current manifest %q: %w", key, err) - } - defer func() { _ = os.Remove(localPath) }() - - manifestStore := &manifest.LocalStore{} - m, err := manifestStore.Load(ctx, localPath) - if err != nil { - return nil, fmt.Errorf("current manifest decode failed: %w", err) - } - return m, nil -} - func LoadCurrentState( ctx context.Context, store storage.ObjectStore, @@ -133,27 +67,19 @@ func LoadCurrentState( if prefix == "" { return nil, fmt.Errorf("session prefix is required") } - currentManifestKey, currentRunIDKey := ResolveCurrentStateKeys(prefix) - runID, err := LoadCurrentRunPointer(ctx, store, currentRunIDKey) - if err != nil { - return nil, err - } - m, err := LoadCurrentManifest(ctx, store, currentManifestKey) - if err != nil { - return nil, err + if store == nil { + return nil, fmt.Errorf("object store is required") } - state := &CurrentState{ - SessionPrefix: prefix, - CurrentRunIDKey: currentRunIDKey, - CurrentManifestKey: currentManifestKey, - RunID: runID, - Manifest: m, + pointerKey := S3CurrentCommitPointerKey(prefix) + pointerExists, err := store.Exists(ctx, pointerKey) + if err != nil { + return nil, fmt.Errorf("check current commit pointer %q: %w", pointerKey, err) } - if err := ValidateCurrentStateIdentity(state, validation); err != nil { - return nil, err + if pointerExists { + return loadCommittedCurrentState(ctx, store, prefix, pointerKey, validation) } - return state, nil + return loadLegacyCurrentState(ctx, store, prefix, validation) } func ValidateCurrentStateIdentity(state *CurrentState, validation CurrentStateValidation) error { diff --git a/internal/artifacts/current_state_commit.go b/internal/artifacts/current_state_commit.go new file mode 100644 index 0000000..7a8dc7a --- /dev/null +++ b/internal/artifacts/current_state_commit.go @@ -0,0 +1,187 @@ +package artifacts + +import ( + "context" + "fmt" + "os" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func loadCommittedCurrentState( + ctx context.Context, + store storage.ObjectStore, + sessionPrefix string, + pointerKey string, + validation CurrentStateValidation, +) (*CurrentState, error) { + pointerData, err := downloadRemoteObject(ctx, store, pointerKey, "narratio-current-commit-pointer-*.json") + if err != nil { + return nil, fmt.Errorf("download current commit pointer %q: %w", pointerKey, err) + } + pointer, err := DecodeCurrentCommitPointer(pointerData) + if err != nil { + return nil, err + } + if err := pointer.ValidateForSessionPrefix(sessionPrefix); err != nil { + return nil, err + } + + commitData, err := readVerifiedRemoteObject(ctx, store, pointer.CommitKey, pointer.CommitSHA256, pointer.CommitSize, pointer.CommitGeneration, "narratio-remote-commit-*.json") + if err != nil { + return nil, fmt.Errorf("read selected remote commit %q: %w", pointer.CommitKey, err) + } + commit, err := DecodeRemoteCommitManifest(commitData) + if err != nil { + return nil, err + } + if err := commit.ValidateForSessionPrefix(sessionPrefix); err != nil { + return nil, err + } + if err := validatePointerCommitIdentity(pointer, commit); err != nil { + return nil, err + } + + sessionManifest, ok := commit.Artifact(RemoteArtifactTypeSessionManifest) + if !ok { + return nil, fmt.Errorf("remote commit must declare exactly one session manifest artifact") + } + manifestData, err := readVerifiedRemoteObject(ctx, store, sessionManifest.DestinationKey, sessionManifest.SHA256, sessionManifest.Size, sessionManifest.Generation, "narratio-remote-session-manifest-*.json") + if err != nil { + return nil, fmt.Errorf("read committed session manifest %q: %w", sessionManifest.DestinationKey, err) + } + m, err := decodeCommittedManifest(ctx, manifestData) + if err != nil { + return nil, err + } + if err := validateCommitManifestIdentity(commit, m); err != nil { + return nil, err + } + + state := &CurrentState{ + SessionPrefix: sessionPrefix, + CurrentRunIDKey: pointerKey, + CurrentPointerKey: pointerKey, + CurrentManifestKey: sessionManifest.DestinationKey, + RunID: commit.RunID, + Manifest: m, + Commit: commit, + Pointer: pointer, + } + if err := ValidateCurrentStateIdentity(state, validation); err != nil { + return nil, err + } + return state, nil +} + +func validatePointerCommitIdentity(pointer *CurrentCommitPointer, commit *RemoteCommitManifest) error { + if pointer == nil || commit == nil { + return fmt.Errorf("current commit pointer and remote commit are required") + } + if pointer.Campaign != commit.Campaign || pointer.SessionID != commit.SessionID || pointer.RunID != commit.RunID { + return fmt.Errorf("current commit pointer identity does not match selected remote commit") + } + return nil +} + +func validateCommitManifestIdentity(commit *RemoteCommitManifest, m *manifest.Manifest) error { + if commit == nil || m == nil { + return fmt.Errorf("remote commit and committed session manifest are required") + } + if commit.Campaign != strings.TrimSpace(m.Campaign) || + commit.SessionID != strings.TrimSpace(m.SessionID) || + commit.RunID != strings.TrimSpace(m.RunID) { + return fmt.Errorf("selected remote commit identity does not match committed session manifest") + } + return nil +} + +func readVerifiedRemoteObject( + ctx context.Context, + store storage.ObjectStore, + key string, + wantSHA256 string, + wantSize int64, + wantGeneration string, + tempPattern string, +) ([]byte, error) { + data, err := downloadRemoteObject(ctx, store, key, tempPattern) + if err != nil { + return nil, err + } + if int64(len(data)) != wantSize { + return nil, fmt.Errorf("size mismatch: got %d, want %d", len(data), wantSize) + } + if checksum := remoteObjectSHA256(data); checksum != wantSHA256 { + return nil, fmt.Errorf("checksum mismatch: got %s, want %s", checksum, wantSHA256) + } + info, err := remoteObjectInfo(ctx, store, key) + if err != nil { + return nil, err + } + if info.Size != wantSize { + return nil, fmt.Errorf("storage size mismatch: got %d, want %d", info.Size, wantSize) + } + if strings.TrimSpace(info.ETag) != wantGeneration { + return nil, fmt.Errorf("generation mismatch: got %q, want %q", info.ETag, wantGeneration) + } + return data, nil +} + +func downloadRemoteObject(ctx context.Context, store storage.ObjectStore, key, tempPattern string) ([]byte, error) { + localPath, err := storage.DownloadObjectToTemp(ctx, store, key, tempPattern) + if err != nil { + return nil, err + } + defer func() { _ = os.Remove(localPath) }() + data, err := os.ReadFile(localPath) + if err != nil { + return nil, fmt.Errorf("read downloaded object %q: %w", key, err) + } + return data, nil +} + +func remoteObjectInfo(ctx context.Context, store storage.ObjectStore, key string) (storage.ObjectInfo, error) { + objects, err := store.List(ctx, key) + if err != nil { + return storage.ObjectInfo{}, fmt.Errorf("list remote object %q: %w", key, err) + } + var found *storage.ObjectInfo + for _, object := range objects { + if object.Key != key { + continue + } + if found != nil { + return storage.ObjectInfo{}, fmt.Errorf("remote object %q is ambiguous", key) + } + copy := object + found = © + } + if found == nil { + return storage.ObjectInfo{}, fmt.Errorf("remote object %q is missing", key) + } + return *found, nil +} + +func decodeCommittedManifest(ctx context.Context, data []byte) (*manifest.Manifest, error) { + file, err := os.CreateTemp("", "narratio-committed-session-manifest-*.json") + if err != nil { + return nil, fmt.Errorf("create committed manifest file: %w", err) + } + path := file.Name() + defer func() { _ = os.Remove(path) }() + if _, err := file.Write(data); err != nil { + _ = file.Close() + return nil, fmt.Errorf("write committed manifest file: %w", err) + } + if err := file.Close(); err != nil { + return nil, fmt.Errorf("close committed manifest file: %w", err) + } + m, err := (&manifest.LocalStore{}).Load(ctx, path) + if err != nil { + return nil, fmt.Errorf("committed session manifest decode failed: %w", err) + } + return m, nil +} diff --git a/internal/artifacts/current_state_legacy.go b/internal/artifacts/current_state_legacy.go new file mode 100644 index 0000000..862c402 --- /dev/null +++ b/internal/artifacts/current_state_legacy.go @@ -0,0 +1,98 @@ +package artifacts + +import ( + "context" + "fmt" + "os" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +// This file is temporary compatibility support for sessions published before +// immutable current commits. It can be removed after legacy remote state is migrated. + +func loadLegacyCurrentState( + ctx context.Context, + store storage.ObjectStore, + sessionPrefix string, + validation CurrentStateValidation, +) (*CurrentState, error) { + currentManifestKey, currentRunIDKey := ResolveCurrentStateKeys(sessionPrefix) + runID, err := loadLegacyCurrentRunPointer(ctx, store, currentRunIDKey) + if err != nil { + return nil, err + } + m, err := loadLegacyCurrentManifest(ctx, store, currentManifestKey) + if err != nil { + return nil, err + } + + state := &CurrentState{ + SessionPrefix: sessionPrefix, + CurrentRunIDKey: currentRunIDKey, + CurrentManifestKey: currentManifestKey, + RunID: runID, + Manifest: m, + } + if err := ValidateCurrentStateIdentity(state, validation); err != nil { + return nil, err + } + return state, nil +} + +func loadLegacyCurrentRunPointer(ctx context.Context, store storage.ObjectStore, currentRunIDKey string) (string, error) { + key := strings.TrimSpace(currentRunIDKey) + if key == "" { + return "", fmt.Errorf("current run pointer key is required") + } + + exists, err := store.Exists(ctx, key) + if err != nil { + return "", fmt.Errorf("check current run pointer %q: %w", key, err) + } + if !exists { + return "", &CurrentRunPointerMissingError{Key: key} + } + + data, err := downloadRemoteObject(ctx, store, key, "narratio-legacy-current-run-id-*.txt") + if err != nil { + return "", fmt.Errorf("download current run pointer %q: %w", key, err) + } + runID := strings.TrimSpace(string(data)) + if runID == "" { + return "", fmt.Errorf("current run pointer %q is empty", key) + } + if err := ValidateRunIdentity(runID); err != nil { + return "", fmt.Errorf("current run pointer %q contains an unsafe legacy run id; migrate remote state before use: %w", key, err) + } + return runID, nil +} + +func loadLegacyCurrentManifest(ctx context.Context, store storage.ObjectStore, currentManifestKey string) (*manifest.Manifest, error) { + key := strings.TrimSpace(currentManifestKey) + if key == "" { + return nil, fmt.Errorf("current manifest key is required") + } + + exists, err := store.Exists(ctx, key) + if err != nil { + return nil, fmt.Errorf("check current manifest %q: %w", key, err) + } + if !exists { + return nil, &CurrentManifestMissingError{Key: key} + } + + localPath, err := storage.DownloadObjectToTemp(ctx, store, key, "narratio-legacy-current-manifest-*.json") + if err != nil { + return nil, fmt.Errorf("download current manifest %q: %w", key, err) + } + defer func() { _ = os.Remove(localPath) }() + + m, err := (&manifest.LocalStore{}).Load(ctx, localPath) + if err != nil { + return nil, fmt.Errorf("current manifest decode failed: %w", err) + } + return m, nil +} diff --git a/internal/artifacts/current_state_test.go b/internal/artifacts/current_state_test.go index 8eded76..920c231 100644 --- a/internal/artifacts/current_state_test.go +++ b/internal/artifacts/current_state_test.go @@ -104,6 +104,33 @@ func TestLoadCurrentStateRunIDMismatchFails(t *testing.T) { } } +func TestLoadCurrentStateReadsCoherentLegacyPair(t *testing.T) { + store := &storage.FakeBackend{} + seedCurrentState(t, store, "2026-05-03", "sample-campaign", "20260519T010203Z-a1b2c3d4") + + state, err := LoadCurrentState(context.Background(), store, testCurrentSessionPrefix(), CurrentStateValidation{ + ExpectedCampaign: "sample-campaign", + ExpectedSessionID: "2026-05-03", + ValidateRunID: true, + }) + if err != nil { + t.Fatalf("LoadCurrentState() error = %v", err) + } + if state.Commit != nil || state.Pointer != nil { + t.Fatalf("legacy current state unexpectedly includes immutable commit data: %#v", state) + } +} + +func TestLoadCurrentStateRejectsTornLegacyPair(t *testing.T) { + store := &storage.FakeBackend{} + seedCurrentState(t, store, "2026-05-03", "sample-campaign", "different-run-id") + + _, err := LoadCurrentState(context.Background(), store, testCurrentSessionPrefix(), CurrentStateValidation{ValidateRunID: true}) + if err == nil || !strings.Contains(err.Error(), "current manifest run_id") { + t.Fatalf("error = %v, want torn legacy pair rejection", err) + } +} + func testCurrentSessionPrefix() string { return S3SessionPrefix("dnd", "sample-campaign", "2026-05-03") } diff --git a/internal/artifacts/remote_commit.go b/internal/artifacts/remote_commit.go new file mode 100644 index 0000000..0c5951f --- /dev/null +++ b/internal/artifacts/remote_commit.go @@ -0,0 +1,264 @@ +package artifacts + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "path" + "strings" +) + +const RemoteCommitFormatVersion = 1 + +// RemoteArtifactType identifies the durable purpose of an object in a remote commit. +type RemoteArtifactType string + +const ( + RemoteArtifactTypeSessionManifest RemoteArtifactType = "session_manifest" + RemoteArtifactTypeRunManifest RemoteArtifactType = "run_manifest" + RemoteArtifactTypePublishedOutput RemoteArtifactType = "published_output" + RemoteArtifactTypePreviousArtifact RemoteArtifactType = "previous_artifact" +) + +// RemoteArtifact maps one typed source to an immutable remote destination. +type RemoteArtifact struct { + Type RemoteArtifactType `json:"type"` + Source string `json:"source"` + DestinationKey string `json:"destination_key"` + SHA256 string `json:"sha256"` + Size int64 `json:"size"` + Generation string `json:"generation"` +} + +// RemoteCommitManifest declares the complete immutable object set for one run. +type RemoteCommitManifest struct { + FormatVersion int `json:"format_version"` + Campaign string `json:"campaign"` + SessionID string `json:"session_id"` + RunID string `json:"run_id"` + Artifacts []RemoteArtifact `json:"artifacts"` +} + +// CurrentCommitPointer is the sole mutable selector for a remote commit manifest. +type CurrentCommitPointer struct { + FormatVersion int `json:"format_version"` + Campaign string `json:"campaign"` + SessionID string `json:"session_id"` + RunID string `json:"run_id"` + CommitKey string `json:"commit_key"` + CommitSHA256 string `json:"commit_sha256"` + CommitSize int64 `json:"commit_size"` + CommitGeneration string `json:"commit_generation"` +} + +func (m RemoteCommitManifest) Validate() error { + if m.FormatVersion != RemoteCommitFormatVersion { + return fmt.Errorf("unsupported remote commit format version %d", m.FormatVersion) + } + if err := ValidateSessionIdentity(m.Campaign, m.SessionID); err != nil { + return fmt.Errorf("remote commit identity: %w", err) + } + if err := ValidateRunIdentity(m.RunID); err != nil { + return fmt.Errorf("remote commit run identity: %w", err) + } + if len(m.Artifacts) == 0 { + return fmt.Errorf("remote commit artifacts are required") + } + + destinations := make(map[string]struct{}, len(m.Artifacts)) + for index, artifact := range m.Artifacts { + if err := artifact.Validate(); err != nil { + return fmt.Errorf("remote commit artifact %d: %w", index, err) + } + if _, exists := destinations[artifact.DestinationKey]; exists { + return fmt.Errorf("remote commit declares duplicate destination %q", artifact.DestinationKey) + } + destinations[artifact.DestinationKey] = struct{}{} + } + return nil +} + +func (m RemoteCommitManifest) ValidateForSessionPrefix(sessionPrefix string) error { + if err := m.Validate(); err != nil { + return err + } + runPrefix := S3RunPrefix(sessionPrefix, m.RunID) + if runPrefix == "" { + return fmt.Errorf("remote commit run prefix is required") + } + for _, artifact := range m.Artifacts { + if !strings.HasPrefix(artifact.DestinationKey, runPrefix) { + return fmt.Errorf("remote commit artifact destination %q is outside immutable run prefix %q", artifact.DestinationKey, runPrefix) + } + } + return nil +} + +func (a RemoteArtifact) Validate() error { + switch a.Type { + case RemoteArtifactTypeSessionManifest, RemoteArtifactTypeRunManifest, RemoteArtifactTypePublishedOutput, RemoteArtifactTypePreviousArtifact: + default: + return fmt.Errorf("unsupported artifact type %q", a.Type) + } + if strings.TrimSpace(a.Source) == "" { + return fmt.Errorf("source is required") + } + if err := validateRemoteObjectKey(a.DestinationKey); err != nil { + return fmt.Errorf("destination key: %w", err) + } + if err := validateSHA256(a.SHA256); err != nil { + return fmt.Errorf("sha256: %w", err) + } + if a.Size < 0 { + return fmt.Errorf("size must not be negative") + } + if strings.TrimSpace(a.Generation) == "" { + return fmt.Errorf("generation is required") + } + return nil +} + +func (p CurrentCommitPointer) Validate() error { + if p.FormatVersion != RemoteCommitFormatVersion { + return fmt.Errorf("unsupported current commit pointer format version %d", p.FormatVersion) + } + if err := ValidateSessionIdentity(p.Campaign, p.SessionID); err != nil { + return fmt.Errorf("current commit pointer identity: %w", err) + } + if err := ValidateRunIdentity(p.RunID); err != nil { + return fmt.Errorf("current commit pointer run identity: %w", err) + } + if err := validateRemoteObjectKey(p.CommitKey); err != nil { + return fmt.Errorf("current commit pointer key: %w", err) + } + if err := validateSHA256(p.CommitSHA256); err != nil { + return fmt.Errorf("current commit pointer checksum: %w", err) + } + if p.CommitSize < 0 { + return fmt.Errorf("current commit pointer size must not be negative") + } + if strings.TrimSpace(p.CommitGeneration) == "" { + return fmt.Errorf("current commit pointer generation is required") + } + return nil +} + +func (p CurrentCommitPointer) ValidateForSessionPrefix(sessionPrefix string) error { + if err := p.Validate(); err != nil { + return err + } + expectedKey := S3RunCommitKey(sessionPrefix, p.RunID) + if p.CommitKey != expectedKey { + return fmt.Errorf("current commit pointer key %q does not match canonical commit key %q", p.CommitKey, expectedKey) + } + return nil +} + +// EncodeRemoteCommitManifest validates and serializes an immutable commit manifest. +func EncodeRemoteCommitManifest(m RemoteCommitManifest) ([]byte, error) { + if err := m.Validate(); err != nil { + return nil, err + } + data, err := json.Marshal(m) + if err != nil { + return nil, fmt.Errorf("encode remote commit manifest: %w", err) + } + return append(data, '\n'), nil +} + +// DecodeRemoteCommitManifest strictly decodes a remote commit manifest. +func DecodeRemoteCommitManifest(data []byte) (*RemoteCommitManifest, error) { + var commit RemoteCommitManifest + if err := decodeStrictJSON(data, &commit); err != nil { + return nil, fmt.Errorf("decode remote commit manifest: %w", err) + } + if err := commit.Validate(); err != nil { + return nil, err + } + return &commit, nil +} + +// EncodeCurrentCommitPointer validates and serializes a current commit pointer. +func EncodeCurrentCommitPointer(p CurrentCommitPointer) ([]byte, error) { + if err := p.Validate(); err != nil { + return nil, err + } + data, err := json.Marshal(p) + if err != nil { + return nil, fmt.Errorf("encode current commit pointer: %w", err) + } + return append(data, '\n'), nil +} + +// DecodeCurrentCommitPointer strictly decodes a current commit pointer. +func DecodeCurrentCommitPointer(data []byte) (*CurrentCommitPointer, error) { + var pointer CurrentCommitPointer + if err := decodeStrictJSON(data, &pointer); err != nil { + return nil, fmt.Errorf("decode current commit pointer: %w", err) + } + if err := pointer.Validate(); err != nil { + return nil, err + } + return &pointer, nil +} + +func (m RemoteCommitManifest) Artifact(artifactType RemoteArtifactType) (RemoteArtifact, bool) { + var found RemoteArtifact + for _, artifact := range m.Artifacts { + if artifact.Type != artifactType { + continue + } + if found.DestinationKey != "" { + return RemoteArtifact{}, false + } + found = artifact + } + return found, found.DestinationKey != "" +} + +func remoteObjectSHA256(data []byte) string { + sum := sha256.Sum256(data) + return hex.EncodeToString(sum[:]) +} + +func validateSHA256(value string) error { + if len(value) != sha256.Size*2 || strings.ToLower(value) != value { + return fmt.Errorf("must be a lowercase SHA-256 digest") + } + if _, err := hex.DecodeString(value); err != nil { + return fmt.Errorf("must be hexadecimal: %w", err) + } + return nil +} + +func validateRemoteObjectKey(value string) error { + if strings.TrimSpace(value) == "" { + return fmt.Errorf("is required") + } + if strings.TrimSpace(value) != value || strings.HasPrefix(value, "/") || strings.Contains(value, `\\`) { + return fmt.Errorf("must be a clean relative object key") + } + if cleaned := path.Clean(value); cleaned != value || cleaned == "." || cleaned == ".." || strings.HasPrefix(cleaned, "../") { + return fmt.Errorf("must be a clean relative object key") + } + return nil +} + +func decodeStrictJSON(data []byte, destination any) error { + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(destination); err != nil { + return err + } + var trailing any + if err := decoder.Decode(&trailing); err != io.EOF { + if err == nil { + return fmt.Errorf("multiple JSON values are not allowed") + } + return err + } + return nil +} diff --git a/internal/artifacts/remote_commit_test.go b/internal/artifacts/remote_commit_test.go new file mode 100644 index 0000000..9f054a3 --- /dev/null +++ b/internal/artifacts/remote_commit_test.go @@ -0,0 +1,253 @@ +package artifacts + +import ( + "bytes" + "context" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" +) + +func TestRemoteCommitManifestRejectsUnsupportedVersion(t *testing.T) { + commit := testRemoteCommit(t, "20260519T010203Z-a1b2c3d4") + pointer := testCommitPointer(t, testCurrentSessionPrefix(), commit) + commit.FormatVersion++ + if _, err := EncodeRemoteCommitManifest(commit); err == nil || !strings.Contains(err.Error(), "unsupported remote commit format version") { + t.Fatalf("EncodeRemoteCommitManifest() error = %v, want unsupported version", err) + } + + pointer.FormatVersion++ + if _, err := EncodeCurrentCommitPointer(pointer); err == nil || !strings.Contains(err.Error(), "unsupported current commit pointer format version") { + t.Fatalf("EncodeCurrentCommitPointer() error = %v, want unsupported version", err) + } +} + +func TestRemoteCommitDecodersRejectUnknownFields(t *testing.T) { + commit := testRemoteCommit(t, "20260519T010203Z-a1b2c3d4") + commitData, err := EncodeRemoteCommitManifest(commit) + if err != nil { + t.Fatalf("EncodeRemoteCommitManifest() error = %v", err) + } + commitData = appendUnknownJSONField(t, commitData) + if _, err := DecodeRemoteCommitManifest(commitData); err == nil || !strings.Contains(err.Error(), "unknown field") { + t.Fatalf("DecodeRemoteCommitManifest() error = %v, want unknown-field rejection", err) + } + + pointer := testCommitPointer(t, testCurrentSessionPrefix(), commit) + pointerData, err := EncodeCurrentCommitPointer(pointer) + if err != nil { + t.Fatalf("EncodeCurrentCommitPointer() error = %v", err) + } + pointerData = appendUnknownJSONField(t, pointerData) + if _, err := DecodeCurrentCommitPointer(pointerData); err == nil || !strings.Contains(err.Error(), "unknown field") { + t.Fatalf("DecodeCurrentCommitPointer() error = %v, want unknown-field rejection", err) + } +} + +func TestLoadCurrentStateReadsVerifiedRemoteCommit(t *testing.T) { + store := &storage.FakeBackend{} + commit := testRemoteCommit(t, "20260519T010203Z-a1b2c3d4") + seedRemoteCommit(t, store, testCurrentSessionPrefix(), commit, "commit-generation", "manifest-generation") + + state, err := LoadCurrentState(context.Background(), store, testCurrentSessionPrefix(), CurrentStateValidation{ + ExpectedCampaign: commit.Campaign, + ExpectedSessionID: commit.SessionID, + ValidateRunID: true, + }) + if err != nil { + t.Fatalf("LoadCurrentState() error = %v", err) + } + if state.Commit == nil || state.Pointer == nil { + t.Fatalf("state does not expose selected immutable commit: %#v", state) + } + if state.CurrentPointerKey != S3CurrentCommitPointerKey(testCurrentSessionPrefix()) { + t.Fatalf("CurrentPointerKey = %q", state.CurrentPointerKey) + } + if state.Manifest.RunID != commit.RunID { + t.Fatalf("manifest run ID = %q, want %q", state.Manifest.RunID, commit.RunID) + } +} + +func TestLoadCurrentStateRejectsPointerCommitIdentityMismatch(t *testing.T) { + store := &storage.FakeBackend{} + prefix := testCurrentSessionPrefix() + pointerRunID := "20260519T010203Z-a1b2c3d4" + commit := testRemoteCommit(t, "20260520T010203Z-a1b2c3d4") + commitData, err := EncodeRemoteCommitManifest(commit) + if err != nil { + t.Fatalf("EncodeRemoteCommitManifest() error = %v", err) + } + store.SeedObject(storage.FakeObject{Key: S3RunCommitKey(prefix, pointerRunID), Data: commitData, ETag: "commit-generation"}) + pointer := CurrentCommitPointer{ + FormatVersion: RemoteCommitFormatVersion, + Campaign: commit.Campaign, + SessionID: commit.SessionID, + RunID: pointerRunID, + CommitKey: S3RunCommitKey(prefix, pointerRunID), + CommitSHA256: remoteObjectSHA256(commitData), + CommitSize: int64(len(commitData)), + CommitGeneration: "commit-generation", + } + pointerData, err := EncodeCurrentCommitPointer(pointer) + if err != nil { + t.Fatalf("EncodeCurrentCommitPointer() error = %v", err) + } + store.SeedObject(storage.FakeObject{Key: S3CurrentCommitPointerKey(prefix), Data: pointerData}) + + _, err = LoadCurrentState(context.Background(), store, prefix, CurrentStateValidation{}) + if err == nil || !strings.Contains(err.Error(), "does not match selected remote commit") { + t.Fatalf("error = %v, want pointer/commit identity mismatch", err) + } +} + +func TestLoadCurrentStateRejectsCommitManifestIdentityMismatch(t *testing.T) { + store := &storage.FakeBackend{} + commit := testRemoteCommit(t, "20260519T010203Z-a1b2c3d4") + pointer := seedRemoteCommit(t, store, testCurrentSessionPrefix(), commit, "commit-generation", "manifest-generation") + + wrongManifest := testManifestJSON(t, "2026-05-03", "wrong-campaign", commit.RunID) + store.SeedObject(storage.FakeObject{ + Key: commit.Artifacts[0].DestinationKey, + Data: wrongManifest, + ETag: "manifest-generation", + }) + commit.Artifacts[0].SHA256 = remoteObjectSHA256(wrongManifest) + commit.Artifacts[0].Size = int64(len(wrongManifest)) + commitData, err := EncodeRemoteCommitManifest(commit) + if err != nil { + t.Fatalf("EncodeRemoteCommitManifest() error = %v", err) + } + store.SeedObject(storage.FakeObject{Key: pointer.CommitKey, Data: commitData, ETag: "commit-generation"}) + pointer.CommitSHA256 = remoteObjectSHA256(commitData) + pointer.CommitSize = int64(len(commitData)) + pointerData, err := EncodeCurrentCommitPointer(*pointer) + if err != nil { + t.Fatalf("EncodeCurrentCommitPointer() error = %v", err) + } + store.SeedObject(storage.FakeObject{Key: S3CurrentCommitPointerKey(testCurrentSessionPrefix()), Data: pointerData}) + + _, err = LoadCurrentState(context.Background(), store, testCurrentSessionPrefix(), CurrentStateValidation{}) + if err == nil || !strings.Contains(err.Error(), "does not match committed session manifest") { + t.Fatalf("error = %v, want commit/manifest identity mismatch", err) + } +} + +func TestLoadCurrentStateRejectsCommitChecksumOrGenerationMismatch(t *testing.T) { + tests := []struct { + name string + mutate func(*CurrentCommitPointer) + want string + }{ + { + name: "checksum", + mutate: func(pointer *CurrentCommitPointer) { + pointer.CommitSHA256 = strings.Repeat("0", 64) + }, + want: "checksum mismatch", + }, + { + name: "generation", + mutate: func(pointer *CurrentCommitPointer) { + pointer.CommitGeneration = "stale-generation" + }, + want: "generation mismatch", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + store := &storage.FakeBackend{} + commit := testRemoteCommit(t, "20260519T010203Z-a1b2c3d4") + pointer := seedRemoteCommit(t, store, testCurrentSessionPrefix(), commit, "commit-generation", "manifest-generation") + test.mutate(pointer) + pointerData, err := EncodeCurrentCommitPointer(*pointer) + if err != nil { + t.Fatalf("EncodeCurrentCommitPointer() error = %v", err) + } + store.SeedObject(storage.FakeObject{Key: S3CurrentCommitPointerKey(testCurrentSessionPrefix()), Data: pointerData}) + + _, err = LoadCurrentState(context.Background(), store, testCurrentSessionPrefix(), CurrentStateValidation{}) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("error = %v, want %q", err, test.want) + } + }) + } +} + +func testRemoteCommit(t *testing.T, runID string) RemoteCommitManifest { + t.Helper() + prefix := testCurrentSessionPrefix() + manifestData := testManifestJSON(t, "2026-05-03", "sample-campaign", runID) + return RemoteCommitManifest{ + FormatVersion: RemoteCommitFormatVersion, + Campaign: "sample-campaign", + SessionID: "2026-05-03", + RunID: runID, + Artifacts: []RemoteArtifact{{ + Type: RemoteArtifactTypeSessionManifest, + Source: "session.manifest", + DestinationKey: S3RunRelativeDestinationKey(S3RunPrefix(prefix, runID), "session-manifest.json"), + SHA256: remoteObjectSHA256(manifestData), + Size: int64(len(manifestData)), + Generation: "manifest-generation", + }}, + } +} + +func testCommitPointer(t *testing.T, prefix string, commit RemoteCommitManifest) CurrentCommitPointer { + t.Helper() + commitData, err := EncodeRemoteCommitManifest(commit) + if err != nil { + t.Fatalf("EncodeRemoteCommitManifest() error = %v", err) + } + return CurrentCommitPointer{ + FormatVersion: RemoteCommitFormatVersion, + Campaign: commit.Campaign, + SessionID: commit.SessionID, + RunID: commit.RunID, + CommitKey: S3RunCommitKey(prefix, commit.RunID), + CommitSHA256: remoteObjectSHA256(commitData), + CommitSize: int64(len(commitData)), + CommitGeneration: "commit-generation", + } +} + +func seedRemoteCommit(t *testing.T, store *storage.FakeBackend, prefix string, commit RemoteCommitManifest, commitGeneration, manifestGeneration string) *CurrentCommitPointer { + t.Helper() + manifestData := testManifestJSON(t, commit.SessionID, commit.Campaign, commit.RunID) + commit.Artifacts[0].SHA256 = remoteObjectSHA256(manifestData) + commit.Artifacts[0].Size = int64(len(manifestData)) + commit.Artifacts[0].Generation = manifestGeneration + commitData, err := EncodeRemoteCommitManifest(commit) + if err != nil { + t.Fatalf("EncodeRemoteCommitManifest() error = %v", err) + } + store.SeedObject(storage.FakeObject{ + Key: S3RunCommitKey(prefix, commit.RunID), + Data: commitData, + ETag: commitGeneration, + }) + store.SeedObject(storage.FakeObject{ + Key: commit.Artifacts[0].DestinationKey, + Data: manifestData, + ETag: manifestGeneration, + }) + pointer := testCommitPointer(t, prefix, commit) + pointer.CommitGeneration = commitGeneration + pointerData, err := EncodeCurrentCommitPointer(pointer) + if err != nil { + t.Fatalf("EncodeCurrentCommitPointer() error = %v", err) + } + store.SeedObject(storage.FakeObject{Key: S3CurrentCommitPointerKey(prefix), Data: pointerData}) + return &pointer +} + +func appendUnknownJSONField(t *testing.T, data []byte) []byte { + t.Helper() + trimmed := bytes.TrimSpace(data) + if len(trimmed) < 2 || trimmed[len(trimmed)-1] != '}' { + t.Fatalf("invalid JSON object %q", trimmed) + } + return append(append([]byte(nil), trimmed[:len(trimmed)-1]...), []byte(`,"unexpected":true}`)...) +} diff --git a/internal/artifacts/s3_keys.go b/internal/artifacts/s3_keys.go index 4af5733..ef5454f 100644 --- a/internal/artifacts/s3_keys.go +++ b/internal/artifacts/s3_keys.go @@ -58,6 +58,18 @@ func S3CurrentRunPointerKey(sessionPrefix string) string { return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), config.S3CurrentSegment, config.S3RunIDFile) } +// S3RunCommitKey returns the immutable commit manifest key for one run. +// Format: {session_prefix}/runs/{run_id}/commit.json +func S3RunCommitKey(sessionPrefix, runID string) string { + return path.Join(strings.TrimSuffix(S3RunPrefix(sessionPrefix, runID), "/"), config.S3CommitFile) +} + +// S3CurrentCommitPointerKey returns the sole mutable selector for a committed run. +// Format: {session_prefix}/current/commit-pointer.json +func S3CurrentCommitPointerKey(sessionPrefix string) string { + return path.Join(strings.TrimSuffix(cleanS3Key(sessionPrefix), "/"), config.S3CurrentSegment, config.S3CommitPointerFile) +} + // S3PublishedOutputKey returns the destination key for one published output. // Format: {session_prefix}/{output.dest} func S3PublishedOutputKey(sessionPrefix, to string) string { diff --git a/internal/artifacts/s3_keys_test.go b/internal/artifacts/s3_keys_test.go index 836d149..4418aef 100644 --- a/internal/artifacts/s3_keys_test.go +++ b/internal/artifacts/s3_keys_test.go @@ -32,6 +32,9 @@ func TestS3KeyConstruction(t *testing.T) { if runPrefix != wantRunPrefix { t.Fatalf("runPrefix = %q, want %q", runPrefix, wantRunPrefix) } + if got, want := S3RunCommitKey(sessionPrefix, runID), wantRunPrefix+"commit.json"; got != want { + t.Fatalf("commit key = %q, want %q", got, want) + } runPointer := S3CurrentRunPointerKey(sessionPrefix) if runPointer != "dnd/campaigns/forsaken/sessions/2026-04-19/current/run_id.txt" { @@ -43,6 +46,11 @@ func TestS3KeyConstruction(t *testing.T) { t.Fatalf("manifest key = %q", manifestKey) } + commitPointer := S3CurrentCommitPointerKey(sessionPrefix) + if commitPointer != "dnd/campaigns/forsaken/sessions/2026-04-19/current/commit-pointer.json" { + t.Fatalf("commit pointer key = %q", commitPointer) + } + publishedKey := S3PublishedOutputKey(sessionPrefix, "transcripts/final.trimmed.json") if publishedKey != "dnd/campaigns/forsaken/sessions/2026-04-19/transcripts/final.trimmed.json" { t.Fatalf("published key = %q", publishedKey) diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 6b6c321..f0123cb 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -83,12 +83,14 @@ const ( PathTranscriptFinal = artifactmodel.TranscriptPathFinal PathTranscriptFinalTrimmed = artifactmodel.TranscriptPathFinalTrimmed - S3CampaignsSegment = "campaigns" - S3SessionsSegment = "sessions" - S3RunsSegment = "runs" - S3CurrentSegment = "current" - S3ManifestFile = "manifest.json" - S3RunIDFile = "run_id.txt" + S3CampaignsSegment = "campaigns" + S3SessionsSegment = "sessions" + S3RunsSegment = "runs" + S3CurrentSegment = "current" + S3ManifestFile = "manifest.json" + S3RunIDFile = "run_id.txt" + S3CommitFile = "commit.json" + S3CommitPointerFile = "commit-pointer.json" ) // DefaultPublishOutputs defines the default publish output rules.