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

Add in-memory pipeline run command

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-04 01:06:04 +00:00
parent 361b1f53f4
commit ae218d7c57
4 changed files with 569 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

182
internal/cli/run.go
View File

@@ -21,6 +21,7 @@ const defaultConfigPath = "/usr/local/etc/notarius/config.yml"
const usage = `Usage:
notarius help
notarius run <pipeline-id> --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b]
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]
`
@@ -55,6 +56,8 @@ func RunWithOptions(args []string, stdout, stderr io.Writer, opts Options) int {
return runConfig(args[1:], stdout, stderr, opts)
case "pipelines":
return runPipelines(args[1:], stdout, stderr, opts)
case "run":
return runPipelineCommand(args[1:], stdout, stderr, opts)
default:
fmt.Fprintf(stderr, "notarius: unknown command %q\n", args[0])
writeUsage(stderr)
@@ -79,6 +82,185 @@ func normalizeOptions(opts Options) Options {
return opts
}
func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) int {
fs := flag.NewFlagSet("run", flag.ContinueOnError)
fs.SetOutput(io.Discard)
configPath := fs.String("config", "", "config file path")
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")
llmProfile := fs.String("llm-profile", "", "LLM profile override")
if err := fs.Parse(reorderRunArgs(args)); err != nil {
fmt.Fprintf(stderr, "notarius: %v\n", err)
return 2
}
if fs.NArg() == 0 {
fmt.Fprintln(stderr, "notarius: run requires a pipeline ID")
return 2
}
if fs.NArg() > 1 {
fmt.Fprintf(stderr, "notarius: unexpected argument %q\n", fs.Arg(1))
return 2
}
pipelineID := strings.TrimSpace(fs.Arg(0))
if pipelineID == "" {
fmt.Fprintln(stderr, "notarius: run requires a pipeline ID")
return 2
}
if strings.TrimSpace(*inputPath) == "" {
fmt.Fprintln(stderr, "notarius: run requires --input")
return 2
}
only, err := parseOnly(*onlyRaw)
if err != nil {
fmt.Fprintf(stderr, "notarius: %v\n", err)
return 2
}
cfg, _, err := loadConfig(*configPath, opts)
if err != nil {
fmt.Fprintf(stderr, "notarius: %v\n", err)
return 1
}
if dir := strings.TrimSpace(*diagnosticsDir); dir != "" {
cfg.Diagnostics.WorkDir = dir
}
catalog, err := effectiveCatalog(opts)
if err != nil {
fmt.Fprintf(stderr, "notarius: %v\n", err)
return 1
}
effective, err := cfg.Resolve(config.ResolveInput{
PipelineID: pipelineID,
Only: only,
Catalog: catalog,
LLMProfileOverride: *llmProfile,
})
if err != nil {
fmt.Fprintf(stderr, "notarius: %v\n", err)
return 1
}
profileIDs := effectiveLLMProfileIDs(effective.ResolvedPipeline)
if len(profileIDs) != 1 {
fmt.Fprintf(stderr, "notarius: pipeline %q uses %d distinct LLM profiles; current runs require exactly one: %s\n", pipelineID, len(profileIDs), strings.Join(profileIDs, ", "))
return 1
}
rawInput, err := os.ReadFile(strings.TrimSpace(*inputPath))
if err != nil {
fmt.Fprintf(stderr, "notarius: read input %q: %v\n", strings.TrimSpace(*inputPath), err)
return 1
}
registries, err := effectiveRegistries(opts)
if err != nil {
fmt.Fprintf(stderr, "notarius: %v\n", err)
return 1
}
ctx := context.Background()
llmClient, llmProfiles, err := opts.LLMClientFactory(ctx, effective.Config, profileIDs[0])
if err != nil {
fmt.Fprintf(stderr, "notarius: create LLM client for profile %q: %v\n", profileIDs[0], err)
return 1
}
output, err := pipeline.New(registries).Run(ctx, pipeline.RunInput{
Pipeline: effective.ResolvedPipeline,
Path: strings.TrimSpace(*inputPath),
RawInput: rawInput,
LLMClient: llmClient,
StartedAt: opts.Now().UTC(),
LLMProfiles: llmProfiles,
Metadata: runMetadata(*outputDir, *diagnosticsDir),
})
if err != nil {
fmt.Fprintf(stderr, "notarius: run pipeline %q: %v\n", pipelineID, err)
return 1
}
fmt.Fprintf(stdout, "pipeline %q complete: approved=%d rejected=%d\n", effective.PipelineID, len(output.Approved), len(output.Rejected))
if len(output.Warnings) > 0 {
fmt.Fprintf(stderr, "notarius: run completed with %d warning(s)\n", len(output.Warnings))
}
return 0
}
func reorderRunArgs(args []string) []string {
var flags []string
var positionals []string
for i := 0; i < len(args); i++ {
arg := args[i]
if arg == "--" {
positionals = append(positionals, args[i+1:]...)
break
}
if strings.HasPrefix(arg, "-") {
flags = append(flags, arg)
if runFlagTakesValue(arg) && !strings.Contains(arg, "=") && i+1 < len(args) {
i++
flags = append(flags, args[i])
}
continue
}
positionals = append(positionals, arg)
}
return append(flags, positionals...)
}
func runFlagTakesValue(arg string) bool {
switch arg {
case "--config", "--input", "--only", "--output-dir", "--diagnostics-dir", "--llm-profile":
return true
default:
return false
}
}
func effectiveLLMProfileIDs(resolved pipeline.ResolvedPipeline) []string {
seen := make(map[string]struct{})
add := func(binding pipeline.ModuleBinding) {
id := strings.TrimSpace(binding.LLMProfile)
if id != "" {
seen[id] = struct{}{}
}
}
add(resolved.Input)
add(resolved.Chunk)
add(resolved.Output)
for _, lane := range resolved.ArtifactLanes {
add(lane.Extract)
add(lane.Merge)
add(lane.Normalize)
for _, validator := range lane.Validators {
add(validator)
}
}
ids := make([]string, 0, len(seen))
for id := range seen {
ids = append(ids, id)
}
sort.Strings(ids)
return ids
}
func runMetadata(outputDir, diagnosticsDir 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 len(metadata) == 0 {
return nil
}
return metadata
}
func runConfig(args []string, stdout, stderr io.Writer, opts Options) int {
if len(args) == 0 {
fmt.Fprintln(stderr, "notarius: config requires a subcommand")