refactor: route workflow sessions through unit of work

Co-authored-by: Copilot App <[email protected]>
This commit is contained in:
Jim Lancaster
2026-08-23 18:43:07 -05:00
co-authored by Copilot App
parent 3873810022
commit efbae26f16
3 changed files with 50 additions and 2 deletions
+23
View File
@@ -64,3 +64,26 @@ def test_no_service_module_imports_another_service_module():
name: sorted(_imported_sibling_modules(tree) & set(modules) - {name}) for name, tree in modules.items()
}
assert {name: found for name, found in violations.items() if found} == {}
def _foreign_session_scope_accesses(tree: ast.Module) -> list[int]:
lines: list[int] = []
for node in ast.walk(tree):
if isinstance(node, ast.Attribute) and node.attr == "_session_scope":
if isinstance(node.value, ast.Name) and node.value.id == "self":
continue
lines.append(node.lineno)
return sorted(lines)
def test_session_scope_is_not_accessed_via_other_services():
"""P4-1: orchestration must use a shared unit-of-work entry point, not private service scopes."""
violations: dict[str, list[int]] = {}
for path in _module_paths():
if path.stem == "base":
continue
tree = ast.parse(path.read_text(encoding="utf-8"))
found = _foreign_session_scope_accesses(tree)
if found:
violations[path.name] = found
assert violations == {}