ผูก Claude CLI เข้าห้อง Telegram บน Windows บอทตอบใน 0 วินาที

พอดีผมอยากสั่งงานเอกสารจากห้องแชทกลุ่มได้ ห้องนี้มีสามสมาชิก คือผม ผู้ร่วมงานหนึ่งคน และบอท เป้าหมายคือพิมพ์สั่งในห้องแล้วได้ PDF กลับเข้าห้องเดิม โดยให้ Claude Code CLI รันแบบ headless ด้วย cwd เป็นโฟลเดอร์โปรเจกต์ มันเลยอ่านไฟล์กติกา memory และ skill ของโปรเจกต์นั้นได้เองเหมือนตอนผมนั่งสั่งหน้าเครื่อง ผมเขียนตัวห่อเป็น Python ไฟล์เดียว จบที่ 366 บรรทัด แต่กว่าจะใช้ได้จริงเจอกับดักต่อกันสี่ข้อในคืนเดียว และทุกข้ออาการหน้าจอชี้ไปผิดทางทั้งนั้น ผมเลยเอามาเขียนเรียงตามลำดับที่เจอ

ชื่อห้อง ชื่อคน ชื่อบอท path และรหัสทุกตัวในบทความเป็นค่าตัวอย่าง ส่วนอาการ ข้อความ error ตัวเลขเวลา และโครงโค้ดเป็นของจริง บทความนี้พูดเฉพาะการห่อ CLI ส่วนกติกาขาเข้าของ Bot API เช่นคิวข้อความกับการเริ่มแชท ผมแยกไปเขียนไว้ที่ บอทอ่านข้อความย้อนหลังไม่ได้ และทักใครก่อนไม่ได้ ส่วนขาออกทางเดียวอยู่ที่ ส่งไฟล์เข้า Telegram กับเพดาน 50 MB

โครงของตัวห่อ ก่อนจะเจอปัญหา

ตัวห่อทำงานแบบ long-poll ธรรมดา เรียก getUpdates ด้วย timeout 25 วินาที ขอเฉพาะ update ชนิด message เก็บ offset ลงไฟล์ state ทุก update ข้ามข้อความที่มาจากบอท และข้ามห้องที่ไม่อยู่ในรายการอนุญาต ไฟล์หรือรูปที่แนบมาจะถูกดาวน์โหลดลงโฟลเดอร์ inbox แยกตามวันที่ แล้วส่ง path ให้ Claude เปิดดูเอง

# วงรอบหลักของตัวห่อ (ย่อ)
import json, os, time, urllib.parse, urllib.request

TOK = os.environ["TG_BOT_TOKEN"]          # ไม่ฝังรหัสลงไฟล์
ALLOWED = {int(os.environ["TG_ROOM_ID"]): "ห้องงานเอกสาร"}

def api(method, **q):
    url = "https://api.telegram.org/bot%s/%s?%s" % (TOK, method, urllib.parse.urlencode(q))
    with urllib.request.urlopen(url, timeout=60) as r:
        return json.load(r)

def main():
    st = load_state()                      # {"offset": 0}
    while True:
        try:
            r = api("getUpdates", offset=st["offset"], timeout=25,
                    allowed_updates=json.dumps(["message"]))
        except Exception as e:
            log("poll err %r" % e); time.sleep(5); continue
        for u in r.get("result", []):
            st["offset"] = u["update_id"] + 1
            save_state(st)
            m = u.get("message")
            if not m or (m.get("from") or {}).get("is_bot"):
                log("ข้าม: ไม่ใช่ข้อความจากคน"); continue
            if m["chat"]["id"] not in ALLOWED:
                log("ข้าม: ห้องที่ไม่อนุญาต"); continue
            handle(m)

บรรทัด log("ข้าม: ...") สองบรรทัดนั้นไม่ได้มีตั้งแต่แรก และมันคือบทเรียนข้อศูนย์ รุ่นแรกผมตั้งกติกาว่าในห้องกลุ่มต้องขึ้นต้นข้อความด้วยคำว่า "ถาม" บอทถึงจะตอบ แล้วผมเองก็ลืมกติกาตัวเอง พิมพ์ปกติเข้าไปบอทก็เงียบ พอไปดู log ก็ว่างเปล่า เพราะข้อความที่ถูกข้ามไม่ถูกบันทึกอะไรเลย มองจาก log ไม่มีทางรู้ว่ามีคนถามเข้ามาแล้วโดนทิ้ง ผมเลยแก้สองอย่างพร้อมกัน คือให้ตอบทุกข้อความในห้องที่อนุญาต และ log ทุกข้อความที่เข้ามาพร้อมเหตุผลที่ข้าม

