devin

#!/usr/bin/env python3
"""
Devin ACP transparent passthrough + usage logger.
Installs as 'devin' in front of the real 'devin acp' binary.
Only intercepts 'devin acp'; all other subcommands are exec'd to the real binary.

jarvis: ceiling Python prototype; rewrite in Go if upstream wants a single-binary CLI.
"""

import atexit
import json
import logging
import os
import queue
import shutil
import signal
import sqlite3
import subprocess
import sys
import threading
import time
from datetime import datetime, timedelta, timezone
from logging.handlers import RotatingFileHandler
from pathlib import Path


def data_dir() -> Path:
    base = os.environ.get('XDG_DATA_HOME', Path.home() / '.local' / 'share')
    d = Path(base) / 'devin_track'
    d.mkdir(parents=True, exist_ok=True)
    return d


def init_logging():
    log_path = data_dir() / 'wrapper.log'
    # Rotating handler caps disk usage. The server's reprocessor resets its
    # byte offset when the file shrinks below the stored value, so rotation
    # is safe: after rotation it re-reads the fresh log from byte 0.
    handler = RotatingFileHandler(
        str(log_path), maxBytes=10_000_000, backupCount=3, encoding='utf-8',
    )
    handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s'))
    root = logging.getLogger()
    root.setLevel(logging.INFO)
    root.addHandler(handler)


def db_path() -> Path:
    return data_dir() / 'usage.db'


