diff --git a/pineforge_codegen/codegen/base.py b/pineforge_codegen/codegen/base.py index 8d5e054..ed37ec0 100644 --- a/pineforge_codegen/codegen/base.py +++ b/pineforge_codegen/codegen/base.py @@ -6,6 +6,8 @@ from __future__ import annotations +from dataclasses import dataclass + from ..ast_nodes import ( ASTNode, Program, StrategyDecl, VarDecl, Assignment, IfStmt, ForStmt, ForInStmt, WhileStmt, SwitchStmt, BreakStmt, ContinueStmt, FuncDef, ExprStmt, @@ -27,6 +29,15 @@ from .. import signatures as sigs from ..errors import CompileError, Diagnostic, Level, Phase, SourceLocation + +@dataclass(frozen=True) +class _StableVarCtorLiteral: + """Exact declaration identity admitted by the TA-length literal route.""" + + value: int + declaration_id: int + top_level_index: int + # --------------------------------------------------------------------------- # Mapping tables — definitions live in ``tables.py``; re-imported here so # inline references inside this module (BAR_FIELDS[name], MATH_FUNC_MAP[fn], @@ -569,6 +580,12 @@ def func_var_storage(owner: str, raw_name: str) -> str: # state are NOT here, so a TA length fed by them is still rejected # by the constructor guard. self._stable_runtime_vars: set[str] = set() + # Narrow subset of persistent scalars that a TA constructor may fold + # directly: one unique top-level ``var [int] name = `` + # binding, never reassigned and never promoted to Series storage. + # Keep this separate from ``_known_vars`` so ordinary identifier use + # still reads the persistent member and only TA buffer sizing changes. + self._stable_var_ctor_literals: dict[str, _StableVarCtorLiteral] = {} # ``_var_names`` (var/varip persistent-state members) is needed by the # stability classifier during _collect_known_vars, so pre-seed it from # the analyzer's var_members before that pass runs; the canonical @@ -1883,11 +1900,14 @@ def _collect_stable_var_scalars(self, reassigned: set[str]) -> None: ``_known_vars`` (no use-site inlining): only the length-analysis path is affected, and the ``var`` member still emits and initializes normally. """ + literal_candidates = self._admitted_stable_var_ctor_literals(reassigned) + for stmt in (self.ctx.ast.body or []): if not isinstance(stmt, VarDecl): continue if not (stmt.is_var or stmt.is_varip): continue + literal_record = literal_candidates.get(stmt.name) if stmt.name in reassigned: continue if self._decl_binding_is_series(id(stmt), stmt.name): @@ -1899,6 +1919,8 @@ def _collect_stable_var_scalars(self, reassigned: set[str]) -> None: continue self._derived_input_expr[stmt.name] = expr_str self._stable_runtime_vars.add(stmt.name) + if literal_record is not None: + self._stable_var_ctor_literals[stmt.name] = literal_record # Mark input-backed iff the init references an input, so the reset # emits override-aware get_input_*() reads for it. import re as _re @@ -1906,6 +1928,215 @@ def _collect_stable_var_scalars(self, reassigned: set[str]) -> None: if any(t in self._input_backed_vars for t in toks): self._input_backed_vars.add(stmt.name) + @staticmethod + def _is_direct_int_var_literal(node) -> bool: + """Whether ``node`` is the one literal persistent shape we admit.""" + + return ( + isinstance(node, VarDecl) + and node.is_var + and not node.is_varip + and node.type_hint in (None, "int") + and isinstance(node.value, NumberLiteral) + and isinstance(node.value.value, int) + and not isinstance(node.value.value, bool) + ) + + @staticmethod + def _walk_candidate_surface(root): + """Yield every AST node reachable through compiler-owned metadata. + + The regular codegen walker intentionally follows only historically + emitted statement/expression slots. Literal admission is a safety + proof and therefore needs the larger authored surface: loop bounds and + iterables, enum/type defaults, function/method parameter defaults kept + in ``annotations``, and any future AST-bearing metadata. ``vars()`` + plus identity de-duplication is both exhaustive and cycle-safe. + """ + + seen: set[int] = set() + + def walk(value): + if value is None or isinstance( + value, (str, bytes, int, float, bool) + ): + return + value_id = id(value) + if value_id in seen: + return + seen.add(value_id) + + if isinstance(value, ASTNode): + yield value + if isinstance(value, dict): + for key, child in value.items(): + yield from walk(key) + yield from walk(child) + return + if isinstance(value, (list, tuple, set, frozenset)): + for child in value: + yield from walk(child) + return + if hasattr(value, "__dict__"): + for child in vars(value).values(): + yield from walk(child) + + yield from walk(root) + + def _stable_var_literal_has_earlier_use( + self, + name: str, + declaration_index: int, + ) -> bool: + """Whether authored code reads ``name`` before its global declaration.""" + + for root in (self.ctx.ast.body or [])[:declaration_index]: + if any( + isinstance(node, Identifier) and node.name == name + for node in self._walk_candidate_surface(root) + ): + return True + return False + + def _candidate_surface_mentions_name(self, root, name: str) -> bool: + return any( + isinstance(node, Identifier) and node.name == name + for node in self._walk_candidate_surface(root) + ) + + def _stable_var_literal_has_unsupported_declaration(self) -> bool: + """Whether authored metadata would require cross-boundary dataflow. + + Stable-var constructor folding is deliberately declaration-local. A + user function, method, type, or enum introduces defaults, aliases, or + receiver effects that require a wider proof, so this narrow feature + fails closed for the whole script instead of approximating that proof. + """ + + return any( + isinstance(node, (FuncDef, MethodDef, TypeDecl, EnumDecl)) + for node in self._walk_candidate_surface(self.ctx.ast) + ) + + def _stable_var_literal_has_parse_recovery(self) -> bool: + """Whether the parser discarded any authored source fragment. + + Recovery intentionally keeps broad corpus transpilation moving, but a + discarded declaration can hide aliases or override a builtin method. + Literal admission therefore requires a lossless parse even though all + existing non-admission codegen paths retain their recovery behavior. + """ + + return bool( + (self.ctx.ast.annotations or {}).get("parse_recovery_count", 0) + ) + + def _stable_var_literal_has_unsupported_receiver_use( + self, + name: str, + ) -> bool: + """Catch primitive extension-method syntax dropped by the parser.""" + + roots: list[object] = [self.ctx.ast] + roots.extend( + pragma.expr_node + for pragma in (getattr(self.ctx, "pf_trace_pragmas", None) or []) + if getattr(pragma, "expr_node", None) is not None + ) + for root in roots: + for node in self._walk_candidate_surface(root): + if ( + isinstance(node, FuncCall) + and isinstance(node.callee, MemberAccess) + and self._candidate_surface_mentions_name( + node.callee.object, name + ) + ): + return True + return False + + def _stable_var_literal_has_history_use(self, name: str) -> bool: + """Whether any authored or post-analyzer expression histories ``name``.""" + + roots: list[object] = [self.ctx.ast] + roots.extend( + pragma.expr_node + for pragma in (getattr(self.ctx, "pf_trace_pragmas", None) or []) + if getattr(pragma, "expr_node", None) is not None + ) + for root in roots: + for node in self._walk_candidate_surface(root): + if not isinstance(node, Subscript): + continue + if any( + isinstance(receiver_node, Identifier) + and receiver_node.name == name + for receiver_node in self._walk_candidate_surface( + node.object + ) + ): + return True + return False + + def _admitted_stable_var_ctor_literals( + self, + reassigned: set[str], + ) -> dict[str, _StableVarCtorLiteral]: + """Return declaration-exact persistent literals safe for TA sizing.""" + + occurrences: dict[str, list[tuple[int, VarDecl]]] = {} + for index, stmt in enumerate(self.ctx.ast.body or []): + if self._is_direct_int_var_literal(stmt): + occurrences.setdefault(stmt.name, []).append((index, stmt)) + + unique = { + name: items[0] + for name, items in occurrences.items() + if len(items) == 1 + } + if not unique: + return {} + + candidate_names = set(unique) + binding_counts = {name: 0 for name in candidate_names} + for node in self._walk_candidate_surface(self.ctx.ast): + names: list[str] = [] + if isinstance(node, VarDecl) and node.name: + names.append(node.name) + elif isinstance(node, TupleAssign): + names.extend(name for name in node.names if name != "_") + elif isinstance(node, ForStmt) and node.var: + names.append(node.var) + elif isinstance(node, ForInStmt): + if node.var: + names.append(node.var) + names.extend(name for name in (node.vars or []) if name != "_") + if isinstance(node, (FuncDef, MethodDef)): + names.extend(name for name in node.params if name != "_") + for bound_name in names: + if bound_name in binding_counts: + binding_counts[bound_name] += 1 + + admitted: dict[str, _StableVarCtorLiteral] = {} + for name, (index, declaration) in unique.items(): + if ( + name in reassigned + or binding_counts.get(name) != 1 + or self._decl_binding_is_series(id(declaration), name) + or self._stable_var_literal_has_earlier_use(name, index) + or self._stable_var_literal_has_history_use(name) + or self._stable_var_literal_has_unsupported_declaration() + or self._stable_var_literal_has_parse_recovery() + or self._stable_var_literal_has_unsupported_receiver_use(name) + ): + continue + admitted[name] = _StableVarCtorLiteral( + value=declaration.value.value, + declaration_id=id(declaration), + top_level_index=index, + ) + return admitted + def _collect_reassigned_stable_scalars(self, reassigned: set[str]) -> None: """Track class-scope scalars that are reassigned but only along stable if/elif paths (see ``test_stable_reassigned_class_scope_length``). @@ -3452,6 +3683,27 @@ def generate(self) -> str: # _if_body_has_ta / _is_result_assignment / _expr_contains_ta / # _hoist_if_body live on TaSiteHelper (codegen/ta.py). + def _resolve_ta_ctor_arg(self, arg_str: str) -> str: + """Resolve one TA constructor argument without widening const folding. + + ``var`` values deliberately stay out of ``_known_vars`` because normal + expression reads must still target persistent storage. A uniquely + bound, never-reassigned top-level integer literal is nevertheless a + faithful static TA buffer length. Admit only that exact bare alias; + arithmetic, input/timeframe expressions, shadows, and all mutable or + history-bearing bindings continue through the existing resolver/reset + guardrails unchanged. + """ + resolved = self._resolve_known(arg_str) + if self._is_compile_time_value(resolved): + return resolved + if ( + arg_str in self._stable_var_ctor_literals + and not self._known_var_is_lexically_shadowed(arg_str) + ): + return str(self._stable_var_ctor_literals[arg_str].value) + return resolved + def _resolve_known(self, arg_str: str) -> str: """Resolve a string arg, replacing known var names with their values. @@ -3937,7 +4189,7 @@ def _collect_ta_runtime_resets(self) -> list[str]: runtime_args.append(rt) any_runtime = True else: - resolved = self._resolve_known(a) + resolved = self._resolve_ta_ctor_arg(a) runtime_args.append(resolved if self._is_compile_time_value(resolved) else "1") if any_runtime: resets.append( @@ -3984,7 +4236,7 @@ def _collect_ta_runtime_resets(self) -> list[str]: any_runtime = False for arg_pos, a in enumerate(ctor_args): rt = self._runtime_ctor_arg_for_reset(a) - resolved = self._resolve_known(a) + resolved = self._resolve_ta_ctor_arg(a) lowered_variant = ctor_arg_stability is not None stable_variant_arg = ( lowered_variant and ctor_arg_stability[arg_pos] diff --git a/pineforge_codegen/codegen/emit_top.py b/pineforge_codegen/codegen/emit_top.py index 393726a..dfd7379 100644 --- a/pineforge_codegen/codegen/emit_top.py +++ b/pineforge_codegen/codegen/emit_top.py @@ -516,7 +516,7 @@ def _emit_constructor(self, lines: list[str]) -> None: # incl. function-derived lengths) are safe: the `!_ta_initialized_` # reset overwrites the placeholder before the first compute. for a in site.ctor_args: - r = self._resolve_known(a) + r = self._resolve_ta_ctor_arg(a) if (not self._is_compile_time_value(r) and self._runtime_ctor_arg_for_reset(a) is None): self._codegen_error( @@ -528,7 +528,7 @@ def _emit_constructor(self, lines: list[str]) -> None: hint=("Use a literal, an input.*() value, or " "arithmetic over those for TA lengths."), ) - resolved = [self._resolve_known(a) for a in site.ctor_args] + resolved = [self._resolve_ta_ctor_arg(a) for a in site.ctor_args] # Compile-time placeholder for the init list; the runtime reset # (when the arg is input-derived) overwrites it on the first bar. safe_resolved = [] @@ -550,7 +550,7 @@ def _emit_constructor(self, lines: list[str]) -> None: site, variant.get("binding_stack", ()), ) - resolved = [self._resolve_known(a) for a in ctor_args] + resolved = [self._resolve_ta_ctor_arg(a) for a in ctor_args] safe_resolved = [] for r in resolved: safe_resolved.append( @@ -1693,7 +1693,7 @@ def _emit_precalculate_and_run(self, lines: list[str]) -> None: if _ti in self._dead_ta_indices: continue if self._ta_site_uses_precalc(site): - resolved = [self._resolve_known(a) for a in site.ctor_args] + resolved = [self._resolve_ta_ctor_arg(a) for a in site.ctor_args] safe_resolved = [] for r in resolved: safe_resolved.append(r if self._is_compile_time_value(r) else "1") @@ -1805,7 +1805,7 @@ def _emit_precalculate_and_run(self, lines: list[str]) -> None: if _ti in self._dead_ta_indices: continue if self._ta_site_uses_precalc(site): - resolved = [self._resolve_known(a) for a in site.ctor_args] + resolved = [self._resolve_ta_ctor_arg(a) for a in site.ctor_args] safe_resolved = [] for r in resolved: safe_resolved.append(r if self._is_compile_time_value(r) else "1") diff --git a/pineforge_codegen/codegen/security.py b/pineforge_codegen/codegen/security.py index 72f9192..a49f2af 100644 --- a/pineforge_codegen/codegen/security.py +++ b/pineforge_codegen/codegen/security.py @@ -2056,10 +2056,22 @@ def _security_ta_ctor_depends_on_mutables( security_mutable_names: set[str], helper_binding_stack: tuple[dict[str, ASTNode], ...] | None = None, ) -> bool: + # A top-level ``var int L = 5`` is normally classified as mutable by + # request.security because persistent state is rebound per requested + # context. The TA constructor does not read that mutable member when + # ``L`` passed the declaration-exact stable-literal admission proof: + # all main/requested-context buffers are sized directly from ``5``. + # Remove only those admitted names for this ctor-dependency query. + # Compute arguments and every other security mutability path retain + # the full set, and helper bindings still resolve transitively before + # checking the reduced set. + ctor_mutable_names = security_mutable_names.difference( + getattr(self, "_stable_var_ctor_literals", {}) + ) return any( self._expr_depends_on_security_mutables( arg, - security_mutable_names, + ctor_mutable_names, helper_binding_stack=helper_binding_stack, ) for arg in self._security_ta_ctor_arg_nodes(site) diff --git a/pineforge_codegen/parser.py b/pineforge_codegen/parser.py index f9864c7..9ca1e7a 100644 --- a/pineforge_codegen/parser.py +++ b/pineforge_codegen/parser.py @@ -56,6 +56,7 @@ def __init__(self, tokens: list[Token], *, source: str = "", filename: str = " Program: self._recover() self._skip_newlines() + if self._recovery_count: + prog.annotations = dict(prog.annotations or {}) + prog.annotations["parse_recovery_count"] = self._recovery_count return prog def _extract_version(self) -> int | None: @@ -156,6 +160,7 @@ def _extract_version(self) -> int | None: def _recover(self) -> None: """Skip tokens until next NEWLINE or EOF for error recovery.""" + self._recovery_count += 1 while not self._at_end() and not self._check(TokenType.NEWLINE): self._advance() if self._check(TokenType.NEWLINE): diff --git a/tests/test_codegen_ta_stable_var_literal.py b/tests/test_codegen_ta_stable_var_literal.py new file mode 100644 index 0000000..1de6eaf --- /dev/null +++ b/tests/test_codegen_ta_stable_var_literal.py @@ -0,0 +1,595 @@ +"""TA constructor lengths backed by immutable persistent literals. + +Pine permits a top-level ``var int`` initialized from a literal to feed a TA +length when that binding is never reassigned. The value is fixed for the +entire run, so it has the same constructor-sizing semantics as an ordinary +literal alias. Keep the admission deliberately narrower than the general +stable-runtime reset path: mutable, history-promoted, and dynamic persistent +bindings must continue to fail closed. +""" + +from __future__ import annotations + +import re + +import pytest + +from pineforge_codegen import transpile +from pineforge_codegen.errors import CompileError, Level, Phase +from pineforge_codegen.lexer import Lexer +from pineforge_codegen.parser import Parser +from tests._compile import compile_cpp + + +def _constructor_initializers(cpp: str) -> str: + match = re.search(r"explicit GeneratedStrategy\(\) : ([^\n]+)", cpp) + assert match is not None, "generated strategy constructor not found" + return match.group(1) + + +def _assert_exact_pivot_ctor_rejection( + source: str, + *, + filename: str, + expected_line: int, +) -> None: + with pytest.raises(CompileError) as caught: + transpile(source, filename=filename) + + assert len(caught.value.diagnostics) == 1 + diagnostic = caught.value.diagnostics[0] + assert diagnostic.level is Level.ERROR + assert diagnostic.phase is Phase.CODEGEN + assert ( + diagnostic.location.file, + diagnostic.location.line, + diagnostic.location.col, + diagnostic.location.end_col, + ) == (filename, expected_line, 17, 18) + assert diagnostic.message == ( + "Unsupported TA constructor length 'p' for ta::PivotHigh: it is " + "neither a compile-time constant nor derived from an input, so " + "PineForge cannot size the indicator buffer." + ) + assert diagnostic.hint == ( + "Use a literal, an input.*() value, or arithmetic over those for TA " + "lengths." + ) + + +def _assert_parse_recovery_fences_sma(source: str, *, expected_line: int) -> None: + program = Parser(Lexer(source).tokenize(), source=source).parse() + assert (program.annotations or {}).get("parse_recovery_count") == 1 + + with pytest.raises(CompileError) as caught: + transpile(source, filename="stable-var-parse-recovery.pine") + + assert len(caught.value.diagnostics) == 1 + diagnostic = caught.value.diagnostics[0] + assert diagnostic.phase is Phase.CODEGEN + assert diagnostic.location.line == expected_line + assert diagnostic.message == ( + "Unsupported TA constructor length 'p' for ta::SMA: it is neither a " + "compile-time constant nor derived from an input, so PineForge cannot " + "size the indicator buffer." + ) + + +@pytest.mark.parametrize( + "declaration", + [ + "int pivotLength = 5", + "var int pivotLength = 5", + ], +) +def test_literal_alias_sizes_both_pivot_sides(declaration: str) -> None: + source = f"""//@version=6 +strategy("stable literal pivot lengths") +{declaration} +highPivot = ta.pivothigh(high, pivotLength, pivotLength) +lowPivot = ta.pivotlow(low, pivotLength, pivotLength) +""" + + cpp = transpile(source) + initializers = _constructor_initializers(cpp) + assert re.search(r"_ta_pivothigh_\d+\(5, 5\)", initializers) + assert re.search(r"_ta_pivotlow_\d+\(5, 5\)", initializers) + assert "ta::PivotHigh(1, 1)" not in cpp + assert "ta::PivotLow(1, 1)" not in cpp + compile_cpp(cpp, label=f"stable-var-pivots-{declaration.startswith('var')}") + + +@pytest.mark.parametrize( + "prefix", + [ + "x = ta.pivothigh(high, p, p)\n", + "prior = p\nvar int p = 5\nx = ta.pivothigh(high, p, p)\n", + ], + ids=["ta-use", "ordinary-use"], +) +def test_literal_declaration_cannot_retroactively_authorize_earlier_use( + prefix: str, +) -> None: + if prefix.startswith("x ="): + body = prefix + "var int p = 5\n" + expected_line = 3 + else: + body = prefix + expected_line = 5 + source = '//@version=6\nstrategy("predecl")\n' + body + + _assert_exact_pivot_ctor_rejection( + source, + filename="stable-var-predecl.pine", + expected_line=expected_line, + ) + + +@pytest.mark.parametrize( + "hidden_history", + [ + "prior = p[1]", + "read(int prior = p[1]) => prior", + "type Sample\n int prior = p", + "type Sample\n int prior = p[1]", + "enum Choice\n First = p[1]", + ( + "type Sample\n" + " int value = 0\n" + "method read(Sample self, int prior = p[1]) => prior" + ), + "for i = p[1] to 2\n plot(i)", + "for i = 0 to p[1]\n plot(i)", + "for i = 0 to 2 by p[1]\n plot(i)", + "for item in array.from(p[1])\n plot(item)", + "// @pf-trace prior=p[1]", + "history(int q = p) => q[1]\nprior = history()", + "method history(int self) => self[1]\nprior = p.history()", + "leaf(int r) => r[1]\nouter(int q) => leaf(q)\nprior = outer(p)", + ( + "leaf(int s) => s[1]\n" + "outer(int r) =>\n" + " int q = r\n" + " leaf(q)\n" + "prior = outer(p)" + ), + ( + "method history(int self) => self[1]\n" + "outer(int q) => q.history()\n" + "prior = outer(p)" + ), + "identity(int q = p) => q\nprior = identity()", + "history(int q) => q[1]\n// @pf-trace prior=history(p)", + ], + ids=[ + "ordinary-ast", + "function-param-default-annotation", + "type-field-plain-default", + "type-field-default", + "enum-member-value", + "method-param-default-annotation", + "for-start", + "for-end", + "for-step", + "for-in-iterable", + "post-analyzer-pf-trace", + "default-param-transitive-history", + "primitive-method-receiver-transitive-history", + "nested-helper-transitive-history", + "local-alias-sensitive-callee", + "nested-method-receiver-transitive-history", + "nonhistory-default-mask", + "pf-trace-helper-transitive-history", + ], +) +def test_any_hidden_history_use_fences_stable_literal_admission( + hidden_history: str, +) -> None: + source = ( + '//@version=6\nstrategy("hidden history")\nvar int p = 5\n' + + hidden_history + + "\nx = ta.pivothigh(high, p, p)\n" + ) + expected_line = source.splitlines().index( + "x = ta.pivothigh(high, p, p)" + ) + 1 + + _assert_exact_pivot_ctor_rejection( + source, + filename="stable-var-hidden-history.pine", + expected_line=expected_line, + ) + + +@pytest.mark.parametrize( + "declaration", + [ + "unused(int q) => q", + "type Sample\n int value = 0", + "enum Choice\n First", + ], + ids=["function", "type", "enum"], +) +def test_any_user_declaration_fences_narrow_literal_admission( + declaration: str, +) -> None: + source = ( + '//@version=6\nstrategy("declaration fence")\nvar int p = 5\n' + + declaration + + "\nx = ta.pivothigh(high, p, p)\n" + ) + expected_line = source.splitlines().index( + "x = ta.pivothigh(high, p, p)" + ) + 1 + + _assert_exact_pivot_ctor_rejection( + source, + filename="stable-var-declaration-fence.pine", + expected_line=expected_line, + ) + + +def test_dropped_primitive_method_declaration_fences_unrelated_receiver_mask() -> None: + source = '''//@version=6 +strategy("parse recovery receiver mask") +method id(int self) => self +x = 1 +y = x.id() +var int p = 5 +z = ta.sma(close, p) +''' + + _assert_parse_recovery_fences_sma(source, expected_line=7) + + +def test_dropped_collection_method_cannot_collide_with_builtin_semantics() -> None: + source = '''//@version=6 +strategy("parse recovery builtin collision") +method push(array self, int x) => array.unshift(self, x) +var array a = array.new() +a.push(1) +var int p = 5 +z = ta.sma(close, p) +''' + + _assert_parse_recovery_fences_sma(source, expected_line=7) + + +@pytest.mark.parametrize( + "bad_fragment", + [ + "var int = 1", + "if close > open\n var int = 1", + ], + ids=["top-level", "nested-block"], +) +def test_any_unrelated_parse_recovery_fences_literal_admission( + bad_fragment: str, +) -> None: + source = ( + '//@version=6\nstrategy("parse recovery fence")\n' + + bad_fragment + + "\nvar int p = 5\nz = ta.sma(close, p)\n" + ) + expected_line = source.splitlines().index("z = ta.sma(close, p)") + 1 + + _assert_parse_recovery_fences_sma(source, expected_line=expected_line) + + +def test_clean_direct_source_has_no_parse_recovery_annotation() -> None: + source = '''//@version=6 +strategy("lossless parse control") +var int p = 5 +z = ta.sma(close, p) +''' + program = Parser(Lexer(source).tokenize(), source=source).parse() + assert not program.annotations + + cpp = transpile(source) + assert re.search(r"_ta_sma_\d+\(5\)", _constructor_initializers(cpp)) + compile_cpp(cpp, label="stable-var-lossless-parse-control") + + +@pytest.mark.parametrize( + "push_call", + [ + "array.push(a, p)", + "a.push(p)", + ], + ids=["namespace", "known-array-receiver"], +) +def test_supported_array_calls_do_not_create_a_recovery_fence( + push_call: str, +) -> None: + source = f'''//@version=6 +strategy("supported array control") +var int p = 5 +var array a = array.new() +{push_call} +z = ta.sma(close, p) +''' + program = Parser(Lexer(source).tokenize(), source=source).parse() + assert not program.annotations + + cpp = transpile(source) + assert re.search(r"_ta_sma_\d+\(5\)", _constructor_initializers(cpp)) + compile_cpp(cpp, label=f"stable-var-array-{push_call.startswith('array.')}") + + +def test_any_recursive_callable_invocation_fences_stable_literal() -> None: + source = '''//@version=6 +strategy("recursive callable control") +identity(q) => identity(q) +var int p = 5 +unused = identity(p) +x = ta.sma(close, p) +''' + + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(source) + + +def test_even_immutable_helper_boundary_fails_closed() -> None: + source = '''//@version=6 +strategy("immutable helper parameter") +requested(int q) => ta.sma(close, q) +var int p = 5 +observed = requested(p) +''' + + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(source) + + +@pytest.mark.parametrize( + "reassignment", + [ + "q := 7", + "q := int(close)", + ], + ids=["literal-reassignment", "series-reassignment"], +) +def test_callable_parameter_reassignment_fences_literal_specialization( + reassignment: str, +) -> None: + source = f'''//@version=6 +strategy("reassigned helper parameter") +var int p = 5 +requested(int q) => + {reassignment} + ta.sma(close, q) +observed = requested(p) +''' + + with pytest.raises(CompileError) as caught: + transpile(source) + + assert len(caught.value.diagnostics) == 1 + diagnostic = caught.value.diagnostics[0] + assert diagnostic.phase is Phase.CODEGEN + assert diagnostic.message == ( + "Unsupported TA constructor length 'p' for ta::SMA: it is neither a " + "compile-time constant nor derived from an input, so PineForge cannot " + "size the indicator buffer." + ) + + +@pytest.mark.parametrize( + "helper_body", + [ + ( + " float out = na\n" + " for q = 1 to 3\n" + " out := ta.sma(close, q)\n" + " out" + ), + ( + " float out = na\n" + " if close > open\n" + " int q = 7\n" + " out := ta.sma(close, q)\n" + " out" + ), + " [q, z] = [7, 8]\n ta.sma(close, q)", + ], + ids=["loop-binder", "block-local", "tuple-binder"], +) +def test_callable_parameter_rebinding_fences_literal_specialization( + helper_body: str, +) -> None: + source = ( + '//@version=6\nstrategy("rebound helper parameter")\n' + "var int p = 5\n" + "requested(int q) =>\n" + + helper_body + + "\nobserved = requested(p)\n" + ) + + with pytest.raises(CompileError) as caught: + transpile(source) + + assert len(caught.value.diagnostics) == 1 + diagnostic = caught.value.diagnostics[0] + assert diagnostic.phase is Phase.CODEGEN + assert diagnostic.message == ( + "Unsupported TA constructor length 'p' for ta::SMA: it is neither a " + "compile-time constant nor derived from an input, so PineForge cannot " + "size the indicator buffer." + ) + + +@pytest.mark.parametrize( + "binding", + [ + "var int pivotLength = 5\npivotLength := close > open ? 6 : 7", + "var int pivotLength = int(close)", + "var int pivotLength = 5\npreviousLength = pivotLength[1]", + "var pivotLength = 5.0", + "var float pivotLength = 5", + ], + ids=[ + "reassigned", + "dynamic-initializer", + "history-promoted", + "inferred-float-literal", + "declared-float", + ], +) +def test_nonimmutable_persistent_lengths_remain_rejected(binding: str) -> None: + source = f"""//@version=6 +strategy("persistent length guard") +{binding} +highPivot = ta.pivothigh(high, pivotLength, pivotLength) +""" + + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(source) + + +def test_same_spelled_callable_var_does_not_capture_global_literal() -> None: + source = """//@version=6 +strategy("persistent length shadow guard") +var int pivotLength = 5 +readPivot() => + var int pivotLength = 7 + ta.pivothigh(high, pivotLength, pivotLength) +highPivot = readPivot() +""" + + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(source) + + +def test_same_spelled_block_local_does_not_capture_global_literal() -> None: + source = """//@version=6 +strategy("persistent block length shadow guard") +var int pivotLength = 5 +if close > open + int pivotLength = int(close) + highPivot = ta.pivothigh(high, pivotLength, pivotLength) +""" + + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(source) + + +def test_same_spelled_loop_binder_does_not_capture_global_literal() -> None: + source = """//@version=6 +strategy("persistent loop length shadow guard") +var int pivotLength = 5 +for pivotLength = 1 to int(close) + highPivot = ta.pivothigh(high, pivotLength, pivotLength) +""" + + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(source) + + +@pytest.mark.parametrize( + "request_call", + [ + 'request.security(syminfo.tickerid, "60", ta.sma(close, L))', + 'request.security_lower_tf(syminfo.tickerid, "1", ta.sma(close, L))', + ], + ids=["direct", "lower-tf"], +) +def test_requested_context_admits_only_proven_stable_var_literal_ctor( + request_call: str, +) -> None: + source = ( + '//@version=6\nstrategy("requested stable literal")\n' + + "var int L = 5\n" + + f"requested = {request_call}\n" + ) + + cpp = transpile(source) + initializers = _constructor_initializers(cpp) + assert re.search(r"_ta_sma_\d+\(5\)", initializers) + assert re.search(r"_sec\d+__ta_sma_\d+\(5\)", initializers) + assert "ta::SMA(1)" not in cpp + assert not re.search(r"_ta_sma_\d+\(1\)", cpp) + assert not re.search(r"_sec\d+__ta_sma_\d+\(1\)", cpp) + compile_cpp( + cpp, + label=( + "stable-var-requested-" + f"{request_call.startswith('request.security_lower_tf')}" + ), + ) + + +def test_requested_helper_boundary_fails_closed() -> None: + source = '''//@version=6 +strategy("requested stable literal helper fence") +requestedSma(src, length) => ta.sma(src, length) +var int L = 5 +requested = request.security( + syminfo.tickerid, "60", requestedSma(close, L)) +''' + + with pytest.raises(CompileError, match="Unsupported TA constructor length"): + transpile(source) + + +@pytest.mark.parametrize( + "boundary", + [ + "var int L = 5\nL := close > open ? 6 : 7", + "var int L = 5\nprior = L[1]", + "shadow(int L) => L\nvar int L = 5", + ], + ids=["reassigned", "history", "same-spelled-parameter"], +) +def test_requested_context_does_not_exempt_unadmitted_literal(boundary: str) -> None: + source = f'''//@version=6 +strategy("requested stable literal negative") +{boundary} +requested = request.security(syminfo.tickerid, "60", ta.sma(close, L)) +''' + + with pytest.raises(CompileError) as caught: + transpile(source) + + assert len(caught.value.diagnostics) == 1 + diagnostic = caught.value.diagnostics[0] + assert diagnostic.phase is Phase.CODEGEN + assert diagnostic.message == ( + "Unsupported TA constructor length 'L' for ta::SMA: it is neither a " + "compile-time constant nor derived from an input, so PineForge cannot " + "size the indicator buffer." + ) + + +def test_input_backed_parameter_keeps_override_aware_reset() -> None: + source = """//@version=6 +strategy("input parameter boundary") +readPivot(int length) => + ta.pivothigh(high, length, length) +pivotLength = input.int(5, "Pivot Length", minval=1) +highPivot = readPivot(pivotLength) +""" + + cpp = transpile(source) + initializers = _constructor_initializers(cpp) + assert re.search(r"_ta_pivothigh_\d+\(5, 5\)", initializers) + assert ( + 'ta::PivotHigh(get_input_int("Pivot Length", 5), ' + 'get_input_int("Pivot Length", 5))' + ) in cpp + compile_cpp(cpp, label="stable-var-input-parameter-boundary") + + +def test_literal_parameter_keeps_existing_static_constructor_path() -> None: + source = """//@version=6 +strategy("literal parameter boundary") +readPivot(int length) => + ta.pivotlow(low, length, length) +lowPivot = readPivot(5) +""" + + cpp = transpile(source) + assert re.search( + r"_ta_pivotlow_\d+\(5, 5\)", _constructor_initializers(cpp) + ) + assert "get_input_int(" not in cpp + compile_cpp(cpp, label="stable-var-literal-parameter-boundary")