กับดักที่ 1: cmd.exe ตัดข้อความเหลือแต่บรรทัดหัว

พอบอทเริ่มตอบ คำตอบแรกที่ได้คือ "ไม่เห็นข้อความ" ทั้งที่ผมพิมพ์คำสั่งหลายบรรทัดเข้าไป อ่านแล้วชวนให้คิดว่าโมเดลตอบไม่ดี หรือ prompt ที่ห่อไว้เขียนไม่ชัด ซึ่งไม่ใช่ทั้งสองอย่าง

ต้นเหตุคือรุ่นแรกผมส่ง prompt เป็น argument ของคำสั่ง แล้วเรียกผ่าน shell หน้าตาประมาณนี้

# รุ่นแรก: ส่ง prompt ทาง argument ผ่าน cmd.exe
prompt = "ข้อความจาก %s ทาง Telegram:\n%s" % (name, text)
cmd = 'claude -p "%s"' % prompt
out = subprocess.run(cmd, shell=True, capture_output=True,
                     text=True, cwd=WORKSPACE).stdout

บน Windows คำว่า shell=True แปลว่าคำสั่งทั้งบรรทัดถูกส่งให้ cmd.exe ตีความ และ cmd.exe มองการขึ้นบรรทัดใหม่เป็นจุดจบคำสั่ง ทุกอย่างหลังบรรทัดแรกถูกตัดทิ้งตรงนั้น สิ่งที่ไปถึง Claude เลยเหลือแค่บรรทัดหัว ข้อความจาก <ชื่อ> ทาง Telegram: แล้วจบ โมเดลตอบถูกแล้วว่าไม่เห็นข้อความ เพราะมันไม่มีข้อความให้เห็นจริง ๆ

ทางแก้คือไม่ส่งเนื้อหาทาง argument เลย ส่งทาง stdin แทน ซึ่งไม่มีใครมาตีความตัวอักษรระหว่างทาง ขึ้นบรรทัดใหม่กี่บรรทัด มีเครื่องหมายคำพูดกี่ตัวก็ผ่านหมด

กับดักที่ 2: ตอบใน 0 วินาที เพราะโปรเซสไม่ได้รัน

แก้ข้อแรกเสร็จผมเพิ่มข้อจำกัดโฟลเดอร์ลงใน --append-system-prompt ข้อความชุดนี้มี path แบบ Windows ที่มี backslash และมีเครื่องหมายคำพูดอยู่ข้างใน หลังจากนั้นทุกคำสั่งที่พิมพ์เข้าห้องได้คำตอบเดียวกันหมด และจอของตัวห่อขึ้นว่าตอบภายใน 0 วินาทีทุกครั้ง

HH:MM:SS [ห้องงานเอกสาร] <ชื่อ> ถาม: <ข้อความที่พิมพ์>
HH:MM:SS ตอบ (0 วินาที):
The system cannot find the file specified.

ข้อความบอกว่าหาไฟล์ไม่เจอ อ่านแล้วชวนให้ไปไล่ดู path ของไฟล์ settings หรือไฟล์ที่สั่งให้เปิด ความจริงตัวเลข 0 วินาทีเป็นเบาะแสที่ชัดกว่าข้อความ โมเดลตอบเร็วแค่ไหนก็ไม่มีทาง 0 วินาที ตัวเลขนี้แปลว่าโปรเซส Claude ไม่ได้เกิดขึ้นมาเลย และข้อความที่เห็นคือ error ของ cmd.exe เอง ที่ตีความ backslash กับเครื่องหมายคำพูดใน system prompt ผิดจนคำสั่งทั้งบรรทัดเสียรูป ตัวห่อของผมเอา stdout มาส่งต่อเข้าห้องตรง ๆ มันเลยดูเหมือนบอทตอบ

ทางแก้คือเลิกใช้ shell=True แล้วส่ง list ให้ Popen เรียก claude.exe ตรง ๆ แต่ละ argument เป็นสมาชิกหนึ่งตัวของ list ไม่มี cmd.exe มาแยกคำให้อีก

def claude_cmd():
    # ห้ามใช้ shell=True: cmd.exe ตีความ backslash กับ quote ใน system prompt ผิด
    return [CLAUDE, "-p",
            "--input-format", "stream-json",
            "--output-format", "stream-json", "--verbose",
            "--model", MODEL,
            "--permission-mode", "acceptEdits",
            "--settings", os.path.join(HERE, "claude-settings.json"),
            "--add-dir", r"C:\projects\web",
            "--allowedTools", "Read", "Write", "Edit", "Glob", "Grep", "Bash",
            "--append-system-prompt", SYSTEM]

