#!/usr/bin/env python3
"""
Codex -> Langfuse notify hook.

Codex calls the configured notify command near the end of a turn. This script
uses that signal to incrementally read the matching Codex session JSONL file and
emit the new assistant/user/tool events to Langfuse.
"""

import json
import os
import re
import subprocess
import sys
import time
import hashlib
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urlparse


def codex_langfuse_config_path() -> Path:
    codex_dir = Path(os.environ.get("CODEX_HOME") or (Path.home() / ".codex"))
    return codex_dir / "langfuse" / "config.json"


def read_langfuse_base_url_from_config() -> Optional[str]:
    try:
        path = codex_langfuse_config_path()
        if not path.exists():
            return None
        data = json.loads(path.read_text(encoding="utf-8-sig"))
        if isinstance(data, dict):
            value = data.get("baseUrl") or data.get("host")
            if isinstance(value, str) and value.strip():
                return value.strip()
    except Exception:
        return None
    return None


def configure_langfuse_no_proxy() -> None:
    hosts = ["localhost", "127.0.0.1"]
    values = [
        *(os.environ.get(key) for key in ("LANGFUSE_HOST", "LANGFUSE_BASEURL", "CODEX_LANGFUSE_BASE_URL")),
        read_langfuse_base_url_from_config(),
    ]
    for value in values:
        if not value:
            continue
        parsed = urlparse(value if "://" in value else f"http://{value}")
        if parsed.hostname:
            hosts.append(parsed.hostname)
        if parsed.netloc:
            hosts.append(parsed.netloc)
    existing = []
    for key in ("NO_PROXY", "no_proxy"):
        existing.extend([item.strip() for item in os.environ.get(key, "").split(",") if item.strip()])
    merged = []
    for item in [*existing, *hosts]:
        if item and item not in merged:
            merged.append(item)
    if merged:
        value = ",".join(merged)
        os.environ["NO_PROXY"] = value
        os.environ["no_proxy"] = value


configure_langfuse_no_proxy()

try:
    from langfuse import Langfuse, propagate_attributes
except Exception:
    sys.exit(0)


CODEX_DIR = Path(os.environ.get("CODEX_HOME") or (Path.home() / ".codex"))
STATE_DIR = CODEX_DIR / "langfuse"
CONFIG_FILE = STATE_DIR / "config.json"
STATE_FILE = STATE_DIR / "state.json"
LOCK_FILE = STATE_DIR / "state.lock"
LOG_FILE = STATE_DIR / "codex_langfuse_notify.log"

DEBUG = os.environ.get("CODEX_LANGFUSE_DEBUG", "").lower() == "true"
MAX_CHARS = int(os.environ.get("CODEX_LANGFUSE_MAX_CHARS", "20000"))
METRICS_SCHEMA_VERSION = "1.1"
AGENT_NAME = "codex"
AGENT_TURN_NAME = "Codex Agent Turn"
REPO_CONTEXT_TTL_S = 30.0
REPO_CONTEXT_GIT_TIMEOUT_S = 0.5
_REPO_CONTEXT_CACHE: Dict[str, Tuple[float, Dict[str, Any]]] = {}


def log(level: str, message: str) -> None:
    try:
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        with open(LOG_FILE, "a", encoding="utf-8") as f:
            f.write(f"{ts} [{level}] {message}\n")
    except Exception:
        pass


def debug(message: str) -> None:
    if DEBUG:
        log("DEBUG", message)


def empty_repo_context(
    cwd: Optional[str] = None,
    cwd_source: str = "unavailable",
    available: bool = False,
    error: str = "",
    is_git_repo: bool = False,
) -> Dict[str, Any]:
    return {
        "cwd": str(cwd or ""),
        "cwd_source": cwd_source or "unavailable",
        "git_context_available": bool(available),
        "git_context_error": error or "",
        "is_git_repo": bool(is_git_repo),
        "git_root": "",
        "git_remote": "",
        "git_remote_host": "",
        "git_repo_owner": "",
        "git_repo_name": "",
        "git_repo_slug": "",
        "git_branch": "",
        "git_commit": "",
    }


def sanitize_git_remote(remote: Optional[str]) -> Dict[str, str]:
    empty = {
        "git_remote": "",
        "git_remote_host": "",
        "git_repo_owner": "",
        "git_repo_name": "",
        "git_repo_slug": "",
    }
    raw = str(remote or "").strip()
    if not raw:
        return empty

    try:
        if "://" in raw:
            parsed = urlparse(raw)
            host = parsed.hostname or ""
            parts = [part for part in parsed.path.strip("/").split("/") if part]
        else:
            scp_like = re.match(r"^[^@]+@([^:]+):(.+)$", raw)
            if scp_like:
                host = scp_like.group(1)
                parts = [part for part in scp_like.group(2).strip("/").split("/") if part]
            else:
                parsed = urlparse(f"ssh://{raw}")
                host = parsed.hostname or ""
                parts = [part for part in parsed.path.strip("/").split("/") if part]

        if len(parts) < 2 or not host:
            return empty
        owner = parts[-2]
        repo = parts[-1]
        if repo.endswith(".git"):
            repo = repo[:-4]
        if not owner or not repo:
            return empty
        slug = f"{owner}/{repo}"
        return {
            "git_remote": f"{host}/{slug}",
            "git_remote_host": host,
            "git_repo_owner": owner,
            "git_repo_name": repo,
            "git_repo_slug": slug,
        }
    except Exception:
        return empty


def run_git_command(cwd: str, args: List[str], timeout_s: float = REPO_CONTEXT_GIT_TIMEOUT_S) -> Tuple[bool, str, str]:
    try:
        result = subprocess.run(
            ["git", "-C", cwd, *args],
            text=True,
            capture_output=True,
            timeout=timeout_s,
            check=False,
        )
        return result.returncode == 0, (result.stdout or "").strip(), (result.stderr or "").strip()
    except subprocess.TimeoutExpired:
        return False, "", "timeout"
    except FileNotFoundError:
        return False, "", "git_unavailable"
    except Exception:
        return False, "", "git_error"


