"""Executed-code view and hierarchical sources export (root + delegated frames). A session's executed code is spread across per-frame ``execution_log`` rows: the root Notebook cells under the root frame, every delegated child Cell under its own ``kind='delegate'`` frame (see ``openai4s/agent/cell_record.py``), nested children recursively below those. This service is the one read-only place that assembles the hierarchy: * :meth:`ExecutionSourcesService.projection` — a bounded, typed JSON tree for the UI's "Executed code" surface. Frames carry name/parent/depth/status/ order and per-frame counts; cells carry order, language, status, source SHA-256, kernel generation, environment identity, and artifact-version links. Code text is deliberately NOT inlined — the existing per-frame ``GET /frames/{fid}/execution-log`` route serves it. * :meth:`ExecutionSourcesService.export` — ``sources.zip``: the executed source files themselves (``root/`` + ``children/__/`` recursively), per-frame ``session.py``/``session.R`` concatenations with ``# %%`` separators, a NEW ``manifest.json`` (this is not the notebook bundle's manifest, which stays untouched), and a bilingual README pair warning that cells ran in a persistent kernel. The export carries ONLY ``execution_log`` fields plus public metadata: never conversation prompts, host payloads, stdout/stderr, or credentials. Failed and interrupted cells are included and marked — dropping them would make the archive describe a run that went smoothly, which is the one thing a reader must not conclude. Both surfaces are deliberately the RAW execution history: rows the read-only Notebook projection hides (protocol-only ``host.submit_output`` completion cells, non-scientific unpinned cells) are counted, listed, and exported here. Per-frame ``counts`` can therefore exceed the entries the frame's own ``/execution-log`` route renders — that route is the Notebook's curated view; this one is the audit trail. Determinism: fixed ZIP timestamps, sorted entries, canonical JSON, and a ``generated_at`` DERIVED from the newest stored ``created_at`` rather than from the wall clock, so exporting the same durable history twice — including across a daemon restart — yields byte-identical archives. """ from __future__ import annotations import hashlib import io import json import time import zipfile from typing import Any, Mapping from openai4s.server.urls import artifact_version_url from openai4s.storage.branch_projection import project_branch_records #: Hard bounds so one projection cannot become an unbounded walk: delegation #: fan-out is capped at 48 with MAX_DEPTH 4, so real trees are far smaller. _MAX_FRAMES = 200 _MAX_CELLS_PER_FRAME = 2000 _MAX_DEPTH = 12 # Source is duplicated into an individual Cell file and a per-language # session file before ZIP compression. Bound the uncompressed input, not the # compressed response: highly compressible source is still expensive while # those copies are being assembled in the shared daemon. _MAX_EXPORT_SOURCE_BYTES = 16 * 1024 * 1024 _MAX_EXPORT_CELL_BYTES = 4 * 1024 * 1024 class ExecutionSourcesExportTooLarge(ValueError): """The requested source archive exceeds its host-memory safety budget.""" _README_EN = """# Executed code sources This archive is the execution history of one OpenAI4S session: the exact source code every recorded Cell ran, for the root session (`root/`) and every delegated child frame (`children/__/`, nested recursively), in true execution order. Failed and interrupted cells are included and marked by status in the file name (e.g. `cell_0002_error.py`). * Cells ran sequentially in a **persistent kernel**: earlier cells created the variables, imports, and files that later cells rely on. A single cell file is therefore **not guaranteed to run standalone**. Each frame's `session.py` / `session.R` concatenates its cells in order with `# %%` separators, which is the closest runnable rendering of the history. * `manifest.json` describes every frame and cell: ids, order, language, status, source SHA-256, kernel generation, environment identity, and artifact-version links. * The archive contains only executed source code and public execution metadata — no conversation prompts, no host payloads, no cell output, and no credentials. """ _README_ZH = """# 已执行代码源文件 本压缩包是一个 OpenAI4S 会话的执行历史：根会话（`root/`）以及每个被委派的 子 frame（`children/<序号>_<名称>_/`，递归嵌套）中每个已记录 Cell 实际执行的源代码，按真实执行顺序排列。失败与被中断的 Cell 一并包含，并在 文件名中标注状态（例如 `cell_0002_error.py`）。 * 这些 Cell 是在**持久内核**中顺序执行的：前面的 Cell 创建了后面 Cell 所 依赖的变量、导入与文件，因此**单个 Cell 文件不保证可以独立运行**。每个 frame 的 `session.py` / `session.R` 用 `# %%` 分隔符按顺序拼接了该 frame 的全部 Cell，是最接近可运行形态的历史呈现。 * `manifest.json` 描述了每个 frame 与 Cell：id、顺序、语言、状态、源代码 SHA-256、内核 generation、环境标识以及 Artifact 版本关联。 * 压缩包只包含已执行的源代码与公开的执行元数据——不含对话提示词、host 载荷、Cell 输出或任何凭据。 """ def output_artifact_bindings( store: Any, frame_id: str, *, cell_ids: set[str] | None = None, producer_frame_id: str | None = None, producer_frames: Mapping[str, str] | None = None, newest_only: bool = False, ) -> dict[str, list[dict[str, str]]]: """Read capture relationships once, bounded by the owning session and Cells. A delegated frame keeps its own execution log but its captures belong to the ancestor session. Exact Cell identities, never names or Artifact heads, join the relationships to the already branch-projected Notebook rows. Compatibility stores without artifact readers have no confirmed bindings; failures of readers that do exist propagate rather than claiming absence. ``newest_only`` keeps one row per ``(cell, filename)``: the version the store recorded last (``created_at``, then version ordinal). A cell that calls ``host.save_artifact`` and then keeps writing the same file owns two versions of it, and the file it left behind is the newer one. """ artifact_reader = getattr(store, "list_artifacts", None) version_reader = getattr(store, "list_versions", None) if ( not callable(artifact_reader) or not callable(version_reader) or cell_ids == set() or producer_frames == {} ): return {} frame_reader = getattr(store, "get_frame", None) frame = frame_reader(frame_id) if callable(frame_reader) else None owning_root = str((frame or {}).get("root_frame_id") or frame_id) observation_reader = getattr(store, "list_artifact_capture_observations", None) bindings: dict[str, dict[tuple[str, str, str], dict[str, str]]] = {} ranks: dict[str, dict[tuple[str, str, str], tuple[int, int]]] = {} for artifact in artifact_reader({"root_frame_id": owning_root}): artifact_id = artifact.get("artifact_id") if not artifact_id or artifact.get("root_frame_id", owning_root) != owning_root: continue versions = version_reader(artifact_id) valid_versions = {row.get("version_id") for row in versions} # Observations point at one of these versions, so a single recency # map (from the version rows, which carry both fields) orders both. recency = { row.get("version_id"): ( _int_or_zero(row.get("created_at")), _int_or_zero(row.get("ordinal")), ) for row in versions } observations = ( observation_reader(artifact_id=artifact_id) if callable(observation_reader) else [] ) for row in [*versions, *observations]: cell_id = row.get("producing_cell_id") version_id = row.get("version_id") filename = row.get("filename") if ( not all( isinstance(value, str) and value for value in (cell_id, version_id, filename) ) or version_id not in valid_versions or (cell_ids is not None and cell_id not in cell_ids) or ( producer_frames is not None and ( cell_id not in producer_frames or row.get("frame_id") not in (None, producer_frames[cell_id]) ) ) or ( producer_frame_id is not None and row.get("frame_id") not in (None, producer_frame_id) ) ): continue try: url = artifact_version_url(version_id) except ValueError: continue # An unrepresentable imported ID is not a latest link. key = (filename, str(artifact_id), version_id) bindings.setdefault(cell_id, {})[key] = { "filename": filename, "artifact_id": str(artifact_id), "version_id": version_id, "url": url, } ranks.setdefault(cell_id, {})[key] = recency.get(version_id, (0, 0)) if newest_only: for cell_id, rows in bindings.items(): newest: dict[str, tuple[str, str, str]] = {} for key in sorted(rows): current = newest.get(key[0]) # Deterministic: recency first, then the sorted key itself. if current is None or ranks[cell_id][key] >= ranks[cell_id][current]: newest[key[0]] = key keep = set(newest.values()) bindings[cell_id] = {key: row for key, row in rows.items() if key in keep} return { cell_id: [rows[key] for key in sorted(rows)] for cell_id, rows in bindings.items() } def _int_or_zero(value: Any) -> int: return value if isinstance(value, int) and not isinstance(value, bool) else 0 class ExecutionSourcesService: """Read-only projection/export of the per-frame executed-code hierarchy.""" def __init__(self, store: Any) -> None: self.store = store # -- JSON projection ------------------------------------------------- def projection( self, root_frame_id: str, *, branch_id: str | None = None ) -> dict[str, Any]: frames, truncated = self._collect(root_frame_id, branch_id) return { "root_frame_id": root_frame_id, "truncated": truncated, "frames": [ { "frame_id": frame["frame_id"], "parent_id": frame["parent_id"], "root_frame_id": frame["root_frame_id"], "name": frame["name"], "kind": frame["kind"], "depth": frame["depth"], "status": frame["status"], "order": frame["order"], "counts": frame["counts"], "cells": [ { "id": cell["id"], "seq": cell["order"], "language": cell["language"], "status": cell["status"], "source_sha256": cell["source_sha256"], "generation_id": cell["generation_id"], "environment": cell["environment"], "artifacts": cell["artifacts"], "interrupted": cell["interrupted"], } for cell in frame["cells"] ], } for frame in frames ], } # -- sources.zip ----------------------------------------------------- def export( self, root_frame_id: str, *, branch_id: str | None = None ) -> dict[str, Any]: """Return immutable ZIP bytes plus the HTTP descriptor Gateway needs.""" frames, truncated = self._collect( root_frame_id, branch_id, max_source_bytes=_MAX_EXPORT_SOURCE_BYTES, ) documents: dict[str, bytes] = { "README.md": _README_EN.encode("utf-8"), "README_zh.md": _README_ZH.encode("utf-8"), } manifest_frames: list[dict[str, Any]] = [] manifest_cells: list[dict[str, Any]] = [] newest = 0 for frame in frames: directory = frame["path"] manifest_frames.append( { "frame_id": frame["frame_id"], "parent_id": frame["parent_id"], "root_frame_id": frame["root_frame_id"], "name": frame["name"], "kind": frame["kind"], "depth": frame["depth"], "status": frame["status"], "order": frame["order"], "path": directory, } ) newest = max(newest, frame["created_at"] or 0) sessions: dict[str, list[str]] = {} for cell in frame["cells"]: newest = max(newest, cell["created_at"] or 0) extension = "R" if cell["language"] == "r" else "py" file_name = f"cell_{cell['order']:04d}_{cell['status']}.{extension}" path = f"{directory}/{file_name}" documents[path] = cell["source_bytes"] manifest_cells.append( { "id": cell["id"], "frame_id": frame["frame_id"], "order": cell["order"], "language": cell["language"], "status": cell["status"], "source_sha256": cell["source_sha256"], "generation_id": cell["generation_id"], "environment": cell["environment"], "artifacts": cell["artifacts"], "interrupted": cell["interrupted"], "path": path, } ) separator = ( f"# %% cell {cell['order']} — {cell['language']} — " f"{cell['status']}\n" ) sessions.setdefault(extension, []).append( separator + cell["source_bytes"].decode("utf-8") ) for extension, blocks in sessions.items(): label = frame["name"] or frame["frame_id"] header = ( f"# Executed code — frame {frame['frame_id']} ({label})\n" "# Cells ran in order in a persistent kernel; single cells\n" "# are not guaranteed to run standalone.\n\n" ) documents[f"{directory}/session.{extension}"] = ( header + "\n".join(blocks) ).encode("utf-8") manifest = { "version": 1, "root_frame_id": root_frame_id, # Derived from the newest stored created_at (never the wall # clock) so identical durable history exports byte-identically. "generated_at": self._stored_timestamp(newest), "truncated": truncated, "frames": manifest_frames, "cells": manifest_cells, } documents["manifest.json"] = ( json.dumps(manifest, ensure_ascii=False, sort_keys=True, indent=1) + "\n" ).encode("utf-8") output = io.BytesIO() with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as archive: for name, data in sorted(documents.items()): info = zipfile.ZipInfo(name) # A fixed timestamp makes equal execution histories byte-stable. info.date_time = (1980, 1, 1, 0, 0, 0) info.compress_type = zipfile.ZIP_DEFLATED info.external_attr = 0o600 << 16 archive.writestr(info, data) data = output.getvalue() return { "filename": f"{self._safe_stem(root_frame_id)}.sources.zip", "content_type": "application/zip", "data": data, "size_bytes": len(data), "sha256": hashlib.sha256(data).hexdigest(), "immutable": True, } # -- collection ------------------------------------------------------ def _collect( self, root_frame_id: str, branch_id: str | None, *, max_source_bytes: int | None = None, ) -> tuple[list[dict[str, Any]], bool]: root = self.store.get_frame(root_frame_id) if root is None: raise KeyError(f"unknown frame {root_frame_id!r}") generation_cache: dict[tuple[str, str], tuple[bool, dict[str, Any] | None]] = {} source_budget = { "used": 0, "maximum": max_source_bytes, } frames: list[dict[str, Any]] = [] truncated = False def visit( frame: Mapping[str, Any], *, parent_id: str | None, depth: int, order: int, directory: str, ) -> None: nonlocal truncated if len(frames) >= _MAX_FRAMES or depth > _MAX_DEPTH: truncated = True return frame_id = str(frame.get("frame_id")) cells, cells_truncated = self._frame_cells( frame_id, root_frame_id=root_frame_id, branch_id=branch_id, is_root=parent_id is None, generation_cache=generation_cache, source_budget=source_budget, ) truncated = truncated or cells_truncated counts = { "cells": len(cells), "ok": sum(1 for cell in cells if cell["status"] == "ok"), "error": sum(1 for cell in cells if cell["status"] == "error"), "interrupted": sum( 1 for cell in cells if cell["status"] == "interrupted" ), } frames.append( { "frame_id": frame_id, "parent_id": parent_id, "root_frame_id": str(frame.get("root_frame_id") or frame_id), "name": frame.get("name"), "kind": str(frame.get("kind") or "turn"), "depth": depth, "status": frame.get("status"), "order": order, "created_at": self._int(frame.get("created_at")), "path": directory, "counts": counts, "cells": cells, } ) for child_order, child in enumerate( self._delegate_children(frame_id), start=1 ): child_id = str(child.get("frame_id")) child_dir = ( f"{directory}/children/" f"{child_order}_{self._safe_component(child.get('name'))}_" f"{self._safe_stem(child_id)[:8]}" ) # Child rows come back name/kind/status/depth only; re-read # the full row so created_at and root scope are real. full = self.store.get_frame(child_id) or dict(child) visit( full, parent_id=frame_id, depth=depth + 1, order=child_order, directory=child_dir, ) visit(root, parent_id=None, depth=0, order=0, directory="root") # Join only the already-visible tree, including each delegate's own # execution identity. A display-only legacy fallback ID is not evidence. producer_frames = { cell["_producer_id"]: frame["frame_id"] for frame in frames for cell in frame["cells"] if cell["_producer_id"] } artifact_links = self._artifact_links(root_frame_id, producer_frames) for frame in frames: for cell in frame["cells"]: cell["artifacts"] = artifact_links.get(cell["_producer_id"], []) # The archive nests children under root/'s sibling namespace, not # inside it: strip the leading "root/" from child paths. for frame in frames: if frame["path"].startswith("root/children/"): frame["path"] = frame["path"][len("root/") :] return frames, truncated def _delegate_children(self, frame_id: str) -> list[dict[str, Any]]: # `_collect` has already resolved this frame. A backend failure while # enumerating its children is not evidence that it has no children; # propagating the error keeps the projection from looking complete # when part of the execution tree could not be read. detail = self.store.frame_detail(frame_id, page=0, page_size=1) children = (detail or {}).get("children") or [] return [ child for child in children if isinstance(child, Mapping) and child.get("kind") == "delegate" ] def _frame_cells( self, frame_id: str, *, root_frame_id: str, branch_id: str | None, is_root: bool, generation_cache: dict[tuple[str, str], tuple[bool, dict[str, Any] | None]], source_budget: dict[str, int | None], ) -> tuple[list[dict[str, Any]], bool]: if is_root: rows = self._branch_cells(frame_id, branch_id or frame_id) else: rows = self.store.list_cells(frame_id) truncated = len(rows) > _MAX_CELLS_PER_FRAME cells: list[dict[str, Any]] = [] for order, row in enumerate(rows[:_MAX_CELLS_PER_FRAME], start=1): code = str(row.get("code") or row.get("source") or "") source_bytes = ( code.encode("utf-8") if code.endswith("\n") else (code + "\n").encode("utf-8") ) maximum = source_budget["maximum"] if maximum is not None: size = len(source_bytes) used = int(source_budget["used"] or 0) if size > _MAX_EXPORT_CELL_BYTES or used + size > maximum: raise ExecutionSourcesExportTooLarge( "executed source archive exceeds the export byte limit" ) source_budget["used"] = used + size language = str(row.get("language") or "python").lower() cell_id = str(row.get("producing_cell_id") or f"cell-{order}") generation_id = row.get("generation_id") generation_owned, environment = self._environment( generation_id, frame_id=frame_id, cache=generation_cache, ) cells.append( { "id": cell_id, "_producer_id": row.get("producing_cell_id"), "order": order, "language": "r" if language == "r" else "python", "status": self._status(row), "source_bytes": source_bytes, "source_sha256": hashlib.sha256(source_bytes).hexdigest(), "generation_id": ( str(generation_id) if generation_id and generation_owned else None ), "environment": environment, "artifacts": [], "interrupted": bool(row.get("interrupted")), "created_at": self._int(row.get("created_at")), } ) return cells, truncated def _branch_cells(self, root_frame_id: str, branch_id: str) -> list[dict]: """The root frame follows its active branch, like every other export.""" def local(selected: str) -> list[dict]: try: return self.store.list_cells(root_frame_id, branch_id=selected) except TypeError as error: if selected != root_frame_id or "branch_id" not in str(error): raise return self.store.list_cells(root_frame_id) return project_branch_records( self.store, root_frame_id, branch_id, list_local=local, record_position=lambda cell: int( cell.get("state_revision") or cell.get("cell_index") or 0 ), cursor_key="cell_cursor", ) def _artifact_links( self, root_frame_id: str, producer_frames: Mapping[str, str] ) -> dict[str, list[str]]: """producing_cell_id -> artifact version ids, versions + observations. ``artifact_versions.producing_cell_id`` is stamped by the capture paths; ``artifact_capture_observations`` records a Cell that re-observed an existing version's bytes without minting a new one. Both are producer attributions the store made itself — never claims. """ return { cell_id: sorted({row["version_id"] for row in rows}) for cell_id, rows in output_artifact_bindings( self.store, root_frame_id, producer_frames=producer_frames ).items() } def _environment( self, generation_id: Any, *, frame_id: str, cache: dict[tuple[str, str], tuple[bool, dict[str, Any] | None]], ) -> tuple[bool, dict[str, Any] | None]: """Return ownership plus environment identity for this exact frame.""" if not generation_id: return False, None key = (str(generation_id), frame_id) if key not in cache: resolved: dict[str, Any] | None = None try: row = self.store.get_kernel_generation(key[0]) except Exception: # noqa: BLE001 - provenance stays optional row = None owned = bool( isinstance(row, Mapping) and str(row.get("root_frame_id") or "") == frame_id ) if owned and isinstance(row, Mapping): environment = row.get("environment") if isinstance(environment, Mapping): resolved = { "name": environment.get("environment_name"), "interpreter": environment.get("interpreter"), } cache[key] = (owned, resolved) return cache[key] @staticmethod def _status(row: Mapping[str, Any]) -> str: status = str(row.get("status") or "").lower() if status in {"ok", "error", "interrupted"}: return status if row.get("interrupted"): return "interrupted" return "error" if row.get("error") else "ok" @staticmethod def _int(value: Any) -> int: try: return int(value) except (TypeError, ValueError): return 0 @staticmethod def _stored_timestamp(newest_ms: int) -> str | None: if newest_ms <= 0: return None base = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(newest_ms / 1000)) return f"{base}.{newest_ms % 1000:03d}Z" @staticmethod def _safe_component(value: Any) -> str: text = str(value or "").strip() safe = "".join( character if (character.isalnum() or character in "-_") else "-" for character in text ).strip("-") while "--" in safe: safe = safe.replace("--", "-") return (safe or "child")[:40] @staticmethod def _safe_stem(value: str) -> str: stem = "".join( character for character in str(value or "") if character.isalnum() or character in "-_" ) if not stem: raise ValueError("frame id must contain a safe filename character") return stem[:120] __all__ = ["ExecutionSourcesExportTooLarge", "ExecutionSourcesService"]