Add multi-language logic exploration

src/docforge/treesitter_logic.py

@@ -0,0 +1,831 @@
+"""Tree-sitter-backed control-flow extraction for JavaScript and C++.
+
+Tree-sitter supplies concrete syntax trees.  This module adds the small amount
+of language-aware control-flow interpretation needed to emit DocForge's
+language-neutral ``LogicProjection`` contract.  Project code is parsed as
+data; it is never imported, compiled, or executed.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Iterable
+from dataclasses import dataclass
+from functools import lru_cache
+
+import tree_sitter_cpp
+import tree_sitter_javascript
+from tree_sitter import Language, Node, Parser
+
+from .errors import DocForgeError
+from .models import LogicEdge, LogicNode, LogicProjection
+
+
+@dataclass(frozen=True)
+class TreeSitterLogicOwner:
+    """One named function or method that should receive a Logic projection."""
+
+    owner_node_id: str
+    qualified_name: str
+    line: int
+
+
+@dataclass(frozen=True)
+class DiscoveredFunction:
+    """One parser-identified callable available to a project adapter."""
+
+    qualified_name: str
+    name: str
+    line: int
+    kind: str
+
+
+@dataclass(frozen=True)
+class _Tail:
+    source_id: str
+    relation: str = "next"
+    label: str | None = None
+
+
+@dataclass(frozen=True)
+class _Condition:
+    entry_id: str
+    when_true: tuple[_Tail, ...]
+    when_false: tuple[_Tail, ...]
+
+
+@dataclass(frozen=True)
+class _Control:
+    break_id: str | None = None
+    continue_id: str | None = None
+
+
+@dataclass(frozen=True)
+class _LanguageProfile:
+    name: str
+    language: Language
+    root_type: str
+    block_types: frozenset[str]
+    function_types: frozenset[str]
+    loop_types: frozenset[str]
+    return_types: frozenset[str]
+    raise_types: frozenset[str]
+    switch_case_types: frozenset[str]
+
+
+@lru_cache(maxsize=1)
+def _javascript_profile() -> _LanguageProfile:
+    return _LanguageProfile(
+        name="javascript",
+        language=Language(tree_sitter_javascript.language()),
+        root_type="program",
+        block_types=frozenset({"program", "statement_block"}),
+        function_types=frozenset(
+            {"function_declaration", "generator_function_declaration", "method_definition"}
+        ),
+        loop_types=frozenset(
+            {"while_statement", "do_statement", "for_statement", "for_in_statement"}
+        ),
+        return_types=frozenset({"return_statement"}),
+        raise_types=frozenset({"throw_statement"}),
+        switch_case_types=frozenset({"switch_case", "switch_default"}),
+    )
+
+
+@lru_cache(maxsize=1)
+def _cpp_profile() -> _LanguageProfile:
+    return _LanguageProfile(
+        name="cpp",
+        language=Language(tree_sitter_cpp.language()),
+        root_type="translation_unit",
+        block_types=frozenset({"translation_unit", "compound_statement"}),
+        function_types=frozenset({"function_definition"}),
+        loop_types=frozenset(
+            {
+                "while_statement",
+                "do_statement",
+                "for_statement",
+                "for_range_loop",
+            }
+        ),
+        return_types=frozenset({"return_statement", "co_return_statement"}),
+        raise_types=frozenset({"throw_statement"}),
+        switch_case_types=frozenset({"case_statement"}),
+    )
+
+
+def analyze_javascript_source(
+    source: str,
+    *,
+    source_id: str,
+    owners: Iterable[TreeSitterLogicOwner],
+    filename: str = "<javascript-source>",
+    max_nodes_per_function: int = 2_000,
+) -> tuple[LogicProjection, ...]:
+    """Build control-flow projections for named JavaScript functions and methods."""
+
+    return _analyze_tree_sitter_source(
+        source,
+        source_id=source_id,
+        owners=owners,
+        filename=filename,
+        profile=_javascript_profile(),
+        max_nodes_per_function=max_nodes_per_function,
+    )
+
+
+def discover_javascript_functions(source: str) -> tuple[DiscoveredFunction, ...]:
+    """Return named JavaScript functions, methods, and assigned arrow functions."""
+
+    return _discover_functions(source, _javascript_profile())
+
+
+def discover_cpp_functions(source: str) -> tuple[DiscoveredFunction, ...]:
+    """Return named C++ functions and methods."""
+
+    return _discover_functions(source, _cpp_profile())
+
+
+def analyze_cpp_source(
+    source: str,
+    *,
+    source_id: str,
+    owners: Iterable[TreeSitterLogicOwner],
+    filename: str = "<cpp-source>",
+    max_nodes_per_function: int = 2_000,
+) -> tuple[LogicProjection, ...]:
+    """Build control-flow projections for named C++ functions and methods."""
+
+    return _analyze_tree_sitter_source(
+        source,
+        source_id=source_id,
+        owners=owners,
+        filename=filename,
+        profile=_cpp_profile(),
+        max_nodes_per_function=max_nodes_per_function,
+    )
+
+
+def _discover_functions(
+    source: str,
+    profile: _LanguageProfile,
+) -> tuple[DiscoveredFunction, ...]:
+    raw = source.encode("utf-8")
+    parser = Parser(profile.language)
+    tree = parser.parse(raw)
+    root = tree.root_node
+    if root.has_error:
+        return ()
+    definitions = _function_definitions(root, raw, profile)
+    result: list[DiscoveredFunction] = []
+    for (qualified_name, line), node in definitions.items():
+        normalized = qualified_name.replace("::", ".")
+        name = normalized.split(".")[-1]
+        result.append(
+            DiscoveredFunction(
+                qualified_name=qualified_name,
+                name=name,
+                line=line,
+                kind=(
+                    "method"
+                    if node.type == "method_definition" or "." in normalized
+                    else "function"
+                ),
+            )
+        )
+    return tuple(
+        sorted(
+            result,
+            key=lambda item: (item.qualified_name, item.line, item.kind),
+        )
+    )
+
+
+def _analyze_tree_sitter_source(
+    source: str,
+    *,
+    source_id: str,
+    owners: Iterable[TreeSitterLogicOwner],
+    filename: str,
+    profile: _LanguageProfile,
+    max_nodes_per_function: int,
+) -> tuple[LogicProjection, ...]:
+    if max_nodes_per_function < 2:
+        raise ValueError("max_nodes_per_function must allow entry and exit nodes")
+    raw = source.encode("utf-8")
+    parser = Parser(profile.language)
+    tree = parser.parse(raw)
+    if tree.root_node.type != profile.root_type or tree.root_node.has_error:
+        error = _first_error(tree.root_node)
+        raise DocForgeError(
+            "invalid_logic_source",
+            f"{profile.name.title()} source cannot be parsed for logic analysis",
+            source=filename,
+            line=(error.start_point.row + 1) if error is not None else 1,
+        )
+    definitions = _function_definitions(tree.root_node, raw, profile)
+    requested = tuple(sorted(owners, key=lambda item: item.owner_node_id))
+    if len({owner.owner_node_id for owner in requested}) != len(requested):
+        raise DocForgeError("invalid_logic_owner", "Logic owner IDs must be unique")
+    projections: list[LogicProjection] = []
+    for owner in requested:
+        function = _resolve_owner(owner, definitions)
+        if function is None:
+            raise DocForgeError(
+                "missing_logic_owner",
+                f"A requested {profile.name} logic owner was not found in its source",
+                owner_node_id=owner.owner_node_id,
+                qualified_name=owner.qualified_name,
+                line=owner.line,
+            )
+        projections.append(
+            _TreeSitterFunctionBuilder(
+                raw=raw,
+                source_id=source_id,
+                owner_node_id=owner.owner_node_id,
+                function=function,
+                profile=profile,
+                max_nodes=max_nodes_per_function,
+            ).build()
+        )
+    return tuple(projections)
+
+
+def _first_error(node: Node) -> Node | None:
+    if node.is_error or node.is_missing:
+        return node
+    for child in node.named_children:
+        error = _first_error(child)
+        if error is not None:
+            return error
+    return None
+
+
+def _function_definitions(
+    root: Node,
+    raw: bytes,
+    profile: _LanguageProfile,
+) -> dict[tuple[str, int], Node]:
+    definitions: dict[tuple[str, int], Node] = {}
+
+    def visit(node: Node, scopes: tuple[str, ...]) -> None:
+        next_scopes = scopes
+        scope_name = _scope_name(node, raw, profile)
+        if scope_name:
+            next_scopes = (*scopes, scope_name)
+        function_name = _function_name(node, raw, profile)
+        if function_name:
+            qualified = (
+                function_name if "::" in function_name else ".".join((*scopes, function_name))
+            )
+            function_node = node
+            if node.type == "variable_declarator":
+                function_node = node.child_by_field_name("value") or node
+            definitions[(qualified, node.start_point.row + 1)] = function_node
+            next_scopes = (*scopes, function_name)
+        for child in node.named_children:
+            visit(child, next_scopes)
+
+    visit(root, ())
+    return definitions
+
+
+def _scope_name(node: Node, raw: bytes, profile: _LanguageProfile) -> str | None:
+    if profile.name == "javascript" and node.type in {"class_declaration", "class"}:
+        return _field_text(node, "name", raw)
+    if profile.name == "cpp" and node.type in {
+        "namespace_definition",
+        "class_specifier",
+        "struct_specifier",
+        "union_specifier",
+    }:
+        return _field_text(node, "name", raw)
+    return None
+
+
+def _function_name(node: Node, raw: bytes, profile: _LanguageProfile) -> str | None:
+    if node.type in profile.function_types:
+        if profile.name == "javascript":
+            return _field_text(node, "name", raw)
+        declarator = node.child_by_field_name("declarator")
+        return _declarator_name(declarator, raw) if declarator is not None else None
+    if profile.name != "javascript" or node.type != "variable_declarator":
+        return None
+    value = node.child_by_field_name("value")
+    if value is None or value.type not in {"arrow_function", "function_expression"}:
+        return None
+    return _field_text(node, "name", raw)
+
+
+def _declarator_name(node: Node, raw: bytes) -> str | None:
+    if node.type in {
+        "identifier",
+        "field_identifier",
+        "operator_name",
+        "destructor_name",
+        "qualified_identifier",
+    }:
+        return _text(node, raw)
+    for field in ("declarator", "name"):
+        child = node.child_by_field_name(field)
+        if child is not None:
+            result = _declarator_name(child, raw)
+            if result:
+                return result
+    for child in node.named_children:
+        result = _declarator_name(child, raw)
+        if result:
+            return result
+    return None
+
+
+def _resolve_owner(
+    owner: TreeSitterLogicOwner,
+    definitions: dict[tuple[str, int], Node],
+) -> Node | None:
+    exact = definitions.get((owner.qualified_name, owner.line))
+    if exact is not None:
+        return exact
+    leaf = owner.qualified_name.replace("::", ".").split(".")[-1]
+    candidates = [
+        node
+        for (qualified_name, line), node in definitions.items()
+        if line == owner.line and qualified_name.replace("::", ".").split(".")[-1] == leaf
+    ]
+    return candidates[0] if len(candidates) == 1 else None
+
+
+class _TreeSitterFunctionBuilder:
+    def __init__(
+        self,
+        *,
+        raw: bytes,
+        source_id: str,
+        owner_node_id: str,
+        function: Node,
+        profile: _LanguageProfile,
+        max_nodes: int,
+    ) -> None:
+        self.raw = raw
+        self.source_id = source_id
+        self.owner_node_id = owner_node_id
+        self.function = function
+        self.profile = profile
+        self.max_nodes = max_nodes
+        self.nodes: list[LogicNode] = []
+        self.edges: list[LogicEdge] = []
+        self._edge_ordinals: dict[str, int] = {}
+        self._sequence = 0
+        self._owner_digest = hashlib.sha256(owner_node_id.encode()).hexdigest()[:12]
+        name = _function_name(function, raw, profile) or owner_node_id.rsplit(".", 1)[-1]
+        self.entry_id = self._node("entry", f"Enter {name}", function)
+        self.exit_id = self._node("exit", f"Exit {name}", function)
+
+    def build(self) -> LogicProjection:
+        body = self.function.child_by_field_name("body")
+        incoming = (_Tail(self.entry_id),)
+        if body is None:
+            tails = incoming
+        elif body.type in self.profile.block_types:
+            tails = self._statements(body.named_children, incoming, control=None)
+        else:
+            tails = self._expression_body(body, incoming)
+        self._connect(tails, self.exit_id)
+        if not self._has_incoming(self.exit_id):
+            self._edge(self.entry_id, "next", self.exit_id, "END")
+        return LogicProjection(
+            owner_node_id=self.owner_node_id,
+            source_id=self.source_id,
+            nodes=tuple(sorted(self.nodes, key=lambda item: item.logic_id)),
+            edges=tuple(
+                sorted(
+                    self.edges,
+                    key=lambda item: (
+                        item.source_id,
+                        item.ordinal,
+                        item.relation,
+                        item.target_id,
+                        item.label or "",
+                    ),
+                )
+            ),
+        )
+
+    def _statements(
+        self,
+        statements: Iterable[Node],
+        incoming: tuple[_Tail, ...],
+        *,
+        control: _Control | None,
+    ) -> tuple[_Tail, ...]:
+        tails = incoming
+        for statement in statements:
+            if not tails:
+                break
+            tails = self._statement(statement, tails, control=control)
+        return tails
+
+    def _statement(
+        self,
+        statement: Node,
+        incoming: tuple[_Tail, ...],
+        *,
+        control: _Control | None,
+    ) -> tuple[_Tail, ...]:
+        if statement.type in self.profile.block_types:
+            return self._statements(statement.named_children, incoming, control=control)
+        if statement.type == "if_statement":
+            return self._if(statement, incoming, control=control)
+        if statement.type in self.profile.loop_types:
+            return self._loop(statement, incoming)
+        if statement.type == "switch_statement":
+            return self._switch(statement, incoming, control=control)
+        if statement.type == "try_statement":
+            return self._try(statement, incoming, control=control)
+        if statement.type in self.profile.return_types:
+            value = next(iter(statement.named_children), None)
+            label = "return" if value is None else f"return {_compact(_text(value, self.raw))}"
+            node_id = self._node("return", label, statement)
+            self._connect(incoming, node_id)
+            self._edge(node_id, "return", self.exit_id, "RETURN")
+            return ()
+        if statement.type in self.profile.raise_types:
+            value = next(iter(statement.named_children), None)
+            keyword = "throw" if self.profile.name in {"javascript", "cpp"} else "raise"
+            label = keyword if value is None else f"{keyword} {_compact(_text(value, self.raw))}"
+            node_id = self._node("raise", label, statement)
+            self._connect(incoming, node_id)
+            self._edge(node_id, "raise", self.exit_id, keyword.upper())
+            return ()
+        if statement.type == "break_statement":
+            node_id = self._node("break", "break", statement)
+            self._connect(incoming, node_id)
+            target = control.break_id if control is not None else None
+            self._edge(node_id, "break" if target else "next", target or self.exit_id, "BREAK")
+            return ()
+        if statement.type == "continue_statement":
+            node_id = self._node("continue", "continue", statement)
+            self._connect(incoming, node_id)
+            target = control.continue_id if control is not None else None
+            self._edge(
+                node_id,
+                "continue" if target else "next",
+                target or self.exit_id,
+                "CONTINUE",
+            )
+            return ()
+        if statement.type in {"function_declaration", "function_definition", "method_definition"}:
+            return incoming
+        if statement.type in {"else_clause", "finally_clause", "catch_clause"}:
+            body = statement.child_by_field_name("body")
+            return (
+                self._statement(body, incoming, control=control) if body is not None else incoming
+            )
+        node_id = self._node(
+            "call" if _contains_type(statement, "call_expression") else "action",
+            _compact(_text(statement, self.raw)),
+            statement,
+        )
+        self._connect(incoming, node_id)
+        return (_Tail(node_id),)
+
+    def _if(
+        self,
+        statement: Node,
+        incoming: tuple[_Tail, ...],
+        *,
+        control: _Control | None,
+    ) -> tuple[_Tail, ...]:
+        expression = statement.child_by_field_name("condition")
+        if expression is None:
+            expression = _first_named(statement)
+        condition = self._condition(_unwrap_condition(expression), incoming)
+        consequence = statement.child_by_field_name("consequence")
+        alternative = statement.child_by_field_name("alternative")
+        body_tails = (
+            self._statement(consequence, condition.when_true, control=control)
+            if consequence is not None
+            else condition.when_true
+        )
+        else_tails = (
+            self._statement(alternative, condition.when_false, control=control)
+            if alternative is not None
+            else condition.when_false
+        )
+        return self._converge("Decision convergence", (*body_tails, *else_tails), statement)
+
+    def _loop(self, statement: Node, incoming: tuple[_Tail, ...]) -> tuple[_Tail, ...]:
+        after_id = self._node("convergence", "Loop exit", statement)
+        condition_node = statement.child_by_field_name("condition")
+        body = statement.child_by_field_name("body")
+        if statement.type in {"for_in_statement", "for_range_loop"}:
+            loop_id = self._node(
+                "loop", _compact(_header_text(statement, body, self.raw)), statement
+            )
+            self._connect(incoming, loop_id)
+            condition = _Condition(
+                loop_id,
+                (_Tail(loop_id, "when_true", "ITEM"),),
+                (_Tail(loop_id, "when_false", "EXHAUSTED"),),
+            )
+        elif condition_node is not None:
+            condition = self._condition(_unwrap_condition(condition_node), incoming)
+            loop_id = condition.entry_id
+        else:
+            loop_id = self._node(
+                "loop", _compact(_header_text(statement, body, self.raw)), statement
+            )
+            self._connect(incoming, loop_id)
+            condition = _Condition(
+                loop_id,
+                (_Tail(loop_id, "when_true", "ITERATE"),),
+                (_Tail(loop_id, "when_false", "EXIT"),),
+            )
+        control = _Control(break_id=after_id, continue_id=loop_id)
+        body_tails = (
+            self._statement(body, condition.when_true, control=control)
+            if body is not None
+            else condition.when_true
+        )
+        for tail in body_tails:
+            self._edge(tail.source_id, "loop", loop_id, "NEXT ITERATION")
+        self._connect(condition.when_false, after_id)
+        return (_Tail(after_id),) if self._has_incoming(after_id) else ()
+
+    def _switch(
+        self,
+        statement: Node,
+        incoming: tuple[_Tail, ...],
+        *,
+        control: _Control | None,
+    ) -> tuple[_Tail, ...]:
+        expression = (
+            statement.child_by_field_name("value")
+            or statement.child_by_field_name("condition")
+            or _first_named(statement)
+        )
+        switch_id = self._node(
+            "condition",
+            f"switch {_compact(_text(_unwrap_condition(expression), self.raw))}",
+            statement,
+        )
+        self._connect(incoming, switch_id)
+        body = statement.child_by_field_name("body")
+        cases = [
+            child
+            for child in (body.named_children if body is not None else ())
+            if child.type in self.profile.switch_case_types
+        ]
+        convergence_id = self._node("convergence", "Case convergence", statement)
+        switch_control = _Control(
+            break_id=convergence_id,
+            continue_id=control.continue_id if control is not None else None,
+        )
+        completed: list[_Tail] = []
+        for case in cases:
+            case_value = case.child_by_field_name("value")
+            label = (
+                "default" if case_value is None else f"case {_compact(_text(case_value, self.raw))}"
+            )
+            case_id = self._node("case", label, case)
+            self._edge(switch_id, "case", case_id, label.upper())
+            body_nodes = tuple(
+                child
+                for child in case.named_children
+                if case_value is None or child.id != case_value.id
+            )
+            completed.extend(
+                self._statements(body_nodes, (_Tail(case_id),), control=switch_control)
+            )
+        self._connect(tuple(completed), convergence_id)
+        return (_Tail(convergence_id),) if self._has_incoming(convergence_id) else ()
+
+    def _try(
+        self,
+        statement: Node,
+        incoming: tuple[_Tail, ...],
+        *,
+        control: _Control | None,
+    ) -> tuple[_Tail, ...]:
+        try_id = self._node("try", "try", statement)
+        self._connect(incoming, try_id)
+        body = statement.child_by_field_name("body")
+        normal = (
+            self._statement(body, (_Tail(try_id),), control=control)
+            if body is not None
+            else (_Tail(try_id),)
+        )
+        branches: list[_Tail] = list(normal)
+        handlers = [child for child in statement.named_children if child.type == "catch_clause"]
+        handler = statement.child_by_field_name("handler")
+        if handler is not None and handler not in handlers:
+            handlers.append(handler)
+        for catch in handlers:
+            parameter = catch.child_by_field_name("parameter") or catch.child_by_field_name(
+                "parameters"
+            )
+            label = (
+                "catch" if parameter is None else f"catch {_compact(_text(parameter, self.raw))}"
+            )
+            catch_id = self._node("except", label, catch)
+            self._edge(try_id, "exception", catch_id, label.upper())
+            catch_body = catch.child_by_field_name("body")
+            branches.extend(
+                self._statement(catch_body, (_Tail(catch_id),), control=control)
+                if catch_body is not None
+                else (_Tail(catch_id),)
+            )
+        converged = self._converge("Exception convergence", tuple(branches), statement)
+        finalizer = statement.child_by_field_name("finalizer")
+        if finalizer is None:
+            finalizer = next(
+                (child for child in statement.named_children if child.type == "finally_clause"),
+                None,
+            )
+        if finalizer is None:
+            return converged
+        final_id = self._node("finally", "finally", finalizer)
+        self._connect(converged, final_id)
+        final_body = finalizer.child_by_field_name("body")
+        return (
+            self._statement(final_body, (_Tail(final_id),), control=control)
+            if final_body is not None
+            else (_Tail(final_id),)
+        )
+
+    def _condition(
+        self,
+        expression: Node,
+        incoming: tuple[_Tail, ...],
+    ) -> _Condition:
+        expression = _unwrap_condition(expression)
+        text = _text(expression, self.raw).strip()
+        if expression.type == "unary_expression" and text.startswith("!"):
+            operand = next(iter(expression.named_children), None)
+            if operand is not None:
+                inner = self._condition(operand, incoming)
+                return _Condition(inner.entry_id, inner.when_false, inner.when_true)
+        if expression.type == "binary_expression":
+            left = expression.child_by_field_name("left")
+            right = expression.child_by_field_name("right")
+            operator = _operator_between(left, right, self.raw)
+            if left is not None and right is not None and operator in {"&&", "||"}:
+                first = self._condition(left, incoming)
+                if operator == "&&":
+                    second = self._condition(right, first.when_true)
+                    return _Condition(
+                        first.entry_id,
+                        second.when_true,
+                        (*first.when_false, *second.when_false),
+                    )
+                second = self._condition(right, first.when_false)
+                return _Condition(
+                    first.entry_id,
+                    (*first.when_true, *second.when_true),
+                    second.when_false,
+                )
+        node_id = self._node("condition", _compact(text), expression)
+        self._connect(incoming, node_id)
+        return _Condition(
+            node_id,
+            (_Tail(node_id, "when_true", "TRUE"),),
+            (_Tail(node_id, "when_false", "FALSE"),),
+        )
+
+    def _expression_body(
+        self,
+        expression: Node,
+        incoming: tuple[_Tail, ...],
+    ) -> tuple[_Tail, ...]:
+        if expression.type == "ternary_expression":
+            condition_node = expression.child_by_field_name("condition")
+            consequence = expression.child_by_field_name("consequence")
+            alternative = expression.child_by_field_name("alternative")
+            if condition_node is not None and consequence is not None and alternative is not None:
+                condition = self._condition(condition_node, incoming)
+                true_id = self._node(
+                    "return",
+                    f"return {_compact(_text(consequence, self.raw))}",
+                    consequence,
+                )
+                false_id = self._node(
+                    "return",
+                    f"return {_compact(_text(alternative, self.raw))}",
+                    alternative,
+                )
+                self._connect(condition.when_true, true_id)
+                self._connect(condition.when_false, false_id)
+                self._edge(true_id, "return", self.exit_id, "RETURN")
+                self._edge(false_id, "return", self.exit_id, "RETURN")
+                return ()
+        node_id = self._node(
+            "return",
+            f"return {_compact(_text(expression, self.raw))}",
+            expression,
+        )
+        self._connect(incoming, node_id)
+        self._edge(node_id, "return", self.exit_id, "RETURN")
+        return ()
+
+    def _converge(
+        self,
+        label: str,
+        incoming: tuple[_Tail, ...],
+        source: Node,
+    ) -> tuple[_Tail, ...]:
+        if not incoming:
+            return ()
+        convergence_id = self._node("convergence", label, source)
+        self._connect(incoming, convergence_id)
+        return (_Tail(convergence_id),)
+
+    def _node(self, kind: str, label: str, source: Node) -> str:
+        if len(self.nodes) >= self.max_nodes:
+            raise DocForgeError(
+                "logic_too_large",
+                "A function exceeds the configured logic-node safety boundary",
+                owner_node_id=self.owner_node_id,
+                maximum=self.max_nodes,
+            )
+        self._sequence += 1
+        line = source.start_point.row + 1
+        column = source.start_point.column
+        logic_id = f"logic.{self._owner_digest}.{kind}.{line}.{column}.{self._sequence}"
+        self.nodes.append(
+            LogicNode(
+                logic_id=logic_id,
+                kind=kind,
+                label=label,
+                source_anchor=f"L{line}",
+            )
+        )
+        return logic_id
+
+    def _connect(self, incoming: tuple[_Tail, ...], target_id: str) -> None:
+        for tail in incoming:
+            self._edge(tail.source_id, tail.relation, target_id, tail.label)
+
+    def _edge(
+        self,
+        source_id: str,
+        relation: str,
+        target_id: str,
+        label: str | None,
+    ) -> None:
+        ordinal = self._edge_ordinals.get(source_id, 0)
+        self._edge_ordinals[source_id] = ordinal + 1
+        self.edges.append(
+            LogicEdge(
+                source_id=source_id,
+                relation=relation,
+                target_id=target_id,
+                label=label,
+                ordinal=ordinal,
+            )
+        )
+
+    def _has_incoming(self, node_id: str) -> bool:
+        return any(edge.target_id == node_id for edge in self.edges)
+
+
+def _field_text(node: Node, field: str, raw: bytes) -> str | None:
+    child = node.child_by_field_name(field)
+    return _text(child, raw) if child is not None else None
+
+
+def _first_named(node: Node) -> Node:
+    return node.named_children[0] if node.named_children else node
+
+
+def _unwrap_condition(node: Node) -> Node:
+    current = node
+    while current.type in {"parenthesized_expression", "condition_clause"}:
+        value = current.child_by_field_name("value")
+        current = value or _first_named(current)
+    return current
+
+
+def _operator_between(left: Node | None, right: Node | None, raw: bytes) -> str:
+    if left is None or right is None:
+        return ""
+    return raw[left.end_byte : right.start_byte].decode("utf-8", errors="replace").strip()
+
+
+def _header_text(statement: Node, body: Node | None, raw: bytes) -> str:
+    end = body.start_byte if body is not None else statement.end_byte
+    return raw[statement.start_byte : end].decode("utf-8", errors="replace").strip()
+
+
+def _contains_type(node: Node, node_type: str) -> bool:
+    if node.type == node_type:
+        return True
+    return any(_contains_type(child, node_type) for child in node.named_children)
+
+
+def _text(node: Node, raw: bytes) -> str:
+    return raw[node.start_byte : node.end_byte].decode("utf-8", errors="replace")
+
+
+def _compact(value: str, limit: int = 240) -> str:
+    compact = " ".join(value.strip().split())
+    return compact if len(compact) <= limit else f"{compact[: limit - 1]}…"