def collect_repo_context(cwd: Optional[str], cwd_source: str = "unavailable") -> Dict[str, Any]:
    cwd_text = str(cwd or "").strip()
    if not cwd_text:
        return empty_repo_context(error="missing_cwd")

    path = Path(cwd_text)
    if not path.exists():
        return empty_repo_context(cwd_text, cwd_source, available=False, error="missing_cwd")

    cache_key = str(path.resolve())
    cached = _REPO_CONTEXT_CACHE.get(cache_key)
    now = time.time()
    if cached and now - cached[0] <= REPO_CONTEXT_TTL_S:
        context = dict(cached[1])
        context["cwd_source"] = cwd_source or context.get("cwd_source") or "unavailable"
        return context

    ok, inside, error = run_git_command(cwd_text, ["rev-parse", "--is-inside-work-tree"])
    if not ok:
        reason = error if error in ("timeout", "git_unavailable") else ""
        if reason:
            context = empty_repo_context(cwd_text, cwd_source, available=False, error=reason)
        else:
            context = empty_repo_context(cwd_text, cwd_source, available=True, is_git_repo=False)
        _REPO_CONTEXT_CACHE[cache_key] = (now, dict(context))
        return context
    if inside.lower() != "true":
        context = empty_repo_context(cwd_text, cwd_source, available=True, is_git_repo=False)
        _REPO_CONTEXT_CACHE[cache_key] = (now, dict(context))
        return context

    context = empty_repo_context(cwd_text, cwd_source, available=True, is_git_repo=True)

    ok_root, root, root_error = run_git_command(cwd_text, ["rev-parse", "--show-toplevel"])
    if not ok_root and root_error in ("timeout", "git_unavailable"):
        context["git_context_available"] = False
        context["git_context_error"] = root_error
        _REPO_CONTEXT_CACHE[cache_key] = (now, dict(context))
        return context
    context["git_root"] = root

    ok_remote, remote, remote_error = run_git_command(cwd_text, ["config", "--get", "remote.origin.url"])
    if ok_remote and remote:
        remote_context = sanitize_git_remote(remote)
        context.update(remote_context)
        if not remote_context["git_remote"]:
            context["git_context_error"] = "remote_parse_failed"
    elif remote_error in ("timeout", "git_unavailable"):
        context["git_context_available"] = False
        context["git_context_error"] = remote_error

    ok_branch, branch, _branch_error = run_git_command(cwd_text, ["branch", "--show-current"])
    if ok_branch:
        context["git_branch"] = branch

    ok_commit, commit, _commit_error = run_git_command(cwd_text, ["rev-parse", "--short", "HEAD"])
    if ok_commit:
        context["git_commit"] = commit

    _REPO_CONTEXT_CACHE[cache_key] = (now, dict(context))
    return context


