342 lines
12 KiB
Go
342 lines
12 KiB
Go
package audita
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/url"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/subprocess"
|
|
)
|
|
|
|
// SubprocessRunnerConfig defines deterministic settings for Audita CLI execution.
|
|
type SubprocessRunnerConfig struct {
|
|
Binary string
|
|
Timeout time.Duration
|
|
LLMAPIKeyEnv string
|
|
Modules []string
|
|
BaseURL string
|
|
Model string
|
|
LLMConcurrency *int
|
|
ValidationModel string
|
|
ValidationLLMConcurrency *int
|
|
Report bool
|
|
}
|
|
|
|
// SubprocessRunner invokes Audita via subprocess.
|
|
type SubprocessRunner struct {
|
|
binary string
|
|
timeout time.Duration
|
|
llmAPIKeyEnv string
|
|
modules []string
|
|
baseURL string
|
|
model string
|
|
llmConcurrency *int
|
|
validationModel string
|
|
validationLLMConcurrency *int
|
|
report bool
|
|
}
|
|
|
|
// NewSubprocessRunnerFromConfigValues parses config-derived values once.
|
|
func NewSubprocessRunnerFromConfigValues(
|
|
binary string,
|
|
timeout string,
|
|
llmAPIKeyEnv string,
|
|
modules []string,
|
|
baseURL string,
|
|
model string,
|
|
llmConcurrency *int,
|
|
validationModel string,
|
|
validationLLMConcurrency *int,
|
|
report bool,
|
|
) (*SubprocessRunner, error) {
|
|
if strings.TrimSpace(timeout) == "" {
|
|
return nil, fmt.Errorf("audita timeout is required")
|
|
}
|
|
parsedTimeout, err := time.ParseDuration(timeout)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("parse audita timeout %q: %w", timeout, err)
|
|
}
|
|
return NewSubprocessRunner(SubprocessRunnerConfig{
|
|
Binary: binary,
|
|
Timeout: parsedTimeout,
|
|
LLMAPIKeyEnv: llmAPIKeyEnv,
|
|
Modules: modules,
|
|
BaseURL: baseURL,
|
|
Model: model,
|
|
LLMConcurrency: llmConcurrency,
|
|
ValidationModel: validationModel,
|
|
ValidationLLMConcurrency: validationLLMConcurrency,
|
|
Report: report,
|
|
})
|
|
}
|
|
|
|
// NewSubprocessRunner constructs a validated Audita subprocess runner.
|
|
func NewSubprocessRunner(cfg SubprocessRunnerConfig) (*SubprocessRunner, error) {
|
|
if strings.TrimSpace(cfg.Binary) == "" {
|
|
return nil, fmt.Errorf("audita binary is required")
|
|
}
|
|
if cfg.Timeout <= 0 {
|
|
return nil, fmt.Errorf("audita timeout must be > 0")
|
|
}
|
|
if len(cfg.Modules) == 0 {
|
|
return nil, fmt.Errorf("audita modules must include at least one module")
|
|
}
|
|
for i, module := range cfg.Modules {
|
|
if strings.TrimSpace(module) == "" {
|
|
return nil, fmt.Errorf("audita module at index %d is empty", i)
|
|
}
|
|
}
|
|
if strings.TrimSpace(cfg.BaseURL) == "" {
|
|
return nil, fmt.Errorf("audita base url is required")
|
|
}
|
|
u, err := url.Parse(cfg.BaseURL)
|
|
if err != nil || u.Scheme == "" || u.Host == "" {
|
|
if err != nil {
|
|
return nil, fmt.Errorf("audita base url %q is invalid: %w", cfg.BaseURL, err)
|
|
}
|
|
return nil, fmt.Errorf("audita base url %q is invalid", cfg.BaseURL)
|
|
}
|
|
if strings.TrimSpace(cfg.Model) == "" {
|
|
return nil, fmt.Errorf("audita model is required")
|
|
}
|
|
if cfg.LLMConcurrency != nil && *cfg.LLMConcurrency <= 0 {
|
|
return nil, fmt.Errorf("audita llm concurrency must be > 0 when provided")
|
|
}
|
|
if cfg.ValidationLLMConcurrency != nil && *cfg.ValidationLLMConcurrency <= 0 {
|
|
return nil, fmt.Errorf("audita validation llm concurrency must be > 0 when provided")
|
|
}
|
|
|
|
modules := make([]string, len(cfg.Modules))
|
|
for i, m := range cfg.Modules {
|
|
modules[i] = strings.TrimSpace(m)
|
|
}
|
|
|
|
return &SubprocessRunner{
|
|
binary: strings.TrimSpace(cfg.Binary),
|
|
timeout: cfg.Timeout,
|
|
llmAPIKeyEnv: strings.TrimSpace(cfg.LLMAPIKeyEnv),
|
|
modules: modules,
|
|
baseURL: strings.TrimSpace(cfg.BaseURL),
|
|
model: strings.TrimSpace(cfg.Model),
|
|
llmConcurrency: cfg.LLMConcurrency,
|
|
validationModel: strings.TrimSpace(cfg.ValidationModel),
|
|
validationLLMConcurrency: cfg.ValidationLLMConcurrency,
|
|
report: cfg.Report,
|
|
}, nil
|
|
}
|
|
|
|
// Run executes Audita process with deterministic flags and validates output artifacts.
|
|
func (r *SubprocessRunner) Run(ctx context.Context, req PolishRequest) (PolishResult, error) {
|
|
if r == nil {
|
|
return PolishResult{}, fmt.Errorf("audita subprocess runner is nil")
|
|
}
|
|
if strings.TrimSpace(req.MergedTranscriptPath) == "" {
|
|
return PolishResult{}, fmt.Errorf("audita merged transcript path is required")
|
|
}
|
|
if strings.TrimSpace(req.GlossaryPath) == "" {
|
|
return PolishResult{}, fmt.Errorf("audita glossary path is required")
|
|
}
|
|
if strings.TrimSpace(req.OutputProcessedPath) == "" {
|
|
return PolishResult{}, fmt.Errorf("audita output processed path is required")
|
|
}
|
|
if strings.TrimSpace(req.WorkDir) == "" {
|
|
return PolishResult{}, fmt.Errorf("audita work dir is required")
|
|
}
|
|
if r.report && strings.TrimSpace(req.ReportPath) == "" {
|
|
return PolishResult{}, fmt.Errorf("audita report is enabled but report path is missing")
|
|
}
|
|
|
|
reqModules := req.Modules
|
|
if len(reqModules) == 0 {
|
|
reqModules = append([]string(nil), r.modules...)
|
|
}
|
|
args := r.buildArgs(req, reqModules)
|
|
|
|
credentialEnvVar := strings.TrimSpace(r.llmAPIKeyEnv)
|
|
credentialPresent := false
|
|
env := map[string]string{}
|
|
if credentialEnvVar != "" {
|
|
credential, ok := os.LookupEnv(credentialEnvVar)
|
|
if !ok || strings.TrimSpace(credential) == "" {
|
|
return PolishResult{}, fmt.Errorf("audita: required credential environment variable %s is not set", credentialEnvVar)
|
|
}
|
|
env["AUDITA_LLM_API_KEY"] = credential
|
|
credentialPresent = true
|
|
}
|
|
primaryConcurrencyViaEnv := false
|
|
if r.llmConcurrency != nil {
|
|
env["AUDITA_LLM_CONCURRENCY"] = strconv.Itoa(*r.llmConcurrency)
|
|
primaryConcurrencyViaEnv = true
|
|
}
|
|
|
|
if req.GeneratedConfigPath != "" {
|
|
if err := r.writeInvocationConfig(req, args, reqModules, credentialPresent, primaryConcurrencyViaEnv); err != nil {
|
|
return PolishResult{}, fmt.Errorf("write audita invocation config %q: %w", req.GeneratedConfigPath, err)
|
|
}
|
|
}
|
|
|
|
runRes, err := subprocess.Run(ctx, subprocess.RunRequest{
|
|
Executable: r.binary,
|
|
Args: args,
|
|
Timeout: r.timeout,
|
|
EnvOverrides: env,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
})
|
|
if err != nil {
|
|
return r.failureResult(req, reqModules, runRes, credentialPresent, primaryConcurrencyViaEnv), fmt.Errorf(
|
|
"run audita process (binary=%q, stdout_log=%q, stderr_log=%q): %w",
|
|
r.binary,
|
|
req.StdoutLogPath,
|
|
req.StderrLogPath,
|
|
err,
|
|
)
|
|
}
|
|
|
|
if err := validateProcessedOutput(req.OutputProcessedPath); err != nil {
|
|
return r.failureResult(req, reqModules, runRes, credentialPresent, primaryConcurrencyViaEnv), fmt.Errorf("validate audita processed output %q: %w", req.OutputProcessedPath, err)
|
|
}
|
|
if r.report {
|
|
if err := validateJSONFile(req.ReportPath); err != nil {
|
|
return r.failureResult(req, reqModules, runRes, credentialPresent, primaryConcurrencyViaEnv), fmt.Errorf("validate audita report output %q: %w", req.ReportPath, err)
|
|
}
|
|
}
|
|
|
|
return PolishResult{
|
|
ProcessedTranscriptPath: req.OutputProcessedPath,
|
|
ReportPath: req.ReportPath,
|
|
WorkDir: req.WorkDir,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
GeneratedConfigPath: req.GeneratedConfigPath,
|
|
ExitCode: runRes.ExitCode,
|
|
Duration: runRes.Duration,
|
|
InvokedBinary: r.binary,
|
|
Metadata: map[string]any{
|
|
"adapter": "audita_subprocess",
|
|
"modules": reqModules,
|
|
"base_url": r.baseURL,
|
|
"model": r.model,
|
|
"validation_model": r.validationModel,
|
|
"validation_llm_concurrency": r.validationLLMConcurrency,
|
|
"credential_env_var": r.llmAPIKeyEnv,
|
|
"credential_present": credentialPresent,
|
|
"primary_llm_concurrency_via_env": primaryConcurrencyViaEnv,
|
|
"primary_llm_concurrency_env_name": "AUDITA_LLM_CONCURRENCY",
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
func (r *SubprocessRunner) failureResult(req PolishRequest, modules []string, runRes subprocess.RunResult, credentialPresent bool, primaryConcurrencyViaEnv bool) PolishResult {
|
|
return PolishResult{
|
|
ProcessedTranscriptPath: req.OutputProcessedPath,
|
|
ReportPath: req.ReportPath,
|
|
WorkDir: req.WorkDir,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
GeneratedConfigPath: req.GeneratedConfigPath,
|
|
ExitCode: runRes.ExitCode,
|
|
Duration: runRes.Duration,
|
|
InvokedBinary: r.binary,
|
|
Metadata: map[string]any{
|
|
"adapter": "audita_subprocess",
|
|
"modules": modules,
|
|
"base_url": r.baseURL,
|
|
"model": r.model,
|
|
"validation_model": r.validationModel,
|
|
"validation_llm_concurrency": r.validationLLMConcurrency,
|
|
"credential_env_var": r.llmAPIKeyEnv,
|
|
"credential_present": credentialPresent,
|
|
"primary_llm_concurrency_via_env": primaryConcurrencyViaEnv,
|
|
"primary_llm_concurrency_env_name": "AUDITA_LLM_CONCURRENCY",
|
|
},
|
|
}
|
|
}
|
|
|
|
func (r *SubprocessRunner) buildArgs(req PolishRequest, modules []string) []string {
|
|
args := []string{
|
|
"process",
|
|
req.MergedTranscriptPath,
|
|
"--glossary", req.GlossaryPath,
|
|
"--output", req.OutputProcessedPath,
|
|
"--modules", strings.Join(modules, ","),
|
|
"--base-url", r.baseURL,
|
|
"--model", r.model,
|
|
"--work-dir", req.WorkDir,
|
|
}
|
|
if r.report {
|
|
args = append(args, "--report-json", req.ReportPath)
|
|
}
|
|
if r.validationModel != "" {
|
|
args = append(args, "--validation-model", r.validationModel)
|
|
}
|
|
if r.validationLLMConcurrency != nil {
|
|
args = append(args, "--validation-llm-concurrency", strconv.Itoa(*r.validationLLMConcurrency))
|
|
}
|
|
return args
|
|
}
|
|
|
|
func (r *SubprocessRunner) writeInvocationConfig(req PolishRequest, args []string, modules []string, credentialPresent bool, primaryConcurrencyViaEnv bool) error {
|
|
payload := map[string]any{
|
|
"schema": "audita.generated.v1",
|
|
"binary": r.binary,
|
|
"args": args,
|
|
"timeout": r.timeout.String(),
|
|
"modules": modules,
|
|
"base_url": r.baseURL,
|
|
"model": r.model,
|
|
"validation_model": r.validationModel,
|
|
"validation_llm_concurrency": r.validationLLMConcurrency,
|
|
"report_enabled": r.report,
|
|
"merged_transcript_path": req.MergedTranscriptPath,
|
|
"glossary_path": req.GlossaryPath,
|
|
"output_path": req.OutputProcessedPath,
|
|
"report_path": req.ReportPath,
|
|
"work_dir": req.WorkDir,
|
|
"credential_env_var": r.llmAPIKeyEnv,
|
|
"credential_present": credentialPresent,
|
|
"primary_llm_concurrency_via_env": primaryConcurrencyViaEnv,
|
|
}
|
|
if r.llmConcurrency != nil {
|
|
payload["llm_concurrency"] = *r.llmConcurrency
|
|
}
|
|
return subprocess.WriteYAMLAtomic(req.GeneratedConfigPath, payload, 0o644)
|
|
}
|
|
|
|
func validateProcessedOutput(path string) error {
|
|
data, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return fmt.Errorf("read file: %w", err)
|
|
}
|
|
|
|
var payload map[string]any
|
|
if err := json.Unmarshal(data, &payload); err != nil {
|
|
return fmt.Errorf("parse json: %w", err)
|
|
}
|
|
segments, ok := payload["segments"]
|
|
if !ok {
|
|
return fmt.Errorf("top-level segments is required")
|
|
}
|
|
if _, ok := segments.([]any); !ok {
|
|
return fmt.Errorf("top-level segments must be an array")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateJSONFile(path string) error {
|
|
data, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return fmt.Errorf("read file: %w", err)
|
|
}
|
|
var v any
|
|
if err := json.Unmarshal(data, &v); err != nil {
|
|
return fmt.Errorf("parse json: %w", err)
|
|
}
|
|
return nil
|
|
}
|