Avoid repeated history materialization in study execution

research/experiments/STUDY_CONTEXT.md

@@ -132,12 +132,16 @@ against the original scientific validator after optional null placeholders are
 restored. A later executor error preserves the completed model response and
 its physical inference timing.
 
-A `context_page_read` ledger event records the exact locally constructed page
-and its source cursor before the parent episode checkpoint. It does not prove
-physical model delivery. The request record and actual CLI trace identify the
-page or reference representation sent; the verified transport outcome records
-whether that request completed. Native AIDE prompt selection must preserve its
-ordinary saved Journal, action receipts and RNG state.
+A `context_page_read` ledger event records the source segment and exact cursor
+before the parent episode checkpoint. The complete delivered page remains in
+the canonical action history; the event points to it instead of copying its
+contents. This follows the same checkpoint-plus-reference pattern used by
+LangGraph's persistence layer and avoids multiplying large evidence pages in
+append-only audit logs. The event does not prove physical model delivery. The
+request record and actual CLI trace identify the page or reference
+representation sent; the verified transport outcome records whether that
+request completed. Native AIDE prompt selection must preserve its ordinary
+saved Journal, action receipts and RNG state.
 
 ## Regression coverage
 

research/experiments/aide_runtime.py

@@ -855,10 +855,14 @@ def _restore_node(record, journal, upstream):
 
 
 def _save(runtime, work_id, state, *, event_type=None, event_payload=None):
