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

Remove legacy workspace and diagnostics implementation

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-18 12:48:21 +00:00
parent 1542a12497
commit 26142f0e05
21 changed files with 3 additions and 8726 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/roadmap/implementation.md
View File

@@ -469,7 +469,7 @@ and CLI tests use the new public contract; and the repository-wide checks pass.
## Stage 4: Remove legacy workspace and diagnostics implementation ## Stage 4: Remove legacy workspace and diagnostics implementation
**Status:** Not started **Status:** Complete
### Objective ### Objective

498
internal/cli/chunk_cache_test.go
View File

@@ -1,498 +0,0 @@
//go:build legacy
package cli
import (
"bytes"
"errors"
"os"
"path/filepath"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
type recordingChunkPlanStore struct {
record pipeline.ChunkPlanRecord
decision pipeline.ChunkPlanDecision
loadErr error
saveErr error
loads int
saves int
}
func (s *recordingChunkPlanStore) Load(string) (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) {
s.loads++
return s.record, s.decision, s.loadErr
}
func (s *recordingChunkPlanStore) Save(record pipeline.ChunkPlanRecord) error {
s.saves++
s.record = record
return s.saveErr
}
type recordingChunkPlanFactory struct {
roots []string
store *recordingChunkPlanStore
err error
}
func (f *recordingChunkPlanFactory) build(root string) (pipeline.ChunkPlanStore, error) {
f.roots = append(f.roots, root)
if f.err != nil {
return nil, f.err
}
if f.store == nil {
f.store = &recordingChunkPlanStore{decision: pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing}}
}
return f.store, nil
}
func TestRunChunkCachePrecedence(t *testing.T) {
tests := []struct {
name string
fileMode string
envMode string
flagMode string
wantMode pipeline.ChunkCacheMode
wantBuild bool
}{
{name: "file refresh", fileMode: "refresh", wantMode: pipeline.ChunkCacheRefresh, wantBuild: true},
{name: "environment over file", fileMode: "bypass", envMode: "refresh", wantMode: pipeline.ChunkCacheRefresh, wantBuild: true},
{name: "flag over environment", fileMode: "refresh", envMode: "auto", flagMode: "bypass", wantMode: pipeline.ChunkCacheBypass},
{name: "explicit auto", fileMode: "bypass", envMode: "refresh", flagMode: "auto", wantMode: pipeline.ChunkCacheAuto, wantBuild: true},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
root := filepath.Join(t.TempDir(), "plans")
configPath := writeTestConfig(t, cacheTestConfig(tc.fileMode, root, ""))
inputPath := writeFile(t, "input.txt", "input")
values := map[string]string{}
if tc.envMode != "" {
values["NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE"] = tc.envMode
}
factory := &recordingChunkPlanFactory{}
args := []string{"run", "example", "--config", configPath, "--input", inputPath, "--output-dir", t.TempDir()}
if tc.flagMode != "" {
args = append(args, "--chunk_cache", tc.flagMode)
}
code, stderr := runCacheCommand(t, args, cacheTestOptions(t, values, factory))
if code != 0 {
t.Fatalf("code = %d stderr = %q", code, stderr)
}
if got := len(factory.roots) > 0; got != tc.wantBuild {
t.Fatalf("store built = %t roots = %#v, want %t", got, factory.roots, tc.wantBuild)
}
if !tc.wantBuild {
return
}
if tc.wantMode == pipeline.ChunkCacheAuto && (factory.store.loads != 1 || factory.store.saves != 1) {
t.Fatalf("auto calls = load %d save %d", factory.store.loads, factory.store.saves)
}
if tc.wantMode == pipeline.ChunkCacheRefresh && (factory.store.loads != 0 || factory.store.saves != 1) {
t.Fatalf("refresh calls = load %d save %d", factory.store.loads, factory.store.saves)
}
})
}
}
func TestRunChunkCacheInvalidValuesHaveEstablishedExitCodes(t *testing.T) {
validConfig := writeTestConfig(t, cacheTestConfig("bypass", "", ""))
invalidFile := writeTestConfig(t, cacheTestConfig("sometimes", "", ""))
inputPath := writeFile(t, "input.txt", "input")
tests := []struct {
name string
config string
env map[string]string
flag string
want int
}{
{name: "flag", config: validConfig, flag: "sometimes", want: 2},
{name: "environment", config: validConfig, env: map[string]string{"NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE": "sometimes"}, want: 1},
{name: "file", config: invalidFile, want: 1},
{name: "flag does not mask invalid file", config: invalidFile, flag: "bypass", want: 1},
{name: "flag does not mask invalid environment", config: validConfig, env: map[string]string{"NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE": "sometimes"}, flag: "bypass", want: 1},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
args := []string{"run", "example", "--config", tc.config, "--input", inputPath, "--output-dir", t.TempDir()}
if tc.flag != "" {
args = append(args, "--chunk_cache", tc.flag)
}
code, _ := runCacheCommand(t, args, cacheTestOptions(t, tc.env, &recordingChunkPlanFactory{}))
if code != tc.want {
t.Fatalf("code = %d, want %d", code, tc.want)
}
})
}
}
func TestRunChunkPlanRootResolution(t *testing.T) {
t.Run("explicit file root", func(t *testing.T) {
factory := &recordingChunkPlanFactory{}
resolverCalls := 0
opts := cacheTestOptions(t, nil, factory)
opts.UserCacheDir = func() (string, error) { resolverCalls++; return "", errors.New("must not be called") }
configPath := writeTestConfig(t, cacheTestConfig("auto", "/var/cache/notarius/chunk-plans", ""))
code, stderr := runCacheCommand(t, cacheRunArgs(t, configPath), opts)
if code != 0 || stderr != "" || resolverCalls != 0 || len(factory.roots) != 1 || factory.roots[0] != "/var/cache/notarius/chunk-plans" {
t.Fatalf("code=%d stderr=%q resolver=%d roots=%#v", code, stderr, resolverCalls, factory.roots)
}
})
t.Run("environment root", func(t *testing.T) {
factory := &recordingChunkPlanFactory{}
environmentRoot := filepath.Join(t.TempDir(), "environment-plans")
configPath := writeTestConfig(t, cacheTestConfig("auto", filepath.Join(t.TempDir(), "file-plans"), ""))
code, stderr := runCacheCommand(t, cacheRunArgs(t, configPath), cacheTestOptions(t, map[string]string{"NOTARIUS_WORKSPACE_CHUNK_CACHE_DIR": environmentRoot}, factory))
if code != 0 || stderr != "" || len(factory.roots) != 1 || factory.roots[0] != environmentRoot {
t.Fatalf("code=%d stderr=%q roots=%#v", code, stderr, factory.roots)
}
})
t.Run("per-user default", func(t *testing.T) {
factory := &recordingChunkPlanFactory{}
cacheDir := filepath.Join(t.TempDir(), "user-cache")
opts := cacheTestOptions(t, nil, factory)
opts.UserCacheDir = func() (string, error) { return cacheDir, nil }
configPath := writeTestConfig(t, cacheTestConfig("auto", "", ""))
code, stderr := runCacheCommand(t, cacheRunArgs(t, configPath), opts)
want := filepath.Join(cacheDir, "notarius", "chunk-plans")
if code != 0 || stderr != "" || len(factory.roots) != 1 || factory.roots[0] != want {
t.Fatalf("code=%d stderr=%q roots=%#v want=%q", code, stderr, factory.roots, want)
}
})
}
func TestRunBypassSkipsRootAndStore(t *testing.T) {
factory := &recordingChunkPlanFactory{err: errors.New("must not build")}
resolverCalls := 0
opts := cacheTestOptions(t, nil, factory)
opts.UserCacheDir = func() (string, error) { resolverCalls++; return "", errors.New("must not resolve") }
configPath := writeTestConfig(t, cacheTestConfig("bypass", "", ""))
code, stderr := runCacheCommand(t, cacheRunArgs(t, configPath), opts)
if code != 0 || stderr != "" || resolverCalls != 0 || len(factory.roots) != 0 {
t.Fatalf("code=%d stderr=%q resolver=%d roots=%#v", code, stderr, resolverCalls, factory.roots)
}
}
func TestRunChunkPlanSetupFailures(t *testing.T) {
for _, mode := range []string{"auto", "refresh"} {
t.Run(mode+" resolver", func(t *testing.T) {
factory := &recordingChunkPlanFactory{}
opts := cacheTestOptions(t, nil, factory)
opts.UserCacheDir = func() (string, error) { return "", errors.New("cache unavailable") }
configPath := writeTestConfig(t, cacheTestConfig(mode, "", ""))
code, stderr := runCacheCommand(t, cacheRunArgs(t, configPath), opts)
if code != 1 || !strings.Contains(stderr, "cache unavailable") || len(factory.roots) != 0 {
t.Fatalf("code=%d stderr=%q roots=%#v", code, stderr, factory.roots)
}
})
}
t.Run("store", func(t *testing.T) {
factory := &recordingChunkPlanFactory{err: errors.New("unwritable store")}
configPath := writeTestConfig(t, cacheTestConfig("auto", t.TempDir(), ""))
code, stderr := runCacheCommand(t, cacheRunArgs(t, configPath), cacheTestOptions(t, nil, factory))
if code != 1 || !strings.Contains(stderr, "unwritable store") {
t.Fatalf("code=%d stderr=%q", code, stderr)
}
})
t.Run("unwritable filesystem store", func(t *testing.T) {
root := writeFile(t, "not-a-directory", "occupied")
configPath := writeTestConfig(t, `version: 2
workspace:
chunk_cache:
mode: auto
directory: `+root+`
pipelines:
dnd-session:
input: seriatim
artifacts:
spells:
extract: dnd/spells
`)
code, stderr := runCacheCommand(t, []string{"run", "dnd-session", "--config", configPath, "--input", writeSeriatimInput(t), "--output-dir", t.TempDir()}, Options{LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), LookupEnv: mapLookup(nil)})
if code != 1 || (!strings.Contains(stderr, "load chunk plan") && !strings.Contains(stderr, "save chunk plan")) {
t.Fatalf("code=%d stderr=%q", code, stderr)
}
})
}
func TestConfigCommandsDoNotResolveOrCreateChunkPlanState(t *testing.T) {
configPath := writeTestConfig(t, cacheTestConfig("auto", "", ""))
for _, args := range [][]string{
{"config", "validate", "--config", configPath},
{"pipelines", "list", "--config", configPath},
} {
factory := &recordingChunkPlanFactory{err: errors.New("must not build")}
resolverCalls := 0
opts := cacheTestOptions(t, nil, factory)
opts.UserCacheDir = func() (string, error) { resolverCalls++; return "", errors.New("must not resolve") }
code, stderr := runCacheCommand(t, args, opts)
if code != 0 || stderr != "" || resolverCalls != 0 || len(factory.roots) != 0 {
t.Fatalf("args=%v code=%d stderr=%q resolver=%d roots=%#v", args, code, stderr, resolverCalls, factory.roots)
}
}
}
func TestRunRecordsExplicitChunkCacheOverrideAndEffectiveMode(t *testing.T) {
diagnosticsDir := t.TempDir()
configPath := writeTestConfig(t, cacheTestConfig("bypass", "", diagnosticsDir))
factory := &recordingChunkPlanFactory{}
args := append(cacheRunArgs(t, configPath), "--chunk_cache", "refresh")
code, stderr := runCacheCommand(t, args, cacheTestOptions(t, nil, factory))
if code != 0 || stderr != "" {
t.Fatalf("code=%d stderr=%q", code, stderr)
}
runDir := onlyChildDir(t, diagnosticsDir)
invocation := string(readFile(t, filepath.Join(runDir, diagnostics.ArtifactInvocationMetadata)))
effective := string(readFile(t, filepath.Join(runDir, diagnostics.ArtifactEffectiveConfig)))
if !strings.Contains(invocation, `"chunk_cache_override": "refresh"`) || !strings.Contains(effective, `"mode": "refresh"`) {
t.Fatalf("invocation=%s effective=%s", invocation, effective)
}
}
func TestRunChunkPlanDiagnosticsRedactStoreDecisionReason(t *testing.T) {
diagnosticsDir := t.TempDir()
configPath := writeTestConfig(t, cacheTestConfig("auto", t.TempDir(), diagnosticsDir))
factory := &recordingChunkPlanFactory{store: &recordingChunkPlanStore{
decision: pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanInvalid, Reason: "SENTINEL_INVALID_RECORD_CONTENT"},
}}
code, stderr := runCacheCommand(t, cacheRunArgs(t, configPath), cacheTestOptions(t, nil, factory))
if code != 0 || stderr != "" {
t.Fatalf("code=%d stderr=%q", code, stderr)
}
summary := string(readFile(t, filepath.Join(onlyChildDir(t, diagnosticsDir), diagnostics.ArtifactChunkPlan)))
if strings.Contains(summary, "SENTINEL_INVALID_RECORD_CONTENT") || !strings.Contains(summary, `"lookup_reason": "stored chunk plan is invalid"`) {
t.Fatalf("chunk plan diagnostic = %s", summary)
}
}
func TestDefaultAutoReusesPlanAcrossIndependentInvocations(t *testing.T) {
cacheBase := filepath.Join(t.TempDir(), "cache")
workspaceDir := filepath.Join(t.TempDir(), "workspace")
configPath := writeTestConfig(t, `version: 2
workspace:
directory: `+workspaceDir+`
debug:
enabled: true
pipelines:
dnd-session:
input: seriatim
chunk: dnd/scenes
artifacts:
spells:
extract: dnd/spells
`)
inputPath := writeFile(t, "source.json", `{
"metadata": {"id": "session-alpha"},
"segments": [
{"id": 1, "start": 0, "end": 1, "speaker": "Aria", "text": "Aria casts Cure Wounds."},
{"id": 2, "start": 1, "end": 2, "speaker": "Borin", "text": "Borin recovers."}
]
}`)
client := newFakeRunLLMClient(false)
opts := Options{
LLMClientFactory: fakeLLMFactory(client, nil),
LookupEnv: mapLookup(nil),
UserCacheDir: func() (string, error) { return cacheBase, nil },
}
for i := 0; i < 2; i++ {
code, stderr := runCacheCommand(t, []string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--output-dir", t.TempDir()}, opts)
if code != 0 {
t.Fatalf("run %d code=%d stderr=%q", i+1, code, stderr)
}
}
if client.calls != 3 {
t.Fatalf("LLM calls = %d, want chunk+extract then extract-only reuse", client.calls)
}
planRoot := filepath.Join(cacheBase, "notarius", "chunk-plans")
entries, err := os.ReadDir(planRoot)
if err != nil || len(entries) != 1 {
t.Fatalf("plan root entries = %v error=%v", entries, err)
}
if _, err := os.Stat(filepath.Join(planRoot, entries[0].Name(), "plan.json")); err != nil {
t.Fatalf("plan file: %v", err)
}
debugRuns := childDirs(t, filepath.Join(workspaceDir, "debug"))
if len(debugRuns) != 2 {
t.Fatalf("debug runs = %#v", debugRuns)
}
attemptCounts := 0
for _, runDir := range debugRuns {
if _, err := os.Stat(filepath.Join(runDir, "chunk", "attempt-01.json")); err == nil {
attemptCounts++
} else if !os.IsNotExist(err) {
t.Fatal(err)
}
}
if attemptCounts != 1 {
t.Fatalf("chunk attempt files across runs = %d, want only generating run", attemptCounts)
}
}
func TestChunkPlanReuseDependsOnlyOnSourceDigest(t *testing.T) {
cacheRoot := filepath.Join(t.TempDir(), "plans")
inputPath := writeSeriatimInput(t)
referencePath := writeFile(t, "players.txt", "Alyx")
profilePath := writeScriptoriumProfileFile(t, "chunk-profile", "http://127.0.0.1:8080/v1", "test-model")
seedConfig := writeTestConfig(t, `version: 2
workspace:
chunk_cache:
mode: auto
directory: `+cacheRoot+`
pipelines:
seed:
input: seriatim
chunk:
module: generic
options:
max_units: 1
overlap_units: 0
artifacts:
spells:
extract: dnd/spells
`)
changedConfig := writeTestConfig(t, `version: 2
scriptorium:
profile_file: `+profilePath+`
workspace:
chunk_cache:
mode: auto
directory: `+cacheRoot+`
pipelines:
changed:
input: seriatim
chunk:
module: dnd/scenes
llm_profile: chunk-profile
references:
players: `+referencePath+`
validators:
- generic/always_accept
artifacts:
spells:
extract: dnd/spells
`)
client := newFakeRunLLMClient(false)
opts := Options{LLMClientFactory: fakeLLMFactory(client, nil), LookupEnv: mapLookup(nil)}
for _, run := range []struct {
pipeline string
config string
}{{pipeline: "seed", config: seedConfig}, {pipeline: "changed", config: changedConfig}} {
code, stderr := runCacheCommand(t, []string{"run", run.pipeline, "--config", run.config, "--input", inputPath, "--output-dir", t.TempDir()}, opts)
if code != 0 {
t.Fatalf("pipeline %q code=%d stderr=%q", run.pipeline, code, stderr)
}
}
if client.calls != 2 {
t.Fatalf("LLM calls = %d, want one extractor call per run and no changed chunker call", client.calls)
}
}
func TestResumeAndChunkCacheModesRemainIndependent(t *testing.T) {
for _, mode := range []pipeline.ChunkCacheMode{pipeline.ChunkCacheAuto, pipeline.ChunkCacheBypass, pipeline.ChunkCacheRefresh} {
t.Run(string(mode), func(t *testing.T) {
workspaceDir := filepath.Join(t.TempDir(), "workspace")
cacheRoot := filepath.Join(t.TempDir(), "plans")
configPath := writeTestConfig(t, cacheTestConfigWithWorkspace(string(mode), cacheRoot, workspaceDir))
factory := &recordingChunkPlanFactory{}
args := append(cacheRunArgs(t, configPath), "--resume")
code, stderr := runCacheCommand(t, args, cacheTestOptions(t, nil, factory))
if code != 0 || stderr != "" {
t.Fatalf("code=%d stderr=%q", code, stderr)
}
if mode == pipeline.ChunkCacheBypass {
if len(factory.roots) != 0 {
t.Fatalf("bypass roots = %#v", factory.roots)
}
} else if len(factory.roots) != 1 || factory.roots[0] != cacheRoot {
t.Fatalf("roots = %#v, want %q", factory.roots, cacheRoot)
}
if entries := childDirs(t, filepath.Join(workspaceDir, "checkpoints")); len(entries) != 1 {
t.Fatalf("checkpoint roots = %#v", entries)
}
if strings.HasPrefix(cacheRoot, workspaceDir+string(filepath.Separator)) || strings.HasPrefix(workspaceDir, cacheRoot+string(filepath.Separator)) {
t.Fatalf("cache root %q and workspace root %q overlap", cacheRoot, workspaceDir)
}
})
}
}
func TestWorkspaceDirectoryDoesNotSelectChunkPlanRoot(t *testing.T) {
cacheBase := filepath.Join(t.TempDir(), "user-cache")
factory := &recordingChunkPlanFactory{}
for _, workspaceDir := range []string{filepath.Join(t.TempDir(), "workspace-one"), filepath.Join(t.TempDir(), "workspace-two")} {
configPath := writeTestConfig(t, cacheTestConfigWithWorkspace("auto", "", workspaceDir))
opts := cacheTestOptions(t, nil, factory)
opts.UserCacheDir = func() (string, error) { return cacheBase, nil }
code, stderr := runCacheCommand(t, cacheRunArgs(t, configPath), opts)
if code != 0 || stderr != "" {
t.Fatalf("workspace=%q code=%d stderr=%q", workspaceDir, code, stderr)
}
}
want := filepath.Join(cacheBase, "notarius", "chunk-plans")
if len(factory.roots) != 2 || factory.roots[0] != want || factory.roots[1] != want {
t.Fatalf("roots = %#v, want %q twice", factory.roots, want)
}
}
func cacheTestOptions(t *testing.T, env map[string]string, factory *recordingChunkPlanFactory) Options {
t.Helper()
registries := fakeExecutionRegistries(t)
return Options{
Catalog: catalogFromRegistries(registries),
Registries: registries,
LLMClientFactory: fakeLLMFactory(nil, nil),
LookupEnv: mapLookup(env),
UserCacheDir: func() (string, error) { return filepath.Join(t.TempDir(), "cache"), nil },
ChunkPlanStoreFactory: factory.build,
}
}
func cacheRunArgs(t *testing.T, configPath string) []string {
t.Helper()
return []string{"run", "example", "--config", configPath, "--input", writeFile(t, "input.txt", "input"), "--output-dir", t.TempDir()}
}
func runCacheCommand(t *testing.T, args []string, opts Options) (int, string) {
t.Helper()
var stdout bytes.Buffer
var stderr bytes.Buffer
code := RunWithOptions(args, &stdout, &stderr, opts)
return code, stderr.String()
}
func cacheTestConfig(mode, directory, diagnosticsDir string) string {
var b strings.Builder
b.WriteString("version: 2\n")
if mode != "" || directory != "" {
b.WriteString("workspace:\n chunk_cache:\n")
if mode != "" {
b.WriteString(" mode: " + mode + "\n")
}
if directory != "" {
b.WriteString(" directory: " + directory + "\n")
}
}
if diagnosticsDir != "" {
b.WriteString("diagnostics:\n work_dir: " + diagnosticsDir + "\n retention: always\n")
}
b.WriteString("pipelines:\n example:\n input: fake/input\n artifacts:\n spells:\n extract: fake/extract\n")
return b.String()
}
func cacheTestConfigWithWorkspace(mode, cacheRoot, workspaceDir string) string {
var b strings.Builder
b.WriteString("version: 2\nworkspace:\n directory: " + workspaceDir + "\n resume:\n enabled: true\n chunk_cache:\n mode: " + mode + "\n")
if cacheRoot != "" {
b.WriteString(" directory: " + cacheRoot + "\n")
}
b.WriteString("pipelines:\n example:\n input: fake/input\n artifacts:\n spells:\n extract: fake/extract\n")
return b.String()
}

602
internal/cli/compatibility_test.go
View File

@@ -1,602 +0,0 @@
//go:build legacy
package cli
import (
"bytes"
"context"
"encoding/json"
"io/fs"
"path/filepath"
"reflect"
"sort"
"strings"
"sync"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/core/config"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
frameworkllm "gitea.maximumdirect.net/eric/notarius/internal/framework/llm"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/chunk/scenes"
"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/spells"
spellshape "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/validate/spells/shape"
spellsourcerefs "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/validate/spells/source_refs"
spellrelatedness "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/validate/spells/source_relatedness"
validjson "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/validate/valid_json"
validjsonschema "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/validate/valid_json_schema"
)
func TestProductionCompatibilitySnapshot(t *testing.T) {
normalizedOptions, err := normalizeOptions(Options{})
if err != nil {
t.Fatalf("normalizeOptions() error = %v, want nil", err)
}
if normalizedOptions.Catalog.Inputs != normalizedOptions.Registries.Inputs ||
normalizedOptions.Catalog.Chunkers != normalizedOptions.Registries.Chunkers ||
normalizedOptions.Catalog.Extractors != normalizedOptions.Registries.Extractors ||
normalizedOptions.Catalog.Mergers != normalizedOptions.Registries.Mergers ||
normalizedOptions.Catalog.Normalizers != normalizedOptions.Registries.Normalizers ||
normalizedOptions.Catalog.Validators != normalizedOptions.Registries.Validators ||
normalizedOptions.Catalog.ValidatorChains != normalizedOptions.Registries.ValidatorChains ||
normalizedOptions.Catalog.Outputs != normalizedOptions.Registries.Outputs {
t.Fatal("production catalog and execution registries do not share one composition")
}
if normalizedOptions.LLMClientFactory == nil {
t.Fatal("production LLM client factory is nil")
}
registries, err := productionRegistries()
if err != nil {
t.Fatalf("productionRegistries() error = %v, want nil", err)
}
keySnapshots := []struct {
name string
got []string
want []string
}{
{name: "inputs", got: registries.Inputs.RegisteredKeys(), want: []string{"seriatim"}},
{name: "chunkers", got: registries.Chunkers.RegisteredKeys(), want: []string{"dnd/scenes", "generic"}},
{name: "extractors", got: registries.Extractors.RegisteredKeys(), want: []string{"dnd/spells"}},
{name: "mergers", got: registries.Mergers.RegisteredKeys(), want: []string{"appendorder"}},
{name: "normalizers", got: registries.Normalizers.RegisteredKeys(), want: []string{"noop"}},
{name: "validators", got: registries.Validators.RegisteredKeys(), want: []string{
"extract/dnd/spells/shape", "extract/dnd/spells/source_refs", "extract/dnd/spells/source_relatedness",
"generic/always_accept", "generic/always_reject", "generic/valid_json", "generic/valid_json_schema",
}},
{name: "outputs", got: registries.Outputs.RegisteredKeys(), want: []string{"json"}},
}
for _, snapshot := range keySnapshots {
t.Run(snapshot.name, func(t *testing.T) {
if !reflect.DeepEqual(snapshot.got, snapshot.want) {
t.Fatalf("registered keys = %#v, want compatibility snapshot %#v", snapshot.got, snapshot.want)
}
})
}
wantChain := []pipeline.ModuleBinding{
pipeline.Binding(validjson.Key),
pipeline.Binding(validjsonschema.Key),
pipeline.Binding(spellshape.Key),
pipeline.Binding(spellsourcerefs.Key),
pipeline.Binding(spellrelatedness.Key),
}
if got := registries.ValidatorChains.Validators(pipeline.StageExtract, spells.Key); !reflect.DeepEqual(got, wantChain) {
t.Fatalf("spell validator chain = %#v, want compatibility snapshot %#v", got, wantChain)
}
assets, err := productionPromptAssets()
if err != nil {
t.Fatalf("productionPromptAssets() error = %v, want nil", err)
}
assertAssetNames(t, assets.PromptFS, []string{
"dnd.scenes/dnd.scenes.yaml",
"dnd.scenes/instructions.md",
"dnd.scenes/sharedassets/common-dnd-references.md",
"dnd.scenes/sharedassets/common-dnd-system.md",
"dnd.scenes/sharedassets/common-dnd-transcript.md",
"dnd.scenes/task.md",
"dnd.spells/dnd.spells.yaml",
"dnd.spells/instructions.md",
"dnd.spells/sharedassets/common-dnd-references.md",
"dnd.spells/sharedassets/common-dnd-system.md",
"dnd.spells/sharedassets/common-dnd-transcript.md",
"dnd.spells/task.md",
})
assertAssetNames(t, assets.SchemaFS, []string{
"dnd_scenes.v1.json",
"dnd_spells_llm.v1.json",
})
identitySnapshot := map[string]map[string]any{
"scenes": sceneManifestMetadata(t),
"spells": spellManifestMetadata(t),
}
for name, metadata := range identitySnapshot {
for _, key := range []string{"prompt_id", "prompt_version", "prompt_sha256", "response_schema_key", "response_schema_id", "response_schema_name", "response_schema_version", "response_schema_sha256"} {
if value, ok := metadata[key].(string); !ok || value == "" {
t.Fatalf("%s metadata[%q] = %#v, want non-empty identity", name, key, metadata[key])
}
}
}
if got := []any{
identitySnapshot["scenes"]["prompt_id"], identitySnapshot["scenes"]["prompt_version"],
identitySnapshot["scenes"]["response_schema_key"], identitySnapshot["scenes"]["response_schema_id"], identitySnapshot["scenes"]["response_schema_name"], identitySnapshot["scenes"]["response_schema_version"],
}; !reflect.DeepEqual(got, []any{"dnd.scenes", "v1", "dnd_scenes", "notarius.dnd.scenes", "notarius_dnd_scenes_v1", "v1"}) {
t.Fatalf("scene identities = %#v, want compatibility snapshot", got)
}
if got := []any{
identitySnapshot["spells"]["prompt_id"], identitySnapshot["spells"]["prompt_version"],
identitySnapshot["spells"]["response_schema_key"], identitySnapshot["spells"]["response_schema_id"], identitySnapshot["spells"]["response_schema_name"], identitySnapshot["spells"]["response_schema_version"],
}; !reflect.DeepEqual(got, []any{"dnd.spells", "v1", "dnd_spells", "notarius.dnd.spells", "notarius_dnd_spells_v1", "v1"}) {
t.Fatalf("spell identities = %#v, want compatibility snapshot", got)
}
fileConfig, err := config.LoadFileConfig(fixturePath(t, "examples/dnd-spells.config.yml"))
if err != nil {
t.Fatalf("LoadFileConfig() error = %v, want nil", err)
}
cfg := config.Default()
if err := cfg.ApplyFileConfig(fileConfig); err != nil {
t.Fatalf("ApplyFileConfig() error = %v, want nil", err)
}
effective, err := cfg.Resolve(config.ResolveInput{PipelineID: "dnd-session", Catalog: catalogFromRegistries(registries)})
if err != nil {
t.Fatalf("Resolve() error = %v, want nil", err)
}
resolved := effective.ResolvedPipeline
if resolved.Input.Module != "seriatim" || resolved.Chunk.Module != "generic" || resolved.Output.Module != "json" || len(resolved.ArtifactLanes) != 1 {
t.Fatalf("resolved example = %#v, want maintained production topology", resolved)
}
lane := resolved.ArtifactLanes[0]
if lane.ID != "spells" || lane.Extract.Module != "dnd/spells" || lane.Merge.Module != "appendorder" || lane.Normalize.Module != "noop" {
t.Fatalf("resolved lane = %#v, want maintained spell lane", lane)
}
if got := resolvedValidatorKeys(resolved.ValidatorChains, pipeline.StageExtract, "spells", spells.Key); !reflect.DeepEqual(got, []string{
"generic/valid_json", "generic/valid_json_schema", "extract/dnd/spells/shape", "extract/dnd/spells/source_refs", "extract/dnd/spells/source_relatedness",
}) {
t.Fatalf("resolved validator keys = %#v, want compatibility snapshot", got)
}
productionFileConfig, err := config.LoadFileConfig(fixturePath(t, "examples/dnd-spells-production.config.yml"))
if err != nil {
t.Fatalf("LoadFileConfig(production example) error = %v, want nil", err)
}
productionConfig := config.Default()
if err := productionConfig.ApplyFileConfig(productionFileConfig); err != nil {
t.Fatalf("ApplyFileConfig(production example) error = %v, want nil", err)
}
productionEffective, err := productionConfig.Resolve(config.ResolveInput{PipelineID: "dnd-session", Catalog: catalogFromRegistries(registries)})
if err != nil {
t.Fatalf("Resolve(production example) error = %v, want nil", err)
}
productionResolved := productionEffective.ResolvedPipeline
if productionConfig.Concurrency.TotalLLM != 1 || productionConfig.Concurrency.StageWorkers["extract"] != 1 || !reflect.DeepEqual(productionResolved.Chunk.Options, map[string]any{"max_units": 50}) {
t.Fatalf("production example concurrency/options = %#v/%#v, want compatibility snapshot", productionConfig.Concurrency, productionResolved.Chunk.Options)
}
bindings := productionResolved.ArtifactLanes[0].ExtractReferences.Bindings
if len(bindings) != 2 || bindings[0].SlotName != "glossary" || bindings[0].Source != "./dnd-spells-glossary.txt" || bindings[1].SlotName != "party" || bindings[1].Source != "./dnd-spells-roster.txt" {
t.Fatalf("production example reference bindings = %#v, want maintained glossary and party bindings", bindings)
}
}
func TestMaintainedSeriatimToDNDCompatibilityBundle(t *testing.T) {
tests := []struct {
name string
client contracts.StructuredLLMClient
wantStatus string
wantLaneFile bool
wantRejectedCount int
}{
{name: "approved", client: newFakeRunLLMClient(false), wantStatus: "approved", wantLaneFile: true},
{name: "validator rejection is nonfatal", client: newFakeRunLLMClient(true), wantStatus: "rejected", wantRejectedCount: 1},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
outputDir := t.TempDir()
var stdout bytes.Buffer
var stderr bytes.Buffer
code := RunWithOptions([]string{
"run", "dnd-session",
"--config", fixturePath(t, "examples/dnd-spells.config.yml"),
"--input", fixturePath(t, "examples/seriatim-minimal-transcript.json"),
"--output-dir", outputDir,
"--diagnostics-dir", t.TempDir(),
}, &stdout, &stderr, Options{LLMClientFactory: fakeLLMFactory(test.client, nil)})
if code != 0 {
t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String())
}
runDir := onlyChildDir(t, outputDir)
wantFiles := []string{"index.json", "manifest.json", "rejected.json", "warnings.json"}
if test.wantLaneFile {
wantFiles = append(wantFiles, "lanes/spells.json")
}
sort.Strings(wantFiles)
if got := relativeFileNames(t, runDir); !reflect.DeepEqual(got, wantFiles) {
t.Fatalf("durable files = %#v, want compatibility snapshot %#v", got, wantFiles)
}
var manifest artifacts.RunManifest
readJSONFile(t, filepath.Join(runDir, "manifest.json"), &manifest)
if manifest.PipelineID != "dnd-session" || manifest.InputModule != "seriatim" || manifest.Chunker != "generic" || manifest.OutputEncoder != "json" {
t.Fatalf("manifest module provenance = %#v, want maintained production modules", manifest)
}
if len(manifest.ArtifactLanes) != 1 || manifest.ArtifactLanes[0].ID != "spells" {
t.Fatalf("artifact lanes = %#v, want one spells lane", manifest.ArtifactLanes)
}
laneManifest := manifest.ArtifactLanes[0]
if laneManifest.Extractor != "dnd/spells" || laneManifest.Merger != "appendorder" || laneManifest.Normalizer != "noop" {
t.Fatalf("manifest lane module provenance = %#v, want maintained production modules", laneManifest)
}
if len(manifest.Extractors) != 0 || manifest.Merger != "" || manifest.Normalizer != "" {
t.Fatalf("legacy top-level lane summaries = %#v/%q/%q, want empty compatibility snapshot", manifest.Extractors, manifest.Merger, manifest.Normalizer)
}
if manifest.ValidationStatus != test.wantStatus || len(manifest.RejectedOutputs) != test.wantRejectedCount {
t.Fatalf("manifest outcome = status %q rejected %#v, want %q/%d", manifest.ValidationStatus, manifest.RejectedOutputs, test.wantStatus, test.wantRejectedCount)
}
if !reflect.DeepEqual(manifest.SourceDigests, []string{"sha256:1c98d94ae632fb10a2b56f684cd4fb1019cedb1a629e57dc0977cf4a54135be0"}) {
t.Fatalf("source digests = %#v, want maintained fixture provenance", manifest.SourceDigests)
}
if got := manifestValidatorKeys(manifestValidatorChain(t, manifest, pipeline.StageExtract, "spells", spells.Key)); !reflect.DeepEqual(got, []string{
"generic/valid_json", "generic/valid_json_schema", "extract/dnd/spells/shape", "extract/dnd/spells/source_refs", "extract/dnd/spells/source_relatedness",
}) {
t.Fatalf("manifest validator chain = %#v, want compatibility snapshot", got)
}
var index struct {
ManifestFile string `json:"manifest_file"`
OutputFiles []struct {
LaneID string `json:"lane_id"`
MediaType string `json:"media_type"`
File string `json:"file"`
ModuleKey string `json:"module_key"`
SchemaID string `json:"schema_id"`
SchemaName string `json:"schema_name"`
SchemaVersion string `json:"schema_version"`
} `json:"output_files"`
RejectedFile string `json:"rejected_file"`
WarningsFile string `json:"warnings_file"`
}
readJSONFile(t, filepath.Join(runDir, "index.json"), &index)
if index.ManifestFile != "manifest.json" || index.RejectedFile != "rejected.json" || index.WarningsFile != "warnings.json" {
t.Fatalf("output index fixed files = %#v, want compatibility snapshot", index)
}
if test.wantLaneFile {
if len(index.OutputFiles) != 1 {
t.Fatalf("output index entries = %#v, want one", index.OutputFiles)
}
wantOutput := struct {
LaneID, MediaType, File, ModuleKey, SchemaID, SchemaName, SchemaVersion string
}{"spells", "application/json", "lanes/spells.json", "noop", "notarius.dnd.spells", "notarius_dnd_spells_v1", "v1"}
gotOutput := index.OutputFiles[0]
got := struct {
LaneID, MediaType, File, ModuleKey, SchemaID, SchemaName, SchemaVersion string
}{gotOutput.LaneID, gotOutput.MediaType, gotOutput.File, gotOutput.ModuleKey, gotOutput.SchemaID, gotOutput.SchemaName, gotOutput.SchemaVersion}
if got != wantOutput {
t.Fatalf("output index entries = %#v, want compatibility snapshot %#v", index.OutputFiles, wantOutput)
}
assertJSONEqual(t, readFile(t, filepath.Join(runDir, "lanes/spells.json")), []byte(`{
"spell_casts": [{
"caster": "Aria",
"spell": "Cure Wounds",
"effect": "Heals a wounded ally.",
"narrative_description": "Aria casts Cure Wounds.",
"source_refs": [{"source_id": "session-alpha", "start_unit_id": 1, "end_unit_id": 1}]
}]
}`))
} else if len(index.OutputFiles) != 0 {
t.Fatalf("output index entries = %#v, want none for rejected lane", index.OutputFiles)
}
var warnings struct {
Warnings []contracts.Warning `json:"warnings"`
}
readJSONFile(t, filepath.Join(runDir, "warnings.json"), &warnings)
if len(warnings.Warnings) != 0 {
t.Fatalf("warnings = %#v, want empty compatibility snapshot", warnings.Warnings)
}
var rejected struct {
Rejected []contracts.RejectedOutput `json:"rejected"`
}
readJSONFile(t, filepath.Join(runDir, "rejected.json"), &rejected)
if len(rejected.Rejected) != test.wantRejectedCount {
t.Fatalf("rejected outputs = %#v, want %d", rejected.Rejected, test.wantRejectedCount)
}
if test.wantRejectedCount == 1 {
got := rejected.Rejected[0]
if got.Stage != "extract" || got.LaneID != "spells" || got.ModuleKey != "dnd/spells" || got.ChunkID != "chunk-000001" || got.ChunkIndex != 0 || got.ValidatorName != "extract/dnd/spells/source_refs" || got.ReasonCode != "invalid_source_refs" || got.AttemptCount != 1 {
t.Fatalf("rejection = %#v, want maintained nonfatal validator outcome", got)
}
}
})
}
}
func TestProductionLLMCallersShareScheduledClient(t *testing.T) {
underlying := newBlockingProductionLLMClient()
scheduler, err := frameworkllm.NewScheduler(1)
if err != nil {
t.Fatalf("NewScheduler() error = %v, want nil", err)
}
client := frameworkllm.NewScheduledClient(underlying, scheduler)
doc := &source.SourceDocument{
ID: "session-alpha", Kind: "transcript", Format: "application/json", Digest: "sha256:source",
Units: []source.SourceUnit{
{ID: 1, Kind: "segment", Text: "Aria casts Cure Wounds.", Ref: source.SourceRef{SourceID: "session-alpha", StartUnitID: 1, EndUnitID: 1}},
{ID: 2, Kind: "segment", Text: "The spell takes effect.", Ref: source.SourceRef{SourceID: "session-alpha", StartUnitID: 2, EndUnitID: 2}},
},
}
chunk := source.Chunk{
ID: "session-alpha:chunk:0", SourceID: doc.ID, Index: 0, Ref: source.SourceRef{SourceID: doc.ID, StartUnitID: 1, EndUnitID: 2},
Content: []byte(`{"scene":"Aria casts Cure Wounds."}`), MediaType: "application/json", Units: append([]source.SourceUnit(nil), doc.Units...),
}
var started sync.WaitGroup
started.Add(2)
errs := make(chan error, 2)
go func() {
started.Done()
chunker, err := scenes.New(client, scenes.Options{})
if err == nil {
_, err = chunker.Plan(context.Background(), contracts.ChunkRequest{Source: doc})
}
errs <- err
}()
go func() {
started.Done()
extractor, err := spells.New(client, spells.Options{})
if err == nil {
_, err = extractor.Extract(context.Background(), contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk})
}
errs <- err
}()
started.Wait()
for i := 0; i < 2; i++ {
<-underlying.entered
underlying.release <- struct{}{}
}
for i := 0; i < 2; i++ {
if err := <-errs; err != nil {
t.Fatalf("production LLM caller error = %v, want nil", err)
}
}
if underlying.maxActive != 1 {
t.Fatalf("maximum concurrent provider calls = %d, want total_llm limit 1", underlying.maxActive)
}
sort.Strings(underlying.stageNames)
if !reflect.DeepEqual(underlying.stageNames, []string{"dnd/scenes", "dnd/spells"}) {
t.Fatalf("scheduled stage names = %#v, want both production LLM callers", underlying.stageNames)
}
}
func TestProductionBundlePreservesLaneAndChunkOrder(t *testing.T) {
configPath := writeTestConfig(t, `version: 2
pipelines:
dnd-session:
input: seriatim
chunk:
module: generic
options:
max_units: 1
artifacts:
zeta:
extract: dnd/spells
alpha:
extract: dnd/spells
`)
outputDir := t.TempDir()
var stdout bytes.Buffer
var stderr bytes.Buffer
code := RunWithOptions([]string{
"run", "dnd-session",
"--config", configPath,
"--input", fixturePath(t, "examples/seriatim-minimal-transcript.json"),
"--output-dir", outputDir,
"--diagnostics-dir", t.TempDir(),
}, &stdout, &stderr, Options{LLMClientFactory: fakeLLMFactory(orderingProductionLLMClient{}, nil)})
if code != 0 {
t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String())
}
runDir := onlyChildDir(t, outputDir)
var index struct {
OutputFiles []struct {
LaneID string `json:"lane_id"`
} `json:"output_files"`
}
readJSONFile(t, filepath.Join(runDir, "index.json"), &index)
if len(index.OutputFiles) != 2 {
t.Fatalf("output index entries = %#v, want two lanes", index.OutputFiles)
}
if got := []string{index.OutputFiles[0].LaneID, index.OutputFiles[1].LaneID}; !reflect.DeepEqual(got, []string{"alpha", "zeta"}) {
t.Fatalf("output lane order = %#v, want resolved lane order", got)
}
for _, laneID := range []string{"alpha", "zeta"} {
var payload struct {
SpellCasts []struct {
Spell string `json:"spell"`
SourceRefs []source.SourceRef `json:"source_refs"`
} `json:"spell_casts"`
}
readJSONFile(t, filepath.Join(runDir, "lanes", laneID+".json"), &payload)
if len(payload.SpellCasts) != 2 {
t.Fatalf("lane %q spell casts = %#v, want one per source chunk", laneID, payload.SpellCasts)
}
got := []any{
payload.SpellCasts[0].Spell, payload.SpellCasts[0].SourceRefs[0].StartUnitID,
payload.SpellCasts[1].Spell, payload.SpellCasts[1].SourceRefs[0].StartUnitID,
}
if !reflect.DeepEqual(got, []any{"Cure Wounds", 1, "Shield", 2}) {
t.Fatalf("lane %q chunk handoff order = %#v, want source chunk order", laneID, got)
}
}
}
type blockingProductionLLMClient struct {
mu sync.Mutex
active int
maxActive int
stageNames []string
entered chan struct{}
release chan struct{}
}
type orderingProductionLLMClient struct{}
func (orderingProductionLLMClient) CompleteStructured(_ context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
material := req.Inputs["transcript"]
unitID := 1
spellName := "Cure Wounds"
if strings.Contains(string(material.Content), "Shield") {
unitID = 2
spellName = "Shield"
}
payload, err := json.Marshal(map[string]any{
"spell_casts": []map[string]any{{
"caster": "Aria",
"spell": spellName,
"effect": "Fixture effect.",
"narrative_description": "Fixture spell cast.",
"source_refs": []map[string]any{{
"start_unit_id": unitID,
"end_unit_id": unitID,
}},
}},
})
if err != nil {
return contracts.StructuredCompletionResponse{}, err
}
if err := json.Unmarshal(payload, out); err != nil {
return contracts.StructuredCompletionResponse{}, err
}
return contracts.StructuredCompletionResponse{Content: payload}, nil
}
func newBlockingProductionLLMClient() *blockingProductionLLMClient {
return &blockingProductionLLMClient{entered: make(chan struct{}, 2), release: make(chan struct{}, 2)}
}
func (client *blockingProductionLLMClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
client.mu.Lock()
client.active++
if client.active > client.maxActive {
client.maxActive = client.active
}
client.stageNames = append(client.stageNames, req.StageName)
client.mu.Unlock()
client.entered <- struct{}{}
select {
case <-ctx.Done():
return contracts.StructuredCompletionResponse{}, ctx.Err()
case <-client.release:
}
client.mu.Lock()
client.active--
client.mu.Unlock()
var payload []byte
switch req.StageName {
case scenes.Key:
payload = []byte(`{"scenes":[{"start_unit_id":1,"end_unit_id":2,"short_title":"Spell","primary_mode":"Narrative","main_participants":["Aria"],"summary":"Aria casts a spell.","boundary_note":"Complete source.","boundary_confidence":"High"}],"boundary_caveats":[]}`)
case spells.Key:
payload = []byte(`{"spell_casts":[{"caster":"Aria","spell":"Cure Wounds","effect":"Healing","narrative_description":"Aria casts Cure Wounds.","source_refs":[{"start_unit_id":1,"end_unit_id":1}]}]}`)
}
if err := json.Unmarshal(payload, out); err != nil {
return contracts.StructuredCompletionResponse{}, err
}
return contracts.StructuredCompletionResponse{Content: payload}, nil
}
func sceneManifestMetadata(t *testing.T) map[string]any {
t.Helper()
chunker, err := scenes.New(orderingProductionLLMClient{}, scenes.Options{})
if err != nil {
t.Fatalf("construct scene chunker: %v", err)
}
return chunker.ManifestMetadata()
}
func spellManifestMetadata(t *testing.T) map[string]any {
t.Helper()
extractor, err := spells.New(orderingProductionLLMClient{}, spells.Options{})
if err != nil {
t.Fatalf("construct spell extractor: %v", err)
}
return extractor.ManifestMetadata()
}
func assertAssetNames(t *testing.T, getFS func() (fs.FS, error), want []string) {
t.Helper()
fSys, err := getFS()
if err != nil {
t.Fatalf("asset filesystem error = %v, want nil", err)
}
var got []string
if err := fs.WalkDir(fSys, ".", func(path string, entry fs.DirEntry, err error) error {
if err == nil && !entry.IsDir() {
got = append(got, path)
}
return err
}); err != nil {
t.Fatalf("walk assets: %v", err)
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("asset names = %#v, want compatibility snapshot %#v", got, want)
}
}
func resolvedValidatorKeys(chains []pipeline.ResolvedValidatorChain, stage pipeline.ModuleStage, laneID, moduleKey string) []string {
for _, chain := range chains {
if chain.Stage == stage && chain.LaneID == laneID && chain.ModuleKey == moduleKey {
keys := make([]string, 0, len(chain.Validators))
for _, validator := range chain.Validators {
keys = append(keys, validator.Binding.Module)
}
return keys
}
}
return nil
}
func relativeFileNames(t *testing.T, root string) []string {
t.Helper()
var names []string
if err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, err error) error {
if err != nil || entry.IsDir() {
return err
}
rel, err := filepath.Rel(root, path)
if err != nil {
return err
}
names = append(names, filepath.ToSlash(rel))
return nil
}); err != nil {
t.Fatalf("walk durable output: %v", err)
}
sort.Strings(names)
return names
}
func assertJSONEqual(t *testing.T, got, want []byte) {
t.Helper()
var gotValue any
var wantValue any
if err := json.Unmarshal(got, &gotValue); err != nil {
t.Fatalf("unmarshal actual JSON: %v", err)
}
if err := json.Unmarshal(want, &wantValue); err != nil {
t.Fatalf("unmarshal expected JSON: %v", err)
}
if !reflect.DeepEqual(gotValue, wantValue) {
t.Fatalf("JSON = %#v, want compatibility snapshot %#v", gotValue, wantValue)
}
}

