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

Publish and inspect configured extraction artifacts

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-10 00:32:24 +00:00
parent 39af7d4f3c
commit 0341e0c7c0
13 changed files with 466 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
internal/app/operator_artifact_rendering.go
View File

@@ -10,9 +10,10 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)
func buildHelperArtifactCatalog(cfg *config.Config) (*artifacts.ArtifactCatalog, error) {
func buildHelperArtifactCatalog(cfg *config.Config, m *manifest.Manifest) (*artifacts.ArtifactCatalog, error) {
catalog := artifacts.NewArtifactCatalog()
if err := catalog.RegisterBuiltIns(); err != nil {
return nil, err
@@ -26,6 +27,14 @@ func buildHelperArtifactCatalog(cfg *config.Config) (*artifacts.ArtifactCatalog,
if err := catalog.RegisterConfiguredArtifacts(configured, nil); err != nil {
return nil, err
}
extractionDefinitions := artifacts.ExtractionDefinitionsFromConfig(cfg.Pipeline.Notarius)
if err := catalog.RegisterExtractionArtifacts(extractionDefinitions); err != nil {
return nil, err
}
paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
if cfg.Pipeline.Notarius != nil && cfg.Pipeline.Notarius.Enabled {
catalog.HydrateExtractionArtifacts(paths, m, extractionDefinitions)
}
return catalog, nil
}
@@ -40,6 +49,14 @@ func writeArtifactList(out io.Writer, cfg *config.Config, catalog *artifacts.Art
for _, entry := range catalog.ListConfigured() {
writeArtifactLine(out, entry.SourceID, lockSet)
}
fmt.Fprintln(out, "Extraction:")
for _, entry := range catalog.ListExtraction() {
state := "unavailable"
if entry.Available {
state = "available"
}
writeExtractionArtifactLine(out, entry.SourceID, state, entry.Provenance, lockSet)
}
fmt.Fprintln(out, "Previous-session:")
for _, req := range artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg)) {
fmt.Fprintf(out, "- %s required=%t\n", artifactpolicy.PreviousSessionSourceID(req.Name), req.Required)
@@ -50,6 +67,17 @@ func writeArtifactList(out io.Writer, cfg *config.Config, catalog *artifacts.Art
}
}
func writeExtractionArtifactLine(out io.Writer, source, state, provenance string, lockSet map[string]config.PublishLockRule) {
parts := []string{source, "planned", state}
if strings.TrimSpace(provenance) != "" {
parts = append(parts, "provenance="+strings.TrimSpace(provenance))
}
if _, ok := lockSet[source]; ok {
parts = append(parts, "locked")
}
fmt.Fprintf(out, "- %s\n", strings.Join(parts, " "))
}
func writeArtifactLine(out io.Writer, source string, lockSet map[string]config.PublishLockRule) {
parts := []string{source}
if _, ok := lockSet[source]; ok {
@@ -103,7 +131,12 @@ func remotePublishedOutputAvailability(ctx context.Context, cfg *config.Config,
func helperPublishedOutputDest(rule config.PublishOutputRule, catalog *artifacts.ArtifactCatalog) (string, bool, error) {
source := strings.TrimSpace(rule.Source)
normalized, err := artifactpolicy.ResolvePublishedDestination(source, rule.Dest, helperConfiguredOutputPathMap(catalog))
normalized, err := artifactpolicy.ResolvePublishedDestinationWithExtractions(
source,
rule.Dest,
helperConfiguredOutputPathMap(catalog),
helperExtractionOutputSet(catalog),
)
if err != nil {
return "", false, err
}
@@ -112,6 +145,19 @@ func helperPublishedOutputDest(rule config.PublishOutputRule, catalog *artifacts
return normalized, showDest, nil
}
func helperExtractionOutputSet(catalog *artifacts.ArtifactCatalog) map[string]struct{} {
out := map[string]struct{}{}
if catalog == nil {
return out
}
for _, entry := range catalog.ListExtraction() {
if strings.TrimSpace(entry.ExtractionKey) != "" {
out[entry.ExtractionKey] = struct{}{}
}
}
return out
}
func helperConfiguredOutputPathMap(catalog *artifacts.ArtifactCatalog) map[string]string {
out := map[string]string{}
if catalog == nil {

4
internal/app/operator_artifacts_list.go
View File

@@ -22,11 +22,11 @@ func ArtifactsList(ctx context.Context, args []string, out io.Writer) error {
if strings.TrimSpace(flags.sessionID) == "" {
return fmt.Errorf("artifacts list: session_id is required")
}
cfg, store, locks, _, err := loadHelperContext(ctx, flags, remote)
cfg, store, locks, m, err := loadHelperContext(ctx, flags, remote)
if err != nil {
return fmt.Errorf("artifacts list: %w", err)
}
catalog, err := buildHelperArtifactCatalog(cfg)
catalog, err := buildHelperArtifactCatalog(cfg, m)
if err != nil {
return fmt.Errorf("artifacts list: %w", err)
}

133
internal/app/operator_helpers_test.go
View File

@@ -8,8 +8,10 @@ import (
"path/filepath"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
@@ -772,6 +774,71 @@ func TestExecuteArtifactsListRemoteReportsPublishedAvailability(t *testing.T) {
}
}
func TestExecuteArtifactsListReportsExtractionLifecycleWithoutPayload(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
addExtractionOutputToPipeline(t, pipelinePath)
addPublishOutputsToPipeline(t, pipelinePath, `
outputs:
- source: narratio.extraction.encounters
dest: artifacts/encounters.json
required: true
`)
lanePath := writeOperatorExtractionManifest(t, workspaceRoot)
fake := &storage.FakeBackend{}
publishedKey := artifacts.S3PublishedOutputKey(
artifacts.S3SessionPrefix("dnd", "sample-campaign", "2026-05-03"),
"artifacts/encounters.json",
)
fake.SeedObject(storage.FakeObject{Key: publishedKey, Data: []byte(`{"secret":"DO_NOT_PRINT"}`)})
var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{
"session", "artifacts", "2026-05-03",
"--config", pipelinePath,
"--campaign-file", campaignPath,
"--session", sessionPath,
"--remote",
}, &stdout, &stderr)
if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
}
out := stdout.String()
for _, want := range []string{
"Extraction:",
"narratio.extraction.encounters planned available provenance=manifest.current_extract_run",
"narratio.extraction.encounters dest=artifacts/encounters.json remote=published",
} {
if !strings.Contains(out, want) {
t.Fatalf("stdout = %q, want %q", out, want)
}
}
if strings.Contains(out, "DO_NOT_PRINT") {
t.Fatalf("operator output exposed extraction payload: %q", out)
}
if err := os.Remove(lanePath); err != nil {
t.Fatal(err)
}
stdout.Reset()
stderr.Reset()
code = Execute([]string{
"session", "artifacts", "2026-05-03",
"--config", pipelinePath,
"--campaign-file", campaignPath,
"--session", sessionPath,
}, &stdout, &stderr)
if code != 0 {
t.Fatalf("unavailable exit code = %d, want 0; stderr=%q", code, stderr.String())
}
if !strings.Contains(stdout.String(), "narratio.extraction.encounters planned unavailable") {
t.Fatalf("stdout = %q, want unavailable extraction state", stdout.String())
}
}
func TestExecuteArtifactsListRemoteUsesPublishOutputDestinations(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
@@ -1041,6 +1108,72 @@ func addPublishOutputsToPipeline(t *testing.T, pipelinePath, publishYAML string)
}
}
func addExtractionOutputToPipeline(t *testing.T, pipelinePath string) {
t.Helper()
data, err := os.ReadFile(pipelinePath)
if err != nil {
t.Fatal(err)
}
data = append(data, []byte(`notarius:
enabled: true
config_path: notarius.yml
pipeline_id: campaign.extract
outputs:
encounters:
lane_id: encounters
media_type: application/json
schema_id: encounters
schema_version: "1"
module_key: encounters
`)...)
if err := os.WriteFile(pipelinePath, data, 0o644); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(filepath.Dir(pipelinePath), "notarius.yml"), []byte("{}\n"), 0o644); err != nil {
t.Fatal(err)
}
}
func writeOperatorExtractionManifest(t *testing.T, workspaceRoot string) string {
t.Helper()
paths := artifacts.NewLocalStore(workspaceRoot).SessionPathsFor("sample-campaign", "2026-05-03")
bundleRoot := filepath.Join(paths.ArtifactsDir, "notarius", "extract-run-1")
lanePath := filepath.Join(bundleRoot, "lanes", "encounters.json")
indexPath := filepath.Join(bundleRoot, "index.json")
mustWriteTestFile(t, lanePath, `{"secret":"DO_NOT_PRINT"}`)
mustWriteTestFile(t, indexPath, `{"lanes":[]}`)
laneChecksum, err := artifacts.SHA256File(lanePath)
if err != nil {
t.Fatal(err)
}
indexChecksum, err := artifacts.SHA256File(indexPath)
if err != nil {
t.Fatal(err)
}
m := manifest.New("2026-05-03", time.Now().UTC())
m.Campaign = "sample-campaign"
m.Stages["extract"] = &manifest.StageRecord{
Name: "extract", Status: manifest.StatusSucceeded,
Metadata: map[string]any{
"narratio_run_id": "extract-run-1", "bundle_root": bundleRoot,
"receipt": map[string]any{"run_id": "notarius-run-1", "pipeline_id": "campaign.extract"},
},
Outputs: []manifest.ArtifactRecord{
{
Kind: "notarius_lane", SourceID: artifacts.ExtractionArtifactSourceID("encounters"), LocalPath: lanePath,
ProducerRunID: "extract-run-1", Checksum: laneChecksum,
Contract: &artifactmodel.ContractMetadata{MediaType: "application/json", SchemaID: "encounters", SchemaVersion: "1", ModuleKey: "encounters"},
ExternalProvenance: &artifactmodel.ExternalProvenance{System: "notarius", RunID: "notarius-run-1", PipelineID: "campaign.extract", ArtifactID: "encounters"},
},
{Kind: "notarius_index", LocalPath: indexPath, ProducerRunID: "extract-run-1", Checksum: indexChecksum},
},
}
if err := (&manifest.LocalStore{}).Save(context.Background(), paths.ManifestPath, m); err != nil {
t.Fatal(err)
}
return lanePath
}
func replaceInFileOrFatal(t *testing.T, path, old, new string) {
t.Helper()
data, err := os.ReadFile(path)

5
internal/app/operator_status.go
View File

@@ -11,6 +11,7 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)
// Status reports effective local/remote session state.
@@ -40,11 +41,13 @@ func Status(ctx context.Context, args []string, out io.Writer) error {
writeStatusStableInputs(out, inspectStableInputs(cfg))
writeStatusLocalAudio(out, inspectLocalAudioPresence(cfg))
var localManifest *manifest.Manifest
if m, err := loadLocalManifest(ctx, paths.ManifestPath); err != nil {
fmt.Fprintf(out, "Local manifest: error: %v\n", err)
} else if m == nil {
fmt.Fprintln(out, "Local manifest: missing")
} else {
localManifest = m
fmt.Fprintf(out, "Local manifest: %s\n", paths.ManifestPath)
writeStageStatuses(out, m)
}
@@ -72,7 +75,7 @@ func Status(ctx context.Context, args []string, out io.Writer) error {
lockChecks := inspectEffectiveLocks(ctx, cfg, store)
locks := lockChecks.Locks
lockErr := lockChecks.Err
if catalog, catalogErr := buildHelperArtifactCatalog(cfg); catalogErr != nil {
if catalog, catalogErr := buildHelperArtifactCatalog(cfg, localManifest); catalogErr != nil {
fmt.Fprintf(out, "Remote outputs: error: %v\n", catalogErr)
} else if storeErr == nil {
catalogLocks := locks

51
internal/app/restore_execution_test.go
View File

@@ -9,8 +9,10 @@ import (
"path/filepath"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
@@ -62,6 +64,55 @@ func TestExecuteRestoreNonDryRunRestoresDurableFiles(t *testing.T) {
}
}
func TestExecuteRestoreRoundTripsPublishedExtractionAndManifestMetadata(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
fake := &storage.FakeBackend{}
cfg, sessionPrefix, manifestKey, runIDKey := seedRestoreCommittedState(t, fake, pipelinePath, campaignPath, sessionPath)
seedRestoreObject(fake, sessionPrefix+"artifacts/encounters.json", []byte(`{"encounters":[]}`))
seedRestoreObject(fake, runIDKey, []byte("20260519T010203Z-a1b2c3d4\n"))
remoteManifest := manifest.New(cfg.Session.SessionID, time.Now().UTC())
remoteManifest.Campaign = cfg.Session.Campaign
remoteManifest.RunID = "20260519T010203Z-a1b2c3d4"
remoteManifest.Stages["extract"] = &manifest.StageRecord{
Name: "extract", Status: manifest.StatusSucceeded,
Outputs: []manifest.ArtifactRecord{{
Kind: "notarius_lane", SourceID: artifacts.ExtractionArtifactSourceID("encounters"),
LocalPath: "/prior/workspace/artifacts/notarius/extract-run-1/lanes/encounters.json",
Contract: &artifactmodel.ContractMetadata{
MediaType: "application/json", SchemaID: "encounters", SchemaVersion: "1",
},
ExternalProvenance: &artifactmodel.ExternalProvenance{
System: "notarius", RunID: "notarius-run-1", PipelineID: "campaign.extract", ArtifactID: "encounters",
},
}},
}
manifestBody, err := json.Marshal(remoteManifest)
if err != nil {
t.Fatal(err)
}
seedRestoreObject(fake, manifestKey, manifestBody)
restoreWithStoreAndRealPhases(t, fake)
var stdout bytes.Buffer
var stderr bytes.Buffer
code := Execute([]string{"session", "restore", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, &stdout, &stderr)
if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
}
sessionRoot := artifacts.SessionWorkDirForCampaign(workspaceRoot, cfg.Session.Campaign, cfg.Session.SessionID)
mustReadEquals(t, filepath.Join(sessionRoot, "artifacts", "encounters.json"), `{"encounters":[]}`)
restored, err := (&manifest.LocalStore{}).Load(context.Background(), filepath.Join(sessionRoot, "manifest.json"))
if err != nil {
t.Fatalf("load restored manifest: %v", err)
}
lane := restored.Stages["extract"].Outputs[0]
if lane.Contract == nil || lane.Contract.SchemaID != "encounters" || lane.ExternalProvenance == nil || lane.ExternalProvenance.RunID != "notarius-run-1" {
t.Fatalf("restored extraction metadata = %#v", lane)
}
}
func TestExecuteRestoreIncludeAudioRestoresAudio(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)