DeepSeek Harness 拆解的维护记录

"""Send a request body (JSON file) to DeepSeek's Anthropic-compatible Messages API via the wiki gateway.
Strips dsh_* top-level fields (session-log upload), never prints the key. Saves raw response."""
import json, os, sys, pathlib, urllib.request

ENV = pathlib.Path('/data2/work/lht/study/26.06/zensical-wiki/.env')
BASE = 'https://deepseek-proxy.liuhetian.work/anthropic/v1/messages'

def key():
    for line in ENV.read_text().splitlines():
        if line.startswith('MY_API_KEY='):
            return line.split('=', 1)[1].strip().strip('"').strip("'")
    raise SystemExit('MY_API_KEY missing')

def call(body, headers=None):
    body = {k: v for k, v in body.items() if not k.startswith('dsh_')}
    h = {'content-type': 'application/json', 'authorization': 'Bearer ' + key(), 'anthropic-version': '2023-06-01'}
    h.update(headers or {})
    req = urllib.request.Request(BASE, data=json.dumps(body).encode(), headers=h, method='POST')
    with urllib.request.urlopen(req, timeout=600) as r:
        return r.read().decode()

if __name__ == '__main__':
    src, dst = sys.argv[1], sys.argv[2]
    body = json.loads(pathlib.Path(src).read_text())
    hdrs = json.loads(sys.argv[3]) if len(sys.argv) > 3 else None
    pathlib.Path(dst).write_text(call(body, hdrs))
    print('saved', dst)

"""Local relay for the agents-line experiment.

dsh runs in a key-less process with DEEPSEEK_BASE_URL=http://127.0.0.1:<port>/anthropic
and a dummy DEEPSEEK_API_KEY. This separate process holds the real key (read by
../ds_call.py from the wiki .env), strips the dsh_* top-level fields (session-log
upload), forwards the request to the DeepSeek Anthropic-compatible endpoint via the
wiki gateway, returns the raw response, and records every call:
  out/calls/NNN.request.json   the body dsh built (after stripping dsh_*)
  out/calls/NNN.response.txt   the raw response (SSE when dsh streams)
  out/calls/index.jsonl        one line per call: model, message count, usage
Usage: python3 relay.py <port> <out-dir>
"""
import json, pathlib, sys, threading, urllib.error, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
import ds_call  # noqa: E402  (shared caller: key lookup + gateway base)

UPSTREAM = ds_call.BASE.rsplit('/v1/messages', 1)[0]  # .../anthropic
PORT, OUT = int(sys.argv[1]), pathlib.Path(sys.argv[2]) / 'calls'
OUT.mkdir(parents=True, exist_ok=True)
lock, counter = threading.Lock(), [0]
PASS_HEADERS = ('content-type', 'accept', 'anthropic-version', 'anthropic-beta')


def usage_of(raw: str) -> dict:
    usage = {}
    for line in raw.splitlines():
        if not line.startswith('data:'):
            continue
        try:
            ev = json.loads(line[5:])
        except ValueError:
            continue
        if ev.get('type') == 'message_start':
            usage.update(ev['message'].get('usage', {}))
        elif ev.get('type') == 'message_delta':
            usage.update(ev.get('usage', {}))
    if not usage:
        try:
            usage = json.loads(raw).get('usage', {})
        except ValueError:
            pass
    return usage


class Relay(BaseHTTPRequestHandler):
    def _forward(self, method: str) -> None:
        path = self.path.split('/anthropic', 1)[-1]
        body = None
        if method == 'POST':
            data = json.loads(self.rfile.read(int(self.headers['content-length'])))
            data = {k: v for k, v in data.items() if not k.startswith('dsh_')}
            body = json.dumps(data).encode()
        headers = {k: self.headers[k] for k in PASS_HEADERS if self.headers.get(k)}
        headers['authorization'] = 'Bearer ' + ds_call.key()
        req = urllib.request.Request(UPSTREAM + path, data=body, headers=headers, method=method)
        try:
            with urllib.request.urlopen(req, timeout=600) as r:
                status, ctype, raw = r.status, r.headers.get('content-type', ''), r.read()
        except urllib.error.HTTPError as e:
            status, ctype, raw = e.code, e.headers.get('content-type', ''), e.read()
        if body is not None:
            with lock:
                counter[0] += 1
                n = counter[0]
            text = raw.decode('utf-8', 'replace')
            (OUT / f'{n:03d}.request.json').write_text(json.dumps(data, ensure_ascii=False, indent=1))
            (OUT / f'{n:03d}.response.txt').write_text(text)
            with lock, open(OUT / 'index.jsonl', 'a') as f:
                f.write(json.dumps({'n': n, 'path': path, 'status': status, 'model': data.get('model'),
                                    'messages': len(data.get('messages', [])), 'usage': usage_of(text)}) + '\n')
        self.send_response(status)
        self.send_header('content-type', ctype)
        self.send_header('content-length', str(len(raw)))
        self.end_headers()
        self.wfile.write(raw)

    def do_POST(self):
        self._forward('POST')

    def do_GET(self):
        self._forward('GET')

    def log_message(self, *args):
        pass