4258
internal/cli/run_test.go
View File

File diff suppressed because it is too large Load Diff

10
internal/cli/test_main_test.go
View File

@@ -1,10 +0,0 @@
package cli
import (
"os"
"testing"
)
func TestMain(m *testing.M) {
os.Exit(m.Run())
}

128
internal/core/config/chunk_cache_test.go
View File

@@ -1,128 +0,0 @@
//go:build legacy
package config
import (
"path/filepath"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
func TestChunkCacheDefaults(t *testing.T) {
cfg := Default()
if cfg.Workspace.ChunkCache.Mode != pipeline.ChunkCacheAuto || cfg.Workspace.ChunkCache.Directory != "" {
t.Fatalf("chunk cache defaults = %#v", cfg.Workspace.ChunkCache)
}
if err := cfg.Validate(); err != nil {
t.Fatalf("Validate() error = %v", err)
}
}
func TestChunkCacheFileConfiguration(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
workspace:
chunk_cache:
mode: refresh
directory: " ./state/../plans "
`)
if cfg.Workspace.ChunkCache.Mode != pipeline.ChunkCacheRefresh {
t.Fatalf("mode = %q", cfg.Workspace.ChunkCache.Mode)
}
if got, want := cfg.Workspace.ChunkCache.Directory, filepath.Clean("./state/../plans"); got != want {
t.Fatalf("directory = %q, want %q", got, want)
}
if err := cfg.Validate(); err != nil {
t.Fatalf("Validate() error = %v", err)
}
}
func TestChunkCacheEnvironmentOverridesFile(t *testing.T) {
fileCfg, err := ParseFileConfigYAML([]byte(`
version: 2
workspace:
chunk_cache:
mode: bypass
directory: /file/plans
`))
if err != nil {
t.Fatal(err)
}
cfg := Default()
if err := cfg.ApplyFileConfigWithLookup(fileCfg, emptyLookup); err != nil {
t.Fatal(err)
}
if err := cfg.ApplyEnvOverridesWithLookup(mapLookup(map[string]string{
"NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE": "refresh",
"NOTARIUS_WORKSPACE_CHUNK_CACHE_DIR": " /environment/../cache/plans ",
})); err != nil {
t.Fatal(err)
}
if cfg.Workspace.ChunkCache.Mode != pipeline.ChunkCacheRefresh || cfg.Workspace.ChunkCache.Directory != filepath.Clean("/environment/../cache/plans") {
t.Fatalf("effective chunk cache = %#v", cfg.Workspace.ChunkCache)
}
}
func TestChunkCacheEmptyDirectoryEnvironmentSelectsDefault(t *testing.T) {
cfg := Default()
cfg.Workspace.ChunkCache.Directory = "/file/plans"
if err := cfg.ApplyEnvOverridesWithLookup(mapLookup(map[string]string{"NOTARIUS_WORKSPACE_CHUNK_CACHE_DIR": " \t "})); err != nil {
t.Fatal(err)
}
if cfg.Workspace.ChunkCache.Directory != "" {
t.Fatalf("directory = %q, want unset", cfg.Workspace.ChunkCache.Directory)
}
}
func TestChunkCacheRejectsInvalidSuppliedModes(t *testing.T) {
fileCfg, err := ParseFileConfigYAML([]byte("version: 2\nworkspace:\n chunk_cache:\n mode: sometimes\n"))
if err != nil {
t.Fatal(err)
}
cfg := Default()
if err := cfg.ApplyFileConfigWithLookup(fileCfg, emptyLookup); err == nil || !strings.Contains(err.Error(), "workspace.chunk_cache.mode") {
t.Fatalf("file mode error = %v", err)
}
cfg = Default()
if err := cfg.ApplyEnvOverridesWithLookup(mapLookup(map[string]string{"NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE": "sometimes"})); err == nil || !strings.Contains(err.Error(), "NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE") {
t.Fatalf("environment mode error = %v", err)
}
cfg = Default()
cfg.Workspace.ChunkCache.Mode = "sometimes"
if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "chunk cache") {
t.Fatalf("Validate() error = %v", err)
}
}
func TestChunkCacheConfigurationClonesAndRedacts(t *testing.T) {
cfg := Default()
cfg.Workspace.ChunkCache = WorkspaceChunkCacheConfig{Mode: pipeline.ChunkCacheRefresh, Directory: "/var/cache/notarius/chunk-plans"}
cloned := cloneConfig(cfg)
redacted := cfg.Redacted()
if cloned.Workspace.ChunkCache != cfg.Workspace.ChunkCache || redacted.Workspace.ChunkCache != cfg.Workspace.ChunkCache {
t.Fatalf("cloned=%#v redacted=%#v", cloned.Workspace.ChunkCache, redacted.Workspace.ChunkCache)
}
redacted.Workspace.ChunkCache.Directory = "/changed"
if cfg.Workspace.ChunkCache.Directory != "/var/cache/notarius/chunk-plans" {
t.Fatal("redacted mutation changed original")
}
if err := cfg.Validate(); err != nil {
t.Fatalf("Validate() error = %v", err)
}
}
func TestChunkCacheDirectoryValidation(t *testing.T) {
cfg := Default()
cfg.Workspace.ChunkCache.Directory = "/var/cache/notarius/chunk-plans"
if err := cfg.Validate(); err != nil {
t.Fatalf("Validate(system root) error = %v", err)
}
cfg.Workspace.ChunkCache.Directory = "bad\x00path"
if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "NUL") {
t.Fatalf("Validate(NUL directory) error = %v", err)
}
}

85
internal/core/config/config_test.go
View File

@@ -1,85 +0,0 @@
//go:build legacy
package config
import (
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics"
)
func TestDefaultValues(t *testing.T) {
cfg := Default()
if cfg.Scriptorium.ProfileDir != "" || cfg.Scriptorium.ProfileFile != "" {
t.Fatalf("unexpected Scriptorium profile source defaults: %+v", cfg.Scriptorium)
}
if len(cfg.Pipelines) != 0 {
t.Fatalf("expected no built-in pipeline profiles, got %v", cfg.Pipelines)
}
if cfg.Concurrency.TotalLLM != 1 {
t.Fatalf("unexpected total LLM concurrency: %d", cfg.Concurrency.TotalLLM)
}
if got := cfg.Concurrency.StageWorkers["extract"]; got != 1 {
t.Fatalf("unexpected extract workers: %d", got)
}
if cfg.Diagnostics.WorkDir != "/tmp/notarius" {
t.Fatalf("unexpected diagnostics work dir: %q", cfg.Diagnostics.WorkDir)
}
if cfg.Diagnostics.Retention != diagnostics.RetentionAuto {
t.Fatalf("unexpected diagnostics retention: %q", cfg.Diagnostics.Retention)
}
if cfg.Workspace.Directory != "" {
t.Fatalf("unexpected workspace directory: %q", cfg.Workspace.Directory)
}
if !cfg.Workspace.Diagnostics.Enabled || !cfg.DiagnosticsEnabled() {
t.Fatalf("expected workspace diagnostics enabled by default: %+v", cfg.Workspace.Diagnostics)
}
if cfg.Workspace.Diagnostics.Retention != "" {
t.Fatalf("unexpected workspace diagnostics retention: %q", cfg.Workspace.Diagnostics.Retention)
}
if cfg.Workspace.Resume.Enabled {
t.Fatalf("workspace resume should be disabled by default")
}
if cfg.Workspace.Debug.Enabled {
t.Fatalf("workspace debug should be disabled by default")
}
}
func TestApplyFileConfigMergesWithDefaults(t *testing.T) {
fileCfg, err := ParseFileConfigYAML([]byte(`
version: 2
scriptorium:
profile_dir: ./profiles
pipelines:
example:
input: fake/input
artifacts:
events:
extract: fake/extract
`))
if err != nil {
t.Fatalf("ParseFileConfigYAML: %v", err)
}
cfg := Default()
if err := cfg.applyFileConfigWithLookup(fileCfg, emptyLookup); err != nil {
t.Fatalf("ApplyFileConfig: %v", err)
}
if cfg.Scriptorium.ProfileDir != "./profiles" {
t.Fatalf("expected Scriptorium profile dir, got %+v", cfg.Scriptorium)
}
if cfg.Concurrency.TotalLLM != 1 {
t.Fatalf("expected default concurrency preserved, got %d", cfg.Concurrency.TotalLLM)
}
if got := cfg.Concurrency.StageWorkers["extract"]; got != 1 {
t.Fatalf("expected default extract workers preserved, got %d", got)
}
if cfg.Diagnostics.Retention != diagnostics.RetentionAuto {
t.Fatalf("expected default diagnostics retention preserved, got %q", cfg.Diagnostics.Retention)
}
if _, ok := cfg.Pipelines["example"]; !ok {
t.Fatalf("expected file pipeline to be applied")
}
}

243
internal/core/config/effective_config_test.go
View File

@@ -1,243 +0,0 @@
//go:build legacy
package config
import (
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
func TestResolveRejectsEmptyAndUnknownPipelineID(t *testing.T) {
tests := []struct {
name string
pipelineID string
want string
}{
{name: "empty", pipelineID: " ", want: "pipeline id"},
{name: "unknown", pipelineID: "missing", want: "not configured"},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
_, err := validConfig().Resolve(ResolveInput{PipelineID: tc.pipelineID, Catalog: fakeCatalog(t)})
if err == nil || !strings.Contains(err.Error(), tc.want) {
t.Fatalf("expected error containing %q, got %v", tc.want, err)
}
})
}
}
func TestResolveMaterializesDefaultExtractWorkersFromEffectiveTotal(t *testing.T) {
cfg := validConfig()
cfg.Concurrency.TotalLLM = 4
effective, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)})
if err != nil {
t.Fatalf("Resolve() error = %v, want nil", err)
}
if got := effective.Config.Concurrency.StageWorkers["extract"]; got != 4 {
t.Fatalf("effective extract workers = %d, want total concurrency 4", got)
}
}
func TestResolveLaneFilteringSuccessAndFailure(t *testing.T) {
effective, err := validConfig().Resolve(ResolveInput{
PipelineID: " example ",
Only: []string{" notes "},
Catalog: fakeCatalog(t),
})
if err != nil {
t.Fatalf("Resolve: %v", err)
}
if effective.PipelineID != "example" {
t.Fatalf("unexpected pipeline ID: %q", effective.PipelineID)
}
if len(effective.ResolvedPipeline.ArtifactLanes) != 1 || effective.ResolvedPipeline.ArtifactLanes[0].ID != "notes" {
t.Fatalf("unexpected resolved lanes: %+v", effective.ResolvedPipeline.ArtifactLanes)
}
if effective.ResolvedPipeline.Digest == "" {
t.Fatalf("expected digest")
}
_, err = validConfig().Resolve(ResolveInput{
PipelineID: "example",
Only: []string{"missing"},
Catalog: fakeCatalog(t),
})
if err == nil || !strings.Contains(err.Error(), "selected artifact lane") {
t.Fatalf("expected invalid lane error, got %v", err)
}
}
func TestResolveUsesTrimmedPipelineMapKeys(t *testing.T) {
cfg := validConfig()
cfg.Pipelines[" example "] = cfg.Pipelines["example"]
delete(cfg.Pipelines, "example")
effective, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)})
if err != nil {
t.Fatalf("Resolve: %v", err)
}
if effective.PipelineID != "example" {
t.Fatalf("unexpected pipeline ID: %q", effective.PipelineID)
}
}
func TestResolveSurfacesUnknownModuleKeyThroughCatalog(t *testing.T) {
cfg := validConfig()
lane := cfg.Pipelines["example"].Artifacts["events"]
lane.Extract = pipeline.Binding("missing/extract")
cfg.Pipelines["example"].Artifacts["events"] = lane
_, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)})
if err == nil || !strings.Contains(err.Error(), "missing/extract") || !strings.Contains(err.Error(), "events") {
t.Fatalf("expected unknown module error with lane context, got %v", err)
}
}
func TestResolveSurfacesMissingCapabilityThroughCatalog(t *testing.T) {
_, err := validConfig().Resolve(ResolveInput{
PipelineID: "example",
Catalog: fakeCatalog(t, pipeline.ModuleSpec{
Key: "json",
Stage: pipeline.StageOutput,
Requires: []string{"missing-capability"},
}),
})
if err == nil || !strings.Contains(err.Error(), "missing capability") || !strings.Contains(err.Error(), "json") {
t.Fatalf("expected missing capability error, got %v", err)
}
}
func TestResolveCanBindSceneChunkerFromCatalog(t *testing.T) {
cfg := validConfig()
profile := cfg.Pipelines["example"]
profile.Chunk = pipeline.Binding("dnd/scenes")
lane := profile.Artifacts["events"]
profile.Artifacts = map[string]pipeline.ArtifactLaneProfile{"events": lane}
cfg.Pipelines["example"] = profile
catalog := fakeCatalog(t,
pipeline.ModuleSpec{
Key: "fake/input",
Stage: pipeline.StageInput,
Provides: []string{"source.transcript"},
},
pipeline.ModuleSpec{
Key: "fake/extract",
Stage: pipeline.StageExtract,
Requires: []string{"chunks", "source.transcript"},
Provides: []string{"artifact"},
},
)
mustRegisterChunker(t, catalog.Chunkers, pipeline.ModuleSpec{
Key: "dnd/scenes",
Stage: pipeline.StageChunk,
Requires: []string{"source.transcript"},
Provides: []string{"chunks"},
})
effective, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: catalog})
if err != nil {
t.Fatalf("Resolve() error = %v, want nil", err)
}
if got := effective.ResolvedPipeline.Chunk.Module; got != "dnd/scenes" {
t.Fatalf("Chunk.Module = %q, want dnd/scenes", got)
}
}
func TestResolveDigestChangesWhenEffectiveConfigChanges(t *testing.T) {
cfg := validConfig()
first, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)})
if err != nil {
t.Fatalf("Resolve first: %v", err)
}
lane := cfg.Pipelines["example"].Artifacts["events"]
lane.Extract.Options = map[string]any{"temperature": 0.2}
cfg.Pipelines["example"].Artifacts["events"] = lane
second, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)})
if err != nil {
t.Fatalf("Resolve second: %v", err)
}
if first.ResolvedPipeline.Digest == second.ResolvedPipeline.Digest {
t.Fatalf("expected digest to change, got %q", first.ResolvedPipeline.Digest)
}
}
func TestResolveLLMProfileOverrideAppliesBeforeDigest(t *testing.T) {
cfg := validConfig()
profile := cfg.Pipelines["example"]
profile.Input.LLMProfile = "input-profile"
profile.Output.LLMProfile = "output-profile"
lane := profile.Artifacts["events"]
lane.Merge.LLMProfile = "merge-profile"
lane.Extract.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{
{Module: "fake/llm-validator", LLMProfile: "validator-profile"},
},
}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
base, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)})
if err != nil {
t.Fatalf("Resolve base: %v", err)
}
effective, err := cfg.Resolve(ResolveInput{
PipelineID: "example",
Catalog: fakeCatalog(t),
LLMProfileOverride: "runtime",
})
if err != nil {
t.Fatalf("Resolve override: %v", err)
}
if base.ResolvedPipeline.Digest == effective.ResolvedPipeline.Digest {
t.Fatalf("expected digest to change after LLM profile override")
}
for _, binding := range llmCapableBindings(effective.ResolvedPipeline) {
if binding.LLMProfile != "runtime" {
t.Fatalf("LLM-capable binding profile = %q, want runtime", binding.LLMProfile)
}
}
if effective.ResolvedPipeline.Input.LLMProfile != "input-profile" {
t.Fatalf("input profile = %q, want original input-profile", effective.ResolvedPipeline.Input.LLMProfile)
}
if effective.ResolvedPipeline.Output.LLMProfile != "output-profile" {
t.Fatalf("output profile = %q, want original output-profile", effective.ResolvedPipeline.Output.LLMProfile)
}
eventLane := effective.ResolvedPipeline.ArtifactLanes[0]
if eventLane.Merge.LLMProfile != "runtime" {
t.Fatalf("merge profile = %q, want runtime", eventLane.Merge.LLMProfile)
}
validatorChain := findEffectiveValidatorChain(effective.ResolvedPipeline.ValidatorChains, pipeline.StageExtract, "events", "fake/extract")
if validatorChain == nil || len(validatorChain.Validators) != 1 {
t.Fatalf("validator chain = %#v, want one extract validator", effective.ResolvedPipeline.ValidatorChains)
}
if validatorChain.Validators[0].Binding.LLMProfile != "validator-profile" {
t.Fatalf("validator profile = %q, want original validator-profile", validatorChain.Validators[0].Binding.LLMProfile)
}
}
func findEffectiveValidatorChain(chains []pipeline.ResolvedValidatorChain, stage pipeline.ModuleStage, laneID string, module string) *pipeline.ResolvedValidatorChain {
for i := range chains {
if chains[i].Stage == stage && chains[i].LaneID == laneID && chains[i].ModuleKey == module {
return &chains[i]
}
}
return nil
}
func llmCapableBindings(resolved pipeline.ResolvedPipeline) []pipeline.ModuleBinding {
bindings := []pipeline.ModuleBinding{resolved.Chunk}
for _, lane := range resolved.ArtifactLanes {
bindings = append(bindings, lane.Extract, lane.Merge, lane.Normalize)
}
return bindings
}

185
internal/core/config/env_test.go
View File

@@ -1,185 +0,0 @@
//go:build legacy
package config
import (
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
func TestApplyEnvOverridesOperationalValues(t *testing.T) {
cfg := Default()
cfg.Pipelines["example"] = pipeline.PipelineProfile{ID: "example", Input: pipeline.Binding("before")}
err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{
"NOTARIUS_TOTAL_LLM_CONCURRENCY": "3",
"NOTARIUS_STAGE_WORKERS_EXTRACT": "2",
"NOTARIUS_WORK_DIR": "/tmp/notarius-env",
"NOTARIUS_DIAGNOSTICS_RETENTION": "never",
"NOTARIUS_WORKSPACE_DIR": "/var/lib/notarius-env",
"NOTARIUS_WORKSPACE_DIAGNOSTICS_ENABLED": "false",
"NOTARIUS_WORKSPACE_DIAGNOSTICS_RETENTION": "always",
"NOTARIUS_WORKSPACE_RESUME_ENABLED": "true",
"NOTARIUS_WORKSPACE_DEBUG_ENABLED": "true",
"NOTARIUS_PIPELINE_INPUT": "after",
}))
if err != nil {
t.Fatalf("ApplyEnvOverrides: %v", err)
}
if cfg.Scriptorium.ProfileDir != "" || cfg.Scriptorium.ProfileFile != "" {
t.Fatalf("LLM environment overrides must not change Scriptorium config: %+v", cfg.Scriptorium)
}
if cfg.Concurrency.TotalLLM != 3 {
t.Fatalf("unexpected total concurrency: %d", cfg.Concurrency.TotalLLM)
}
if got := cfg.Concurrency.StageWorkers["extract"]; got != 2 {
t.Fatalf("extract workers = %d, want 2", got)
}
if cfg.Workspace.Directory != "/var/lib/notarius-env" {
t.Fatalf("unexpected workspace directory: %q", cfg.Workspace.Directory)
}
if cfg.DiagnosticsEnabled() {
t.Fatalf("expected workspace diagnostics disabled")
}
if cfg.Diagnostics.WorkDir != "/var/lib/notarius-env/diagnostics" || cfg.Diagnostics.Retention != diagnostics.RetentionAlways {
t.Fatalf("unexpected diagnostics config: %+v", cfg.Diagnostics)
}
if !cfg.Workspace.Resume.Enabled {
t.Fatalf("expected workspace resume enabled")
}
if !cfg.Workspace.Debug.Enabled {
t.Fatalf("expected workspace debug enabled")
}
if cfg.Pipelines["example"].Input.Module != "before" {
t.Fatalf("environment overrides must not change pipeline wiring: %+v", cfg.Pipelines["example"])
}
}
func TestApplyEnvOverridesRejectsInvalidIntegers(t *testing.T) {
for _, name := range []string{"NOTARIUS_TOTAL_LLM_CONCURRENCY", "NOTARIUS_STAGE_WORKERS_EXTRACT"} {
t.Run(name, func(t *testing.T) {
cfg := Default()
err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{name: "many"}))
if err == nil || !strings.Contains(err.Error(), name) {
t.Fatalf("expected named integer error, got %v", err)
}
})
}
}
func TestStageWorkerEnvironmentPrecedenceAndDefaulting(t *testing.T) {
fileCfg, err := ParseFileConfigYAML([]byte(`
version: 2
concurrency:
total_llm: 4
stage_workers:
extract: 2
`))
if err != nil {
t.Fatalf("ParseFileConfigYAML() error = %v", err)
}
cfg := Default()
if err := cfg.applyFileConfigWithLookup(fileCfg, emptyLookup); err != nil {
t.Fatalf("ApplyFileConfig() error = %v", err)
}
if err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{
"NOTARIUS_TOTAL_LLM_CONCURRENCY": "5",
"NOTARIUS_STAGE_WORKERS_EXTRACT": "3",
})); err != nil {
t.Fatalf("ApplyEnvOverrides() error = %v", err)
}
if cfg.Concurrency.TotalLLM != 5 || cfg.Concurrency.StageWorkers["extract"] != 3 {
t.Fatalf("effective concurrency = %#v, want total 5 and extract 3", cfg.Concurrency)
}
if err := cfg.Validate(); err != nil {
t.Fatalf("Validate(overridden) error = %v, want nil", err)
}
defaulted := Default()
if err := defaulted.applyEnvOverridesWithLookup(mapLookup(map[string]string{"NOTARIUS_TOTAL_LLM_CONCURRENCY": "6"})); err != nil {
t.Fatalf("ApplyEnvOverrides(defaulted) error = %v", err)
}
if got := defaulted.Concurrency.StageWorkers["extract"]; got != 6 {
t.Fatalf("defaulted extract workers = %d, want effective total 6", got)
}
}
func TestStageWorkerRangeValidationUsesFinalEnvironmentTotal(t *testing.T) {
fileCfg, err := ParseFileConfigYAML([]byte(`
version: 2
concurrency:
total_llm: 4
stage_workers:
extract: 5
`))
if err != nil {
t.Fatalf("ParseFileConfigYAML() error = %v", err)
}
cfg := Default()
if err := cfg.applyFileConfigWithLookup(fileCfg, emptyLookup); err != nil {
t.Fatalf("ApplyFileConfig() error = %v", err)
}
if err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{"NOTARIUS_TOTAL_LLM_CONCURRENCY": "6"})); err != nil {
t.Fatalf("ApplyEnvOverrides() error = %v", err)
}
if err := cfg.Validate(); err != nil {
t.Fatalf("Validate() error = %v, want final total to make extract workers valid", err)
}
}
func TestApplyEnvOverridesRejectsInvalidBooleans(t *testing.T) {
for _, name := range []string{
"NOTARIUS_WORKSPACE_DIAGNOSTICS_ENABLED",
"NOTARIUS_WORKSPACE_RESUME_ENABLED",
"NOTARIUS_WORKSPACE_DEBUG_ENABLED",
} {
t.Run(name, func(t *testing.T) {
cfg := Default()
err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{name: "maybe"}))
if err == nil || !strings.Contains(err.Error(), name) {
t.Fatalf("expected named boolean error, got %v", err)
}
})
}
}
func TestApplyEnvOverridesLegacyDiagnosticsRemainCompatibleWithoutWorkspace(t *testing.T) {
cfg := Default()
err := cfg.applyEnvOverridesWithLookup(mapLookup(map[string]string{
"NOTARIUS_WORK_DIR": "/tmp/notarius-env",
"NOTARIUS_DIAGNOSTICS_RETENTION": "never",
}))
if err != nil {
t.Fatalf("ApplyEnvOverrides: %v", err)
}
if cfg.Diagnostics.WorkDir != "/tmp/notarius-env" {
t.Fatalf("diagnostics work dir = %q, want legacy env", cfg.Diagnostics.WorkDir)
}
if cfg.Diagnostics.Retention != diagnostics.RetentionNever {
t.Fatalf("diagnostics retention = %q, want legacy env", cfg.Diagnostics.Retention)
}
}
func TestLoadFromEnvUsesDefaultConfig(t *testing.T) {
t.Setenv("NOTARIUS_TOTAL_LLM_CONCURRENCY", "2")
cfg, err := LoadFromEnv()
if err != nil {
t.Fatalf("LoadFromEnv: %v", err)
}
if cfg.Scriptorium.ProfileDir != "" || cfg.Scriptorium.ProfileFile != "" {
t.Fatalf("unexpected Scriptorium config from env: %+v", cfg.Scriptorium)
}
if cfg.Concurrency.TotalLLM != 2 {
t.Fatalf("expected env concurrency override, got %+v", cfg.Concurrency)
}
if got := cfg.Concurrency.StageWorkers["extract"]; got != 2 {
t.Fatalf("expected extract workers to default to total, got %d", got)
}
}

725
internal/core/config/file_config_test.go
View File

@@ -1,725 +0,0 @@
//go:build legacy
package config
import (
"os"
"path/filepath"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics"
)
func TestParseMinimalValidConfig(t *testing.T) {
fileCfg, err := ParseFileConfigYAML([]byte(`
version: 2
`))
if err != nil {
t.Fatalf("ParseFileConfigYAML: %v", err)
}
if fileCfg.Version != SupportedFileConfigVersion {
t.Fatalf("unexpected version: %d", fileCfg.Version)
}
}
func TestLoadFileConfig(t *testing.T) {
path := filepath.Join(t.TempDir(), "config.yml")
if err := os.WriteFile(path, []byte("version: 2\n"), 0o644); err != nil {
t.Fatalf("write config: %v", err)
}
fileCfg, err := LoadFileConfig(path)
if err != nil {
t.Fatalf("LoadFileConfig: %v", err)
}
if fileCfg.Version != SupportedFileConfigVersion {
t.Fatalf("unexpected version: %d", fileCfg.Version)
}
}
func TestParseFileConfigRejectsUnknownYAMLFields(t *testing.T) {
_, err := ParseFileConfigYAML([]byte(`
version: 2
unexpected: true
`))
if err == nil || !strings.Contains(err.Error(), "field unexpected not found") {
t.Fatalf("expected unknown field error, got %v", err)
}
}
func TestParseFileConfigRejectsUnknownModuleBindingFields(t *testing.T) {
_, err := ParseFileConfigYAML([]byte(`
version: 2
pipelines:
example:
input:
module: fake/input
unexpected: true
artifacts:
events:
extract: fake/extract
`))
if err == nil || !strings.Contains(err.Error(), "field unexpected not found") {
t.Fatalf("expected unknown binding field error, got %v", err)
}
}
func TestParseFileConfigRejectsMissingAndUnsupportedVersion(t *testing.T) {
tests := []struct {
name string
data string
want string
}{
{name: "missing", data: `scriptorium: {}`, want: "version is required"},
{name: "unsupported", data: `version: 1`, want: "unsupported config version"},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
_, err := ParseFileConfigYAML([]byte(tc.data))
if err == nil || !strings.Contains(err.Error(), tc.want) {
t.Fatalf("expected error containing %q, got %v", tc.want, err)
}
})
}
}
func TestParseFileConfigRejectsStaleLLMProfiles(t *testing.T) {
_, err := ParseFileConfigYAML([]byte(`
version: 2
llm_profiles:
default: {}
`))
if err == nil || !strings.Contains(err.Error(), "llm_profiles") {
t.Fatalf("expected stale llm_profiles error, got %v", err)
}
}
func TestParseFileConfigScriptoriumProfileSources(t *testing.T) {
t.Run("profile dir", func(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
scriptorium:
profile_dir: ./profiles
`)
if cfg.Scriptorium.ProfileDir != "./profiles" || cfg.Scriptorium.ProfileFile != "" {
t.Fatalf("Scriptorium = %+v, want profile_dir", cfg.Scriptorium)
}
})
t.Run("profile file", func(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
scriptorium:
profile_file: ./profiles.yml
`)
if cfg.Scriptorium.ProfileFile != "./profiles.yml" || cfg.Scriptorium.ProfileDir != "" {
t.Fatalf("Scriptorium = %+v, want profile_file", cfg.Scriptorium)
}
})
}
func TestParseFileConfigModuleBindingForms(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
pipelines:
example:
input: fake/input
chunk:
module: generic
retries: 2
options:
size: 10
flags:
- alpha
nested:
enabled: true
artifacts:
events:
extract:
module: fake/extract
llm_profile: fast
retries: 3
options:
temperature: 0
merge:
module: appendorder
retries: 1
normalize:
module: noop
output: json
`)
profile := cfg.Pipelines["example"]
if profile.Input.Module != "fake/input" {
t.Fatalf("unexpected input binding: %+v", profile.Input)
}
if profile.Chunk.Module != "generic" {
t.Fatalf("unexpected chunk binding: %+v", profile.Chunk)
}
if profile.Chunk.Retries != 2 {
t.Fatalf("chunk retries = %d, want 2", profile.Chunk.Retries)
}
if profile.Chunk.Options["size"] != 10 {
t.Fatalf("expected chunk options to preserve scalar, got %#v", profile.Chunk.Options)
}
if !reflect.DeepEqual(profile.Chunk.Options["flags"], []any{"alpha"}) {
t.Fatalf("expected list option, got %#v", profile.Chunk.Options["flags"])
}
nested, ok := profile.Chunk.Options["nested"].(map[string]any)
if !ok || nested["enabled"] != true {
t.Fatalf("expected nested map option, got %#v", profile.Chunk.Options["nested"])
}
lane := profile.Artifacts["events"]
if lane.Extract.Module != "fake/extract" || lane.Extract.LLMProfile != "fast" {
t.Fatalf("unexpected extract binding: %+v", lane.Extract)
}
if lane.Extract.Retries != 3 || lane.Merge.Retries != 1 {
t.Fatalf("unexpected retries: extract=%d merge=%d", lane.Extract.Retries, lane.Merge.Retries)
}
if lane.Extract.Options["temperature"] != 0 {
t.Fatalf("expected object options, got %#v", lane.Extract.Options)
}
if lane.Merge.Module != "appendorder" || lane.Normalize.Module != "noop" {
t.Fatalf("unexpected lane defaults: %+v", lane)
}
if profile.Output.Module != "json" {
t.Fatalf("unexpected output binding: %+v", profile.Output)
}
}
func TestParseFileConfigReferenceMaps(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
pipelines:
example:
input: fake/input
references:
" roster ": " ./shared-roster.yml "
artifacts:
events:
extract: fake/extract
references:
" lore ": " ./lore.md "
`)
profile := cfg.Pipelines["example"]
if !reflect.DeepEqual(profile.References, map[string]string{"roster": "./shared-roster.yml"}) {
t.Fatalf("pipeline references = %#v, want trimmed map", profile.References)
}
gotLaneRefs := profile.Artifacts["events"].References
if !reflect.DeepEqual(gotLaneRefs, map[string]string{"lore": "./lore.md"}) {
t.Fatalf("lane references = %#v, want trimmed map", gotLaneRefs)
}
}
func TestParseFileConfigStageLocalReferenceMaps(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
pipelines:
example:
input: fake/input
chunk:
module: generic
references:
" scene_guide ": " ./scenes.md "
artifacts:
events:
extract:
module: fake/extract
references:
" glossary ": " ./glossary.md "
" roster ": " ./extract-roster.yml "
references:
roster: ./legacy-roster.yml
lore: ./lore.md
merge:
module: appendorder
references:
" merge_notes ": " ./merge.md "
normalize:
module: noop
references:
" normalization_notes ": " ./normalization.md "
`)
profile := cfg.Pipelines["example"]
if !reflect.DeepEqual(profile.Chunk.References, map[string]string{"scene_guide": "./scenes.md"}) {
t.Fatalf("chunk references = %#v, want trimmed map", profile.Chunk.References)
}
lane := profile.Artifacts["events"]
if !reflect.DeepEqual(lane.References, map[string]string{"lore": "./lore.md", "roster": "./legacy-roster.yml"}) {
t.Fatalf("lane references = %#v, want trimmed map", lane.References)
}
wantExtract := map[string]string{
"glossary": "./glossary.md",
"lore": "./lore.md",
"roster": "./extract-roster.yml",
}
if !reflect.DeepEqual(lane.Extract.References, wantExtract) {
t.Fatalf("extract references = %#v, want legacy merged with extract override %#v", lane.Extract.References, wantExtract)
}
if !reflect.DeepEqual(lane.Merge.References, map[string]string{"merge_notes": "./merge.md"}) {
t.Fatalf("merge references = %#v, want trimmed map", lane.Merge.References)
}
if !reflect.DeepEqual(lane.Normalize.References, map[string]string{"normalization_notes": "./normalization.md"}) {
t.Fatalf("normalize references = %#v, want trimmed map", lane.Normalize.References)
}
}
func TestParseFileConfigValidatorMixedBindingForms(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
pipelines:
example:
input: fake/input
artifacts:
events:
extract: fake/extract
validators:
- fake/validator
- module: fake/llm-validator
llm_profile: careful
options:
threshold: 0.7
`)
validators := cfg.Pipelines["example"].Artifacts["events"].Validators
if len(validators) != 2 {
t.Fatalf("expected two validators, got %d", len(validators))
}
if validators[0].Module != "fake/validator" {
t.Fatalf("unexpected shorthand validator: %+v", validators[0])
}
if validators[1].Module != "fake/llm-validator" || validators[1].LLMProfile != "careful" {
t.Fatalf("unexpected object validator: %+v", validators[1])
}
if validators[1].Options["threshold"] != 0.7 {
t.Fatalf("unexpected validator options: %#v", validators[1].Options)
}
}
func TestParseFileConfigStageLocalValidatorOverrides(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
pipelines:
example:
input: fake/input
chunk:
module: generic
validators: []
artifacts:
events:
extract:
module: fake/extract
validators:
- fake/validator
- module: fake/llm-validator
llm_profile: careful
options:
threshold: 0.7
merge:
module: appendorder
validators: []
normalize:
module: noop
`)
profile := cfg.Pipelines["example"]
if !profile.Chunk.Validators.Set || len(profile.Chunk.Validators.Validators) != 0 {
t.Fatalf("chunk validator override = %#v, want explicit empty", profile.Chunk.Validators)
}
lane := profile.Artifacts["events"]
if !lane.Extract.Validators.Set {
t.Fatalf("extract validator override Set = false, want true")
}
validators := lane.Extract.Validators.Validators
if len(validators) != 2 {
t.Fatalf("extract validators = %#v, want two validators", validators)
}
if validators[0].Module != "fake/validator" {
t.Fatalf("first validator = %#v, want fake/validator", validators[0])
}
if validators[1].Module != "fake/llm-validator" || validators[1].LLMProfile != "careful" {
t.Fatalf("second validator = %#v, want LLM validator with profile", validators[1])
}
if validators[1].Options["threshold"] != 0.7 {
t.Fatalf("second validator options = %#v, want threshold", validators[1].Options)
}
if !lane.Merge.Validators.Set || len(lane.Merge.Validators.Validators) != 0 {
t.Fatalf("merge validator override = %#v, want explicit empty", lane.Merge.Validators)
}
if lane.Normalize.Validators.Set {
t.Fatalf("normalize validator override Set = true, want omitted")
}
}
func TestApplyFileConfigRejectsDuplicateTrimmedPipelineIDs(t *testing.T) {
fileCfg, err := ParseFileConfigYAML([]byte(`
version: 2
pipelines:
example:
input: fake/input
" example ":
input: fake/other-input
`))
if err != nil {
t.Fatalf("ParseFileConfigYAML: %v", err)
}
cfg := Default()
err = cfg.applyFileConfigWithLookup(fileCfg, emptyLookup)
if err == nil || !strings.Contains(err.Error(), "pipeline id") || !strings.Contains(err.Error(), "duplicated") {
t.Fatalf("expected duplicate pipeline ID error, got %v", err)
}
}
func TestApplyFileConfigRejectsDuplicateTrimmedArtifactLaneIDs(t *testing.T) {
fileCfg, err := ParseFileConfigYAML([]byte(`
version: 2
pipelines:
example:
input: fake/input
artifacts:
events:
extract: fake/extract
" events ":
extract: fake/other-extract
`))
if err != nil {
t.Fatalf("ParseFileConfigYAML: %v", err)
}
cfg := Default()
err = cfg.applyFileConfigWithLookup(fileCfg, emptyLookup)
if err == nil || !strings.Contains(err.Error(), `pipeline "example" artifact lane id`) || !strings.Contains(err.Error(), "duplicated") {
t.Fatalf("expected duplicate artifact lane ID error, got %v", err)
}
}
func TestApplyFileConfigRejectsDuplicateTrimmedReferenceSlots(t *testing.T) {
tests := []struct {
name string
raw string
want string
}{
{
name: "pipeline",
raw: `
version: 2
pipelines:
example:
input: fake/input
references:
roster: ./first.yml
" roster ": ./second.yml
`,
want: `pipeline "example" reference slot`,
},
{
name: "lane",
raw: `
version: 2
pipelines:
example:
input: fake/input
artifacts:
events:
extract: fake/extract
references:
roster: ./first.yml
" roster ": ./second.yml
`,
want: `pipeline "example" lane "events" reference slot`,
},
{
name: "chunk",
raw: `
version: 2
pipelines:
example:
input: fake/input
chunk:
module: generic
references:
roster: ./first.yml
" roster ": ./second.yml
`,
want: `pipeline "example" chunk reference slot`,
},
{
name: "extract",
raw: `
version: 2
pipelines:
example:
input: fake/input
artifacts:
events:
extract:
module: fake/extract
references:
roster: ./first.yml
" roster ": ./second.yml
`,
want: `pipeline "example" lane "events" extract reference slot`,
},
{
name: "normalize",
raw: `
version: 2
pipelines:
example:
input: fake/input
artifacts:
events:
extract: fake/extract
normalize:
module: noop
references:
roster: ./first.yml
" roster ": ./second.yml
`,
want: `pipeline "example" lane "events" normalize reference slot`,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
fileCfg, err := ParseFileConfigYAML([]byte(tc.raw))
if err != nil {
t.Fatalf("ParseFileConfigYAML: %v", err)
}
cfg := Default()
err = cfg.applyFileConfigWithLookup(fileCfg, emptyLookup)
if err == nil || !strings.Contains(err.Error(), tc.want) || !strings.Contains(err.Error(), "duplicated") {
t.Fatalf("expected duplicate reference slot error, got %v", err)
}
})
}
}
func TestApplyFileConfigRejectsInvalidScriptoriumSources(t *testing.T) {
tests := []struct {
name string
raw string
want string
}{
{name: "empty profile dir", raw: "profile_dir: ' '", want: "profile_dir"},
{name: "empty profile file", raw: "profile_file: ' '", want: "profile_file"},
{name: "both sources", raw: "profile_dir: ./profiles\n profile_file: ./profiles.yml", want: "mutually exclusive"},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
cfg := Default()
fileCfg, err := ParseFileConfigYAML([]byte(`
version: 2
scriptorium:
` + tc.raw + `
`))
if err != nil {
t.Fatalf("ParseFileConfigYAML: %v", err)
}
err = cfg.applyFileConfigWithLookup(fileCfg, emptyLookup)
if err == nil {
err = cfg.Validate()
}
if err == nil || !strings.Contains(err.Error(), tc.want) {
t.Fatalf("expected error containing %q, got %v", tc.want, err)
}
})
}
}
func TestApplyFileConfigOperationalSections(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
concurrency:
total_llm: 4
diagnostics:
work_dir: /tmp/notarius-test
retention: always
`)
if cfg.Concurrency.TotalLLM != 4 {
t.Fatalf("unexpected total concurrency: %d", cfg.Concurrency.TotalLLM)
}
if got := cfg.Concurrency.StageWorkers["extract"]; got != 4 {
t.Fatalf("default extract workers = %d, want total concurrency", got)
}
if cfg.Diagnostics.WorkDir != "/tmp/notarius-test" {
t.Fatalf("unexpected work dir: %q", cfg.Diagnostics.WorkDir)
}
if cfg.Diagnostics.Retention != diagnostics.RetentionAlways {
t.Fatalf("unexpected retention: %q", cfg.Diagnostics.Retention)
}
}
func TestApplyFileConfigStageWorkers(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
concurrency:
total_llm: 4
stage_workers:
extract: 3
`)
if got := cfg.Concurrency.StageWorkers["extract"]; got != 3 {
t.Fatalf("extract workers = %d, want 3", got)
}
if err := cfg.Validate(); err != nil {
t.Fatalf("Validate() error = %v, want nil", err)
}
}
func TestApplyFileConfigEmptyStageWorkersDefaultsExtractToTotal(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
concurrency:
total_llm: 4
stage_workers: {}
`)
if got := cfg.Concurrency.StageWorkers["extract"]; got != 4 {
t.Fatalf("extract workers = %d, want total concurrency 4", got)
}
}
func TestApplyFileConfigRejectsUnsupportedStageWorkerKeys(t *testing.T) {
for _, test := range []struct {
name string
key string
want string
}{
{name: "empty", key: "' '", want: "must not be empty"},
{name: "unknown", key: "merge", want: "not supported"},
} {
t.Run(test.name, func(t *testing.T) {
fileCfg, err := ParseFileConfigYAML([]byte("version: 2\nconcurrency:\n stage_workers:\n " + test.key + ": 1\n"))
if err != nil {
t.Fatalf("ParseFileConfigYAML() error = %v", err)
}
cfg := Default()
err = cfg.applyFileConfigWithLookup(fileCfg, emptyLookup)
if err == nil || !strings.Contains(err.Error(), test.want) {
t.Fatalf("ApplyFileConfig() error = %v, want %q", err, test.want)
}
})
}
}
func TestApplyFileConfigWorkspaceSection(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
workspace:
directory: /var/lib/notarius
diagnostics:
enabled: false
retention: never
resume:
enabled: true
debug:
enabled: true
diagnostics:
work_dir: /tmp/legacy
retention: always
`)
if cfg.Workspace.Directory != "/var/lib/notarius" {
t.Fatalf("workspace directory = %q, want /var/lib/notarius", cfg.Workspace.Directory)
}
if cfg.DiagnosticsEnabled() {
t.Fatalf("expected diagnostics disabled")
}
if cfg.Diagnostics.WorkDir != "/var/lib/notarius/diagnostics" {
t.Fatalf("effective diagnostics work dir = %q, want workspace diagnostics root", cfg.Diagnostics.WorkDir)
}
if cfg.Diagnostics.Retention != diagnostics.RetentionNever {
t.Fatalf("effective diagnostics retention = %q, want workspace override", cfg.Diagnostics.Retention)
}
if !cfg.Workspace.Resume.Enabled {
t.Fatalf("expected resume enabled")
}
if !cfg.Workspace.Debug.Enabled {
t.Fatalf("expected debug enabled")
}
}
func TestApplyFileConfigLegacyDiagnosticsRemainCompatible(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
diagnostics:
work_dir: /tmp/legacy
retention: always
`)
if cfg.Workspace.Directory != "" {
t.Fatalf("workspace directory = %q, want unset", cfg.Workspace.Directory)
}
if !cfg.DiagnosticsEnabled() {
t.Fatalf("expected diagnostics enabled")
}
if cfg.Diagnostics.WorkDir != "/tmp/legacy" {
t.Fatalf("effective diagnostics work dir = %q, want legacy", cfg.Diagnostics.WorkDir)
}
if cfg.Diagnostics.Retention != diagnostics.RetentionAlways {
t.Fatalf("effective diagnostics retention = %q, want legacy", cfg.Diagnostics.Retention)
}
}
func TestApplyFileConfigWorkspaceRetentionOverridesLegacyRetentionOnlyWhenSet(t *testing.T) {
t.Run("legacy retained", func(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
workspace:
directory: /var/lib/notarius
diagnostics:
retention: never
`)
if cfg.Diagnostics.Retention != diagnostics.RetentionNever {
t.Fatalf("effective diagnostics retention = %q, want legacy", cfg.Diagnostics.Retention)
}
})
t.Run("workspace overrides", func(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 2
workspace:
directory: /var/lib/notarius
diagnostics:
retention: always
diagnostics:
retention: never
`)
if cfg.Diagnostics.Retention != diagnostics.RetentionAlways {
t.Fatalf("effective diagnostics retention = %q, want workspace", cfg.Diagnostics.Retention)
}
})
}
func parseAndApplyConfig(t *testing.T, raw string) Config {
t.Helper()
fileCfg, err := ParseFileConfigYAML([]byte(raw))
if err != nil {
t.Fatalf("ParseFileConfigYAML: %v", err)
}
cfg := Default()
if err := cfg.applyFileConfigWithLookup(fileCfg, emptyLookup); err != nil {
t.Fatalf("ApplyFileConfig: %v", err)
}
return cfg
}
func emptyLookup(string) (string, bool) {
return "", false
}
func mapLookup(values map[string]string) func(string) (string, bool) {
return func(key string) (string, bool) {
value, ok := values[key]
return value, ok
}
}

174
internal/core/config/redaction_test.go
View File

@@ -1,174 +0,0 @@
//go:build legacy
package config
import (
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
func TestRedactedConfigCopiesScriptoriumConfig(t *testing.T) {
cfg := Default()
cfg.Scriptorium.ProfileDir = "./profiles"
cfg.Workspace.Directory = "/var/lib/notarius"
cfg.Workspace.Resume.Enabled = true
cfg.Concurrency.StageWorkers["extract"] = 1
redacted := cfg.Redacted()
if redacted.Scriptorium.ProfileDir != "./profiles" {
t.Fatalf("expected Scriptorium profile source preserved, got %+v", redacted.Scriptorium)
}
redacted.Scriptorium.ProfileDir = "./changed"
if cfg.Scriptorium.ProfileDir != "./profiles" {
t.Fatalf("redaction mutated original config")
}
if redacted.Workspace.Directory != "/var/lib/notarius" || !redacted.Workspace.Resume.Enabled {
t.Fatalf("expected workspace config preserved, got %+v", redacted.Workspace)
}
redacted.Workspace.Directory = "/changed"
if cfg.Workspace.Directory != "/var/lib/notarius" {
t.Fatalf("redaction mutated original workspace config")
}
redacted.Concurrency.StageWorkers["extract"] = 9
if cfg.Concurrency.StageWorkers["extract"] != 1 {
t.Fatalf("redaction aliased stage worker map")
}
}
func TestConfigRedactedSummaryPayloadCopiesConfig(t *testing.T) {
cfg := Default()
cfg.Scriptorium.ProfileFile = "./profiles.yml"
payload, ok := cfg.RedactedSummaryPayload().(Config)
if !ok {
t.Fatalf("expected Config payload, got %T", cfg.RedactedSummaryPayload())
}
if payload.Scriptorium.ProfileFile != "./profiles.yml" {
t.Fatalf("expected Scriptorium profile file preserved, got %+v", payload.Scriptorium)
}
}
func TestEffectiveConfigRedactedSummaryPayloadCopies(t *testing.T) {
cfg := validConfig()
cfg.Concurrency.TotalLLM = 4
cfg.Concurrency.StageWorkers["extract"] = 2
lane := cfg.Pipelines["example"].Artifacts["events"]
lane.Extract.Options = map[string]any{"temperature": 0.2}
lane.References = map[string]string{"roster": "./roster.yml"}
lane.Extract.References = map[string]string{"glossary": "./glossary.md"}
lane.Normalize.References = map[string]string{"notes": "./normalize.md"}
cfg.Pipelines["example"].Artifacts["events"] = lane
pipelineProfile := cfg.Pipelines["example"]
pipelineProfile.Chunk.References = map[string]string{"scene_guide": "./scene.md"}
cfg.Pipelines["example"] = pipelineProfile
effective, err := cfg.Resolve(ResolveInput{
PipelineID: "example",
Only: []string{"events"},
Catalog: fakeCatalog(t,
pipeline.ModuleSpec{
Key: "generic",
Stage: pipeline.StageChunk,
Requires: []string{"source"},
Provides: []string{"chunks"},
ReferenceSlots: []contracts.ReferenceSlot{
{Name: "scene_guide"},
},
},
pipeline.ModuleSpec{
Key: "fake/extract",
Stage: pipeline.StageExtract,
Requires: []string{"chunks"},
Provides: []string{"artifact"},
ReferenceSlots: []contracts.ReferenceSlot{
{Name: "glossary"},
{Name: "roster"},
},
},
pipeline.ModuleSpec{
Key: "noop",
Stage: pipeline.StageNormalize,
Requires: []string{"merged"},
Provides: []string{"normalized"},
ReferenceSlots: []contracts.ReferenceSlot{
{Name: "notes"},
},
},
),
})
if err != nil {
t.Fatalf("Resolve: %v", err)
}
payload, ok := effective.RedactedSummaryPayload().(EffectiveConfig)
if !ok {
t.Fatalf("expected EffectiveConfig payload, got %T", effective.RedactedSummaryPayload())
}
if payload.PipelineID != effective.PipelineID || payload.ResolvedPipeline.Digest != effective.ResolvedPipeline.Digest {
t.Fatalf("expected pipeline metadata preserved, got %+v", payload)
}
payload.Config.Concurrency.StageWorkers["extract"] = 4
if effective.Config.Concurrency.StageWorkers["extract"] != 2 {
t.Fatalf("expected effective stage worker map to be copied")
}
payload.Only[0] = "changed"
if effective.Only[0] != "events" {
t.Fatalf("expected only lanes to be copied")
}
payload.ResolvedPipeline.ArtifactLanes[0].Extract.Options["temperature"] = 1.0
if effective.ResolvedPipeline.ArtifactLanes[0].Extract.Options["temperature"] != 0.2 {
t.Fatalf("expected resolved pipeline options to be copied")
}
payload.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.Bindings[0].Source = "./changed.yml"
if referenceBindingSource(effective.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.Bindings, "roster") != "./roster.yml" {
t.Fatalf("expected resolved pipeline references to be copied")
}
payload.ResolvedPipeline.Chunk.References["scene_guide"] = "./changed-scene.md"
if effective.ResolvedPipeline.Chunk.References["scene_guide"] != "./scene.md" {
t.Fatalf("expected chunk references to be copied")
}
payload.ResolvedPipeline.ArtifactLanes[0].Extract.References["glossary"] = "./changed-glossary.md"
if effective.ResolvedPipeline.ArtifactLanes[0].Extract.References["glossary"] != "./glossary.md" {
t.Fatalf("expected extract references to be copied")
}
payload.ResolvedPipeline.ArtifactLanes[0].Normalize.References["notes"] = "./changed-normalize.md"
if effective.ResolvedPipeline.ArtifactLanes[0].Normalize.References["notes"] != "./normalize.md" {
t.Fatalf("expected normalize references to be copied")
}
effective.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.ReferenceSet = contracts.ReferenceSet{
Slots: map[string]contracts.ResolvedReferenceSlot{
"roster": {
Slot: contracts.ReferenceSlot{Name: "roster"},
Items: []contracts.ReferenceItem{
{
SlotName: "roster",
Content: []byte("reference content"),
},
},
},
},
}
payload, ok = effective.RedactedSummaryPayload().(EffectiveConfig)
if !ok {
t.Fatalf("expected EffectiveConfig payload, got %T", effective.RedactedSummaryPayload())
}
payload.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0].Content[0] = 'X'
got := effective.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0].Content
if string(got) != "reference content" {
t.Fatalf("expected materialized reference content to be copied, got %q", got)
}
}
func referenceBindingSource(bindings []pipeline.ReferenceBinding, slotName string) string {
for _, binding := range bindings {
if binding.SlotName == slotName {
return binding.Source
}
}
return ""
}

666
internal/core/config/validation_test.go
View File

@@ -1,666 +0,0 @@
//go:build legacy
package config
import (
"fmt"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
func TestValidateSuccessForValidConfig(t *testing.T) {
cfg := validConfig()
if err := cfg.Validate(); err != nil {
t.Fatalf("Validate: %v", err)
}
}
func TestValidateAllowsExplicitScriptoriumProfileIDOnBinding(t *testing.T) {
cfg := validConfig()
lane := cfg.Pipelines["example"].Artifacts["events"]
lane.Extract.LLMProfile = "scriptorium-profile"
cfg.Pipelines["example"].Artifacts["events"] = lane
if err := cfg.Validate(); err != nil {
t.Fatalf("Validate() error = %v, want nil", err)
}
}
func TestValidateRejectsWhitespaceOnlyExplicitLLMProfile(t *testing.T) {
cfg := validConfig()
lane := cfg.Pipelines["example"].Artifacts["events"]
lane.Extract.LLMProfile = " "
cfg.Pipelines["example"].Artifacts["events"] = lane
err := cfg.Validate()
if err == nil || !strings.Contains(err.Error(), "llm_profile") || !strings.Contains(err.Error(), "events") {
t.Fatalf("expected llm_profile error with lane context, got %v", err)
}
}
func TestValidateRejectsInvalidNumericFields(t *testing.T) {
tests := []struct {
name string
mutate func(Config) Config
want string
}{
{
name: "total concurrency",
mutate: func(cfg Config) Config {
cfg.Concurrency.TotalLLM = 0
return cfg
},
want: "total LLM concurrency",
},
{
name: "negative retries",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Merge.Retries = -1
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: "retries",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
err := tc.mutate(validConfig()).Validate()
if err == nil || !strings.Contains(err.Error(), tc.want) {
t.Fatalf("expected error containing %q, got %v", tc.want, err)
}
})
}
}
func TestValidateStageWorkerBoundaries(t *testing.T) {
for _, test := range []struct {
name string
workers int
wantErr bool
}{
{name: "below minimum", workers: 0, wantErr: true},
{name: "minimum", workers: 1},
{name: "maximum", workers: 4},
{name: "above maximum", workers: 5, wantErr: true},
} {
t.Run(test.name, func(t *testing.T) {
cfg := validConfig()
cfg.Concurrency.TotalLLM = 4
cfg.Concurrency.StageWorkers["extract"] = test.workers
err := cfg.Validate()
if test.wantErr && (err == nil || !strings.Contains(err.Error(), "stage_workers.extract")) {
t.Fatalf("Validate() error = %v, want extract worker range error", err)
}
if !test.wantErr && err != nil {
t.Fatalf("Validate() error = %v, want nil", err)
}
})
}
}
func TestValidateRejectsUnknownEffectiveStageWorkerKey(t *testing.T) {
cfg := validConfig()
cfg.Concurrency.StageWorkers["merge"] = 1
err := cfg.Validate()
if err == nil || !strings.Contains(err.Error(), "stage_workers key") || !strings.Contains(err.Error(), "merge") {
t.Fatalf("Validate() error = %v, want unknown stage worker key", err)
}
}
func TestValidateRejectsMutuallyExclusiveScriptoriumProfileSources(t *testing.T) {
cfg := validConfig()
cfg.Scriptorium.ProfileDir = "./profiles"
cfg.Scriptorium.ProfileFile = "./profiles.yml"
err := cfg.Validate()
if err == nil || !strings.Contains(err.Error(), "mutually exclusive") {
t.Fatalf("expected Scriptorium source conflict, got %v", err)
}
}
func TestValidateRejectsInvalidDiagnosticsRetention(t *testing.T) {
cfg := validConfig()
cfg.Diagnostics.Retention = diagnostics.RetentionMode("sometimes")
err := cfg.Validate()
if err == nil || !strings.Contains(err.Error(), "retention") {
t.Fatalf("expected retention error, got %v", err)
}
}
func TestValidateRejectsInvalidWorkspaceDiagnosticsRetention(t *testing.T) {
cfg := validConfig()
cfg.Workspace.Diagnostics.Retention = diagnostics.RetentionMode("sometimes")
cfg.Workspace.Diagnostics.retentionSet = true
err := cfg.Validate()
if err == nil || !strings.Contains(err.Error(), "workspace diagnostics retention") {
t.Fatalf("expected workspace retention error, got %v", err)
}
}
func TestValidateRejectsInvalidReferenceMaps(t *testing.T) {
tests := []struct {
name string
mutate func(Config) Config
want []string
}{
{
name: "empty chunk slot",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
profile.Chunk.References = map[string]string{" ": "./roster.yml"}
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "chunk", "reference slot", "empty"},
},
{
name: "empty chunk source",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
profile.Chunk.References = map[string]string{"roster": " "}
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "chunk", "roster", "source", "empty"},
},
{
name: "empty extract slot",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Extract.References = map[string]string{" ": "./roster.yml"}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "extract", "reference slot", "empty"},
},
{
name: "empty extract source",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Extract.References = map[string]string{"roster": " "}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "extract", "roster", "source", "empty"},
},
{
name: "empty normalize slot",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Normalize.References = map[string]string{" ": "./roster.yml"}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "normalize", "reference slot", "empty"},
},
{
name: "empty normalize source",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Normalize.References = map[string]string{"roster": " "}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "normalize", "roster", "source", "empty"},
},
{
name: "empty pipeline slot",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
profile.References = map[string]string{" ": "./roster.yml"}
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "reference slot", "empty"},
},
{
name: "empty pipeline source",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
profile.References = map[string]string{"roster": " "}
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "roster", "source", "empty"},
},
{
name: "empty lane slot",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.References = map[string]string{" ": "./roster.yml"}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "reference slot", "empty"},
},
{
name: "empty lane source",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.References = map[string]string{"roster": " "}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "roster", "source", "empty"},
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
err := tc.mutate(validConfig()).Validate()
if err == nil {
t.Fatal("Validate() error = nil, want error")
}
for _, want := range tc.want {
if !strings.Contains(err.Error(), want) {
t.Fatalf("Validate() error = %q, want substring %q", err.Error(), want)
}
}
})
}
}
func TestValidateRejectsReferencesOnUnsupportedBindings(t *testing.T) {
tests := []struct {
name string
mutate func(Config) Config
want []string
}{
{
name: "input",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
profile.Input.References = map[string]string{"roster": "./roster.yml"}
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "input", "references", "not supported"},
},
{
name: "output",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
profile.Output.References = map[string]string{"roster": "./roster.yml"}
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "output", "references", "not supported"},
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
err := tc.mutate(validConfig()).Validate()
if err == nil {
t.Fatal("Validate() error = nil, want error")
}
for _, want := range tc.want {
if !strings.Contains(err.Error(), want) {
t.Fatalf("Validate() error = %q, want substring %q", err.Error(), want)
}
}
})
}
}
func TestValidateRejectsEmptyIDs(t *testing.T) {
tests := []struct {
name string
mutate func(Config) Config
want string
}{
{
name: "pipeline",
mutate: func(cfg Config) Config {
cfg.Pipelines[" "] = pipeline.PipelineProfile{}
return cfg
},
want: "pipeline id",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
err := tc.mutate(validConfig()).Validate()
if err == nil || !strings.Contains(err.Error(), tc.want) {
t.Fatalf("expected error containing %q, got %v", tc.want, err)
}
})
}
}
func TestValidateRejectsIDsDuplicatedAfterTrimming(t *testing.T) {
tests := []struct {
name string
mutate func(Config) Config
want string
}{
{
name: "pipeline",
mutate: func(cfg Config) Config {
cfg.Pipelines[" example "] = cfg.Pipelines["example"]
return cfg
},
want: "duplicated",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
err := tc.mutate(validConfig()).Validate()
if err == nil || !strings.Contains(err.Error(), tc.want) {
t.Fatalf("expected error containing %q, got %v", tc.want, err)
}
})
}
}
func TestValidateRejectsConfiguredValidators(t *testing.T) {
cfg := validConfig()
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Validators = []pipeline.ModuleBinding{pipeline.Binding("fake/validator")}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
err := cfg.Validate()
if err == nil {
t.Fatal("Validate() error = nil, want configured validators error")
}
for _, want := range []string{"example", "events", "validators", "extract.validators", "merge.validators", "normalize.validators"} {
if !strings.Contains(err.Error(), want) {
t.Fatalf("Validate() error = %q, want substring %q", err.Error(), want)
}
}
}
func TestValidateAcceptsStageLocalValidatorOverrides(t *testing.T) {
cfg := validConfig()
profile := cfg.Pipelines["example"]
profile.Chunk.Validators = pipeline.ValidatorOverride{Set: true}
lane := profile.Artifacts["events"]
lane.Extract.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{
pipeline.Binding("fake/validator"),
{Module: "fake/llm-validator", LLMProfile: "careful", Options: map[string]any{"threshold": 0.7}},
},
}
lane.Merge.Validators = pipeline.ValidatorOverride{Set: true}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
if err := cfg.Validate(); err != nil {
t.Fatalf("Validate() error = %v, want nil", err)
}
}
func TestValidateRejectsInvalidValidatorBindings(t *testing.T) {
tests := []struct {
name string
binding pipeline.ModuleBinding
want string
}{
{
name: "empty module",
binding: pipeline.ModuleBinding{},
want: "module must not be empty",
},
{
name: "references",
binding: pipeline.ModuleBinding{Module: "fake/validator", References: map[string]string{"roster": "./roster.txt"}},
want: "references are not supported",
},
{
name: "nested validators",
binding: pipeline.ModuleBinding{Module: "fake/validator", Validators: pipeline.ValidatorOverride{Set: true}},
want: "nested validators are not supported",
},
{
name: "retries",
binding: pipeline.ModuleBinding{Module: "fake/validator", Retries: 1},
want: "retries are not supported",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
cfg := validConfig()
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Extract.Validators = pipeline.ValidatorOverride{
Set: true,
Validators: []pipeline.ModuleBinding{test.binding},
}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
err := cfg.Validate()
if err == nil || !strings.Contains(err.Error(), test.want) {
t.Fatalf("Validate() error = %v, want %q", err, test.want)
}
})
}
}
func validConfig() Config {
cfg := Default()
cfg.Pipelines["example"] = pipeline.PipelineProfile{
Input: pipeline.Binding("fake/input"),
Artifacts: map[string]pipeline.ArtifactLaneProfile{
"events": {
Extract: pipeline.Binding("fake/extract"),
},
"notes": {
Extract: pipeline.Binding("fake/extract"),
},
},
}
return cfg
}
func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.ModuleCatalog {
t.Helper()
specs := map[string]pipeline.ModuleSpec{
"fake/input": {
Key: "fake/input",
Stage: pipeline.StageInput,
Provides: []string{"source"},
},
"generic": {
Key: "generic",
Stage: pipeline.StageChunk,
Requires: []string{"source"},
Provides: []string{"chunks"},
},
"fake/extract": {
Key: "fake/extract",
Stage: pipeline.StageExtract,
Requires: []string{"chunks"},
Provides: []string{"artifact"},
},
"appendorder": {
Key: "appendorder",
Stage: pipeline.StageMerge,
Requires: []string{"artifact"},
Provides: []string{"merged"},
},
"noop": {
Key: "noop",
Stage: pipeline.StageNormalize,
Requires: []string{"merged"},
Provides: []string{"normalized"},
},
"fake/validator": {
Key: "fake/validator",
Stage: pipeline.StageValidate,
Requires: []string{"normalized"},
Provides: []string{"validated"},
},
"fake/llm-validator": {
Key: "fake/llm-validator",
Stage: pipeline.StageValidate,
Requires: []string{"normalized"},
Provides: []string{"validated"},
},
"json": {
Key: "json",
Stage: pipeline.StageOutput,
Requires: []string{"normalized"},
},
}
for _, override := range overrides {
specs[override.Key] = override
}
for _, key := range []string{"fake/extract", "appendorder", "noop"} {
spec := specs[key]
spec.ArtifactKind = fakeArtifactKind
specs[key] = spec
}
inputs := pipeline.NewInputAdapterRegistry()
chunkers := pipeline.NewChunkerRegistry()
extractors := pipeline.NewExtractorRegistry()
mergers := pipeline.NewMergerRegistry()
normalizers := pipeline.NewNormalizerRegistry()
validators := pipeline.NewValidatorRegistry()
outputs := pipeline.NewOutputEncoderRegistry()
mustRegisterInput(t, inputs, specs["fake/input"])
mustRegisterChunker(t, chunkers, specs["generic"])
mustRegisterExtractor(t, extractors, specs["fake/extract"])
mustRegisterMerger(t, mergers, specs["appendorder"])
mustRegisterNormalizer(t, normalizers, specs["noop"])
mustRegisterValidator(t, validators, specs["fake/validator"])
mustRegisterValidator(t, validators, specs["fake/llm-validator"])
mustRegisterOutput(t, outputs, specs["json"])
codecs := pipeline.NewArtifactCodecRegistry()
if err := pipeline.RegisterArtifactCodec(codecs, fakeArtifactCodec{}); err != nil {
t.Fatalf("register artifact codec: %v", err)
}
return pipeline.ModuleCatalog{
Inputs: inputs,
Chunkers: chunkers,
ArtifactCodecs: codecs,
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,
Validators: validators,
ValidatorChains: pipeline.NewValidatorChainRegistry(),
Outputs: outputs,
}
}
func mustRegisterInput(t *testing.T, registry *pipeline.InputAdapterRegistry, spec pipeline.ModuleSpec) {
t.Helper()
if err := registry.RegisterWithSpec(spec, func() (contracts.InputAdapter, error) { return nil, nil }); err != nil {
t.Fatalf("register input: %v", err)
}
}
func mustRegisterChunker(t *testing.T, registry *pipeline.ChunkerRegistry, spec pipeline.ModuleSpec) {
t.Helper()
if err := registry.RegisterWithSpec(spec, func() (contracts.Chunker, error) { return nil, nil }); err != nil {
t.Fatalf("register chunker: %v", err)
}
}
func mustRegisterExtractor(t *testing.T, registry *pipeline.ExtractorRegistry, spec pipeline.ModuleSpec) {
t.Helper()
validateOptions := func(options map[string]any) error {
if err := pipeline.RejectUnknownOptions(options, "temperature"); err != nil {
return err
}
if value, ok := options["temperature"]; ok {
if _, ok := value.(float64); !ok {
return fmt.Errorf("temperature must be a number")
}
}
return nil
}
if err := pipeline.RegisterExtractorBuilder[fakeArtifact](registry, spec, validateOptions, func(pipeline.BuildRequest) (contracts.Extractor[fakeArtifact], error) { return nil, nil }); err != nil {
t.Fatalf("register extractor: %v", err)
}
}
func mustRegisterMerger(t *testing.T, registry *pipeline.MergerRegistry, spec pipeline.ModuleSpec) {
t.Helper()
if err := pipeline.RegisterMerger[fakeArtifact](registry, spec, func() (contracts.Merger[fakeArtifact], error) { return nil, nil }); err != nil {
t.Fatalf("register merger: %v", err)
}
}
func mustRegisterNormalizer(t *testing.T, registry *pipeline.NormalizerRegistry, spec pipeline.ModuleSpec) {
t.Helper()
if err := pipeline.RegisterNormalizer[fakeArtifact](registry, spec, func() (contracts.Normalizer[fakeArtifact], error) { return nil, nil }); err != nil {
t.Fatalf("register normalizer: %v", err)
}
}
func mustRegisterValidator(t *testing.T, registry *pipeline.ValidatorRegistry, spec pipeline.ModuleSpec) {
t.Helper()
executionClass := contracts.ExecutionClassDeterministic
if spec.Key == "fake/llm-validator" {
executionClass = contracts.ExecutionClassLLMBacked
}
validatorSpec := pipeline.ValidatorSpec{Key: spec.Key, ExecutionClass: executionClass}
if err := pipeline.RegisterTypedValidator[fakeArtifact](registry, fakeArtifactKind, validatorSpec, func() (contracts.TypedValidator[fakeArtifact], error) { return nil, nil }); err != nil {
t.Fatalf("register validator: %v", err)
}
}
const fakeArtifactKind contracts.ArtifactKind = "test/artifact"
type fakeArtifact string
type fakeArtifactCodec struct{}
func (fakeArtifactCodec) Kind() contracts.ArtifactKind { return fakeArtifactKind }
func (fakeArtifactCodec) Schema() contracts.ArtifactSchema {
return contracts.ArtifactSchema{ID: "urn:notarius:test:artifact", Name: "Test artifact", Version: "1", JSONSchema: []byte(`{"type":"string"}`)}
}
func (fakeArtifactCodec) MediaType() string { return "application/json" }
func (fakeArtifactCodec) EncodeCandidate(value fakeArtifact) ([]byte, error) {
return []byte(fmt.Sprintf("%q", value)), nil
}
func (fakeArtifactCodec) Encode(value fakeArtifact) ([]byte, error) {
return []byte(fmt.Sprintf("%q", value)), nil
}
func (fakeArtifactCodec) Decode(content []byte) (fakeArtifact, error) {
if len(content) < 2 {
return "", fmt.Errorf("invalid test artifact")
}
return fakeArtifact(content[1 : len(content)-1]), nil
}
func mustRegisterOutput(t *testing.T, registry *pipeline.OutputEncoderRegistry, spec pipeline.ModuleSpec) {
t.Helper()
if err := registry.RegisterWithSpec(spec, func() (contracts.OutputEncoder, error) { return nil, nil }); err != nil {
t.Fatalf("register output: %v", err)
}
}

15
internal/core/diagnostics/artifacts.go
View File

@@ -1,15 +0,0 @@
package diagnostics
const (
ArtifactInvocationMetadata = "invocation.json"
ArtifactEffectiveConfig = "effective-config.json"
ArtifactResolvedPipeline = "resolved-pipeline.json"
ArtifactResolvedReferences = "resolved-references.json"
ArtifactCheckpointEvents = "checkpoint-events.json"
ArtifactSourceDocument = "source-document.json"
ArtifactRunManifest = "run-manifest.json"
ArtifactChunkPlan = "chunk-plan.json"
ArtifactRunReport = "run-report.json"
ArtifactWarnings = "warnings.json"
ArtifactErrorLog = "error.log"
)

24
internal/core/diagnostics/artifacts_test.go
View File

@@ -1,24 +0,0 @@
package diagnostics
import "testing"
func TestArtifactNamesUseExtractionOrientedNames(t *testing.T) {
names := []string{
ArtifactInvocationMetadata,
ArtifactEffectiveConfig,
ArtifactResolvedPipeline,
ArtifactResolvedReferences,
ArtifactSourceDocument,
ArtifactRunManifest,
ArtifactChunkPlan,
ArtifactRunReport,
ArtifactWarnings,
ArtifactErrorLog,
}
for _, name := range names {
if name == "" {
t.Fatalf("artifact name must not be empty")
}
}
}

293
internal/core/diagnostics/run_dir.go
View File

@@ -1,293 +0,0 @@
package diagnostics
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"time"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
const (
defaultWorkDir = "/tmp/notarius"
maxRunDirectoryCreateAttempts = 16
)
var utcNow = func() time.Time {
return time.Now().UTC()
}
// RunDirectory represents a per-run diagnostics directory.
type RunDirectory struct {
path string
retention RetentionMode
createdAt time.Time
}
type RetentionMode string
const (
RetentionAuto RetentionMode = "auto"
RetentionAlways RetentionMode = "always"
RetentionNever RetentionMode = "never"
)
type RetentionDecisionInput struct {
RetentionMode RetentionMode
RunSucceeded bool
HasWarnings bool
}
type RedactedEffectiveConfigPayload interface {
RedactedSummaryPayload() any
}
// InvocationMetadata captures non-secret invocation details for diagnostics.
type InvocationMetadata struct {
Operation string `json:"operation"`
PipelineID string `json:"pipeline_id,omitempty"`
PipelineDigest string `json:"pipeline_digest,omitempty"`
Resume bool `json:"resume,omitempty"`
InputPath string `json:"input_path,omitempty"`
ConfigPath string `json:"config_path,omitempty"`
ConfigSource string `json:"config_source,omitempty"`
OnlyLanes []string `json:"only_lanes,omitempty"`
ChunkCacheOverride string `json:"chunk_cache_override,omitempty"`
RunID string `json:"run_id"`
StartedAt time.Time `json:"started_at"`
}
func ShouldRetainRunDirectory(input RetentionDecisionInput) bool {
if !input.RunSucceeded {
return true
}
switch input.RetentionMode {
case RetentionAlways:
return true
case RetentionNever:
return false
case RetentionAuto, "":
return input.HasWarnings
default:
return true
}
}
func NewRunDirectory(workDir string, retention RetentionMode) (*RunDirectory, error) {
if strings.TrimSpace(workDir) == "" {
workDir = defaultWorkDir
}
if retention == "" {
retention = RetentionAuto
}
if err := os.MkdirAll(workDir, 0o755); err != nil {
return nil, fmt.Errorf("create diagnostics work directory %q: %w", workDir, err)
}
var lastRunPath string
for attempt := 0; attempt < maxRunDirectoryCreateAttempts; attempt++ {
createdAt := utcNow()
runID := fmt.Sprintf("run-%d", createdAt.UnixNano())
runPath := filepath.Join(workDir, runID)
lastRunPath = runPath
if err := os.Mkdir(runPath, 0o755); err != nil {
if os.IsExist(err) {
continue
}
return nil, fmt.Errorf("create diagnostics run directory %q: %w", runPath, err)
}
return &RunDirectory{
path: runPath,
retention: retention,
createdAt: createdAt,
}, nil
}
return nil, fmt.Errorf("create diagnostics run directory %q: exhausted unique run ID attempts", lastRunPath)
}
func (r *RunDirectory) Path() string {
if r == nil {
return ""
}
return r.path
}
func (r *RunDirectory) RunID() string {
if r == nil {
return ""
}
return filepath.Base(r.path)
}
func (r *RunDirectory) WriteInvocationMetadata(metadata InvocationMetadata) error {
if r == nil {
return fmt.Errorf("run directory must not be nil")
}
if metadata.RunID == "" {
metadata.RunID = r.RunID()
}
if metadata.StartedAt.IsZero() {
metadata.StartedAt = r.createdAt
}
return r.WriteJSONArtifact(ArtifactInvocationMetadata, metadata)
}
func (r *RunDirectory) WriteRedactedEffectiveConfig(payload RedactedEffectiveConfigPayload) error {
if payload == nil {
return fmt.Errorf("redacted effective config payload must not be nil")
}
return r.WriteJSONArtifact(ArtifactEffectiveConfig, payload.RedactedSummaryPayload())
}
func (r *RunDirectory) WriteResolvedPipeline(payload any) error {
return r.WriteJSONArtifact(ArtifactResolvedPipeline, payload)
}
func (r *RunDirectory) WriteResolvedReferences(payload any) error {
return r.WriteJSONArtifact(ArtifactResolvedReferences, payload)
}
func (r *RunDirectory) WriteCheckpointEvents(payload any) error {
return r.WriteJSONArtifact(ArtifactCheckpointEvents, payload)
}
func (r *RunDirectory) WriteSourceDocument(payload any) error {
return r.WriteJSONArtifact(ArtifactSourceDocument, payload)
}
func (r *RunDirectory) WriteRunManifest(manifest artifacts.RunManifest) error {
return r.WriteJSONArtifact(ArtifactRunManifest, manifest)
}
func (r *RunDirectory) WriteChunkPlan(summary artifacts.ChunkPlanSummary) error {
return r.WriteJSONArtifact(ArtifactChunkPlan, summary)
}
func (r *RunDirectory) WriteRunReport(payload any) error {
return r.WriteJSONArtifact(ArtifactRunReport, payload)
}
func (r *RunDirectory) WriteWarnings(warnings []contracts.Warning) error {
return r.WriteJSONArtifact(ArtifactWarnings, warnings)
}
func (r *RunDirectory) WriteErrorLog(errorMessage string) error {
if r == nil {
return fmt.Errorf("run directory must not be nil")
}
path, err := r.artifactPath(ArtifactErrorLog)
if err != nil {
return err
}
if err := writeFileAtomic(path, []byte(errorMessage+"\n"), 0o644); err != nil {
return fmt.Errorf("write diagnostics artifact %q: %w", ArtifactErrorLog, err)
}
return nil
}
func (r *RunDirectory) WriteJSONArtifact(name string, payload any) error {
if r == nil {
return fmt.Errorf("run directory must not be nil")
}
path, err := r.artifactPath(name)
if err != nil {
return err
}
data, err := json.MarshalIndent(payload, "", " ")
if err != nil {
return fmt.Errorf("marshal diagnostics artifact %q: %w", name, err)
}
data = append(data, '\n')
if err := writeFileAtomic(path, data, 0o644); err != nil {
return fmt.Errorf("write diagnostics artifact %q: %w", name, err)
}
return nil
}
func (r *RunDirectory) ApplyRetention(input RetentionDecisionInput) error {
if r == nil {
return fmt.Errorf("run directory must not be nil")
}
decision := input
if decision.RetentionMode == "" {
decision.RetentionMode = r.retention
}
if ShouldRetainRunDirectory(decision) {
return nil
}
if err := os.RemoveAll(r.path); err != nil {
return fmt.Errorf("remove diagnostics run directory %q: %w", r.path, err)
}
return nil
}
func (r *RunDirectory) artifactPath(name string) (string, error) {
name = strings.TrimSpace(name)
if name == "" {
return "", fmt.Errorf("diagnostics artifact name must not be empty")
}
if filepath.IsAbs(name) {
return "", fmt.Errorf("diagnostics artifact name %q must not be absolute", name)
}
if name != filepath.Base(name) || strings.Contains(name, "/") || strings.Contains(name, `\`) {
return "", fmt.Errorf("diagnostics artifact name %q must not contain path separators", name)
}
runPath, err := filepath.Abs(r.path)
if err != nil {
return "", fmt.Errorf("resolve diagnostics run directory %q: %w", r.path, err)
}
artifactPath, err := filepath.Abs(filepath.Join(runPath, name))
if err != nil {
return "", fmt.Errorf("resolve diagnostics artifact %q: %w", name, err)
}
if filepath.Dir(artifactPath) != runPath {
return "", fmt.Errorf("diagnostics artifact name %q resolves outside run directory", name)
}
return artifactPath, nil
}
func writeFileAtomic(path string, data []byte, perm os.FileMode) error {
dir := filepath.Dir(path)
if err := os.MkdirAll(dir, 0o755); err != nil {
return err
}
temp, err := os.CreateTemp(dir, "."+filepath.Base(path)+".tmp-*")
if err != nil {
return err
}
tempPath := temp.Name()
removeTemp := true
defer func() {
if removeTemp {
_ = os.Remove(tempPath)
}
}()
if _, err := temp.Write(data); err != nil {
_ = temp.Close()
return err
}
if err := temp.Chmod(perm); err != nil {
_ = temp.Close()
return err
}
if err := temp.Close(); err != nil {
return err
}
if err := os.Rename(tempPath, path); err != nil {
return err
}
removeTemp = false
return nil
}

383
internal/core/diagnostics/run_dir_test.go
View File

@@ -1,383 +0,0 @@
package diagnostics
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"regexp"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestNewRunDirectoryCreatesRunDirectoryAndRunID(t *testing.T) {
workDir := t.TempDir()
runDir, err := NewRunDirectory(workDir, RetentionAuto)
if err != nil {
t.Fatalf("NewRunDirectory: %v", err)
}
if filepath.Dir(runDir.Path()) != workDir {
t.Fatalf("unexpected run directory parent: %q", runDir.Path())
}
if ok := regexp.MustCompile(`^run-\d+$`).MatchString(runDir.RunID()); !ok {
t.Fatalf("unexpected run ID: %q", runDir.RunID())
}
info, err := os.Stat(runDir.Path())
if err != nil {
t.Fatalf("stat run directory: %v", err)
}
if !info.IsDir() {
t.Fatalf("expected run path to be a directory")
}
}
func TestNewRunDirectoryRetriesOnRunIDCollision(t *testing.T) {
workDir := t.TempDir()
first := time.Unix(0, 100).UTC()
second := first.Add(time.Nanosecond)
if err := os.Mkdir(filepath.Join(workDir, fmt.Sprintf("run-%d", first.UnixNano())), 0o755); err != nil {
t.Fatalf("create existing run directory: %v", err)
}
restoreUTCNow := replaceUTCNow(func() func() time.Time {
calls := 0
return func() time.Time {
calls++
if calls == 1 {
return first
}
return second
}
}())
t.Cleanup(restoreUTCNow)
runDir, err := NewRunDirectory(workDir, RetentionAuto)
if err != nil {
t.Fatalf("NewRunDirectory: %v", err)
}
wantRunID := fmt.Sprintf("run-%d", second.UnixNano())
if runDir.RunID() != wantRunID {
t.Fatalf("RunID = %q, want %q", runDir.RunID(), wantRunID)
}
if _, err := os.Stat(runDir.Path()); err != nil {
t.Fatalf("stat run directory: %v", err)
}
}
func TestNewRunDirectoryReturnsErrorAfterRunIDCollisionsExhausted(t *testing.T) {
workDir := t.TempDir()
collisionTime := time.Unix(0, 200).UTC()
collisionPath := filepath.Join(workDir, fmt.Sprintf("run-%d", collisionTime.UnixNano()))
if err := os.Mkdir(collisionPath, 0o755); err != nil {
t.Fatalf("create existing run directory: %v", err)
}
restoreUTCNow := replaceUTCNow(func() time.Time {
return collisionTime
})
t.Cleanup(restoreUTCNow)
_, err := NewRunDirectory(workDir, RetentionAuto)
if err == nil || !strings.Contains(err.Error(), "exhausted unique run ID attempts") {
t.Fatalf("expected exhausted collision error, got %v", err)
}
}
func TestNewRunDirectoryUsesDefaultWorkDirectory(t *testing.T) {
runDir, err := NewRunDirectory("", RetentionAuto)
if err != nil {
t.Fatalf("NewRunDirectory: %v", err)
}
t.Cleanup(func() {
_ = os.RemoveAll(runDir.Path())
_ = os.Remove(defaultWorkDir)
})
if filepath.Dir(runDir.Path()) != defaultWorkDir {
t.Fatalf("expected default work directory %q, got %q", defaultWorkDir, filepath.Dir(runDir.Path()))
}
}
func TestWriteJSONArtifactWritesIndentedNewlineTerminatedJSON(t *testing.T) {
runDir := newTestRunDirectory(t)
if err := runDir.WriteJSONArtifact("artifact.json", map[string]any{"value": "ok"}); err != nil {
t.Fatalf("WriteJSONArtifact: %v", err)
}
data := readArtifact(t, runDir, "artifact.json")
if !strings.HasSuffix(string(data), "\n") {
t.Fatalf("expected trailing newline, got %q", data)
}
if !strings.Contains(string(data), "\n \"value\": \"ok\"\n") {
t.Fatalf("expected indented JSON, got %s", data)
}
}
func TestWriteJSONArtifactLeavesNoTemporaryFiles(t *testing.T) {
runDir := newTestRunDirectory(t)
if err := runDir.WriteJSONArtifact("artifact.json", map[string]any{"value": "ok"}); err != nil {
t.Fatalf("WriteJSONArtifact: %v", err)
}
entries, err := os.ReadDir(runDir.Path())
if err != nil {
t.Fatalf("read run directory: %v", err)
}
for _, entry := range entries {
if strings.Contains(entry.Name(), ".tmp-") {
t.Fatalf("temporary diagnostics file remains after success: %s", entry.Name())
}
}
}
func TestWriteInvocationMetadataFillsMissingRunIDAndStartTime(t *testing.T) {
runDir := newTestRunDirectory(t)
if err := runDir.WriteInvocationMetadata(InvocationMetadata{Operation: "validate"}); err != nil {
t.Fatalf("WriteInvocationMetadata: %v", err)
}
var got InvocationMetadata
if err := json.Unmarshal(readArtifact(t, runDir, ArtifactInvocationMetadata), &got); err != nil {
t.Fatalf("unmarshal invocation metadata: %v", err)
}
if got.RunID != runDir.RunID() {
t.Fatalf("unexpected run ID: got %q want %q", got.RunID, runDir.RunID())
}
if got.StartedAt.IsZero() {
t.Fatalf("expected started_at to be filled")
}
if got.Operation != "validate" {
t.Fatalf("unexpected operation: %q", got.Operation)
}
}
func TestWriteInvocationMetadataPreservesProvidedRunIDAndStartTime(t *testing.T) {
runDir := newTestRunDirectory(t)
startedAt := time.Date(2026, 7, 3, 12, 0, 0, 0, time.UTC)
if err := runDir.WriteInvocationMetadata(InvocationMetadata{
Operation: "validate",
RunID: "provided",
StartedAt: startedAt,
}); err != nil {
t.Fatalf("WriteInvocationMetadata: %v", err)
}
var got InvocationMetadata
if err := json.Unmarshal(readArtifact(t, runDir, ArtifactInvocationMetadata), &got); err != nil {
t.Fatalf("unmarshal invocation metadata: %v", err)
}
if got.RunID != "provided" {
t.Fatalf("unexpected run ID: %q", got.RunID)
}
if !got.StartedAt.Equal(startedAt) {
t.Fatalf("unexpected started_at: %s", got.StartedAt)
}
}
func TestWriteTypedArtifacts(t *testing.T) {
runDir := newTestRunDirectory(t)
if err := runDir.WriteRedactedEffectiveConfig(fakeRedactedEffectiveConfig{payload: map[string]any{"redacted": true}}); err != nil {
t.Fatalf("WriteRedactedEffectiveConfig: %v", err)
}
if err := runDir.WriteResolvedPipeline(map[string]any{"pipeline": "test"}); err != nil {
t.Fatalf("WriteResolvedPipeline: %v", err)
}
if err := runDir.WriteResolvedReferences([]artifacts.ReferenceProvenance{{LaneID: "events", SlotName: "roster"}}); err != nil {
t.Fatalf("WriteResolvedReferences: %v", err)
}
if err := runDir.WriteSourceDocument(map[string]any{"source_id": "source-1"}); err != nil {
t.Fatalf("WriteSourceDocument: %v", err)
}
if err := runDir.WriteRunManifest(artifacts.RunManifest{RunID: "run-1"}); err != nil {
t.Fatalf("WriteRunManifest: %v", err)
}
if err := runDir.WriteChunkPlan(artifacts.ChunkPlanSummary{Mode: "auto", RequestedModule: "chunk/test", LookupStatus: "invalid", LookupReason: "stored chunk plan failed validation", ValidationStatus: "not_run", PublicationStatus: "not_published"}); err != nil {
t.Fatalf("WriteChunkPlan: %v", err)
}
if err := runDir.WriteRunReport(map[string]any{"ok": true}); err != nil {
t.Fatalf("WriteRunReport: %v", err)
}
if err := runDir.WriteWarnings([]contracts.Warning{{ReasonCode: "test", Message: "warning"}}); err != nil {
t.Fatalf("WriteWarnings: %v", err)
}
for _, name := range []string{
ArtifactEffectiveConfig,
ArtifactResolvedPipeline,
ArtifactResolvedReferences,
ArtifactSourceDocument,
ArtifactRunManifest,
ArtifactChunkPlan,
ArtifactRunReport,
ArtifactWarnings,
} {
if _, err := os.Stat(filepath.Join(runDir.Path(), name)); err != nil {
t.Fatalf("expected artifact %q: %v", name, err)
}
}
chunkSummary, err := os.ReadFile(filepath.Join(runDir.Path(), ArtifactChunkPlan))
if err != nil {
t.Fatal(err)
}
for _, forbidden := range []string{"units", "annotations", "source content", "prompt", "response", "raw invalid"} {
if strings.Contains(string(chunkSummary), forbidden) {
t.Fatalf("chunk plan summary leaked %q: %s", forbidden, chunkSummary)
}
}
}
func TestWriteRedactedEffectiveConfigWritesPayloadReturnedByProvider(t *testing.T) {
runDir := newTestRunDirectory(t)
if err := runDir.WriteRedactedEffectiveConfig(fakeRedactedEffectiveConfig{
payload: map[string]any{
"api_key": "[REDACTED]",
"model": "test-model",
},
}); err != nil {
t.Fatalf("WriteRedactedEffectiveConfig: %v", err)
}
data := string(readArtifact(t, runDir, ArtifactEffectiveConfig))
if !strings.Contains(data, `"api_key": "[REDACTED]"`) || !strings.Contains(data, `"model": "test-model"`) {
t.Fatalf("unexpected effective config artifact: %s", data)
}
}
func TestWriteErrorLogWritesPlainTextWithTrailingNewline(t *testing.T) {
runDir := newTestRunDirectory(t)
if err := runDir.WriteErrorLog("something failed"); err != nil {
t.Fatalf("WriteErrorLog: %v", err)
}
if got := string(readArtifact(t, runDir, ArtifactErrorLog)); got != "something failed\n" {
t.Fatalf("unexpected error log: %q", got)
}
}
func TestArtifactPathRejectsUnsafeNames(t *testing.T) {
runDir := newTestRunDirectory(t)
tests := []string{
"",
" ",
"/absolute.json",
"nested/artifact.json",
`nested\artifact.json`,
"../escape.json",
}
for _, name := range tests {
t.Run(name, func(t *testing.T) {
if err := runDir.WriteJSONArtifact(name, map[string]any{}); err == nil {
t.Fatalf("expected unsafe artifact name %q to be rejected", name)
}
})
}
}
func TestShouldRetainRunDirectoryDecisions(t *testing.T) {
tests := []struct {
name string
input RetentionDecisionInput
want bool
}{
{name: "failed auto retained", input: RetentionDecisionInput{RetentionMode: RetentionAuto, RunSucceeded: false}, want: true},
{name: "failed always retained", input: RetentionDecisionInput{RetentionMode: RetentionAlways, RunSucceeded: false}, want: true},
{name: "failed never retained", input: RetentionDecisionInput{RetentionMode: RetentionNever, RunSucceeded: false}, want: true},
{name: "successful always retained", input: RetentionDecisionInput{RetentionMode: RetentionAlways, RunSucceeded: true}, want: true},
{name: "successful never removed", input: RetentionDecisionInput{RetentionMode: RetentionNever, RunSucceeded: true}, want: false},
{name: "successful auto without warnings removed", input: RetentionDecisionInput{RetentionMode: RetentionAuto, RunSucceeded: true}, want: false},
{name: "successful auto with warnings retained", input: RetentionDecisionInput{RetentionMode: RetentionAuto, RunSucceeded: true, HasWarnings: true}, want: true},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
if got := ShouldRetainRunDirectory(tc.input); got != tc.want {
t.Fatalf("ShouldRetainRunDirectory() = %v, want %v", got, tc.want)
}
})
}
}
func TestApplyRetentionRemovesOnlyRunDirectory(t *testing.T) {
workDir := t.TempDir()
runDir, err := NewRunDirectory(workDir, RetentionNever)
if err != nil {
t.Fatalf("NewRunDirectory: %v", err)
}
siblingPath := filepath.Join(workDir, "sibling")
if err := os.WriteFile(siblingPath, []byte("keep"), 0o644); err != nil {
t.Fatalf("write sibling: %v", err)
}
if err := runDir.ApplyRetention(RetentionDecisionInput{RunSucceeded: true}); err != nil {
t.Fatalf("ApplyRetention: %v", err)
}
if _, err := os.Stat(runDir.Path()); !os.IsNotExist(err) {
t.Fatalf("expected run directory removed, stat err=%v", err)
}
if _, err := os.Stat(workDir); err != nil {
t.Fatalf("expected work directory retained: %v", err)
}
if _, err := os.Stat(siblingPath); err != nil {
t.Fatalf("expected sibling retained: %v", err)
}
}
func TestApplyRetentionKeepsRetainedRunDirectory(t *testing.T) {
runDir := newTestRunDirectory(t)
if err := runDir.ApplyRetention(RetentionDecisionInput{RunSucceeded: true, HasWarnings: true}); err != nil {
t.Fatalf("ApplyRetention: %v", err)
}
if _, err := os.Stat(runDir.Path()); err != nil {
t.Fatalf("expected run directory retained: %v", err)
}
}
func newTestRunDirectory(t *testing.T) *RunDirectory {
t.Helper()
runDir, err := NewRunDirectory(t.TempDir(), RetentionAuto)
if err != nil {
t.Fatalf("NewRunDirectory: %v", err)
}
return runDir
}
func replaceUTCNow(replacement func() time.Time) func() {
original := utcNow
utcNow = replacement
return func() {
utcNow = original
}
}
func readArtifact(t *testing.T, runDir *RunDirectory, name string) []byte {
t.Helper()
data, err := os.ReadFile(filepath.Join(runDir.Path(), name))
if err != nil {
t.Fatalf("read artifact %q: %v", name, err)
}
return data
}
type fakeRedactedEffectiveConfig struct {
payload any
}
func (f fakeRedactedEffectiveConfig) RedactedSummaryPayload() any {
return f.payload
}

114
internal/core/workspace/files.go
View File

@@ -1,114 +0,0 @@
package workspace
import (
"encoding/json"
"fmt"
"os"
"path"
"path/filepath"
"strings"
)
func SafePath(root string, name string) (string, error) {
root = strings.TrimSpace(root)
if root == "" {
return "", fmt.Errorf("workspace root must not be empty")
}
name = strings.TrimSpace(name)
if name == "" {
return "", fmt.Errorf("workspace artifact name must not be empty")
}
if strings.Contains(name, `\`) {
return "", fmt.Errorf("workspace artifact name %q must use slash-separated relative paths", name)
}
if path.IsAbs(name) || filepath.IsAbs(name) {
return "", fmt.Errorf("workspace artifact name %q must be relative", name)
}
if name == "." || strings.Contains(name, "..") {
return "", fmt.Errorf("workspace artifact name %q must not contain ..", name)
}
cleaned := path.Clean(name)
if cleaned != name {
return "", fmt.Errorf("workspace artifact name %q must be clean", name)
}
absRoot, err := filepath.Abs(root)
if err != nil {
return "", fmt.Errorf("resolve workspace root %q: %w", root, err)
}
target, err := filepath.Abs(filepath.Join(absRoot, filepath.FromSlash(cleaned)))
if err != nil {
return "", fmt.Errorf("resolve workspace artifact %q: %w", name, err)
}
rel, err := filepath.Rel(absRoot, target)
if err != nil {
return "", fmt.Errorf("resolve workspace artifact %q: %w", name, err)
}
if rel == "." || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
return "", fmt.Errorf("workspace artifact name %q resolves outside workspace root", name)
}
return target, nil
}
func WriteJSON(root string, name string, payload any) error {
target, err := SafePath(root, name)
if err != nil {
return err
}
data, err := json.MarshalIndent(payload, "", " ")
if err != nil {
return fmt.Errorf("marshal workspace artifact %q: %w", name, err)
}
data = append(data, '\n')
if err := writeFileAtomic(target, data, 0o644); err != nil {
return fmt.Errorf("write workspace artifact %q: %w", name, err)
}
return nil
}
func WriteBytes(root string, name string, data []byte) error {
target, err := SafePath(root, name)
if err != nil {
return err
}
if err := writeFileAtomic(target, data, 0o644); err != nil {
return fmt.Errorf("write workspace artifact %q: %w", name, err)
}
return nil
}
func writeFileAtomic(target string, data []byte, perm os.FileMode) error {
dir := filepath.Dir(target)
if err := os.MkdirAll(dir, 0o755); err != nil {
return err
}
temp, err := os.CreateTemp(dir, "."+filepath.Base(target)+".tmp-*")
if err != nil {
return err
}
tempPath := temp.Name()
removeTemp := true
defer func() {
if removeTemp {
_ = os.Remove(tempPath)
}
}()
if _, err := temp.Write(data); err != nil {
_ = temp.Close()
return err
}
if err := temp.Chmod(perm); err != nil {
_ = temp.Close()
return err
}
if err := temp.Close(); err != nil {
return err
}
if err := os.Rename(tempPath, target); err != nil {
return err
}
removeTemp = false
return nil
}

148
internal/core/workspace/files_test.go
View File

@@ -1,148 +0,0 @@
package workspace
import (
"os"
"path/filepath"
"strings"
"testing"
)
func TestSafePathAcceptsCleanRelativePaths(t *testing.T) {
root := t.TempDir()
got, err := SafePath(root, "source/manifest.json")
if err != nil {
t.Fatalf("SafePath: %v", err)
}
want := filepath.Join(root, "source", "manifest.json")
if got != want {
t.Fatalf("SafePath = %q, want %q", got, want)
}
}
func TestSafePathRejectsUnsafeNames(t *testing.T) {
root := t.TempDir()
tests := []struct {
name string
path string
want string
}{
{name: "empty", path: " ", want: "empty"},
{name: "absolute", path: filepath.Join(root, "artifact.json"), want: "relative"},
{name: "parent segment", path: "../artifact.json", want: ".."},
{name: "embedded parent", path: "source/../artifact.json", want: ".."},
{name: "backslash", path: `source\artifact.json`, want: "slash-separated"},
{name: "unclean", path: "source//artifact.json", want: "clean"},
{name: "dot", path: ".", want: ".."},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
got, err := SafePath(root, tc.path)
if err == nil {
t.Fatalf("SafePath returned %q, want error", got)
}
if !strings.Contains(err.Error(), tc.want) {
t.Fatalf("SafePath error = %v, want containing %q", err, tc.want)
}
})
}
}
func TestSafePathRejectsEmptyRoot(t *testing.T) {
got, err := SafePath(" ", "artifact.json")
if err == nil {
t.Fatalf("SafePath returned %q, want error", got)
}
if !strings.Contains(err.Error(), "root") {
t.Fatalf("SafePath error = %v, want root error", err)
}
}
func TestSafePathDoesNotPermitEscapingRoot(t *testing.T) {
root := t.TempDir()
for _, name := range []string{
"..",
"../outside.json",
"nested/../../outside.json",
} {
t.Run(name, func(t *testing.T) {
got, err := SafePath(root, name)
if err == nil {
t.Fatalf("SafePath returned %q, want error", got)
}
})
}
}
func TestWriteJSONWritesIndentedAtomicArtifact(t *testing.T) {
root := t.TempDir()
err := WriteJSON(root, "source/manifest.json", map[string]any{
"status": "succeeded",
"count": 2,
})
if err != nil {
t.Fatalf("WriteJSON: %v", err)
}
got := string(readFile(t, filepath.Join(root, "source", "manifest.json")))
if !strings.HasSuffix(got, "\n") {
t.Fatalf("expected trailing newline, got %q", got)
}
if !strings.Contains(got, `"status": "succeeded"`) || !strings.Contains(got, `"count": 2`) {
t.Fatalf("unexpected JSON: %s", got)
}
assertNoTempFiles(t, filepath.Join(root, "source"))
}
func TestWriteBytesWritesNestedArtifact(t *testing.T) {
root := t.TempDir()
if err := WriteBytes(root, "chunk/chunks.json", []byte("payload")); err != nil {
t.Fatalf("WriteBytes: %v", err)
}
got := string(readFile(t, filepath.Join(root, "chunk", "chunks.json")))
if got != "payload" {
t.Fatalf("bytes = %q, want payload", got)
}
assertNoTempFiles(t, filepath.Join(root, "chunk"))
}
func TestWritersRejectUnsafePaths(t *testing.T) {
root := t.TempDir()
if err := WriteBytes(root, "../outside.json", []byte("payload")); err == nil {
t.Fatalf("WriteBytes accepted unsafe path")
}
if err := WriteJSON(root, `debug\trace.json`, map[string]string{"x": "y"}); err == nil {
t.Fatalf("WriteJSON accepted unsafe path")
}
if _, err := os.Stat(filepath.Join(root, "..", "outside.json")); !os.IsNotExist(err) {
t.Fatalf("outside path stat err = %v, want not exist", err)
}
}
func readFile(t *testing.T, path string) []byte {
t.Helper()
data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read %q: %v", path, err)
}
return data
}
func assertNoTempFiles(t *testing.T, dir string) {
t.Helper()
entries, err := os.ReadDir(dir)
if err != nil {
t.Fatalf("read dir %q: %v", dir, err)
}
for _, entry := range entries {
if strings.Contains(entry.Name(), ".tmp-") {
t.Fatalf("temporary file was not cleaned up: %s", entry.Name())
}
}
}

54
internal/core/workspace/settings.go
View File

@@ -1,54 +0,0 @@
package workspace
import (
"fmt"
"path/filepath"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/core/config"
)
type Settings struct {
RootDir string
DiagnosticsRoot string
CheckpointsRoot string
DebugRoot string
DiagnosticsEnabled bool
ResumeEnabled bool
DebugEnabled bool
}
func FromConfig(cfg config.Config) Settings {
_ = cfg
return Settings{}
}
func (s Settings) DiagnosticsRunDirectory(runID string) (string, error) {
if !s.DiagnosticsEnabled || strings.TrimSpace(s.DiagnosticsRoot) == "" {
return "", nil
}
return safeSingleDirectory(s.DiagnosticsRoot, runID, "diagnostics run ID")
}
func (s Settings) DebugRunDirectory(runID string) (string, error) {
if !s.DebugEnabled || strings.TrimSpace(s.DebugRoot) == "" {
return "", nil
}
return safeSingleDirectory(s.DebugRoot, runID, "debug run ID")
}
func cleanPath(path string) string {
path = strings.TrimSpace(path)
if path == "" {
return ""
}
return filepath.Clean(path)
}
func safeSingleDirectory(root string, name string, label string) (string, error) {
name = strings.TrimSpace(name)
if strings.Contains(name, "/") || strings.Contains(name, `\`) {
return "", fmt.Errorf("%s %q must be a single directory name", label, name)
}
return SafePath(root, name)
}

120
internal/core/workspace/settings_test.go
View File

@@ -1,120 +0,0 @@
//go:build legacy
package workspace
import (
"path/filepath"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/config"
)
func TestFromConfigBuildsWorkspaceRoots(t *testing.T) {
cfg := config.Default()
cfg.Workspace.Directory = "/var/lib/notarius"
cfg.Workspace.Resume.Enabled = true
cfg.Workspace.Debug.Enabled = true
cfg.RecomputeEffectiveDiagnostics()
settings := FromConfig(cfg)
if settings.RootDir != "/var/lib/notarius" {
t.Fatalf("RootDir = %q, want /var/lib/notarius", settings.RootDir)
}
if settings.DiagnosticsRoot != "/var/lib/notarius/diagnostics" || !settings.DiagnosticsEnabled {
t.Fatalf("diagnostics settings = %+v, want workspace diagnostics root enabled", settings)
}
if settings.CheckpointsRoot != "/var/lib/notarius/checkpoints" || !settings.ResumeEnabled {
t.Fatalf("checkpoint settings = %+v, want workspace checkpoints root enabled", settings)
}
if settings.DebugRoot != "/var/lib/notarius/debug" || !settings.DebugEnabled {
t.Fatalf("debug settings = %+v, want workspace debug root enabled", settings)
}
}
func TestFromConfigKeepsLegacyDiagnosticsRootWithoutWorkspaceRoot(t *testing.T) {
cfg := config.Default()
cfg.Diagnostics.WorkDir = "/tmp/notarius-legacy"
cfg.Workspace.Resume.Enabled = true
cfg.Workspace.Debug.Enabled = true
settings := FromConfig(cfg)
if settings.RootDir != "" {
t.Fatalf("RootDir = %q, want empty", settings.RootDir)
}
if settings.DiagnosticsRoot != "/tmp/notarius-legacy" || !settings.DiagnosticsEnabled {
t.Fatalf("diagnostics settings = %+v, want legacy diagnostics root enabled", settings)
}
if settings.CheckpointsRoot != "" || settings.ResumeEnabled {
t.Fatalf("checkpoint settings = %+v, want disabled empty root", settings)
}
if settings.DebugRoot != "" || settings.DebugEnabled {
t.Fatalf("debug settings = %+v, want disabled empty root", settings)
}
}
func TestPathConstructors(t *testing.T) {
root := t.TempDir()
settings := Settings{
RootDir: root,
DiagnosticsRoot: filepath.Join(root, "diagnostics"),
CheckpointsRoot: filepath.Join(root, "checkpoints"),
DebugRoot: filepath.Join(root, "debug"),
DiagnosticsEnabled: true,
ResumeEnabled: true,
DebugEnabled: true,
}
diagnosticsDir, err := settings.DiagnosticsRunDirectory("run-123")
if err != nil {
t.Fatalf("DiagnosticsRunDirectory: %v", err)
}
if diagnosticsDir != filepath.Join(root, "diagnostics", "run-123") {
t.Fatalf("diagnostics dir = %q", diagnosticsDir)
}
debugDir, err := settings.DebugRunDirectory("run-456")
if err != nil {
t.Fatalf("DebugRunDirectory: %v", err)
}
if debugDir != filepath.Join(root, "debug", "run-456") {
t.Fatalf("debug dir = %q", debugDir)
}
}
func TestDisabledPathConstructorsReturnEmptyPaths(t *testing.T) {
settings := Settings{}
for name, call := range map[string]func() (string, error){
"diagnostics": func() (string, error) { return settings.DiagnosticsRunDirectory("run-1") },
"debug": func() (string, error) { return settings.DebugRunDirectory("run-1") },
} {
t.Run(name, func(t *testing.T) {
got, err := call()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got != "" {
t.Fatalf("path = %q, want empty", got)
}
})
}
}
func TestRunDirectoryConstructorsRejectNestedNames(t *testing.T) {
root := t.TempDir()
settings := Settings{
DiagnosticsRoot: filepath.Join(root, "diagnostics"),
DebugRoot: filepath.Join(root, "debug"),
DiagnosticsEnabled: true,
DebugEnabled: true,
}
if got, err := settings.DiagnosticsRunDirectory("run-1/nested"); err == nil {
t.Fatalf("DiagnosticsRunDirectory returned %q, want error", got)
}
if got, err := settings.DebugRunDirectory("run-1/nested"); err == nil {
t.Fatalf("DebugRunDirectory returned %q, want error", got)
}
}

2
internal/framework/checkpoint/manifest.go
View File

@@ -3,6 +3,8 @@ package checkpoint
import "time" import "time"
const ( const (
// These names and values are frozen checkpoint wire-compatibility
// identifiers. They intentionally retain the former terminology.
WorkspaceSchemaVersion = "notarius.workspace.v2" WorkspaceSchemaVersion = "notarius.workspace.v2"
WorkspaceSchemaVersionV1 = "notarius.workspace.v1" WorkspaceSchemaVersionV1 = "notarius.workspace.v1"
) )