Skip to content

feat(eval): export Pi conversation transcripts #3470

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
TomCC7 wants to merge 1 commit into from
Draft

feat(eval): export Pi conversation transcripts #3470

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
194 changes: 188 additions & 6 deletions dimos/benchmark/evaluation/pi_process.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import json
import os
from pathlib import Path
import queue
import subprocess
import threading
import time
Expand All @@ -37,6 +38,7 @@

PI_VERSION = "0.80.10"
MAX_STDERR_BYTES = 64 * 1024
EXPORT_TIMEOUT_SECONDS = 30.0


@dataclass(frozen=True)
Expand All @@ -49,9 +51,20 @@ class PiRunResult:


class PiRunError(RuntimeError):
def __init__(self, message: str, *, stderr: str = "") -> None:
def __init__(
self,
message: str,
*,
stderr: str = "",
transcript_path: Path | None = None,
) -> None:
super().__init__(message)
self.stderr = stderr
self.transcript_path = transcript_path


class PiExportError(RuntimeError):
pass


class PiCliRunner:
Expand Down Expand Up @@ -199,27 +212,191 @@ def read_stderr() -> None:
reader.join(timeout=1)

stderr = "".join(stderr_parts)
transcript_path = _latest_transcript(session_dir)
if timed_out:
raise PiRunError(
f"Pi timed out after {self.timeout_s:g}s",
stderr=stderr,
transcript_path=transcript_path,
)
duration = time.monotonic() - started
if process.returncode != 0:
raise PiRunError(f"Pi exited with status {process.returncode}", stderr=stderr)
raise PiRunError(
f"Pi exited with status {process.returncode}",
stderr=stderr,
transcript_path=transcript_path,
)
if events.stop_error is not None:
raise PiRunError(events.stop_error, stderr=stderr)
raise PiRunError(
events.stop_error,
stderr=stderr,
transcript_path=transcript_path,
)
if events.final_text is None:
raise PiRunError("Pi produced no final assistant response", stderr=stderr)
transcripts = sorted(session_dir.rglob("*.jsonl")) if session_dir.exists() else []
raise PiRunError(
"Pi produced no final assistant response",
stderr=stderr,
transcript_path=transcript_path,
)
return PiRunResult(
final_text=events.final_text,
tool_call_count=events.tool_count,
duration_seconds=duration,
transcript_path=transcripts[-1] if transcripts else None,
transcript_path=transcript_path,
stderr=stderr,
)

def export_transcript(
self,
*,
transcript_path: Path,
output_path: Path,
mcp_url: str,
api_key: str,
) -> None:
"""Export a saved session through Pi RPC so extension renderers are active."""
request_id = "dimos-transcript-export"
command = (
"node",
str(self.cli),
"--mode",
"rpc",
"--model",
f"openai/{self.model}",
"--thinking",
self.thinking_level,
"--session",
str(transcript_path),
"--no-builtin-tools",
"--tools",
"python_exec",
"--no-extensions",
"--extension",
str(self.extension),
"--no-skills",
"--no-prompt-templates",
"--no-themes",
"--no-context-files",
"--no-approve",
"--offline",
)
env = {
"PATH": os.environ.get("PATH", ""),
"OPENAI_API_KEY": api_key,
"DIMOS_CODE_POLICY_MCP_URL": mcp_url,
"PI_CODING_AGENT_DIR": str(transcript_path.parent / ".pi-agent-export"),
"PI_SKIP_VERSION_CHECK": "1",
"PI_TELEMETRY": "0",
}
process = subprocess.Popen(
command,
cwd=transcript_path.parent,
env=env,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
)
stdin = process.stdin
stdout = process.stdout
stderr_stream = process.stderr
assert stdin is not None
assert stdout is not None
assert stderr_stream is not None
responses: queue.Queue[dict[str, Any]] = queue.Queue()
stderr_parts: list[str] = []

def read_stdout() -> None:
for line in stdout:
try:
response = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(response, dict) and response.get("id") == request_id:
responses.put(response)

def read_stderr() -> None:
retained = 0
for line in stderr_stream:
line = line.replace(api_key, "[REDACTED]")
remaining = MAX_STDERR_BYTES - retained
if remaining <= 0:
continue
value = line.encode()[:remaining].decode(errors="ignore")
stderr_parts.append(value)
retained += len(value.encode())

