diff --git a/.gitignore b/.gitignore index 710de0b..813e34e 100644 --- a/.gitignore +++ b/.gitignore @@ -34,3 +34,5 @@ credentials .cache/ # claim checker output (local) claims_report.md +spec_coverage_report.md +voice_report.md diff --git a/scripts/docs/check_claims.py b/scripts/docs/check_claims.py index 04635ed..9c07a43 100644 --- a/scripts/docs/check_claims.py +++ b/scripts/docs/check_claims.py @@ -17,24 +17,18 @@ Exit code 1 when any claim is contradicted or has no source, so the script can g from __future__ import annotations import argparse -import hashlib -import json -import os import re import sys -import time -import urllib.error -import urllib.request from dataclasses import dataclass, field from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "jev")) +from client import MODEL, api_key, system_one + ROOT = Path(__file__).resolve().parents[2] -API_URL = "https://api.typesafe.ai/v1/systemone" -MODEL = "jev-latest" AUTO_ACCEPT = 0.8 # cookbook default; evaluate on our own claims before trusting it TOP_K = 3 DEFAULT_SOURCES = ["docs/validation", "docs/known_unknowns.md", "SIMURGARRAY_FUNCTIONAL_SPEC.md"] -CACHE = ROOT / ".cache" / "claims" RELATION = { "type": "choice", @@ -255,33 +249,6 @@ def build_request(claim: Claim, cands: list[Passage]) -> dict[str, object]: return {"model": MODEL, "state": state, "questions": questions} -def call_api(payload: dict[str, object], key: str, retries: int = 5) -> dict[str, object]: - body = json.dumps(payload).encode() - digest = hashlib.sha256(body).hexdigest() - cached = CACHE / f"{digest}.json" - if cached.exists(): - return json.loads(cached.read_text()) # type: ignore[no-any-return] - headers = { - "Authorization": f"Bearer {key}", - "Content-Type": "application/json", - "User-Agent": "simurgarray-claims/1", - } - req = urllib.request.Request(API_URL, data=body, method="POST", headers=headers) - for attempt in range(retries): - try: - with urllib.request.urlopen(req, timeout=60) as r: - res = json.loads(r.read()) - CACHE.mkdir(parents=True, exist_ok=True) - cached.write_text(json.dumps(res)) - return res # type: ignore[no-any-return] - except urllib.error.HTTPError as e: - if e.code in (429, 529) and attempt < retries - 1: - time.sleep(2**attempt) - continue - raise RuntimeError(f"TypeSafe API error {e.code}: {e.read()[:200]!r}") from e - raise RuntimeError("TypeSafe API: retries exhausted") - - def verdict(answers: dict[str, dict[str, object]], threshold: float = AUTO_ACCEPT) -> tuple[str, str | None, float]: """Combine per-section answers: any confident 'supports' wins; else a confident 'contradicts'; else a confident 'says_nothing' from every section; anything else goes to a person.""" @@ -306,15 +273,6 @@ def verdict(answers: dict[str, dict[str, object]], threshold: float = AUTO_ACCEP # ------------------------------------------------------------------ report -def _key_from_file(path: Path) -> str: - """``TYPESAFE_API_KEY=...`` from a private local file (never from the repository).""" - if path.exists(): - for ln in path.read_text().splitlines(): - if ln.startswith("TYPESAFE_API_KEY="): - return ln.split("=", 1)[1].strip() - return "" - - def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("files", nargs="+", type=Path, help="texts to check (markdown or plain text)") @@ -323,7 +281,7 @@ def main(argv: list[str] | None = None) -> int: ap.add_argument("--threshold", type=float, default=AUTO_ACCEPT) ap.add_argument("--dry-run", action="store_true", help="extract claims and evidence, no API calls") a = ap.parse_args(argv) - key = os.environ.get("TYPESAFE_API_KEY", "") or _key_from_file(Path.home() / ".config" / "typesafe" / "credentials") + key = api_key() if not a.dry_run and not key: print("TYPESAFE_API_KEY is not set (use --dry-run to check extraction without the API).", file=sys.stderr) return 2 @@ -338,8 +296,8 @@ def main(argv: list[str] | None = None) -> int: if a.dry_run: rows.append(("planned", c, cands[0], 0.0, f"{len(cands)} candidate(s)")) continue - res = call_api(build_request(c, cands), key) - answers = res["answers"] # type: ignore[index] + req = build_request(c, cands) + answers = system_one(req["state"], req["questions"], key) # type: ignore[arg-type] v, qid, conf = verdict(answers, a.threshold) # type: ignore[arg-type] src = cands[int(qid[1:])] if qid else cands[0] probs = ", ".join(f"{q}: {answers[q]['choice']} {answers[q]['confidence']:.2f}" for q in answers) # type: ignore[index] diff --git a/scripts/jev/__init__.py b/scripts/jev/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/scripts/jev/client.py b/scripts/jev/client.py new file mode 100644 index 0000000..cc3a6ab --- /dev/null +++ b/scripts/jev/client.py @@ -0,0 +1,86 @@ +"""Minimal client for TypeSafe System One (Jev), shared by the project's maintenance tools in ``scripts/``. + +Only the standard library: one POST per request to https://api.typesafe.ai/v1/systemone, retries with exponential +backoff on 429 and 529, and a content addressed cache in ``.cache/jev`` so unchanged questions are never paid twice. +The API key comes from ``TYPESAFE_API_KEY`` or ``~/.config/typesafe/credentials`` (``TYPESAFE_API_KEY=...``, mode +600) and never enters the repository. These tools send documentation, test code, issue text and public abstracts; +they are never used inside the ``simurg`` analysis commands, which stay offline. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import time +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[2] +API_URL = "https://api.typesafe.ai/v1/systemone" +MODEL = "jev-latest" +CACHE = ROOT / ".cache" / "jev" +KEY_FILE = Path.home() / ".config" / "typesafe" / "credentials" + +Question = dict[str, Any] +Answers = dict[str, dict[str, Any]] + + +def api_key() -> str: + """The key from the environment, else from the private credentials file; empty when absent.""" + key = os.environ.get("TYPESAFE_API_KEY", "") + if not key and KEY_FILE.exists(): + for ln in KEY_FILE.read_text().splitlines(): + if ln.startswith("TYPESAFE_API_KEY="): + key = ln.split("=", 1)[1].strip() + return key + + +def noul(instructions: str, true: str | None = None, false: str | None = None) -> Question: + q: Question = {"type": "noul", "instructions": instructions} + if true and false: + q["criteria"] = {"true": true, "false": false} + return q + + +def choice(instructions: str, criteria: dict[str, str | None]) -> Question: + return {"type": "choice", "instructions": instructions, "criteria": criteria} + + +def score(instructions: str, levels: list[str]) -> Question: + """``levels`` from the low end of the scale to the high end (2 to 10 concrete situations).""" + return {"type": "score", "instructions": instructions, "criteria": levels} + + +def system_one(state: Any, questions: dict[str, Question], key: str | None = None, retries: int = 5) -> Answers: + """Ask ``questions`` about ``state`` in one request; returns the ``answers`` map (cached by request content).""" + payload = {"model": MODEL, "state": state, "questions": questions} + body = json.dumps(payload, sort_keys=True).encode() + cached = CACHE / f"{hashlib.sha256(body).hexdigest()}.json" + if cached.exists(): + return json.loads(cached.read_text())["answers"] # type: ignore[no-any-return] + key = key or api_key() + if not key: + raise RuntimeError("TYPESAFE_API_KEY is not set (environment or ~/.config/typesafe/credentials)") + headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": "simurgarray-jev/1"} + req = urllib.request.Request(API_URL, data=body, method="POST", headers=headers) + for attempt in range(retries): + try: + with urllib.request.urlopen(req, timeout=90) as r: + res = json.loads(r.read()) + CACHE.mkdir(parents=True, exist_ok=True) + cached.write_text(json.dumps(res)) + return res["answers"] # type: ignore[no-any-return] + except urllib.error.HTTPError as e: + if e.code in (429, 529) and attempt < retries - 1: + time.sleep(2**attempt) + continue + raise RuntimeError(f"TypeSafe API error {e.code}: {e.read()[:300]!r}") from e + except (urllib.error.URLError, TimeoutError) as e: + if attempt < retries - 1: + time.sleep(2**attempt) + continue + raise RuntimeError(f"TypeSafe API unreachable: {e}") from e + raise RuntimeError("TypeSafe API: retries exhausted") diff --git a/scripts/jev/nightly_triage.py b/scripts/jev/nightly_triage.py new file mode 100644 index 0000000..5316930 --- /dev/null +++ b/scripts/jev/nightly_triage.py @@ -0,0 +1,132 @@ +"""Triage of a failed nightly run: which known finding explains each failing step, if any? + +Code collects the evidence of every step that did not pass (its verdict and counts, the fields of the compare +summary that differ with their first example rows, or the failing lines of the run log) and the 30 or so entries of +docs/known_unknowns.md. Jev answers three questions per step in one request: + +* which known finding explains the failure, an environment problem (memory, disk, network, missing data), or a new + issue (Choice); +* does it look like a behaviour change introduced by recent code (Noul); +* how severe is it, from metadata only to a crash (Score). + +Only new issues and likely regressions need a person. The evidence comes from public test data (the GSA demo and the +1000 Genomes Omni2.5 subset). + + python scripts/jev/nightly_triage.py data/nightly/2026-10-01 # writes triage.md in that folder +""" + +from __future__ import annotations + +import argparse +import gzip +import json +import re +import sys +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from client import ROOT, choice, noul, score, system_one + +KNOWN = ROOT / "docs" / "known_unknowns.md" +SEVERITY = [ + "Only metadata, timestamps or report formatting differ; every computed value is the same", + "Small numeric differences in some values, genotype calls unchanged", + "Genotype calls, file contents or counts differ for some samples or loci", + "A step crashed or produced no output", +] + + +def known_findings(text: str) -> dict[str, str]: + """``K-NNN -> 'topic: finding'`` from the table of docs/known_unknowns.md.""" + out: dict[str, str] = {} + for ln in text.splitlines(): + m = re.match(r"\|\s*(K-\d+)\s*\|[^|]*\|\s*([^|]*)\|\s*([^|]*)\|", ln) + if m: + out[m.group(1)] = f"{m.group(2).strip()}: {m.group(3).strip()}"[:420] + return out + + +def _log_lines(log: Path, n: int = 30) -> list[str]: + if not log.exists(): + return [] + keep = [ln for ln in log.read_text(errors="replace").splitlines() + if re.search(r"FAILED|ERROR|Error|Traceback|Killed|MemAvailable|No space|exit [1-9]", ln)] # fmt: skip + return keep[-n:] + + +def evidence(base: Path, name: str, step: dict[str, Any]) -> dict[str, Any]: + ev: dict[str, Any] = {"step": name, "result": {k: v for k, v in step.items() if k not in ("report", "seconds")}} + rep = step.get("report") or (step.get("regression") or {}).get("report") + if rep: + folder = (base / rep).parent + s = folder / "summary.json" + if s.exists(): + summ = json.loads(s.read_text()) + fields = {f: {"tolerance": d.get("tolerance"), "discrepancies": d.get("discrepancies")} + for f, d in summ.get("fields", {}).items() if d.get("discrepancies")} # fmt: skip + ev["differing_fields"] = fields + ev["notes"] = summ.get("notes", [])[:10] + tsv = folder / "discrepancies.tsv.gz" + if tsv.exists(): + with gzip.open(tsv, "rt") as fh: + ev["first_rows"] = [ln.rstrip("\n")[:300] for _, ln in zip(range(16), fh, strict=False)] + if "differing_fields" not in ev and (step.get("verdict") in ("FAIL", "ERROR") or "error" in step): + ev["log"] = _log_lines(base / "run.log") # the shared log only when the step left no compare evidence + return ev + + +def triage_step(ev: dict[str, Any], findings: dict[str, str]) -> dict[str, Any]: + criteria: dict[str, str | None] = dict(findings) + criteria["environment"] = ("The run itself failed: out of memory, disk full, network, missing data or tools, a " + "killed or timed out process") # fmt: skip + criteria["new_issue"] = "None of the known findings or environment problems explains it: a new difference" + qs = { + "match": choice("Which known finding explains `failure`, if any?", criteria), + "regression": noul( + "Does `failure` look like a behaviour change introduced by recent code (outputs changed), rather than a " + "known quirk or an environment problem?" + ), + "severity": score("How severe is `failure`?", SEVERITY), + } + ans = system_one({"failure": ev}, qs) + m = ans["match"] + return { + "match": m["choice"], + "match_confidence": round(float(m["confidence"]), 2), + "regression": round(float(ans["regression"]["noul"]), 2), + "severity": round(float(ans["severity"]["score"]), 2), + "needs_person": m["choice"] == "new_issue" + or float(ans["regression"]["noul"]) >= 0.5 + or float(m["confidence"]) < 0.6, + } + + +def triage(base: Path) -> dict[str, Any]: + night = json.loads((base / "summary.json").read_text()) + findings = known_findings(KNOWN.read_text(encoding="utf-8")) + out: dict[str, Any] = {} + for name, step in night["steps"].items(): + if step.get("verdict") in ("PASS", "SKIPPED"): + continue + out[name] = triage_step(evidence(base, name, step), findings) + lines = [f"# Nightly triage {night['date']}", ""] + for name, t in out.items(): + who = "a person should look" if t["needs_person"] else "explained" + lines.append(f"* **{name}**: {t['match']} (confidence {t['match_confidence']}), regression p " + f"{t['regression']}, severity {t['severity']} of 3: {who}") # fmt: skip + (base / "triage.md").write_text("\n".join(lines) + "\n") + return out + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("night", type=Path, help="folder of one night, for example data/nightly/2026-10-01") + a = ap.parse_args(argv) + res = triage(a.night) + print(json.dumps(res, indent=1) if res else "every step passed; nothing to triage") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/jev/spec_coverage.py b/scripts/jev/spec_coverage.py new file mode 100644 index 0000000..e88c6bd --- /dev/null +++ b/scripts/jev/spec_coverage.py @@ -0,0 +1,238 @@ +"""Which functional spec items have a test? Gaps between SIMURGARRAY_FUNCTIONAL_SPEC.md and tests/. + +1. Code parses the spec items (``- **CLI-021** ...`` up to the next item or heading) and every test function. +2. An item cited by ID in a test file counts as covered by citation. +3. For the other items, code ranks test functions by word overlap (BM25) and keeps the best three; Jev answers one + yes/no question per candidate: does this test check what the item requires? +4. The report lists, per prefix, the items covered by citation, covered by judgment, maybe covered and uncovered. + +Open questions (OQ) and scope statements (SCOPE) are not behaviours and are skipped. + + python scripts/jev/spec_coverage.py # writes spec_coverage_report.md + python scripts/jev/spec_coverage.py --dry-run # parsing and candidates only, no API calls +""" + +from __future__ import annotations + +import argparse +import ast +import math +import re +import sys +from collections import Counter +from dataclasses import dataclass +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from client import ROOT, noul, system_one + +SPEC = ROOT / "SIMURGARRAY_FUNCTIONAL_SPEC.md" +TESTS = ROOT / "tests" +SKIP_PREFIXES = {"OQ", "SCOPE"} +COVERED, MAYBE = 0.8, 0.5 +_ITEM = re.compile(r"^- \*\*([A-Z]+)-(\d+)\*\*\s*(`[^`]*`)?\s*(.*)") +_ID = re.compile(r"\b([A-Z]{2,5})-(\d{3})\b") +_WORD = re.compile(r"[a-z][a-z0-9_]{2,}") +STOP = frozenset( + { + "the", + "and", + "for", + "with", + "from", + "that", + "this", + "are", + "not", + "any", + "all", + "its", + "into", + "each", + "when", + "then", + "than", + "per", + "but", + "has", + "have", + "was", + "which", + "their", + "them", + "they", + "one", + "two", + "set", + "get", + "use", + "used", + "uses", + "test", + "tests", + "should", + "must", + "may", + "can", + } +) + + +@dataclass +class Item: + id: str + prefix: str + tag: str + text: str + + +@dataclass +class TestFn: + path: str + line: int + name: str + text: str + + +def parse_spec(text: str) -> list[Item]: + items: list[Item] = [] + cur: Item | None = None + for ln in text.splitlines(): + m = _ITEM.match(ln) + if m or ln.startswith("#"): + if cur: + items.append(cur) + cur = None + if m: + cur = Item(f"{m.group(1)}-{m.group(2)}", m.group(1), (m.group(3) or "").strip("`"), m.group(4)) + continue + if cur and len(cur.text) < 1800: + cur.text += "\n" + ln + if cur: + items.append(cur) + return [i for i in items if i.prefix not in SKIP_PREFIXES] + + +def parse_tests(root: Path) -> list[TestFn]: + """Every ``test_*`` function: name, docstring and source (truncated), with the module docstring as context.""" + out: list[TestFn] = [] + for f in sorted(root.rglob("test_*.py")): + src = f.read_text(encoding="utf-8") + tree = ast.parse(src) + mod_doc = (ast.get_docstring(tree) or "").split("\n")[0] + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef) and node.name.startswith("test_"): + body = ast.get_source_segment(src, node) or "" + text = f"# {f.relative_to(ROOT)} ({mod_doc})\n{body}"[:1600] + out.append(TestFn(str(f.relative_to(ROOT)), node.lineno, node.name, text)) + return out + + +def cited_ids(root: Path) -> dict[str, set[str]]: + """Spec IDs mentioned anywhere in each test file.""" + out: dict[str, set[str]] = {} + for f in sorted(root.rglob("*.py")): + for m in _ID.finditer(f.read_text(encoding="utf-8")): + out.setdefault(f"{m.group(1)}-{m.group(2)}", set()).add(str(f.relative_to(ROOT))) + return out + + +def _tokens(s: str) -> list[str]: + return [w for w in _WORD.findall(s.lower().replace("-", "_")) if w not in STOP] + + +class Bm25: + def __init__(self, docs: list[str], k1: float = 1.4, b: float = 0.75) -> None: + self.toks = [Counter(_tokens(d)) for d in docs] + self.len = [sum(t.values()) for t in self.toks] + self.avg = sum(self.len) / max(len(self.len), 1) + df = Counter(w for t in self.toks for w in t) + n = len(docs) + self.idf = {w: math.log(1 + (n - c + 0.5) / (c + 0.5)) for w, c in df.items()} + self.k1, self.b = k1, b + + def top(self, query: str, k: int) -> list[int]: + q = set(_tokens(query)) + scores = [] + for i, t in enumerate(self.toks): + s = 0.0 + for w in q & t.keys(): + tf = t[w] + s += self.idf[w] * tf * (self.k1 + 1) / (tf + self.k1 * (1 - self.b + self.b * self.len[i] / self.avg)) + scores.append((s, i)) + scores.sort(reverse=True) + return [i for s, i in scores[:k] if s > 0] + + +QUESTION = ( + "Does the test function in `tests[{i}]` check the behaviour that `requirement` specifies? Read the test's " + "assertions: a test that only touches the same area without asserting this behaviour does not count." +) + + +def judge(item: Item, cands: list[TestFn]) -> list[float]: + state = {"requirement": f"{item.id} ({item.tag}): {item.text.strip()}", "tests": [c.text for c in cands]} + qs = { + f"t{i}": noul( + QUESTION.format(i=i), + true="The test exercises this behaviour and asserts on its outcome", + false="The test checks something else, or only touches the area", + ) + for i in range(len(cands)) + } + ans = system_one(state, qs) + return [float(ans[f"t{i}"]["noul"]) for i in range(len(cands))] + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("--out", type=Path, default=ROOT / "spec_coverage_report.md") + ap.add_argument("--dry-run", action="store_true") + ap.add_argument("--prefix", nargs="*", help="only these prefixes, for example CLI OUT") + a = ap.parse_args(argv) + items = parse_spec(SPEC.read_text(encoding="utf-8")) + if a.prefix: + items = [i for i in items if i.prefix in a.prefix] + tests = parse_tests(TESTS) + cited = cited_ids(TESTS) + bm = Bm25([t.text for t in tests]) + rows: list[tuple[Item, str, float, TestFn | None, str]] = [] + for it in items: + if it.id in cited: + rows.append((it, "cited", 1.0, None, ", ".join(sorted(cited[it.id])))) + continue + cands = [tests[i] for i in bm.top(f"{it.id} {it.text}", 5)] + if not cands: + rows.append((it, "uncovered", 0.0, None, "no candidate test")) + continue + if a.dry_run: + rows.append((it, "planned", 0.0, cands[0], f"{len(cands)} candidates")) + continue + ps = judge(it, cands) + best = max(range(len(ps)), key=lambda k: ps[k]) + status = "covered" if ps[best] >= COVERED else "maybe" if ps[best] >= MAYBE else "uncovered" + rows.append((it, status, ps[best], cands[best], "")) + order = ["uncovered", "maybe", "covered", "cited", "planned"] + lines = ["# Spec coverage by tests", ""] + tot = Counter(r[1] for r in rows) + lines.append(f"{len(rows)} items: " + " · ".join(f"{k} {tot[k]}" for k in order if tot[k])) + lines += ["", "| Prefix | Items | cited | covered | maybe | uncovered |", "|---|---|---|---|---|---|"] + for pfx in sorted({r[0].prefix for r in rows}): + c = Counter(r[1] for r in rows if r[0].prefix == pfx) + lines.append(f"| {pfx} | {sum(c.values())} | {c['cited']} | {c['covered']} | {c['maybe']} | {c['uncovered']} |") + for st in order: + sel = [r for r in rows if r[1] == st and st != "cited"] + if not sel: + continue + lines += ["", f"## {st} ({len(sel)})", ""] + for it, _, p, t, extra in sel: + first = it.text.strip().split("\n")[0][:140] + where = f" · best test `{t.path}:{t.line}` {t.name} (p {p:.2f})" if t else "" + lines.append(f"* **{it.id}** `{it.tag}` {first}{where}{(' · ' + extra) if extra else ''}") + a.out.write_text("\n".join(lines) + "\n") + print(f"{len(rows)} items: " + ", ".join(f"{k} {tot[k]}" for k in order if tot[k]) + f" → {a.out}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/jev/voice_check.py b/scripts/jev/voice_check.py new file mode 100644 index 0000000..79b9224 --- /dev/null +++ b/scripts/jev/voice_check.py @@ -0,0 +1,152 @@ +"""Check public texts against the brand voice in branding/README.md: plain, exact, every claim next to its limitation. + +Code checks the mechanical rules (em and en dashes in prose, banned marketing words). For every paragraph of at least +15 words, Jev answers three questions in one request: does the paragraph claim something about SimurgArray without +any limitation or condition (Noul), does it use promotional exaggeration (Noul), and how plain is the tone (Score, +four concrete levels). The tool marks paragraphs to look at; it never rewrites them. + + python scripts/jev/voice_check.py README.md blog/*.md my_post.txt # writes voice_report.md + python scripts/jev/voice_check.py --rules-only blog/*.md # mechanical rules only, no API calls +""" + +from __future__ import annotations + +import argparse +import re +import sys +from dataclasses import dataclass, field +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from client import ROOT, noul, score, system_one + +BANNED = ["revolutionary", "cutting edge", "cutting-edge", "ai powered", "ai-powered", "next generation", + "next-generation", "seamless", "seamlessly", "unlock", "game changer", "game-changing", "best in class", + "world class", "blazing"] # fmt: skip +LIMIT_T, HYPE_T, TONE_T = 0.7, 0.6, 1.5 +TONE_LEVELS = [ + "Plain and exact: short sentences, concrete facts or numbers, no salesmanship", + "Mostly plain, with an occasional vague or enthusiastic phrase", + "Noticeably promotional or vague: several adjectives or benefits stated without specifics", + "Marketing copy: superlatives, buzzwords and promises dominate", +] + + +@dataclass +class Para: + path: str + line: int + heading: str + text: str + issues: list[str] = field(default_factory=list) + + +def paragraphs(path: Path, text: str) -> list[Para]: + """Prose paragraphs and bullets; code blocks, tables, headings and front matter lines are skipped.""" + out: list[Para] = [] + heading, buf, start, in_code = "", [], 0, False + + def flush() -> None: + if buf: + out.append(Para(str(path), start, heading, " ".join(buf))) + buf.clear() + + for i, raw in enumerate([*text.splitlines(), ""], 1): + ln = raw.rstrip() + if ln.lstrip().startswith("```"): + flush() + in_code = not in_code + continue + if in_code: + continue + if not ln.strip() or ln.lstrip().startswith("|") or ln.startswith("#"): + flush() + if ln.startswith("#"): + heading = ln.lstrip("#").strip() + continue + m = re.match(r"\s*([*>-]|\d+\.)\s+", ln) + if m: + flush() + ln = ln[m.end() :] + if ln.lstrip().startswith("<"): # raw HTML blocks (banners, badges) are layout, not prose + continue + if not buf: + start = i + buf.append(ln.strip()) + return out + + +def mechanical(p: Para) -> list[str]: + prose = re.sub(r"`[^`]*`|https?://\S+|\([^)]*\.(md|png|jpg|svg)\)", "", p.text) + prose = re.sub(r"\"[^\"]*\"|\u201c[^\u201d]*\u201d", "", prose) # quoted words are examples, not usage + if re.match(r"\s*(Never|No|Do not)\b", prose): + prose = "" + issues = [] + if re.search("[—–]", prose): + issues.append("em or en dash in prose") + low = prose.lower() + found = [w for w in BANNED if re.search(rf"\b{re.escape(w)}\b", low)] + if found: + issues.append("banned words: " + ", ".join(found)) + return issues + + +QUESTIONS = { + "no_limit": noul( + "Does `paragraph` claim that SimurgArray achieves, guarantees or provides something (accuracy, speed, " + "compatibility, safety, a benefit) without stating any limitation, condition, number or caveat in the same " + "paragraph?", + true="A capability or benefit is asserted with nothing that bounds it", + false="There is no such claim, or the claim comes with a number, condition or limitation", + ), + "hype": noul( + "Does `paragraph` use promotional exaggeration: superlatives, buzzwords or unqualified promises instead of " + "plain description?", + true="Contains marketing language or overstatement", + false="Plain description, even if enthusiastic in a modest way", + ), + "tone": score("How plain and exact is the tone of `paragraph`?", TONE_LEVELS), +} + + +def judge(p: Para, doc: str) -> dict[str, float]: + ans = system_one({"document": doc, "section": p.heading, "paragraph": p.text}, QUESTIONS) + return {"no_limit": float(ans["no_limit"]["noul"]), "hype": float(ans["hype"]["noul"]), + "tone": float(ans["tone"]["score"])} # fmt: skip + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("files", nargs="+", type=Path) + ap.add_argument("--out", type=Path, default=ROOT / "voice_report.md") + ap.add_argument("--rules-only", action="store_true", help="mechanical rules only, no API calls") + a = ap.parse_args(argv) + flagged: list[tuple[Para, dict[str, float]]] = [] + n = 0 + for f in a.files: + doc = f.name + for p in paragraphs(f, f.read_text(encoding="utf-8")): + p.issues = mechanical(p) + vals: dict[str, float] = {} + if not a.rules_only and len(p.text.split()) >= 15: + n += 1 + vals = judge(p, doc) + if vals["no_limit"] >= LIMIT_T: + p.issues.append(f"claim without a limitation (p {vals['no_limit']:.2f})") + if vals["hype"] >= HYPE_T: + p.issues.append(f"promotional language (p {vals['hype']:.2f})") + if vals["tone"] >= TONE_T: + p.issues.append(f"tone {vals['tone']:.1f} of 3") + if p.issues: + flagged.append((p, vals)) + lines = ["# Voice check", "", f"{n} paragraphs judged, {len(flagged)} flagged.", ""] + for p, _ in flagged: + lines.append(f"* `{p.path}:{p.line}` ({p.heading}) **{'; '.join(p.issues)}**") + lines.append(f" > {p.text[:400]}") + a.out.write_text("\n".join(lines) + "\n") + print(f"{n} paragraphs judged, {len(flagged)} flagged → {a.out}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/nightly.py b/scripts/nightly.py index 0bde8c4..3179b04 100644 --- a/scripts/nightly.py +++ b/scripts/nightly.py @@ -175,6 +175,23 @@ def render(night: dict[str, Any]) -> str: "
| step | verdict | seconds | details |
|---|
Triage failed: {html.escape(str(tri['error']))}
" + rows = "".join( + f"| step | explained by | confidence | regression p | " + f"severity (0 to 3) | needs a person |
|---|