"""execution_trace.jsonl -> insan-okunabilir rapor (Faz 8). Bu modül `tracer.py`/`store.py` mimarisini DEĞİŞTİRMEZ: `ExecutionTraceEntry` JSONL'i hâlâ TEK source of truth'tur. Renderer onu yalnızca OKUR; hiçbir şekilde yazmaz/değiştirmez ve WorkflowState'i mutasyona uğratmaz - salt sunum (presentation) katmanıdır. `render_trace_jsonl`, diskteki bir trace dosyasını `TraceStore` ÜZERİNDEN okur (kendi JSONL parse mantığını icat ETMEZ - bkz. `_load_via_trace_store`). `TraceRenderContext`, `ExecutionTraceEntry` şemasında BULUNMAYAN (final karar, toplam iterasyon gibi) isteğe bağlı özet bilgiyi taşır; bu, trace şemasını GENİŞLETMEZ - yalnızca WorkflowState'e erişimi olan çağıranların (CLI/demo) daha okunabilir bir başlık eklemesine izin veren, tamamen opsiyonel bir sunum yardımcısıdır. Context olmadan da renderer tam işlevseldir. Güvenlik: yalnızca `ExecutionTraceEntry`'nin zaten taşıdığı kısa özetleri render eder (`input_context_summary`, `output_summary` vb.) - ham prompt, ham LLM response, tam dosya içeriği veya secret/API key BURADA YOKTUR çünkü zaten kaynak şemada (tracer.py, Faz 5) hiç bulunmuyorlar. HTML çıktısında tüm metin `html.escape()` ile kaçışlanır (harici bağımlılık veya JS yok). """ from __future__ import annotations import html as _html from dataclasses import dataclass from pathlib import Path from orchestrator.schemas.enums import Action from orchestrator.schemas.execution_trace import ExecutionTraceEntry from orchestrator.tracing.store import TraceStore @dataclass(frozen=True) class TraceRenderContext: """`ExecutionTraceEntry` JSONL'inde bulunmayan, isteğe bağlı özet bilgisi.""" run_id: str | None = None final_outcome: str | None = None final_summary: str | None = None total_iterations: int | None = None coding_revisions: int | None = None def _load_via_trace_store(jsonl_path: Path) -> list[ExecutionTraceEntry]: """`//execution_trace.jsonl` yolunu `TraceStore` ile okur. `TraceStore`'u bypass ETMEZ: run_id/runs_dir, dosya yolunun kendi (`TraceStore`'un ürettiği) dizin yapısından türetilip aynı sınıfa geri verilir. """ run_id = jsonl_path.parent.name runs_dir = jsonl_path.parent.parent return TraceStore(runs_dir, run_id).read_all() def _stage_counts(entries: list[ExecutionTraceEntry]) -> list[tuple[str, int]]: counts: dict[str, int] = {} order: list[str] = [] for entry in entries: key = f"{entry.stage.value} ({entry.agent_role.value})" if key not in counts: order.append(key) counts[key] = counts.get(key, 0) + 1 return [(key, counts[key]) for key in order] def _header_lines(context: TraceRenderContext | None) -> list[str]: if context is None: return [] lines = ["=" * 70] if context.run_id: lines.append(f"Run ID: {context.run_id}") if context.final_outcome: lines.append(f"Final Outcome: {context.final_outcome}") if context.final_summary: lines.append(f"Final Summary: {context.final_summary}") if context.total_iterations is not None: lines.append(f"Total Iterations: {context.total_iterations}") if context.coding_revisions is not None: lines.append(f"Coding Revisions: {context.coding_revisions}") lines.append("=" * 70) return lines def render_trace_text( entries: list[ExecutionTraceEntry], *, context: TraceRenderContext | None = None ) -> str: """`ExecutionTraceEntry` listesinden terminal-dostu, insan-okunabilir bir rapor üretir.""" lines: list[str] = [] lines.extend(_header_lines(context)) if not entries: lines.append("(execution trace boş)") return "\n".join(lines) stage_counts = _stage_counts(entries) lines.append("Stage/Agent call sayıları:") for key, count in stage_counts: lines.append(f" - {key}: {count}") lines.append("") for index, entry in enumerate(entries, start=1): status_marker = "OK" if entry.action == Action.COMPLETED else "FAIL" lines.append( f"[{index}] {entry.stage.value} | {entry.agent_role.value} | " f"{status_marker} | iter={entry.iteration} | {entry.duration_ms}ms" ) lines.append(f" in: {entry.input_context_summary}") lines.append(f" out: {entry.output_summary}") if entry.routing_decision_ref: lines.append(f" routing_ref: {entry.routing_decision_ref}") lines.append("") return "\n".join(lines).rstrip() + "\n" def render_trace_jsonl(path: Path, *, context: TraceRenderContext | None = None) -> str: """Diskteki bir `execution_trace.jsonl` dosyasını (TraceStore üzerinden) render eder.""" entries = _load_via_trace_store(path) return render_trace_text(entries, context=context) _HTML_STYLE = """ body { font-family: -apple-system, Segoe UI, Arial, sans-serif; margin: 2rem; background: #0b0f14; color: #e6edf3; } h1 { font-size: 1.3rem; } table.summary { border-collapse: collapse; margin-bottom: 1.5rem; } table.summary td { padding: 0.15rem 0.75rem 0.15rem 0; } .entry { border: 1px solid #30363d; border-radius: 6px; padding: 0.75rem 1rem; margin-bottom: 0.75rem; background: #161b22; } .entry.failed { border-color: #f85149; } .entry .title { font-weight: 600; margin-bottom: 0.35rem; } .entry .ok { color: #3fb950; } .entry .fail { color: #f85149; } .entry .row { font-family: Consolas, Menlo, monospace; font-size: 0.85rem; color: #c9d1d9; margin: 0.15rem 0; } """ def render_trace_html( entries: list[ExecutionTraceEntry], *, context: TraceRenderContext | None = None ) -> str: """`ExecutionTraceEntry` listesinden basit, dependency-free bir HTML raporu üretir.""" esc = _html.escape parts: list[str] = [ "", "", "Execution Trace Report", f"", "

Execution Trace Report

", ] if context is not None: parts.append("") if context.run_id: parts.append(f"") if context.final_outcome: parts.append(f"") if context.final_summary: parts.append(f"") if context.total_iterations is not None: parts.append(f"") if context.coding_revisions is not None: parts.append(f"") parts.append("

Run ID	{esc(context.run_id)}
Final Outcome	{esc(context.final_outcome)}
Final Summary	{esc(context.final_summary)}
Total Iterations	{context.total_iterations}
Coding Revisions	{context.coding_revisions}

") if not entries: parts.append("

(execution trace boş)

") return "\n".join(parts) parts.append("") for key, count in _stage_counts(entries): parts.append(f"") parts.append("

{esc(key)}

{count}

") for index, entry in enumerate(entries, start=1): is_ok = entry.action == Action.COMPLETED css_class = "entry" if is_ok else "entry failed" status_span = "OK" if is_ok else "FAIL" parts.append(f"

") parts.append( f"

[{index}] {esc(entry.stage.value)} | " f"{esc(entry.agent_role.value)} | {status_span} | " f"iter={entry.iteration} | {entry.duration_ms}ms

" ) parts.append(f"

in: {esc(entry.input_context_summary)}

") parts.append(f"

out: {esc(entry.output_summary)}

") if entry.routing_decision_ref: parts.append(f"

routing_ref: {esc(entry.routing_decision_ref)}

") parts.append("

") parts.append("") return "\n".join(parts) def write_html_report( entries: list[ExecutionTraceEntry], output_path: Path, *, context: TraceRenderContext | None = None, ) -> None: """`render_trace_html`'in sonucunu `output_path`'e yazan ince bir yardımcı.""" output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(render_trace_html(entries, context=context), encoding="utf-8")

v1.0.0 — GitHub import: MuradHasanov07/multi-agent-software-orchestrator			v1.1.0 — Editör commit — 6 ajan
		Expand 492 lines ...

Diff