refactor: apply multi-reviewer findings to switch-model

- extract _copy_tree -> shared cowork2code/_fs.py (Protocol-typed); rehome + cowork_session use it
- audit.py: type _text_of, sequential assistant message ids, __all__
- cowork_session.py: Callable typing, transcript assert, outputs_dir on dataclass,
  --to-model format guard, 0o600 writes for the new config + audit
- cli.py: dedupe selector resolution into _resolve_session; Namespace annotations;
  dry-run no longer prints a Cleanup/quit line for files never written
- tests: strengthen audit assertions; add symlink-skip-via-execute, exit-5,
  invalid-model, --title-as, --id, multi-turn, empty-records, source-read-only-on-disk
- 73 passing, 93% coverage, ruff clean

Author: ebeckner-looptech

cowork2code/_fs.py

@@ -0,0 +1,40 @@
+"""Shared filesystem helper: copy a file tree without clobbering or following symlinks."""
+
+import shutil
+from pathlib import Path
+from typing import Protocol
+
+
+class CopyTarget(Protocol):
+    """Anything that records what was copied and what was skipped."""
+
+    copied: list[str]
+    warnings: list[str]
+
+
+def copy_tree(
+    src_root: Path, dst_root: Path, rel_prefix: Path, target: CopyTarget, dry_run: bool
+) -> None:
+    """Copy files from ``src_root`` into ``dst_root`` (no clobber, never follow symlinks).
+
+    Appends copied relative paths to ``target.copied`` and skip reasons to
+    ``target.warnings``. Symlinks are skipped (avoids exfiltrating files outside the
+    source tree into the destination).
+    """
+    for src in sorted(src_root.rglob("*")):
+        rel = rel_prefix / src.relative_to(src_root)
+        if src.is_symlink():
+            target.warnings.append(f"skip symlink {rel}")
+            continue
+        if src.is_dir():
+            continue
+        dst = dst_root / rel
+        if dst.exists() and src.resolve() == dst.resolve():
+            continue
+        if dst.exists():
+            target.warnings.append(f"skip existing {rel}")
+            continue
+        target.copied.append(str(rel))
+        if not dry_run:
+            dst.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy2(src, dst)

cowork2code/audit.py

