eric/audita
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Implemented invocation-level failure diagnostics

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-04 12:10:27 -05:00
parent 434f812b98
commit a70f481a93
9 changed files with 501 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
audita
View File

@@ -1,23 +1,134 @@
#!/usr/bin/env python3
import json
import os
from datetime import datetime
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Optional
from uuid import uuid4
DEFAULT_WORK_DIR = "/tmp/audita"
_SECRET_FLAGS = {"--llm-api-key", "--validation-llm-api-key"}
def _redact_argv(argv: list[str]) -> list[str]:
redacted: list[str] = []
index = 0
while index < len(argv):
arg = argv[index]
matched_flag = next((flag for flag in _SECRET_FLAGS if arg == flag or arg.startswith(flag + "=")), None)
if matched_flag is None:
redacted.append(arg)
index += 1
continue
if arg == matched_flag:
redacted.append(arg)
if index + 1 < len(argv):
redacted.append("[REDACTED]")
index += 2
else:
index += 1
continue
redacted.append(f"{matched_flag}=[REDACTED]")
index += 1
return redacted
def _resolve_work_root(argv: list[str]) -> Path:
for index, arg in enumerate(argv):
if arg == "--work-dir" and index + 1 < len(argv):
return Path(argv[index + 1])
if arg.startswith("--work-dir="):
return Path(arg.split("=", 1)[1])
return Path(os.environ.get("AUDITA_WORK_DIR") or DEFAULT_WORK_DIR)
def _create_run_dir(root: Path) -> Path:
root.mkdir(parents=True, exist_ok=True)
timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
run_dir = root / f"run-{timestamp}-{uuid4().hex[:8]}"
run_dir.mkdir(parents=False, exist_ok=False)
return run_dir
def _capture_run_dirs(root: Path) -> set[str]:
if not root.exists():
return set()
return {path.name for path in root.iterdir() if path.is_dir() and path.name.startswith("run-")}
def _find_new_run_dir(root: Path, before: set[str]) -> Optional[Path]:
if not root.exists():
return None
candidates = [
path for path in root.iterdir() if path.is_dir() and path.name.startswith("run-") and path.name not in before
]
if not candidates:
return None
return max(candidates, key=lambda path: path.name)
def _write_launcher_error_log(
path: Path,
*,
message: str,
exit_code: int,
argv: list[str],
command: Optional[list[str]],
) -> None:
payload = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"message": message,
"exit_code": exit_code,
"argv": argv,
"cwd": os.getcwd(),
"command": command,
}
path.write_text(
"Audita Launcher Diagnostics\n" + json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def main() -> int:
argv = list(sys.argv[1:])
work_root = _resolve_work_root(argv)
redacted_argv = _redact_argv(argv)
uv = shutil.which("uv")
if uv is None:
print(
"audita: error: uv is required to run this launcher. Install uv and run `uv sync` in the Audita project.",
file=sys.stderr,
)
run_dir = _create_run_dir(work_root)
error_log = run_dir / "error.log"
message = "uv is required to run this launcher. Install uv and run `uv sync` in the Audita project."
_write_launcher_error_log(error_log, message=message, exit_code=1, argv=redacted_argv, command=None)
print(f"audita: error: {message}", file=sys.stderr)
print(f"audita: exit code: 1", file=sys.stderr)
print(f"audita: run directory: {run_dir}", file=sys.stderr)
print(f"audita: error log: {error_log}", file=sys.stderr)
return 1
project_root = Path(__file__).resolve().parent
command = [uv, "run", "--project", str(project_root), "python", "-m", "audita", *sys.argv[1:]]
os.execv(uv, command)
return 1
before = _capture_run_dirs(work_root)
result = subprocess.run(command, cwd=project_root, check=False)
if result.returncode == 0:
return 0
if _find_new_run_dir(work_root, before) is None:
run_dir = _create_run_dir(work_root)
error_log = run_dir / "error.log"
_write_launcher_error_log(
error_log,
message=f"Audita subprocess exited with status {result.returncode}.",
exit_code=result.returncode,
argv=redacted_argv,
command=_redact_argv(command),
)
print(f"audita: subprocess exited with status {result.returncode}", file=sys.stderr)
print(f"audita: run directory: {run_dir}", file=sys.stderr)
print(f"audita: error log: {error_log}", file=sys.stderr)
return result.returncode
if __name__ == "__main__":

66
src/audita/cli.py
View File

@@ -1,11 +1,14 @@
import argparse
import os
import sys
from pathlib import Path
from typing import Optional, Sequence
from .core.config import AuditaConfig, ConfigOverrides
from .core.diagnostics import build_error_details, create_run_dir, format_stderr_summary, format_traceback_text, redact_argv, write_error_log
from .core.errors import AuditaError
from .core.io import load_glossary, load_transcript, write_report, write_transcript
from .core.reporting import RunReport
from .core.schemas import transcript_to_json
from .pipeline import process_transcript_result
@@ -14,7 +17,7 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
parser = _build_parser()
args = parser.parse_args(argv)
if args.command == "process":
return _process(args)
return _process(args, raw_argv=argv)
parser.print_help(sys.stderr)
return 2
@@ -130,7 +133,14 @@ def _build_parser() -> argparse.ArgumentParser:
return parser
def _process(args: argparse.Namespace) -> int:
def _process(args: argparse.Namespace, raw_argv: Optional[Sequence[str]] = None) -> int:
argv = list(raw_argv) if raw_argv is not None else list(sys.argv[1:])
redacted_argv = redact_argv(argv)
run_dir = create_run_dir(_resolve_work_root(args))
report_path = run_dir / "report.json"
error_log_path = run_dir / "error.log"
config: Optional[AuditaConfig] = None
phase = "config"
try:
config = AuditaConfig.from_sources(
overrides=ConfigOverrides(
@@ -163,13 +173,18 @@ def _process(args: argparse.Namespace) -> int:
work_dir_retention=args.work_dir_retention,
)
)
phase = "transcript_load"
transcript = load_transcript(args.transcript)
phase = "glossary_load"
glossary = load_glossary(args.glossary)
phase = "pipeline"
result = process_transcript_result(
transcript,
glossary,
config,
progress=lambda message: print(message, file=sys.stderr),
run_dir=run_dir,
invocation_details={"argv": redacted_argv, "cwd": os.getcwd()},
)
if args.output is not None:
write_transcript(args.output, result.transcript)
@@ -178,6 +193,49 @@ def _process(args: argparse.Namespace) -> int:
if args.report_json is not None:
write_report(args.report_json, result.report)
return 0
except AuditaError as exc:
print(f"audita: error: {exc}", file=sys.stderr)
except Exception as exc:
error_details = getattr(exc, "audita_error_details", None)
if error_details is None:
traceback_text = format_traceback_text(exc)
error_details = build_error_details(
exc=exc,
exit_code=1,
run_dir=run_dir,
report_path=report_path,
error_log_path=error_log_path,
phase=phase,
module_instance=None,
argv=redacted_argv,
cwd=os.getcwd(),
traceback_text=traceback_text,
)
write_error_log(error_log_path, error_details=error_details, traceback_text=traceback_text)
report_config = config.to_report_dict() if isinstance(config, AuditaConfig) else {}
pipeline = [] if not isinstance(config, AuditaConfig) else list(config.module_keys)
retention = "always" if not isinstance(config, AuditaConfig) else config.work_dir_retention
report = RunReport(
status="failed",
config=report_config,
normalization=None,
pipeline=pipeline,
modules=[],
applied_changes=[],
skipped_corrections=[],
totals={"output_segment_count": 0, "applied_change_count": 0, "skipped_correction_count": 0},
work_dir_retention=retention,
work_dir_retained=True,
work_dir=str(run_dir),
error=str(exc),
error_details=error_details,
)
write_report(report_path, report)
if args.report_json is not None and report_path.exists():
args.report_json.write_text(report_path.read_text(encoding="utf-8"), encoding="utf-8")
print(format_stderr_summary(error_details=error_details), file=sys.stderr)
return 1
def _resolve_work_root(args: argparse.Namespace) -> Path:
if args.work_dir is not None:
return args.work_dir
return Path(os.environ.get("AUDITA_WORK_DIR") or "/tmp/audita")

148
src/audita/core/diagnostics.py Normal file
View File

@@ -0,0 +1,148 @@
from __future__ import annotations
import json
import traceback
from datetime import datetime
from pathlib import Path
import re
from typing import Any, Mapping, Optional, Sequence
from uuid import uuid4
_SECRET_FLAGS = {"--llm-api-key", "--validation-llm-api-key"}
_STAGE_PATTERN = re.compile(r"stage '([^']+)'")
def create_run_dir(root: Path) -> Path:
root.mkdir(parents=True, exist_ok=True)
timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
run_dir = root / f"run-{timestamp}-{uuid4().hex[:8]}"
run_dir.mkdir(parents=False, exist_ok=False)
return run_dir
def redact_argv(argv: Sequence[str]) -> list[str]:
redacted: list[str] = []
index = 0
while index < len(argv):
arg = argv[index]
matched_flag = next((flag for flag in _SECRET_FLAGS if arg == flag or arg.startswith(flag + "=")), None)
if matched_flag is None:
redacted.append(arg)
index += 1
continue
if arg == matched_flag:
redacted.append(arg)
if index + 1 < len(argv):
redacted.append("[REDACTED]")
index += 2
else:
index += 1
continue
redacted.append(f"{matched_flag}=[REDACTED]")
index += 1
return redacted
def format_traceback_text(exc: BaseException) -> str:
return "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
def extract_stage_name(message: str) -> Optional[str]:
match = _STAGE_PATTERN.search(message)
return match.group(1) if match is not None else None
def build_error_details(
*,
exc: BaseException,
exit_code: int,
run_dir: Path,
report_path: Path,
error_log_path: Path,
phase: str,
module_instance: Optional[str],
argv: Optional[Sequence[str]] = None,
cwd: Optional[str] = None,
traceback_text: Optional[str] = None,
) -> dict[str, Any]:
message = str(exc)
stage_name = extract_stage_name(message)
excerpt_source = traceback_text or format_traceback_text(exc)
excerpt_lines = excerpt_source.strip().splitlines()
traceback_excerpt = "\n".join(excerpt_lines[-12:]) if excerpt_lines else ""
details: dict[str, Any] = {
"type": type(exc).__name__,
"message": message,
"exit_code": exit_code,
"phase": phase,
"stage": stage_name,
"module_instance": module_instance,
"run_dir": str(run_dir),
"report_path": str(report_path),
"error_log": str(error_log_path),
"traceback_excerpt": traceback_excerpt,
}
if argv is not None:
details["argv"] = list(argv)
if cwd is not None:
details["cwd"] = cwd
return details
def write_error_log(
path: Path,
*,
error_details: Mapping[str, Any],
traceback_text: str,
) -> None:
lines = [
"Audita Error Diagnostics",
f"timestamp: {datetime.utcnow().isoformat()}Z",
f"type: {error_details.get('type')}",
f"message: {error_details.get('message')}",
f"exit_code: {error_details.get('exit_code')}",
f"phase: {error_details.get('phase')}",
f"stage: {error_details.get('stage') or ''}",
f"module_instance: {error_details.get('module_instance') or ''}",
f"run_dir: {error_details.get('run_dir')}",
f"report_path: {error_details.get('report_path')}",
f"error_log: {error_details.get('error_log')}",
]
argv = error_details.get("argv")
if argv is not None:
lines.append(f"argv: {json.dumps(argv, ensure_ascii=False)}")
cwd = error_details.get("cwd")
if cwd is not None:
lines.append(f"cwd: {cwd}")
lines.extend(
[
"",
"Traceback:",
traceback_text.rstrip(),
"",
]
)
path.write_text("\n".join(lines), encoding="utf-8")
def format_stderr_summary(*, error_details: Mapping[str, Any]) -> str:
lines = [
f"audita: error: {error_details.get('message')}",
f"audita: exit code: {error_details.get('exit_code')}",
f"audita: phase: {error_details.get('phase')}",
]
stage = error_details.get("stage")
if stage:
lines.append(f"audita: stage: {stage}")
module_instance = error_details.get("module_instance")
if module_instance:
lines.append(f"audita: module: {module_instance}")
lines.extend(
[
f"audita: run directory: {error_details.get('run_dir')}",
f"audita: error log: {error_details.get('error_log')}",
f"audita: report: {error_details.get('report_path')}",
]
)
return "\n".join(lines)

2
src/audita/core/reporting.py
View File

@@ -91,6 +91,7 @@ class RunReport:
work_dir_retained: bool
work_dir: Optional[str]
error: Optional[str] = None
error_details: Optional[dict] = None
def to_dict(self) -> dict:
return {
@@ -106,6 +107,7 @@ class RunReport:
"work_dir_retained": self.work_dir_retained,
"work_dir": self.work_dir,
"error": self.error,
"error_details": self.error_details,
}
def to_json(self) -> str:

4
src/audita/framework/runner.py
View File

@@ -50,6 +50,7 @@ class PipelineRunError(Exception):
applied_changes: List[AppliedChange],
skipped_corrections: List[ReportedSkip],
cause: Exception,
module_instance: Optional[str],
) -> None:
super().__init__(str(cause))
self.transcript = transcript
@@ -57,6 +58,7 @@ class PipelineRunError(Exception):
self.applied_changes = applied_changes
self.skipped_corrections = skipped_corrections
self.cause = cause
self.module_instance = module_instance
class PipelineRunner:
@@ -114,6 +116,7 @@ class PipelineRunner:
applied_changes=[*applied_changes, *exc.applied_changes],
skipped_corrections=[*skipped_corrections, *exc.skipped_corrections],
cause=exc.cause,
module_instance=run_spec.instance_name,
) from exc
except Exception as exc:
raise PipelineRunError(
@@ -122,6 +125,7 @@ class PipelineRunner:
applied_changes=list(applied_changes),
skipped_corrections=list(skipped_corrections),
cause=exc,
module_instance=run_spec.instance_name,
) from exc
working = result.transcript
module_reports.append(result.module_report)