def init_db():
    with sqlite3.connect(db_path()) as conn:
        conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS sessions (
                session_id TEXT PRIMARY KEY,
                created_at TEXT,
                updated_at TEXT,
                model TEXT,
                cwd TEXT,
                first_prompt TEXT,
                description TEXT
            );

            CREATE TABLE IF NOT EXISTS usage (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT NOT NULL,
                session_id TEXT,
                request_id TEXT,
                model TEXT,
                input_tokens INTEGER DEFAULT 0,
                output_tokens INTEGER DEFAULT 0,
                total_tokens INTEGER DEFAULT 0,
                cached_read_tokens INTEGER DEFAULT 0,
                cached_write_tokens INTEGER DEFAULT 0,
                duration_ms INTEGER,
                stop_reason TEXT,
                error_message TEXT,
                created_at TEXT DEFAULT CURRENT_TIMESTAMP
            );
            """
        )
        # Add columns introduced after the first schema version.
        for col in ('cached_read_tokens', 'cached_write_tokens', 'error_message',
                    'model_display_name', 'reasoning_effort', 'source', 'prompt_text'):
            try:
                if 'tokens' in col:
                    conn.execute(f'ALTER TABLE usage ADD COLUMN {col} INTEGER DEFAULT 0')
                else:
                    conn.execute(f'ALTER TABLE usage ADD COLUMN {col} TEXT')
            except sqlite3.OperationalError:
                pass
        # Sessions table additions.
        for col in ('cwd', 'first_prompt', 'description',
                    'model_display_name', 'reasoning_effort', 'source'):
            try:
                conn.execute(f'ALTER TABLE sessions ADD COLUMN {col} TEXT')
            except sqlite3.OperationalError:
                pass


def find_real_devin() -> str:
    if os.environ.get('DEVIN_TRACK_REAL'):
        return os.environ['DEVIN_TRACK_REAL']

    # Resolve symlinks: when the wrapper is reached via a symlink chain
    # (e.g. the CLI's ~/.local/bin/devin -> _versions/current/bin/devin ->
    # wrapper), sys.argv[0] is the outer link. realpath lands on the real
    # wrapper file so the devin.real backup sitting next to it is found.
    my_path = os.path.realpath(sys.argv[0])
    my_dir = os.path.dirname(my_path)

    # Check for backup alongside the wrapper.
    alt_names = ('devin.real', 'devin.bin', 'devin.original',
                 'devin.exe.real', 'devin.exe.bak')
    for alt in alt_names:
        candidate = os.path.join(my_dir, alt)
        if os.path.isfile(candidate):
            return candidate

    # Avoid resolving to ourself in PATH.
    found = shutil.which('devin')
    if found and os.path.abspath(found) != my_path:
        return found

    # Platform-specific fallback paths.
    candidates = []
    if sys.platform == 'win32':
        candidates = [
            os.path.join(os.environ.get('LOCALAPPDATA', ''), 'Programs', 'Devin Desktop', 'resources', 'app', 'extensions', 'windsurf', 'devin', 'bin', 'devin.exe'),
            os.path.join(os.environ.get('PROGRAMFILES', r'C:\Program Files'), 'Devin Desktop', 'resources', 'app', 'extensions', 'windsurf', 'devin', 'bin', 'devin.exe'),
        ]
    elif sys.platform == 'darwin':
        candidates = [
            '/Applications/Devin Desktop.app/Contents/Resources/app/extensions/windsurf/devin/bin/devin',
            os.path.expanduser('~/.local/bin/devin'),
        ]
    else:  # Linux and other Unix
        candidates = [
            os.path.expanduser('~/.local/bin/devin'),
            '/usr/share/devin-desktop/resources/app/extensions/windsurf/devin/bin/devin',
            '/usr/share/devin-desktop/resources/app/extensions/windsurf/devin/bin/devin.real',
            '/usr/local/bin/devin',
        ]

    for candidate in candidates:
        if os.path.isfile(candidate):
            return candidate

    raise FileNotFoundError('real devin binary not found')


# Subcommands that never create an agent session. For these, exec the real
# binary directly; for everything else, run the real binary and backfill.
# jarvis: heuristic list; update if Devin adds non-session subcommands.
_NON_SESSION = frozenset({
    'auth', 'mcp', 'models', 'doctor', 'rules', 'skills', 'plugins', 'cloud',
    'desktop', 'list', 'ls', 'rm', 'ssh', 'update', 'version', 'migrate',
    'sandbox', 'setup', 'uninstall', 'help',
})


def is_cli_session(argv):
    """True if the given argv starts an agent session.

    `devin`, `devin acp`, `devin --print`, and `devin <path> [prompt]` are
    sessions. `devin auth`, `devin list`, `devin --help`, etc. are not.
    """
    if not argv:
        return True
    if any(a in argv for a in ('--help', '-h', '--version', '-v')):
        return False
    if argv[0] in _NON_SESSION:
        return False
    # Anything else (including options like --print or a path/prompt) is a
    # session that the CLI's own database will record.
    return True


def _backfill_cli_path() -> Path | None:
    """Locate backfill_cli.py next to the wrapper, via env, or in source repo."""
    if os.environ.get('DEVIN_TRACK_BACKFILL'):
        p = Path(os.environ['DEVIN_TRACK_BACKFILL'])
        if p.is_file():
            return p

    p = Path(__file__).with_name('backfill_cli.py')
    if p.is_file():
        return p

    # Source-repo fallback for development / manual runs.
    # jarvis: hardcoded; deploy.py installs backfill_cli.py next to the
    # wrapper, so this only fires during development from the repo itself.
    p = Path('/home/tdvorak/Desktop/PROG+HTML/DevinTrack/backfill_cli.py')
    if p.is_file():
        return p

    return None


def run_backfill():
    """Spawn backfill_cli.py in the background to import CLI sessions."""
    # jarvis: scans the last 3 days of message_nodes; targeted per-session
    # backfill would be lighter, but this is safe, idempotent, and reuses the
    # existing backfill script.
    path = _backfill_cli_path()
    if not path:
        logging.warning('backfill_cli.py not found; skipping CLI backfill')
        return

    today = datetime.now().astimezone()
    from_date = (today - timedelta(days=2)).strftime('%Y-%m-%d')
    to_date = today.strftime('%Y-%m-%d')

    try:
        subprocess.Popen(
            [sys.executable, str(path), '--from', from_date, '--to', to_date],
            start_new_session=True,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        logging.info('spawned CLI backfill (%s -> %s)', from_date, to_date)
    except Exception as e:
        logging.exception('failed to spawn CLI backfill: %s', e)


def run_session(real_devin: str, argv: list):
    """Run an interactive `devin` session and backfill after it exits."""
    proc = subprocess.Popen([real_devin] + argv)
    try:
        return proc.wait()
    except KeyboardInterrupt:
        _terminate_child(proc)
        return 130
    finally:
        run_backfill()


def iso_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def extract_model(config_options):
    if not isinstance(config_options, list):
        return None
    for opt in config_options:
        if isinstance(opt, dict) and opt.get('id') == 'model':
            return opt.get('currentValue')
    return None


def extract_model_display_name(config_options, model_value):
    """Search configOptions options tree for the display name matching model_value."""
    if not isinstance(config_options, list) or not model_value:
        return None

    def search_options(options):
        if not isinstance(options, list):
            return None
        for opt in options:
            if isinstance(opt, dict):
                if opt.get('value') == model_value:
                    return opt.get('name')
                # Recurse into nested groups (opt has "options" key)
                found = search_options(opt.get('options'))
                if found:
                    return found
        return None

    # Find the model config option, then search its options tree.
    for cfg in config_options:
        if isinstance(cfg, dict) and cfg.get('id') == 'model':
            return search_options(cfg.get('options'))
    return None


# Reasoning effort suffixes in order of specificity (longest first to avoid
# partial matches like "high" matching "xhigh").
_EFFORT_SUFFIXES = ('xhigh', 'medium', 'high', 'low', 'max', 'none')


def extract_reasoning_effort(model_value):
    """Parse reasoning effort from model value suffix (e.g. 'claude-opus-5-high' -> 'high')."""
    if not model_value:
        return None
    val = model_value.lower()
    # Check for "-fast" suffix first, strip it to avoid mismatch.
    val = val.removesuffix('-fast')
    for suffix in _EFFORT_SUFFIXES:
        if val.endswith('-' + suffix):
            return suffix
    return None


def extract_prompt_text(prompt):
    """Extract text from ACP prompt content blocks.
    prompt is a list of {"type": "text", "text": "..."} objects.
    Returns the concatenated text, truncated to 500 chars.
    """
    if not isinstance(prompt, list):
        return None
    parts = []
    for block in prompt:
        if isinstance(block, dict) and block.get('type') == 'text':
            t = block.get('text')
            if t:
                parts.append(t)
    if not parts:
        return None
    text = ' '.join(parts)
    return text[:500] if len(text) > 500 else text


def maybe_json(line_bytes: bytes):
    """Return (parsed_dict, text) for a line, or (None, None) if not valid JSON."""
    # Strip the line terminator(s) before parsing, but keep the original bytes for forwarding.
    text = line_bytes.decode('utf-8', errors='replace').rstrip('\r\n')
    try:
        obj = json.loads(text)
    except json.JSONDecodeError:
        return None, None
    if not isinstance(obj, dict):
        return None, None
    return obj, text


class Tracker:
    def __init__(self):
        self._conn = sqlite3.connect(db_path(), check_same_thread=False)
        self._lock = threading.Lock()
        # id -> {session_id, request_id, start}
        self._pending = {}
        # session_id -> {'iso': str, 'perf': float} for the first session/prompt in the current turn.
        self._turn_start = {}
        # session_id -> True once agent_stopped has been recorded for the current turn.
        # Late session/prompt responses check this to avoid inserting a duplicate numeric row.
        self._turn_done = {}
        # id -> cwd from session/new (sessionId is not known until the response).
        self._session_cwd = {}
        # session_id -> prompt_text for the current turn (set by session/prompt request).
        self._session_prompts = {}

    def _execute(self, *args, **kwargs):
        with self._lock:
            return self._conn.execute(*args, **kwargs)

    def upsert_session(self, session_id: str, model=None, created_at=None, cwd=None,
                       first_prompt=None, description=None,
                       model_display_name=None, reasoning_effort=None):
        if not session_id:
            return
        now = iso_now()
        with self._lock:
            cur = self._conn.execute(
                'SELECT 1 FROM sessions WHERE session_id = ?', (session_id,)
            )
            exists = cur.fetchone()
            if exists:
                self._conn.execute(
                    """UPDATE sessions SET
                       updated_at = ?,
                       model = COALESCE(?, model),
                       cwd = COALESCE(?, cwd),
                       first_prompt = COALESCE(first_prompt, ?),
                       description = COALESCE(description, ?),
                       model_display_name = COALESCE(?, model_display_name),
                       reasoning_effort = COALESCE(?, reasoning_effort)
                       WHERE session_id = ?""",
                    (now, model, cwd, first_prompt, description,
                     model_display_name, reasoning_effort, session_id),
                )
            else:
                self._conn.execute(
                    """INSERT INTO sessions
                       (session_id, created_at, updated_at, model, cwd, first_prompt, description,
                        model_display_name, reasoning_effort)
                       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                    (session_id, created_at or now, now, model, cwd, first_prompt, description,
                     model_display_name, reasoning_effort),
                )
            self._conn.commit()

    def on_request(self, line_bytes: bytes):
        obj, _ = maybe_json(line_bytes)
        if obj is None:
            return

        method = obj.get('method')
        params = obj.get('params') or {}
        msg_id = obj.get('id')

        # Log only the methods the server's reprocessor depends on at INFO:
        # session/new, session/load, session/resume (session metadata + cwd),
        # session/prompt (prompt text reconstruction).
        # Config/model-change methods are logged so model switches are recoverable.
        # Everything else is DEBUG to avoid unbounded log growth (document/didOpen,
        # listSteps, session/list, etc. are high-volume and not needed for replay).
        _REPLAY_METHODS = ('session/new', 'session/load', 'session/resume', 'session/prompt')
        if method:
            is_replay = method in _REPLAY_METHODS
            is_config = 'mode' in method.lower() or 'model' in method.lower() or 'config' in method.lower()
            if is_replay or is_config:
                logging.info('request method=%s params=%s', method, json.dumps(params, ensure_ascii=False)[:2000])
            else:
                logging.debug('request method=%s params=%s', method, json.dumps(params, ensure_ascii=False)[:2000])

        if method in ('session/new', 'session/resume', 'session/load'):
            session_id = params.get('sessionId')
            cwd = params.get('cwd')
            if session_id:
                self.upsert_session(session_id, cwd=cwd)
            # session/new does not include sessionId in the request; stash cwd
            # by msg_id so the response can attach it to the new session.
            if method == 'session/new' and not session_id and msg_id is not None:
                with self._lock:
                    self._session_cwd[msg_id] = cwd
            return

        # Catch model/config change methods
        if method and ('mode' in method.lower() or 'model' in method.lower() or 'config' in method.lower()):
            session_id = params.get('sessionId')
            if session_id:
                # session/set_config_option uses configId + value, not configOptions.
                model = None
                if params.get('configId') == 'model':
                    model = params.get('value')
                if not model:
                    model = extract_model(params.get('configOptions')) or params.get('model')
                if model:
                    effort = extract_reasoning_effort(model)
                    self.upsert_session(session_id, model=model,
                                        reasoning_effort=effort)
                # Stash session_id by msg_id so we can extract display name from response.
                if msg_id is not None:
                    with self._lock:
                        self._pending[msg_id] = {
                            'session_id': session_id,
                            'request_id': str(msg_id),
                            'start': time.perf_counter(),
                            'config_change': True,
                        }

        if method == 'session/prompt' and msg_id is not None:
            session_id = params.get('sessionId')
            prompt_text = extract_prompt_text(params.get('prompt'))
            if session_id:
                # Store first_prompt only if not already set.
                self.upsert_session(session_id, first_prompt=prompt_text)
                with self._lock:
                    self._pending[msg_id] = {
                        'session_id': session_id,
                        'request_id': str(msg_id),
                        'start': time.perf_counter(),
                    }
                    # Store prompt text for this session's current turn.
                    # agent_stopped will pick it up when recording the final usage row.
                    if prompt_text:
                        self._session_prompts[session_id] = prompt_text
                    # Mark the start of a new agent turn so agent_stopped can
                    # clean up intermediate session/prompt rows.
                    if session_id not in self._turn_start:
                        self._turn_start[session_id] = {
                            'iso': iso_now(),
                            'perf': time.perf_counter(),
                        }
                        self._turn_done.pop(session_id, None)

    def on_response(self, line_bytes: bytes):
        obj, _ = maybe_json(line_bytes)
        if obj is None:
            return

        msg_id = obj.get('id')
        result = obj.get('result')
        error = obj.get('error')

        # DEBUG: log all responses with sessionId or configOptions or model info
        if isinstance(result, dict) and ('sessionId' in result or 'configOptions' in result or 'model' in result):
            logging.info('session response: %s', json.dumps(result, ensure_ascii=False)[:3000])

        # Log notification-like responses (method field on response side).
        # Only agent_stopped is needed by the reprocessor (it scans for the
        # method string + full params JSON). Other notifications (didOpen,
        # didClose, listSteps, etc.) are high-volume -> DEBUG.
        method = obj.get('method')
        if method:
            if method == '_cognition.ai/agent_stopped':
                logging.info('response method=%s: %s', method, json.dumps(obj, ensure_ascii=False)[:2000])
            else:
                logging.debug('response method=%s: %s', method, json.dumps(obj, ensure_ascii=False)[:2000])

        if method == '_cognition.ai/agent_stopped':
            self.handle_agent_stopped(obj)

        # session/new, session/resume, etc. expose the model in their result.
        if isinstance(result, dict) and 'sessionId' in result:
            session_id = result.get('sessionId')
            config_opts = result.get('configOptions')
            model = extract_model(config_opts)
            description = result.get('description') or result.get('title')
            display_name = extract_model_display_name(config_opts, model) if model else None
            effort = extract_reasoning_effort(model) if model else None
            with self._lock:
                cwd = self._session_cwd.pop(msg_id, None)
            self.upsert_session(session_id, cwd=cwd, model=model, description=description,
                                model_display_name=display_name, reasoning_effort=effort)

        # Also handle set_config_option responses (no sessionId but has configOptions).
        # The session_id was stashed in on_request via self._pending.
        if isinstance(result, dict) and 'configOptions' in result and 'sessionId' not in result:
            config_opts = result.get('configOptions')
            model = extract_model(config_opts)
            if model and msg_id is not None:
                with self._lock:
                    pending_entry = self._pending.get(msg_id)
                if pending_entry and pending_entry.get('config_change'):
                    session_id = pending_entry['session_id']
                    display_name = extract_model_display_name(config_opts, model)
                    effort = extract_reasoning_effort(model)
                    self.upsert_session(session_id, model=model,
                                        model_display_name=display_name,
                                        reasoning_effort=effort)

        if msg_id is None or msg_id not in self._pending:
            return

        with self._lock:
            pending = self._pending.pop(msg_id)

        # Config change responses don't have usage data; early return after cleanup.
        if pending.get('config_change'):
            return

        duration_ms = int((time.perf_counter() - pending['start']) * 1000)
        session_id = pending['session_id']

        # If agent_stopped already recorded the final cumulative usage for this
        # turn, skip the per-call session/prompt row to avoid duplicates.
        with self._lock:
            if self._turn_done.get(session_id):
                logging.info(
                    'skipped late session/prompt row: session=%s request=%s (agent_stopped already recorded)',
                    session_id, pending['request_id'],
                )
                return

        stop_reason = None
        error_message = None
        input_tokens = 0
        output_tokens = 0
        total_tokens = 0
        cached_read_tokens = 0
        cached_write_tokens = 0

        if isinstance(result, dict):
            stop_reason = result.get('stopReason')
            usage = result.get('usage') or {}
            input_tokens = usage.get('inputTokens', 0) or 0
            output_tokens = usage.get('outputTokens', 0) or 0
            total_tokens = usage.get('totalTokens')
            if total_tokens is None:
                total_tokens = input_tokens + output_tokens
            cached_read_tokens = usage.get('cachedReadTokens', 0) or 0
            cached_write_tokens = usage.get('cachedWriteTokens', 0) or 0
        if error:
            stop_reason = 'error'
            error_message = (error.get('message') or str(error))[:1000]

        # Get the model and metadata from the sessions table, if known.
        model = None
        model_display_name = None
        reasoning_effort = None
        try:
            cur = self._execute(
                'SELECT model, model_display_name, reasoning_effort FROM sessions WHERE session_id = ?',
                (session_id,)
            )
            row = cur.fetchone()
            if row:
                model = row[0]
                model_display_name = row[1]
                reasoning_effort = row[2]
        except Exception:
            pass

        # Pick up prompt text for this request.
        prompt_text = None
        with self._lock:
            prompt_text = self._session_prompts.pop(session_id, None)

        self._execute(
            """
            INSERT INTO usage
            (timestamp, session_id, request_id, model,
             input_tokens, output_tokens, total_tokens,
             cached_read_tokens, cached_write_tokens,
             duration_ms, stop_reason, error_message,
             model_display_name, reasoning_effort, source, prompt_text)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (iso_now(), session_id, pending['request_id'], model,
             input_tokens, output_tokens, total_tokens,
             cached_read_tokens, cached_write_tokens,
             duration_ms, stop_reason, error_message,
             model_display_name, reasoning_effort, 'local', prompt_text),
        )
        self._conn.commit()
        logging.info(
            'recorded usage: session=%s request=%s in=%d out=%d total=%d cached=%d/%d ms=%d reason=%s',
            session_id, pending['request_id'], input_tokens, output_tokens,
            total_tokens, cached_read_tokens, cached_write_tokens,
            duration_ms, stop_reason,
        )

    def handle_agent_stopped(self, obj):
        """Record final cumulative usage from _cognition.ai/agent_stopped.

        The per-call 'usage' object inside session/prompt responses only covers
        the final model inference. The agent_stopped notification carries
        responseDimensions with the cumulative token counts shown in the Devin UI.
        """
        params = obj.get('params') or {}
        stats = params.get('stats') or {}
        session_id = params.get('sessionId')
        if not session_id:
            return

        # Build a uid -> dimension lookup.
        dims = {
            d.get('uid'): d
            for d in (stats.get('responseDimensions') or [])
            if isinstance(d, dict)
        }

        def _dim_value(uid, default=0):
            dim = dims.get(uid)
            if not isinstance(dim, dict):
                return default
            kind = dim.get('kind') or {}
            val = kind.get('value')
            try:
                return int(val) if val is not None else default
            except (TypeError, ValueError):
                return default

        input_new = _dim_value('input_tokens')
        output = _dim_value('output_tokens')
        cached = _dim_value('cached_input_tokens')

        # responseDimensions.input_tokens is *new* input, while
        # cached_input_tokens is the cached portion. Store gross input so the
        # server can compute new_input_tokens = input - cached.
        input_gross = input_new + cached
        total = input_gross + output

        cause = params.get('cause') or 'complete'
        stop_reason = 'end_turn' if cause == 'complete' else cause

        request_id = stats.get('requestId') or 'unknown'

        with self._lock:
            turn_start = self._turn_start.pop(session_id, None)

        # Wall-clock turn duration is more useful than the final-call totalTimeMs.
        if turn_start:
            duration_ms = int((time.perf_counter() - turn_start['perf']) * 1000)
        else:
            duration_ms = stats.get('totalTimeMs') or 0

        # Model metadata: prefer the session's canonical model, fall back to
        # modelLabel / responseDimensions model.
        dim_model = None
        dim = dims.get('model')
        if isinstance(dim, dict):
            kind = dim.get('kind') or {}
            dim_model = kind.get('value')
        model_label = stats.get('modelLabel') or dim_model

        model = None
        model_display_name = None
        reasoning_effort = None
        try:
            row = self._execute(
                'SELECT model, model_display_name, reasoning_effort FROM sessions WHERE session_id = ?',
                (session_id,)
            ).fetchone()
            if row:
                model = row[0]
                model_display_name = row[1]
                reasoning_effort = row[2]
        except Exception:
            pass

        if model_label:
            model_display_name = model_label
            # Derive reasoning effort from a dashed form of the model label.
            label_dashed = model_label.lower().replace(' ', '-')
            reasoning_effort = extract_reasoning_effort(label_dashed) or reasoning_effort
            # modelLabel is the ground truth for this specific request — use
            # it even if the sessions table has a different (stale) model.
            model = label_dashed

        # Deduplicate on (session_id, request_id) in case agent_stopped is seen twice.
        try:
            self._execute(
                'DELETE FROM usage WHERE session_id = ? AND request_id = ?',
                (session_id, request_id)
            )
        except Exception:
            pass

        if turn_start:
            # Remove intermediate session/prompt rows for this turn.
            try:
                self._execute(
                    'DELETE FROM usage WHERE session_id = ? AND timestamp >= ?',
                    (session_id, turn_start['iso'])
                )
            except Exception:
                pass

        # Pick up the prompt text for this session's current turn.
        prompt_text = None
        with self._lock:
            prompt_text = self._session_prompts.pop(session_id, None)

        self._execute(
            """
            INSERT INTO usage
            (timestamp, session_id, request_id, model,
             input_tokens, output_tokens, total_tokens,
             cached_read_tokens, cached_write_tokens,
             duration_ms, stop_reason, error_message,
             model_display_name, reasoning_effort, source, prompt_text)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (iso_now(), session_id, request_id, model,
             input_gross, output, total,
             cached, 0,
             duration_ms, stop_reason, None,
             model_display_name, reasoning_effort, 'local', prompt_text),
        )

        # Mark turn as done so any late session/prompt response skips its numeric row.
        with self._lock:
            self._turn_done[session_id] = True
            # Clean up the prompt text for this session.
            self._session_prompts.pop(session_id, None)

        # Update session metadata with final model / display name.
        if model or model_display_name or reasoning_effort:
            self.upsert_session(
                session_id,
                model=model,
                model_display_name=model_display_name,
                reasoning_effort=reasoning_effort,
            )

        self._conn.commit()
        logging.info(
            'recorded agent_stopped: session=%s request=%s in=%d out=%d total=%d cached=%d ms=%d',
            session_id, request_id, input_gross, output, total, cached, duration_ms
        )


