Consult card keeps parsed packets in memory, prewarms each turn, writes 0600 (issue #129, BUG-1476)

- consult_packet_view keeps the last 3 parsed packets with their shared domain
  base (~30 MB each). A hit reads no file and copies nothing: attach for one
  domain 0.106 s -> 0.002 s. Memory holds the file's JSON round trip, i.e.
  exactly what a hit returned before; consult_card_facts sha256 equal old/new.
- Cache files are created 0600, same bytes as before.
- The consult route fires a warm for the turn's chart before reserving credit,
  so the day's first question does not build the packet on the answer path.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

scripts/consult_full_data_cache.py

@@ -2,16 +2,19 @@
 
 The key is chart identity + ayanamsa + node mode + engine version
 + the reference date used for current dasha, transits and the year chart.
-The file name is a hash. Birth data stays inside the scratch file.
-Files older than two days are removed.
+The file name is a hash. Birth data stays inside the scratch file, mode 0600.
+Files older than two days are removed. The consult card also keeps the last
+few parsed packets in memory with their shared domain base (issue #129).
 """
 
 from __future__ import annotations
 
 import hashlib
 import json
+import os
 import threading
 import time
+from collections import OrderedDict
 from datetime import date
 from pathlib import Path
 from types import SimpleNamespace
@@ -24,6 +27,13 @@ KEEP_SECONDS = 2 * 24 * 60 * 60
 
 PacketBuilder = Callable[[dict[str, Any]], dict[str, Any]]
 
+# Parsed consult packets kept in this process, newest last. One entry is about
+# 18 MB parsed plus its shared base copy (~30 MB together), so three stay well
+# inside the 4 GB host. A hit skips the 5.5 MB file read and the base copy.
+MEMORY_SLOTS = 3
+_memory: "OrderedDict[str, tuple[dict[str, Any], dict[str, Any]]]" = OrderedDict()
+_memory_guard = threading.Lock()
+
 # One build per cache key. A later caller waits and reads the stored packet.
 class _Flight:
     def __init__(self) -> None:
@@ -137,6 +147,17 @@ def prune_cache(now: float | None = None, *, name: str = "consult_full_data_cach
     return removed
 
 
+def _write_private_text(path: Path, text: str) -> None:
+    """Birth data: the file is created 0600, never readable by other users.
+
+    Same bytes as before (no key sorting), so a packet read back is unchanged.
+    """
+    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
+        handle.write(text)
+    os.chmod(path, 0o600)
+
+
 def _read_cached(path: Path, started: float, now: Callable[[], float]) -> dict[str, Any]:
     packet = json.loads(path.read_text(encoding="utf-8"))
     return {"packet": packet, "hit": True, "elapsed_s": now() - started, "slow": False}
@@ -186,7 +207,7 @@ def get_or_build(
         dump_kwargs: dict[str, Any] = {"ensure_ascii": False, "separators": (",", ":")}
         if json_default is not None:
             dump_kwargs["default"] = json_default
-        temporary.write_text(json.dumps(packet, **dump_kwargs), encoding="utf-8")
+        _write_private_text(temporary, json.dumps(packet, **dump_kwargs))
         temporary.replace(path)
         elapsed = now() - started
         return {
@@ -205,6 +226,51 @@ def get_or_build(
                 _flights.pop(key, None)
 
 
+def consult_packet_view(identity: dict[str, Any], builder: PacketBuilder | None = None) -> dict[str, Any]:
+    """The consult card's packet and its shared domain base, both read only.
+
+    Kept in memory for the last ``MEMORY_SLOTS`` charts. Callers must not
+    mutate ``packet`` or ``base``: later turns read the same objects. Other
+    users of ``get_or_build`` (report reference, marriage match) still get
+    their own copy from the file.
+    """
+    try:
+        from pl9_full_data_export import full_data_domain_base
+    except ImportError:  # pragma: no cover - package import path
+        from scripts.pl9_full_data_export import full_data_domain_base
+    key = cache_key(identity)
+    started = time.perf_counter()
+    with _memory_guard:
+        entry = _memory.get(key)
+        if entry is not None:
+            _memory.move_to_end(key)
+    if entry is not None:
+        return {"packet": entry[0], "base": entry[1], "hit": True, "memory": True,
+                "elapsed_s": time.perf_counter() - started, "slow": False}
+    cached = get_or_build(identity, builder)
+    base = full_data_domain_base(cached["packet"])
+    # Remember what a file hit returns: a fresh build is the builder's own
+    # object, and its JSON round trip (lists for tuples, string keys) is what
+    # every later turn used to read. This turn still answers from the build.
+    remembered = (cached["packet"], base)
+    if not cached["hit"]:
+        path = _cache_path(identity)
+        if path.is_file():
+            parsed = json.loads(path.read_text(encoding="utf-8"))
+            remembered = (parsed, full_data_domain_base(parsed))
+    with _memory_guard:
+        _memory[key] = remembered
+        _memory.move_to_end(key)
+        while len(_memory) > MEMORY_SLOTS:
+            _memory.popitem(last=False)
+    return {**cached, "base": base, "memory": False}
+
+
+def clear_memory_cache() -> None:
+    with _memory_guard:
+        _memory.clear()
+
+
 def _attach_graha_drishti(packet: dict[str, Any], reading: dict[str, Any]) -> dict[str, Any]:
     """Copy the engine's whole-sign aspects onto the consult packet. Do not recompute."""
     modules = reading.get("modules") if isinstance(reading, dict) else None
@@ -274,7 +340,7 @@ def warm_consult_packet(body: dict[str, Any] | None) -> dict[str, Any]:
         }
     try:
         try:
-            cached = get_or_build(identity_from_body(dict(body or {})))
+            cached = consult_packet_view(identity_from_body(dict(body or {})))
         except Exception as exc:
             return {
                 "success": False,
@@ -317,12 +383,13 @@ def attach_consult_full_data(result: dict[str, Any], body: dict[str, Any] | None
         from scripts.consult_card_domain_facts import build_consult_card_facts
     try:
         identity = identity_from_body(payload)
-        cached = get_or_build(identity)
+        cached = consult_packet_view(identity)
         facts = build_consult_card_facts(
             cached["packet"],
             _requested_domains(payload),
             minute_confirmed=None,
             body=payload,
+            base=cached["base"],
         )
         facts["timing"] = {
             "hit": bool(cached["hit"]),