diff --git a/examples/semantic-vocabulary-drift-smoke.py b/examples/semantic-vocabulary-drift-smoke.py index 5e3b06d189..2049a4e8cf 100755 --- a/examples/semantic-vocabulary-drift-smoke.py +++ b/examples/semantic-vocabulary-drift-smoke.py @@ -463,6 +463,21 @@ def _producer_literals(field: str, source: SourceFile) -> set[str]: return set().union(*(row.values for row in rows)) +def summarise_blockers(sites: list[str]) -> str: + """Count reported sites by blocker so the total is actionable, not opaque. + + `argument_name_only` and `annotation_only` can never become evidence: the + first is a field-named keyword argument, the second a bare declaration. + Counting them with resolvable paths would make the total look reducible. + """ + + counts: dict[str, int] = {} + for site in sites: + label = site.rpartition('[')[2].rstrip(']') or 'other' + counts[label] = counts.get(label, 0) + 1 + return ','.join(f"{label}={counts[label]}" for label in sorted(counts)) + + def check_producers(registry: dict[str, Any], sources: list[SourceFile]) -> list[str]: unknown: list[str] = [] for name, vocabulary in registry['vocabularies'].items(): @@ -710,6 +725,7 @@ def main() -> int: print(" " + " ".join(budgets)) print(" " + twins) print(f" unresolved_producer_sites={len(unknown_producers)} (not proven safe)") + print(" unresolved_producer_blockers=" + summarise_blockers(unknown_producers)) uncovered = [name for name, v in registry['vocabularies'].items() if v['tier'] == 'kernel' and 'producers' not in v] print(f" kernel_producer_coverage_pending={','.join(uncovered)}") if '--report' in sys.argv[1:]: diff --git a/loopx/semantics/production.py b/loopx/semantics/production.py index 14cc6faeba..0b12cbaa61 100644 --- a/loopx/semantics/production.py +++ b/loopx/semantics/production.py @@ -126,7 +126,8 @@ def _typescript_scan( and isinstance(error.get('line'), int) and error['line'] > 0): raise ValueError(f"{error['path']}:{error['line']}: invalid TypeScript source; repair syntax before semantic scanning") raise ValueError('TypeScript production parser failed; run npm ci --ignore-scripts and check the Node runtime') - rows.extend(Production(r['site'], r['line'], r['form'], frozenset(r['values']), r['unresolved']) + rows.extend(Production(r['site'], r['line'], r['form'], frozenset(r['values']), r['unresolved'], + 'typescript_dynamic' if r['unresolved'] else None) for r in json.loads(completed.stdout)) return rows @@ -191,7 +192,10 @@ def validate_production( undeclared = sorted({row.site for row in outputs if row.values and row.site not in declared}) if undeclared: raise ValueError(f'{name}: undeclared producer sites: {undeclared}') - return sorted({f'{row.site}:{row.line}' for row in rows if row.unresolved}) + # The label says why the unknown stayed unknown, so the count is actionable: + # `argument_name_only` and `annotation_only` can never become evidence. + return sorted({f'{row.site}:{row.line} [{row.blocker or "other"}]' + for row in rows if row.unresolved}) def probe_turn_result_input_domain(vocabulary: dict[str, Any]) -> list[Production]: diff --git a/loopx/semantics/python_production.py b/loopx/semantics/python_production.py index 8fb398cf31..3bcf1be656 100644 --- a/loopx/semantics/python_production.py +++ b/loopx/semantics/python_production.py @@ -21,6 +21,18 @@ class Production: form: str values: frozenset[str] unresolved: bool + blocker: str | None = None + """Why the unknown portion stayed unknown; ``None`` when fully resolved. + + ``argument_name_only`` a field-named keyword argument, which never proves an + output role. ``unstable_local`` a parameter, reassignment or shadowed name. + ``call_result`` the value comes back from a call. ``dynamic_key`` a computed + or non-literal subscript. ``serialized_value`` a string where an enum object + was required. ``annotation_only`` a bare annotation that declares the field + without a value. ``attribute_read`` an attribute of an unresolved object. + ``typescript_dynamic`` the TypeScript scanner could not resolve the write. + ``other`` anything else; it keeps the site visible. + """ def _module(path: str) -> str: @@ -373,29 +385,35 @@ def index_value(node: ast.AST) -> str | int | None: def lookup(container: ast.AST | None, key: str | int | None) -> tuple[list[ast.AST], bool]: if isinstance(container, (ast.Tuple, ast.List)): if type(key) is int: - return ([container.elts[key]], False) if -len(container.elts) <= key < len(container.elts) else ([], True) + return ([container.elts[key]], False) if -len(container.elts) <= key < len(container.elts) else ([], blocked('dynamic_key')) if key is not None: - return [], True - return list(container.elts), True + return [], blocked('dynamic_key') + return list(container.elts), blocked('dynamic_key') if isinstance(container, ast.Dict): keys = [index_value(k) if k is not None else None for k in container.keys] if key is not None and all(k is not None for k in keys): # Python dict construction keeps the last duplicate key. found = [v for k, v in zip(keys, container.values, strict=True) if k == key] - return ([found[-1]], False) if found else ([], True) - return list(container.values), True - return [], True + return ([found[-1]], False) if found else ([], blocked('dynamic_key')) + return list(container.values), blocked('dynamic_key') + return [], blocked('unstable_local' if isinstance(container, ast.Name) else 'other') + + blockers: list[str] = [] + + def blocked(label: str) -> bool: + blockers.append(label) + return True def resolve(node: ast.AST | None, seen: frozenset[str] = frozenset(), *, enum_only: bool = False) -> tuple[set[str], bool]: node, seen = bound(node, seen) if isinstance(node, ast.Constant): if isinstance(node.value, str): return ({node.value} if node.value and not enum_only else set()), False - return set(), node.value is not None + return set(), node.value is not None and blocked('other') if isinstance(node, ast.Subscript): if isinstance(node.slice, ast.Slice) or (isinstance(node.slice, ast.Constant) and type(node.slice.value) not in (str, int)): - return set(), True + return set(), blocked('dynamic_key') container, visited = bound(node.value, seen) choices, unknown = lookup(container, index_value(node.slice)) known: set[str] = set() @@ -422,7 +440,14 @@ def resolve(node: ast.AST | None, seen: frozenset[str] = frozenset(), *, enum_on return {members[member.attr]}, False if node.attr == 'value' and isinstance(node.value, ast.Name): return enum_object_value(node.value, seen) - return set(), True + return set(), blocked('attribute_read') + if isinstance(node, ast.Call): + return set(), blocked('call_result') + if isinstance(node, ast.Name): + return set(), blocked('unstable_local') + if node is None: + return set(), blocked('other') + return set(), blocked('other') def enum_object_value(node: ast.AST, seen: frozenset[str]) -> tuple[set[str], bool]: node, seen = bound(node, seen) @@ -433,7 +458,7 @@ def enum_object_value(node: ast.AST, seen: frozenset[str]) -> tuple[set[str], bo return resolve(node, seen, enum_only=True) # A serialized string (including Action.RUN.value) is not an enum # object with another .value attribute. - return set(), True + return set(), blocked('serialized_value') def returned(node: ast.AST | None, path: tuple[str | int, ...], seen: frozenset[str] = frozenset()) -> tuple[set[str], bool]: if not path: @@ -444,10 +469,17 @@ def returned(node: ast.AST | None, path: tuple[str | int, ...], seen: frozenset[ return set().union(*(v for v, _ in parts)), unknown or any(u for _, u in parts) def record(node: ast.AST | None, form: str, location: ast.AST) -> None: + blockers.clear() values, unknown = (returned(node, return_paths.get(scope, ())) if form == 'return' else resolve(node)) + blocker = blockers[0] if blockers else None + if node is None and form == 'assignment': + # A bare annotation declares the field; there is no value to resolve. + blocker = 'annotation_only' if form == 'keyword_unproved': - unknown = True # Argument name alone does not prove an output role. - result.append(Production(f'{source.path}::{scope}', location.lineno, form, frozenset(values), unknown)) + # Argument name alone does not prove an output role. + unknown, blocker = True, 'argument_name_only' + result.append(Production(f'{source.path}::{scope}', location.lineno, form, + frozenset(values), unknown, blocker if unknown else None)) for node in nodes: if isinstance(node, ast.Assign): diff --git a/tests/architecture/test_semantic_production.py b/tests/architecture/test_semantic_production.py index 68080a929a..dc80f88bd3 100644 --- a/tests/architecture/test_semantic_production.py +++ b/tests/architecture/test_semantic_production.py @@ -53,8 +53,8 @@ def vocabulary(): 'literal_scan': {'field': 'action'}} -def row(value, *, site=SITE, unresolved=False): - return Production(site, 1, 'return', frozenset([value]) if value else frozenset(), unresolved) +def row(value, *, site=SITE, unresolved=False, blocker=None): + return Production(site, 1, 'return', frozenset([value]) if value else frozenset(), unresolved, blocker) def test_owner_values_never_satisfy_production_liveness(): @@ -74,9 +74,12 @@ def test_registering_an_unrelated_function_does_not_cover_a_writer(): def test_dynamic_path_remains_visible_and_cannot_supply_missing_value(): with pytest.raises(ValueError, match='no observed producer'): - validate_production('action', vocabulary(), [row('run'), row(None, unresolved=True)]) - unknown = validate_production('action', vocabulary(), [row('run'), row('wait'), row(None, unresolved=True)]) - assert unknown == [SITE + ':1'] + validate_production('action', vocabulary(), [row('run'), row(None, unresolved=True, blocker='call_result')]) + unknown = validate_production('action', vocabulary(), + [row('run'), row('wait'), row(None, unresolved=True, blocker='call_result')]) + # The reported entry names the site and why it stayed unknown. Passing the + # blocker explicitly keeps a lost label from passing as the `other` fallback. + assert unknown == [f'{SITE}:1 [call_result]'] def test_compatibility_values_must_have_no_observed_production(): @@ -288,3 +291,23 @@ def test_input_witness_runs_only_for_the_anchored_site(): rows = collect_production(ROOT, anchored, []) assert {r.form for r in rows} == {'input_witness'} assert set().union(*(r.values for r in rows)) == set(values) + + +def test_reported_sites_carry_a_blocker_label_and_summarise(): + import runpy + + from loopx.semantics.inventory import load_sources + + smoke = runpy.run_path(str(ROOT / 'examples/semantic-vocabulary-drift-smoke.py'), + run_name='not_main') + registry = smoke['load_registry']() + sites = smoke['check_producers'](registry, load_sources(ROOT)) + assert sites, 'the repository still has unresolved producer sites to describe' + assert all(site.endswith(']') and ' [' in site for site in sites) + summary = smoke['summarise_blockers'](sites) + counted = sum(int(part.split('=')[1]) for part in summary.split(',')) + assert counted == len(sites) + # These two can never become evidence, so they must stay separable from the + # paths a future slice could still resolve. + assert 'argument_name_only=' in summary + assert 'annotation_only=' in summary diff --git a/tests/architecture/test_semantic_python_production.py b/tests/architecture/test_semantic_python_production.py index 532deacfa5..faf6299705 100644 --- a/tests/architecture/test_semantic_python_production.py +++ b/tests/architecture/test_semantic_python_production.py @@ -317,3 +317,40 @@ def test_second_reexport_hop_and_untracked_module_stay_unknown(): assert known(rows) == set() and all(r.unresolved for r in rows) rows = scan('from .compat import Action\ndef emit():\n return Action.RUN.value\n', returns=['emit'], modules={}) assert known(rows) == set() and all(r.unresolved for r in rows) + + +def blockers(rows): + return {r.blocker for r in rows if r.unresolved} + + +@pytest.mark.parametrize('text, expected', [ + # a field-named keyword argument never proves an output role + ('from .owner import Action\ndef emit(p):\n return Packet(action=p)\n', 'argument_name_only'), + # a bare annotation declares the field; there is no value to resolve + ('class Packet:\n action: str\n', 'annotation_only'), + # a parameter cannot borrow owner values + ('def emit(action):\n p = {"action": action}\n return p\n', 'unstable_local'), + # the value comes back from a call + ('def emit():\n return {"action": compute()}\n', 'call_result'), + # a computed subscript is not a literal key + ('def emit(key, table):\n return {"action": table[key]}\n', 'unstable_local'), +]) +def test_unresolved_rows_say_why_they_stayed_unknown(text, expected): + rows = scan(text) + assert expected in blockers(rows), (expected, [(r.form, r.blocker) for r in rows]) + + +def test_resolved_rows_carry_no_blocker(): + rows = scan('from .owner import Action\ndef emit():\n return {"action": Action.RUN.value}\n') + assert rows and all(r.blocker is None for r in rows if not r.unresolved) + + +def test_every_unresolved_row_is_labelled(): + """An unlabelled unknown would be invisible in the report breakdown.""" + + rows = scan('from .owner import Action\ndef emit(p, table, key):\n' + ' a = {"action": p}\n' + ' b = {"action": table[key]}\n' + ' c = Packet(action=compute())\n' + ' return [a, b, c]\n') + assert all(r.blocker for r in rows if r.unresolved)