กับดักที่ 3: เปิดโปรเซสใหม่ทุกข้อความ 22 วินาที

พอทำงานได้ถูกแล้วก็เจอว่าช้า ทุกข้อความใช้เวลาราว 22 วินาทีแม้จะถามสั้น ๆ เหตุผลคือการเปิดโปรเซสใหม่ทุกครั้งทำให้ CLI ต้องอ่านไฟล์กติกาของโปรเจกต์กับ memory อีก 41 ไฟล์ (ราว 24 KB) ใหม่ทั้งหมด แล้วสร้าง cache ใหม่ครั้งละ 3 ถึง 5 หมื่น token ก่อนจะได้เริ่มคิดเรื่องที่ถาม จ่ายทั้งเวลาและค่าใช้จ่ายซ้ำทุกข้อความ

ผมเปลี่ยนเป็นเปิดโปรเซสค้างไว้หนึ่งตัวต่อหนึ่งห้อง ใช้โหมด stream-json ทั้งขาเข้าและขาออก เขียนข้อความใหม่เป็น JSON หนึ่งบรรทัดลง stdin แล้วอ่าน stdout ทีละบรรทัดจนเจอ event ที่ type เป็น result

class Session:
    """โปรเซส Claude ที่เปิดค้างไว้ต่อหนึ่งห้อง อ่านกติกากับ memory ครั้งเดียวตอนเปิด"""
    procs = {}

    @classmethod
    def get(cls, key):
        p = cls.procs.get(key)
        if p and p.poll() is None:
            return p
        p = subprocess.Popen(claude_cmd(), cwd=WORKSPACE,
                             stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE, encoding="utf-8",
                             errors="replace", bufsize=1)
        cls.procs[key] = p
        return p

def run_claude(prompt, key):
    for attempt in (1, 2):                 # โปรเซสตาย = เปิดใหม่ ลองซ้ำครั้งเดียว
        p = Session.get(key)
        msg = {"type": "user", "message": {"role": "user",
               "content": [{"type": "text", "text": prompt}]}}
        p.stdin.write(json.dumps(msg, ensure_ascii=False) + "\n")
        p.stdin.flush()
        deadline = time.time() + 1800
        while time.time() < deadline:
            line = p.stdout.readline()
            if not line:
                break                      # ท่อปิด แปลว่าโปรเซสตาย
            try:
                ev = json.loads(line)
            except ValueError:
                continue
            if ev.get("type") == "assistant":
                for c in (ev.get("message") or {}).get("content", []):
                    if c.get("type") == "tool_use":
                        log("   ใช้เครื่องมือ: %s" % c.get("name"))
            elif ev.get("type") == "result":
                if ev.get("is_error"):
                    return "ผิดพลาด: %s" % ev.get("result")
                return ev.get("result") or "(ไม่มีข้อความตอบ)"
        Session.close(key)
    return "session หลุด ลองใหม่อีกครั้ง"

วัดจริงได้ 22 วินาทีสำหรับข้อความแรกของห้อง (ยังต้องโหลดทุกอย่างหนึ่งรอบ) แล้วเหลือ 6 วินาทีตั้งแต่ข้อความถัดไป ของแถมที่ได้คือ event ชนิด assistant ที่มี tool_use ผมเอาชื่อเครื่องมือไปพิมพ์ลง log ทำให้เห็นว่าระหว่างรอ agent กำลังอ่านไฟล์ แก้ไฟล์ หรือรันคำสั่งอะไรอยู่ ซึ่งจะกลายเป็นหลักฐานสำคัญในกับดักข้อถัดไป

ข้อแลกเปลี่ยนของโปรเซสค้างคือบริบทของห้องสะสมไปเรื่อย ๆ ในโปรเซสเดียว และถ้าโปรเซสตายกลางทางต้องมีทางฟื้น ผมใส่ไว้สองอย่าง คืออ่าน stdout ไม่ได้ให้ถือว่าโปรเซสตาย ปิดแล้วเปิดใหม่ลองซ้ำหนึ่งครั้ง กับ timeout ต่อข้อความ 1800 วินาทีสำหรับงานทำเอกสารที่ยาวจริง

กับดักที่ 4: agent บอกว่าส่งไฟล์แล้ว แต่ในห้องไม่มี

งานแรกที่สั่งจริงคือให้ทำ PDF แล้วส่งเข้าห้อง agent ตอบว่าทำเสร็จและส่งให้แล้ว แต่ในห้องไม่มีไฟล์ คำถามที่ตามมาคือ "ทำไมในห้องไม่มี PDF"

