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
16 changes: 16 additions & 0 deletions examples/semantic-vocabulary-drift-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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:]:
Expand Down
8 changes: 6 additions & 2 deletions loopx/semantics/production.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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]:
Expand Down
56 changes: 44 additions & 12 deletions loopx/semantics/python_production.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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()
Expand All @@ -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)
Expand All @@ -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:
Expand All @@ -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):
Expand Down
33 changes: 28 additions & 5 deletions tests/architecture/test_semantic_production.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand All @@ -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():
Expand Down Expand Up @@ -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
37 changes: 37 additions & 0 deletions tests/architecture/test_semantic_python_production.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)