T TelemHQ

Track CodexCodexAn AI coding assistant workflow. TelemHQ can record Codex usage by branch, task, model, token counts, cost, and run result.View glossary entrySource: OpenAI token guide Usage

Send TelemHQ pingspingA request sent to TelemHQ after a job runs. A ping can be a simple heartbeat or include JSON payload data about what happened.View glossary entrySource: TelemHQ docs from your local Codex usage logs so you can chart modelmodelThe AI system that processes input and returns output. For monitoring, the model name helps explain which tool or provider produced a run and how its token usage should be priced.View glossary entrySource: Anthropic model docs, token usagetokensThe pieces of text an AI model processes. Token counts are often used to measure usage and calculate model cost.View glossary entrySource: OpenAI token guide, cached tokens, reasoning tokensreasoning tokensTokens used internally by some reasoning models while working toward an answer. They can count toward usage even when they are not shown as final output.View glossary entrySource: OpenAI token guide, project names, gitGitA distributed version control system used to track code changes. TelemHQ can store Git metadata like branch or issue identifiers with a run.View glossary entrySource: Git glossary branchesbranchA line of development in Git. Tracking branch names helps connect AI coding usage or job cost to a specific task or change.View glossary entrySource: Git glossary, and usage trends over time.

Before You Start

  1. Create a trackertrackerA monitored job, AI pipeline, worker, script, or automation in TelemHQ. Each tracker has its own ping URL and run history.View glossary entrySource: TelemHQ docs in TelemHQ named Codex usage.
  2. Leave the schedule blank. Codex usage is ad hocad hoc jobA task that runs whenever needed instead of on a fixed schedule. TelemHQ records each run but does not fail the tracker just because no scheduled ping arrived.View glossary entrySource: AWS EventBridge Scheduler docs, so TelemHQ should not mark missed pings as failures.
  3. Copy the tracker ping URL from the success dialog or tracker details page.
  4. Run the setup commands below on the machine where you use Codex.
This integration reads Codex's local SQLite logs under ~/.codex. It sends usage metadatametadataData about a run rather than the private content of the run itself, such as model name, duration, branch, item counts, or token totals.View glossary entrySource: MDN API glossary only, not prompts, generated code, file contents, model output, or raw local project paths by default. Those logs are a local implementation detail and may change in future Codex releases.

1. Save Your Ping URL

Use your real TelemHQ ping URL. The script also supports TELEMHQ_PING_URL if you already use that name elsewhere. 

mkdir -p ~/.telemhq
export TELEMHQ_CODEX_PING_URL="https://telemhq.com/ping/YOUR_TRACKING_TOKEN"
printf 'TELEMHQ_CODEX_PING_URL="%s"\n' "$TELEMHQ_CODEX_PING_URL" > ~/.telemhq/codex-usage.env
chmod 600 ~/.telemhq/codex-usage.env

The private env file lets the watcher run from macOS launchd or another background service that does not inherit your terminal environment.

Optional: add CODEX_USAGE_INCLUDE_PROJECT_PATH="1" to the env file if you want to include the raw local working directory. Most users should leave this off and use the default project_path_hash instead.

2. Create The Watcher Script

This script sends one TelemHQ ping per completed Codex model response. By default it runs once and exits. Use --watch when you want it to keep polling in the background. 

cat > ~/.telemhq/codex-usage-watch.py <<'PY'
#!/usr/bin/env python3
import argparse
import datetime as dt
import hashlib
import json
import os
import re
import sqlite3
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


DEFAULT_DB = Path.home() / ".codex" / "logs_2.sqlite"
DEFAULT_STATE_DB = Path.home() / ".codex" / "state_5.sqlite"
DEFAULT_ENV = Path.home() / ".telemhq" / "codex-usage.env"
DEFAULT_STATE = Path.home() / ".telemhq" / "codex-usage-watch.state.json"


def load_env_file(path):
    if not path.exists():
        return
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip().strip('"').strip("'")
        if key and key not in os.environ:
            os.environ[key] = value


