fix(agent-wrapper): stop assuming POSIX-only calls and path styles

Windows runs hit three platform assumptions in the agent wrappers:

- `CodexAgentWrapper._effective_config_snapshot()` called `os.fchmod`,
  which only exists on POSIX. Building the MCP config snapshot therefore
  raised `AttributeError` on Windows and the Codex wrapper could not
  start. The call is now guarded by a capability check
  (`hasattr(os, "fchmod")`), which also keeps the guard testable on
  POSIX hosts.
- `CcFileSessionStore.list_subkeys()` rendered subpaths with `str()`,
  which yields backslashes on Windows. The session-store contract - and
  the SDK conformance check shipped with `claude_agent_sdk` - expects
  POSIX subpaths such as `subagents/agent-2`. Render them with
  `as_posix()`.
- `AutoMemoryStep` wrote session JSONL with the platform default newline
  translation, so Windows stored CRLF even though the writer appends
  `"\n"`. Pass `newline=""` on the append and rewrite paths so one
  message stays one LF-terminated line on every platform.

Add a regression test that pins the snapshot path while `os.fchmod` is
unavailable, so the guard is covered on POSIX CI too.
This commit is contained in:
BeiMu-new 2026-10-23 22:29:59 +08:00
parent c773fa7b77
commit a265fbf8cd
4 changed files with 18 additions and 4 deletions

View file

@ -104,7 +104,7 @@ class CcFileSessionStore:
if not session_dir.is_dir():
return []
return [
str(path.relative_to(session_dir).with_suffix(""))
path.relative_to(session_dir).with_suffix("").as_posix()
for path in session_dir.rglob("*.jsonl")
if path.is_file()
]

View file

@ -175,7 +175,8 @@ class CodexAgentWrapper(BaseAgentWrapper):
snapshot_dir.mkdir(parents=True, exist_ok=True)
fd, raw_path = tempfile.mkstemp(prefix="config-", suffix=".json", dir=snapshot_dir)
try:
os.fchmod(fd, 0o600)
if hasattr(os, "fchmod"):
os.fchmod(fd, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as stream:
json.dump(self.app_context.app_config.model_dump(mode="json"), stream)
except BaseException:

View file

@ -331,7 +331,7 @@ class AutoMemoryStep(BaseStep):
if can_append:
new_msgs = merged[len(existing) :]
if new_msgs:
async with aiofiles.open(path, "a", encoding="utf-8") as f:
async with aiofiles.open(path, "a", encoding="utf-8", newline="") as f:
for msg in new_msgs:
await f.write(_sanitize_msg_for_save(msg).model_dump_json() + "\n")
self.logger.info(
@ -344,7 +344,7 @@ class AutoMemoryStep(BaseStep):
f"existing={len(existing)} total={len(merged)}",
)
else:
async with aiofiles.open(path, "w", encoding="utf-8") as f:
async with aiofiles.open(path, "w", encoding="utf-8", newline="") as f:
for msg in merged:
await f.write(_sanitize_msg_for_save(msg).model_dump_json() + "\n")
self.logger.info(

View file

@ -5,6 +5,7 @@
import asyncio
from dataclasses import dataclass
import json
import os
from pathlib import Path
import stat
import sys
@ -575,6 +576,18 @@ def test_effective_mcp_config_snapshot_is_private_and_removed_on_close(tmp_path)
assert not snapshot.exists()
def test_effective_mcp_config_snapshot_needs_no_fchmod(tmp_path, monkeypatch):
"""A platform without ``os.fchmod`` (Windows) still gets a usable snapshot."""
monkeypatch.delattr(os, "fchmod", raising=False)
wrapper, _job = _wrapper(tmp_path)
config = wrapper._mcp_server_config({"job_tools": ["search"]})
snapshot = Path(config["args"][config["args"].index("--config") + 1])
assert snapshot.exists()
assert json.loads(snapshot.read_text(encoding="utf-8"))["workspace_dir"] == str(tmp_path)
@pytest.mark.asyncio
async def test_effective_snapshot_exposes_parent_only_custom_job(tmp_path):
from fastmcp import Client