diff --git a/README.md b/README.md index eed0b71..efc2cba 100644 --- a/README.md +++ b/README.md @@ -121,7 +121,6 @@ Environment: - `AUDITA_BASE_URL` - `AUDITA_LLM_TIMEOUT_SECONDS` - `AUDITA_MAX_RETRIES` -- `AUDITA_LLM_CONCURRENCY` CLI: - `--llm-api-key` @@ -129,7 +128,6 @@ CLI: - `--base-url` - `--llm-timeout-seconds` - `--max-retries` -- `--llm-concurrency` ### Validation LLM @@ -151,9 +149,26 @@ CLI: - `--validation-llm-concurrency` - `--validation-max-prompt-tokens` -Validation concurrency behavior: -- when validation concurrency is unset, it inherits primary `llm-concurrency` -- when explicitly set, validation concurrency must be `<= llm-concurrency` +### LLM Concurrency + +Environment: +- `AUDITA_TOTAL_LLM_CONCURRENCY` +- `AUDITA_PROPOSAL_LLM_CONCURRENCY` +- `AUDITA_VALIDATION_LLM_CONCURRENCY` +- `AUDITA_LLM_CONCURRENCY` (legacy alias for `AUDITA_TOTAL_LLM_CONCURRENCY`) + +CLI: +- `--total-llm-concurrency` +- `--proposal-llm-concurrency` +- `--validation-llm-concurrency` +- `--llm-concurrency` (legacy alias for `--total-llm-concurrency`) + +Behavior: +- all proposal and validation LLM calls are bounded by total LLM concurrency +- proposal LLM calls are additionally bounded by proposal LLM concurrency +- when validation concurrency is unset, it inherits total LLM concurrency +- when explicitly set, proposal and validation concurrency must each be `<= total-llm-concurrency` +- canonical total settings win when both canonical and legacy alias settings are provided at the same precedence layer ### Confidence Thresholds diff --git a/docs/architecture.md b/docs/architecture.md index 41cf9a6..190a3a4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -22,7 +22,7 @@ Implemented today: - Deterministic validator-chain execution in the runner with cardinality enforcement. - Module-level validator decision/rejection reporting. - Internal structured LLM client contract plus an `instructor-go`-backed adapter package. -- Bounded LLM scheduler/semaphore infrastructure with context-aware permit handling. +- Bounded FIFO LLM scheduler infrastructure with context-aware permit handling. - Runtime primary/validation LLM effective-config resolution helpers with validation inheritance. - Generic JSON prompt/response diagnostics writer primitives with secret redaction. - LLM-backed validator models, prompt builders, batching, and runtime execution. @@ -162,6 +162,7 @@ Current runtime flow (`internal/cli/run.go`): - default run path uses configured default sequence (`glossary,homophones,glossary,spoken_word,grammar`); - explicit `--modules` overrides the default sequence; - test/injected module factory path remains available for deterministic runtime tests. + - each module recomputes chunks from the current working transcript, runs chunk proposal work concurrently, aggregates deterministically, validates, and applies approved proposals once. 13. Output working transcript to `--output` file or stdout. 14. Build process report (`phase` currently set to `default_pipeline`). 15. Optionally write `--report-json`; always write run-dir `report.json`. @@ -219,6 +220,7 @@ Precedence: Implemented config surfaces include: - module list - primary and validation LLM settings +- total/proposal/validation LLM concurrency controls - section token controls and target sections - confidence thresholds - normalization controls @@ -246,10 +248,17 @@ Current runtime boundary: - normal `go test ./...` does not require real LLM credentials or Python dependencies. `internal/framework/llm` also provides: -- a bounded `Scheduler` for controlled concurrent LLM calls with reliable permit release; -- primary/validation effective-config resolution helpers, including validation inheritance fallback to primary settings; +- a bounded FIFO `Scheduler` for controlled concurrent LLM calls with reliable permit release on success, error, and cancellation; +- primary/validation effective-config resolution helpers, including validation inheritance fallback to total LLM concurrency settings; - generic interaction diagnostics primitives that write machine-readable JSON artifacts for request metadata, request payload, response payload, and optional error payload with secret redaction. +LLM concurrency runtime behavior: +- `total` concurrency bounds all proposal and validation LLM calls. +- `proposal` concurrency adds a proposal-only sub-cap, composed with total. +- `validation` concurrency adds a validation-only sub-cap, composed with total. +- legacy `llm-concurrency` inputs remain compatibility aliases for total concurrency. +- modules execute serially, chunk proposals run concurrently within each module, and approved proposals are applied once per module in deterministic order. + ## Implemented normalization behavior Normalization (`internal/core/normalization`) currently: - sorts by segment start time; diff --git a/docs/llm-concurrency-audit.md b/docs/llm-concurrency-audit.md index bc96dc5..4dc62d3 100644 --- a/docs/llm-concurrency-audit.md +++ b/docs/llm-concurrency-audit.md @@ -20,6 +20,15 @@ Main gap versus the requested target architecture: - global concurrency is currently represented by existing `--llm-concurrency`, - scheduler implementation is semaphore-based and does not explicitly guarantee FIFO ordering. +## Implementation Status (2026-05-12 Update) + +The targeted concurrency gaps identified in this audit have now been addressed: +- explicit total/proposal/validation LLM concurrency controls are implemented in config/env/CLI, +- legacy `llm-concurrency` settings are preserved as compatibility aliases to total concurrency, +- proposal and validation schedulers are composed with a global total-cap scheduler, +- scheduler default behavior is FIFO with context-aware queued cancellation and reliable permit release, +- runner proposal worker fan-out is aligned with effective proposal concurrency. + ## Audit Findings (Questions 1-14) 1. **Does the current runner execute modules serially?** diff --git a/docs/subprocess-operations.md b/docs/subprocess-operations.md index edbd630..b17385b 100644 --- a/docs/subprocess-operations.md +++ b/docs/subprocess-operations.md @@ -16,6 +16,7 @@ audita process \ Recommended additions: - `--work-dir ` to control diagnostics location. - `--work-dir-retention ` to control retained run directories. +- `--total-llm-concurrency`, `--proposal-llm-concurrency`, and `--validation-llm-concurrency` when orchestration needs explicit LLM throughput controls. - `--modules ...` only when intentionally overriding the default full sequence. ## Stdout behavior diff --git a/internal/cli/run.go b/internal/cli/run.go index fd3a847..1c4fed4 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -196,8 +196,7 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio } globalScheduler := proposalScheduler if globalScheduler == nil { - primaryCfg := llm.ResolvePrimaryConfig(inv.Config) - s, sErr := llm.NewScheduler(primaryCfg.Concurrency) + s, sErr := llm.NewScheduler(inv.Config.TotalLLMConcurrency) if sErr != nil { return fail("runner_setup", sErr, nil) } @@ -205,12 +204,18 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio } if proposalScheduler == nil { proposalScheduler = globalScheduler + if inv.Config.EffectiveProposalLLMConcurrency() < inv.Config.TotalLLMConcurrency { + s, sErr := llm.NewScheduler(inv.Config.EffectiveProposalLLMConcurrency()) + if sErr != nil { + return fail("runner_setup", sErr, nil) + } + proposalScheduler = composeSchedulers(globalScheduler, s) + } } if validationScheduler == nil { validationScheduler = globalScheduler - if inv.Config.ValidationLLM.Concurrency != nil && inv.Config.PrimaryLLM.Concurrency > inv.Config.EffectiveValidationLLMConfig().Concurrency { - validationCfg := llm.ResolveValidationConfig(inv.Config) - s, sErr := llm.NewScheduler(validationCfg.Concurrency) + if inv.Config.ValidationLLMConcurrency != nil && inv.Config.EffectiveValidationLLMConcurrency() < inv.Config.TotalLLMConcurrency { + s, sErr := llm.NewScheduler(inv.Config.EffectiveValidationLLMConcurrency()) if sErr != nil { return fail("runner_setup", sErr, nil) } @@ -392,6 +397,10 @@ func runProcess(args []string, stdout, stderr io.Writer) int { overrides.ValidationBaseURL = pFlags.validationBaseURL case "llm-timeout-seconds": overrides.PrimaryLLMTimeoutSeconds = pFlags.llmTimeoutSeconds + case "total-llm-concurrency": + overrides.TotalLLMConcurrency = pFlags.totalLLMConcurrency + case "proposal-llm-concurrency": + overrides.ProposalLLMConcurrency = pFlags.proposalLLMConcurrency case "llm-concurrency": overrides.PrimaryLLMConcurrency = pFlags.llmConcurrency case "validation-llm-timeout-seconds": @@ -682,6 +691,8 @@ type processFlags struct { baseURL *string validationBaseURL *string llmTimeoutSeconds *int + totalLLMConcurrency *int + proposalLLMConcurrency *int llmConcurrency *int validationLLMTimeoutSeconds *int validationMaxPromptTokens *int @@ -717,9 +728,9 @@ func newProcessFlagSet(cfg config.Config, stderr io.Writer) (*flag.FlagSet, proc validationMaxRetriesDefault = *cfg.ValidationLLM.MaxRetries } - validationLLMConcurrencyDefault := cfg.PrimaryLLM.Concurrency - if cfg.ValidationLLM.Concurrency != nil { - validationLLMConcurrencyDefault = *cfg.ValidationLLM.Concurrency + validationLLMConcurrencyDefault := cfg.TotalLLMConcurrency + if cfg.ValidationLLMConcurrency != nil { + validationLLMConcurrencyDefault = *cfg.ValidationLLMConcurrency } targetSectionsDefault := 0 @@ -739,13 +750,15 @@ func newProcessFlagSet(cfg config.Config, stderr io.Writer) (*flag.FlagSet, proc baseURL: fs.String("base-url", cfg.PrimaryLLM.BaseURL, "Primary OpenAI-compatible base URL"), validationBaseURL: fs.String("validation-base-url", cfg.ValidationLLM.BaseURL, "Validation OpenAI-compatible base URL"), llmTimeoutSeconds: fs.Int("llm-timeout-seconds", cfg.PrimaryLLM.TimeoutSeconds, "Primary LLM timeout in seconds"), - llmConcurrency: fs.Int("llm-concurrency", cfg.PrimaryLLM.Concurrency, "Primary LLM concurrency"), + totalLLMConcurrency: fs.Int("total-llm-concurrency", cfg.TotalLLMConcurrency, "Total concurrent LLM calls across proposal and validation"), + proposalLLMConcurrency: fs.Int("proposal-llm-concurrency", cfg.EffectiveProposalLLMConcurrency(), "Concurrent proposal-generation LLM calls"), + llmConcurrency: fs.Int("llm-concurrency", cfg.TotalLLMConcurrency, "Alias for --total-llm-concurrency"), validationLLMTimeoutSeconds: fs.Int("validation-llm-timeout-seconds", validationTimeoutSecondsDefault, "Validation LLM timeout in seconds"), validationMaxPromptTokens: fs.Int("validation-max-prompt-tokens", cfg.ValidationMaxPromptTokens, "Validation max prompt tokens"), targetSections: fs.Int("target-sections", targetSectionsDefault, "Target number of transcript sections"), maxRetries: fs.Int("max-retries", cfg.PrimaryLLM.MaxRetries, "Maximum structured-output retries"), validationMaxRetries: fs.Int("validation-max-retries", validationMaxRetriesDefault, "Validation structured-output retries"), - validationLLMConcurrency: fs.Int("validation-llm-concurrency", validationLLMConcurrencyDefault, "Validation LLM concurrency"), + validationLLMConcurrency: fs.Int("validation-llm-concurrency", validationLLMConcurrencyDefault, "Concurrent validation LLM calls (inherits total when unset)"), maxSectionTokens: fs.Int("max-section-tokens", cfg.MaxSectionTokens, "Maximum section tokens"), minSectionTokens: fs.Int("min-section-tokens", cfg.MinSectionTokens, "Minimum section tokens"), glossaryConfidenceThreshold: fs.Float64("glossary-confidence-threshold", cfg.Thresholds.Glossary, "Glossary confidence threshold"), diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index d651e90..e27e6f3 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -63,6 +63,8 @@ func TestRunProcessHelpListsExpectedFlags(t *testing.T) { "--base-url", "--validation-base-url", "--llm-timeout-seconds", + "--total-llm-concurrency", + "--proposal-llm-concurrency", "--llm-concurrency", "--validation-llm-timeout-seconds", "--validation-max-prompt-tokens", @@ -246,7 +248,7 @@ func TestRunProcessCLIOverridesEnvironment(t *testing.T) { } } -func TestRunProcessRejectsValidationConcurrencyAbovePrimaryConcurrency(t *testing.T) { +func TestRunProcessRejectsValidationConcurrencyAboveTotalConcurrency(t *testing.T) { var stdout bytes.Buffer var stderr bytes.Buffer @@ -259,7 +261,7 @@ func TestRunProcessRejectsValidationConcurrencyAbovePrimaryConcurrency(t *testin transcriptPath, "--glossary", fixturePath("tiny_glossary.yaml"), - "--llm-concurrency", + "--total-llm-concurrency", "1", "--validation-llm-concurrency", "2", @@ -273,12 +275,12 @@ func TestRunProcessRejectsValidationConcurrencyAbovePrimaryConcurrency(t *testin if !strings.Contains(stderr.String(), "invalid CLI configuration") { t.Fatalf("expected invalid CLI configuration error, got %q", stderr.String()) } - if !strings.Contains(stderr.String(), "validation llm concurrency must be less than or equal to primary llm concurrency") { - t.Fatalf("expected validation/primary concurrency error details, got %q", stderr.String()) + if !strings.Contains(stderr.String(), "validation llm concurrency must be less than or equal to total llm concurrency") { + t.Fatalf("expected validation/total concurrency error details, got %q", stderr.String()) } } -func TestRunProcessLLMConcurrencyDrivesEffectiveValidationConcurrencyWhenUnset(t *testing.T) { +func TestRunProcessTotalLLMConcurrencyDrivesEffectiveValidationConcurrencyWhenUnset(t *testing.T) { processModuleFactory = fakeModuleFactory{modules: map[string]contracts.TranscriptModule{ "m": fakeModule{ key: "m", @@ -288,14 +290,96 @@ func TestRunProcessLLMConcurrencyDrivesEffectiveValidationConcurrencyWhenUnset(t if req.Config == nil { t.Fatal("expected config in validation request") } - if req.Config.PrimaryLLM.Concurrency != 4 { - t.Fatalf("expected primary llm concurrency 4, got %d", req.Config.PrimaryLLM.Concurrency) + if req.Config.TotalLLMConcurrency != 4 { + t.Fatalf("expected total llm concurrency 4, got %d", req.Config.TotalLLMConcurrency) } - if req.Config.ValidationLLM.Concurrency != nil { - t.Fatalf("expected validation concurrency unset, got %#v", req.Config.ValidationLLM.Concurrency) + if req.Config.ProposalLLMConcurrency != 4 { + t.Fatalf("expected proposal llm concurrency to inherit total 4, got %d", req.Config.ProposalLLMConcurrency) } - if req.Config.EffectiveValidationLLMConfig().Concurrency != 4 { - t.Fatalf("expected inherited effective validation concurrency 4, got %d", req.Config.EffectiveValidationLLMConfig().Concurrency) + if req.Config.ValidationLLMConcurrency != nil { + t.Fatalf("expected validation concurrency unset, got %#v", req.Config.ValidationLLMConcurrency) + } + if req.Config.EffectiveValidationLLMConcurrency() != 4 { + t.Fatalf("expected inherited effective validation concurrency 4, got %d", req.Config.EffectiveValidationLLMConcurrency()) + } + return validators.Result{ValidatorName: "capture-config", Decisions: nil}, nil + }}, + }, + proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return nil, nil + }, + }, + }} + t.Cleanup(func() { processModuleFactory = nil }) + + var stdout bytes.Buffer + var stderr bytes.Buffer + transcriptPath := writeFile(t, "transcript.json", `[ + {"id":1,"speaker":"Alice","start":0.0,"end":1.0,"text":"Hello"} +]`) + + exitCode := Run([]string{ + "process", + transcriptPath, + "--glossary", + fixturePath("tiny_glossary.yaml"), + "--modules", + "m", + "--total-llm-concurrency", + "4", + }, &stdout, &stderr) + if exitCode != 0 { + t.Fatalf("expected success, got %d stderr=%q", exitCode, stderr.String()) + } +} + +func TestRunProcessRejectsProposalConcurrencyAboveTotalConcurrency(t *testing.T) { + var stdout bytes.Buffer + var stderr bytes.Buffer + + transcriptPath := writeFile(t, "transcript.json", `[ + {"id":1,"speaker":"Alice","start":0.0,"end":1.0,"text":"Hello"} +]`) + + exitCode := Run([]string{ + "process", + transcriptPath, + "--glossary", + fixturePath("tiny_glossary.yaml"), + "--total-llm-concurrency", + "1", + "--proposal-llm-concurrency", + "2", + }, &stdout, &stderr) + if exitCode == 0 { + t.Fatalf("expected nonzero exit code for invalid llm concurrency combination") + } + if stdout.Len() != 0 { + t.Fatalf("expected empty stdout, got %q", stdout.String()) + } + if !strings.Contains(stderr.String(), "invalid CLI configuration") { + t.Fatalf("expected invalid CLI configuration error, got %q", stderr.String()) + } + if !strings.Contains(stderr.String(), "proposal llm concurrency must be less than or equal to total llm concurrency") { + t.Fatalf("expected proposal/total concurrency error details, got %q", stderr.String()) + } +} + +func TestRunProcessLegacyLLMConcurrencyAliasSetsTotalAndProposal(t *testing.T) { + processModuleFactory = fakeModuleFactory{modules: map[string]contracts.TranscriptModule{ + "m": fakeModule{ + key: "m", + policy: proposals.ReplacementPolicyRequireUnique, + validators: []contracts.Validator{ + fakeValidator{name: "capture-config", validateF: func(req contracts.ValidationRequest) (validators.Result, error) { + if req.Config == nil { + t.Fatal("expected config in validation request") + } + if req.Config.TotalLLMConcurrency != 3 { + t.Fatalf("expected total llm concurrency 3, got %d", req.Config.TotalLLMConcurrency) + } + if req.Config.ProposalLLMConcurrency != 3 { + t.Fatalf("expected proposal llm concurrency inherited from alias, got %d", req.Config.ProposalLLMConcurrency) } return validators.Result{ValidatorName: "capture-config", Decisions: nil}, nil }}, @@ -321,7 +405,62 @@ func TestRunProcessLLMConcurrencyDrivesEffectiveValidationConcurrencyWhenUnset(t "--modules", "m", "--llm-concurrency", + "3", + }, &stdout, &stderr) + if exitCode != 0 { + t.Fatalf("expected success, got %d stderr=%q", exitCode, stderr.String()) + } +} + +func TestRunProcessLLMConcurrencyFlagsOverrideEnvironment(t *testing.T) { + processModuleFactory = fakeModuleFactory{modules: map[string]contracts.TranscriptModule{ + "m": fakeModule{ + key: "m", + policy: proposals.ReplacementPolicyRequireUnique, + validators: []contracts.Validator{ + fakeValidator{name: "capture-config", validateF: func(req contracts.ValidationRequest) (validators.Result, error) { + if req.Config == nil { + t.Fatal("expected config in validation request") + } + if req.Config.TotalLLMConcurrency != 4 { + t.Fatalf("expected cli total llm concurrency 4, got %d", req.Config.TotalLLMConcurrency) + } + if req.Config.ProposalLLMConcurrency != 3 { + t.Fatalf("expected cli proposal llm concurrency 3, got %d", req.Config.ProposalLLMConcurrency) + } + if req.Config.ValidationLLMConcurrency == nil || *req.Config.ValidationLLMConcurrency != 2 { + t.Fatalf("expected env validation llm concurrency 2, got %#v", req.Config.ValidationLLMConcurrency) + } + return validators.Result{ValidatorName: "capture-config", Decisions: nil}, nil + }}, + }, + proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + return nil, nil + }, + }, + }} + t.Cleanup(func() { processModuleFactory = nil }) + t.Setenv("AUDITA_TOTAL_LLM_CONCURRENCY", "2") + t.Setenv("AUDITA_PROPOSAL_LLM_CONCURRENCY", "2") + t.Setenv("AUDITA_VALIDATION_LLM_CONCURRENCY", "2") + + var stdout bytes.Buffer + var stderr bytes.Buffer + transcriptPath := writeFile(t, "transcript.json", `[ + {"id":1,"speaker":"Alice","start":0.0,"end":1.0,"text":"Hello"} +]`) + + exitCode := Run([]string{ + "process", + transcriptPath, + "--glossary", + fixturePath("tiny_glossary.yaml"), + "--modules", + "m", + "--total-llm-concurrency", "4", + "--proposal-llm-concurrency", + "3", }, &stdout, &stderr) if exitCode != 0 { t.Fatalf("expected success, got %d stderr=%q", exitCode, stderr.String()) @@ -373,6 +512,60 @@ func TestComposeSchedulersEnforcesStricterSubcap(t *testing.T) { } } +func TestComposedProposalAndValidationSchedulersShareGlobalTotalCap(t *testing.T) { + global, err := llm.NewScheduler(2) + if err != nil { + t.Fatalf("NewScheduler(global): %v", err) + } + proposalSubcap, err := llm.NewScheduler(2) + if err != nil { + t.Fatalf("NewScheduler(proposalSubcap): %v", err) + } + validationSubcap, err := llm.NewScheduler(2) + if err != nil { + t.Fatalf("NewScheduler(validationSubcap): %v", err) + } + proposalScheduler := composeSchedulers(global, proposalSubcap) + validationScheduler := composeSchedulers(global, validationSubcap) + + var inFlight int32 + var maxInFlight int32 + var wg sync.WaitGroup + for i := 0; i < 12; i++ { + wg.Add(1) + go func(idx int) { + defer wg.Done() + scheduler := proposalScheduler + if idx%2 == 1 { + scheduler = validationScheduler + } + runErr := scheduler.Run(context.Background(), func(context.Context) error { + current := atomic.AddInt32(&inFlight, 1) + for { + prior := atomic.LoadInt32(&maxInFlight) + if current <= prior || atomic.CompareAndSwapInt32(&maxInFlight, prior, current) { + break + } + } + time.Sleep(20 * time.Millisecond) + atomic.AddInt32(&inFlight, -1) + return nil + }) + if runErr != nil { + t.Errorf("scheduler run error: %v", runErr) + } + }(i) + } + wg.Wait() + + if maxInFlight > 2 { + t.Fatalf("expected combined proposal+validation concurrency <= global total cap (2), got %d", maxInFlight) + } + if maxInFlight < 2 { + t.Fatalf("expected observed combined concurrency of at least 2, got %d", maxInFlight) + } +} + func TestRunProcessReportJSONSuccessIncludesNormalizationSummary(t *testing.T) { var stdout bytes.Buffer var stderr bytes.Buffer @@ -2694,6 +2887,12 @@ func TestRunProcessWritesRedactedRunMetadataArtifacts(t *testing.T) { fixturePath("tiny_transcript.json"), "--glossary", fixturePath("tiny_glossary.yaml"), + "--total-llm-concurrency", + "3", + "--proposal-llm-concurrency", + "2", + "--validation-llm-concurrency", + "1", "--output", outputPath, "--report-json", @@ -2736,6 +2935,24 @@ func TestRunProcessWritesRedactedRunMetadataArtifacts(t *testing.T) { t.Fatalf("effective config artifact leaked API key material") } + var effectiveConfig struct { + TotalLLMConcurrency int `json:"TotalLLMConcurrency"` + ProposalLLMConcurrency int `json:"ProposalLLMConcurrency"` + ValidationLLMConcurrency *int `json:"ValidationLLMConcurrency"` + } + if err := json.Unmarshal(configBytes, &effectiveConfig); err != nil { + t.Fatalf("failed to parse effective config metadata: %v", err) + } + if effectiveConfig.TotalLLMConcurrency != 3 { + t.Fatalf("expected total_llm_concurrency=3 in effective config, got %d", effectiveConfig.TotalLLMConcurrency) + } + if effectiveConfig.ProposalLLMConcurrency != 2 { + t.Fatalf("expected proposal_llm_concurrency=2 in effective config, got %d", effectiveConfig.ProposalLLMConcurrency) + } + if effectiveConfig.ValidationLLMConcurrency == nil || *effectiveConfig.ValidationLLMConcurrency != 1 { + t.Fatalf("expected validation_llm_concurrency=1 in effective config, got %#v", effectiveConfig.ValidationLLMConcurrency) + } + var invocation struct { Operation string `json:"operation"` TranscriptPath string `json:"transcript_path"` diff --git a/internal/core/config/config.go b/internal/core/config/config.go index d00d998..ad00094 100644 --- a/internal/core/config/config.go +++ b/internal/core/config/config.go @@ -36,6 +36,9 @@ type Config struct { Modules []string PrimaryLLM LLMConfig ValidationLLM ValidationLLMConfig + TotalLLMConcurrency int + ProposalLLMConcurrency int + ValidationLLMConcurrency *int ValidationMaxPromptTokens int MaxSectionTokens int MinSectionTokens int @@ -52,7 +55,9 @@ type LLMConfig struct { BaseURL string TimeoutSeconds int MaxRetries int - Concurrency int + // Concurrency is retained as a backward-compatible alias for + // TotalLLMConcurrency. + Concurrency int } type ValidationLLMConfig struct { @@ -61,7 +66,9 @@ type ValidationLLMConfig struct { BaseURL string TimeoutSeconds *int MaxRetries *int - Concurrency *int + // Concurrency is retained as a backward-compatible alias for + // ValidationLLMConcurrency. + Concurrency *int } type ConfidenceThresholds struct { @@ -91,6 +98,9 @@ func Default() Config { Concurrency: DefaultLLMConcurrency, }, ValidationLLM: ValidationLLMConfig{}, + TotalLLMConcurrency: DefaultLLMConcurrency, + ProposalLLMConcurrency: DefaultLLMConcurrency, + ValidationLLMConcurrency: nil, ValidationMaxPromptTokens: DefaultValidationMaxPromptTokens, MaxSectionTokens: DefaultMaxSectionTokens, MinSectionTokens: DefaultMinSectionTokens, @@ -146,9 +156,37 @@ func (c Config) EffectiveValidationLLMConfig() LLMConfig { if c.ValidationLLM.MaxRetries != nil { effective.MaxRetries = *c.ValidationLLM.MaxRetries } - if c.ValidationLLM.Concurrency != nil { - effective.Concurrency = *c.ValidationLLM.Concurrency - } + effective.Concurrency = c.EffectiveValidationLLMConcurrency() return effective } + +func (c Config) EffectiveValidationLLMConcurrency() int { + if c.ValidationLLMConcurrency != nil { + return *c.ValidationLLMConcurrency + } + return c.TotalLLMConcurrency +} + +func (c Config) EffectiveProposalLLMConcurrency() int { + if c.ProposalLLMConcurrency > 0 { + return c.ProposalLLMConcurrency + } + return c.TotalLLMConcurrency +} + +func (c *Config) syncLegacyConcurrencyAliases() { + if c == nil { + return + } + c.PrimaryLLM.Concurrency = c.TotalLLMConcurrency + c.ValidationLLM.Concurrency = intPtr(c.ValidationLLMConcurrency) +} + +func intPtr(v *int) *int { + if v == nil { + return nil + } + x := *v + return &x +} diff --git a/internal/core/config/config_test.go b/internal/core/config/config_test.go index eb15bae..75c22e4 100644 --- a/internal/core/config/config_test.go +++ b/internal/core/config/config_test.go @@ -24,8 +24,17 @@ func TestDefaultConfigValues(t *testing.T) { if cfg.PrimaryLLM.MaxRetries != DefaultMaxRetries { t.Fatalf("unexpected default max retries: %d", cfg.PrimaryLLM.MaxRetries) } - if cfg.PrimaryLLM.Concurrency != DefaultLLMConcurrency { - t.Fatalf("unexpected default llm concurrency: %d", cfg.PrimaryLLM.Concurrency) + if cfg.TotalLLMConcurrency != DefaultLLMConcurrency { + t.Fatalf("unexpected default total llm concurrency: %d", cfg.TotalLLMConcurrency) + } + if cfg.ProposalLLMConcurrency != DefaultLLMConcurrency { + t.Fatalf("unexpected default proposal llm concurrency: %d", cfg.ProposalLLMConcurrency) + } + if cfg.ValidationLLMConcurrency != nil { + t.Fatalf("expected validation llm concurrency to be unset by default") + } + if cfg.PrimaryLLM.Concurrency != cfg.TotalLLMConcurrency { + t.Fatalf("expected primary llm concurrency alias to mirror total, got primary=%d total=%d", cfg.PrimaryLLM.Concurrency, cfg.TotalLLMConcurrency) } if cfg.ValidationLLM.TimeoutSeconds != nil { t.Fatalf("expected validation timeout to be unset by default") @@ -34,7 +43,7 @@ func TestDefaultConfigValues(t *testing.T) { t.Fatalf("expected validation max retries to be unset by default") } if cfg.ValidationLLM.Concurrency != nil { - t.Fatalf("expected validation llm concurrency to be unset by default") + t.Fatalf("expected legacy validation llm concurrency alias to be unset by default") } if cfg.TargetSections != nil { t.Fatalf("expected target sections to be unset by default") @@ -56,7 +65,8 @@ func TestLoadFromEnvOverridesAndFallback(t *testing.T) { "AUDITA_BASE_URL": "https://api.openai.com/v1", "AUDITA_LLM_TIMEOUT_SECONDS": "120", "AUDITA_MAX_RETRIES": "7", - "AUDITA_LLM_CONCURRENCY": "6", + "AUDITA_TOTAL_LLM_CONCURRENCY": "6", + "AUDITA_PROPOSAL_LLM_CONCURRENCY": "4", "AUDITA_VALIDATION_LLM_CONCURRENCY": "2", "AUDITA_VALIDATION_MAX_PROMPT_TOKENS": "4096", "AUDITA_MAX_SECTION_TOKENS": "9000", @@ -92,17 +102,63 @@ func TestLoadFromEnvOverridesAndFallback(t *testing.T) { if cfg.TargetSections == nil || *cfg.TargetSections != 5 { t.Fatalf("unexpected target sections: %#v", cfg.TargetSections) } + if cfg.TotalLLMConcurrency != 6 { + t.Fatalf("unexpected total llm concurrency: %d", cfg.TotalLLMConcurrency) + } + if cfg.ProposalLLMConcurrency != 4 { + t.Fatalf("unexpected proposal llm concurrency: %d", cfg.ProposalLLMConcurrency) + } + if cfg.ValidationLLMConcurrency == nil || *cfg.ValidationLLMConcurrency != 2 { + t.Fatalf("unexpected validation llm concurrency: %#v", cfg.ValidationLLMConcurrency) + } if cfg.PrimaryLLM.Concurrency != 6 { - t.Fatalf("unexpected primary llm concurrency: %d", cfg.PrimaryLLM.Concurrency) + t.Fatalf("expected primary alias concurrency 6, got %d", cfg.PrimaryLLM.Concurrency) } if cfg.ValidationLLM.Concurrency == nil || *cfg.ValidationLLM.Concurrency != 2 { - t.Fatalf("unexpected validation llm concurrency: %#v", cfg.ValidationLLM.Concurrency) + t.Fatalf("expected validation alias concurrency 2, got %#v", cfg.ValidationLLM.Concurrency) } if cfg.WorkDirRetention != WorkDirRetentionAlways { t.Fatalf("unexpected work dir retention: %q", cfg.WorkDirRetention) } } +func TestLoadFromEnvLegacyLLMConcurrencyAliasForTotalAndProposal(t *testing.T) { + env := map[string]string{ + "AUDITA_LLM_CONCURRENCY": "5", + } + + cfg, err := loadFromLookup(mapLookup(env)) + if err != nil { + t.Fatalf("loadFromLookup returned error: %v", err) + } + + if cfg.TotalLLMConcurrency != 5 { + t.Fatalf("expected total concurrency from legacy alias, got %d", cfg.TotalLLMConcurrency) + } + if cfg.ProposalLLMConcurrency != 5 { + t.Fatalf("expected proposal concurrency to inherit legacy total, got %d", cfg.ProposalLLMConcurrency) + } +} + +func TestLoadFromEnvCanonicalTotalWinsLegacyAlias(t *testing.T) { + env := map[string]string{ + "AUDITA_TOTAL_LLM_CONCURRENCY": "4", + "AUDITA_LLM_CONCURRENCY": "9", + } + + cfg, err := loadFromLookup(mapLookup(env)) + if err != nil { + t.Fatalf("loadFromLookup returned error: %v", err) + } + + if cfg.TotalLLMConcurrency != 4 { + t.Fatalf("expected canonical total to win over legacy alias, got %d", cfg.TotalLLMConcurrency) + } + if cfg.ProposalLLMConcurrency != 4 { + t.Fatalf("expected proposal to inherit canonical total when unset, got %d", cfg.ProposalLLMConcurrency) + } +} + func TestLoadFromEnvUsesAuditaLLMAPIKeyOverFallback(t *testing.T) { env := map[string]string{ "AUDITA_LLM_API_KEY": "primary-key", @@ -127,12 +183,14 @@ func TestApplyCLIOverridesPrecedence(t *testing.T) { model := "cli-model" workDir := "/cli/work" modules := "grammar" - llmConcurrency := 5 + totalLLMConcurrency := 5 + proposalLLMConcurrency := 3 overrides := CLIOverrides{ - PrimaryModel: &model, - WorkDir: &workDir, - ModulesCSV: &modules, - PrimaryLLMConcurrency: &llmConcurrency, + PrimaryModel: &model, + WorkDir: &workDir, + ModulesCSV: &modules, + TotalLLMConcurrency: &totalLLMConcurrency, + ProposalLLMConcurrency: &proposalLLMConcurrency, } if err := cfg.ApplyCLIOverrides(overrides); err != nil { @@ -148,17 +206,54 @@ func TestApplyCLIOverridesPrecedence(t *testing.T) { if !reflect.DeepEqual(cfg.Modules, []string{"grammar"}) { t.Fatalf("unexpected modules: %#v", cfg.Modules) } - if cfg.PrimaryLLM.Concurrency != 5 { - t.Fatalf("expected CLI llm concurrency override, got %d", cfg.PrimaryLLM.Concurrency) + if cfg.TotalLLMConcurrency != 5 { + t.Fatalf("expected CLI total concurrency override, got %d", cfg.TotalLLMConcurrency) + } + if cfg.ProposalLLMConcurrency != 3 { + t.Fatalf("expected CLI proposal concurrency override, got %d", cfg.ProposalLLMConcurrency) + } +} + +func TestApplyCLIOverridesLegacyLLMConcurrencyAlias(t *testing.T) { + cfg := Default() + aliasConcurrency := 6 + + if err := cfg.ApplyCLIOverrides(CLIOverrides{PrimaryLLMConcurrency: &aliasConcurrency}); err != nil { + t.Fatalf("ApplyCLIOverrides failed: %v", err) + } + + if cfg.TotalLLMConcurrency != 6 { + t.Fatalf("expected legacy --llm-concurrency alias to set total, got %d", cfg.TotalLLMConcurrency) + } + if cfg.ProposalLLMConcurrency != 6 { + t.Fatalf("expected proposal to inherit aliased total when unset, got %d", cfg.ProposalLLMConcurrency) + } +} + +func TestApplyCLIOverridesCanonicalTotalWinsLegacyAlias(t *testing.T) { + cfg := Default() + canonicalTotal := 4 + legacyAlias := 9 + + if err := cfg.ApplyCLIOverrides(CLIOverrides{TotalLLMConcurrency: &canonicalTotal, PrimaryLLMConcurrency: &legacyAlias}); err != nil { + t.Fatalf("ApplyCLIOverrides failed: %v", err) + } + + if cfg.TotalLLMConcurrency != 4 { + t.Fatalf("expected canonical total concurrency to win, got %d", cfg.TotalLLMConcurrency) + } + if cfg.ProposalLLMConcurrency != 4 { + t.Fatalf("expected proposal to inherit canonical total when proposal is unset, got %d", cfg.ProposalLLMConcurrency) } } func TestValidationFailures(t *testing.T) { cfg := Default() cfg.PrimaryLLM.TimeoutSeconds = -1 - cfg.PrimaryLLM.Concurrency = 0 + cfg.TotalLLMConcurrency = 0 + cfg.ProposalLLMConcurrency = 0 validationConcurrency := 5 - cfg.ValidationLLM.Concurrency = &validationConcurrency + cfg.ValidationLLMConcurrency = &validationConcurrency cfg.ValidationMaxPromptTokens = 0 cfg.MaxSectionTokens = 100 cfg.MinSectionTokens = 200 @@ -173,8 +268,9 @@ func TestValidationFailures(t *testing.T) { message := err.Error() for _, expected := range []string{ "primary llm timeout seconds", - "primary llm concurrency", - "validation llm concurrency must be less than or equal to primary llm concurrency", + "total llm concurrency", + "proposal llm concurrency", + "validation llm concurrency must be less than or equal to total llm concurrency", "validation max prompt tokens", "min section tokens", "grammar confidence threshold", @@ -193,7 +289,8 @@ func TestEffectiveValidationLLMInheritance(t *testing.T) { cfg.PrimaryLLM.BaseURL = "https://primary.example/v1" cfg.PrimaryLLM.TimeoutSeconds = 111 cfg.PrimaryLLM.MaxRetries = 2 - cfg.PrimaryLLM.Concurrency = 7 + cfg.TotalLLMConcurrency = 7 + cfg.syncLegacyConcurrencyAliases() effective := cfg.EffectiveValidationLLMConfig() if effective.APIKey != "primary-key" || effective.Model != "primary-model" || effective.BaseURL != "https://primary.example/v1" || effective.TimeoutSeconds != 111 || effective.MaxRetries != 2 || effective.Concurrency != 7 { @@ -208,7 +305,8 @@ func TestEffectiveValidationLLMInheritance(t *testing.T) { cfg.ValidationLLM.TimeoutSeconds = &validationTimeout cfg.ValidationLLM.MaxRetries = &validationRetries validationConcurrency := 4 - cfg.ValidationLLM.Concurrency = &validationConcurrency + cfg.ValidationLLMConcurrency = &validationConcurrency + cfg.syncLegacyConcurrencyAliases() effective = cfg.EffectiveValidationLLMConfig() if effective.APIKey != "validation-key" || effective.Model != "validation-model" || effective.BaseURL != "https://validation.example/v1" || effective.TimeoutSeconds != 222 || effective.MaxRetries != 9 || effective.Concurrency != 4 { @@ -216,34 +314,50 @@ func TestEffectiveValidationLLMInheritance(t *testing.T) { } } -func TestValidationLLMConcurrencyCannotExceedPrimary(t *testing.T) { +func TestValidationLLMConcurrencyCannotExceedTotal(t *testing.T) { cfg := Default() - cfg.PrimaryLLM.Concurrency = 2 + cfg.TotalLLMConcurrency = 2 + cfg.ProposalLLMConcurrency = 2 validationConcurrency := 3 - cfg.ValidationLLM.Concurrency = &validationConcurrency + cfg.ValidationLLMConcurrency = &validationConcurrency if err := cfg.Validate(); err == nil { - t.Fatal("expected validation error when validation llm concurrency exceeds primary") + t.Fatal("expected validation error when validation llm concurrency exceeds total") } validationConcurrency = 2 - cfg.ValidationLLM.Concurrency = &validationConcurrency + cfg.ValidationLLMConcurrency = &validationConcurrency if err := cfg.Validate(); err != nil { t.Fatalf("expected equal concurrency to validate, got %v", err) } } -func TestCLIPrimaryLLMConcurrencyOverrideDrivesEffectiveValidationConcurrencyWhenValidationUnset(t *testing.T) { +func TestProposalLLMConcurrencyCannotExceedTotal(t *testing.T) { cfg := Default() - llmConcurrency := 6 - if err := cfg.ApplyCLIOverrides(CLIOverrides{PrimaryLLMConcurrency: &llmConcurrency}); err != nil { + cfg.TotalLLMConcurrency = 2 + cfg.ProposalLLMConcurrency = 3 + + if err := cfg.Validate(); err == nil { + t.Fatal("expected validation error when proposal llm concurrency exceeds total") + } + + cfg.ProposalLLMConcurrency = 2 + if err := cfg.Validate(); err != nil { + t.Fatalf("expected equal concurrency to validate, got %v", err) + } +} + +func TestCLITotalLLMConcurrencyOverrideDrivesEffectiveValidationConcurrencyWhenValidationUnset(t *testing.T) { + cfg := Default() + totalLLMConcurrency := 6 + if err := cfg.ApplyCLIOverrides(CLIOverrides{TotalLLMConcurrency: &totalLLMConcurrency}); err != nil { t.Fatalf("ApplyCLIOverrides failed: %v", err) } - if cfg.ValidationLLM.Concurrency != nil { - t.Fatalf("expected validation concurrency to remain unset, got %#v", cfg.ValidationLLM.Concurrency) + if cfg.ValidationLLMConcurrency != nil { + t.Fatalf("expected validation concurrency to remain unset, got %#v", cfg.ValidationLLMConcurrency) } - if cfg.EffectiveValidationLLMConfig().Concurrency != 6 { - t.Fatalf("expected inherited validation concurrency 6, got %d", cfg.EffectiveValidationLLMConfig().Concurrency) + if cfg.EffectiveValidationLLMConcurrency() != 6 { + t.Fatalf("expected inherited validation concurrency 6, got %d", cfg.EffectiveValidationLLMConcurrency()) } } diff --git a/internal/core/config/env.go b/internal/core/config/env.go index 0ef70c3..33d19a8 100644 --- a/internal/core/config/env.go +++ b/internal/core/config/env.go @@ -68,12 +68,37 @@ func loadFromLookup(lookup func(string) (string, bool)) (Config, error) { } cfg.PrimaryLLM.MaxRetries = value } + totalConcurrencySet := false + if raw, ok := lookup("AUDITA_TOTAL_LLM_CONCURRENCY"); ok { + value, err := parseInt(raw) + if err != nil { + return Config{}, fmt.Errorf("AUDITA_TOTAL_LLM_CONCURRENCY: %w", err) + } + cfg.TotalLLMConcurrency = value + totalConcurrencySet = true + } if raw, ok := lookup("AUDITA_LLM_CONCURRENCY"); ok { value, err := parseInt(raw) if err != nil { return Config{}, fmt.Errorf("AUDITA_LLM_CONCURRENCY: %w", err) } - cfg.PrimaryLLM.Concurrency = value + if !totalConcurrencySet { + cfg.TotalLLMConcurrency = value + totalConcurrencySet = true + } + } + + proposalConcurrencySet := false + if raw, ok := lookup("AUDITA_PROPOSAL_LLM_CONCURRENCY"); ok { + value, err := parseInt(raw) + if err != nil { + return Config{}, fmt.Errorf("AUDITA_PROPOSAL_LLM_CONCURRENCY: %w", err) + } + cfg.ProposalLLMConcurrency = value + proposalConcurrencySet = true + } + if totalConcurrencySet && !proposalConcurrencySet { + cfg.ProposalLLMConcurrency = cfg.TotalLLMConcurrency } if raw, ok := lookup("AUDITA_VALIDATION_MAX_RETRIES"); ok { @@ -88,7 +113,7 @@ func loadFromLookup(lookup func(string) (string, bool)) (Config, error) { if err != nil { return Config{}, fmt.Errorf("AUDITA_VALIDATION_LLM_CONCURRENCY: %w", err) } - cfg.ValidationLLM.Concurrency = &value + cfg.ValidationLLMConcurrency = &value } if raw, ok := lookup("AUDITA_VALIDATION_MAX_PROMPT_TOKENS"); ok { @@ -188,6 +213,8 @@ func loadFromLookup(lookup func(string) (string, bool)) (Config, error) { cfg.WorkDirRetention = WorkDirRetention(raw) } + cfg.syncLegacyConcurrencyAliases() + if err := cfg.Validate(); err != nil { return Config{}, err } diff --git a/internal/core/config/flags.go b/internal/core/config/flags.go index 7e2cef2..4add17c 100644 --- a/internal/core/config/flags.go +++ b/internal/core/config/flags.go @@ -11,6 +11,8 @@ type CLIOverrides struct { PrimaryBaseURL *string ValidationBaseURL *string PrimaryLLMTimeoutSeconds *int + TotalLLMConcurrency *int + ProposalLLMConcurrency *int PrimaryLLMConcurrency *int ValidationLLMTimeoutSeconds *int MaxRetries *int @@ -62,8 +64,24 @@ func (c *Config) ApplyCLIOverrides(overrides CLIOverrides) error { if overrides.PrimaryLLMTimeoutSeconds != nil { c.PrimaryLLM.TimeoutSeconds = *overrides.PrimaryLLMTimeoutSeconds } - if overrides.PrimaryLLMConcurrency != nil { - c.PrimaryLLM.Concurrency = *overrides.PrimaryLLMConcurrency + totalConcurrencySet := false + if overrides.TotalLLMConcurrency != nil { + c.TotalLLMConcurrency = *overrides.TotalLLMConcurrency + totalConcurrencySet = true + } + // Backward-compatible alias: --llm-concurrency maps to total concurrency + // only when --total-llm-concurrency is not set in the same CLI invocation. + if overrides.PrimaryLLMConcurrency != nil && !totalConcurrencySet { + c.TotalLLMConcurrency = *overrides.PrimaryLLMConcurrency + totalConcurrencySet = true + } + proposalConcurrencySet := false + if overrides.ProposalLLMConcurrency != nil { + c.ProposalLLMConcurrency = *overrides.ProposalLLMConcurrency + proposalConcurrencySet = true + } + if totalConcurrencySet && !proposalConcurrencySet { + c.ProposalLLMConcurrency = c.TotalLLMConcurrency } if overrides.ValidationLLMTimeoutSeconds != nil { value := *overrides.ValidationLLMTimeoutSeconds @@ -78,7 +96,7 @@ func (c *Config) ApplyCLIOverrides(overrides CLIOverrides) error { } if overrides.ValidationLLMConcurrency != nil { value := *overrides.ValidationLLMConcurrency - c.ValidationLLM.Concurrency = &value + c.ValidationLLMConcurrency = &value } if overrides.ValidationMaxPromptTokens != nil { c.ValidationMaxPromptTokens = *overrides.ValidationMaxPromptTokens @@ -124,5 +142,7 @@ func (c *Config) ApplyCLIOverrides(overrides CLIOverrides) error { c.WorkDirRetention = WorkDirRetention(*overrides.WorkDirRetention) } + c.syncLegacyConcurrencyAliases() + return c.Validate() } diff --git a/internal/core/config/validation.go b/internal/core/config/validation.go index b549bed..92c509c 100644 --- a/internal/core/config/validation.go +++ b/internal/core/config/validation.go @@ -24,8 +24,14 @@ func (c Config) Validate() error { if c.PrimaryLLM.MaxRetries < 0 { issues = append(issues, "max retries must be zero or greater") } - if c.PrimaryLLM.Concurrency <= 0 { - issues = append(issues, "primary llm concurrency must be greater than zero") + if c.TotalLLMConcurrency <= 0 { + issues = append(issues, "total llm concurrency must be greater than zero") + } + if c.ProposalLLMConcurrency <= 0 { + issues = append(issues, "proposal llm concurrency must be greater than zero") + } + if c.ProposalLLMConcurrency > c.TotalLLMConcurrency { + issues = append(issues, "proposal llm concurrency must be less than or equal to total llm concurrency") } if c.ValidationLLM.TimeoutSeconds != nil && *c.ValidationLLM.TimeoutSeconds <= 0 { @@ -34,11 +40,11 @@ func (c Config) Validate() error { if c.ValidationLLM.MaxRetries != nil && *c.ValidationLLM.MaxRetries < 0 { issues = append(issues, "validation max retries must be zero or greater") } - if c.ValidationLLM.Concurrency != nil && *c.ValidationLLM.Concurrency <= 0 { + if c.ValidationLLMConcurrency != nil && *c.ValidationLLMConcurrency <= 0 { issues = append(issues, "validation llm concurrency must be greater than zero") } - if c.ValidationLLM.Concurrency != nil && *c.ValidationLLM.Concurrency > c.PrimaryLLM.Concurrency { - issues = append(issues, "validation llm concurrency must be less than or equal to primary llm concurrency") + if c.ValidationLLMConcurrency != nil && *c.ValidationLLMConcurrency > c.TotalLLMConcurrency { + issues = append(issues, "validation llm concurrency must be less than or equal to total llm concurrency") } if c.ValidationMaxPromptTokens <= 0 { diff --git a/internal/framework/llm/effective_config.go b/internal/framework/llm/effective_config.go index 6517f00..ceb0469 100644 --- a/internal/framework/llm/effective_config.go +++ b/internal/framework/llm/effective_config.go @@ -21,13 +21,13 @@ type EffectiveConfig struct { // ResolvePrimaryConfig resolves the primary LLM settings into runtime form. func ResolvePrimaryConfig(cfg config.Config) EffectiveConfig { - return resolveFromLLMConfig(cfg.PrimaryLLM) + return resolveFromLLMConfig(cfg.PrimaryLLM, cfg.TotalLLMConcurrency) } // ResolveValidationConfig resolves validation LLM settings with inheritance // from primary values when validation overrides are unset. func ResolveValidationConfig(cfg config.Config) EffectiveConfig { - return resolveFromLLMConfig(cfg.EffectiveValidationLLMConfig()) + return resolveFromLLMConfig(cfg.EffectiveValidationLLMConfig(), cfg.EffectiveValidationLLMConcurrency()) } // ToInstructorClientConfig converts an effective runtime config into adapter @@ -44,13 +44,13 @@ func (c EffectiveConfig) ToInstructorClientConfig(mode Mode, httpClient *http.Cl } } -func resolveFromLLMConfig(raw config.LLMConfig) EffectiveConfig { +func resolveFromLLMConfig(raw config.LLMConfig, concurrency int) EffectiveConfig { return EffectiveConfig{ APIKey: strings.TrimSpace(raw.APIKey), Model: strings.TrimSpace(raw.Model), BaseURL: strings.TrimSpace(raw.BaseURL), RequestTimeout: time.Duration(raw.TimeoutSeconds) * time.Second, MaxRetries: raw.MaxRetries, - Concurrency: raw.Concurrency, + Concurrency: concurrency, } } diff --git a/internal/framework/llm/effective_config_test.go b/internal/framework/llm/effective_config_test.go index f39c9c3..84a8d39 100644 --- a/internal/framework/llm/effective_config_test.go +++ b/internal/framework/llm/effective_config_test.go @@ -14,7 +14,8 @@ func TestResolvePrimaryConfig(t *testing.T) { cfg.PrimaryLLM.BaseURL = " https://example.test/v1 " cfg.PrimaryLLM.TimeoutSeconds = 42 cfg.PrimaryLLM.MaxRetries = 7 - cfg.PrimaryLLM.Concurrency = 3 + cfg.TotalLLMConcurrency = 3 + cfg.ProposalLLMConcurrency = 3 effective := ResolvePrimaryConfig(cfg) if effective.APIKey != "primary-key" { @@ -44,7 +45,8 @@ func TestResolveValidationConfigInheritsPrimaryWhenUnset(t *testing.T) { cfg.PrimaryLLM.BaseURL = "https://primary.example/v1" cfg.PrimaryLLM.TimeoutSeconds = 90 cfg.PrimaryLLM.MaxRetries = 4 - cfg.PrimaryLLM.Concurrency = 2 + cfg.TotalLLMConcurrency = 2 + cfg.ProposalLLMConcurrency = 2 effective := ResolveValidationConfig(cfg) if effective.APIKey != "primary-key" || @@ -64,7 +66,8 @@ func TestResolveValidationConfigOverridesPrimaryWhenSet(t *testing.T) { cfg.PrimaryLLM.BaseURL = "https://primary.example/v1" cfg.PrimaryLLM.TimeoutSeconds = 90 cfg.PrimaryLLM.MaxRetries = 4 - cfg.PrimaryLLM.Concurrency = 2 + cfg.TotalLLMConcurrency = 2 + cfg.ProposalLLMConcurrency = 2 timeout := 12 retries := 9 @@ -74,7 +77,7 @@ func TestResolveValidationConfigOverridesPrimaryWhenSet(t *testing.T) { cfg.ValidationLLM.BaseURL = "https://validation.example/v1" cfg.ValidationLLM.TimeoutSeconds = &timeout cfg.ValidationLLM.MaxRetries = &retries - cfg.ValidationLLM.Concurrency = &concurrency + cfg.ValidationLLMConcurrency = &concurrency effective := ResolveValidationConfig(cfg) if effective.APIKey != "validation-key" || diff --git a/internal/framework/llm/scheduler.go b/internal/framework/llm/scheduler.go index 78f80bc..7aafa01 100644 --- a/internal/framework/llm/scheduler.go +++ b/internal/framework/llm/scheduler.go @@ -8,7 +8,16 @@ import ( // Scheduler bounds concurrent backend LLM calls. type Scheduler struct { - permits chan struct{} + maxConcurrency int + mu sync.Mutex + inFlight int + queue []*waiter +} + +type waiter struct { + ready chan struct{} + queued bool + granted bool } // NewScheduler creates a scheduler with a fixed concurrency limit. @@ -17,26 +26,86 @@ func NewScheduler(maxConcurrency int) (*Scheduler, error) { return nil, fmt.Errorf("max concurrency must be greater than zero") } return &Scheduler{ - permits: make(chan struct{}, maxConcurrency), + maxConcurrency: maxConcurrency, }, nil } // Acquire blocks until a permit is available or the context is canceled. // The returned release function is safe to call multiple times. func (s *Scheduler) Acquire(ctx context.Context) (func(), error) { + if err := ctx.Err(); err != nil { + return nil, err + } + + s.mu.Lock() + if s.inFlight < s.maxConcurrency && len(s.queue) == 0 { + s.inFlight++ + s.mu.Unlock() + return s.releaseFunc(), nil + } + w := &waiter{ + ready: make(chan struct{}), + queued: true, + } + s.queue = append(s.queue, w) + s.mu.Unlock() + select { - case s.permits <- struct{}{}: - var once sync.Once - return func() { - once.Do(func() { - <-s.permits - }) - }, nil + case <-w.ready: + return s.releaseFunc(), nil case <-ctx.Done(): + s.mu.Lock() + if w.queued { + s.removeQueuedWaiterLocked(w) + s.mu.Unlock() + return nil, ctx.Err() + } + if w.granted { + // Permit was granted concurrently with cancellation; release it to + // avoid leaks and unblock queued callers. + s.inFlight-- + s.grantQueuedLocked() + } + s.mu.Unlock() return nil, ctx.Err() } } +func (s *Scheduler) releaseFunc() func() { + var once sync.Once + return func() { + once.Do(func() { + s.mu.Lock() + if s.inFlight > 0 { + s.inFlight-- + s.grantQueuedLocked() + } + s.mu.Unlock() + }) + } +} + +func (s *Scheduler) grantQueuedLocked() { + for s.inFlight < s.maxConcurrency && len(s.queue) > 0 { + w := s.queue[0] + s.queue = s.queue[1:] + w.queued = false + w.granted = true + s.inFlight++ + close(w.ready) + } +} + +func (s *Scheduler) removeQueuedWaiterLocked(target *waiter) { + for i, w := range s.queue { + if w == target { + w.queued = false + s.queue = append(s.queue[:i], s.queue[i+1:]...) + return + } + } +} + // Run acquires a permit, executes fn, and releases the permit. func (s *Scheduler) Run(ctx context.Context, fn func(context.Context) error) error { release, err := s.Acquire(ctx) diff --git a/internal/framework/llm/scheduler_test.go b/internal/framework/llm/scheduler_test.go index 1981029..7d51f64 100644 --- a/internal/framework/llm/scheduler_test.go +++ b/internal/framework/llm/scheduler_test.go @@ -3,12 +3,64 @@ package llm import ( "context" "errors" + "reflect" "sync" "sync/atomic" "testing" "time" ) +func TestSchedulerFIFOOrdering(t *testing.T) { + s, err := NewScheduler(1) + if err != nil { + t.Fatalf("NewScheduler: %v", err) + } + + firstRelease, err := s.Acquire(context.Background()) + if err != nil { + t.Fatalf("Acquire(first): %v", err) + } + + gotOrder := make(chan int, 3) + waitChans := []chan struct{}{ + make(chan struct{}), + make(chan struct{}), + make(chan struct{}), + } + var wg sync.WaitGroup + for i := 0; i < 3; i++ { + id := i + wg.Add(1) + go func() { + defer wg.Done() + runErr := s.Run(context.Background(), func(context.Context) error { + gotOrder <- id + <-waitChans[id] + return nil + }) + if runErr != nil { + t.Errorf("Run[%d] error: %v", id, runErr) + } + }() + time.Sleep(10 * time.Millisecond) + } + + firstRelease() + + order := make([]int, 0, 3) + for i := 0; i < 3; i++ { + id := <-gotOrder + order = append(order, id) + close(waitChans[id]) + } + + wg.Wait() + + if !reflect.DeepEqual(order, []int{0, 1, 2}) { + t.Fatalf("expected FIFO order [0 1 2], got %v", order) + } +} + func TestSchedulerEnforcesMaxConcurrency(t *testing.T) { s, err := NewScheduler(2) if err != nil { @@ -79,7 +131,6 @@ func TestSchedulerReleasesPermitOnError(t *testing.T) { t.Fatalf("expected %v, got %v", expectedErr, err) } - // Must be able to run again after an error, proving permit release. if err := s.Run(context.Background(), func(ctx context.Context) error { _ = ctx return nil @@ -88,7 +139,7 @@ func TestSchedulerReleasesPermitOnError(t *testing.T) { } } -func TestSchedulerRespectsContextCancellation(t *testing.T) { +func TestSchedulerContextCancellationWhileQueued(t *testing.T) { s, err := NewScheduler(1) if err != nil { t.Fatalf("NewScheduler: %v", err) @@ -111,3 +162,27 @@ func TestSchedulerRespectsContextCancellation(t *testing.T) { t.Fatalf("expected deadline exceeded, got %v", err) } } + +func TestSchedulerNoPermitLeakAfterQueuedCancellation(t *testing.T) { + s, err := NewScheduler(1) + if err != nil { + t.Fatalf("NewScheduler: %v", err) + } + + firstRelease, err := s.Acquire(context.Background()) + if err != nil { + t.Fatalf("Acquire(first): %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := s.Acquire(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("expected context canceled, got %v", err) + } + + firstRelease() + + if err := s.Run(context.Background(), func(context.Context) error { return nil }); err != nil { + t.Fatalf("expected scheduler to accept new work after cancellation, got %v", err) + } +} diff --git a/internal/framework/proposal_generation/generate_test.go b/internal/framework/proposal_generation/generate_test.go index bb5cb4c..1328fbc 100644 --- a/internal/framework/proposal_generation/generate_test.go +++ b/internal/framework/proposal_generation/generate_test.go @@ -7,10 +7,14 @@ import ( "path/filepath" "reflect" "strings" + "sync" + "sync/atomic" "testing" + "time" "gitea.maximumdirect.net/eric/audita/internal/core/config" "gitea.maximumdirect.net/eric/audita/internal/framework/contracts" + "gitea.maximumdirect.net/eric/audita/internal/framework/llm" "gitea.maximumdirect.net/eric/audita/internal/framework/proposals" ) @@ -47,6 +51,40 @@ func (s *countingScheduler) Run(ctx context.Context, fn func(context.Context) er return fn(ctx) } +type sleepingStructuredClient struct { + inFlight int32 + maxInFlight int32 +} + +func (c *sleepingStructuredClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { + _ = req + current := atomic.AddInt32(&c.inFlight, 1) + for { + prior := atomic.LoadInt32(&c.maxInFlight) + if current <= prior || atomic.CompareAndSwapInt32(&c.maxInFlight, prior, current) { + break + } + } + select { + case <-time.After(20 * time.Millisecond): + case <-ctx.Done(): + atomic.AddInt32(&c.inFlight, -1) + return contracts.StructuredCompletionResponse{}, ctx.Err() + } + atomic.AddInt32(&c.inFlight, -1) + + target, ok := out.(*StructuredCorrectionSet) + if !ok { + return contracts.StructuredCompletionResponse{}, errors.New("unexpected output type") + } + *target = StructuredCorrectionSet{ + Corrections: []StructuredCorrectionProposal{ + {TargetSegmentID: 1, OriginalText: "x", CorrectedText: "y", Confidence: 0.9}, + }, + } + return contracts.StructuredCompletionResponse{}, nil +} + func defaultRequest(t *testing.T) Request { t.Helper() cfg := config.Default() @@ -265,3 +303,35 @@ func TestGenerateCandidatesClientError(t *testing.T) { t.Fatalf("expected error diagnostics artifact under %s", req.DiagnosticsDir) } } + +func TestGenerateCandidatesRespectsSchedulerConcurrency(t *testing.T) { + scheduler, err := llm.NewScheduler(2) + if err != nil { + t.Fatalf("NewScheduler: %v", err) + } + client := &sleepingStructuredClient{} + baseReq := defaultRequest(t) + baseReq.LLMClient = client + baseReq.Scheduler = scheduler + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func(idx int) { + defer wg.Done() + req := baseReq + req.StartIndex = idx + if _, runErr := GenerateCandidates(context.Background(), req); runErr != nil { + t.Errorf("GenerateCandidates[%d] error: %v", idx, runErr) + } + }(i) + } + wg.Wait() + + if got := atomic.LoadInt32(&client.maxInFlight); got > 2 { + t.Fatalf("expected scheduler cap <= 2, got %d", got) + } + if got := atomic.LoadInt32(&client.maxInFlight); got < 2 { + t.Fatalf("expected observed concurrency of at least 2, got %d", got) + } +} diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go index 9e253ca..cc159a8 100644 --- a/internal/framework/runner/runner.go +++ b/internal/framework/runner/runner.go @@ -296,8 +296,8 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI } maxWorkers := 1 - if input.Config != nil && input.Config.PrimaryLLM.Concurrency > 1 { - maxWorkers = input.Config.PrimaryLLM.Concurrency + if input.Config != nil && input.Config.EffectiveProposalLLMConcurrency() > 1 { + maxWorkers = input.Config.EffectiveProposalLLMConcurrency() } if maxWorkers > len(input.Sections) { maxWorkers = len(input.Sections) diff --git a/internal/framework/runner/runner_test.go b/internal/framework/runner/runner_test.go index 227200b..923c576 100644 --- a/internal/framework/runner/runner_test.go +++ b/internal/framework/runner/runner_test.go @@ -163,11 +163,12 @@ func TestRunnerProposalsExecutePerChunkSection(t *testing.T) { } } -func TestRunnerProposalSectionConcurrencyBoundedByPrimaryLLMConcurrency(t *testing.T) { +func TestRunnerProposalSectionConcurrencyBoundedByProposalLLMConcurrency(t *testing.T) { cfg := config.Default() cfg.MaxSectionTokens = 3 cfg.MinSectionTokens = 0 - cfg.PrimaryLLM.Concurrency = 2 + cfg.TotalLLMConcurrency = 2 + cfg.ProposalLLMConcurrency = 2 transcript := &schema.Transcript{Segments: []schema.Segment{ {ID: 1, Text: "one two"}, @@ -215,7 +216,8 @@ func TestRunnerProposalIndexOrderingIsDeterministicAcrossParallelSections(t *tes cfg := config.Default() cfg.MaxSectionTokens = 3 cfg.MinSectionTokens = 0 - cfg.PrimaryLLM.Concurrency = 3 + cfg.TotalLLMConcurrency = 3 + cfg.ProposalLLMConcurrency = 3 transcript := &schema.Transcript{Segments: []schema.Segment{ {ID: 1, Text: "alpha one"}, @@ -259,6 +261,12 @@ func TestRunnerProposalIndexOrderingIsDeterministicAcrossParallelSections(t *tes t.Fatalf("expected deterministic applied-change order by section/segment, got %+v", out.ModuleResults[0].AppliedChanges) } } + wantFinal := []string{"ALPHA ONE", "BRAVO TWO", "CHARLIE THREE"} + for i, seg := range out.FinalTranscript.Segments { + if seg.Text != wantFinal[i] { + t.Fatalf("expected deterministic final transcript regardless of section completion order; got %+v", out.FinalTranscript.Segments) + } + } } func TestRunnerSkippedRecorded(t *testing.T) { diff --git a/internal/framework/validators/llm_validators_test.go b/internal/framework/validators/llm_validators_test.go index 1a0c31d..bb741e3 100644 --- a/internal/framework/validators/llm_validators_test.go +++ b/internal/framework/validators/llm_validators_test.go @@ -4,7 +4,10 @@ import ( "context" "errors" "strings" + "sync" + "sync/atomic" "testing" + "time" "gitea.maximumdirect.net/eric/audita/internal/core/chunking" "gitea.maximumdirect.net/eric/audita/internal/core/config" @@ -18,6 +21,54 @@ type fakeStructuredLLMClient struct { calls []StructuredCompletionRequest } +type sleepingValidationClient struct { + inFlight int32 + maxInFlight int32 +} + +type boundedScheduler struct { + permits chan struct{} +} + +func newBoundedScheduler(max int) *boundedScheduler { + return &boundedScheduler{permits: make(chan struct{}, max)} +} + +func (s *boundedScheduler) Run(ctx context.Context, fn func(context.Context) error) error { + select { + case s.permits <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + } + defer func() { <-s.permits }() + return fn(ctx) +} + +func (c *sleepingValidationClient) CompleteStructured(ctx context.Context, req StructuredCompletionRequest, out any) (StructuredCompletionResponse, error) { + _ = req + current := atomic.AddInt32(&c.inFlight, 1) + for { + prior := atomic.LoadInt32(&c.maxInFlight) + if current <= prior || atomic.CompareAndSwapInt32(&c.maxInFlight, prior, current) { + break + } + } + select { + case <-time.After(20 * time.Millisecond): + case <-ctx.Done(): + atomic.AddInt32(&c.inFlight, -1) + return StructuredCompletionResponse{}, ctx.Err() + } + atomic.AddInt32(&c.inFlight, -1) + target := out.(*LLMValidationResponse) + *target = LLMValidationResponse{ + Validations: []LLMValidationDecision{ + {CorrectionIndex: 0, Approved: true, Confidence: 0.9, Reason: "ok"}, + }, + } + return StructuredCompletionResponse{}, nil +} + func (f *fakeStructuredLLMClient) CompleteStructured(ctx context.Context, req StructuredCompletionRequest, out any) (StructuredCompletionResponse, error) { _ = ctx f.calls = append(f.calls, req) @@ -191,3 +242,34 @@ func TestLLMBackedValidatorUnknownProposalIndexFails(t *testing.T) { t.Fatalf("expected unknown index error, got %v", err) } } + +func TestLLMBackedValidatorRespectsSchedulerConcurrency(t *testing.T) { + scheduler := newBoundedScheduler(2) + client := &sleepingValidationClient{} + v, err := NewLLMBackedValidator("spoken_form_plausibility_review", LLMValidatorTypeSpokenFormPlausibility, "test-model") + if err != nil { + t.Fatalf("new validator error: %v", err) + } + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + req := makeReq([]proposals.EnrichedCorrectionProposal{mk(0, "gestures", "Jesters")}) + req.LLMClient = client + req.Scheduler = scheduler + if _, runErr := v.Validate(context.Background(), req); runErr != nil { + t.Errorf("Validate error: %v", runErr) + } + }() + } + wg.Wait() + + if got := atomic.LoadInt32(&client.maxInFlight); got > 2 { + t.Fatalf("expected scheduler cap <= 2, got %d", got) + } + if got := atomic.LoadInt32(&client.maxInFlight); got < 2 { + t.Fatalf("expected observed concurrency of at least 2, got %d", got) + } +}