# -*- coding: utf-8 -*-
"""versus adapter for the DeepSeek Harness (dsh), version 0.1.0-rc.7.

One node process per scenario, holding one live Agent, driven a turn at a time
over a line protocol. The grader hashes the project tree between turns, so the
turns cannot be batched, and the harness cannot be restarted between them without
throwing away the memory that three of the five scenarios exist to measure.

Nothing in the dsh package is modified. What this file creates is a profile of its
own under DSH_HOME, composed from the same two bundles the shipped headless
profile uses, with the one-shot runner switched off and the multi-turn runner from
`dsh/versus_runner.mjs` inserted in its place. That is the same kind of scaffolding
paper 04's Hermes adapter was, and it is published with the paper.

Provenance note, which belongs in the paper rather than in a footnote: for Axium
and Orange the tool calls are captured by wrapping the dispatcher in process, so
the record is what ran. For dsh the record comes from its own append-only session
log. Both are evidence rather than self-report, but they are not the same
evidence, and a harness that fails to log a call would look like a harness that
did not make one.
"""
import io
import json
import os
import queue
import subprocess
import threading
import time
from datetime import datetime, timezone

import tiers as TI
from versus import adapters, graders as G

DSH_ROOT = r'C:\agent-eval-thirdparty\dsh'
DSH_BIN = os.path.join(DSH_ROOT, 'node_modules', '@deepseek-ai', 'dsh', 'lib', 'bin.js')
DSH_HOME = r'C:\agent-eval-thirdparty\dsh-home'
PROFILE = 'versus'
HERE = os.path.dirname(os.path.abspath(__file__))
RUNNER_SRC = os.path.join(HERE, 'dsh', 'versus_runner.mjs')
NODE = 'node'
TURN_TIMEOUT_S = int(os.environ.get('DSH_TURN_TIMEOUT', '600'))


def _args(value):
    """Tool arguments as a dict, whatever the harness handed over.

    dsh streams tool arguments and a block can carry them as a partial JSON
    string rather than an object. The graders iterate `.values()` on it, so a
    string reaches them as an AttributeError that kills the session after the
    turn has already been paid for. It is stored under `raw` instead, which keeps
    the evidence and lets the argument-matching checks see the text.
    """
    if isinstance(value, dict):
        return value
    if isinstance(value, str):
        try:
            parsed = json.loads(value)
            if isinstance(parsed, dict):
                return parsed
        except ValueError:
            pass
        return {'raw': value[:2000]}
    if value is None:
        return {}
    return {'raw': str(value)[:2000]}


def _write(path, body):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with io.open(path, 'w', encoding='utf-8') as f:
        f.write(body)


def ensure_profile():
    """Create the versus profile and install the multi-turn runner into it.

    Idempotent, and it rewrites the runner every time so an edit to the published
    source cannot silently fail to reach the profile.
    """
    prof = os.path.join(DSH_HOME, 'profiles', PROFILE)
    _write(os.path.join(prof, 'package.json'), json.dumps({
        'name': 'dsh-profile-versus',
        'private': True,
        'dependencies': {},
        'dsh': {'profile': {'bundles': ['@deepseek-ai/dsh-base',
                                        '@deepseek-ai/dsh-headless']}},
    }, indent=2) + '\n')
    _write(os.path.join(prof, 'cordis.yml'), '[]\n')
    # The shipped one-shot rows go quiet; everything else about the headless
    # bundle stays, so the persona, the tool mode and Code Mode are identical.
    _write(os.path.join(prof, 'cordis.patch.yml'), json.dumps([
        {'id': 'headless-runner', 'disabled': True},
        {'id': 'headless-startup', 'disabled': True},
        {'insert': [{'id': 'versus-runner', 'name': 'dsh-versus-runner'}]},
    ], indent=2) + '\n')

    pkg = os.path.join(DSH_HOME, 'profiles', 'node_modules', 'dsh-versus-runner')
    _write(os.path.join(pkg, 'package.json'), json.dumps({
        'name': 'dsh-versus-runner', 'version': '1.0.0',
        'type': 'module', 'main': 'index.mjs',
    }, indent=2) + '\n')
    with io.open(RUNNER_SRC, encoding='utf-8') as f:
        _write(os.path.join(pkg, 'index.mjs'), f.read())
    return prof


def tier_patch_path(tier):
    prof = os.path.join(DSH_HOME, 'profiles', PROFILE)
    path = os.path.join(prof, 'tier-%s.patch.json' % tier)
    # A JSON document is valid YAML, so the loader takes this as written and no
    # YAML emitter has to be trusted with the quoting.
    _write(path, json.dumps(TI.dsh_patch(tier), indent=2) + '\n')
    return path


