# -*- coding: utf-8 -*- """versus adapter for windows-agent (`pc-agent.py`), the deliberately simple one. Eight tools, one LangGraph loop, no memory store, no compaction beyond a summariser, no classifier and no planner. It is in this study as the control for "how much scaffolding do you actually need", and it is the author's, which is declared. Driven in process rather than as a subprocess, for the same reason Axium and Orange are: the versus grader hashes the project tree between turns, so the turns cannot be batched, and its one-shot entry point (`run_single_task`) starts a fresh message list every call, which would deny the harness the history that three of the five scenarios exist to measure. This adapter keeps the message list across turns, which is what its own interactive loop does. Nothing is written to the windows-agent repository. Four state paths that otherwise sit beside the script are redirected into the build, and the local rung gets a base URL that `create_llm` has no argument for, wrapped in memory here rather than edited into the file so the measured artefact stays as it ships. """ import importlib.util import io import os import sys import time import tiers as TI from versus import adapters, graders as G AGENT_PATH = r'C:\xampp\htdocs\windows-agent\pc-agent.py' VENV_HINT = r'C:\xampp\htdocs\windows-agent\.venv\Lib\site-packages' _MODULE = None def _load(): """Import pc-agent.py once. Its main loop is guarded, so import is safe.""" global _MODULE if _MODULE is not None: return _MODULE try: import langchain_deepseek # noqa: F401 except ImportError: # The harness's own venv holds its pinned versions. Borrowing it works # only when the interpreter matches: its native extensions are built for # the Python that created it, and a 3.14 process loading a 3.12 venv dies # on pydantic_core rather than on anything to do with the harness. if os.path.isdir(VENV_HINT) and VENV_HINT not in sys.path: sys.path.insert(0, VENV_HINT) spec = importlib.util.spec_from_file_location('pc_agent', AGENT_PATH) mod = importlib.util.module_from_spec(spec) sys.modules['pc_agent'] = mod spec.loader.exec_module(mod) _MODULE = mod return mod class WinAgentAdapter: name = 'winagent' def __init__(self, tier): self.tier = tier self.t = TI.TIERS[tier] self.m = _load() self._cwd0 = None self._messages = [] self._paths0 = {} self._pin_model() def label(self): return 'winagent[%s]' % self.t['model'] # ── rung ───────────────────────────────────────────────────────────────── def _pin_model(self): m = self.m t = self.t provider = 'openai' if t['wire'] in ('openai', 'openai-cloud') else 'deepseek' key = TI.key_for(self.tier) if provider == 'openai': m.OPENAI_API_KEY = key # Only the local rung repoints the endpoint. The hosted one is the # vendor's own, which the client already knows. base = t['base_url'] if t['wire'] == 'openai' else None else: m.DEEPSEEK_API_KEY = key base = None # Current OpenAI models refuse function tools on chat completions with any # reasoning level, so a reasoning rung has to go through the Responses # API. langchain exposes that as a flag, which is the whole of the change # here; without it the rung is unreachable for this harness. hosted = (provider == 'openai' and t['wire'] == 'openai-cloud' and str(t['model']).startswith(('gpt-5', 'o1', 'o3', 'o4'))) effort = t['effort'] if not getattr(m, '_versus_base_url_patch', False): real = m.create_llm def create_llm(prov, model, _real=real): llm = _real(prov, model) if hosted: for attr, value in (('use_responses_api', True), ('reasoning', {'effort': effort or 'medium'})): try: setattr(llm, attr, value) except Exception: # noqa: BLE001 pass if base: # langchain's OpenAI client keeps the endpoint on the client # object; setting it after construction avoids depending on # which keyword this version of the package accepts. for attr in ('openai_api_base', 'base_url'): if hasattr(llm, attr): try: setattr(llm, attr, base) except Exception: # noqa: BLE001 pass client = getattr(llm, 'client', None) for obj in (client, getattr(llm, 'async_client', None)): inner = getattr(obj, '_client', None) if inner is not None and hasattr(inner, 'base_url'): try: inner.base_url = base except Exception: # noqa: BLE001 pass return llm m.create_llm = create_llm m._versus_base_url_patch = True # noqa: SLF001 m.rebuild_graph(provider, t['model']) # ── session ────────────────────────────────────────────────────────────── def open_session(self, build, project): m = self.m home = os.path.join(build, '.pc-agent') os.makedirs(home, exist_ok=True) self._paths0 = {k: getattr(m, k) for k in ('SESSION_FILE', 'BACKUP_DIR', 'LEDGER_FILE')} m.SESSION_FILE = os.path.join(home, 'session.json') m.BACKUP_DIR = os.path.join(home, 'backups') m.LEDGER_FILE = os.path.join(home, 'ledger.json') self._cwd0 = os.getcwd() os.chdir(build) # its tools are cwd-relative facts = '[THIS MACHINE - measured, not assumed]\n' + m.machine_profile() self._messages = [m.SystemMessage(content=m.SYSTEM_PROMPT), m.SystemMessage(content=facts)] return adapters.Session(self.name, build, project, home, time.time()) def send(self, session, text): m = self.m before = G.tree_hash(session.build) t0 = time.time() start = len(self._messages) self._messages.append(m.HumanMessage(content=text)) m.CURRENT_USER_REQUEST = text error = None try: out = m._graph.invoke({'messages': self._messages}, config={'recursion_limit': m.MAX_STEPS_LIMIT}) self._messages = out['messages'] except Exception as e: # noqa: BLE001 error = '%s: %s' % (type(e).__name__, e) fresh = self._messages[start:] tools, calls, usage, asked = [], 0, {'in': 0, 'out': 0, 'cache': 0}, [] final = '' for msg in fresh: if not isinstance(msg, m.AIMessage): continue calls += 1 for c in (getattr(msg, 'tool_calls', None) or []): name = c.get('name') if isinstance(c, dict) else getattr(c, 'name', '') args = c.get('args') if isinstance(c, dict) else getattr(c, 'args', {}) if not isinstance(args, dict): # LangChain can hand back a string when the model streamed # partial arguments. The graders call .values() on it. args = {'raw': str(args)[:2000]} if args else {} tools.append({'name': name or '', 'args': dict(args)}) if 'ask' in (name or '').lower() or 'question' in (name or '').lower(): asked.append(str(args)[:400]) u = getattr(msg, 'usage_metadata', None) or {} usage['in'] += int(u.get('input_tokens') or 0) usage['out'] += int(u.get('output_tokens') or 0) details = u.get('input_token_details') or {} usage['cache'] += int(details.get('cache_read') or 0) if msg.content: final = str(msg.content) from datetime import datetime, timezone usd, rate = TI.price(self.t['model'], usage['in'], usage['out'], usage['cache'], datetime.now(timezone.utc)) res = adapters.TurnResult( text=final, tool_calls=tools, llm_calls=calls, cost_usd=usd if self.t['priced'] else 0.0, input_tokens=usage['in'], output_tokens=usage['out'], cached_tokens=usage['cache'], wall_s=time.time() - t0, asked=asked, error=error) res.rate = rate res.before, res.after = before, G.tree_hash(session.build) return res def close_session(self, session): m = self.m if self._cwd0: try: os.chdir(self._cwd0) except OSError: pass self._cwd0 = None for k, v in (self._paths0 or {}).items(): setattr(m, k, v) self._paths0 = {} self._messages = []