Run ordered processing with independent settings
This commit is contained in:
@@ -110,7 +110,16 @@ func prepareProcess(ctx context.Context, input processInput, loaded *processInpu
|
||||
}
|
||||
debug = &preflight.DebugInput{WorkDir: cfg.Diagnostics.WorkDir, RunID: runID, Secrets: secrets, EffectiveConfig: effective.Bytes(), Metadata: diagnostics.InvocationMetadata{Operation: "process", TranscriptPath: input.Transcript, OutputPath: input.Output.Value, Overlay: input.Overlay.Value, ConfigVersion: &cfg.Version, RunID: runID, StartedAt: started}}
|
||||
}
|
||||
prepared, err := preflight.Prepare(ctx, preflight.Input{Config: cfg, Transcript: loaded.Source, Output: diagnostics.OutputSelection{InputTranscriptPath: input.Transcript, TranscriptPath: input.Output.Value, ReportDir: reportDir, ReportingEnabled: cfg.Reporting.Enabled, ProtectedPaths: loaded.ProtectedPaths, Stdout: stdout}, InitialReports: initial, FailedReports: failed, Debug: debug})
|
||||
prepared, err := preflight.Prepare(ctx, preflight.Input{InspectProfile: func(ctx context.Context, id string) error {
|
||||
profile, err := loaded.Runtime.Engine.InspectProfile(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if profile.Backend != nil {
|
||||
_, err = loaded.Runtime.Engine.InspectBackend(ctx, *profile.Backend)
|
||||
}
|
||||
return err
|
||||
}, Config: cfg, Transcript: loaded.Source, Output: diagnostics.OutputSelection{InputTranscriptPath: input.Transcript, TranscriptPath: input.Output.Value, ReportDir: reportDir, ReportingEnabled: cfg.Reporting.Enabled, ProtectedPaths: loaded.ProtectedPaths, Stdout: stdout}, InitialReports: initial, FailedReports: failed, Debug: debug})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
166
internal/cli/process_ordered_integration_test.go
Normal file
166
internal/cli/process_ordered_integration_test.go
Normal file
@@ -0,0 +1,166 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/config"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/schema"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/llm"
|
||||
ck "go.maximumdirect.net/configkit"
|
||||
)
|
||||
|
||||
type controlledTransport func(*http.Request) (*http.Response, error)
|
||||
|
||||
func (f controlledTransport) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
|
||||
|
||||
func TestProcessingUsesPerEntryProfilesInOrder(t *testing.T) {
|
||||
var mu sync.Mutex
|
||||
var models []string
|
||||
var prompts []string
|
||||
transport := controlledTransport(func(r *http.Request) (*http.Response, error) {
|
||||
var request struct {
|
||||
Model string `json:"model"`
|
||||
Messages []struct {
|
||||
Content string `json:"content"`
|
||||
} `json:"messages"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var contents string
|
||||
for _, message := range request.Messages {
|
||||
contents += message.Content
|
||||
}
|
||||
mu.Lock()
|
||||
models = append(models, request.Model)
|
||||
prompts = append(prompts, contents)
|
||||
mu.Unlock()
|
||||
result := `{"validations":[{"correction_index":1,"approved":true,"confidence":1,"reason":"ok"}]}`
|
||||
switch request.Model {
|
||||
case "model-one":
|
||||
result = `{"corrections":[{"id":1,"original_text":"Alyce","corrected_text":"Alice","confidence":1}]}`
|
||||
case "model-two":
|
||||
result = `{"corrections":[{"id":1,"original_text":"Sheeld","corrected_text":"Shield","confidence":1}]}`
|
||||
case "model-three":
|
||||
result = `{"corrections":[{"id":1,"original_text":"casts","corrected_text":"cast","confidence":1}]}`
|
||||
}
|
||||
encoded, _ := json.Marshal(result)
|
||||
body := fmt.Sprintf(`{"model":"fixture","choices":[{"message":{"role":"assistant","content":%s}}]}`, encoded)
|
||||
return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(body)), Request: r}, nil
|
||||
})
|
||||
dir := t.TempDir()
|
||||
files := map[string]string{
|
||||
"config.yml": `version: 3
|
||||
reporting: {enabled: false}
|
||||
glossary_collections:
|
||||
party:
|
||||
entries: [{id: alice, name: Alice, category: person}]
|
||||
spells:
|
||||
entries: [{id: shield, name: Shield, category: spell}]
|
||||
protected_terms: [party, spells]
|
||||
llm: {validation_profile: review}
|
||||
pipeline:
|
||||
stages:
|
||||
- module: glossary
|
||||
glossaries: [party]
|
||||
llm: {proposal_profile: one}
|
||||
- module: glossary
|
||||
glossaries: [spells]
|
||||
llm: {proposal_profile: two}
|
||||
- module: grammar
|
||||
glossaries: [party, spells]
|
||||
llm: {proposal_profile: three}
|
||||
`,
|
||||
"prompts.yml": fmt.Sprintf("profiles:\n - {id: one, endpoint: %q, model: model-one, api_key_required: false}\n - {id: two, endpoint: %q, model: model-two, api_key_required: false}\n - {id: three, endpoint: %q, model: model-three, api_key_required: false}\n - {id: review, endpoint: %q, model: review, api_key_required: false}\n", "https://example.invalid/v1", "https://example.invalid/v1", "https://example.invalid/v1", "https://example.invalid/v1"),
|
||||
"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"Alyce casts Sheeld."}]`,
|
||||
}
|
||||
for name, value := range files {
|
||||
if err := os.WriteFile(filepath.Join(dir, name), []byte(value), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
loaded, err := config.Load(ck.Input{Source: ck.File(filepath.Join(dir, "config.yml"))}, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
runtime, err := llm.NewRuntime(context.Background(), loaded, &http.Client{Transport: transport})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
source, err := schema.ParseSourceTranscriptJSON([]byte(files["transcript.json"]))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
inputs := &processInputs{Configuration: loaded, Runtime: runtime, Source: source}
|
||||
prepared, err := prepareProcess(context.Background(), processInput{Transcript: filepath.Join(dir, "transcript.json")}, inputs, &bytes.Buffer{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer prepared.Prepared.Output.Close()
|
||||
result, err := executePipeline(context.Background(), inputs, prepared, sourceToCanonicalTranscript(source), func(error) {})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if result.Transcript.Segments[0].Text != "Alice cast Shield." {
|
||||
t.Fatalf("result=%q models=%v", result.Transcript.Segments[0].Text, models)
|
||||
}
|
||||
if len(prompts) > 3 && !strings.Contains(prompts[3], "Alice casts Sheeld.") {
|
||||
t.Fatalf("second proposal missed earlier edit: %s", prompts[3])
|
||||
}
|
||||
if !reflect.DeepEqual(models, []string{"model-one", "review", "review", "model-two", "review", "review", "model-three", "review", "review"}) {
|
||||
t.Fatalf("models=%v", models)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecoverableModelFailureKeepsNormalizedTranscript(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
files := map[string]string{
|
||||
"config.yml": "version: 3\nreporting: {enabled: false}\nllm: {proposal_profile: local, transient_retries: 0}\npipeline:\n stages: [{module: grammar}]\n",
|
||||
"prompts.yml": "profiles:\n - {id: local, endpoint: 'https://example.invalid/v1', model: fixture, api_key_required: false}\n",
|
||||
"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`,
|
||||
}
|
||||
for name, value := range files {
|
||||
if err := os.WriteFile(filepath.Join(dir, name), []byte(value), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
loaded, err := config.Load(ck.Input{Source: ck.File(filepath.Join(dir, "config.yml"))}, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
transport := controlledTransport(func(r *http.Request) (*http.Response, error) {
|
||||
return &http.Response{StatusCode: http.StatusServiceUnavailable, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"error":{"message":"temporarily unavailable"}}`)), Request: r}, nil
|
||||
})
|
||||
runtime, err := llm.NewRuntime(context.Background(), loaded, &http.Client{Transport: transport})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
source, err := schema.ParseSourceTranscriptJSON([]byte(files["transcript.json"]))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
inputs := &processInputs{Configuration: loaded, Runtime: runtime, Source: source}
|
||||
prepared, err := prepareProcess(context.Background(), processInput{Transcript: filepath.Join(dir, "transcript.json")}, inputs, &bytes.Buffer{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer prepared.Prepared.Output.Close()
|
||||
result, err := executePipeline(context.Background(), inputs, prepared, sourceToCanonicalTranscript(source), func(error) {})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if result.Transcript.Segments[0].Text != "hello" || len(result.Snapshot.Errors) != 1 || result.Snapshot.Errors[0].Scope != "work_unit" {
|
||||
t.Fatalf("result=%+v errors=%+v", result.Transcript, result.Snapshot.Errors)
|
||||
}
|
||||
}
|
||||
53
internal/cli/process_pipeline.go
Normal file
53
internal/cli/process_pipeline.go
Normal file
@@ -0,0 +1,53 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/schema"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/contracts"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/llm"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/modules"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/processreport"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/runner"
|
||||
)
|
||||
|
||||
func executePipeline(ctx context.Context, loaded *processInputs, prepared *preparedProcess, normalized *schema.Transcript, warn func(error)) (processResult, error) {
|
||||
cfg := &loaded.Configuration.Config
|
||||
specs := contracts.ResolveModuleRunSpecs(cfg)
|
||||
observer := llm.NewObservationCollector()
|
||||
for i := range specs {
|
||||
item := &specs[i]
|
||||
proposal, err := loaded.Runtime.NewRoleClient("proposal", *item.Settings.LLM.ProposalProfile, item.Settings.LLM, observer.Hooks())
|
||||
if err != nil {
|
||||
return processResult{}, fmt.Errorf("%s proposal client: %w", item.InstanceName, err)
|
||||
}
|
||||
validation, err := loaded.Runtime.NewRoleClient("validation", *item.Settings.LLM.ValidationProfile, item.Settings.LLM, observer.Hooks())
|
||||
if err != nil {
|
||||
return processResult{}, fmt.Errorf("%s validation client: %w", item.InstanceName, err)
|
||||
}
|
||||
item.ProposalClient, item.ValidationClient = proposal, validation
|
||||
}
|
||||
debugPath := prepared.DebugPath()
|
||||
output, runErr := runner.New(modules.NewFactory()).Run(ctx, runner.RunInput{
|
||||
Config: cfg, Transcript: normalized, ModuleSpecs: specs,
|
||||
ProposalDiagnosticsDir: debugPath, ValidationDiagnosticsDir: debugPath,
|
||||
CaptureCandidates: debugPath != "", ReportDiagnosticError: warn,
|
||||
})
|
||||
output.Observations = observer.Snapshot()
|
||||
output.ObservationsAvailable = true
|
||||
output.LLMSummary = llm.Summarize(output.Observations)
|
||||
if debug := prepared.Prepared.Debug; debug != nil {
|
||||
warn(debug.WriteJSONArtifact(diagnostics.ArtifactLLMExecution, processreport.BuildLLMExecution(output.Observations)))
|
||||
warn(debug.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{RunDirectoryPath: debug.Path(), RunOutput: &output})))
|
||||
}
|
||||
var failures []processreport.RunFailure
|
||||
if runErr != nil {
|
||||
failures = append(failures, processreport.RunFailure{Phase: "processing", Reason: outcomes.ReasonInternalFailure})
|
||||
}
|
||||
snapshot := processreport.Build(processreport.BuildInput{RunID: prepared.RunID, Status: reporting.StatusInProgress, StartedAt: prepared.Started, ModuleSpecs: specs, RunOutput: &output, RunFailures: failures})
|
||||
return processResult{Transcript: output.FinalTranscript, Snapshot: snapshot}, runErr
|
||||
}
|
||||
Reference in New Issue
Block a user