Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 43 additions & 21 deletions scripts/notebook_tools/audit_engine_named_not_invoked.py
Original file line number Diff line number Diff line change
Expand Up @@ -194,33 +194,51 @@ class SurfaceHits:
simulation_hits: List[Tuple[int, str]] = field(default_factory=list)


def _strip_comments(src: str) -> str:
"""Retire les commentaires `#` et les docstrings pour le scan wiring/proof.
def _strip_comments_and_strings(src: str) -> str:
"""Retire commentaires, docstrings, et chaines de caracteres pour le scan wiring.

Un match dans un commentaire n'est pas une preuve d'invocation -- c'est
juste une mention discursive. On garde les chaines de caracteres
(print("openai...")) parce qu'elles peuvent etre des appels caches,
mais on retire les commentaires Python purs.
Un match dans un commentaire OU dans une string n'est pas une preuve
d'invocation SDK -- c'est une mention discursive ou un exemple pedagogique.
On garde uniquement le code actif (appels reels, imports reels).
"""
lines = src.split("\n")
out = []
for line in lines:
stripped = line.lstrip()
if stripped.startswith("#"):
continue
# Retrait d'un commentaire en milieu de ligne (apres du code)
# Heuristique : `#` non precede d'un caractere alphanumerique (donc pas dans une string)
# Simplification : on retire tout apres le premier `#` qui n'est pas dans une string.
# Les strings simples avec `#` (`"#hashtag"`) ne sont pas affectees par notre regex wiring
# (qui cherche `google.adk` etc.), donc cette approximation suffit.
in_str = False
for i, c in enumerate(line):
if c == '"' or c == "'":
in_str = not in_str
elif c == "#" and not in_str:
line = line[:i]
# Parse caractere par caractere pour reperer les commentaires et strings.
in_str = None # None si hors string, '"' ou "'" sinon
new_line_chars = []
i = 0
while i < len(line):
c = line[i]
if in_str:
if c == in_str:
in_str = None
new_line_chars.append(c) if False else None # on jette le contenu des strings
i += 1
elif c == "#":
# Commentaire in-line : on tronque
break
out.append(line)
elif c == '"' or c == "'":
in_str = c
# On remplace le contenu de la string par des espaces (preserve positions pour l'erreur)
new_line_chars.append(c)
# Skip le contenu de la string
j = i + 1
while j < len(line) and line[j] != in_str:
if line[j] == "\\" and j + 1 < len(line):
j += 2
else:
j += 1
# Ajouter des espaces a la place du contenu
new_line_chars.append(" " * (j - i - 1))
i = j + 1 if j < len(line) else j
else:
new_line_chars.append(c)
i += 1
out.append("".join(new_line_chars))
return "\n".join(out)


Expand All @@ -233,7 +251,7 @@ def _scan_engine(notebook: dict, spec: EngineSpec) -> SurfaceHits:
out_text = _cell_output_text(cell) if ctype == "code" else ""
# Code sans commentaires : utilise pour wiring (un import dans un
# commentaire n'est pas un import reel).
code_clean = _strip_comments(src) if ctype == "code" else ""
code_clean = _strip_comments_and_strings(src) if ctype == "code" else ""

# --- claim : markdown prose ---
if ctype == "markdown":
Expand Down Expand Up @@ -363,10 +381,14 @@ def classify_notebook(
verdict = "ENGINE_EXEC_PROVED"
elif disclosed and (siblings is None or _detect_disclosed_sequence(notebook_path, spec, hits, siblings) is True):
verdict = "DISCLOSED_SEQUENCE_PROVED"
elif has_wiring and not has_proof:
verdict = "WIRING_ONLY"
elif has_simulation:
# Simulation detectee : un import SDK + un output simule = le wiring
# est cosmétique, le moteur reel n'est pas invoqué. Prend precedence
# sur WIRING_ONLY car le verdict `SIMULATED_TERMINAL` est plus
# informatif pour le lecteur (cycle 93 feedback).
verdict = "SIMULATED_TERMINAL"
elif has_wiring and not has_proof:
verdict = "WIRING_ONLY"
elif not has_wiring:
verdict = "NAMED_NOT_INVOKED"
else:
Expand Down
202 changes: 201 additions & 1 deletion scripts/notebook_tools/tests/test_audit_engine_named_not_invoked.py
Original file line number Diff line number Diff line change
Expand Up @@ -388,4 +388,204 @@ def test_scan_repo_finds_track2_adk_baseline(tmp_path):
if "google_adk" in nb_results:
assert nb_results["google_adk"]["verdict"] == "NAMED_NOT_INVOKED"
found = True
assert found, "Lab10 Track2 devrait declencher NAMED_NOT_INVOKED"
assert found, "Lab10 Track2 devrait declencher NAMED_NOT_INVOKED"


# --- Coverage gaps identified by empirical --scan-all cycle 92 -------------

def test_wiring_inside_string_literal_ignored():
"""Un import SDK place dans une string (print('import openai')) n'est PAS
un wiring reel. Le scanner doit ignorer les chaines et considerer que
la cellule n'a pas d'import.
"""
nb = _nb(
md_cells=["# GPT-4 integration"],
code_cells=[
(
"msg = 'import openai'\nprint(msg)",
_out_stream("import openai"),
),
],
)
p = Path("/tmp/_test_string_import.ipynb")
p.write_text(json.dumps(nb))
try:
results = classify_notebook(p, nb, ["openai_llm"], {p.parent.resolve(): [p]})
# L'output textuel contient 'import openai' mais ce n'est pas un wiring
# car il est dans une string. La cellule n'a pas d'import SDK reel.
# Cependant l'output (print(msg)) est reel, et il y a un claim LLM :
# c'est WIRING_ONLY si on considere le wiring absent, ou NAMED_NOT_INVOKED
# si on est strict. Le scanner doit signaler NAMED_NOT_INVOKED
# (pas de vrai wiring SDK).
assert results["openai_llm"]["verdict"] == "NAMED_NOT_INVOKED"
finally:
p.unlink()


def test_simulation_marker_only_in_output():
"""Si seul l'output contient 'Reponse simulee' (pas le code source),
le scanner doit toujours detecter la simulation.
"""
nb = _nb(
md_cells=["# GPT-4 integration via API"],
code_cells=[
(
"import openai\nresponse = openai.ChatCompletion.create()\nprint(response.choices[0].message.content)",
_out_stream("Reponse simulee : bonjour"),
),
],
)
p = Path("/tmp/_test_simulation_in_output.ipynb")
p.write_text(json.dumps(nb))
try:
results = classify_notebook(p, nb, ["openai_llm"], {p.parent.resolve(): [p]})
assert results["openai_llm"]["verdict"] == "SIMULATED_TERMINAL"
assert results["openai_llm"]["simulation"], "simulation_marker devrait etre detecte"
assert not results["openai_llm"]["proof"], "proof devrait etre vide a cause de la simulation"
finally:
p.unlink()


def test_claim_only_in_objectives_metadata():
"""Le claim peut etre dans la cellule objectifs (premiere markdown) au lieu
du titre. Le scanner doit scanner toutes les cellules markdown du notebook,
pas seulement la premiere.
"""
nb = _nb(
md_cells=[
"# Lab standard\nObjectifs pedagogiques.",
"## Details\nCe notebook integre le **Google ADK** pour demonstrer les patterns.",
],
code_cells=[
("print('placeholder')", _out_stream("placeholder")),
],
)
p = Path("/tmp/_test_claim_in_objectives.ipynb")
p.write_text(json.dumps(nb))
try:
results = classify_notebook(p, nb, ["google_adk"], {p.parent.resolve(): [p]})
assert "google_adk" in results
assert results["google_adk"]["verdict"] == "NAMED_NOT_INVOKED"
finally:
p.unlink()


def test_engine_filter_limits_scan():
"""Le filtre --engine doit limiter le scan au(x) moteur(s) demande(s)."""
nb = _nb(
md_cells=["# Lab\nGoogle ADK et BigQuery sont mentionnes ici."],
code_cells=[
("from google.cloud import bigquery", _out_stream("Client() OK")),
],
)
p = Path("/tmp/_test_engine_filter.ipynb")
p.write_text(json.dumps(nb))
try:
# Scan limite a bigquery : seul bigquery devrait apparaitre dans les results
results_bq = classify_notebook(p, nb, ["bigquery"], {p.parent.resolve(): [p]})
assert "bigquery" in results_bq
assert "google_adk" not in results_bq
# Scan complet (tous moteurs) : les deux devraient apparaitre
results_all = classify_notebook(p, nb, None, {p.parent.resolve(): [p]})
assert "google_adk" in results_all
assert "bigquery" in results_all
finally:
p.unlink()


def test_disclosed_deterministic_with_successor_executed():
"""Le verdict DISCLOSED_SEQUENCE_PROVED exige un successeur dans la meme
serie-sibling qui ait wiring+proof. Ce test verifie que le scanner cherche
bien dans le cache sibling, pas seulement le notebook courant.
"""
p_deterministic = Path("/tmp/_test_seq_det.ipynb")
p_real = Path("/tmp/_test_seq_real.ipynb")

# Notebook deterministe qui mentionne Google ADK dans sa prose comme
# etant realise dans le prochain notebook.
nb_det = _nb(
md_cells=[
"# Setup deterministe\nCe notebook est deterministe (sans LLM).\n"
"Il prepare les inputs pour le **Google ADK** runtime du notebook suivant.",
],
code_cells=[
("x = 1\nprint(x)", _out_stream("1")),
],
)
# Notebook suivant dans la serie : wiring et proof reels.
nb_real = _nb(
md_cells=["# Real Google ADK runtime"],
code_cells=[
("from google.adk import Agent\nprint('agent ready')", _out_stream("agent ready")),
],
)
p_deterministic.write_text(json.dumps(nb_det))
p_real.write_text(json.dumps(nb_real))
try:
# Le cache sibling DOIT inclure les deux notebooks dans le bon ordre.
siblings = [p_deterministic, p_real]
cache = {p_deterministic.parent.resolve(): siblings}
results = classify_notebook(p_deterministic, nb_det, ["google_adk"], cache)
assert "google_adk" in results
# Verdict attendu : DISCLOSED_SEQUENCE_PROVED (le notebook suivant dans
# la meme serie a wiring+proof reels).
assert results["google_adk"]["verdict"] == "DISCLOSED_SEQUENCE_PROVED"
finally:
p_deterministic.unlink()
p_real.unlink()


def test_disclosed_deterministic_without_successor_falls_back():
"""Si un notebook est deterministe ET claim Google ADK mais qu'il n'y a
PAS de successeur dans la serie, le scanner doit retomber sur
NAMED_NOT_INVOKED (pas inventer un successeur).
"""
p = Path("/tmp/_test_det_no_successor.ipynb")
nb = _nb(
md_cells=[
"# Lab isole\nDeterministe, sans LLM.\nMais le **Google ADK** est mentionne.",
],
code_cells=[
("x = 1", _out_stream("1")),
],
)
p.write_text(json.dumps(nb))
try:
# Pas de successeur dans la cache (un seul notebook)
siblings = [p]
cache = {p.parent.resolve(): siblings}
results = classify_notebook(p, nb, ["google_adk"], cache)
assert "google_adk" in results
# Pas de successeur -> NAMED_NOT_INVOKED (le claim n'est pas prouve)
assert results["google_adk"]["verdict"] == "NAMED_NOT_INVOKED"
finally:
p.unlink()


def test_scan_repo_excludes_nested_output_dir(tmp_path):
"""_iter_notebooks doit exclure les sous-dossiers _output/_executed
pour eviter de scanner des artefacts en double.
"""
# Un notebook normal
(tmp_path / "main.ipynb").write_text(json.dumps(_nb(
md_cells=["# Lab\nGoogle ADK integration"],
code_cells=[("print(1)", _out_stream("1"))],
)))
# Un artefact _output au meme endroit
(tmp_path / "main_output.ipynb").write_text(json.dumps(_nb(
md_cells=["# _output artefact"],
code_cells=[("print(1)", _out_stream("1"))],
)))
# Un sous-dossier _archive (doit etre scanne si pas exclu par le suffix)
archive_dir = tmp_path / "_archive"
archive_dir.mkdir()
(archive_dir / "old.ipynb").write_text(json.dumps(_nb(
md_cells=["# Archive"],
code_cells=[("print(1)", _out_stream("1"))],
)))
results = scan_repo(tmp_path)
# Doit inclure main mais pas main_output
found_main = any("main.ipynb" in nb_path and "_output" not in nb_path for nb_path in results)
found_output_artefact = any("main_output.ipynb" in nb_path for nb_path in results)
assert found_main, "main.ipynb doit etre scanne"
assert not found_output_artefact, "main_output.ipynb doit etre exclu"
Loading