Handling Semantic Failures

# script_id: day_017_semantic_failures/atlas_base
# atlas_base.py: compact copy of Day 15 (schemas, data), Day 12 (client, cost) and Day 16 (extractor).
# Install: uv add "openai>=3.46,<4" pydantic
# Env: ORBIT_DATA (optional). OPENAI_API_KEY only when a call is made: the client is built lazily.
import json
import os
from functools import cache
from pathlib import Path
from typing import Literal

from openai import OpenAI
from openai.types.responses import ParsedResponse
from pydantic import BaseModel, ConfigDict, Field, ValidationError

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))

IssueType = Literal["bug", "feature", "question", "docs"]
Severity = Literal["S1", "S2", "S3", "S4"]
Component = Literal["core", "scheduler", "flows", "broker-redis", "broker-postgres", "broker-sqs",
                    "cli", "dashboard", "retries", "results", "observability", "auth", "docs"]


class Comment(BaseModel):
    author: str
    body: str


class Issue(BaseModel):
    """One row of issues/issues.jsonl, trimmed to the fields used today (extra keys are ignored)."""
    id: int
    title: str
    body: str
    labels: list[str]
    comments: list[Comment]

# script_id: day_017_semantic_failures/atlas_base
class IssueRecord(BaseModel):
    """The facts Atlas extracts from one issue. Fields mirror eval/extraction_gold.jsonl."""
    model_config = ConfigDict(extra="forbid")

    type: IssueType = Field(description="bug, feature, question or docs")
    components: list[Component] = Field(description="Every Orbit component the issue is about")
    severity: Severity | None = Field(description="Bugs only: S1 outage or data loss, S2 major "
                                      "breakage, S3 has a workaround, S4 cosmetic. null otherwise")
    orbit_version: str | None = Field(description="Full Orbit version, e.g. 3.1.2. null when not stated")
    python_version: str | None = Field(description="Python version as written, e.g. 3.12.6. "
                                       "null when not stated")
    error_codes: list[str] = Field(description="ORB-NNNN codes quoted in the issue or its comments")
    summary: str = Field(description="One or two English sentences: symptom, trigger, version")


def _jsonl(path: Path) -> list[dict]:
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]


def load_issues() -> list[Issue]:
    return [Issue.model_validate(row) for row in _jsonl(DATA_DIR / "issues" / "issues.jsonl")]


def load_gold() -> dict[int, IssueRecord]:
    rows = _jsonl(DATA_DIR / "eval" / "extraction_gold.jsonl")
    return {row.pop("issue_id"): IssueRecord.model_validate(row) for row in rows}


# USD per 1M tokens, as of 2026-11; verify at https://developers.openai.com/api/docs/pricing
PRICES_PER_1M = {"gpt-6-luna": {"input": 0.10, "cached_input": 0.01, "output": 0.50},
                 "gpt-6.1-terra": {"input": 1.00, "cached_input": 0.05, "output": 5.00}}


@cache
def get_client() -> OpenAI:
    # Tier 1 of two-tier retries: the SDK retries 408/409/429/5xx and connection errors itself.
    return OpenAI(max_retries=2, timeout=30.0)


def cost_from_usage(usage, model: str = MODEL) -> float:
    # Approximation: ignores the cache-write surcharge, which only matters for prefixes over 1,024 tokens (Day 18).
    p, cached = PRICES_PER_1M[model], usage.input_tokens_details.cached_tokens
    return ((usage.input_tokens - cached) * p["input"] + cached * p["cached_input"]
            + usage.output_tokens * p["output"]) / 1_000_000   # reasoning is inside output_tokens

# script_id: day_017_semantic_failures/atlas_base
EXTRACT_INSTRUCTIONS = """You extract facts from GitHub issues filed against Orbit, an open-source
Python background-jobs library. Fill every field of the schema from the issue text only.

Rules:
- Use null when a fact is not stated. Never guess a version from dates or context.
- severity is for bugs only; use null for features, questions and docs.
- error_codes: every ORB-NNNN code that appears anywhere, including logs and comments. [] if none.
- components: usually one; list two only when the issue really spans both.
- summary: one or two sentences in English, even when the issue is in another language.

Component glossary:
core = queues, workers, leases, job execution | scheduler = cron schedules and delays
flows = Flows, steps and step state | broker-redis, broker-postgres, broker-sqs = that broker
cli = the orbit command | dashboard = the web UI | retries = retry policy and backoff
results = the result backend | observability = metrics and tracing
auth = dashboard login and OIDC | docs = the documentation itself"""