@@ -5,17 +5,17 @@
 valid signatures.
 """
 
+__all__ = ["build_audit_log"]
+
 _PLACEHOLDER_HMAC = "0" * 64
 
 
-def _text_of(content) -> str:
+def _text_of(content: str | list[dict] | None) -> str:
     if isinstance(content, str):
         return content
     if isinstance(content, list):
         return "".join(
-            b.get("text", "")
-            for b in content
-            if isinstance(b, dict) and b.get("type") == "text"
+            b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text"
         )
     return ""
 
@@ -26,13 +26,21 @@ def build_audit_log(
     """Return Cowork audit records: system/init, per-turn user/assistant, result."""
     out: list[dict] = [
         {
-            "type": "system", "subtype": "init", "cwd": cwd, "session_id": cli_id,
-            "tools": [], "mcp_servers": [], "model": model, "permissionMode": "default",
-            "_audit_timestamp": timestamp, "_audit_hmac": _PLACEHOLDER_HMAC,
+            "type": "system",
+            "subtype": "init",
+            "cwd": cwd,
+            "session_id": cli_id,
+            "tools": [],
+            "mcp_servers": [],
+            "model": model,
+            "permissionMode": "default",
+            "_audit_timestamp": timestamp,
+            "_audit_hmac": _PLACEHOLDER_HMAC,
         }
     ]
     last_text = ""
-    turns = 0
+    user_count = 0
+    asst_count = 0
     for rec in records:
         if rec.get("type") not in ("user", "assistant"):
             continue
@@ -41,32 +49,53 @@ def build_audit_log(
         if msg.get("role") == "user":
             if not text:  # skip tool-result-only / empty user turns
                 continue
-            out.append({
-                "type": "user", "uuid": rec.get("uuid") or f"audit-u{turns}",
-                "session_id": cli_id, "parent_tool_use_id": None,
-                "client_platform": "desktop_app",
-                "message": {"role": "user", "content": text},
-                "_audit_timestamp": timestamp, "_audit_hmac": _PLACEHOLDER_HMAC,
-            })
-            turns += 1
+            out.append(
+                {
+                    "type": "user",
+                    "uuid": rec.get("uuid") or f"audit-u{user_count}",
+                    "session_id": cli_id,
+                    "parent_tool_use_id": None,
+                    "client_platform": "desktop_app",
+                    "message": {"role": "user", "content": text},
+                    "_audit_timestamp": timestamp,
+                    "_audit_hmac": _PLACEHOLDER_HMAC,
+                }
+            )
+            user_count += 1
         elif msg.get("role") == "assistant":
             last_text = text or last_text
-            out.append({
-                "type": "assistant",
-                "message": {
-                    "model": model, "id": f"msg_{turns}", "type": "message",
-                    "role": "assistant", "content": [{"type": "text", "text": text}],
-                    "stop_reason": "end_turn", "stop_sequence": None,
-                    "usage": {"input_tokens": 0, "output_tokens": 0},
-                },
-                "parent_tool_use_id": None, "session_id": cli_id,
-                "uuid": rec.get("uuid") or f"audit-a{turns}",
-                "_audit_timestamp": timestamp, "_audit_hmac": _PLACEHOLDER_HMAC,
-            })
-            turns += 1
-    out.append({
-        "type": "result", "subtype": "success", "is_error": False,
-        "result": last_text, "session_id": cli_id, "num_turns": turns, "duration_ms": 0,
-        "_audit_timestamp": timestamp, "_audit_hmac": _PLACEHOLDER_HMAC,
-    })
+            out.append(
+                {
+                    "type": "assistant",
+                    "message": {
+                        "model": model,
+                        "id": f"msg_{asst_count}",
+                        "type": "message",
+                        "role": "assistant",
+                        "content": [{"type": "text", "text": text}],
+                        "stop_reason": "end_turn",
+                        "stop_sequence": None,
+                        "usage": {"input_tokens": 0, "output_tokens": 0},
+                    },
+                    "parent_tool_use_id": None,
+                    "session_id": cli_id,
+                    "uuid": rec.get("uuid") or f"audit-a{asst_count}",
+                    "_audit_timestamp": timestamp,
+                    "_audit_hmac": _PLACEHOLDER_HMAC,
+                }
+            )
+            asst_count += 1
+    out.append(
+        {
+            "type": "result",
+            "subtype": "success",
+            "is_error": False,
+            "result": last_text,
+            "session_id": cli_id,
+            "num_turns": user_count + asst_count,
+            "duration_ms": 0,
+            "_audit_timestamp": timestamp,
+            "_audit_hmac": _PLACEHOLDER_HMAC,
+        }
+    )
     return out

cowork2code/cli.py

@@ -63,7 +63,25 @@ def _build_parser() -> argparse.ArgumentParser:
     return parser
 
 
-def _cmd_list(args) -> int:
+def _resolve_session(
+    root: Path, args: argparse.Namespace
+) -> tuple[discover.SessionRef | None, int]:
+    """Resolve the selected session, or print an error and return an exit code."""
+    try:
+        ref = discover.resolve(root, title=args.title, sid=args.id, from_dir=args.from_dir)
+        return ref, 0
+    except discover.NoMatch as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return None, 2
+    except discover.AmbiguousMatch as exc:
+        print("error: selector matched multiple sessions:", file=sys.stderr)
+        for r in exc.candidates:
+            print(f"  {r.title}  (id={r.local_session_id}, model={r.model})", file=sys.stderr)
+        print("Refine --title or use --id.", file=sys.stderr)
+        return None, 3
+
+
+def _cmd_list(args: argparse.Namespace) -> int:
     root = Path(args.sessions_root).expanduser()
     refs = discover.list_sessions(root, include_archived=args.all)
     if args.json:
@@ -80,19 +98,11 @@ def _cmd_list(args) -> int:
     return 0
 
 
-def _cmd_rehome(args) -> int:
+def _cmd_rehome(args: argparse.Namespace) -> int:
     root = Path(args.sessions_root).expanduser()
-    try:
-        ref = discover.resolve(root, title=args.title, sid=args.id, from_dir=args.from_dir)
-    except discover.NoMatch as exc:
-        print(f"error: {exc}", file=sys.stderr)
-        return 2
-    except discover.AmbiguousMatch as exc:
-        print("error: selector matched multiple sessions:", file=sys.stderr)
-        for r in exc.candidates:
-            print(f"  {r.title}  (id={r.local_session_id}, model={r.model})", file=sys.stderr)
-        print("Refine --title or use --id.", file=sys.stderr)
-        return 3
+    ref, rc = _resolve_session(root, args)
+    if ref is None:
+        return rc
 
     try:
         plan = rehome.build_plan(
@@ -129,19 +139,11 @@ def _cmd_rehome(args) -> int:
     return 0
 
 
-def _cmd_switch_model(args) -> int:
+def _cmd_switch_model(args: argparse.Namespace) -> int:
     root = Path(args.sessions_root).expanduser()
-    try:
-        ref = discover.resolve(root, title=args.title, sid=args.id, from_dir=args.from_dir)
-    except discover.NoMatch as exc:
-        print(f"error: {exc}", file=sys.stderr)
-        return 2
-    except discover.AmbiguousMatch as exc:
-        print("error: selector matched multiple sessions:", file=sys.stderr)
-        for r in exc.candidates:
-            print(f"  {r.title}  (id={r.local_session_id}, model={r.model})", file=sys.stderr)
-        print("Refine --title or use --id.", file=sys.stderr)
-        return 3
+    ref, rc = _resolve_session(root, args)
+    if ref is None:
+        return rc
 
     try:
         session = cowork_session.build(ref, to_model=args.to_model, title=args.title_as)
@@ -157,13 +159,14 @@ def _cmd_switch_model(args) -> int:
     print(f"{tag}Copied {len(session.copied)} file(s)")
     for warning in session.warnings:
         print(f"{tag}warn: {warning}")
-    print()
-    print("Quit Cowork fully (Cmd+Q) and relaunch to see it on the new model.")
-    print(f"Cleanup: {session.cleanup_cmd}")
+    if not args.dry_run:
+        print()
+        print("Quit Cowork fully (Cmd+Q) and relaunch to see it on the new model.")
+        print(f"Cleanup: {session.cleanup_cmd}")
     return 0
 
 
-def main(argv=None) -> int:
+def main(argv: list[str] | None = None) -> int:
     args = _build_parser().parse_args(list(sys.argv[1:] if argv is None else argv))
     if args.cmd == "list":
         return _cmd_list(args)