ThreadingHTTPServer(('127.0.0.1', PORT), Relay).serve_forever()

"""本机转发：dsh 的 DeepSeek 适配器把 baseURL 指到这里，这个进程换上 key 转给 DeepSeek 的
Anthropic 兼容接口（经本站中转），原样回传 SSE。跑 dsh 代码的进程里没有任何密钥。

每次请求存一对文件：out/proxy/NNN-request.json（剥掉 dsh_* 顶层字段后的请求体）、
NNN-response.sse（原始返回），并在 out/proxy/calls.jsonl 追加一行用量。

用法：python3 proxy.py <port> <out_dir>
"""
import json, pathlib, sys, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
from ds_call import key  # noqa: E402  共享脚本里读 .env 的 MY_API_KEY，不打印

UPSTREAM = 'https://deepseek-proxy.liuhetian.work/anthropic'
PASS_HEADERS = ('content-type', 'anthropic-version', 'anthropic-beta', 'accept')

PORT, OUT = int(sys.argv[1]), pathlib.Path(sys.argv[2])
OUT.mkdir(parents=True, exist_ok=True)
count = 0


def usage_of(sse: str) -> dict:
    usage = {}
    for line in sse.splitlines():
        if not line.startswith('data: '):
            continue
        try:
            event = json.loads(line[6:])
        except ValueError:
            continue
        if event.get('type') == 'message_start':
            usage.update(event['message'].get('usage', {}))
        elif event.get('type') == 'message_delta':
            usage.update(event.get('usage', {}))
    return usage


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        global count
        count += 1
        n = count
        raw = self.rfile.read(int(self.headers['content-length']))
        body = {k: v for k, v in json.loads(raw).items() if not k.startswith('dsh_')}
        (OUT / f'{n:03d}-request.json').write_text(json.dumps(body, ensure_ascii=False, indent=2))
        headers = {h: self.headers[h] for h in PASS_HEADERS if self.headers.get(h)}
        headers['authorization'] = 'Bearer ' + key()
        req = urllib.request.Request(UPSTREAM + self.path, data=json.dumps(body).encode(),
                                     headers=headers, method='POST')
        try:
            with urllib.request.urlopen(req, timeout=900) as r:
                status, ctype, data = r.status, r.headers.get('content-type'), r.read()
        except urllib.error.HTTPError as e:
            status, ctype, data = e.code, e.headers.get('content-type'), e.read()
        text = data.decode()
        (OUT / f'{n:03d}-response.sse').write_text(text)
        purpose = 'compaction' if self.headers.get('x-deepseek-harness-compact') == '1' else 'conversation'
        with open(OUT / 'calls.jsonl', 'a') as f:
            f.write(json.dumps({'n': n, 'status': status, 'purpose': purpose, 'model': body.get('model'),
                                'messages': len(body.get('messages', [])), 'usage': usage_of(text)}) + '\n')
        self.send_response(status)
        self.send_header('content-type', ctype or 'text/event-stream')
        self.send_header('content-length', str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, *args):
        pass


ThreadingHTTPServer(('127.0.0.1', PORT), Handler).serve_forever()

