App Server 握手、双向事件与审批

双向 RPC 不只是输出日志

App Server 为 IDE 或应用宿主管理 thread、turn、item 与审批。客户端既发请求，也可能收到服务端请求，必须回复相同 id。notification 不需要响应。它与 MCP 的 initialize 形状不同：共享 JSON-RPC 风格不代表共享同一协议。

先运行 codex app-server --help 核对本机支持。下面只做 initialize 握手，不启动模型 turn。Python 3.12 保存 handshake.py，机器需已安装 codex：

import json,subprocess,selectors,time
p=subprocess.Popen(['codex','app-server'],stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,text=True,bufsize=1)
sel=selectors.DefaultSelector();sel.register(p.stdout,selectors.EVENT_READ)
try:
    request={'id':1,'method':'initialize','params':{'clientInfo':{
        'name':'coderpath_lab','title':'CoderPath Lab','version':'0.1.0'}}}
    p.stdin.write(json.dumps(request)+'\n');p.stdin.flush()
    deadline=time.monotonic()+10
    while time.monotonic()<deadline:
        if not sel.select(timeout=1):continue
        line=p.stdout.readline()
        if not line:raise RuntimeError('server closed')
        msg=json.loads(line)
        if msg.get('id')==1:
            if 'error' in msg:raise RuntimeError(msg['error'])
            print('initialized',sorted(msg['result']))
            p.stdin.write(json.dumps({'method':'initialized','params':{}})+'\n')
            p.stdin.flush()
            break
    else:raise TimeoutError('initialize deadline')
finally:
    sel.close();p.terminate()
    try:p.wait(timeout=3)
    except subprocess.TimeoutExpired:p.kill();p.wait()

握手成功只证明服务可接收客户端，不证明已登录或模型可用。正式宿主应保留 stderr 到受控诊断日志；这里丢弃仅为缩小握手实验输出。

从 thread 到 turn

握手后依次发送如下协议消息。它们是协议消息示意，不应直接拼接到 shell；宿主需等待前一个响应并提取真实 ID：

{"id":2,"method":"thread/start","params":{"cwd":"/absolute/codex-lab","sandbox":"read-only","approvalPolicy":"never"}}

取返回 result.thread.id 后再发送：

{"id":3,"method":"turn/start","params":{"threadId":"返回的线程ID","input":[{"type":"text","text":"读取 README.md 并总结"}]}}

turn/start 响应不是完成通知。宿主继续消费 item 与 turn 通知，按 threadId/turnId 路由到正确窗口，最终观察 turn/completed。断连后按本机协议支持读取 thread 状态或恢复 thread；没有持久事件游标的版本不能假装可以从任意字节重放。

审批请求必须显示资源、命令、工作目录与参数，决定回传原 request id。多个并行审批不能只存一个全局 pending 变量，否则晚点的决定可能批准另一动作。服务端仍执行权限检查，UI 按钮不能绕过沙箱。

正式协议使用安装版本生成的 schema 核对字段：codex app-server generate-json-schema --help。实验基线和官网最新字段可能不同；新增实验能力要显式协商，不凭看到一个字段名就默认可用。参阅 App Server。

把审批接到真实协议 schema

固定本实验CLI为 @openai/codex@0.153.4（npm install -g @openai/codex@0.153.4）；生成安装版本schema：codex app-server generate-json-schema --out ./schema。Python环境安装 jsonschema==4.26.0。保存dispatcher.py，执行 python dispatcher.py ./schema，不启动模型、不执行任何审批中的命令。

import json,sys
from pathlib import Path
from jsonschema import Draft7Validator
root=Path(sys.argv[1])
params_validator=Draft7Validator(json.loads((root/'CommandExecutionRequestApprovalParams.json').read_text()))
reply_validator=Draft7Validator(json.loads((root/'CommandExecutionRequestApprovalResponse.json').read_text()))
class Host:
    def __init__(self):self.pending={};self.closed=set();self.notifications=[]
    def receive(self,msg):
        if 'method' not in msg: return ('response',msg) # own outbound requests have separate map
        if 'id' not in msg:
            self.notifications.append(msg)
            if msg['method']=='turn/completed':
                p=msg['params'];self.closed.add((p['threadId'],p['turn']['id']))
            return None
        if msg['method']!='item/commandExecution/requestApproval':
            return {'id':msg['id'],'error':{'code':-32601,'message':'unsupported request'}}
        params_validator.validate(msg['params'])
        key=(type(msg['id']).__name__,msg['id'])
        if key in self.pending:raise ValueError('duplicate pending id')
        self.pending[key]=msg['params']
    def resolve(self,id,decision):
        key=(type(id).__name__,id)
        if key not in self.pending:raise ValueError('unknown or resolved id')
        p=self.pending[key]
        if (p['threadId'],p['turnId']) in self.closed:raise ValueError('turn already closed')
        # Restrict this UI to one-shot decisions; session/policy grants need separate UI.
        if decision not in {'accept','decline','cancel'}:raise ValueError('unsupported decision')
        result={'decision':decision};reply_validator.validate(result)
        del self.pending[key]
        return {'id':id,'result':result}
def approval(id,item):
    return {'id':id,'method':'item/commandExecution/requestApproval','params':{
        'itemId':item,'threadId':'thread-1','turnId':'turn-1','startedAtMs':1,
        'command':'cat README.md','cwd':'/tmp/lab'}}
h=Host();h.receive(approval(11,'i1'));h.receive(approval(12,'i2'))
assert h.resolve(11,'decline')=={'id':11,'result':{'decision':'decline'}}
assert ('int',12) in h.pending
try:h.resolve(11,'accept')
except ValueError:pass
else:raise AssertionError('duplicate response')
assert h.resolve(12,'cancel')['result']['decision']=='cancel'
h.receive(approval(13,'i3'))
assert h.receive({'method':'turn/completed','params':{'threadId':'thread-1','turn':{'id':'turn-1'}}}) is None
try:h.resolve(13,'accept')
except ValueError:pass
else:raise AssertionError('stale approval')
print('schema and bidirectional dispatcher passed')

真实服务请求method是item/commandExecution/requestApproval；回复result.decision中decline拒绝该命令但允许turn继续，cancel还请求立即中断turn。批准某项不等于永久改变策略，本小宿主不提供session或策略扩权选项。command、cwd及thread/turn/item应一起展示，决定来源必须是实际操作者，不能由模型消息代填。

fixture使用生成的Params与Response schema验证。完成通知这里只验证宿主路由，不声称是服务真实发出的完整通知；在线宿主还需生成的完整ServerNotification校验、维护自己发出的请求表及线程重连状态。本实验确实验证两个并行待审批、拒绝、取消响应、重复和过期输入；握手实验验证真正的进程传输，两类证据不可混称为真实付费turn审批已跑通。

练习

复用dispatcher.py，构造整数21与字符串"21"两条真实形状请求。拒绝整数21后，字符串请求应仍待处理，并可用cancel答复；再次处理整数21及未知999均应失败。解释为什么不能把所有request id转成字符串。

先记录自己的输出与解释，再核对本章答案。