eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move chunks into the canonical source model

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-17 05:32:11 +00:00
parent 40709e4ad8
commit 075888c97f
33 changed files with 548 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
internal/framework/checkpoint/loader.go
View File

@@ -78,7 +78,11 @@ func (l *WorkspaceLoader) Chunk(moduleKey string, sourceDigest string) (pipeline
if len(chunks) == 0 {
return pipeline.ChunkCheckpoint{}, invalidDecision("chunk checkpoint payload has no chunks")
}
if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), chunkOutputDigests(chunks)) {
outputDigests, err := chunkOutputDigests(chunks)
if err != nil {
return pipeline.ChunkCheckpoint{}, invalidDecision("chunk checkpoint output cannot be digested: %v", err)
}
if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), outputDigests) {
return pipeline.ChunkCheckpoint{}, invalidDecision("chunk checkpoint output digests do not match payload")
}
return pipeline.ChunkCheckpoint{Chunks: chunks, Warnings: cloneWarnings(payload.Warnings)}, reusedDecision()
@@ -180,6 +184,9 @@ func (l *WorkspaceLoader) validateManifest(manifest coreworkspace.StageManifest,
}
func (l *WorkspaceLoader) validateLaneManifest(manifest coreworkspace.StageManifest, stage coreworkspace.StageName, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, statuses ...coreworkspace.StageStatus) pipeline.CheckpointDecision {
if manifest.WorkspaceSchemaVersion == coreworkspace.WorkspaceSchemaVersionV1 {
return invalidDecision("checkpoint workspace schema version %q is incompatible with %q and must be recomputed", manifest.WorkspaceSchemaVersion, coreworkspace.WorkspaceSchemaVersion)
}
if manifest.WorkspaceSchemaVersion != coreworkspace.WorkspaceSchemaVersion {
return invalidDecision("checkpoint workspace schema version %q is not supported", manifest.WorkspaceSchemaVersion)
}
@@ -211,26 +218,25 @@ func (l *WorkspaceLoader) validateLaneManifest(manifest coreworkspace.StageManif
return reusedDecision()
}
func sourceChunksFromEnvelope(values []chunkEnvelope) ([]contracts.SourceChunk, error) {
func sourceChunksFromEnvelope(values []chunkEnvelope) ([]source.Chunk, error) {
if len(values) == 0 {
return nil, nil
}
out := make([]contracts.SourceChunk, 0, len(values))
out := make([]source.Chunk, 0, len(values))
for _, value := range values {
content, err := contentFromEnvelope(value.Content)
if err != nil {
return nil, err
}
out = append(out, contracts.SourceChunk{
ID: value.ID,
SourceID: value.SourceID,
Index: value.Index,
StartUnitID: value.StartUnitID,
EndUnitID: value.EndUnitID,
Content: content,
MediaType: value.Content.MediaType,
Units: cloneSourceUnits(value.Units),
Metadata: cloneMetadata(value.Metadata),
out = append(out, source.Chunk{
ID: value.ID,
SourceID: value.SourceID,
Index: value.Index,
Ref: value.Ref,
Content: content,
MediaType: value.Content.MediaType,
Units: cloneSourceUnits(value.Units),
Metadata: cloneMetadata(value.Metadata),
})
}
return out, nil

50
internal/framework/checkpoint/recorder.go
View File

@@ -73,7 +73,11 @@ func (r *WorkspaceRecorder) ChunkRunning(moduleKey string, sourceDigest string)
return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{StageManifest: manifest})
}
func (r *WorkspaceRecorder) ChunkSucceeded(moduleKey string, sourceDigest string, chunks []contracts.SourceChunk, warnings []contracts.Warning) error {
func (r *WorkspaceRecorder) ChunkSucceeded(moduleKey string, sourceDigest string, chunks []source.Chunk, warnings []contracts.Warning) error {
outputDigests, err := chunkOutputDigests(chunks)
if err != nil {
return fmt.Errorf("digest chunk checkpoint output: %w", err)
}
payload := chunksEnvelope{Chunks: chunkEnvelopes(chunks), Warnings: cloneWarnings(warnings)}
if err := r.writePayload("chunk/chunks.json", payload); err != nil {
return err
@@ -81,7 +85,7 @@ func (r *WorkspaceRecorder) ChunkSucceeded(moduleKey string, sourceDigest string
manifest := r.newStageManifest(coreworkspace.StageChunk, coreworkspace.StatusSucceeded)
manifest.ModuleKey = moduleKey
manifest.DependencyFingerprints = workspaceFingerprints(digestFingerprints("source_document", sourceDigest))
manifest.OutputDigests = workspaceFingerprints(chunkOutputDigests(chunks))
manifest.OutputDigests = workspaceFingerprints(outputDigests)
manifest.ValidationStatus = validationStatusString(warnings, nil)
manifest.CompletedAt = timePtr(r.timestamp())
return r.writeManifest("chunk/manifest.json", coreworkspace.ChunkManifest{
@@ -260,14 +264,13 @@ type chunksEnvelope struct {
}
type chunkEnvelope struct {
ID string `json:"id"`
SourceID string `json:"source_id"`
Index int `json:"index"`
StartUnitID int `json:"start_unit_id"`
EndUnitID int `json:"end_unit_id"`
Content binaryEnvelope `json:"content"`
Units []source.SourceUnit `json:"units,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
ID string `json:"id"`
SourceID string `json:"source_id"`
Index int `json:"index"`
Ref source.SourceRef `json:"ref"`
Content binaryEnvelope `json:"content"`
Units []source.SourceUnit `json:"units,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type extractOutputsEnvelope struct {
@@ -320,21 +323,20 @@ type binaryEnvelope struct {
Warnings []contracts.Warning `json:"warnings,omitempty"`
}
func chunkEnvelopes(chunks []contracts.SourceChunk) []chunkEnvelope {
func chunkEnvelopes(chunks []source.Chunk) []chunkEnvelope {
if len(chunks) == 0 {
return nil
}
out := make([]chunkEnvelope, 0, len(chunks))
for _, chunk := range chunks {
out = append(out, chunkEnvelope{
ID: chunk.ID,
SourceID: chunk.SourceID,
Index: chunk.Index,
StartUnitID: chunk.StartUnitID,
EndUnitID: chunk.EndUnitID,
Content: binaryEnvelopeFromContent(chunk.Content, chunk.MediaType, chunk.Metadata, nil),
Units: cloneSourceUnits(chunk.Units),
Metadata: cloneMetadata(chunk.Metadata),
ID: chunk.ID,
SourceID: chunk.SourceID,
Index: chunk.Index,
Ref: chunk.Ref,
Content: binaryEnvelopeFromContent(chunk.Content, chunk.MediaType, chunk.Metadata, nil),
Units: cloneSourceUnits(chunk.Units),
Metadata: cloneMetadata(chunk.Metadata),
})
}
return out
@@ -468,15 +470,19 @@ func extractPayloads(outputs []contracts.ExtractOutput) []contracts.RawPayload {
return payloads
}
func chunkOutputDigests(chunks []contracts.SourceChunk) []pipeline.CheckpointFingerprint {
func chunkOutputDigests(chunks []source.Chunk) ([]pipeline.CheckpointFingerprint, error) {
values := make([]pipeline.CheckpointFingerprint, 0, len(chunks))
for _, chunk := range chunks {
digest, err := source.DigestChunk(chunk)
if err != nil {
return nil, fmt.Errorf("chunk %q: %w", chunk.ID, err)
}
values = append(values, pipeline.CheckpointFingerprint{
Name: chunk.ID,
Value: contentDigest(chunk.Content),
Value: digest,
})
}
return normalizeFingerprints(values)
return normalizeFingerprints(values), nil
}
func digestFingerprints(name string, digest string) []pipeline.CheckpointFingerprint {

72
internal/framework/checkpoint/recorder_test.go
View File

@@ -24,16 +24,15 @@ func TestWorkspaceRecorderWritesSuccessfulCheckpointFiles(t *testing.T) {
Digest: "sha256:source",
Units: []source.SourceUnit{{ID: 1, Kind: "line", Text: "hello", Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1}}},
}
chunks := []contracts.SourceChunk{
chunks := []source.Chunk{
{
ID: "chunk-1",
SourceID: "source-1",
Index: 0,
StartUnitID: 1,
EndUnitID: 1,
Content: []byte("chunk content"),
MediaType: "text/plain",
Units: doc.Units,
ID: "chunk-1",
SourceID: "source-1",
Index: 0,
Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1},
Content: []byte("chunk content"),
MediaType: "text/plain",
Units: doc.Units,
},
}
@@ -91,16 +90,15 @@ func TestWorkspaceLoaderReusesSuccessfulCheckpointFiles(t *testing.T) {
Digest: "sha256:source",
Units: []source.SourceUnit{{ID: 1, Kind: "line", Text: "hello", Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1}}},
}
chunks := []contracts.SourceChunk{
chunks := []source.Chunk{
{
ID: "chunk-1",
SourceID: "source-1",
Index: 0,
StartUnitID: 1,
EndUnitID: 1,
Content: []byte("chunk content"),
MediaType: "text/plain",
Units: doc.Units,
ID: "chunk-1",
SourceID: "source-1",
Index: 0,
Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1},
Content: []byte("chunk content"),
MediaType: "text/plain",
Units: doc.Units,
},
}
extractOutput := contracts.ExtractOutput{
@@ -162,6 +160,9 @@ func TestWorkspaceLoaderReusesSuccessfulCheckpointFiles(t *testing.T) {
if !decision.Reused || len(chunkCheckpoint.Chunks) != 1 || string(chunkCheckpoint.Chunks[0].Content) != "chunk content" {
t.Fatalf("chunk decision = %#v checkpoint=%#v, want reused", decision, chunkCheckpoint)
}
if got, want := chunkCheckpoint.Chunks[0].Ref, chunks[0].Ref; got != want {
t.Fatalf("checkpoint chunk ref = %#v, want %#v", got, want)
}
extractCheckpoint, decision := loader.Extract("spells", "dnd/spells", extractDeps)
if !decision.Reused || len(extractCheckpoint.Outputs) != 1 || string(extractCheckpoint.Outputs[0].Payload.Content) != `{"spell":"cure wounds"}` {
t.Fatalf("extract decision = %#v checkpoint=%#v, want reused", decision, extractCheckpoint)
@@ -187,7 +188,7 @@ func TestWorkspaceLoaderInvalidatesMissingCorruptAndMismatchedCheckpoints(t *tes
t.Run("dependency mismatch", func(t *testing.T) {
root := t.TempDir()
recorder := newTestRecorder(t, root)
chunks := []contracts.SourceChunk{{
chunks := []source.Chunk{{
ID: "chunk-1",
SourceID: "source-1",
Content: []byte("chunk content"),
@@ -202,10 +203,41 @@ func TestWorkspaceLoaderInvalidatesMissingCorruptAndMismatchedCheckpoints(t *tes
}
})
t.Run("incompatible workspace schema remains untouched", func(t *testing.T) {
root := t.TempDir()
recorder := newTestRecorder(t, root)
doc := &source.SourceDocument{
ID: "source-1", Kind: "document", Format: "text/plain", Digest: "sha256:source",
Units: []source.SourceUnit{{ID: 1, Kind: "line", Text: "hello", Ref: source.SourceRef{SourceID: "source-1", StartUnitID: 1, EndUnitID: 1}}},
}
if err := recorder.SourceSucceeded("seriatim", doc); err != nil {
t.Fatalf("SourceSucceeded: %v", err)
}
manifestPath := filepath.Join(root, "source", "manifest.json")
manifest := strings.Replace(string(readFile(t, manifestPath)), coreworkspace.WorkspaceSchemaVersion, coreworkspace.WorkspaceSchemaVersionV1, 1)
if err := os.WriteFile(manifestPath, []byte(manifest), 0o644); err != nil {
t.Fatalf("write legacy manifest: %v", err)
}
beforeManifest := readFile(t, manifestPath)
payloadPath := filepath.Join(root, "source", "source-document.json")
beforePayload := readFile(t, payloadPath)
loader := &WorkspaceLoader{root: root}
if _, decision := loader.Source("seriatim"); decision.Reused || !strings.Contains(decision.Reason, "incompatible") || !strings.Contains(decision.Reason, coreworkspace.WorkspaceSchemaVersionV1) {
t.Fatalf("decision = %#v, want incompatible legacy schema invalidation", decision)
}
if got := readFile(t, manifestPath); string(got) != string(beforeManifest) {
t.Fatal("legacy manifest changed during reuse decision")
}
if got := readFile(t, payloadPath); string(got) != string(beforePayload) {
t.Fatal("legacy payload changed during reuse decision")
}
})
t.Run("corrupt payload", func(t *testing.T) {
root := t.TempDir()
recorder := newTestRecorder(t, root)
chunks := []contracts.SourceChunk{{
chunks := []source.Chunk{{
ID: "chunk-1",
SourceID: "source-1",
Content: []byte("chunk content"),