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

Simplify stage and Audita contracts

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-10 22:43:22 +00:00
parent 131ffd9887
commit b89224bbde
21 changed files with 47 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
internal/stage/analyze.go
View File

@@ -21,17 +21,6 @@ type analyzeStage struct{}
func (analyzeStage) Name() string { return "analyze" }
func (analyzeStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{
{Kind: "transcript_polished", Category: "transcripts", RelativePath: "transcripts/polished.json"},
{Kind: "transcript_final", Category: "transcripts", RelativePath: "transcripts/final.json"},
{Kind: "transcript_final_trimmed", Category: "transcripts", RelativePath: "transcripts/final.trimmed.json"},
},
Outputs: nil,
}
}
type analyzeArtifactExecutionPlan struct {
Name string
Cfg config.ScriptoriumArtifactConfig

15
internal/stage/extract.go
View File

@@ -33,21 +33,6 @@ type extractStage struct{}
func (extractStage) Name() string { return "extract" }
func (extractStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{{
Kind: artifactmodel.TranscriptOutputKindFinalTrimmed,
SourceID: artifactmodel.SourceTranscriptFinalTrimmed,
Category: "transcripts",
RelativePath: artifactmodel.TranscriptPathFinalTrimmed,
}},
Outputs: []artifacts.Ref{
{Kind: extractLaneOutputKind, Category: "artifacts", RelativePath: "artifacts/notarius/<run-id>/lanes/*.json"},
{Kind: extractIndexOutputKind, Category: "artifacts", RelativePath: "artifacts/notarius/<run-id>/index.json"},
},
}
}
func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil {
return nil, fmt.Errorf("extract: resolved stage environment config is required")

14
internal/stage/merge.go
View File

@@ -19,20 +19,6 @@ type mergeStage struct{}
func (mergeStage) Name() string { return "merge" }
func (mergeStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{
{Kind: "transcript_raw", Category: "transcripts", RelativePath: "transcripts/raw/*.json"},
{Kind: "speakers", Category: "inputs", RelativePath: "inputs/speakers.yml"},
{Kind: "autocorrect", Category: "inputs", RelativePath: "inputs/autocorrect.yml"},
},
Outputs: []artifacts.Ref{
{Kind: "transcript_base", Category: "transcripts", RelativePath: "transcripts/base.json"},
{Kind: "seriatim_report", Category: "artifacts", RelativePath: "artifacts/seriatim.report.json"},
},
}
}
func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
if env == nil || env.Config == nil {
return nil, fmt.Errorf("merge: stage environment config is required")

12
internal/stage/normalize.go
View File

@@ -16,18 +16,6 @@ type normalizeStage struct{}
func (normalizeStage) Name() string { return "normalize" }
func (normalizeStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{
{Kind: "transcript_polished", Category: "transcripts", RelativePath: "transcripts/polished.json"},
},
Outputs: []artifacts.Ref{
{Kind: "transcript_final", Category: "transcripts", RelativePath: "transcripts/final.json"},
{Kind: "seriatim_normalize_report", Category: "artifacts", RelativePath: "artifacts/seriatim.normalize.report.json"},
},
}
}
func (normalizeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
if env == nil || env.Config == nil {
return nil, fmt.Errorf("normalize: stage environment config is required")

8
internal/stage/placeholders.go
View File

@@ -5,7 +5,6 @@ import (
"fmt"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/notify"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)
@@ -17,13 +16,6 @@ type placeholderStage struct {
func (s placeholderStage) Name() string { return s.name }
func (s placeholderStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{{Kind: "artifact", Category: "input", RelativePath: s.name + ".input.placeholder"}},
Outputs: []artifacts.Ref{{Kind: "artifact", Category: "output", RelativePath: s.name + ".output.placeholder"}},
}
}
func (s placeholderStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
result := &StageResult{
Metadata: map[string]any{

56
internal/stage/polish.go
View File

@@ -16,19 +16,6 @@ type polishStage struct{}
func (polishStage) Name() string { return "polish" }
func (polishStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{
{Kind: "transcript_base", Category: "transcripts", RelativePath: "transcripts/base.json"},
{Kind: "glossary", Category: "inputs", RelativePath: "inputs/glossary.yml"},
},
Outputs: []artifacts.Ref{
{Kind: "transcript_polished", Category: "transcripts", RelativePath: "transcripts/polished.json"},
{Kind: "audita_report", Category: "artifacts", RelativePath: "artifacts/audita.report.json"},
},
}
}
func (polishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
if env == nil || env.Config == nil {
return nil, fmt.Errorf("polish: stage environment config is required")
@@ -101,25 +88,14 @@ func (polishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*St
reportEnabled := env.Config.Pipeline.Audita.Report != nil && *env.Config.Pipeline.Audita.Report
req := audita.PolishRequest{
GeneratedConfigPath: generatedConfigPath,
MergedTranscriptPath: mergedPath,
OutputProcessedPath: processedPath,
GlossaryPath: glossaryPath,
ReportPath: "",
WorkDir: workDir,
Modules: append([]string(nil), env.Config.Pipeline.Audita.Modules...),
BaseURL: env.Config.Pipeline.Audita.BaseURL,
Model: env.Config.Pipeline.Audita.Model,
TranscriptDescription: env.Config.Pipeline.Audita.TranscriptDescription,
ConfigPath: env.Config.Pipeline.Audita.ConfigPath,
OutputSchema: env.Config.Pipeline.Audita.OutputSchema,
WorkDirRetention: env.Config.Pipeline.Audita.WorkDirRetention,
TotalLLMConcurrency: env.Config.Pipeline.Audita.TotalLLMConcurrency,
ProposalLLMConcurrency: env.Config.Pipeline.Audita.ProposalLLMConcurrency,
ValidationModel: env.Config.Pipeline.Audita.ValidationModel,
ValidationLLMConcurrency: env.Config.Pipeline.Audita.ValidationLLMConcurrency,
StdoutLogPath: stdoutPath,
StderrLogPath: stderrPath,
GeneratedConfigPath: generatedConfigPath,
MergedTranscriptPath: mergedPath,
OutputProcessedPath: processedPath,
GlossaryPath: glossaryPath,
ReportPath: "",
WorkDir: workDir,
StdoutLogPath: stdoutPath,
StderrLogPath: stderrPath,
}
if reportEnabled {
req.ReportPath = reportPath
@@ -197,14 +173,14 @@ func (polishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*St
"report_path": reportCanonicalPath,
"audita_work_dir": workDir,
"report_enabled": reportEnabled,
"modules": append([]string(nil), req.Modules...),
"base_url": req.BaseURL,
"model": req.Model,
"transcript_description": req.TranscriptDescription,
"config_path": req.ConfigPath,
"output_schema": req.OutputSchema,
"work_dir_retention": req.WorkDirRetention,
"validation_model": req.ValidationModel,
"modules": append([]string(nil), env.Config.Pipeline.Audita.Modules...),
"base_url": env.Config.Pipeline.Audita.BaseURL,
"model": env.Config.Pipeline.Audita.Model,
"transcript_description": env.Config.Pipeline.Audita.TranscriptDescription,
"config_path": env.Config.Pipeline.Audita.ConfigPath,
"output_schema": env.Config.Pipeline.Audita.OutputSchema,
"work_dir_retention": env.Config.Pipeline.Audita.WorkDirRetention,
"validation_model": env.Config.Pipeline.Audita.ValidationModel,
"total_llm_concurrency": totalLLMConcurrency,
"proposal_llm_concurrency": proposalLLMConcurrency,
"validation_llm_concurrency": validationConcurrency,

34
internal/stage/polish_test.go
View File

@@ -48,40 +48,6 @@ func TestPolishStagePolishesMergedTranscriptAndRecordsMetadata(t *testing.T) {
if req.WorkDir != filepath.Join(paths.ArtifactsDir, "audita-work") {
t.Fatalf("work dir = %q", req.WorkDir)
}
if strings.Join(req.Modules, ",") != "glossary,homophones,grammar" {
t.Fatalf("modules = %#v", req.Modules)
}
if req.BaseURL != "https://openrouter.ai/api/v1" {
t.Fatalf("base url = %q", req.BaseURL)
}
if req.Model != "openrouter/google/gemma-4-31b-it" {
t.Fatalf("model = %q", req.Model)
}
if req.ValidationModel != "openrouter/google/gemma-4-31b-it" {
t.Fatalf("validation model = %q", req.ValidationModel)
}
if req.TranscriptDescription != "Campaign Session 42" {
t.Fatalf("transcript description = %q", req.TranscriptDescription)
}
if req.ConfigPath != "/etc/audita/config.yml" {
t.Fatalf("config path = %q", req.ConfigPath)
}
if req.OutputSchema != "audita-v1" {
t.Fatalf("output schema = %q", req.OutputSchema)
}
if req.WorkDirRetention != "auto" {
t.Fatalf("work dir retention = %q", req.WorkDirRetention)
}
if req.TotalLLMConcurrency == nil || *req.TotalLLMConcurrency != 3 {
t.Fatalf("total llm concurrency = %#v, want 3", req.TotalLLMConcurrency)
}
if req.ProposalLLMConcurrency == nil || *req.ProposalLLMConcurrency != 2 {
t.Fatalf("proposal llm concurrency = %#v, want 2", req.ProposalLLMConcurrency)
}
if req.ValidationLLMConcurrency == nil || *req.ValidationLLMConcurrency != 2 {
t.Fatalf("validation llm concurrency = %#v, want 2", req.ValidationLLMConcurrency)
}
if len(result.Outputs) != 2 {
t.Fatalf("outputs len = %d, want 2", len(result.Outputs))
}

16
internal/stage/prepare.go
View File

@@ -24,22 +24,6 @@ type prepareStage struct{}
func (prepareStage) Name() string { return "prepare" }
func (prepareStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{
{Kind: "config", Category: "inputs", RelativePath: "campaign.yml"},
{Kind: "config", Category: "inputs", RelativePath: "session.yml"},
{Kind: "config", Category: "inputs", RelativePath: "pipeline.resolved.yml"},
{Kind: "config", Category: "inputs", RelativePath: "speakers.yml"},
{Kind: "config", Category: "inputs", RelativePath: "autocorrect.yml"},
{Kind: "config", Category: "inputs", RelativePath: "glossary.yml"},
{Kind: "config", Category: "inputs", RelativePath: "players.yml"},
{Kind: "config", Category: "inputs", RelativePath: "party.yml"},
{Kind: "audio", Category: "audio", RelativePath: "*.flac"},
},
}
}
func (prepareStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
if env == nil || env.Config == nil {
return nil, fmt.Errorf("prepare: stage environment config is required")

26
internal/stage/publish.go
View File

@@ -43,14 +43,6 @@ var publishPrerequisiteStages = []string{
func (publishStage) Name() string { return "publish" }
func (publishStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{
{Kind: "manifest", Category: "input", RelativePath: "manifest.json"},
},
}
}
func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil {
return nil, fmt.Errorf("publish: resolved config must include pipeline and session")
@@ -671,24 +663,6 @@ func resolvePublishRunManifestPath(env *Env, m *manifest.Manifest) (string, erro
return canonical, nil
}
func resolvePublishSessionRoot(env *Env, m *manifest.Manifest) (string, error) {
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
if sessionID == "" && m != nil {
sessionID = strings.TrimSpace(m.SessionID)
}
campaign := strings.TrimSpace(env.Config.Session.Campaign)
if campaign == "" && m != nil {
campaign = strings.TrimSpace(m.Campaign)
}
if sessionID == "" {
return "", fmt.Errorf("session id is required")
}
if campaign == "" {
return "", fmt.Errorf("campaign is required")
}
return filepath.Clean(artifacts.SessionWorkDirForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID)), nil
}
func publishSessionPaths(env *Env, m *manifest.Manifest) artifacts.SessionPaths {
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
if sessionID == "" && m != nil {

13
internal/stage/render.go
View File

@@ -17,19 +17,6 @@ type renderStage struct{}
func (renderStage) Name() string { return "render" }
func (renderStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{
{Kind: "transcript_final", Category: "transcripts", RelativePath: "transcripts/final.json"},
{Kind: "transcript_final_trimmed", Category: "transcripts", RelativePath: "transcripts/final.trimmed.json"},
},
Outputs: []artifacts.Ref{
{Kind: artifacts.TranscriptOutputKindFinalMarkdown, Category: "transcripts", RelativePath: artifacts.TranscriptPathFinalMarkdown},
{Kind: artifacts.TranscriptOutputKindFinalTrimmedMarkdown, Category: "transcripts", RelativePath: artifacts.TranscriptPathFinalTrimmedMarkdown},
},
}
}
func (renderStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
if env == nil || env.Config == nil {
return nil, fmt.Errorf("render: stage environment config is required")

7
internal/stage/stage.go
View File

@@ -40,16 +40,9 @@ type Env struct {
RevalidatePublishLocks func(context.Context) ([]config.PublishLockRule, error)
}
// IODecl declares the intended input/output artifact kinds for a stage.
type IODecl struct {
Inputs []artifacts.Ref
Outputs []artifacts.Ref
}
// Stage is the pipeline unit contract.
type Stage interface {
Name() string
Declares() IODecl
Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error)
}

11
internal/stage/transcribe.go
View File

@@ -19,17 +19,6 @@ type transcribeStage struct{}
func (transcribeStage) Name() string { return "transcribe" }
func (transcribeStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{
{Kind: "audio", Category: "audio", RelativePath: "audio/*.flac"},
},
Outputs: []artifacts.Ref{
{Kind: "transcript_raw", Category: "transcripts", RelativePath: "transcripts/raw/*.json"},
},
}
}
func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
if env == nil || env.Config == nil {
return nil, fmt.Errorf("transcribe: stage environment config is required")

12
internal/stage/trim.go
View File

@@ -20,18 +20,6 @@ type trimStage struct{}
func (trimStage) Name() string { return "trim" }
func (trimStage) Declares() IODecl {
return IODecl{
Inputs: []artifacts.Ref{
{Kind: "transcript_final", Category: "transcripts", RelativePath: "transcripts/final.json"},
},
Outputs: []artifacts.Ref{
{Kind: "transcript_final_trimmed", Category: "transcripts", RelativePath: "transcripts/final.trimmed.json"},
{Kind: "session_bounds", Category: "artifacts", RelativePath: "artifacts/session_bounds.json"},
},
}
}
func (trimStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
if env == nil || env.Config == nil {
return nil, fmt.Errorf("trim: stage environment config is required")