Knowledge Wiki Orchestrata Ultimate

scripts.serialize_codebase_ast

Reference › scripts › serialize_codebase_ast

scripts/serialize_codebase_ast.py is the AST toolkit: a full codebase AST serializer, a deep semantic comparison engine and a large family of read-only code-intelligence subcommands in one 13,800-line module. It is the main way to inspect Python structure in this project without reading source files. A human or an agent runs it, usually through the deployed toolkit or the resident ast_daemon (the --no-daemon flag forces a direct run); orchestrata_ultimate.agents.volks.codedirector.implementation and orchestrata_ultimate.factory.evidence.mutation import it, and gen_brief_pack and gen_codebase_map call it. Parsing, the data model, scope and inheritance analysis come from the orchestrata_ast package (analyzer, inheritance, models, roots, scope, serialize) and from orchestrata_tools.

_build_parser defines the command line, with a global --verbose, and _dispatch runs the chosen subcommand. main calls _dispatch and always flushes the warm-cache sidecar in a finally block, because some handlers exit mid-dispatch (for example find-cycles --strict). The subcommands include serialize (write the full AST of one or more --source trees to a JSON --output under a --name; the first source is the primary root for dotted module paths and the default comes from ORCHESTRATA_AST_ROOTS), compare and deep-compare, symbols, refs, who-imports, source, call-graph, find-cycles, orphans, unused, dead-params, untyped, complexity, nesting, long-params, branches, interface, docs, diagram, audit, pattern, check-deps, check-traces, trace-audit, trace-lint, ast-to-kg, and refactoring helpers such as rename, inline, extract-method, extract-class and move-symbol, plus generators (gen-tests, testgen, gen-dto, gen-enum, gen-volk, gen-protocol, gen-skeleton, gen-facade, gen-serde, skeleton-diff).

A parse cache keyed by a content hash of each file (never its modification time) makes repeat runs cheap: configure_parse_cache and parse_cache_stats manage it, and a persisted sidecar is loaded lazily and written back at exit. Stale temporary sidecar files left by dead processes are removed by _reap_dead_sidecar_temps. configure_repo_root sets the served tree.

At a glance
Source scripts/serialize_codebase_ast.py
Lines of code 13843
Classes 14
Functions 368
Imports (internal) orchestrata_ast.analyzer, orchestrata_ast.inheritance, orchestrata_ast.models, orchestrata_ast.roots, orchestrata_ast.scope, orchestrata_ast.serialize, orchestrata_tools.test_generator.testgen, orchestrata_ultimate.core.atomic_json, orchestrata_ultimate.core.graph_factory
Imported by orchestrata_ultimate.agents.volks.codedirector.implementation, orchestrata_ultimate.factory.evidence.mutation
Tests that import it none

Class diagram

Class diagram scripts.serialize_codebase_ast

Dependency diagram

Dependency diagram scripts.serialize_codebase_ast

API

Constants