64
src/audita/pipeline.py
View File

@@ -1,11 +1,10 @@
import json
import shutil
from datetime import datetime
from pathlib import Path
from typing import Callable, List, Optional, Sequence
from uuid import uuid4
from typing import Any, Callable, List, Optional, Sequence
from .core.config import AuditaConfig
from .core.diagnostics import build_error_details, create_run_dir, format_traceback_text, write_error_log
from .core.errors import AuditaError
from .core.normalization import normalize_transcript
from .core.reporting import AppliedChange, ModuleRunReport, ProcessResult, ReportedSkip, RunReport
@@ -44,12 +43,16 @@ def process_transcript_result(
module_keys: Optional[Sequence[str]] = None,
llm_client: Optional[StructuredLLMClient] = None,
progress: Optional[ProgressCallback] = None,
run_dir: Optional[Path] = None,
invocation_details: Optional[dict[str, Any]] = None,
) -> ProcessResult:
run_dir = _create_run_dir(config.work_dir)
run_dir = create_run_dir(config.work_dir) if run_dir is None else run_dir
module_specs = resolve_module_specs(config.module_keys if module_keys is None else module_keys)
normalized_transcript: List[TranscriptSegment] = []
normalization_summary: Optional[dict] = None
report: Optional[RunReport] = None
report_path = run_dir / "report.json"
error_log_path = run_dir / "error.log"
try:
_log(progress, f"Created work directory {run_dir}")
normalization_result = normalize_transcript(
@@ -95,8 +98,9 @@ def process_transcript_result(
work_dir_retained=work_dir_retained,
work_dir=str(run_dir) if work_dir_retained else None,
error=None,
error_details=None,
)
_write_run_report(run_dir / "report.json", report)
_write_run_report(report_path, report)
if work_dir_retained:
_log(progress, f"Work directory preserved at {run_dir}")
else:
@@ -110,6 +114,21 @@ def process_transcript_result(
)
except Exception as exc:
failed_report_data = _failure_report_data(exc, normalized_transcript)
root_error = failed_report_data["error"]
traceback_text = format_traceback_text(root_error)
error_details = build_error_details(
exc=root_error,
exit_code=1,
run_dir=run_dir,
report_path=report_path,
error_log_path=error_log_path,
phase=failed_report_data["phase"],
module_instance=failed_report_data["module_instance"],
argv=None if invocation_details is None else invocation_details.get("argv"),
cwd=None if invocation_details is None else invocation_details.get("cwd"),
traceback_text=traceback_text,
)
write_error_log(error_log_path, error_details=error_details, traceback_text=traceback_text)
report = _build_run_report(
status="failed",
config=config.to_report_dict(),
@@ -122,21 +141,24 @@ def process_transcript_result(
work_dir_retention=config.work_dir_retention,
work_dir_retained=True,
work_dir=str(run_dir),
error=str(failed_report_data["error"]),
error=str(root_error),
error_details=error_details,
)
_write_run_report(run_dir / "report.json", report)
error = failed_report_data["error"]
_write_run_report(report_path, report)
error = root_error
if isinstance(error, AuditaError):
raise type(error)(f"{error} Diagnostics preserved at {run_dir}") from error
raise AuditaError(f"{error} Diagnostics preserved at {run_dir}") from error
def _create_run_dir(root: Path) -> Path:
root.mkdir(parents=True, exist_ok=True)
timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
run_dir = root / f"run-{timestamp}-{uuid4().hex[:8]}"
run_dir.mkdir(parents=False, exist_ok=False)
return run_dir
reraised = type(error)(f"{error} Diagnostics preserved at {run_dir}")
setattr(reraised, "audita_run_dir", run_dir)
setattr(reraised, "audita_report_path", report_path)
setattr(reraised, "audita_error_log_path", error_log_path)
setattr(reraised, "audita_error_details", error_details)
raise reraised from error
reraised = AuditaError(f"{error} Diagnostics preserved at {run_dir}")
setattr(reraised, "audita_run_dir", run_dir)
setattr(reraised, "audita_report_path", report_path)
setattr(reraised, "audita_error_log_path", error_log_path)
setattr(reraised, "audita_error_details", error_details)
raise reraised from error
def _should_retain_run_dir(retention: str, has_skips: bool) -> bool:
@@ -188,6 +210,7 @@ def _build_run_report(
work_dir_retained: bool,
work_dir: Optional[str],
error: Optional[str],
error_details: Optional[dict],
) -> RunReport:
return RunReport(
status=status,
@@ -206,6 +229,7 @@ def _build_run_report(
work_dir_retained=work_dir_retained,
work_dir=work_dir,
error=error,
error_details=error_details,
)
@@ -220,6 +244,8 @@ def _failure_report_data(
"skipped_corrections": exc.skipped_corrections,
"transcript": _sort_transcript_chronologically(exc.transcript),
"error": exc.cause,
"phase": "pipeline",
"module_instance": exc.module_instance,
}
return {
"modules": [],
@@ -227,4 +253,6 @@ def _failure_report_data(
"skipped_corrections": [],
"transcript": _sort_transcript_chronologically(normalized_transcript),
"error": exc,
"phase": "normalization",
"module_instance": None,
}

68
tests/test_new_cli.py
View File

@@ -1,6 +1,8 @@
import json
import pytest
from audita.cli import main
from audita.core.errors import AuditaConfigError
from audita.core.reporting import ProcessResult, RunReport
from audita.core.schemas import parse_transcript_json
@@ -501,3 +503,69 @@ def test_cli_process_passes_min_section_tokens_override_to_config(monkeypatch, t
assert exit_code == 0
assert captured["min_section_tokens"] == 5000
def test_cli_process_writes_failure_diagnostics_and_keeps_stdout_empty(monkeypatch, tmp_path, capsys):
monkeypatch.setattr(
"audita.cli.AuditaConfig.from_sources",
lambda overrides=None: (_ for _ in ()).throw(AuditaConfigError("bad config")),
)
report_path = tmp_path / "external-report.json"
exit_code = main(
[
"process",
"transcript.json",
"--glossary",
"glossary.yaml",
"--work-dir",
str(tmp_path / "work"),
"--report-json",
str(report_path),
]
)
captured = capsys.readouterr()
assert exit_code == 1
assert captured.out == ""
assert "audita: error: bad config" in captured.err
assert "audita: exit code: 1" in captured.err
run_dir = next((tmp_path / "work").iterdir())
assert f"audita: run directory: {run_dir}" in captured.err
assert f"audita: error log: {run_dir / 'error.log'}" in captured.err
assert f"audita: report: {run_dir / 'report.json'}" in captured.err
assert (run_dir / "error.log").exists()
report = json.loads((run_dir / "report.json").read_text(encoding="utf-8"))
assert report["status"] == "failed"
assert report["error"] == "bad config"
assert report["error_details"]["phase"] == "config"
assert report["error_details"]["type"] == "AuditaConfigError"
external = json.loads(report_path.read_text(encoding="utf-8"))
assert external["error"] == "bad config"
def test_cli_process_writes_failure_diagnostics_for_unexpected_exceptions(monkeypatch, tmp_path, capsys):
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", lambda overrides=None: object())
monkeypatch.setattr("audita.cli.load_transcript", lambda path: (_ for _ in ()).throw(RuntimeError("boom")))
exit_code = main(
[
"process",
"transcript.json",
"--glossary",
"glossary.yaml",
"--work-dir",
str(tmp_path / "work"),
]
)
captured = capsys.readouterr()
assert exit_code == 1
assert captured.out == ""
assert "audita: error: boom" in captured.err
run_dir = next((tmp_path / "work").iterdir())
report = json.loads((run_dir / "report.json").read_text(encoding="utf-8"))
assert report["status"] == "failed"
assert report["error_details"]["phase"] == "transcript_load"
assert report["error_details"]["type"] == "RuntimeError"
assert "RuntimeError: boom" in (run_dir / "error.log").read_text(encoding="utf-8")

44
tests/test_new_launcher.py
View File

@@ -1,6 +1,7 @@
import os
import shutil
import subprocess
import sys
from pathlib import Path
import pytest
@@ -29,3 +30,46 @@ def test_root_launcher_help_smoke():
assert result.returncode == 0
assert result.stdout.startswith("usage: audita ")
def test_root_launcher_writes_error_log_when_uv_is_missing(tmp_path):
work_dir = tmp_path / "work"
result = subprocess.run(
[sys.executable, str(LAUNCHER), "process", "transcript.json", "--glossary", "glossary.yaml", "--work-dir", str(work_dir)],
cwd=ROOT,
text=True,
capture_output=True,
env={**os.environ, "PATH": ""},
check=False,
)
assert result.returncode == 1
assert result.stdout == ""
assert "audita: error: uv is required to run this launcher" in result.stderr
run_dir = next(work_dir.iterdir())
assert f"audita: run directory: {run_dir}" in result.stderr
assert (run_dir / "error.log").exists()
def test_root_launcher_preserves_child_exit_code_and_writes_fallback_error_log(tmp_path):
work_dir = tmp_path / "work"
fake_bin = tmp_path / "bin"
fake_bin.mkdir()
fake_uv = fake_bin / "uv"
fake_uv.write_text("#!/bin/sh\nexit 120\n", encoding="utf-8")
fake_uv.chmod(0o755)
result = subprocess.run(
[sys.executable, str(LAUNCHER), "process", "transcript.json", "--glossary", "glossary.yaml", "--work-dir", str(work_dir)],
cwd=ROOT,
text=True,
capture_output=True,
env={**os.environ, "PATH": str(fake_bin)},
check=False,
)
assert result.returncode == 120
assert result.stdout == ""
assert "audita: subprocess exited with status 120" in result.stderr
run_dir = next(work_dir.iterdir())
assert (run_dir / "error.log").exists()

10
tests/test_new_pipeline.py
View File

@@ -414,6 +414,10 @@ def test_process_transcript_result_missing_api_key_writes_failed_report(tmp_path
assert "AUDITA_LLM_API_KEY" in report["error"]
assert "OPENROUTER_API_KEY" in report["error"]
assert "OpenRouter endpoint" in report["error"]
assert report["error_details"]["type"] == "AuditaLLMError"
assert report["error_details"]["phase"] == "pipeline"
assert report["error_details"]["error_log"] == str(run_dir / "error.log")
assert (run_dir / "error.log").exists()
def test_process_transcript_result_allows_missing_api_key_for_nondefault_proposal_endpoint(tmp_path):
@@ -590,6 +594,9 @@ def test_process_transcript_result_preserves_partial_progress_when_later_module_
assert report["skipped_corrections"] == []
assert report["pipeline"][1] == "homophones"
assert "Simulated homophones proposal failure." in report["error"]
assert report["error_details"]["module_instance"] == "homophones"
assert report["error_details"]["phase"] == "pipeline"
assert (run_dir / "error.log").exists()
def test_process_transcript_result_preserves_partial_skips_and_validator_diagnostics_on_failure(tmp_path):
@@ -658,6 +665,9 @@ def test_process_transcript_result_preserves_partial_skips_and_validator_diagnos
assert report["skipped_corrections"][0]["reason"] == "proposal confidence below threshold"
assert report["skipped_corrections"][0]["source"] == "validator:proposal_confidence_guard"
assert "unknown correction_index" in report["error"]
assert report["error_details"]["module_instance"] == "glossary_1"
assert report["error_details"]["phase"] == "pipeline"
assert (run_dir / "error.log").exists()
assert (validator_dir / "spoken_form_plausibility_review-prompt-0000.json").exists()
assert (validator_dir / "spoken_form_plausibility_review-response-0000.json").exists()