|
|
|
|
@@ -17,8 +17,7 @@ import (
|
|
|
|
|
|
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
|
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/core/config"
|
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics"
|
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/core/workspace"
|
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/core/debugbundle"
|
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/framework/checkpoint"
|
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/framework/chunkplan"
|
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
|
|
|
|
@@ -27,11 +26,9 @@ import (
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
const defaultConfigPath = "/usr/local/etc/notarius/config.yml"
|
|
|
|
|
const defaultOutputRoot = "./notarius-output"
|
|
|
|
|
|
|
|
|
|
const usage = `Usage:
|
|
|
|
|
notarius help
|
|
|
|
|
notarius run <pipeline-id> --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--chunk_cache auto|bypass|refresh] [--resume] [--session-id id] [--reference selector=path] [--without-reference selector]
|
|
|
|
|
notarius run <pipeline-id> --input path/to/source.json [--config path/to/config.yml] [--output-dir path] [--chunk_cache auto|bypass|refresh] [--resume] [--debug [--debug-dir path]] [--only lane-a,lane-b] [--session-id id] [--reference selector=path] [--without-reference selector]
|
|
|
|
|
notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b]
|
|
|
|
|
notarius pipelines list --config path/to/config.yml [--json]
|
|
|
|
|
`
|
|
|
|
|
@@ -123,9 +120,10 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
|
|
|
|
|
inputPath := fs.String("input", "", "source input file path")
|
|
|
|
|
onlyRaw := fs.String("only", "", "comma-separated artifact lanes")
|
|
|
|
|
outputDir := fs.String("output-dir", "", "output directory")
|
|
|
|
|
diagnosticsDir := fs.String("diagnostics-dir", "", "diagnostics directory")
|
|
|
|
|
debug := fs.Bool("debug", false, "write a debug bundle")
|
|
|
|
|
debugDir := fs.String("debug-dir", "", "debug bundle directory")
|
|
|
|
|
llmProfile := fs.String("llm-profile", "", "LLM profile override")
|
|
|
|
|
resume := fs.Bool("resume", false, "reuse valid workspace checkpoints")
|
|
|
|
|
resume := fs.Bool("resume", false, "reuse and record compatible checkpoints")
|
|
|
|
|
chunkCache := chunkCacheFlag{}
|
|
|
|
|
sessionID := sessionIDFlag{}
|
|
|
|
|
referenceFlags := stringListFlag{}
|
|
|
|
|
@@ -159,6 +157,18 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
|
|
|
|
|
fmt.Fprintln(stderr, "notarius: run requires --input")
|
|
|
|
|
return 2
|
|
|
|
|
}
|
|
|
|
|
if strings.TrimSpace(*debugDir) != "" && !*debug {
|
|
|
|
|
fmt.Fprintln(stderr, "notarius: --debug-dir requires --debug")
|
|
|
|
|
return 2
|
|
|
|
|
}
|
|
|
|
|
if strings.TrimSpace(*outputDir) == "" && flagWasProvided(args, "--output-dir") {
|
|
|
|
|
fmt.Fprintln(stderr, "notarius: --output-dir must not be empty")
|
|
|
|
|
return 2
|
|
|
|
|
}
|
|
|
|
|
if strings.TrimSpace(*debugDir) == "" && flagWasProvided(args, "--debug-dir") {
|
|
|
|
|
fmt.Fprintln(stderr, "notarius: --debug-dir must not be empty")
|
|
|
|
|
return 2
|
|
|
|
|
}
|
|
|
|
|
if sessionID.set && strings.TrimSpace(sessionID.value) == "" {
|
|
|
|
|
fmt.Fprintln(stderr, "notarius: --session-id must not be empty")
|
|
|
|
|
return 2
|
|
|
|
|
@@ -185,26 +195,38 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
|
|
|
|
|
return 1
|
|
|
|
|
}
|
|
|
|
|
if chunkCache.set {
|
|
|
|
|
cfg.Workspace.ChunkCache.Mode = chunkCache.value
|
|
|
|
|
cfg.Cache.ChunkPlans.Mode = chunkCache.value
|
|
|
|
|
}
|
|
|
|
|
workspaceSettings := workspace.FromConfig(cfg)
|
|
|
|
|
if dir := strings.TrimSpace(*diagnosticsDir); dir != "" {
|
|
|
|
|
workspaceSettings.DiagnosticsRoot = dir
|
|
|
|
|
if dir := strings.TrimSpace(*outputDir); dir != "" {
|
|
|
|
|
cfg.Output.Directory = dir
|
|
|
|
|
}
|
|
|
|
|
if dir := strings.TrimSpace(*debugDir); dir != "" {
|
|
|
|
|
cfg.Debug.Directory = dir
|
|
|
|
|
}
|
|
|
|
|
if err := cfg.Validate(); err != nil {
|
|
|
|
|
fmt.Fprintf(stderr, "notarius: %v\n", err)
|
|
|
|
|
return 1
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
startedAt := opts.Now().UTC()
|
|
|
|
|
runID := fmt.Sprintf("run-%d", startedAt.UnixNano())
|
|
|
|
|
var runDir *diagnostics.RunDirectory
|
|
|
|
|
if workspaceSettings.DiagnosticsEnabled {
|
|
|
|
|
var err error
|
|
|
|
|
runDir, err = diagnostics.NewRunDirectory(workspaceSettings.DiagnosticsRoot, cfg.Diagnostics.Retention)
|
|
|
|
|
var summary *debugbundle.SummaryWriter
|
|
|
|
|
debugPath := ""
|
|
|
|
|
debugRecorder := pipeline.NoopDebugRecorder()
|
|
|
|
|
if *debug {
|
|
|
|
|
bundle, err := debugbundle.Allocate(cfg.Debug.Directory)
|
|
|
|
|
if err != nil {
|
|
|
|
|
fmt.Fprintf(stderr, "notarius: %v\n", err)
|
|
|
|
|
return 1
|
|
|
|
|
}
|
|
|
|
|
runID = runDir.RunID()
|
|
|
|
|
runID, debugPath, summary = bundle.RunID(), bundle.Path(), bundle.Summary()
|
|
|
|
|
debugRecorder, err = frameworkdebug.NewFilesystemRecorder(bundle.TraceRoot())
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("create debug recorder: %w", err), true)
|
|
|
|
|
}
|
|
|
|
|
debugRecorder = pipeline.SynchronizedDebugRecorder(debugRecorder)
|
|
|
|
|
}
|
|
|
|
|
invocation := diagnostics.InvocationMetadata{
|
|
|
|
|
invocation := debugbundle.Invocation{
|
|
|
|
|
Operation: "run",
|
|
|
|
|
PipelineID: pipelineID,
|
|
|
|
|
InputPath: strings.TrimSpace(*inputPath),
|
|
|
|
|
@@ -216,29 +238,17 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
|
|
|
|
|
RunID: runID,
|
|
|
|
|
StartedAt: startedAt,
|
|
|
|
|
}
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error { return runDir.WriteInvocationMetadata(invocation) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics invocation metadata: %w", err))
|
|
|
|
|
if err := writeSummary(summary, func() error { return summary.WriteInvocation(invocation) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("write debug invocation metadata: %w", err), false)
|
|
|
|
|
}
|
|
|
|
|
if *resume && !workspaceSettings.ResumeEnabled {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("--resume requires workspace.resume.enabled: true"))
|
|
|
|
|
}
|
|
|
|
|
debugRoot, err := workspaceSettings.DebugRunDirectory(runID)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("resolve debug root: %w", err))
|
|
|
|
|
}
|
|
|
|
|
debugRecorder, err := frameworkdebug.NewFilesystemRecorder(debugRoot)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("create debug recorder: %w", err))
|
|
|
|
|
}
|
|
|
|
|
debugRecorder = pipeline.SynchronizedDebugRecorder(debugRecorder)
|
|
|
|
|
|
|
|
|
|
catalog, err := effectiveCatalog(opts)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, err, true)
|
|
|
|
|
}
|
|
|
|
|
referenceOverrides, referenceUnbinds, err := resolveCLIReferenceRequests(cfg, pipelineID, only, catalog, referenceRequests, referenceUnbindRequests)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, err, true)
|
|
|
|
|
}
|
|
|
|
|
effective, err := cfg.Resolve(config.ResolveInput{
|
|
|
|
|
PipelineID: pipelineID,
|
|
|
|
|
@@ -249,43 +259,43 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
|
|
|
|
|
ReferenceUnbinds: referenceUnbinds,
|
|
|
|
|
})
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, err, true)
|
|
|
|
|
}
|
|
|
|
|
profileIDs := effectiveLLMProfileIDs(effective.ResolvedPipeline)
|
|
|
|
|
if err := validateExplicitScriptoriumProfiles(context.Background(), effective.Config, profileIDs); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, err, true)
|
|
|
|
|
}
|
|
|
|
|
workingDir, err := os.Getwd()
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("resolve working directory: %w", err))
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("resolve working directory: %w", err), true)
|
|
|
|
|
}
|
|
|
|
|
materialized, referenceWarnings, err := pipeline.MaterializeReferences(effective.ResolvedPipeline, catalog, pipeline.ReferenceMaterializationOptions{
|
|
|
|
|
ConfigPath: loadedConfigPath,
|
|
|
|
|
WorkingDir: workingDir,
|
|
|
|
|
})
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, err, true)
|
|
|
|
|
}
|
|
|
|
|
effective.ResolvedPipeline = materialized
|
|
|
|
|
invocation.PipelineDigest = effective.ResolvedPipeline.Digest
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error { return runDir.WriteInvocationMetadata(invocation) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics invocation metadata: %w", err))
|
|
|
|
|
if err := writeSummary(summary, func() error { return summary.WriteInvocation(invocation) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("write debug invocation metadata: %w", err), false)
|
|
|
|
|
}
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error { return runDir.WriteRedactedEffectiveConfig(effective) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics effective config: %w", err))
|
|
|
|
|
if err := writeSummary(summary, func() error { return summary.WriteRedactedEffectiveConfig(effective) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("write debug effective config: %w", err), false)
|
|
|
|
|
}
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error { return runDir.WriteResolvedPipeline(effective.ResolvedPipeline) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics resolved pipeline: %w", err))
|
|
|
|
|
if err := writeSummary(summary, func() error { return summary.WriteResolvedPipeline(effective.ResolvedPipeline) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("write debug resolved pipeline: %w", err), false)
|
|
|
|
|
}
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error {
|
|
|
|
|
return runDir.WriteResolvedReferences(pipeline.ReferenceProvenance(effective.ResolvedPipeline))
|
|
|
|
|
if err := writeSummary(summary, func() error {
|
|
|
|
|
return summary.WriteResolvedReferences(pipeline.ReferenceProvenance(effective.ResolvedPipeline))
|
|
|
|
|
}); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics resolved references: %w", err))
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("write debug resolved references: %w", err), false)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
registries, err := effectiveRegistries(opts)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, err, true)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
ctx := context.Background()
|
|
|
|
|
@@ -295,24 +305,24 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
|
|
|
|
|
}
|
|
|
|
|
llmClient, llmProfiles, err := opts.LLMClientFactory(ctx, effective.Config, factoryProfileID)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("create LLM client for profile %q: %w", factoryProfileID, err))
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("create LLM client for profile %q: %w", factoryProfileID, err), true)
|
|
|
|
|
}
|
|
|
|
|
llmClient = pipeline.WithDebugLLMRecording(llmClient, debugRecorder)
|
|
|
|
|
prepared, err := pipeline.Prepare(effective.ResolvedPipeline, registries, pipeline.ModuleDependencies{LLM: llmClient})
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("prepare pipeline %q: %w", pipelineID, err))
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("prepare pipeline %q: %w", pipelineID, err), true)
|
|
|
|
|
}
|
|
|
|
|
rawInput, err := os.ReadFile(strings.TrimSpace(*inputPath))
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("read input %q: %w", strings.TrimSpace(*inputPath), err))
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("read input %q: %w", strings.TrimSpace(*inputPath), err), true)
|
|
|
|
|
}
|
|
|
|
|
chunkPlans, err := chunkPlanStoreForRun(effective.Config.Workspace.ChunkCache, opts)
|
|
|
|
|
chunkPlans, err := chunkPlanStoreForRun(effective.Config.Cache.ChunkPlans, opts)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, err, true)
|
|
|
|
|
}
|
|
|
|
|
checkpointRecorder, checkpointLoader, err := checkpointHandlersForRun(workspaceSettings, effective.ResolvedPipeline, rawInput, only, llmProfiles, strings.TrimSpace(*llmProfile), strings.TrimSpace(sessionID.value), *resume)
|
|
|
|
|
checkpointRecorder, checkpointLoader, err := checkpointHandlersForRun(effective.Config.Cache.Checkpoints, opts, effective.ResolvedPipeline, rawInput, only, llmProfiles, strings.TrimSpace(*llmProfile), strings.TrimSpace(sessionID.value), *resume)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, err, true)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
output, err := pipeline.New().Run(ctx, pipeline.RunInput{
|
|
|
|
|
@@ -323,9 +333,9 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
|
|
|
|
|
RunID: runID,
|
|
|
|
|
StartedAt: startedAt,
|
|
|
|
|
LLMProfiles: llmProfiles,
|
|
|
|
|
Metadata: runMetadata(*outputDir, *diagnosticsDir),
|
|
|
|
|
Metadata: runMetadata(effective.Config.Output.Directory, debugPath),
|
|
|
|
|
Warnings: referenceWarnings,
|
|
|
|
|
ChunkCacheMode: effective.Config.Workspace.ChunkCache.Mode,
|
|
|
|
|
ChunkCacheMode: effective.Config.Cache.ChunkPlans.Mode,
|
|
|
|
|
ChunkPlans: chunkPlans,
|
|
|
|
|
Checkpoints: checkpointRecorder,
|
|
|
|
|
Checkpoint: checkpointLoader,
|
|
|
|
|
@@ -333,101 +343,78 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
|
|
|
|
|
ExtractWorkers: cfg.Concurrency.StageWorkers["extract"],
|
|
|
|
|
})
|
|
|
|
|
if err != nil {
|
|
|
|
|
if output.Manifest.PipelineID != "" && runDir != nil {
|
|
|
|
|
_ = runDir.WriteRunManifest(output.Manifest)
|
|
|
|
|
if output.ChunkPlan != nil {
|
|
|
|
|
_ = runDir.WriteChunkPlan(*output.ChunkPlan)
|
|
|
|
|
if output.Manifest.PipelineID != "" {
|
|
|
|
|
if summaryErr := writePartialSummary(summary, output); summaryErr != nil {
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("run pipeline %q: %w; write debug summary: %v", pipelineID, err, summaryErr), false)
|
|
|
|
|
}
|
|
|
|
|
_ = runDir.WriteCheckpointEvents(output.CheckpointEvents)
|
|
|
|
|
}
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("run pipeline %q: %w", pipelineID, err))
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("run pipeline %q: %w", pipelineID, err), true)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
runOutputDir := filepath.Join(outputRoot(*outputDir), runID)
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error { return runDir.WriteRunManifest(output.Manifest) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics run manifest: %w", err))
|
|
|
|
|
}
|
|
|
|
|
if output.ChunkPlan != nil {
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error { return runDir.WriteChunkPlan(*output.ChunkPlan) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics chunk plan: %w", err))
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error { return runDir.WriteWarnings(output.Warnings) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics warnings: %w", err))
|
|
|
|
|
}
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error { return runDir.WriteCheckpointEvents(output.CheckpointEvents) }); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics checkpoint events: %w", err))
|
|
|
|
|
}
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error {
|
|
|
|
|
return runDir.WriteRunReport(runReport{
|
|
|
|
|
RunID: runDir.RunID(),
|
|
|
|
|
PipelineID: effective.PipelineID,
|
|
|
|
|
OutputPath: runOutputDir,
|
|
|
|
|
DiagnosticsPath: runDir.Path(),
|
|
|
|
|
OutputCount: len(output.NormalizeOutputs),
|
|
|
|
|
RejectedCount: len(output.Rejected),
|
|
|
|
|
WarningCount: len(output.Warnings),
|
|
|
|
|
ValidationStatus: output.Manifest.ValidationStatus,
|
|
|
|
|
})
|
|
|
|
|
}); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics run report: %w", err))
|
|
|
|
|
runOutputDir := filepath.Join(effective.Config.Output.Directory, runID)
|
|
|
|
|
if err := writePartialSummary(summary, output); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("write debug summary: %w", err), false)
|
|
|
|
|
}
|
|
|
|
|
if err := writeOutputFiles(runOutputDir, output.OutputFiles); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, err, true)
|
|
|
|
|
}
|
|
|
|
|
if err := writeDiagnostics(runDir, func() error {
|
|
|
|
|
return runDir.ApplyRetention(diagnostics.RetentionDecisionInput{
|
|
|
|
|
RetentionMode: cfg.Diagnostics.Retention,
|
|
|
|
|
RunSucceeded: true,
|
|
|
|
|
HasWarnings: len(output.Warnings) > 0,
|
|
|
|
|
})
|
|
|
|
|
if err := writeSummary(summary, func() error {
|
|
|
|
|
return summary.WriteRunReport(debugbundle.RunReport{RunID: runID, PipelineID: effective.PipelineID, OutputPath: runOutputDir, DebugPath: debugPath, Succeeded: true, OutputCount: len(output.NormalizeOutputs), RejectedCount: len(output.Rejected), WarningCount: len(output.Warnings), ValidationStatus: output.Manifest.ValidationStatus})
|
|
|
|
|
}); err != nil {
|
|
|
|
|
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("apply diagnostics retention: %w", err))
|
|
|
|
|
return failPipelineCommand(stderr, summary, debugPath, fmt.Errorf("write debug run report: %w", err), false)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fmt.Fprintf(stdout, "pipeline %q complete: outputs=%d rejected=%d output=%s\n", effective.PipelineID, len(output.NormalizeOutputs), len(output.Rejected), runOutputDir)
|
|
|
|
|
if debugPath != "" {
|
|
|
|
|
fmt.Fprintf(stdout, "debug=%s\n", debugPath)
|
|
|
|
|
}
|
|
|
|
|
if len(output.Warnings) > 0 {
|
|
|
|
|
fmt.Fprintf(stderr, "notarius: run completed with %d warning(s)\n", len(output.Warnings))
|
|
|
|
|
}
|
|
|
|
|
return 0
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
type runReport struct {
|
|
|
|
|
RunID string `json:"run_id"`
|
|
|
|
|
PipelineID string `json:"pipeline_id"`
|
|
|
|
|
OutputPath string `json:"output_path"`
|
|
|
|
|
DiagnosticsPath string `json:"diagnostics_path,omitempty"`
|
|
|
|
|
OutputCount int `json:"output_count"`
|
|
|
|
|
RejectedCount int `json:"rejected_count"`
|
|
|
|
|
WarningCount int `json:"warning_count"`
|
|
|
|
|
ValidationStatus string `json:"validation_status,omitempty"`
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func failPipelineCommand(stderr io.Writer, runDir *diagnostics.RunDirectory, retention diagnostics.RetentionMode, err error) int {
|
|
|
|
|
func failPipelineCommand(stderr io.Writer, summary *debugbundle.SummaryWriter, debugPath string, err error, recordError bool) int {
|
|
|
|
|
fmt.Fprintf(stderr, "notarius: %v\n", err)
|
|
|
|
|
if runDir != nil {
|
|
|
|
|
if logErr := runDir.WriteErrorLog(err.Error()); logErr != nil {
|
|
|
|
|
fmt.Fprintf(stderr, "notarius: write diagnostics error log: %v\n", logErr)
|
|
|
|
|
}
|
|
|
|
|
if retentionErr := runDir.ApplyRetention(diagnostics.RetentionDecisionInput{
|
|
|
|
|
RetentionMode: retention,
|
|
|
|
|
RunSucceeded: false,
|
|
|
|
|
}); retentionErr != nil {
|
|
|
|
|
fmt.Fprintf(stderr, "notarius: apply diagnostics retention: %v\n", retentionErr)
|
|
|
|
|
if recordError && summary != nil {
|
|
|
|
|
if summaryErr := summary.WriteError(err.Error()); summaryErr != nil {
|
|
|
|
|
fmt.Fprintf(stderr, "notarius: write debug error log: %v\n", summaryErr)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
if debugPath != "" {
|
|
|
|
|
fmt.Fprintf(stderr, "notarius: debug=%s\n", debugPath)
|
|
|
|
|
}
|
|
|
|
|
return 1
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func writeDiagnostics(runDir *diagnostics.RunDirectory, write func() error) error {
|
|
|
|
|
if runDir == nil {
|
|
|
|
|
func writeSummary(summary *debugbundle.SummaryWriter, write func() error) error {
|
|
|
|
|
if summary == nil {
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
return write()
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func writePartialSummary(summary *debugbundle.SummaryWriter, output pipeline.RunOutput) error {
|
|
|
|
|
if summary == nil {
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
if err := summary.WriteRunManifest(output.Manifest); err != nil {
|
|
|
|
|
return err
|
|
|
|
|
}
|
|
|
|
|
if output.ChunkPlan != nil {
|
|
|
|
|
if err := summary.WriteChunkPlan(*output.ChunkPlan); err != nil {
|
|
|
|
|
return err
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
if err := summary.WriteWarnings(output.Warnings); err != nil {
|
|
|
|
|
return err
|
|
|
|
|
}
|
|
|
|
|
return summary.WriteCheckpointEvents(output.CheckpointEvents)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func checkpointHandlersForRun(
|
|
|
|
|
settings workspace.Settings,
|
|
|
|
|
settings config.CheckpointCacheConfig,
|
|
|
|
|
opts Options,
|
|
|
|
|
resolved pipeline.ResolvedPipeline,
|
|
|
|
|
rawInput []byte,
|
|
|
|
|
only []string,
|
|
|
|
|
@@ -436,6 +423,9 @@ func checkpointHandlersForRun(
|
|
|
|
|
sessionID string,
|
|
|
|
|
resume bool,
|
|
|
|
|
) (pipeline.CheckpointRecorder, pipeline.CheckpointLoader, error) {
|
|
|
|
|
if !resume {
|
|
|
|
|
return pipeline.NoopCheckpointRecorder(), pipeline.NoopCheckpointLoader(), nil
|
|
|
|
|
}
|
|
|
|
|
identity, err := checkpoint.NewIdentity(checkpoint.IdentityInput{
|
|
|
|
|
Pipeline: resolved,
|
|
|
|
|
InputKey: resolved.Input.Module,
|
|
|
|
|
@@ -448,20 +438,20 @@ func checkpointHandlersForRun(
|
|
|
|
|
if err != nil {
|
|
|
|
|
return nil, nil, fmt.Errorf("create checkpoint identity: %w", err)
|
|
|
|
|
}
|
|
|
|
|
checkpointRoot := ""
|
|
|
|
|
if settings.ResumeEnabled {
|
|
|
|
|
checkpointRoot = settings.CheckpointsRoot
|
|
|
|
|
checkpointRoot := strings.TrimSpace(settings.Directory)
|
|
|
|
|
if checkpointRoot == "" {
|
|
|
|
|
checkpointRoot, err = config.DefaultCheckpointRoot(opts.UserCacheDir)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return nil, nil, fmt.Errorf("resolve checkpoint root: %w", err)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
recorder, err := checkpoint.NewFilesystemRecorder(checkpointRoot, identity)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return nil, nil, fmt.Errorf("create checkpoint recorder: %w", err)
|
|
|
|
|
}
|
|
|
|
|
loader := pipeline.NoopCheckpointLoader()
|
|
|
|
|
if resume {
|
|
|
|
|
loader, err = checkpoint.NewFilesystemLoader(checkpointRoot, identity)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return nil, nil, fmt.Errorf("create checkpoint loader: %w", err)
|
|
|
|
|
}
|
|
|
|
|
loader, err := checkpoint.NewFilesystemLoader(checkpointRoot, identity)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return nil, nil, fmt.Errorf("create checkpoint loader: %w", err)
|
|
|
|
|
}
|
|
|
|
|
return recorder, loader, nil
|
|
|
|
|
}
|
|
|
|
|
@@ -507,13 +497,6 @@ func configSource(configPath string) string {
|
|
|
|
|
return "discovered"
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func outputRoot(outputDir string) string {
|
|
|
|
|
if dir := strings.TrimSpace(outputDir); dir != "" {
|
|
|
|
|
return dir
|
|
|
|
|
}
|
|
|
|
|
return defaultOutputRoot
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func writeOutputFiles(runOutputDir string, files []contracts.OutputFile) error {
|
|
|
|
|
type outputTarget struct {
|
|
|
|
|
path string
|
|
|
|
|
@@ -635,7 +618,7 @@ func reorderRunArgs(args []string) []string {
|
|
|
|
|
|
|
|
|
|
func runFlagTakesValue(arg string) bool {
|
|
|
|
|
switch arg {
|
|
|
|
|
case "--config", "--input", "--only", "--output-dir", "--diagnostics-dir", "--llm-profile", "--session-id", "--chunk_cache", "--reference", "--without-reference":
|
|
|
|
|
case "--config", "--input", "--only", "--output-dir", "--debug-dir", "--llm-profile", "--session-id", "--chunk_cache", "--reference", "--without-reference":
|
|
|
|
|
return true
|
|
|
|
|
default:
|
|
|
|
|
return false
|
|
|
|
|
@@ -671,7 +654,7 @@ func (f chunkCacheFlag) explicitValue() string {
|
|
|
|
|
return string(f.value)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func chunkPlanStoreForRun(cfg config.WorkspaceChunkCacheConfig, opts Options) (pipeline.ChunkPlanStore, error) {
|
|
|
|
|
func chunkPlanStoreForRun(cfg config.ChunkPlanCacheConfig, opts Options) (pipeline.ChunkPlanStore, error) {
|
|
|
|
|
if cfg.Mode == pipeline.ChunkCacheBypass {
|
|
|
|
|
return nil, nil
|
|
|
|
|
}
|
|
|
|
|
@@ -705,6 +688,15 @@ func validateRunFlagValues(args []string) error {
|
|
|
|
|
return nil
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func flagWasProvided(args []string, name string) bool {
|
|
|
|
|
for _, arg := range args {
|
|
|
|
|
if arg == name || strings.HasPrefix(arg, name+"=") {
|
|
|
|
|
return true
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
return false
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func effectiveLLMProfileIDs(resolved pipeline.ResolvedPipeline) []string {
|
|
|
|
|
seen := make(map[string]struct{})
|
|
|
|
|
add := func(binding pipeline.ModuleBinding) {
|
|
|
|
|
@@ -734,13 +726,13 @@ func effectiveLLMProfileIDs(resolved pipeline.ResolvedPipeline) []string {
|
|
|
|
|
return ids
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func runMetadata(outputDir, diagnosticsDir string) map[string]any {
|
|
|
|
|
func runMetadata(outputDir, debugDir string) map[string]any {
|
|
|
|
|
metadata := make(map[string]any)
|
|
|
|
|
if dir := strings.TrimSpace(outputDir); dir != "" {
|
|
|
|
|
metadata["output_dir"] = dir
|
|
|
|
|
}
|
|
|
|
|
if dir := strings.TrimSpace(diagnosticsDir); dir != "" {
|
|
|
|
|
metadata["diagnostics_dir"] = dir
|
|
|
|
|
if dir := strings.TrimSpace(debugDir); dir != "" {
|
|
|
|
|
metadata["debug_dir"] = dir
|
|
|
|
|
}
|
|
|
|
|
if len(metadata) == 0 {
|
|
|
|
|
return nil
|
|
|
|
|
|