def read_state(path):
    if not path.exists():
        return {"last_id": 0, "response_ids": []}
    raw = path.read_text().strip()
    if not raw:
        return {"last_id": 0, "response_ids": []}
    try:
        data = json.loads(raw)
        return {
            "last_id": int(data.get("last_id", 0)),
            "response_ids": list(data.get("response_ids", [])),
        }
    except json.JSONDecodeError:
        return {"last_id": int(raw), "response_ids": []}


def write_state(path, state):
    path.parent.mkdir(parents=True, exist_ok=True)
    state["response_ids"] = state.get("response_ids", [])[-500:]
    path.write_text(json.dumps(state, indent=2))


def extract_event(body):
    for marker in ("Received message ", "websocket event: "):
        index = body.find(marker)
        if index != -1:
            raw = body[index + len(marker):].strip()
            if raw.startswith("{"):
                return json.loads(raw)
    return None


def model_from_body(body, response):
    if response.get("model"):
        return response["model"]
    matches = re.findall(r"\bmodel=([^ }\]]+)", body)
    return matches[-1] if matches else "unknown"


def thread_from_body(body, row_thread_id):
    if row_thread_id:
        return row_thread_id
    match = re.search(r"(?:thread_id|thread\.id)=([0-9a-zA-Z-]+)", body)
    return match.group(1) if match else None


def sqlite_readonly_uri(path):
    return "file:{}?mode=ro".format(urllib.parse.quote(str(path), safe="/:"))


def git_repo_slug(origin_url):
    if not origin_url:
        return None
    value = origin_url.strip()
    if value.endswith(".git"):
        value = value[:-4]
    match = re.search(r"(?:github\.com[:/])([^/]+/[^/]+)$", value)
    if match:
        return match.group(1)
    if "/" in value:
        parts = [part for part in value.rsplit("/", 2) if part]
        if len(parts) >= 2:
            return "/".join(parts[-2:])
    if ":" in value:
        return value.rsplit(":", 1)[-1]
    return value or None


def project_metadata_from_thread(state_db_path, thread_id):
    if not thread_id or not state_db_path.exists():
        return {}
    try:
        with sqlite3.connect(sqlite_readonly_uri(state_db_path), uri=True) as connection:
            row = connection.execute(
                """
                SELECT cwd, git_branch, git_origin_url, git_sha
                FROM threads
                WHERE id = ?
                LIMIT 1
                """,
                (thread_id,),
            ).fetchone()
    except sqlite3.Error:
        return {}
    if not row:
        return {}

    cwd, git_branch, git_origin_url, git_sha = row
    metadata = {}
    if cwd:
        metadata["project"] = Path(cwd).name or "unknown"
        metadata["project_path_hash"] = hashlib.sha256(cwd.encode("utf-8")).hexdigest()[:12]
        if os.getenv("CODEX_USAGE_INCLUDE_PROJECT_PATH") == "1":
            metadata["project_cwd"] = cwd
    repo = git_repo_slug(git_origin_url)
    if repo:
        metadata["git_repo"] = repo
    if git_branch:
        metadata["git_branch"] = git_branch
    if git_sha:
        metadata["git_sha"] = git_sha[:12]
    return metadata


def build_payload(row_id, ts, row_thread_id, body, state_db_path):
    event = extract_event(body)
    if not event or event.get("type") != "response.completed":
        return None

    response = event.get("response") or {}
    usage = response.get("usage") or {}
    if not usage:
        return None

    input_tokens = usage.get("input_tokens")
    output_tokens = usage.get("output_tokens")
    total_tokens = usage.get("total_tokens")
    input_details = usage.get("input_tokens_details") or {}
    output_details = usage.get("output_tokens_details") or {}
    cached_input_tokens = input_details.get("cached_tokens")
    reasoning_tokens = output_details.get("reasoning_tokens")
    thread_id = thread_from_body(body, row_thread_id)

    payload = {
        "tool": "codex",
        "provider": "openai",
        "status": response.get("status") or "completed",
        "model": model_from_body(body, response),
        "session_type": "local",
        "thread_id": thread_id,
        "response_id": response.get("id"),
        "log_id": row_id,
        "timestamp": dt.datetime.fromtimestamp(ts, dt.timezone.utc).isoformat(),
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "total_tokens": total_tokens,
        "cached_input_tokens": cached_input_tokens,
        "reasoning_tokens": reasoning_tokens,
    }
    payload.update(project_metadata_from_thread(state_db_path, thread_id))

    return {key: value for key, value in payload.items() if value is not None}