def forward_stdin(child):
    """Read our stdin, observe, then forward to the child stdin as bytes."""
    stdin = sys.stdin.buffer
    try:
        for line in iter(stdin.readline, b''):
            child.tracker.on_request(line)
            child.proc.stdin.write(line)
            child.proc.stdin.flush()
    except BrokenPipeError:
        pass
    finally:
        try:
            child.proc.stdin.close()
        except Exception:
            pass


def forward_stdout(child):
    """Read child stdout, observe, then enqueue for the main thread to forward."""
    try:
        for line in iter(child.proc.stdout.readline, b''):
            child.tracker.on_response(line)
            child.out_q.put(line)
    except Exception:
        logging.exception('stdout reader failed')
    finally:
        child.out_q.put(None)


def forward_stderr(child):
    """Forward child stderr to our stderr, byte-for-byte."""
    try:
        for line in iter(child.proc.stderr.readline, b''):
            sys.stderr.buffer.write(line)
            sys.stderr.buffer.flush()
    except Exception:
        logging.exception('stderr reader failed')


class Child:
    def __init__(self, proc, tracker, out_q):
        self.proc = proc
        self.tracker = tracker
        self.out_q = out_q


def _terminate_child(proc):
    if proc.poll() is None:
        proc.terminate()
        try:
            proc.wait(timeout=3)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait()


