fix: immutable completed instances, no past generation, strict UTC timestamps, hardened append-only guards

- upsert refuses due_dates before today (never generate in the past)
  and refuses regeneration of any instance carrying a completion event
  (immutability); the ON CONFLICT refresh only ever touches open
  instances
- delete_future_uncompleted clamps the cutoff to today so backdating
  cannot rewrite past open instances
- occurred_at enforces ONE strict UTC shape (YYYY-MM-DDTHH:MM:SS+00:00)
  at append; occurred-at range scope uses exact half-open bounds
- was_on_time: bool | None with type(was_on_time) is bool validation -
  0.5 silently coercing to False is rejected
- append-only guards hardened: package-wide mutation regex covers
  UPDATE OR IGNORE/qualified names/quotes, DAO INSERT exempted
  narrowly; behavioral call-interception test proves only INSERT/SELECT
  ever reach the DB for completion_events; compensating self-scan for
  the instances test file's rules-table exemption
This commit is contained in:
Joshua King
2026-09-14 08:17:48 -04:00
parent ea6cf73bdc
commit e48123e488
3 changed files with 444 additions and 102 deletions
+44
View File
@@ -818,6 +818,50 @@ def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
)
def test_dao_instances_test_file_uses_dao_not_raw_rules_sql() -> None:
"""Compensating self-scan for the test_dao_instances.py exemption:
that file may query task_instances/completion_events raw (its own
guard's scope) but must go through the DAO for schedule_rules and
task_definitions, except its single sanctioned definition-create
helper usage and guard spans.
"""
import ast
import re
from pathlib import Path
instances_test = Path(__file__).parent / "test_dao_instances.py"
text = instances_test.read_text()
lines = text.splitlines(keepends=True)
tree = ast.parse(text)
guard_names = {
"test_rules_and_definitions_sql_lives_only_in_dao_module",
"test_no_mutation_sql_for_completion_events_anywhere",
}
excluded: set[int] = set()
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
if node.name in guard_names:
excluded.update(
range(node.lineno, (node.end_lineno or node.lineno) + 1)
)
remaining = "".join(
line
for number, line in enumerate(lines, start=1)
if number not in excluded
)
sql_pattern = re.compile(
r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|task_definitions)\b",
re.IGNORECASE,
)
assert sql_pattern.search(remaining) is None, (
"test_dao_instances.py must go through the DAO, not raw SQL, "
"for schedule_rules/task_definitions"
)
def test_dao_rules_test_file_uses_dao_not_raw_table_sql() -> None:
"""This test module must exercise the DAO, not raw SQL, for these
tables. Exceptions: the guard functions' own spans (their regexes