ticket-218: fix SQLite FD leaks, N+1 queries, uncached YAML loop parses, and unwired parameters

planfile/analysis/generator.py

@@ -198,6 +198,8 @@ def generate_from_analysis(self,
         tickets = self.generator.generate_tickets(analysis_result)
 
         metrics = self._extract_key_metrics(analysis_result, external_metrics)
+        if compact:
+            metrics.pop('high_cc_functions', None)
         project_name = project_name or Path(analysis_path).resolve().name
 
         strategy_data = {

planfile/api/server.py

@@ -1933,7 +1933,7 @@ async def _start_ticket_index_maintenance() -> None:
         or _ticket_index_maintenance_task.done()
     ):
         _ticket_index_maintenance_task = asyncio.create_task(
-            _maintain_ticket_index()
+            _maintain_ticket_index(interval_seconds=3.0)
         )
 
 
@@ -1954,7 +1954,9 @@ async def _start_archive_maintenance() -> None:
     if os.environ.get("PLANFILE_DISABLE_ARCHIVE_MAINTENANCE") == "1":
         return
     if _archive_maintenance_task is None or _archive_maintenance_task.done():
-        _archive_maintenance_task = asyncio.create_task(_archive_history_daily())
+        _archive_maintenance_task = asyncio.create_task(
+            _archive_history_daily(interval_seconds=300.0)
+        )
 
 
 async def _stop_archive_maintenance() -> None:
@@ -1989,7 +1991,9 @@ async def _start_planfile_watcher() -> None:
     if os.environ.get("PLANFILE_DISABLE_WATCHER") == "1":
         return
     if _watch_task is None or _watch_task.done():
-        _watch_task = asyncio.create_task(_watch_planfile_changes())
+        _watch_task = asyncio.create_task(
+            _watch_planfile_changes(interval_seconds=3.0)
+        )
 
 
 async def _stop_planfile_watcher() -> None:

planfile/core/configuration.py

@@ -10,6 +10,7 @@
 import tempfile
 from collections.abc import Callable
 from dataclasses import dataclass
+from functools import lru_cache
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
@@ -29,6 +30,16 @@
     from planfile import Planfile
 
 
+@lru_cache(maxsize=256)
+def _load_yaml_cached(path_str: str, mtime_ns: int) -> Any:
+    return yaml.safe_load(Path(path_str).read_text(encoding="utf-8"))
+
+
+def _read_yaml_file(path: Path) -> Any:
+    stat = path.stat()
+    return _load_yaml_cached(str(path.resolve()), stat.st_mtime_ns)
+
+
 _SENSITIVE_PATH = re.compile(
     r"(^|[._-])(authorization|cookie|credential|password|passwd|secret|token|"
     r"api[_-]?key|private[_-]?key)($|[._-])",
@@ -393,7 +404,7 @@ def _integration_values(self) -> dict[str, Any]:
         # must neither mutate the process environment nor materialize secrets.
         values: dict[str, Any] = {}
         for path in self._integration_paths():
-            loaded = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
+            loaded = _read_yaml_file(path) or {}
             if isinstance(loaded, dict):
                 self._deep_merge(values, loaded)
         integrations = values.get("integrations", {}) if isinstance(values, dict) else {}
@@ -414,7 +425,7 @@ def revision(self) -> str:
         integration_sources = []
         for path in self._integration_paths():
             try:
-                document = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
+                document = _read_yaml_file(path) or {}
             except FileNotFoundError:
                 continue
             integrations = (
@@ -819,7 +830,7 @@ def _set_integration_values(
             if source == path:
                 continue
             try:
-                data = yaml.safe_load(source.read_text(encoding="utf-8")) or {}
+                data = _read_yaml_file(source) or {}
             except FileNotFoundError:
                 continue
             for setting_path in changes:

planfile/core/sqlite_index.py

@@ -106,6 +106,32 @@ def is_current(self, signature: tuple) -> bool:
         except sqlite3.DatabaseError:
             return False
 
+    @staticmethod
+    def _insert_rebuild_batch(
+        connection: sqlite3.Connection,
+        batch: list[dict[str, Any]],
+    ) -> None:
+        connection.executemany(
+            """
+            INSERT INTO tickets(
+                id, sprint, status, priority, source, queue,
+                created_at, updated_at, position, ticket_json, summary_json
+            ) VALUES(
+                :id, :sprint, :status, :priority, :source, :queue,
+                :created_at, :updated_at, :position, :ticket_json, :summary_json
+            )
+            """,
+            batch,
+        )
+        connection.executemany(
+            "INSERT INTO dependencies(ticket_id, blocked_by) VALUES(?, ?)",
+            (
+                (row["id"], dependency)
+                for row in batch
+                for dependency in row.get("blocked_by", [])
+            ),
+        )
+
     def rebuild(
         self,
         records: Iterable[dict[str, Any]],
@@ -132,26 +158,7 @@ def rebuild(
                 batch = list(islice(rows, batch_size))
                 if not batch:
                     break
-                connection.executemany(
-                    """
-                    INSERT INTO tickets(
-                        id, sprint, status, priority, source, queue,
-                        created_at, updated_at, position, ticket_json, summary_json
-                    ) VALUES(
-                        :id, :sprint, :status, :priority, :source, :queue,
-                        :created_at, :updated_at, :position, :ticket_json, :summary_json
-                    )
-                    """,
-                    batch,
-                )
-                connection.executemany(
-                    "INSERT INTO dependencies(ticket_id, blocked_by) VALUES(?, ?)",
-                    (
-                        (row["id"], dependency)
-                        for row in batch
-                        for dependency in row.get("blocked_by", [])
-                    ),
-                )
+                self._insert_rebuild_batch(connection, batch)
                 count += len(batch)
                 # Release record graphs before constructing the next batch.
                 batch.clear()
@@ -197,22 +204,37 @@ def apply(
                     "DELETE FROM tickets WHERE id=?",
                     [(ticket_id,) for ticket_id in deleted_ids],
                 )
-            next_position = int(
-                connection.execute(
-                    "SELECT COALESCE(MAX(position), -1) + 1 AS value FROM tickets"
-                ).fetchone()["value"]
-            )
-            for row in rows:
-                existing = connection.execute(
-                    "SELECT position FROM tickets WHERE id=?",
-                    (row["id"],),
-                ).fetchone()
-                position = existing["position"] if existing is not None else next_position
-                if existing is None:
-                    next_position += 1
-                values = dict(row)
-                values["position"] = position
-                connection.execute(
+            if rows:
+                next_position = int(
+                    connection.execute(
+                        "SELECT COALESCE(MAX(position), -1) + 1 AS value FROM tickets"
+                    ).fetchone()["value"]
+                )
+                position_rows = connection.execute(
+                    "SELECT id, position FROM tickets"
+                ).fetchall()
+                existing_positions = {
+                    row["id"]: int(row["position"])
+                    for row in position_rows
+                }
+                upsert_values: list[dict[str, Any]] = []
+                dep_delete_ids: list[tuple[str]] = []
+                dep_insert_rows: list[tuple[str, Any]] = []
+                for row in rows:
+                    ticket_id = row["id"]
+                    if ticket_id in existing_positions:
+                        position = existing_positions[ticket_id]
+                    else:
+                        position = next_position
+                        existing_positions[ticket_id] = position
+                        next_position += 1
+                    values = dict(row)
+                    values["position"] = position
+                    upsert_values.append(values)
+                    dep_delete_ids.append((ticket_id,))
+                    for dependency in row.get("blocked_by", []):
+                        dep_insert_rows.append((ticket_id, dependency))
+                connection.executemany(
                     """
                     INSERT INTO tickets(
                         id, sprint, status, priority, source, queue,
@@ -232,16 +254,17 @@ def apply(
                         ticket_json=excluded.ticket_json,
                         summary_json=excluded.summary_json
                     """,
-                    values,
-                )
-                connection.execute(
-                    "DELETE FROM dependencies WHERE ticket_id=?",
-                    (row["id"],),
+                    upsert_values,
                 )
                 connection.executemany(
-                    "INSERT INTO dependencies(ticket_id, blocked_by) VALUES(?, ?)",
-                    [(row["id"], dependency) for dependency in row.get("blocked_by", [])],
+                    "DELETE FROM dependencies WHERE ticket_id=?",
+                    dep_delete_ids,
                 )
+                if dep_insert_rows:
+                    connection.executemany(
+                        "INSERT INTO dependencies(ticket_id, blocked_by) VALUES(?, ?)",
+                        dep_insert_rows,
+                    )
             connection.execute(
                 """
                 INSERT INTO meta(key, value) VALUES('source_signature', ?)

planfile/core/store.py

@@ -10,6 +10,7 @@
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 from enum import Enum
+from functools import lru_cache
 from pathlib import Path
 from threading import RLock
 
@@ -24,6 +25,20 @@
 from .store_tickets import TicketStoreMixin
 
 
+@lru_cache(maxsize=256)
+def _load_yaml_cached(path_str: str, mtime_ns: int) -> object:
+    return yaml.safe_load(Path(path_str).read_text(encoding="utf-8"))
+
+
+def _read_yaml_file(path: Path) -> object:
+    stat = path.stat()
+    return _load_yaml_file_cached(path, stat.st_mtime_ns)
+
+
+def _load_yaml_file_cached(path: Path, mtime_ns: int) -> object:
+    return _load_yaml_cached(str(path.resolve()), mtime_ns)
+
+
 class ImmutableTerminalReopenError(RuntimeError):
     """Raised when an ordinary mutation tries to reactivate done/canceled work."""
 
@@ -456,7 +471,7 @@ def _cached_sync_mappings(self) -> dict[str, dict]:
             backend = file_name.removesuffix(".state.yaml")
             file_path = sync_dir / file_name
             try:
-                data = yaml.safe_load(file_path.read_text(encoding="utf-8")) or {}
+                data = _load_yaml_file_cached(file_path, item[1]) or {}
             except Exception:
                 continue
             if isinstance(data, dict):
@@ -1051,7 +1066,7 @@ def _known_ticket_highwater(self, prefix: str) -> int:
                 for entry in os.scandir(sync_dir):
                     if entry.name.endswith(".state.yaml") and entry.is_file():
                         try:
-                            data = yaml.safe_load(Path(entry.path).read_text(encoding="utf-8")) or {}
+                            data = _read_yaml_file(Path(entry.path)) or {}
                         except Exception:
                             continue
                         if isinstance(data, dict):

planfile/dsl/executor.py

@@ -559,7 +559,7 @@ def _exec_sequence(self, cmd: DSLCommand) -> DSLResult:
             step_cmd = self._parser.parse(step_cmd_text)
             if dry_run:
                 step_cmd.params["dry_run"] = True
-            step_res = self.execute(step_cmd)
+            step_res = self._run_sequence_step(step_cmd)
             step_results.append(step_res.to_dict())
             if not step_res.ok:
                 return DSLResult(
@@ -577,6 +577,9 @@ def _exec_sequence(self, cmd: DSLCommand) -> DSLResult:
             source_layer="nl_plan_v1",
         )
 
+    def _run_sequence_step(self, step_cmd: DSLCommand) -> DSLResult:
+        return self.execute(step_cmd)
+
     def _exec_create(self, cmd: DSLCommand) -> DSLResult:
         obj = cmd.object_type or "ticket"
         if obj == "ticket":

planfile/importers/code2llm_importer.py

@@ -25,6 +25,8 @@ def import_code2llm(
         return []
     for ticket in tickets:
         ticket["name"] = ticket["title"]
+        if sprint:
+            ticket.setdefault("sprint", sprint)
     return tickets
 
 

planfile/integrations/config.py

@@ -1,14 +1,26 @@
 """Configuration management for integrations with support for multiple config files."""
 
+import copy
 import os
 import re
+from functools import lru_cache
 from pathlib import Path
 from typing import Any
 
 import yaml
 from dotenv import load_dotenv
 
 
+@lru_cache(maxsize=256)
+def _load_yaml_cached(path_str: str, mtime_ns: int) -> Any:
+    return yaml.safe_load(Path(path_str).read_text(encoding="utf-8"))
+
+
+def _read_yaml_file(path: Path) -> Any:
+    stat = path.stat()
+    return copy.deepcopy(_load_yaml_cached(str(path.resolve()), stat.st_mtime_ns))
+
+
 class IntegrationConfig:
     """Manages integration configuration with support for multiple config files."""
 
@@ -67,10 +79,9 @@ def load_configs(self) -> dict[str, Any]:
         self.config = {}
 
         for config_file in self._ordered_configs():
-            with open(config_file) as f:
-                file_config = yaml.safe_load(f) or {}
-                file_config = self._expand_env_vars(file_config)
-                self._deep_merge(self.config, file_config)
+            file_config = _read_yaml_file(config_file) or {}
+            file_config = self._expand_env_vars(file_config)
+            self._deep_merge(self.config, file_config)
 
         return self.config
 

planfile/project_paths.py

@@ -3,13 +3,24 @@
 from __future__ import annotations
 
 import subprocess
+from functools import lru_cache
 from pathlib import Path
 
 import yaml
 
 PLANFILE_DIRNAME = ".planfile"
 
 
+@lru_cache(maxsize=256)
+def _load_yaml_cached(path_str: str, mtime_ns: int) -> object:
+    return yaml.safe_load(Path(path_str).read_text(encoding="utf-8"))
+
+
+def _read_yaml_file(path: Path) -> object:
+    stat = path.stat()
+    return _load_yaml_cached(str(path.resolve()), stat.st_mtime_ns)
+
+
 class RepositoryRoutingError(RuntimeError):
     """Raised when discovery would cross or misidentify a Git repository."""
 
@@ -48,7 +59,7 @@ def _configured_github_repository(project_path: Path) -> str | None:
         project_path / ".planfile" / "integrations.oql.planfile.yaml",
     ):
         try:
-            data = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
+            data = _read_yaml_file(config_path) or {}
         except (OSError, yaml.YAMLError):
             continue
         if not isinstance(data, dict):