class FileLock:
    def __init__(self, path: Path, timeout_s: float = 2.0):
        self.path = path
        self.timeout_s = timeout_s
        self._fh = None

    def __enter__(self):
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        self._fh = open(self.path, "a+", encoding="utf-8")
        try:
            import fcntl

            deadline = time.time() + self.timeout_s
            while True:
                try:
                    fcntl.flock(self._fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                    break
                except BlockingIOError:
                    if time.time() > deadline:
                        break
                    time.sleep(0.05)
        except Exception:
            pass
        return self

    def __exit__(self, exc_type, exc, tb):
        try:
            import fcntl

            fcntl.flock(self._fh.fileno(), fcntl.LOCK_UN)
        except Exception:
            pass
        try:
            self._fh.close()
        except Exception:
            pass


@dataclass
class SessionState:
    offset: int = 0
    buffer: str = ""
    turn_count: int = 0
    uploaded_turn_ids: set = field(default_factory=set)
    uploaded_turn_fallback_keys: set = field(default_factory=set)


@dataclass
class CodexToolCall:
    id: str
    name: str
    input: Any = None
    output: Any = None
    start_time: Optional[str] = None
    end_time: Optional[str] = None
    status: Optional[str] = None
    error: Optional[str] = None


@dataclass
class CodexStep:
    start_time: Optional[str] = None
    end_time: Optional[str] = None
    assistant_text: str = ""
    reasoning: str = ""
    usage: Dict[str, Any] = field(default_factory=dict)
    tool_calls: List[CodexToolCall] = field(default_factory=list)


@dataclass
class CodexTurn:
    turn_id: Optional[str] = None
    start_time: Optional[str] = None
    end_time: Optional[str] = None
    completed: bool = False
    aborted: bool = False
    user_text: str = ""
    user_text_fallback: str = ""
    final_output: str = ""
    model: Optional[str] = None
    invocation_params: Dict[str, Any] = field(default_factory=dict)
    steps: List[CodexStep] = field(default_factory=list)
    total_usage: Dict[str, Any] = field(default_factory=dict)
    subagent_thread_ids: List[str] = field(default_factory=list)


@dataclass
class CodexParsedSession:
    session_meta: Dict[str, Any]
    turns: List[CodexTurn]


def read_json_if_exists(path: Path) -> Dict[str, Any]:
    try:
        if not path.exists():
            return {}
        text = path.read_text(encoding="utf-8-sig")
        if not text.strip():
            return {}
        value = json.loads(text)
        return value if isinstance(value, dict) else {}
    except Exception as e:
        debug(f"read_json_if_exists failed for {path}: {e}")
        return {}


def write_json_atomic(path: Path, obj: Dict[str, Any]) -> None:
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        tmp = path.with_suffix(".tmp")
        tmp.write_text(json.dumps(obj, indent=2, sort_keys=True), encoding="utf-8")
        os.replace(tmp, path)
    except Exception as e:
        debug(f"write_json_atomic failed for {path}: {e}")


def read_stdin_payload() -> Dict[str, Any]:
    try:
        text = sys.stdin.read()
        if not text.strip():
            return {}
        data = json.loads(text)
        return data if isinstance(data, dict) else {}
    except Exception as e:
        debug(f"stdin payload parse failed: {e}")
        return {}


def first_string(*values: Any) -> Optional[str]:
    for value in values:
        if isinstance(value, str) and value.strip():
            return value.strip()
    return None


def find_value(obj: Any, keys: Tuple[str, ...]) -> Optional[str]:
    if isinstance(obj, dict):
        for key, value in obj.items():
            if key in keys and isinstance(value, str) and value.strip():
                return value.strip()
            found = find_value(value, keys)
            if found:
                return found
    elif isinstance(obj, list):
        for item in obj:
            found = find_value(item, keys)
            if found:
                return found
    return None


def normalize_session_path(raw: Optional[str]) -> Optional[Path]:
    if not raw:
        return None
    try:
        path = Path(raw).expanduser()
        if path.exists() and path.is_file():
            return path.resolve()
    except Exception:
        return None
    return None


def latest_session_file() -> Optional[Path]:
    sessions_dir = CODEX_DIR / "sessions"
    if not sessions_dir.exists():
        return None
    newest: Optional[Path] = None
    newest_mtime = -1.0
    try:
        for path in sessions_dir.rglob("*.jsonl"):
            try:
                mtime = path.stat().st_mtime
            except Exception:
                continue
            if mtime > newest_mtime:
                newest = path
                newest_mtime = mtime
    except Exception as e:
        debug(f"latest_session_file failed: {e}")
    return newest


def resolve_session_file(payload: Dict[str, Any]) -> Optional[Path]:
    raw_path = first_string(
        find_value(payload, ("session_path", "sessionPath", "transcript_path", "transcriptPath", "rollout_path", "rolloutPath")),
        os.environ.get("CODEX_SESSION_PATH"),
    )
    session_path = normalize_session_path(raw_path)
    if session_path:
        return session_path

    session_id = first_string(
        find_value(payload, ("session_id", "sessionId", "conversation_id", "conversationId", "id")),
        os.environ.get("CODEX_SESSION_ID"),
    )
    if session_id:
        sessions_dir = CODEX_DIR / "sessions"
        try:
            matches = list(sessions_dir.rglob(f"*{session_id}*.jsonl"))
            if matches:
                return max(matches, key=lambda p: p.stat().st_mtime).resolve()
        except Exception as e:
            debug(f"session_id lookup failed: {e}")

    return latest_session_file()


def session_key(path: Path) -> str:
    return hashlib.sha256(str(path).encode("utf-8")).hexdigest()


def load_session_state(state: Dict[str, Any], key: str) -> SessionState:
    raw = state.get(key, {})
    if not isinstance(raw, dict):
        raw = {}
    return SessionState(
        offset=int(raw.get("offset", 0) or 0),
        buffer=str(raw.get("buffer", "") or ""),
        turn_count=int(raw.get("turn_count", 0) or 0),
        uploaded_turn_ids={str(item) for item in raw.get("uploaded_turn_ids", []) if item},
        uploaded_turn_fallback_keys={str(item) for item in raw.get("uploaded_turn_fallback_keys", []) if item},
    )


def save_session_state(state: Dict[str, Any], key: str, ss: SessionState) -> None:
    state[key] = {
        "offset": ss.offset,
        "buffer": ss.buffer,
        "turn_count": ss.turn_count,
        "uploaded_turn_ids": sorted(ss.uploaded_turn_ids),
        "uploaded_turn_fallback_keys": sorted(ss.uploaded_turn_fallback_keys),
        "updated": datetime.now(timezone.utc).isoformat(),
    }


def read_all_jsonl(path: Path) -> List[Dict[str, Any]]:
    out: List[Dict[str, Any]] = []
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            for line in f:
                raw = line.strip()
                if not raw:
                    continue
                try:
                    item = json.loads(raw)
                    if isinstance(item, dict):
                        out.append(item)
                except Exception as e:
                    debug(f"jsonl row parse failed: {e}")
    except Exception as e:
        debug(f"read_all_jsonl failed: {e}")
    return out


def read_new_jsonl(path: Path, ss: SessionState) -> Tuple[List[Dict[str, Any]], SessionState]:
    out: List[Dict[str, Any]] = []
    try:
        size = path.stat().st_size
        if ss.offset > size:
            ss = SessionState()
        with open(path, "rb") as f:
            f.seek(ss.offset)
            chunk = f.read().decode("utf-8", errors="replace")
            ss.offset = f.tell()
    except Exception as e:
        debug(f"read_new_jsonl failed: {e}")
        return out, ss

    text = ss.buffer + chunk
    if not text:
        return out, ss

    lines = text.splitlines(keepends=True)
    ss.buffer = ""
    if lines and not lines[-1].endswith(("\n", "\r")):
        ss.buffer = lines.pop()

    for line in lines:
        raw = line.strip()
        if not raw:
            continue
        try:
            item = json.loads(raw)
            if isinstance(item, dict):
                out.append(item)
        except Exception as e:
            debug(f"jsonl row parse failed: {e}")

    return out, ss


def extract_text(content: Any) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: List[str] = []
        for item in content:
            if isinstance(item, str):
                parts.append(item)
            elif isinstance(item, dict):
                text = item.get("text") or item.get("output_text") or item.get("input_text")
                if isinstance(text, str):
                    parts.append(text)
        return "\n".join(parts)
    if isinstance(content, dict):
        text = content.get("text") or content.get("message")
        return text if isinstance(text, str) else ""
    return ""


def truncate(value: Any, max_chars: int = MAX_CHARS) -> Tuple[Any, Dict[str, Any]]:
    if not isinstance(value, str):
        try:
            text = json.dumps(value, ensure_ascii=False)
        except Exception:
            text = str(value)
    else:
        text = value

    orig_len = len(text)
    if orig_len <= max_chars:
        return value if isinstance(value, str) else value, {"truncated": False, "orig_len": orig_len}
    kept = text[:max_chars]
    return kept, {
        "truncated": True,
        "orig_len": orig_len,
        "kept_len": len(kept),
        "sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
    }


def build_interaction_id(source: str, session_id: str, turn_number: int) -> str:
    return f"{source or 'unknown'}:{session_id or 'unknown'}:{int(turn_number or 0)}"


def _num_or_none(value: Any) -> Optional[int]:
    if isinstance(value, bool):
        return None
    if isinstance(value, int) and value >= 0:
        return value
    if isinstance(value, float) and value >= 0:
        return int(value)
    if isinstance(value, str):
        try:
            n = int(value)
            return n if n >= 0 else None
        except Exception:
            return None
    return None


def _first_num(raw: Dict[str, Any], *keys: str) -> Optional[int]:
    for key in keys:
        if key in raw:
            value = _num_or_none(raw.get(key))
            if value is not None:
                return value
    return None


def normalize_token_metrics(raw: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    if not isinstance(raw, dict) or not raw:
        return {
            "token_metrics_available": False,
            "input_tokens": None,
            "output_tokens": None,
            "total_tokens": None,
            "cache_read_tokens": None,
            "reasoning_tokens": None,
        }
    input_tokens = _first_num(raw, "input", "input_tokens", "inputTokens")
    output_tokens = _first_num(raw, "output", "output_tokens", "outputTokens")
    total_tokens = _first_num(raw, "total", "total_tokens", "totalTokens")
    if total_tokens is None and input_tokens is not None and output_tokens is not None:
        total_tokens = input_tokens + output_tokens
    cache_read_tokens = _first_num(raw, "cache_read_tokens", "cachedInputTokens", "cacheRead")
    reasoning_tokens = _first_num(raw, "reasoning_tokens", "reasoningTokens", "reasoning")
    available = any(v is not None for v in [input_tokens, output_tokens, total_tokens, cache_read_tokens, reasoning_tokens])
    return {
        "token_metrics_available": available,
        "input_tokens": input_tokens if available else None,
        "output_tokens": output_tokens if available else None,
        "total_tokens": total_tokens if available else None,
        "cache_read_tokens": cache_read_tokens if available else None,
        "reasoning_tokens": reasoning_tokens if available else None,
    }


def _ratio(numerator: Optional[int], denominator: Optional[int]) -> Optional[float]:
    if numerator is None or denominator in (None, 0):
        return None
    return numerator / denominator


def build_interaction_metadata(
    source: str,
    user_id: Optional[str],
    session_id: str,
    turn_number: int,
    token_metrics: Optional[Dict[str, Any]],
    tool_call_count: int,
    tool_result_count: int,
    skill_use_count: int,
    model: Optional[str],
    user_message_count: int = 1,
    assistant_message_count: int = 1,
    skill_use_events: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    tokens = normalize_token_metrics(token_metrics)
    interaction_id = build_interaction_id(source, session_id, turn_number)
    events = list(skill_use_events or [])
    skill_names_all = [str(event.get("skill_name") or "") for event in events if event.get("skill_name")]
    unique_skill_names = list(dict.fromkeys(skill_names_all))
    skill_invocation_modes = [str(event.get("skill_invocation_mode") or "") for event in events if event.get("skill_invocation_mode")]
    skill_agent_paths = [str(event.get("skill_agent_path") or "") for event in events if event.get("skill_agent_path")]
    effective_skill_count = len(events) if events else int(skill_use_count or 0)
    return {
        "source": source,
        "agent": source,
        "user_id": user_id or "",
        "session_id": session_id,
        "interaction_id": interaction_id,
        "metrics_schema_version": METRICS_SCHEMA_VERSION,
        "interaction_count": 1,
        "user_message_count": user_message_count,
        "assistant_message_count": assistant_message_count,
        "tool_call_count": int(tool_call_count or 0),
        "tool_result_count": int(tool_result_count or 0),
        "skill_use_count": effective_skill_count,
        "unique_skill_count": len(unique_skill_names),
        "repeated_skill_count": max(0, effective_skill_count - len(unique_skill_names)),
        **tokens,
        "model": model,
        "turn_number": int(turn_number or 0),
        "efficiency": {
            "tokens_per_interaction": tokens.get("total_tokens"),
            "tool_calls_per_interaction": int(tool_call_count or 0),
            "skills_per_interaction": effective_skill_count,
            "output_input_token_ratio": _ratio(tokens.get("output_tokens"), tokens.get("input_tokens")),
            "tokens_per_tool_call": _ratio(tokens.get("total_tokens"), int(tool_call_count or 0)),
        },
        **({
            "skill_names": unique_skill_names,
            "skill_names_all": skill_names_all,
            "skill_invocation_modes": skill_invocation_modes,
            "skill_agent_paths": skill_agent_paths,
        } if events else {}),
    }


def discover_known_skills(extra_roots: Optional[List[Path]] = None) -> set:
    roots = [
        CODEX_DIR / "skills",
        CODEX_DIR / "plugins" / "cache",
        Path.home() / ".claude" / "skills",
        Path.home() / ".config" / "opencode" / "skill",
    ]
    if extra_roots:
        roots.extend(extra_roots)
    names = set()
    for root in roots:
        try:
            if not root.exists():
                continue
            for skill_file in root.rglob("SKILL.md"):
                names.add(skill_file.parent.name)
        except Exception:
            continue
    return names


def _skill_namespace(name: str) -> str:
    return name.split(":", 1)[0] if ":" in name else ""


def _skill_agent_from_interaction_id(interaction_id: str) -> str:
    return str(interaction_id or "unknown").split(":", 1)[0] or "unknown"


def _skill_agent_path(agent: str, detected_by: str) -> str:
    if agent == "codex":
        if detected_by == "codex_explicit_injection":
            return "codex_native_skill_injection"
        if detected_by == "codex_implicit_script":
            return "codex_skill_script_exec"
        if detected_by == "codex_implicit_doc_read":
            return "codex_skill_doc_read"
        if detected_by == "codex_skill_tool":
            return "codex_skill_tool_invoke"
        if detected_by == "codex_skill_doc_read":
            return "codex_skill_doc_read_tool"
        if detected_by == "tool_call":
            return "codex_unsupported_skill_tool"
    if detected_by == "skill_file_path":
        return "skill_file_path"
    return detected_by or "metadata"


def _skill_invocation_mode(agent: str, detected_by: str) -> str:
    if agent == "codex" and detected_by == "codex_explicit_injection":
        return "explicit"
    if agent == "codex" and detected_by in ("codex_implicit_script", "codex_implicit_doc_read", "codex_skill_tool", "codex_skill_doc_read"):
        return "implicit"
    return "detected"


def _skill_event_type(detected_by: str, agent: str = "unknown") -> str:
    if agent == "codex" and detected_by == "tool_call":
        return "detected"
    return "invoked" if detected_by in ("codex_explicit_injection", "codex_implicit_script", "codex_implicit_doc_read", "codex_skill_tool", "codex_skill_doc_read") else "detected"


def _skill_id_segment(name: str) -> str:
    segment = re.sub(r"[^A-Za-z0-9_.:-]+", "-", str(name or "").strip()).strip("-")
    return (segment or "unknown")[:96]


def _skill_usage(name: str, detected_by: str, skill_call_id: str = "") -> Dict[str, str]:
    clean = str(name or "").strip()
    return {
        "name": clean,
        "skill_namespace": _skill_namespace(clean),
        "detected_by": detected_by,
        "skill_call_id": str(skill_call_id or "").strip(),
    }


_CODEX_SCRIPT_RUNNERS = {"python", "python3", "bash", "zsh", "sh", "node", "deno", "ruby", "perl", "pwsh"}
_CODEX_DOC_READERS = {"cat", "sed", "head", "tail", "less", "more", "bat", "awk"}


def _command_from_tool_input(input_obj: Any) -> str:
    if isinstance(input_obj, dict):
        command = input_obj.get("command")
        if isinstance(command, str) and command.strip():
            return command.strip()
        argv = input_obj.get("cmd") or input_obj.get("argv")
        if isinstance(argv, list):
            return " ".join(str(item) for item in argv)
    if isinstance(input_obj, str):
        return input_obj.strip()
    return ""


def _first_command_token(command: str) -> str:
    match = re.match(r"\s*(?:[A-Za-z]:)?[^\"'\s]*?([^\\/\"'\s]+)(?:\.(?:exe|cmd))?(?=\s|$)", command, re.IGNORECASE)
    return match.group(1).lower() if match else ""


def _accept_known_skill(name: str, known_skills: set) -> str:
    clean = str(name or "").strip()
    return clean if clean and (clean in known_skills or not known_skills) else ""


def _detect_codex_skill_command(command: str, known_skills: set) -> List[Dict[str, str]]:
    found: List[Dict[str, str]] = []
    if not command:
        return found

    first = _first_command_token(command)
    if first in _CODEX_SCRIPT_RUNNERS:
        script_pattern = r"[\\/](?:skills|skill)[\\/]([^\\/\"'\s]+)[\\/]scripts[\\/][^\"'\s]+\.(?:py|sh|js|ts|rb|pl|ps1)(?=$|[\"'\s])"
        for match in re.finditer(script_pattern, command, re.IGNORECASE):
            name = _accept_known_skill(match.group(1), known_skills)
            if name:
                found.append(_skill_usage(name, "codex_implicit_script"))

    if first in _CODEX_DOC_READERS:
        doc_pattern = r"[\\/](?:skills|skill)[\\/]([^\\/\"'\s]+)[\\/]SKILL\.md(?=$|[\"'\s])"
        for match in re.finditer(doc_pattern, command, re.IGNORECASE):
            name = _accept_known_skill(match.group(1), known_skills)
            if name:
                found.append(_skill_usage(name, "codex_implicit_doc_read"))

    return found


def detect_skill_usages(tool_calls: List[Dict[str, Any]], known_skills: set) -> List[Dict[str, str]]:
    found: List[Dict[str, str]] = []
    for call in tool_calls or []:
        call_name = str(call.get("name") or "").strip().lower()
        input_obj = call.get("input") if isinstance(call.get("input"), (dict, list, str)) else {}

        if call_name == "skill":
            skill_input_name = None
            if isinstance(input_obj, dict):
                skill_input_name = str(input_obj.get("name") or "").strip()
            if skill_input_name:
                name = _accept_known_skill(skill_input_name, known_skills)
                if name:
                    found.append(_skill_usage(name, "codex_skill_tool"))
            continue

        if call_name == "read":
            if isinstance(input_obj, dict):
                file_path = str(input_obj.get("filePath") or "").strip()
                if file_path:
                    doc_pattern = r"[\\/](?:skills|skill)[\\/]([^\\/\"'\s]+)[\\/]SKILL\.md$"
                    match = re.search(doc_pattern, file_path, re.IGNORECASE)
                    if match:
                        name = _accept_known_skill(match.group(1), known_skills)
                        if name:
                            found.append(_skill_usage(name, "codex_skill_doc_read"))
            continue

        found.extend(_detect_codex_skill_command(_command_from_tool_input(input_obj), known_skills))
    return found


def _detect_codex_explicit_injections(material: Dict[str, Any], known_skills: set) -> List[Dict[str, str]]:
    found: List[Dict[str, str]] = []
    sources = [material.get("user_text"), *(material.get("skill_detection_sources") or [])]
    for source in sources:
        text = extract_text(source) if not isinstance(source, str) else source
        if not text:
            try:
                text = json.dumps(source, ensure_ascii=False)
            except Exception:
                text = str(source)
        for match in re.finditer(r"<skill>\s*<name>\s*([^<\s]+)\s*</name>.*?</skill>", text, re.IGNORECASE | re.DOTALL):
            name = _accept_known_skill(match.group(1), known_skills)
            if name:
                found.append(_skill_usage(name, "codex_explicit_injection"))
    return found


def _dedupe_turn_skill_usages(usages: List[Dict[str, str]]) -> List[Dict[str, str]]:
    out: List[Dict[str, str]] = []
    seen_call_ids: set = set()
    seen_detected: set = set()
    for usage in usages or []:
        name = str(usage.get("name") or "").strip()
        if not name:
            continue
        call_id = str(usage.get("skill_call_id") or "").strip()
        if call_id:
            key = f"call:{call_id}"
            if key in seen_call_ids:
                continue
            seen_call_ids.add(key)
            out.append(usage)
            continue
        detected_by = str(usage.get("detected_by") or "")
        if detected_by == "skill_file_path":
            key = f"{name}:{detected_by}"
            if key in seen_detected:
                continue
            seen_detected.add(key)
        out.append(usage)
    return out


def detect_turn_skill_usages(material: Dict[str, Any], known_skills: set) -> List[Dict[str, str]]:
    found = list(_detect_codex_explicit_injections(material, known_skills))
    found.extend(detect_skill_usages(material.get("tool_calls") or [], known_skills))
    return _dedupe_turn_skill_usages(found)


def build_skill_use_events(interaction_id: str, skill_usages: List[Dict[str, str]]) -> List[Dict[str, Any]]:
    events: List[Dict[str, Any]] = []
    deduped: List[Dict[str, str]] = []
    seen_call_ids: set = set()
    agent = _skill_agent_from_interaction_id(interaction_id)
    for skill in skill_usages or []:
        call_id = str(skill.get("skill_call_id") or "").strip()
        if call_id:
            dedupe_key = f"call:{call_id}"
            if dedupe_key in seen_call_ids:
                continue
            seen_call_ids.add(dedupe_key)
        deduped.append(skill)
    total = len(deduped)
    for index, skill in enumerate(deduped, start=1):
        name = str(skill.get("name") or "").strip()
        if not name:
            continue
        detected_by = str(skill.get("detected_by") or "metadata")
        call_id = str(skill.get("skill_call_id") or "").strip()
        invocation_mode = _skill_invocation_mode(agent, detected_by)
        events.append({
            "skill_use_id": f"{interaction_id}:skill:{index}:{_skill_id_segment(name)}",
            "skill_use_index": index,
            "skill_use_count_in_interaction": total,
            "skill_event_type": _skill_event_type(detected_by, agent),
            "skill_trigger": invocation_mode,
            "skill_invocation_mode": invocation_mode,
            "skill_agent_path": _skill_agent_path(agent, detected_by),
            "skill_name": name,
            "skill_use_count": 1,
            "skill_namespace": skill.get("skill_namespace") or _skill_namespace(name),
            "detected_by": detected_by,
            **({"skill_call_id": call_id} if call_id else {}),
        })
    return events


def summarize_skill_usages(skill_usages: List[Dict[str, str]]) -> List[Dict[str, Any]]:
    summary: Dict[str, Dict[str, Any]] = {}
    for item in skill_usages or []:
        name = item.get("name")
        if not name:
            continue
        entry = summary.setdefault(name, {"name": name, "count": 0, "detected_by": item.get("detected_by")})
        entry["count"] += 1
        detected_by = str(item.get("detected_by") or "metadata")
        entry.setdefault("skill_invocation_mode", _skill_invocation_mode("codex", detected_by))
        entry.setdefault("skill_agent_path", _skill_agent_path("codex", detected_by))
    return list(summary.values())


def is_codex_wrapper_text(text: str) -> bool:
    stripped = (text or "").strip()
    return stripped.startswith("<environment_context") or stripped.startswith("<user_instructions")


def append_text(existing: str, text: str) -> str:
    text = text or ""
    if not text:
        return existing or ""
    return f"{existing}\n{text}" if existing else text


def parse_tool_input(value: Any) -> Any:
    if isinstance(value, str):
        try:
            return json.loads(value)
        except Exception:
            return value
    return value


def get_payload(row: Dict[str, Any]) -> Dict[str, Any]:
    payload = row.get("payload")
    return payload if isinstance(payload, dict) else {}


def get_session_meta(rows: List[Dict[str, Any]], session_path: Path) -> Dict[str, Any]:
    meta: Dict[str, Any] = {"session_path": str(session_path)}
    for row in rows:
        if row.get("type") == "session_meta":
            payload = get_payload(row)
            if payload:
                meta.update(payload)
    return meta


def extract_usage(rows: List[Dict[str, Any]]) -> Dict[str, Any]:
    usage: Dict[str, Any] = {}
    for row in rows:
        payload = get_payload(row)
        if row.get("type") == "event_msg" and payload.get("type") == "token_count":
            info = payload.get("info")
            if isinstance(info, dict):
                last = info.get("last_token_usage")
                total = info.get("total_token_usage")
                if isinstance(last, dict):
                    usage["last_token_usage"] = last
                if isinstance(total, dict):
                    usage["total_token_usage"] = total
    return usage


def parse_codex_session(rows: List[Dict[str, Any]]) -> CodexParsedSession:
    session_meta: Dict[str, Any] = {"session_path": ""}
    turns: List[CodexTurn] = []
    turn: Optional[CodexTurn] = None
    step: Optional[CodexStep] = None
    tool_calls_by_id: Dict[str, CodexToolCall] = {}
    last_ts: Optional[str] = None

    def ensure_turn(ts: Optional[str]) -> CodexTurn:
        nonlocal turn
        if turn is None:
            turn = CodexTurn(start_time=ts, end_time=ts)
        return turn

    def ensure_step(ts: Optional[str]) -> CodexStep:
        nonlocal step
        if step is None:
            step = CodexStep(start_time=ts, end_time=ts)
        return step

    def close_step(ts: Optional[str], usage: Optional[Dict[str, Any]] = None) -> None:
        nonlocal step
        if step is None or turn is None:
            return
        step.end_time = ts or step.end_time
        if isinstance(usage, dict):
            step.usage = usage
        turn.steps.append(step)
        step = None

    def finish_turn(ts: Optional[str], completed: bool, aborted: bool) -> None:
        nonlocal turn, step, tool_calls_by_id
        if turn is None:
            return
        close_step(ts)
        turn.end_time = ts or turn.end_time
        turn.completed = completed
        turn.aborted = aborted
        if not turn.user_text:
            turn.user_text = turn.user_text_fallback
        if not turn.final_output:
            for candidate in reversed(turn.steps):
                if candidate.assistant_text:
                    turn.final_output = candidate.assistant_text
                    break
        turns.append(turn)
        turn = None
        step = None
        tool_calls_by_id = {}

    for row in rows:
        ts = str(row.get("timestamp") or last_ts or "")
        last_ts = ts
        row_type = row.get("type")
        payload = get_payload(row)

        if row_type == "session_meta":
            session_meta.update(payload)
            continue

        if row_type == "turn_context":
            t = ensure_turn(ts)
            model = payload.get("model")
            if isinstance(model, str) and model.strip():
                t.model = model.strip()
            if isinstance(payload, dict):
                t.invocation_params = dict(payload)
            continue

        if row_type == "response_item":
            item_type = payload.get("type")
            t = ensure_turn(ts)
            if item_type == "message":
                role = payload.get("role")
                text = extract_text(payload.get("content"))
                if role == "assistant" and text:
                    s = ensure_step(ts)
                    s.assistant_text = append_text(s.assistant_text, text)
                elif role == "user" and text and not is_codex_wrapper_text(text) and not t.user_text_fallback:
                    t.user_text_fallback = text
            elif item_type == "reasoning":
                text = extract_text(payload.get("content") or payload.get("summary"))
                if text:
                    s = ensure_step(ts)
                    s.reasoning = append_text(s.reasoning, text)
            elif item_type in ("function_call", "custom_tool_call"):
                call_id = first_string(str(payload.get("call_id") or ""), str(payload.get("id") or "")) or ""
                call = CodexToolCall(
                    id=call_id,
                    name=first_string(str(payload.get("name") or "")) or "tool",
                    input=parse_tool_input(payload.get("arguments") if item_type == "function_call" else payload.get("input")),
                    start_time=ts,
                )
                ensure_step(ts).tool_calls.append(call)
                if call_id:
                    tool_calls_by_id[call_id] = call
            elif item_type in ("function_call_output", "custom_tool_call_output"):
                call_id = str(payload.get("call_id") or "")
                call = tool_calls_by_id.get(call_id)
                if call:
                    call.output = payload.get("output")
                    call.end_time = ts
            continue

        if row_type != "event_msg":
            continue

        event_type = payload.get("type")
        if event_type == "task_started":
            if turn is not None:
                finish_turn(ts, completed=False, aborted=False)
            turn = CodexTurn(
                turn_id=first_string(str(payload.get("turn_id") or "")),
                start_time=ts,
                end_time=ts,
            )
            step = None
            tool_calls_by_id = {}
            continue

        t = ensure_turn(ts)
        if event_type == "user_message" and isinstance(payload.get("message"), str):
            if not t.user_text:
                t.user_text = payload["message"]
        elif event_type == "agent_message" and isinstance(payload.get("message"), str):
            t.final_output = payload["message"]
        elif event_type == "token_count":
            info = payload.get("info")
            if isinstance(info, dict):
                total = info.get("total_token_usage")
                if isinstance(total, dict):
                    t.total_usage = total
                last = info.get("last_token_usage")
                close_step(ts, last if isinstance(last, dict) else None)
        elif event_type == "task_complete":
            finish_turn(ts, completed=True, aborted=False)
        elif event_type == "turn_aborted":
            finish_turn(ts, completed=True, aborted=True)
        else:
            if event_type == "collab_agent_spawn_end" and isinstance(payload.get("new_thread_id"), str):
                t.subagent_thread_ids.append(payload["new_thread_id"])
            call_id = payload.get("call_id")
            if isinstance(call_id, str) and isinstance(event_type, str) and event_type.endswith("_end"):
                call = tool_calls_by_id.get(call_id)
                if call:
                    call.end_time = ts
                    if isinstance(payload.get("status"), str):
                        call.status = payload["status"]
                    if payload.get("status") in ("failed", "declined"):
                        call.error = first_string(
                            str(payload.get("stderr") or ""),
                            str(payload.get("error") or ""),
                            str(payload.get("message") or ""),
                        )
                    if call.output is None:
                        call.output = {
                            "stdout": payload.get("stdout"),
                            "stderr": payload.get("stderr"),
                            "success": payload.get("success"),
                            "status": payload.get("status"),
                            "aggregated_output": payload.get("aggregated_output"),
                        }
                    elif isinstance(call.output, str) and payload.get("stdout") is not None:
                        call.output = {
                            "stdout": payload.get("stdout"),
                            "stderr": payload.get("stderr"),
                            "success": payload.get("success"),
                            "status": payload.get("status"),
                            "aggregated_output": payload.get("aggregated_output"),
                        }

    if turn is not None:
        finish_turn(last_ts, completed=False, aborted=False)

    return CodexParsedSession(session_meta=session_meta, turns=turns)


def usage_details_from_codex(usage: Dict[str, Any]) -> Dict[str, int]:
    raw = usage.get("last_token_usage")
    if not isinstance(raw, dict):
        return {}
    out: Dict[str, int] = {}
    mapping = {
        "input_tokens": "input",
        "output_tokens": "output",
        "cached_input_tokens": "cache_read_input_tokens",
        "reasoning_output_tokens": "reasoning_output_tokens",
    }
    for src, dst in mapping.items():
        value = raw.get(src)
        if isinstance(value, int) and value >= 0:
            out[dst] = value
    return out


def collect_turn_material(rows: List[Dict[str, Any]]) -> Dict[str, Any]:
    user_texts: List[str] = []
    assistant_texts: List[str] = []
    tool_calls: List[Dict[str, Any]] = []
    tool_results: List[Dict[str, Any]] = []
    skill_detection_sources: List[Any] = []

    for row in rows:
        row_type = row.get("type")
        payload = get_payload(row)
        skill_detection_sources.append(payload or row)

        if row_type == "response_item":
            item_type = payload.get("type")
            if item_type == "message":
                role = payload.get("role")
                text = extract_text(payload.get("content"))
                if text:
                    if role == "user":
                        user_texts.append(text)
                    elif role == "assistant":
                        assistant_texts.append(text)
            elif item_type == "function_call":
                tool_calls.append(
                    {
                        "id": payload.get("call_id") or payload.get("id") or "",
                        "name": payload.get("name") or "tool",
                        "input": payload.get("arguments") or payload.get("input") or {},
                    }
                )

        if row_type == "event_msg":
            event_type = payload.get("type")
            if event_type == "agent_message" and isinstance(payload.get("message"), str):
                assistant_texts.append(payload["message"])
            elif isinstance(event_type, str) and event_type.endswith("_end"):
                tool_results.append(
                    {
                        "id": payload.get("call_id") or "",
                        "name": event_type,
                        "output": {
                            "stdout": payload.get("stdout"),
                            "stderr": payload.get("stderr"),
                            "success": payload.get("success"),
                            "status": payload.get("status"),
                            "aggregated_output": payload.get("aggregated_output"),
                        },
                    }
                )

    return {
        "user_text": "\n\n".join(user_texts[-3:]),
        "assistant_text": "\n\n".join(assistant_texts),
        "tool_calls": tool_calls,
        "tool_results": tool_results,
        "skill_detection_sources": skill_detection_sources,
    }


def codex_turn_usage_details(turn: CodexTurn) -> Dict[str, int]:
    for step in reversed(turn.steps):
        details = usage_details_from_codex({"last_token_usage": step.usage})
        if details:
            return details
    return {}


def codex_turn_material(turn: CodexTurn) -> Dict[str, Any]:
    tool_calls: List[Dict[str, Any]] = []
    tool_results: List[Dict[str, Any]] = []
    skill_sources: List[Any] = [turn.invocation_params]
    for step in turn.steps:
        skill_sources.append(step.reasoning)
        for call in step.tool_calls:
            tool_calls.append({"id": call.id, "name": call.name, "input": call.input})
            tool_results.append({
                "id": call.id,
                "name": call.name,
                "output": call.output,
                "status": call.status,
                "error": call.error,
                "start_time": call.start_time,
                "end_time": call.end_time,
            })
            skill_sources.append({"name": call.name, "input": call.input, "output": call.output})
    return {
        "user_text": turn.user_text,
        "assistant_text": turn.final_output or "\n\n".join(step.assistant_text for step in turn.steps if step.assistant_text),
        "tool_calls": tool_calls,
        "tool_results": tool_results,
        "skill_detection_sources": skill_sources,
    }


def codex_turn_upload_key(session_path: Path, turn: CodexTurn, index: int) -> Tuple[str, str]:
    if turn.turn_id:
        return "turn_id", str(turn.turn_id)
    seed = {
        "session_path": str(session_path),
        "index": index,
        "start_time": turn.start_time,
        "end_time": turn.end_time,
        "user_text": turn.user_text,
        "final_output": turn.final_output,
        "aborted": turn.aborted,
    }
    digest = hashlib.sha256(json.dumps(seed, ensure_ascii=False, sort_keys=True).encode("utf-8")).hexdigest()
    return "fallback", digest


def should_upload_codex_turn(ss: SessionState, session_path: Path, turn: CodexTurn, index: int) -> bool:
    if not (turn.completed or turn.aborted):
        return False
    material = codex_turn_material(turn)
    if not material.get("assistant_text") and not material.get("tool_calls") and not material.get("tool_results"):
        return False
    kind, value = codex_turn_upload_key(session_path, turn, index)
    if kind == "turn_id":
        return value not in ss.uploaded_turn_ids
    return value not in ss.uploaded_turn_fallback_keys


def mark_codex_turn_uploaded(ss: SessionState, session_path: Path, turn: CodexTurn, index: int) -> None:
    kind, value = codex_turn_upload_key(session_path, turn, index)
    if kind == "turn_id":
        ss.uploaded_turn_ids.add(value)
    else:
        ss.uploaded_turn_fallback_keys.add(value)


def emit_codex_turn(
    langfuse: Langfuse,
    session_id: str,
    user_id: Optional[str],
    turn_num: int,
    session_path: Path,
    meta: Dict[str, Any],
    turn: CodexTurn,
) -> None:
    material = codex_turn_material(turn)
    user_text, user_meta = truncate(material.get("user_text") or "")
    assistant_text, assistant_meta = truncate(material.get("assistant_text") or "")
    usage_details = codex_turn_usage_details(turn)
    model = first_string(turn.model, meta.get("model"), meta.get("model_provider")) or "codex"
    tool_calls = material.get("tool_calls") or []
    tool_results = material.get("tool_results") or []
    skill_usages = detect_turn_skill_usages(material, discover_known_skills())
    interaction_id = build_interaction_id("codex", session_id, turn_num)
    skill_use_events = build_skill_use_events(interaction_id, skill_usages)
    interaction_meta = build_interaction_metadata(
        "codex",
        user_id,
        session_id,
        turn_num,
        usage_details,
        len(tool_calls),
        len(tool_results),
        len(skill_use_events),
        model,
        user_message_count=1 if material.get("user_text") else 0,
        assistant_message_count=1 if material.get("assistant_text") else 0,
        skill_use_events=skill_use_events,
    )
    skill_summary = summarize_skill_usages(skill_usages)
    repo_context = dict(meta.get("repo_context") or collect_repo_context(first_string(meta.get("cwd")), "session_meta"))

    with propagate_attributes(
        user_id=user_id,
        session_id=session_id,
        trace_name=AGENT_TURN_NAME,
        tags=[AGENT_NAME],
    ):
        with langfuse.start_as_current_observation(
            name=AGENT_TURN_NAME,
            input={"role": "user", "content": user_text},
            output={"role": "assistant", "content": assistant_text},
            metadata={
                **interaction_meta,
                "source": AGENT_NAME,
                "agent": AGENT_NAME,
                "session_id": session_id,
                "turn_number": turn_num,
                "session_path": str(session_path),
                **repo_context,
                "originator": meta.get("originator"),
                "cli_version": meta.get("cli_version"),
                "user_text": user_meta,
                "usage": {"last_token_usage": usage_details},
                "skills": skill_summary,
                "codex_turn_id": turn.turn_id or "",
                "completed": turn.completed,
                "aborted": turn.aborted,
                "subagent_thread_ids": turn.subagent_thread_ids,
                "step_count": len(turn.steps),
            },
        ) as trace_span:
            with langfuse.start_as_current_observation(
                name="Agent Response",
                as_type="generation",
                model=model,
                input={"role": "user", "content": user_text},
                output={"role": "assistant", "content": assistant_text},
                usage_details=usage_details or None,
                metadata={
                    "assistant_text": assistant_meta,
                    "source": AGENT_NAME,
                    "agent": AGENT_NAME,
                    "user_id": user_id or "",
                    "session_id": session_id,
                    "interaction_id": interaction_meta["interaction_id"],
                    "turn_number": turn_num,
                },
            ):
                pass

            for call in tool_calls:
                tool_input, input_meta = truncate(call.get("input"))
                with langfuse.start_as_current_observation(
                    name="Tool Call",
                    as_type="tool",
                    input=tool_input,
                    metadata={
                        "source": AGENT_NAME,
                        "agent": AGENT_NAME,
                        "user_id": user_id or "",
                        "session_id": session_id,
                        "interaction_id": interaction_meta["interaction_id"],
                        "tool_id": call.get("id"),
                        "tool_name": call.get("name"),
                        "turn_number": turn_num,
                        "input_meta": input_meta,
                        "metrics_schema_version": METRICS_SCHEMA_VERSION,
                    },
                ):
                    pass

            for result in tool_results:
                output, output_meta = truncate(result.get("output"))
                with langfuse.start_as_current_observation(
                    name="Tool Result",
                    as_type="tool",
                    metadata={
                        "source": AGENT_NAME,
                        "agent": AGENT_NAME,
                        "user_id": user_id or "",
                        "session_id": session_id,
                        "interaction_id": interaction_meta["interaction_id"],
                        "tool_id": result.get("id"),
                        "tool_name": result.get("name"),
                        "turn_number": turn_num,
                        "output_meta": output_meta,
                        "status": result.get("status"),
                        "error": result.get("error"),
                        "start_time": result.get("start_time"),
                        "end_time": result.get("end_time"),
                        "metrics_schema_version": METRICS_SCHEMA_VERSION,
                    },
                ) as tool_obs:
                    tool_obs.update(output=output)

            trace_span.update(output={"role": "assistant", "content": assistant_text})


def main() -> int:
    if os.environ.get("TRACE_TO_LANGFUSE", "").lower() == "false":
        return 0

    payload = read_stdin_payload()
    config = read_json_if_exists(CONFIG_FILE)

    public_key = (
        os.environ.get("CODEX_LANGFUSE_PUBLIC_KEY")
        or os.environ.get("LANGFUSE_PUBLIC_KEY")
        or config.get("publicKey")
    )
    secret_key = (
        os.environ.get("CODEX_LANGFUSE_SECRET_KEY")
        or os.environ.get("LANGFUSE_SECRET_KEY")
        or config.get("secretKey")
    )
    host = (
        os.environ.get("CODEX_LANGFUSE_BASE_URL")
        or os.environ.get("LANGFUSE_BASEURL")
        or os.environ.get("LANGFUSE_HOST")
        or config.get("baseUrl")
        or "https://cloud.langfuse.com"
    )
    user_id = (
        find_value(payload, ("user_id", "userId", "username", "userName"))
        or os.environ.get("CODEX_LANGFUSE_USER_ID")
        or os.environ.get("LANGFUSE_USER_ID")
        or config.get("userId")
        or os.environ.get("USERNAME")
    )

    if not public_key or not secret_key:
        debug("missing Langfuse credentials")
        return 0

    session_path = resolve_session_file(payload)
    if not session_path or not session_path.exists():
        debug("missing Codex session file")
        return 0

    try:
        langfuse = Langfuse(public_key=public_key, secret_key=secret_key, host=host)
    except Exception as e:
        debug(f"Langfuse init failed: {e}")
        return 0

    try:
        with FileLock(LOCK_FILE):
            state = read_json_if_exists(STATE_FILE)
            key = session_key(session_path)
            ss = load_session_state(state, key)
            rows = read_all_jsonl(session_path)
            if not rows:
                try:
                    ss.offset = session_path.stat().st_size
                    ss.buffer = ""
                except Exception:
                    pass
                save_session_state(state, key, ss)
                write_json_atomic(STATE_FILE, state)
                return 0

            try:
                ss.offset = session_path.stat().st_size
                ss.buffer = ""
            except Exception:
                pass

            parsed = parse_codex_session(rows)
            if not parsed.turns:
                save_session_state(state, key, ss)
                write_json_atomic(STATE_FILE, state)
                return 0

            meta = get_session_meta(rows, session_path)
            meta.update({k: v for k, v in parsed.session_meta.items() if v not in (None, "")})
            meta.setdefault("session_path", str(session_path))
            meta["repo_context"] = collect_repo_context(first_string(meta.get("cwd")), "session_meta")
            session_id = first_string(str(meta.get("id")) if meta.get("id") else "", session_path.stem) or session_path.stem
            uploaded = 0

            for index, turn in enumerate(parsed.turns, start=1):
                if not should_upload_codex_turn(ss, session_path, turn, index):
                    continue
                turn_num = ss.turn_count + 1
                try:
                    emit_codex_turn(langfuse, session_id, user_id, turn_num, session_path, meta, turn)
                    mark_codex_turn_uploaded(ss, session_path, turn, index)
                    ss.turn_count = turn_num
                    uploaded += 1
                except Exception as e:
                    debug(f"emit_codex_turn failed: {e}")

            save_session_state(state, key, ss)
            write_json_atomic(STATE_FILE, state)

        try:
            langfuse.flush()
        except Exception:
            pass
        log("INFO", f"Processed {uploaded} Codex turn(s), total {ss.turn_count} for {session_path}")
        return 0
    except Exception as e:
        debug(f"unexpected failure: {e}")
        return 0
    finally:
        try:
            langfuse.shutdown()
        except Exception:
            pass


if __name__ == "__main__":
    sys.exit(main())