def run_acp(real_devin: str, argv: list):
    init_db()
    tracker = Tracker()

    env = os.environ.copy()
    proc = subprocess.Popen(
        [real_devin] + argv,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=False,
        bufsize=-1,
        env=env,
    )

    # Ensure the child is terminated if the wrapper exits or is signalled.
    atexit.register(lambda p=proc: _terminate_child(p))

    # Signal handlers: forward SIGTERM/SIGHUP/SIGINT to the child, then exit.
    for sig in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT):
        signal.signal(sig, lambda _s, _f, p=proc: (_terminate_child(p), sys.exit(1))[1])

    out_q = queue.Queue()
    child = Child(proc, tracker, out_q)

    t_in = threading.Thread(target=forward_stdin, args=(child,), daemon=True)
    t_out = threading.Thread(target=forward_stdout, args=(child,), daemon=True)
    t_err = threading.Thread(target=forward_stderr, args=(child,), daemon=True)
    t_in.start()
    t_out.start()
    t_err.start()

    try:
        while True:
            line = out_q.get()
            if line is None:
                break
            sys.stdout.buffer.write(line)
            sys.stdout.buffer.flush()
    except KeyboardInterrupt:
        pass
    finally:
        _terminate_child(proc)
        run_backfill()

    return proc.returncode