class DshAdapter:
    name = 'dsh'

    def __init__(self, tier, permission_mode=None):
        if not os.path.exists(DSH_BIN):
            raise SystemExit('dsh is not installed at %s' % DSH_BIN)
        self.tier = tier
        self.t = TI.TIERS[tier]
        self.permission_mode = permission_mode or os.environ.get(
            'DSH_PERMISSION_MODE', 'workspace-write')
        ensure_profile()
        self.patch = tier_patch_path(tier)
        self.proc = None
        self._q = None
        self._stderr = []

    def label(self):
        return 'dsh[%s, %s]' % (self.t['model'], self.permission_mode)

    # ── process ──────────────────────────────────────────────────────────────
    def open_session(self, build, project):
        home = os.path.join(build, '.dsh-session')
        os.makedirs(home, exist_ok=True)
        env = dict(os.environ)
        env['DSH_HOME'] = DSH_HOME
        env['DSH_PERMISSION_MODE'] = self.permission_mode
        env['DSH_TELEMETRY_MODE'] = 'DISABLED'
        env['DSH_VERSUS_OUT'] = os.path.join(home, 'turns.jsonl')
        env[self.t['key_env']] = TI.key_for(self.tier)
        env['NO_COLOR'] = '1'

        self.proc = subprocess.Popen(
            [NODE, DSH_BIN, '--profile', PROFILE, '--patch', self.patch],
            cwd=build, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
            stderr=subprocess.PIPE, text=True, encoding='utf-8', errors='replace',
            bufsize=1)
        self._q = queue.Queue()
        self._stderr = []
        threading.Thread(target=self._pump, args=(self.proc.stdout, self._q),
                         daemon=True).start()
        threading.Thread(target=self._drain, args=(self.proc.stderr,),
                         daemon=True).start()
        return adapters.Session(self.name, build, project, home, time.time())

    @staticmethod
    def _pump(stream, q):
        for line in stream:
            q.put(line)
        q.put(None)

    def _drain(self, stream):
        for line in stream:
            self._stderr.append(line.rstrip())
            if len(self._stderr) > 400:
                del self._stderr[:200]

    # ── one turn ─────────────────────────────────────────────────────────────
    def send(self, session, text):
        before = G.tree_hash(session.build)
        t0 = time.time()
        result, error = None, None
        try:
            self.proc.stdin.write(json.dumps({'task': text}) + '\n')
            self.proc.stdin.flush()
            result = self._await_turn()
        except Exception as e:                              # noqa: BLE001
            error = '%s: %s' % (type(e).__name__, e)

        if result is None and error is None:
            rc = self.proc.poll() if self.proc else 'gone'
            how = ('harness exited rc=%s' % rc) if rc is not None \
                else 'no result before %ds' % TURN_TIMEOUT_S
            error = '%s; stderr tail: %s' % (
                how, ' | '.join(self._stderr[-3:])[:400])
        result = result or {}
        usage = result.get('usage') or {}
        when = datetime.now(timezone.utc)
        usd, rate = TI.price(self.t['model'], usage.get('input', 0),
                             usage.get('output', 0), usage.get('cached', 0), when)

        out = adapters.TurnResult(
            text=result.get('text', ''),
            tool_calls=[{'name': c.get('name', ''), 'args': _args(c.get('args'))}
                        for c in (result.get('tools') or [])],
            llm_calls=int(result.get('llm_calls') or 0),
            cost_usd=usd if self.t['priced'] else 0.0,
            input_tokens=int(usage.get('input') or 0),
            output_tokens=int(usage.get('output') or 0),
            cached_tokens=int(usage.get('cached') or 0),
            wall_s=time.time() - t0,
            asked=self._asked_from(result),
            error=error or self._error_from(result))
        out.rate = rate
        out.before, out.after = before, G.tree_hash(session.build)
        return out

    def _await_turn(self):
        deadline = time.time() + TURN_TIMEOUT_S
        while time.time() < deadline:
            # A dead process cannot answer, and waiting ten minutes to find that
            # out turns one crash into an hour of wasted sweep. The pump also
            # posts a sentinel, but only after the stream closes, which a hung
            # child never does.
            if self.proc.poll() is not None and self._q.empty():
                return None
            try:
                line = self._q.get(timeout=min(5.0, max(0.1, deadline - time.time())))
            except queue.Empty:
                continue
            if line is None:                                # process ended
                return None
            line = line.strip()
            if not line or not line.startswith('{'):
                continue                                    # harness chatter
            try:
                obj = json.loads(line)
            except ValueError:
                continue
            if 'ok' in obj:
                return obj
        return None

    @staticmethod
    def _asked_from(result):
        """Questions dsh put to the user, through its own question tool.

        Read from the tool calls rather than from the prose, so it lands in the
        same field Axium's and Orange's ask_user populate and V4 reads one
        channel for all three.
        """
        out = []
        for c in (result.get('tools') or []):
            if 'question' in (c.get('name') or '').lower():
                args = c.get('args') or {}
                q = args.get('question') or args.get('prompt') or json.dumps(args)
                out.append(str(q)[:400])
        return out

    @staticmethod
    def _error_from(result):
        if result.get('error'):
            return str(result['error'])[:400]
        reason = result.get('reason') or {}
        if reason.get('kind') == 'error':
            err = reason.get('error') or {}
            return '%s: %s' % (err.get('code', 'error'), err.get('message', ''))[:400]
        return None

    def close_session(self, session):
        if not self.proc:
            return
        try:
            self.proc.stdin.close()
        except Exception:                                   # noqa: BLE001
            pass
        try:
            self.proc.wait(timeout=60)
        except Exception:                                   # noqa: BLE001
            self.proc.kill()
        if self._stderr:
            with io.open(os.path.join(session.agent_home, 'stderr.log'), 'w',
                         encoding='utf-8') as f:
                f.write('\n'.join(self._stderr))
        self.proc = None