stack/packages/mosaic/framework/tools/_scripts/mosaic-ensure-sequential-thinking

#!/usr/bin/env bash
set -euo pipefail

# Fleet launches execute this source through an already-validated absolute bash
# capability and pass all interpreter capabilities explicitly.  Do not add PATH
# lookup here: this helper is intentionally capability-minimal.
MODE="apply"
RUNTIME="all"
STRICT_CHECK=0
CLAUDE_CONFIG_DIR=""
PYTHON_BIN=""
NODE_BIN=""
NPX_BIN=""
TIMEOUT_BIN=""
PKG="@modelcontextprotocol/server-sequential-thinking"

err() { echo "[mosaic-seq] ERROR: $*" >&2; }
log() { echo "[mosaic-seq] $*"; }

while [[ $# -gt 0 ]]; do
  case "$1" in
    --check) MODE="check"; shift ;;
    --runtime) RUNTIME="${2:?--runtime requires a value}"; shift 2 ;;
    --strict) STRICT_CHECK=1; shift ;;
    --claude-config-dir) CLAUDE_CONFIG_DIR="${2:?--claude-config-dir requires a value}"; shift 2 ;;
    --python-bin) PYTHON_BIN="${2:?--python-bin requires a value}"; shift 2 ;;
    --node-bin) NODE_BIN="${2:?--node-bin requires a value}"; shift 2 ;;
    --npx-bin) NPX_BIN="${2:?--npx-bin requires a value}"; shift 2 ;;
    --timeout-bin) TIMEOUT_BIN="${2:?--timeout-bin requires a value}"; shift 2 ;;
    *) err "Unknown argument: $1"; exit 2 ;;
  esac
done
case "$RUNTIME" in all|claude|codex|opencode) ;; *) err "Invalid runtime: $RUNTIME"; exit 2;; esac
# Explicit fleet-seat operation is capability-minimal. Legacy operator repair
# keeps its documented PATH-based compatibility contract.
if [[ -n "$CLAUDE_CONFIG_DIR" || -n "$PYTHON_BIN$NODE_BIN$NPX_BIN$TIMEOUT_BIN" ]]; then
  [[ -n "$PYTHON_BIN" && -n "$NODE_BIN" && -n "$NPX_BIN" ]] || { err "Fleet capabilities are required"; exit 2; }
else
  PYTHON_BIN=python3
  NODE_BIN=node
  NPX_BIN=npx
  TIMEOUT_BIN=timeout
fi

warm_package() {
  local timeout_sec="${MOSAIC_SEQ_WARM_TIMEOUT_SEC:-15}"
  if [[ -n "$TIMEOUT_BIN" ]]; then "$TIMEOUT_BIN" "$timeout_sec" "$NPX_BIN" -y "$PKG" --help >/dev/null 2>&1
  else "$NPX_BIN" -y "$PKG" --help >/dev/null 2>&1; fi
}

claude_config_python='import json, os, stat, tempfile
from pathlib import Path

def die(): raise SystemExit(1)
def secure_dir(p):
    p=Path(p)
    if not p.is_absolute(): die()
    # Every parent may be sticky /tmp, but none may be a symlink.  The fleet
    # config root itself must be private and owned by the invoking principal.
    for q in [p, *p.parents]:
        try: s=os.lstat(q)
        except OSError: die()
        if stat.S_ISLNK(s.st_mode) or not stat.S_ISDIR(s.st_mode): die()
        if q != p and s.st_mode & 0o022 and not (s.st_mode & stat.S_ISVTX): die()
    s=os.lstat(p)
    if s.st_uid not in (os.geteuid(), 0) or s.st_mode & 0o022: die()
    return p

def read_private(p):
    try: fd=os.open(p, os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK)
    except OSError: die()
    try:
        s=os.fstat(fd)
        if not stat.S_ISREG(s.st_mode) or s.st_uid not in (os.geteuid(),0) or s.st_mode & 0o077 or s.st_size>1048576: die()
        data=b""
        while len(data)<=1048576:
            c=os.read(fd,65536)
            if not c: break
            data+=c
        if len(data)>1048576: die()
        return data, (s.st_dev,s.st_ino)
    finally: os.close(fd)

def entry_ok(data):
    try: d=json.loads(data.decode()); e=d.get("mcpServers",{}).get("sequential-thinking",{})
    except Exception: return False
    return e.get("command")=="npx" and e.get("args")==["-y","@modelcontextprotocol/server-sequential-thinking"]

def explicit_check_or_apply(apply):
    root=secure_dir(os.environ["CLAUDE_CONFIG_DIR"]); p=root/".claude.json"
    if not apply: return 0 if entry_ok(read_private(str(p))[0]) else 1
    old={}; identity=None
    if os.path.lexists(p):
        raw,identity=read_private(str(p))
        try: old=json.loads(raw.decode())
        except Exception: old={}
    mcp=old.get("mcpServers") if isinstance(old.get("mcpServers"),dict) else {}
    mcp["sequential-thinking"]={"command":"npx","args":["-y","@modelcontextprotocol/server-sequential-thinking"]}; old["mcpServers"]=mcp
    fd,tmp=tempfile.mkstemp(prefix=".claude.json.",dir=root)
    try:
        os.fchmod(fd,0o600); os.write(fd,(json.dumps(old,indent=2)+"\n").encode()); os.fsync(fd); os.close(fd)
        try: now=os.lstat(p); current=(now.st_dev,now.st_ino)
        except FileNotFoundError: current=None
        if current!=identity: die()
        os.replace(tmp,p)
    finally:
        try: os.close(fd)
        except OSError: pass
        try: os.unlink(tmp)
        except FileNotFoundError: pass
    return 0