def issue_to_input(issue: Issue) -> str:
    """Title, body and comments; never the labels. (Day 16 also truncates very long issues.)"""
    parts = [f"# {issue.title}", issue.body]
    parts += [f"--- comment by {c.author} ---\n{c.body}" for c in issue.comments]
    return "\n\n".join(parts)


class ExtractionError(Exception):
    """The API answered, but not with a usable record."""

    def __init__(self, reason: Literal["refusal", "incomplete"], response: ParsedResponse | None = None,
                 detail: str = "", cost_usd: float = 0.0) -> None:
        super().__init__(f"{reason}: {detail}")
        self.reason, self.response, self.cost_usd = reason, response, cost_usd

# script_id: day_017_semantic_failures/atlas_base
def parse_issue(items: list, *, model: str = MODEL, effort: str = "low",
                max_output_tokens: int = 25_000) -> tuple[ParsedResponse, float]:
    """Day 16's extract_issue(), refactored to take the input list and return (response, cost)."""
    try:
        resp = get_client().responses.parse(
            model=model, instructions=EXTRACT_INSTRUCTIONS, input=items, text_format=IssueRecord,
            reasoning={"effort": effort}, max_output_tokens=max_output_tokens)
    except ValidationError as err:   # strict mode fixes the shape, so this means cut-off JSON
        raise ExtractionError("incomplete", detail="truncated JSON") from err   # cost unknown
    cost = cost_from_usage(resp.usage, model)
    if resp.status == "incomplete":
        raise ExtractionError("incomplete", resp, str(resp.incomplete_details.reason), cost)
    refusals = [part.refusal for item in resp.output if item.type == "message"
                for part in item.content if part.type == "refusal"]
    if refusals:
        raise ExtractionError("refusal", resp, refusals[0], cost)
    if resp.output_parsed is None:
        raise ExtractionError("incomplete", resp, "no final answer", cost)
    return resp, cost

# script_id: day_017_semantic_failures/semantic_checks
# semantic_checks.py: three deterministic rules that compare a record with its source issue. Offline, $0.
# Install: uv add "openai>=3.46,<4" pydantic · Env: ORBIT_DATA (optional)
import re
from collections.abc import Callable, Sequence

from atlas_base import DATA_DIR, Issue, IssueRecord, load_issues

KNOWN_CODES = frozenset(re.findall(r"ORB-\d{4}", (DATA_DIR / "docs" / "error-codes.md").read_text("utf-8")))
Rule = Callable[[IssueRecord, str], list[str]]      # (record, issue text) -> error messages


def issue_text(issue: Issue) -> str:
    return "\n".join([issue.title, issue.body, *(c.body for c in issue.comments)])


def codes_in_text(rec: IssueRecord, text: str) -> list[str]:
    return [f"error_codes: {c} is not in the issue text" for c in rec.error_codes if c not in text]


def codes_known(rec: IssueRecord, text: str) -> list[str]:
    return [f"error_codes: {c} is not an Orbit error code" for c in rec.error_codes if c not in KNOWN_CODES]


def severity_matches_type(rec: IssueRecord, text: str) -> list[str]:
    if (rec.severity is None) == (rec.type == "bug"):
        return [f"severity: {rec.severity} on a {rec.type}; set it for bugs only, null otherwise"]
    return []


BASIC_RULES: list[Rule] = [codes_in_text, codes_known, severity_matches_type]


def semantic_errors(record: IssueRecord, issue: Issue, rules: Sequence[Rule] | None = None) -> list[str]:
    text = issue_text(issue)
    return [msg for rule in (rules or BASIC_RULES) for msg in rule(record, text)]


if __name__ == "__main__":
    issue = next(i for i in load_issues() if i.id == 113)
    bad = IssueRecord(type="question", components=["core", "cli"], severity="S3", orbit_version="3.0.1",
                      python_version="3.11.8", error_codes=["ORB-1102"], summary="orbit doctor fails.")
    print(f"{len(KNOWN_CODES)} known codes")
    print(semantic_errors(bad, issue))

20 known codes
['error_codes: ORB-1102 is not in the issue text', 'severity: S3 on a question; set it for bugs only, null otherwise']

# script_id: day_017_semantic_failures/reask_minimal
# One re-ask round on issue #113: extract, check, feed the errors back, extract again. ≈ $0.001.
# Install: uv add "openai>=3.46,<4" pydantic · Env: OPENAI_API_KEY, ORBIT_DATA
from atlas_base import issue_to_input, load_issues, parse_issue
from semantic_checks import semantic_errors