def send_ping(ping_url, payload):
    data = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        ping_url,
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=15) as response:
        if response.status >= 300:
            raise RuntimeError(f"TelemHQ returned HTTP {response.status}")


def scan_once(db_path, codex_state_db_path, state_path, ping_url):
    state = read_state(state_path)
    sent_response_ids = set(state.get("response_ids", []))

    with sqlite3.connect(sqlite_readonly_uri(db_path), uri=True) as connection:
        rows = connection.execute(
            """
            SELECT id, ts, thread_id, feedback_log_body
            FROM logs
            WHERE id > ?
              AND feedback_log_body LIKE '%response.completed%'
              AND feedback_log_body LIKE '%"usage":%'
            ORDER BY id ASC
            """,
            (state["last_id"],),
        ).fetchall()

    sent = 0
    for row_id, ts, row_thread_id, body in rows:
        state["last_id"] = row_id
        try:
            payload = build_payload(row_id, ts, row_thread_id, body or "", codex_state_db_path)
        except json.JSONDecodeError:
            write_state(state_path, state)
            continue

        if not payload:
            write_state(state_path, state)
            continue

        response_id = payload.get("response_id")
        if response_id and response_id in sent_response_ids:
            write_state(state_path, state)
            continue

        send_ping(ping_url, payload)
        if response_id:
            sent_response_ids.add(response_id)
            state["response_ids"] = list(sent_response_ids)
        write_state(state_path, state)
        sent += 1
        print(
            "Sent Codex usage ping: "
            f"model={payload.get('model')} project={payload.get('project', 'unknown')} total_tokens={payload.get('total_tokens')}"
        )

    return sent


def main():
    parser = argparse.ArgumentParser(description="Send Codex usage metrics to TelemHQ.")
    parser.add_argument("--watch", action="store_true", help="Keep polling for new Codex responses.")
    parser.add_argument("--interval", type=int, default=int(os.getenv("CODEX_USAGE_POLL_SECONDS", "60")))
    parser.add_argument("--db", default=os.getenv("CODEX_USAGE_DB", str(DEFAULT_DB)))
    parser.add_argument("--codex-state-db", default=os.getenv("CODEX_STATE_DB", str(DEFAULT_STATE_DB)))
    parser.add_argument("--state", default=os.getenv("CODEX_USAGE_STATE_FILE", str(DEFAULT_STATE)))
    parser.add_argument("--env-file", default=os.getenv("CODEX_USAGE_ENV_FILE", str(DEFAULT_ENV)))
    args = parser.parse_args()

    load_env_file(Path(args.env_file).expanduser())
    ping_url = os.getenv("TELEMHQ_CODEX_PING_URL") or os.getenv("TELEMHQ_PING_URL")
    if not ping_url:
        sys.exit("Set TELEMHQ_CODEX_PING_URL in your shell or ~/.telemhq/codex-usage.env.")

    db_path = Path(args.db).expanduser()
    codex_state_db_path = Path(args.codex_state_db).expanduser()
    state_path = Path(args.state).expanduser()
    if not db_path.exists():
        sys.exit(f"Codex log database not found: {db_path}")

    while True:
        sent = scan_once(db_path, codex_state_db_path, state_path, ping_url)
        if not args.watch:
            if sent == 0:
                print("No new completed Codex usage events found.")
            return
        time.sleep(max(args.interval, 10))


if __name__ == "__main__":
    main()
PY

chmod +x ~/.telemhq/codex-usage-watch.py

3. Send A Test Ping