if os.environ.get("CLAUDE_CONFIG_DIR"):
    raise SystemExit(explicit_check_or_apply(os.environ.get("SEQ_APPLY")=="1"))
# Compatibility path is intentionally not fleet-authoritative.
p=Path.home()/".claude.json"
if not p.exists() and not os.environ.get("SEQ_APPLY")=="1": p=Path.home()/".claude"/"settings.json"
if os.environ.get("SEQ_APPLY")=="1":
    try: d=json.loads(p.read_text()) if p.exists() else {}
    except Exception: d={}
    m=d.get("mcpServers") if isinstance(d.get("mcpServers"),dict) else {}
    m["sequential-thinking"]={"command":"npx","args":["-y","@modelcontextprotocol/server-sequential-thinking"]}; d["mcpServers"]=m
    p.parent.mkdir(parents=True,exist_ok=True); p.write_text(json.dumps(d,indent=2)+"\n")
    raise SystemExit(0)
try: raise SystemExit(0 if entry_ok(p.read_bytes()) else 1)
except Exception: raise SystemExit(1)'

check_claude_config() { CLAUDE_CONFIG_DIR="$CLAUDE_CONFIG_DIR" SEQ_APPLY=0 "$PYTHON_BIN" -c "$claude_config_python"; }
apply_claude_config() { CLAUDE_CONFIG_DIR="$CLAUDE_CONFIG_DIR" SEQ_APPLY=1 "$PYTHON_BIN" -c "$claude_config_python"; }
check_codex_config() { CODEX_CFG="${CODEX_HOME:-$HOME/.codex}/config.toml" "$PYTHON_BIN" -c 'import os,re; from pathlib import Path; s=Path(os.environ["CODEX_CFG"]).read_text(); ok=bool(re.search(r"^\[mcp_servers\.(sequential-thinking|sequential_thinking)\]",s,re.M) and "command = \"npx\"" in s and "@modelcontextprotocol/server-sequential-thinking" in s); raise SystemExit(0 if ok else 1)'; }
apply_codex_config() { CODEX_CFG="${CODEX_HOME:-$HOME/.codex}/config.toml" "$PYTHON_BIN" -c 'import os,re; from pathlib import Path; p=Path(os.environ["CODEX_CFG"]); p.parent.mkdir(parents=True,exist_ok=True); out=[]; skip=False
for line in (p.read_text().splitlines() if p.exists() else []):
  if re.match(r"^\[mcp_servers\.(sequential-thinking|sequential_thinking)\]$",line): skip=True; continue
  if skip and line.startswith("["): skip=False
  if not skip: out.append(line)
p.write_text("\n".join(out).rstrip()+"\n\n[mcp_servers.sequential-thinking]\ncommand = \"npx\"\nargs = [\"-y\", \"@modelcontextprotocol/server-sequential-thinking\"]\n")'; }
check_opencode_config() { XDG_CONFIG_HOME="${XDG_CONFIG_HOME:-}" "$PYTHON_BIN" -c 'import json,os; from pathlib import Path; p=Path(os.environ["XDG_CONFIG_HOME"])/"opencode/config.json" if os.environ.get("XDG_CONFIG_HOME") else Path.home()/".config/opencode/config.json"; d=json.loads(p.read_text()); e=d.get("mcp",{}).get("sequential-thinking"); expected={"type":"local","command":["npx","-y","@modelcontextprotocol/server-sequential-thinking"],"enabled":True}; raise SystemExit(0 if e==expected else 1)' ; }
apply_opencode_config() { XDG_CONFIG_HOME="${XDG_CONFIG_HOME:-}" "$PYTHON_BIN" -c 'import json,os; from pathlib import Path; p=Path(os.environ["XDG_CONFIG_HOME"])/"opencode/config.json" if os.environ.get("XDG_CONFIG_HOME") else Path.home()/".config/opencode/config.json"; p.parent.mkdir(parents=True,exist_ok=True); d=json.loads(p.read_text()) if p.exists() else {}; m=d.get("mcp") if isinstance(d.get("mcp"),dict) else {}; m["sequential-thinking"]={"type":"local","command":["npx","-y","@modelcontextprotocol/server-sequential-thinking"],"enabled":True}; d["mcp"]=m; p.write_text(json.dumps(d,indent=2)+"\n")'; }
check_runtime_config() { case "$RUNTIME" in all) check_claude_config && check_codex_config && check_opencode_config;; claude) check_claude_config;; codex) check_codex_config;; opencode) check_opencode_config;; esac; }
apply_runtime_config() { case "$RUNTIME" in claude) apply_claude_config;; codex) apply_codex_config;; opencode) apply_opencode_config;; all) apply_claude_config && apply_codex_config && apply_opencode_config;; esac; }
if [[ "$MODE" == check ]]; then
  check_runtime_config
  if [[ "$STRICT_CHECK" == 1 || "${MOSAIC_SEQ_CHECK_WARM:-0}" == 1 ]]; then warm_package || { err "sequential-thinking package warm-up failed in strict mode"; exit 1; }; fi
  log "sequential-thinking MCP is configured and available (${RUNTIME})"; exit 0
fi
warm_package || { err "sequential-thinking package warm-up failed"; exit 1; }
apply_runtime_config
log "sequential-thinking MCP configured (${RUNTIME})"