| """Optional JSONL trace logging for sharing Borderless agent sessions."""
|
|
|
| from __future__ import annotations
|
|
|
| import json
|
| import os
|
| import time
|
| from pathlib import Path
|
| from typing import Any
|
|
|
| TRACE_DIR = Path(os.environ.get("BORDERLESS_TRACE_DIR", "agent_traces"))
|
| TRACE_RESULT_LIMIT = 6000
|
|
|
|
|
| def _compact_result(result: str) -> str:
|
| if len(result) <= TRACE_RESULT_LIMIT:
|
| return result
|
| return result[:TRACE_RESULT_LIMIT] + "\n... (truncated)"
|
|
|
|
|
| def record_tool_trace(
|
| *,
|
| tool_name: str,
|
| arguments: str,
|
| result: str,
|
| duration: float,
|
| ) -> None:
|
| """Write one tool call to JSONL so sessions can be shared after a demo."""
|
| if os.environ.get("BORDERLESS_DISABLE_TRACE_LOGS") == "1":
|
| return
|
|
|
| try:
|
| TRACE_DIR.mkdir(parents=True, exist_ok=True)
|
| path = TRACE_DIR / time.strftime("borderless-%Y%m%d.jsonl")
|
| payload: dict[str, Any] = {
|
| "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
| "tool_name": tool_name,
|
| "arguments": arguments,
|
| "duration_seconds": round(duration, 3),
|
| "result": _compact_result(result),
|
| }
|
| with path.open("a", encoding="utf-8") as handle:
|
| handle.write(json.dumps(payload, ensure_ascii=False) + "\n")
|
| except OSError:
|
|
|
| return
|
|
|
