Forward Notarius cache controls during extraction
This commit is contained in:
@@ -89,7 +89,13 @@ This stage alone does not enable delivery and is not a finished feature.
|
|||||||
|
|
||||||
## Stage 2 — Forward Cache Intent and Apply Force Precedence
|
## Stage 2 — Forward Cache Intent and Apply Force Precedence
|
||||||
|
|
||||||
Status: Not started
|
Status: Complete (2026-09-16)
|
||||||
|
|
||||||
|
Completion: Carried typed cache intent through extract requests, applied force
|
||||||
|
precedence in stage orchestration, mapped validated effective values to exact
|
||||||
|
Notarius flags, and confirmed both controls remain outside semantic resume
|
||||||
|
evidence. Verified normal and race tests for `./internal/stage` and
|
||||||
|
`./internal/adapters/notarius`.
|
||||||
|
|
||||||
Read: `internal/stage/extract.go`, `stage.go`, `extract_test.go`,
|
Read: `internal/stage/extract.go`, `stage.go`, `extract_test.go`,
|
||||||
`semantic_contracts_delivery.go`, `semantic_contracts_delivery_test.go`;
|
`semantic_contracts_delivery.go`, `semantic_contracts_delivery_test.go`;
|
||||||
|
|||||||
@@ -35,6 +35,8 @@ type RunRequest struct {
|
|||||||
ReceiptPath string
|
ReceiptPath string
|
||||||
LogPath string
|
LogPath string
|
||||||
Timeout time.Duration
|
Timeout time.Duration
|
||||||
|
Resume bool
|
||||||
|
ChunkCache string
|
||||||
References []ReferenceBinding
|
References []ReferenceBinding
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -73,6 +73,12 @@ func (r *SubprocessRunner) Run(ctx context.Context, req RunRequest) (RunResult,
|
|||||||
"--input", req.InputPath,
|
"--input", req.InputPath,
|
||||||
"--output-dir", req.OutputRoot,
|
"--output-dir", req.OutputRoot,
|
||||||
}
|
}
|
||||||
|
if req.Resume {
|
||||||
|
args = append(args, "--resume")
|
||||||
|
}
|
||||||
|
if req.ChunkCache != "" {
|
||||||
|
args = append(args, "--chunk_cache", req.ChunkCache)
|
||||||
|
}
|
||||||
for _, reference := range references {
|
for _, reference := range references {
|
||||||
args = append(args, "--reference", reference.Selector+"="+reference.Path)
|
args = append(args, "--reference", reference.Selector+"="+reference.Path)
|
||||||
}
|
}
|
||||||
@@ -159,6 +165,9 @@ func validateRunRequest(req RunRequest) ([]ReferenceBinding, error) {
|
|||||||
if req.Timeout <= 0 {
|
if req.Timeout <= 0 {
|
||||||
return nil, fmt.Errorf("notarius timeout must be positive")
|
return nil, fmt.Errorf("notarius timeout must be positive")
|
||||||
}
|
}
|
||||||
|
if !validChunkCacheMode(req.ChunkCache) {
|
||||||
|
return nil, fmt.Errorf("notarius chunk cache mode must be empty, auto, bypass, or refresh")
|
||||||
|
}
|
||||||
for label, path := range map[string]string{
|
for label, path := range map[string]string{
|
||||||
"config": req.ConfigPath,
|
"config": req.ConfigPath,
|
||||||
"input": req.InputPath,
|
"input": req.InputPath,
|
||||||
@@ -217,6 +226,15 @@ func validateRunRequest(req RunRequest) ([]ReferenceBinding, error) {
|
|||||||
return references, nil
|
return references, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func validChunkCacheMode(mode string) bool {
|
||||||
|
switch mode {
|
||||||
|
case "", "auto", "bypass", "refresh":
|
||||||
|
return true
|
||||||
|
default:
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
type receiptDocument struct {
|
type receiptDocument struct {
|
||||||
SchemaVersion string `json:"schema_version"`
|
SchemaVersion string `json:"schema_version"`
|
||||||
RunID string `json:"run_id"`
|
RunID string `json:"run_id"`
|
||||||
|
|||||||
@@ -77,6 +77,68 @@ func TestSubprocessRunnerBuildsExactInvocationAndDiscoversBundle(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestSubprocessRunnerMapsCacheControlsToArguments(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
resume bool
|
||||||
|
chunkCache string
|
||||||
|
want []string
|
||||||
|
}{
|
||||||
|
{name: "resume", resume: true, want: []string{"--resume"}},
|
||||||
|
{name: "auto", chunkCache: "auto", want: []string{"--chunk_cache", "auto"}},
|
||||||
|
{name: "bypass", chunkCache: "bypass", want: []string{"--chunk_cache", "bypass"}},
|
||||||
|
{name: "refresh", chunkCache: "refresh", want: []string{"--chunk_cache", "refresh"}},
|
||||||
|
{name: "resume and mode", resume: true, chunkCache: "auto", want: []string{"--resume", "--chunk_cache", "auto"}},
|
||||||
|
}
|
||||||
|
for _, test := range tests {
|
||||||
|
t.Run(test.name, func(t *testing.T) {
|
||||||
|
req := validRunRequest(t)
|
||||||
|
req.Resume = test.resume
|
||||||
|
req.ChunkCache = test.chunkCache
|
||||||
|
var captured sharedsubprocess.RunRequest
|
||||||
|
runner := &SubprocessRunner{run: func(_ context.Context, processReq sharedsubprocess.RunRequest) (sharedsubprocess.RunResult, error) {
|
||||||
|
captured = processReq
|
||||||
|
writeValidBundleAndReceipt(t, req, false)
|
||||||
|
return sharedsubprocess.RunResult{ExitCode: 0}, nil
|
||||||
|
}}
|
||||||
|
|
||||||
|
if _, err := runner.Run(context.Background(), req); err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
want := []string{
|
||||||
|
"run", req.PipelineID, "--config", req.ConfigPath, "--input", req.InputPath,
|
||||||
|
"--output-dir", req.OutputRoot,
|
||||||
|
}
|
||||||
|
want = append(want, test.want...)
|
||||||
|
want = append(want, "--json")
|
||||||
|
if !reflect.DeepEqual(captured.Args, want) {
|
||||||
|
t.Fatalf("subprocess args = %#v, want %#v", captured.Args, want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSubprocessRunnerRejectsInvalidChunkCacheBeforeLaunch(t *testing.T) {
|
||||||
|
for _, mode := range []string{"AUTO", "invalid", " auto "} {
|
||||||
|
t.Run(mode, func(t *testing.T) {
|
||||||
|
req := validRunRequest(t)
|
||||||
|
req.ChunkCache = mode
|
||||||
|
started := false
|
||||||
|
runner := &SubprocessRunner{run: func(context.Context, sharedsubprocess.RunRequest) (sharedsubprocess.RunResult, error) {
|
||||||
|
started = true
|
||||||
|
return sharedsubprocess.RunResult{}, nil
|
||||||
|
}}
|
||||||
|
_, err := runner.Run(context.Background(), req)
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "chunk cache mode") {
|
||||||
|
t.Fatalf("Run() error = %v, want chunk cache validation failure", err)
|
||||||
|
}
|
||||||
|
if started {
|
||||||
|
t.Fatal("subprocess started with invalid chunk cache mode")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestSubprocessRunnerForwardsLoadedCredentialWithoutRereadingEnvironment(t *testing.T) {
|
func TestSubprocessRunnerForwardsLoadedCredentialWithoutRereadingEnvironment(t *testing.T) {
|
||||||
const envName = "NARRATIO_NOTARIUS_CUSTOM_LLM_KEY"
|
const envName = "NARRATIO_NOTARIUS_CUSTOM_LLM_KEY"
|
||||||
const loadedValue = "loaded-notarius-credential"
|
const loadedValue = "loaded-notarius-credential"
|
||||||
|
|||||||
@@ -146,10 +146,16 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (re
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("extract: build configuration fingerprint: %w", err)
|
return nil, fmt.Errorf("extract: build configuration fingerprint: %w", err)
|
||||||
}
|
}
|
||||||
|
resume := notariusConfig.Resume
|
||||||
|
chunkCache := notariusConfig.ChunkCache
|
||||||
|
if env.Force {
|
||||||
|
resume = false
|
||||||
|
chunkCache = "refresh"
|
||||||
|
}
|
||||||
request := notarius.RunRequest{
|
request := notarius.RunRequest{
|
||||||
Binary: resolvedBinary, LLMAPIKey: notariusConfig.LLMAPIKey, ConfigPath: configPath, PipelineID: notariusConfig.PipelineID,
|
Binary: resolvedBinary, LLMAPIKey: notariusConfig.LLMAPIKey, ConfigPath: configPath, PipelineID: notariusConfig.PipelineID,
|
||||||
InputPath: inputPath, OutputRoot: outputRoot, WorkingDirectory: workingDirectory,
|
InputPath: inputPath, OutputRoot: outputRoot, WorkingDirectory: workingDirectory,
|
||||||
ReceiptPath: receiptPath, LogPath: logPath, Timeout: timeout, References: referenceBindings,
|
ReceiptPath: receiptPath, LogPath: logPath, Timeout: timeout, Resume: resume, ChunkCache: chunkCache, References: referenceBindings,
|
||||||
}
|
}
|
||||||
stageSucceeded := false
|
stageSucceeded := false
|
||||||
bundleMoved := false
|
bundleMoved := false
|
||||||
|
|||||||
@@ -96,6 +96,9 @@ notarius:
|
|||||||
if len(req.References) != 0 {
|
if len(req.References) != 0 {
|
||||||
t.Fatalf("references = %#v, want none", req.References)
|
t.Fatalf("references = %#v, want none", req.References)
|
||||||
}
|
}
|
||||||
|
if req.Resume || req.ChunkCache != "" {
|
||||||
|
t.Fatalf("default cache controls = resume %v, mode %q", req.Resume, req.ChunkCache)
|
||||||
|
}
|
||||||
if req.LLMAPIKey != loadedPipeline.Notarius.LLMAPIKey || req.LLMAPIKey.EnvName() != credentialName || !req.LLMAPIKey.IsResolved() {
|
if req.LLMAPIKey != loadedPipeline.Notarius.LLMAPIKey || req.LLMAPIKey.EnvName() != credentialName || !req.LLMAPIKey.IsResolved() {
|
||||||
t.Fatal("request did not retain the loaded Notarius credential reference")
|
t.Fatal("request did not retain the loaded Notarius credential reference")
|
||||||
}
|
}
|
||||||
@@ -114,6 +117,41 @@ notarius:
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestExtractStageAppliesNotariusCacheControlsAndForcePrecedence(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
configuredResume bool
|
||||||
|
configuredMode string
|
||||||
|
force bool
|
||||||
|
wantResume bool
|
||||||
|
wantMode string
|
||||||
|
}{
|
||||||
|
{name: "ordinary defaults"},
|
||||||
|
{name: "ordinary configured", configuredResume: true, configuredMode: "bypass", wantResume: true, wantMode: "bypass"},
|
||||||
|
{name: "force overrides configured", configuredResume: true, configuredMode: "bypass", force: true, wantMode: "refresh"},
|
||||||
|
{name: "force overrides defaults", force: true, wantMode: "refresh"},
|
||||||
|
}
|
||||||
|
for _, test := range tests {
|
||||||
|
t.Run(test.name, func(t *testing.T) {
|
||||||
|
env, m, fake := setupExtractEnv(t)
|
||||||
|
env.Config.Pipeline.Notarius.Resume = test.configuredResume
|
||||||
|
env.Config.Pipeline.Notarius.ChunkCache = test.configuredMode
|
||||||
|
env.Force = test.force
|
||||||
|
|
||||||
|
if _, err := (extractStage{}).Run(context.Background(), env, m); err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(fake.Requests) != 1 {
|
||||||
|
t.Fatalf("requests = %d, want 1", len(fake.Requests))
|
||||||
|
}
|
||||||
|
request := fake.Requests[0]
|
||||||
|
if request.Resume != test.wantResume || request.ChunkCache != test.wantMode {
|
||||||
|
t.Fatalf("cache controls = resume %v, mode %q; want resume %v, mode %q", request.Resume, request.ChunkCache, test.wantResume, test.wantMode)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T) {
|
func TestExtractStageResolvesAllPreparedReferencesInSelectorOrder(t *testing.T) {
|
||||||
env, m, fake := setupExtractEnv(t)
|
env, m, fake := setupExtractEnv(t)
|
||||||
configured := []struct {
|
configured := []struct {
|
||||||
|
|||||||
@@ -76,6 +76,26 @@ func TestExtractSemanticConfigSensitivityAndOperationalExclusions(t *testing.T)
|
|||||||
if !want.Equal(got) {
|
if !want.Equal(got) {
|
||||||
t.Fatalf("operational settings changed fingerprint: got %#v want %#v", got, want)
|
t.Fatalf("operational settings changed fingerprint: got %#v want %#v", got, want)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
for _, test := range []struct {
|
||||||
|
name string
|
||||||
|
mutate func(*config.NotariusConfig)
|
||||||
|
}{
|
||||||
|
{name: "resume", mutate: func(cfg *config.NotariusConfig) { cfg.Resume = true }},
|
||||||
|
{name: "chunk cache", mutate: func(cfg *config.NotariusConfig) { cfg.ChunkCache = "refresh" }},
|
||||||
|
} {
|
||||||
|
t.Run("operational "+test.name, func(t *testing.T) {
|
||||||
|
env := downstreamSemanticEnv()
|
||||||
|
test.mutate(env.Config.Pipeline.Notarius)
|
||||||
|
got, err := (extractStage{}).SemanticConfigFingerprint(env)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if !want.Equal(got) {
|
||||||
|
t.Fatalf("%s changed semantic fingerprint: got %#v want %#v", test.name, got, want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestExtractResumeValidationExcludesOperationalSettings(t *testing.T) {
|
func TestExtractResumeValidationExcludesOperationalSettings(t *testing.T) {
|
||||||
@@ -85,6 +105,8 @@ func TestExtractResumeValidationExcludesOperationalSettings(t *testing.T) {
|
|||||||
env.Config.Pipeline.Notarius.ConfigPath = "/not/present/private.yml"
|
env.Config.Pipeline.Notarius.ConfigPath = "/not/present/private.yml"
|
||||||
env.Config.Pipeline.Notarius.Timeout = "not-a-duration"
|
env.Config.Pipeline.Notarius.Timeout = "not-a-duration"
|
||||||
env.Config.Pipeline.Notarius.WorkingDirectory = "/not/present/work"
|
env.Config.Pipeline.Notarius.WorkingDirectory = "/not/present/work"
|
||||||
|
env.Config.Pipeline.Notarius.Resume = true
|
||||||
|
env.Config.Pipeline.Notarius.ChunkCache = "bypass"
|
||||||
|
|
||||||
validation, err := (extractStage{}).ValidateResume(context.Background(), env, m)
|
validation, err := (extractStage{}).ValidateResume(context.Background(), env, m)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user