readers = (
threading.Thread(target=read_stdout, name="pi-export-stdout", daemon=True),
threading.Thread(target=read_stderr, name="pi-export-stderr", daemon=True),
)
for reader in readers:
reader.start()
try:
stdin.write(
json.dumps(
{
"id": request_id,
"type": "export_html",
"outputPath": str(output_path),
}
)
+ "\n"
)
stdin.flush()
deadline = time.monotonic() + EXPORT_TIMEOUT_SECONDS
response = None
while response is None and time.monotonic() < deadline:
try:
response = responses.get(timeout=min(0.1, deadline - time.monotonic()))
except queue.Empty:
if process.poll() is not None:
break
if response is None:
for reader in readers:
reader.join(timeout=1)
detail = "".join(stderr_parts).strip()
suffix = f": {detail}" if detail else ""
if process.poll() is not None:
raise PiExportError(
f"Pi transcript exporter exited with status {process.returncode}{suffix}"
)
raise PiExportError(
f"Pi transcript export timed out after {EXPORT_TIMEOUT_SECONDS:g}s{suffix}"
)
if not response.get("success"):
message = str(response.get("error") or "Pi transcript export failed")
raise PiExportError(message.replace(api_key, "[REDACTED]"))
data = response.get("data")
exported = data.get("path") if isinstance(data, dict) else None
if not isinstance(exported, str) or Path(exported).resolve() != output_path.resolve():
raise PiExportError("Pi transcript export returned an unexpected output path")
finally:
stdin.close()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
for reader, stream in zip(readers, (stdout, stderr_stream), strict=True):
reader.join(timeout=1)
if reader.is_alive():
stream.close()
reader.join(timeout=1)
if process.returncode != 0:
detail = "".join(stderr_parts).strip()
suffix = f": {detail}" if detail else ""
raise PiExportError(
f"Pi transcript exporter exited with status {process.returncode}{suffix}"
)
if not output_path.is_file():
raise PiExportError("Pi transcript exporter produced no HTML file")


def parse_pi_events(stream: str) -> tuple[str | None, int, str | None]:
"""Return the final assistant text, tool count, and terminal error."""
Expand Down Expand Up @@ -307,3 +484,8 @@ def _bounded_stderr(value: str) -> str:
if len(encoded) <= MAX_STDERR_BYTES:
return value
return encoded[:MAX_STDERR_BYTES].decode(errors="ignore")


def _latest_transcript(session_dir: Path) -> Path | None:
transcripts = sorted(session_dir.rglob("*.jsonl")) if session_dir.exists() else []
return transcripts[-1] if transcripts else None
65 changes: 58 additions & 7 deletions dimos/benchmark/evaluation/runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@
from dimos.agents.code_policy_server import CodePolicyMcpServer
from dimos.benchmark.evaluation.models import ArtifactReference, RuntimeIdentity
from dimos.benchmark.evaluation.pi_process import PI_VERSION, PiCliRunner, PiRunError
from dimos.benchmark.evaluation.progress import ProgressSink
from dimos.benchmark.evaluation.progress import ProgressSink, StatusProgress, emit_progress
from dimos.benchmark.evaluation.protocol import (
DebugTrialSubmitter,
ExplorationOutcome,
Expand Down Expand Up @@ -144,13 +144,21 @@ def explore(
)
except PiRunError as exc:
self._record_text(path, relative, "stderr.log", exc.stderr, "Pi stderr")
raise
if result.transcript_path is not None:
target = path / "pi-transcript.jsonl"
shutil.copy2(result.transcript_path, target)
self._runtime_artifacts.append(
_artifact(relative / target.name, "Pi transcript", "application/x-ndjson")
self._record_pi_transcript(
path=path,
relative=relative,
source=exc.transcript_path,
runner=runner,
mcp_url=server.mcp_url,
)
raise
self._record_pi_transcript(
path=path,
relative=relative,
source=result.transcript_path,
runner=runner,
mcp_url=server.mcp_url,
)
if result.stderr:
self._record_text(path, relative, "stderr.log", result.stderr, "Pi stderr")
policy = manager.last_policy
Expand Down Expand Up @@ -293,6 +301,49 @@ def _record_text(
(path / filename).write_text(value, encoding="utf-8")
self._runtime_artifacts.append(_artifact(relative / filename, label, "text/plain"))

def _record_pi_transcript(
self,
*,
path: Path,
relative: Path,
source: Path | None,
runner: PiCliRunner,
mcp_url: str,
) -> None:
if source is None:
return
transcript = path / "pi-transcript.jsonl"
shutil.copy2(source, transcript)
self._runtime_artifacts.append(
_artifact(relative / transcript.name, "Pi transcript", "application/x-ndjson")
)
viewer = path / "pi-transcript.html"
try:
runner.export_transcript(
transcript_path=source,
output_path=viewer,
mcp_url=mcp_url,
api_key=self.api_key,
)
except Exception as exc:
viewer.unlink(missing_ok=True)
message = f"{type(exc).__name__}: {exc}".replace(self.api_key, "[REDACTED]")
emit_progress(
self.progress,
StatusProgress(channel="pi", message=f"transcript export failed: {message}"),
)
self._record_text(
path,
relative,
"pi-transcript-export-error.log",
message + "\n",
"Pi transcript export error",
)
return
self._runtime_artifacts.append(
_artifact(relative / viewer.name, "Pi transcript viewer", "text/html")
)


class _SubmissionManager:
def __init__(
Expand Down
Loading
Loading