72 lines
3.0 KiB
Python
72 lines
3.0 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Add explicit parent lineage to Hermes API-created sessions."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
|
||
|
|
BEFORE = ''' model = body.get("model") or self._model_name
|
||
|
|
system_prompt = body.get("system_prompt")
|
||
|
|
if system_prompt is not None and not isinstance(system_prompt, str):
|
||
|
|
return web.json_response(_openai_error("system_prompt must be a string", code="invalid_system_prompt"), status=400)
|
||
|
|
db.create_session(session_id, "api_server", model=str(model) if model else None, system_prompt=system_prompt)
|
||
|
|
'''
|
||
|
|
|
||
|
|
AFTER = ''' model = body.get("model") or self._model_name
|
||
|
|
system_prompt = body.get("system_prompt")
|
||
|
|
if system_prompt is not None and not isinstance(system_prompt, str):
|
||
|
|
return web.json_response(_openai_error("system_prompt must be a string", code="invalid_system_prompt"), status=400)
|
||
|
|
|
||
|
|
# API workers are first-class children of the objective that launched
|
||
|
|
# them. Accept a JSON field for normal clients and a header for thin
|
||
|
|
# relays that cannot extend their request schema.
|
||
|
|
metadata = body.get("metadata")
|
||
|
|
metadata_parent = metadata.get("parent_session_id") if isinstance(metadata, dict) else None
|
||
|
|
raw_parent = body.get("parent_session_id") or metadata_parent or request.headers.get(
|
||
|
|
"X-Hermes-Parent-Session-Id"
|
||
|
|
)
|
||
|
|
parent_session_id = str(raw_parent).strip() if raw_parent else None
|
||
|
|
if parent_session_id:
|
||
|
|
if (
|
||
|
|
len(parent_session_id) > self._MAX_SESSION_HEADER_LEN
|
||
|
|
or re.search(r'[\\r\\n\\x00]', parent_session_id)
|
||
|
|
or _is_path_unsafe(parent_session_id)
|
||
|
|
or parent_session_id == session_id
|
||
|
|
):
|
||
|
|
return web.json_response(_openai_error("Invalid parent session ID", code="invalid_parent_session_id"), status=400)
|
||
|
|
if not db.get_session(parent_session_id):
|
||
|
|
return web.json_response(_openai_error(f"Parent session not found: {parent_session_id}", code="parent_session_not_found"), status=404)
|
||
|
|
|
||
|
|
db.create_session(
|
||
|
|
session_id,
|
||
|
|
"api_server",
|
||
|
|
model=str(model) if model else None,
|
||
|
|
system_prompt=system_prompt,
|
||
|
|
parent_session_id=parent_session_id,
|
||
|
|
)
|
||
|
|
'''
|
||
|
|
|
||
|
|
|
||
|
|
def patch(source: Path, destination: Path) -> None:
|
||
|
|
"""Apply the narrow session-lineage extension and fail on upstream drift."""
|
||
|
|
content = source.read_text(encoding="utf-8")
|
||
|
|
if BEFORE not in content:
|
||
|
|
raise RuntimeError("Hermes API session patch context changed")
|
||
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
destination.write_text(content.replace(BEFORE, AFTER, 1), encoding="utf-8")
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
||
|
|
parser.add_argument("source", type=Path)
|
||
|
|
parser.add_argument("destination", type=Path)
|
||
|
|
args = parser.parse_args()
|
||
|
|
patch(args.source, args.destination)
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|