Require committed remote current state

This commit is contained in:
2026-09-02 13:47:20 +00:00
parent 82d75e7beb
commit 6c7a45d8d0
27 changed files with 218 additions and 662 deletions

View File

@@ -2,6 +2,8 @@ package stage
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"io"
"os"
@@ -116,6 +118,7 @@ func TestHydratePreviousSessionArtifactsFailsMissingRequiredArtifact(t *testing.
}
seedPreviousCurrentState(t, env, fake, previousStateSeedOptions{
includeRunPointerObject: true,
includeManifestObject: true,
includeArtifactObject: false,
})
@@ -175,8 +178,8 @@ func TestHydratePreviousSessionArtifactsRespectsCurrentCommitMarker(t *testing.T
})
_, err := hydratePreviousSessionArtifacts(context.Background(), env, sessionPaths, requirements)
if err == nil || !strings.Contains(err.Error(), "remote current run pointer missing") {
t.Fatalf("error = %v, want current run pointer missing failure", err)
if err == nil || !strings.Contains(err.Error(), "remote current commit pointer missing") {
t.Fatalf("error = %v, want current commit pointer missing failure", err)
}
}
@@ -218,9 +221,6 @@ func TestHydratePreviousSessionArtifactsUsesExplicitStorageKeys(t *testing.T) {
t.Fatalf("hydratePreviousSessionArtifacts() error = %v", err)
}
if !containsString(capture.existsKeys, seed.ArtifactKey) {
t.Fatalf("exists keys = %#v, want %q", capture.existsKeys, seed.ArtifactKey)
}
if !containsString(capture.accessKeys, seed.RunPointerKey) {
t.Fatalf("access keys = %#v, want %q", capture.accessKeys, seed.RunPointerKey)
}
@@ -268,7 +268,8 @@ func seedPreviousCurrentState(
campaign := strings.TrimSpace(env.Config.Session.Campaign)
rootPrefix := strings.TrimSpace(env.Config.Pipeline.Storage.S3.RootPrefix)
previousSessionPrefix := artifacts.S3SessionPrefix(rootPrefix, campaign, previousSessionID)
manifestKey, runPointerKey := artifacts.ResolveCurrentStateKeys(previousSessionPrefix)
manifestKey := artifacts.S3RunSessionManifestKey(previousSessionPrefix, "20260510T010203Z-a1b2c3d4")
runPointerKey := artifacts.S3CurrentCommitPointerKey(previousSessionPrefix)
previousRunID := "20260510T010203Z-a1b2c3d4"
previousSessionRoot := filepath.Join(t.TempDir(), "work", campaign, previousSessionID)
@@ -282,12 +283,6 @@ func seedPreviousCurrentState(
artifactLocalPath,
)
if options.includeRunPointerObject || (!options.includeRunPointerObject && !options.includeManifestObject && !options.includeArtifactObject) {
fake.SeedObject(storage.FakeObject{
Key: runPointerKey,
Data: []byte(previousRunID + "\n"),
})
}
if options.includeManifestObject || options.includeArtifactObject {
fake.SeedObject(storage.FakeObject{
Key: manifestKey,
@@ -295,7 +290,7 @@ func seedPreviousCurrentState(
})
}
artifactKey := artifacts.S3PublishedOutputKey(previousSessionPrefix, "artifacts/session_recap.md")
artifactKey := artifacts.S3RunRelativeDestinationKey(artifacts.S3RunPrefix(previousSessionPrefix, previousRunID), "artifacts/session_recap.md")
if options.includeArtifactObject {
body := options.artifactBody
if body == "" {
@@ -306,6 +301,27 @@ func seedPreviousCurrentState(
Data: []byte(body),
})
}
if options.includeRunPointerObject || (!options.includeRunPointerObject && !options.includeManifestObject && !options.includeArtifactObject) {
manifestObject := fake.Objects[manifestKey]
entries := []artifacts.RemoteArtifact{{Type: artifacts.RemoteArtifactTypeSessionManifest, Source: "session.manifest", DestinationKey: manifestKey, SHA256: preparePreviousSHA256(previousManifest), Size: int64(len(previousManifest)), Generation: manifestObject.ETag}}
if options.includeArtifactObject {
artifactObject := fake.Objects[artifactKey]
entries = append(entries, artifacts.RemoteArtifact{Type: artifacts.RemoteArtifactTypePublishedOutput, Source: artifacts.ConfiguredArtifactSourceID("session_recap"), DestinationKey: artifactKey, SHA256: preparePreviousSHA256(artifactObject.Data), Size: int64(len(artifactObject.Data)), Generation: artifactObject.ETag})
}
commit := artifacts.RemoteCommitManifest{FormatVersion: artifacts.RemoteCommitFormatVersion, Campaign: campaign, SessionID: previousSessionID, RunID: previousRunID, Artifacts: entries}
commitData, err := artifacts.EncodeRemoteCommitManifest(commit)
if err != nil {
t.Fatalf("encode previous commit: %v", err)
}
commitKey := artifacts.S3RunCommitKey(previousSessionPrefix, previousRunID)
fake.SeedObject(storage.FakeObject{Key: commitKey, Data: commitData})
commitObject := fake.Objects[commitKey]
pointerData, err := artifacts.EncodeCurrentCommitPointer(artifacts.CurrentCommitPointer{FormatVersion: artifacts.RemoteCommitFormatVersion, Campaign: campaign, SessionID: previousSessionID, RunID: previousRunID, CommitKey: commitKey, CommitSHA256: preparePreviousSHA256(commitData), CommitSize: int64(len(commitData)), CommitGeneration: commitObject.ETag})
if err != nil {
t.Fatalf("encode previous pointer: %v", err)
}
fake.SeedObject(storage.FakeObject{Key: runPointerKey, Data: pointerData})
}
return previousStateSeedResult{
PreviousSessionPrefix: previousSessionPrefix,
@@ -316,6 +332,11 @@ func seedPreviousCurrentState(
}
}
func preparePreviousSHA256(data []byte) string {
sum := sha256.Sum256(data)
return hex.EncodeToString(sum[:])
}
func previousHydrationFixture(t *testing.T) (*Env, artifacts.SessionPaths, *storage.FakeBackend) {
t.Helper()
env, m := setupPrepareEnv(t)

View File

@@ -790,7 +790,7 @@ func TestPrepareStageRerunOverwritesPreviousCache(t *testing.T) {
fake := &storage.FakeBackend{}
env.ObjectStore = fake
seed := seedPreviousCurrentState(t, env, fake, previousStateSeedOptions{
seedPreviousCurrentState(t, env, fake, previousStateSeedOptions{
includeRunPointerObject: true,
includeArtifactObject: true,
artifactBody: "# old recap\n",
@@ -810,7 +810,11 @@ func TestPrepareStageRerunOverwritesPreviousCache(t *testing.T) {
}
writeFile(t, filepath.Join(paths.PreviousArtifactsDir, "stale.txt"), "stale")
fake.SeedObject(storage.FakeObject{Key: seed.ArtifactKey, Data: []byte("# new recap\n")})
seedPreviousCurrentState(t, env, fake, previousStateSeedOptions{
includeRunPointerObject: true,
includeArtifactObject: true,
artifactBody: "# new recap\n",
})
if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil {
t.Fatalf("second prepare run error = %v", err)