| 1 | """Convention enforcement for `log.source_log(..., tty_only=False)`. |
| 2 | |
| 3 | `log.source_log` defaults to `tty_only=True`, silently dropping every line |
| 4 | when stderr isn't a real TTY (every Claude Code / Codex / CI / captured- |
| 5 | output run). The default exists to keep interactive output uncluttered, |
| 6 | but it weaponizes any source module that forgets to opt out: error logs, |
| 7 | query heartbeats, and success signals all disappear. |
| 8 | |
| 9 | Ten source modules quietly shipped with this bug. This test prevents the |
| 10 | eleventh. Every `log.source_log(...)` call site under |
| 11 | `skills/last30days/scripts/lib/` must pass `tty_only=False` explicitly. |
| 12 | The cost is one kwarg per call; the value is that source observability |
| 13 | never goes silent again, even when the next contributor copies an old |
| 14 | `_log` template without thinking. |
| 15 | |
| 16 | Implementation uses `ast.parse` (not regex) so the convention check is |
| 17 | robust against multi-line calls, nested parens in f-strings, calls inside |
| 18 | docstrings or comments, whitespace variation (`tty_only = False`), and |
| 19 | future indentation styles. |
| 20 | """ |
| 21 | |
| 22 | import ast |
| 23 | import io |
| 24 | import pathlib |
| 25 | import sys |
| 26 | import unittest |
| 27 | from unittest.mock import patch |
| 28 | |
| 29 | from lib import bluesky, perplexity |
| 30 | |
| 31 | REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent |
| 32 | LIB_DIR = REPO_ROOT / "skills" / "last30days" / "scripts" / "lib" |
| 33 | |
| 34 | |
| 35 | class _SourceLogCallFinder(ast.NodeVisitor): |
| 36 | """Collect every `log.source_log(...)` call from a parsed module.""" |
| 37 | |
| 38 | def __init__(self) -> None: |
| 39 | self.calls: list[ast.Call] = [] |
| 40 | |
| 41 | def visit_Call(self, node: ast.Call) -> None: |
| 42 | func = node.func |
| 43 | if ( |
| 44 | isinstance(func, ast.Attribute) |
| 45 | and func.attr == "source_log" |
| 46 | and isinstance(func.value, ast.Name) |
| 47 | and func.value.id == "log" |
| 48 | ): |
| 49 | self.calls.append(node) |
| 50 | self.generic_visit(node) |
| 51 | |
| 52 | |
| 53 | def _has_tty_only_false_kwarg(call: ast.Call) -> bool: |
| 54 | """Return True iff the call passes `tty_only=False` as a keyword arg.""" |
| 55 | for kw in call.keywords: |
| 56 | if kw.arg == "tty_only" and isinstance(kw.value, ast.Constant) and kw.value.value is False: |
| 57 | return True |
| 58 | return False |
| 59 | |
| 60 | |
| 61 | def _iter_source_modules() -> list[pathlib.Path]: |
| 62 | """Yield production source modules under lib/ (excluding vendor and __init__).""" |
| 63 | paths: list[pathlib.Path] = [] |
| 64 | for path in LIB_DIR.rglob("*.py"): |
| 65 | if path.name == "log.py": |
| 66 | continue # definition site; not a caller |
| 67 | if "vendor" in path.parts: |
| 68 | continue # vendored third-party code |
| 69 | if path.name == "__init__.py": |
| 70 | continue # bare package marker per AGENTS.md |
| 71 | paths.append(path) |
| 72 | return sorted(paths) |
| 73 | |
| 74 | |
| 75 | class SourceLogConventionTests(unittest.TestCase): |
| 76 | """Enforce the project convention at the codebase level.""" |
| 77 | |
| 78 | def test_every_source_log_call_passes_tty_only_false(self): |
| 79 | violations: list[str] = [] |
| 80 | for path in _iter_source_modules(): |
| 81 | text = path.read_text(encoding="utf-8") |
| 82 | try: |
| 83 | tree = ast.parse(text, filename=str(path)) |
| 84 | except SyntaxError as exc: |
| 85 | self.fail(f"Could not parse {path.name}: {exc}") |
| 86 | finder = _SourceLogCallFinder() |
| 87 | finder.visit(tree) |
| 88 | for call in finder.calls: |
| 89 | if not _has_tty_only_false_kwarg(call): |
| 90 | violations.append(f"{path.relative_to(REPO_ROOT)}:{call.lineno}") |
| 91 | if violations: |
| 92 | self.fail( |
| 93 | "Source modules must call `log.source_log(..., tty_only=False)` so " |
| 94 | "lines stay visible under non-TTY contexts (Claude Code, Codex, CI, " |
| 95 | "captured output). See AGENTS.md for the convention. Violations:\n - " |
| 96 | + "\n - ".join(violations) |
| 97 | ) |
| 98 | |
| 99 | |
| 100 | class _NonTTYStringIO(io.StringIO): |
| 101 | """StringIO subclass that reports as a non-TTY stream.""" |
| 102 | |
| 103 | def isatty(self) -> bool: |
| 104 | return False |
| 105 | |
| 106 | |
| 107 | class PerplexityAndBlueskyVisibilityTests(unittest.TestCase): |
| 108 | """Targeted regression tests for the two original bug instances. |
| 109 | |
| 110 | Kept alongside the convention test so a future refactor of either |
| 111 | module immediately surfaces a regression if the opt-out is dropped. |
| 112 | """ |
| 113 | |
| 114 | def _captured_stderr_under_non_tty(self, log_callable) -> str: |
| 115 | fake_stderr = _NonTTYStringIO() |
| 116 | with patch.object(sys, "stderr", fake_stderr): |
| 117 | log_callable("visibility probe") |
| 118 | return fake_stderr.getvalue() |
| 119 | |
| 120 | def test_perplexity_log_visible_under_non_tty(self): |
| 121 | out = self._captured_stderr_under_non_tty(perplexity._log) |
| 122 | self.assertIn("[Perplexity] visibility probe", out) |
| 123 | |
| 124 | def test_bluesky_log_visible_under_non_tty(self): |
| 125 | out = self._captured_stderr_under_non_tty(bluesky._log) |
| 126 | self.assertIn("[Bluesky] visibility probe", out) |
| 127 | |
| 128 | |
| 129 | if __name__ == "__main__": |
| 130 | unittest.main() |
| 131 |