+    from .study_runtime import _compact_action_history_frontiers
+    stored_state = dict(state)
+    if isinstance(stored_state.get("action_history"), list):
+        stored_state["action_history"] = _compact_action_history_frontiers(stored_state["action_history"])
     with runtime.ledger.connect() as db:
         db.execute("BEGIN IMMEDIATE")
         db.execute("INSERT INTO forest_study_aide_sessions VALUES(?,?,?,?) ON CONFLICT(campaign_id,work_id) DO UPDATE SET state_json=excluded.state_json,updated_at=excluded.updated_at",
-                   (runtime.campaign_id, work_id, _json(state), time.time()))
+                   (runtime.campaign_id, work_id, _json(stored_state), time.time()))
         if event_type is not None:
             db.execute("INSERT INTO forest_study_events VALUES(?,?,?,?,?,?)",
                 (str(uuid.uuid4()), runtime.campaign_id, work_id, event_type,
@@ -923,6 +927,18 @@ def run_aide_episode(runtime, episode_item: Mapping[str, Any], *, client,
         row = db.execute("SELECT state_json FROM forest_study_aide_sessions WHERE campaign_id=? AND work_id=?",
                          (runtime.campaign_id, work_id)).fetchone()
     state = json.loads(row[0]) if row else None
+    if state is not None:
+        from .study_runtime import _compact_action_history_frontiers, _expand_action_history_frontiers
+        state["action_history"] = _expand_action_history_frontiers(state.get("action_history", []))
+        compacted = dict(state)
+        compacted["action_history"] = _compact_action_history_frontiers(state["action_history"])
+        compacted_json = _json(compacted)
+        if len(compacted_json) < len(row[0]):
+            # Migrate legacy full snapshots losslessly on recovery. Event
+            # records and the reconstructed in-memory state remain untouched.
+            with runtime.ledger.connect() as db:
+                db.execute("UPDATE forest_study_aide_sessions SET state_json=?,updated_at=? WHERE campaign_id=? AND work_id=?",
+                    (compacted_json, time.time(), runtime.campaign_id, work_id))
     if state is not None and state.get("finished"):
         return dict(state["final_outcome"])
     _restore_pending_task_context(state, payload)

research/experiments/end_to_end_runtime.py

@@ -13,6 +13,8 @@
 from typing import Any, Mapping, Sequence
 import uuid
 
+from .study_runtime import _decode_study_result_json
+
 SCHEMA = """
 CREATE TABLE IF NOT EXISTS forest_study_end_to_end (
  campaign_id TEXT NOT NULL, work_id TEXT NOT NULL, state_json TEXT NOT NULL,
@@ -159,7 +161,7 @@ def _episode_history(runtime, work_id):
                          (runtime.campaign_id, work_id)).fetchone()
         calls = db.execute("SELECT request_id,status FROM forest_study_model_calls WHERE campaign_id=? AND work_id=? ORDER BY recorded_at,request_id",
                            (runtime.campaign_id, work_id)).fetchall()
-    result = json.loads(row["result_json"] or "{}") if row else {}
+    result = _decode_study_result_json(row["result_json"]) if row else {}
     return {"last_actual_decision": json.loads(row["decision_json"]) if row and row["decision_json"] else None,
             "actual_action_history": result.get("action_history", []),
             "model_response_records": [dict(call) for call in calls]}
@@ -267,7 +269,7 @@ def _checked_scores(runtime, state, observations):
         if row is None or row["state"] != "completed":
             pending.append(binding)
             continue
-        result = json.loads(row["result_json"] or "{}")
+        result = _decode_study_result_json(row["result_json"])
         observation = next(value for value in observations if value["receipt_path"] == binding["reference_receipt_path"])
         actual = result.get("metrics") or result.get("validation_metrics") or {}
         if result.get("status") != "SUCCEEDED" or not isinstance(actual, Mapping):
@@ -318,7 +320,7 @@ def build_end_to_end_episode_request(runtime, payload: Mapping[str, Any]) -> dic
     known_rows = _work_rows(runtime, [value["work_id"] for value in state["bindings"].values()])
     select_values = [(binding, known_rows.get(binding["work_id"])) for binding in state["bindings"].values()
                      if binding.get("analysis_operation") == "end_to_end_select"]
-    completed_selects = [(binding, json.loads(row["result_json"] or "{}")) for binding, row in select_values
+    completed_selects = [(binding, _decode_study_result_json(row["result_json"])) for binding, row in select_values
                          if row and row["state"] == "completed"]
     recorded_selection_ids = {value["selection_work_id"] for value in state["comparison_records"]}
     new_selections = [(binding, result) for binding, result in completed_selects
@@ -556,7 +558,7 @@ def verify_end_to_end_submission(runtime, payload, history, submission):
         row = rows.get(binding["work_id"], {})
         if row.get("state") != "completed" or json.loads(row["payload_json"]) != binding:
             continue
-        result = json.loads(row["result_json"] or "{}")
+        result = _decode_study_result_json(row["result_json"])
         differences = result.get("paired_differences_right_minus_left") or []
         if (result.get("status") != "SUCCEEDED" or result.get("operation") != "compare"
                 or result.get("comparison_record") != current or result.get("metric") != binding["metric"]

research/experiments/study_memory.py

@@ -11,6 +11,9 @@
 from research.agents.context_store import ContextStore
 
 
+_LOSSLESS_STORAGE_ENCODING = 'forest.lossless-json-subtrees.v1'
+
+
 class ContextQueryError(ValueError):
     """A caller supplied an invalid query for existing public evidence."""
 
@@ -35,9 +38,26 @@ def save_json(self, key, value, *, audience='arm-public', origin=None):
         if audience != 'arm-public':
             raise ValueError('Study memory accepts only an already filtered arm-public view')
         text = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False)
-        identifier = self.store.put('study-public:' + str(key), text,
+        stored_text = text
+        stored_encoding = None
+        from .study_context import intern_json_with_stats
+        encoded, stats = intern_json_with_stats(json.loads(text))
+        if stats['saved_characters'] > 0:
+            stored_text = json.dumps(encoded, ensure_ascii=False, sort_keys=True,
+                                     separators=(',', ':'), allow_nan=False)
+            stored_encoding = _LOSSLESS_STORAGE_ENCODING
+        identifier = self.store.put('study-public:' + str(key), stored_text,
             origin=origin or str(key), authority='untrusted_evidence')
         metadata = self.store.index['segments'][identifier]
+        # Catalog character counts and page cursors describe the logical source,
+        # while the immutable backing file can use a lossless shared-subtree
+        # representation. Existing plain segments remain readable unchanged.
+        metadata['characters'] = len(text)
+        if stored_encoding:
+            metadata['storage_encoding'] = stored_encoding
+        else:
+            metadata.pop('storage_encoding', None)
+        self.store.save()
         return {**metadata, 'segment_id': identifier, 'audience': audience,
                 'read_operation': 'read_context_segment'}
 
@@ -76,6 +96,11 @@ def read_json(self, segment_id, *, pointer='', offset=0, limit=6000,
             metadata = self.store.index['segments'][segment_id]
             path = self.store.root / (segment_id + '.txt')
             value = json.loads(path.read_text(encoding='utf-8'))
+            if metadata.get('storage_encoding') == _LOSSLESS_STORAGE_ENCODING:
+                from .study_context import restore_json
+                value = restore_json(value)
+            elif metadata.get('storage_encoding') is not None:
+                raise ContextQueryError('Unknown study-memory storage encoding')
         if not isinstance(pointer, str) or pointer and not pointer.startswith('/'):
             raise ContextQueryError('A JSON Pointer must be empty or start with /')
         if pointer: