← Files Portable ResumeARCHIVED FILE
skills/.portable-resume/runtime/portable_resume/adapters/opencode.py
46.3 KB · Oct 3, 2026 · 06:34 UTC
"""Read supported OpenCode stores without invoking OpenCode or opening a live DB.
The provider signatures in this module are deliberately structural and closed:
* ``opencode-sqlite-v1`` requires the ``session``/``message``/``part``
relations and their documented join columns.
* ``opencode-file-store-v1`` requires the legacy
``storage/{session,message,part}`` tree and explicit ID fields.
Supported on-disk keys for show are session-scoped
``storage/message/<sessionID>/`` and message-scoped
``storage/part/<messageID>/`` (see fixtures). Unrelated sessions are never
enumerated during show.
* an explicit OpenCode export is accepted only when it has the closed
``info`` plus ``messages[].{info,parts}`` shape. It is still reported as a
file-store capability; the provider string distinguishes it.
Product bound: an export is one **source document** limited by
``source_read_bytes`` (aggregate admitted payload), not the generic single
record ``record_bytes`` ceiling. Internal JSON depth/width still use the
shared structural limits.
List selection applies ``WHERE id = ?`` before any newest-session ``LIMIT`` so
older exact IDs remain reachable. SQLite show charges joined message/part rows
against ``transcript_records`` (with a ``LIMIT n+1`` fail-closed overflow),
never the discovery ``scanned_records`` ceiling.
Unknown schemas are never guessed. SQLite is queried only after the common
snapshot primitive has copied a stable main/WAL family to private storage.
"""
from __future__ import annotations
import contextlib
import json
import os
import sqlite3
import stat
from dataclasses import replace
from datetime import datetime, timedelta, timezone
from typing import Any, Iterable, Iterator, Mapping
from .base import CapabilityReport, ResolvedRef
from ..bounds import DEFAULT_BOUNDS, ReadBudget
from ..diagnostics import DiagnosticError
from ..model import Query, Session, SessionSummary, Turn
from ..paths import canonical_root, canonicalize_cwd, is_within, same_cwd
from ..sanitize import sanitize_turn_record
from ..snapshot import (
private_sqlite_connection,
private_sqlite_connection_live_wal_cow,
query_only_live_sqlite,
stable_read_bytes,
)
SQLITE_FORMAT = "opencode-sqlite-v1"
FILE_FORMAT = "opencode-file-store-v1"
EXPORT_PROVIDER = "opencode-export-file-v1"
_DEGRADABLE_SQLITE_CODES = frozenset({"E_SQLITE_LIVE_WAL", "E_SOURCE_BUSY"})
_DATABASE_NAMES = ("opencode.db", "opencode.sqlite")
_REQUIRED_COLUMNS: dict[str, dict[str, str]] = {
"session": {
"id": "TEXT",
"directory": "TEXT",
"title": "TEXT",
"time_created": "INTEGER",
"time_updated": "INTEGER",
},
"message": {
"id": "TEXT",
"session_id": "TEXT",
"time_created": "INTEGER",
"data": "TEXT",
},
"part": {
"id": "TEXT",
"message_id": "TEXT",
"session_id": "TEXT",
"time_created": "INTEGER",
"data": "TEXT",
},
}
_CONTROL_PARTS = frozenset(
{
"reasoning",
"thinking",
"step-start",
"step-finish",
"snapshot",
"patch",
"control",
"system",
}
)
def _opencode_sqlite_diagnostic(error: DiagnosticError) -> DiagnosticError:
"""Bind a shared snapshot diagnostic to this source without losing details."""
return DiagnosticError(
error.code,
source="opencode",
provider=error.provider or SQLITE_FORMAT,
attempts=error.attempts,
family=error.family,
)
_BINARY_PARTS = frozenset({"file", "image", "audio", "video", "attachment"})
class _DuplicateKey(ValueError):
pass
def _closed_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
value: dict[str, Any] = {}
for key, item in pairs:
if key in value:
raise _DuplicateKey(key)
value[key] = item
return value
def _json_value(data: bytes, *, unsupported: bool = False) -> Any:
try:
value = json.loads(data.decode("utf-8"), object_pairs_hook=_closed_object)
except (UnicodeDecodeError, json.JSONDecodeError, _DuplicateKey, RecursionError) as error:
code = "E_UNSUPPORTED_FORMAT" if unsupported else "E_CORRUPT_RECORD"
raise DiagnosticError(code, source="opencode") from error
_check_depth(value)
return value
def _json_text(value: object, *, unsupported: bool = False) -> Mapping[str, Any]:
if not isinstance(value, str):
raise DiagnosticError("E_UNSUPPORTED_FORMAT" if unsupported else "E_CORRUPT_RECORD", source="opencode")
parsed = _json_value(value.encode("utf-8"), unsupported=unsupported)
if not isinstance(parsed, Mapping):
raise DiagnosticError("E_UNSUPPORTED_FORMAT" if unsupported else "E_CORRUPT_RECORD", source="opencode")
return parsed
def _check_depth(value: Any, depth: int = 0) -> None:
if depth > 32:
raise DiagnosticError.limit_exceeded()
if isinstance(value, Mapping):
if len(value) > 512:
raise DiagnosticError.limit_exceeded()
for item in value.values():
_check_depth(item, depth + 1)
elif isinstance(value, list):
if len(value) > DEFAULT_BOUNDS.scanned_records:
raise DiagnosticError.limit_exceeded()
for item in value:
_check_depth(item, depth + 1)
def _identifier(value: object) -> str:
if not isinstance(value, str) or not value or len(value) > DEFAULT_BOUNDS.ref_chars:
raise DiagnosticError("E_CORRUPT_RECORD", source="opencode")
if value in {".", ".."} or any(ord(char) < 0x20 for char in value):
raise DiagnosticError("E_CORRUPT_RECORD", source="opencode")
return value
def _timestamp(value: object) -> str | None:
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
seconds = float(value)
if abs(seconds) >= 100_000_000_000:
seconds /= 1000.0
try:
return datetime.fromtimestamp(seconds, timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
except (OverflowError, OSError, ValueError):
return None
if isinstance(value, str):
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
return None
if parsed.tzinfo is None:
return None
return parsed.isoformat(timespec="seconds").replace("+00:00", "Z")
return None
def _time_container(value: object, field: str) -> str | None:
return _timestamp(value.get(field)) if isinstance(value, Mapping) else None
def _within_window(updated_at: str | None, minutes: int) -> bool:
if minutes is not None and minutes <= 0:
return True
if updated_at is None:
return False
try:
parsed = datetime.fromisoformat(updated_at.replace("Z", "+00:00"))
except ValueError:
return False
return parsed >= datetime.now(timezone.utc) - timedelta(minutes=minutes)
def _eligible(summary: SessionSummary, query: Query) -> bool:
if query.cwd is not None and (summary.cwd is None or not same_cwd(summary.cwd, query.cwd)):
return False
ref = query.ref.strip() if query.ref else None
if ref == summary.session_id:
return True
if ref and os.path.isabs(ref) and summary.source_path is not None:
if canonicalize_cwd(ref) == canonicalize_cwd(summary.source_path):
return True
minutes = query.within_min if query.within_min is not None else DEFAULT_BOUNDS.listing_age_minutes
return _within_window(summary.updated_at, minutes)
def _exact_session_ref(query: Query | None) -> str | None:
"""Return a non-path, non-latest ref that may be an exact session ID.
Absolute paths and ``latest`` are never treated as session IDs. Callers
must still confirm a matching row exists before treating the ref as exact.
"""
if query is None or query.ref is None:
return None
ref = query.ref.strip()
if not ref or ref == "latest" or os.path.isabs(ref):
return None
if len(ref) > DEFAULT_BOUNDS.ref_chars:
return None
return ref
def _field_utf8_len(value: object) -> int:
if value is None:
return 0
if isinstance(value, bool):
raise DiagnosticError("E_CORRUPT_RECORD", source="opencode", provider=SQLITE_FORMAT)
if isinstance(value, (bytes, bytearray, memoryview)):
return len(bytes(value))
if isinstance(value, str):
return len(value.encode("utf-8"))
raise DiagnosticError("E_CORRUPT_RECORD", source="opencode", provider=SQLITE_FORMAT)
def _charge_sql_text_field(value: object, budget: ReadBudget) -> None:
"""Charge one message/part payload against record_bytes + source_read_bytes."""
size = _field_utf8_len(value)
max_record = min(budget.limits.record_bytes, DEFAULT_BOUNDS.record_bytes)
if size > max_record:
raise DiagnosticError.limit_exceeded()
if size:
budget.consume_bytes(size)
def _export_read_cap(budget: ReadBudget | None) -> int:
"""Explicit export documents use source_read_bytes, not record_bytes."""
if budget is None:
return DEFAULT_BOUNDS.source_read_bytes
return min(budget.limits.source_read_bytes, DEFAULT_BOUNDS.source_read_bytes)
def _regular_json_files(base: str, root: str) -> list[str]:
"""Return a bounded, no-symlink set of JSON files below ``base``."""
if not os.path.isdir(base) or not is_within(base, root):
return []
pending = [base]
files: list[str] = []
observed = 0
while pending:
directory = pending.pop()
try:
entries = sorted(os.scandir(directory), key=lambda entry: entry.name)
except OSError as error:
raise DiagnosticError.source_busy(provider=FILE_FORMAT) from error
for entry in entries:
observed += 1
if observed > DEFAULT_BOUNDS.scanned_records:
raise DiagnosticError.limit_exceeded()
if entry.is_symlink():
raise DiagnosticError.unsafe_path()
try:
mode = entry.stat(follow_symlinks=False).st_mode
except OSError as error:
raise DiagnosticError.source_busy(provider=FILE_FORMAT) from error
if stat.S_ISDIR(mode):
pending.append(entry.path)
elif stat.S_ISREG(mode) and entry.name.endswith(".json"):
files.append(entry.path)
elif not stat.S_ISREG(mode):
raise DiagnosticError.unsafe_path()
return sorted(files)
class OpenCodeAdapter:
key = "opencode"
def __init__(self, *, root: str | None = None, read_hook: Any = None, sqlite_hook: Any = None):
self._configured_root = root
self._read_hook = read_hook
self._sqlite_hook = sqlite_hook
def _root(self, query: Query, *, required: bool = False) -> str | None:
if query.source_root is not None:
candidate = query.source_root
elif self._configured_root is not None:
candidate = self._configured_root
else:
data_home = os.environ.get("XDG_DATA_HOME")
candidate = os.path.join(data_home, "opencode") if data_home else os.path.expanduser("~/.local/share/opencode")
if not os.path.isdir(candidate):
if required:
raise DiagnosticError("E_CAPABILITY_UNAVAILABLE", source=self.key)
return None
return canonical_root(candidate)
def approved_roots(self, query: Query) -> tuple[str, ...]:
root = self._root(query)
return (root,) if root is not None else ()
def _database_paths(self, root: str) -> list[str]:
return [os.path.join(root, name) for name in _DATABASE_NAMES if os.path.isfile(os.path.join(root, name))]
def _file_store(self, root: str) -> str | None:
storage = os.path.join(root, "storage")
required = tuple(os.path.join(storage, name) for name in ("session", "message", "part"))
if all(os.path.isdir(path) and not os.path.islink(path) for path in required):
return storage
return None
def _explicit_exports(self, root: str, query: Query) -> list[str]:
values: list[str] = []
if query.ref and os.path.isabs(query.ref) and query.ref.endswith(".json"):
if not is_within(query.ref, root):
raise DiagnosticError.unsafe_path()
if os.path.isfile(query.ref):
values.append(canonicalize_cwd(query.ref))
export_root = os.path.join(root, "exports")
if os.path.isdir(export_root) and not os.path.islink(export_root):
values.extend(_regular_json_files(export_root, root))
return list(dict.fromkeys(values))
def _sqlite_supported(self, database: str, root: str) -> bool:
try:
with self._sqlite_connection(database, root) as connection:
self._require_schema(connection)
return True
except sqlite3.DatabaseError:
return False
except DiagnosticError as error:
if error.code == "E_UNSUPPORTED_FORMAT":
return False
raise
@contextlib.contextmanager
def _sqlite_connection(self, database: str, root: str) -> Iterator[sqlite3.Connection]:
"""Open only an accepted private snapshot for a live WAL family.
Small quiescent stores retain the existing byte-copy path. Oversized
rollback-mode stores retain the descriptor-pinned live read. A live or
persistent WAL family is handed to the capability-gated Darwin/APFS COW
backend; unsupported hosts preserve ``E_SQLITE_LIVE_WAL`` for the
provider-local file/export fallback policy.
"""
try:
size = os.path.getsize(database)
except OSError as error:
raise DiagnosticError(
"E_SOURCE_BUSY", source=self.key, provider=SQLITE_FORMAT
) from error
if size <= DEFAULT_BOUNDS.sqlite_snapshot_bytes:
yielded = False
try:
with private_sqlite_connection(
database,
root=root,
hook=self._sqlite_hook,
provider=SQLITE_FORMAT,
) as connection:
yielded = True
yield connection
return
except DiagnosticError as error:
if yielded or error.code != "E_SOURCE_BUSY":
raise
# The COW backend is specifically a live-WAL recovery path.
# A main-only rewrite/race must retain E_SOURCE_BUSY rather
# than being reclassified by a WAL backend that has no
# sidecars to pin.
if not any(os.path.lexists(database + suffix) for suffix in ("-wal", "-shm")):
raise
else:
yielded = False
try:
with query_only_live_sqlite(
database,
root=root,
provider=SQLITE_FORMAT,
) as connection:
yielded = True
yield connection
return
except DiagnosticError as error:
if yielded or error.code != "E_SQLITE_LIVE_WAL":
raise
with private_sqlite_connection_live_wal_cow(
database,
root=root,
bounds=DEFAULT_BOUNDS,
hook=self._sqlite_hook,
provider=SQLITE_FORMAT,
) as connection:
yield connection
@staticmethod
def _require_schema(connection: sqlite3.Connection) -> None:
try:
integrity = connection.execute("PRAGMA integrity_check(1)").fetchone()
if integrity != ("ok",):
raise DiagnosticError("E_CORRUPT_RECORD", source="opencode", provider=SQLITE_FORMAT)
tables = {
row[0]
for row in connection.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name IN ('session','message','part')"
)
}
if tables != set(_REQUIRED_COLUMNS):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source="opencode", provider=SQLITE_FORMAT)
for table, required in _REQUIRED_COLUMNS.items():
columns = {
str(row[1]): str(row[2]).upper().split("(", 1)[0]
for row in connection.execute(f'PRAGMA table_info("{table}")')
}
if any(columns.get(name) != affinity for name, affinity in required.items()):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source="opencode", provider=SQLITE_FORMAT)
except sqlite3.DatabaseError as error:
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source="opencode", provider=SQLITE_FORMAT) from error
def probe(self, query: Query) -> CapabilityReport:
root = self._root(query)
if root is None:
return CapabilityReport(self.key, None, "unavailable")
evidence: list[str] = []
unsupported = False
sqlite_ok = False
skipped_sqlite: DiagnosticError | None = None
for database in self._database_paths(root):
try:
if self._sqlite_supported(database, root):
sqlite_ok = True
evidence.append(f"sqlite:{os.path.basename(database)}")
else:
unsupported = True
except DiagnosticError as error:
if error.code not in _DEGRADABLE_SQLITE_CODES:
raise
skipped_sqlite = _opencode_sqlite_diagnostic(error)
storage = self._file_store(root)
file_ok = False
if not sqlite_ok and storage is not None:
file_ok = any(
_eligible(item, query)
for item in self._list_file_store(storage, root, ReadBudget())
)
if file_ok:
evidence.append("file-store:storage")
export_ok = False
if not sqlite_ok:
export_budget = ReadBudget()
for path in self._explicit_exports(root, query):
try:
summary, _ = self._read_export(path, root, budget=export_budget)
if _eligible(summary, query):
export_ok = True
evidence.append("export:explicit-json")
except DiagnosticError as error:
if error.code in {"E_CORRUPT_RECORD", "E_UNSUPPORTED_FORMAT"}:
unsupported = True
else:
raise
if not (sqlite_ok or file_ok or export_ok):
if skipped_sqlite is not None:
raise skipped_sqlite
return CapabilityReport(self.key, SQLITE_FORMAT if self._database_paths(root) else None, "unsupported", root=root)
format_id = SQLITE_FORMAT if sqlite_ok else FILE_FORMAT
# The warning means output actually came from an independent fallback,
# not merely that one alternate SQLite filename was unavailable.
warnings = (
("W_SOURCE_PROVIDER_SKIPPED",)
if skipped_sqlite is not None and not sqlite_ok and (file_ok or export_ok)
else ()
)
state = "partial" if unsupported or skipped_sqlite is not None else "supported"
return CapabilityReport(
self.key,
format_id,
state,
root=root,
evidence=tuple(evidence),
warnings=warnings,
)
def list(self, query: Query, budget: ReadBudget) -> list[SessionSummary]:
root = self._root(query, required=True)
assert root is not None
output: list[SessionSummary] = []
databases = self._database_paths(root)
supported_database = False
skipped_sqlite: DiagnosticError | None = None
for database in databases:
try:
summaries = self._list_sqlite(database, root, budget, query=query)
supported_database = True
output.extend(item for item in summaries if _eligible(item, query))
except DiagnosticError as error:
if error.code in _DEGRADABLE_SQLITE_CODES:
skipped_sqlite = _opencode_sqlite_diagnostic(error)
continue
if error.code != "E_UNSUPPORTED_FORMAT":
raise
storage = self._file_store(root)
if storage is not None:
output.extend(item for item in self._list_file_store(storage, root, budget) if _eligible(item, query))
for export in self._explicit_exports(root, query):
summary, _ = self._read_export(export, root, budget=budget)
if _eligible(summary, query):
output.append(summary)
if skipped_sqlite is not None:
if not output:
raise skipped_sqlite
output = [
replace(
item,
warnings=tuple(
dict.fromkeys((*item.warnings, "W_SOURCE_PROVIDER_SKIPPED"))
),
)
if item.provider != SQLITE_FORMAT
else item
for item in output
]
if not output and databases and not supported_database and storage is None:
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key, provider=SQLITE_FORMAT)
if not (databases or storage is not None or self._explicit_exports(root, query)):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key)
return output
def _list_sqlite(
self, database: str, root: str, budget: ReadBudget, *, query: Query | None = None
) -> list[SessionSummary]:
def _fetch(connection: sqlite3.Connection) -> list[tuple]:
self._require_schema(connection)
# Exact ID before any newest-session LIMIT so older sessions remain
# selectable (issue #13). On miss, fall through for text / latest.
exact = _exact_session_ref(query)
if exact is not None:
exact_rows = connection.execute(
'SELECT id,directory,title,time_created,time_updated FROM "session" '
"WHERE id = ? LIMIT 2",
(exact,),
).fetchall()
if len(exact_rows) > 1:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=SQLITE_FORMAT)
if exact_rows:
return exact_rows
# Prefer cwd-scoped list (Codex-style) so large live DBs stay bounded.
limit = DEFAULT_BOUNDS.listed_sessions
if query is not None and query.cwd:
rows = connection.execute(
'SELECT id,directory,title,time_created,time_updated FROM "session" '
"WHERE directory = ? ORDER BY time_updated DESC,id ASC LIMIT ?",
(query.cwd, limit),
).fetchall()
if rows:
return rows
# Directory string miss: page newest sessions for same_cwd in Python.
return connection.execute(
'SELECT id,directory,title,time_created,time_updated FROM "session" '
"ORDER BY time_updated DESC,id ASC LIMIT ?",
(DEFAULT_BOUNDS.scanned_records,),
).fetchall()
return connection.execute(
'SELECT id,directory,title,time_created,time_updated FROM "session" '
"ORDER BY time_updated DESC,id ASC LIMIT ?",
(limit,),
).fetchall()
try:
with self._sqlite_connection(database, root) as connection:
rows = _fetch(connection)
except sqlite3.DatabaseError as error:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=SQLITE_FORMAT) from error
except DiagnosticError as error:
if error.code in _DEGRADABLE_SQLITE_CODES and error.source is None:
raise _opencode_sqlite_diagnostic(error) from error
raise
# Do not hard-fail when the DB has more sessions than LIMIT (live homes).
budget.consume_records(len(rows))
output: list[SessionSummary] = []
seen: set[str] = set()
for row in rows:
session_id = _identifier(row[0])
if session_id in seen:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=SQLITE_FORMAT)
seen.add(session_id)
if not isinstance(row[1], str):
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=SQLITE_FORMAT)
try:
cwd = canonicalize_cwd(row[1])
except DiagnosticError:
continue
output.append(
SessionSummary(
source=self.key,
session_id=session_id,
source_path=database,
title=row[2] if isinstance(row[2], str) else None,
cwd=cwd,
created_at=_timestamp(row[3]),
updated_at=_timestamp(row[4]),
provider=SQLITE_FORMAT,
)
)
return output
def _read_json_file(
self,
path: str,
root: str,
budget: ReadBudget | None,
*,
max_bytes: int | None = None,
) -> Mapping[str, Any]:
ceiling = DEFAULT_BOUNDS.record_bytes if max_bytes is None else max_bytes
read = stable_read_bytes(
path,
root=root,
max_bytes=ceiling,
budget=budget,
hook=self._read_hook,
)
value = _json_value(read.data)
if not isinstance(value, Mapping):
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key)
return value
def _list_file_store(self, storage: str, root: str, budget: ReadBudget) -> list[SessionSummary]:
output: list[SessionSummary] = []
seen: set[str] = set()
for path in _regular_json_files(os.path.join(storage, "session"), root):
value = self._read_json_file(path, root, budget)
summary = self._file_summary(path, value)
if summary.session_id in seen:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=FILE_FORMAT)
seen.add(summary.session_id)
output.append(summary)
return output
def _file_summary(self, path: str, value: Mapping[str, Any]) -> SessionSummary:
session_id = _identifier(value.get("id"))
directory = value.get("directory")
times = value.get("time")
if not isinstance(directory, str) or not isinstance(times, Mapping):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key, provider=FILE_FORMAT)
return SessionSummary(
source=self.key,
session_id=session_id,
source_path=path,
title=value.get("title") if isinstance(value.get("title"), str) else None,
cwd=canonicalize_cwd(directory),
created_at=_time_container(times, "created"),
updated_at=_time_container(times, "updated"),
provider=FILE_FORMAT,
)
@staticmethod
def _export_shape(value: Mapping[str, Any]) -> tuple[Mapping[str, Any], list[Any]]:
info = value.get("info")
messages = value.get("messages")
if not isinstance(info, Mapping) or not isinstance(messages, list):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source="opencode", provider=EXPORT_PROVIDER)
if not all(isinstance(item, Mapping) and isinstance(item.get("info"), Mapping) and isinstance(item.get("parts"), list) for item in messages):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source="opencode", provider=EXPORT_PROVIDER)
return info, messages
def _read_export(
self, path: str, root: str, *, budget: ReadBudget | None
) -> tuple[SessionSummary, tuple[Mapping[str, Any], list[Any]]]:
# Explicit product bound: one export document uses source_read_bytes.
value = self._read_json_file(path, root, budget, max_bytes=_export_read_cap(budget))
info, messages = self._export_shape(value)
session_id = _identifier(info.get("id"))
directory = info.get("directory")
times = info.get("time")
if not isinstance(directory, str) or not isinstance(times, Mapping):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key, provider=EXPORT_PROVIDER)
summary = SessionSummary(
source=self.key,
session_id=session_id,
source_path=path,
title=info.get("title") if isinstance(info.get("title"), str) else None,
cwd=canonicalize_cwd(directory),
created_at=_time_container(times, "created"),
updated_at=_time_container(times, "updated"),
provider=EXPORT_PROVIDER,
)
return summary, (info, messages)
def show(self, ref: ResolvedRef, query: Query, budget: ReadBudget) -> Session:
root = self._root(query, required=True)
assert root is not None
provider = ref.provider
if provider == SQLITE_FORMAT:
session = self._show_sqlite(ref, query, root, budget)
elif provider == FILE_FORMAT:
session = self._show_file_store(ref, query, root, budget)
elif provider == EXPORT_PROVIDER:
session = self._show_export(ref, query, root, budget)
else:
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key, provider=provider)
return replace(
session,
warnings=tuple(dict.fromkeys((*ref.warnings, *session.warnings))),
)
def _show_sqlite(self, ref: ResolvedRef, query: Query, root: str, budget: ReadBudget) -> Session:
if (
ref.source_path is None
or not is_within(ref.source_path, root)
or canonicalize_cwd(ref.source_path)
not in {canonicalize_cwd(path) for path in self._database_paths(root)}
):
raise DiagnosticError.unsafe_path()
row_cap = min(budget.limits.transcript_records, DEFAULT_BOUNDS.transcript_records)
try:
with self._sqlite_connection(ref.source_path, root) as connection:
self._require_schema(connection)
summary_row = connection.execute(
'SELECT id,directory,title,time_created,time_updated FROM "session" WHERE id=?',
(ref.session_id,),
).fetchone()
if summary_row is None:
raise DiagnosticError("E_NO_MATCH", source=self.key)
# Transcript ceiling + 1: overflow fails closed (no silent prefix).
# Stream rows (no fetchall) so record_bytes / source_read_bytes
# can fail closed without materializing the full join first.
cursor = connection.execute(
'SELECT m.id,m.time_created,m.data,p.id,p.time_created,p.data '
'FROM "message" AS m LEFT JOIN "part" AS p '
"ON p.message_id=m.id AND p.session_id=m.session_id "
"WHERE m.session_id=? "
"ORDER BY m.time_created ASC,m.id ASC,p.time_created ASC,p.id ASC LIMIT ?",
(ref.session_id, row_cap + 1),
)
rows = []
charged_messages: set[str] = set()
try:
for row in cursor:
if len(rows) >= row_cap:
raise DiagnosticError.limit_exceeded()
# Charge once per admitted row field as it streams in.
message_key = str(row[0])
message_data = row[2]
part_data = row[5]
if message_key not in charged_messages:
charged_messages.add(message_key)
_charge_sql_text_field(message_data, budget)
if part_data is not None:
_charge_sql_text_field(part_data, budget)
budget.consume_transcript_records()
rows.append(row)
finally:
cursor.close()
orphan_count = connection.execute(
'SELECT COUNT(*) FROM "part" AS p LEFT JOIN "message" AS m '
"ON m.id=p.message_id AND m.session_id=p.session_id "
"WHERE p.session_id=? AND m.id IS NULL",
(ref.session_id,),
).fetchone()
except sqlite3.DatabaseError as error:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=SQLITE_FORMAT) from error
except DiagnosticError as error:
if error.code in _DEGRADABLE_SQLITE_CODES and error.source is None:
raise _opencode_sqlite_diagnostic(error) from error
raise
warnings: list[str] = []
if orphan_count and int(orphan_count[0]) > 0:
warnings.append("W_BROKEN_CHAIN")
turns = self._turns_from_sql_rows(rows, query, budget, warnings)
summary = SessionSummary(
source=self.key,
session_id=_identifier(summary_row[0]),
source_path=ref.source_path,
title=summary_row[2] if isinstance(summary_row[2], str) else None,
cwd=canonicalize_cwd(summary_row[1]),
created_at=_timestamp(summary_row[3]),
updated_at=_timestamp(summary_row[4]),
provider=SQLITE_FORMAT,
warnings=tuple(dict.fromkeys(warnings)),
)
return _session_from(summary, turns, warnings)
def _turns_from_sql_rows(
self, rows: Iterable[tuple[Any, ...]], query: Query, budget: ReadBudget, warnings: list[str]
) -> list[Turn]:
turns: list[Turn] = []
seen_message: str | None = None
message_values: dict[str, tuple[object, object]] = {}
empty_messages: set[str] = set()
seen_parts: set[str] = set()
role = ""
for message_id, message_time, message_data, part_id, part_time, part_data in rows:
current = _identifier(message_id)
fingerprint = (message_time, message_data)
if current in message_values and message_values[current] != fingerprint:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=SQLITE_FORMAT)
message_values[current] = fingerprint
if current != seen_message:
seen_message = current
info = _json_text(message_data)
role_value = info.get("role")
if not isinstance(role_value, str):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key, provider=SQLITE_FORMAT)
role = role_value.casefold()
if part_id is None:
if current in empty_messages:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=SQLITE_FORMAT)
empty_messages.add(current)
warnings.append("W_MISSING_BLOB")
continue
current_part = _identifier(part_id)
if current_part in seen_parts:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=SQLITE_FORMAT)
seen_parts.add(current_part)
part = _json_text(part_data)
self._append_part(turns, part, role, _timestamp(part_time if part_time is not None else message_time), query, budget, warnings)
return turns
def _show_file_store(self, ref: ResolvedRef, query: Query, root: str, budget: ReadBudget) -> Session:
storage = self._file_store(root)
if storage is None or ref.source_path is None:
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key, provider=FILE_FORMAT)
session_root = os.path.join(storage, "session")
if not is_within(ref.source_path, session_root):
raise DiagnosticError.unsafe_path()
approved = {canonicalize_cwd(path) for path in _regular_json_files(session_root, root)}
if canonicalize_cwd(ref.source_path) not in approved:
raise DiagnosticError.unsafe_path()
summary_value = self._read_json_file(ref.source_path, root, budget)
summary = self._file_summary(ref.source_path, summary_value)
if summary.session_id != ref.session_id:
raise DiagnosticError("E_NO_MATCH", source=self.key)
if not _eligible(summary, query):
raise DiagnosticError.unsafe_path()
# Scope to storage/message/<sessionID>/ — do not enumerate unrelated sessions.
message_dir = os.path.join(storage, "message", ref.session_id)
messages: dict[str, tuple[Mapping[str, Any], str]] = {}
if os.path.lexists(message_dir):
if os.path.islink(message_dir) or not os.path.isdir(message_dir):
raise DiagnosticError.unsafe_path()
if not is_within(message_dir, root):
raise DiagnosticError.unsafe_path()
for path in _regular_json_files(message_dir, root):
budget.consume_transcript_records()
value = self._read_json_file(path, root, budget)
if value.get("sessionID") != ref.session_id:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=FILE_FORMAT)
message_id = _identifier(value.get("id"))
role = value.get("role")
if not isinstance(role, str):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key, provider=FILE_FORMAT)
if message_id in messages:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=FILE_FORMAT)
messages[message_id] = (value, path)
parts: dict[str, list[Mapping[str, Any]]] = {message_id: [] for message_id in messages}
warnings: list[str] = []
seen_parts: set[str] = set()
# Scope to storage/part/<messageID>/ for admitted messages only.
for message_id in messages:
part_dir = os.path.join(storage, "part", message_id)
if not os.path.lexists(part_dir):
continue
if os.path.islink(part_dir) or not os.path.isdir(part_dir):
raise DiagnosticError.unsafe_path()
if not is_within(part_dir, root):
raise DiagnosticError.unsafe_path()
for path in _regular_json_files(part_dir, root):
budget.consume_transcript_records()
value = self._read_json_file(path, root, budget)
if value.get("sessionID") != ref.session_id:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=FILE_FORMAT)
part_message = _identifier(value.get("messageID"))
if part_message != message_id:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=FILE_FORMAT)
part_id = _identifier(value.get("id"))
if part_id in seen_parts:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=FILE_FORMAT)
seen_parts.add(part_id)
parts[message_id].append(value)
turns: list[Turn] = []
ordered_messages = sorted(
messages.items(),
key=lambda item: (_numeric_sort(_mapping_get(item[1][0], "time", "created")), item[0]),
)
for message_id, (message, _) in ordered_messages:
role = str(message["role"]).casefold()
ordered_parts = sorted(
parts[message_id],
key=lambda item: (_numeric_sort(_mapping_get(item, "time", "created")), str(item.get("id", ""))),
)
if not ordered_parts:
warnings.append("W_MISSING_BLOB")
for part in ordered_parts:
self._append_part(
turns,
part,
role,
_time_container(part.get("time"), "created") or _time_container(message.get("time"), "created"),
query,
budget,
warnings,
)
return _session_from(summary, turns, warnings)
def _show_export(self, ref: ResolvedRef, query: Query, root: str, budget: ReadBudget) -> Session:
if ref.source_path is None:
raise DiagnosticError.unsafe_path()
approved = {canonicalize_cwd(path) for path in self._explicit_exports(root, query)}
if canonicalize_cwd(ref.source_path) not in approved:
raise DiagnosticError.unsafe_path()
summary, (_, messages) = self._read_export(ref.source_path, root, budget=budget)
if summary.session_id != ref.session_id:
raise DiagnosticError("E_NO_MATCH", source=self.key)
turns: list[Turn] = []
warnings: list[str] = []
seen_messages: set[str] = set()
seen_parts: set[str] = set()
ordered = sorted(
messages,
key=lambda item: (
_numeric_sort(_mapping_get(item["info"], "time", "created")),
str(item["info"].get("id", "")),
),
)
for message in ordered:
info = message["info"]
if info.get("sessionID", ref.session_id) != ref.session_id:
warnings.append("W_BROKEN_CHAIN")
continue
message_id = _identifier(info.get("id"))
if message_id in seen_messages:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=EXPORT_PROVIDER)
seen_messages.add(message_id)
role = info.get("role")
if not isinstance(role, str):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key, provider=EXPORT_PROVIDER)
parts = sorted(
message["parts"],
key=lambda item: (_numeric_sort(_mapping_get(item, "time", "created")), str(item.get("id", ""))),
)
for part in parts:
if not isinstance(part, Mapping):
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=EXPORT_PROVIDER)
part_id = _identifier(part.get("id"))
if part_id in seen_parts:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key, provider=EXPORT_PROVIDER)
seen_parts.add(part_id)
self._append_part(
turns,
part,
role.casefold(),
_time_container(part.get("time"), "created") or _time_container(info.get("time"), "created"),
query,
budget,
warnings,
)
return _session_from(summary, turns, warnings)
def _append_part(
self,
turns: list[Turn],
part: Mapping[str, Any],
message_role: str,
timestamp: str | None,
query: Query,
budget: ReadBudget,
warnings: list[str],
) -> None:
kind = part.get("type")
if not isinstance(kind, str):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key)
kind = kind.casefold()
if kind in _CONTROL_PARTS or message_role in {"system", "reasoning", "control"}:
return
if kind in _BINARY_PARTS:
warnings.append("W_BINARY_OMITTED")
return
record: dict[str, Any]
if kind == "text":
text = part.get("text")
if not isinstance(text, str) or message_role not in {"user", "assistant"}:
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key)
record = {"role": message_role, "content": text, "timestamp": timestamp}
elif kind in {"tool", "tool-result", "tool_result"}:
tool_name = part.get("tool") or part.get("name")
content: object = part.get("content")
state = part.get("state")
if content is None and isinstance(state, Mapping):
content = state.get("output")
if not isinstance(content, str):
if content is None:
warnings.append("W_MISSING_BLOB")
return
raise DiagnosticError("E_CORRUPT_RECORD", source=self.key)
record = {
"role": "tool",
"content": content,
"timestamp": timestamp,
"tool_name": tool_name if isinstance(tool_name, str) else None,
}
else:
# Unknown content-bearing parts could hide a user/assistant turn.
if any(key in part for key in ("text", "content", "output")):
raise DiagnosticError("E_UNSUPPORTED_FORMAT", source=self.key)
warnings.append("W_BROKEN_CHAIN")
return
bounds = replace(DEFAULT_BOUNDS, tool_output_chars=query.max_tool_chars)
turn, turn_warnings = sanitize_turn_record(record, ordinal=len(turns), bounds=bounds)
warnings.extend(turn_warnings)
if turn is not None:
budget.consume_turns()
turns.append(turn)
def _mapping_get(value: object, container: str, key: str) -> object:
nested = value.get(container) if isinstance(value, Mapping) else None
return nested.get(key) if isinstance(nested, Mapping) else None
def _numeric_sort(value: object) -> tuple[int, str]:
if isinstance(value, bool):
return (1, "")
if isinstance(value, (int, float)):
return (0, f"{float(value):030.6f}")
if isinstance(value, str):
return (0, value)
return (1, "")
def _session_from(summary: SessionSummary, turns: Iterable[Turn], warnings: Iterable[str]) -> Session:
values = tuple(turns)
last_user = next((turn.content for turn in reversed(values) if turn.role == "user"), None)
last_assistant = next((turn.content for turn in reversed(values) if turn.role == "assistant"), None)
return Session(
source=summary.source,
session_id=summary.session_id,
source_path=summary.source_path,
title=summary.title,
cwd=summary.cwd,
branch=summary.branch,
created_at=summary.created_at,
updated_at=summary.updated_at,
source_repo_root=summary.source_repo_root,
last_user_request=last_user,
last_assistant_action=last_assistant,
turns=values,
warnings=tuple(dict.fromkeys((*summary.warnings, *warnings))),
)
ADAPTER = OpenCodeAdapter()
def get_adapter() -> OpenCodeAdapter:
return ADAPTER
SHA-256: 2caaecc608d1c595c179bc6adcad0a9d96956a7492619c2b27d0fbfce222367c