openai4s/agent/compaction.py

• CompactionSummaryError · class · L95-L96 — class CompactionSummaryError(RuntimeError)
• CompactionCancelled · class · L99-L100 — class CompactionCancelled(RuntimeError)
• ContextEstimate · class · L104-L136 — class ContextEstimate
• total · method · L123-L133 — def total(self) -> int
• as_dict · method · L135-L136 — def as_dict(self) -> dict[str, int]
• ContextSegment · class · L140-L145 — class ContextSegment
• CompactionArchiveMetadata · class · L149-L219 — class CompactionArchiveMetadata
• from_mapping · method · L166-L212 — def from_mapping( cls, value: Mapping[str, Any] | "CompactionArchiveMetadata" | None ) -> "CompactionArchiveMetadata"
• namespace_continuity_known · method · L215-L216 — def namespace_continuity_known(self) -> bool
• as_dict · method · L218-L219 — def as_dict(self) -> dict[str, Any]
• _first_present · function · L222-L226 — def _first_present(source: Mapping[str, Any], *keys: str) -> Any
• _json_safe · function · L229-L237 — def _json_safe(value: Any) -> Any
• _json_text · function · L240-L243 — def _json_text(value: Any) -> str
• _chars_to_tokens · function · L246-L259 — def _chars_to_tokens(value: str) -> int
• _content_estimate · function · L262-L298 — def _content_estimate(content: Any) -> tuple[int, int]
• _artifact_reference_tokens · function · L301-L318 — def _artifact_reference_tokens(message: Mapping[str, Any]) -> int
• estimate_context · function · L321-L357 — def estimate_context( messages: Iterable[Mapping[str, Any]], tool_schemas: Iterable[Mapping[str, Any]] = (), ) -> ContextEstimate
• estimate_tokens · function · L360-L362 — def estimate_tokens(messages: list[dict]) -> int
• should_compact · function · L365-L375 — def should_compact( messages: list[dict], cfg: Config, *, tool_schemas: Iterable[Mapping[str, Any]] = (), context_budget: int | None = None, ) -> bool
• _has_code_action · function · L378-L381 — def _has_code_action(message: Mapping[str, Any]) -> bool
• segment_messages · function · L384-L422 — def segment_messages( messages: Sequence[Mapping[str, Any]], ) -> tuple[ContextSegment, ...]
• safe_keep_recent · function · L425-L440 — def safe_keep_recent(messages: list[dict], minimum: int = 4) -> int
• keep_recent_by_tokens · function · L443-L463 — def keep_recent_by_tokens(messages: list[dict], token_budget: int) -> int
• _existing_handoff_text · function · L466-L477 — def _existing_handoff_text(messages: Sequence[Mapping[str, Any]]) -> str | None: # The note's banner is framing for the live model, not handoff content. # Shown under "PREVIOUS HANDOFF (carry every fact forward)" the # summarizer echoes it, and compact() would then prepend it a second # time while the ledger restore prepends it once.
• _summary_pieces · function · L480-L495 — def _summary_pieces( messages: Sequence[Mapping[str, Any]], metadata: CompactionArchiveMetadata ) -> list[tuple[list[dict], int]]
• _next_summary_batch · function · L498-L518 — def _next_summary_batch( pieces: Sequence[tuple[list[dict], int]], start: int, limit: int ) -> tuple[list[dict], int]
• _compaction_windows · function · L521-L558 — def _compaction_windows( projected: list[dict], *, keep_recent: int, context_budget: int ) -> tuple[list[dict], list[dict], list[dict]] | None
• _summary_preamble · function · L561-L567 — def _summary_preamble(prev: str | None) -> str
• _summary_user_content · function · L570-L575 — def _summary_user_content( batch: Sequence[Mapping[str, Any]], metadata: CompactionArchiveMetadata, prev: str | None, ) -> str
• _content_chars · function · L578-L579 — def _content_chars(content: Any) -> int
• _large_output_candidate · function · L582-L600 — def _large_output_candidate(messages: Sequence[Mapping[str, Any]], index: int) -> bool
• _confine · function · L603-L613 — def _confine(root: Path, relative: str | Path, what: str) -> Path
• _confined_blob_path · function · L616-L623 — def _confined_blob_path(archive_dir: Path, digest: str) -> Path
• _content_blob_payload · function · L629-L649 — def _content_blob_payload( content: Any, message: Mapping[str, Any], metadata: CompactionArchiveMetadata, ) -> tuple[str, dict[str, Any]]
• _verify_blob_payload · function · L652-L665 — def _verify_blob_payload(existing: Any, digest: str) -> Any
• _TornBlob · class · L668-L669 — class _TornBlob(ValueError)
• _read_bounded · function · L693-L708 — def _read_bounded(name: str, limit: int, *, dir_fd: int | None) -> bytes
• content_digest · function · L711-L713 — def content_digest(content: Any) -> str
• _verify_existing_blob · function · L716-L752 — def _verify_existing_blob( name: str, payload: Mapping[str, Any], *, dir_fd: int | None ) -> None
• _open_below · function · L756-L786 — def _open_below(root: Path, relative_dir: PurePosixPath) -> Iterator[int]
• _write_exclusive_json_at · function · L789-L841 — def _write_exclusive_json_at( dir_fd: int | None, name: str, payload: Mapping[str, Any], *, host_owned: bool ) -> None
• _write_json_below · function · L844-L859 — def _write_json_below( root: Path, relative: str, payload: Mapping[str, Any], *, host_owned: bool ) -> None
• _write_content_blob · function · L862-L870 — def _write_content_blob( archive_dir: Path, digest: str, payload: Mapping[str, Any], ) -> str
• _workspace_context_ref · function · L873-L881 — def _workspace_context_ref(digest: str) -> str
• _confined_workspace_blob_path · function · L884-L885 — def _confined_workspace_blob_path(workspace: Path, digest: str) -> Path
• _write_workspace_content_blob · function · L888-L897 — def _write_workspace_content_blob( workspace: Path | str, digest: str, payload: Mapping[str, Any], ) -> str
• _preview · function · L900-L905 — def _preview(content: Any, limit: int) -> str
• _read_back_hint · function · L908-L924 — def _read_back_hint(workspace_ref: str | None) -> str | None
• _archived_marker · function · L927-L944 — def _archived_marker( title: str, fields: Sequence[tuple[str, Any]], preview: str, guidance: str | None, ) -> str
• externalize_large_outputs · function · L947-L1087 — def externalize_large_outputs( messages: Sequence[Mapping[str, Any]], archive_dir: Path | str | None, *, threshold_chars: int = DEFAULT_LARGE_OUTPUT_CHARS, preview_chars: int = DEFAULT_PREVIEW_CHARS, archive_metadata: Mapping[str, Any] | CompactionArchiveMetadata | None = None, artifact_archiver: ( Callable[[Any, Mapping[str, Any], dict[str, Any]], Mapping[str, Any]] | None ) = None, workspace: Path | str | None = None, ) -> list[dict]
• load_archived_content · function · L1090-L1093 — def load_archived_content(archive_dir: Path | str, digest: str) -> Any
• _runtime_handoff_value · function · L1096-L1113 — def _runtime_handoff_value(metadata: CompactionArchiveMetadata) -> str
• _handoff_titles_complete · function · L1116-L1118 — def _handoff_titles_complete(text: str) -> bool
• _normalize_handoff · function · L1121-L1152 — def _normalize_handoff(summary: str, metadata: CompactionArchiveMetadata) -> str
• _truncate_summary_arg · function · L1155-L1170 — def _truncate_summary_arg(value: Any, limit: int = _SUMMARY_TOOL_ARG_CHARS) -> Any
• _summary_tool_call · function · L1173-L1188 — def _summary_tool_call(call: Any) -> Any
• _summary_content · function · L1194-L1222 — def _summary_content(content: Any) -> Any
• _summary_message · function · L1225-L1237 — def _summary_message(message: Mapping[str, Any]) -> dict[str, Any]
• _summary_transcript · function · L1240-L1245 — def _summary_transcript(middle: Sequence[Mapping[str, Any]]) -> str
• _summary_header · function · L1248-L1253 — def _summary_header(metadata: CompactionArchiveMetadata) -> str
• _summary_input · function · L1256-L1259 — def _summary_input( middle: Sequence[Mapping[str, Any]], metadata: CompactionArchiveMetadata ) -> str
• _summary_output_cap · function · L1262-L1278 — def _summary_output_cap(cfg: Config) -> int | None
• _summary_max_tokens · function · L1281-L1291 — def _summary_max_tokens(cfg: Config, cap: int | None = None) -> int
• _summary_truncated · function · L1303-L1307 — def _summary_truncated(reply: Mapping[str, Any]) -> bool
• _require_usable_summary · function · L1310-L1322 — def _require_usable_summary(reply: Mapping[str, Any]) -> str
• _summary_chunk · function · L1331-L1368 — def _summary_chunk( request: list[dict[str, Any]], cfg: Config, max_tokens: int, cap: int | None, should_cancel: Callable[[], bool] | None = None, ) -> str
• compact · function · L1371-L1489 — def compact( messages: list[dict], cfg: Config, *, keep_recent: int = 4, archive_dir: Path | str | None = None, archive_metadata: Mapping[str, Any] | CompactionArchiveMetadata | None = None, large_output_chars: int = DEFAULT_LARGE_OUTPUT_CHARS, artifact_archiver: ( Callable[[Any, Mapping[str, Any], dict[str, Any]], Mapping[str, Any]] | None ) = None, archive_sink: Callable[[Mapping[str, Any]], Any] | None = None, tool_schemas: Iterable[Mapping[str, Any]] = (), context_budget: int | None = None, workspace: Path | str | None = None, should_cancel: Callable[[], bool] | None = None, ) -> list[dict]
• _archive · function · L1492-L1537 — def _archive( archive_dir: Path, middle: list[dict], summary: str, handoff: str | None = None, metadata: CompactionArchiveMetadata | None = None, before: ContextEstimate | None = None, after: ContextEstimate | None = None, *, archive_sink: Callable[[Mapping[str, Any]], Any] | None = None, summary_chunk_estimates: Sequence[int] | None = None, ) -> Path