定期タスクの冒頭で WS が空になった朝、シェルが二つあることを知りました

# 1) 代入の終了コード
WS="$(ls -d /sessions/*/mnt/Workspace 2>/dev/null | head -1)"
echo "assign_status=$? ws=[$WS]"
# → assign_status=0 ws=[]
 
# 2) pipefail を有効にすると
set -o pipefail
WS="$(ls -d /sessions/*/mnt/Workspace 2>/dev/null | head -1)"
echo "pipefail_assign_status=$? ws=[$WS]"
# → pipefail_assign_status=2 ws=[]
 
# 3) 空のまま後続で使うと
cat "$WS/_skills/x/SKILL.md"
# → cat: /_skills/x/SKILL.md: No such file or directory

: "${WS:?作業フォルダが見えないシェルです。端末側のシェルに切り替えてください}"

WS="$(ls -d /sessions/*/mnt/Workspace 2>/dev/null | head -1)"
echo "shell=$(hostname) py=$(python3 -c 'import platform;print(platform.python_version())') ws=[${WS:+found}]"
: "${WS:?作業フォルダが見えないシェルです。端末側のシェルに切り替えてください}"

"""二つのシェルを持つエージェントのための薄いガード。標準ライブラリだけで動きます。"""
import glob, json, os, platform, socket, subprocess, sys
 
WS_NAME = "Workspace"  # 作業フォルダ名（例）
 
# ツールごとの「契約」。説明文と同じ内容を機械側でも持つ。
CONTRACT = {
    "run_in_sandbox": {"needs_workspace": False},
    "run_on_desktop": {"needs_workspace": True},
}
 
 
def fingerprint():
    """シェルの素性。空文字ではなく、必ず何かを返す。"""
    found = glob.glob(f"/sessions/*/mnt/{WS_NAME}") + glob.glob(
        os.path.expanduser(f"~/mnt/{WS_NAME}")
    )
    return {
        "host": socket.gethostname(),
        "python": platform.python_version(),
        "has_sessions": os.path.isdir("/sessions"),
        "workspace": found[0] if found else None,
    }
 
 
def dispatch(tool_use_id, name, command, fp, expect_empty=False):
    """tool_use を実行し、Messages API にそのまま返せる tool_result を作る。"""
    need = CONTRACT[name]["needs_workspace"]
    if need and not fp["workspace"]:
        return {
            "type": "tool_result", "tool_use_id": tool_use_id, "is_error": True,
            "content": f"{name} の実行先 (host={fp['host']}) に作業フォルダがありません。"
                       f"ツールと実行先の対応を確認してください。",
        }
    p = subprocess.run(["bash", "-c", command], capture_output=True, text=True, timeout=30)
    out = p.stdout.strip()
    empty_is_error = (not out) and (not expect_empty)
    if p.returncode != 0 or empty_is_error:
        why = f"exit={p.returncode}" if p.returncode else "exit=0 だが出力が空"
        return {
            "type": "tool_result", "tool_use_id": tool_use_id, "is_error": True,
            "content": f"{why}。stderr={p.stderr.strip()[:200]!r}。"
                       f"パスの解決結果が空でないか、実行先のシェルが正しいかを確認してください。",
        }
    return {"type": "tool_result", "tool_use_id": tool_use_id, "content": out}

{"host": "vm", "python": "3.13.16", "has_sessions": false, "workspace": null}
→ is_error: true
  "run_on_desktop の実行先 (host=vm) に作業フォルダがありません。ツールと実行先の対応を確認してください。"

→ is_error: true
  "exit=0 だが出力が空。stderr=''。パスの解決結果が空でないか、実行先のシェルが正しいかを確認してください。"

import anthropic
import twoshell_guard as g
 
MODEL = "claude-sonnet-5-5"
 
TOOLS = [
    {
        "name": "run_in_sandbox",
        "description": (
            "使い捨てのクラウド側サンドボックスでコマンドを実行します。"
            "見えるのは /tmp と作業用ディレクトリだけで、ユーザーのフォルダは見えません。"
            "外部への通信が制限されることがあります。"
        ),
        "input_schema": {
            "type": "object",
            "properties": {"command": {"type": "string"}},
            "required": ["command"],
        },
    },
    {
        "name": "run_on_desktop",
        "description": (
            "ユーザーの端末側のシェルでコマンドを実行します。"
            "接続された作業フォルダは $HOME/mnt/Workspace にあります。"
            "ユーザーのファイルを読み書きするときは、必ずこちらを使います。"
            "接続したフォルダの中のファイルは、既定では削除できません。"
        ),
        "input_schema": {
            "type": "object",
            "properties": {"command": {"type": "string"}},
            "required": ["command"],
        },
    },
]
 
 
def run_turn(client, messages, fps):
    """1ターン進める。fps は {ツール名: fingerprint()} を起動時に集めたもの。"""
    resp = client.messages.create(
        model=MODEL, max_tokens=2048, tools=TOOLS, messages=messages
    )
    messages.append({"role": "assistant", "content": resp.content})
    results = [
        g.dispatch(b.id, b.name, b.input["command"], fps[b.name])
        for b in resp.content if b.type == "tool_use"
    ]
    if results:
        messages.append({"role": "user", "content": results})
    return resp.stop_reason