v0.1.2-alpha.1中PTC模式被历史session写为code #5099
Unanswered
Ghost-Kokoro
asked this question in
Q&A
Replies: 3 comments 1 reply
|
这是 预设 ID 重命名 导致的旧会话锁定，不是 skill/模型配置丢了。 报错已经点名：
临时绕过（先备份）：
更稳妥： 在新版本下用 若改完仍报错，贴一下 session 里和 preset 相关的那几行字段（可打码路径），以及 |
1 reply
|
感谢回复，旧会话已无法继续，agent-preset预设写成ptc也无济于事，只能新开session。 |
|
写了一个恢复用的脚本(ptc-fix)，基本恢复了，我自己是恢复了 ./ptc-fix fix --all # code->ptc
./ptc-fix cache # 重建subagent缓存
./ptc-fix desc --all # 修复subagent 的消息格式版本#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
fix - DSH 会话预设修复工具（code -> ptc）
用法:
fix 显示帮助
fix fix --all 修复全部 code 会话 -> ptc
fix fix --session <ID>... 只修复指定会话
fix fix --dir <DIR>... 只修复指定会话目录
fix fix --list 列出 code 会话（只读）
fix fix ... --dry-run 预览，不修改
fix desc --all 修复子代理 descriptor 版本 v2 -> v3
fix desc --list 列出 descriptor 版本不兼容的会话
fix restore 从备份还原已修复会话
fix restore --dry-run 预览还原
子命令:
fix 修复 code 预设会话为 ptc（备份 -> 修改 -> 异常还原 -> 终验）
desc 修复子代理 descriptor 版本 v2 -> v3（解决"会话记录损坏"）
restore 从 .bak-code 备份还原会话
全局选项:
--root <DIR> 会话根目录（默认 ~/.dsh/sessions）
"""
import argparse
import json
import os
import shutil
import subprocess
import sys
DEFAULT_ROOT = os.path.expanduser("~/.dsh/sessions")
BACKUP_SUFFIX = ".bak-code"
MANIFEST_NAME = ".fix-preset-manifest.json"
SESSION_FILE = "session.jsonl.zstd"
NL = b"\n"
# ---------------- 基础 ----------------
def zstd():
return os.environ.get("ZSTD_BIN", "zstd")
def run(cmd, data=None, timeout=300):
return subprocess.run(cmd, input=data, capture_output=True, timeout=timeout)
def decompress(path):
p = run([zstd(), "-d", "-c", path])
if p.returncode != 0:
raise RuntimeError("zstd 解压失败: " + p.stderr.decode(errors="replace")[:200])
return p.stdout
def compress(data, out_path):
p = run([zstd(), "-f", "-q", "-o", out_path], data=data)
if p.returncode != 0:
raise RuntimeError("zstd 压缩失败: " + p.stderr.decode(errors="replace")[:200])
def zstd_ok(path):
return run([zstd(), "-t", "-q", path]).returncode == 0
ZSTD_MAGIC = 4247762216 # 0xFD2FB528
def scan_frames(data):
"""扫描 zstd 多帧结构，返回 [(start, end), ...] 与是否 torn。
完全复刻 DSH dsh-session-persistence-jsonl 的 scanZstdFrames。"""
frames = []
offset = 0
n = len(data)
while offset < n:
start = offset
if n - offset < 4:
return frames, True
if int.from_bytes(data[offset:offset+4], "little") != ZSTD_MAGIC:
raise RuntimeError("invalid frame magic at byte %d" % offset)
offset += 4
if offset == n:
return frames, True
descriptor = data[offset]
offset += 1
if (descriptor & 24) != 0:
raise RuntimeError("reserved frame-header bit")
content_size_flag = descriptor >> 6
single_segment = (descriptor & 32) != 0
checksum = (descriptor & 4) != 0
dictionary_flag = descriptor & 3
dictionary_bytes = 4 if dictionary_flag == 3 else dictionary_flag
content_size_bytes = (1 if single_segment else 0) if content_size_flag == 0 else 1 << content_size_flag
remaining = (0 if single_segment else 1) + dictionary_bytes + content_size_bytes
if n - offset < remaining:
return frames, True
offset += remaining
while True:
if n - offset < 3:
return frames, True
block_header = int.from_bytes(data[offset:offset+3], "little")
offset += 3
last_block = (block_header & 1) != 0
block_type = (block_header >> 1) & 3
block_size = block_header >> 3
if block_type == 3:
raise RuntimeError("reserved block type")
payload = 1 if block_type == 1 else block_size
if n - offset < payload:
return frames, True
offset += payload
if last_block:
break
if checksum:
if n - offset < 4:
return frames, True
offset += 4
frames.append((start, offset))
return frames, False
def read_header(path):
"""读取会话 header：只解压第一个 zstd frame（header 帧），返回 (header, 首行, 剩余字节)"""
data = open(path, "rb").read()
frames, torn = scan_frames(data)
if not frames:
raise RuntimeError("无完整首帧（torn=%s）" % torn)
start, end = frames[0]
frame = data[start:end]
p = run([zstd(), "-d", "-c"], data=frame)
if p.returncode != 0:
raise RuntimeError("首帧解压失败: " + p.stderr.decode(errors="replace")[:200])
plain = p.stdout
# DSH assertZstdHeaderFrame: 恰好一行
if len(plain) == 0 or plain.find(NL) != len(plain) - 1:
raise RuntimeError("首帧不是单行 header")
first = plain[:-1]
try:
h = json.loads(first.decode("utf-8"))
except Exception as e:
raise RuntimeError("首行 JSON 解析失败: %s" % e)
if not isinstance(h, dict):
raise RuntimeError("首行不是对象")
return h, first, data[end:] # 剩余 = 第一个 frame 之后的所有字节（保留后续帧）
def find_sessions(root):
"""yield (session_id, dir) 为每个含 session.jsonl.zstd 的目录"""
if not os.path.isdir(root):
return
for wd in sorted(os.listdir(root)):
wpath = os.path.join(root, wd)
if not os.path.isdir(wpath):
continue
for sd in sorted(os.listdir(wpath)):
d = os.path.join(wpath, sd)
if os.path.isdir(d) and os.path.isfile(os.path.join(d, SESSION_FILE)):
yield sd, d
# ---------------- 目标收集 ----------------
def collect(args, root):
targets, seen = [], set()
skip = set(args.skip or [])
def add(sid, d):
if sid in skip:
return
if d not in seen:
seen.add(d)
targets.append((sid, d))
if args.dir:
for d in args.dir:
d = os.path.abspath(d)
if not os.path.isfile(os.path.join(d, SESSION_FILE)):
print("[警告] 目录里没有 %s: %s" % (SESSION_FILE, d), file=sys.stderr)
continue
add(os.path.basename(d.rstrip(os.sep)), d)
if args.session:
wanted = set(args.session)
for sid, d in find_sessions(root):
if sid in wanted:
add(sid, d)
missing = sorted(wanted - {s for s, _ in targets})
for m in missing:
print("[警告] 找不到会话: %s" % m, file=sys.stderr)
if args.all:
for sid, d in find_sessions(root):
add(sid, d)
return targets
# ---------------- 修复 ----------------
def migrate_one(sid, d, to, backup):
"""返回 (status, msg)。status: ok / skip / failed（failed 时已尽量还原）"""
zpath = os.path.join(d, SESSION_FILE)
bak = zpath + BACKUP_SUFFIX
try:
h, first, rest = read_header(zpath)
cur = h.get("agentPreset")
if cur is None:
return "skip", "无 agentPreset 字段"
if cur != "code":
return "skip", "预设=%r（非 code）" % cur
if cur == to:
return "skip", "已是 %r" % to
if backup and not os.path.exists(bak):
shutil.copy2(zpath, bak) # 1. 备份
h["agentPreset"] = to # 2. 修改 header 帧，保留后续帧（多帧结构不变）
new_first = json.dumps(h, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + NL
# 新 header 帧：独立压缩（zstd CLI 默认带 checksum，与 DSH 一致）
tmp_header = zpath + ".fix-hdr"
try:
with open(tmp_header, "wb") as f:
f.write(new_first)
tmp = zpath + ".fix-tmp"
p = run([zstd(), "-f", "-q", "-o", tmp, tmp_header])
if p.returncode != 0:
raise RuntimeError("header 帧压缩失败: " + p.stderr.decode(errors="replace")[:200])
os.unlink(tmp_header)
# 拼接：新 header 帧 + 原始后续帧
new_hdr = open(tmp, "rb").read()
with open(zpath, "rb") as f:
raw = f.read()
_, _torn = scan_frames(raw)
frames, _t = scan_frames(raw)
end1 = frames[0][1]
result = new_hdr + raw[end1:]
with open(tmp, "wb") as f:
f.write(result)
os.replace(tmp, zpath)
except Exception as e:
for t in (tmp_header, tmp) if 'tmp' in dir() else (tmp_header,):
pass
for t in (tmp_header, zpath + ".fix-tmp"):
if os.path.exists(t):
os.unlink(t)
if os.path.exists(bak):
shutil.copy2(bak, zpath) # 3. 异常还原
return "failed", "修改失败已还原: %s" % e
return "failed", "修改失败且无备份: %s" % e
# 终验：frame 结构 + 首帧单行 + header 值
try:
raw2 = open(zpath, "rb").read()
frames2, torn2 = scan_frames(raw2)
h2, _, _ = read_header(zpath)
ok = (not torn2) and len(frames2) >= 1 and h2.get("agentPreset") == to
except Exception:
ok = False
if not ok:
if os.path.exists(bak):
shutil.copy2(bak, zpath)
return "failed", "校验失败已还原"
return "failed", "校验失败且无备份"
return "ok", "code -> %s" % to
except Exception as e:
if os.path.exists(bak):
try:
shutil.copy2(bak, zpath)
return "failed", "异常已还原: %s" % e
except Exception:
pass
return "failed", "异常: %s" % e
def do_fix(args, root):
# --list 未指定选择器时默认扫描全部 code 会话
if args.list and not (args.all or args.session or args.dir):
args.all = True
targets = collect(args, root)
plan = []
skipped_reasons = []
for sid, d in targets:
try:
h, _, _ = read_header(os.path.join(d, SESSION_FILE))
plan.append((sid, d, h.get("agentPreset"), h.get("origin", "?")))
except Exception as e:
skipped_reasons.append((sid, str(e)))
print("[跳过] %s: %s" % (sid, e), file=sys.stderr)
code_plan = [p for p in plan if p[2] == "code"]
if not code_plan:
print("没有 code 预设会话需要修复")
return 0
if args.dry_run or args.list:
print("== 待修复 %d 个 code 会话 -> '%s' ==" % (len(code_plan), args.to))
for sid, d, _, origin in sorted(code_plan):
print(" %s origin=%s" % (sid, origin))
print("（dry-run，未修改任何文件）")
return 0
print("== 迁移 %d 个会话: code -> %s ==" % (len(code_plan), args.to))
stats = {"ok": 0, "skip": 0, "failed": 0}
manifest = {"to": args.to, "entries": []}
for sid, d, _, _ in sorted(code_plan):
status, msg = migrate_one(sid, d, args.to, backup=not args.no_backup)
stats[status] += 1
tag = {"ok": "OK ", "skip": "跳过", "failed": "失败"}.get(status, status)
print("[%s] %s %s" % (tag, sid, msg))
manifest["entries"].append({"sessionId": sid, "dir": d, "status": status, "msg": msg})
print("== 终验 ==")
vok = 0
for sid, d, _, _ in sorted(code_plan):
zpath = os.path.join(d, SESSION_FILE)
try:
if read_header(zpath)[0].get("agentPreset") == args.to and zstd_ok(zpath):
vok += 1
else:
print("[终验失败] %s" % sid, file=sys.stderr)
except Exception as e:
print("[终验失败] %s: %s" % (sid, e), file=sys.stderr)
print("通过 %d/%d" % (vok, len(code_plan)))
mpath = os.path.join(root, MANIFEST_NAME)
try:
with open(mpath, "w") as f:
json.dump(manifest, f, ensure_ascii=False, indent=2)
print("清单: %s" % mpath)
except Exception as e:
print("[警告] 清单写入失败: %s" % e, file=sys.stderr)
skipped_by_user = len(set(args.skip or []))
print("结果: ok=%d skip=%d failed=%d（备份: *.bak-code）" % (
stats["ok"], stats["skip"] + len(skipped_reasons), stats["failed"]))
if args.skip:
print("已跳过用户指定会话: %s" % ", ".join(sorted(set(args.skip))))
if skipped_reasons:
print("已跳过损坏会话: %s" % ", ".join(s for s, _ in skipped_reasons))
return 0 if stats["failed"] == 0 else 1
# ---------------- 还原 ----------------
def do_restore(args, root):
pairs = []
for sid, d in find_sessions(root):
bak = os.path.join(d, SESSION_FILE) + BACKUP_SUFFIX
if os.path.isfile(bak):
pairs.append((sid, d, bak))
if not pairs:
print("没有可还原的备份（*.bak-code）")
return 0
if args.dry_run:
print("== 待还原 %d 个会话（从 *.bak-code）==" % len(pairs))
for sid, d, _ in sorted(pairs):
print(" %s" % sid)
print("（dry-run，未修改任何文件）")
return 0
ok = fail = 0
for sid, d, bak in sorted(pairs):
try:
shutil.copy2(bak, os.path.join(d, SESSION_FILE))
ok += 1
print("[还原] %s" % sid)
except Exception as e:
fail += 1
print("[还原失败] %s: %s" % (sid, e), file=sys.stderr)
print("还原完成: %d 成功, %d 失败" % (ok, fail))
return 0 if fail == 0 else 1
# ---------------- desc 修复（descriptor v2 -> v3） ----------------
ZSTD_MAGIC = 4247762216 # 0xFD2FB528
def scan_frames_bin(data):
"""扫描 zstd 多帧结构，返回 [(start, end), ...] 与是否 torn。"""
frames = []
offset = 0
n = len(data)
while offset < n:
start = offset
if n - offset < 4:
return frames, True
if int.from_bytes(data[offset:offset+4], "little") != ZSTD_MAGIC:
raise RuntimeError("invalid frame magic at byte %d" % offset)
offset += 4
if offset == n:
return frames, True
descriptor = data[offset]
offset += 1
if (descriptor & 24) != 0:
raise RuntimeError("reserved frame-header bit")
csf = descriptor >> 6
ss = (descriptor & 32) != 0
ck = (descriptor & 4) != 0
df = descriptor & 3
db = 4 if df == 3 else df
cb = (1 if ss else 0) if csf == 0 else 1 << csf
rh = (0 if ss else 1) + db + cb
if n - offset < rh:
return frames, True
offset += rh
while True:
if n - offset < 3:
return frames, True
bh = int.from_bytes(data[offset:offset+3], "little")
offset += 3
last = (bh & 1) != 0
bt = (bh >> 1) & 3
bs = bh >> 3
if bt == 3:
raise RuntimeError("reserved block type")
payload = 1 if bt == 1 else bs
if n - offset < payload:
return frames, True
offset += payload
if last:
break
if ck:
if n - offset < 4:
return frames, True
offset += 4
frames.append((start, offset))
return frames, False
def compress_one_frame(data_bytes):
"""把一段 JSONL 独立压缩成一个带 checksum 的 zstd frame（与 DSH 一致）"""
tmp_in = "/tmp/fix-desc-in.jsonl"
tmp_out = "/tmp/fix-desc-out.zst"
with open(tmp_in, "wb") as f:
f.write(data_bytes)
p = run([zstd(), "-f", "-q", "-o", tmp_out, tmp_in])
if p.returncode != 0:
raise RuntimeError("frame 压缩失败: " + p.stderr.decode(errors="replace")[:200])
with open(tmp_out, "rb") as f:
return f.read()
def fix_descriptor_in_file(zpath):
"""修复一个会话文件：descriptor v2 -> v3。返回 (changed, 是否成功)"""
with open(zpath, "rb") as f:
raw = f.read()
frames, torn = scan_frames_bin(raw)
if not frames or torn:
return False, False # 结构不完整，无法处理
# 解压所有帧
decoded = []
for (start, end) in frames:
p = run([zstd(), "-d", "-c"], data=raw[start:end])
if p.returncode != 0:
return False, False
decoded.append(p.stdout)
# 逐帧查找 descriptor v2 并修改
changed_any = False
for i in range(1, len(decoded)):
text = decoded[i].decode("utf-8", errors="replace")
lines = text.split(chr(10))
modified = False
out_lines = []
for line in lines:
if not line.strip():
out_lines.append(line)
continue
try:
parsed = json.loads(line)
except Exception:
out_lines.append(line)
continue
arr = parsed if isinstance(parsed, list) else [parsed]
hit = False
for ev in arr:
if (isinstance(ev, dict) and ev.get("type") == "subagent/descriptor"
and isinstance(ev.get("data"), dict)
and ev["data"].get("version") == 2):
ev["data"]["version"] = 3
hit = True
if hit:
modified = True
if isinstance(parsed, list):
out_lines.append(json.dumps(parsed, ensure_ascii=False, separators=(",", ":")))
else:
out_lines.append(json.dumps(parsed, ensure_ascii=False, separators=(",", ":")))
else:
out_lines.append(line)
if modified:
decoded[i] = chr(10).join(out_lines).encode("utf-8")
changed_any = True
if not changed_any:
return False, True # 无 v2 descriptor（无需改）
# 重新压缩所有帧并拼接
result = b""
for i, frame_data in enumerate(decoded):
if i == 0:
# header 帧必须保持单行
result += compress_one_frame(frame_data)
else:
result += compress_one_frame(frame_data)
with open(zpath, "wb") as f:
f.write(result)
return True, True
def do_desc(args, root):
"""修复子代理 descriptor v2 -> v3（只处理有 .bak-code 备份的子代理）"""
targets = []
for sid, d in find_sessions(root):
if args.session and sid not in set(args.session):
continue
if args.skip and sid in set(args.skip):
continue
# 只处理有 .bak-code 备份的子代理（其余一律不碰）
if not os.path.isfile(os.path.join(d, SESSION_FILE) + BACKUP_SUFFIX):
continue
try:
h, _, _ = read_header(os.path.join(d, SESSION_FILE))
except Exception:
continue
if not (h.get("origin") == "subagent" or h.get("parentSession")):
continue
targets.append((sid, d))
# 扫描阶段：找出所有含 v2 descriptor 的会话
need_fix = []
for sid, d in targets:
zpath = os.path.join(d, SESSION_FILE)
try:
with open(zpath, "rb") as f:
raw = f.read()
frames, torn = scan_frames_bin(raw)
if not frames or torn:
continue
has_v2 = False
for (start, end) in frames[1:]:
p = run([zstd(), "-d", "-c"], data=raw[start:end])
if p.returncode != 0:
continue
text = p.stdout.decode("utf-8", errors="replace")
for line in text.split(chr(10)):
if not line.strip():
continue
try:
parsed = json.loads(line)
except Exception:
continue
arr = parsed if isinstance(parsed, list) else [parsed]
for ev in arr:
if (isinstance(ev, dict) and ev.get("type") == "subagent/descriptor"
and isinstance(ev.get("data"), dict)
and ev["data"].get("version") == 2):
has_v2 = True
break
if has_v2:
break
if has_v2:
break
if has_v2:
need_fix.append((sid, d))
except Exception:
continue
if args.dry_run or args.list:
print("== 待修复 %d 个 descriptor v2 会话 -> v3 ==" % len(need_fix))
for sid, d in sorted(need_fix):
print(" %s" % sid)
if args.dry_run or args.list:
print("（dry-run，未修改任何文件）")
return 0
if not need_fix:
print("没有 descriptor v2 会话需要修复")
return 0
print("== 修复 %d 个会话 descriptor v2 -> v3 ==" % len(need_fix))
ok = fail = 0
for sid, d in sorted(need_fix):
zpath = os.path.join(d, SESSION_FILE)
bak = zpath + BACKUP_SUFFIX
try:
# 备份（若不存在）
if not os.path.exists(bak):
shutil.copy2(zpath, bak)
changed, success = fix_descriptor_in_file(zpath)
if changed and success:
ok += 1
print("[OK ] %s descriptor v2 -> v3" % sid)
elif not changed:
ok += 1
print("[SKIP] %s 无 v2 descriptor" % sid)
else:
fail += 1
print("[FAIL] %s 修复失败" % sid, file=sys.stderr)
except Exception as e:
fail += 1
print("[FAIL] %s: %s" % (sid, e), file=sys.stderr)
print("结果: ok=%d failed=%d（备份: *.bak-code）" % (ok, fail))
return 0 if fail == 0 else 1
# ---------------- cache 补写（projcache subagent 缓存） ----------------
DEFAULT_CACHE_DIR = os.path.expanduser("~/.dsh/storages/session_projcache/sessions")
CACHE_VERSION = 4
def count_events(zpath):
"""统计会话事件总数（所有有 seq 的事件的最大 seq + 1）"""
data = decompress(zpath)
max_seq = -1
for line in data.split(NL):
if not line.strip():
continue
try:
parsed = json.loads(line)
except Exception:
continue
arr = parsed if isinstance(parsed, list) else [parsed]
for ev in arr:
if isinstance(ev, dict) and isinstance(ev.get("seq"), int):
if ev["seq"] > max_seq:
max_seq = ev["seq"]
return max_seq + 1
def extract_descriptor_from_log(zpath):
"""从会话日志提取 subagent/descriptor 事件，返回 (data, seq) 或 (None, None)"""
data = decompress(zpath)
for line in data.split(NL):
if not line.strip():
continue
try:
parsed = json.loads(line)
except Exception:
continue
arr = parsed if isinstance(parsed, list) else [parsed]
for ev in arr:
if isinstance(ev, dict) and ev.get("type") == "subagent/descriptor":
return ev.get("data"), ev.get("seq")
return None, None
def build_cache_file(zpath, identity):
"""为一个子代理会话构造 projcache 缓存 JSON（只含必需字段）"""
desc, desc_seq = extract_descriptor_from_log(zpath)
if not desc:
return None, "无 subagent/descriptor 事件"
mode = desc.get("mode")
label = desc.get("label")
if mode not in ("one-shot", "continuable"):
return None, "descriptor mode 非法: %r" % mode
# identity.seq = descriptor 在会话内的相对位置（standard 缓存里是 0）
# row.seq = 会话事件总数（投影 checkpoint 时的事件计数）
total_events = count_events(zpath)
identity_val = {
"mode": mode,
"seq": 0,
}
if label is not None:
identity_val["label"] = label
rows = {
"subagent": {
"ver": 2,
"seq": total_events,
"val": {"identity": identity_val},
},
"subagentTiming": {
"ver": 2,
"seq": total_events,
"val": {"descriptorSeen": True, "settledMs": 0},
},
}
record = {
"version": CACHE_VERSION,
"record": {
"identity": identity,
"rows": rows,
},
}
return record, None
def do_cache(args, root):
"""给有 .bak-code 备份的子代理会话补写 projcache 缓存"""
cache_dir = args.cache_dir
if not os.path.isdir(cache_dir):
print("[错误] 缓存目录不存在: %s" % cache_dir, file=sys.stderr)
return 2
# 收集有 .bak-code 的会话
bak_sessions = []
for sid, d in find_sessions(root):
if os.path.isfile(os.path.join(d, SESSION_FILE) + BACKUP_SUFFIX):
bak_sessions.append((sid, d))
# 过滤子代理（origin=subagent 或有 parentSession）
subagents = []
for sid, d in bak_sessions:
if args.session and sid not in set(args.session):
continue
if args.skip and sid in set(args.skip):
continue
try:
h, _, _ = read_header(os.path.join(d, SESSION_FILE))
except Exception:
continue
if h.get("origin") == "subagent" or h.get("parentSession"):
subagents.append((sid, d, h))
# 判断哪些需要补
need = []
for sid, d, h in subagents:
cache_file = os.path.join(cache_dir, sid + ".json")
if os.path.isfile(cache_file) and not args.force:
continue
need.append((sid, d, h, cache_file))
if args.dry_run or args.list:
print("== 待补缓存 %d 个子代理（有 .bak-code 备份）==" % len(need))
for sid, d, h, cf in sorted(need):
print(" %s" % sid)
if args.dry_run or args.list:
print("（dry-run，未写任何文件）")
return 0
if not need:
print("没有需要补写的子代理缓存")
return 0
ok = fail = 0
for sid, d, h, cache_file in sorted(need):
zpath = os.path.join(d, SESSION_FILE)
identity = {
"createdAt": h.get("createdAt") or 0,
"cwd": h.get("cwd") or "",
}
record, err = build_cache_file(zpath, identity)
if record is None:
fail += 1
print("[FAIL] %s %s" % (sid, err), file=sys.stderr)
continue
try:
with open(cache_file, "w") as f:
json.dump(record, f, ensure_ascii=False, separators=(",", ":"))
ok += 1
print("[OK ] %s 已写缓存 %s" % (sid, os.path.basename(cache_file)))
except Exception as e:
fail += 1
print("[FAIL] %s 写入失败: %s" % (sid, e), file=sys.stderr)
print("结果: ok=%d failed=%d" % (ok, fail))
return 0 if fail == 0 else 1
# ---------------- main ----------------
def main():
ap = argparse.ArgumentParser(
prog="fix",
description="DSH 会话预设修复工具（code -> ptc）",
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--root", default=DEFAULT_ROOT,
help="会话根目录（默认 %s）" % DEFAULT_ROOT)
ap.add_argument("--cache-dir", default=DEFAULT_CACHE_DIR,
help="projcache 缓存目录（默认 %s）" % DEFAULT_CACHE_DIR)
sub = ap.add_subparsers(dest="command", metavar="<command>")
p_fix = sub.add_parser("fix", help="修复 code 预设会话 -> ptc")
p_fix.add_argument("--all", action="store_true", help="修复全部 code 会话")
p_fix.add_argument("--list", action="store_true", help="列出 code 会话（只读）")
p_fix.add_argument("--session", action="append", metavar="ID", help="按会话 id（可多个）")
p_fix.add_argument("--dir", action="append", metavar="DIR", help="按会话目录（可多个）")
p_fix.add_argument("--skip", action="append", metavar="ID", help="跳过指定会话（可多个）")
p_fix.add_argument("--dry-run", action="store_true", help="只预览不修改")
p_fix.add_argument("--to", default="ptc", help="目标预设（默认 ptc）")
p_fix.add_argument("--no-backup", action="store_true", help="不备份（危险）")
p_fix.set_defaults(fn=do_fix)
p_desc = sub.add_parser("desc", help="修复子代理 descriptor v2 -> v3")
p_desc.add_argument("--all", action="store_true", help="修复全部 v2 descriptor 会话")
p_desc.add_argument("--list", action="store_true", help="列出需要修复的会话（只读）")
p_desc.add_argument("--session", action="append", metavar="ID", help="按会话 id（可多个）")
p_desc.add_argument("--skip", action="append", metavar="ID", help="跳过指定会话（可多个）")
p_desc.add_argument("--dry-run", action="store_true", help="只预览不修改")
p_desc.set_defaults(fn=do_desc)
p_cache = sub.add_parser("cache", help="给有 .bak-code 的子代理补写 projcache 缓存")
p_cache.add_argument("--list", action="store_true", help="列出需要补写的会话（只读）")
p_cache.add_argument("--session", action="append", metavar="ID", help="按会话 id（可多个）")
p_cache.add_argument("--skip", action="append", metavar="ID", help="跳过指定会话（可多个）")
p_cache.add_argument("--force", action="store_true", help="缓存已存在也覆盖")
p_cache.add_argument("--dry-run", action="store_true", help="只预览不写")
p_cache.set_defaults(fn=do_cache)
p_restore = sub.add_parser("restore", help="从备份还原已修复会话")
p_restore.add_argument("--dry-run", action="store_true", help="只预览不修改")
p_restore.set_defaults(fn=do_restore)
args = ap.parse_args()
# 无子命令 -> 只显示帮助，不做任何动作
if args.command is None:
ap.print_help()
return 0
if not os.path.isdir(args.root):
print("会话根目录不存在: %s" % args.root, file=sys.stderr)
return 2
return args.fn(args, args.root)
if __name__ == "__main__":
sys.exit(main()) |
0 replies
|
我是让dsh自己修自己： |
0 replies
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Uh oh!
There was an error while loading. Please reload this page.
如题，session无法识别code模式并且无法加载skill与模型
All reactions