ผมเปิด log ของตัวห่อดู ไม่มีบรรทัดส่งไฟล์สักบรรทัด ไล่ดูแล้วไฟล์ PDF ถูกสร้างขึ้นจริง แล้ว agent ไปเจอสคริปต์ส่ง Telegram อีกตัวที่มีอยู่ในเครื่อง ซึ่งผมทำไว้ส่งไฟล์หาตัวเอง ปลายทางของสคริปต์นั้นเป็นแชทส่วนตัวของผม มันเรียกสคริปต์นั้น ได้ผลว่าสำเร็จ แล้วรายงานตามที่เห็นว่าส่งแล้ว ไม่มีใครพูดผิดสักคน แต่ไฟล์ไปอยู่ผิดที่

ผมเลยตัดทางนั้นออกทั้งหมด ไม่ให้ agent ถือรหัสบอทหรือเรียก API เอง แล้วกำหนดเป็นโปรโตคอลแทน คือถ้าต้องการส่งไฟล์ ให้จบคำตอบด้วยบรรทัด SEND: <path> หรือวางไฟล์ลงโฟลเดอร์ outbox ตัวห่อจะเป็นคนส่งเข้าห้องที่กำหนดไว้ในไฟล์ตั้งค่าเสมอ และใน system prompt เขียนชัดว่าห้ามพูดว่าส่งแล้วถ้าคำตอบนั้นไม่มีบรรทัด SEND

SEND_RE = re.compile(r"^\s*SEND:\s*(.+?)\s*$")

def deliver(out, chat, reply_to):
    keep = []
    for line in out.splitlines():
        m = SEND_RE.match(line)
        if not m:
            keep.append(line); continue
        fp = m.group(1).strip("\"'")
        if not os.path.isabs(fp):
            fp = os.path.join(WORKSPACE, fp)
        if os.path.isfile(fp):
            send_file(FILE_CHAT, fp, caption=os.path.basename(fp))   # log ทุกครั้ง
        else:
            keep.append("(ไม่พบไฟล์ %s)" % fp)
    flush_outbox(FILE_CHAT)               # ไฟล์ใน outbox ส่งแล้วย้ายเข้า sent/
    return "\n".join(keep).strip() or "(ส่งไฟล์แล้ว)"

หลังเปลี่ยนเป็นแบบนี้ บรรทัดใน log กลายเป็นหลักฐานชุดเดียวที่เชื่อได้ มีชื่อไฟล์ ขนาด และผลของ API ครบ ถ้ามีคนถามว่าไฟล์ไปไหน เปิด log แล้วตอบได้ทันทีโดยไม่ต้องถาม agent

จำกัดสิทธิ์ และแยกคนสั่งกับคนที่เก็บข้อความอย่างเดียว

ช่วงทดลองคืนแรกผมเปิดด้วย --dangerously-skip-permissions เพื่อให้มันทำงานได้โดยไม่ติดถาม ซึ่งไม่ควรอยู่นานกว่าช่วงทดลอง เพราะช่องทางนี้รับคำสั่งจากห้องแชท ผมเลยเปลี่ยนเป็นโหมด acceptEdits คู่กับไฟล์ settings ที่มี deny list เป็นรายโฟลเดอร์ ให้แตะได้เฉพาะโฟลเดอร์โปรเจกต์กับโฟลเดอร์เว็บในเครื่อง

{
  "permissions": {
    "defaultMode": "acceptEdits",
    "additionalDirectories": ["C:\\projects\\web"],
    "deny": [
      "Read(//C:/Users/**)",
      "Read(//C:/Windows/**)",
      "Read(//D:/other-project/**)",
      "Edit(//C:/Users/**)",
      "Edit(//D:/other-project/**)",
      "Write(//C:/Users/**)",
      "Write(//C:/Windows/**)",
      "Write(//D:/other-project/**)",
      "Bash(format:*)",
      "Bash(rd /s:*)",
      "Bash(rmdir /s:*)"
    ]
  }
}

deny list กันได้เฉพาะเครื่องมืออ่านเขียนไฟล์กับคำสั่งที่เขียนตรงตามรูปแบบ ผมไม่ถือว่ามันเป็นกำแพงที่สมบูรณ์ มันคือรั้วกันพลาด ส่วนแนวคิดเรื่องการคุม AI ที่มีสิทธิ์รันคำสั่ง ผมเขียนไว้ละเอียดกว่านี้ใน คุม Linux server ด้วย AI ผ่าน SSH ให้ปลอดภัย

