Sensor

Sensor 是 Agent 的感官中枢：把外部世界的事件（消息、RSS 更新、闹钟到点……）推送给 Agent，并可将 Agent 的回复发回外部渠道。每个 Sensor 都是一个独立的常驻 Python 子进程。 A sensor is the agent's sensory system: it pushes outside events (messages, feed updates, alarms) into the agent and can carry replies back out. Each sensor is its own long-running Python subprocess.

• 由主进程 SensorManager 用 uv run <name>.py 拉起，依赖通过 PEP 723 内联元数据声明（uv 自动建环境）。 The SensorManager starts it with uv run <name>.py; dependencies are declared as PEP 723 inline metadata and uv builds the environment.
• 与主进程通过 stdin/stdout 的单行 JSON-RPC（ACP 方言）双向通信：stdout 只能输出协议 JSON，调试日志必须走 stderr。 It speaks one-line JSON-RPC over stdin/stdout (ACP dialect). stdout carries protocol JSON only; debug output must go to stderr.
• 子进程崩溃会被守护线程拉起；同名 Sensor 已在运行时，会被先整树击杀再重启，避免双实例抢占外部连接。 A watchdog restarts it if it dies. If the same sensor is already running, the old process tree is killed first so two instances never fight over the same external connection.

目录与命名约定Layout and naming

sensors/<sensor-name>/     # 市场仓库中的单元
├── <sensor-name>.py       # 单文件脚本，名称 = 文件夹名
└── config.json            # 市场/配置清单

协议方法Protocol methods

stdio Sensor 固定当前活跃会话——网关在 session/new 时无条件注入 follow 标记。你无需、也不应实现任何会话切换逻辑。 A stdio sensor is pinned to the active session: the gateway injects a follow marker on session/new. Never implement session switching yourself.

开发铁律Hard rules

1. stdout 只能输出协议 JSON（一行一条）。启动时 sys.stdout = sys.stderr，把真实 stdout 另存引用，调试 print 走 stderr。 stdout carries protocol JSON only, one object per line. Reassign sys.stdout = sys.stderr at start-up, keep the real stdout in a reference, and send debug output to stderr.
2. 严禁退出主进程。连接崩溃要无限重连（带 sleep 缓冲）；守护线程只救意外退出。 Never exit. Reconnect in a loop with a sleep backoff; the watchdog is only there for the truly unexpected.
3. 凭证一律从环境变量读取，禁止硬编码密钥。 Read credentials from the environment. Never hard-code them.
4. 有游标 / 登录态的必须持久化到脚本同目录的 state 文件——Sensor 会被热重启，不落盘就丢消息、丢登录态。 Persist cursors and login state to a file next to the script. Sensors restart hot; anything held in memory is lost.
5. 一切外部网络调用放后台线程，主线程只跑 stdin 读取循环。 Put all network calls on a background thread. The main thread only reads stdin.
6. 鉴权求助必须写清凭证名与获取入口（控制台网址 + 步骤）；缺凭证时严禁静默退出或空转，应立刻用 session/prompt 向 Agent 求助。 Ask for credentials properly: name them and say exactly where to get them. When credentials are missing, never exit silently — immediately ask the agent over session/prompt.

最小骨架Minimal skeleton

协议体检（合并硬性判据）覆盖：协议纯净、永不退出、契约自述、依赖可装、握手时序、鉴权自服务、sessionId 回填、故障可见。沙盒根目录自带真网关夹具 evals/gateway_probe.py，凭证留空来逼出鉴权路径。 The merge checklist covers protocol purity, never exiting, a truthful manifest, installable dependencies, handshake order, self-service credentials, sessionId echo, and visible failures. The sandbox root ships a real gateway fixture at evals/gateway_probe.py that runs with empty credentials to force the auth path.

# /// script
# requires-python = ">=3.10"
# dependencies = []
# ///

import sys, json, os, threading

# stdout 让给协议，日志走 stderr
_REAL_STDOUT = sys.stdout
sys.stdout = sys.stderr

SENSOR_NAME = "my-sensor"
MY_TOKEN = os.environ.get("MY_TOKEN", "")   # 必须在 config.json 的 env 里声明

_SID = ""
_SID_READY = threading.Event()
_ID = 2  # 1/2 留给握手


def send(method, params, rid=None):
    global _ID
    if rid is None:
        _ID += 1
        rid = _ID
    _REAL_STDOUT.write(json.dumps(
        {"jsonrpc": "2.0", "id": rid, "method": method, "params": params},
        ensure_ascii=False) + "\n")
    _REAL_STDOUT.flush()


def connect():
    send("initialize", {"clientInfo": {"name": SENSOR_NAME}}, rid=1)
    send("session/new", {"clientInfo": {"name": SENSOR_NAME}}, rid=2)


def prompt(text):
    _SID_READY.wait(timeout=30)
    if _SID:
        send("session/prompt", {"sessionId": _SID,
                                "prompt": [{"type": "text", "text": text}]})


def on_line(line):
    global _SID
    try:
        msg = json.loads(line)
    except Exception:
        return
    if msg.get("id") == 2 and "result" in msg:      # 回填 sessionId
        _SID = (msg["result"] or {}).get("sessionId", "")
        _SID_READY.set()
        return
    upd = (msg.get("params") or {}).get("update") or {}
    if upd.get("sessionUpdate") == "agent_message_chunk":
        text = (upd.get("content") or {}).get("text", "")
        # TODO: 把 Agent 回复转发到外部渠道（放后台线程，别阻塞 stdin）


def external_loop():                                # 后台线程：收外部事件
    if not MY_TOKEN:
        prompt("【my-sensor 需要配置】缺少凭证 MY_TOKEN。请到 <控制台网址> 按步骤创建后，"
               "在前端配置中心 → Sensor 设置里填写并启用。配置后我将自动开始监听。")
        return
    while True:
        # TODO: 长轮询 / WebSocket 收外部消息
        pass


connect()
threading.Thread(target=external_loop, daemon=True).start()
print("[my-sensor] 已启动")

for line in sys.stdin:                              # 主线程只跑 stdin 读取
    if line.strip():
        on_line(line)

安装与启用Install & enable

• 安装：市场接口 /api/tools/market/sensors/install，把 config.json 与 <name>.py 落到 ~/.purrcat/sensor/，并合并进 ~/.purrcat/activate_sensor.json。 Install via /api/tools/market/sensors/install, which writes config.json and <name>.py into ~/.purrcat/sensor/ and merges an entry into ~/.purrcat/activate_sensor.json.
• 启用/禁用与填 env：写 activate_sensor.json 对应条目的 enabled / env。 Enable, disable, and fill in env by editing the matching entry in activate_sensor.json.
• 重载：POST /api/config/sensor/reload（停掉全部旧进程，按配置重新拉起启用项）。 Reload with POST /api/config/sensor/reload: every old process is stopped and the enabled entries are started again.