Run a Codex task, then run the watcher once. It should exit after sending any new completed usage events. 

~/.telemhq/codex-usage-watch.py

In TelemHQ, open the tracker and check Ping Historyrun historyThe stored record of previous job runs. TelemHQ uses run history to show payloads, failures, timing, token totals, and trends over time.View glossary entrySource: TelemHQ docs. You should see a payloadpayloadThe structured data sent with a request. In TelemHQ, payloads should contain safe operational metadata, not prompts, completions, secrets, customer data, or private paths.View glossary entrySource: MDN API glossary with fields such as model, input_tokens, output_tokens, total_tokens, cached_input_tokens, reasoning_tokens, project, git_repo, and git_branch.

4. Keep It Running

Use watch mode when you want TelemHQ to receive pings automatically as you use Codex. This command keeps running until you stop it. 

~/.telemhq/codex-usage-watch.py --watch

To run it in the background for your current terminal sessionsessionServer-side or cookie-backed state that remembers a signed-in user between requests.View glossary entrySource: MDN glossary: 

nohup ~/.telemhq/codex-usage-watch.py --watch > ~/.telemhq/codex-usage-watch.log 2>&1 &

5. Start It Automatically On macOS

If you use Codex on macOS, create a LaunchAgent so the watcher starts when you log in and restarts if it exits. This is more reliable than leaving a terminal tab open. 

mkdir -p ~/Library/LaunchAgents

cat > ~/Library/LaunchAgents/com.telemhq.codex-usage-watch.plist <<PLIST
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
  <key>Label</key>
  <string>com.telemhq.codex-usage-watch</string>

  <key>ProgramArguments</key>
  <array>
    <string>$HOME/.telemhq/codex-usage-watch.py</string>
    <string>--watch</string>
    <string>--interval</string>
    <string>30</string>
  </array>

  <key>RunAtLoad</key>
  <true/>

  <key>KeepAlive</key>
  <true/>

  <key>StandardOutPath</key>
  <string>$HOME/.telemhq/codex-usage-watch.log</string>

  <key>StandardErrorPath</key>
  <string>$HOME/.telemhq/codex-usage-watch.err.log</string>

  <key>WorkingDirectory</key>
  <string>$HOME</string>
</dict>
</plist>
PLIST

launchctl bootout gui/$(id -u) ~/Library/LaunchAgents/com.telemhq.codex-usage-watch.plist 2>/dev/null || true
launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.telemhq.codex-usage-watch.plist
launchctl kickstart -k gui/$(id -u)/com.telemhq.codex-usage-watch

Verify that it is running: 

launchctl print gui/$(id -u)/com.telemhq.codex-usage-watch
pgrep -af codex-usage-watch.py
tail -f ~/.telemhq/codex-usage-watch.log ~/.telemhq/codex-usage-watch.err.log

Recommended Charts

Open the Analytics tab on your Codex tracker. TelemHQ will detect numeric payload fields and recommend charts automatically. Use Manage charts to pin the fields you want to see every time.

  • total_tokens
  • input_tokens
  • output_tokens
  • cached_input_tokens
  • reasoning_tokens
  • project in the payload summary
  • git_repo and git_branch in the payload summary
  • model in the payload summary

When recent Codex payloads include more than one project, TelemHQ also shows a project usage chart. Codex usage defaults that chart to total_tokens.

Privacy And Troubleshooting

The watcher sends usage metadata only. It does not send prompts, generated code, command arguments, file contents, model output, or raw local project paths unless you explicitly enable CODEX_USAGE_INCLUDE_PROJECT_PATH.

If you see No new completed Codex usage events found, run a new Codex task and try the one-shot command again.

If the databasedatabaseA system for storing and querying structured data. TelemHQ uses database records to keep users, trackers, job runs, teams, and billing state.View glossary entrySource: PostgreSQL overview path changes, set CODEX_USAGE_DB to the new SQLite file path before running the watcher.

Official references: OpenAI Codex CLI getting started and Using Codex with your ChatGPT plan.