Name Type Value
_DATA_ROOT Path | None None
_CACHE_STATE dict[str, Any] {'enabled': False, 'sidecar_path': None, 'data': None, 'dirty': False, 'hits': …
_PARSE_CACHE dict[str, tuple[tuple[int, int], tuple[str, ast.Module]]] {}
_DEF_BEARING_NODES tuple[type[ast.AST], ...] (ast.stmt, ast.excepthandler, ast.match_case)
_SOURCE_ROWS_CACHE dict[str, tuple[ast.Module, tuple[tuple[str, str, int, int], ...]]] {}
_SKEL_KNOWN_TYPE_IMPORTS dict[str, str] {'Tool': 'orchestrata_ultimate.tools.base', 'ToolResult': 'orchestrata_ultimate…
_SKELETON_EMITTERS dict[str, _SkeletonEmitter] {'Protocol': _ProtocolEmitter(), 'dataclass': _DataclassEmitter(), 'enum': _Enu…
_COMMANDS dict[str, Callable[[argparse.Namespace], None]] {'serialize': _cmd_serialize, 'compare': _cmd_compare, 'deep-compare': _cmd_dee…

Classes

_DaemonMiss

Line 577

Sentinel: no usable daemon result — caller must compute locally.

_UnixHTTPConnection

Bases: http.client.HTTPConnection · Line 588

http.client.HTTPConnection over an AF_UNIX socket.

Method Signature Summary
__init__ __init__(socket_path: str, timeout: float) -> None
connect connect() -> None

_SkelParam (dataclass)

Decorators: dataclass · Line 5139

One parsed method parameter (name + optional Mermaid type + default).

Attribute Type Default
name str
type str | None None
default str | None None

_SkelField (dataclass)

Decorators: dataclass · Line 5148

One parsed class member (+name: str / -_sandbox: ReplSandbox).

Attribute Type Default
name str
type str | None None
default str | None None
forced_property bool False
Method Signature Summary
is_public property is_public() -> bool

_SkelMethod (dataclass)

Decorators: dataclass · Line 5166

One parsed method (+execute(args: ReplArgs) ToolResult).

Attribute Type Default
name str
params list[_SkelParam] field(default_factory=list)
return_type str | None None
is_async bool False
reqs list[str] | None None
role str | None None
is_classmethod bool False
is_staticmethod bool False
is_property bool False
is_abstract bool False
Method Signature Summary
is_public property is_public() -> bool

_SkelClass (dataclass)

Decorators: dataclass · Line 5189

One parsed class X { <<stereotype>> ... } block.

Attribute Type Default
name str
stereotype str ''
bases list[str] field(default_factory=list)
fields list[_SkelField] field(default_factory=list)
methods list[_SkelMethod] field(default_factory=list)
spec str | None None
story str | None None
reqs list[str] | None None
line int 0
namespace str ''
type_params list[str] field(default_factory=list)
mode str 'generate'
extend_target str | None None
_pending_method_reqs dict[str, list[str]] field(default_factory=dict)
_pending_method_roles dict[str, str] field(default_factory=dict)

_SkelEmitContext (dataclass)

Decorators: dataclass · Line 5675

Threaded through every emitter + method-rendering helper below:

Attribute Type Default
classes dict[str, _SkelClass]
known_types dict[str, str]
type_params set[str] field(default_factory=set)
_property_cache dict[tuple[str, str, str], bool | None] field(default_factory=dict)
Method Signature Summary
base_property_status base_property_status(module_path: str, class_name: str, attr_name: str) -> bool | None

_SkeletonEmitResult

Line 6101

What one emitter contributes to the combined module.

Method Signature Summary
__init__ __init__(lines: list[str], imports: set[str], unresolved: set[str], leaf_spans: list[tuple[str, int, int]] | None = None) -> None

_SkeletonEmitter

Line 6121

Structural base for the per-stereotype Strategy emitters, selected by the _select_skeleton_emitter Factory.

Method Signature Summary
emit emit(ctx: _SkelEmitContext) -> _SkeletonEmitResult

_ProtocolEmitter

Bases: _SkeletonEmitter · Line 6132

<<Protocol>> -> typing.Protocol, ... method bodies.

Method Signature Summary
emit emit(ctx: _SkelEmitContext) -> _SkeletonEmitResult

_DataclassEmitter

Bases: _SkeletonEmitter · Line 6181

<<dataclass>> -> @dataclass(slots=True) typed fields.

Method Signature Summary
emit emit(ctx: _SkelEmitContext) -> _SkeletonEmitResult

_EnumEmitter

Bases: _SkeletonEmitter · Line 6247

<<enum>> -> StrEnum with each field name a lowercased member, UNLESS the diagram declares an explicit = value (fix 4).

Method Signature Summary
emit emit(ctx: _SkelEmitContext) -> _SkeletonEmitResult

_ToolEmitter

Bases: _SkeletonEmitter · Line 6409

<<Tool>> -> a class COMPOSING injected collaborators (never a deep inheritance chain), realizing whatever ABC/Protocol its ..|> X relationship names (Tool — gen-skeleton's own baseline…

Method Signature Summary
emit emit(ctx: _SkelEmitContext) -> _SkeletonEmitResult

_PlainClassEmitter

Bases: _SkeletonEmitter · Line 6470

<<class>> — a plain composed service class with no ABC/Protocol of its own (unless it realizes one via ..|> X); needed once a real analyst contract (SPEC-072) used it for ordinary composed…

Method Signature Summary
emit emit(ctx: _SkelEmitContext) -> _SkeletonEmitResult

Functions

Function Signature Summary
configure_repo_root configure_repo_root(path: Path | None) -> Path Set (or, with None, clear) the served tree's root; idempotent.
_served_root _served_root() -> Path The served tree's root: the configured data root, else the toolkit's own.
_locate _locate(path: str | Path) -> Path path anchored on the configured data root when it is relative.
configure_parse_cache configure_parse_cache(cache_path: str | Path | None, enabled: bool = True) -> None Configure the process-wide read-only AST warm cache.
parse_cache_stats parse_cache_stats() -> dict[str, int] {"hits": N, "misses": M} since the last configure_parse_cache.
_file_sha1 _file_sha1(text: str) -> str Content hash of file text — the freshness authority (never mtime).
_load_sidecar _load_sidecar() -> dict[str, Any] Lazily load (and memoize) the warm-cache sidecar for this process.
_pid_alive _pid_alive(pid: int) -> bool True while a process with pid exists (a dead-PID janitor's liveness check).
_parse_temp_pid _parse_temp_pid(middle: str) -> int | None Best-effort pid parse from the <middle> of <sidecar>.<middle>.tmp.
_reap_dead_sidecar_temps _reap_dead_sidecar_temps(path: Path) -> None Unlink sibling <sidecar>.<pid>.tmp files that are both OLD and (pid dead or unresolvable).
_save_sidecar _save_sidecar() -> None Persist the sidecar if it changed, pruning entries for deleted files.
_atomic_write_json _atomic_write_json(path: Path, payload: Any, *, default: Callable[[Any], Any] | None = None) -> None Write payload as JSON so a concurrent writer cannot shred it.
_cached_extract _cached_extract(path: Path, text: str, kind: str, compute: Callable[[], T]) -> T Per-file DERIVED-extraction cache (kind-scoped), sha1-keyed.
_daemon_socket_path _daemon_socket_path() -> str | None The daemon's Unix socket path, or None if nothing is listening there.
_daemon_http _daemon_http(sock_path: str, method: str, path: str, timeout: float, body: bytes | None = None) -> Any One request/response round-trip over the daemon's socket, or None on ANY failure (timeout, refused, non-200, malformed JSON) — the caller treats None as "delegation unavailable right now".
_log_daemon_identity _log_daemon_identity(health: Any) -> None Log the daemon's /health identity at DEBUG, once per process.
query_ast_daemon query_ast_daemon(cmd: str, params: dict, use_daemon: bool = True) -> Any Delegate a read-only query to the resident AST daemon.
_add_daemon_arg _add_daemon_arg(p: argparse.ArgumentParser) -> None --no-daemon for a subcommand that can delegate to the resident scripts/ast_daemon.py (see query_ast_daemon).
compare_codebases compare_codebases(ast_a: dict, ast_b: dict) -> dict Compare two serialized codebase ASTs.
index_codebase index_codebase(ast_data: dict) -> dict Build fast lookup dicts from serialized AST data.
resolve_chain resolve_chain(qualified: str, graph: dict, visited: set | None = None) -> list[str] Resolve full inheritance chain for a class (recursive).
summarize_method summarize_method(m: dict) -> dict Compact method summary for reporting.
compare_methods_deep compare_methods_deep(a_methods: list, b_methods: list) -> dict Deep method-level comparison with full signature diffs.
classify_readiness classify_readiness(a_cls: dict | None, b_cls: dict | None) -> str Classify porting readiness of a class pair based on method coverage ratio.
pair_classes pair_classes(a_idx: dict, b_idx: dict, class_mappings: dict, name_transforms: list[dict]) -> tuple[list[dict], set[str], set[str]] Pair classes semantically across two codebases using 3-pass strategy.
analyze_pair analyze_pair(pair: dict, a_graph: dict, b_graph: dict) -> dict Full analysis of a paired class: methods, bases, inheritance, vars.
path_to_package path_to_package(path: str) -> str Map an orchestrata module path to its source package name.
analyze_subsystems analyze_subsystems(b_data: dict, a_idx: dict, tier_data: dict, mod_map: dict) -> list[dict] Per-package analysis of orchestrata with ultimate coverage check.
coverage_gaps_by_tier coverage_gaps_by_tier(subsystems: list[dict]) -> dict Aggregate subsystem data by porting tier.
audit_imports audit_imports(b_data: dict, mod_map: dict) -> dict Audit all orchestrata internal imports and determine translations.
compare_dep_graphs compare_dep_graphs(a_data: dict, b_data: dict) -> dict Compare module dependency graphs — external package analysis.
complexity_metrics complexity_metrics(a_idx: dict, b_idx: dict) -> dict Complexity metrics for both codebases.
deep_compare deep_compare(ast_a: dict, ast_b: dict, mapping: dict) -> dict Deep semantic comparison: ultimate (a) vs orchestrata (b) using mapping table.
_normalize_dist _normalize_dist(spec: str) -> str Normalize a dependency spec/name to a comparable distribution name.
_declared_dependencies _declared_dependencies(pyproject_path: Path) -> set[str] All distribution names declared in pyproject (deps + groups + extras).
_collect_imports _collect_imports(sources: list[Path]) -> dict[str, list[dict]] Walk every .py under sources and collect each top-level external import module -> [{file, line, lazy}].
_local_module_names _local_module_names(repo_root: Path) -> set[str] Every module/package name defined locally under src/ scripts/ tests/ claude/.
find_undeclared_dependencies find_undeclared_dependencies(sources: list[Path], pyproject_path: Path, repo_root: Path) -> dict Find third-party imports that are NOT declared in pyproject.
_parse_cached _parse_cached(path: Path) -> tuple[str, ast.Module] | None Parse path to (text, tree), memoized by (mtime_ns, size).
_iter_py_files _iter_py_files(sources: SourceList) -> Iterator[tuple[Path, str, ast.Module]] Yield (path, text, tree) for every parseable .py under sources.
_func_signature _func_signature(node: ast.FunctionDef | ast.AsyncFunctionDef) -> str
_class_signature _class_signature(node: ast.ClassDef) -> str
_extract_symbols _extract_symbols(tree: ast.AST, path: str | Path) -> list[dict] Every class/func/method row in tree (one file) — the per-file unit list_symbols caches (kind "symbols") and _find_function reuses for a fast, non-walking lookup.
list_symbols list_symbols(sources: SourceList, pattern: str | None = None, kind: str | None = None) -> list[dict] List classes/functions/methods (qualified name + signature + location).
_extract_refs_index _extract_refs_index(tree: ast.AST, path: str | Path) -> dict Every definition/use/import occurrence in tree (one file), each tagged with the name it binds/refers to — the per-file unit find_references caches (kind "refs_index").
_refs_index_for_file _refs_index_for_file(args: tuple[Path, str, ast.Module]) -> dict The refs_index derived extraction for ONE file — the thread-pool unit of work (cold refs walks every tree serially: ~8s of a 17s cold call measured 2026-08-11; read-only extractions scale…
find_references find_references(name: str, sources: SourceList) -> dict Every reference to name, split into definitions / uses / imports.
_owning_module_for _owning_module_for(file_path: Path, repo_root: Path) -> str | None Dotted import path for file_path, matching how the rest of the repo actually spells it: a leading src path component is stripped (the installed package root — src/orchestrata_ultimate/x.py…
_resolve_symbol_aliases _resolve_symbol_aliases(symbol: str, owning_module: str, sources: SourceList) -> dict Every genuine reference to symbol (defined in owning_module) anywhere under sources — resolving each file's LOCAL binding for the symbol first, then searching that same file for uses of the…
_find_string_literal_uses _find_string_literal_uses(name: str, sources: SourceList) -> list[dict] Every occurrence of name as an EXACT string constant under sources — WEAK evidence (the same graded-evidence concept as find_unused's string_ref_locations: dynamic dispatch via…
alias_aware_references alias_aware_references(name: str, sources: SourceList, repo_root: Path, module: str | None = None, base: dict | None = None) -> dict find_references enriched with alias resolution (DEFECT 2) and graded string-literal evidence (the same false-negative fix as find_unused's ref_sources grading).
_pick_module_root _pick_module_root(path: str | Path, roots: list[Path]) -> Path | None The most specific roots entry (longest matching path) path is nested under, or None if it isn't under any of them.
_dotted_module_name _dotted_module_name(path: str | Path, root: Path) -> str Dotted module name for path relative to root — the toolkit's module-path convention (e.g. core.lru_cache for <root>/core/lru_cache.py, __init__ stripped to the package name).
_iter_source_modules _iter_source_modules(sources: SourceList) -> Iterator[tuple[str, Path, str, ast.Module]] Yield (module_name, path, text, tree) for every file _iter_py_files finds under sources, with module_name computed relative to whichever --source root contains it (a bare file source…
_extract_symbol_nodes _extract_symbol_nodes(tree: ast.AST) -> Iterator[tuple[str, str, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef]] Yield (qualname, kind, node) for every class/func/method in tree, mirroring _extract_symbols's traversal (qualname is the bare name for classes/top-level funcs, Class.method inside a…
_node_span _node_span(node: ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef) -> tuple[int, int] (start_line, end_line), 1-indexed inclusive. start is pulled back to the earliest decorator line — ast.get_source_segment excludes decorators, but source must include them.
_source_symbol_rows _source_symbol_rows(path: str | Path, tree: ast.Module) -> tuple[tuple[str, str, int, int], ...] (qualname, kind, start, end) for every class/func/method in tree, computed once per parsed tree (see _SOURCE_ROWS_CACHE).
_slice_source _slice_source(text: str, start: int, end: int) -> str Bytes [start, end] (1-indexed, inclusive line numbers) of text, keeping line endings — byte-identical to that span of the file.
_module_residue _module_residue(text: str, tree: ast.Module) -> str text minus every TOP-LEVEL def/class span (imports, constants, the module docstring, other module-level statements survive) — nested defs are already excluded as part of their parent's top-level…
_qualify _qualify(candidate: dict) -> str Display name for a resolve_source symbol candidate: its module dotted-prefixed onto its qualname (or the bare qualname for a rootless module, e.g. a lone top-level script).
resolve_source resolve_source(name: str, sources: SourceList) -> dict Resolve name — a qualified symbol (module.Class.method, module.Class, module.function, a dotted module path) or a bare name — to exactly one class/func/method/module under sources.
resolve_source_payload resolve_source_payload(name: str, sources: SourceList, *, residue: bool = False) -> dict The full source subcommand result: resolve name via :func:resolve_source, then materialize the exact byte-identical source span (or --residue) into a plain, JSON-serializable payload —…
_def_name_span _def_name_span(line: str, name: str, keyword: str) -> tuple[int, int] | None Column span of name right after def/class on a def line.
_binds_old_in_stmt _binds_old_in_stmt(node: ast.AST, old: str) -> bool Scope-respecting search for a genuine (non-import) binding of old.
_scope_binds_old _scope_binds_old(func_node: ast.FunctionDef | ast.AsyncFunctionDef | ast.Lambda, old: str) -> bool True if old is a GENUINE local binding of func_node's own scope.
_shadow_line_ranges _shadow_line_ranges(source: str, old: str) -> list[tuple[int, int]] Line ranges of function scopes where old is a local binding.
_imported_module_names _imported_module_names(tree: ast.AST) -> set Local names that refer to imported modules — roots of qualified references.
_attr_root_name _attr_root_name(node: ast.expr) -> str | None Root identifier of an attribute chain (a.b.c -> a); None if dynamic.
_module_path_span _module_path_span(line: str, old: str, start_col: int, end_col: int) -> tuple[int, int] | None Column span of old as a dotted-path segment within line[start:end].
_rename_spans _rename_spans(text: str, tree: ast.AST, old: str, new: str, kind: str, class_name: str | None = None) -> list[tuple[int, int, int]] List of (lineno, col_start, col_end) identifier spans to rewrite.
_apply_spans _apply_spans(text: str, spans: list[tuple[int, int, int]], new: str) -> str Rewrite each (line, c0, c1) span to new; spans must be sorted desc.
rename_symbol rename_symbol(old: str, new: str, sources: SourceList, kind: str = 'name', apply: bool = False, class_name: str | None = None) -> dict Rename old -> new across sources (AST-token-precise, dry-run default).
_pytest_hook_names _pytest_hook_names() -> frozenset[str] The canonical set of pytest hookspec names (pytest_*) — DEFECT 3 of the 2026-08-19 false-dead-code audit (92% FP rate on a 101-item sweep).
_pytest_plugin_registration_sites _pytest_plugin_registration_sites(ref_sources: SourceList) -> dict[str, list[dict]] Best-effort structural evidence for a DEFECT-3 exemption message: every [..., "-p", "<dotted.module>", ...]-shaped argv literal (subprocess pytest invocations) under ref_sources, keyed by the…
find_unused find_unused(sources: SourceList, include_traced: bool = False, ref_sources: SourceList | None = None, exempted: list[dict] | None = None) -> list[dict] Module-level functions/classes defined but never referenced anywhere in the project.
_cyclomatic _cyclomatic(node: ast.AST) -> int McCabe cyclomatic complexity of a function body (includes nested).
measure_complexity measure_complexity(sources: SourceList, threshold: int = 1) -> list[dict] Cyclomatic complexity of every function/method, descending, >= threshold.
_int_line _int_line(d: dict, key: str = 'line') -> int | None Safely pull an int line number out of a serialized JSON-ish dict.
_augment_end_line_and_complexity _augment_end_line_and_complexity(data: object, trees: dict[str, ast.Module] | None = None) -> object Additively inject end_line (classes + functions/methods) and complexity (functions/methods only) into a serialize()d CodebaseAST dict.
generate_interface generate_interface(class_name: str, sources: SourceList) -> str | None Emit a typing.Protocol (structural interface) for a class's public API.
generate_protocol generate_protocol(class_name: str, sources: SourceList, kind: str = 'protocol', name: str | None = None) -> str | None Emit an interface for class_name's public API in one of three shapes:
generate_docs generate_docs(sources: SourceList) -> str Render a markdown API reference from docstrings + signatures.
generate_mermaid generate_mermaid(sources: SourceList, kind: str = 'class') -> str Mermaid diagram source: class (inheritance) or deps (module imports).
_try_unparse _try_unparse(node: ast.AST) -> str
_branches_of _branches_of(func: ast.AST) -> list[dict] Every decision point inside func (includes nested bodies).
list_branches list_branches(func_name: str | None, sources: SourceList) -> list[dict] Enumerate branches of each function (optionally only func_name).
_fixed_label_query _fixed_label_query(query: str) -> LiteralString Assert a Cypher query built from a FIXED (non-user-input) label is safe to run.
ast_to_kg ast_to_kg(ast_path: str, project_id: str, *, graph_uri: str | None = None, graph_user: str | None = None, graph_password: str | None = None, dry_run: bool = False) -> dict Write AST-derived signals (complexity, branch count, call-graph edges) from a serialized AST into the knowledge graph.
_dotted_module _dotted_module(path: str | Path) -> str Dotted import path for a .py file (walks up while init.py exists).
_default_for_annotation _default_for_annotation(ann: str | None) -> str A literal source for a plausible value of a type annotation.
_other_literal _other_literal(value: object) -> str
_value_for_compare _value_for_compare(op: ast.cmpop, comp: ast.expr, want_true: bool) -> str | None Literal src for the left operand of left OP comp to make it want_true.
_infer_param_values _infer_param_values(test: ast.expr, want_true: bool) -> dict Best-effort {param: literal_src} making boolean test == want_true.
_literal_return _literal_return(body: list[ast.stmt] | None) -> str | None Repr of the first literal return reachable in body (skips nested defs).
_raised_exception _raised_exception(body: list[ast.stmt] | None) -> str | None Name of the exception an arm directly raises (its first meaningful stmt).
_build_call _build_call(callee: str, params: list[dict], vals: dict) -> str callee(p=v, ...) supplying inferred or type-default values for required params (params with their own defaults are omitted unless constrained).
_trivial_ctor _trivial_ctor(class_node: ast.ClassDef) -> str ClassName(...) with inferred/type-default args for required init params.
_class_node _class_node(class_name: str, sources: SourceList) -> ast.ClassDef | None
generate_branch_tests generate_branch_tests(func_name: str, sources: SourceList) -> str Generate RUNNABLE pytest tests covering each branch of a function.
audit_error_handling audit_error_handling(sources: SourceList) -> list[dict] Advisory audit for unified logging + exception handling.
_main_guard_line_ranges _main_guard_line_ranges(tree: ast.AST) -> list[tuple[int, int]] Line ranges of if __name__ == "__main__": blocks at module level.
_max_nesting _max_nesting(func: ast.AST) -> int Deepest chain of nested compound blocks in a function (excludes nested defs).
measure_nesting measure_nesting(sources: SourceList, threshold: int = 1) -> list[dict] Max nesting depth per function/method, descending, >= threshold.
_find_class _find_class(class_name: str, sources: SourceList) -> tuple[ast.ClassDef, Path] | tuple[None, None]
_public_methods _public_methods(class_node: ast.ClassDef) -> list
_call_args _call_args(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> str
generate_pattern generate_pattern(class_name: str, pattern: str, sources: SourceList) -> str | None Scaffold a Gang-of-Four pattern around class_name (reviewable code).
_params_of _params_of(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> list[dict] Parameters of fn (excluding self/cls) with annotation + default.
_locate_function_node _locate_function_node(tree: ast.AST, qual_name: str, line: int) -> tuple[ast.FunctionDef | ast.AsyncFunctionDef | None, str | None] (FunctionDef, class_name_or_None) for a _extract_symbols row (qualified name + line) within its own already-parsed tree — a single targeted walk of ONE file, not a whole-codebase scan.…
_find_function _find_function(func_name: str, sources: SourceList) -> tuple[ast.FunctionDef | ast.AsyncFunctionDef | None, str | None, Path | None] Return (fn_node, class_name_or_None, path) for the first match.
_snake _snake(name: str) -> str CamelCase -> snake_case (ServiceSubject -> service_subject).
_dir_dotted_prefix _dir_dotted_prefix(d: str | Path) -> str Dotted package path for directory d (treating it as a package).
split_pattern_files split_pattern_files(class_name: str, pattern: str, sources: SourceList, target_dir: str | None = None, apply: bool = False) -> dict | None Generate a GoF pattern as ONE module file per generated class.
find_long_param_lists find_long_param_lists(sources: SourceList, threshold: int = 5) -> list[dict] Functions/methods whose parameter count (excl. self/cls) exceeds threshold.
generate_dto generate_dto(func_name: str, sources: SourceList, kind: str = 'dataclass') -> str | None Generate a DTO (dataclass or pydantic model) from a function's parameters.
generate_enum generate_enum(name: str, members: list[str], base: str = 'StrEnum', spec: str | None = None, story: str | None = None, reqs: list[str] | None = None, values: dict | None = None, docstring: str | None = None, traces_specs: list[dict] | None = None, import_prefix: str = 'orchestrata_ultimate') -> str Emit an enum class from a member-name list.
generate_enum_from_spec generate_enum_from_spec(spec: dict, import_prefix: str = 'orchestrata_ultimate') -> str Generate an enum class from a JSON-style specification dict.
_load_enum_spec _load_enum_spec(path: str) -> dict Load an enum spec JSON file.
_extract_class_text _extract_class_text(text: str, class_name: str) -> str Return the source text of class_name from text.
_extract_enum_from_source _extract_enum_from_source(path: str, class_name: str) -> str Return the source text of class_name from path, or raise ValueError.
check_enum_drift check_enum_drift(spec_path: str) -> list[str] Verify that enums in spec_path match their generated counterparts.
_volk_handler_ident _volk_handler_ident(subject: str) -> str Sanitize a CXP subject string into a valid _handle_<x> method name suffix — hyphens/spaces/other non-word characters become underscores, and a leading digit gets an underscore guard.
generate_volk generate_volk(class_name: str, alias: str, role: str, subjects: list[str], spec: str | None = None, reqs: list[str] | None = None, import_prefix: str = 'orchestrata_ultimate') -> str Emit the deterministic scaffold shell of a new volk: a BaseAgent subclass with alias/role fields, an __init__ calling super().__init__(), and a process() router dispatching…
generate_volk_init generate_volk_init(class_name: str, module: str) -> str Emit an __init__.py that re-exports class_name from module (the dotted import path of the sibling volk.py).
_convert_skeleton_tildes _convert_skeleton_tildes(text: str) -> str Convert Mermaid tilde generics to Python brackets: X~T~ -> X[T].
_skeleton_kv_pragma _skeleton_kv_pragma(kv: str) -> dict[str, str] Parse spec=SPEC-063 story=US-063-001 reqs=A,B into a dict.
_skeleton_reqs_list _skeleton_reqs_list(kv: dict[str, str]) -> list[str]
_split_top_level _split_top_level(s: str, sep: str) -> list[str] Split s on sep ignoring occurrences nested inside ()/[]/{}.
_split_skeleton_on_first_eq _split_skeleton_on_first_eq(s: str) -> list[str] Split s on the FIRST top-level = (ignoring ()/[]/{} nesting) — ported from diagram_to_code.DiagramToCode._split_on_first_eq so param DEFAULTS survive the round-trip (fix: param…
_parse_skeleton_params _parse_skeleton_params(raw: str) -> list[_SkelParam]
_render_skeleton_param _render_skeleton_param(p: _SkelParam) -> str
_parse_skeleton_method_line _parse_skeleton_method_line(line: str) -> None
_parse_skeleton_field_line _parse_skeleton_field_line(line: str) -> None
_parse_skeleton_pragma _parse_skeleton_pragma(line: str) -> None
_skeleton_is_self_returning_factory_name _skeleton_is_self_returning_factory_name(m: _SkelMethod) -> bool Bare create/default/from_* names on a class returning its OWN type are classmethod factories by convention, even with no explicit {classmethod} bracket marker — legacy contracts…
_finalize_skeleton_class _finalize_skeleton_class() -> None
_parse_skeleton_diagram _parse_skeleton_diagram(text: str) -> dict[str, _SkelClass] Parse a Mermaid classDiagram CONTRACT into {name: _SkelClass}, preserving diagram declaration order (see module convention above).
_normalize_skeleton_module_path _normalize_skeleton_module_path(module: str) -> str
_parse_skeleton_reuse_imports _parse_skeleton_reuse_imports(diagram_text: str) -> dict[str, str] Parse the contract's own %% REUSE (...): Name1 / Name2 (module.path), ... comment block into {name: dotted_module_path}.
_build_skeleton_known_types _build_skeleton_known_types(diagram_text: str) -> dict[str, str]
_skeleton_type_tokens _skeleton_type_tokens(annotation: str | None) -> set[str]
_skel_generic_suffix _skel_generic_suffix() -> str PEP 695 type-param suffix for a class header: [T] or "".
_resolve_skeleton_type _resolve_skeleton_type(token: str, local_names: set[str], imports: set[str], unresolved: set[str], ctx: _SkelEmitContext) -> None
_resolve_skeleton_annotation _resolve_skeleton_annotation(annotation: str | None, local_names: set[str], imports: set[str], unresolved: set[str], ctx: _SkelEmitContext) -> None
_introspect_base_readonly_property _introspect_base_readonly_property(module_path: str, class_name: str, attr_name: str) -> bool | None Real introspection of a RESOLVABLE base class's attribute (fix 3 — "never a guess"): True when attr_name is a property with no setter (assigning self.<attr_name> = ... in a composing class…
_skeleton_field_is_readonly_property _skeleton_field_is_readonly_property(f: _SkelField, ctx: _SkelEmitContext) -> tuple[bool, bool] (is_property, resolved) for one composed field vs. cls's realized base(s) (fix 3). resolved=False means no base could be confirmed either way (the base itself isn't resolvable) — the…
_traces_call _traces_call(spec: str | None, story: str | None, reqs: list[str]) -> str
_not_implemented_message _not_implemented_message(cls_name: str, method_name: str, spec: str | None) -> str
_skeleton_method_role _skeleton_method_role(m: _SkelMethod) -> str patternRole for one method.
_skeleton_method_decorator_lines _skeleton_method_decorator_lines(m: _SkelMethod) -> list[str] Bracket-decorator convention (ported from orchestrata_tools.diagram .diagram_to_code.DiagramToCode._parse_class_member/_generate_method)…
_note_method_decorator_imports _note_method_decorator_imports(m: _SkelMethod, imports: set[str]) -> None
_skeleton_method_first_param _skeleton_method_first_param(m: _SkelMethod) -> str | None
_emit_leaf_method_lines _emit_leaf_method_lines(m: _SkelMethod, local_names: set[str], imports: set[str], unresolved: set[str], ctx: _SkelEmitContext) -> tuple[list[str], list[tuple[str, int, int]]] Signature-faithful stub whose body is always raise NotImplementedError(...) — the leaf role, and the safe fallback for factory/non-Tool command roles: their body can't be fabricated…
_emit_accessor_method_lines _emit_accessor_method_lines(m: _SkelMethod, local_names: set[str], imports: set[str], unresolved: set[str], ctx: _SkelEmitContext) -> tuple[list[str], list[tuple[str, int, int]]] | None return self._<field> when m takes no params and a composed field's name AND type match exactly (get_x -> _x) — deterministic wiring, never a guess. None when no such clean…
_find_delegate_leaf _find_delegate_leaf(args_type: str | None, ctx: _SkelEmitContext) -> _SkelMethod | None The delegation target for execute()'s single # IMPLEMENT: line: the first non-execute method whose parameters all name-match a field of the Args dataclass (so the call can be wired as…
_emit_tool_execute_lines _emit_tool_execute_lines(m: _SkelMethod, ctx: _SkelEmitContext, local_names: set[str], imports: set[str], unresolved: set[str]) -> tuple[list[str], list[tuple[str, int, int]]] The Command patternRole's FULLY GENERATED boilerplate: arg parse/ validate from the Args dataclass (when one is locally recognizable), try/except -> ToolResult wrapping, a config-gated tracer span —…
_dispatch_method_lines _dispatch_method_lines(m: _SkelMethod, ctx: _SkelEmitContext, local_names: set[str], imports: set[str], unresolved: set[str]) -> tuple[list[str], list[tuple[str, int, int]]] Shape one method's generated body by its resolved patternRole, uniformly across the Tool/class/dataclass stereotypes (Protocol bodies are handled separately — always ...).
_emit_composed_init_lines _emit_composed_init_lines(fields: list[_SkelField], local_names: set[str], imports: set[str], unresolved: set[str], ctx: _SkelEmitContext) -> tuple[list[str], list[str]] Composition/DI __init__ — wiring of fields as injected collaborators (private -> self._x) or plain values (public -> self.x).
_select_skeleton_emitter _select_skeleton_emitter(stereotype: str) -> _SkeletonEmitter | None Factory: select the Strategy emitter for a class's <<stereotype>>, or None when unsupported.
_build_skeleton_trace_map _build_skeleton_trace_map(classes: dict[str, _SkelClass], output_file: str) -> dict Companion trace-map: REQ -> {file, class, method} for KG/RTM ingestion (method is None for a class-level requirement).
_build_skeleton_leafmap _build_skeleton_leafmap(class_blocks: list[tuple[_SkelClass, _SkeletonEmitResult, int]]) -> dict Per-class, per-method LINE SPANS (1-indexed, inclusive) of the generated # IMPLEMENT: leaf regions in the FINAL assembled module — the substrate for a later structural-escape gate (fix 8).…
_merge_skeleton_imports _merge_skeleton_imports(imports: set[str]) -> list[str] Merge {"from X import A", "from X import B"} -> one sorted from X import A, B line per module (matches project isort style).
_render_skeleton_module _render_skeleton_module(classes_to_emit: dict[str, _SkelClass], ctx: _SkelEmitContext) -> tuple[str, list[tuple[_SkelClass, _SkeletonEmitResult, int]]] Emit ONE module's Python source for classes_to_emit (the whole diagram, or one namespace's slice of it per fix 7); cross-class type references resolve against the FULL ctx.classes table so a…
_generate_skeleton_full _generate_skeleton_full(diagram_text: str, *, output_file: str = '<generated>') -> tuple[str, dict, dict] Core implementation shared by generate_skeleton (the stable (code, trace_map) public API, unchanged for every existing caller) and the CLI, which also wants the companion leafmap (fix 8).
generate_skeleton generate_skeleton(diagram_text: str, *, output_file: str = '<generated>') -> tuple[str, dict] Convert a Mermaid classDiagram CONTRACT into typed, @traces'd Python skeletons — the build-factory's linchpin generator (see plans/build-factory-architecture.md).
_group_skeleton_classes_by_namespace _group_skeleton_classes_by_namespace(classes: dict[str, _SkelClass]) -> dict[str, dict[str, _SkelClass]]
_namespace_output_file _namespace_output_file(namespace: str) -> str
generate_skeleton_modules generate_skeleton_modules(diagram_text: str) -> dict[str, tuple[str, dict, dict]] Multi-module variant of generate_skeleton (fix 7): classes inside a Mermaid namespace &lt;dotted.path&gt; { ... } block route to their OWN module; classes outside any namespace land in the ""…
_cheapest_dummy_for_annotation _cheapest_dummy_for_annotation(ann_text: str | None) -> object The cheapest REAL dummy value for one parameter, derived from its annotation TEXT alone — Gate 0 never imports an annotated type just to probe construction. "" for str, 0 for…
_gate0_init_required_params _gate0_init_required_params(node: ast.ClassDef) -> list[tuple[str, str | None]] | None [(name, annotation_text)] for every __init__ parameter with NO default (self excluded), read straight off the AST — never imports the annotated type. None when the class declares no…
_gate0_required_ctor_dummies _gate0_required_ctor_dummies(node: ast.ClassDef, cls_obj: type) -> dict[str, object] Cheapest-dummy kwargs to construct cls_obj once — every required __init__ parameter, resolved from the AST when the class writes its own __init__ (Tool/class emitters), or via real…
_gate0_zero_arg_callables _gate0_zero_arg_callables(node: ast.ClassDef) -> list[tuple[str, bool, bool]] [(method_name, is_property, is_async)] for every concrete, zero-arg method the class defines directly (__init__ and dunders excluded) — Gate 0 calls (or, for a @property, merely…
gate0_construction_smoke gate0_construction_smoke(code: str, label: str) -> tuple[list[str], list[str]] Gate 0 — construction smoke (hard-fail): exec code in an isolated, sys.modules-registered namespace, then attempt to construct every generated non-Protocol/non-enum concrete class (cheapest…
gate2_leaf_fraction_table gate2_leaf_fraction_table(code: str, leafmap: dict) -> list[dict] Gate 2 (advisory only — never affects exit code): per generated class, leaves/methods computed from the leafmap + a fresh AST parse of the written code.
_print_gate2_table _print_gate2_table(rows: list[dict]) -> None
_gen_skeleton_self_verify_traces _gen_skeleton_self_verify_traces(output_path: Path, specs_dir: str) -> bool plans/build-factory-architecture.md documents that 'gen-skeleton re-parses its own output (ast.parse) + runs check-traces to self-verify' — this makes the code match that doc (it never…
run_gen_skeleton_gates run_gen_skeleton_gates(code: str, label: str, leafmap: dict, specs_dir: str, no_smoke: bool, output_path: Path | None) -> bool Run Gate 0 (unless no_smoke), Gate 2, and — when the generated code carries at least one @traces(...) AND it was actually written to disk (output_path is not None) — the check-traces…
_skeldiff_class_index _skeldiff_class_index(tree: ast.Module) -> dict[str, ast.ClassDef]
_skeldiff_toplevel_func_index _skeldiff_toplevel_func_index(tree: ast.Module) -> dict[str, ast.FunctionDef | ast.AsyncFunctionDef]
_skeldiff_method_index _skeldiff_method_index() -> dict[str, ast.FunctionDef | ast.AsyncFunctionDef]
_skeldiff_signature _skeldiff_signature(node: ast.FunctionDef | ast.AsyncFunctionDef) -> str AST-dump-normalized signature: params/annotations/defaults, return annotation, decorators, and sync/async-ness — never source text (pure whitespace/reflow must never register as a change).
_skeldiff_body _skeldiff_body(node: ast.FunctionDef | ast.AsyncFunctionDef) -> str
_skeldiff_import_lines _skeldiff_import_lines(tree: ast.Module) -> set[str]
skeleton_diff skeleton_diff(skeleton_path: str, final_path: str, leafmap_path: str) -> dict Gate 3: STRUCTURAL ESCAPE report — differences between skeleton_path (the gen-skeleton output) and final_path (the builder-completed version) OUTSIDE the leaf regions leafmap_path recorded.
_module_export_names _module_export_names(tree: ast.Module) -> list[str] Public export names for one already-parsed module.
_isort_sort_key _isort_sort_key(name: str) -> tuple[int, str] Ruff/isort's default order-by-type bucketing for names sharing one from ... import statement: CONSTANT_CASE names first, then ClassCase names, then lower_case names/functions —…
_format_facade_import _format_facade_import(module: str, names: list[str]) -> str from module import a, b — wrapped in parens, one name per line with a trailing comma (black/ruff style) once the single-line form would exceed the project's 88-column limit.
generate_facade generate_facade(package_dir: str | Path, exclude: list[str] | None = None) -> tuple[str, list[str]] Generate a facade __init__.py re-exporting every sibling module's public API under package_dir (direct children only — not recursive).
_loaded_stored_names _loaded_stored_names(node: ast.AST) -> tuple[set, set] (loaded, stored) plain Name ids referenced anywhere under node.
_range_stmts _range_stmts(fn: ast.FunctionDef | ast.AsyncFunctionDef, start: int, end: int) -> list Top-level statements of fn's body whose lineno falls in [start, end].
_names_bound_before _names_bound_before(fn: ast.FunctionDef | ast.AsyncFunctionDef, before_line: int) -> set Names bound in fn before before_line — its params plus anything Stored on an earlier top-level statement (the candidate inputs).
_param_annotations _param_annotations(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> dict Map each fn parameter name -> its annotation source (or None).
extract_method extract_method(func_name: str, start: int, end: int, new_name: str, sources: SourceList, apply: bool = False) -> dict Extract the contiguous statement run [start, end] from func_name into a new callable new_name.
_dto_fields _dto_fields(dto_node: ast.ClassDef) -> list[dict] Fields of a DTO class: init params if present, else module-level AnnAssign attributes (name + annotation + default), required first.
_rewrite_signature_inline _rewrite_signature_inline(fn_node: ast.FunctionDef | ast.AsyncFunctionDef, dto_param: str, fields: list[dict]) -> str Return the new args source so the dto_param is gone and the DTO fields are spliced in (annotations + defaults preserved, self/cls kept in front).
inline_dto_param inline_dto_param(func_name: str, dto_param: str, sources: SourceList, apply: bool = False) -> dict Inline a DTO parameter of func_name: expand its fields into the signature, rewrite dto.field -> field in the body, and best-effort expand keyword-constructed call sites.
_inline_rewrite_callsites _inline_rewrite_callsites(path: str | Path, text: str, tree: ast.Module, func_name: str, dto_param: str, dto_name: str, resolver: tuple[dict, dict, str]) -> tuple[str | None, list[str]] Rewrite each call ``func_name(...
_class_is_enum _class_is_enum(node: ast.ClassDef) -> bool Same rule CodebaseAnalyzer uses for ClassInfo.is_enum (bases naming Enum/IntEnum, which also covers StrEnum).
_class_has_serde _class_has_serde(node: ast.ClassDef) -> bool True if node already defines both to_dict and from_dict — a nested value this codebase already knows how to (de)serialize on its own.
_class_lookup_by_name _class_lookup_by_name(sources: SourceList) -> dict[str, ast.ClassDef] First ClassDef per bare name across sources — enough to classify a field's annotation as an Enum vs. a serde-able nested dataclass without re-walking the tree once per field.
_unwrap_annotation _unwrap_annotation(ann: str | None) -> tuple[bool, str | None, str | None] (is_optional, container, element) from a field's annotation source. container is "list"/"dict"/None; element is the bare inner type (a dict's value type, a list's item type, or…
_classify_element _classify_element(elem: str | None, lookup: dict[str, ast.ClassDef]) -> str "datetime"/"date"/"enum"/"dataclass"/"plain" for one bare element type name.
_factory_call_text _factory_call_text(factory_node: ast.expr) -> str Call-text for a default_factory= expression — bare name() for a simple reference (matching e.g. created_at=... else _utc_now()), parenthesized for anything more complex (e.g. a…
_default_fallback _default_fallback(default_src: str | None, container: str | None) -> str The from_dict fallback for a field: its own dataclass default where present (field(default_factory=X) unwraps to a call of X; field(default=X) unwraps to X), else a type-sensible…
_to_dict_field_expr _to_dict_field_expr(accessor: str, kind: str, container: str | None, optional: bool) -> str The to_dict value expression for one field, e.g. self.f.isoformat() if self.f else None.
_from_dict_field_expr _from_dict_field_expr(key: str, kind: str, container: str | None, optional: bool, type_name: str | None, default_src: str | None) -> str The from_dict kwarg expression for one field — the inverse of _to_dict_field_expr's transforms.
generate_serde generate_serde(class_name: str, sources: SourceList) -> str | None Emit to_dict/from_dict for class_name from its fields — type-aware per the conventions hand-written across the codebase (agents/types.py et al): datetime/date -> .isoformat(), an…
_returns_in _returns_in(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> list Every Return value node directly inside fn's own scope (mirrors _returns_value's nested-scope skip, but collects the values instead of just checking whether one exists).
_to_dict_keys _to_dict_keys(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> list[str] | None String-literal keys of the dict a to_dict returns, or None if it isn't a single literal-dict return (too dynamic to verify statically).
_from_dict_kwargs _from_dict_kwargs(fn: ast.FunctionDef | ast.AsyncFunctionDef, class_name: str) -> list[str] | None Keyword names of the cls(...)/&lt;Class&gt;(...) call a from_dict returns, or None if it isn't a single direct-construction return (too dynamic to verify statically).
_self_attrs_used _self_attrs_used(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> set Attr names x where self.x is read anywhere in fn — used to tell whether a field's VALUE is serialized regardless of the dict key it lands under (so a keyword-avoidance rename like…
_serde_exclude _serde_exclude(node: ast.ClassDef) -> set Field names a class deliberately keeps OUT of its dict serde, declared via a class-level __serde_exclude__ = {"f", ...} (also accepts frozenset/set/ list/tuple of string literals).
check_serde_drift check_serde_drift(sources: SourceList, class_name: str | None = None) -> dict Verify every @dataclass that defines BOTH to_dict+from_dict wires every field into both.
_funcs_named _funcs_named(func_name: str | None, tree: ast.AST) -> list[ast.FunctionDef | ast.AsyncFunctionDef] All FunctionDef/AsyncFunctionDef matching func_name (bare or Class.method).
_entry_stmt _entry_stmt(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> ast.stmt | None First real statement of a function body (skips a leading docstring).
_breakpoint_points _breakpoint_points(fn: ast.FunctionDef | ast.AsyncFunctionDef, mode: str) -> list[tuple[int, int]] List of (lineno, indent_col) at which to insert a breakpoint.
insert_breakpoints insert_breakpoints(sources: SourceList, func_name: str | None = None, mode: str = 'entry', line: int | None = None, apply: bool = False) -> dict Inject breakpoint() # ast-debug at function entry / branches / a line.
clear_breakpoints clear_breakpoints(sources: SourceList, apply: bool = False) -> dict Remove every breakpoint() # ast-debug line previously injected.
_who_imports_file _who_imports_file(args: tuple[Path, str, ast.Module], target: str, tlast: str) -> list[dict] Import hits for target in ONE file — the thread-pool unit of work (the walk dominates a warm who-imports: 2.9s of 8.5s measured 2026-08-11; trees are read-only here so the free-threaded build…
who_imports who_imports(target: str, sources: SourceList) -> list[dict] Files that import target — a module path or a symbol name — with the statement.
attr_refs attr_refs(attrs: list[str], sources: SourceList, calls_only: bool = False) -> list[dict] Attribute access/call sites by attribute NAME (e.g. session, _driver) across sources.
_has_traces_decorator _has_traces_decorator(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool
_imports_traces _imports_traces(tree: ast.AST) -> bool
_import_insert_line _import_insert_line(tree: ast.Module) -> int 1-based line to insert a new import after (last import, else docstring).
find_untraced find_untraced(sources: SourceList) -> list[dict] Public (non-underscore) functions/methods lacking an @traces decorator.
add_traces add_traces(sources: SourceList, apply: bool = False) -> dict Insert an import-safe placeholder @traces above each untraced public function/method (and the traces import if missing).
_module_root _module_root(path: str | Path) -> Path Directory containing the top-level package of path (dotted<->path base).
_free_names _free_names(node: ast.AST) -> set Names loaded in node but not bound within it (approximate free vars).
_module_level_names _module_level_names(tree: ast.AST, exclude: str) -> set Module-level names DEFINED via assignment or def/class (not imports).
_rewrite_import_module _rewrite_import_module(text: str, tree: ast.AST, name: str, src_module: str, dest_module: str) -> str Repoint from src_module import …name… to dest_module, splitting the statement when other names are imported alongside name.
_splice_module_imports _splice_module_imports(text: str, imports: str) -> str Insert imports (newline-terminated) after the module docstring and any leading run of top-level imports; fall back to the very top when absent.
move_symbol move_symbol(name: str, dest_module: str, sources: SourceList, apply: bool = False) -> dict Move a top-level class/function name to dest_module and repoint every importer.
apply_pattern_inplace apply_pattern_inplace(class_name: str, pattern: str, sources: SourceList, apply: bool = False) -> dict | None Inject a singleton/factory pattern directly INTO the class body (the wrapper patterns stay separate — use --output for those).
_resolve_relative_module _resolve_relative_module(importer: str, module: str | None, level: int) -> str Resolve a from . import x target to an absolute dotted module.
_resolve_local_node _resolve_local_node(name: str, local_modules: set[str]) -> str | None Map a dotted name to its graph node: itself, or — since :func:_dotted_module names a package's own module pkg.__init__, never bare pkg — its package __init__ node.
_import_edges _import_edges(importer: str, tree: ast.AST, local_modules: set[str]) -> set[str] First-party modules importer (a dotted module) depends on.
_extract_raw_imports _extract_raw_imports(tree: ast.AST, path: str | Path) -> list[tuple] Every Import/ImportFrom node in tree, decoupled from local_modules resolution (which needs the FULL codebase's module set first, so it can't be computed per-file) — the per-file unit…
_import_edges_from_raw _import_edges_from_raw(importer: str, raw: list[tuple], local_modules: set[str]) -> set[str] Resolve one file's raw import list (see :func:_extract_raw_imports) into first-party edges — the exact same resolution rules as :func:_import_edges, split out so they can run over a CACHED…
_build_import_graph _build_import_graph(sources: SourceList) -> dict[str, set[str]] First-party import graph at module granularity: module -&gt; {deps}.
_tarjan_sccs _tarjan_sccs(graph: dict[str, set[str]]) -> list[list[str]] Strongly connected components of graph, via iterative Tarjan.
_order_cycle _order_cycle(component: set[str], graph: dict[str, set[str]]) -> list[str] Order an SCC's nodes into a closed chain following real edges.
find_import_cycles find_import_cycles(sources: SourceList) -> list[list[str]] Detect first-party import cycles among the modules under sources.
_script_root_edges _script_root_edges(scripts_dir: str | Path, local_modules: set[str]) -> set[str] First-party src modules imported (directly) by top-level scripts/*.py.
_pyproject_entrypoint_modules _pyproject_entrypoint_modules(pyproject_path: str | Path, local_modules: set[str]) -> set[str] src modules named by [project.scripts]/[project.entry-points.*].
_main_guard_modules _main_guard_modules(sources: SourceList, local_modules: set[str]) -> set[str] src modules with a module-level if __name__ == "__main__": guard — self-executing entrypoints, reused from the existing serializer helper.
_module_ancestors _module_ancestors(module: str) -> list[str] Dotted package-ancestor NODES of module, nearest first.
find_orphan_modules find_orphan_modules(src_root: str | Path = 'src/orchestrata_ultimate', scripts_dir: str | Path = 'scripts', pyproject_path: str | Path = 'pyproject.toml') -> dict Which src modules are production-unreachable (AGENTS.md Hard Rule 8).
_load_orphan_baseline _load_orphan_baseline(path: str | Path) -> set[str]
_write_orphan_baseline _write_orphan_baseline(path: str | Path, orphans: list[str]) -> None
_self_attr_names _self_attr_names(nodes: Sequence[ast.AST]) -> set Every self.&lt;attr&gt; attribute name read or written within nodes.
_self_calls _self_calls(nodes: Sequence[ast.AST]) -> set Names of methods invoked as self.&lt;m&gt;(...) within nodes.
_find_init _find_init(class_node: ast.ClassDef) -> ast.FunctionDef | ast.AsyncFunctionDef | None The __init__ def of class_node, or None.
_init_assigned_attrs _init_assigned_attrs(init_node: ast.FunctionDef | ast.AsyncFunctionDef | None) -> set Names assigned via self.&lt;x&gt; = ... (incl. annotated) in init_node.
_class_by_name _class_by_name(text: str, class_name: str) -> ast.ClassDef Locate class_name's ClassDef in freshly parsed text.
extract_class extract_class(class_name: str, methods: list[str], new_name: str, sources: SourceList, apply: bool = False) -> dict Pull methods out of class_name into a new collaborator new_name and leave class_name delegating to an instance of it.
_inject_init _inject_init(text: str, class_name: str, body_pad: str, attr: str, new_name: str, wire_args: str) -> str Add an __init__ to class_name that wires the collaborator.
_splice_delegators _splice_delegators(text: str, class_name: str, delegators_block: str) -> str Append delegators_block at the end of class_name's body.
_insert_before_class _insert_before_class(text: str, class_name: str, new_cls_block: str) -> str Insert new_cls_block on the lines just before class_name's class.
_iter_funcs_with_class _iter_funcs_with_class(tree: ast.Module) -> Iterator[tuple[str | None, ast.FunctionDef | ast.AsyncFunctionDef]] Yield (class_name_or_None, func_node) for every function/method at any nesting depth.
_iter_own_scope _iter_own_scope(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> Iterator[ast.AST] Yield descendant nodes within fn's own body, without descending into nested def / async def / class scopes.
_all_arg_nodes _all_arg_nodes(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> list[ast.arg]
_is_self_attr _is_self_attr(node: ast.expr) -> TypeIs[ast.Attribute]
_ann_type_name _ann_type_name(node: ast.expr | None) -> str | None Base class name of a type annotation (Optional[X]/X | None -> X; list[X] -> list; forward-ref str -> its base); None if not a name.
_call_type_name _call_type_name(func_node: ast.expr) -> str | None
_dotted_from_attr _dotted_from_attr(node: ast.expr) -> str | None a.b.c for an attribute chain rooted at a Name, else None.
_file_import_map _file_import_map(tree: ast.AST) -> dict Bound name -> ('module', qual) or ('symbol', qual).
_extract_symbol_index_entry _extract_symbol_index_entry(tree: ast.Module, path: str | Path) -> dict One file's modules[mod] entry for :func:_build_symbol_index (functions with return types, classes with methods/fields/bases, and the file's import map) — the per-file unit that function caches…
_build_symbol_index _build_symbol_index(sources: SourceList) -> tuple[dict, dict] (modules, class_index) — per-module functions (with return types), classes (methods+returns, field types, bases), and import maps.
_resolve_type_name _resolve_type_name(name: str | None, module: str, modules: dict, class_index: dict) -> str | None A bare type name -> class qualname via imports, local classes, then a unique global match.
_resolve_method _resolve_method(class_qual: str, method: str, modules: dict, class_index: dict, seen: set | None = None) -> str | None Resolve method on class_qual via MRO -> the defining class's method qualname, or None (method defined on an external base / not found).
_resolve_field_type _resolve_field_type(class_qual: str, field: str, modules: dict, class_index: dict, seen: set | None = None) -> str | None
_resolve_module_ref _resolve_module_ref(prefix: str, module: str, modules: dict) -> str | None
_func_type_env _func_type_env(fn: ast.FunctionDef | ast.AsyncFunctionDef, class_qual: str | None, module: str, modules: dict, class_index: dict) -> dict Map local names to class qualnames within fn (self, annotated params, annotated locals, x = Cls(...)).
_resolve_call_target _resolve_call_target(func: ast.expr, env: dict, module: str, modules: dict, class_index: dict) -> str | None Fully-qualified target of a call's func node, or None if the type is not statically determinable.
_extract_call_names _extract_call_names(tree: ast.AST, path: str | Path) -> frozenset[str] Every syntactic call-site name in tree (f(...) -> "f", x.m(...) -> "m") — the per-file unit build_call_graph's caller-scan caches (kind "call_names") as a SOUND pre-filter…
build_call_graph build_call_graph(func_name: str, sources: SourceList) -> dict | None Type-resolved forward (callees) and reverse (callers) call graph.
_iter_calls_with_env _iter_calls_with_env(node: ast.AST, env: dict, class_qual: str | None, module: str, modules: dict, class_index: dict) -> Iterator[tuple[ast.Call, dict]] Yield (call_node, type_env) for every Call under node, carrying the enclosing scope's name->type environment (for receiver resolution).
_class_defines _class_defines(class_qual: str, method: str, modules: dict) -> bool
_ancestors _ancestors(class_qual: str, modules: dict, class_index: dict, seen: set) -> Iterator[str]
_children_map _children_map(modules: dict, class_index: dict) -> dict
_descendants _descendants(root: str, children: dict, seen: set | None = None) -> set
_method_roots _method_roots(c_qual: str, old: str, modules: dict, class_index: dict) -> set Top-most ancestor(s) of c_qual that define old (no super defines it).
_method_family _method_family(class_name: str, old: str, modules: dict, class_index: dict) -> tuple[set, set] All class qualnames whose .old belongs to class_name's override family: the root definer(s) reachable from it, plus every descendant.
_module_type_env _module_type_env(tree: ast.Module, module: str, modules: dict, class_index: dict) -> dict
_attr_receiver_type _attr_receiver_type(recv: ast.expr, env: dict, module: str, modules: dict, class_index: dict) -> str | None
_scan_method_calls _scan_method_calls(node: ast.AST, env: dict, class_qual: str | None, module: str, modules: dict, class_index: dict, old: str, family: set, spans: set, counts: dict) -> None
_type_scoped_method_rename _type_scoped_method_rename(old: str, new: str, sources: SourceList, class_name: str, apply: bool) -> dict Rename method old -> new on class_name and its inheritance family, rewriting only call sites whose receiver type resolves into that family.
_decorator_name _decorator_name(dec: ast.expr) -> str | None
_is_declaration_only _is_declaration_only(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> bool True if fn is a declaration with no implementation — body is only a docstring and/or .../pass/raise.
_is_interface_class _is_interface_class(node: ast.ClassDef) -> bool True if node is a Protocol / ABC — its methods are interface contracts.
_dead_params_of _dead_params_of(fn: ast.FunctionDef | ast.AsyncFunctionDef, cname: str | None, path: str | Path) -> dict | None
find_dead_params find_dead_params(sources: SourceList) -> list[dict] Parameters never referenced in their function body.
_returns_value _returns_value(fn: ast.FunctionDef | ast.AsyncFunctionDef) -> bool True if fn has a return &lt;non-None&gt; (ignoring nested scopes).
_is_generated_file _is_generated_file(path: Path, text: str) -> bool True for machine-generated modules (protobuf stubs, @generated files) that must not be hand-edited — so type-hint enforcement skips them.
find_untyped find_untyped(sources: SourceList) -> list[dict] Public functions/methods missing type hints — the precondition for reliable type resolution.
_requirement_source_files _requirement_source_files(specs_dir: str | Path) -> list[Path] The NORMATIVE requirement-definition files, in precedence order.
_load_spec_requirements _load_spec_requirements(specs_dir: str | Path) -> dict[str, tuple[str, int]] REQ ids defined in the normative files of &lt;specs_dir&gt;/*/.
_is_traces_call _is_traces_call(dec: ast.expr) -> TypeIs[ast.Call] True for a @traces(...) call (mirrors generate_rtm.py's _is_traces_decorator).
_collect_code_traces _collect_code_traces(sources: SourceList) -> tuple[dict[str, list[tuple[str, int]]], list[dict]] Walk sources for @traces(reqs=[...]) decorators on classes AND functions/methods, at any nesting depth (ast.walk covers it).
check_traces check_traces(sources: SourceList, specs_dir: str | Path) -> dict Requirement-traceability VERIFY GATE — read-only, never writes @traces.
_traces_kwarg_str _traces_kwarg_str(dec: ast.Call, name: str) -> str | None A @traces(...) string keyword value (spec=/story=), or None if absent / not a string literal.
_traces_kwarg_reqs _traces_kwarg_reqs(dec: ast.Call) -> list[str] The string constants of a @traces(reqs=[...]) list literal (non-list or non-string entries are silently skipped, mirroring _collect_code_traces).
_package_for _package_for(path: str | Path, root: str | Path) -> str The package dir directly under root that path lives in (e.g. gateway for src/orchestrata_ultimate/gateway/handlers/auth.py), or "" for a top-level module directly in root.
_trace_records _trace_records(sources: SourceList) -> list[dict] Census walk: every @traces(...) decorator under sources -> {file, pkg, line, name, kind, spec, story, reqs}.
_req_family _req_family(req_id: str) -> str | None
_family_matrix _family_matrix(records: list[dict]) -> dict[str, dict[str, int]] package -> {req family -> occurrence count}.
_shotgun _shotgun(records: list[dict], shotgun_min: int) -> list[dict] Decorators with len(reqs) &gt;= shotgun_min, sorted desc by count.
_fanout _fanout(records: list[dict], fanout_min: int) -> list[dict] Req ids used on >= fanout_min DISTINCT decorators, sorted desc.
_prefix_and_num _prefix_and_num(req_id: str) -> tuple[str, int] | None
_is_sequential_pair _is_sequential_pair(a: dict, b: dict) -> bool True if some req in a and some req in b share a prefix and b's trailing integer is a's + 1 (mechanical enumeration, not mapping).
_sequential_runs _sequential_runs(records: list[dict], seq_min: int) -> list[dict] Per file: count of adjacent (by line) decorator pairs that are sequential (see _is_sequential_pair).
_classify_story _classify_story(story: str | None) -> str Coarse shape classification for a story= value (convention-drift analytic — not a validity check).
_convention_stats _convention_stats(records: list[dict]) -> dict Counts of spec= SPEC-prefixed vs. bare vs. missing, and story= format-shape distribution.
trace_audit trace_audit(sources: SourceList, shotgun_min: int = 10, fanout_min: int = 20, seq_min: int = 3) -> dict Census + analytics INSTRUMENT over every @traces(...) decorator under sources — the wave-sizing and acceptance instrument for the trace remediation program (plans/trace-remediation-plan.md §3).
_load_requirement_ground_truth _load_requirement_ground_truth(specs_dir: str | Path) -> set[str] REQ ids defined anywhere in the normative files of &lt;specs_dir&gt;/*/ as either a ### REQ-... heading or a **REQ-...** bold span.
_ground_truth_by_spec _ground_truth_by_spec(specs_dir: str | Path) -> dict[str, set[str]] Map each spec NUMBER (dir-name prefix, e.g. 007) to the REQ ids its own checklists/requirements.md defines.
_known_spec_numbers _known_spec_numbers(specs_dir: str | Path) -> set[str] Known spec numbers = REQ families present in the ground truth UNION the numeric prefixes of the spec dir names themselves (covers a spec that has a directory but no requirements.md yet).
trace_lint trace_lint(sources: SourceList, specs_dir: str | Path = 'specs') -> dict Deterministic PREVENTION gate for @traces(...): dangling ids, placeholder ids (REQ-000-000), reqs-per-decorator ceilings (error >25, warning 11-25), spec= values naming no known spec…
_add_cache_args _add_cache_args(p: argparse.ArgumentParser) -> None --cache/--no-cache for a read-only, tree-scanning subcommand.
main main() -> None Thin wrapper: dispatch (_dispatch), then ALWAYS flush the warm cache — a try/finally rather than trailing code, because several subcommand handlers sys.exit(...) mid-dispatch (e.g.…
_build_parser _build_parser() -> argparse.ArgumentParser Build the full serialize_codebase_ast argument parser.
_declared_ast_roots _declared_ast_roots() -> list[str] The root set from $ORCHESTRATA_AST_ROOTS (whitespace/colon separated).
_cmd_serialize _cmd_serialize(args: argparse.Namespace) -> None serialize subcommand.
_cmd_compare _cmd_compare(args: argparse.Namespace) -> None compare subcommand.
_cmd_deep_compare _cmd_deep_compare(args: argparse.Namespace) -> None deep-compare subcommand.
_cmd_check_deps _cmd_check_deps(args: argparse.Namespace) -> None check-deps subcommand.
_cmd_symbols _cmd_symbols(args: argparse.Namespace) -> None symbols subcommand.
_cmd_refs _cmd_refs(args: argparse.Namespace) -> None refs subcommand.
_cmd_source _cmd_source(args: argparse.Namespace) -> None source subcommand.
_cmd_rename _cmd_rename(args: argparse.Namespace) -> None rename subcommand.
_cmd_unused _cmd_unused(args: argparse.Namespace) -> None unused subcommand.
_cmd_complexity _cmd_complexity(args: argparse.Namespace) -> None complexity subcommand.
_cmd_ast_to_kg _cmd_ast_to_kg(args: argparse.Namespace) -> None ast-to-kg subcommand.
_cmd_interface _cmd_interface(args: argparse.Namespace) -> None interface subcommand. --kind protocol (default) is byte-identical to the original Protocol-only behavior; --kind abc is additive, delegating to generate_protocol's ABC shape.
_cmd_docs _cmd_docs(args: argparse.Namespace) -> None docs subcommand.
_cmd_diagram _cmd_diagram(args: argparse.Namespace) -> None diagram subcommand.
_cmd_branches _cmd_branches(args: argparse.Namespace) -> None branches subcommand.
_cmd_gen_tests _cmd_gen_tests(args: argparse.Namespace) -> None gen-tests subcommand.
_cmd_testgen _cmd_testgen(args: argparse.Namespace) -> None testgen subcommand — restored Layer 1+2 generator (orchestrata_tools.test_generator.testgen, restored 2026-08-19 after its 2026-07-09 deletion left tests/generated_recovery/ claiming a…
_cmd_audit _cmd_audit(args: argparse.Namespace) -> None audit subcommand.
_cmd_nesting _cmd_nesting(args: argparse.Namespace) -> None nesting subcommand.
_cmd_pattern _cmd_pattern(args: argparse.Namespace) -> None pattern subcommand.
_cmd_long_params _cmd_long_params(args: argparse.Namespace) -> None long-params subcommand.
_cmd_gen_dto _cmd_gen_dto(args: argparse.Namespace) -> None gen-dto subcommand.
_cmd_gen_enum _cmd_gen_enum(args: argparse.Namespace) -> None gen-enum subcommand.
_cmd_gen_volk _cmd_gen_volk(args: argparse.Namespace) -> None gen-volk subcommand.
_cmd_gen_protocol _cmd_gen_protocol(args: argparse.Namespace) -> None gen-protocol subcommand.
_cmd_gen_skeleton _cmd_gen_skeleton(args: argparse.Namespace) -> None gen-skeleton subcommand.
_cmd_skeleton_diff _cmd_skeleton_diff(args: argparse.Namespace) -> None skeleton-diff subcommand.
_cmd_gen_facade _cmd_gen_facade(args: argparse.Namespace) -> None gen-facade subcommand.
_cmd_gen_serde _cmd_gen_serde(args: argparse.Namespace) -> None gen-serde subcommand.
_cmd_extract_method _cmd_extract_method(args: argparse.Namespace) -> None extract-method subcommand.
_cmd_inline _cmd_inline(args: argparse.Namespace) -> None inline subcommand.
_cmd_debug _cmd_debug(args: argparse.Namespace) -> None debug subcommand.
_cmd_who_imports _cmd_who_imports(args: argparse.Namespace) -> None who-imports subcommand.
_cmd_attr_refs _cmd_attr_refs(args: argparse.Namespace) -> None attr-refs subcommand.
_cmd_add_trace _cmd_add_trace(args: argparse.Namespace) -> None add-trace subcommand.
_cmd_move_symbol _cmd_move_symbol(args: argparse.Namespace) -> None move-symbol subcommand.
_cmd_find_cycles _cmd_find_cycles(args: argparse.Namespace) -> None find-cycles subcommand.
_cmd_orphans _cmd_orphans(args: argparse.Namespace) -> None orphans subcommand.
_cmd_extract_class _cmd_extract_class(args: argparse.Namespace) -> None extract-class subcommand.
_cmd_call_graph _cmd_call_graph(args: argparse.Namespace) -> None call-graph subcommand.
_cmd_dead_params _cmd_dead_params(args: argparse.Namespace) -> None dead-params subcommand.
_cmd_untyped _cmd_untyped(args: argparse.Namespace) -> None untyped subcommand.
_cmd_check_traces _cmd_check_traces(args: argparse.Namespace) -> None check-traces subcommand.
_cmd_trace_audit _cmd_trace_audit(args: argparse.Namespace) -> None trace-audit subcommand.
_cmd_trace_lint _cmd_trace_lint(args: argparse.Namespace) -> None trace-lint subcommand.
_dispatch _dispatch() -> None Parse argv and run the selected subcommand.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/serialize_codebase_ast.py

File has 13844 lines (over 3000); see the permalink above.