Update the analyze stage to utilize the new artifact package
This commit is contained in:
294
internal/artifacts/artifact_resolver.go
Normal file
294
internal/artifacts/artifact_resolver.go
Normal file
@@ -0,0 +1,294 @@
|
|||||||
|
package artifacts
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
ArtifactTranscriptMerged = "narratio.transcript.merged"
|
||||||
|
ArtifactTranscriptPolished = "narratio.transcript.polished"
|
||||||
|
ArtifactTranscriptFull = "narratio.transcript.full"
|
||||||
|
ArtifactTranscriptTrimmed = "narratio.transcript.trimmed"
|
||||||
|
ArtifactBoundsSession = "narratio.bounds.session"
|
||||||
|
ArtifactSessionRecap = "narratio.artifact.session_recap"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ErrSessionArtifactNotFound is returned when no readable artifact exists for a known ID.
|
||||||
|
var ErrSessionArtifactNotFound = errors.New("session artifact not found")
|
||||||
|
|
||||||
|
type artifactContentKind string
|
||||||
|
|
||||||
|
const (
|
||||||
|
contentTranscriptJSON artifactContentKind = "transcript_json"
|
||||||
|
contentJSON artifactContentKind = "json"
|
||||||
|
contentText artifactContentKind = "text"
|
||||||
|
)
|
||||||
|
|
||||||
|
type artifactSpec struct {
|
||||||
|
ID string
|
||||||
|
CanonicalRelPath string
|
||||||
|
ProducerStage string
|
||||||
|
OutputKind string
|
||||||
|
ContentKind artifactContentKind
|
||||||
|
}
|
||||||
|
|
||||||
|
var artifactRegistry = map[string]artifactSpec{
|
||||||
|
ArtifactTranscriptMerged: {
|
||||||
|
ID: ArtifactTranscriptMerged,
|
||||||
|
CanonicalRelPath: "transcripts/merged.json",
|
||||||
|
ProducerStage: "merge",
|
||||||
|
OutputKind: "transcript_merged",
|
||||||
|
ContentKind: contentTranscriptJSON,
|
||||||
|
},
|
||||||
|
ArtifactTranscriptPolished: {
|
||||||
|
ID: ArtifactTranscriptPolished,
|
||||||
|
CanonicalRelPath: "transcripts/processed.json",
|
||||||
|
ProducerStage: "polish",
|
||||||
|
OutputKind: "transcript_processed",
|
||||||
|
ContentKind: contentTranscriptJSON,
|
||||||
|
},
|
||||||
|
ArtifactTranscriptFull: {
|
||||||
|
ID: ArtifactTranscriptFull,
|
||||||
|
CanonicalRelPath: "transcripts/normalized.json",
|
||||||
|
ProducerStage: "normalize",
|
||||||
|
OutputKind: "transcript_normalized",
|
||||||
|
ContentKind: contentTranscriptJSON,
|
||||||
|
},
|
||||||
|
ArtifactTranscriptTrimmed: {
|
||||||
|
ID: ArtifactTranscriptTrimmed,
|
||||||
|
CanonicalRelPath: "transcripts/trimmed.json",
|
||||||
|
ProducerStage: "trim",
|
||||||
|
OutputKind: "transcript_trimmed",
|
||||||
|
ContentKind: contentTranscriptJSON,
|
||||||
|
},
|
||||||
|
ArtifactBoundsSession: {
|
||||||
|
ID: ArtifactBoundsSession,
|
||||||
|
CanonicalRelPath: "artifacts/session_bounds.json",
|
||||||
|
ProducerStage: "trim",
|
||||||
|
OutputKind: "session_bounds",
|
||||||
|
ContentKind: contentJSON,
|
||||||
|
},
|
||||||
|
ArtifactSessionRecap: {
|
||||||
|
ID: ArtifactSessionRecap,
|
||||||
|
CanonicalRelPath: "artifacts/session_recap.md",
|
||||||
|
ProducerStage: "analyze",
|
||||||
|
OutputKind: "session_recap",
|
||||||
|
ContentKind: contentText,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
var artifactAliases = map[string]string{
|
||||||
|
"processed_transcript": ArtifactTranscriptPolished,
|
||||||
|
"normalized_transcript": ArtifactTranscriptFull,
|
||||||
|
"trimmed_transcript": ArtifactTranscriptTrimmed,
|
||||||
|
}
|
||||||
|
|
||||||
|
// ResolvedSessionArtifact describes one session-level artifact lookup result.
|
||||||
|
type ResolvedSessionArtifact struct {
|
||||||
|
ID string
|
||||||
|
Path string
|
||||||
|
ProducerStage string
|
||||||
|
OutputKind string
|
||||||
|
ProducerRunID string
|
||||||
|
Provenance string
|
||||||
|
}
|
||||||
|
|
||||||
|
// SessionArtifactNotFoundError includes context when a known artifact cannot be read.
|
||||||
|
type SessionArtifactNotFoundError struct {
|
||||||
|
ArtifactID string
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *SessionArtifactNotFoundError) Error() string {
|
||||||
|
return fmt.Sprintf("%s: %q", ErrSessionArtifactNotFound, e.ArtifactID)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *SessionArtifactNotFoundError) Unwrap() error {
|
||||||
|
return ErrSessionArtifactNotFound
|
||||||
|
}
|
||||||
|
|
||||||
|
// NormalizeSessionArtifactSource maps legacy aliases to canonical IDs and validates IDs.
|
||||||
|
func NormalizeSessionArtifactSource(source string) (string, error) {
|
||||||
|
normalized := strings.TrimSpace(source)
|
||||||
|
if normalized == "" {
|
||||||
|
return "", fmt.Errorf("artifact source is required")
|
||||||
|
}
|
||||||
|
if alias, ok := artifactAliases[normalized]; ok {
|
||||||
|
normalized = alias
|
||||||
|
}
|
||||||
|
if _, ok := artifactRegistry[normalized]; !ok {
|
||||||
|
return "", fmt.Errorf("unsupported artifact source %q", source)
|
||||||
|
}
|
||||||
|
return normalized, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// ResolveSessionArtifact resolves a symbolic source to a readable local session artifact path.
|
||||||
|
// Resolution order is manifest producer outputs first, then canonical session path fallback.
|
||||||
|
func ResolveSessionArtifact(paths SessionPaths, m *manifest.Manifest, source string) (ResolvedSessionArtifact, error) {
|
||||||
|
id, err := NormalizeSessionArtifactSource(source)
|
||||||
|
if err != nil {
|
||||||
|
return ResolvedSessionArtifact{}, err
|
||||||
|
}
|
||||||
|
spec := artifactRegistry[id]
|
||||||
|
|
||||||
|
for _, candidate := range manifestArtifactCandidates(paths, m, spec) {
|
||||||
|
exists, isDir, statErr := pathExists(candidate.Path)
|
||||||
|
if statErr != nil {
|
||||||
|
return ResolvedSessionArtifact{}, fmt.Errorf("stat %q: %w", candidate.Path, statErr)
|
||||||
|
}
|
||||||
|
if !exists || isDir {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
resolved := candidate
|
||||||
|
resolved.ID = spec.ID
|
||||||
|
resolved.ProducerStage = spec.ProducerStage
|
||||||
|
resolved.OutputKind = spec.OutputKind
|
||||||
|
if err := validateResolvedContent(resolved.Path, spec.ContentKind); err != nil {
|
||||||
|
return ResolvedSessionArtifact{}, fmt.Errorf("validate %q: %w", resolved.ID, err)
|
||||||
|
}
|
||||||
|
return resolved, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
fallbackPath := filepath.Join(paths.Root, filepath.FromSlash(spec.CanonicalRelPath))
|
||||||
|
exists, isDir, statErr := pathExists(fallbackPath)
|
||||||
|
if statErr != nil {
|
||||||
|
return ResolvedSessionArtifact{}, fmt.Errorf("stat %q: %w", fallbackPath, statErr)
|
||||||
|
}
|
||||||
|
if exists && !isDir {
|
||||||
|
if err := validateResolvedContent(fallbackPath, spec.ContentKind); err != nil {
|
||||||
|
return ResolvedSessionArtifact{}, fmt.Errorf("validate %q: %w", spec.ID, err)
|
||||||
|
}
|
||||||
|
return ResolvedSessionArtifact{
|
||||||
|
ID: spec.ID,
|
||||||
|
Path: filepath.Clean(fallbackPath),
|
||||||
|
ProducerStage: spec.ProducerStage,
|
||||||
|
OutputKind: spec.OutputKind,
|
||||||
|
Provenance: "fallback.canonical_path",
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return ResolvedSessionArtifact{}, &SessionArtifactNotFoundError{ArtifactID: spec.ID}
|
||||||
|
}
|
||||||
|
|
||||||
|
func manifestArtifactCandidates(paths SessionPaths, m *manifest.Manifest, spec artifactSpec) []ResolvedSessionArtifact {
|
||||||
|
if m == nil || len(m.Stages) == 0 || spec.ProducerStage == "" || spec.OutputKind == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
sr := m.Stages[spec.ProducerStage]
|
||||||
|
if sr == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
candidates := make([]ResolvedSessionArtifact, 0, len(sr.Outputs))
|
||||||
|
for _, out := range sr.Outputs {
|
||||||
|
if strings.TrimSpace(out.Kind) != spec.OutputKind {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if strings.TrimSpace(out.LocalPath) == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
resolved := filepath.Clean(ResolveSessionLocalPathForRead(paths, out.LocalPath))
|
||||||
|
if resolved == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
candidates = append(candidates, ResolvedSessionArtifact{
|
||||||
|
Path: resolved,
|
||||||
|
ProducerRunID: strings.TrimSpace(out.ProducerRunID),
|
||||||
|
Provenance: "manifest." + spec.ProducerStage + ".outputs",
|
||||||
|
})
|
||||||
|
}
|
||||||
|
return dedupeResolvedArtifacts(candidates)
|
||||||
|
}
|
||||||
|
|
||||||
|
func dedupeResolvedArtifacts(values []ResolvedSessionArtifact) []ResolvedSessionArtifact {
|
||||||
|
seen := map[string]struct{}{}
|
||||||
|
out := make([]ResolvedSessionArtifact, 0, len(values))
|
||||||
|
for _, value := range values {
|
||||||
|
key := filepath.Clean(strings.TrimSpace(value.Path))
|
||||||
|
if key == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, ok := seen[key]; ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
seen[key] = struct{}{}
|
||||||
|
value.Path = key
|
||||||
|
out = append(out, value)
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
func pathExists(path string) (exists bool, isDir bool, err error) {
|
||||||
|
info, err := os.Stat(path)
|
||||||
|
if err == nil {
|
||||||
|
return true, info.IsDir(), nil
|
||||||
|
}
|
||||||
|
if errors.Is(err, os.ErrNotExist) {
|
||||||
|
return false, false, nil
|
||||||
|
}
|
||||||
|
return false, false, err
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateResolvedContent(path string, kind artifactContentKind) error {
|
||||||
|
switch kind {
|
||||||
|
case contentTranscriptJSON:
|
||||||
|
return validateTranscriptSegmentsJSON(path)
|
||||||
|
case contentJSON:
|
||||||
|
return validateJSONContent(path)
|
||||||
|
case contentText:
|
||||||
|
return validateNonEmptyContent(path)
|
||||||
|
default:
|
||||||
|
return fmt.Errorf("unsupported content kind %q", kind)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateTranscriptSegmentsJSON(path string) error {
|
||||||
|
data, err := os.ReadFile(path)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("read file: %w", err)
|
||||||
|
}
|
||||||
|
var payload map[string]any
|
||||||
|
if err := json.Unmarshal(data, &payload); err != nil {
|
||||||
|
return fmt.Errorf("decode json: %w", err)
|
||||||
|
}
|
||||||
|
segments, ok := payload["segments"]
|
||||||
|
if !ok {
|
||||||
|
return fmt.Errorf("top-level segments is required")
|
||||||
|
}
|
||||||
|
if _, ok := segments.([]any); !ok {
|
||||||
|
return fmt.Errorf("top-level segments must be an array")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateJSONContent(path string) error {
|
||||||
|
data, err := os.ReadFile(path)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("read file: %w", err)
|
||||||
|
}
|
||||||
|
var payload any
|
||||||
|
if err := json.Unmarshal(data, &payload); err != nil {
|
||||||
|
return fmt.Errorf("decode json: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateNonEmptyContent(path string) error {
|
||||||
|
info, err := os.Stat(path)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("stat file: %w", err)
|
||||||
|
}
|
||||||
|
if info.IsDir() {
|
||||||
|
return fmt.Errorf("path is a directory")
|
||||||
|
}
|
||||||
|
if info.Size() <= 0 {
|
||||||
|
return fmt.Errorf("file is empty")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
139
internal/artifacts/artifact_resolver_test.go
Normal file
139
internal/artifacts/artifact_resolver_test.go
Normal file
@@ -0,0 +1,139 @@
|
|||||||
|
package artifacts
|
||||||
|
|
||||||
|
import (
|
||||||
|
"errors"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestNormalizeSessionArtifactSource(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
source string
|
||||||
|
wantID string
|
||||||
|
wantErr string
|
||||||
|
}{
|
||||||
|
{name: "legacy alias processed", source: "processed_transcript", wantID: ArtifactTranscriptPolished},
|
||||||
|
{name: "legacy alias normalized", source: "normalized_transcript", wantID: ArtifactTranscriptFull},
|
||||||
|
{name: "legacy alias trimmed", source: "trimmed_transcript", wantID: ArtifactTranscriptTrimmed},
|
||||||
|
{name: "canonical", source: ArtifactTranscriptTrimmed, wantID: ArtifactTranscriptTrimmed},
|
||||||
|
{name: "unsupported", source: "narratio.unknown", wantErr: "unsupported artifact source"},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
got, err := NormalizeSessionArtifactSource(tt.source)
|
||||||
|
if tt.wantErr != "" {
|
||||||
|
if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
|
||||||
|
t.Fatalf("NormalizeSessionArtifactSource() error = %v, want contains %q", err, tt.wantErr)
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("NormalizeSessionArtifactSource() error = %v", err)
|
||||||
|
}
|
||||||
|
if got != tt.wantID {
|
||||||
|
t.Fatalf("NormalizeSessionArtifactSource() = %q, want %q", got, tt.wantID)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestResolveSessionArtifactPrefersManifestOutput(t *testing.T) {
|
||||||
|
workspace := t.TempDir()
|
||||||
|
paths := buildSessionPaths(workspace, "campaign", "session")
|
||||||
|
if err := os.MkdirAll(paths.ArtifactsDir, 0o755); err != nil {
|
||||||
|
t.Fatalf("MkdirAll() error = %v", err)
|
||||||
|
}
|
||||||
|
manifestPath := filepath.Join(paths.ArtifactsDir, "normalized.from-manifest.json")
|
||||||
|
if err := os.WriteFile(manifestPath, []byte(`{"segments":[]}`), 0o644); err != nil {
|
||||||
|
t.Fatalf("WriteFile() error = %v", err)
|
||||||
|
}
|
||||||
|
canonicalPath := filepath.Join(paths.TranscriptsDir, "normalized.json")
|
||||||
|
if err := os.MkdirAll(filepath.Dir(canonicalPath), 0o755); err != nil {
|
||||||
|
t.Fatalf("MkdirAll() error = %v", err)
|
||||||
|
}
|
||||||
|
if err := os.WriteFile(canonicalPath, []byte(`{"segments":[{"id":123}]}`), 0o644); err != nil {
|
||||||
|
t.Fatalf("WriteFile() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
m := manifest.New("session", time.Now().UTC())
|
||||||
|
m.MarkStageSucceeded("normalize", time.Now().UTC(), []manifest.ArtifactRecord{
|
||||||
|
{Kind: "transcript_normalized", LocalPath: manifestPath, ProducerRunID: "run-123"},
|
||||||
|
})
|
||||||
|
|
||||||
|
resolved, err := ResolveSessionArtifact(paths, m, "normalized_transcript")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ResolveSessionArtifact() error = %v", err)
|
||||||
|
}
|
||||||
|
if resolved.Path != manifestPath {
|
||||||
|
t.Fatalf("resolved path = %q, want %q", resolved.Path, manifestPath)
|
||||||
|
}
|
||||||
|
if resolved.Provenance != "manifest.normalize.outputs" {
|
||||||
|
t.Fatalf("provenance = %q, want manifest.normalize.outputs", resolved.Provenance)
|
||||||
|
}
|
||||||
|
if resolved.ProducerRunID != "run-123" {
|
||||||
|
t.Fatalf("producer run id = %q, want run-123", resolved.ProducerRunID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestResolveSessionArtifactFallsBackToCanonicalPath(t *testing.T) {
|
||||||
|
workspace := t.TempDir()
|
||||||
|
paths := buildSessionPaths(workspace, "campaign", "session")
|
||||||
|
canonicalPath := filepath.Join(paths.TranscriptsDir, "trimmed.json")
|
||||||
|
if err := os.MkdirAll(filepath.Dir(canonicalPath), 0o755); err != nil {
|
||||||
|
t.Fatalf("MkdirAll() error = %v", err)
|
||||||
|
}
|
||||||
|
if err := os.WriteFile(canonicalPath, []byte(`{"segments":[]}`), 0o644); err != nil {
|
||||||
|
t.Fatalf("WriteFile() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
resolved, err := ResolveSessionArtifact(paths, nil, ArtifactTranscriptTrimmed)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ResolveSessionArtifact() error = %v", err)
|
||||||
|
}
|
||||||
|
if resolved.Path != canonicalPath {
|
||||||
|
t.Fatalf("resolved path = %q, want %q", resolved.Path, canonicalPath)
|
||||||
|
}
|
||||||
|
if resolved.Provenance != "fallback.canonical_path" {
|
||||||
|
t.Fatalf("provenance = %q, want fallback.canonical_path", resolved.Provenance)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestResolveSessionArtifactMissingReturnsTypedError(t *testing.T) {
|
||||||
|
workspace := t.TempDir()
|
||||||
|
paths := buildSessionPaths(workspace, "campaign", "session")
|
||||||
|
|
||||||
|
_, err := ResolveSessionArtifact(paths, nil, ArtifactTranscriptTrimmed)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("expected error, got nil")
|
||||||
|
}
|
||||||
|
if !errors.Is(err, ErrSessionArtifactNotFound) {
|
||||||
|
t.Fatalf("errors.Is(err, ErrSessionArtifactNotFound) = false; err=%v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestResolveSessionArtifactValidatesTranscriptShape(t *testing.T) {
|
||||||
|
workspace := t.TempDir()
|
||||||
|
paths := buildSessionPaths(workspace, "campaign", "session")
|
||||||
|
canonicalPath := filepath.Join(paths.TranscriptsDir, "processed.json")
|
||||||
|
if err := os.MkdirAll(filepath.Dir(canonicalPath), 0o755); err != nil {
|
||||||
|
t.Fatalf("MkdirAll() error = %v", err)
|
||||||
|
}
|
||||||
|
if err := os.WriteFile(canonicalPath, []byte(`{"not_segments":[]}`), 0o644); err != nil {
|
||||||
|
t.Fatalf("WriteFile() error = %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err := ResolveSessionArtifact(paths, nil, ArtifactTranscriptPolished)
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("expected error, got nil")
|
||||||
|
}
|
||||||
|
if !strings.Contains(err.Error(), "top-level segments is required") {
|
||||||
|
t.Fatalf("error = %q, want segments validation error", err.Error())
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -116,6 +116,37 @@ func TestScriptoriumLoadAndValidate(t *testing.T) {
|
|||||||
output_kind: session_recap
|
output_kind: session_recap
|
||||||
`,
|
`,
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
name: "canonical artifact source is accepted",
|
||||||
|
scriptoriumYAML: `scriptorium:
|
||||||
|
binary: scriptorium
|
||||||
|
artifacts:
|
||||||
|
session_recap:
|
||||||
|
enabled: true
|
||||||
|
prompt_id: dnd.session_recap
|
||||||
|
output_path: artifacts/session_recap.md
|
||||||
|
inputs:
|
||||||
|
transcript:
|
||||||
|
source: narratio.transcript.trimmed
|
||||||
|
required: true
|
||||||
|
`,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "unknown artifact source fails validation",
|
||||||
|
scriptoriumYAML: `scriptorium:
|
||||||
|
binary: scriptorium
|
||||||
|
artifacts:
|
||||||
|
session_recap:
|
||||||
|
enabled: true
|
||||||
|
prompt_id: dnd.session_recap
|
||||||
|
output_path: artifacts/session_recap.md
|
||||||
|
inputs:
|
||||||
|
transcript:
|
||||||
|
source: narratio.unknown
|
||||||
|
required: true
|
||||||
|
`,
|
||||||
|
wantValidateErr: `pipeline.scriptorium.artifacts.session_recap.inputs.transcript.source "narratio.unknown" is unsupported`,
|
||||||
|
},
|
||||||
{
|
{
|
||||||
name: "artifact render_debug override is accepted",
|
name: "artifact render_debug override is accepted",
|
||||||
scriptoriumYAML: `scriptorium:
|
scriptoriumYAML: `scriptorium:
|
||||||
|
|||||||
@@ -343,9 +343,13 @@ func validateScriptorium(cfg *ScriptoriumConfig) error {
|
|||||||
if trimmedInputName == "" {
|
if trimmedInputName == "" {
|
||||||
return fmt.Errorf("pipeline.scriptorium.artifacts.%s.inputs keys must be non-empty", artifactName)
|
return fmt.Errorf("pipeline.scriptorium.artifacts.%s.inputs keys must be non-empty", artifactName)
|
||||||
}
|
}
|
||||||
if strings.TrimSpace(inputCfg.Source) == "" {
|
source := strings.TrimSpace(inputCfg.Source)
|
||||||
|
if source == "" {
|
||||||
return fmt.Errorf("pipeline.scriptorium.artifacts.%s.inputs.%s.source is required", artifactName, inputName)
|
return fmt.Errorf("pipeline.scriptorium.artifacts.%s.inputs.%s.source is required", artifactName, inputName)
|
||||||
}
|
}
|
||||||
|
if !isSupportedScriptoriumInputSource(source) {
|
||||||
|
return fmt.Errorf("pipeline.scriptorium.artifacts.%s.inputs.%s.source %q is unsupported", artifactName, inputName, inputCfg.Source)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
for varName, varValue := range artifactCfg.Vars {
|
for varName, varValue := range artifactCfg.Vars {
|
||||||
if strings.TrimSpace(varName) == "" {
|
if strings.TrimSpace(varName) == "" {
|
||||||
@@ -435,6 +439,33 @@ func archiveUploadConfiguredForS3(pipeline *PipelineConfig) bool {
|
|||||||
return enabled && upload
|
return enabled && upload
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func isSupportedScriptoriumInputSource(source string) bool {
|
||||||
|
switch strings.TrimSpace(source) {
|
||||||
|
case "previous_session_artifact":
|
||||||
|
return true
|
||||||
|
case "processed_transcript":
|
||||||
|
return true
|
||||||
|
case "normalized_transcript":
|
||||||
|
return true
|
||||||
|
case "trimmed_transcript":
|
||||||
|
return true
|
||||||
|
case "narratio.transcript.merged":
|
||||||
|
return true
|
||||||
|
case "narratio.transcript.polished":
|
||||||
|
return true
|
||||||
|
case "narratio.transcript.full":
|
||||||
|
return true
|
||||||
|
case "narratio.transcript.trimmed":
|
||||||
|
return true
|
||||||
|
case "narratio.bounds.session":
|
||||||
|
return true
|
||||||
|
case "narratio.artifact.session_recap":
|
||||||
|
return true
|
||||||
|
default:
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
var windowsAbsPathRE = regexp.MustCompile(`^[A-Za-z]:[\\/].*`)
|
var windowsAbsPathRE = regexp.MustCompile(`^[A-Za-z]:[\\/].*`)
|
||||||
var envVarNameRE = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
|
var envVarNameRE = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
|
||||||
|
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package stage
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -87,27 +88,7 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
processedTranscriptPath, processedSource, err := discoverProcessedTranscript(m, paths)
|
transcriptRefs := discoverAnalyzeTranscriptRefs(m, paths)
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("analyze: resolve processed transcript: %w", err)
|
|
||||||
}
|
|
||||||
normalizedTranscriptPath, normalizedSource, err := discoverNormalizedTranscript(m, paths)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("analyze: resolve normalized transcript: %w", err)
|
|
||||||
}
|
|
||||||
trimmedTranscriptPath, trimmedSource, err := discoverTrimmedTranscript(m, paths)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("analyze: resolve trimmed transcript: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
transcriptInputs := analyzeTranscriptInputs{
|
|
||||||
ProcessedPath: processedTranscriptPath,
|
|
||||||
ProcessedSource: processedSource,
|
|
||||||
NormalizedPath: normalizedTranscriptPath,
|
|
||||||
NormalizedSource: normalizedSource,
|
|
||||||
TrimmedPath: trimmedTranscriptPath,
|
|
||||||
TrimmedSource: trimmedSource,
|
|
||||||
}
|
|
||||||
|
|
||||||
inputPaths := map[string]string{}
|
inputPaths := map[string]string{}
|
||||||
omittedOptionalInputs := []string{}
|
omittedOptionalInputs := []string{}
|
||||||
@@ -115,7 +96,7 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
inputNames := sortedScriptoriumInputNames(artifactCfg.Inputs)
|
inputNames := sortedScriptoriumInputNames(artifactCfg.Inputs)
|
||||||
for _, inputName := range inputNames {
|
for _, inputName := range inputNames {
|
||||||
inputCfg := artifactCfg.Inputs[inputName]
|
inputCfg := artifactCfg.Inputs[inputName]
|
||||||
resolvedPath, resolved, resolveErr := resolveScriptoriumInput(inputName, inputCfg, transcriptInputs, paths, sessionDir)
|
resolvedPath, resolved, resolveErr := resolveScriptoriumInput(inputName, inputCfg, m, paths, sessionDir)
|
||||||
if resolveErr != nil {
|
if resolveErr != nil {
|
||||||
return nil, fmt.Errorf("analyze: resolve input %q: %w", inputName, resolveErr)
|
return nil, fmt.Errorf("analyze: resolve input %q: %w", inputName, resolveErr)
|
||||||
}
|
}
|
||||||
@@ -170,12 +151,12 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
|
|||||||
"omitted_optional_inputs": omittedOptionalInputs,
|
"omitted_optional_inputs": omittedOptionalInputs,
|
||||||
"vars": vars,
|
"vars": vars,
|
||||||
"timeout": timeout.String(),
|
"timeout": timeout.String(),
|
||||||
"processed_transcript_path": processedTranscriptPath,
|
"processed_transcript_path": transcriptRefs.ProcessedPath,
|
||||||
"processed_transcript_source": processedSource,
|
"processed_transcript_source": transcriptRefs.ProcessedSource,
|
||||||
"normalized_transcript_path": normalizedTranscriptPath,
|
"normalized_transcript_path": transcriptRefs.NormalizedPath,
|
||||||
"normalized_transcript_source": normalizedSource,
|
"normalized_transcript_source": transcriptRefs.NormalizedSource,
|
||||||
"trimmed_transcript_path": trimmedTranscriptPath,
|
"trimmed_transcript_path": transcriptRefs.TrimmedPath,
|
||||||
"trimmed_transcript_source": trimmedSource,
|
"trimmed_transcript_source": transcriptRefs.TrimmedSource,
|
||||||
"render_debug_enabled": resolveRenderDebugEnabled(env.Config.Pipeline.Scriptorium.RenderDebug, artifactCfg.RenderDebug),
|
"render_debug_enabled": resolveRenderDebugEnabled(env.Config.Pipeline.Scriptorium.RenderDebug, artifactCfg.RenderDebug),
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -373,40 +354,6 @@ func discoverProcessedTranscript(m *manifest.Manifest, paths artifacts.SessionPa
|
|||||||
return "", "", nil
|
return "", "", nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func discoverTrimmedTranscript(m *manifest.Manifest, paths artifacts.SessionPaths) (string, string, error) {
|
|
||||||
candidates := []string{}
|
|
||||||
if m != nil && m.Stages != nil {
|
|
||||||
if sr := m.Stages["trim"]; sr != nil {
|
|
||||||
for _, out := range sr.Outputs {
|
|
||||||
if out.Kind != "transcript_trimmed" {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
p := strings.TrimSpace(out.LocalPath)
|
|
||||||
if p == "" {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
resolved := artifacts.ResolveSessionLocalPathForRead(paths, p)
|
|
||||||
candidates = append(candidates, filepath.Clean(resolved))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
deduped := dedupeAndSortPaths(candidates)
|
|
||||||
for _, p := range deduped {
|
|
||||||
if info, err := os.Stat(p); err == nil && !info.IsDir() {
|
|
||||||
return p, "manifest.trim.outputs", nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fallback := filepath.Join(paths.TranscriptsDir, "trimmed.json")
|
|
||||||
if info, err := os.Stat(fallback); err == nil && !info.IsDir() {
|
|
||||||
return filepath.Clean(fallback), "fallback.transcripts_dir", nil
|
|
||||||
}
|
|
||||||
if len(deduped) > 0 {
|
|
||||||
return deduped[0], "manifest.trim.outputs", nil
|
|
||||||
}
|
|
||||||
return "", "", nil
|
|
||||||
}
|
|
||||||
|
|
||||||
type analyzeTranscriptInputs struct {
|
type analyzeTranscriptInputs struct {
|
||||||
ProcessedPath string
|
ProcessedPath string
|
||||||
ProcessedSource string
|
ProcessedSource string
|
||||||
@@ -416,38 +363,36 @@ type analyzeTranscriptInputs struct {
|
|||||||
TrimmedSource string
|
TrimmedSource string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func discoverAnalyzeTranscriptRefs(m *manifest.Manifest, paths artifacts.SessionPaths) analyzeTranscriptInputs {
|
||||||
|
processedPath, processedSource := discoverAnalyzeArtifactRef(m, paths, artifacts.ArtifactTranscriptPolished)
|
||||||
|
normalizedPath, normalizedSource := discoverAnalyzeArtifactRef(m, paths, artifacts.ArtifactTranscriptFull)
|
||||||
|
trimmedPath, trimmedSource := discoverAnalyzeArtifactRef(m, paths, artifacts.ArtifactTranscriptTrimmed)
|
||||||
|
return analyzeTranscriptInputs{
|
||||||
|
ProcessedPath: processedPath,
|
||||||
|
ProcessedSource: processedSource,
|
||||||
|
NormalizedPath: normalizedPath,
|
||||||
|
NormalizedSource: normalizedSource,
|
||||||
|
TrimmedPath: trimmedPath,
|
||||||
|
TrimmedSource: trimmedSource,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func discoverAnalyzeArtifactRef(m *manifest.Manifest, paths artifacts.SessionPaths, source string) (string, string) {
|
||||||
|
resolved, err := artifacts.ResolveSessionArtifact(paths, m, source)
|
||||||
|
if err != nil {
|
||||||
|
return "", ""
|
||||||
|
}
|
||||||
|
return resolved.Path, resolved.Provenance
|
||||||
|
}
|
||||||
|
|
||||||
func resolveScriptoriumInput(
|
func resolveScriptoriumInput(
|
||||||
inputName string,
|
inputName string,
|
||||||
inputCfg config.ScriptoriumInputConfig,
|
inputCfg config.ScriptoriumInputConfig,
|
||||||
transcriptInputs analyzeTranscriptInputs,
|
m *manifest.Manifest,
|
||||||
paths artifacts.SessionPaths,
|
paths artifacts.SessionPaths,
|
||||||
sessionDir string,
|
sessionDir string,
|
||||||
) (string, bool, error) {
|
) (string, bool, error) {
|
||||||
switch strings.TrimSpace(inputCfg.Source) {
|
switch strings.TrimSpace(inputCfg.Source) {
|
||||||
case "processed_transcript":
|
|
||||||
if strings.TrimSpace(transcriptInputs.ProcessedPath) == "" {
|
|
||||||
return "", false, nil
|
|
||||||
}
|
|
||||||
if err := validateProcessedTranscriptOutput(transcriptInputs.ProcessedPath); err != nil {
|
|
||||||
return "", false, fmt.Errorf("processed transcript %q invalid: %w", transcriptInputs.ProcessedPath, err)
|
|
||||||
}
|
|
||||||
return transcriptInputs.ProcessedPath, true, nil
|
|
||||||
case "normalized_transcript":
|
|
||||||
if strings.TrimSpace(transcriptInputs.NormalizedPath) == "" {
|
|
||||||
return "", false, fmt.Errorf("normalized transcript input is unavailable; run normalize stage first")
|
|
||||||
}
|
|
||||||
if err := validateProcessedTranscriptOutput(transcriptInputs.NormalizedPath); err != nil {
|
|
||||||
return "", false, fmt.Errorf("normalized transcript %q invalid: %w", transcriptInputs.NormalizedPath, err)
|
|
||||||
}
|
|
||||||
return transcriptInputs.NormalizedPath, true, nil
|
|
||||||
case "trimmed_transcript":
|
|
||||||
if strings.TrimSpace(transcriptInputs.TrimmedPath) == "" {
|
|
||||||
return "", false, fmt.Errorf("trimmed transcript input is unavailable; run trim stage first")
|
|
||||||
}
|
|
||||||
if err := validateProcessedTranscriptOutput(transcriptInputs.TrimmedPath); err != nil {
|
|
||||||
return "", false, fmt.Errorf("trimmed transcript %q invalid: %w", transcriptInputs.TrimmedPath, err)
|
|
||||||
}
|
|
||||||
return transcriptInputs.TrimmedPath, true, nil
|
|
||||||
case "previous_session_artifact":
|
case "previous_session_artifact":
|
||||||
if strings.TrimSpace(inputCfg.Path) == "" {
|
if strings.TrimSpace(inputCfg.Path) == "" {
|
||||||
return "", false, nil
|
return "", false, nil
|
||||||
@@ -458,7 +403,27 @@ func resolveScriptoriumInput(
|
|||||||
}
|
}
|
||||||
return resolved, true, nil
|
return resolved, true, nil
|
||||||
default:
|
default:
|
||||||
return "", false, fmt.Errorf("unsupported source %q", inputCfg.Source)
|
resolved, err := artifacts.ResolveSessionArtifact(paths, m, inputCfg.Source)
|
||||||
|
if err == nil {
|
||||||
|
return resolved.Path, true, nil
|
||||||
|
}
|
||||||
|
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) {
|
||||||
|
normalized, normalizeErr := artifacts.NormalizeSessionArtifactSource(inputCfg.Source)
|
||||||
|
if normalizeErr != nil {
|
||||||
|
return "", false, normalizeErr
|
||||||
|
}
|
||||||
|
switch normalized {
|
||||||
|
case artifacts.ArtifactTranscriptPolished:
|
||||||
|
return "", false, nil
|
||||||
|
case artifacts.ArtifactTranscriptFull:
|
||||||
|
return "", false, fmt.Errorf("normalized transcript input is unavailable; run normalize stage first")
|
||||||
|
case artifacts.ArtifactTranscriptTrimmed:
|
||||||
|
return "", false, fmt.Errorf("trimmed transcript input is unavailable; run trim stage first")
|
||||||
|
default:
|
||||||
|
return "", false, nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return "", false, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -473,6 +473,30 @@ func TestAnalyzeSupportsProcessedTranscriptSourceWhenConfigured(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestAnalyzeSupportsCanonicalTrimmedTranscriptSourceWhenConfigured(t *testing.T) {
|
||||||
|
env, m, fake := setupAnalyzeEnv(t)
|
||||||
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
|
writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "trimmed.json"), `{"segments":[]}`)
|
||||||
|
|
||||||
|
artifact := env.Config.Pipeline.Scriptorium.Artifacts["session_recap"]
|
||||||
|
artifact.Inputs["transcript"] = config.ScriptoriumInputConfig{
|
||||||
|
Source: "narratio.transcript.trimmed",
|
||||||
|
Required: true,
|
||||||
|
}
|
||||||
|
env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] = artifact
|
||||||
|
|
||||||
|
_, err := (analyzeStage{}).Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(fake.RunRequests) != 1 {
|
||||||
|
t.Fatalf("run requests = %d, want 1", len(fake.RunRequests))
|
||||||
|
}
|
||||||
|
if fake.RunRequests[0].InputPaths["transcript"] != filepath.Join(paths.TranscriptsDir, "trimmed.json") {
|
||||||
|
t.Fatalf("transcript input = %q, want trimmed transcript path", fake.RunRequests[0].InputPaths["transcript"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestAnalyzeSupportsNormalizedTranscriptSourceWhenConfigured(t *testing.T) {
|
func TestAnalyzeSupportsNormalizedTranscriptSourceWhenConfigured(t *testing.T) {
|
||||||
env, m, fake := setupAnalyzeEnv(t)
|
env, m, fake := setupAnalyzeEnv(t)
|
||||||
paths := sessionPathsForEnv(env, m.SessionID)
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
@@ -498,6 +522,36 @@ func TestAnalyzeSupportsNormalizedTranscriptSourceWhenConfigured(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestAnalyzeSupportsCanonicalNormalizedTranscriptSourceFromManifestOutput(t *testing.T) {
|
||||||
|
env, m, fake := setupAnalyzeEnv(t)
|
||||||
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
|
fallbackPath := filepath.Join(paths.TranscriptsDir, "normalized.json")
|
||||||
|
manifestPath := filepath.Join(paths.ArtifactsDir, "normalized.from-manifest.json")
|
||||||
|
writeAnalyzeFile(t, fallbackPath, `{"segments":[{"id":999}]}`)
|
||||||
|
writeAnalyzeFile(t, manifestPath, `{"segments":[{"id":10}]}`)
|
||||||
|
m.MarkStageSucceeded("normalize", time.Now().UTC(), []manifest.ArtifactRecord{
|
||||||
|
{Kind: "transcript_normalized", LocalPath: manifestPath},
|
||||||
|
})
|
||||||
|
|
||||||
|
artifact := env.Config.Pipeline.Scriptorium.Artifacts["session_recap"]
|
||||||
|
artifact.Inputs["transcript"] = config.ScriptoriumInputConfig{
|
||||||
|
Source: "narratio.transcript.full",
|
||||||
|
Required: true,
|
||||||
|
}
|
||||||
|
env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] = artifact
|
||||||
|
|
||||||
|
_, err := (analyzeStage{}).Run(context.Background(), env, m)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Run() error = %v", err)
|
||||||
|
}
|
||||||
|
if len(fake.RunRequests) != 1 {
|
||||||
|
t.Fatalf("run requests = %d, want 1", len(fake.RunRequests))
|
||||||
|
}
|
||||||
|
if fake.RunRequests[0].InputPaths["transcript"] != manifestPath {
|
||||||
|
t.Fatalf("transcript input = %q, want manifest normalized transcript path", fake.RunRequests[0].InputPaths["transcript"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestAnalyzeSupportsNormalizedTranscriptSourceFromManifestOutput(t *testing.T) {
|
func TestAnalyzeSupportsNormalizedTranscriptSourceFromManifestOutput(t *testing.T) {
|
||||||
env, m, fake := setupAnalyzeEnv(t)
|
env, m, fake := setupAnalyzeEnv(t)
|
||||||
paths := sessionPathsForEnv(env, m.SessionID)
|
paths := sessionPathsForEnv(env, m.SessionID)
|
||||||
|
|||||||
Reference in New Issue
Block a user