"""对照组：同一段被压的对话，按 baseline（pi 1.3.0）的格式另起一次摘要请求，看前缀缓存命中多少。

输入是 dsh 实跑里那次压缩请求（proxy 记下的 NNN-request.json）：去掉末尾 dsh 的压缩指令，
其余消息按 pi 的 serializeConversation 格式转成一段文本（[User]: / [Assistant]: /
[Assistant tool calls]: / [Tool result]:，工具结果截到 2000 字符），包进 <conversation>，
后接 pi 的 SUMMARIZATION_PROMPT；system 换成 pi 的 SUMMARIZATION_SYSTEM_PROMPT，不带 tools。
模型、max_tokens、thinking、effort 和 dsh 那次一致。pi 的两段提示词原文照抄自
@earendil-works/pi-coding-agent 1.3.0 的 dist/core/compaction/{utils,compaction}.js。

用法：python3 control.py <dsh-compaction-request.json> <out_dir>
"""
import json, pathlib, sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
from ds_call import call  # noqa: E402

PI_SYSTEM = """You are a context summarization assistant. Your task is to read a conversation between a user and an AI assistant, then produce a structured summary following the exact format specified.

Do NOT continue the conversation. Do NOT respond to any questions in the conversation. ONLY output the structured summary."""

PI_PROMPT = """The messages above are a conversation to summarize. Create a structured context checkpoint summary that another LLM will use to continue the work.

Use this EXACT format:

## Goal
[What is the user trying to accomplish? Can be multiple items if the session covers different tasks.]

## Constraints & Preferences
- [Any constraints, preferences, or requirements mentioned by user]
- [Or "(none)" if none were mentioned]

## Progress
### Done
- [x] [Completed tasks/changes]

### In Progress
- [ ] [Current work]

### Blocked
- [Issues preventing progress, if any]

## Key Decisions
- **[Decision]**: [Brief rationale]

## Next Steps
1. [Ordered list of what should happen next]

## Critical Context
- [Any data, examples, or references needed to continue]
- [Or "(none)" if not applicable]

Keep each section concise. Preserve exact file paths, function names, and error messages."""

TOOL_RESULT_MAX_CHARS = 2000


def text_of(blocks):
    if isinstance(blocks, str):
        return blocks
    return '\n'.join(b['text'] for b in blocks if b.get('type') == 'text')


def truncate(text, limit):
    return text if len(text) <= limit else f'{text[:limit]}\n\n[... {len(text) - limit} more characters truncated]'


def serialize(messages):
    parts = []
    for m in messages:
        blocks = m['content'] if isinstance(m['content'], list) else [{'type': 'text', 'text': m['content']}]
        if m['role'] == 'user':
            for b in blocks:
                if b['type'] == 'text' and b['text']:
                    parts.append(f"[User]: {b['text']}")
                elif b['type'] == 'tool_result':
                    content = text_of(b.get('content', ''))
                    if content:
                        parts.append(f'[Tool result]: {truncate(content, TOOL_RESULT_MAX_CHARS)}')
        else:
            texts = [b['text'] for b in blocks if b['type'] == 'text']
            calls = [f"{b['name']}(" + ', '.join(f'{k}={json.dumps(v, ensure_ascii=False)}' for k, v in b['input'].items()) + ')'
                     for b in blocks if b['type'] == 'tool_use']
            if texts:
                parts.append('[Assistant]: ' + '\n'.join(texts))
            if calls:
                parts.append('[Assistant tool calls]: ' + '; '.join(calls))
    return '\n\n'.join(parts)


src, out = pathlib.Path(sys.argv[1]), pathlib.Path(sys.argv[2])
dsh = json.loads(src.read_text())
messages = json.loads(json.dumps(dsh['messages']))
last = messages[-1]
# dsh 把压缩指令作为最后一条 user 的最后一个文本块；对照组把它拿掉，换成 pi 的提示词
assert last['role'] == 'user' and last['content'][-1]['type'] == 'text' and 'compaction engine' in last['content'][-1]['text']
last['content'] = last['content'][:-1]
if not last['content']:
    messages.pop()

body = {
    'model': dsh['model'],
    'max_tokens': dsh['max_tokens'],
    'thinking': dsh['thinking'],
    'output_config': dsh['output_config'],
    'system': PI_SYSTEM,
    'messages': [{'role': 'user', 'content': [{'type': 'text', 'text': f'<conversation>\n{serialize(messages)}\n</conversation>\n\n{PI_PROMPT}'}]}],
}
out.mkdir(parents=True, exist_ok=True)
(out / 'control-request.json').write_text(json.dumps(body, ensure_ascii=False, indent=2))
resp = call(body)
(out / 'control-response.json').write_text(resp)
print(json.dumps(json.loads(resp)['usage']))