feat(competences): extract batch candidates

This commit is contained in:
Dom
2026-05-29 11:25:00 +02:00
parent 4ba426c205
commit e8a0fb0e42
60 changed files with 18176 additions and 0 deletions

View File

@@ -0,0 +1,996 @@
from __future__ import annotations
import json
import sys
from pathlib import Path
import yaml
ROOT = Path(__file__).parent.parent.parent
sys.path.insert(0, str(ROOT))
from tools.competence_validator import validate_competence_file, validate_primitive_file
from tools import competence_validator
P0_COMPETENCE = ROOT / "data/competences/candidate/open_windows_search.yaml"
P1_SEARCH_COMPETENCE = ROOT / "data/competences/observed/saisir_requete_recherche.yaml"
P2_WORD_COMPETENCE = ROOT / "data/competences/candidate/saisir_texte_word.yaml"
P3_RUN_COMPETENCE = ROOT / "data/competences/observed/open_application_via_run.yaml"
P3_SCROLL_COMPETENCE = ROOT / "data/competences/observed/scroll_down_pdf_edge.yaml"
P4_CLICK_SEARCH_COMPETENCE = ROOT / "data/competences/candidate/open_windows_search_taskbar_click.yaml"
KEY_COMBO_PRIMITIVE = ROOT / "data/primitives/key_combo.yaml"
TEXT_INPUT_FOCUSED_PRIMITIVE = ROOT / "data/primitives/text_input_focused.yaml"
SCROLL_VIEW_PRIMITIVE = ROOT / "data/primitives/scroll_view.yaml"
CLICK_ANCHOR_PRIMITIVE = ROOT / "data/primitives/click_anchor.yaml"
WAIT_FOR_STATE_PRIMITIVE = ROOT / "data/primitives/wait_for_state.yaml"
def _issue_codes(path: Path) -> set[str]:
return {issue.code for issue in validate_competence_file(path, repo_root=ROOT).issues}
def _sequence_competence_data() -> dict:
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
data["methods_execution"] = "sequence"
data["chain_refs"]["cleaned_segment"]["keep_event_indices"] = [3, 5, 6, 7, 8, 9, 10, 11, 12, 13]
data["chain_refs"]["cleaned_segment"]["method_event_indices"] = [3, 5, 6, 8, 9, 10, 12]
data["chain_refs"]["cleaned_segment"]["success_event_indices"] = [13]
data["methods"] = [
{
"id": "step_1_open_search",
"kind": "key_combo",
"primitive_ref": "key_combo",
"parameters": {"keys": ["win", "s"]},
"keys": ["win", "s"],
"observed": True,
"trace_source": "live_events.jsonl",
"trace_event_indices": [3],
},
{
"id": "step_2_type_query",
"kind": "text_input",
"primitive_ref": "text_input_focused",
"parameters": {
"text": "test lea apprentissage",
"concat_rule": "concat_in_order",
},
"observed": True,
"trace_source": "live_events.jsonl",
"trace_event_indices": [5, 6, 8, 9, 10, 12],
"reconstructed_text": "test lea apprentissage",
},
]
return data
def _write_nested_session(path: Path, events: list[dict]) -> None:
path.write_text(
json.dumps(
{
"session_id": "sess_nested",
"events": [
{
"session_id": "sess_nested",
"timestamp": float(index),
"event": event,
}
for index, event in enumerate(events)
],
}
),
encoding="utf-8",
)
def _scroll_competence_data(tmp_path: Path, events: list[dict]) -> dict:
session_path = tmp_path / "nested_scroll_session.json"
live_events_path = tmp_path / "live_events.jsonl"
_write_nested_session(session_path, events)
live_events_path.write_text("", encoding="utf-8")
return {
"schema_version": 1,
"id": "scroll_test",
"name": "Scroll test",
"version": 1,
"learning_state": "observed",
"intent": {"fr": "tester un scroll"},
"parameters": {},
"preconditions": [{"id": "app_active", "kind": "active_window", "any_of": [{"process_active": "msedge.exe"}]}],
"methods": [
{
"id": "scroll_down",
"kind": "scroll",
"primitive_ref": "scroll_view",
"parameters": {"direction": "down", "amount": 3, "unit": "lines"},
"observed": True,
"trace_source": "live_events.jsonl",
"trace_event_indices": [1],
}
],
"success_marker": {
"mode": "all_of",
"timeout_ms": 5000,
"markers": [{"kind": "active_process_name_is", "value": "msedge.exe"}],
},
"failure_message_template": {
"intention": "tester un scroll",
"attendu": "la fenetre doit rester active apres le scroll",
"vu": "{observed_human_state}",
"demande": "indiquer si la fenetre active peut defiler vers le bas",
},
"chain_refs": {
"source_session": "sess_nested",
"machine_id": "DESKTOP-58D5CAC_windows",
"streaming_session_path": str(session_path),
"live_events_path": str(live_events_path),
"cleaned_segment": {
"status": "documented_offline",
"keep_event_indices": [0, 1, 2],
"method_event_indices": [1],
"success_event_indices": [2],
"excluded_event_indices": [],
"stop_before_event_index": 3,
"stop_before": ["end_of_synthetic_scroll_trace"],
},
},
"promotion": {
"candidate_requires": ["cleaned_segment_validated"],
"supervised_requires": ["replay_verified_once"],
"stable_requires": {"min_successes": 3, "distinct_contexts": 3, "max_unexplained_failures": 0},
},
"generalisation": {"seen_contexts": [], "method_success_rate": {}, "variance_log": []},
"failure_log": [],
"created_at": "2026-05-28T13:45:00+02:00",
"last_updated_at": "2026-05-28T13:45:00+02:00",
}
def _click_competence_data(tmp_path: Path, events: list[dict]) -> dict:
session_path = tmp_path / "nested_click_session.json"
live_events_path = tmp_path / "live_events.jsonl"
_write_nested_session(session_path, events)
live_events_path.write_text("", encoding="utf-8")
return {
"schema_version": 1,
"id": "click_test",
"name": "Click test",
"version": 1,
"learning_state": "observed",
"intent": {"fr": "tester un clic sur ancre"},
"parameters": {},
"preconditions": [{"id": "desktop_active", "kind": "active_window", "any_of": [{"process_active": "explorer.exe"}]}],
"methods": [
{
"id": "click_search",
"kind": "click",
"primitive_ref": "click_anchor",
"parameters": {
"anchor_ref": "windows_search_button",
"button": "left",
"click_count": 1,
"relative_offset": {"x_pct": 0.5, "y_pct": 0.5},
},
"observed": True,
"trace_source": "live_events.jsonl",
"trace_event_indices": [1],
}
],
"success_marker": {
"mode": "all_of",
"timeout_ms": 5000,
"markers": [{"kind": "active_process_name_is", "value": "SearchHost.exe"}],
},
"failure_message_template": {
"intention": "cliquer sur le bouton de recherche",
"attendu": "la fenetre rechercher doit s'ouvrir",
"vu": "{observed_human_state}",
"demande": "me montrer le bouton rechercher dans la barre des taches",
},
"chain_refs": {
"source_session": "sess_nested",
"machine_id": "windows_vm",
"streaming_session_path": str(session_path),
"live_events_path": str(live_events_path),
"cleaned_segment": {
"status": "documented_offline",
"keep_event_indices": [0, 1, 2],
"method_event_indices": [1],
"success_event_indices": [2],
"excluded_event_indices": [],
"stop_before_event_index": 3,
"stop_before": ["end_of_synthetic_click_trace"],
},
},
"promotion": {
"candidate_requires": ["cleaned_segment_validated"],
"supervised_requires": ["replay_verified_once"],
"stable_requires": {"min_successes": 3, "distinct_contexts": 3, "max_unexplained_failures": 0},
"t2_known_gaps": [
{
"id": "click_target_semantics_not_observed_offline",
"description": "la trace prouve le clic mais pas l'ancre semantique sans OCR offline",
"impact": "candidate requiert replay ou validation humaine de l'ancre",
"proposed_resolution": "ajouter preuve OCR ou screenshot diff au replay supervise",
}
],
},
"generalisation": {"seen_contexts": [], "method_success_rate": {}, "variance_log": []},
"failure_log": [],
"created_at": "2026-05-28T15:35:00+02:00",
"last_updated_at": "2026-05-28T15:35:00+02:00",
}
def _wait_state_competence_data(tmp_path: Path, events: list[dict]) -> dict:
session_path = tmp_path / "nested_wait_state_session.json"
live_events_path = tmp_path / "live_events.jsonl"
_write_nested_session(session_path, events)
live_events_path.write_text("", encoding="utf-8")
return {
"schema_version": 1,
"id": "wait_state_test",
"name": "Wait state test",
"version": 1,
"learning_state": "observed",
"intent": {"fr": "tester une attente d'etat"},
"parameters": {},
"preconditions": [{"id": "desktop_active", "kind": "active_window", "any_of": [{"process_active": "explorer.exe"}]}],
"methods": [
{
"id": "wait_search_visible",
"kind": "wait_state",
"primitive_ref": "wait_for_state",
"parameters": {
"expected_state": {
"window_title_in": ["Rechercher"],
"process_active": "SearchHost.exe",
},
"timeout_ms": 3000,
"poll_interval_ms": 250,
"evidence_required": "window_or_process",
},
"observed": True,
"trace_source": "live_events.jsonl",
"trace_event_indices": [1],
}
],
"success_marker": {
"mode": "all_of",
"timeout_ms": 5000,
"markers": [
{"kind": "active_window_title_in", "values": ["Rechercher"]},
{"kind": "active_process_name_is", "value": "SearchHost.exe"},
],
},
"failure_message_template": {
"intention": "attendre l'apparition de la recherche Windows",
"attendu": "la fenetre rechercher doit etre visible",
"vu": "{observed_human_state}",
"demande": "me montrer la fenetre rechercher ou son libelle visible",
},
"chain_refs": {
"source_session": "sess_nested",
"machine_id": "windows_vm",
"streaming_session_path": str(session_path),
"live_events_path": str(live_events_path),
"cleaned_segment": {
"status": "documented_offline",
"keep_event_indices": [0, 1, 2],
"method_event_indices": [1],
"success_event_indices": [2],
"excluded_event_indices": [],
"stop_before_event_index": 3,
"stop_before": ["end_of_synthetic_wait_state_trace"],
},
},
"promotion": {
"candidate_requires": ["cleaned_segment_validated"],
"supervised_requires": ["replay_verified_once"],
"stable_requires": {"min_successes": 3, "distinct_contexts": 3, "max_unexplained_failures": 0},
},
"generalisation": {"seen_contexts": [], "method_success_rate": {}, "variance_log": []},
"failure_log": [],
"created_at": "2026-05-28T16:35:00+02:00",
"last_updated_at": "2026-05-28T16:35:00+02:00",
}
def test_validator_imports_message_contract():
assert competence_validator.format_supervised_pause_message is not None, (
"message_contract introuvable: le validateur ignorerait silencieusement "
"failure_message_template"
)
def test_open_windows_search_candidate_validates_against_source_trace():
report = validate_competence_file(P0_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_saisir_requete_recherche_competence_validates_against_source_trace():
report = validate_competence_file(P1_SEARCH_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_saisir_texte_word_competence_validates_against_source_trace():
report = validate_competence_file(P2_WORD_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_open_application_via_run_competence_validates_against_source_trace():
report = validate_competence_file(P3_RUN_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_scroll_down_pdf_edge_competence_validates_against_source_trace():
report = validate_competence_file(P3_SCROLL_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_open_windows_search_taskbar_click_validates_against_source_trace():
report = validate_competence_file(P4_CLICK_SEARCH_COMPETENCE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_validator_handles_nested_event_format(tmp_path):
session_path = tmp_path / "nested_session.json"
live_events_path = tmp_path / "live_events.jsonl"
session_path.write_text(
json.dumps(
{
"session_id": "sess_nested",
"events": [
{
"session_id": "sess_nested",
"timestamp": 1.0,
"event": {
"type": "key_combo",
"keys": ["win", "s"],
"window": {"title": "Desktop", "app_name": "explorer.exe"},
},
},
{
"session_id": "sess_nested",
"timestamp": 2.0,
"event": {
"type": "window_focus_change",
"to": {"title": "Rechercher", "app_name": "SearchHost.exe"},
},
},
],
}
),
encoding="utf-8",
)
live_events_path.write_text("", encoding="utf-8")
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["chain_refs"]["source_session"] = "sess_nested"
data["chain_refs"]["streaming_session_path"] = str(session_path)
data["chain_refs"]["live_events_path"] = str(live_events_path)
data["chain_refs"]["cleaned_segment"]["keep_event_indices"] = [0, 1]
data["chain_refs"]["cleaned_segment"]["method_event_indices"] = [0]
data["chain_refs"]["cleaned_segment"]["success_event_indices"] = [1]
data["chain_refs"]["cleaned_segment"]["excluded_event_indices"] = []
data["chain_refs"]["cleaned_segment"]["stop_before_event_index"] = 2
data["chain_refs"]["cleaned_segment"]["stop_before"] = ["end_of_synthetic_nested_trace"]
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_saisir_texte_word_documents_t2_known_gap():
data = yaml.safe_load(P2_WORD_COMPETENCE.read_text(encoding="utf-8"))
gaps = data["promotion"]["t2_known_gaps"]
assert gaps[0]["id"] == "marker_continuation_human"
assert "success_event #40" in gaps[0]["description"]
assert gaps[0]["proposed_resolution"]
def test_bootstrap_primitives_validate():
for path in (
KEY_COMBO_PRIMITIVE,
TEXT_INPUT_FOCUSED_PRIMITIVE,
SCROLL_VIEW_PRIMITIVE,
CLICK_ANCHOR_PRIMITIVE,
WAIT_FOR_STATE_PRIMITIVE,
):
report = validate_primitive_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_primitive_click_anchor_validates():
report = validate_primitive_file(CLICK_ANCHOR_PRIMITIVE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_primitive_wait_for_state_validates():
report = validate_primitive_file(WAIT_FOR_STATE_PRIMITIVE, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_existing_competences_reference_bootstrap_primitives():
p0 = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
p1 = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
assert p0["methods"][0]["primitive_ref"] == "key_combo"
assert p0["methods"][0]["parameters"]["keys"] == ["win", "s"]
assert p1["methods"][0]["primitive_ref"] == "text_input_focused"
assert p1["methods"][0]["parameters"]["text"] == "test lea apprentissage"
def test_observed_dependency_accepts_promoted_candidate():
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
assert data["preconditions"][0]["state"] == "observed"
assert validate_competence_file(P1_SEARCH_COMPETENCE, repo_root=ROOT).valid
def test_validator_rejects_missing_observed_key_combo_in_cleaned_segment(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["keys"] = ["ctrl", "k"]
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_rejects_id_filename_mismatch(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
path = tmp_path / "wrong_filename.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "id_filename_mismatch" in _issue_codes(path)
def test_validator_full_competence_corpus():
competence_paths = sorted((ROOT / "data/competences").glob("*/*.yaml"))
primitive_paths = sorted((ROOT / "data/primitives").glob("*.yaml"))
assert competence_paths, "no competence YAML found"
assert primitive_paths, "no primitive YAML found"
failures = {
str(path.relative_to(ROOT)): [
f"{issue.code}: {issue.detail}"
for issue in validate_competence_file(path, repo_root=ROOT).issues
]
for path in competence_paths
}
failures.update(
{
str(path.relative_to(ROOT)): [
f"{issue.code}: {issue.detail}"
for issue in validate_primitive_file(path, repo_root=ROOT).issues
]
for path in primitive_paths
}
)
failures = {path: issues for path, issues in failures.items() if issues}
assert failures == {}
def test_validator_rejects_primitive_forbidden_field(tmp_path):
data = yaml.safe_load(KEY_COMBO_PRIMITIVE.read_text(encoding="utf-8"))
data["learning_state"] = "observed"
path = tmp_path / "key_combo.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
issue_codes = {issue.code for issue in validate_primitive_file(path, repo_root=ROOT).issues}
assert "primitive_forbidden_field" in issue_codes
def test_validator_rejects_primitive_empty_enum(tmp_path):
data = yaml.safe_load(SCROLL_VIEW_PRIMITIVE.read_text(encoding="utf-8"))
data["parameters_schema"]["direction"]["constraints"]["enum"] = []
path = tmp_path / "scroll_view.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
issue_codes = {issue.code for issue in validate_primitive_file(path, repo_root=ROOT).issues}
assert "primitive_schema_invalid" in issue_codes
def test_primitive_click_anchor_rejects_pos_in_parameters(tmp_path):
data = yaml.safe_load(CLICK_ANCHOR_PRIMITIVE.read_text(encoding="utf-8"))
data["parameters_schema"]["pos"] = {
"type": "list[str]",
"required": False,
"description": "coordonnees a refuser",
}
path = tmp_path / "click_anchor.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
issue_codes = {issue.code for issue in validate_primitive_file(path, repo_root=ROOT).issues}
assert "durable_coordinate_key" in issue_codes
def test_primitive_click_count_out_of_range_rejected(tmp_path):
data = _click_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["click_count"] = 3
path = tmp_path / "click_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_click_count_out_of_range" in _issue_codes(path)
def test_primitive_relative_offset_pct_out_of_range_rejected(tmp_path):
data = _click_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["relative_offset"] = {"x_pct": 1.5, "y_pct": 0.5}
path = tmp_path / "click_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
issue_codes = _issue_codes(path)
assert "primitive_relative_offset_invalid" in issue_codes
assert "durable_coordinate_key" not in issue_codes
def test_validator_click_method_requires_mouse_click_events(tmp_path):
data = _click_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "key_combo", "keys": ["win", "s"], "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
path = tmp_path / "click_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_click_method_with_valid_mouse_click_passes(tmp_path):
data = _click_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
path = tmp_path / "click_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_wait_for_state_method_with_window_focus_change_passes(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{
"type": "window_focus_change",
"to": {"title": "Rechercher", "app_name": "SearchHost.exe"},
},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_wait_for_state_expected_state_required(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"].pop("expected_state")
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_expected_state_invalid" in _issue_codes(path)
def test_wait_for_state_expected_state_must_be_non_empty_dict(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["expected_state"] = {}
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_expected_state_invalid" in _issue_codes(path)
def test_wait_for_state_timeout_out_of_range_rejected(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["timeout_ms"] = 50
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_wait_timeout_invalid" in _issue_codes(path)
def test_wait_for_state_poll_interval_out_of_range_rejected(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["poll_interval_ms"] = 10000
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_poll_interval_invalid" in _issue_codes(path)
def test_wait_for_state_evidence_required_enum_validated(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "window_focus_change", "to": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
data["methods"][0]["parameters"]["evidence_required"] = "foo"
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_schema_invalid" in _issue_codes(path)
def test_wait_for_state_method_rejects_human_continuation_event(tmp_path):
data = _wait_state_competence_data(
tmp_path,
[
{"type": "mouse_click", "button": "left", "window": {"title": "Bureau", "app_name": "explorer.exe"}},
{"type": "text_input", "text": "test", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
{"type": "heartbeat", "window": {"title": "Rechercher", "app_name": "SearchHost.exe"}},
],
)
path = tmp_path / "wait_state_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_rejects_bad_t2_known_gaps_type(tmp_path):
data = yaml.safe_load(P2_WORD_COMPETENCE.read_text(encoding="utf-8"))
data["promotion"]["t2_known_gaps"] = "marker_continuation_human"
path = tmp_path / "saisir_texte_word.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "t2_known_gap_invalid" in _issue_codes(path)
def test_validator_rejects_t2_known_gap_missing_required_field(tmp_path):
data = yaml.safe_load(P2_WORD_COMPETENCE.read_text(encoding="utf-8"))
data["promotion"]["t2_known_gaps"] = [
{
"id": "marker_continuation_human",
"description": "success_event #40 est un text_input humain post-methode.",
"proposed_resolution": "Ajouter wait_state ou OCR runtime.",
}
]
path = tmp_path / "saisir_texte_word.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "t2_known_gap_invalid" in _issue_codes(path)
def test_validator_accepts_methods_execution_sequence_with_step_trace_indices(tmp_path):
data = _sequence_competence_data()
path = tmp_path / "saisir_requete_recherche.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_validator_rejects_invalid_methods_execution_mode(tmp_path):
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
data["methods_execution"] = "serial"
path = tmp_path / "saisir_requete_recherche.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "methods_sequence_invalid" in _issue_codes(path)
def test_validator_rejects_sequence_without_two_methods(tmp_path):
data = _sequence_competence_data()
data["methods"] = data["methods"][:1]
path = tmp_path / "saisir_requete_recherche.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "methods_sequence_invalid" in _issue_codes(path)
def test_validator_rejects_sequence_observed_step_without_trace_indices(tmp_path):
data = _sequence_competence_data()
data["methods"][1].pop("trace_event_indices")
path = tmp_path / "saisir_requete_recherche.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_accepts_trace_event_indices_in_alternatives_mode(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["trace_event_indices"] = [3]
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_validator_rejects_trace_event_indices_outside_keep_indices_in_alternatives_mode(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["trace_event_indices"] = [5]
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_rejects_trace_event_indices_outside_method_indices_in_alternatives_mode(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["chain_refs"]["cleaned_segment"]["keep_event_indices"] = [0, 1, 2, 3, 4, 7]
data["chain_refs"]["cleaned_segment"]["method_event_indices"] = [3]
data["methods"][0]["trace_event_indices"] = [4]
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_alternatives_trace_event_indices_have_no_order_constraint(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["chain_refs"]["cleaned_segment"]["method_event_indices"] = [3, 7]
data["methods"][0]["trace_event_indices"] = [7]
data["methods"][1]["observed"] = True
data["methods"][1]["trace_source"] = "live_events.jsonl"
data["methods"][1]["trace_event_indices"] = [3]
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "methods_sequence_invalid" not in _issue_codes(path)
def test_validator_accepts_scroll_method_with_trace_event_indices(tmp_path):
data = _scroll_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "mouse_scroll", "delta": [0, -1], "window": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "heartbeat", "window": {"title": "PDF", "app_name": "msedge.exe"}},
],
)
path = tmp_path / "scroll_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_validator_accepts_scroll_method_with_method_indices_fallback(tmp_path):
data = _scroll_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "mouse_scroll", "delta": [0, -1], "window": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "heartbeat", "window": {"title": "PDF", "app_name": "msedge.exe"}},
],
)
data["methods"][0].pop("trace_event_indices")
path = tmp_path / "scroll_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
report = validate_competence_file(path, repo_root=ROOT)
assert report.valid, [f"{issue.code}: {issue.detail}" for issue in report.issues]
def test_validator_rejects_scroll_method_with_non_scroll_events(tmp_path):
data = _scroll_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "mouse_click", "window": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "heartbeat", "window": {"title": "PDF", "app_name": "msedge.exe"}},
],
)
path = tmp_path / "scroll_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_rejects_scroll_method_without_delta(tmp_path):
data = _scroll_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "mouse_scroll", "window": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "heartbeat", "window": {"title": "PDF", "app_name": "msedge.exe"}},
],
)
path = tmp_path / "scroll_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_scroll_delta_missing" in _issue_codes(path)
def test_validator_rejects_scroll_method_direction_mismatch(tmp_path):
data = _scroll_competence_data(
tmp_path,
[
{"type": "window_focus_change", "to": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "mouse_scroll", "delta": [0, 1], "window": {"title": "PDF", "app_name": "msedge.exe"}},
{"type": "heartbeat", "window": {"title": "PDF", "app_name": "msedge.exe"}},
],
)
path = tmp_path / "scroll_test.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_scroll_direction_mismatch" in _issue_codes(path)
def test_validator_rejects_unknown_primitive_ref(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["primitive_ref"] = "missing_primitive"
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_ref_unknown" in _issue_codes(path)
def test_validator_rejects_primitive_kind_mismatch(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["primitive_ref"] = "text_input_focused"
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_kind_mismatch" in _issue_codes(path)
def test_validator_rejects_missing_primitive_parameter(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["parameters"] = {}
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_schema_invalid" in _issue_codes(path)
def test_validator_rejects_missing_scroll_direction_parameter(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["kind"] = "scroll"
data["methods"][0]["primitive_ref"] = "scroll_view"
data["methods"][0]["parameters"] = {"amount": 3, "unit": "lines"}
data["methods"][0].pop("keys", None)
path = tmp_path / "open_windows_search.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "primitive_schema_invalid" in _issue_codes(path)
def test_validator_rejects_durable_coordinates(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["success_marker"]["coordinates"] = {"x": 120, "y": 340}
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "durable_coordinate_key" in _issue_codes(path)
def test_validator_rejects_bad_failure_message_contract(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["failure_message_template"]["vu"] = "target_not_found score=0.87"
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "failure_message_contract" in _issue_codes(path)
def test_validator_rejects_success_marker_before_method(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["chain_refs"]["cleaned_segment"]["keep_event_indices"] = [0, 1, 2, 3, 4]
data["chain_refs"]["cleaned_segment"]["success_event_indices"] = [2]
data["chain_refs"]["cleaned_segment"]["stop_before_event_index"] = 5
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "success_marker_pre_method" in _issue_codes(path)
def test_validator_rejects_stable_state_without_3_contexts(tmp_path):
data = yaml.safe_load(P0_COMPETENCE.read_text(encoding="utf-8"))
data["learning_state"] = "stable"
data["generalisation"]["seen_contexts"] = [
{"dpi": 150, "screen": "2560x1600", "method_used": "keyboard_win_s"},
{"dpi": 150, "screen": "2560x1600", "method_used": "keyboard_win_s"},
]
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "learning_state_premature" in _issue_codes(path)
def test_validator_rejects_text_input_reconstruction_mismatch(tmp_path):
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
data["methods"][0]["reconstructed_text"] = "test lea"
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_reconstructed_text_mismatch" in _issue_codes(path)
def test_validator_rejects_text_input_method_indices_with_heartbeat(tmp_path):
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
data["chain_refs"]["cleaned_segment"]["method_event_indices"] = [5, 6, 7]
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "method_trace_missing" in _issue_codes(path)
def test_validator_rejects_missing_competence_dependency(tmp_path):
data = yaml.safe_load(P1_SEARCH_COMPETENCE.read_text(encoding="utf-8"))
data["preconditions"][0]["competence"] = "missing_competence"
path = tmp_path / "bad_competence.yaml"
path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True), encoding="utf-8")
assert "competence_dependency_missing" in _issue_codes(path)

View File

@@ -0,0 +1,580 @@
from __future__ import annotations
import json
import pytest
import tools.extract_competences_from_session as extractor
from tools.extract_competences_from_session import build_report, render_markdown_report
def _write_raw_jsonl(path, events):
lines = [
json.dumps(
{
"session_id": "sess_extract_test",
"timestamp": float(index),
"event": event,
"machine_id": "windows_vm",
}
)
for index, event in enumerate(events)
]
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
def test_dry_run_extracts_click_wait_state_candidate(tmp_path):
session_path = tmp_path / "live_events.jsonl"
output_dir = tmp_path / "observed"
_write_raw_jsonl(
session_path,
[
{"type": "heartbeat", "active_window_title": "Bureau"},
{
"type": "mouse_click",
"button": "left",
"window": {"title": "Bureau", "app_name": "explorer.exe"},
"uia_snapshot": {
"name": "Rechercher",
"control_type": "bouton",
"automation_id": "SearchButton",
"parent_path": [{"name": "Barre des taches", "control_type": "volet"}],
},
},
{
"type": "window_focus_change",
"to": {"title": "Rechercher", "app_name": "SearchHost.exe"},
"window": {"title": "Rechercher", "app_name": "SearchHost.exe"},
},
],
)
report = build_report(
session_path=session_path,
machine_id="windows_vm",
output_dir=output_dir,
)
assert report["mode"] == "dry_run"
assert report["summary"]["would_write"] == 0
assert report["summary"]["candidates_generated"] == 1
candidate = report["candidates"][0]
assert candidate["validator_status"] == "would_pass"
assert candidate["apply_eligible"] is True
assert candidate["primitive_refs"] == ["click_anchor", "wait_for_state"]
assert candidate["segment"] == {"keep": [0, 1, 2], "method": [1, 2], "success": [2]}
assert candidate["t2_gaps_detected"] == [
"click_target_semantics_not_observed_offline",
"no_ocr_offline",
]
assert not (output_dir / f"{candidate['competence_id']}.yaml").exists()
def test_dry_run_rejects_click_without_uia_anchor(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "mouse_click",
"button": "left",
"window": {"title": "Bureau", "app_name": "explorer.exe"},
},
{
"type": "window_focus_change",
"to": {"title": "Rechercher", "app_name": "SearchHost.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
assert report["summary"]["candidates_generated"] == 0
assert report["summary"]["candidates_rejected"] == 1
assert report["rejected"][0]["reason"] == "click without uia_snapshot anchor"
assert report["rejected"][0]["validator_codes"] == ["anchor_ref_uia_missing"]
def test_dry_run_rejects_weak_uia_click_anchor(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "mouse_click",
"button": "left",
"window": {"title": "Rechercher", "app_name": "SearchHost.exe"},
"uia_snapshot": {
"name": "Aujourd'hui",
"control_type": "Groupe",
"automation_id": "0",
},
},
{
"type": "window_focus_change",
"to": {"title": "unknown_window", "app_name": "explorer.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
assert report["summary"]["candidates_generated"] == 0
assert report["rejected"][0]["reason"] == "click with too generic anchor"
assert report["rejected"][0]["validator_codes"] == ["anchor_ref_too_generic"]
def test_dry_run_rejects_systemtrayicon_anchor(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "mouse_click",
"button": "left",
"window": {"title": "Shell_TrayWnd", "app_name": "explorer.exe"},
"uia_snapshot": {
"name": "SystemTrayIcon",
"control_type": "bouton",
"automation_id": "SystemTrayIcon",
},
},
{
"type": "window_focus_change",
"to": {"title": "unknown_window", "app_name": "explorer.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
assert report["summary"]["candidates_generated"] == 0
assert report["rejected"][0]["reason"] == "click on fragile system tray anchor"
assert report["rejected"][0]["validator_codes"] == ["anchor_ref_systray_fragile"]
def test_dry_run_rejects_dom_autogenerated_anchor(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "mouse_click",
"button": "left",
"window": {"title": "Chrome", "app_name": "chrome.exe"},
"uia_snapshot": {
"name": "Continuer",
"control_type": "bouton",
"automation_id": "so_iazxhgsedkduppcyhoay_73",
},
},
{
"type": "window_focus_change",
"to": {"title": "Chrome", "app_name": "chrome.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
assert report["summary"]["candidates_generated"] == 0
assert report["rejected"][0]["reason"] == "click on autogenerated DOM anchor"
assert report["rejected"][0]["validator_codes"] == ["anchor_ref_dom_autogenerated"]
def test_dry_run_rejects_unknown_window_title(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "mouse_click",
"button": "left",
"window": {"title": "unknown_window", "app_name": "explorer.exe"},
"uia_snapshot": {
"name": "Ouvrir",
"control_type": "bouton",
"automation_id": "OpenButton",
},
},
{
"type": "window_focus_change",
"to": {"title": "Explorateur", "app_name": "explorer.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
assert report["summary"]["candidates_generated"] == 0
assert report["rejected"][0]["reason"] == "click in unknown or overflow window"
assert report["rejected"][0]["validator_codes"] == ["anchor_ref_unknown_window"]
def test_dry_run_rejects_browser_contextual_anchor(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "mouse_click",
"button": "left",
"window": {"title": "Dashboard - Google Chrome", "app_name": "chrome.exe"},
"uia_snapshot": {
"name": "Nouvel onglet",
"control_type": "Bouton",
"class_name": "TabStripControlButton",
"automation_id": "",
"parent_path": [{"name": "", "control_type": "tabulation"}],
},
},
{
"type": "window_focus_change",
"to": {"title": "Nouvel onglet - Google Chrome", "app_name": "chrome.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
assert report["summary"]["candidates_generated"] == 0
assert report["rejected"][0]["reason"] == "click on contextual browser chrome anchor"
assert report["rejected"][0]["validator_codes"] == ["anchor_ref_browser_contextual"]
def test_dry_run_rejects_contextual_add_tab_button_anchor(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "mouse_click",
"button": "left",
"window": {"title": "agent_debug.log - Bloc-notes", "app_name": "Notepad.exe"},
"uia_snapshot": {
"name": "Ajouter un nouvel onglet",
"control_type": "bouton",
"class_name": "Button",
"automation_id": "AddButton",
"parent_path": [
{"name": "Bureau 1", "control_type": "volet"},
{"name": "agent_debug.log - Bloc-notes", "control_type": "fenetre"},
{"name": "", "control_type": "volet"},
{"name": "", "control_type": "onglet"},
],
},
},
{
"type": "window_focus_change",
"to": {"title": "agent_debug.log - Bloc-notes", "app_name": "Notepad.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
assert report["summary"]["candidates_generated"] == 0
assert report["rejected"][0]["reason"] == "click on contextual UI chrome button"
assert report["rejected"][0]["validator_codes"] == ["anchor_ref_contextual_button"]
def test_dry_run_rejects_too_generic_anchor(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "mouse_click",
"button": "left",
"window": {"title": "Application", "app_name": "app.exe"},
"uia_snapshot": {
"name": "button_12",
"control_type": "bouton",
"automation_id": "",
},
},
{
"type": "window_focus_change",
"to": {"title": "Application", "app_name": "app.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
assert report["summary"]["candidates_generated"] == 0
assert report["rejected"][0]["reason"] == "click with too generic anchor"
assert report["rejected"][0]["validator_codes"] == ["anchor_ref_too_generic"]
def test_dry_run_rejects_empty_region_anchor(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "mouse_click",
"button": "left",
"window": {"title": "Application", "app_name": "app.exe"},
"uia_snapshot": {
"name": "",
"control_type": "région",
"automation_id": "",
},
},
{
"type": "window_focus_change",
"to": {"title": "Application", "app_name": "app.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
assert report["summary"]["candidates_generated"] == 0
assert report["rejected"][0]["reason"] == "click with too generic anchor"
assert report["rejected"][0]["validator_codes"] == ["anchor_ref_too_generic"]
def test_dry_run_hard_caps_candidates(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(session_path, [])
with pytest.raises(ValueError, match="hard-cap"):
build_report(session_path=session_path, machine_id="windows_vm", max_candidates=11)
def test_apply_requires_allow_list(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(session_path, [])
with pytest.raises(ValueError, match="--allow-list is required"):
build_report(session_path=session_path, machine_id="windows_vm", mode="apply")
def test_apply_rejects_unknown_id_in_allow_list(tmp_path):
session_path = tmp_path / "live_events.jsonl"
output_dir = tmp_path / "observed"
_write_raw_jsonl(
session_path,
[
{
"type": "key_combo",
"keys": ["win", "e"],
"window": {"title": "Bureau", "app_name": "explorer.exe"},
},
{
"type": "window_focus_change",
"to": {"title": "Executer", "app_name": "explorer.exe"},
},
],
)
with pytest.raises(ValueError, match="--allow-list-id-not-found: missing_id"):
build_report(
session_path=session_path,
machine_id="windows_vm",
output_dir=output_dir,
mode="apply",
allow_list=["missing_id"],
)
assert not list(output_dir.glob("*.yaml"))
def test_apply_atomic_rollback_on_validation_failure(tmp_path, monkeypatch):
session_path = tmp_path / "live_events.jsonl"
output_dir = tmp_path / "observed"
_write_raw_jsonl(
session_path,
[
{
"type": "key_combo",
"keys": ["win", "e"],
"window": {"title": "Bureau", "app_name": "explorer.exe"},
},
{
"type": "window_focus_change",
"to": {"title": "Executer", "app_name": "explorer.exe"},
},
],
)
def fail_validation(paths, *, repo_root):
raise ValueError("apply-validation-failed: forced")
monkeypatch.setattr(extractor, "_validate_apply_yaml_files", fail_validation)
with pytest.raises(ValueError, match="apply-validation-failed: forced"):
build_report(
session_path=session_path,
machine_id="windows_vm",
output_dir=output_dir,
mode="apply",
allow_list=["key_win_e_wait_explorer_exe"],
)
assert not list(output_dir.glob("*.yaml"))
def test_apply_writes_only_allowed_ids(tmp_path):
session_path = tmp_path / "live_events.jsonl"
output_dir = tmp_path / "observed"
_write_raw_jsonl(
session_path,
[
{
"type": "key_combo",
"keys": ["win", "e"],
"window": {"title": "Bureau", "app_name": "explorer.exe"},
},
{
"type": "window_focus_change",
"to": {"title": "Executer", "app_name": "explorer.exe"},
},
{
"type": "key_combo",
"keys": ["ctrl", "p"],
"window": {"title": "Bloc-notes", "app_name": "Notepad.exe"},
},
{
"type": "window_focus_change",
"to": {"title": "Enregistrer sous", "app_name": "Notepad.exe"},
},
],
)
report = build_report(
session_path=session_path,
machine_id="windows_vm",
output_dir=output_dir,
mode="apply",
allow_list=["key_win_e_wait_explorer_exe"],
)
assert report["mode"] == "apply"
assert report["allow_list"] == ["key_win_e_wait_explorer_exe"]
assert report["summary"]["would_write"] == 1
assert report["summary"]["written"] == 1
assert report["applied"] == [
{
"competence_id": "key_win_e_wait_explorer_exe",
"path": str(output_dir / "key_win_e_wait_explorer_exe.yaml"),
}
]
assert (output_dir / "key_win_e_wait_explorer_exe.yaml").is_file()
assert not (output_dir / "key_ctrl_p_wait_notepad_exe.yaml").exists()
def test_apply_respects_max_candidates_cap(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(session_path, [])
with pytest.raises(ValueError, match="hard-cap"):
build_report(
session_path=session_path,
machine_id="windows_vm",
mode="apply",
allow_list=["key_win_r_wait_explorer_exe"],
max_candidates=11,
)
def test_markdown_report_includes_candidate_summary(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "key_combo",
"keys": ["ctrl", "s"],
"window": {"title": "Bloc-notes", "app_name": "Notepad.exe"},
},
{
"type": "window_focus_change",
"to": {"title": "Enregistrer sous", "app_name": "Notepad.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
markdown = render_markdown_report(report)
assert "# Extraction report" in markdown
assert "key_ctrl_s_wait_notepad_exe" in markdown
assert "wait_for_state" in markdown
def test_azerty_ctrl_s_trace_is_normalized_for_candidate(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "key_combo",
"keys": ["shift", "ctrl", "@"],
"window": {"title": "WordPad", "app_name": "WordPad.exe"},
},
{
"type": "window_focus_change",
"to": {"title": "Enregistrer sous", "app_name": "WordPad.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
candidate = report["candidates"][0]
assert candidate["competence_id"] == "key_ctrl_s_wait_wordpad_exe"
assert candidate["validator_status"] == "would_pass"
assert candidate["apply_eligible"] is True
def test_ctrl_s_control_character_trace_is_normalized_for_candidate(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "key_combo",
"keys": ["shift", "ctrl", "\x13"],
"window": {"title": "Bloc-notes", "app_name": "Notepad.exe"},
},
{
"type": "window_focus_change",
"to": {"title": "Enregistrer sous", "app_name": "Notepad.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
candidate = report["candidates"][0]
assert candidate["competence_id"] == "key_ctrl_s_wait_notepad_exe"
assert candidate["validator_status"] == "would_pass"
def test_text_input_candidate_is_below_apply_threshold(tmp_path):
session_path = tmp_path / "live_events.jsonl"
_write_raw_jsonl(
session_path,
[
{
"type": "text_input",
"text": "hello",
"window": {"title": "Bloc-notes", "app_name": "Notepad.exe"},
},
{
"type": "heartbeat",
"window": {"title": "Bloc-notes", "app_name": "Notepad.exe"},
},
],
)
report = build_report(session_path=session_path, machine_id="windows_vm")
candidate = report["candidates"][0]
assert candidate["primitive_refs"] == ["text_input_focused"]
assert candidate["confidence"] < report["summary"]["apply_min_confidence"]
assert candidate["apply_eligible"] is False
assert "below_apply_confidence_threshold" in candidate["quality_flags"]