545 lines
15 KiB
Go
545 lines
15 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/audita"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/stage"
|
|
)
|
|
|
|
type RunOptions struct {
|
|
Force bool
|
|
SelectedArtifacts []string
|
|
EffectiveArtifacts artifacts.EffectiveArtifactSet
|
|
Env *Env
|
|
}
|
|
|
|
type RunSummary struct {
|
|
SessionID string
|
|
RunID string
|
|
ManifestPath string
|
|
StageNames []string
|
|
Executed []string
|
|
Skipped []string
|
|
}
|
|
|
|
type pipelineExecutor func(context.Context, *config.Config, BoundedPlan, RunOptions) (*RunSummary, error)
|
|
|
|
func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (summary *RunSummary, resultErr error) {
|
|
return executePlanWithDependencies(ctx, cfg, plan, opts, invocationDependencies{})
|
|
}
|
|
|
|
func executePlanWithDependencies(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions, dependencies invocationDependencies) (summary *RunSummary, resultErr error) {
|
|
dependencies = dependencies.normalized()
|
|
runtime, err := prepareInvocation(ctx, cfg, plan, opts, dependencies)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := prepareInvocationLayout(runtime); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
lock, err := runtime.artifactStore.AcquireSessionLockForContext(ctx, runtime.identity.Campaign, runtime.identity.SessionID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("acquire session lock: %w", err)
|
|
}
|
|
defer func() {
|
|
if releaseErr := runtime.artifactStore.ReleaseSessionLock(lock); releaseErr != nil {
|
|
if resultErr == nil {
|
|
resultErr = fmt.Errorf("release session lock: %w", releaseErr)
|
|
} else {
|
|
resultErr = errors.Join(resultErr, fmt.Errorf("release session lock: %w", releaseErr))
|
|
}
|
|
}
|
|
}()
|
|
|
|
if err := initializeLockedInvocation(runtime); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := composeInvocation(runtime); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, selected := range runtime.stages {
|
|
runtime.stageNames = append(runtime.stageNames, selected.Name())
|
|
outcome, err := executeSelectedStage(runtime, selected)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if outcome.executed {
|
|
runtime.executed = append(runtime.executed, selected.Name())
|
|
}
|
|
if outcome.skipped {
|
|
runtime.skipped = append(runtime.skipped, selected.Name())
|
|
}
|
|
}
|
|
return finalizeInvocation(runtime)
|
|
}
|
|
|
|
func stagesContainAny(stages []stage.Stage, names ...string) bool {
|
|
wanted := make(map[string]struct{}, len(names))
|
|
for _, name := range names {
|
|
wanted[name] = struct{}{}
|
|
}
|
|
for _, candidate := range stages {
|
|
if candidate == nil {
|
|
continue
|
|
}
|
|
if _, ok := wanted[candidate.Name()]; ok {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func persistSessionTerminalFailure(
|
|
ctx context.Context,
|
|
sessionStore manifest.Store,
|
|
sessionManifestPath string,
|
|
sessionManifest *manifest.Manifest,
|
|
operationErr error,
|
|
at time.Time,
|
|
) error {
|
|
if operationErr == nil {
|
|
return nil
|
|
}
|
|
|
|
if sessionManifest != nil {
|
|
sessionManifest.RecordFailure(at, operationErr.Error())
|
|
}
|
|
|
|
sessionErr := sessionStore.Save(ctx, sessionManifestPath, sessionManifest)
|
|
return errors.Join(
|
|
operationErr,
|
|
wrapTerminalPersistenceError("save terminal session manifest", sessionErr),
|
|
)
|
|
}
|
|
|
|
func wrapTerminalPersistenceError(operation string, err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("%s: %w", operation, err)
|
|
}
|
|
|
|
func buildDefaultWhisperXClient(cfg *config.Config) (whisperx.Client, error) {
|
|
if cfg == nil || cfg.Pipeline == nil {
|
|
return &whisperx.NoopClient{}, nil
|
|
}
|
|
|
|
wx := cfg.Pipeline.WhisperX
|
|
if strings.TrimSpace(wx.TranscribeURL) == "" {
|
|
// Compatibility fallback for tests or internal call paths that bypass config validation.
|
|
return &whisperx.NoopClient{}, nil
|
|
}
|
|
|
|
retries := 0
|
|
if wx.Retries != nil {
|
|
retries = *wx.Retries
|
|
}
|
|
|
|
client, err := whisperx.NewHTTPClientFromConfigValues(
|
|
wx.TranscribeURL,
|
|
wx.Language,
|
|
wx.Timeout,
|
|
wx.RetryDelay,
|
|
retries,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("from pipeline.whisperx: %w", err)
|
|
}
|
|
return client, nil
|
|
}
|
|
|
|
func buildDefaultSeriatimRunner(cfg *config.Config) (seriatim.Runner, error) {
|
|
if cfg == nil || cfg.Pipeline == nil {
|
|
return &seriatim.NoopRunner{}, nil
|
|
}
|
|
|
|
s := cfg.Pipeline.Seriatim
|
|
if strings.TrimSpace(s.Binary) == "" || strings.TrimSpace(s.Timeout) == "" || strings.TrimSpace(s.OutputSchema) == "" {
|
|
// Compatibility fallback for tests or internal call paths that bypass config validation/defaults.
|
|
return &seriatim.NoopRunner{}, nil
|
|
}
|
|
|
|
report := false
|
|
if s.Report != nil {
|
|
report = *s.Report
|
|
}
|
|
runner, err := seriatim.NewSubprocessRunnerFromConfigValues(
|
|
s.Binary,
|
|
s.Timeout,
|
|
s.OutputSchema,
|
|
s.CoalesceGap,
|
|
report,
|
|
seriatim.EnvConfig{
|
|
OverlapWordRunGap: s.Env.OverlapWordRunGap,
|
|
OverlapWordRunReorderWindow: s.Env.OverlapWordRunReorderWindow,
|
|
BackchannelMaxDuration: s.Env.BackchannelMaxDuration,
|
|
FillerMaxDuration: s.Env.FillerMaxDuration,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("from pipeline.seriatim: %w", err)
|
|
}
|
|
return runner, nil
|
|
}
|
|
|
|
func buildDefaultAuditaRunner(cfg *config.Config) (audita.Runner, error) {
|
|
if cfg == nil || cfg.Pipeline == nil {
|
|
return &audita.NoopRunner{}, nil
|
|
}
|
|
|
|
a := cfg.Pipeline.Audita
|
|
if strings.TrimSpace(a.Binary) == "" || strings.TrimSpace(a.Timeout) == "" {
|
|
// Compatibility fallback for tests or internal call paths that bypass config validation/defaults.
|
|
return &audita.NoopRunner{}, nil
|
|
}
|
|
|
|
report := false
|
|
if a.Report != nil {
|
|
report = *a.Report
|
|
}
|
|
llmAPIKey := ""
|
|
llmAPIKeyReference := ""
|
|
if a.LLMAPIKey != nil {
|
|
llmAPIKey = a.LLMAPIKey.Value()
|
|
llmAPIKeyReference = a.LLMAPIKey.EnvName()
|
|
}
|
|
|
|
runner, err := audita.NewSubprocessRunnerFromConfigValues(
|
|
a.Binary,
|
|
a.Timeout,
|
|
llmAPIKey,
|
|
llmAPIKeyReference,
|
|
append([]string(nil), a.Modules...),
|
|
a.BaseURL,
|
|
a.Model,
|
|
a.TranscriptDescription,
|
|
a.ConfigPath,
|
|
a.OutputSchema,
|
|
a.WorkDirRetention,
|
|
a.TotalLLMConcurrency,
|
|
a.ProposalLLMConcurrency,
|
|
a.ValidationModel,
|
|
a.ValidationLLMConcurrency,
|
|
report,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("from pipeline.audita: %w", err)
|
|
}
|
|
return runner, nil
|
|
}
|
|
|
|
func loadManifestAtPathIfPresent(ctx context.Context, store manifest.Store, path string) (*manifest.Manifest, bool, error) {
|
|
exists, err := fileExists(path)
|
|
if err != nil {
|
|
return nil, false, fmt.Errorf("check manifest existence %q: %w", path, err)
|
|
}
|
|
if exists {
|
|
m, err := store.Load(ctx, path)
|
|
if err != nil {
|
|
return nil, false, fmt.Errorf("load manifest %q: %w", path, err)
|
|
}
|
|
return m, true, nil
|
|
}
|
|
return nil, false, nil
|
|
}
|
|
|
|
func fileExists(path string) (bool, error) {
|
|
_, err := os.Stat(path)
|
|
if err == nil {
|
|
return true, nil
|
|
}
|
|
if os.IsNotExist(err) {
|
|
return false, nil
|
|
}
|
|
return false, err
|
|
}
|
|
|
|
func mapResultOutputs(stageName string, result *stage.StageResult, runID string) []manifest.ArtifactRecord {
|
|
if result == nil || len(result.Outputs) == 0 {
|
|
return nil
|
|
}
|
|
|
|
runID = strings.TrimSpace(runID)
|
|
out := make([]manifest.ArtifactRecord, 0, len(result.Outputs))
|
|
for _, ref := range result.Outputs {
|
|
localPath := ref.AbsolutePath
|
|
if localPath == "" {
|
|
localPath = ref.RelativePath
|
|
}
|
|
kind := ref.Kind
|
|
sourceID := strings.TrimSpace(ref.SourceID)
|
|
if sourceID == "" {
|
|
if stageName == "analyze" {
|
|
sourceID = artifacts.ConfiguredArtifactSourceID(ref.Kind)
|
|
kind = "scriptorium_artifact"
|
|
} else {
|
|
sourceID = sourceIDForOutputKind(kind)
|
|
}
|
|
}
|
|
producerRunID := strings.TrimSpace(ref.ProducerRunID)
|
|
if producerRunID == "" {
|
|
producerRunID = runID
|
|
}
|
|
out = append(out, manifest.ArtifactRecord{
|
|
Kind: kind,
|
|
SourceID: sourceID,
|
|
LocalPath: localPath,
|
|
ArchivePath: ref.ArchivePath,
|
|
Contract: cloneContractMetadata(ref.Contract),
|
|
ExternalProvenance: cloneExternalProvenance(ref.ExternalProvenance),
|
|
ProducerRunID: producerRunID,
|
|
RemoteKey: ref.RemoteKey,
|
|
Checksum: ref.Checksum,
|
|
})
|
|
}
|
|
|
|
return out
|
|
}
|
|
|
|
func validateStageResult(result *stage.StageResult) error {
|
|
if result == nil {
|
|
return nil
|
|
}
|
|
switch result.Disposition {
|
|
case stage.StageDispositionSucceeded:
|
|
if strings.TrimSpace(result.SkipReason) != "" {
|
|
return fmt.Errorf("successful result contains a skip reason")
|
|
}
|
|
return nil
|
|
case stage.StageDispositionSkipped:
|
|
if strings.TrimSpace(result.SkipReason) == "" {
|
|
return fmt.Errorf("skipped result requires a skip reason")
|
|
}
|
|
if len(result.Outputs) != 0 {
|
|
return fmt.Errorf("skipped result contains %d output(s)", len(result.Outputs))
|
|
}
|
|
return nil
|
|
default:
|
|
return fmt.Errorf("unsupported stage result disposition %q", result.Disposition)
|
|
}
|
|
}
|
|
|
|
func cloneContractMetadata(value *artifactmodel.ContractMetadata) *artifactmodel.ContractMetadata {
|
|
if value == nil {
|
|
return nil
|
|
}
|
|
cloned := *value
|
|
return &cloned
|
|
}
|
|
|
|
func cloneExternalProvenance(value *artifactmodel.ExternalProvenance) *artifactmodel.ExternalProvenance {
|
|
if value == nil {
|
|
return nil
|
|
}
|
|
cloned := *value
|
|
return &cloned
|
|
}
|
|
|
|
func sourceIDForOutputKind(kind string) string {
|
|
trimmed := strings.TrimSpace(kind)
|
|
if trimmed == "" {
|
|
return ""
|
|
}
|
|
if trimmed == "session_bounds" {
|
|
return artifacts.ArtifactBoundsSession
|
|
}
|
|
for _, spec := range artifactmodel.RuntimeTranscriptArtifacts() {
|
|
if spec.OutputKind == trimmed {
|
|
return spec.SourceID
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func applyStageResultToManifest(m *manifest.Manifest, stageName string, result *stage.StageResult) {
|
|
if m == nil || result == nil {
|
|
return
|
|
}
|
|
sr := m.Stages[stageName]
|
|
if sr == nil {
|
|
return
|
|
}
|
|
if len(result.Logs) > 0 {
|
|
sr.Logs = append([]string(nil), result.Logs...)
|
|
}
|
|
if len(result.GeneratedConfigs) > 0 {
|
|
sr.GeneratedConfigs = append([]string(nil), result.GeneratedConfigs...)
|
|
}
|
|
if len(result.Metadata) > 0 {
|
|
sr.Metadata = result.Metadata
|
|
}
|
|
}
|
|
|
|
type invocationIdentity struct {
|
|
Campaign string
|
|
SessionID string
|
|
RunID string
|
|
LocalWorkDir string
|
|
LocalSpoolDir string
|
|
S3Bucket string
|
|
S3SessionPrefix string
|
|
S3RunPrefix string
|
|
}
|
|
|
|
func resolveInvocationIdentity(cfg *config.Config, runID string) (invocationIdentity, error) {
|
|
if cfg == nil || cfg.Pipeline == nil || cfg.Campaign == nil || cfg.Session == nil {
|
|
return invocationIdentity{}, fmt.Errorf("resolved configuration requires pipeline, campaign, and session")
|
|
}
|
|
|
|
campaign := strings.TrimSpace(cfg.Session.Campaign)
|
|
configuredCampaign := strings.TrimSpace(config.CampaignID(cfg.Campaign))
|
|
if campaign == "" {
|
|
return invocationIdentity{}, fmt.Errorf("configured session campaign is required")
|
|
}
|
|
if configuredCampaign == "" {
|
|
return invocationIdentity{}, fmt.Errorf("configured campaign is required")
|
|
}
|
|
if campaign != configuredCampaign {
|
|
return invocationIdentity{}, fmt.Errorf("configured session campaign %q does not match configured campaign %q", campaign, configuredCampaign)
|
|
}
|
|
|
|
sessionID := strings.TrimSpace(cfg.Session.SessionID)
|
|
if sessionID == "" {
|
|
return invocationIdentity{}, fmt.Errorf("configured session_id is required")
|
|
}
|
|
runID = strings.TrimSpace(runID)
|
|
if runID == "" {
|
|
return invocationIdentity{}, fmt.Errorf("run id is required")
|
|
}
|
|
|
|
identity := invocationIdentity{
|
|
Campaign: campaign,
|
|
SessionID: sessionID,
|
|
RunID: runID,
|
|
LocalWorkDir: artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, campaign, sessionID, runID),
|
|
}
|
|
if spoolRoot := strings.TrimSpace(cfg.Pipeline.Spool.Root); spoolRoot != "" {
|
|
identity.LocalSpoolDir = artifacts.SessionSpoolAudioDir(spoolRoot, campaign, sessionID, runID)
|
|
}
|
|
if s3 := cfg.Pipeline.Storage.S3; s3 != nil {
|
|
identity.S3Bucket = strings.TrimSpace(s3.Bucket)
|
|
identity.S3SessionPrefix = artifacts.S3SessionPrefix(s3.RootPrefix, campaign, sessionID)
|
|
identity.S3RunPrefix = artifacts.S3RunPrefix(identity.S3SessionPrefix, runID)
|
|
}
|
|
return identity, nil
|
|
}
|
|
|
|
func (identity invocationIdentity) validateSessionManifest(m *manifest.Manifest) error {
|
|
if m == nil {
|
|
return fmt.Errorf("persisted manifest is nil")
|
|
}
|
|
if sessionID := strings.TrimSpace(m.SessionID); sessionID != "" && sessionID != identity.SessionID {
|
|
return fmt.Errorf("persisted manifest session_id %q does not match configured session_id %q", sessionID, identity.SessionID)
|
|
}
|
|
if campaign := strings.TrimSpace(m.Campaign); campaign != "" && campaign != identity.Campaign {
|
|
return fmt.Errorf("persisted manifest campaign %q does not match configured campaign %q", campaign, identity.Campaign)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (identity invocationIdentity) applyToSessionManifest(m *manifest.Manifest) {
|
|
if m == nil {
|
|
return
|
|
}
|
|
m.SessionID = identity.SessionID
|
|
m.Campaign = identity.Campaign
|
|
m.RunID = identity.RunID
|
|
m.LocalWorkDir = identity.LocalWorkDir
|
|
m.LocalSpoolDir = identity.LocalSpoolDir
|
|
m.S3Bucket = identity.S3Bucket
|
|
m.S3SessionPrefix = identity.S3SessionPrefix
|
|
m.S3RunPrefix = identity.S3RunPrefix
|
|
}
|
|
|
|
func manifestPathFor(cfg *config.Config) string {
|
|
return artifacts.SessionManifestPathForCampaign(
|
|
cfg.Pipeline.Workspace.Root,
|
|
cfg.Session.Campaign,
|
|
cfg.Session.SessionID,
|
|
)
|
|
}
|
|
|
|
func needsObjectStoreForRun(cfg *config.Config, stages []stage.Stage, effectiveSets ...artifacts.EffectiveArtifactSet) bool {
|
|
if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil {
|
|
return false
|
|
}
|
|
effective := artifacts.EffectiveArtifactSet{}
|
|
if len(effectiveSets) > 0 {
|
|
effective = effectiveSets[0]
|
|
}
|
|
if !effective.Resolved() && cfg.Pipeline.Scriptorium != nil {
|
|
var err error
|
|
effective, err = artifacts.ResolveEffectiveArtifactSet(
|
|
artifacts.ConfiguredArtifactDefinitions(cfg.Pipeline.Scriptorium.Artifacts),
|
|
nil,
|
|
)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
}
|
|
stageRequested := func(name string) bool {
|
|
for _, s := range stages {
|
|
if s != nil && s.Name() == name {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
if cfg.Session.Inputs.AudioS3 != nil && stageRequested("prepare") {
|
|
return true
|
|
}
|
|
if stageRequested("prepare") {
|
|
requirements := artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg), effective)
|
|
if len(requirements) > 0 && strings.TrimSpace(cfg.Session.PreviousSessionID) != "" {
|
|
return true
|
|
}
|
|
}
|
|
if !stageRequested("publish") {
|
|
return false
|
|
}
|
|
if cfg.Pipeline.Publish == nil {
|
|
return false
|
|
}
|
|
if cfg.Pipeline.Publish.Enabled != nil && !*cfg.Pipeline.Publish.Enabled {
|
|
return false
|
|
}
|
|
if cfg.Pipeline.Publish.UploadRun != nil && !*cfg.Pipeline.Publish.UploadRun {
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func needsNotariusForRun(cfg *config.Config, stages []stage.Stage) bool {
|
|
if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Notarius == nil || !cfg.Pipeline.Notarius.Enabled {
|
|
return false
|
|
}
|
|
for _, candidate := range stages {
|
|
if candidate != nil && candidate.Name() == "extract" {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func configuredScriptoriumArtifacts(cfg *config.Config) map[string]config.ScriptoriumArtifactConfig {
|
|
if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Scriptorium == nil {
|
|
return nil
|
|
}
|
|
return cfg.Pipeline.Scriptorium.Artifacts
|
|
}
|