Spaces:
Paused
Paused
| """Research DATA v2: compact AI packets plus immutable compressed raw runs. | |
| The raw experiment observations remain available for audit, but the default | |
| Research Assistant path is intentionally bounded. `DATA/runs.json` is a compact | |
| projection for the current experiment, while each complete run is archived as an | |
| immutable gzip member under `DATA/raw/` and indexed by SHA-256. | |
| """ | |
| from __future__ import annotations | |
| import gzip | |
| import hashlib | |
| import json | |
| import re | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| from typing import Any, Mapping, Sequence, cast | |
| AI_PACKET_MAX_BYTES = 1_000_000 | |
| RUN_SUMMARY_MAX_BYTES = 5_000_000 | |
| DETAIL_PACKET_MAX_BYTES = 1_000_000 | |
| SEQUENCE_PREVIEW_ITEMS = 16 | |
| class ResearchDataArtifacts: | |
| runs_path: Path | |
| raw_index_path: Path | |
| current_run_path: Path | |
| ai_packet_path: Path | |
| ai_packet_digest_path: Path | |
| raw_paths: tuple[Path, ...] | |
| def _json_bytes(value: object) -> bytes: | |
| return json.dumps( | |
| value, | |
| ensure_ascii=False, | |
| sort_keys=True, | |
| separators=(",", ":"), | |
| default=list, | |
| ).encode("utf-8") | |
| def _sha256_file(path: Path) -> str: | |
| return hashlib.sha256(path.read_bytes()).hexdigest() | |
| def _safe_name(value: object) -> str: | |
| name = re.sub(r"[^A-Za-z0-9_.-]+", "-", str(value)).strip("-") | |
| return name or "run" | |
| def compact_for_storage( | |
| value: object, *, preview_items: int = SEQUENCE_PREVIEW_ITEMS | |
| ) -> object: | |
| """Return a deterministic bounded projection without inventing statistics.""" | |
| if isinstance(value, dict): | |
| mapping = cast(dict[object, object], value) | |
| return { | |
| str(key): compact_for_storage(item, preview_items=preview_items) | |
| for key, item in mapping.items() | |
| } | |
| if isinstance(value, list): | |
| items = cast(list[object], value) | |
| if len(items) <= preview_items * 2: | |
| return [ | |
| compact_for_storage(item, preview_items=preview_items) for item in items | |
| ] | |
| return { | |
| "_projection": "bounded_sequence", | |
| "item_count": len(items), | |
| "head": [ | |
| compact_for_storage(item, preview_items=preview_items) | |
| for item in items[:preview_items] | |
| ], | |
| "tail": [ | |
| compact_for_storage(item, preview_items=preview_items) | |
| for item in items[-preview_items:] | |
| ], | |
| } | |
| if isinstance(value, tuple): | |
| items_tuple = cast(tuple[object, ...], value) | |
| return compact_for_storage(list(items_tuple), preview_items=preview_items) | |
| return value | |
| def _write_json(path: Path, value: object) -> None: | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_text( | |
| json.dumps(value, indent=2, ensure_ascii=False, sort_keys=True, default=list) | |
| + "\n", | |
| encoding="utf-8", | |
| ) | |
| def _write_gzip_json(path: Path, value: object) -> tuple[str, int, int]: | |
| payload = _json_bytes(value) | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| with gzip.open(path, "wb", compresslevel=9) as handle: | |
| handle.write(payload) | |
| return _sha256_file(path), path.stat().st_size, len(payload) | |
| def _statistics_projection(statistics: Mapping[str, Any]) -> dict[str, Any]: | |
| allowed = ( | |
| "schema_version", | |
| "run_count", | |
| "conditions", | |
| "two_condition_effects", | |
| "inter_spike_intervals", | |
| "temporal_horizons", | |
| "formulas", | |
| ) | |
| return {key: statistics[key] for key in allowed if key in statistics} | |
| def prepare_research_data_v2( | |
| experiment_dir: Path, | |
| runs: list[dict[str, Any]], | |
| statistics: Mapping[str, Any], | |
| ) -> ResearchDataArtifacts: | |
| """Archive full runs and replace the mutable list with compact summaries.""" | |
| data_dir = experiment_dir / "DATA" | |
| raw_dir = data_dir / "raw" | |
| analysis_dir = experiment_dir / "analysis" | |
| runs_path = data_dir / "runs.json" | |
| index_path = data_dir / "runs_index.json" | |
| current_path = data_dir / "current_run.json" | |
| ai_packet_path = analysis_dir / "ai_packet.json" | |
| ai_digest_path = analysis_dir / "ai_packet_digest.json" | |
| raw_entries: list[dict[str, Any]] = [] | |
| summaries: list[dict[str, Any]] = [] | |
| raw_paths: list[Path] = [] | |
| for run_index, run in enumerate(tuple(runs)): | |
| condition = _safe_name(run.get("condition", "unknown")) | |
| seed = run.get("seed", "unknown") | |
| raw_path = raw_dir / f"run-{run_index:04d}-{condition}-seed-{seed}.json.gz" | |
| _write_json( | |
| current_path, | |
| { | |
| "status": "archiving", | |
| "run_index": run_index, | |
| "condition": run.get("condition"), | |
| "seed": seed, | |
| }, | |
| ) | |
| digest, size_bytes, uncompressed_bytes = _write_gzip_json(raw_path, run) | |
| raw_paths.append(raw_path) | |
| raw_ref: dict[str, Any] = { | |
| "path": raw_path.relative_to(experiment_dir).as_posix(), | |
| "format": "json.gz", | |
| "sha256": digest, | |
| "size_bytes": size_bytes, | |
| "uncompressed_bytes": uncompressed_bytes, | |
| } | |
| compact = compact_for_storage(run) | |
| if not isinstance(compact, dict): | |
| raise ValueError("Compact run projection must remain an object.") | |
| summary = cast(dict[str, Any], compact) | |
| summary["_raw_artifact"] = raw_ref | |
| summaries.append(summary) | |
| raw_entries.append( | |
| { | |
| "run_index": run_index, | |
| "condition": run.get("condition"), | |
| "seed": seed, | |
| **raw_ref, | |
| } | |
| ) | |
| _write_json( | |
| current_path, | |
| {"status": "idle", "last_archived_run": run_index, "raw_artifact": raw_ref}, | |
| ) | |
| index_payload = { | |
| "schema_version": "2.0", | |
| "storage_policy": "immutable_raw_runs_plus_compact_current_experiment", | |
| "run_count": len(raw_entries), | |
| "runs": raw_entries, | |
| } | |
| _write_json(index_path, index_payload) | |
| compact_bytes = json.dumps( | |
| summaries, indent=2, ensure_ascii=False, sort_keys=True, default=list | |
| ).encode("utf-8") | |
| if len(compact_bytes) > RUN_SUMMARY_MAX_BYTES: | |
| raise ValueError( | |
| f"Compact DATA/runs.json would exceed {RUN_SUMMARY_MAX_BYTES} bytes; " | |
| "increase deterministic aggregation instead of sending it to AI." | |
| ) | |
| runs[:] = summaries | |
| ai_packet = { | |
| "schema_version": "2.0", | |
| "generated_by": "deterministic_research_data_v2", | |
| "ai_input_policy": "compact_only_raw_on_explicit_deterministic_extract", | |
| "run_count": len(summaries), | |
| "statistics": _statistics_projection(statistics), | |
| "run_preview": compact_for_storage(summaries, preview_items=8), | |
| "raw_index": { | |
| "path": index_path.relative_to(experiment_dir).as_posix(), | |
| "sha256": _sha256_file(index_path), | |
| "run_count": len(raw_entries), | |
| }, | |
| "limits": { | |
| "ai_packet_max_bytes": AI_PACKET_MAX_BYTES, | |
| "run_summary_max_bytes": RUN_SUMMARY_MAX_BYTES, | |
| "sequence_preview_items": SEQUENCE_PREVIEW_ITEMS, | |
| }, | |
| } | |
| packet_bytes = json.dumps( | |
| ai_packet, indent=2, ensure_ascii=False, sort_keys=True, default=list | |
| ).encode("utf-8") | |
| if len(packet_bytes) > AI_PACKET_MAX_BYTES: | |
| raise ValueError( | |
| f"AI packet exceeds hard limit of {AI_PACKET_MAX_BYTES} bytes; " | |
| "the deterministic projection must be tightened." | |
| ) | |
| ai_packet_path.parent.mkdir(parents=True, exist_ok=True) | |
| ai_packet_path.write_bytes(packet_bytes + b"\n") | |
| _write_json( | |
| ai_digest_path, | |
| { | |
| "schema_version": "1.0", | |
| "path": ai_packet_path.relative_to(experiment_dir).as_posix(), | |
| "sha256": _sha256_file(ai_packet_path), | |
| "size_bytes": ai_packet_path.stat().st_size, | |
| "raw_index_sha256": _sha256_file(index_path), | |
| }, | |
| ) | |
| return ResearchDataArtifacts( | |
| runs_path=runs_path, | |
| raw_index_path=index_path, | |
| current_run_path=current_path, | |
| ai_packet_path=ai_packet_path, | |
| ai_packet_digest_path=ai_digest_path, | |
| raw_paths=tuple(raw_paths), | |
| ) | |
| def _select_metrics( | |
| run: dict[str, Any], metrics: Sequence[str] | None | |
| ) -> dict[str, Any]: | |
| if not metrics: | |
| return run | |
| selected = dict(run) | |
| raw_metrics = run.get("metrics") | |
| if isinstance(raw_metrics, dict): | |
| selected["metrics"] = { | |
| name: raw_metrics[name] for name in metrics if name in raw_metrics | |
| } | |
| return selected | |
| def build_detail_packet( | |
| experiment_dir: Path, | |
| *, | |
| condition: str, | |
| metrics: Sequence[str] | None = None, | |
| preview_items: int = 64, | |
| ) -> dict[str, Any]: | |
| """Build a bounded condition-specific packet from indexed raw runs only.""" | |
| index_path = experiment_dir / "DATA" / "runs_index.json" | |
| raw_index: object = json.loads(index_path.read_text(encoding="utf-8")) | |
| if not isinstance(raw_index, dict): | |
| raise ValueError("Raw run index must be a JSON object.") | |
| index_object = cast(dict[str, object], raw_index) | |
| entries_value = index_object.get("runs", []) | |
| if not isinstance(entries_value, list): | |
| raise ValueError("Raw run index 'runs' must be a list.") | |
| entries = cast(list[object], entries_value) | |
| detail_runs: list[dict[str, Any]] = [] | |
| source_refs: list[dict[str, Any]] = [] | |
| for entry_value in entries: | |
| if not isinstance(entry_value, dict): | |
| continue | |
| entry = cast(dict[str, object], entry_value) | |
| condition_value = entry.get("condition") | |
| if not isinstance(condition_value, str) or condition_value != condition: | |
| continue | |
| path_value = entry.get("path") | |
| digest_value = entry.get("sha256") | |
| if not isinstance(path_value, str) or not isinstance(digest_value, str): | |
| raise ValueError("Raw run index entry lacks path or SHA-256.") | |
| path = experiment_dir / path_value | |
| if _sha256_file(path) != digest_value: | |
| raise ValueError(f"Raw run digest mismatch: {path}") | |
| with gzip.open(path, "rt", encoding="utf-8") as handle: | |
| raw_value: object = json.load(handle) | |
| if not isinstance(raw_value, dict): | |
| continue | |
| raw = cast(dict[str, Any], raw_value) | |
| compact = compact_for_storage( | |
| _select_metrics(raw, metrics), preview_items=preview_items | |
| ) | |
| if not isinstance(compact, dict): | |
| raise ValueError("Detail run projection must remain an object.") | |
| detail_runs.append(cast(dict[str, Any], compact)) | |
| source_refs.append( | |
| {"path": path_value, "sha256": digest_value, "seed": entry.get("seed")} | |
| ) | |
| packet: dict[str, Any] = { | |
| "schema_version": "1.0", | |
| "generated_by": "deterministic_detail_extractor", | |
| "condition": condition, | |
| "metrics": list(metrics or ()), | |
| "run_count": len(detail_runs), | |
| "runs": detail_runs, | |
| "sources": source_refs, | |
| } | |
| encoded = json.dumps(packet, ensure_ascii=False, sort_keys=True).encode("utf-8") | |
| if len(encoded) > DETAIL_PACKET_MAX_BYTES: | |
| raise ValueError( | |
| f"Detail packet exceeds {DETAIL_PACKET_MAX_BYTES} bytes; request fewer metrics or a smaller preview." | |
| ) | |
| return packet | |
| def write_detail_packet( | |
| experiment_dir: Path, | |
| *, | |
| condition: str, | |
| metrics: Sequence[str] | None = None, | |
| preview_items: int = 64, | |
| ) -> Path: | |
| packet = build_detail_packet( | |
| experiment_dir, | |
| condition=condition, | |
| metrics=metrics, | |
| preview_items=preview_items, | |
| ) | |
| path = experiment_dir / "analysis" / "detail" / f"{_safe_name(condition)}.json" | |
| _write_json(path, packet) | |
| return path | |