issue = next(i for i in load_issues() if i.id == 113)
items = [{"role": "user", "content": issue_to_input(issue)}]
resp, cost = parse_issue(items)
errors = semantic_errors(resp.output_parsed, issue)
print("attempt 1:", errors)

if errors:
    feedback = ("Your record breaks these rules:\n" + "\n".join(f"- {e}" for e in errors)
                + "\nReturn the full corrected record and change only those fields. "
                "If the issue doesn't state a fact, use null (or [] for lists). Never guess.")
    items = [*items, *resp.output, {"role": "user", "content": feedback}]   # Day 8 history pattern
    resp, cost2 = parse_issue(items)
    print("attempt 2:", semantic_errors(resp.output_parsed, issue), f"· ${cost + cost2:.5f}")

attempt 1: ['error_codes: ORB-1102 is not in the issue text']
attempt 2: [] · $0.00083

✘ "Try again."                                    -> same answer, double cost
✘ "orbit_version is required."                    -> fabricated 3.2.1
✔ "orbit_version: 3.2.1 does not appear in the issue; use null when the version isn't stated"

# script_id: day_017_semantic_failures/rules
# rules.py: the full rule set (7 rules) plus a sanity check against gold. Offline, $0.
# Install: uv add "openai>=3.46,<4" pydantic · Env: ORBIT_DATA (optional)
import re

from collections.abc import Sequence

from atlas_base import DATA_DIR, Issue, IssueRecord, load_gold, load_issues
from semantic_checks import BASIC_RULES, Rule
import semantic_checks as basic

_CHANGELOG = (DATA_DIR / "docs" / "changelog.md").read_text("utf-8")
CHANGELOG_VERSIONS = frozenset(re.findall(r"^## (\d+\.\d+\.\d+) ", _CHANGELOG, re.M))
KNOWN_VERSIONS = CHANGELOG_VERSIONS | {"3.0.1"}   # 3.0.1 shipped without its own entry; issues cite it


def mentioned(value: str, text: str) -> bool:
    return re.search(rf"(?<![\d.]){re.escape(value)}(?!\d)", text) is not None


def version_rule(known: frozenset[str]) -> Rule:
    def orbit_version_ok(rec: IssueRecord, text: str) -> list[str]:
        v = rec.orbit_version
        if v is not None and v not in known:
            return [f"orbit_version: {v} is not a known Orbit release; use null if unsure"]
        if v is not None and not mentioned(v, text):
            return [f"orbit_version: {v} does not appear in the issue; use null when it isn't stated"]
        return []
    return orbit_version_ok

# script_id: day_017_semantic_failures/rules
def python_version_stated(rec: IssueRecord, text: str) -> list[str]:
    v = rec.python_version
    return [] if v is None or mentioned(v, text) else [f"python_version: {v} does not appear in the issue; use null"]


def components_present(rec: IssueRecord, text: str) -> list[str]:
    if rec.type in ("bug", "feature") and not rec.components:
        return [f"components: empty on a {rec.type}; list the component(s) the issue is about"]
    return []


def summary_short(rec: IssueRecord, text: str) -> list[str]:
    return [f"summary: {len(rec.summary)} characters; keep it under 300"] if len(rec.summary) > 300 else []


RULES: list[Rule] = [*BASIC_RULES, version_rule(KNOWN_VERSIONS), python_version_stated,
                     components_present, summary_short]


def semantic_errors(record: IssueRecord, issue: Issue, rules: Sequence[Rule] | None = None) -> list[str]:
    """All seven rules by default; pass `rules` to check a subset or a custom list."""
    return basic.semantic_errors(record, issue, rules or RULES)


if __name__ == "__main__":
    gold, issues = load_gold(), {i.id: i for i in load_issues()}
    for label, rules in [("changelog versions only", [*BASIC_RULES, version_rule(CHANGELOG_VERSIONS)]),
                         ("full RULES", RULES)]:
        flagged = {iid: errs for iid, rec in gold.items() if (errs := semantic_errors(rec, issues[iid], rules))}
        print(f"{label}: {len(flagged)}/{len(gold)} gold records flagged {flagged}")

changelog versions only: 1/30 gold records flagged {113: ['orbit_version: 3.0.1 is not a known Orbit release; use null if unsure']}
full RULES: 0/30 gold records flagged {}

# script_id: day_017_semantic_failures/reliable_extract
# reliable_extract.py: the bounded re-ask loop for atlas/extract.py. ≈ $0.0005-0.002 per issue on gpt-6-luna.
# Install: uv add "openai>=3.46,<4" pydantic · Env: OPENAI_API_KEY, ORBIT_DATA, ATLAS_MODEL (optional)
from collections.abc import Callable, Sequence
from dataclasses import dataclass, field

from atlas_base import MODEL, ExtractionError, Issue, IssueRecord, issue_to_input, load_issues, parse_issue
from rules import RULES, Rule, semantic_errors

ParseFn = Callable[..., tuple]   # parse_issue() in production; a fake with the same shape in tests


@dataclass
class ExtractResult:
    record: IssueRecord | None          # best record seen, even when needs_review is True
    attempts: int                       # API calls made
    errors: list[str]                   # errors still open on `record`
    cost_usd: float
    needs_review: bool                  # True -> send to the review queue
    stop_reason: str = "ok"             # ok | max_attempts | max_usd | refusal | incomplete
    history: list[dict] = field(default_factory=list)   # [{"errors": [...], "cost_usd": x}, ...]


def build_feedback(errors: list[str]) -> str:
    """Name each field, its bad value and the rule, and always allow null."""
    lines = "\n".join(f"- {e}" for e in errors)
    return (f"Your record breaks these rules:\n{lines}\nReturn the full corrected record and change "
            "only those fields. If the issue doesn't state a fact, use null (or [] for lists). Never guess.")

# script_id: day_017_semantic_failures/reliable_extract
def reliable_extract(issue: Issue, *, max_attempts: int = 3, max_usd: float = 0.01, model: str = MODEL,
                     effort: str = "low", rules: Sequence[Rule] = RULES, parse: ParseFn = parse_issue) -> ExtractResult:
    """Extract, check, re-ask with feedback. Never raises for a bad answer, never loops forever."""
    items: list = [{"role": "user", "content": issue_to_input(issue)}]
    best, best_errors, spent, history = None, [], 0.0, []
    max_tokens, stop = 25_000, "max_attempts"
    for _ in range(max_attempts):
        if spent >= max_usd:
            stop = "max_usd"
            break
        try:
            resp, cost = parse(items, model=model, effort=effort, max_output_tokens=max_tokens)
        except ExtractionError as err:
            spent += err.cost_usd
            history.append({"errors": [str(err)], "cost_usd": err.cost_usd})
            stop = err.reason
            if err.reason == "refusal":
                break                   # a re-ask pays twice for the same "no"
            max_tokens = 50_000         # incomplete: same input, more room
            continue
        spent += cost
        errors = semantic_errors(resp.output_parsed, issue, rules)
        history.append({"errors": errors, "cost_usd": cost})
        if best is None or len(errors) < len(best_errors):
            best, best_errors = resp.output_parsed, errors
        if not errors:
            stop = "ok"
            break
        stop = "max_attempts"
        items = [*items, *resp.output, {"role": "user", "content": build_feedback(errors)}]
    return ExtractResult(record=best, attempts=len(history), errors=best_errors, cost_usd=spent,
                         needs_review=stop != "ok", stop_reason=stop, history=history)


if __name__ == "__main__":
    issue = next(i for i in load_issues() if i.id == 113)
    r = reliable_extract(issue)
    print(f"#{issue.id} attempts={r.attempts} needs_review={r.needs_review} "
          f"stop={r.stop_reason} cost=${r.cost_usd:.5f} errors={r.errors}")

#113 attempts=2 needs_review=False stop=ok cost=$0.00082 errors=[]

# script_id: day_017_semantic_failures/loop_offline
# loop_offline.py: drive reliable_extract() with a scripted fake model, so every exit runs offline. $0.
# Install: uv add "openai>=3.46,<4" pydantic · Env: ORBIT_DATA (optional). No API key needed.
from types import SimpleNamespace

from atlas_base import ExtractionError, IssueRecord, load_gold, load_issues
from reliable_extract import reliable_extract


def scripted(*replies: IssueRecord | str, cost: float = 0.0004):
    """Stands in for parse_issue(): returns replies in order; the string "refusal" raises like a refusal."""
    queue, seen = list(replies), []

    def fake_parse(items: list, **_: object) -> tuple:
        seen.append(len(items))                  # the input grows by output + feedback on each re-ask
        reply = queue.pop(0)
        if reply == "refusal":
            raise ExtractionError("refusal", detail="I can't help with that.", cost_usd=cost)
        message = {"role": "assistant", "content": reply.model_dump_json()}   # stands in for resp.output
        return SimpleNamespace(output=[message], output_parsed=reply), cost

    fake_parse.seen = seen
    return fake_parse


issue = next(i for i in load_issues() if i.id == 113)
good = load_gold()[113]
bad = good.model_copy(update={"error_codes": ["ORB-1102"], "severity": "S3"})
cases = [("fixed on re-ask", scripted(bad, good), {}),
         ("never fixed", scripted(bad, bad, bad), {}),
         ("refusal", scripted("refusal"), {}),
         ("over budget", scripted(bad, bad, bad, cost=0.006), {"max_usd": 0.01})]
for label, fake, kwargs in cases:
    r = reliable_extract(issue, parse=fake, **kwargs)
    print(f"{label:<16} attempts={r.attempts} stop={r.stop_reason:<12} needs_review={r.needs_review!s:<5} "
          f"cost=${r.cost_usd:.4f} input items per call={fake.seen}")

fixed on re-ask  attempts=2 stop=ok           needs_review=False cost=$0.0008 input items per call=[1, 3]
never fixed      attempts=3 stop=max_attempts needs_review=True  cost=$0.0012 input items per call=[1, 3, 5]
refusal          attempts=1 stop=refusal      needs_review=True  cost=$0.0004 input items per call=[1]
over budget      attempts=2 stop=max_usd      needs_review=True  cost=$0.0120 input items per call=[1, 3]

# script_id: day_017_semantic_failures/attempt_math
# Worst-case requests and wall time for one issue, right vs wrong nesting. Offline, $0. Stdlib only.
from math import prod

SDK_TRIES = 1 + 2          # max_retries=2 on the client: the first try plus 2 retries
SEMANTIC_ATTEMPTS = 3      # reliable_extract(max_attempts=3)
OUTER_WRAPPER = 3          # a generic "retry 3 times" decorator someone adds around everything
BACKOFF_CAP_S = 8          # SDK backoff grows from 0.5 s to 8 s (a retry-after header can ask for more)


def worst_case(timeout_s: float, *layers: int) -> tuple[int, float]:
    requests = prod(layers)
    logical_calls = requests // SDK_TRIES
    return requests, requests * timeout_s + logical_calls * (SDK_TRIES - 1) * BACKOFF_CAP_S


for label, layers in [("right: semantic x SDK", (SEMANTIC_ATTEMPTS, SDK_TRIES)),
                      ("wrong: wrapper x semantic x SDK", (OUTER_WRAPPER, SEMANTIC_ATTEMPTS, SDK_TRIES))]:
    for timeout in (30, 600):                 # 600 s is the SDK's default timeout
        n, secs = worst_case(timeout, *layers)
        print(f"{label:<32} timeout={timeout:>3}s  {n:>2} requests  up to {secs / 60:5.1f} min")

right: semantic x SDK            timeout= 30s   9 requests  up to   5.3 min
right: semantic x SDK            timeout=600s   9 requests  up to  90.8 min
wrong: wrapper x semantic x SDK  timeout= 30s  27 requests  up to  15.9 min
wrong: wrapper x semantic x SDK  timeout=600s  27 requests  up to 272.4 min

# script_id: day_017_semantic_failures/measure
# R1 scoreboard: semantic errors before vs after re-ask on the 30 gold issues. ≈ $0.02.
# Install: uv add "openai>=3.46,<4" pydantic · Env: OPENAI_API_KEY, ORBIT_DATA
from collections import Counter

from atlas_base import load_gold, load_issues
from reliable_extract import reliable_extract

gold = load_gold()
issues = [i for i in load_issues() if i.id in gold]
results = [reliable_extract(issue) for issue in issues]

n = len(results)
before = sum(bool(r.history[0]["errors"]) for r in results)      # attempt 1 alone
after = sum(r.needs_review for r in results)
first_cost = sum(r.history[0]["cost_usd"] for r in results)
total = sum(r.cost_usd for r in results)
print(f"records with semantic errors: {before}/{n} -> {after}/{n}")
print(f"re-ask rate {sum(r.attempts > 1 for r in results) / n:.0%} · "
      f"extra cost +{(total - first_cost) / max(first_cost, 1e-9):.0%} · total ${total:.4f}")
print("attempts histogram:", dict(sorted(Counter(r.attempts for r in results).items())))
print("needs_review:", [f"#{i.id} {r.stop_reason}" for i, r in zip(issues, results) if r.needs_review])

records with semantic errors: 7/30 -> 2/30
re-ask rate 23% · extra cost +27% · total $0.0161
attempts histogram: {1: 23, 2: 5, 3: 2}
needs_review: ['#10 max_attempts', '#176 max_attempts']