diff --git a/.claude/rules/agent-archive.md b/.claude/rules/agent-archive.md index 28f7db17..8870be10 100644 --- a/.claude/rules/agent-archive.md +++ b/.claude/rules/agent-archive.md @@ -235,6 +235,10 @@ paths: (`qwen_session.QwenProjectionTail`) MIRRORS its bytes into the store, append-only, on its own cursor. A running qwen session is NOT un-excluded from the manifest (below) — qwen has no Trajectory viewer, so its native log is retention only and archives once the session ends. + Migration ([Qwen K], XERK-516) does NOT carry this `/qwen/` mirror: it is the display/metrics + feed, and qwen resumes from its OWN native log under `~/.qwen/projects/` (the file the bundle + carries under `.qwen-store/` instead). The target rebuilds this mirror from new events, exactly as + dsh's `/dsh/` feed is rebuilt. Keep the resumable log and this feed straight. - **A RUNNING dsh session is NOT excluded from the manifest** (`_running_slugs` subtracts `_live_dsh_slugs`), and that exception is what makes the Trajectory work at all. **A running QWEN session gets NO such carve-out** (XERK-512): qwen has a real ttyd TUI and no Trajectory diff --git a/.claude/rules/qwen-migration.md b/.claude/rules/qwen-migration.md new file mode 100644 index 00000000..42ee74c9 --- /dev/null +++ b/.claude/rules/qwen-migration.md @@ -0,0 +1,60 @@ +--- +paths: + - "agent/hub-agent.py" + - "agent/tests/test_hub_agent.py" +--- + +# Qwen session migration + resume ([Qwen K], XERK-516) + +Split out of `.claude/rules/qwen.md` (which reached its size ceiling once every XERK-504 child had a +section). This is the migration/resume child; read `qwen.md` for the qwen runtime around it +(launcher [Qwen B], projection [Qwen S1], drive/liveness [Qwen C], archive [Qwen E]). + +XERK-101 (`CLAUDE.md`, "Migrating a session to another agent") extended to qwen — the dsh [K] +(XERK-475, `.claude/rules/dsh.md`) analogue. The **load-bearing difference from dsh**: qwen is +Claude-shaped, so its NATIVE LOG at `~/.qwen/projects//chats/.jsonl` *is* the durable store +`qwen --resume ` reloads from — there is NO separate store like dsh's `DSH_SESSIONS_ROOT`, which +is distinct from dsh's `/dsh/` display feed. So a qwen migration carries the native log itself. +Full mechanics + rationale are in the `QWEN_STORE_ARCNAME` comment in `hub-agent.py`; the invariants +a change must not undo: + +- **RESUME (process-death `--resume`, boot-adopt) was already wired by [Qwen B]/[Qwen C]; [K] only + PINS it.** `_launch_tmux(resume=True)` dispatches to `_launch_qwen(resume=True)` → `qwen --resume + ` at the transcript's origin cwd, and `_start_qwen_tail(resume=True)` restarts the projection + at the native log's EOF so it never re-projects/doubles the kept `.jsonl` (the deterministic- + uuid projection is already there; qwen appends in place). Boot-adopt reattaches the tail the same + way. Never add a resume path that re-reads the native log from 0. +- **MIGRATION carries the NATIVE LOG, not the projection feed.** `export_session` GLOB-locates it + (`_qwen_native_log`, `.jsonl` across `QWEN_PROJECTS_ROOT/*/chats/`, the slug-rule-independent + discipline the tail/`_qwen_runtime_file` use) and packs it under the reserved `.qwen-store/chat.jsonl` + prefix (twin of dsh's `.dsh-store/`), truncated to its last complete line like the main transcript + (a live log is appended by its process). `_unpack_transcript` routes that member to a single target + FILE (`_qwen_store_dest`), **NEVER a dir** — the shared `chats/` dir holds every cwd-cohabiting + session's log, so it places exactly this session's `.jsonl`. It is un-droppable resumable data, + so an oversized bundle is refused (like the dsh store) rather than shipped un-resumable. +- **The `//qwen/` raw-archive mirror ([Qwen E]) is the DISPLAY/metrics feed and is NEVER + carried by migration** — the target rebuilds it from new events past the log's EOF (the tail's + `_mirror_native` primes from the fresh mirror's size 0 and copies the whole placed native log). + Keep the two straight: resume reads the native log, not the mirror. This is the [K] correction the + ticket names. +- **Cross-mount re-key is MANDATORY (upstream issue #2373: qwen keys the store on the working dir).** + The log lands under the TARGET cwd's slug — and qwen's slug rule is `_project_slug` (every + non-alnum→`-`, VERIFIED against real on-disk qwen project dirs; the G0 note's `/`→`-` was + imprecise), not a qwen-specific port. `_reconcile_qwen_store_cwd` then re-points the `cwd` carried + on EVERY native-log row (qwen has no single header line like dsh) from the source cwd (read from the + first row) to the localized worktree, so the placed log is self-consistent with where it now lives. + A no-op on a same-mount move (source cwd == target); never raises the migration over a store detail. +- **A qwen session migrated to a host without qwen falls back to CLAUDE cleanly** — the existing + `agent_type_configured` rebuild guard in `_resume_at_cwd` (per-runtime, not dsh-only). `want_qwen` + (`agentType=="qwen" and qwen_configured()`) gates the unpack, so a claude-fallback import DROPS the + `.qwen-store/` member (no resume to feed it), exactly as a stray dsh store is dropped. +- **Model/endpoint re-validation is against the TARGET host's config on every launch.** `_launch_qwen` + reads the TARGET's `QWEN_MODEL_BASE_URL`/`QWEN_MODEL_API_KEY_ENV` (never the source's) and keeps a + carried `model` only if it passes `QWEN_IDENT_RE`, else the host default. qwen has NO model-list + discovery (unlike local-model failover), so an id the target's endpoint does not actually serve is + caught at `_confirm_qwen_launch` (clean teardown + a reason via `_refuse_start`), not pre-validated. +- Tests: the qwen cases in `TestMigrateSession` (`test_a_qwen_bundle_carries_the_native_log_*`, + `test_a_claude_import_drops_a_stray_qwen_log`, `test_import_places_and_rekeys_the_qwen_log_cross_mount`, + `test_import_drops_the_qwen_log_when_target_lacks_qwen`, `test_qwen_store_dest_uses_the_project_slug_rule`, + `test_reconcile_qwen_log_cwd_is_a_noop_on_same_mount`). A real cross-host move actually resumed by + qwen is host-proof only — qwen is not installed in CI — the footing [Qwen C]/[Qwen E] shipped on. diff --git a/.claude/rules/qwen.md b/.claude/rules/qwen.md index 80683b4f..e66b5036 100644 --- a/.claude/rules/qwen.md +++ b/.claude/rules/qwen.md @@ -461,3 +461,12 @@ ticket-branch directive — no new launch code). The mechanics — the `/runtime gate, the per-runtime `findTicketHost` capability filter, the board `qwenAvailable`/picker, and the Android parity — live in **`.claude/rules/turma-board.md`**'s Runtime-row section (board-scoped, where the dsh [I] board detail's twin belongs), whose `paths:` load for the board files this touched. + +## [Qwen K] (XERK-516) shipped: session migration + resume + +Session migration + resume for a qwen session (the dsh [K] analogue) lives in +**`.claude/rules/qwen-migration.md`** — split out to keep this file under its size ceiling; its +`paths:` include `hub-agent.py`, so it co-loads with this file when the migration code is touched. +The load-bearing point: qwen is Claude-shaped, so its NATIVE LOG *is* the store `qwen --resume` +reloads from (no separate store like dsh's `DSH_SESSIONS_ROOT`) — a migration carries that log under +`.qwen-store/`, places it at the target cwd's slug and re-keys its per-row `cwd`. diff --git a/CLAUDE.md b/CLAUDE.md index d5e875eb..6635a420 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -36,6 +36,7 @@ that component's files. | `.claude/rules/routing-eval.md` | `bench/archive/**`, the routing docs | archive-sourced replay eval: the requestId turn unit, curation gates, why routing is per-session | | `.claude/rules/dsh.md` | `poc/turma-2.0-poc/**`, `agent/**` | dsh ADR + per-child mechanics (XERK-460) | | `.claude/rules/qwen.md` | `agent/hub-agent.py`, `turma/server.js`, `sessions.html`, `android/**` | Qwen Code runtime plumbing (XERK-504) | +| `.claude/rules/qwen-migration.md` | `agent/hub-agent.py` | qwen session migration + resume (XERK-516 [Qwen K]): the native log as the resumable store | | `.claude/rules/dsh-input.md` | `agent/dsh_session.py` | driving a dsh session (XERK-467 [C]): socket, driver, input | | `.claude/rules/dsh-delegation.md` | `agent/dsh_transcript.py` | dsh delegation (XERK-474 [J]): bg-agent/workflow rows + `subagentHistory` | | `.claude/rules/dsh-guard.md` | `agent/dsh/guard/**` | dsh safety guard (XERK-470 [F]): deny policy on dsh's pipeline | diff --git a/agent/hub-agent.py b/agent/hub-agent.py index 11ca6d76..e2587398 100644 --- a/agent/hub-agent.py +++ b/agent/hub-agent.py @@ -1261,6 +1261,29 @@ def _positive_int_env(name, default): or os.path.join(os.path.expanduser("~"), ".qwen")) QWEN_PROJECTS_ROOT = (os.environ.get("QWEN_PROJECTS_ROOT") or os.path.join(QWEN_HOME, "projects")) +# Reserved tar arcname carrying qwen's native session log into a migration bundle +# ([Qwen K], XERK-516), the qwen twin of DSH_STORE_ARCNAME. The CRUCIAL +# difference from dsh: dsh resumes from a SEPARATE store (DSH_SESSIONS_ROOT), +# distinct from its `/dsh/` display feed — so dsh migration carries that +# store. qwen is Claude-shaped: its native log at ~/.qwen/projects//chats/ +# .jsonl IS what `qwen --resume ` reloads from (there is no separate +# store), so THAT log is what a qwen migration carries. It rides under this +# reserved prefix (never a slug-relative member — a claude session id is never a +# dotfile) so _unpack_transcript routes it into the target's OWN +# QWEN_PROJECTS_ROOT at the localized cwd's slug, not into the projection slug +# tree. qwen keys the native log on the cwd (issue #2373: not yet portable), so a +# cross-mount move MUST place it under the TARGET cwd's slug (== _project_slug, +# verified: qwen's slug rule is Claude's every-non-alnum->'-', not the G0 note's +# '/'->'-') and re-point the log's per-row cwd, or a resume in the new worktree +# finds nothing. The `//qwen/` raw-archive mirror ([Qwen E]) is the +# DISPLAY/metrics feed and is NOT what resume reads — keep the two straight. +QWEN_STORE_ARCNAME = ".qwen-store" +# The single member name the native log rides as inside QWEN_STORE_ARCNAME. Fixed +# (not the session-id filename) so _unpack_transcript writes it to the ONE target +# path — the shared chats/ dir holds every cwd-cohabiting session's log, so a +# migration must place exactly this session's .jsonl, never unpack a dir into +# it. +QWEN_STORE_MEMBER = "chat.jsonl" # Per-session transient files (the sourced 0600 env file with the model key) # live under the agent-owned ~/.turma, never a worktree — the same discipline as # the dsh socket dir and the local-model env file. @@ -13747,9 +13770,27 @@ def _launch_qwen(self, sess, resume=False, prompt=None, resume_id=None): except OSError as e: raise RuntimeError(f"qwen runtime dir: {e}") env_file = os.path.join(QWEN_RUNTIME_DIR, f"{sid}.env") + # The model/endpoint are re-validated against THIS host's qwen config on + # every launch, which is what makes a MIGRATED session ([Qwen K], + # XERK-516) land cleanly: the endpoint/key are the TARGET host's + # (QWEN_MODEL_BASE_URL / QWEN_MODEL_API_KEY_ENV read here, never the + # source's), and a carried model is kept only if it passes qwen's ident + # charset — a garbage/foreign one falls back to the host default rather + # than being argv'd through. qwen has NO model-list discovery (unlike the + # local-model failover), so an id the target's endpoint does not actually + # serve is caught at _confirm_qwen_launch (clean teardown + a reason via + # _refuse_start), not pre-validated here. A target with no qwen at all + # never reaches this: agent_type_configured already fell it back to claude. + carried_model = sess.get("model") + model = (carried_model if carried_model + and QWEN_IDENT_RE.fullmatch(carried_model) else QWEN_MODEL) + if carried_model and model != carried_model: + log(f"qwen session {sid}: carried model {carried_model!r} is not a " + f"valid qwen model id on this host; using the host default " + f"{QWEN_MODEL!r}") env_map = { "OPENAI_BASE_URL": QWEN_MODEL_BASE_URL, - "OPENAI_MODEL": sess.get("model") or QWEN_MODEL, + "OPENAI_MODEL": model, # A pinned fleet must not let qwen auto-update the binary out from # under the parsers (the G0 spike caught it upgrading mid-run) — belt # and suspenders with settings.general.disableAutoUpdate. @@ -14009,6 +14050,28 @@ def _qwen_runtime_file(self, claude_sid): f"{claude_sid}.runtime.json")) return hits[0] if hits else None + def _qwen_native_log(self, claude_sid): + """qwen's native session log for a pinned id — the file `qwen --resume` + reloads from ([Qwen K], XERK-516) — located by the id ACROSS project dirs + (glob), the same slug-rule-independent discipline as _qwen_runtime_file + and the projection tail. Returns the path if present, else None (an empty + conversation writes no `.jsonl` until its first turn).""" + if not (claude_sid and VALID_CLAUDE_SID_RE.fullmatch(claude_sid)): + return None + hits = glob.glob(os.path.join(QWEN_PROJECTS_ROOT, "*", "chats", + f"{claude_sid}.jsonl")) + return hits[0] if hits else None + + def _qwen_store_dest(self, cwd, claude_sid): + """Where a migrated qwen native log must land on THIS host so + `qwen --resume ` (run in the localized worktree) finds it: under the + TARGET cwd's project slug. qwen's slug rule is Claude's _project_slug + (every non-alnum->'-', verified against real on-disk qwen project dirs — + the G0 note's '/'->'-' was imprecise), so a cross-mount move re-keys by + computing the slug from the LOCALIZED cwd, not the source's.""" + return os.path.join(QWEN_PROJECTS_ROOT, _project_slug(cwd), "chats", + f"{claude_sid}.jsonl") + def _qwen_runtime_pid(self, path): """The pid recorded in a qwen `.runtime.json`, or None if unreadable. Best-effort and never raises — it runs in the launch-confirm loop.""" @@ -15828,8 +15891,21 @@ def refuse(reason): cand = _dsh_store_dir(worktree, sid) if os.path.isdir(cand): dsh_store = cand + # A qwen session carries its NATIVE LOG ([Qwen K], XERK-516) — for qwen, + # unlike dsh, that log IS what `qwen --resume` reloads from (qwen is + # Claude-shaped, no separate store). The top-level transcript above is the + # lossy display PROJECTION, which qwen cannot resume; without the native + # log the target resumes into a blank conversation. Located by the pinned + # id across project dirs (glob), so the exact cwd->slug rule is not relied + # on. The `//qwen/` archive mirror is the DISPLAY/metrics feed, + # NOT carried here (the target rebuilds it from new events past the log's + # EOF — the tail's resume=True), so the two never get crossed. + qwen_store = None + if sess.get("agentType") == "qwen": + qwen_store = self._qwen_native_log(sess.get("claudeSessionId") or "") try: - blob = self._pack_transcript(path, dsh_store=dsh_store) + blob = self._pack_transcript(path, dsh_store=dsh_store, + qwen_store=qwen_store) except Exception as e: refuse(f"packing the transcript failed: {e}") return @@ -15891,14 +15967,32 @@ def refuse(reason): want_dsh = cmd.get("agentType") == "dsh" and dsh_configured() dsh_store_dest = (_dsh_store_dir(os.path.normpath(cwd), transcript_id) if want_dsh else None) + # A qwen session carries its NATIVE LOG under `.qwen-store/` ([Qwen K], + # XERK-516); unpack it into THIS host's QWEN_PROJECTS_ROOT at the LOCALIZED + # cwd's slug so `qwen --resume ` (run in the new worktree) finds it. + # Only when the target will actually launch qwen — a qwen session falling + # back to claude here (no qwen, via agent_type_configured in _resume_at_cwd) + # has no resume to consume the log, so it is dropped, exactly as a stray + # dsh store is. The per-row cwd is re-pointed to the localized worktree so + # a cross-mount move stays self-consistent (issue #2373: qwen keys on cwd). + want_qwen = cmd.get("agentType") == "qwen" and qwen_configured() + qwen_store_dest = (self._qwen_store_dest(os.path.normpath(cwd), + transcript_id) + if want_qwen else None) try: os.makedirs(slug_dir, exist_ok=True) if dsh_store_dest: os.makedirs(dsh_store_dest, exist_ok=True) - self._unpack_transcript(blob, slug_dir, dsh_store_dest=dsh_store_dest) + if qwen_store_dest: + os.makedirs(os.path.dirname(qwen_store_dest), exist_ok=True) + self._unpack_transcript(blob, slug_dir, dsh_store_dest=dsh_store_dest, + qwen_store_dest=qwen_store_dest) if dsh_store_dest: self._reconcile_dsh_store_cwd(dsh_store_dest, os.path.normpath(cwd)) + if qwen_store_dest: + self._reconcile_qwen_store_cwd(qwen_store_dest, + os.path.normpath(cwd)) except Exception as e: refuse(f"unpacking the transcript bundle failed: {e}") return @@ -15918,18 +16012,22 @@ def refuse(reason): self._resume_at_cwd(transcript_id, cwd, cmd_id=cmd.get("cmdId"), extra=extra, migration_id=migration_id) - def _pack_bytes(self, path, runs, dsh_store=None): + def _pack_bytes(self, path, runs, dsh_store=None, qwen_store=None): """The bundle itself: `.jsonl` (truncated to its last complete line), plus `/subagents/...` when present, plus the `runs` dir as `/workflows/...` when one is given, plus a dsh session's durable - store as `.dsh-store/...` when one is given. See _pack_transcript. - - The dsh store rides under the reserved `.dsh-store/` prefix (never a - slug-relative member) so `_unpack_transcript` routes it to the target's - own DSH_SESSIONS_ROOT rather than into the project-slug tree. Unlike the - workflow records, it is NOT droppable — it is the resumable data — so it - is packed on every path, and if that pushes the bundle over the ceiling - the move is refused (export_session) rather than shipped un-resumable.""" + store as `.dsh-store/...` OR a qwen session's native log as + `.qwen-store/chat.jsonl` when one is given. See _pack_transcript. + + The dsh store / qwen native log ride under a reserved dotfile prefix + (never a slug-relative member) so `_unpack_transcript` routes them to the + target's own store root rather than into the project-slug tree. Unlike the + workflow records, they are NOT droppable — they are the resumable data — + so they are packed on every path, and if that pushes the bundle over the + ceiling the move is refused (export_session) rather than shipped + un-resumable. A live qwen log is appended by its process, so — like the + main transcript — it is truncated to its last complete line, or a resume + could choke on a half-written tail.""" tid = os.path.basename(path)[:-len(".jsonl")] with open(path, "rb") as f: raw = f.read() @@ -15947,9 +16045,18 @@ def _pack_bytes(self, path, runs, dsh_store=None): tar.add(runs, arcname=os.path.join(tid, WORKFLOW_RUNS_SUBDIR)) if dsh_store and os.path.isdir(dsh_store): tar.add(dsh_store, arcname=DSH_STORE_ARCNAME) + if qwen_store and os.path.isfile(qwen_store): + with open(qwen_store, "rb") as f: + qraw = f.read() + qnl = qraw.rfind(b"\n") + qcomplete = qraw[:qnl + 1] if qnl >= 0 else qraw + qti = tarfile.TarInfo( + name=QWEN_STORE_ARCNAME + "/" + QWEN_STORE_MEMBER) + qti.size = len(qcomplete) + tar.addfile(qti, io.BytesIO(qcomplete)) return buf.getvalue() - def _pack_transcript(self, path, dsh_store=None): + def _pack_transcript(self, path, dsh_store=None, qwen_store=None): """Bundle a transcript file (+ its subagents/ and workflows/ dirs, if any) into gzipped tar bytes, laid out relative to the project-slug dir so the target unpacks straight into PROJECTS_ROOT//: `.jsonl` and, @@ -15982,7 +16089,8 @@ def _pack_transcript(self, path, dsh_store=None): f"{WORKFLOW_PACK_MAX_BYTES} bytes; not carrying them") else: try: - blob = self._pack_bytes(path, runs, dsh_store=dsh_store) + blob = self._pack_bytes(path, runs, dsh_store=dsh_store, + qwen_store=qwen_store) except OSError as e: # An unreadable file in the RECORDS tree (a leftover # root-owned one after a PUID change, say) must not refuse @@ -16010,9 +16118,11 @@ def _pack_transcript(self, path, dsh_store=None): return blob log(f"migration: workflow records for {tid} would put the " f"bundle over {MIGRATION_BLOB_MAX} bytes; not carrying them") - return self._pack_bytes(path, None, dsh_store=dsh_store) + return self._pack_bytes(path, None, dsh_store=dsh_store, + qwen_store=qwen_store) - def _unpack_transcript(self, blob, dest_dir, dsh_store_dest=None): + def _unpack_transcript(self, blob, dest_dir, dsh_store_dest=None, + qwen_store_dest=None): """Extract a _pack_transcript bundle. Slug-relative members go to dest_dir; members under the reserved `.dsh-store/` prefix go to `dsh_store_dest` instead (a dsh session's durable store, which lives @@ -16024,10 +16134,17 @@ def _unpack_transcript(self, blob, dest_dir, dsh_store_dest=None): A `.dsh-store/` member with no `dsh_store_dest` (a claude import, or a dsh session falling back to claude on a host without dsh) is skipped — - the store is useless without a dsh resume to consume it.""" + the store is useless without a dsh resume to consume it. A `.qwen-store/` + member ([Qwen K], XERK-516) routes the SAME way, but to a single FILE + (`qwen_store_dest`, the target native log `.jsonl`) rather than a dir: + the shared chats/ dir holds every cwd-cohabiting session's log, so a + migration writes exactly this session's file, never a tree into it. It is + likewise skipped without a `qwen_store_dest` (a qwen session falling back + to claude on a host without qwen).""" root = os.path.realpath(dest_dir) store_root = os.path.realpath(dsh_store_dest) if dsh_store_dest else None store_prefix = DSH_STORE_ARCNAME + "/" + qwen_prefix = QWEN_STORE_ARCNAME + "/" buf = io.BytesIO(blob) with tarfile.open(fileobj=buf, mode="r:gz") as tar: for m in tar.getmembers(): @@ -16040,6 +16157,13 @@ def _unpack_transcript(self, blob, dest_dir, dsh_store_dest=None): continue rel = m.name[len(store_prefix):] if m.name != DSH_STORE_ARCNAME else "" base, check_root = dsh_store_dest, store_root + # Route .qwen-store/* to the single target native-log FILE. + elif m.name == QWEN_STORE_ARCNAME or m.name.startswith(qwen_prefix): + if qwen_store_dest is None or not m.isreg(): + continue # no qwen resume to feed, or a stray dir member + base = os.path.dirname(qwen_store_dest) + rel = os.path.basename(qwen_store_dest) + check_root = os.path.realpath(base) else: rel, base, check_root = m.name, dest_dir, root out = os.path.join(base, rel) if rel else base @@ -16090,6 +16214,63 @@ def _reconcile_dsh_store_cwd(self, store_dir, cwd): except OSError as e: log(f"migration: could not re-key dsh store cwd for {store_dir}: {e}") + def _reconcile_qwen_store_cwd(self, store_path, cwd): + """Re-point a migrated qwen native log's per-row `cwd` to the localized + worktree ([Qwen K], XERK-516). qwen keys the log on the working dir + (issue #2373: not yet portable) — a cross-mount move already placed the + log under the TARGET cwd's slug (_qwen_store_dest), and re-pointing the + cwd carried on every row keeps the log self-consistent with where it now + lives (its raw-archive mirror, any tool reading cwd, and a resume that + may check it). Every native-log row carries `cwd` (unlike dsh's single + header line), so this rewrites the WHOLE file, replacing only rows whose + cwd equals the SOURCE cwd (read from the first row) with the target cwd. + + A no-op when the source cwd already equals the target (same-mount move) + or the shape is unexpected — like the dsh twin, it never raises the + migration over a store detail. Bounded like _reconcile_dsh_store_cwd: the + log was just written from the in-memory bundle, so reading it back is the + same order of memory the migration already spent.""" + try: + with open(store_path, "r", encoding="utf-8") as f: + lines = f.readlines() + except OSError: + return + # The source cwd is whatever the first parseable row records — robust to + # the command's cwd not matching the log byte-for-byte. + source_cwd = None + for line in lines: + try: + row = json.loads(line) + except ValueError: + continue + if isinstance(row, dict) and isinstance(row.get("cwd"), str): + source_cwd = row["cwd"] + break + if not source_cwd or source_cwd == cwd: + return # nothing to re-key (same-mount move, or no cwd in the log) + out = [] + for line in lines: + stripped = line.strip() + if not stripped: + out.append(line) + continue + try: + row = json.loads(stripped) + except ValueError: + out.append(line) # keep an unparseable line verbatim + continue + if isinstance(row, dict) and row.get("cwd") == source_cwd: + row["cwd"] = cwd + out.append(json.dumps(row, ensure_ascii=False) + "\n") + else: + out.append(line) + try: + with open(store_path, "w", encoding="utf-8") as f: + f.writelines(out) + except OSError as e: + log(f"migration: could not re-key qwen native log cwd for " + f"{store_path}: {e}") + # A bundle is the largest thing an agent ever sends, so it is the body most # likely to meet the hub's in-flight budget (XERK-258) and be told 503 "busy, # try again". Nothing else would retry it — a lost bundle strands the whole diff --git a/agent/tests/test_hub_agent.py b/agent/tests/test_hub_agent.py index 838795d3..50957cc4 100644 --- a/agent/tests/test_hub_agent.py +++ b/agent/tests/test_hub_agent.py @@ -6921,6 +6921,155 @@ def test_import_drops_the_dsh_store_when_target_lacks_dsh(self): self.assertFalse(os.path.exists(sessions_root)) self.assertEqual(sm.registry[0]["agentType"], "claude") + # --- qwen session migration: the native log IS the store (XERK-516) ------ + # + # qwen is Claude-shaped: `qwen --resume ` reloads from its native log at + # ~/.qwen/projects//chats/.jsonl (there is no separate store like + # dsh). So migration carries THAT log, places it under the TARGET cwd's slug + # (qwen keys on cwd — issue #2373), and re-points its per-row cwd. The + # `//qwen/` archive mirror is the DISPLAY feed, never carried. + + def _write_qwen_native_log(self, path, cwd, sid, tail="\n"): + """A qwen native log as the tail reads it: uuid/parentUuid-linked rows, + each carrying `cwd` (unlike dsh's single header). `tail` lets a test add a + half-written final line to prove the truncation.""" + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as f: + f.write(json.dumps({"uuid": "u1", "parentUuid": None, + "sessionId": sid, "cwd": cwd, "type": "user", + "message": {"role": "user", + "parts": [{"text": "hi"}]}}) + "\n") + f.write(json.dumps({"uuid": "u2", "parentUuid": "u1", + "sessionId": sid, "cwd": cwd, "type": "assistant", + "message": {"role": "assistant", + "parts": [{"text": "ok"}]}}) + "\n") + f.write(tail) + return path + + def test_a_qwen_bundle_carries_the_native_log_under_its_own_prefix(self): + wt = os.path.join(ha.WORKTREES_ROOT, "Turma", "qwmig") + path = self._write_transcript(wt, "transQ") + native = os.path.join(self.tmp, "qwen-src", "chats", "transQ.jsonl") + # A half-written final line must not travel — a resume could choke on it. + self._write_qwen_native_log(native, wt, "transQ", tail='{"partial') + sm = self._manager() + blob = sm._pack_transcript(path, qwen_store=native) + dest = os.path.join(self.tmp, "dest-q") + os.makedirs(dest) + store_dest = os.path.join(self.tmp, "qwen-dst", "chats", "transQ.jsonl") + sm._unpack_transcript(blob, dest, qwen_store_dest=store_dest) + # The projection still rides the slug tree; the native log rides its own + # prefix and landed at the single target FILE, never in the slug tree. + self.assertTrue(os.path.isfile(os.path.join(dest, "transQ.jsonl"))) + self.assertFalse(os.path.exists(os.path.join(dest, ha.QWEN_STORE_ARCNAME))) + self.assertTrue(os.path.isfile(store_dest)) + with open(store_dest) as f: + body = f.read() + self.assertNotIn("partial", body) # truncated at the last newline + self.assertTrue(body.endswith("\n")) + self.assertEqual(len([l for l in body.splitlines() if l.strip()]), 2) + + def test_a_claude_import_drops_a_stray_qwen_log(self): + # No qwen_store_dest -> `.qwen-store/` members are skipped, never written + # into the slug tree (they have no resume to feed). + wt = os.path.join(ha.WORKTREES_ROOT, "Turma", "qwstray") + path = self._write_transcript(wt, "transQS") + native = os.path.join(self.tmp, "qwen-src2", "chats", "transQS.jsonl") + self._write_qwen_native_log(native, wt, "transQS") + sm = self._manager() + blob = sm._pack_transcript(path, qwen_store=native) + dest = os.path.join(self.tmp, "dest-qs") + os.makedirs(dest) + sm._unpack_transcript(blob, dest, qwen_store_dest=None) + self.assertTrue(os.path.isfile(os.path.join(dest, "transQS.jsonl"))) + self.assertFalse(os.path.exists(os.path.join(dest, ha.QWEN_STORE_ARCNAME))) + self.assertEqual(sorted(os.listdir(dest)), ["transQS.jsonl"]) + + def test_import_places_and_rekeys_the_qwen_log_cross_mount(self): + """The end-to-end target half for a qwen session: the native log lands at + the LOCALIZED cwd's slug (where `qwen --resume` in the new worktree + looks), and its per-row cwd is re-pointed from the source's to match.""" + foreign = "/home/otheruser/src/.turma/worktrees/Turma/qwx" + local = os.path.join(ha.WORKTREES_ROOT, "Turma", "qwx") + projects_root = os.path.join(self.tmp, "qwen-home", "projects") + sm = self._manager() + sm._worktree_add = mock.Mock() + with mock.patch.object(ha, "QWEN_PROJECTS_ROOT", projects_root), \ + mock.patch.object(ha, "qwen_configured", lambda: True): + src_path = self._write_transcript(local, "transQX") + # The source native log carries the FOREIGN cwd on every row. + native = os.path.join(self.tmp, "qwen-foreign", "chats", + "transQX.jsonl") + self._write_qwen_native_log(native, foreign, "transQX") + blob = sm._pack_transcript(src_path, qwen_store=native) + shutil.rmtree(os.path.join(ha.PROJECTS_ROOT, ha._project_slug(local))) + sm._migration_download = lambda mid: blob + cmd = {"type": "importSession", "cmdId": "cQX", "migrationId": "migQX", + "transcriptId": "transQX", "cwd": foreign, "repo": "Turma", + "agentType": "qwen"} + with mock.patch.object(ha, "resolve_base_ref", return_value="origin/main"): + sm.import_session(cmd) + # The native log landed at the LOCAL cwd's slug (not the foreign one). + placed = sm._qwen_store_dest(local, "transQX") + self.assertTrue(os.path.isfile(placed)) + self.assertFalse(os.path.isfile( + sm._qwen_store_dest(foreign, "transQX"))) + # ...and every row's cwd was re-pointed to the local worktree. + with open(placed) as f: + rows = [json.loads(l) for l in f if l.strip()] + self.assertTrue(rows and all(r["cwd"] == local for r in rows)) + self.assertEqual(rows[0]["uuid"], "u1") # every other field kept + self.assertEqual(sm.registry[0]["agentType"], "qwen") + self.assertEqual(sm._launch_tmux.call_args.kwargs["resume_id"], "transQX") + + def test_import_drops_the_qwen_log_when_target_lacks_qwen(self): + # agentType=qwen but the target does not offer qwen -> the session falls + # back to claude (agent_type_configured), so the native log has no resume + # to feed and is not placed under QWEN_PROJECTS_ROOT. + wt = os.path.join(ha.WORKTREES_ROOT, "Turma", "qwno") + projects_root = os.path.join(self.tmp, "qwen-home2", "projects") + sm = self._manager() + sm._worktree_add = mock.Mock() + with mock.patch.object(ha, "QWEN_PROJECTS_ROOT", projects_root), \ + mock.patch.object(ha, "qwen_configured", lambda: False): + src_path = self._write_transcript(wt, "transQN") + native = os.path.join(self.tmp, "qwen-src3", "chats", "transQN.jsonl") + self._write_qwen_native_log(native, wt, "transQN") + blob = sm._pack_transcript(src_path, qwen_store=native) + shutil.rmtree(os.path.join(ha.PROJECTS_ROOT, ha._project_slug(wt))) + sm._migration_download = lambda mid: blob + cmd = {"type": "importSession", "cmdId": "cQN", "migrationId": "migQN", + "transcriptId": "transQN", "cwd": wt, "repo": "Turma", + "agentType": "qwen"} + with mock.patch.object(ha, "resolve_base_ref", return_value="origin/main"): + sm.import_session(cmd) + self.assertFalse(os.path.exists(projects_root)) + self.assertEqual(sm.registry[0]["agentType"], "claude") + + def test_qwen_store_dest_uses_the_project_slug_rule(self): + # qwen's native-log dir is keyed by the cwd's project slug (every + # non-alnum -> '-', == _project_slug, verified against real qwen dirs), + # so a cross-mount move that re-keys must compute it the same way qwen + # will when it runs `--resume` in the localized worktree. + sm = self._manager() + with mock.patch.object(ha, "QWEN_PROJECTS_ROOT", "/q/projects"): + self.assertEqual( + sm._qwen_store_dest("/home/me/git/.turma/worktrees/Turma/abc", + "sid1"), + "/q/projects/-home-me-git--turma-worktrees-Turma-abc/chats/" + "sid1.jsonl") + + def test_reconcile_qwen_log_cwd_is_a_noop_on_same_mount(self): + # Same-mount move (source cwd == target): the log is left byte-for-byte + # unchanged — never rewritten over a store detail resume re-validates. + wt = os.path.join(ha.WORKTREES_ROOT, "Turma", "qwsame") + native = os.path.join(self.tmp, "qwen-same", "chats", "transS.jsonl") + self._write_qwen_native_log(native, wt, "transS") + before = open(native, "rb").read() + sm = self._manager() + sm._reconcile_qwen_store_cwd(native, wt) + self.assertEqual(open(native, "rb").read(), before) + class TestDshProjectKey(unittest.TestCase): """`_dsh_project_key` ports dsh's `projectKey(cwd)` byte-for-byte — the store