"""Bounded selected export v1. See docs/plan/PHASE_11_EXPORT_FORMAT.md.""" import hashlib import json import os from pathlib import Path import re import stat from uuid import UUID from agenthotel import MAX_BYTES, segment FORMAT = "agenthotel.selected-export" MAX_MANIFEST = 4 * 1024 * 1024 MAX_TOTAL = 1024 * 1024 * 1024 MAX_RECORDS = 10000 UNAVAILABLE = {"resource_not_found", "retained_excess_inaccessible", "context_revision_conflict", "history_unavailable", "cursor_expired"} class ExportError(Exception): """Sanitized failure; committed observations remain resumable.""" def _uuid(value): try: return isinstance(value, str) and str(UUID(value)) == value and UUID(value).int != 0 except ValueError: return False def _sequence(value): return (isinstance(value, str) and re.fullmatch(r"0|[1-9][0-9]{0,19}", value) and int(value) <= 2**64 - 1) def _selection(project, selection): if not _uuid(project) or not isinstance(selection, list) or not 1 <= len(selection) <= 100: raise ExportError("invalid selection") for item in selection: if not isinstance(item, dict): raise ExportError("invalid selector") kind = item.get("kind") if kind == "state" and set(item) == {"kind", "namespace", "key"}: for field in ("namespace", "key"): if not isinstance(item[field], str) or len(item[field].encode("utf-8")) > 256: raise ExportError("invalid state selector") try: segment(item[field]) except ValueError: raise ExportError("invalid state selector") from None elif kind == "context" and set(item) == {"kind", "id"} and _uuid(item["id"]): pass elif (kind == "inbox" and set(item) == {"kind", "id", "after_sequence"} and _uuid(item["id"]) and _sequence(item["after_sequence"])): pass else: raise ExportError("invalid selector") def _regular(path): info = path.lstat() if (not stat.S_ISREG(info.st_mode) or path.is_symlink() or getattr(info, "st_file_attributes", 0) & 0x400): raise ExportError("unsafe export file") return info.st_size def _json(data): try: value = json.loads(data) if not isinstance(value, dict): raise ValueError() return value except (ValueError, UnicodeError, RecursionError): raise ExportError("invalid export response or manifest") from None def _atomic(path, data, temporary=None): temporary = temporary or path.with_suffix(".pending") if temporary.exists(): _regular(temporary) # Directory must be caller-protected and single-writer; reject symlinks above. with temporary.open("wb") as stream: if os.name == "posix": os.chmod(temporary, 0o600) stream.write(data) stream.flush() os.fsync(stream.fileno()) os.replace(temporary, path) def _save(directory, manifest, *, writer=None): raw = json.dumps(manifest, ensure_ascii=True, separators=(",", ":")).encode() if len(raw) > MAX_MANIFEST: raise ExportError("manifest limit reached") (writer or _atomic)(directory / "manifest.json", raw) def _open(directory, project, selection, *, writer=None, persist=True): if not directory.exists(): directory.mkdir(mode=0o700) info = directory.lstat() if (not stat.S_ISDIR(info.st_mode) or directory.is_symlink() or getattr(info, "st_file_attributes", 0) & 0x400): raise ExportError("unsafe export directory") total = 0 for path in directory.iterdir(): size = _regular(path) if re.fullmatch(r"[a-f0-9]{64}\.bin", path.name): total += size if size > MAX_BYTES or total > MAX_TOTAL: raise ExportError("export size limit reached") elif path.name not in ("manifest.json", "manifest.pending", "body.pending"): raise ExportError("foreign file in export directory") elif size > MAX_MANIFEST and path.name != "body.pending": raise ExportError("manifest size limit reached") elif size > MAX_BYTES: raise ExportError("pending body size limit reached") path = directory / "manifest.json" if not path.exists(): m = {"format": FORMAT, "version": 1, "project": project, "selection": selection, "records": [], "index": 0, "stage": "initial", "cursor": "", "etag": "", "last_sequence": "", "high_water": ""} if persist: _save(directory, m, writer=writer) else: m = _json(path.read_bytes()) if (m.get("format") != FORMAT or m.get("version") != 1 or m.get("project") != project or m.get("selection") != selection or type(m.get("index")) is not int or not 0 <= m["index"] <= len(selection) or m.get("stage") not in ("initial", "content") or not isinstance(m.get("cursor"), str) or len(m["cursor"]) > 1024 or not isinstance(m.get("etag"), str) or any(m.get(k) != "" and not _sequence(m.get(k)) for k in ("last_sequence", "high_water")) or not isinstance(m.get("records"), list) or len(m["records"]) > MAX_RECORDS): raise ExportError("export selection or checkpoint mismatch") for record in m["records"]: if not isinstance(record, dict): raise ExportError("invalid export record") digest = record.get("sha256") if digest is not None: if not isinstance(digest, str) or not re.fullmatch(r"[a-f0-9]{64}", digest): raise ExportError("invalid body reference") blob = directory / (digest + ".bin") if _regular(blob) != record.get("bytes"): raise ExportError("incomplete export body") if hashlib.sha256(blob.read_bytes()).hexdigest() != digest: raise ExportError("export body checksum mismatch") return m, total def export_batch(client, directory, project, selection, *, requests=20, writer=None): """Perform up to requests reads; return True when all selectors have outcomes. Completion can include unavailable records: inspect those before using data. Call again explicitly to resume. No credential is written to the directory. An optional atomic writer lets workspace orchestration enforce aggregate limits before every body and serialized-manifest write. """ _selection(project, selection) if type(requests) is not int or not 1 <= requests <= 20: raise ExportError("invalid request bound") directory = Path(directory) manifest, total = _open(directory, project, selection, writer=writer) for _ in range(requests): index = manifest["index"] if index == len(selection): return True if len(manifest["records"]) >= MAX_RECORDS: raise ExportError("record limit reached") item = selection[index] base = "/api/v1/projects/" + project query, match = None, None if item["kind"] == "state": path = base + "/state/" + segment(item["namespace"]) + "/" + segment(item["key"]) elif item["kind"] == "context": path = base + "/context-files/" + item["id"] if manifest["stage"] == "content": path += "/content" match = manifest["etag"] if not re.fullmatch(r'"[1-9][0-9]{0,19}"', match): raise ExportError("invalid context revision") else: path = base + "/inboxes/" + item["id"] + "/events" query = {"limit": 100} if manifest["cursor"]: query["cursor"] = manifest["cursor"] else: query["after"] = item["after_sequence"] response = client.request("GET", path, query=query, if_match=match) if len(response.body) > MAX_BYTES: raise ExportError("response size limit reached") record = {"selector": index, "stage": manifest["stage"], "status": response.status} next_cursor, next_stage, next_etag = "", "initial", "" next_last, next_high = "", "" advance = True if response.status != 200: error = _json(response.body) code = error.get("code") if response.status not in (403, 404, 409, 410) or code not in UNAVAILABLE: raise ExportError("retrieval stopped; inspect authority, quotas or service availability") record["unavailable"] = code history = error.get("history", {}) if not isinstance(history, dict): history = {} for field in ("earliest_available_sequence", "latest_sequence"): value = history.get(field) if type(value) is int and 0 <= value <= 2**64 - 1: record[field] = str(value) else: headers = {key.lower(): value for key, value in response.headers.items()} etag = headers.get("etag", "") if etag and re.fullmatch(r'"[1-9][0-9]{0,19}"', etag): record["etag"] = etag if item["kind"] == "context" and manifest["stage"] == "initial": if "etag" not in record: raise ExportError("context revision missing") next_stage, next_etag, advance = "content", etag, False elif item["kind"] == "context" and etag != match: raise ExportError("context revision changed") elif item["kind"] == "inbox": page = _json(response.body) next_cursor = page.get("next_cursor", "") events, high = page.get("events"), page.get("latest_sequence") last = int(manifest["last_sequence"] or item["after_sequence"]) if (not isinstance(events, list) or len(events) > 100 or type(high) is not int or not last <= high <= 2**64 - 1 or manifest["high_water"] and high != int(manifest["high_water"])): raise ExportError("invalid event page") for event in events: if (not isinstance(event, dict) or type(event.get("sequence")) is not int or event["sequence"] != last + 1 or event["sequence"] > high): raise ExportError("invalid event sequence") last = event["sequence"] if (not isinstance(next_cursor, str) or len(next_cursor) > 1024 or next_cursor and (next_cursor == manifest["cursor"] or not events or last >= high) or not next_cursor and last != high): raise ExportError("invalid event cursor") advance = not next_cursor if not advance: next_last, next_high = str(last), str(high) digest = hashlib.sha256(response.body).hexdigest() blob = directory / (digest + ".bin") if not blob.exists(): if total + len(response.body) > MAX_TOTAL: raise ExportError("export size limit reached") (writer or _atomic)(blob, response.body, directory / "body.pending") total += len(response.body) else: _regular(blob) if blob.read_bytes() != response.body: raise ExportError("export body checksum mismatch") record.update(sha256=digest, bytes=len(response.body)) manifest["records"].append(record) manifest.update(index=index + int(advance), stage=next_stage, cursor=next_cursor, etag=next_etag, last_sequence=next_last, high_water=next_high) _save(directory, manifest, writer=writer) return manifest["index"] == len(selection)