From 4812032c9ed874c1f7cfcc4b54880cd2ff8ba614 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 24 Sep 2026 02:18:31 +0000 Subject: [PATCH] Connect processing to shared PromptKit backend capacity --- docs/internal/llm-runtime.md | 16 +- docs/internal/pipeline.md | 7 +- docs/policy/architecture.md | 16 +- docs/roadmap/implementation.md | 15 +- internal/cli/run.go | 180 ++---- internal/core/diagnostics/artifacts.go | 2 +- internal/core/diagnostics/run_dir.go | 11 +- internal/framework/contracts/contracts.go | 10 +- .../framework/llm/adapter_capacity_test.go | 214 +++++++ internal/framework/llm/adapter_test.go | 29 +- internal/framework/llm/scheduler.go | 117 ---- internal/framework/llm/scheduler_test.go | 223 ------- internal/framework/modules/registry.go | 24 +- .../framework/proposal_generation/generate.go | 19 +- .../proposal_generation/generate_test.go | 128 ---- .../proposal_generation/module_proposal.go | 1 - internal/framework/runner/observability.go | 350 ----------- internal/framework/runner/runner.go | 83 +-- internal/framework/runner/runner_test.go | 567 ++---------------- .../framework/validators/llm_validators.go | 25 +- .../validators/llm_validators_test.go | 111 ---- internal/framework/validators/models.go | 6 - internal/modules/glossary/module_test.go | 17 +- internal/modules/grammar/module_test.go | 17 +- internal/modules/homophones/module_test.go | 17 +- internal/modules/spoken_word/module_test.go | 17 +- 26 files changed, 406 insertions(+), 1816 deletions(-) create mode 100644 internal/framework/llm/adapter_capacity_test.go delete mode 100644 internal/framework/llm/scheduler.go delete mode 100644 internal/framework/llm/scheduler_test.go delete mode 100644 internal/framework/runner/observability.go diff --git a/docs/internal/llm-runtime.md b/docs/internal/llm-runtime.md index deb74f2..f262873 100644 --- a/docs/internal/llm-runtime.md +++ b/docs/internal/llm-runtime.md @@ -23,12 +23,12 @@ message and a typed cause; only exhausted structural repair is a malformed output failure. Secrets from all configured backends, including unselected ones, are available to the diagnostics redaction boundary. -## Current Orchestration +## Orchestration -The old `llm.Scheduler` and runner instrumentation still exist pending the -orchestration migration. The final runtime uses PromptKit backend admission, -retry, and repair rather than that scheduler. `llm.DiagnosticsWriter` remains -the current protected artifact writer. - -Focused adapter tests use controlled local HTTP responses. Complete runner and -CLI integration checks follow the caller and orchestration migration. +CLI processing supplies two role clients from the same engine to the runner. +PromptKit's per-backend admission applies to both roles and every generation, +retry, and repair. The runner still processes module instances in order, +collects concurrent section work deterministically, and cancels remaining +section work on a runtime failure. `llm.DiagnosticsWriter` remains the protected +request/response artifact writer. Execution observations and report summaries +are completed by the following roadmap work. diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 584237e..5956b59 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -38,10 +38,11 @@ Pipeline inputs are: `runner.Run` returns: - final transcript; -- per-module results; -- utilization diagnostics. +- per-module results. -CLI/reporting then map this into process report and diagnostics artifacts. +CLI/reporting then map these results into process reports and diagnostics +artifacts. PromptKit execution observations are collected at the role-client +boundary for the new report summary. ## Failure Behavior diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index 29cb053..a526735 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -22,7 +22,7 @@ and [integration documentation](../integrations/subprocess.md) own external cont normalization, chunking, output, diagnostics, and reporting contracts. Domain behavior must not require CLI or provider-transport knowledge. - **Framework:** owns orchestration contracts and shared runtime mechanics for - modules, validators, proposals, LLM adapters, scheduling, and report assembly. + modules, validators, proposals, LLM adapters, and report assembly. It composes domain operations through explicit inputs and outputs. - **Modules:** own narrowly scoped correction behavior and module-specific prompt choices. Keep module packages separate rather than collapsing their distinct @@ -45,9 +45,10 @@ hidden global state. - Validate corrections before final mutation, including local proposal checks, decision cardinality, and apply-time safety. Provider-side schema enforcement does not replace these checks. -- Keep concurrency bounded, configurable, context-aware, and observable through - the existing scheduler and utilization diagnostics. Preserve permit release on - success, failure, and cancellation. +- Keep LLM concurrency bounded and context-aware through PromptKit's engine-local + per-backend admission. Proposal and validation calls selecting the same backend + share that capacity, including retries and repairs. Preserve deterministic + collection order and cancellation of remaining work on runtime failure. - Keep module and validator identities, reason codes, and execution-class distinctions stable where exposed as contracts. @@ -84,9 +85,10 @@ responsibilities and should remain separate. Implementation details belong in ## Dependencies Keep Audita dependency-light and prefer the Go standard library for CLI parsing, -HTTP, JSON, filesystems, synchronization, and tests. Third-party dependencies -should be narrow, justified, and preferably established for their purpose. -YAML parsing is the existing direct dependency exception. +filesystems, synchronization, and tests. ConfigKit owns configuration loading, +presence, provenance, and secret resolution; PromptKit owns prompt/profile +loading, OpenAI-compatible transport, retries, repairs, and backend admission. +Audita owns domain meaning, role selection, and deterministic processing. Do not introduce broad CLI, dependency-injection, workflow, logging, or plugin frameworks without a concrete implemented need and focused tests. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 87109d7..b164f54 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -382,7 +382,20 @@ full runner behavior is Stage 8. ## Stage 8: Connect the Runner and Remove the Old Scheduling Hierarchy -**Status:** Pending. **Depends on:** Stages 2 and 4–7. +**Status:** Complete (2026-09-24). CLI processing now supplies both role clients +from one PromptKit engine; removed global/role schedulers, scheduler contracts, +the old utilization instrumentation, and retired concurrency settings. Runner +ordering, cancellation, partial failure handling, and domain checks remain. +Updated architecture and runtime guidance. `go build -o /tmp/audita-stage8-build +./cmd/audita` passed; focused diagnostics, module, proposal, validator, runner, +and LLM suites passed; `go test -race ./internal/framework/llm` passed. Local +HTTP tests covered shared backend capacity, independent backend identities, +zero-wait rejection, queued deadline cancellation, and later capacity reuse. +A real CLI-to-local-HTTP run completed with one proposal and two validation +calls and wrote the corrected transcript. The legacy CLI test file still uses +retired configuration and report contracts; assembled CLI tests remain for +Stages 10–12. `git diff --check` passed. +**Depends on:** Stages 2 and 4–7. **Owns:** CLI process assembly, `internal/framework/runner`, old scheduler contracts, and runner/CLI integration tests. diff --git a/internal/cli/run.go b/internal/cli/run.go index 588a518..f5f895f 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -1,11 +1,10 @@ package cli import ( + "bytes" "context" "fmt" "io" - "os" - "path/filepath" "strings" "time" @@ -22,29 +21,9 @@ import ( "gitea.maximumdirect.net/eric/audita/internal/framework/llm" "gitea.maximumdirect.net/eric/audita/internal/framework/modules" "gitea.maximumdirect.net/eric/audita/internal/framework/processreport" - "gitea.maximumdirect.net/eric/audita/internal/framework/proposal_generation" "gitea.maximumdirect.net/eric/audita/internal/framework/runner" - "gitea.maximumdirect.net/eric/audita/internal/framework/validators" ) -type noOpStructuredLLMClient struct{} - -func (c noOpStructuredLLMClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { - _ = ctx - _ = req - switch target := out.(type) { - case *validators.LLMValidationResponse: - *target = validators.LLMValidationResponse{Validations: []validators.LLMValidationDecision{}} - case *proposal_generation.StructuredCorrectionSet: - *target = proposal_generation.StructuredCorrectionSet{Corrections: nil} - } - return contracts.StructuredCompletionResponse{}, nil -} - -func shouldUseNoOpLLMClientForTests() bool { - return strings.HasSuffix(filepath.Base(os.Args[0]), ".test") || os.Getenv("GO_WANT_HELPER_PROCESS") == "1" -} - type processInvocation struct { TranscriptPath string GlossaryPath string @@ -61,15 +40,13 @@ type processInvocation struct { var processModuleFactory runner.ModuleFactory var processProposalLLMClient contracts.StructuredLLMClient -var processProposalLLMScheduler runner.ValidationScheduler var processValidationLLMClient contracts.StructuredLLMClient -var processValidationLLMScheduler runner.ValidationScheduler var processRunnerContext = func() (context.Context, context.CancelFunc) { return context.Background(), func() {} } var processRunner = func(inv processInvocation, stdout io.Writer) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) { - runDir, err := diagnostics.NewRunDirectory(inv.Config.WorkDir, string(inv.Config.WorkDirRetention)) + runDir, err := diagnostics.NewRunDirectory(inv.Config.Diagnostics.WorkDir, string(inv.Config.Diagnostics.Retention)) if err != nil { return nil, nil, nil, nil, fmt.Errorf("run_dir_creation: %w", err) } @@ -88,13 +65,19 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio ConfigPath: inv.ConfigPath, ConfigSource: inv.ConfigSource, ConfigVersion: inv.ConfigVersion, - TranscriptDescription: inv.Config.TranscriptDescription, - Modules: append([]string(nil), inv.Config.Modules...), + TranscriptDescription: inv.Config.Context.Description, + Modules: append([]string(nil), inv.Config.Pipeline.Modules...), }); err != nil { _ = runDir.WriteErrorLog(fmt.Sprintf("invocation_metadata: %v", err)) } - if err := runDir.WriteEffectiveConfig(inv.Config); err != nil { - _ = runDir.WriteErrorLog(fmt.Sprintf("effective_config: %v", err)) + if inv.ConfigResult != nil { + var effective bytes.Buffer + if err := inv.ConfigResult.WriteEffective(&effective); err == nil { + err = runDir.WriteEffectiveConfig(effective.Bytes()) + } + if err != nil { + _ = runDir.WriteErrorLog(fmt.Sprintf("effective_config: %v", err)) + } } transcriptBytes, err := coreio.ReadRequiredFile(inv.TranscriptPath, "transcript") @@ -140,9 +123,9 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio // Compute chunks after normalization chunker := chunking.NewChunker(chunking.ChunkingConfig{ - MaxSectionTokens: inv.Config.MaxSectionTokens, - MinSectionTokens: inv.Config.MinSectionTokens, - TargetSections: inv.Config.TargetSections, + MaxSectionTokens: inv.Config.Chunking.MaxSectionTokens, + MinSectionTokens: inv.Config.Chunking.MinSectionTokens, + TargetSections: inv.Config.Chunking.TargetSections, }) sections, chunkErr := chunker.ChunkTranscript(normalizedTranscript) @@ -151,9 +134,9 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio } chunkConfig := chunking.ChunkingConfig{ - MaxSectionTokens: inv.Config.MaxSectionTokens, - MinSectionTokens: inv.Config.MinSectionTokens, - TargetSections: inv.Config.TargetSections, + MaxSectionTokens: inv.Config.Chunking.MaxSectionTokens, + MinSectionTokens: inv.Config.Chunking.MinSectionTokens, + TargetSections: inv.Config.Chunking.TargetSections, } chunkSummary := chunking.ComputeSummary(sections, chunkConfig) chunkDetailedSummary := chunking.ComputeDetailedSummary(sections, chunkConfig) @@ -175,66 +158,23 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio if moduleFactory != nil { proposalLLMClient := processProposalLLMClient validationLLMClient := processValidationLLMClient - proposalScheduler := processProposalLLMScheduler - validationScheduler := processValidationLLMScheduler - - if processModuleFactory == nil { - // Production runtime path: construct clients/schedulers from config. - if proposalLLMClient == nil { - if shouldUseNoOpLLMClientForTests() { - proposalLLMClient = noOpStructuredLLMClient{} - } else { - primaryCfg := llm.ResolvePrimaryConfig(inv.Config) - client, clientErr := llm.NewOpenAICompatibleClient(primaryCfg.ToOpenAICompatibleClientConfig(nil)) - if clientErr != nil { - return fail("runner_setup", clientErr, nil) - } - proposalLLMClient = client - } + if inv.LLMRuntime == nil && (proposalLLMClient == nil || validationLLMClient == nil) { + return fail("runner_setup", fmt.Errorf("LLM runtime is required"), nil) + } + if proposalLLMClient == nil { + proposalLLMClient, err = inv.LLMRuntime.NewRoleClient("proposal", llm.CallHooks{}) + if err != nil { + return fail("runner_setup", err, nil) } - if validationLLMClient == nil { - if shouldUseNoOpLLMClientForTests() { - validationLLMClient = noOpStructuredLLMClient{} - } else { - validationCfg := llm.ResolveValidationConfig(inv.Config) - client, clientErr := llm.NewOpenAICompatibleClient(validationCfg.ToOpenAICompatibleClientConfig(nil)) - if clientErr != nil { - return fail("runner_setup", clientErr, nil) - } - validationLLMClient = client - } - } - globalScheduler := proposalScheduler - if globalScheduler == nil { - s, sErr := llm.NewScheduler(inv.Config.TotalLLMConcurrency) - if sErr != nil { - return fail("runner_setup", sErr, nil) - } - globalScheduler = s - } - 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.ValidationLLMConcurrency != nil && inv.Config.EffectiveValidationLLMConcurrency() < inv.Config.TotalLLMConcurrency { - s, sErr := llm.NewScheduler(inv.Config.EffectiveValidationLLMConcurrency()) - if sErr != nil { - return fail("runner_setup", sErr, nil) - } - validationScheduler = composeSchedulers(globalScheduler, s) - } + } + if validationLLMClient == nil { + validationLLMClient, err = inv.LLMRuntime.NewRoleClient("validation", llm.CallHooks{}) + if err != nil { + return fail("runner_setup", err, nil) } } - moduleSpecs, err := contracts.ResolveModuleRunSpecs(inv.Config.Modules) + moduleSpecs, err := contracts.ResolveModuleRunSpecs(inv.Config.Pipeline.Modules) if err != nil { return fail("runner_setup", err, nil) } @@ -246,16 +186,10 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio Transcript: normalizedTranscript, Glossary: glossary, ModuleSpecs: moduleSpecs, - EffectiveConcurrency: runner.EffectiveConcurrencyLimits{ - TotalLLM: inv.Config.TotalLLMConcurrency, - ProposalLLM: inv.Config.EffectiveProposalLLMConcurrency(), - ValidationLLM: inv.Config.EffectiveValidationLLMConcurrency(), - }, + ProposalLLMClient: proposalLLMClient, - ProposalLLMScheduler: proposalScheduler, ProposalDiagnosticsDir: runDir.Path(), ValidationLLMClient: validationLLMClient, - ValidationLLMScheduler: validationScheduler, ValidationDiagnosticsDir: runDir.Path(), }) runOutput = &runnerResult @@ -265,7 +199,7 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio workingTranscript = runnerResult.FinalTranscript } - encoderDef, err := outputschema.Resolve(inv.Config.OutputSchema) + encoderDef, err := outputschema.Resolve(inv.Config.Output.Schema) if err != nil { return fail("output_schema", err, runOutput) } @@ -307,42 +241,6 @@ func sourceToCanonicalTranscript(source *schema.SourceTranscript) *schema.Transc return &schema.Transcript{Segments: segments} } -type chainedScheduler struct { - schedulers []runner.ValidationScheduler -} - -func (s chainedScheduler) Run(ctx context.Context, fn func(context.Context) error) error { - if len(s.schedulers) == 0 { - return fn(ctx) - } - run := fn - for i := len(s.schedulers) - 1; i >= 0; i-- { - scheduler := s.schedulers[i] - next := run - run = func(callCtx context.Context) error { - return scheduler.Run(callCtx, next) - } - } - return run(ctx) -} - -func composeSchedulers(schedulers ...runner.ValidationScheduler) runner.ValidationScheduler { - filtered := make([]runner.ValidationScheduler, 0, len(schedulers)) - for _, scheduler := range schedulers { - if scheduler != nil { - filtered = append(filtered, scheduler) - } - } - switch len(filtered) { - case 0: - return nil - case 1: - return filtered[0] - default: - return chainedScheduler{schedulers: filtered} - } -} - // Run executes the Audita CLI with the provided arguments and streams. func Run(args []string, stdout, stderr io.Writer) int { if len(args) == 0 { @@ -428,9 +326,7 @@ func runProcess(args []string, stdout, stderr io.Writer) int { if runErr != nil { if runDir != nil && runOutput != nil { - if runOutput.Utilization != nil { - _ = runDir.WriteJSONArtifact(diagnostics.ArtifactUtilizationSummary, runOutput.Utilization) - } + _ = runDir.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{ RunDirectoryPath: runDir.Path(), RunOutput: runOutput, @@ -460,9 +356,7 @@ func runProcess(args []string, stdout, stderr io.Writer) int { } if runDir != nil && runOutput != nil { - if runOutput.Utilization != nil { - _ = runDir.WriteJSONArtifact(diagnostics.ArtifactUtilizationSummary, runOutput.Utilization) - } + _ = runDir.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{ RunDirectoryPath: runDir.Path(), RunOutput: runOutput, @@ -606,8 +500,8 @@ func processReportInput(status string, inv processInvocation, runDir *diagnostic TranscriptPath: inv.TranscriptPath, GlossaryPath: inv.GlossaryPath, OutputPath: inv.OutputPath, - Modules: inv.Config.Modules, - OutputSchema: inv.Config.OutputSchema, + Modules: inv.Config.Pipeline.Modules, + OutputSchema: inv.Config.Output.Schema, ConfigVersion: inv.ConfigVersion, StartedAt: startedAt, CompletedAt: completedAt, diff --git a/internal/core/diagnostics/artifacts.go b/internal/core/diagnostics/artifacts.go index 13f4f41..4946e8a 100644 --- a/internal/core/diagnostics/artifacts.go +++ b/internal/core/diagnostics/artifacts.go @@ -15,7 +15,7 @@ const ( ArtifactUtilizationSummary = "utilization-diagnostics.json" ArtifactCorrectionLedger = "correction-ledger.json" ArtifactInvocationMetadata = "invocation.json" - ArtifactEffectiveConfig = "effective-config.json" + ArtifactEffectiveConfig = "effective-config.yaml" ArtifactReport = "report.json" ArtifactErrorLog = "error.log" ) diff --git a/internal/core/diagnostics/run_dir.go b/internal/core/diagnostics/run_dir.go index aa9d232..34affed 100644 --- a/internal/core/diagnostics/run_dir.go +++ b/internal/core/diagnostics/run_dir.go @@ -8,7 +8,6 @@ import ( "time" "gitea.maximumdirect.net/eric/audita/internal/core/chunking" - "gitea.maximumdirect.net/eric/audita/internal/core/config" "gitea.maximumdirect.net/eric/audita/internal/core/normalization" "gitea.maximumdirect.net/eric/audita/internal/core/reporting" "gitea.maximumdirect.net/eric/audita/internal/core/schema" @@ -119,15 +118,9 @@ func (r *RunDirectory) WriteInvocationMetadata(metadata InvocationMetadata) erro } // WriteEffectiveConfig writes redacted effective config metadata for this run. -func (r *RunDirectory) WriteEffectiveConfig(cfg config.Config) error { +func (r *RunDirectory) WriteEffectiveConfig(effectiveYAML []byte) error { path := filepath.Join(r.path, ArtifactEffectiveConfig) - redacted := cfg.Redacted() - bytes, err := json.MarshalIndent(redacted, "", " ") - if err != nil { - return fmt.Errorf("failed to marshal effective config: %w", err) - } - bytes = append(bytes, '\n') - if err := os.WriteFile(path, bytes, 0o644); err != nil { + if err := os.WriteFile(path, effectiveYAML, 0o644); err != nil { return fmt.Errorf("failed to write effective config: %w", err) } return nil diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index e350b3d..4678539 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -15,11 +15,6 @@ import ( type StructuredLLMClient = structuredoutput.Client -// LLMScheduler provides bounded execution for LLM call sites. -type LLMScheduler interface { - Run(ctx context.Context, fn func(context.Context) error) error -} - // TranscriptModule is the minimal contract for framework-integrated modules. type TranscriptModule interface { Key() string @@ -74,9 +69,8 @@ type ExecutionContext struct { // ProposalRequest is the input to module proposal generation. type ProposalRequest struct { ExecutionContext - RunSpec ModuleRunSpec `json:"run_spec"` - LLMClient StructuredLLMClient `json:"-"` - LLMScheduler LLMScheduler `json:"-"` + RunSpec ModuleRunSpec `json:"run_spec"` + LLMClient StructuredLLMClient `json:"-"` } type ProposalResult struct { diff --git a/internal/framework/llm/adapter_capacity_test.go b/internal/framework/llm/adapter_capacity_test.go new file mode 100644 index 0000000..4518da0 --- /dev/null +++ b/internal/framework/llm/adapter_capacity_test.go @@ -0,0 +1,214 @@ +package llm + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + ck "gitea.maximumdirect.net/eric/ConfigKit" + "gitea.maximumdirect.net/eric/audita/internal/core/config" + "gitea.maximumdirect.net/eric/audita/internal/framework/responseschema" + "gitea.maximumdirect.net/eric/audita/internal/framework/structuredoutput" + "gitea.maximumdirect.net/eric/audita/internal/prompts" + pk "gitea.maximumdirect.net/eric/promptkit" +) + +func TestRolesShareBackendCapacityAndReleaseAfterCancellation(t *testing.T) { + entered := make(chan struct{}, 3) + release := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + entered <- struct{}{} + <-release + content, _ := json.Marshal(`{"corrections":[]}`) + _, _ = fmt.Fprintf(w, `{"choices":[{"message":{"content":%s}}]}`, content) + })) + defer server.Close() + dir := t.TempDir() + if err := os.Mkdir(filepath.Join(dir, "profiles"), 0700); err != nil { + t.Fatal(err) + } + main := filepath.Join(dir, "config.yml") + body := fmt.Sprintf("version: 2\nbackends:\n shared:\n endpoint: %s/v1\n concurrency: 1\n queue_capacity: 0\nllm:\n proposal_profile: proposal\n validation_profile: validation\n transient_retries: 0\n repair_attempts: 0\n", server.URL) + if err := os.WriteFile(main, []byte(body), 0600); err != nil { + t.Fatal(err) + } + for _, name := range []string{"proposal", "validation"} { + profile := "id: " + name + "\nbackend: shared\nmodel: local\n" + if err := os.WriteFile(filepath.Join(dir, "profiles", name+".yaml"), []byte(profile), 0600); err != nil { + t.Fatal(err) + } + } + loaded, err := config.Load(ck.File(main), "") + if err != nil { + t.Fatal(err) + } + runtime, err := NewRuntime(context.Background(), loaded.Config, main, server.Client()) + if err != nil { + t.Fatal(err) + } + proposal, err := runtime.NewRoleClient("proposal", CallHooks{}) + if err != nil { + t.Fatal(err) + } + validation, err := runtime.NewRoleClient("validation", CallHooks{}) + if err != nil { + t.Fatal(err) + } + request := structuredoutput.Request{PromptID: prompts.PromptIDModuleGlossaryProposal, PromptVersion: prompts.VersionV1, Variables: map[string]string{"TranscriptDescriptionBlock": "", "GlossaryJSON": "[]", "SectionJSON": "[]"}, Schema: responseschema.CorrectionSetKey} + firstDone := make(chan error, 1) + go func() { + var output map[string]any + _, callErr := proposal.CompleteStructured(context.Background(), request, &output) + firstDone <- callErr + }() + select { + case <-entered: + case <-time.After(2 * time.Second): + t.Fatal("first call did not enter provider") + } + var output map[string]any + _, err = validation.CompleteStructured(context.Background(), request, &output) + if !errors.Is(err, pk.ErrCapacityExceeded) { + t.Fatalf("other role did not share capacity: %v", err) + } + select { + case <-entered: + t.Fatal("second role reached provider despite full backend") + default: + } + close(release) + if err := <-firstDone; err != nil { + t.Fatal(err) + } + _, err = validation.CompleteStructured(context.Background(), request, &output) + if err != nil { + t.Fatalf("capacity was not released: %v", err) + } + for len(entered) > 0 { + <-entered + } + release = make(chan struct{}) + if err := os.WriteFile(main, []byte(strings.Replace(body, "queue_capacity: 0", "queue_capacity: 1", 1)), 0600); err != nil { + t.Fatal(err) + } + loaded, err = config.Load(ck.File(main), "") + if err != nil { + t.Fatal(err) + } + runtime, err = NewRuntime(context.Background(), loaded.Config, main, server.Client()) + if err != nil { + t.Fatal(err) + } + proposal, err = runtime.NewRoleClient("proposal", CallHooks{}) + if err != nil { + t.Fatal(err) + } + validation, err = runtime.NewRoleClient("validation", CallHooks{}) + if err != nil { + t.Fatal(err) + } + go func() { + var out map[string]any + _, callErr := proposal.CompleteStructured(context.Background(), request, &out) + firstDone <- callErr + }() + select { + case <-entered: + case <-time.After(2 * time.Second): + t.Fatal("first queued-capacity call did not dispatch") + } + queuedCtx, cancelQueued := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancelQueued() + _, err = validation.CompleteStructured(queuedCtx, request, &output) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("queued call did not honor deadline: %v", err) + } + close(release) + if err := <-firstDone; err != nil { + t.Fatal(err) + } + _, err = validation.CompleteStructured(context.Background(), request, &output) + if err != nil { + t.Fatalf("queued cancellation leaked capacity: %v", err) + } +} + +func TestDistinctBackendsHaveIndependentCapacity(t *testing.T) { + entered := make(chan struct{}, 2) + release := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + entered <- struct{}{} + <-release + schema := map[string]any{} + _ = json.NewDecoder(r.Body).Decode(&schema) + content, _ := json.Marshal(`{"corrections":[]}`) + _, _ = fmt.Fprintf(w, `{"choices":[{"message":{"content":%s}}]}`, content) + })) + defer server.Close() + dir := t.TempDir() + if err := os.Mkdir(filepath.Join(dir, "profiles"), 0700); err != nil { + t.Fatal(err) + } + main := filepath.Join(dir, "config.yml") + body := fmt.Sprintf("version: 2\nbackends:\n first:\n endpoint: %s/v1\n concurrency: 1\n queue_capacity: 0\n second:\n endpoint: %s/v1\n concurrency: 1\n queue_capacity: 0\nllm:\n proposal_profile: proposal\n validation_profile: validation\n transient_retries: 0\n repair_attempts: 0\n", server.URL, server.URL) + if err := os.WriteFile(main, []byte(body), 0600); err != nil { + t.Fatal(err) + } + for name, backend := range map[string]string{"proposal": "first", "validation": "second"} { + if err := os.WriteFile(filepath.Join(dir, "profiles", name+".yaml"), []byte("id: "+name+"\nbackend: "+backend+"\nmodel: local\n"), 0600); err != nil { + t.Fatal(err) + } + } + loaded, err := config.Load(ck.File(main), "") + if err != nil { + t.Fatal(err) + } + runtime, err := NewRuntime(context.Background(), loaded.Config, main, server.Client()) + if err != nil { + t.Fatal(err) + } + proposal, err := runtime.NewRoleClient("proposal", CallHooks{}) + if err != nil { + t.Fatal(err) + } + validation, err := runtime.NewRoleClient("validation", CallHooks{}) + if err != nil { + t.Fatal(err) + } + request := structuredoutput.Request{PromptID: prompts.PromptIDModuleGlossaryProposal, PromptVersion: prompts.VersionV1, Variables: map[string]string{"TranscriptDescriptionBlock": "", "GlossaryJSON": "[]", "SectionJSON": "[]"}, Schema: responseschema.CorrectionSetKey} + done := make(chan error, 2) + go func() { + var out map[string]any + _, e := proposal.CompleteStructured(context.Background(), request, &out) + done <- e + }() + select { + case <-entered: + case <-time.After(2 * time.Second): + t.Fatal("first backend did not dispatch") + } + go func() { + var out map[string]any + _, e := validation.CompleteStructured(context.Background(), request, &out) + done <- e + }() + select { + case <-entered: + case <-time.After(2 * time.Second): + t.Fatal("second backend was blocked by first") + } + close(release) + for range 2 { + if err := <-done; err != nil { + t.Fatal(err) + } + } +} diff --git a/internal/framework/llm/adapter_test.go b/internal/framework/llm/adapter_test.go index c900033..9140546 100644 --- a/internal/framework/llm/adapter_test.go +++ b/internal/framework/llm/adapter_test.go @@ -11,6 +11,7 @@ import ( "path/filepath" "strings" "sync" + "sync/atomic" "testing" ck "gitea.maximumdirect.net/eric/ConfigKit" @@ -26,17 +27,17 @@ type reply struct { content string } -func adapterFixture(t *testing.T, replies []reply, retries, repairs int, hook func(*http.Request)) (*Runtime, *int, string) { +func adapterFixture(t *testing.T, replies []reply, retries, repairs int, hook func(*http.Request)) (*Runtime, *atomic.Int32, string) { t.Helper() - count := 0 + var count atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - count++ + call := int(count.Add(1)) if hook != nil { hook(r) } step := replies[len(replies)-1] - if count <= len(replies) { - step = replies[count-1] + if call <= len(replies) { + step = replies[call-1] } if step.status == 0 { connection, _, hijackErr := w.(http.Hijacker).Hijack() @@ -128,8 +129,8 @@ func TestPreparedAdapterRepairsAndClassifies(t *testing.T) { } var out map[string]any response, err := client.CompleteStructured(context.Background(), adapterRequest(schema), &out) - if *count != tc.attempts { - t.Fatalf("HTTP attempts=%d, want %d", *count, tc.attempts) + if int(count.Load()) != tc.attempts { + t.Fatalf("HTTP attempts=%d, want %d", count.Load(), tc.attempts) } if tc.malformed != structuredoutput.IsMalformedError(err) { t.Fatalf("malformed classification: %v", err) @@ -188,7 +189,7 @@ func TestPreparedAdapterMetadataAndTargetGuard(t *testing.T) { if err != nil { t.Fatal(err) } - if !requestSeen || *count != 1 || response.Prepared == nil || len(response.Prepared.Messages) == 0 { + if !requestSeen || count.Load() != 1 || response.Prepared == nil || len(response.Prepared.Messages) == 0 { t.Fatal("prepared call missing") } if len(posted.Messages) != len(response.Prepared.Messages) { @@ -222,8 +223,8 @@ func TestPreparedAdapterRejectsTargetDriftAndPreparationErrors(t *testing.T) { req := adapterRequest(responseschema.CorrectionSetKey) req.PromptID = "missing.prompt" var out map[string]any - if _, err := client.CompleteStructured(context.Background(), req, &out); err == nil || *count != 0 { - t.Fatalf("preparation failure dispatched: %v, %d", err, *count) + if _, err := client.CompleteStructured(context.Background(), req, &out); err == nil || count.Load() != 0 { + t.Fatalf("preparation failure dispatched: %v, %d", err, count.Load()) } // PrepareExecution reloads the profile, so a changed target must be blocked. if err := os.WriteFile(filepath.Join(dir, "profiles", "base.yaml"), []byte("id: base\nbackend: offline\nendpoint: http://127.0.0.1:1/v1\nmodel: full-model\n"), 0600); err != nil { @@ -231,8 +232,8 @@ func TestPreparedAdapterRejectsTargetDriftAndPreparationErrors(t *testing.T) { } req = adapterRequest(responseschema.CorrectionSetKey) _, err = client.CompleteStructured(context.Background(), req, &out) - if err == nil || !strings.Contains(err.Error(), "target changed") || *count != 0 { - t.Fatalf("target drift dispatched: %v, %d", err, *count) + if err == nil || !strings.Contains(err.Error(), "target changed") || count.Load() != 0 { + t.Fatalf("target drift dispatched: %v, %d", err, count.Load()) } } @@ -251,7 +252,7 @@ func TestPreparedAdapterCancellationStopsRetryWait(t *testing.T) { } var out map[string]any _, err = client.CompleteStructured(ctx, adapterRequest(responseschema.CorrectionSetKey), &out) - if !errors.Is(err, context.Canceled) || *count != 1 { - t.Fatalf("cancellation: %v, HTTP calls=%d", err, *count) + if !errors.Is(err, context.Canceled) || count.Load() != 1 { + t.Fatalf("cancellation: %v, HTTP calls=%d", err, count.Load()) } } diff --git a/internal/framework/llm/scheduler.go b/internal/framework/llm/scheduler.go deleted file mode 100644 index 7aafa01..0000000 --- a/internal/framework/llm/scheduler.go +++ /dev/null @@ -1,117 +0,0 @@ -package llm - -import ( - "context" - "fmt" - "sync" -) - -// Scheduler bounds concurrent backend LLM calls. -type Scheduler 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. -func NewScheduler(maxConcurrency int) (*Scheduler, error) { - if maxConcurrency <= 0 { - return nil, fmt.Errorf("max concurrency must be greater than zero") - } - return &Scheduler{ - 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 <-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) - if err != nil { - return err - } - defer release() - return fn(ctx) -} diff --git a/internal/framework/llm/scheduler_test.go b/internal/framework/llm/scheduler_test.go deleted file mode 100644 index 8bd40d1..0000000 --- a/internal/framework/llm/scheduler_test.go +++ /dev/null @@ -1,223 +0,0 @@ -package llm - -import ( - "context" - "errors" - "reflect" - "runtime" - "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) - } - }() - waitForQueueDepth(t, s, i+1) - } - - 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 { - t.Fatalf("NewScheduler: %v", err) - } - - var inFlight int32 - var maxInFlight int32 - release := make(chan struct{}) - var wg sync.WaitGroup - - for i := 0; i < 12; i++ { - wg.Add(1) - go func() { - defer wg.Done() - err := s.Run(context.Background(), func(ctx context.Context) error { - _ = ctx - current := atomic.AddInt32(&inFlight, 1) - for { - seen := atomic.LoadInt32(&maxInFlight) - if current <= seen || atomic.CompareAndSwapInt32(&maxInFlight, seen, current) { - break - } - } - <-release - atomic.AddInt32(&inFlight, -1) - return nil - }) - if err != nil { - t.Errorf("Run error: %v", err) - } - }() - } - - waitForMinInFlight(t, &maxInFlight, 2) - close(release) - - wg.Wait() - if got := atomic.LoadInt32(&maxInFlight); got > 2 { - t.Fatalf("expected max in-flight <= 2, got %d", got) - } -} - -func TestSchedulerReleasesPermitOnSuccess(t *testing.T) { - s, err := NewScheduler(1) - if err != nil { - t.Fatalf("NewScheduler: %v", err) - } - - for i := 0; i < 3; i++ { - if err := s.Run(context.Background(), func(ctx context.Context) error { - _ = ctx - return nil - }); err != nil { - t.Fatalf("Run[%d] error: %v", i, err) - } - } -} - -func TestSchedulerReleasesPermitOnError(t *testing.T) { - s, err := NewScheduler(1) - if err != nil { - t.Fatalf("NewScheduler: %v", err) - } - - expectedErr := errors.New("boom") - err = s.Run(context.Background(), func(ctx context.Context) error { - _ = ctx - return expectedErr - }) - if !errors.Is(err, expectedErr) { - t.Fatalf("expected %v, got %v", expectedErr, err) - } - - if err := s.Run(context.Background(), func(ctx context.Context) error { - _ = ctx - return nil - }); err != nil { - t.Fatalf("second Run error: %v", err) - } -} - -func TestSchedulerContextCancellationWhileQueued(t *testing.T) { - s, err := NewScheduler(1) - if err != nil { - t.Fatalf("NewScheduler: %v", err) - } - - release, err := s.Acquire(context.Background()) - if err != nil { - t.Fatalf("Acquire: %v", err) - } - defer release() - - ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond) - defer cancel() - - _, err = s.Acquire(ctx) - if err == nil { - t.Fatalf("expected context cancellation error") - } - if !errors.Is(err, context.DeadlineExceeded) { - 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) - } -} - -func waitForQueueDepth(t *testing.T, s *Scheduler, want int) { - t.Helper() - deadline := time.Now().Add(250 * time.Millisecond) - for time.Now().Before(deadline) { - s.mu.Lock() - depth := len(s.queue) - s.mu.Unlock() - if depth >= want { - return - } - runtime.Gosched() - } - s.mu.Lock() - depth := len(s.queue) - s.mu.Unlock() - t.Fatalf("timed out waiting for queue depth >= %d (got %d)", want, depth) -} - -func waitForMinInFlight(t *testing.T, maxInFlight *int32, want int32) { - t.Helper() - deadline := time.Now().Add(250 * time.Millisecond) - for time.Now().Before(deadline) { - if atomic.LoadInt32(maxInFlight) >= want { - return - } - runtime.Gosched() - } - t.Fatalf("timed out waiting for max in-flight >= %d (got %d)", want, atomic.LoadInt32(maxInFlight)) -} diff --git a/internal/framework/modules/registry.go b/internal/framework/modules/registry.go index 7eeae8d..1179812 100644 --- a/internal/framework/modules/registry.go +++ b/internal/framework/modules/registry.go @@ -34,13 +34,11 @@ func IsKnownModuleKey(key string) bool { // Dependencies holds explicit constructor dependencies for module creation. type Dependencies struct { - Config *config.Config - Glossary *schema.Glossary - ProposalLLMClient contracts.StructuredLLMClient - ProposalLLMScheduler contracts.LLMScheduler - ValidationLLMClient contracts.StructuredLLMClient - ValidationLLMScheduler contracts.LLMScheduler - DiagnosticsDir string + Config *config.Config + Glossary *schema.Glossary + ProposalLLMClient contracts.StructuredLLMClient + ValidationLLMClient contracts.StructuredLLMClient + DiagnosticsDir string } // ConstructRequest is one module-construction request. @@ -130,13 +128,11 @@ func (f *Factory) ModuleForSpec(spec contracts.ModuleRunSpec) (contracts.Transcr module, err := constructor(context.Background(), ConstructRequest{ RunSpec: spec, Dependencies: Dependencies{ - Config: f.deps.Config, - Glossary: f.deps.Glossary, - ProposalLLMClient: f.deps.ProposalLLMClient, - ProposalLLMScheduler: f.deps.ProposalLLMScheduler, - ValidationLLMClient: f.deps.ValidationLLMClient, - ValidationLLMScheduler: f.deps.ValidationLLMScheduler, - DiagnosticsDir: f.deps.DiagnosticsDir, + Config: f.deps.Config, + Glossary: f.deps.Glossary, + ProposalLLMClient: f.deps.ProposalLLMClient, + ValidationLLMClient: f.deps.ValidationLLMClient, + DiagnosticsDir: f.deps.DiagnosticsDir, }, }) if err != nil { diff --git a/internal/framework/proposal_generation/generate.go b/internal/framework/proposal_generation/generate.go index 758398c..588ea07 100644 --- a/internal/framework/proposal_generation/generate.go +++ b/internal/framework/proposal_generation/generate.go @@ -62,7 +62,6 @@ type Request struct { StageName string `json:"stage_name,omitempty"` StartIndex int `json:"start_index"` LLMClient contracts.StructuredLLMClient - Scheduler contracts.LLMScheduler DiagnosticsDir string DiagnosticsWriter InteractionDiagnosticsWriter } @@ -117,19 +116,11 @@ func GenerateCandidates(ctx context.Context, req Request) (Result, error) { callErr error artifacts InteractionArtifacts ) - call := func(callCtx context.Context) error { - completion, callErr = req.LLMClient.CompleteStructured(callCtx, contracts.StructuredCompletionRequest{ - PromptID: req.PromptID, PromptVersion: req.PromptVersion, - Variables: req.Variables, Schema: responseschema.CorrectionSetKey, - Context: structuredoutput.DiagnosticContext{Role: "proposal", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: sectionIndexPtr(req.Section)}, - }, &response) - return callErr - } - if req.Scheduler != nil { - callErr = req.Scheduler.Run(ctx, call) - } else { - callErr = call(ctx) - } + completion, callErr = req.LLMClient.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ + PromptID: req.PromptID, PromptVersion: req.PromptVersion, + Variables: req.Variables, Schema: responseschema.CorrectionSetKey, + Context: structuredoutput.DiagnosticContext{Role: "proposal", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: sectionIndexPtr(req.Section)}, + }, &response) requestMetadata := map[string]any{ "module_key": req.ModuleKey, diff --git a/internal/framework/proposal_generation/generate_test.go b/internal/framework/proposal_generation/generate_test.go index d449693..dd18ea3 100644 --- a/internal/framework/proposal_generation/generate_test.go +++ b/internal/framework/proposal_generation/generate_test.go @@ -7,17 +7,12 @@ import ( "os" "path/filepath" "reflect" - "runtime" "strings" - "sync" - "sync/atomic" "testing" - "time" ck "gitea.maximumdirect.net/eric/ConfigKit" "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" "gitea.maximumdirect.net/eric/audita/internal/framework/responseschema" "gitea.maximumdirect.net/eric/audita/internal/framework/structuredoutput" @@ -51,15 +46,6 @@ func (f *fakeStructuredClient) CompleteStructured(ctx context.Context, req contr return contracts.StructuredCompletionResponse{Prepared: &structuredoutput.PreparedSnapshot{Messages: []structuredoutput.Message{{Role: "user", Content: content.String()}}}}, nil } -type countingScheduler struct { - runs int -} - -func (s *countingScheduler) Run(ctx context.Context, fn func(context.Context) error) error { - s.runs++ - return fn(ctx) -} - type captureDiagnosticsWriter struct { lastStage string lastRequestMetadata any @@ -75,45 +61,6 @@ func (w *captureDiagnosticsWriter) WriteInteraction(stage string, requestMetadat return InteractionArtifacts{}, nil } -type sleepingStructuredClient struct { - inFlight int32 - maxInFlight int32 - entered chan struct{} - release chan struct{} -} - -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 - } - } - if c.entered != nil { - c.entered <- struct{}{} - } - select { - case <-c.release: - 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.Config{} @@ -401,26 +348,6 @@ func TestGenerateCandidatesDiagnosticsWrittenAndRedacted(t *testing.T) { } } -func TestGenerateCandidatesSchedulerUsage(t *testing.T) { - client := &fakeStructuredClient{ - responses: []StructuredCorrectionSet{ - {Corrections: []StructuredCorrectionProposal{{TargetSegmentID: 1, OriginalText: "x", CorrectedText: "y", Confidence: 0.9}}}, - }, - } - scheduler := &countingScheduler{} - req := defaultRequest(t) - req.LLMClient = client - req.Scheduler = scheduler - - _, err := GenerateCandidates(context.Background(), req) - if err != nil { - t.Fatalf("GenerateCandidates error: %v", err) - } - if scheduler.runs != 1 { - t.Fatalf("expected scheduler to run once, got %d", scheduler.runs) - } -} - func TestGenerateCandidatesClientError(t *testing.T) { client := &fakeStructuredClient{err: errors.New("boom")} req := defaultRequest(t) @@ -442,58 +369,3 @@ 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{ - entered: make(chan struct{}, 16), - release: make(chan struct{}), - } - 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) - } - - waitForEntries(t, client.entered, 2) - close(client.release) - 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) - } -} - -func waitForEntries(t *testing.T, entered <-chan struct{}, want int) { - t.Helper() - deadline := time.Now().Add(300 * time.Millisecond) - got := 0 - for got < want && time.Now().Before(deadline) { - select { - case <-entered: - got++ - default: - runtime.Gosched() - } - } - if got < want { - t.Fatalf("timed out waiting for %d entered calls (got %d)", want, got) - } -} diff --git a/internal/framework/proposal_generation/module_proposal.go b/internal/framework/proposal_generation/module_proposal.go index 37fd5df..8afa8c1 100644 --- a/internal/framework/proposal_generation/module_proposal.go +++ b/internal/framework/proposal_generation/module_proposal.go @@ -68,7 +68,6 @@ func ExecuteModuleProposal(ctx context.Context, req ModuleProposalRequest) (cont StageName: stagename.ModuleProposal(req.ProposalRequest.RunSpec.InstanceName, sectionIndexPtr(req.ProposalRequest.Section)), StartIndex: 0, LLMClient: req.ProposalRequest.LLMClient, - Scheduler: req.ProposalRequest.LLMScheduler, DiagnosticsDir: req.ProposalRequest.DiagnosticsDir, }) if err != nil { diff --git a/internal/framework/runner/observability.go b/internal/framework/runner/observability.go deleted file mode 100644 index b7eaa19..0000000 --- a/internal/framework/runner/observability.go +++ /dev/null @@ -1,350 +0,0 @@ -package runner - -import ( - "context" - "sort" - "strings" - "sync" - "time" - - "gitea.maximumdirect.net/eric/audita/internal/framework/contracts" - validatormetadata "gitea.maximumdirect.net/eric/audita/internal/validators/metadata" -) - -type UtilizationDiagnostics struct { - EffectiveConcurrency EffectiveConcurrencyLimits `json:"effective_concurrency"` - RunTiming RunTimingSummary `json:"run_timing"` - LLMCalls LLMCallSummary `json:"llm_calls"` - Modules []ModuleTimingSummary `json:"modules,omitempty"` - Validators []ValidatorTimingSummary `json:"validators,omitempty"` -} - -type EffectiveConcurrencyLimits struct { - TotalLLM int `json:"total_llm"` - ProposalLLM int `json:"proposal_llm"` - ValidationLLM int `json:"validation_llm"` -} - -type RunTimingSummary struct { - RunWallTimeMS int64 `json:"run_wall_time_ms"` - SchedulerQueueWaitMS int64 `json:"scheduler_queue_wait_ms"` - LLMExecutionTimeMS int64 `json:"llm_execution_time_ms"` - DeterministicValidationMS int64 `json:"deterministic_validation_time_ms"` - MaxInFlightLLMCalls int `json:"max_in_flight_llm_calls"` - AverageInFlightLLMCalls int `json:"average_in_flight_llm_calls"` -} - -type LLMCallSummary struct { - TotalProposalCalls int `json:"total_proposal_calls"` - TotalValidationCalls int `json:"total_validation_calls"` -} - -type ModuleTimingSummary struct { - ModuleKey string `json:"module_key"` - ModuleInstance string `json:"module_instance"` - ProposalLLMCalls int `json:"proposal_llm_calls"` - ValidationLLMCalls int `json:"validation_llm_calls"` - SchedulerQueueWaitMS int64 `json:"scheduler_queue_wait_ms"` - LLMExecutionTimeMS int64 `json:"llm_execution_time_ms"` - DeterministicValidationMS int64 `json:"deterministic_validation_time_ms"` - ModuleWallTimeMS int64 `json:"module_wall_time_ms"` -} - -type ValidatorTimingSummary struct { - ModuleKey string `json:"module_key"` - ModuleInstance string `json:"module_instance"` - ValidatorKey string `json:"validator_key"` - LLMBacked bool `json:"llm_backed"` - Calls int `json:"calls"` - ElapsedMS int64 `json:"elapsed_ms"` -} - -type runInstrumentation struct { - startedAt time.Time - - mu sync.Mutex - totalProposalCalls int - totalValidationCalls int - schedulerQueueWait time.Duration - llmExecutionTime time.Duration - deterministicValidation time.Duration - perModuleProposalCalls map[string]int - perModuleValidationCalls map[string]int - perModuleSchedulerQueueWait map[string]time.Duration - perModuleLLMExecutionTime map[string]time.Duration - perModuleDeterministicValidation map[string]time.Duration - validatorSummaries map[string]*ValidatorTimingSummary - - inflight int - maxInflight int - inflightArea float64 - lastInflightChange time.Time - averageInflightComputed bool -} - -func newRunInstrumentation(startedAt time.Time) *runInstrumentation { - return &runInstrumentation{ - startedAt: startedAt, - perModuleProposalCalls: make(map[string]int), - perModuleValidationCalls: make(map[string]int), - perModuleSchedulerQueueWait: make(map[string]time.Duration), - perModuleLLMExecutionTime: make(map[string]time.Duration), - perModuleDeterministicValidation: make(map[string]time.Duration), - validatorSummaries: make(map[string]*ValidatorTimingSummary), - lastInflightChange: startedAt, - } -} - -func (i *runInstrumentation) wrapProposalClient(client contracts.StructuredLLMClient) contracts.StructuredLLMClient { - if client == nil { - return nil - } - return measuredClient{ - inner: client, - collector: i, - isProposal: true, - } -} - -func (i *runInstrumentation) wrapValidationClient(client contracts.StructuredLLMClient) contracts.StructuredLLMClient { - if client == nil { - return nil - } - return measuredClient{ - inner: client, - collector: i, - isProposal: false, - } -} - -func (i *runInstrumentation) wrapScheduler(s contracts.LLMScheduler, isProposal bool) contracts.LLMScheduler { - if s == nil { - return nil - } - return measuredScheduler{ - inner: s, - collector: i, - isProposal: isProposal, - } -} - -func (i *runInstrumentation) recordValidatorTiming(moduleKey, moduleInstance, validatorKey string, llmBacked bool, elapsed time.Duration) { - i.mu.Lock() - defer i.mu.Unlock() - if !llmBacked { - i.deterministicValidation += elapsed - i.perModuleDeterministicValidation[moduleInstance] += elapsed - } - k := moduleInstance + "::" + validatorKey - summary, ok := i.validatorSummaries[k] - if !ok { - summary = &ValidatorTimingSummary{ - ModuleKey: moduleKey, - ModuleInstance: moduleInstance, - ValidatorKey: validatorKey, - LLMBacked: llmBacked, - } - i.validatorSummaries[k] = summary - } - summary.Calls++ - summary.ElapsedMS += elapsed.Milliseconds() -} - -func (i *runInstrumentation) recordModuleWallTime(moduleInstance string, elapsed time.Duration) { - i.mu.Lock() - defer i.mu.Unlock() - // Derived later from started/completed, but we keep a slot for sanity. - _ = moduleInstance - _ = elapsed -} - -func (i *runInstrumentation) beginInFlight() { - now := time.Now().UTC() - i.mu.Lock() - defer i.mu.Unlock() - i.inflightArea += float64(i.inflight) * now.Sub(i.lastInflightChange).Seconds() - i.lastInflightChange = now - i.inflight++ - if i.inflight > i.maxInflight { - i.maxInflight = i.inflight - } -} - -func (i *runInstrumentation) endInFlight() { - now := time.Now().UTC() - i.mu.Lock() - defer i.mu.Unlock() - i.inflightArea += float64(i.inflight) * now.Sub(i.lastInflightChange).Seconds() - i.lastInflightChange = now - if i.inflight > 0 { - i.inflight-- - } -} - -func (i *runInstrumentation) finalize(runOutput *RunOutput, completedAt time.Time, limits EffectiveConcurrencyLimits) *UtilizationDiagnostics { - i.mu.Lock() - defer i.mu.Unlock() - - if !i.averageInflightComputed { - i.inflightArea += float64(i.inflight) * completedAt.Sub(i.lastInflightChange).Seconds() - i.lastInflightChange = completedAt - i.averageInflightComputed = true - } - - runSeconds := completedAt.Sub(i.startedAt).Seconds() - avgInflight := 0 - if runSeconds > 0 { - avgInflight = int(i.inflightArea / runSeconds) - } - - moduleSummaries := make([]ModuleTimingSummary, 0) - if runOutput != nil { - moduleSummaries = make([]ModuleTimingSummary, 0, len(runOutput.ModuleResults)) - for _, moduleResult := range runOutput.ModuleResults { - moduleSummaries = append(moduleSummaries, ModuleTimingSummary{ - ModuleKey: moduleResult.ModuleKey, - ModuleInstance: moduleResult.ModuleInstance, - ProposalLLMCalls: i.perModuleProposalCalls[moduleResult.ModuleInstance], - ValidationLLMCalls: i.perModuleValidationCalls[moduleResult.ModuleInstance], - SchedulerQueueWaitMS: i.perModuleSchedulerQueueWait[moduleResult.ModuleInstance].Milliseconds(), - LLMExecutionTimeMS: i.perModuleLLMExecutionTime[moduleResult.ModuleInstance].Milliseconds(), - DeterministicValidationMS: i.perModuleDeterministicValidation[moduleResult.ModuleInstance].Milliseconds(), - ModuleWallTimeMS: moduleResult.CompletedAt.Sub(moduleResult.StartedAt).Milliseconds(), - }) - } - } - - validatorSummaries := make([]ValidatorTimingSummary, 0, len(i.validatorSummaries)) - for _, summary := range i.validatorSummaries { - validatorSummaries = append(validatorSummaries, *summary) - } - sortValidatorSummaries(validatorSummaries) - - return &UtilizationDiagnostics{ - EffectiveConcurrency: limits, - RunTiming: RunTimingSummary{ - RunWallTimeMS: completedAt.Sub(i.startedAt).Milliseconds(), - SchedulerQueueWaitMS: i.schedulerQueueWait.Milliseconds(), - LLMExecutionTimeMS: i.llmExecutionTime.Milliseconds(), - DeterministicValidationMS: i.deterministicValidation.Milliseconds(), - MaxInFlightLLMCalls: i.maxInflight, - AverageInFlightLLMCalls: avgInflight, - }, - LLMCalls: LLMCallSummary{ - TotalProposalCalls: i.totalProposalCalls, - TotalValidationCalls: i.totalValidationCalls, - }, - Modules: moduleSummaries, - Validators: validatorSummaries, - } -} - -type measuredClient struct { - inner contracts.StructuredLLMClient - collector *runInstrumentation - isProposal bool -} - -func (c measuredClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { - if c.collector != nil { - c.collector.beginInFlight() - } - started := time.Now().UTC() - resp, err := c.inner.CompleteStructured(ctx, req, out) - elapsed := time.Since(started) - if c.collector != nil { - c.collector.endInFlight() - c.collector.mu.Lock() - moduleInstance := moduleInstanceFromStage(req.StageName) - c.collector.llmExecutionTime += elapsed - c.collector.perModuleLLMExecutionTime[moduleInstance] += elapsed - if c.isProposal { - c.collector.totalProposalCalls++ - c.collector.perModuleProposalCalls[moduleInstance]++ - } else { - c.collector.totalValidationCalls++ - c.collector.perModuleValidationCalls[moduleInstance]++ - } - c.collector.mu.Unlock() - } - return resp, err -} - -type measuredScheduler struct { - inner contracts.LLMScheduler - collector *runInstrumentation - isProposal bool -} - -func (s measuredScheduler) Run(ctx context.Context, fn func(context.Context) error) error { - started := time.Now().UTC() - var execDuration time.Duration - err := s.inner.Run(ctx, func(runCtx context.Context) error { - execStart := time.Now().UTC() - callErr := fn(runCtx) - execDuration += time.Since(execStart) - return callErr - }) - totalDuration := time.Since(started) - waitDuration := totalDuration - execDuration - if waitDuration < 0 { - waitDuration = 0 - } - if s.collector != nil { - stageModule := moduleInstanceFromContext(ctx) - s.collector.mu.Lock() - s.collector.schedulerQueueWait += waitDuration - if stageModule != "" { - s.collector.perModuleSchedulerQueueWait[stageModule] += waitDuration - } - s.collector.mu.Unlock() - } - _ = s.isProposal - return err -} - -func moduleInstanceFromStage(stage string) string { - stage = strings.TrimSpace(stage) - if stage == "" { - return "" - } - parts := strings.Split(stage, ":") - if len(parts) == 0 { - return stage - } - return strings.TrimSpace(parts[0]) -} - -func moduleInstanceFromContext(ctx context.Context) string { - if ctx == nil { - return "" - } - if v := ctx.Value(moduleInstanceContextKey{}); v != nil { - if s, ok := v.(string); ok { - return strings.TrimSpace(s) - } - } - return "" -} - -type moduleInstanceContextKey struct{} - -func withModuleInstanceContext(ctx context.Context, moduleInstance string) context.Context { - if ctx == nil { - ctx = context.Background() - } - return context.WithValue(ctx, moduleInstanceContextKey{}, strings.TrimSpace(moduleInstance)) -} - -func isLLMBackedValidator(v contracts.Validator) bool { - return validatormetadata.ClassOf(v) == validatormetadata.ExecutionClassLLMBacked -} - -func sortValidatorSummaries(in []ValidatorTimingSummary) { - sort.SliceStable(in, func(i, j int) bool { - if in[i].ModuleInstance != in[j].ModuleInstance { - return in[i].ModuleInstance < in[j].ModuleInstance - } - return in[i].ValidatorKey < in[j].ValidatorKey - }) -} diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go index 0ab6dbf..d3719c1 100644 --- a/internal/framework/runner/runner.go +++ b/internal/framework/runner/runner.go @@ -34,8 +34,6 @@ type Runner struct { factory ModuleFactory } -type ValidationScheduler = contracts.LLMScheduler - // ModuleResult captures deterministic per-module execution output. type ModuleResult struct { ModuleKey string `json:"module_key"` @@ -80,12 +78,9 @@ type RunInput struct { Transcript *schema.Transcript Glossary *schema.Glossary ModuleSpecs []contracts.ModuleRunSpec - EffectiveConcurrency EffectiveConcurrencyLimits ProposalLLMClient contracts.StructuredLLMClient - ProposalLLMScheduler contracts.LLMScheduler ProposalDiagnosticsDir string ValidationLLMClient contracts.StructuredLLMClient - ValidationLLMScheduler ValidationScheduler ValidationDiagnosticsDir string } @@ -93,7 +88,6 @@ type RunInput struct { type RunOutput struct { FinalTranscript *schema.Transcript `json:"-"` ModuleResults []ModuleResult `json:"module_results"` - Utilization *UtilizationDiagnostics } func New(factory ModuleFactory) *Runner { @@ -105,14 +99,6 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { return RunOutput{}, fmt.Errorf("runner module factory is required") } - startedAt := time.Now().UTC() - instrumentation := newRunInstrumentation(startedAt) - - input.ProposalLLMClient = instrumentation.wrapProposalClient(input.ProposalLLMClient) - input.ValidationLLMClient = instrumentation.wrapValidationClient(input.ValidationLLMClient) - input.ProposalLLMScheduler = instrumentation.wrapScheduler(input.ProposalLLMScheduler, true) - input.ValidationLLMScheduler = instrumentation.wrapScheduler(input.ValidationLLMScheduler, false) - working := cloneTranscript(input.Transcript) results := make([]ModuleResult, 0, len(input.ModuleSpecs)) @@ -132,7 +118,6 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { return RunOutput{ FinalTranscript: working, ModuleResults: results, - Utilization: instrumentation.finalize(&RunOutput{ModuleResults: results}, time.Now().UTC(), input.EffectiveConcurrency), }, fmt.Errorf("module %q setup failed: %w", spec.InstanceName, err) } @@ -152,7 +137,6 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { return RunOutput{ FinalTranscript: working, ModuleResults: results, - Utilization: instrumentation.finalize(&RunOutput{ModuleResults: results}, time.Now().UTC(), input.EffectiveConcurrency), }, fmt.Errorf("module %q chunking failed: %w", spec.InstanceName, err) } @@ -164,12 +148,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { Glossary: input.Glossary, Sections: sections, ProposalClient: input.ProposalLLMClient, - ProposalScheduler: input.ProposalLLMScheduler, DiagnosticsDir: input.ProposalDiagnosticsDir, ValidationClient: input.ValidationLLMClient, - ValidationScheduler: input.ValidationLLMScheduler, ValidationDiagnosticsDir: input.ValidationDiagnosticsDir, - Instrumentation: instrumentation, }) if pipelineErr != nil { failed := ModuleResult{ @@ -189,7 +170,6 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { return RunOutput{ FinalTranscript: working, ModuleResults: results, - Utilization: instrumentation.finalize(&RunOutput{ModuleResults: results}, time.Now().UTC(), input.EffectiveConcurrency), }, fmt.Errorf("module %q failed: %w", spec.InstanceName, pipelineErr) } @@ -213,7 +193,6 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) { } output := RunOutput{FinalTranscript: working, ModuleResults: results} - output.Utilization = instrumentation.finalize(&output, time.Now().UTC(), input.EffectiveConcurrency) return output, nil } @@ -225,12 +204,9 @@ type collectSectionProposalsInput struct { Glossary *schema.Glossary Sections []chunking.Section ProposalClient contracts.StructuredLLMClient - ProposalScheduler contracts.LLMScheduler DiagnosticsDir string ValidationClient contracts.StructuredLLMClient - ValidationScheduler ValidationScheduler ValidationDiagnosticsDir string - Instrumentation *runInstrumentation } type sectionProposals struct { @@ -279,7 +255,7 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI go func() { defer wg.Done() meta := contracts.SectionMetadataFromSection(section) - proposalResult, err := input.Module.Propose(withModuleInstanceContext(runCtx, input.Spec.InstanceName), contracts.ProposalRequest{ + proposalResult, err := input.Module.Propose(runCtx, contracts.ProposalRequest{ ExecutionContext: contracts.ExecutionContext{ Config: input.Config, WorkingTranscript: transcriptFromSection(section), @@ -292,8 +268,7 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI InstanceName: input.Spec.InstanceName, ReplacementPolicy: input.Policy, }, - LLMClient: input.ProposalClient, - LLMScheduler: input.ProposalScheduler, + LLMClient: input.ProposalClient, }) select { case results <- sectionProposalResult{ @@ -402,9 +377,7 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) Validators: validatorsOrdered, SectionEnriched: enriched, ValidationLLMClient: input.ValidationClient, - ValidationScheduler: input.ValidationScheduler, DiagnosticsDir: input.ValidationDiagnosticsDir, - Instrumentation: input.Instrumentation, }) validationResults <- sectionValidationResult{ sectionPos: sectionPos, @@ -503,9 +476,7 @@ type validateSectionCandidatesInput struct { Validators []contracts.Validator SectionEnriched []proposals.EnrichedCorrectionProposal ValidationLLMClient contracts.StructuredLLMClient - ValidationScheduler ValidationScheduler DiagnosticsDir string - Instrumentation *runInstrumentation } type validateSectionCandidatesResult struct { @@ -532,8 +503,7 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida } } - validatorStartedAt := time.Now().UTC() - vResult, err := validator.Validate(withModuleInstanceContext(ctx, input.Spec.InstanceName), contracts.ValidationRequest{ + vResult, err := validator.Validate(ctx, contracts.ValidationRequest{ WorkingTranscript: input.WorkingTranscript, CandidateProposal: eligible, ModuleKey: input.Spec.ModuleKey, @@ -541,19 +511,10 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida ReplacementPolicy: input.Policy, Glossary: input.Glossary, Config: input.Config, - LLMClient: validationLLMClientAdapter{client: input.ValidationLLMClient}, - Scheduler: input.ValidationScheduler, + LLMClient: input.ValidationLLMClient, DiagnosticsWriter: diagnosticsWriter, }) - if input.Instrumentation != nil { - input.Instrumentation.recordValidatorTiming( - input.Spec.ModuleKey, - input.Spec.InstanceName, - validator.Name(), - isLLMBackedValidator(validator), - time.Since(validatorStartedAt), - ) - } + if err != nil { return validateSectionCandidatesResult{ approved: eligible, @@ -650,9 +611,9 @@ func chunkWorkingTranscript(cfg *config.Config, transcript *schema.Transcript) ( }, nil } chunker := chunking.NewChunker(chunking.ChunkingConfig{ - MaxSectionTokens: cfg.MaxSectionTokens, - MinSectionTokens: cfg.MinSectionTokens, - TargetSections: cfg.TargetSections, + MaxSectionTokens: cfg.Chunking.MaxSectionTokens, + MinSectionTokens: cfg.Chunking.MinSectionTokens, + TargetSections: cfg.Chunking.TargetSections, }) return chunker.ChunkTranscript(transcript) } @@ -697,34 +658,6 @@ func cloneTranscript(t *schema.Transcript) *schema.Transcript { return &schema.Transcript{Segments: segments} } -type validationLLMClientAdapter struct { - client contracts.StructuredLLMClient -} - -func (a validationLLMClientAdapter) CompleteStructured(ctx context.Context, req validators.StructuredCompletionRequest, out any) (validators.StructuredCompletionResponse, error) { - if a.client == nil { - return validators.StructuredCompletionResponse{}, fmt.Errorf("validation structured LLM client is not configured") - } - messages := make([]contracts.LLMMessage, len(req.Messages)) - for i, m := range req.Messages { - messages[i] = contracts.LLMMessage{Role: m.Role, Content: m.Content} - } - resp, err := a.client.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ - StageName: req.StageName, - Messages: messages, - Model: req.Model, - ResponseSchema: req.ResponseSchema, - }, out) - if err != nil { - return validators.StructuredCompletionResponse{}, err - } - return validators.StructuredCompletionResponse{ - PromptTokens: resp.PromptTokens, - CompletionTokens: resp.CompletionTokens, - TotalTokens: resp.TotalTokens, - }, nil -} - type llmDiagnosticsWriterAdapter struct { writer *llm.DiagnosticsWriter } diff --git a/internal/framework/runner/runner_test.go b/internal/framework/runner/runner_test.go index 559bde5..3ace1d2 100644 --- a/internal/framework/runner/runner_test.go +++ b/internal/framework/runner/runner_test.go @@ -5,26 +5,27 @@ import ( "errors" "os" "path/filepath" - "regexp" "strconv" "strings" "sync" - "sync/atomic" "testing" "time" + ck "gitea.maximumdirect.net/eric/ConfigKit" "gitea.maximumdirect.net/eric/audita/internal/core/config" "gitea.maximumdirect.net/eric/audita/internal/core/schema" "gitea.maximumdirect.net/eric/audita/internal/framework/contracts" - "gitea.maximumdirect.net/eric/audita/internal/framework/llm" "gitea.maximumdirect.net/eric/audita/internal/framework/modules" "gitea.maximumdirect.net/eric/audita/internal/framework/proposal_generation" "gitea.maximumdirect.net/eric/audita/internal/framework/proposals" - "gitea.maximumdirect.net/eric/audita/internal/framework/responseschema" "gitea.maximumdirect.net/eric/audita/internal/framework/validators" validatormetadata "gitea.maximumdirect.net/eric/audita/internal/validators/metadata" ) +func runnerTestConfig() config.Config { + return config.Config{Chunking: config.Chunking{MaxSectionTokens: 8192, MinSectionTokens: 2048}, Validation: config.Validation{MaxPromptTokens: 2048}} +} + type fakeFactory struct { modules map[string]contracts.TranscriptModule } @@ -162,9 +163,9 @@ func TestRunnerModulesRunSequentially(t *testing.T) { } func TestRunnerProposalsExecutePerChunkSection(t *testing.T) { - cfg := config.Default() - cfg.MaxSectionTokens = 3 - cfg.MinSectionTokens = 0 + cfg := runnerTestConfig() + cfg.Chunking.MaxSectionTokens = 3 + cfg.Chunking.MinSectionTokens = 0 transcript := &schema.Transcript{Segments: []schema.Segment{ {ID: 1, Text: "one two"}, @@ -220,84 +221,10 @@ func TestRunnerProposalsExecutePerChunkSection(t *testing.T) { } } -func TestRunnerProposalSectionConcurrencyBoundedByProposalLLMConcurrency(t *testing.T) { - cfg := config.Default() - cfg.MaxSectionTokens = 3 - cfg.MinSectionTokens = 0 - cfg.TotalLLMConcurrency = 2 - cfg.ProposalLLMConcurrency = 2 - - transcript := &schema.Transcript{Segments: []schema.Segment{ - {ID: 1, Text: "one two"}, - {ID: 2, Text: "three four"}, - {ID: 3, Text: "five six"}, - {ID: 4, Text: "seven eight"}, - }} - - var inFlight int32 - var maxInFlight int32 - entered := make(chan struct{}, len(transcript.Segments)) - release := make(chan struct{}) - scheduler, err := llm.NewScheduler(2) - if err != nil { - t.Fatalf("NewScheduler: %v", err) - } - r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ - "m": fakeModule{key: "m", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { - err := req.LLMScheduler.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 - } - } - entered <- struct{}{} - <-release - atomic.AddInt32(&inFlight, -1) - return nil - }) - if err != nil { - return nil, err - } - return nil, nil - }}, - }}) - - resultCh := make(chan error, 1) - go func() { - _, err := r.Run(context.Background(), RunInput{ - Config: &cfg, - Transcript: transcript, - ProposalLLMScheduler: scheduler, - ModuleSpecs: []contracts.ModuleRunSpec{ - {ModuleKey: "m", InstanceName: "m"}, - }, - }) - resultCh <- err - }() - - waitForRunnerEntries(t, entered, 2, "proposal workers to enter") - close(release) - - err = <-resultCh - if err != nil { - t.Fatalf("Run error: %v", err) - } - if maxInFlight < 2 { - t.Fatalf("expected proposal execution to run concurrently, got max in-flight %d", maxInFlight) - } - if maxInFlight > 2 { - t.Fatalf("expected proposal concurrency <= 2, got %d", maxInFlight) - } -} - func TestRunnerProposalIndexOrderingIsDeterministicAcrossParallelSections(t *testing.T) { - cfg := config.Default() - cfg.MaxSectionTokens = 3 - cfg.MinSectionTokens = 0 - cfg.TotalLLMConcurrency = 3 - cfg.ProposalLLMConcurrency = 3 + cfg := runnerTestConfig() + cfg.Chunking.MaxSectionTokens = 3 + cfg.Chunking.MinSectionTokens = 0 transcript := &schema.Transcript{Segments: []schema.Segment{ {ID: 1, Text: "alpha one"}, @@ -411,11 +338,9 @@ func mapKeys(values map[int]struct{}) []int { } func TestRunnerValidationStartsBeforeAllSectionProposalsComplete(t *testing.T) { - cfg := config.Default() - cfg.MaxSectionTokens = 3 - cfg.MinSectionTokens = 0 - cfg.TotalLLMConcurrency = 2 - cfg.ProposalLLMConcurrency = 2 + cfg := runnerTestConfig() + cfg.Chunking.MaxSectionTokens = 3 + cfg.Chunking.MinSectionTokens = 0 transcript := &schema.Transcript{Segments: []schema.Segment{ {ID: 1, Text: "alpha one"}, @@ -488,298 +413,6 @@ func TestRunnerValidationStartsBeforeAllSectionProposalsComplete(t *testing.T) { } } -func TestRunnerProposalAndValidationLLMCanOverlapWithinModule(t *testing.T) { - cfg := config.Default() - cfg.MaxSectionTokens = 3 - cfg.MinSectionTokens = 0 - cfg.TotalLLMConcurrency = 2 - cfg.ProposalLLMConcurrency = 2 - validationCap := 2 - cfg.ValidationLLMConcurrency = &validationCap - - transcript := &schema.Transcript{Segments: []schema.Segment{ - {ID: 1, Text: "alpha one"}, - {ID: 2, Text: "bravo two"}, - }} - - global, err := llm.NewScheduler(2) - if err != nil { - t.Fatalf("NewScheduler: %v", err) - } - scheduler := &trackingScheduler{inner: global} - - proposalSectionOneStarted := make(chan struct{}, 1) - releaseProposalSectionOne := make(chan struct{}) - - releaseValidation := make(chan struct{}) - client := &stageAwareStructuredClient{ - startedSection: make(chan int, 8), - releaseValidation: releaseValidation, - } - llmValidator, err := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility, "") - if err != nil { - t.Fatalf("NewLLMBackedValidator: %v", err) - } - - r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ - "m": fakeModule{ - key: "m", - policy: proposals.ReplacementPolicyRequireUnique, - validators: []contracts.Validator{llmValidator}, - proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { - seg := req.WorkingTranscript.Segments[0] - err := req.LLMScheduler.Run(context.Background(), func(context.Context) error { - if req.Section != nil && req.Section.Index == 1 { - select { - case proposalSectionOneStarted <- struct{}{}: - default: - } - <-releaseProposalSectionOne - } - return nil - }) - if err != nil { - return nil, err - } - return []proposals.CorrectionProposal{ - {TargetSegmentID: seg.ID, OriginalText: seg.Text, CorrectedText: strings.ToUpper(seg.Text), Confidence: 1}, - }, nil - }, - }, - }}) - - resultCh := make(chan struct { - out RunOutput - err error - }, 1) - go func() { - out, err := r.Run(context.Background(), RunInput{ - Config: &cfg, - Transcript: transcript, - ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}, - ProposalLLMScheduler: scheduler, - ValidationLLMScheduler: scheduler, - ValidationLLMClient: client, - }) - resultCh <- struct { - out RunOutput - err error - }{out: out, err: err} - }() - - select { - case <-proposalSectionOneStarted: - case <-time.After(350 * time.Millisecond): - t.Fatal("expected section-1 proposal to start") - } - - select { - case section := <-client.startedSection: - if section != 0 { - t.Fatalf("expected first validation to start for section 0, got section %d", section) - } - case <-time.After(350 * time.Millisecond): - t.Fatal("expected validation call for section 0 while later section proposal still running") - } - - close(releaseProposalSectionOne) - close(releaseValidation) - - result := <-resultCh - if result.err != nil { - t.Fatalf("Run error: %v", result.err) - } - if got := atomic.LoadInt32(&scheduler.maxInFlight); got > 2 { - t.Fatalf("expected combined proposal+validation in-flight <= 2, got %d", got) - } - if got := atomic.LoadInt32(&scheduler.maxInFlight); got < 2 { - t.Fatalf("expected observed overlap/in-flight utilization of at least 2, got %d", got) - } -} - -func TestRunnerValidationLLMConcurrencyRespected(t *testing.T) { - cfg := config.Default() - cfg.MaxSectionTokens = 3 - cfg.MinSectionTokens = 0 - cfg.TotalLLMConcurrency = 4 - cfg.ProposalLLMConcurrency = 4 - validationCap := 1 - cfg.ValidationLLMConcurrency = &validationCap - - transcript := &schema.Transcript{Segments: []schema.Segment{ - {ID: 1, Text: "one one"}, - {ID: 2, Text: "two two"}, - {ID: 3, Text: "three three"}, - }} - - validationSchedulerInner, err := llm.NewScheduler(1) - if err != nil { - t.Fatalf("NewScheduler(validation): %v", err) - } - validationScheduler := &trackingScheduler{inner: validationSchedulerInner} - - releaseValidation := make(chan struct{}) - client := &stageAwareStructuredClient{ - startedSection: make(chan int, 16), - releaseValidation: releaseValidation, - } - llmValidator, err := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility, "") - if err != nil { - t.Fatalf("NewLLMBackedValidator: %v", err) - } - - r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ - "m": fakeModule{ - key: "m", - policy: proposals.ReplacementPolicyRequireUnique, - validators: []contracts.Validator{llmValidator}, - proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { - seg := req.WorkingTranscript.Segments[0] - return []proposals.CorrectionProposal{ - {TargetSegmentID: seg.ID, OriginalText: seg.Text, CorrectedText: strings.ToUpper(seg.Text), Confidence: 1}, - }, nil - }, - }, - }}) - - resultCh := make(chan error, 1) - go func() { - _, err := r.Run(context.Background(), RunInput{ - Config: &cfg, - Transcript: transcript, - ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}, - ValidationLLMScheduler: validationScheduler, - ValidationLLMClient: client, - }) - resultCh <- err - }() - - select { - case <-client.startedSection: - case <-time.After(350 * time.Millisecond): - t.Fatal("expected validation to start") - } - close(releaseValidation) - - if err := <-resultCh; err != nil { - t.Fatalf("Run error: %v", err) - } - if got := atomic.LoadInt32(&validationScheduler.maxInFlight); got > 1 { - t.Fatalf("expected validation in-flight <= 1, got %d", got) - } -} - -func TestRunnerMixedProposalValidationFIFOOrder(t *testing.T) { - cfg := config.Default() - cfg.MaxSectionTokens = 3 - cfg.MinSectionTokens = 0 - cfg.TotalLLMConcurrency = 1 - cfg.ProposalLLMConcurrency = 1 - validationCap := 1 - cfg.ValidationLLMConcurrency = &validationCap - - transcript := &schema.Transcript{Segments: []schema.Segment{ - {ID: 1, Text: "alpha one"}, - {ID: 2, Text: "bravo two"}, - }} - - global, err := llm.NewScheduler(1) - if err != nil { - t.Fatalf("NewScheduler: %v", err) - } - proposalScheduler := global - validationScheduler := global - - events := make(chan string, 8) - releaseProposalSectionOne := make(chan struct{}) - releaseValidation := make(chan struct{}) - sectionZeroEntered := make(chan struct{}) - sectionOneAttempted := make(chan struct{}, 1) - - client := &stageAwareStructuredClient{ - startedSection: make(chan int, 8), - releaseValidation: releaseValidation, - eventSink: events, - } - llmValidator, err := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility, "") - if err != nil { - t.Fatalf("NewLLMBackedValidator: %v", err) - } - - r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ - "m": fakeModule{ - key: "m", - policy: proposals.ReplacementPolicyRequireUnique, - validators: []contracts.Validator{llmValidator}, - proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { - seg := req.WorkingTranscript.Segments[0] - if req.Section != nil && req.Section.Index == 1 { - <-sectionZeroEntered - select { - case sectionOneAttempted <- struct{}{}: - default: - } - } - err := req.LLMScheduler.Run(context.Background(), func(context.Context) error { - if req.Section != nil { - events <- "p" + strconv.Itoa(req.Section.Index) - if req.Section.Index == 0 { - select { - case sectionZeroEntered <- struct{}{}: - default: - } - <-sectionOneAttempted - } - if req.Section.Index == 1 { - <-releaseProposalSectionOne - } - } - return nil - }) - if err != nil { - return nil, err - } - return []proposals.CorrectionProposal{ - {TargetSegmentID: seg.ID, OriginalText: seg.Text, CorrectedText: strings.ToUpper(seg.Text), Confidence: 1}, - }, nil - }, - }, - }}) - - resultCh := make(chan error, 1) - go func() { - _, err := r.Run(context.Background(), RunInput{ - Config: &cfg, - Transcript: transcript, - ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}, - ProposalLLMScheduler: proposalScheduler, - ValidationLLMScheduler: validationScheduler, - ValidationLLMClient: client, - }) - resultCh <- err - }() - - if got := <-events; got != "p0" { - t.Fatalf("expected first event p0, got %q", got) - } - if got := <-events; got != "p1" { - t.Fatalf("expected second event p1 (FIFO queued proposal), got %q", got) - } - - close(releaseProposalSectionOne) - close(releaseValidation) - - if got := <-events; !strings.HasPrefix(got, "v") { - t.Fatalf("expected validator event after queued p1, got %q", got) - } - - if err := <-resultCh; err != nil { - t.Fatalf("Run error: %v", err) - } -} - -var sectionStagePattern = regexp.MustCompile(`section-(\d+)`) - type stageAwareStructuredClient struct { startedSection chan int releaseValidation <-chan struct{} @@ -804,7 +437,10 @@ func (c *captureContractStructuredClient) CompleteStructured(ctx context.Context } func (c *stageAwareStructuredClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { - section := parseSectionFromStage(req.StageName) + section := 0 + if req.Context.BatchIndex != nil { + section = *req.Context.BatchIndex + } if c.startedSection != nil { select { case c.startedSection <- section: @@ -831,67 +467,6 @@ func (c *stageAwareStructuredClient) CompleteStructured(ctx context.Context, req return contracts.StructuredCompletionResponse{}, nil } -func parseSectionFromStage(stage string) int { - match := sectionStagePattern.FindStringSubmatch(stage) - if len(match) != 2 { - return 0 - } - n, err := strconv.Atoi(match[1]) - if err != nil { - return 0 - } - return n -} - -func TestValidationLLMClientAdapterPassesResponseSchema(t *testing.T) { - capture := &captureContractStructuredClient{} - adapter := validationLLMClientAdapter{client: capture} - schema := responseschema.MustLookup(responseschema.ValidatorDecisionSetKey) - - _, err := adapter.CompleteStructured(context.Background(), validators.StructuredCompletionRequest{ - StageName: "module:validator:batch-0000", - Messages: []validators.LLMMessage{ - {Role: "system", Content: "system"}, - {Role: "user", Content: "user"}, - }, - Model: "test-model", - ResponseSchema: &schema, - }, &validators.LLMValidationResponse{}) - if err != nil { - t.Fatalf("CompleteStructured error: %v", err) - } - - if capture.lastRequest.ResponseSchema == nil { - t.Fatalf("expected response schema to be forwarded") - } - if capture.lastRequest.ResponseSchema.ID != schema.ID || - capture.lastRequest.ResponseSchema.Version != schema.Version || - capture.lastRequest.ResponseSchema.Name != schema.Name || - capture.lastRequest.ResponseSchema.SHA256 != schema.SHA256 { - t.Fatalf("unexpected forwarded schema metadata: got=%+v want=%+v", *capture.lastRequest.ResponseSchema, schema) - } -} - -type trackingScheduler struct { - inner contracts.LLMScheduler - inFlight int32 - maxInFlight int32 -} - -func (s *trackingScheduler) Run(ctx context.Context, fn func(context.Context) error) error { - return s.inner.Run(ctx, func(callCtx context.Context) error { - current := atomic.AddInt32(&s.inFlight, 1) - for { - prior := atomic.LoadInt32(&s.maxInFlight) - if current <= prior || atomic.CompareAndSwapInt32(&s.maxInFlight, prior, current) { - break - } - } - defer atomic.AddInt32(&s.inFlight, -1) - return fn(callCtx) - }) -} - func TestRunnerSkippedRecorded(t *testing.T) { transcript := &schema.Transcript{Segments: []schema.Segment{{ID: 1, Speaker: "A", Start: 0, End: 1, Text: "word word"}}} r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ @@ -940,7 +515,7 @@ func TestRunnerDoesNotMutateInputTranscript(t *testing.T) { }}, }}) - _, err := r.Run(context.Background(), RunInput{Config: ptrConfig(config.Default()), Transcript: transcript, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}}) + _, err := r.Run(context.Background(), RunInput{Config: ptrConfig(runnerTestConfig()), Transcript: transcript, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}}) if err != nil { t.Fatalf("Run error: %v", err) } @@ -1128,20 +703,9 @@ func (f *fakeStructuredClient) CompleteStructured(ctx context.Context, req contr return contracts.StructuredCompletionResponse{}, nil } -type countingScheduler struct{ runs int } - -func (s *countingScheduler) Run(ctx context.Context, fn func(context.Context) error) error { - s.runs++ - return fn(ctx) -} - -type lenEstimator struct{} - -func (lenEstimator) EstimateTokens(text string) int { return len(text) } - func TestRunnerLLMValidatorApprovalApplied(t *testing.T) { client := &fakeStructuredClient{responses: []validators.LLMValidationResponse{{Validations: []validators.LLMValidationDecision{{CorrectionIndex: 0, Approved: true, Confidence: 0.9, Reason: "ok"}}}}} - llmValidator, err := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility, "") + llmValidator, err := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility) if err != nil { t.Fatalf("NewLLMBackedValidator: %v", err) } @@ -1151,7 +715,7 @@ func TestRunnerLLMValidatorApprovalApplied(t *testing.T) { return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "gestures", CorrectedText: "Jesters", Confidence: 1}}, nil }}, }}) - cfg := config.Default() + cfg := runnerTestConfig() out, err := r.Run(context.Background(), RunInput{ Config: &cfg, Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "There were gestures at the temple."}}}, @@ -1168,13 +732,13 @@ func TestRunnerLLMValidatorApprovalApplied(t *testing.T) { func TestRunnerLLMValidatorRejectionPreventsApplication(t *testing.T) { client := &fakeStructuredClient{responses: []validators.LLMValidationResponse{{Validations: []validators.LLMValidationDecision{{CorrectionIndex: 0, Approved: false, Confidence: 0.95, Reason: "reject"}}}}} - llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility, "") + llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility) r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ "m": fakeModule{key: "m", policy: proposals.ReplacementPolicyRequireUnique, validators: []contracts.Validator{llmValidator}, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "gestures", CorrectedText: "Jesters", Confidence: 1}}, nil }}, }}) - cfg := config.Default() + cfg := runnerTestConfig() out, err := r.Run(context.Background(), RunInput{ Config: &cfg, Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "There were gestures at the temple."}}}, @@ -1194,7 +758,7 @@ func TestRunnerLLMValidatorRejectionPreventsApplication(t *testing.T) { func TestRunnerLLMValidatorMalformedResponseRejectsBatchAndKeepsPartialProgress(t *testing.T) { client := &fakeStructuredClient{responses: []validators.LLMValidationResponse{{Validations: []validators.LLMValidationDecision{{CorrectionIndex: 99, Approved: true, Confidence: 0.9, Reason: "bad index"}}}}} - llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility, "") + llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility) r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ "m1": fakeModule{key: "m1", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "teh", CorrectedText: "the", Confidence: 1}}, nil @@ -1203,7 +767,7 @@ func TestRunnerLLMValidatorMalformedResponseRejectsBatchAndKeepsPartialProgress( return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "cat", CorrectedText: "dog", Confidence: 1}}, nil }}, }}) - cfg := config.Default() + cfg := runnerTestConfig() out, err := r.Run(context.Background(), RunInput{ Config: &cfg, Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "teh cat"}}}, @@ -1226,13 +790,13 @@ func TestRunnerLLMValidatorMalformedResponseRejectsBatchAndKeepsPartialProgress( func TestRunnerLLMValidatorMissingDecisionRejectsBatch(t *testing.T) { client := &fakeStructuredClient{responses: []validators.LLMValidationResponse{{Validations: []validators.LLMValidationDecision{}}}} - llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility, "") + llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility) r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ "m": fakeModule{key: "m", policy: proposals.ReplacementPolicyRequireUnique, validators: []contracts.Validator{llmValidator}, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "gestures", CorrectedText: "Jesters", Confidence: 1}}, nil }}, }}) - cfg := config.Default() + cfg := runnerTestConfig() _, err := r.Run(context.Background(), RunInput{ Config: &cfg, Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "There were gestures at the temple."}}}, @@ -1249,13 +813,13 @@ func TestRunnerLLMValidatorDuplicateDecisionRejectsBatch(t *testing.T) { {CorrectionIndex: 0, Approved: true, Confidence: 0.9, Reason: "ok"}, {CorrectionIndex: 0, Approved: false, Confidence: 0.9, Reason: "dup"}, }}}} - llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility, "") + llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility) r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ "m": fakeModule{key: "m", policy: proposals.ReplacementPolicyRequireUnique, validators: []contracts.Validator{llmValidator}, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "gestures", CorrectedText: "Jesters", Confidence: 1}}, nil }}, }}) - cfg := config.Default() + cfg := runnerTestConfig() _, err := r.Run(context.Background(), RunInput{ Config: &cfg, Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "There were gestures at the temple."}}}, @@ -1267,16 +831,20 @@ func TestRunnerLLMValidatorDuplicateDecisionRejectsBatch(t *testing.T) { } } -func TestRunnerLLMValidatorBatchingAndSchedulerUsage(t *testing.T) { +type lenEstimator struct{} + +func (lenEstimator) EstimateTokens(text string) int { return len(text) } + +func TestRunnerLLMValidatorBatching(t *testing.T) { client := &fakeStructuredClient{responses: []validators.LLMValidationResponse{ {Validations: []validators.LLMValidationDecision{{CorrectionIndex: 0, Approved: true, Confidence: 0.9, Reason: "ok"}}}, {Validations: []validators.LLMValidationDecision{{CorrectionIndex: 1, Approved: true, Confidence: 0.9, Reason: "ok"}}}, }} - llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility, "") + llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility) llmValidator.SetTokenEstimator(lenEstimator{}) - scheduler := &countingScheduler{} - cfg := config.Default() - cfg.ValidationMaxPromptTokens = 260 + + cfg := runnerTestConfig() + cfg.Validation.MaxPromptTokens = 260 r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ "m": fakeModule{key: "m", policy: proposals.ReplacementPolicyReplaceAll, validators: []contracts.Validator{llmValidator}, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { return []proposals.CorrectionProposal{ @@ -1286,18 +854,15 @@ func TestRunnerLLMValidatorBatchingAndSchedulerUsage(t *testing.T) { }}, }}) out, err := r.Run(context.Background(), RunInput{ - Config: &cfg, - Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "alpha gamma"}}}, - ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}, - ValidationLLMClient: client, - ValidationLLMScheduler: scheduler, + Config: &cfg, + Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "alpha gamma"}}}, + ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}, + ValidationLLMClient: client, }) if err != nil { t.Fatalf("unexpected run error: %v", err) } - if scheduler.runs < 2 { - t.Fatalf("expected scheduler to run per batch, got %d", scheduler.runs) - } + if client.calls < 2 { t.Fatalf("expected multiple llm calls for batching, got %d", client.calls) } @@ -1310,10 +875,14 @@ func TestRunnerLLMValidatorDiagnosticsWrittenAndRedacted(t *testing.T) { primarySecret := "runner-primary-secret" validationSecret := "runner-validation-secret" client := &fakeStructuredClient{responses: []validators.LLMValidationResponse{{Validations: []validators.LLMValidationDecision{{CorrectionIndex: 0, Approved: true, Confidence: 0.9, Reason: validationSecret}}}}} - llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility, "") - cfg := config.Default() - cfg.PrimaryLLM.APIKey = primarySecret - cfg.ValidationLLM.APIKey = validationSecret + llmValidator, _ := validators.NewLLMBackedValidator("spoken_form_plausibility_review", validators.LLMValidatorTypeSpokenFormPlausibility) + t.Setenv("RUNNER_PRIMARY_KEY", primarySecret) + t.Setenv("RUNNER_VALIDATION_KEY", validationSecret) + loaded, loadErr := config.Load(ck.Bytes("config.yml", []byte("version: 2\nbackends:\n primary:\n endpoint: https://primary.invalid/v1\n credential: RUNNER_PRIMARY_KEY\n validation:\n endpoint: https://validation.invalid/v1\n credential: RUNNER_VALIDATION_KEY\nllm:\n proposal_profile: base\n")), "") + if loadErr != nil { + t.Fatal(loadErr) + } + cfg := loaded.Config diagDir := t.TempDir() r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ "m": fakeModule{key: "m", policy: proposals.ReplacementPolicyRequireUnique, validators: []contracts.Validator{llmValidator}, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { @@ -1359,16 +928,6 @@ func TestRunnerLLMValidatorDiagnosticsWrittenAndRedacted(t *testing.T) { } } -func TestRunnerAcceptsLLMSchedulerType(t *testing.T) { - s, err := llm.NewScheduler(1) - if err != nil { - t.Fatalf("NewScheduler: %v", err) - } - if s == nil { - t.Fatal("expected scheduler instance") - } -} - type proposalGenerationModule struct { key string policy proposals.ReplacementPolicy @@ -1385,13 +944,11 @@ func (m proposalGenerationModule) Propose(ctx context.Context, req contracts.Pro WorkingTranscript: req.WorkingTranscript, Config: req.Config, Glossary: req.Glossary, - Messages: []contracts.LLMMessage{ - {Role: "system", Content: "return transcript corrections"}, - {Role: "user", Content: "produce one safe correction"}, - }, - LLMClient: req.LLMClient, - Scheduler: req.LLMScheduler, - DiagnosticsDir: req.DiagnosticsDir, + PromptID: "modules.grammar.proposal", + PromptVersion: "v1", + Variables: map[string]string{"TranscriptDescriptionBlock": "", "GlossaryJSON": "[]", "SectionJSON": "[]"}, + LLMClient: req.LLMClient, + DiagnosticsDir: req.DiagnosticsDir, }) if err != nil { return contracts.ProposalResult{}, err @@ -1428,8 +985,8 @@ func TestRunnerProposalGenerationHelperFlowsThroughPipeline(t *testing.T) { }, }, } - scheduler := &countingScheduler{} - cfg := config.Default() + + cfg := runnerTestConfig() diagDir := t.TempDir() r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ @@ -1441,7 +998,6 @@ func TestRunnerProposalGenerationHelperFlowsThroughPipeline(t *testing.T) { Transcript: &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "teh cat"}}}, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "m", InstanceName: "m"}}, ProposalLLMClient: client, - ProposalLLMScheduler: scheduler, ProposalDiagnosticsDir: diagDir, }) if err != nil { @@ -1450,9 +1006,6 @@ func TestRunnerProposalGenerationHelperFlowsThroughPipeline(t *testing.T) { if out.FinalTranscript.Segments[0].Text != "the cat" { t.Fatalf("expected proposal-generated correction to apply, got %q", out.FinalTranscript.Segments[0].Text) } - if scheduler.runs != 1 { - t.Fatalf("expected proposal scheduler use, got %d runs", scheduler.runs) - } if len(out.ModuleResults) != 1 || len(out.ModuleResults[0].AppliedChanges) != 1 { t.Fatalf("expected one applied change, got %+v", out.ModuleResults) } @@ -1475,7 +1028,7 @@ func TestRunnerGrammarModuleUsesConfidenceThreshold(t *testing.T) { }, }, } - cfg := config.Default() + cfg := runnerTestConfig() cfg.Thresholds.Grammar = 0.9 factory := modules.NewFactory(modules.Dependencies{}) out, err := New(factory).Run(context.Background(), RunInput{ @@ -1509,7 +1062,7 @@ func TestRunnerGlossaryModuleUsesConfidenceThreshold(t *testing.T) { }, }, } - cfg := config.Default() + cfg := runnerTestConfig() cfg.Thresholds.Glossary = 0.9 factory := modules.NewFactory(modules.Dependencies{}) out, err := New(factory).Run(context.Background(), RunInput{ diff --git a/internal/framework/validators/llm_validators.go b/internal/framework/validators/llm_validators.go index 158e178..568a715 100644 --- a/internal/framework/validators/llm_validators.go +++ b/internal/framework/validators/llm_validators.go @@ -53,12 +53,12 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result, if v == nil { return Result{}, fmt.Errorf("validator is nil") } - if req.LLMClient == nil { - return Result{}, fmt.Errorf("LLM-backed validator %q requires a structured LLM client", v.name) - } if len(req.CandidateProposal) == 0 { return Result{ValidatorName: v.name, Decisions: nil}, nil } + if req.LLMClient == nil { + return Result{}, fmt.Errorf("LLM-backed validator %q requires a structured LLM client", v.name) + } validationReq, immediate := BuildLLMValidationRequest(v.name, v.validatorType, req) if len(validationReq.Items) == 0 { all := append([]Decision(nil), immediate...) @@ -108,19 +108,12 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result, responseSchema := responseschema.MustLookup(responseschema.ValidatorDecisionSetKey) promptMetadata := validatorPromptMetadata(v.validatorType) stage := stagename.ValidatorBatch(req.ModuleInstance, v.name, batch.BatchIndex) - call := func(callCtx context.Context) error { - completion, err = req.LLMClient.CompleteStructured(callCtx, StructuredCompletionRequest{ - PromptID: promptMetadata.PromptID, PromptVersion: promptMetadata.PromptVersion, - Variables: variables, Schema: responseschema.ValidatorDecisionSetKey, - Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, BatchIndex: &batch.BatchIndex, ValidatorName: v.name}, - }, &response) - return err - } - if req.Scheduler != nil { - err = req.Scheduler.Run(ctx, call) - } else { - err = call(ctx) - } + completion, err = req.LLMClient.CompleteStructured(ctx, StructuredCompletionRequest{ + PromptID: promptMetadata.PromptID, PromptVersion: promptMetadata.PromptVersion, + Variables: variables, Schema: responseschema.ValidatorDecisionSetKey, + Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, BatchIndex: &batch.BatchIndex, ValidatorName: v.name}, + }, &response) + artifacts := InteractionArtifacts{} if req.DiagnosticsWriter != nil { artifacts, _ = req.DiagnosticsWriter.WriteInteraction( diff --git a/internal/framework/validators/llm_validators_test.go b/internal/framework/validators/llm_validators_test.go index 50bf112..665836a 100644 --- a/internal/framework/validators/llm_validators_test.go +++ b/internal/framework/validators/llm_validators_test.go @@ -3,12 +3,8 @@ package validators import ( "context" "errors" - "runtime" "strings" - "sync" - "sync/atomic" "testing" - "time" "gitea.maximumdirect.net/eric/audita/internal/core/chunking" "gitea.maximumdirect.net/eric/audita/internal/core/config" @@ -25,17 +21,6 @@ type fakeStructuredLLMClient struct { calls []StructuredCompletionRequest } -type sleepingValidationClient struct { - inFlight int32 - maxInFlight int32 - entered chan struct{} - release chan struct{} -} - -type boundedScheduler struct { - permits chan struct{} -} - type captureValidationDiagnosticsWriter struct { lastRequestMetadata any } @@ -49,48 +34,6 @@ func (w *captureValidationDiagnosticsWriter) WriteInteraction(stage string, requ return InteractionArtifacts{}, nil } -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 - } - } - if c.entered != nil { - c.entered <- struct{}{} - } - select { - case <-c.release: - 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) @@ -313,42 +256,6 @@ func TestLLMBackedValidatorOversizedSingleProposalRejectsOnlyThatProposal(t *tes } } -func TestLLMBackedValidatorRespectsSchedulerConcurrency(t *testing.T) { - scheduler := newBoundedScheduler(2) - client := &sleepingValidationClient{ - entered: make(chan struct{}, 16), - release: make(chan struct{}), - } - v, err := NewLLMBackedValidator("spoken_form_plausibility_review", LLMValidatorTypeSpokenFormPlausibility) - 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) - } - }() - } - waitForValidationEntries(t, client.entered, 2) - close(client.release) - 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) - } -} - func TestLLMBackedValidatorDiagnosticsIncludeSchemaMetadata(t *testing.T) { client := &fakeStructuredLLMClient{responses: []LLMValidationResponse{{Validations: []LLMValidationDecision{ {CorrectionIndex: 0, Approved: true, Confidence: 0.9, Reason: "ok"}, @@ -379,24 +286,6 @@ func TestLLMBackedValidatorDiagnosticsIncludeSchemaMetadata(t *testing.T) { } } - -func waitForValidationEntries(t *testing.T, entered <-chan struct{}, want int) { - t.Helper() - deadline := time.Now().Add(300 * time.Millisecond) - got := 0 - for got < want && time.Now().Before(deadline) { - select { - case <-entered: - got++ - default: - runtime.Gosched() - } - } - if got < want { - t.Fatalf("timed out waiting for %d validator calls to enter (got %d)", want, got) - } -} - func TestEveryValidationTypeSelectsEmbeddedPromptAndVariables(t *testing.T) { cases := []struct { kind LLMValidatorType diff --git a/internal/framework/validators/models.go b/internal/framework/validators/models.go index 59dc9ed..079c815 100644 --- a/internal/framework/validators/models.go +++ b/internal/framework/validators/models.go @@ -37,7 +37,6 @@ type Request struct { Glossary *schema.Glossary `json:"-"` Config *config.Config `json:"-"` LLMClient StructuredLLMClient `json:"-"` - Scheduler ValidationScheduler `json:"-"` DiagnosticsWriter InteractionDiagnosticsWriter `json:"-"` } @@ -57,11 +56,6 @@ type Result struct { Warnings []stagewarnings.StageWarning `json:"warnings,omitempty"` } -// ValidationScheduler provides bounded execution for validator LLM calls. -type ValidationScheduler interface { - Run(ctx context.Context, fn func(context.Context) error) error -} - type InteractionDiagnosticsWriter interface { WriteInteraction(stage string, requestMetadata any, requestPayload any, responsePayload any, errorPayload any) (InteractionArtifacts, error) } diff --git a/internal/modules/glossary/module_test.go b/internal/modules/glossary/module_test.go index 99fd250..e9820fa 100644 --- a/internal/modules/glossary/module_test.go +++ b/internal/modules/glossary/module_test.go @@ -41,15 +41,6 @@ func (f *fakeLLMClient) CompleteStructured(ctx context.Context, req contracts.St return contracts.StructuredCompletionResponse{}, nil } -type countingScheduler struct { - runs int -} - -func (s *countingScheduler) Run(ctx context.Context, fn func(context.Context) error) error { - s.runs++ - return fn(ctx) -} - func tinyTranscript() *schema.Transcript { return &schema.Transcript{Segments: []schema.Segment{ {ID: 1, Speaker: "A", Start: 0, End: 1, Text: "There were gestures in the hall.", Categories: []string{"session"}}, @@ -107,7 +98,6 @@ func TestGlossaryModuleProposeMapsCorrectionsAndWritesDiagnostics(t *testing.T) }, }, } - scheduler := &countingScheduler{} t.Setenv("MODULE_SECRET", secret) loaded, loadErr := config.Load(ck.Bytes("config.yml", []byte("version: 2\nbackends:\n offline:\n endpoint: https://offline.invalid/v1\n credential: MODULE_SECRET\nllm:\n proposal_profile: base\n")), "") if loadErr != nil { @@ -131,15 +121,12 @@ func TestGlossaryModuleProposeMapsCorrectionsAndWritesDiagnostics(t *testing.T) InstanceName: "glossary", ReplacementPolicy: proposals.ReplacementPolicyReplaceAll, }, - LLMClient: client, - LLMScheduler: scheduler, + LLMClient: client, }) if err != nil { t.Fatalf("Propose error: %v", err) } - if scheduler.runs != 1 { - t.Fatalf("expected scheduler run count 1, got %d", scheduler.runs) - } + if len(client.calls) != 1 || client.calls[0].PromptID != prompts.PromptIDModuleGlossaryProposal { t.Fatalf("expected one glossary:proposal call, got %+v", client.calls) } diff --git a/internal/modules/grammar/module_test.go b/internal/modules/grammar/module_test.go index 6fb3df7..eae684a 100644 --- a/internal/modules/grammar/module_test.go +++ b/internal/modules/grammar/module_test.go @@ -51,15 +51,6 @@ func (f *fakeLLMClient) CompleteStructured(ctx context.Context, req contracts.St return contracts.StructuredCompletionResponse{}, nil } -type countingScheduler struct { - runs int -} - -func (s *countingScheduler) Run(ctx context.Context, fn func(context.Context) error) error { - s.runs++ - return fn(ctx) -} - func tinyTranscript() *schema.Transcript { return &schema.Transcript{Segments: []schema.Segment{ {ID: 1, Speaker: "A", Start: 0, End: 1, Text: "hello ,world"}, @@ -117,7 +108,6 @@ func TestGrammarModuleProposeMapsCorrectionsAndWritesDiagnostics(t *testing.T) { }, }, } - scheduler := &countingScheduler{} t.Setenv("MODULE_SECRET", secret) loaded, loadErr := config.Load(ck.Bytes("config.yml", []byte("version: 2\nbackends:\n offline:\n endpoint: https://offline.invalid/v1\n credential: MODULE_SECRET\nllm:\n proposal_profile: base\n")), "") if loadErr != nil { @@ -141,15 +131,12 @@ func TestGrammarModuleProposeMapsCorrectionsAndWritesDiagnostics(t *testing.T) { InstanceName: "grammar", ReplacementPolicy: proposals.ReplacementPolicyRequireUnique, }, - LLMClient: client, - LLMScheduler: scheduler, + LLMClient: client, }) if err != nil { t.Fatalf("Propose error: %v", err) } - if scheduler.runs != 1 { - t.Fatalf("expected scheduler run count 1, got %d", scheduler.runs) - } + if len(client.calls) != 1 || client.calls[0].PromptID != prompts.PromptIDModuleGrammarProposal { t.Fatalf("expected one grammar:proposal call, got %+v", client.calls) } diff --git a/internal/modules/homophones/module_test.go b/internal/modules/homophones/module_test.go index 883d622..c1c983a 100644 --- a/internal/modules/homophones/module_test.go +++ b/internal/modules/homophones/module_test.go @@ -42,15 +42,6 @@ func (f *fakeLLMClient) CompleteStructured(ctx context.Context, req contracts.St return contracts.StructuredCompletionResponse{}, nil } -type countingScheduler struct { - runs int -} - -func (s *countingScheduler) Run(ctx context.Context, fn func(context.Context) error) error { - s.runs++ - return fn(ctx) -} - func tinyTranscript() *schema.Transcript { return &schema.Transcript{Segments: []schema.Segment{ {ID: 1, Speaker: "A", Start: 0, End: 1, Text: "There were gestures in the hall.", Categories: []string{"session"}}, @@ -108,7 +99,6 @@ func TestHomophonesModuleProposeMapsCorrectionsAndWritesDiagnostics(t *testing.T }, }, } - scheduler := &countingScheduler{} t.Setenv("MODULE_SECRET", secret) loaded, loadErr := config.Load(ck.Bytes("config.yml", []byte("version: 2\nbackends:\n offline:\n endpoint: https://offline.invalid/v1\n credential: MODULE_SECRET\nllm:\n proposal_profile: base\n")), "") if loadErr != nil { @@ -132,15 +122,12 @@ func TestHomophonesModuleProposeMapsCorrectionsAndWritesDiagnostics(t *testing.T InstanceName: "homophones", ReplacementPolicy: proposals.ReplacementPolicyRequireUnique, }, - LLMClient: client, - LLMScheduler: scheduler, + LLMClient: client, }) if err != nil { t.Fatalf("Propose error: %v", err) } - if scheduler.runs != 1 { - t.Fatalf("expected scheduler run count 1, got %d", scheduler.runs) - } + if len(client.calls) != 1 || client.calls[0].PromptID != prompts.PromptIDModuleHomophonesProposal { t.Fatalf("expected one homophones:proposal call, got %+v", client.calls) } diff --git a/internal/modules/spoken_word/module_test.go b/internal/modules/spoken_word/module_test.go index a1390c6..0c6821f 100644 --- a/internal/modules/spoken_word/module_test.go +++ b/internal/modules/spoken_word/module_test.go @@ -42,15 +42,6 @@ func (f *fakeLLMClient) CompleteStructured(ctx context.Context, req contracts.St return contracts.StructuredCompletionResponse{}, nil } -type countingScheduler struct { - runs int -} - -func (s *countingScheduler) Run(ctx context.Context, fn func(context.Context) error) error { - s.runs++ - return fn(ctx) -} - func tinyTranscript() *schema.Transcript { return &schema.Transcript{Segments: []schema.Segment{ {ID: 1, Speaker: "A", Start: 0, End: 1, Text: "I I think, you know, we should go.", Categories: []string{"session"}}, @@ -108,7 +99,6 @@ func TestSpokenWordModuleProposeMapsCorrectionsAndWritesDiagnostics(t *testing.T }, }, } - scheduler := &countingScheduler{} t.Setenv("MODULE_SECRET", secret) loaded, loadErr := config.Load(ck.Bytes("config.yml", []byte("version: 2\nbackends:\n offline:\n endpoint: https://offline.invalid/v1\n credential: MODULE_SECRET\nllm:\n proposal_profile: base\n")), "") if loadErr != nil { @@ -132,15 +122,12 @@ func TestSpokenWordModuleProposeMapsCorrectionsAndWritesDiagnostics(t *testing.T InstanceName: "spoken_word", ReplacementPolicy: proposals.ReplacementPolicyRequireUnique, }, - LLMClient: client, - LLMScheduler: scheduler, + LLMClient: client, }) if err != nil { t.Fatalf("Propose error: %v", err) } - if scheduler.runs != 1 { - t.Fatalf("expected scheduler run count 1, got %d", scheduler.runs) - } + if len(client.calls) != 1 || client.calls[0].PromptID != prompts.PromptIDModuleSpokenWordProposal { t.Fatalf("expected one spoken_word:proposal call, got %+v", client.calls) }