ชั้นที่สำคัญกว่าคือใครสั่งได้ ในห้องมีคนสองคน ผมแยกเป็นสองรายชื่อในไฟล์ตั้งค่า คือ command_users ที่สั่งงานได้ มีคนเดียว กับ archive_users ที่ข้อความและไฟล์จะถูกเก็บอย่างเดียว เขียนต่อท้ายไฟล์ archive/messages.md และย้ายไฟล์แนบไปไว้ใน archive/<วันที่>/ ไม่รันคำสั่ง ไม่ตอบ อีกข้อคือข้อความที่ผมพิมพ์ขึ้นต้นด้วยชื่อผู้ร่วมงาน ตัวห่อจะข้าม เพราะนั่นคือคนคุยกับคน ไม่ใช่คำสั่งถึงบอท

สคริปต์เปิด โมเดล และ error ระหว่าง poll

ตัวห่อเปิดด้วย PowerShell เป็นหน้าต่างคอนโซลที่มองเห็นได้ ปิดหน้าต่างคือหยุด สิ่งที่ต้องตั้งก่อนเรียก Python มีสามบรรทัด ไม่งั้นภาษาไทยในคอนโซลเพี้ยนและ log ไม่ออกทันที

# start_agent.ps1
$here = Split-Path -Parent $MyInvocation.MyCommand.Path
Set-Location $here
[Console]::OutputEncoding = [System.Text.Encoding]::UTF8
$env:PYTHONIOENCODING = 'utf-8'
$env:PYTHONUNBUFFERED = '1'
try {
    & python (Join-Path $here 'agent.py')
}
finally {
    Write-Host '*** agent หยุดทำงานแล้ว ***'
    Read-Host 'กด Enter เพื่อปิดหน้าต่าง'
}

เรื่องโมเดล ผมเริ่มด้วยรุ่นกลาง (sonnet) งานจัด layout เอกสารออกมาไม่ดีพอ เลยเปลี่ยนเป็นรุ่นใหญ่ (opus) ซึ่งช้ากว่าและแพงกว่าราว 5 เท่า อันนี้เป็นการตัดสินใจตามชนิดงาน ถ้างานเป็นถามตอบสั้น ๆ รุ่นกลางก็พอ

ส่วนวงรอบ poll เจอ error จากเครือข่ายเป็นระยะตามปกติ ที่เห็นใน log จริงมีสามแบบ คือ HTTPError 502, ConnectionResetError 10054 และ getaddrinfo failed 11001 ทั้งหมดจัดการแบบเดียวกันคือ log แล้ว sleep 5 วินาทีแล้ววนต่อ ไม่ให้ตัวห่อตายเพราะเน็ตสะดุด

1. log ทุกข้อความที่เข้ามา รวมทั้งที่ถูกข้าม พร้อมเหตุผล
2. ส่งเนื้อหาที่คนพิมพ์เข้าทาง stdin เท่านั้น ไม่ผ่าน command line
3. เรียกโปรแกรมด้วย list ไม่ใช้ shell=True บน Windows
4. จับเวลาทุกคำตอบ คำตอบที่เร็วผิดธรรมชาติให้สงสัยชั้นเรียกโปรเซสก่อน
5. เปิดโปรเซสค้างต่อห้องด้วย stream-json มีทางฟื้นเมื่อโปรเซสตาย และมี timeout
6. การส่งไฟล์ออกให้ตัวห่อทำเอง ผ่านบรรทัด SEND หรือ outbox และ log ทุกครั้ง
7. ใช้ acceptEdits กับ deny list รายโฟลเดอร์ แทนการข้ามสิทธิ์ทั้งหมด
8. แยกรายชื่อคนสั่งงานออกจากคนที่เก็บข้อความอย่างเดียว

ตอนจบของเรื่องนี้ไม่ได้หวือหวา แปดวันหลังจากนั้นผมถอดตัวห่อออกจาก Startup ของเครื่อง เหลือเป็นไอคอนเปิดด้วยมือเวลาจะใช้ เพราะ agent ที่รับคำสั่งจากห้องแชทและมีสิทธิ์แก้ไฟล์ในเครื่อง ส่วนตัวผมว่าไม่ควรเปิดทิ้งไว้ตลอดถ้าไม่ได้นั่งดูอยู่ ใครจะเปิดค้างเป็นบริการก็ตามสะดวก แต่ควรมีรายชื่อคนสั่ง deny list และ log การส่งไฟล์ครบก่อน ตรงไหนผมอธิบายตกหล่นหรือพิมพ์ผิด เดี๋ยวกลับมาแก้เพิ่มให้ภายหลังครับ