gpt skill ergünzt

2026-08-26 16:54:32 +02:00
parent 844b4e5569
commit affde3a45f
3 changed files with 376 additions and 33 deletions
@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""Materialize Codex result files and normalize its JSONL measurements."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import Counter
+from datetime import datetime
+from pathlib import Path, PurePosixPath
+from typing import Any
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("run_directory", type=Path)
+    parser.add_argument("--model", required=True)
+    parser.add_argument("--effort", required=True)
+    return parser.parse_args()
+
+
+def read_jsonl(path: Path) -> tuple[list[dict[str, Any]], list[str]]:
+    events: list[dict[str, Any]] = []
+    malformed: list[str] = []
+    for number, line in enumerate(path.read_text(encoding="utf-8-sig").splitlines(), 1):
+        if not line.strip():
+            continue
+        try:
+            value = json.loads(line)
+        except json.JSONDecodeError as exc:
+            malformed.append(f"line {number}: {exc}")
+            continue
+        if isinstance(value, dict):
+            events.append(value)
+        else:
+            malformed.append(f"line {number}: JSON value is not an object")
+    return events, malformed
+
+
+def safe_result_path(results_dir: Path, raw_path: str) -> Path:
+    relative = PurePosixPath(raw_path.replace("\\", "/"))
+    if relative.is_absolute() or not relative.parts or ".." in relative.parts:
+        raise ValueError(f"unsafe result path: {raw_path!r}")
+    if any(part in ("", ".") or ":" in part for part in relative.parts):
+        raise ValueError(f"invalid result path: {raw_path!r}")
+    target = results_dir.joinpath(*relative.parts).resolve()
+    root = results_dir.resolve()
+    if root != target and root not in target.parents:
+        raise ValueError(f"result path escapes Ergebnisse: {raw_path!r}")
+    return target
+
+
+def parse_iso(path: Path) -> datetime | None:
+    if not path.exists():
+        return None
+    value = path.read_text(encoding="utf-8-sig").strip()
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def main() -> int:
+    args = parse_args()
+    run_dir = args.run_directory.resolve()
+    meta_dir = run_dir / "_meta"
+    results_dir = run_dir / "Ergebnisse"
+    events, malformed = read_jsonl(run_dir / "RawEvents.jsonl")
+    envelope = json.loads((meta_dir / "final_response.json").read_text(encoding="utf-8-sig"))
+    if not isinstance(envelope, dict) or not isinstance(envelope.get("files"), list):
+        raise ValueError("final_response.json does not match the Codex output envelope")
+
+    results_dir.mkdir(parents=True, exist_ok=True)
+    seen: set[str] = set()
+    materialized: list[str] = []
+    for entry in envelope["files"]:
+        if not isinstance(entry, dict):
+            raise ValueError("file entry is not an object")
+        raw_path = entry.get("path")
+        content = entry.get("content")
+        if not isinstance(raw_path, str) or not isinstance(content, str):
+            raise ValueError("file entry requires string path and content")
+        target = safe_result_path(results_dir, raw_path)
+        key = str(target).casefold()
+        if key in seen:
+            raise ValueError(f"duplicate result path: {raw_path!r}")
+        seen.add(key)
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(content, encoding="utf-8", newline="\n")
+        materialized.append(str(target.relative_to(results_dir)).replace("\\", "/"))
+
+    usage = Counter()
+    item_types = Counter()
+    errors: list[Any] = list(malformed)
+    thread_id = None
+    turns = 0
+    for event in events:
+        event_type = event.get("type")
+        if event_type == "thread.started":
+            thread_id = event.get("thread_id")
+        elif event_type == "turn.completed":
+            turns += 1
+            turn_usage = event.get("usage") or {}
+            for field in ("input_tokens", "cached_input_tokens", "output_tokens", "reasoning_output_tokens"):
+                value = turn_usage.get(field, 0)
+                if isinstance(value, int):
+                    usage[field] += value
+        elif event_type in ("turn.failed", "error"):
+            errors.append(event)
+        if event_type == "item.completed":
+            item = event.get("item") or {}
+            if isinstance(item.get("type"), str):
+                item_types[item["type"]] += 1
+
+    start = parse_iso(meta_dir / "startzeit.txt")
+    end = parse_iso(meta_dir / "endzeit.txt")
+    duration_ms = round((end - start).total_seconds() * 1000) if start and end else None
+
+    exit_code = None
+    exit_code_path = meta_dir / "exitcode.txt"
+    if exit_code_path.exists():
+        try:
+            exit_code = int(exit_code_path.read_text(encoding="utf-8-sig").strip())
+        except ValueError:
+            errors.append("invalid exitcode.txt")
+    if exit_code not in (None, 0):
+        errors.append({"exit_code": exit_code})
+
+    normalized = {
+        "adapter": "codex-cli",
+        "is_error": bool(errors),
+        "subtype": "success" if not errors else "error",
+        "session_id": thread_id,
+        "duration_ms": duration_ms,
+        "duration_api_ms": None,
+        "num_turns": turns,
+        "requested_model": args.model,
+        "actual_models": None,
+        "model_control": "not_verifiable_from_codex_exec_jsonl",
+        "effort": args.effort,
+        "usage": {
+            "input_tokens": usage["input_tokens"],
+            "cached_input_tokens": usage["cached_input_tokens"],
+            "output_tokens": usage["output_tokens"],
+            "reasoning_output_tokens": usage["reasoning_output_tokens"],
+            "total_tokens": usage["input_tokens"] + usage["output_tokens"],
+            "semantics": "cached_input_tokens is a subset of input_tokens and is not added again",
+        },
+        "item_counts": dict(sorted(item_types.items())),
+        "permission_denials": None,
+        "subagent_stats": {"spawned": 0, "source": "multi-agent disabled by configuration"},
+        "materialized_files": materialized,
+        "result": envelope.get("summary", ""),
+        "errors": errors,
+    }
+    (run_dir / "RawResult.json").write_text(
+        json.dumps(normalized, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+        newline="\n",
+    )
+    return 1 if errors else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())