def run_print_mode(real_devin: str, argv: list):
    """Handle --print mode by adding --export, running the real binary,
    then parsing the export file for token usage."""
    import tempfile
    import atexit

    init_db()

    # Add --export to a temp file if not already present.
    has_export = '--export' in argv
    if has_export:
        # Find the existing export path.
        idx = argv.index('--export')
        export_path = argv[idx + 1] if idx + 1 < len(argv) else None
    else:
        fd, export_path = tempfile.mkstemp(suffix='.json', prefix='devin-track-')
        os.close(fd)
        argv = argv + ['--export', export_path]

    # Extract model from --model flag.
    model = None
    if '--model' in argv:
        idx = argv.index('--model')
        if idx + 1 < len(argv):
            model = argv[idx + 1]

    # Extract prompt from --print or --prompt-file.
    prompt_text = ''
    if '--prompt-file' in argv:
        idx = argv.index('--prompt-file')
        if idx + 1 < len(argv):
            try:
                with open(argv[idx + 1]) as f:
                    prompt_text = f.read()[:500]
            except Exception:
                pass
    elif '--print' in argv:
        idx = argv.index('--print')
        if idx + 1 < len(argv) and not argv[idx + 1].startswith('-'):
            prompt_text = argv[idx + 1][:500]

    # Run the real binary, capturing stdout/stderr.
    proc = subprocess.run(
        [real_devin] + argv,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    # Forward stdout/stderr to the user.
    sys.stdout.buffer.write(proc.stdout)
    sys.stdout.buffer.flush()
    sys.stderr.buffer.write(proc.stderr)
    sys.stderr.buffer.flush()

    # Parse the export file for token usage.
    try:
        with open(export_path) as f:
            data = json.load(f)

        session_id = data.get('session_id', 'unknown')
        metrics = data.get('final_metrics', {})
        agent = data.get('agent', {})
        model_name = agent.get('model_name', model or 'unknown')

        input_tokens = metrics.get('total_prompt_tokens', 0)
        output_tokens = metrics.get('total_completion_tokens', 0)
        cached_tokens = metrics.get('total_cached_tokens', 0)
        total_tokens = input_tokens + output_tokens

        # Insert into DB.
        tracker = Tracker()
        tracker._execute(
            """INSERT INTO usage
               (timestamp, session_id, request_id, model,
                input_tokens, output_tokens, total_tokens,
                cached_read_tokens, cached_write_tokens,
                duration_ms, stop_reason, error_message,
                model_display_name, reasoning_effort, source)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (iso_now(), session_id, 'print-mode', model,
             input_tokens, output_tokens, total_tokens,
             cached_tokens, 0,
             None, 'end_turn', None,
             model_name, None, os.environ.get('DEVIN_TRACK_SOURCE', 'local')),
        )

        # Upsert session.
        tracker._execute(
            """INSERT INTO sessions (session_id, created_at, updated_at, model,
               model_display_name, cwd, first_prompt, description, source)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
               ON CONFLICT(session_id) DO UPDATE SET
               updated_at=excluded.updated_at,
               model=COALESCE(excluded.model, sessions.model),
               model_display_name=COALESCE(excluded.model_display_name, sessions.model_display_name),
               first_prompt=COALESCE(excluded.first_prompt, sessions.first_prompt),
               source=excluded.source""",
            (session_id, iso_now(), iso_now(), model,
             model_name, os.getcwd(), prompt_text, prompt_text[:80] if prompt_text else None,
             os.environ.get('DEVIN_TRACK_SOURCE', 'local')),
        )
        tracker._conn.commit()

        logging.info('captured print-mode usage: session=%s model=%s in=%d out=%d cached=%d',
                     session_id, model_name, input_tokens, output_tokens, cached_tokens)
    except Exception as e:
        logging.exception('failed to parse export file: %s', e)
    finally:
        # Clean up temp export file if we created it.
        if not has_export:
            try:
                os.unlink(export_path)
            except Exception:
                pass
        run_backfill()

    return proc.returncode


def main():
    init_logging()
    argv = sys.argv[1:]

    real_devin = find_real_devin()
    is_acp = argv and argv[0] == 'acp'
    is_print = '--print' in argv or '--prompt-file' in argv
    session = is_cli_session(argv)

    if not session:
        # Non-session subcommand: exec the real binary directly for speed.
        os.execvp(real_devin, [real_devin] + argv)

    logging.info('starting wrapper; real devin=%s', real_devin)

    if is_print and not is_acp:
        try:
            return run_print_mode(real_devin, argv)
        except Exception as e:
            logging.exception('wrapper error (print mode)')
            raise

    if is_acp:
        try:
            return run_acp(real_devin, argv)
        except Exception as e:
            logging.exception('wrapper error')
            raise

    # Interactive / prompt-based session (e.g. `devin` or `devin path prompt`).
    try:
        return run_session(real_devin, argv)
    except Exception as e:
        logging.exception('wrapper error (session)')
        raise


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